fix(documents): keep deferred imports local and meter every re-send

A document uploaded with aiRead=defer is now indexed locally by any
automatic path, because the indexer reads the marker from the row rather
than from the job. Index all for search leaves such documents out and no
longer counts them. Only an explicit Read with AI or Generate summary ends
the hold.

Uploads that send a body keep their slot of the hourly limit, duplicates
included; handing it back let a token send the same bytes under id after
id for free. One document takes at most 20 further source ids, beyond
which the upload answers 409 documents.inbound.sourceAliasLimit. The
query-string key check is metered on the lookup allowance it shares with
GET /api/documents/inbound/source.

Refs #1038

src/app/api/documents/inbound/__tests__/route.test.ts

@@ -56,6 +56,7 @@ vi.mock("@/lib/db", () => {
       documentSourceAlias: {
         findUnique: vi.fn(),
         createMany: vi.fn(),
+        count: vi.fn(),
       },
       extractedFact: {
         groupBy: vi.fn(),
@@ -247,6 +248,7 @@ beforeEach(() => {
   vi.mocked(prisma.documentSourceAlias.createMany).mockResolvedValue({
     count: 1,
   } as never);
+  vi.mocked(prisma.documentSourceAlias.count).mockResolvedValue(0 as never);
   // No settings row / no user override → policy defaults (25 MiB / 1 GiB).
   vi.mocked(prisma.appSettings.findUnique).mockResolvedValue(null as never);
   vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never);
@@ -844,7 +846,28 @@ describe("POST /api/documents/inbound — document import", () => {
     );
     expect(res.status).toBe(200);
     expect((await res.json()).data).toEqual({ id: "doc-src", duplicate: true });
-    expect(checkRateLimit).not.toHaveBeenCalled();
+    // Metered on the lookup bucket only; no upload slot is spent.
+    expect(
+      vi.mocked(checkRateLimit).mock.calls.map((c) => [c[0], c[1]]),
+    ).toEqual([["documents-source:token:tok-1", 5000]]);
+  });
+
+  it("429s the query-string check once the lookup bucket is spent", async () => {
+    armToken([DOCUMENTS_WRITE_SCOPE]);
+    vi.mocked(checkRateLimit).mockResolvedValueOnce({
+      allowed: false,
+      remaining: 0,
+      limit: 5000,
+      resetAt: Date.now() + 60_000,
+    } as never);
+    const res = await post(
+      new Request(
+        "http://localhost/api/documents/inbound?sourceSystem=PAPRA&sourceId=doc_1",
+        { method: "POST", body: "x" },
+      ),
+    );
+    expect(res.status).toBe(429);
+    expect(prisma.inboundDocument.findFirst).not.toHaveBeenCalled();
   });
 
   it("422s half a query-string key", async () => {
@@ -871,7 +894,7 @@ describe("POST /api/documents/inbound — document import", () => {
     expect(txCreate).not.toHaveBeenCalled();
   });
 
-  it("remembers the key when keyed bytes are already stored, and refunds the slot", async () => {
+  it("remembers the key when keyed bytes are already stored, and keeps the slot charged", async () => {
     // Own-key lookup misses; the content lookup finds a live row.
     vi.mocked(prisma.inboundDocument.findFirst)
       .mockResolvedValueOnce(null as never)
@@ -891,10 +914,27 @@ describe("POST /api/documents/inbound — document import", () => {
       ],
       skipDuplicates: true,
     });
-    expect(refundRateLimit).toHaveBeenCalledWith("documents-upload:user-1");
+    // The body was read: the slot stays spent, so the same bytes under id
+    // after id are not free.
+    expect(refundRateLimit).not.toHaveBeenCalled();
     expect(txCreate).not.toHaveBeenCalled();
   });
 
+  it("refuses a further id for bytes already held under the maximum", async () => {
+    vi.mocked(prisma.inboundDocument.findFirst)
+      .mockResolvedValueOnce(null as never)
+      .mockResolvedValueOnce(docRow({ id: "doc-manual" }) as never);
+    vi.mocked(prisma.documentSourceAlias.count).mockResolvedValue(20 as never);
+    const res = await post(
+      mkUpload({ sourceSystem: "PAPRA", sourceId: "doc_21" }),
+    );
+    expect(res.status).toBe(409);
+    expect((await res.json()).meta.errorCode).toBe(
+      "documents.inbound.sourceAliasLimit",
+    );
+    expect(prisma.documentSourceAlias.createMany).not.toHaveBeenCalled();
+  });
+
   it("resolves a key held as an alias of a deleted document to deleted", async () => {
     vi.mocked(prisma.documentSourceAlias.findUnique).mockResolvedValue({
       document: docRow({ id: "doc-gone", deletedAt: new Date() }),
@@ -908,22 +948,18 @@ describe("POST /api/documents/inbound — document import", () => {
       duplicate: true,
       deleted: true,
     });
-    expect(refundRateLimit).toHaveBeenCalled();
+    expect(refundRateLimit).not.toHaveBeenCalled();
     expect(txCreate).not.toHaveBeenCalled();
   });
 
-  it("refunds the slot on a plain content duplicate", async () => {
+  it("keeps a plain content duplicate charged and writes no alias", async () => {
     vi.mocked(prisma.inboundDocument.findFirst).mockResolvedValue(
       docRow({ id: "doc-existing" }) as never,
     );
     await post(mkUpload());
-    expect(refundRateLimit).toHaveBeenCalledWith("documents-upload:user-1");
-    expect(prisma.documentSourceAlias.createMany).not.toHaveBeenCalled();
-  });
-
-  it("does not refund a stored upload", async () => {
-    await post(mkUpload());
+    expect(checkRateLimit).toHaveBeenCalledTimes(1);
     expect(refundRateLimit).not.toHaveBeenCalled();
+    expect(prisma.documentSourceAlias.createMany).not.toHaveBeenCalled();
   });
 
   it("persists the deferral on the row", async () => {

src/app/api/documents/inbound/reindex/__tests__/route.test.ts

@@ -114,11 +114,16 @@ describe("POST /api/documents/inbound/reindex", () => {
 
   it("scopes both counts to the caller's LIVE documents (the gauge's own figures)", async () => {
     await POST(req() as never);
+    // Imports held back from AI reading (#1038) are not part of the run and
+    // stay out of both figures.
     expect(prisma.inboundDocument.count).toHaveBeenCalledWith({
-      where: { userId: "user-1", deletedAt: null },
+      where: { userId: "user-1", deletedAt: null, aiReadDeferred: false },
     });
     expect(prisma.documentContentIndex.count).toHaveBeenCalledWith({
-      where: { userId: "user-1", document: { deletedAt: null } },
+      where: {
+        userId: "user-1",
+        document: { deletedAt: null, aiReadDeferred: false },
+      },
     });
   });
 });

src/app/api/documents/inbound/reindex/route.ts

@@ -67,11 +67,16 @@ export const POST = apiHandler(async (request) => {
   let enqueued = 0;
   if (jobCreated) {
     const [indexedCount, totalCount] = await Promise.all([
+      // Imports held back from AI reading (#1038) are not part of this run
+      // and are left out of both figures.
       prisma.documentContentIndex.count({
-        where: { userId: user.id, document: { deletedAt: null } },
+        where: {
+          userId: user.id,
+          document: { deletedAt: null, aiReadDeferred: false },
+        },
       }),
       prisma.inboundDocument.count({
-        where: { userId: user.id, deletedAt: null },
+        where: { userId: user.id, deletedAt: null, aiReadDeferred: false },
       }),
     ]);
     enqueued = Math.max(0, totalCount - indexedCount);

src/app/api/documents/inbound/route.ts

@@ -70,6 +70,7 @@ import { enqueueDocumentSummary } from "@/lib/jobs/document-summary";
 import { enqueueDocumentThumbnail } from "@/lib/jobs/document-thumbnail";
 import { DOCUMENTS_WRITE_SCOPE } from "@/lib/documents/scopes";
 import {
+  checkSourceLookupRateLimit,
   findSourceKey,
   rememberSourceAlias,
   type SourceKeyMatch,
@@ -86,11 +87,7 @@ import { linkTargets } from "@/lib/links";
 import { annotate } from "@/lib/logging/context";
 import { requireModuleEnabled } from "@/lib/modules/gate";
 import { isP2002 } from "@/lib/prisma-errors";
-import {
-  checkRateLimit,
-  rateLimitHeaders,
-  refundRateLimit,
-} from "@/lib/rate-limit";
+import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit";
 import {
   documentCreateSchema,
   documentListQuerySchema,
@@ -203,6 +200,19 @@ function deletedResponse(ctx: UploadContext, id: string | null): NextResponse {
   return receiptResponse(id, 200, { duplicate: true, deleted: true });
 }
 
+/**
+ * The same bytes have already been sent under too many source keys
+ * (`MAX_SOURCE_ALIASES_PER_DOCUMENT`). A conflict with what is stored, not a
+ * rate: nothing is remembered and nothing stored.
+ */
+function aliasLimitResponse(): NextResponse {
+  return apiError(
+    "This file is already stored under too many source ids.",
+    409,
+    { errorCode: "documents.inbound.sourceAliasLimit" },
+  );
+}
+
 /** Answer a source key that is already held: duplicate, or deleted. */
 function answerSourceKey(
   ctx: UploadContext,
@@ -307,6 +317,22 @@ async function processUpload(
       );
     }
     queryKey = parsedKey.data;
+    // Metered like the lookup route, on the same bucket: the check stores
+    // nothing and costs no upload slot, but it is not a free oracle either.
+    const lookupRl = await checkSourceLookupRateLimit(
+      scoped,
+      auth.session.id,
+      user.id,
+    );
+    if (!lookupRl.allowed) {
+      const response = apiError("Too many lookups. Try again later.", 429, {
+        errorCode: "documents.inbound.rateLimited",
+      });
+      for (const [k, v] of Object.entries(rateLimitHeaders(lookupRl))) {
+        response.headers.set(k, v);
+      }
+      return response;
+    }
     const match = await findSourceKey(
       user.id,
       queryKey.sourceSystem,
@@ -337,11 +363,10 @@ async function processUpload(
     scoped ? resolveDocumentUploadLimitPerHour() : UPLOAD_LIMIT_PER_HOUR,
     UPLOAD_WINDOW_MS,
   );
-  // A slot pays for a document stored. An upload that turns out to be one
-  // already held (same key, same bytes) hands its slot back, so re-sends —
-  // a Paperless workflow firing on every edit, a re-run over an archive —
-  // never eat into the allowance.
-  const refundSlot = () => refundRateLimit(bucketKey).catch(() => {});
+  // Every upload that reaches the body pays its slot, stored or not. The cheap
+  // way to re-send is the query-string key above, answered before this line;
+  // handing slots back after a full body read would let a leaked token send
+  // the same bytes under id after id for free.
   if (!rl.allowed) {
     const response = apiError("Too many uploads. Try again later.", 429, {
       errorCode: "documents.inbound.rateLimited",
@@ -452,10 +477,7 @@ async function processUpload(
   // the ledger remembers.
   if (sourceSystem && sourceId && !queryKey) {
     const match = await findSourceKey(user.id, sourceSystem, sourceId);
-    if (match) {
-      await refundSlot();
-      return answerSourceKey(ctx, match);
-    }
+    if (match) return answerSourceKey(ctx, match);
   }
 
   let buffer: Buffer;
@@ -522,9 +544,14 @@ async function processUpload(
     // none) is answered with that document — and the key is remembered for
     // it, so once the person deletes the document this key stays deleted too.
     if (sourceSystem && sourceId) {
-      await rememberSourceAlias(user.id, existing.id, sourceSystem, sourceId);
+      const remembered = await rememberSourceAlias(
+        user.id,
+        existing.id,
+        sourceSystem,
+        sourceId,
+      );
+      if (remembered === "limit") return aliasLimitResponse();
     }
-    await refundSlot();
     return duplicateResponse(ctx, existing);
   }
 
@@ -626,10 +653,7 @@ async function processUpload(
       // paths above would have.
       if (sourceSystem && sourceId) {
         const match = await findSourceKey(user.id, sourceSystem, sourceId);
-        if (match) {
-          await refundSlot();
-          return answerSourceKey(ctx, match);
-        }
+        if (match) return answerSourceKey(ctx, match);
       }
       const winner = await prisma.inboundDocument.findFirst({
         where: { userId: user.id, contentSha256, deletedAt: null },
@@ -637,9 +661,14 @@ async function processUpload(
       });
       if (winner) {
         if (sourceSystem && sourceId) {
-          await rememberSourceAlias(user.id, winner.id, sourceSystem, sourceId);
+          const remembered = await rememberSourceAlias(
+            user.id,
+            winner.id,
+            sourceSystem,
+            sourceId,
+          );
+          if (remembered === "limit") return aliasLimitResponse();
         }
-        await refundSlot();
         return duplicateResponse(ctx, winner);
       }
     }

src/app/api/documents/inbound/source/route.ts

@@ -20,22 +20,17 @@ import {
   sanitiseZodIssues,
 } from "@/lib/api-response";
 import { DOCUMENTS_WRITE_SCOPE } from "@/lib/documents/scopes";
-import { findSourceKey } from "@/lib/documents/source-key";
+import {
+  checkSourceLookupRateLimit,
+  findSourceKey,
+} from "@/lib/documents/source-key";
 import { annotate } from "@/lib/logging/context";
 import { requireModuleEnabled } from "@/lib/modules/gate";
-import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit";
+import { rateLimitHeaders } from "@/lib/rate-limit";
 import { documentSourceKeySchema } from "@/lib/validations/inbound-documents";
 
 export const dynamic = "force-dynamic";
 
-/**
- * Lookups per caller per hour. Generous on purpose: an importer asks once per
- * document in the source, so a re-sync of a few thousand documents has to fit
- * in an hour. A lookup reads three indexed rows and stores nothing.
- */
-const LOOKUP_LIMIT_PER_HOUR = 5000;
-const LOOKUP_WINDOW_MS = 60 * 60 * 1000;
-
 export const GET = apiHandler(async (request: Request) => {
   const auth = await requireAuth(DOCUMENTS_WRITE_SCOPE);
   const { user } = auth;
@@ -43,12 +38,10 @@ export const GET = apiHandler(async (request: Request) => {
   const gate = await requireModuleEnabled(user.id, "inboundDocuments");
   if (!gate.enabled) return gate.response;
 
-  const rl = await checkRateLimit(
-    isScopedCredential(auth)
-      ? `documents-source:token:${auth.session.id}`
-      : `documents-source:${user.id}`,
-    LOOKUP_LIMIT_PER_HOUR,
-    LOOKUP_WINDOW_MS,
+  const rl = await checkSourceLookupRateLimit(
+    isScopedCredential(auth),
+    auth.session.id,
+    user.id,
   );
   if (!rl.allowed) {
     const response = apiError("Too many lookups. Try again later.", 429, {