diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index e0fbf79..ffa2be8 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -169,53 +169,29 @@ export function usePdfDocument(): PdfDocumentState { // Guards for setCurrentDocument to prevent stale loads from overwriting newer selections. const docLoadSeqRef = useRef(0); const docLoadAbortRef = useRef(null); - const parsePollAbortRef = useRef(null); + const parseStreamAbortRef = useRef(null); const parseSseCloseRef = useRef<(() => void) | null>(null); const lastPreparedPlaybackPageRef = useRef(null); - const fetchParsedDocument = useCallback(async ( + const loadParsedDocumentOnce = useCallback(async ( documentId: string, - initialStatus: PdfParseStatus | null, signal: AbortSignal, - initialOpId?: string | null, ): Promise => { - // Legacy PDFs may have null parseStatus; treat as pending so opening the - // document backfills parse output via the parsed endpoint polling path. - const effectiveInitialStatus: PdfParseStatus = initialStatus ?? 'pending'; - setParseStatus(effectiveInitialStatus); + if (signal.aborted) return; + const parsed = await getParsedPdfDocument(documentId, { signal }); + if (signal.aborted) return; + setParsedDocument(parsed); + setParseStatus('ready'); setParseProgress(null); - const delayMs = 1200; - const retryFailed = effectiveInitialStatus === 'failed'; - let attempt = 0; - let effectiveOpId = initialOpId?.trim() || null; - while (!signal.aborted) { - if (signal.aborted) return; - const result = await getParsedPdfDocument(documentId, { - signal, - retryFailed: retryFailed && attempt === 0, - ...(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)); - } + setActiveParseOpId(null); + }, []); + + const resetParsedDocumentState = useCallback(() => { + setParsedDocument(null); + setCurrDocText(undefined); + setIsPlaybackReady(false); + lastPreparedPlaybackPageRef.current = null; + pageTextCacheRef.current.clear(); }, []); const fetchDocumentSettings = useCallback(async (documentId: string, signal: AbortSignal): Promise => { @@ -229,14 +205,14 @@ export function usePdfDocument(): PdfDocumentState { } }, []); - const startParsedPolling = useCallback((documentId: string, initialOpId?: string | null) => { - parsePollAbortRef.current?.abort(); + const startParsedEventStream = useCallback((documentId: string, initialOpId?: string | null) => { + parseStreamAbortRef.current?.abort(); parseSseCloseRef.current?.(); parseSseCloseRef.current = null; setParseProgress(null); setActiveParseOpId(initialOpId?.trim() || null); const controller = new AbortController(); - parsePollAbortRef.current = controller; + parseStreamAbortRef.current = controller; const closeSse = subscribeParsedPdfDocumentEvents(documentId, { opId: initialOpId?.trim() || null, @@ -248,25 +224,27 @@ export function usePdfDocument(): PdfDocumentState { } setParseStatus(snapshot.parseStatus); setParseProgress(snapshot.parseProgress); - if (snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed') { - 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), - ); - } + if (snapshot.parseStatus === 'ready') { closeSse(); parseSseCloseRef.current = null; - if (parsePollAbortRef.current === controller) { - parsePollAbortRef.current = null; + if (parseStreamAbortRef.current === controller) { + 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; } }, @@ -286,11 +264,11 @@ export function usePdfDocument(): PdfDocumentState { if (parseSseCloseRef.current === closeSse) { parseSseCloseRef.current = null; } - if (parsePollAbortRef.current === controller) { - parsePollAbortRef.current = null; + if (parseStreamAbortRef.current === controller) { + parseStreamAbortRef.current = null; } }, { once: true }); - }, [fetchParsedDocument, setActiveParseOpId]); + }, [loadParsedDocumentOnce, resetParsedDocumentState, setActiveParseOpId]); useEffect(() => { pdfDocumentRef.current = pdfDocument; @@ -472,8 +450,8 @@ export function usePdfDocument(): PdfDocumentState { // or fast refresh. pdfDocGenerationRef.current += 1; loadSeqRef.current += 1; - parsePollAbortRef.current?.abort(); - parsePollAbortRef.current = null; + parseStreamAbortRef.current?.abort(); + parseStreamAbortRef.current = null; parseSseCloseRef.current?.(); parseSseCloseRef.current = null; pageTextCacheRef.current.clear(); @@ -502,7 +480,7 @@ export function usePdfDocument(): PdfDocumentState { setParseStatus(initialParseStatus); setParseProgress(null); setActiveParseOpId(null); - startParsedPolling(id, null); + startParsedEventStream(id, null); void fetchDocumentSettings(id, controller.signal); } @@ -537,7 +515,7 @@ export function usePdfDocument(): PdfDocumentState { setCurrDocText, setPdfDocument, fetchDocumentSettings, - startParsedPolling, + startParsedEventStream, ]); const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise => { @@ -564,11 +542,11 @@ export function usePdfDocument(): PdfDocumentState { setParseStatus(forced.status); setParseProgress(null); setActiveParseOpId(forced.opId ?? null); - startParsedPolling(currDocId, forced.opId ?? null); + startParsedEventStream(currDocId, forced.opId ?? null); } catch (error) { console.error('Failed to force PDF reparse:', error); } - }, [currDocId, startParsedPolling]); + }, [currDocId, startParsedEventStream]); /** * Clears the current document state @@ -582,8 +560,8 @@ export function usePdfDocument(): PdfDocumentState { docLoadSeqRef.current += 1; docLoadAbortRef.current?.abort(); docLoadAbortRef.current = null; - parsePollAbortRef.current?.abort(); - parsePollAbortRef.current = null; + parseStreamAbortRef.current?.abort(); + parseStreamAbortRef.current = null; parseSseCloseRef.current?.(); parseSseCloseRef.current = null; setCurrDocId(undefined); diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 79d6801..3406252 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -5,7 +5,11 @@ import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; 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 { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { @@ -83,13 +87,20 @@ async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Pr && workerState.opId === opId && !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 { snapshot: { - ...snapshotFromWorkerState(workerState), + ...merged, 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 signature = JSON.stringify(current); let currentOpId = requestedOpId ?? initialState.opId; - let currentFromWorker = initialState.fromWorker; let lastEventId: number | null = null; let loggedMissingOpId = false; const pinnedRequestedOp = requestedOpId ?? null; - const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot, fromWorker: boolean): boolean => { + const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot): boolean => { const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed'; if (!isTerminal) return false; - // If caller pinned an opId, keep streaming until worker confirms that - // op state. DB fallback can report stale terminal status for other rows. - if (pinnedRequestedOp && snapshot.opId === pinnedRequestedOp && !fromWorker) return false; + if (pinnedRequestedOp && snapshot.opId !== pinnedRequestedOp) return false; return true; }; writeSnapshot(current); - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + if (shouldCloseForTerminalSnapshot(current)) { closeStream(); return; } @@ -304,7 +312,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - currentFromWorker = next.fromWorker; if (!requestedOpId && next.opId !== currentOpId) { currentOpId = next.opId; lastEventId = null; @@ -313,7 +320,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string workerAbort = null; } } - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + if (shouldCloseForTerminalSnapshot(current)) { closeStream(); } }).catch((error) => { @@ -350,7 +357,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - currentFromWorker = next.fromWorker; currentOpId = requestedOpId ?? next.opId; if (!currentOpId) { if (!loggedMissingOpId) { @@ -358,7 +364,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string logger.warn({ event: 'documents.parsed.events.missing_opid_non_terminal', degraded: true, - step: 'poll_without_worker_op', + step: 'missing_opid_fallback', documentId: id, storageUserIdHash, parseStatus: current.parseStatus, @@ -368,7 +374,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } else if (loggedMissingOpId) { loggedMissingOpId = false; } - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + if (shouldCloseForTerminalSnapshot(current)) { closeStream(); return; } @@ -472,19 +478,23 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string ); if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; + const mergedSnapshot = mergeNonReadyParseSnapshot({ + parseStatus: current.parseStatus, + parseProgress: current.parseProgress, + workerState: workerSnapshot, + }); const nextSnapshot: ParsedSnapshot = { - ...snapshotFromWorkerState(workerSnapshot), + ...mergedSnapshot, opId: workerSnapshot.opId, }; const nextSignature = JSON.stringify(nextSnapshot); if (nextSignature !== signature) { current = nextSnapshot; signature = nextSignature; - currentFromWorker = true; writeSnapshot(current); } - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + if (shouldCloseForTerminalSnapshot(current)) { closeStream(); return; } @@ -523,12 +533,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - currentFromWorker = next.fromWorker; if (!requestedOpId && next.opId !== currentOpId) { currentOpId = next.opId; lastEventId = null; } - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + if (shouldCloseForTerminalSnapshot(current)) { closeStream(); return; } diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 11158ea..9375ee2 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -4,41 +4,36 @@ import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; import { - documentParseStateFromWorkerState, isWorkerOperationStateStale, snapshotFromWorkerState, } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { - documentKey, getParsedDocumentBlob, getParsedDocumentBlobByKey, isMissingBlobError, isValidDocumentId, - putParsedDocumentBlob, } from '@/lib/server/documents/blobstore'; import { normalizeDocumentParseStateForCurrentParserVersion, normalizeParseStatus, parseDocumentParseState, - resolveParsedPdfParserVersion, stringifyDocumentParseState, } 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 { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; 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 { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; 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 { logDegraded } from '@/lib/server/errors/logging'; import type { ParsedPdfDocument } from '@/types/parsed-pdf'; 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'; @@ -102,105 +97,12 @@ async function writeParseRowState(input: { .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); } -async function finalizeFromWorkerState(input: { - workerState: WorkerOperationState; - row: ParseRow; - namespace: string | null; - logger: ServerLogger; -}): Promise { - 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 }> }) { const { logger } = createRequestLogger({ route: '/api/documents/[id]/parsed', request: req, }); try { - const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); 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 id = (params.id || '').trim().toLowerCase(); - const retryFailed = req.nextUrl.searchParams.get('retry') === '1'; - const requestedOpId = normalizeOpId(req.nextUrl.searchParams.get('opId')); if (!isValidDocumentId(id)) { 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 }); } - if (requestedOpId) { - const workerState = await fetchWorkerOperationState(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 state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)); const effectiveStatus = normalizeParseStatus(state.status); const effectiveProgress = state.progress ?? null; const effectiveOpId = normalizeOpId(state.opId); - if (effectiveOpId && effectiveStatus !== 'ready') { - const workerState = await fetchWorkerOperationState(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') { return NextResponse.json({ parseStatus: effectiveStatus, parseProgress: effectiveProgress, opId: effectiveOpId, - }, { status: 202 }); + }, { status: 409 }); } try { @@ -446,25 +251,33 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname }); } - const created = await createOrReusePdfWorkerOperation({ + const forceToken = randomUUID(); + const startedParse = await startPdfParseOperation({ documentId: id, + userId: authCtxOrRes.userId, namespace: testNamespace, - documentObjectKey: documentKey(id, testNamespace), - forceToken: randomUUID(), + forceToken, }); - await recordJobEvent(authCtxOrRes.userId, 'pdf_layout', created.opId, rateConfig); - - const snapshot = snapshotFromWorkerState(created); + const snapshot = snapshotFromWorkerState(startedParse.workerState); await writeParseRowState({ documentId: row.id, 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({ parseStatus: snapshot.parseStatus, parseProgress: snapshot.parseProgress, - opId: created.opId, + opId: startedParse.workerState.opId, }, { status: 202 }); } catch (error) { return errorResponse(error, { diff --git a/src/app/api/documents/docx-to-pdf/upload/route.ts b/src/app/api/documents/docx-to-pdf/upload/route.ts index 10d1f23..834b4ab 100644 --- a/src/app/api/documents/docx-to-pdf/upload/route.ts +++ b/src/app/api/documents/docx-to-pdf/upload/route.ts @@ -10,16 +10,14 @@ import { db } from '@/db'; import { documents } from '@/db/schema'; import { safeDocumentName } from '@/lib/server/documents/utils'; 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 { 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 { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import { putDocumentBlob } from '@/lib/server/documents/blobstore'; import { errorToLog, serverLogger } from '@/lib/server/logger'; 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 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 lastModified = Number.isFinite(file.lastModified) ? file.lastModified : Date.now(); + const startedParse = await startPdfParseOperation({ + documentId: id, + userId: storageUserId, + namespace: testNamespace, + }); await db .insert(documents) @@ -143,12 +146,7 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, - parseState: stringifyDocumentParseState({ - status: 'pending', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }), + parseState: stringifyDocumentParseState(startedParse.parseState), parsedJsonKey: null, }) .onConflictDoUpdate({ @@ -159,12 +157,7 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, - parseState: stringifyDocumentParseState({ - status: 'pending', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }), + parseState: stringifyDocumentParseState(startedParse.parseState), parsedJsonKey: null, }, }); @@ -186,13 +179,13 @@ export async function POST(req: NextRequest) { }, '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({ documentId: id, userId: storageUserId, namespace: testNamespace, + initialOpId: startedParse.workerState.opId, + initialJobId: startedParse.workerState.jobId, + initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending', }); return NextResponse.json({ diff --git a/src/lib/client/api/documents.ts b/src/lib/client/api/documents.ts index 360914f..d6a80ce 100644 --- a/src/lib/client/api/documents.ts +++ b/src/lib/client/api/documents.ts @@ -82,31 +82,20 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort export async function getParsedPdfDocument( id: string, - options?: { signal?: AbortSignal; retryFailed?: boolean; opId?: string }, -): Promise< - | { status: 'ready'; parsed: ParsedPdfDocument } - | { 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}`, { + options?: { signal?: AbortSignal }, +): Promise { + const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, { signal: options?.signal, cache: 'no-store', }); - if (res.status === 202) { + if (res.status === 409) { const data = (await res.json().catch(() => null)) as { parseStatus?: string; parseProgress?: PdfParseProgress | null; opId?: string | null; } | null; - const parseStatus = data?.parseStatus; - 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 }; + throw new Error(data?.parseStatus ? `Parsed PDF is not ready (${data.parseStatus})` : 'Parsed PDF is not ready'); } if (!res.ok) { @@ -114,8 +103,7 @@ export async function getParsedPdfDocument( throw new Error(data?.error || 'Failed to load parsed PDF'); } - const parsed = (await res.json()) as ParsedPdfDocument; - return { status: 'ready', parsed }; + return (await res.json()) as ParsedPdfDocument; } export function subscribeParsedPdfDocumentEvents( diff --git a/src/lib/server/compute/worker-parse-state.ts b/src/lib/server/compute/worker-parse-state.ts index 6798a22..4b954b7 100644 --- a/src/lib/server/compute/worker-parse-state.ts +++ b/src/lib/server/compute/worker-parse-state.ts @@ -68,7 +68,7 @@ export function mergeNonReadyParseSnapshot(input: { workerState: WorkerOperationState; }): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } { const workerSnapshot = snapshotFromWorkerState(input.workerState); - if (workerSnapshot.parseStatus === 'ready') { + if (workerSnapshot.parseStatus === 'ready' || workerSnapshot.parseStatus === 'failed') { return { parseStatus: input.parseStatus, parseProgress: input.parseProgress, diff --git a/src/lib/server/documents/parse-state-backfill.ts b/src/lib/server/documents/parse-state-backfill.ts index 97f9cae..b972553 100644 --- a/src/lib/server/documents/parse-state-backfill.ts +++ b/src/lib/server/documents/parse-state-backfill.ts @@ -1,7 +1,5 @@ -import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; -import { documentKey } from '@/lib/server/documents/blobstore'; -import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; -import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter'; +import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation'; +import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; import { normalizeParseStatus, type DocumentParseState } from '@/lib/server/documents/parse-state'; 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 (normalizeOpId(input.state.opId)) return null; - const created = await createOrReusePdfWorkerOperation({ + const startedParse = await startPdfParseOperation({ documentId: input.documentId, + userId: input.userId, namespace: input.namespace, - documentObjectKey: documentKey(input.documentId, input.namespace), }); - - const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); - await recordJobEvent(input.userId, 'pdf_layout', created.opId, rateConfig); - return created; + enqueueParsePdfJob({ + documentId: input.documentId, + userId: input.userId, + namespace: input.namespace, + initialOpId: startedParse.workerState.opId, + initialJobId: startedParse.workerState.jobId, + initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending', + }); + return startedParse.workerState; } diff --git a/src/lib/server/documents/pdf-parse-operation.ts b/src/lib/server/documents/pdf-parse-operation.ts new file mode 100644 index 0000000..c356f86 --- /dev/null +++ b/src/lib/server/documents/pdf-parse-operation.ts @@ -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; + 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), + }; +} diff --git a/src/lib/server/documents/register-upload.ts b/src/lib/server/documents/register-upload.ts index f63daea..06bafd2 100644 --- a/src/lib/server/documents/register-upload.ts +++ b/src/lib/server/documents/register-upload.ts @@ -1,4 +1,3 @@ -import { randomUUID } from 'node:crypto'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { @@ -6,10 +5,9 @@ import { } from '@/lib/server/documents/previews'; import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse'; 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 { 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 type { BaseDocument, DocumentType } from '@/types/documents'; @@ -27,12 +25,19 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn const reusableParsedPdf = input.type === 'pdf' ? await findReusableParsedPdfResult(input.documentId) : 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 parseState = input.type === 'pdf' ? stringifyDocumentParseState( reusableParsedPdf ? { 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; @@ -79,13 +84,14 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn }, 'Failed to enqueue document preview'); }); - if (input.type === 'pdf' && !reusableParsedPdf) { - const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); - await recordJobEvent(input.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig); + if (startedParse) { enqueueParsePdfJob({ documentId: input.documentId, userId: input.userId, namespace: input.namespace, + initialOpId: startedParse.workerState.opId, + initialJobId: startedParse.workerState.jobId, + initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending', }); } diff --git a/src/lib/server/jobs/user-pdf-layout-job.ts b/src/lib/server/jobs/user-pdf-layout-job.ts index 531dfd0..0f44043 100644 --- a/src/lib/server/jobs/user-pdf-layout-job.ts +++ b/src/lib/server/jobs/user-pdf-layout-job.ts @@ -27,6 +27,9 @@ import { PDF_PARSER_VERSION } from '@openreader/compute-core'; type UserPdfLayoutJobRequest = PdfLayoutJobBase & { userId: string; forceToken?: string; + initialOpId?: string; + initialJobId?: string; + initialStatus?: 'pending' | 'running'; }; const running = new Set(); @@ -49,6 +52,11 @@ function keyFor(input: UserPdfLayoutJobRequest): string { return `shared:${input.documentId}`; } +function normalizeOpId(value: string | undefined): string | undefined { + const normalized = value?.trim(); + return normalized ? normalized : undefined; +} + function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -212,6 +220,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise } running.add(key); + let activeOpId = normalizeOpId(input.initialOpId); + let activeJobId = normalizeOpId(input.initialJobId); try { const scopedRows = await loadScopedRows(input); const scopedUserIds = userIdsFromRows(scopedRows); @@ -257,18 +267,21 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0]; if (!coordinator) return; - const runningStateData: DocumentParseState = { - status: 'running', + const initialInflightStatus = input.initialStatus === 'running' ? 'running' : 'pending'; + const inflightStateData: DocumentParseState = { + status: initialInflightStatus, progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION, + ...(activeOpId ? { opId: activeOpId } : {}), + ...(activeJobId ? { jobId: activeJobId } : {}), }; - const runningState = stringifyDocumentParseState(runningStateData); + const inflightState = stringifyDocumentParseState(inflightStateData); const claimRows = (await db .update(documents) .set({ - parseState: runningState, + parseState: inflightState, }) .where( and( @@ -291,12 +304,10 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise await updateParseStateForUsers({ documentId: input.documentId, userIds: cohortUserIds, - parseState: runningState, + parseState: inflightState, }); const compute = await getCompute(); - let activeOpId: string | undefined; - let activeJobId: string | undefined; let lastProgressWriteAt = 0; let lastSnapshotWriteAt = 0; let lastSnapshotStatus: 'pending' | 'running' | null = null; @@ -395,6 +406,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise progress: null, updatedAt: Date.now(), parserVersion: resolveParsedPdfParserVersion(parsedDoc), + ...(activeOpId ? { opId: activeOpId } : {}), + ...(activeJobId ? { jobId: activeJobId } : {}), }; const readyUserIds = finalUserIds.length > 0 ? finalUserIds : cohortUserIds; await updateParseStateForUsers({ @@ -448,6 +461,8 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise updatedAt: Date.now(), error: message, parserVersion: PDF_PARSER_VERSION, + ...(activeOpId ? { opId: activeOpId } : {}), + ...(activeJobId ? { jobId: activeJobId } : {}), }; const failedUserIds = scopedUserIds.length > 0 ? scopedUserIds : [input.userId]; await updateParseStateForUsers({ diff --git a/tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts b/tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts new file mode 100644 index 0000000..d39d9f5 --- /dev/null +++ b/tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts @@ -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; + update: ReturnType; + } | 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"'); + }); +}); diff --git a/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts index 15b4bf1..1c48cb6 100644 --- a/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts +++ b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts @@ -69,7 +69,7 @@ vi.mock('@/lib/server/logger', () => ({ hashForLog: vi.fn(() => 'user-hash'), })); -describe('GET /api/documents/[id]/parsed legacy backfill', () => { +describe('GET /api/documents/[id]/parsed pure data fetch', () => { beforeEach(async () => { process.env.BASE_URL = 'http://localhost:3003'; process.env.AUTH_SECRET = 'test-secret'; @@ -108,40 +108,19 @@ describe('GET /api/documents/[id]/parsed legacy backfill', () => { hoisted.healStaleDocumentParseState.mockClear(); }); - 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(), - }); - + test('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => { const { GET } = await import('../../src/app/api/documents/[id]/parsed/route'); const request = new NextRequest('http://localhost/api/documents/doc-1/parsed'); const response = await GET(request, { params: Promise.resolve({ id: 'doc-1' }), }); - expect(response.status).toBe(202); + expect(response.status).toBe(409); await expect(response.json()).resolves.toMatchObject({ parseStatus: 'pending', - opId: 'op-legacy-1', + opId: null, }); - 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"'); + expect(hoisted.backfillPendingPdfParseOperation).not.toHaveBeenCalled(); + expect(hoisted.row.parseState).toBeNull(); }); });