feat(documents): catch up unread documents when auto-read is switched on

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.
This commit is contained in:
Marc-André Bombeck committed 2026-07-19 11:09:57 +02:00
1 parent 8e938a4f45
commit 80258914f2
5 files changed
+523 -1

No files matched your search

@@ -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 },
});
});
});
@@ -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),
@@ -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<typeof getGlobalBoss>);
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<typeof getGlobalBoss>,
);
await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({
enqueued: false,
});
vi.mocked(getGlobalBoss).mockReturnValue({
send: vi.fn().mockRejectedValue(new Error("down")),
} as unknown as ReturnType<typeof getGlobalBoss>);
await expect(enqueueSummaryCatchUp("user-1")).resolves.toEqual({
enqueued: false,
});
});
});
+164
View File
@@ -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<SummaryCatchUpSummary> {
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 };
}
}
@@ -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<SummaryCatchUpPayload>(
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