cleanup of dedupe keys lol

Author: n1ru4l

packages/migrations/src/actions/2025.12.12T00-00-00.workflows-deduplication.ts

@@ -9,5 +9,9 @@ export default {
           "expires_at"  timestamptz NOT NULL,
           CONSTRAINT "dedupe_pk" PRIMARY KEY ("task_name", "dedupe_key")
       );
+
+      CREATE INDEX "graphile_worker_deduplication_expires_at_idx"
+        ON "graphile_worker_deduplication" ("expires_at")
+      ;
   `,
 } satisfies MigrationExecutor;

packages/services/storage/src/db/types.ts

@@ -135,6 +135,12 @@ export interface document_preflight_scripts {
   updated_at: Date;
 }
 
+export interface graphile_worker_deduplication {
+  dedupe_key: string;
+  expires_at: Date;
+  task_name: string;
+}
+
 export interface migration {
   date: Date;
   hash: string;
@@ -441,6 +447,7 @@ export interface DBTables {
   document_collection_documents: document_collection_documents;
   document_collections: document_collections;
   document_preflight_scripts: document_preflight_scripts;
+  graphile_worker_deduplication: graphile_worker_deduplication;
   migration: migration;
   oidc_integrations: oidc_integrations;
   organization_access_tokens: organization_access_tokens;

packages/services/workflows/src/index.ts

@@ -35,6 +35,7 @@ const modules = await Promise.all([
   import('./tasks/organization-invitation.js'),
   import('./tasks/organization-ownership-transfer.js'),
   import('./tasks/password-reset.js'),
+  import('./tasks/purge-expired-dedupe-keys.js'),
   import('./tasks/purge-expired-schema-checks.js'),
   import('./tasks/schema-change-notification.js'),
   import('./tasks/usage-rate-limit-exceeded.js'),
@@ -44,6 +45,8 @@ const modules = await Promise.all([
 const crontab = `
   # Purge expired schema checks every Sunday at 10:00AM
   0 10 * * 0 purgeExpiredSchemaChecks
+  # Every day at 3:00 AM
+  0 3 * * * purgeExpiredDedupeKeys
 `;
 
 const pg = await createPool(env.postgres.connectionString);

packages/services/workflows/src/tasks/purge-expired-dedupe-keys.ts

@@ -0,0 +1,25 @@
+import { sql } from 'slonik';
+import { z } from 'zod';
+import { defineTask, implementTask } from '../kit.js';
+
+export const PurgeExpiredDedupeKeysTask = defineTask({
+  name: 'purgeExpiredDedupeKeys',
+  schema: z.unknown(),
+});
+
+export const task = implementTask(PurgeExpiredDedupeKeysTask, async args => {
+  args.logger.debug('purging expired postgraphile task dedupe keys');
+  const result = await args.context.pg.oneFirst(sql`
+      WITH "deleted" AS (
+        DELETE FROM "graphile_worker_deduplication"
+        WHERE "expires_at" < NOW()
+          RETURNING 1
+      )
+      SELECT COUNT(*) FROM "deleted";
+  `);
+  const amount = z.number().parse(result);
+  args.logger.debug(
+    { purgedCount: amount },
+    'finished purging expired postgraphile task dedupe keys',
+  );
+});