From 80258914f224f8dbf56a8bdfa739466767709749 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 19 Jul 2026 11:09:57 +0200 Subject: [PATCH] feat(documents): catch up unread documents when auto-read is switched on MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-document summary job is enqueued at upload time and no-ops while the documentsAutoAiRead opt-in is OFF. Nothing ever re-enqueued it, so the switch only applied to documents uploaded AFTER the flip — a user who filled the vault first and opted in later had documents that would never be read, and the setting looked inert. A genuine OFF->ON transition now schedules one bounded catch-up pass over the user's un-summarised documents. It grants nothing of its own: it only enqueues the ordinary summary job, which still re-reads the opt-in, re-asserts the per-document egress consent and reserves the daily token budget. The pass itself re-reads the opt-in first, so a toggle flipped back OFF before it runs egresses nothing. Bounded by an id-cursor walk capped at 200 documents per pass (a summary is a provider call, unlike the local-compute thumbnail backfill). Idempotent at four levels: the trigger fires only on a real transition, the per-user singleton collapses a double toggle, the candidate set excludes summarised documents, and the summary job skips one that already has a summary. --- .../__tests__/route.test.ts | 128 +++++++++++++ .../auth/me/documents-auto-ai-read/route.ts | 17 +- .../document-summary-catchup.test.ts | 171 ++++++++++++++++++ src/lib/jobs/document-summary-catchup.ts | 164 +++++++++++++++++ src/lib/jobs/reminder/register-maintenance.ts | 44 +++++ 5 files changed, 523 insertions(+), 1 deletion(-) create mode 100644 src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts create mode 100644 src/lib/jobs/__tests__/document-summary-catchup.test.ts create mode 100644 src/lib/jobs/document-summary-catchup.ts diff --git a/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts b/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts new file mode 100644 index 000000000..68d2a90f6 --- /dev/null +++ b/src/app/api/auth/me/documents-auto-ai-read/__tests__/route.test.ts @@ -0,0 +1,128 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +/** + * The auto-read opt-in endpoint, and specifically the catch-up it schedules. + * + * The summary job is enqueued at UPLOAD time and no-ops while the flag is OFF, + * so without a catch-up on the flip the toggle only ever applied to future + * uploads — a user who filled the vault first saw the switch do nothing. These + * tests pin that a genuine OFF→ON transition schedules the pass, and that a + * no-op re-save or a flip to OFF does not. + */ + +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: vi.fn(), update: vi.fn() } }, +})); +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/auth/audit", () => ({ + auditLog: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/consent/web-grant", () => ({ + ensureWebAiConsentReceipt: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/document-summary-catchup", () => ({ + enqueueSummaryCatchUp: vi.fn().mockResolvedValue({ enqueued: true }), +})); +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn().mockResolvedValue({ + allowed: true, + remaining: 59, + resetAt: Date.now() + 60_000, + }), + rateLimitHeaders: () => ({}), +})); +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { PATCH } from "../route"; +import { prisma } from "@/lib/db"; +import { getSession } from "@/lib/auth/session"; +import { ensureWebAiConsentReceipt } from "@/lib/consent/web-grant"; +import { enqueueSummaryCatchUp } from "@/lib/jobs/document-summary-catchup"; + +const SESSION_OK = { + session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "testuser", role: "USER" as const }, +}; + +function mkPatch(documentsAutoAiRead: boolean): Request { + return new Request("http://localhost/api/auth/me/documents-auto-ai-read", { + method: "PATCH", + body: JSON.stringify({ documentsAutoAiRead }), + headers: { "Content-Type": "application/json" }, + }); +} + +/** Seed the flag's value BEFORE the PATCH under test. */ +function withPrevious(documentsAutoAiRead: boolean) { + vi.mocked(prisma.user.findUnique).mockResolvedValue({ + documentsAutoAiRead, + } as never); +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + vi.mocked(prisma.user.update).mockResolvedValue({} as never); +}); + +describe("PATCH /api/auth/me/documents-auto-ai-read — catch-up scheduling", () => { + it("schedules a catch-up when the opt-in flips OFF→ON", async () => { + withPrevious(false); + + const res = await PATCH(mkPatch(true)); + + expect(res.status).toBe(200); + expect(enqueueSummaryCatchUp).toHaveBeenCalledWith("user-1"); + expect(enqueueSummaryCatchUp).toHaveBeenCalledTimes(1); + }); + + it("does not re-schedule when the opt-in was already ON", async () => { + // Idempotency at the trigger: re-saving an already-ON setting must not + // queue a second pass over the same documents. + withPrevious(true); + + await PATCH(mkPatch(true)); + + expect(enqueueSummaryCatchUp).not.toHaveBeenCalled(); + }); + + it("does not schedule anything when the opt-in is turned OFF", async () => { + withPrevious(true); + + await PATCH(mkPatch(false)); + + expect(enqueueSummaryCatchUp).not.toHaveBeenCalled(); + expect(ensureWebAiConsentReceipt).not.toHaveBeenCalled(); + }); + + it("still mints the consent receipt on the flip that schedules the pass", async () => { + // The catch-up rides the same act of consent, never around it. + withPrevious(false); + + await PATCH(mkPatch(true)); + + expect(ensureWebAiConsentReceipt).toHaveBeenCalledWith("user-1"); + }); + + it("persists the flag field-by-field", async () => { + withPrevious(false); + + await PATCH(mkPatch(true)); + + expect(prisma.user.update).toHaveBeenCalledWith({ + where: { id: "user-1" }, + data: { documentsAutoAiRead: true }, + }); + }); +}); diff --git a/src/app/api/auth/me/documents-auto-ai-read/route.ts b/src/app/api/auth/me/documents-auto-ai-read/route.ts index 4bd8f1a62..20cd90c06 100644 --- a/src/app/api/auth/me/documents-auto-ai-read/route.ts +++ b/src/app/api/auth/me/documents-auto-ai-read/route.ts @@ -12,7 +12,10 @@ * Flipping it ON is itself the standing consent act, so the write also mints an * append-only `ai_full` consent receipt (`ensureWebAiConsentReceipt`) — the * durable audit record that sits alongside the runtime short-circuit the - * document consent gate reads. Mirrors `auth/me/labs-local-ocr`: 60/min rate + * document consent gate reads. An OFF→ON flip additionally schedules a bounded + * catch-up over the documents already stored (`enqueueSummaryCatchUp`), because + * the summary job is enqueued at upload time and would otherwise never revisit + * a vault filled before the opt-in. Mirrors `auth/me/labs-local-ocr`: 60/min rate * limit, Zod `safeParse` → 422 via `returnAllZodIssues`, audit-log row, * field-by-field write (no mass assignment). Idempotent — always returns the * resolved next state so the client can hard-set the optimistic update. @@ -31,6 +34,7 @@ import { annotate } from "@/lib/logging/context"; import { auditLog } from "@/lib/auth/audit"; import { ensureWebAiConsentReceipt } from "@/lib/consent/web-grant"; import { prisma } from "@/lib/db"; +import { enqueueSummaryCatchUp } from "@/lib/jobs/document-summary-catchup"; import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit"; const patchBodySchema = z.object({ @@ -110,6 +114,17 @@ export const PATCH = apiHandler(async (req: Request) => { await ensureWebAiConsentReceipt(user.id); } + // A genuine OFF→ON flip schedules a catch-up over the documents already in + // the vault. Without it the switch only ever applied to FUTURE uploads: the + // summary job is enqueued at upload time and no-ops while the flag is OFF, so + // a user who uploaded first and opted in later saw the toggle do nothing. + // Fire-and-forget and bounded; the pass only enqueues, and every consent and + // budget gate still runs per document inside the summary job itself. + const wasEnabled = previous?.documentsAutoAiRead ?? false; + if (next && !wasEnabled) { + void enqueueSummaryCatchUp(user.id); + } + await auditLog("user.documentsAutoAiRead.update", { userId: user.id, ipAddress: getClientIp(req), diff --git a/src/lib/jobs/__tests__/document-summary-catchup.test.ts b/src/lib/jobs/__tests__/document-summary-catchup.test.ts new file mode 100644 index 000000000..e3666f4d4 --- /dev/null +++ b/src/lib/jobs/__tests__/document-summary-catchup.test.ts @@ -0,0 +1,171 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +/** + * The auto-read catch-up pass. + * + * Documents are summarised by a job enqueued at UPLOAD time that no-ops while + * the `documentsAutoAiRead` opt-in is OFF, so a vault filled before the flip was + * never read and the toggle appeared to do nothing. These tests pin the catch-up + * that closes that hole, and the three properties it must not lose: it stays + * bounded, it stays idempotent, and it grants no consent or budget of its own. + */ + +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/db", () => ({ + prisma: { inboundDocument: { findMany: vi.fn() } }, +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: vi.fn() })); +vi.mock("@/lib/documents/document-settings", () => ({ + documentAutoReadEnabled: vi.fn(), +})); +vi.mock("@/lib/jobs/document-summary", () => ({ + enqueueDocumentSummary: vi.fn(), +})); + +import { + enqueueSummaryCatchUp, + runSummaryCatchUpForUser, + MAX_ENQUEUES_PER_RUN, + DOCUMENT_SUMMARY_CATCHUP_QUEUE, +} from "../document-summary-catchup"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { documentAutoReadEnabled } from "@/lib/documents/document-settings"; +import { enqueueDocumentSummary } from "@/lib/jobs/document-summary"; +import { annotate } from "@/lib/logging/context"; + +const findMany = vi.mocked(prisma.inboundDocument.findMany); +const mockEnqueueSummary = vi.mocked(enqueueDocumentSummary); +const mockAutoRead = vi.mocked(documentAutoReadEnabled); + +/** Serve `total` document ids across the job's id-cursor paged walk. */ +function serveDocuments(total: number) { + const ids = Array.from({ length: total }, (_, i) => ({ + id: `doc-${String(i).padStart(5, "0")}`, + })); + findMany.mockImplementation((async (args: { + take: number; + cursor?: { id: string }; + }) => { + const start = args.cursor + ? ids.findIndex((d) => d.id === args.cursor!.id) + 1 + : 0; + return ids.slice(start, start + args.take); + }) as unknown as typeof findMany); +} + +beforeEach(() => { + vi.clearAllMocks(); + mockAutoRead.mockResolvedValue(true); + mockEnqueueSummary.mockResolvedValue({ enqueued: true }); +}); + +describe("runSummaryCatchUpForUser", () => { + it("enqueues a summary for every already-stored un-summarised document", async () => { + serveDocuments(3); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result).toEqual({ enqueued: 3, capped: false }); + expect(mockEnqueueSummary).toHaveBeenCalledTimes(3); + expect(mockEnqueueSummary).toHaveBeenCalledWith("user-1", "doc-00000"); + expect(mockEnqueueSummary).toHaveBeenCalledWith("user-1", "doc-00002"); + }); + + it("only considers documents that have no summary yet", async () => { + // Idempotency floor: a re-run cannot redo finished work because a + // summarised document is not in the candidate set at all. + serveDocuments(1); + + await runSummaryCatchUpForUser("user-1"); + + const where = findMany.mock.calls[0]![0]!.where; + expect(where).toMatchObject({ + userId: "user-1", + deletedAt: null, + summaryEncrypted: null, + }); + }); + + it("stops at the documented cap instead of queueing a whole vault", async () => { + serveDocuments(MAX_ENQUEUES_PER_RUN + 50); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result.enqueued).toBe(MAX_ENQUEUES_PER_RUN); + expect(result.capped).toBe(true); + expect(mockEnqueueSummary).toHaveBeenCalledTimes(MAX_ENQUEUES_PER_RUN); + }); + + it("re-reads the opt-in and does nothing when it was flipped back OFF", async () => { + // Consent race: the PATCH that scheduled the pass is not authority enough. + serveDocuments(5); + mockAutoRead.mockResolvedValue(false); + + const result = await runSummaryCatchUpForUser("user-1"); + + expect(result).toEqual({ enqueued: 0, capped: false }); + expect(mockEnqueueSummary).not.toHaveBeenCalled(); + expect(findMany).not.toHaveBeenCalled(); + }); + + it("routes work through the ordinary summary job, granting nothing itself", async () => { + // The per-document job is what re-asserts egress consent and reserves the + // daily budget. The catch-up must not have its own provider or budget path. + serveDocuments(2); + + await runSummaryCatchUpForUser("user-1"); + + expect(mockEnqueueSummary).toHaveBeenCalledTimes(2); + for (const call of mockEnqueueSummary.mock.calls) { + expect(call[0]).toBe("user-1"); + } + }); + + it("emits the catch-up wide event", async () => { + serveDocuments(2); + + await runSummaryCatchUpForUser("user-1"); + + expect(annotate).toHaveBeenCalledWith({ + action: { name: "documents.autoRead.catchUp" }, + meta: { enqueued: 2, capped: false }, + }); + }); +}); + +describe("enqueueSummaryCatchUp", () => { + it("coalesces a double toggle onto one per-user singleton key", async () => { + const send = vi.fn().mockResolvedValue("job-1"); + vi.mocked(getGlobalBoss).mockReturnValue({ + send, + } as unknown as ReturnType); + + await enqueueSummaryCatchUp("user-1"); + await enqueueSummaryCatchUp("user-1"); + + expect(send).toHaveBeenCalledTimes(2); + for (const call of send.mock.calls) { + expect(call[0]).toBe(DOCUMENT_SUMMARY_CATCHUP_QUEUE); + expect(call[2]).toMatchObject({ + singletonKey: "document-summary-catchup|user-1", + }); + } + }); + + it("is a no-op without a boss and never throws on a send failure", async () => { + vi.mocked(getGlobalBoss).mockReturnValue( + null as unknown as ReturnType, + ); + await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({ + enqueued: false, + }); + + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn().mockRejectedValue(new Error("down")), + } as unknown as ReturnType); + await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({ + enqueued: false, + }); + }); +}); diff --git a/src/lib/jobs/document-summary-catchup.ts b/src/lib/jobs/document-summary-catchup.ts new file mode 100644 index 000000000..9acb154fe --- /dev/null +++ b/src/lib/jobs/document-summary-catchup.ts @@ -0,0 +1,164 @@ +/** + * Catch-up pass for documents stored BEFORE the auto-read opt-in was turned on. + * + * The per-document summary job is enqueued at UPLOAD time and no-ops when + * `documentsAutoAiRead` is OFF. That leaves an obvious hole: a user who uploads + * their vault first and flips the toggle afterwards gets nothing. Every document + * they already hold was enqueued while the flag was OFF, the job no-opped, and + * nothing ever re-enqueues them — so the switch reads as broken. It only ever + * worked for documents uploaded after the flip. + * + * A genuine OFF→ON transition on `PATCH /api/auth/me/documents-auto-ai-read` + * enqueues one pass of this job, which walks the user's un-summarised documents + * and enqueues the ordinary per-document summary job for each. + * + * What this job does NOT do is grant anything. It only enqueues; every gate + * still runs inside `runDocumentSummaryJob` — the opt-in is re-read, the egress + * consent is re-asserted per document, and the daily token budget is reserved + * and reconciled exactly as on the upload path. Retroactive work is not a reason + * to skip a consent receipt or a budget ceiling, so it does not. + * + * Bounded and idempotent: + * - the candidate set is documents with NO summary, so a document drops out + * the moment its summary lands and a re-run cannot redo finished work; + * - an id-cursor forward walk capped at `MAX_ENQUEUES_PER_RUN` (see the + * constant for why that number) keeps one flip from queueing a whole vault; + * - the per-user `singletonKey` plus the queue's `short` policy collapses a + * second flip while a pass is still queued, and the per-document enqueue + * carries its own singleton on top; + * - `runDocumentSummaryJob` skips any document that already has a summary, so + * even a duplicate that slips past both singletons is a cheap no-op. + * + * The queue MUST be registered in the maintenance registrar + * (`src/lib/jobs/reminder/register-maintenance.ts`) so pg-boss provisions it. + */ +import { documentAutoReadEnabled } from "@/lib/documents/document-settings"; +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { enqueueDocumentSummary } from "@/lib/jobs/document-summary"; +import { annotate } from "@/lib/logging/context"; + +export const DOCUMENT_SUMMARY_CATCHUP_QUEUE = "document-summary-catchup"; + +/** Serial concurrency — cheap discovery + fan-out enqueues. */ +export const DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY = 1; + +/** Discovery page size (id-only). */ +const PAGE_SIZE = 100; + +/** + * Max per-document summary jobs one catch-up pass enqueues. + * + * Deliberately far below the thumbnail backfill's cap: a thumbnail is local + * compute, a summary is a provider call against the user's daily token budget. + * The budget gate already stops runaway spend (every job past the ceiling + * no-ops), so this cap is about not dumping an entire vault onto the queue in + * one go — a large library converges over repeated flips and the ordinary + * upload path rather than in a single burst. Raising it costs queue depth, not + * money. + */ +export const MAX_ENQUEUES_PER_RUN = 200; + +export interface SummaryCatchUpPayload { + userId: string; + enqueuedAt?: string; +} + +export interface SummaryCatchUpSummary { + enqueued: number; + /** True when the pass stopped at the cap with candidates still outstanding. */ + capped: boolean; +} + +/** + * Enqueue summary jobs for one user's un-summarised documents. Re-reads the + * opt-in first: a user who flipped the toggle back OFF between the PATCH and + * this job running gets nothing, which keeps the pass fail-closed against a + * consent race the same way the per-document job is. + */ +export async function runSummaryCatchUpForUser( + userId: string, +): Promise { + if (!userId) return { enqueued: 0, capped: false }; + + // Fail-closed on the toggle. The PATCH that scheduled this pass is not + // authority enough — the flag is re-read at run time, exactly as the + // per-document job does before it egresses anything. + if (!(await documentAutoReadEnabled(userId))) { + annotate({ + action: { name: "documents.autoRead.catchUpSkipped" }, + meta: { reason: "opt-out" }, + }); + return { enqueued: 0, capped: false }; + } + + let enqueued = 0; + let capped = false; + let cursor: string | null = null; + + for (;;) { + if (enqueued >= MAX_ENQUEUES_PER_RUN) { + capped = true; + break; + } + const rows: { id: string }[] = await prisma.inboundDocument.findMany({ + where: { userId, deletedAt: null, summaryEncrypted: null }, + select: { id: true }, + orderBy: { id: "asc" }, + take: PAGE_SIZE, + ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}), + }); + if (rows.length === 0) break; + + for (const { id } of rows) { + if (enqueued >= MAX_ENQUEUES_PER_RUN) { + capped = true; + break; + } + const { enqueued: ok } = await enqueueDocumentSummary(userId, id); + if (ok) enqueued += 1; + } + + cursor = rows[rows.length - 1]!.id; + if (rows.length < PAGE_SIZE) break; + } + + annotate({ + action: { name: "documents.autoRead.catchUp" }, + meta: { enqueued, capped }, + }); + return { enqueued, capped }; +} + +/** + * Enqueue one catch-up pass for a user. Called on a genuine OFF→ON flip of the + * auto-read opt-in. Coalesced by `singletonKey` per user so flipping the toggle + * twice cannot queue two passes. Fire-and-forget — a missing boss (worker not + * up) is a no-op, and a `boss.send` failure is swallowed, so the catch-up can + * never fail the settings write that scheduled it. + */ +export async function enqueueSummaryCatchUp( + userId: string, +): Promise<{ enqueued: boolean }> { + const boss = getGlobalBoss(); + if (!boss) return { enqueued: false }; + const payload: SummaryCatchUpPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + try { + const jobId = await boss.send(DOCUMENT_SUMMARY_CATCHUP_QUEUE, payload, { + retryLimit: 2, + retryDelay: 60, + retryBackoff: true, + singletonKey: `document-summary-catchup|${userId}`, + }); + return { enqueued: Boolean(jobId) }; + } catch (err) { + annotate({ + action: { name: "documents.autoRead.catchUpEnqueueFailed" }, + meta: { reason: err instanceof Error ? err.name : "unknown" }, + }); + return { enqueued: false }; + } +} diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index f7950400f..c258a79ca 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -93,6 +93,12 @@ import { runDocumentSummaryJob, type DocumentSummaryPayload, } from "@/lib/jobs/document-summary"; +import { + DOCUMENT_SUMMARY_CATCHUP_QUEUE, + DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY, + runSummaryCatchUpForUser, + type SummaryCatchUpPayload, +} from "@/lib/jobs/document-summary-catchup"; import { DOCUMENT_THUMBNAIL_BACKFILL_QUEUE, DOCUMENT_THUMBNAIL_BACKFILL_CONCURRENCY, @@ -359,6 +365,11 @@ const allQueues = [ // egress consent + budget; persists the summary encrypted. Without this entry // pg-boss never provisions the queue and every upload enqueue silently no-ops. DOCUMENT_SUMMARY_QUEUE, + // v1.30.31 — auto-read catch-up. Scheduled on a genuine OFF→ON flip of the + // `documentsAutoAiRead` opt-in; fans out summary jobs for documents that were + // uploaded while the flag was still OFF. Without this entry pg-boss never + // provisions the queue and the catch-up enqueue silently no-ops. + DOCUMENT_SUMMARY_CATCHUP_QUEUE, ]; const schedules: ScheduleEntry[] = [ @@ -507,6 +518,11 @@ const queuePolicies: QueuePolicyTable = { reason: "Per-document, no re-converging discovery pass. Collapse queued duplicates only.", }, + [DOCUMENT_SUMMARY_CATCHUP_QUEUE]: { + policy: "short", + reason: + "Per-user, keyed on the opt-in flip. Collapse a double toggle into one pass; the pass re-reads its candidate set when it starts.", + }, }; /** @@ -890,6 +906,34 @@ export async function registerMaintenanceQueues( }, ); + // Document AI — auto-read catch-up worker. A genuine OFF→ON flip of the + // `documentsAutoAiRead` opt-in sends one per-user job; this handler fans out + // per-document summary jobs for that user's already-stored, un-summarised + // documents (the upload-time enqueue no-opped while the flag was OFF). It + // only enqueues — the opt-in, egress consent and budget gates all still run + // per document inside the summary job. Serial concurrency. + await boss.work( + DOCUMENT_SUMMARY_CATCHUP_QUEUE, + { localConcurrency: DOCUMENT_SUMMARY_CATCHUP_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + if (!userId) continue; + try { + await runSummaryCatchUpForUser(userId); + } catch (err) { + recordError(); + workerLog( + "error", + `[document-summary-catchup] user=${userId} failed`, + err, + ); + throw err; + } + } + }, + ); + // Document vault — boot-time preview-thumbnail backfill worker. The boot // discovery sends one per-user job; this handler fans out per-document // thumbnail jobs for that user's not-yet-thumbnailed documents. Serial