refactor(documents): move the upload's store path into one ingest function

The upload route keeps authentication, its rate buckets, multipart parsing
and the response shape; the source-key answer, size and type checks,
pre-link narrowing, content-hash dedup with alias memory, the quota-gated
insert, the audit row and the background jobs move to
src/lib/documents/ingest.ts unchanged, so a second way in can store
through the same rules.

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

@@ -32,7 +32,6 @@
  * instruction inside it. Storing it does nothing with its contents.
  */
 import { Buffer } from "node:buffer";
-import { createHash } from "node:crypto";
 
 import { NextResponse } from "next/server";
 
@@ -50,43 +49,38 @@ import {
   getClientIp,
   sanitiseZodIssues,
 } from "@/lib/api-response";
-import { getAiCapability } from "@/lib/ai/capabilities/gate";
-import { auditLog } from "@/lib/auth/audit";
 import { prisma } from "@/lib/db";
 import { hashQueryTokens } from "@/lib/documents/content-index";
+import {
+  ingestDocument,
+  isoDateToUtc,
+  personalUploadBucket,
+  UPLOAD_LIMIT_PER_HOUR,
+  UPLOAD_WINDOW_MS,
+} from "@/lib/documents/ingest";
 import {
   loadConditionLinks,
   loadDocumentEncounterLinks,
-  narrowOwnedEncounterIds,
-  narrowOwnedEpisodeIds,
 } from "@/lib/documents/links";
 import {
-  encryptDocumentContent,
   serialiseDocument,
   type SerialisableDocument,
 } from "@/lib/documents/store";
-import { enqueueDocumentIndex } from "@/lib/jobs/document-index";
-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,
 } from "@/lib/documents/source-key";
 import {
   acquireDocumentUploadSlot,
-  detectDocumentType,
   resolveDocumentLimits,
   resolveDocumentUploadLimitPerHour,
 } from "@/lib/documents/upload-policy";
 import { withIdempotency } from "@/lib/idempotency";
 import { BodyTooLargeError, readBoundedBody } from "@/lib/labs/ocr-upload";
-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 } from "@/lib/rate-limit";
 import {
   documentCreateSchema,
@@ -99,13 +93,6 @@ import type { Prisma } from "@/generated/prisma/client";
 
 export const dynamic = "force-dynamic";
 
-/**
- * A store-only upload touches no provider — the only abuse vector is disk, so
- * a generous per-user ceiling is enough (extraction has its own tight gate).
- */
-const UPLOAD_LIMIT_PER_HOUR = 60;
-const UPLOAD_WINDOW_MS = 60 * 60 * 1000;
-
 /**
  * Multipart envelope allowance on top of the per-file cap: boundaries plus
  * the small metadata fields (title / kind / documentDate / episodeIds). The
@@ -114,11 +101,6 @@ const UPLOAD_WINDOW_MS = 60 * 60 * 1000;
 const MULTIPART_OVERHEAD_BYTES = 64 * 1024;
 const UPLOAD_BODY_TIMEOUT_MS = 30_000;
 
-/** Parse a YYYY-MM-DD form field into a UTC midnight Date. */
-function isoDateToUtc(value: string): Date {
-  return new Date(`${value}T00:00:00.000Z`);
-}
-
 /** §3.2 — 413 fileTooLarge with the configured limit in `meta`. */
 function fileTooLarge(maxFileBytes: number): NextResponse {
   return apiError("File is too large.", 413, {
@@ -128,14 +110,6 @@ function fileTooLarge(maxFileBytes: number): NextResponse {
   });
 }
 
-/** Internal signal: the quota gate inside the insert transaction tripped. */
-class QuotaExceededError extends Error {
-  constructor(public readonly usedBytes: number) {
-    super("Document quota exceeded");
-    this.name = "QuotaExceededError";
-  }
-}
-
 /** What a single upload asked for, carried into the annotations. */
 interface UploadContext {
   userId: string;
@@ -357,7 +331,7 @@ async function processUpload(
   // draw on. The cookie / wildcard bucket is unchanged.
   const bucketKey = scoped
     ? `documents-upload:token:${auth.session.id}`
-    : `documents-upload:${user.id}`;
+    : personalUploadBucket(user.id);
   const rl = await checkRateLimit(
     bucketKey,
     scoped ? resolveDocumentUploadLimitPerHour() : UPLOAD_LIMIT_PER_HOUR,
@@ -486,157 +460,55 @@ async function processUpload(
   } catch {
     return apiError("Failed to read uploaded file", 400);
   }
-  if (buffer.byteLength > limits.maxFileBytes) {
-    return fileTooLarge(limits.maxFileBytes);
-  }
-  if (buffer.byteLength === 0) {
-    return apiError("Uploaded file is empty", 422, {
-      errorCode: "documents.inbound.invalidMetadata",
-    });
-  }
 
-  // §3.1 — magic-byte classification; the wire Content-Type is never trusted.
-  const detected = detectDocumentType(
-    buffer,
-    typeof file.name === "string" ? file.name : null,
-  );
-  if (!detected) {
-    return apiError("This file type is not supported.", 415, {
-      errorCode: "documents.inbound.fileType",
-      reason: "unsupportedType",
-    });
-  }
-
-  // Pre-linking: every episode id must be a LIVE episode of the caller.
-  const episodeIds = await narrowOwnedEpisodeIds(
-    user.id,
-    parsed.data.episodeIds ?? [],
-  );
-  if (episodeIds === null) {
-    return apiError("Episode not found", 404, {
-      errorCode: "documents.inbound.episodeNotFound",
-    });
-  }
-
-  // Same for the visit ids the review step offered. A refusal rather than a
-  // silent drop: the person saw a visit named and would otherwise watch the
-  // upload succeed without the link they asked for.
-  const encounterIds = await narrowOwnedEncounterIds(
-    user.id,
-    parsed.data.encounterIds ?? [],
-  );
-  if (encounterIds === null) {
-    return apiError("Visit not found", 404, {
-      errorCode: "documents.inbound.encounterNotFound",
-    });
-  }
-
-  // sha256 of the PLAINTEXT for same-user duplicate detection.
-  const contentSha256 = createHash("sha256").update(buffer).digest("hex");
-
-  // Fast-path dedupe check; the partial unique index closes the race below.
-  const existing = await prisma.inboundDocument.findFirst({
-    where: { userId: user.id, contentSha256, deletedAt: null },
-    omit: { contentEncrypted: true },
+  // Everything from here to the stored row is the shared ingest path
+  // (`src/lib/documents/ingest.ts`), the one the document picker's import
+  // stores through as well.
+  const result = await ingestDocument({
+    userId: user.id,
+    scoped,
+    ipAddress: getClientIp(request),
+    bytes: buffer,
+    filename: typeof file.name === "string" ? file.name : null,
+    title: parsed.data.title ?? null,
+    kind: parsed.data.kind ?? null,
+    documentDate: parsed.data.documentDate ?? null,
+    episodeIds: parsed.data.episodeIds ?? [],
+    encounterIds: parsed.data.encounterIds ?? [],
+    sourceSystem,
+    sourceId,
+    aiDeferred: ctx.aiDeferred,
+    sourceKeyChecked: true,
+    limits,
   });
-  if (existing) {
-    // An import sending bytes that are already stored under another key (or
-    // 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) {
-      const remembered = await rememberSourceAlias(
-        user.id,
-        existing.id,
-        sourceSystem,
-        sourceId,
-      );
-      if (remembered === "limit") return aliasLimitResponse();
-    }
-    return duplicateResponse(ctx, existing);
-  }
 
-  // Quota gate + insert + pre-links in ONE transaction. Usage counts every
-  // non-purged row — tombstones still hold TOAST bytes, so "deleted" bytes
-  // are never invisible weight (undo-delete never changes usage).
-  let document: SerialisableDocument;
-  try {
-    document = await prisma.$transaction(async (tx) => {
-      // Serialise the quota gate per user: without this, N concurrent
-      // uploads all read the same SUM before any of them commits and the
-      // quota can be overshot by up to N × cap in one burst. The advisory
-      // lock is transaction-scoped (released on commit/rollback) and keyed
-      // on the user id, so uploads by different users never queue on each
-      // other.
-      // (`pg_advisory_xact_lock` returns void, which the client cannot
-      // deserialize as a column — selecting FROM it yields a plain int row.)
-      await tx.$queryRaw`
-        SELECT 1 AS locked
-        FROM pg_advisory_xact_lock(hashtextextended('documents-quota:' || ${user.id}, 0))
-      `;
-      const rows = await tx.$queryRaw<Array<{ used: bigint }>>`
-        SELECT COALESCE(SUM(byte_size), 0)::bigint AS used
-        FROM inbound_documents
-        WHERE user_id = ${user.id}
-      `;
-      const usedBytes = Number(rows[0]?.used ?? 0);
-      if (usedBytes + buffer.byteLength > limits.quotaBytes) {
-        throw new QuotaExceededError(usedBytes);
-      }
-
-      const { content, codec } = encryptDocumentContent(buffer);
-
-      // No mass assignment — every column is set field-by-field; `userId`
-      // comes from the session, never the body. `documentDate` defaults to
-      // the upload day so display == sort == filter (user-editable later).
-      const created = await tx.inboundDocument.create({
-        data: {
-          userId: user.id,
-          kind: parsed.data.kind ?? "OTHER",
-          // Stored plaintext on purpose (mirrors `filename`) so the list can
-          // ILIKE-search + ORDER BY it. It MAY hold PHI the user types; that
-          // is the accepted tradeoff for server-side search/sort — the
-          // document body stays encrypted.
-          title: parsed.data.title ?? null,
-          filename:
-            typeof file.name === "string" ? file.name.slice(0, 255) : null,
-          mimeType: detected.mimeType,
-          byteSize: buffer.byteLength,
-          contentEncrypted: content,
-          contentCodec: codec,
-          contentSha256,
-          status: "STORED",
-          documentDate: parsed.data.documentDate
-            ? isoDateToUtc(parsed.data.documentDate)
-            : new Date(),
-          sourceSystem,
-          sourceId,
-          aiReadDeferred: ctx.aiDeferred,
-        },
-        omit: { contentEncrypted: true },
+  switch (result.kind) {
+    case "sourceKey":
+      return answerSourceKey(ctx, result.match);
+    case "tooLarge":
+      return fileTooLarge(result.maxFileBytes);
+    case "empty":
+      return apiError("Uploaded file is empty", 422, {
+        errorCode: "documents.inbound.invalidMetadata",
       });
-
-      if (episodeIds.length > 0) {
-        await linkTargets(tx, {
-          userId: user.id,
-          sourceKind: "document",
-          sourceId: created.id,
-          targetKind: "conditionEpisode",
-          targetIds: episodeIds,
-        });
-      }
-      if (encounterIds.length > 0) {
-        await linkTargets(tx, {
-          userId: user.id,
-          sourceKind: "document",
-          sourceId: created.id,
-          targetKind: "encounter",
-          targetIds: encounterIds,
-        });
-      }
-      return created;
-    });
-  } catch (err) {
-    if (err instanceof QuotaExceededError) {
+    case "unsupportedType":
+      return apiError("This file type is not supported.", 415, {
+        errorCode: "documents.inbound.fileType",
+        reason: "unsupportedType",
+      });
+    case "episodeNotFound":
+      return apiError("Episode not found", 404, {
+        errorCode: "documents.inbound.episodeNotFound",
+      });
+    case "encounterNotFound":
+      return apiError("Visit not found", 404, {
+        errorCode: "documents.inbound.encounterNotFound",
+      });
+    case "aliasLimit":
+      return aliasLimitResponse();
+    case "duplicate":
+      return duplicateResponse(ctx, result.document);
+    case "quotaExceeded":
       // The figures are the owner's; a write-only token learns only that the
       // vault is full.
       return apiError("Storage quota exceeded.", 413, {
@@ -644,107 +516,13 @@ async function processUpload(
         reason: "quotaExceeded",
         ...(scoped
           ? {}
-          : { quotaBytes: limits.quotaBytes, usedBytes: err.usedBytes }),
+          : { quotaBytes: result.quotaBytes, usedBytes: result.usedBytes }),
       });
-    }
-    if (isP2002(err)) {
-      // A racing upload won one of the two partial unique indexes — the same
-      // source key, or the same bytes. Surface the winner exactly as the fast
-      // paths above would have.
-      if (sourceSystem && sourceId) {
-        const match = await findSourceKey(user.id, sourceSystem, sourceId);
-        if (match) return answerSourceKey(ctx, match);
-      }
-      const winner = await prisma.inboundDocument.findFirst({
-        where: { userId: user.id, contentSha256, deletedAt: null },
-        omit: { contentEncrypted: true },
-      });
-      if (winner) {
-        if (sourceSystem && sourceId) {
-          const remembered = await rememberSourceAlias(
-            user.id,
-            winner.id,
-            sourceSystem,
-            sourceId,
-          );
-          if (remembered === "limit") return aliasLimitResponse();
-        }
-        return duplicateResponse(ctx, winner);
-      }
-    }
-    throw err;
-  }
-
-  await auditLog("documents.inbound.store", {
-    userId: user.id,
-    ipAddress: getClientIp(request),
-    details: {
-      documentId: document.id,
-      mime: detected.mimeType,
-      ...(scoped ? { scoped: true } : {}),
-      ...(sourceSystem ? { sourceSystem } : {}),
-    },
-  });
-
-  annotate({
-    action: { name: "documents.vault.upload" },
-    meta: {
-      documentId: document.id,
-      byteSize: document.byteSize,
-      servingClass: detected.servingClass,
-      linked: episodeIds.length,
-      linkedVisits: encounterIds.length,
-      scoped,
-      sourceSystem,
-      aiDeferred: ctx.aiDeferred,
-    },
-  });
-
-  // Auto-index the freshly stored document for content search: enqueue a
-  // fire-and-forget background job (provider-first, local text-layer fallback).
-  // The upload response never blocks on or fails because of indexing — the
-  // enqueue is not awaited and swallows its own errors (a missing boss or a
-  // transient send failure is a silent no-op). Only fresh inserts enqueue — a
-  // duplicate upload returns early above and never reaches here.
-  //
-  // `aiRead=defer` narrows it to the local text layer: an import of a whole
-  // archive must not turn into one provider call per file. The person reads
-  // them later, deliberately, from the document itself.
-  if (ctx.aiDeferred) {
-    void enqueueDocumentIndex(user.id, document.id, { localOnly: true });
-  } else {
-    void enqueueDocumentIndex(user.id, document.id);
-  }
-
-  // Render a preview thumbnail in the background too (pure local compute — no
-  // egress). Same fire-and-forget contract: the upload never blocks on or fails
-  // because of it, and a dropped enqueue is recoverable via the boot backfill.
-  // Only fresh inserts reach here (a duplicate returns early above).
-  void enqueueDocumentThumbnail(user.id, document.id);
-
-  // Summarise the freshly stored document in the background — but ONLY when the
-  // `documentsAutoAiRead` opt-in is ON (the job re-checks it and the egress
-  // consent, and no-ops otherwise) and the `documentAi` capability is open for
-  // this record. The upload itself never depends on AI: the document is stored
-  // above whatever the capability says, and the index job still runs its
-  // provider-free text-layer path. The persisted summary shows on the detail
-  // view. Same fire-and-forget contract: the upload never blocks on or fails
-  // because of it. Only fresh inserts reach here (a duplicate returns early).
-  //
-  // A deferred upload skips it outright and keeps `summaryState: NONE`, so
-  // the detail sheet offers "Generate summary" rather than a pending state.
-  const documentAi = ctx.aiDeferred
-    ? { available: false as const, reason: "deferred" }
-    : await getAiCapability("documentAi");
-  if (documentAi.available) {
-    void enqueueDocumentSummary(user.id, document.id);
-  } else {
-    annotate({
-      action: { name: "documents.summary.enqueueSkipped" },
-      meta: { documentId: document.id, reason: documentAi.reason },
-    });
+    case "stored":
+      break;
   }
 
+  const document = result.document;
   if (scoped) return receiptResponse(document.id, 201, { duplicate: false });
 
   const [links, visitLinks] = await Promise.all([

src/lib/documents/ingest.ts

@@ -0,0 +1,378 @@
+/**
+ * One way into the vault for a file that has already been read into memory.
+ *
+ * The upload route (`POST /api/documents/inbound`, cookie, `["*"]` or a
+ * `documents:write` token) and the document picker's import route
+ * (`POST /api/documents/sources/{system}/import`, which fetches a picked
+ * document from Paperless-ngx or Papra) both store through here, so the rules a
+ * stored document is held to cannot differ by the way it arrived:
+ *
+ *   - a source key answers before the bytes are looked at: a key already held
+ *     (live, tombstoned, remembered as an alias or kept in the purge ledger) is
+ *     a duplicate or a "deleted", never a second copy;
+ *   - the per-file cap, the empty-file refusal and magic-byte classification
+ *     (the wire Content-Type is never trusted);
+ *   - pre-links to the caller's own live episodes and visits, refused when an
+ *     id names nothing the caller owns;
+ *   - sha256 duplicate detection against live rows, remembering the source key
+ *     for the existing document so a later delete keeps that key deleted too;
+ *   - the quota gate and the insert in one transaction under a per-user
+ *     advisory lock, and the race on either partial unique index resolved the
+ *     way the fast paths above would have answered;
+ *   - the audit row, the wide-event annotation, and the background index,
+ *     thumbnail and summary jobs (the summary only when the `documentAi`
+ *     capability is open and the upload did not ask to defer AI reading).
+ *
+ * The callers own authentication, rate buckets, request parsing and the shape
+ * of their responses. This module returns what happened; it builds no HTTP.
+ */
+import type { Buffer } from "node:buffer";
+import { createHash } from "node:crypto";
+
+import { getAiCapability } from "@/lib/ai/capabilities/gate";
+import { auditLog } from "@/lib/auth/audit";
+import { prisma } from "@/lib/db";
+import {
+  narrowOwnedEncounterIds,
+  narrowOwnedEpisodeIds,
+} from "@/lib/documents/links";
+import {
+  findSourceKey,
+  rememberSourceAlias,
+  type SourceKeyMatch,
+} from "@/lib/documents/source-key";
+import {
+  encryptDocumentContent,
+  type SerialisableDocument,
+} from "@/lib/documents/store";
+import { detectDocumentType } from "@/lib/documents/upload-policy";
+import { enqueueDocumentIndex } from "@/lib/jobs/document-index";
+import { enqueueDocumentSummary } from "@/lib/jobs/document-summary";
+import { enqueueDocumentThumbnail } from "@/lib/jobs/document-thumbnail";
+import { linkTargets } from "@/lib/links";
+import { annotate } from "@/lib/logging/context";
+import { isP2002 } from "@/lib/prisma-errors";
+import type {
+  DocumentSourceSystemValue,
+  InboundDocumentKindValue,
+} from "@/lib/validations/inbound-documents";
+
+/**
+ * The person's own uploads per hour, from the web or the phone, and the
+ * picker's imports with them. A store-only upload touches no provider — the
+ * only abuse vector is disk, so a generous ceiling is enough. A narrow
+ * `documents:write` token draws on its own bucket instead
+ * (`resolveDocumentUploadLimitPerHour`).
+ */
+export const UPLOAD_LIMIT_PER_HOUR = 60;
+export const UPLOAD_WINDOW_MS = 60 * 60 * 1000;
+
+/** The bucket a person's own uploads and picker imports share. */
+export function personalUploadBucket(userId: string): string {
+  return `documents-upload:${userId}`;
+}
+
+/** Parse a YYYY-MM-DD value into a UTC midnight Date. */
+export function isoDateToUtc(value: string): Date {
+  return new Date(`${value}T00:00:00.000Z`);
+}
+
+export interface IngestInput {
+  userId: string;
+  /** A narrow `documents:write` token rather than a session or `["*"]`. */
+  scoped: boolean;
+  /** For the audit row. */
+  ipAddress: string | null;
+  bytes: Buffer;
+  filename: string | null;
+  title: string | null;
+  kind: InboundDocumentKindValue | null;
+  /** YYYY-MM-DD; the upload day when absent. */
+  documentDate: string | null;
+  episodeIds: string[];
+  encounterIds: string[];
+  sourceSystem: DocumentSourceSystemValue | null;
+  sourceId: string | null;
+  /** Hold back automatic AI reading (`aiRead=defer`). */
+  aiDeferred: boolean;
+  /**
+   * The caller already asked about this source key (the upload's query-string
+   * form answers before the bucket is charged). Skips the second lookup.
+   */
+  sourceKeyChecked: boolean;
+  /** The per-file cap and the quota, resolved by the caller. */
+  limits: { maxFileBytes: number; quotaBytes: number };
+}
+
+export type IngestResult =
+  | { kind: "sourceKey"; match: SourceKeyMatch }
+  | { kind: "tooLarge"; maxFileBytes: number }
+  | { kind: "empty" }
+  | { kind: "unsupportedType" }
+  | { kind: "episodeNotFound" }
+  | { kind: "encounterNotFound" }
+  | { kind: "aliasLimit" }
+  | { kind: "duplicate"; document: SerialisableDocument }
+  | { kind: "quotaExceeded"; usedBytes: number; quotaBytes: number }
+  | {
+      kind: "stored";
+      document: SerialisableDocument;
+      servingClass: string;
+      linkedEpisodes: number;
+      linkedVisits: number;
+    };
+
+/** Internal signal: the quota gate inside the insert transaction tripped. */
+class QuotaExceededError extends Error {
+  constructor(public readonly usedBytes: number) {
+    super("Document quota exceeded");
+    this.name = "QuotaExceededError";
+  }
+}
+
+/** A live row with the same bytes, remembering the source key for it. */
+async function answerDuplicate(
+  input: IngestInput,
+  existing: SerialisableDocument,
+): Promise<IngestResult> {
+  // An import sending bytes that are already stored under another key (or
+  // 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 (input.sourceSystem && input.sourceId) {
+    const remembered = await rememberSourceAlias(
+      input.userId,
+      existing.id,
+      input.sourceSystem,
+      input.sourceId,
+    );
+    if (remembered === "limit") return { kind: "aliasLimit" };
+  }
+  return { kind: "duplicate", document: existing };
+}
+
+export async function ingestDocument(
+  input: IngestInput,
+): Promise<IngestResult> {
+  const { userId, sourceSystem, sourceId, limits, bytes } = input;
+
+  // A source key answers before the bytes are looked at: an import re-sending
+  // what it sent before is a duplicate, and one re-sending a document the
+  // person deleted is refused a second copy — live and tombstoned rows both
+  // count (the unique index has no `deleted_at` predicate), and past the purge
+  // the ledger remembers.
+  if (sourceSystem && sourceId && !input.sourceKeyChecked) {
+    const match = await findSourceKey(userId, sourceSystem, sourceId);
+    if (match) return { kind: "sourceKey", match };
+  }
+
+  if (bytes.byteLength > limits.maxFileBytes) {
+    return { kind: "tooLarge", maxFileBytes: limits.maxFileBytes };
+  }
+  if (bytes.byteLength === 0) return { kind: "empty" };
+
+  // Magic-byte classification; the wire Content-Type is never trusted.
+  const detected = detectDocumentType(bytes, input.filename);
+  if (!detected) return { kind: "unsupportedType" };
+
+  // Pre-linking: every episode id must be a LIVE episode of the caller.
+  const episodeIds = await narrowOwnedEpisodeIds(userId, input.episodeIds);
+  if (episodeIds === null) return { kind: "episodeNotFound" };
+
+  // Same for the visit ids the review step offered. A refusal rather than a
+  // silent drop: the person saw a visit named and would otherwise watch the
+  // upload succeed without the link they asked for.
+  const encounterIds = await narrowOwnedEncounterIds(
+    userId,
+    input.encounterIds,
+  );
+  if (encounterIds === null) return { kind: "encounterNotFound" };
+
+  // sha256 of the PLAINTEXT for same-user duplicate detection.
+  const contentSha256 = createHash("sha256").update(bytes).digest("hex");
+
+  // Fast-path dedupe check; the partial unique index closes the race below.
+  const existing = await prisma.inboundDocument.findFirst({
+    where: { userId, contentSha256, deletedAt: null },
+    omit: { contentEncrypted: true },
+  });
+  if (existing) return answerDuplicate(input, existing);
+
+  // Quota gate + insert + pre-links in ONE transaction. Usage counts every
+  // non-purged row — tombstones still hold TOAST bytes, so "deleted" bytes
+  // are never invisible weight (undo-delete never changes usage).
+  let document: SerialisableDocument;
+  try {
+    document = await prisma.$transaction(async (tx) => {
+      // Serialise the quota gate per user: without this, N concurrent
+      // uploads all read the same SUM before any of them commits and the
+      // quota can be overshot by up to N × cap in one burst. The advisory
+      // lock is transaction-scoped (released on commit/rollback) and keyed
+      // on the user id, so uploads by different users never queue on each
+      // other.
+      // (`pg_advisory_xact_lock` returns void, which the client cannot
+      // deserialize as a column — selecting FROM it yields a plain int row.)
+      await tx.$queryRaw`
+        SELECT 1 AS locked
+        FROM pg_advisory_xact_lock(hashtextextended('documents-quota:' || ${userId}, 0))
+      `;
+      const rows = await tx.$queryRaw<Array<{ used: bigint }>>`
+        SELECT COALESCE(SUM(byte_size), 0)::bigint AS used
+        FROM inbound_documents
+        WHERE user_id = ${userId}
+      `;
+      const usedBytes = Number(rows[0]?.used ?? 0);
+      if (usedBytes + bytes.byteLength > limits.quotaBytes) {
+        throw new QuotaExceededError(usedBytes);
+      }
+
+      const { content, codec } = encryptDocumentContent(bytes);
+
+      // No mass assignment — every column is set field-by-field; `userId`
+      // comes from the session, never the body. `documentDate` defaults to
+      // the upload day so display == sort == filter (user-editable later).
+      const created = await tx.inboundDocument.create({
+        data: {
+          userId,
+          kind: input.kind ?? "OTHER",
+          // Stored plaintext on purpose (mirrors `filename`) so the list can
+          // ILIKE-search + ORDER BY it. It MAY hold PHI the user types; that
+          // is the accepted tradeoff for server-side search/sort — the
+          // document body stays encrypted.
+          title: input.title,
+          filename: input.filename ? input.filename.slice(0, 255) : null,
+          mimeType: detected.mimeType,
+          byteSize: bytes.byteLength,
+          contentEncrypted: content,
+          contentCodec: codec,
+          contentSha256,
+          status: "STORED",
+          documentDate: input.documentDate
+            ? isoDateToUtc(input.documentDate)
+            : new Date(),
+          sourceSystem,
+          sourceId,
+          aiReadDeferred: input.aiDeferred,
+        },
+        omit: { contentEncrypted: true },
+      });
+
+      if (episodeIds.length > 0) {
+        await linkTargets(tx, {
+          userId,
+          sourceKind: "document",
+          sourceId: created.id,
+          targetKind: "conditionEpisode",
+          targetIds: episodeIds,
+        });
+      }
+      if (encounterIds.length > 0) {
+        await linkTargets(tx, {
+          userId,
+          sourceKind: "document",
+          sourceId: created.id,
+          targetKind: "encounter",
+          targetIds: encounterIds,
+        });
+      }
+      return created;
+    });
+  } catch (err) {
+    if (err instanceof QuotaExceededError) {
+      return {
+        kind: "quotaExceeded",
+        usedBytes: err.usedBytes,
+        quotaBytes: limits.quotaBytes,
+      };
+    }
+    if (isP2002(err)) {
+      // A racing upload won one of the two partial unique indexes — the same
+      // source key, or the same bytes. Surface the winner exactly as the fast
+      // paths above would have.
+      if (sourceSystem && sourceId) {
+        const match = await findSourceKey(userId, sourceSystem, sourceId);
+        if (match) return { kind: "sourceKey", match };
+      }
+      const winner = await prisma.inboundDocument.findFirst({
+        where: { userId, contentSha256, deletedAt: null },
+        omit: { contentEncrypted: true },
+      });
+      if (winner) return answerDuplicate(input, winner);
+    }
+    throw err;
+  }
+
+  await auditLog("documents.inbound.store", {
+    userId,
+    ipAddress: input.ipAddress,
+    details: {
+      documentId: document.id,
+      mime: detected.mimeType,
+      ...(input.scoped ? { scoped: true } : {}),
+      ...(sourceSystem ? { sourceSystem } : {}),
+    },
+  });
+
+  annotate({
+    action: { name: "documents.vault.upload" },
+    meta: {
+      documentId: document.id,
+      byteSize: document.byteSize,
+      servingClass: detected.servingClass,
+      linked: episodeIds.length,
+      linkedVisits: encounterIds.length,
+      scoped: input.scoped,
+      sourceSystem,
+      aiDeferred: input.aiDeferred,
+    },
+  });
+
+  // Auto-index the freshly stored document for content search: enqueue a
+  // fire-and-forget background job (provider-first, local text-layer fallback).
+  // The caller never blocks on or fails because of indexing — the enqueue is
+  // not awaited and swallows its own errors (a missing boss or a transient
+  // send failure is a silent no-op). Only fresh inserts enqueue — a duplicate
+  // returns early above and never reaches here.
+  //
+  // `aiRead=defer` narrows it to the local text layer: an import of a whole
+  // archive must not turn into one provider call per file. The person reads
+  // them later, deliberately, from the document itself.
+  if (input.aiDeferred) {
+    void enqueueDocumentIndex(userId, document.id, { localOnly: true });
+  } else {
+    void enqueueDocumentIndex(userId, document.id);
+  }
+
+  // Render a preview thumbnail in the background too (pure local compute — no
+  // egress). Same fire-and-forget contract; a dropped enqueue is recoverable
+  // via the boot backfill.
+  void enqueueDocumentThumbnail(userId, document.id);
+
+  // Summarise the freshly stored document in the background — but ONLY when the
+  // `documentsAutoAiRead` opt-in is ON (the job re-checks it and the egress
+  // consent, and no-ops otherwise) and the `documentAi` capability is open for
+  // this record. Storing never depends on AI: the document is stored above
+  // whatever the capability says, and the index job still runs its
+  // provider-free text-layer path.
+  //
+  // A deferred upload skips it outright and keeps `summaryState: NONE`, so
+  // the detail sheet offers "Generate summary" rather than a pending state.
+  const documentAi = input.aiDeferred
+    ? { available: false as const, reason: "deferred" }
+    : await getAiCapability("documentAi");
+  if (documentAi.available) {
+    void enqueueDocumentSummary(userId, document.id);
+  } else {
+    annotate({
+      action: { name: "documents.summary.enqueueSkipped" },
+      meta: { documentId: document.id, reason: documentAi.reason },
+    });
+  }
+
+  return {
+    kind: "stored",
+    document,
+    servingClass: detected.servingClass,
+    linkedEpisodes: episodeIds.length,
+    linkedVisits: encounterIds.length,
+  };
+}