Add search_queries cleanup job to match event log retention (#13035)

* Add search_query cleanup job to match event log retention

* fix: findAllInBatches to use key based pagination rather than offset

server/commands/teamPermanentDeleter.ts

@@ -42,33 +42,32 @@ async function teamPermanentDeleter(team: Team) {
 
   // Attachments are destroyed as individual instances (rather than a bulk
   // delete) so the BeforeDestroy hook runs and removes the associated file from
-  // storage. We cannot use findAllInBatches with an advancing offset here –
-  // deleting a batch shifts the remaining rows backwards, so advancing the
-  // offset would skip records and leave attachments that still reference the
-  // team, causing a foreign key violation when the team itself is destroyed.
-  // Instead we repeatedly fetch and delete the first batch until none remain.
-  let attachments: Attachment[];
-  do {
-    attachments = await Attachment.findAll<Attachment>({
+  // storage.
+  await Attachment.findAllInBatches<Attachment>(
+    {
       where: {
         teamId,
       },
-      limit: 100,
-    });
-
-    if (attachments.length > 0) {
-      await sequelize.transaction(async (transaction) => {
-        Logger.info("commands", `Deleting ${attachments.length} attachments…`);
-        await Promise.all(
-          attachments.map((attachment) =>
-            attachment.destroy({
-              transaction,
-            })
-          )
-        );
-      });
+      batchLimit: 100,
+    },
+    async (attachments) => {
+      if (attachments.length > 0) {
+        await sequelize.transaction(async (transaction) => {
+          Logger.info(
+            "commands",
+            `Deleting ${attachments.length} attachments…`
+          );
+          await Promise.all(
+            attachments.map((attachment) =>
+              attachment.destroy({
+                transaction,
+              })
+            )
+          );
+        });
+      }
     }
-  } while (attachments.length > 0);
+  );
 
   // Destroy user-relation models
   await User.findAllInBatches<User>(

server/migrations/20260718134528-add-search-queries-created-at-index.js

@@ -0,0 +1,16 @@
+"use strict";
+
+/** @type {import('sequelize-cli').Migration} */
+module.exports = {
+  async up(queryInterface) {
+    await queryInterface.sequelize.query(
+      'CREATE INDEX CONCURRENTLY IF NOT EXISTS "search_queries_created_at" ON "search_queries" ("createdAt");'
+    );
+  },
+
+  async down(queryInterface) {
+    await queryInterface.sequelize.query(
+      'DROP INDEX CONCURRENTLY IF EXISTS "search_queries_created_at";'
+    );
+  },
+};

server/models/base/Model.test.ts

@@ -1,7 +1,12 @@
 import { faker } from "@faker-js/faker";
 import { randomUUID } from "node:crypto";
 import { CommentingAccess, TeamPreference } from "@shared/types";
-import { buildDocument, buildTeam } from "@server/test/factories";
+import {
+  buildDocument,
+  buildSearchQuery,
+  buildTeam,
+} from "@server/test/factories";
+import SearchQuery from "../SearchQuery";
 import User from "../User";
 
 describe("Model", () => {
@@ -96,5 +101,59 @@ describe("Model", () => {
       expect(usersBatch[0].length).toEqual(2);
       expect(usersBatch[1].length).toEqual(2);
     });
+
+    it("should not skip records when the callback deletes them", async () => {
+      const team = await buildTeam();
+      await Promise.all(
+        [...Array(5)].map(() => buildSearchQuery({ teamId: team.id }))
+      );
+
+      const total = await SearchQuery.findAllInBatches<SearchQuery>(
+        {
+          attributes: ["id"],
+          where: { teamId: team.id },
+          order: [["createdAt", "ASC"]],
+          batchLimit: 2,
+        },
+        async (searchQueries) => {
+          await SearchQuery.destroy({
+            where: { id: searchQueries.map((searchQuery) => searchQuery.id) },
+          });
+        }
+      );
+
+      expect(total).toEqual(5);
+      expect(await SearchQuery.count({ where: { teamId: team.id } })).toEqual(
+        0
+      );
+    });
+
+    it("should not skip or repeat records when ordering by a non-unique column", async () => {
+      const team = await buildTeam();
+      await User.bulkCreate(
+        [...Array(10)].map(() => ({
+          email: faker.internet.email().toLowerCase(),
+          name: faker.person.fullName(),
+          teamId: team.id,
+        }))
+      );
+
+      const seen: string[] = [];
+
+      await User.findAllInBatches<User>(
+        {
+          attributes: ["id"],
+          where: { teamId: team.id },
+          order: [["createdAt", "ASC"]],
+          batchLimit: 3,
+        },
+        async (foundUsers) => {
+          seen.push(...foundUsers.map((user) => user.id));
+        }
+      );
+
+      expect(seen.length).toEqual(10);
+      expect(new Set(seen).size).toEqual(10);
+    });
   });
 });

server/models/base/Model.ts

@@ -10,7 +10,7 @@ import type {
   NonAttribute,
   SaveOptions,
 } from "sequelize";
-import { DataTypes, UniqueConstraintError } from "sequelize";
+import { DataTypes, Op, UniqueConstraintError } from "sequelize";
 import {
   AfterCreate,
   AfterDestroy,
@@ -344,31 +344,85 @@ class Model<
    * @return The total number of results processed.
    */
   static async findAllInBatches<T extends Model>(
-    query: Replace<FindOptions<T>, "limit", "batchLimit"> & {
+    query: Omit<Replace<FindOptions<T>, "limit", "batchLimit">, "order"> & {
       /** The maximum number of results to return, after which the query will stop. */
       totalLimit?: number;
+      /** The order of results, as plain [column, direction] pairs. */
+      order?: Array<[string, "ASC" | "DESC"]>;
     },
     callback: (results: Array<T>, query: FindOptions<T>) => Promise<void>
   ): Promise<number> {
-    let total = 0;
-    const mappedQuery = {
-      ...query,
-      offset: query.offset ?? 0,
-      limit: query.batchLimit ?? 10,
-    };
+    const { batchLimit = 10, totalLimit = Infinity, offset, ...rest } = query;
 
-    let results;
+    // A raw query with an explicit order is trusted to be unique (it may
+    // select aggregates where the primary key is unavailable), otherwise
+    // append the primary key column(s) to guarantee a total, stable order.
+    const order = [...(rest.order ?? [])];
+    if (!rest.raw || order.length === 0) {
+      for (const pk of this.primaryKeyAttributes) {
+        if (!order.some(([column]) => column === pk)) {
+          order.push([pk, "ASC"]);
+        }
+      }
+    }
+
+    // The cursor columns must be selected for their values to be read back,
+    // so add any missing ones when the caller restricts attributes.
+    let attributes = rest.attributes;
+    if (
+      Array.isArray(attributes) &&
+      attributes.every((attr): attr is string => typeof attr === "string")
+    ) {
+      attributes = [
+        ...new Set([...attributes, ...order.map(([column]) => column)]),
+      ];
+    }
+
+    const cursorValue = (row: T, column: string) =>
+      typeof row.get === "function"
+        ? row.get(column)
+        : (row as Record<string, unknown>)[column];
+
+    const buildCursor = (lastRow: T) => ({
+      [Op.or]: order.map(([column, direction], index) => {
+        const clause: Record<string, unknown> = {};
+        for (const [prevColumn] of order.slice(0, index)) {
+          clause[prevColumn] = { [Op.eq]: cursorValue(lastRow, prevColumn) };
+        }
+        clause[column] = {
+          [direction === "DESC" ? Op.lt : Op.gt]: cursorValue(lastRow, column),
+        };
+        return clause;
+      }),
+    });
+
+    let cursor: ReturnType<typeof buildCursor> | undefined;
+    let currentOffset = offset ?? 0;
+    let total = 0;
+    let limit: number;
+    let results: T[];
 
     do {
+      limit = Math.min(batchLimit, totalLimit - total);
+      const findOptions: FindOptions<T> = {
+        ...rest,
+        attributes,
+        order,
+        limit,
+        offset: currentOffset,
+        where: cursor ? { [Op.and]: [rest.where ?? {}, cursor] } : rest.where,
+      };
       // @ts-expect-error this T
-      results = await this.findAll<T>(mappedQuery);
+      results = await this.findAll<T>(findOptions);
       total += results.length;
-      await callback(results, mappedQuery);
-      mappedQuery.offset += mappedQuery.limit;
-    } while (
-      results.length >= mappedQuery.limit &&
-      (mappedQuery.totalLimit ?? Infinity) > mappedQuery.offset
-    );
+      await callback(results, findOptions);
+
+      if (results.length > 0) {
+        cursor = buildCursor(results[results.length - 1]);
+      }
+      // The cursor supersedes offset-based pagination after the first batch.
+      currentOffset = 0;
+    } while (results.length >= limit && total < totalLimit);
 
     return total;
   }

server/queues/tasks/CleanupOldSearchQueriesTask.test.ts

@@ -0,0 +1,29 @@
+import { subDays } from "date-fns";
+import { SearchQuery } from "@server/models";
+import { buildSearchQuery, buildTeam } from "@server/test/factories";
+import CleanupOldSearchQueriesTask from "./CleanupOldSearchQueriesTask";
+
+describe("CleanupOldSearchQueriesTask", () => {
+  it("deletes search queries older than the retention window", async () => {
+    const team = await buildTeam();
+    const old = await buildSearchQuery({
+      teamId: team.id,
+      createdAt: subDays(new Date(), 400),
+    });
+    const recent = await buildSearchQuery({
+      teamId: team.id,
+      createdAt: subDays(new Date(), 5),
+    });
+
+    await new CleanupOldSearchQueriesTask().perform({
+      limit: 10000,
+      partition: {
+        partitionIndex: 0,
+        partitionCount: 1,
+      },
+    });
+
+    expect(await SearchQuery.findByPk(old.id)).toBeNull();
+    expect(await SearchQuery.findByPk(recent.id)).not.toBeNull();
+  });
+});

server/queues/tasks/CleanupOldSearchQueriesTask.ts

@@ -0,0 +1,64 @@
+import { subDays } from "date-fns";
+import { Op } from "sequelize";
+import Logger from "@server/logging/Logger";
+import { SearchQuery } from "@server/models";
+import { TaskPriority } from "./base/BaseTask";
+import type { Props } from "./base/CronTask";
+import { CronTask, TaskInterval } from "./base/CronTask";
+import { Minute } from "@shared/utils/time";
+
+export default class CleanupOldSearchQueriesTask extends CronTask {
+  public async perform({ partition }: Props) {
+    // TODO: Hardcoded right now, configurable later
+    const retentionDays = 365;
+    const cutoffDate = subDays(new Date(), retentionDays);
+    const maxSearchQueriesPerTask = 100000;
+    let totalSearchQueriesDeleted = 0;
+
+    try {
+      await SearchQuery.findAllInBatches(
+        {
+          attributes: ["id"],
+          where: {
+            createdAt: {
+              [Op.lt]: cutoffDate,
+            },
+            ...this.getPartitionWhereClause("id", partition),
+          },
+          batchLimit: 1000,
+          totalLimit: maxSearchQueriesPerTask,
+          order: [["createdAt", "ASC"]],
+        },
+        async (searchQueries) => {
+          totalSearchQueriesDeleted += await SearchQuery.destroy({
+            where: {
+              id: {
+                [Op.in]: searchQueries.map((searchQuery) => searchQuery.id),
+              },
+            },
+          });
+        }
+      );
+    } finally {
+      if (totalSearchQueriesDeleted > 0) {
+        Logger.info("task", `Deleted old search queries`, {
+          totalSearchQueriesDeleted,
+        });
+      }
+    }
+  }
+
+  public get cron() {
+    return {
+      interval: TaskInterval.Hour,
+      partitionWindow: 15 * Minute.ms,
+    };
+  }
+
+  public get options() {
+    return {
+      attempts: 1,
+      priority: TaskPriority.Background,
+    };
+  }
+}

server/queues/tasks/UpdateTeamsAttachmentsSizeTask.ts

@@ -30,6 +30,9 @@ export default class UpdateTeamsAttachmentsSizeTask extends CronTask {
             [Op.gt]: subDays(new Date(), 1),
           },
         },
+        // A unique order is required for keyset pagination; the primary key
+        // cannot be used with a DISTINCT select.
+        order: [["teamId", "ASC"]],
         batchLimit: 100,
         raw: true,
       },