fix(documents): key imports by source instance, keep a token on its origin, and harden the import path

- Source keys carry the normalised origin of the Paperless-ngx or Papra
  they came from (source_instance on the document, the alias and the
  purge ledger; unique NULLS NOT DISTINCT). A key stored without one
  (v1.39.2 imports) matches any instance of its system, and a lookup
  without one matches any stored instance. The upload route, the lookup
  route and the import script accept or send it; backups carry it.
- Saving a connection reuses the stored token only for the same origin;
  a new origin needs the token again (422 tokenRequired).
- Document ids are checked against what each system issues (Paperless
  digits, Papra [A-Za-z0-9_-], never . or ..) before any URL is built;
  the Papra organization id likewise.
- Paperless API 9 filters dates on created__gte / created__lte, in the
  picker and the script. A search page the local re-check empties reads
  on, up to five source pages. Tags are read past the first 250.
- The import's key pre-check uses the lookup allowance, a failed fetch
  no longer charges the upload allowance, and one import per key runs at
  a time under an advisory lock, so a double click downloads once.
- Reading and deleting one's own connection works with the documents
  module off and the operator list unset.
This commit is contained in:
Marc-André Bombeck committed 2026-09-26 14:20:17 +02:00
1 parent 6029c7c30d
commit 1eac05b454
24 files changed
+886 -209

No files matched your search

@@ -51,10 +51,10 @@ vi.mock("@/lib/db", () => {
findFirst: vi.fn(),
},
documentImportKey: {
findUnique: vi.fn(),
findFirst: vi.fn(),
},
documentSourceAlias: {
findUnique: vi.fn(),
findFirst: vi.fn(),
createMany: vi.fn(),
count: vi.fn(),
},
@@ -239,10 +239,10 @@ beforeEach(() => {
[] as never,
);
vi.mocked(prisma.inboundDocument.findFirst).mockResolvedValue(null as never);
vi.mocked(prisma.documentImportKey.findUnique).mockResolvedValue(
vi.mocked(prisma.documentImportKey.findFirst).mockResolvedValue(
null as never,
);
vi.mocked(prisma.documentSourceAlias.findUnique).mockResolvedValue(
vi.mocked(prisma.documentSourceAlias.findFirst).mockResolvedValue(
null as never,
);
vi.mocked(prisma.documentSourceAlias.createMany).mockResolvedValue({
@@ -783,7 +783,7 @@ describe("POST /api/documents/inbound — document import", () => {
});
it("remembers a purged import through the ledger", async () => {
vi.mocked(prisma.documentImportKey.findUnique).mockResolvedValue({
vi.mocked(prisma.documentImportKey.findFirst).mockResolvedValue({
id: "key-1",
} as never);
const res = await post(
@@ -909,6 +909,8 @@ describe("POST /api/documents/inbound — document import", () => {
userId: "user-1",
documentId: "doc-manual",
sourceSystem: "PAPRA",
// Sent without an instance: the alias matches any Papra.
sourceInstance: null,
sourceId: "doc_9",
},
],
@@ -936,7 +938,7 @@ describe("POST /api/documents/inbound — document import", () => {
});
it("resolves a key held as an alias of a deleted document to deleted", async () => {
vi.mocked(prisma.documentSourceAlias.findUnique).mockResolvedValue({
vi.mocked(prisma.documentSourceAlias.findFirst).mockResolvedValue({
document: docRow({ id: "doc-gone", deletedAt: new Date() }),
} as never);
const res = await post(
+20 -2
View File
@@ -274,14 +274,17 @@ async function processUpload(
const url = new URL(request.url);
const querySystem = url.searchParams.get("sourceSystem");
const queryId = url.searchParams.get("sourceId");
const queryInstance = url.searchParams.get("sourceInstance");
let queryKey: {
sourceSystem: DocumentSourceSystemValue;
sourceId: string;
sourceInstance?: string;
} | null = null;
if (querySystem !== null || queryId !== null) {
const parsedKey = documentSourceKeySchema.safeParse({
sourceSystem: querySystem ?? undefined,
sourceId: queryId ?? undefined,
sourceInstance: queryInstance ?? undefined,
});
if (!parsedKey.success) {
return apiValidationError(
@@ -312,6 +315,7 @@ async function processUpload(
user.id,
queryKey.sourceSystem,
queryKey.sourceId,
queryKey.sourceInstance ?? null,
);
if (match) {
return answerSourceKey(
@@ -408,6 +412,7 @@ async function processUpload(
encounterIds: rawEncounterIds.length > 0 ? rawEncounterIds : undefined,
sourceSystem: formData.get("sourceSystem") ?? undefined,
sourceId: formData.get("sourceId") ?? undefined,
sourceInstance: formData.get("sourceInstance") ?? undefined,
aiRead: formData.get("aiRead") ?? undefined,
});
if (!parsed.success) {
@@ -425,10 +430,14 @@ async function processUpload(
// both they have to agree.
const formSystem = parsed.data.sourceSystem ?? null;
const formId = formSystem ? (parsed.data.sourceId ?? null) : null;
const formInstance = formId ? (parsed.data.sourceInstance ?? null) : null;
if (
queryKey &&
formSystem !== null &&
(formSystem !== queryKey.sourceSystem || formId !== queryKey.sourceId)
(formSystem !== queryKey.sourceSystem ||
formId !== queryKey.sourceId ||
(formInstance !== null &&
formInstance !== (queryKey.sourceInstance ?? null)))
) {
return apiError(
"The source key in the address and in the form differ.",
@@ -438,6 +447,9 @@ async function processUpload(
}
const sourceSystem = queryKey?.sourceSystem ?? formSystem;
const sourceId = queryKey?.sourceId ?? formId;
const sourceInstance = sourceId
? (queryKey?.sourceInstance ?? formInstance ?? null)
: null;
const ctx: UploadContext = {
userId: user.id,
scoped,
@@ -451,7 +463,12 @@ async function processUpload(
// count (the unique index has no `deleted_at` predicate), and past the purge
// the ledger remembers.
if (sourceSystem && sourceId && !queryKey) {
const match = await findSourceKey(user.id, sourceSystem, sourceId);
const match = await findSourceKey(
user.id,
sourceSystem,
sourceId,
sourceInstance,
);
if (match) return answerSourceKey(ctx, match);
}
@@ -478,6 +495,7 @@ async function processUpload(
encounterIds: parsed.data.encounterIds ?? [],
sourceSystem,
sourceId,
sourceInstance,
aiDeferred: ctx.aiDeferred,
sourceKeyChecked: true,
limits,
@@ -57,6 +57,7 @@ export const GET = apiHandler(async (request: Request) => {
const parsed = documentSourceKeySchema.safeParse({
sourceSystem: url.searchParams.get("sourceSystem") ?? undefined,
sourceId: url.searchParams.get("sourceId") ?? undefined,
sourceInstance: url.searchParams.get("sourceInstance") ?? undefined,
});
if (!parsed.success) {
return apiValidationError(
@@ -71,6 +72,7 @@ export const GET = apiHandler(async (request: Request) => {
user.id,
parsed.data.sourceSystem,
parsed.data.sourceId,
parsed.data.sourceInstance ?? null,
);
annotate({
action: { name: "documents.vault.sourceLookup" },
@@ -12,13 +12,16 @@
* imported (or already holding this key as an alias) is a `duplicate`,
* and one the person deleted here is `deleted` — nothing is fetched and
* nothing is stored, exactly as the token upload answers the same key;
* 3. only then the vault's own upload allowance (60 an hour, shared with
* uploads from the web and the phone) and an upload slot are taken, the
* document's details and its original file are fetched (bounded by the
* vault's per-file cap), and the file goes through `ingestDocument`, the
* same function the upload route stores through: type check, content
* dedup, quota, provenance (`sourceSystem` / `sourceId`), the background
* jobs;
* 3. only then is the source asked (on the search allowance, like every
* request to it) and an upload slot taken. One import per key and person
* runs at a time under an advisory lock, so a double click downloads the
* document once. The details and the original file are fetched (bounded
* by the vault's per-file cap); only a file that arrived is charged to
* the vault's own upload allowance (60 an hour, shared with uploads from
* the web and the phone), and it goes through `ingestDocument`, the same
* function the upload route stores through: type check, content dedup,
* quota, provenance (`sourceSystem` / `sourceId` / `sourceInstance`), the
* background jobs;
* 4. the document is linked to the record through `src/lib/links/`.
*
* AI reading follows the person's own setting, unlike the token upload's
@@ -48,18 +51,26 @@ import {
UPLOAD_LIMIT_PER_HOUR,
UPLOAD_WINDOW_MS,
} from "@/lib/documents/ingest";
import { findSourceKey, type SourceKeyMatch } from "@/lib/documents/source-key";
import {
checkSourceLookupRateLimit,
findSourceKey,
type SourceKeyMatch,
} from "@/lib/documents/source-key";
import type { SourceClient } from "@/lib/documents/sources/clients";
import { clientFor, loadConnection } from "@/lib/documents/sources/connections";
import {
admitDocumentSourceCaller,
checkSearchRateLimit,
responseForSourceFailure,
sourceErrorResponse,
systemParam,
} from "@/lib/documents/sources/route-support";
import type {
DocumentImportOutcome,
DocumentPickerLinkKind,
DocumentSourceImportDto,
import {
isSourceDocumentId,
type DocumentImportOutcome,
type DocumentPickerLinkKind,
type DocumentPickerSystem,
type DocumentSourceImportDto,
} from "@/lib/documents/sources/types";
import {
acquireDocumentUploadSlot,
@@ -69,11 +80,17 @@ import { linkTargets } from "@/lib/links";
import { annotate } from "@/lib/logging/context";
import { checkRateLimit, rateLimitHeaders } from "@/lib/rate-limit";
import { documentSourceImportSchema } from "@/lib/validations/document-sources";
import type { InboundDocumentKindValue } from "@/lib/validations/inbound-documents";
export const dynamic = "force-dynamic";
type RouteParams = { params: Promise<{ system: string }> };
/** How long an import waits for a database connection to hold its lock on. */
const IMPORT_LOCK_WAIT_MS = 10_000;
/** Longest a single import may hold its key: the download plus the store. */
const IMPORT_LOCK_HOLD_MS = 150_000;
interface LinkTarget {
kind: DocumentPickerLinkKind;
id: string;
@@ -172,6 +189,10 @@ export const POST = apiHandler(
);
}
const { sourceId, kind, link } = parsed.data;
// Only an id this system could have issued goes into a request path.
if (!isSourceDocumentId(system, sourceId)) {
return apiError("Not a document id of this archive.", 422);
}
const row = await loadConnection(userId, system);
if (!row) return sourceErrorResponse("notConnected");
@@ -181,25 +202,38 @@ export const POST = apiHandler(
} catch (err) {
return responseForSourceFailure(err);
}
const instance = client.instance;
if (link && !(await ownsLinkTarget(userId, link))) {
return sourceErrorResponse("linkTargetNotFound");
}
// The source key answers before anything is downloaded or charged.
const held = await findSourceKey(userId, system, sourceId);
// The source key answers before anything is downloaded or charged, on the
// same metered lookup allowance the upload's key check draws on.
const lookupRl = await checkSourceLookupRateLimit(
false,
auth.session.id,
userId,
);
if (!lookupRl.allowed) {
return sourceErrorResponse(
"rateLimited",
undefined,
rateLimitHeaders(lookupRl),
);
}
const held = await findSourceKey(userId, system, sourceId, instance);
if (held) return answerHeldKey(system, userId, link, held);
const rl = await checkRateLimit(
personalUploadBucket(userId),
UPLOAD_LIMIT_PER_HOUR,
UPLOAD_WINDOW_MS,
);
if (!rl.allowed) {
return apiError("Too many uploads. Try again later.", 429, {
errorCode: "documents.inbound.rateLimited",
headers: rateLimitHeaders(rl),
});
// A request to the source counts against the search allowance like any
// other; the upload allowance is charged only once a file arrived.
const fetchRl = await checkSearchRateLimit(userId);
if (!fetchRl.allowed) {
return sourceErrorResponse(
"rateLimited",
undefined,
rateLimitHeaders(fetchRl),
);
}
const release = acquireDocumentUploadSlot(userId);
@@ -216,83 +250,136 @@ export const POST = apiHandler(
}
try {
const limits = await resolveDocumentLimits(userId);
let meta;
let file;
try {
meta = await client.document(sourceId);
file = await client.download(sourceId, limits.maxFileBytes);
} catch (err) {
return responseForSourceFailure(err);
}
const result = await ingestDocument({
userId,
scoped: false,
ipAddress: getClientIp(request),
bytes: Buffer.from(file.bytes),
filename: meta.filename ?? file.filename,
title: meta.title,
kind: kind ?? null,
documentDate: meta.date,
episodeIds: [],
encounterIds: [],
sourceSystem: system,
sourceId,
aiDeferred: false,
sourceKeyChecked: true,
limits,
// The import is data; only the summary queued after a fresh insert
// follows the capability (and the person's auto-read setting).
documentAi: () => getAiCapability("documentAi"),
});
switch (result.kind) {
case "sourceKey":
return answerHeldKey(system, userId, link, result.match);
case "duplicate": {
// The same bytes were already in the vault (an upload, or another
// archive); the key is now remembered for that document.
const linked = await linkDocument(userId, link, result.document.id);
return respond(system, "duplicate", result.document.id, linked);
}
case "stored": {
const linked = await linkDocument(userId, link, result.document.id);
return respond(system, "imported", result.document.id, linked);
}
case "tooLarge":
return apiError("File is too large.", 413, {
errorCode: "documents.inbound.fileTooLarge",
reason: "fileTooLarge",
maxFileBytes: result.maxFileBytes,
// One import of one key at a time per person: a second tab or a double
// click waits here, then finds the key the first one stored, so the
// document is downloaded once. The lock is transaction-scoped and the
// transaction holds nothing else; the store runs in its own.
return await prisma.$transaction(
async (tx) => {
await tx.$queryRaw`
SELECT 1 AS locked
FROM pg_advisory_xact_lock(hashtextextended(${`documents-import:${userId}:${system}:${instance}:${sourceId}`}, 0))
`;
const raced = await findSourceKey(userId, system, sourceId, instance);
if (raced) return answerHeldKey(system, userId, link, raced);
return importOne({
request,
client,
userId,
system,
instance,
sourceId,
kind: kind ?? null,
link,
});
case "empty":
case "unsupportedType":
return apiError("This file type is not supported.", 415, {
errorCode: "documents.inbound.fileType",
reason: "unsupportedType",
});
case "quotaExceeded":
return apiError("Storage quota exceeded.", 413, {
errorCode: "documents.inbound.quotaExceeded",
reason: "quotaExceeded",
quotaBytes: result.quotaBytes,
usedBytes: result.usedBytes,
});
case "aliasLimit":
return apiError(
"This file is already stored under too many source ids.",
409,
{ errorCode: "documents.inbound.sourceAliasLimit" },
);
case "episodeNotFound":
case "encounterNotFound":
// No pre-links are passed on this path; unreachable, but a result
// the switch does not name must not fall through silently.
return sourceErrorResponse("linkTargetNotFound");
}
},
{ maxWait: IMPORT_LOCK_WAIT_MS, timeout: IMPORT_LOCK_HOLD_MS },
);
} finally {
release();
}
},
);
/** Fetch one document and store it; the caller holds the key's lock. */
async function importOne(args: {
request: Request;
client: SourceClient;
userId: string;
system: DocumentPickerSystem;
instance: string;
sourceId: string;
kind: InboundDocumentKindValue | null;
link: LinkTarget | undefined;
}): Promise<Response> {
const { request, client, userId, system, instance, sourceId, kind, link } =
args;
const limits = await resolveDocumentLimits(userId);
let meta;
let file;
try {
meta = await client.document(sourceId);
file = await client.download(sourceId, limits.maxFileBytes);
} catch (err) {
return responseForSourceFailure(err);
}
const rl = await checkRateLimit(
personalUploadBucket(userId),
UPLOAD_LIMIT_PER_HOUR,
UPLOAD_WINDOW_MS,
);
if (!rl.allowed) {
return apiError("Too many uploads. Try again later.", 429, {
errorCode: "documents.inbound.rateLimited",
headers: rateLimitHeaders(rl),
});
}
const result = await ingestDocument({
userId,
scoped: false,
ipAddress: getClientIp(request),
bytes: Buffer.from(file.bytes),
filename: meta.filename ?? file.filename,
title: meta.title,
kind,
documentDate: meta.date,
episodeIds: [],
encounterIds: [],
sourceSystem: system,
sourceId,
sourceInstance: instance,
aiDeferred: false,
sourceKeyChecked: true,
limits,
// The import is data; only the summary queued after a fresh insert
// follows the capability (and the person's auto-read setting).
documentAi: () => getAiCapability("documentAi"),
});
switch (result.kind) {
case "sourceKey":
return answerHeldKey(system, userId, link, result.match);
case "duplicate": {
// The same bytes were already in the vault (an upload, or another
// archive); the key is now remembered for that document.
const linked = await linkDocument(userId, link, result.document.id);
return respond(system, "duplicate", result.document.id, linked);
}
case "stored": {
const linked = await linkDocument(userId, link, result.document.id);
return respond(system, "imported", result.document.id, linked);
}
case "tooLarge":
return apiError("File is too large.", 413, {
errorCode: "documents.inbound.fileTooLarge",
reason: "fileTooLarge",
maxFileBytes: result.maxFileBytes,
});
case "empty":
case "unsupportedType":
return apiError("This file type is not supported.", 415, {
errorCode: "documents.inbound.fileType",
reason: "unsupportedType",
});
case "quotaExceeded":
return apiError("Storage quota exceeded.", 413, {
errorCode: "documents.inbound.quotaExceeded",
reason: "quotaExceeded",
quotaBytes: result.quotaBytes,
usedBytes: result.usedBytes,
});
case "aliasLimit":
return apiError(
"This file is already stored under too many source ids.",
409,
{ errorCode: "documents.inbound.sourceAliasLimit" },
);
case "episodeNotFound":
case "encounterNotFound":
// No pre-links are passed on this path; unreachable, but a result
// the switch does not name must not fall through silently.
return sourceErrorResponse("linkTargetNotFound");
}
}
@@ -6,8 +6,9 @@
* (`DOCUMENT_SOURCE_ORIGINS`), tests the connection live before storing
* anything (a wrong token or an unreachable host is reported here, not on the
* first search), then stores the token encrypted. A PUT without a token keeps
* the stored one, so the address or the Papra organization can change without
* typing the key again; the first save must carry it. The response is the
* the stored one, so the base path or the Papra organization can change
* without typing the key again, but only on the same origin: a new origin, and
* the first save, must carry it (422 `documents.sources.tokenRequired`). The response is the
* connection without its token, like every read of it.
*
* DELETE removes the row, and with it the token.
@@ -100,8 +101,16 @@ export const PUT = apiHandler(
}
const existing = await loadConnection(userId, system);
// The stored token is only ever sent back to the origin it was saved
// for. A new address on another origin needs the token typed again, or
// saving an address would hand the old archive's token to whatever the
// new one is.
const reusable =
existing !== null &&
evaluateSourceBaseUrl(existing.baseUrl, new Set([verdict.origin])).ok;
const token =
parsed.data.token ?? (existing ? decrypt(existing.tokenEncrypted) : null);
parsed.data.token ??
(existing && reusable ? decrypt(existing.tokenEncrypted) : null);
if (!token) {
return apiError("Enter the API token.", 422, {
errorCode: "documents.sources.tokenRequired",
@@ -165,9 +174,13 @@ export const PUT = apiHandler(
export const DELETE = apiHandler(
async (request: NextRequest, { params }: RouteParams) => {
const auth = await requireAuth();
// No list needed to forget a connection: an operator who removed the
// origin must not leave the person unable to delete the stored token.
const refused = await admitDocumentSourceCaller(auth, { needsList: false });
// Neither the list nor the module is needed to forget a connection: an
// operator who removed the origin, or a person who switched Documents
// off, must still be able to delete the stored token.
const refused = await admitDocumentSourceCaller(auth, {
needsList: false,
needsModule: false,
});
if (refused) return refused;
const system = await systemParam(params);
if (system instanceof Response) return system;
@@ -9,6 +9,10 @@
* `new`, `imported` (with the vault document's id) or `deleted` (the person
* deleted it here; importing it again is refused). Nothing is stored.
*
* When the source ignores a filter and the local re-check empties a page, the
* route reads on (up to five source pages) and answers with the last page it
* read, so the client's "load more" continues after it.
*
* Sixty searches a minute, shared with the tag read. Cookie-only and
* owner-only; see `admitDocumentSourceCaller`.
*/
@@ -30,9 +34,10 @@ import {
sourceErrorResponse,
systemParam,
} from "@/lib/documents/sources/route-support";
import type {
DocumentSourceResultDto,
DocumentSourceSearchDto,
import {
DOCUMENT_PICKER_MAX_PAGE,
type DocumentSourceResultDto,
type DocumentSourceSearchDto,
} from "@/lib/documents/sources/types";
import { annotate } from "@/lib/logging/context";
import { rateLimitHeaders } from "@/lib/rate-limit";
@@ -40,6 +45,9 @@ import { documentSourceSearchSchema } from "@/lib/validations/document-sources";
export const dynamic = "force-dynamic";
/** Source pages one search request may read to find a non-empty one. */
const SEARCH_PAGES_PER_REQUEST = 5;
type RouteParams = { params: Promise<{ system: string }> };
export const GET = apiHandler(
@@ -77,16 +85,34 @@ export const GET = apiHandler(
);
}
let found: { items: SourceListItem[]; hasMore: boolean };
// A page the local tag/date re-check emptied is not an answer: the next
// source page may hold matches. Read on, a bounded number of pages, and
// report the last one read so "load more" continues after it.
let found: { items: SourceListItem[]; hasMore: boolean } = {
items: [],
hasMore: false,
};
let lastPage = page;
let instance: string;
try {
const client = await openConnection(userId, system);
found = await client.search({
q,
tagId: tag ?? null,
from: from ?? null,
to: to ?? null,
page,
});
instance = client.instance;
for (
let next = page;
next < page + SEARCH_PAGES_PER_REQUEST &&
next <= DOCUMENT_PICKER_MAX_PAGE;
next += 1
) {
found = await client.search({
q,
tagId: tag ?? null,
from: from ?? null,
to: to ?? null,
page: next,
});
lastPage = next;
if (found.items.length > 0 || !found.hasMore) break;
}
} catch (err) {
return responseForSourceFailure(err);
}
@@ -95,6 +121,7 @@ export const GET = apiHandler(
userId,
system,
found.items.map((item) => item.sourceId),
instance,
);
const results: DocumentSourceResultDto[] = found.items.map((item) => {
const held = states.get(item.sourceId);
@@ -116,6 +143,7 @@ export const GET = apiHandler(
meta: {
system,
page,
pagesRead: lastPage - page + 1,
count: results.length,
filtered: Boolean(q || tag || from || to),
},
@@ -123,8 +151,8 @@ export const GET = apiHandler(
const body: DocumentSourceSearchDto = {
results,
page,
hasMore: found.hasMore,
page: lastPage,
hasMore: found.hasMore && lastPage < DOCUMENT_PICKER_MAX_PAGE,
};
return apiSuccess(body);
},
+13 -6
View File
@@ -8,9 +8,10 @@
* No request leaves the server here.
*
* Cookie-only and owner-only like every route under this path
* (`admitDocumentSourceCaller`). With the list unset it answers 200
* `available: false` rather than 404, because this is the one read the client
* makes to decide whether to show the picker at all.
* (`admitDocumentSourceCaller`), but it needs neither the documents module nor
* the list: with the list unset it answers 200 `available: false` (plus any
* rows still stored, so they can be deleted) rather than 404, because this is
* the one read the client makes to decide what to show.
*/
import { apiHandler, requireAuth } from "@/lib/api-handler";
import { apiSuccess } from "@/lib/api-response";
@@ -30,21 +31,27 @@ export const dynamic = "force-dynamic";
export const GET = apiHandler(async () => {
const auth = await requireAuth();
const refused = await admitDocumentSourceCaller(auth, { needsList: false });
const refused = await admitDocumentSourceCaller(auth, {
needsList: false,
needsModule: false,
});
if (refused) return refused;
annotate({ action: { name: "documents.sources.status" } });
// Rows are listed even with the picker off: a connection whose origin the
// operator removed (or whose list is gone) must stay visible so it can be
// deleted. `originAllowed` is false for each of them then.
const rows = await listConnections(auth.user.id);
if (!documentSourcesAvailable()) {
const off: DocumentSourcesStatusDto = {
available: false,
allowedOrigins: [],
connections: [],
connections: rows.map(toConnectionDto),
};
return apiSuccess(off);
}
const rows = await listConnections(auth.user.id);
const status: DocumentSourcesStatusDto = {
available: true,
allowedOrigins: listedDocumentSourceOrigins(),