danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(pdf): refactor parse operation orchestration and client readiness flow

Browse source 
Introduce startPdfParseOperation to encapsulate worker operation creation, job rate tracking, and parse state initialization for PDFs. Replace scattered operation setup logic in register-upload, docx-to-pdf upload, and parse-state-backfill with this unified helper. Update job enqueuing to propagate initial opId/jobId/status for accurate state tracking.

Change getParsedPdfDocument to throw on non-ready states and remove polling logic from usePdfDocument, shifting readiness detection to event streams. Switch API responses for non-ready parses from 202 to 409 to clarify client expectations and simplify error handling.

Add targeted tests to verify legacy backfill behavior and ensure no worker operation is created for pure data fetches of pending legacy PDFs.

BREAKING CHANGE: API now returns 409 Conflict for non-ready parsed PDFs instead of 202, and clients must subscribe to event streams for parse progress.
This commit is contained in:
Richard R 2026-06-04 10:52:49 -06:00
parent d489356b00
commit 62c913e96d
12 changed files with 357 additions and 387 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

118
src/app/(app)/pdf/[id]/usePdfDocument.ts
View file

@ -169,53 +169,29 @@ export function usePdfDocument(): PdfDocumentState {
// Guards for setCurrentDocument to prevent stale loads from overwriting newer selections. // Guards for setCurrentDocument to prevent stale loads from overwriting newer selections.
const docLoadSeqRef = useRef(0); const docLoadSeqRef = useRef(0);
const docLoadAbortRef = useRef<AbortController | null>(null); const docLoadAbortRef = useRef<AbortController | null>(null);
const parsePollAbortRef = useRef<AbortController | null>(null); const parseStreamAbortRef = useRef<AbortController | null>(null);
const parseSseCloseRef = useRef<(() => void) | null>(null); const parseSseCloseRef = useRef<(() => void) | null>(null);
const lastPreparedPlaybackPageRef = useRef<number | null>(null); const lastPreparedPlaybackPageRef = useRef<number | null>(null);
const fetchParsedDocument = useCallback(async ( const loadParsedDocumentOnce = useCallback(async (
documentId: string, documentId: string,
initialStatus: PdfParseStatus | null,
signal: AbortSignal, signal: AbortSignal,
initialOpId?: string | null,
): Promise<void> => { ): Promise<void> => {
// Legacy PDFs may have null parseStatus; treat as pending so opening the if (signal.aborted) return;
// document backfills parse output via the parsed endpoint polling path. const parsed = await getParsedPdfDocument(documentId, { signal });
const effectiveInitialStatus: PdfParseStatus = initialStatus ?? 'pending'; if (signal.aborted) return;
setParseStatus(effectiveInitialStatus); setParsedDocument(parsed);
setParseStatus('ready');
setParseProgress(null); setParseProgress(null);
const delayMs = 1200; setActiveParseOpId(null);
const retryFailed = effectiveInitialStatus === 'failed'; }, []);
let attempt = 0;
let effectiveOpId = initialOpId?.trim() || null; const resetParsedDocumentState = useCallback(() => {
while (!signal.aborted) { setParsedDocument(null);
if (signal.aborted) return; setCurrDocText(undefined);
const result = await getParsedPdfDocument(documentId, { setIsPlaybackReady(false);
signal, lastPreparedPlaybackPageRef.current = null;
retryFailed: retryFailed && attempt === 0, pageTextCacheRef.current.clear();
...(effectiveOpId ? { opId: effectiveOpId } : {}),
});
if (result.status === 'ready') {
setParsedDocument(result.parsed);
setParseStatus('ready');
setParseProgress(null);
setActiveParseOpId(null);
return;
}
if ('opId' in result && typeof result.opId === 'string' && result.opId.trim()) {
effectiveOpId = result.opId.trim();
setActiveParseOpId(effectiveOpId);
}
setParseStatus(result.status);
setParseProgress(result.parseProgress ?? null);
if (result.status === 'failed') {
setParsedDocument(null);
setActiveParseOpId(null);
return;
}
attempt += 1;
await new Promise((resolve) => setTimeout(resolve, delayMs));
}
}, []); }, []);
const fetchDocumentSettings = useCallback(async (documentId: string, signal: AbortSignal): Promise<void> => { const fetchDocumentSettings = useCallback(async (documentId: string, signal: AbortSignal): Promise<void> => {
@ -229,14 +205,14 @@ export function usePdfDocument(): PdfDocumentState {
} }
}, []); }, []);
const startParsedPolling = useCallback((documentId: string, initialOpId?: string | null) => { const startParsedEventStream = useCallback((documentId: string, initialOpId?: string | null) => {
parsePollAbortRef.current?.abort(); parseStreamAbortRef.current?.abort();
parseSseCloseRef.current?.(); parseSseCloseRef.current?.();
parseSseCloseRef.current = null; parseSseCloseRef.current = null;
setParseProgress(null); setParseProgress(null);
setActiveParseOpId(initialOpId?.trim() || null); setActiveParseOpId(initialOpId?.trim() || null);
const controller = new AbortController(); const controller = new AbortController();
parsePollAbortRef.current = controller; parseStreamAbortRef.current = controller;
const closeSse = subscribeParsedPdfDocumentEvents(documentId, { const closeSse = subscribeParsedPdfDocumentEvents(documentId, {
opId: initialOpId?.trim() || null, opId: initialOpId?.trim() || null,
@ -248,25 +224,27 @@ export function usePdfDocument(): PdfDocumentState {
} }
setParseStatus(snapshot.parseStatus); setParseStatus(snapshot.parseStatus);
setParseProgress(snapshot.parseProgress); setParseProgress(snapshot.parseProgress);
if (snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed') { if (snapshot.parseStatus === 'ready') {
if (snapshot.parseStatus === 'failed') {
setParsedDocument(null);
setIsPlaybackReady(false);
lastPreparedPlaybackPageRef.current = null;
setActiveParseOpId(null);
} else {
void fetchParsedDocument(
documentId,
'ready',
controller.signal,
typeof snapshot.opId === 'string' ? snapshot.opId : (initialOpId ?? null),
);
}
closeSse(); closeSse();
parseSseCloseRef.current = null; parseSseCloseRef.current = null;
if (parsePollAbortRef.current === controller) { if (parseStreamAbortRef.current === controller) {
parsePollAbortRef.current = null; parseStreamAbortRef.current = null;
} }
void loadParsedDocumentOnce(documentId, controller.signal).catch((error) => {
if (error instanceof DOMException && error.name === 'AbortError') return;
console.error('Failed to load parsed PDF after ready status:', error);
resetParsedDocumentState();
});
return;
}
if (snapshot.parseStatus === 'failed') {
closeSse();
parseSseCloseRef.current = null;
if (parseStreamAbortRef.current === controller) {
parseStreamAbortRef.current = null;
}
resetParsedDocumentState();
setActiveParseOpId(null);
return; return;
} }
}, },
@ -286,11 +264,11 @@ export function usePdfDocument(): PdfDocumentState {
if (parseSseCloseRef.current === closeSse) { if (parseSseCloseRef.current === closeSse) {
parseSseCloseRef.current = null; parseSseCloseRef.current = null;
} }
if (parsePollAbortRef.current === controller) { if (parseStreamAbortRef.current === controller) {
parsePollAbortRef.current = null; parseStreamAbortRef.current = null;
} }
}, { once: true }); }, { once: true });
}, [fetchParsedDocument, setActiveParseOpId]); }, [loadParsedDocumentOnce, resetParsedDocumentState, setActiveParseOpId]);
useEffect(() => { useEffect(() => {
pdfDocumentRef.current = pdfDocument; pdfDocumentRef.current = pdfDocument;
@ -472,8 +450,8 @@ export function usePdfDocument(): PdfDocumentState {
// or fast refresh. // or fast refresh.
pdfDocGenerationRef.current += 1; pdfDocGenerationRef.current += 1;
loadSeqRef.current += 1; loadSeqRef.current += 1;
parsePollAbortRef.current?.abort(); parseStreamAbortRef.current?.abort();
parsePollAbortRef.current = null; parseStreamAbortRef.current = null;
parseSseCloseRef.current?.(); parseSseCloseRef.current?.();
parseSseCloseRef.current = null; parseSseCloseRef.current = null;
pageTextCacheRef.current.clear(); pageTextCacheRef.current.clear();
@ -502,7 +480,7 @@ export function usePdfDocument(): PdfDocumentState {
setParseStatus(initialParseStatus); setParseStatus(initialParseStatus);
setParseProgress(null); setParseProgress(null);
setActiveParseOpId(null); setActiveParseOpId(null);
startParsedPolling(id, null); startParsedEventStream(id, null);
void fetchDocumentSettings(id, controller.signal); void fetchDocumentSettings(id, controller.signal);
} }
@ -537,7 +515,7 @@ export function usePdfDocument(): PdfDocumentState {
setCurrDocText, setCurrDocText,
setPdfDocument, setPdfDocument,
fetchDocumentSettings, fetchDocumentSettings,
startParsedPolling, startParsedEventStream,
]); ]);
const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise<void> => { const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise<void> => {
@ -564,11 +542,11 @@ export function usePdfDocument(): PdfDocumentState {
setParseStatus(forced.status); setParseStatus(forced.status);
setParseProgress(null); setParseProgress(null);
setActiveParseOpId(forced.opId ?? null); setActiveParseOpId(forced.opId ?? null);
startParsedPolling(currDocId, forced.opId ?? null); startParsedEventStream(currDocId, forced.opId ?? null);
} catch (error) { } catch (error) {
console.error('Failed to force PDF reparse:', error); console.error('Failed to force PDF reparse:', error);
} }
}, [currDocId, startParsedPolling]); }, [currDocId, startParsedEventStream]);
/** /**
* Clears the current document state * Clears the current document state
@ -582,8 +560,8 @@ export function usePdfDocument(): PdfDocumentState {
docLoadSeqRef.current += 1; docLoadSeqRef.current += 1;
docLoadAbortRef.current?.abort(); docLoadAbortRef.current?.abort();
docLoadAbortRef.current = null; docLoadAbortRef.current = null;
parsePollAbortRef.current?.abort(); parseStreamAbortRef.current?.abort();
parsePollAbortRef.current = null; parseStreamAbortRef.current = null;
parseSseCloseRef.current?.(); parseSseCloseRef.current?.();
parseSseCloseRef.current = null; parseSseCloseRef.current = null;
setCurrDocId(undefined); setCurrDocId(undefined);

47
src/app/api/documents/[id]/parsed/events/route.ts
View file

@ -5,7 +5,11 @@ import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth'; import { requireAuthContext } from '@/lib/server/auth/auth';
import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker';
import { isAbortLikeError } from '@/lib/server/compute/abort-like-error'; import { isAbortLikeError } from '@/lib/server/compute/abort-like-error';
import { isWorkerOperationStateStale, snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; import {
isWorkerOperationStateStale,
mergeNonReadyParseSnapshot,
snapshotFromWorkerState,
} from '@/lib/server/compute/worker-parse-state';
import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state';
import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { isValidDocumentId } from '@/lib/server/documents/blobstore';
import { import {
@ -83,13 +87,20 @@ async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Pr
&& workerState.opId === opId && workerState.opId === opId
&& !isWorkerOperationStateStale(workerState, opStaleMs) && !isWorkerOperationStateStale(workerState, opStaleMs)
) { ) {
const merged = mergeNonReadyParseSnapshot({
parseStatus,
parseProgress: state.progress ?? null,
workerState,
});
const workerSnapshot = snapshotFromWorkerState(workerState);
const fromWorker = workerSnapshot.parseStatus === 'pending' || workerSnapshot.parseStatus === 'running';
return { return {
snapshot: { snapshot: {
...snapshotFromWorkerState(workerState), ...merged,
opId: workerState.opId, opId: workerState.opId,
}, },
opId: workerState.opId, opId: workerState.opId,
fromWorker: true, fromWorker,
}; };
} }
} }
@ -262,22 +273,19 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
let current = initialState.snapshot; let current = initialState.snapshot;
let signature = JSON.stringify(current); let signature = JSON.stringify(current);
let currentOpId = requestedOpId ?? initialState.opId; let currentOpId = requestedOpId ?? initialState.opId;
let currentFromWorker = initialState.fromWorker;
let lastEventId: number | null = null; let lastEventId: number | null = null;
let loggedMissingOpId = false; let loggedMissingOpId = false;
const pinnedRequestedOp = requestedOpId ?? null; const pinnedRequestedOp = requestedOpId ?? null;
const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot, fromWorker: boolean): boolean => { const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot): boolean => {
const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed'; const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed';
if (!isTerminal) return false; if (!isTerminal) return false;
// If caller pinned an opId, keep streaming until worker confirms that if (pinnedRequestedOp && snapshot.opId !== pinnedRequestedOp) return false;
// op state. DB fallback can report stale terminal status for other rows.
if (pinnedRequestedOp && snapshot.opId === pinnedRequestedOp && !fromWorker) return false;
return true; return true;
}; };
writeSnapshot(current); writeSnapshot(current);
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { if (shouldCloseForTerminalSnapshot(current)) {
closeStream(); closeStream();
return; return;
} }
@ -304,7 +312,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
current = next.snapshot; current = next.snapshot;
signature = next.signature; signature = next.signature;
currentFromWorker = next.fromWorker;
if (!requestedOpId && next.opId !== currentOpId) { if (!requestedOpId && next.opId !== currentOpId) {
currentOpId = next.opId; currentOpId = next.opId;
lastEventId = null; lastEventId = null;
@ -313,7 +320,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
workerAbort = null; workerAbort = null;
} }
} }
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { if (shouldCloseForTerminalSnapshot(current)) {
closeStream(); closeStream();
} }
}).catch((error) => { }).catch((error) => {
@ -350,7 +357,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
current = next.snapshot; current = next.snapshot;
signature = next.signature; signature = next.signature;
currentFromWorker = next.fromWorker;
currentOpId = requestedOpId ?? next.opId; currentOpId = requestedOpId ?? next.opId;
if (!currentOpId) { if (!currentOpId) {
if (!loggedMissingOpId) { if (!loggedMissingOpId) {
@ -358,7 +364,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
logger.warn({ logger.warn({
event: 'documents.parsed.events.missing_opid_non_terminal', event: 'documents.parsed.events.missing_opid_non_terminal',
degraded: true, degraded: true,
step: 'poll_without_worker_op', step: 'missing_opid_fallback',
documentId: id, documentId: id,
storageUserIdHash, storageUserIdHash,
parseStatus: current.parseStatus, parseStatus: current.parseStatus,
@ -368,7 +374,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} else if (loggedMissingOpId) { } else if (loggedMissingOpId) {
loggedMissingOpId = false; loggedMissingOpId = false;
} }
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { if (shouldCloseForTerminalSnapshot(current)) {
closeStream(); closeStream();
return; return;
} }
@ -472,19 +478,23 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
); );
if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue;
const mergedSnapshot = mergeNonReadyParseSnapshot({
parseStatus: current.parseStatus,
parseProgress: current.parseProgress,
workerState: workerSnapshot,
});
const nextSnapshot: ParsedSnapshot = { const nextSnapshot: ParsedSnapshot = {
...snapshotFromWorkerState(workerSnapshot), ...mergedSnapshot,
opId: workerSnapshot.opId, opId: workerSnapshot.opId,
}; };
const nextSignature = JSON.stringify(nextSnapshot); const nextSignature = JSON.stringify(nextSnapshot);
if (nextSignature !== signature) { if (nextSignature !== signature) {
current = nextSnapshot; current = nextSnapshot;
signature = nextSignature; signature = nextSignature;
currentFromWorker = true;
writeSnapshot(current); writeSnapshot(current);
} }
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { if (shouldCloseForTerminalSnapshot(current)) {
closeStream(); closeStream();
return; return;
} }
@ -523,12 +533,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
current = next.snapshot; current = next.snapshot;
signature = next.signature; signature = next.signature;
currentFromWorker = next.fromWorker;
if (!requestedOpId && next.opId !== currentOpId) { if (!requestedOpId && next.opId !== currentOpId) {
currentOpId = next.opId; currentOpId = next.opId;
lastEventId = null; lastEventId = null;
} }
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { if (shouldCloseForTerminalSnapshot(current)) {
closeStream(); closeStream();
return; return;
} }

233
src/app/api/documents/[id]/parsed/route.ts
View file

@ -4,41 +4,36 @@ import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db'; import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth'; import { requireAuthContext } from '@/lib/server/auth/auth';
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create';
import { import {
documentParseStateFromWorkerState,
isWorkerOperationStateStale, isWorkerOperationStateStale,
snapshotFromWorkerState, snapshotFromWorkerState,
} from '@/lib/server/compute/worker-parse-state'; } from '@/lib/server/compute/worker-parse-state';
import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state';
import { import {
documentKey,
getParsedDocumentBlob, getParsedDocumentBlob,
getParsedDocumentBlobByKey, getParsedDocumentBlobByKey,
isMissingBlobError, isMissingBlobError,
isValidDocumentId, isValidDocumentId,
putParsedDocumentBlob,
} from '@/lib/server/documents/blobstore'; } from '@/lib/server/documents/blobstore';
import { import {
normalizeDocumentParseStateForCurrentParserVersion, normalizeDocumentParseStateForCurrentParserVersion,
normalizeParseStatus, normalizeParseStatus,
parseDocumentParseState, parseDocumentParseState,
resolveParsedPdfParserVersion,
stringifyDocumentParseState, stringifyDocumentParseState,
} from '@/lib/server/documents/parse-state'; } from '@/lib/server/documents/parse-state';
import { backfillPendingPdfParseOperation } from '@/lib/server/documents/parse-state-backfill'; import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing'; import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { checkJobRate, recordJobEvent, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter'; import { checkJobRate, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter';
import { buildComputeRateLimitedResponse } from '@/lib/server/rate-limit/problem-response'; import { buildComputeRateLimitedResponse } from '@/lib/server/rate-limit/problem-response';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { isS3Configured } from '@/lib/server/storage/s3'; import { isS3Configured } from '@/lib/server/storage/s3';
import { createRequestLogger, hashForLog, type ServerLogger } from '@/lib/server/logger'; import { createRequestLogger, hashForLog } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response'; import { errorResponse } from '@/lib/server/errors/next-response';
import { logDegraded } from '@/lib/server/errors/logging';
import type { ParsedPdfDocument } from '@/types/parsed-pdf'; import type { ParsedPdfDocument } from '@/types/parsed-pdf';
import { getComputeOpStaleMs } from '@openreader/compute-core'; import { getComputeOpStaleMs } from '@openreader/compute-core';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; import type { PdfLayoutJobResult } from '@openreader/compute-core/api-contracts';
export const dynamic = 'force-dynamic'; export const dynamic = 'force-dynamic';
@ -102,105 +97,12 @@ async function writeParseRowState(input: {
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
} }
async function finalizeFromWorkerState(input: {
workerState: WorkerOperationState<PdfLayoutJobResult>;
row: ParseRow;
namespace: string | null;
logger: ServerLogger;
}): Promise<NextResponse> {
const snapshot = snapshotFromWorkerState(input.workerState);
const workerParseState = documentParseStateFromWorkerState(input.workerState);
if (snapshot.parseStatus === 'pending' || snapshot.parseStatus === 'running') {
await writeParseRowState({
documentId: input.row.id,
userId: input.row.userId,
parseState: stringifyDocumentParseState(workerParseState),
});
return NextResponse.json({
parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress,
opId: input.workerState.opId,
}, { status: 202 });
}
if (snapshot.parseStatus === 'failed') {
await writeParseRowState({
documentId: input.row.id,
userId: input.row.userId,
parseState: stringifyDocumentParseState(workerParseState),
});
return NextResponse.json({
parseStatus: 'failed',
parseProgress: null,
opId: input.workerState.opId,
error: input.workerState.error?.message ?? 'Worker parse failed',
}, { status: 202 });
}
let parsedJsonKey: string | null = null;
if (input.workerState.result && 'parsedObjectKey' in input.workerState.result) {
parsedJsonKey = typeof input.workerState.result.parsedObjectKey === 'string'
? input.workerState.result.parsedObjectKey
: null;
}
if (!parsedJsonKey && input.workerState.result && 'parsed' in input.workerState.result && input.workerState.result.parsed) {
const parsedJson = Buffer.from(JSON.stringify(input.workerState.result.parsed));
parsedJsonKey = await putParsedDocumentBlob(input.row.id, parsedJson, input.namespace);
}
if (!parsedJsonKey) {
return errorResponse(new Error('Worker completed without parsed output'), {
apiErrorMessage: 'Worker completed without parsed output',
normalize: { code: 'DOCUMENTS_PARSED_WORKER_OUTPUT_MISSING', errorClass: 'upstream' },
});
}
const json = await getParsedDocumentBlobByKey(parsedJsonKey);
let parsedDoc: ParsedPdfDocument | null = null;
try {
parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument;
} catch {
parsedDoc = null;
}
await writeParseRowState({
documentId: input.row.id,
userId: input.row.userId,
parseState: stringifyDocumentParseState({
...workerParseState,
parserVersion: resolveParsedPdfParserVersion(parsedDoc),
}),
parsedJsonKey,
});
if (!hasAnyParsedBlocks(parsedDoc)) {
input.logger.warn({
event: 'documents.parsed.no_blocks_in_output',
documentId: input.row.id,
userIdHash: hashForLog(input.row.userId),
parsedJsonKey,
}, 'Worker output parsed successfully but contained no blocks');
}
return new NextResponse(new Uint8Array(json), {
status: 200,
headers: {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
},
});
}
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
const { logger } = createRequestLogger({ const { logger } = createRequestLogger({
route: '/api/documents/[id]/parsed', route: '/api/documents/[id]/parsed',
request: req, request: req,
}); });
try { try {
const opStaleMs = getComputeOpStaleMs();
if (!isS3Configured()) return s3NotConfiguredResponse(); if (!isS3Configured()) return s3NotConfiguredResponse();
const authCtxOrRes = await requireAuthContext(req); const authCtxOrRes = await requireAuthContext(req);
@ -209,8 +111,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
const params = await ctx.params; const params = await ctx.params;
const id = (params.id || '').trim().toLowerCase(); const id = (params.id || '').trim().toLowerCase();
const retryFailed = req.nextUrl.searchParams.get('retry') === '1';
const requestedOpId = normalizeOpId(req.nextUrl.searchParams.get('opId'));
if (!isValidDocumentId(id)) { if (!isValidDocumentId(id)) {
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
} }
@ -225,112 +125,17 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
return NextResponse.json({ error: 'Not found' }, { status: 404 }); return NextResponse.json({ error: 'Not found' }, { status: 404 });
} }
if (requestedOpId) { const state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState));
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(requestedOpId);
if (workerState && workerState.opId === requestedOpId) {
return finalizeFromWorkerState({
workerState,
row,
namespace: testNamespace,
logger,
});
}
logDegraded(logger, {
event: 'documents.parsed.requested_op_unavailable',
msg: 'Requested worker operation id was unavailable',
step: 'requested_op_lookup',
context: {
documentId: id,
userIdHash: hashForLog(row.userId),
opId: requestedOpId,
error: {
name: 'RequestedOperationUnavailable',
message: `Requested worker operation ${requestedOpId} was unavailable`,
},
},
});
}
let state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState));
state = await healStaleDocumentParseState({
documentId: id,
userId: row.userId,
state,
});
const effectiveStatus = normalizeParseStatus(state.status); const effectiveStatus = normalizeParseStatus(state.status);
const effectiveProgress = state.progress ?? null; const effectiveProgress = state.progress ?? null;
const effectiveOpId = normalizeOpId(state.opId); const effectiveOpId = normalizeOpId(state.opId);
if (effectiveOpId && effectiveStatus !== 'ready') {
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(effectiveOpId);
if (
workerState
&& workerState.opId === effectiveOpId
&& !isWorkerOperationStateStale(workerState, opStaleMs)
) {
return finalizeFromWorkerState({
workerState,
row,
namespace: testNamespace,
logger,
});
}
}
if (!effectiveOpId && (effectiveStatus === 'pending' || effectiveStatus === 'running')) {
const created = await backfillPendingPdfParseOperation({
documentId: id,
userId: row.userId,
namespace: testNamespace,
state,
});
if (created) {
const snapshot = snapshotFromWorkerState(created);
await writeParseRowState({
documentId: row.id,
userId: row.userId,
parseState: stringifyDocumentParseState(documentParseStateFromWorkerState(created)),
});
return NextResponse.json({
parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress,
opId: created.opId,
}, { status: 202 });
}
}
if (effectiveStatus === 'failed' && retryFailed) {
const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
const rateDecision = await checkJobRate(authCtxOrRes.userId, 'pdf_layout', rateConfig);
if (!rateDecision.allowed) {
return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname });
}
const created = await createOrReusePdfWorkerOperation({
documentId: id,
namespace: testNamespace,
documentObjectKey: documentKey(id, testNamespace),
});
await recordJobEvent(authCtxOrRes.userId, 'pdf_layout', created.opId, rateConfig);
const snapshot = snapshotFromWorkerState(created);
await writeParseRowState({
documentId: row.id,
userId: row.userId,
parseState: stringifyDocumentParseState(documentParseStateFromWorkerState(created)),
});
return NextResponse.json({
parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress,
opId: created.opId,
}, { status: 202 });
}
if (effectiveStatus !== 'ready') { if (effectiveStatus !== 'ready') {
return NextResponse.json({ return NextResponse.json({
parseStatus: effectiveStatus, parseStatus: effectiveStatus,
parseProgress: effectiveProgress, parseProgress: effectiveProgress,
opId: effectiveOpId, opId: effectiveOpId,
}, { status: 202 }); }, { status: 409 });
} }
try { try {
@ -446,25 +251,33 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname }); return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname });
} }
const created = await createOrReusePdfWorkerOperation({ const forceToken = randomUUID();
const startedParse = await startPdfParseOperation({
documentId: id, documentId: id,
userId: authCtxOrRes.userId,
namespace: testNamespace, namespace: testNamespace,
documentObjectKey: documentKey(id, testNamespace), forceToken,
forceToken: randomUUID(),
}); });
await recordJobEvent(authCtxOrRes.userId, 'pdf_layout', created.opId, rateConfig); const snapshot = snapshotFromWorkerState(startedParse.workerState);
const snapshot = snapshotFromWorkerState(created);
await writeParseRowState({ await writeParseRowState({
documentId: row.id, documentId: row.id,
userId: row.userId, userId: row.userId,
parseState: stringifyDocumentParseState(documentParseStateFromWorkerState(created)), parseState: stringifyDocumentParseState(startedParse.parseState),
});
enqueueParsePdfJob({
documentId: id,
userId: row.userId,
namespace: testNamespace,
forceToken,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
}); });
return NextResponse.json({ return NextResponse.json({
parseStatus: snapshot.parseStatus, parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress, parseProgress: snapshot.parseProgress,
opId: created.opId, opId: startedParse.workerState.opId,
}, { status: 202 }); }, { status: 202 });
} catch (error) { } catch (error) {
return errorResponse(error, { return errorResponse(error, {

29
src/app/api/documents/docx-to-pdf/upload/route.ts
View file

@ -10,16 +10,14 @@ import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { safeDocumentName } from '@/lib/server/documents/utils'; import { safeDocumentName } from '@/lib/server/documents/utils';
import { enqueueDocumentPreview } from '@/lib/server/documents/previews'; import { enqueueDocumentPreview } from '@/lib/server/documents/previews';
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { recordJobEvent, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state'; import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3'; import { isS3Configured } from '@/lib/server/storage/s3';
import { putDocumentBlob } from '@/lib/server/documents/blobstore'; import { putDocumentBlob } from '@/lib/server/documents/blobstore';
import { errorToLog, serverLogger } from '@/lib/server/logger'; import { errorToLog, serverLogger } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response'; import { errorResponse } from '@/lib/server/errors/next-response';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
const DOCSTORE_DIR = path.join(process.cwd(), 'docstore'); const DOCSTORE_DIR = path.join(process.cwd(), 'docstore');
const TEMP_DIR = path.join(DOCSTORE_DIR, 'tmp'); const TEMP_DIR = path.join(DOCSTORE_DIR, 'tmp');
@ -132,6 +130,11 @@ export async function POST(req: NextRequest) {
const derivedName = safeDocumentName(`${path.parse(file.name).name}.pdf`, `${id}.pdf`); const derivedName = safeDocumentName(`${path.parse(file.name).name}.pdf`, `${id}.pdf`);
const lastModified = Number.isFinite(file.lastModified) ? file.lastModified : Date.now(); const lastModified = Number.isFinite(file.lastModified) ? file.lastModified : Date.now();
const startedParse = await startPdfParseOperation({
documentId: id,
userId: storageUserId,
namespace: testNamespace,
});
await db await db
.insert(documents) .insert(documents)
@ -143,12 +146,7 @@ export async function POST(req: NextRequest) {
size: pdfContent.length, size: pdfContent.length,
lastModified, lastModified,
filePath: id, filePath: id,
parseState: stringifyDocumentParseState({ parseState: stringifyDocumentParseState(startedParse.parseState),
status: 'pending',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
}),
parsedJsonKey: null, parsedJsonKey: null,
}) })
.onConflictDoUpdate({ .onConflictDoUpdate({
@ -159,12 +157,7 @@ export async function POST(req: NextRequest) {
size: pdfContent.length, size: pdfContent.length,
lastModified, lastModified,
filePath: id, filePath: id,
parseState: stringifyDocumentParseState({ parseState: stringifyDocumentParseState(startedParse.parseState),
status: 'pending',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
}),
parsedJsonKey: null, parsedJsonKey: null,
}, },
}); });
@ -186,13 +179,13 @@ export async function POST(req: NextRequest) {
}, 'Failed to enqueue preview for converted DOCX'); }, 'Failed to enqueue preview for converted DOCX');
}); });
// Record upload-driven parse load (see register route for rationale).
const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
await recordJobEvent(storageUserId, 'pdf_layout', `docx:${randomUUID()}`, pdfRateConfig);
enqueueParsePdfJob({ enqueueParsePdfJob({
documentId: id, documentId: id,
userId: storageUserId, userId: storageUserId,
namespace: testNamespace, namespace: testNamespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
}); });
return NextResponse.json({ return NextResponse.json({

24
src/lib/client/api/documents.ts
View file

@ -82,31 +82,20 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort
export async function getParsedPdfDocument( export async function getParsedPdfDocument(
id: string, id: string,
options?: { signal?: AbortSignal; retryFailed?: boolean; opId?: string }, options?: { signal?: AbortSignal },
): Promise< ): Promise<ParsedPdfDocument> {
| { status: 'ready'; parsed: ParsedPdfDocument } const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, {
| { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null; opId?: string | null }
> {
const params = new URLSearchParams();
if (options?.retryFailed) params.set('retry', '1');
if (options?.opId) params.set('opId', options.opId);
const query = params.size > 0 ? `?${params.toString()}` : '';
const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed${query}`, {
signal: options?.signal, signal: options?.signal,
cache: 'no-store', cache: 'no-store',
}); });
if (res.status === 202) { if (res.status === 409) {
const data = (await res.json().catch(() => null)) as { const data = (await res.json().catch(() => null)) as {
parseStatus?: string; parseStatus?: string;
parseProgress?: PdfParseProgress | null; parseProgress?: PdfParseProgress | null;
opId?: string | null; opId?: string | null;
} | null; } | null;
const parseStatus = data?.parseStatus; throw new Error(data?.parseStatus ? `Parsed PDF is not ready (${data.parseStatus})` : 'Parsed PDF is not ready');
if (parseStatus === 'pending' || parseStatus === 'running' || parseStatus === 'failed') {
return { status: parseStatus, parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null };
}
return { status: 'pending', parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null };
} }
if (!res.ok) { if (!res.ok) {
@ -114,8 +103,7 @@ export async function getParsedPdfDocument(
throw new Error(data?.error || 'Failed to load parsed PDF'); throw new Error(data?.error || 'Failed to load parsed PDF');
} }
const parsed = (await res.json()) as ParsedPdfDocument; return (await res.json()) as ParsedPdfDocument;
return { status: 'ready', parsed };
} }
export function subscribeParsedPdfDocumentEvents( export function subscribeParsedPdfDocumentEvents(

2
src/lib/server/compute/worker-parse-state.ts
View file

@ -68,7 +68,7 @@ export function mergeNonReadyParseSnapshot(input: {
workerState: WorkerOperationState<PdfLayoutJobResult>; workerState: WorkerOperationState<PdfLayoutJobResult>;
}): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } { }): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } {
const workerSnapshot = snapshotFromWorkerState(input.workerState); const workerSnapshot = snapshotFromWorkerState(input.workerState);
if (workerSnapshot.parseStatus === 'ready') { if (workerSnapshot.parseStatus === 'ready' || workerSnapshot.parseStatus === 'failed') {
return { return {
parseStatus: input.parseStatus, parseStatus: input.parseStatus,
parseProgress: input.parseProgress, parseProgress: input.parseProgress,

23
src/lib/server/documents/parse-state-backfill.ts
View file

@ -1,7 +1,5 @@
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { documentKey } from '@/lib/server/documents/blobstore'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter';
import { normalizeParseStatus, type DocumentParseState } from '@/lib/server/documents/parse-state'; import { normalizeParseStatus, type DocumentParseState } from '@/lib/server/documents/parse-state';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
@ -20,13 +18,18 @@ export async function backfillPendingPdfParseOperation(input: {
if (parseStatus === 'ready' || parseStatus === 'failed') return null; if (parseStatus === 'ready' || parseStatus === 'failed') return null;
if (normalizeOpId(input.state.opId)) return null; if (normalizeOpId(input.state.opId)) return null;
const created = await createOrReusePdfWorkerOperation({ const startedParse = await startPdfParseOperation({
documentId: input.documentId, documentId: input.documentId,
userId: input.userId,
namespace: input.namespace, namespace: input.namespace,
documentObjectKey: documentKey(input.documentId, input.namespace),
}); });
enqueueParsePdfJob({
const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); documentId: input.documentId,
await recordJobEvent(input.userId, 'pdf_layout', created.opId, rateConfig); userId: input.userId,
return created; namespace: input.namespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
});
return startedParse.workerState;
} }

30
src/lib/server/documents/pdf-parse-operation.ts Normal file
View file

@ -0,0 +1,30 @@
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create';
import { documentParseStateFromWorkerState } from '@/lib/server/compute/worker-parse-state';
import { documentKey } from '@/lib/server/documents/blobstore';
import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import type { DocumentParseState } from '@/lib/server/documents/parse-state';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
export async function startPdfParseOperation(input: {
documentId: string;
userId: string;
namespace: string | null;
forceToken?: string;
}): Promise<{
workerState: WorkerOperationState<PdfLayoutJobResult>;
parseState: DocumentParseState;
}> {
const workerState = await createOrReusePdfWorkerOperation({
documentId: input.documentId,
namespace: input.namespace,
documentObjectKey: documentKey(input.documentId, input.namespace),
...(input.forceToken ? { forceToken: input.forceToken } : {}),
});
const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
await recordJobEvent(input.userId, 'pdf_layout', workerState.opId, rateConfig);
return {
workerState,
parseState: documentParseStateFromWorkerState(workerState),
};
}

20
src/lib/server/documents/register-upload.ts
View file

@ -1,4 +1,3 @@
import { randomUUID } from 'node:crypto';
import { db } from '@/db'; import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { import {
@ -6,10 +5,9 @@ import {
} from '@/lib/server/documents/previews'; } from '@/lib/server/documents/previews';
import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse'; import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse';
import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state'; import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state';
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { errorToLog, serverLogger } from '@/lib/server/logger'; import { errorToLog, serverLogger } from '@/lib/server/logger';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter';
import { PDF_PARSER_VERSION } from '@openreader/compute-core'; import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import type { BaseDocument, DocumentType } from '@/types/documents'; import type { BaseDocument, DocumentType } from '@/types/documents';
@ -27,12 +25,19 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn
const reusableParsedPdf = input.type === 'pdf' const reusableParsedPdf = input.type === 'pdf'
? await findReusableParsedPdfResult(input.documentId) ? await findReusableParsedPdfResult(input.documentId)
: null; : null;
const startedParse = input.type === 'pdf' && !reusableParsedPdf
? await startPdfParseOperation({
documentId: input.documentId,
userId: input.userId,
namespace: input.namespace,
})
: null;
const parsedJsonKey = reusableParsedPdf?.parsedJsonKey ?? null; const parsedJsonKey = reusableParsedPdf?.parsedJsonKey ?? null;
const parseState = input.type === 'pdf' const parseState = input.type === 'pdf'
? stringifyDocumentParseState( ? stringifyDocumentParseState(
reusableParsedPdf reusableParsedPdf
? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }
: { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }, : (startedParse?.parseState ?? { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }),
) )
: null; : null;
@ -79,13 +84,14 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn
}, 'Failed to enqueue document preview'); }, 'Failed to enqueue document preview');
}); });
if (input.type === 'pdf' && !reusableParsedPdf) { if (startedParse) {
const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
await recordJobEvent(input.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig);
enqueueParsePdfJob({ enqueueParsePdfJob({
documentId: input.documentId, documentId: input.documentId,
userId: input.userId, userId: input.userId,
namespace: input.namespace, namespace: input.namespace,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
}); });
} }

29
src/lib/server/jobs/user-pdf-layout-job.ts
View file

@ -27,6 +27,9 @@ import { PDF_PARSER_VERSION } from '@openreader/compute-core';
type UserPdfLayoutJobRequest = PdfLayoutJobBase & { type UserPdfLayoutJobRequest = PdfLayoutJobBase & {
userId: string; userId: string;
forceToken?: string; forceToken?: string;
initialOpId?: string;
initialJobId?: string;
initialStatus?: 'pending' | 'running';
}; };
const running = new Set<string>(); const running = new Set<string>();
@ -49,6 +52,11 @@ function keyFor(input: UserPdfLayoutJobRequest): string {
return `shared:${input.documentId}`; return `shared:${input.documentId}`;
} }
function normalizeOpId(value: string | undefined): string | undefined {
const normalized = value?.trim();
return normalized ? normalized : undefined;
}
function sleep(ms: number): Promise<void> { function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms)); return new Promise((resolve) => setTimeout(resolve, ms));
} }
@ -212,6 +220,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
} }
running.add(key); running.add(key);
let activeOpId = normalizeOpId(input.initialOpId);
let activeJobId = normalizeOpId(input.initialJobId);
try { try {
const scopedRows = await loadScopedRows(input); const scopedRows = await loadScopedRows(input);
const scopedUserIds = userIdsFromRows(scopedRows); const scopedUserIds = userIdsFromRows(scopedRows);
@ -257,18 +267,21 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0]; const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0];
if (!coordinator) return; if (!coordinator) return;
const runningStateData: DocumentParseState = { const initialInflightStatus = input.initialStatus === 'running' ? 'running' : 'pending';
status: 'running', const inflightStateData: DocumentParseState = {
status: initialInflightStatus,
progress: null, progress: null,
updatedAt: Date.now(), updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION, parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
}; };
const runningState = stringifyDocumentParseState(runningStateData); const inflightState = stringifyDocumentParseState(inflightStateData);
const claimRows = (await db const claimRows = (await db
.update(documents) .update(documents)
.set({ .set({
parseState: runningState, parseState: inflightState,
}) })
.where( .where(
and( and(
@ -291,12 +304,10 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
await updateParseStateForUsers({ await updateParseStateForUsers({
documentId: input.documentId, documentId: input.documentId,
userIds: cohortUserIds, userIds: cohortUserIds,
parseState: runningState, parseState: inflightState,
}); });
const compute = await getCompute(); const compute = await getCompute();
let activeOpId: string | undefined;
let activeJobId: string | undefined;
let lastProgressWriteAt = 0; let lastProgressWriteAt = 0;
let lastSnapshotWriteAt = 0; let lastSnapshotWriteAt = 0;
let lastSnapshotStatus: 'pending' | 'running' | null = null; let lastSnapshotStatus: 'pending' | 'running' | null = null;
@ -395,6 +406,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
progress: null, progress: null,
updatedAt: Date.now(), updatedAt: Date.now(),
parserVersion: resolveParsedPdfParserVersion(parsedDoc), parserVersion: resolveParsedPdfParserVersion(parsedDoc),
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
}; };
const readyUserIds = finalUserIds.length > 0 ? finalUserIds : cohortUserIds; const readyUserIds = finalUserIds.length > 0 ? finalUserIds : cohortUserIds;
await updateParseStateForUsers({ await updateParseStateForUsers({
@ -448,6 +461,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
updatedAt: Date.now(), updatedAt: Date.now(),
error: message, error: message,
parserVersion: PDF_PARSER_VERSION, parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
}; };
const failedUserIds = scopedUserIds.length > 0 ? scopedUserIds : [input.userId]; const failedUserIds = scopedUserIds.length > 0 ? scopedUserIds : [input.userId];
await updateParseStateForUsers({ await updateParseStateForUsers({

156
tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts Normal file
View file

@ -0,0 +1,156 @@
import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
import { NextRequest } from 'next/server';
const hoisted = vi.hoisted(() => ({
db: null as {
select: ReturnType<typeof vi.fn>;
update: ReturnType<typeof vi.fn>;
} | null,
row: {
id: 'doc-1',
userId: 'user-1',
parseState: null as string | null,
},
requireAuthContext: vi.fn(),
backfillPendingPdfParseOperation: vi.fn(),
healStaleDocumentParseState: vi.fn(async ({ state }) => state),
}));
vi.mock('@/db', () => ({
get db() {
return hoisted.db;
},
}));
vi.mock('@/lib/server/auth/auth', () => ({
requireAuthContext: hoisted.requireAuthContext,
}));
vi.mock('@/lib/server/documents/parse-state-backfill', () => ({
backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation,
}));
vi.mock('@/lib/server/documents/parse-state-healing', () => ({
healStaleDocumentParseState: hoisted.healStaleDocumentParseState,
}));
vi.mock('@/lib/server/documents/blobstore', () => ({
isValidDocumentId: vi.fn(() => true),
}));
vi.mock('@/lib/server/storage/s3', () => ({
isS3Configured: vi.fn(() => true),
}));
vi.mock('@/lib/server/testing/test-namespace', () => ({
getOpenReaderTestNamespace: vi.fn(() => null),
}));
vi.mock('@/lib/server/logger', () => ({
createRequestLogger: vi.fn(() => ({
logger: {
warn: vi.fn(),
info: vi.fn(),
error: vi.fn(),
},
requestId: 'req-test',
})),
hashForLog: vi.fn(() => 'user-hash'),
}));
describe('GET /api/documents/[id]/parsed/events legacy backfill', () => {
const originalFetch = global.fetch;
beforeEach(() => {
process.env.COMPUTE_WORKER_URL = 'http://localhost:4010';
process.env.COMPUTE_WORKER_TOKEN = 'worker-test-token';
hoisted.row = {
id: 'doc-1',
userId: 'user-1',
parseState: null,
};
hoisted.db = {
select: vi.fn(() => ({
from: vi.fn(() => ({
where: vi.fn(async () => ([{ ...hoisted.row }])),
})),
})),
update: vi.fn(() => ({
set: vi.fn((values: { parseState?: string | null }) => ({
where: vi.fn(async () => {
if (typeof values.parseState !== 'undefined') {
hoisted.row.parseState = values.parseState;
}
return [];
}),
})),
})),
};
hoisted.requireAuthContext.mockReset();
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
hoisted.backfillPendingPdfParseOperation.mockReset();
hoisted.healStaleDocumentParseState.mockClear();
global.fetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => new Promise((_, reject) => {
const signal = init?.signal;
if (signal) {
signal.addEventListener('abort', () => {
reject(new DOMException('Aborted', 'AbortError'));
}, { once: true });
}
})) as typeof fetch;
});
afterEach(() => {
global.fetch = originalFetch;
});
test('creates a worker op for legacy pending PDFs without opId', async () => {
hoisted.backfillPendingPdfParseOperation.mockResolvedValue({
opId: 'op-legacy-1',
opKey: 'pdf_layout|v1|doc-1||doc-1|',
jobId: 'job-legacy-1',
kind: 'pdf_layout',
status: 'queued',
queuedAt: Date.now(),
progress: null,
result: undefined,
error: undefined,
updatedAt: Date.now(),
});
const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route');
const controller = new AbortController();
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed/events', {
signal: controller.signal,
});
const response = await GET(request, {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(200);
const reader = response.body?.getReader();
expect(reader).toBeDefined();
const first = await reader!.read();
const chunk = new TextDecoder().decode(first.value);
expect(chunk).toContain('event: snapshot');
expect(chunk).toContain('"parseStatus":"pending"');
expect(chunk).toContain('"opId":"op-legacy-1"');
controller.abort();
await reader!.cancel().catch(() => {});
expect(hoisted.backfillPendingPdfParseOperation).toHaveBeenCalledWith(expect.objectContaining({
documentId: 'doc-1',
userId: 'user-1',
namespace: null,
state: expect.objectContaining({ status: 'pending' }),
}));
const parseState = String(hoisted.row.parseState ?? '');
expect(parseState).toContain('"status":"pending"');
expect(parseState).toContain('"opId":"op-legacy-1"');
expect(parseState).toContain('"jobId":"job-legacy-1"');
});
});

33
tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts
View file

@ -69,7 +69,7 @@ vi.mock('@/lib/server/logger', () => ({
hashForLog: vi.fn(() => 'user-hash'), hashForLog: vi.fn(() => 'user-hash'),
})); }));
describe('GET /api/documents/[id]/parsed legacy backfill', () => { describe('GET /api/documents/[id]/parsed pure data fetch', () => {
beforeEach(async () => { beforeEach(async () => {
process.env.BASE_URL = 'http://localhost:3003'; process.env.BASE_URL = 'http://localhost:3003';
process.env.AUTH_SECRET = 'test-secret'; process.env.AUTH_SECRET = 'test-secret';
@ -108,40 +108,19 @@ describe('GET /api/documents/[id]/parsed legacy backfill', () => {
hoisted.healStaleDocumentParseState.mockClear(); hoisted.healStaleDocumentParseState.mockClear();
}); });
test('creates a worker op for legacy pending PDFs without opId', async () => { test('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => {
hoisted.backfillPendingPdfParseOperation.mockResolvedValue({
opId: 'op-legacy-1',
opKey: 'pdf_layout|v1|doc-1||doc-1|',
jobId: 'job-legacy-1',
kind: 'pdf_layout',
status: 'queued',
queuedAt: Date.now(),
progress: null,
result: undefined,
error: undefined,
updatedAt: Date.now(),
});
const { GET } = await import('../../src/app/api/documents/[id]/parsed/route'); const { GET } = await import('../../src/app/api/documents/[id]/parsed/route');
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed'); const request = new NextRequest('http://localhost/api/documents/doc-1/parsed');
const response = await GET(request, { const response = await GET(request, {
params: Promise.resolve({ id: 'doc-1' }), params: Promise.resolve({ id: 'doc-1' }),
}); });
expect(response.status).toBe(202); expect(response.status).toBe(409);
await expect(response.json()).resolves.toMatchObject({ await expect(response.json()).resolves.toMatchObject({
parseStatus: 'pending', parseStatus: 'pending',
opId: 'op-legacy-1', opId: null,
}); });
expect(hoisted.backfillPendingPdfParseOperation).toHaveBeenCalledWith(expect.objectContaining({ expect(hoisted.backfillPendingPdfParseOperation).not.toHaveBeenCalled();
documentId: 'doc-1', expect(hoisted.row.parseState).toBeNull();
userId: 'user-1',
namespace: null,
state: expect.objectContaining({ status: 'pending' }),
}));
const parseState = String(hoisted.row.parseState ?? '');
expect(parseState).toContain('"status":"pending"');
expect(parseState).toContain('"opId":"op-legacy-1"');
expect(parseState).toContain('"jobId":"job-legacy-1"');
}); });
}); });