perf(database): add contact-tag/embedding indexes and orphan-attachment cleanup

apps/worker/__tests__/purge-orphaned-attachments.test.ts

@@ -0,0 +1,60 @@
+import { beforeEach, describe, expect, test, vi } from "vitest"
+
+const purgeOrphanedAttachments = vi.fn()
+const runExclusive = vi.fn(async ({ fn }: { fn: () => Promise<unknown> }) =>
+  fn(),
+)
+const lockExists = vi.fn()
+const info = vi.fn()
+
+vi.mock("@chatbotx.io/business", () => ({
+  messageCleanupService: { purgeOrphanedAttachments },
+}))
+vi.mock("@chatbotx.io/redis", () => ({
+  distributedLock: { runExclusive },
+  distributedStore: { exists: lockExists },
+}))
+vi.mock("@chatbotx.io/logger", () => ({
+  getChildLogger: () => ({ info, warn: vi.fn() }),
+}))
+
+const { purgeOrphanedAttachments: handlePurgeOrphanedAttachments } =
+  await import("../src/schedule/handlers/purge-orphaned-attachments")
+
+beforeEach(() => {
+  purgeOrphanedAttachments.mockReset()
+  purgeOrphanedAttachments.mockResolvedValue(0)
+  runExclusive.mockClear()
+  info.mockReset()
+})
+
+describe("purgeOrphanedAttachments", () => {
+  test("deletes one bounded orphan batch under the distributed lock", async () => {
+    await handlePurgeOrphanedAttachments()
+
+    expect(runExclusive).toHaveBeenCalledWith(
+      expect.objectContaining({
+        key: "schedule:purge-orphaned-attachments",
+        timeoutInSeconds: 60 * 60,
+      }),
+    )
+    expect(purgeOrphanedAttachments).toHaveBeenCalledWith({ limit: 1000 })
+  })
+
+  test("logs the deleted count when orphaned attachments are purged", async () => {
+    purgeOrphanedAttachments.mockResolvedValue(2)
+
+    await handlePurgeOrphanedAttachments()
+
+    expect(info).toHaveBeenCalledWith(
+      { deleted: 2 },
+      "purgeOrphanedAttachments: orphaned attachments purged",
+    )
+  })
+
+  test("does not log when no orphaned attachments exist", async () => {
+    await handlePurgeOrphanedAttachments()
+
+    expect(info).not.toHaveBeenCalled()
+  })
+})

apps/worker/src/schedule/handlers/purge-orphaned-attachments.ts

@@ -0,0 +1,68 @@
+import { messageCleanupService } from "@chatbotx.io/business"
+import { getChildLogger } from "@chatbotx.io/logger"
+import { distributedLock, distributedStore } from "@chatbotx.io/redis"
+
+const LOCK_KEY = "schedule:purge-orphaned-attachments"
+const log = getChildLogger("purge-orphaned-attachments")
+const BATCH_SIZE = 1000
+const LOCK_TTL_SECONDS = 60 * 60
+const LOCK_ACQUIRE_RETRY_SECONDS = 5
+let isPurgeOrphanedAttachmentsRunning = false
+
+export async function purgeOrphanedAttachments(): Promise<void> {
+  if (isPurgeOrphanedAttachmentsRunning) {
+    log.warn(
+      "purgeOrphanedAttachments: skipped because a local run is still in progress",
+    )
+    return
+  }
+
+  isPurgeOrphanedAttachmentsRunning = true
+  try {
+    await distributedLock.runExclusive({
+      key: LOCK_KEY,
+      timeoutInSeconds: LOCK_TTL_SECONDS,
+      retryTimeoutInSeconds: LOCK_ACQUIRE_RETRY_SECONDS,
+      fn: async () => {
+        const deleted = await messageCleanupService.purgeOrphanedAttachments({
+          limit: BATCH_SIZE,
+        })
+
+        if (deleted > 0) {
+          log.info(
+            { deleted },
+            "purgeOrphanedAttachments: orphaned attachments purged",
+          )
+        }
+      },
+    })
+  } catch (err) {
+    if (
+      isLockAcquisitionFailure(err) &&
+      (await distributedStore.exists(LOCK_KEY))
+    ) {
+      log.warn(
+        { err },
+        "purgeOrphanedAttachments: skipped because another run still holds the lock",
+      )
+      return
+    }
+
+    throw err
+  } finally {
+    isPurgeOrphanedAttachmentsRunning = false
+  }
+}
+
+function isLockAcquisitionFailure(err: unknown): boolean {
+  return (
+    typeof err === "object" &&
+    err !== null &&
+    "name" in err &&
+    "code" in err &&
+    "key" in err &&
+    err.name === "LockAcquisitionError" &&
+    err.code === "LOCK_ACQUISITION_FAILED" &&
+    err.key === LOCK_KEY
+  )
+}

apps/worker/src/schedule/handlers/register-schedules.ts

@@ -269,6 +269,22 @@ export const registerSchedules = async () => {
       },
     },
   )
+  // Orphan cleanup is a reconciliation path for the Message/Attachment
+  // hypertables, which cannot enforce their parent FK. Keep it apart from the
+  // other retention sweeps to avoid concurrent compressed-chunk deletes.
+  await scheduleQueue.upsertJobScheduler(
+    ScheduleJobData.purgeOrphanedAttachments,
+    {
+      pattern: "30 3 * * *",
+    },
+    {
+      name: ScheduleJobData.purgeOrphanedAttachments,
+      data: {
+        type: ScheduleJobData.purgeOrphanedAttachments,
+        data: {},
+      },
+    },
+  )
 
   // Same "retention applies to every edition" reasoning as `purgeErrorLogs`
   // above; offset 15 minutes so the two chunked deletes do not contend.

apps/worker/src/schedule/worker.ts

@@ -28,6 +28,7 @@ import { purgeBroadcasts } from "./handlers/purge-broadcasts"
 import { purgeCoexistStaging } from "./handlers/purge-coexist-staging"
 import { purgeCommentAutomationEvents } from "./handlers/purge-comment-automation-events"
 import { purgeErrorLogs } from "./handlers/purge-error-logs"
+import { purgeOrphanedAttachments } from "./handlers/purge-orphaned-attachments"
 import { purgeWhatsappSignupSessions } from "./handlers/purge-whatsapp-signup-sessions"
 import { purgeWorkspaces } from "./handlers/purge-workspaces"
 import { reconcileBroadcasts } from "./handlers/reconcile-broadcasts"
@@ -171,6 +172,10 @@ async function startScheduleWorker() {
               await purgeCommentAutomationEvents()
               return
 
+            case ScheduleJobData.purgeOrphanedAttachments:
+              await purgeOrphanedAttachments()
+              return
+
             case ScheduleJobData.refreshChannelTokens:
               await refreshChannelTokens(job.data.data.channels)
               return

packages/business/src/message-cleanup/service.ts

@@ -40,6 +40,7 @@ const byInboxSourceKey = (
 
 const PROCESS_DEFAULT_LIMIT = 100
 const CONVERSATION_DELETE_BATCH_SIZE = 100
+const ORPHANED_ATTACHMENT_BATCH_SIZE = 1000
 
 /**
  * Tracks Message/Attachment rows orphaned by contact deletes.
@@ -206,6 +207,61 @@ class MessageCleanupService extends BaseService {
     return { processed, failed }
   }
 
+  /**
+   * Deletes one bounded batch of attachments whose parent Message no longer
+   * exists. Attachment cannot have an FK to the Message hypertable, so this
+   * reconciles rows left behind by deletes outside the contact tombstone flow.
+   */
+  async purgeOrphanedAttachments(props?: { limit?: number }): Promise<number> {
+    const limit = props?.limit ?? ORPHANED_ATTACHMENT_BATCH_SIZE
+    const result = await db.transaction(async (tx) => {
+      await liftDecompressionLimit(tx)
+
+      return tx.execute<{
+        originPath: string
+        thumbnailPath: string | null
+      }>(sql`
+        WITH orphaned AS (
+          SELECT attachment."id", attachment."createdAt"
+          FROM "Attachment" AS attachment
+          WHERE NOT EXISTS (
+            SELECT 1
+            FROM "Message" AS message
+            WHERE message."id" = attachment."messageId"
+              AND message."createdAt" = attachment."messageCreatedAt"
+          )
+          ORDER BY attachment."createdAt" ASC
+          LIMIT ${limit}
+        )
+        DELETE FROM "Attachment" AS attachment
+        USING orphaned
+        WHERE attachment."id" = orphaned."id"
+          AND attachment."createdAt" = orphaned."createdAt"
+        RETURNING attachment."originPath", attachment."thumbnailPath"
+      `)
+    })
+
+    const deleteResults = await Promise.allSettled(
+      result.rows
+        .flatMap((attachment) =>
+          attachment.thumbnailPath
+            ? [attachment.originPath, attachment.thumbnailPath]
+            : [attachment.originPath],
+        )
+        .map((path) => uploader.deleteObject(path)),
+    )
+    for (const deleteResult of deleteResults) {
+      if (deleteResult.status === "rejected") {
+        logger.warn(
+          { err: deleteResult.reason },
+          "Orphaned attachment file deletion failed",
+        )
+      }
+    }
+
+    return result.rows.length
+  }
+
   private async purgeRow(row: MessageCleanupModel): Promise<void> {
     const attachmentPaths: string[] = []
 

packages/database/drizzle/20260916062120_add_contact_tag_and_embedding_indexes/migration.sql

@@ -0,0 +1,2 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "AIEmbedding_embedding_idx" ON "AIEmbedding" USING hnsw ("embedding" vector_cosine_ops);--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "ContactToTag_tagId_contactId_idx" ON "ContactToTag" ("tagId","contactId");