diff --git a/src/app/api/documents/inbound/route.ts b/src/app/api/documents/inbound/route.ts index 70097db7b..f2f08dd2d 100644 --- a/src/app/api/documents/inbound/route.ts +++ b/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>` - 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([ diff --git a/src/lib/documents/ingest.ts b/src/lib/documents/ingest.ts new file mode 100644 index 000000000..06614a5c9 --- /dev/null +++ b/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 { + // 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 { + 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>` + 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, + }; +}