diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index e208e63..7b14f0f 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -30,6 +30,7 @@ import { getComputeOpStaleMs, getAvailableCpuCores, getOnnxThreadsPerJob, + PDF_PARSER_VERSION, withIdleTimeoutAndHardCap, withTimeout, } from '@openreader/compute-core'; @@ -124,6 +125,7 @@ interface OperationEventStreamLike { interface OperationStateStoreLike { getOpState(opId: string): Promise; getOpStateRecord?(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + getOpIndex?(opKey: string): Promise<{ opId: string } | null>; listOpStates?(): Promise; } @@ -243,13 +245,18 @@ function sanitizeNamespace(namespace: string | null): string | null { return SAFE_NAMESPACE_REGEX.test(namespace) ? namespace : null; } +function encodeParserVersion(parserVersion: string): string { + const normalized = parserVersion.trim() || 'unknown-parser'; + return encodeURIComponent(normalized); +} + function documentParsedKey(id: string, namespace: string | null, prefix: string): string { if (!DOCUMENT_ID_REGEX.test(id)) { throw new Error(`Invalid document id: ${id}`); } const ns = sanitizeNamespace(namespace); const nsSegment = ns ? `ns/${ns}/` : ''; - return `${prefix}/documents_v1/parsed_v1/${nsSegment}${id}.json`; + return `${prefix}/documents_v1/parsed_v2/${nsSegment}${id}/${encodeParserVersion(PDF_PARSER_VERSION)}.json`; } function buildS3Client(): S3Client { @@ -425,6 +432,10 @@ const operationCreateSchema = z.discriminatedUnion('kind', [ }), ]); +const operationLookupSchema = z.object({ + opKey: z.string().trim().min(1).max(1024), +}); + async function ensureJetStreamResources( jsm: JetStreamManager, whisperTimeoutMs: number, @@ -901,6 +912,36 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti return op; }); + app.post('/ops/lookup', async (request, reply) => { + const parsed = operationLookupSchema.safeParse(request.body); + if (!parsed.success) { + reply.code(400); + return { + error: 'Invalid request body', + issues: parsed.error.issues, + }; + } + + await ensureOrphanedOpRecovery(); + if (typeof operationStateStore.getOpIndex !== 'function') { + reply.code(501); + return { error: 'Operation lookup by opKey is not supported by this state store' }; + } + const indexEntry = await operationStateStore.getOpIndex(parsed.data.opKey); + if (!indexEntry?.opId) { + reply.code(404); + return { error: 'Operation not found' }; + } + + const state = await operationStateStore.getOpState(indexEntry.opId); + if (!state) { + reply.code(404); + return { error: 'Operation not found' }; + } + + return state; + }); + app.get('/ops/:opId', async (request, reply) => { const params = z.object({ opId: z.string().trim().min(1) }).safeParse(request.params); if (!params.success) { diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index c916bb1..9cb6d45 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -19,10 +19,12 @@ import { import type { PDFDocumentProxy } from 'pdfjs-dist'; import { + ensureParsedPdfDocumentOperation, forceReparsePdfDocument, getDocumentMetadata, getDocumentSettings, getParsedPdfDocument, + ParsedPdfNotReadyError, putDocumentSettings, subscribeParsedPdfDocumentEvents, } from '@/lib/client/api/documents'; @@ -205,18 +207,18 @@ export function usePdfDocument(): PdfDocumentState { } }, []); - const startParsedEventStream = useCallback((documentId: string, initialOpId?: string | null) => { + const startParsedEventStream = useCallback((documentId: string, initialOpId: string) => { parseStreamAbortRef.current?.abort(); parseSseCloseRef.current?.(); parseSseCloseRef.current = null; setParseProgress(null); - setActiveParseOpId(initialOpId?.trim() || null); + setActiveParseOpId(initialOpId.trim() || null); const controller = new AbortController(); parseStreamAbortRef.current = controller; let isResolvingTerminalState = false; const closeSse = subscribeParsedPdfDocumentEvents(documentId, { - opId: initialOpId?.trim() || null, + opId: initialOpId.trim(), }, { onSnapshot: (snapshot) => { if (controller.signal.aborted) return; @@ -281,6 +283,60 @@ export function usePdfDocument(): PdfDocumentState { }, { once: true }); }, [loadParsedDocumentOnce, resetParsedDocumentState, setActiveParseOpId]); + const resolveParsedDocumentState = useCallback(async ( + documentId: string, + signal: AbortSignal, + ): Promise => { + try { + await loadParsedDocumentOnce(documentId, signal); + } catch (error) { + if (signal.aborted) return; + if (!(error instanceof ParsedPdfNotReadyError)) { + throw error; + } + + resetParsedDocumentState(); + setParseStatus(error.parseStatus); + setParseProgress(error.parseProgress); + setActiveParseOpId(error.opId); + + if (error.parseStatus === 'failed') { + return; + } + + let nextOpId = error.opId; + let nextStatus: PdfParseStatus = error.parseStatus; + let nextProgress = error.parseProgress; + + if (!nextOpId) { + const ensured = await ensureParsedPdfDocumentOperation(documentId, { signal }); + if (signal.aborted) return; + nextOpId = ensured.opId; + nextStatus = ensured.parseStatus; + nextProgress = ensured.parseProgress; + setParseStatus(nextStatus); + setParseProgress(nextProgress); + setActiveParseOpId(nextOpId); + } + + if (nextStatus === 'ready') { + await loadParsedDocumentOnce(documentId, signal); + return; + } + + if (nextStatus === 'failed' || !nextOpId) { + return; + } + + startParsedEventStream(documentId, nextOpId); + } + }, [ + loadParsedDocumentOnce, + resetParsedDocumentState, + setActiveParseOpId, + startParsedEventStream, + ]); + useEffect(() => { pdfDocumentRef.current = pdfDocument; }, [pdfDocument]); @@ -487,12 +543,11 @@ export function usePdfDocument(): PdfDocumentState { return false; } if (meta.type === 'pdf') { - const initialParseStatus = (meta.parseStatus ?? null) as PdfParseStatus | null; - setParseStatus(initialParseStatus); - setParseProgress(null); - setActiveParseOpId(null); - startParsedEventStream(id, null); void fetchDocumentSettings(id, controller.signal); + void resolveParsedDocumentState(id, controller.signal).catch((error) => { + if (controller.signal.aborted) return; + console.error('Failed to resolve parsed PDF state:', error); + }); } const doc = await ensureCachedDocument(meta, { signal: controller.signal }); @@ -526,7 +581,7 @@ export function usePdfDocument(): PdfDocumentState { setCurrDocText, setPdfDocument, fetchDocumentSettings, - startParsedEventStream, + resolveParsedDocumentState, ]); const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise => { @@ -553,7 +608,9 @@ export function usePdfDocument(): PdfDocumentState { setParseStatus(forced.status); setParseProgress(null); setActiveParseOpId(forced.opId ?? null); - startParsedEventStream(currDocId, forced.opId ?? null); + if (forced.opId) { + startParsedEventStream(currDocId, forced.opId); + } } catch (error) { console.error('Failed to force PDF reparse:', error); } diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 3406252..a0d3dfd 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -4,56 +4,23 @@ import { db } from '@/db'; 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, - 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 { - normalizeDocumentParseStateForCurrentParserVersion, - normalizeParseStatus, - parseDocumentParseState, - stringifyDocumentParseState, -} from '@/lib/server/documents/parse-state'; -import { backfillPendingPdfParseOperation } from '@/lib/server/documents/parse-state-backfill'; -import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing'; -import { documentParseStateFromWorkerState } from '@/lib/server/compute/worker-parse-state'; + fetchPdfParseOperation, + isPdfParseOperationForDocument, +} from '@/lib/server/pdf-parse/operation'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; -import { createRequestLogger, hashForLog } from '@/lib/server/logger'; +import { createRequestLogger } from '@/lib/server/logger'; import { errorResponse } from '@/lib/server/errors/next-response'; -import { logDegraded, logServerError } from '@/lib/server/errors/logging'; -import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; -import { parseSseEventId, parseSsePayload } from '@openreader/compute-core'; -import { getComputeOpStaleMs } from '@openreader/compute-core'; -import type { PdfLayoutJobResult, WorkerOperationEvent, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; export const runtime = 'nodejs'; export const maxDuration = 300; -const SSE_KEEPALIVE_MS = 15_000; -const SSE_RESYNC_INTERVAL_MS = 30_000; - -type ParseRow = { +type DocumentRow = { id: string; - userId: string; - parseState: string | null; -}; - -type ParsedSnapshot = { - parseStatus: PdfParseStatus; - parseProgress: PdfParseProgress | null; - opId?: string | null; -}; - -type SnapshotState = { - snapshot: ParsedSnapshot; - opId: string | null; - fromWorker: boolean; + type: string; }; function s3NotConfiguredResponse(): NextResponse { @@ -63,120 +30,35 @@ function s3NotConfiguredResponse(): NextResponse { ); } -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - -async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise { - const opStaleMs = getComputeOpStaleMs(); - const state = await healStaleDocumentParseState({ - documentId: row.id, - userId: row.userId, - state: normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)), - }); - const parseStatus = normalizeParseStatus(state.status); - const dbOpId = typeof state.opId === 'string' && state.opId.trim() ? state.opId.trim() : null; - const requestedOpId = preferredOpId?.trim() || null; - const opId = requestedOpId ?? dbOpId; - // When a caller pins an opId, that op is the live source of truth even if - // the per-user document row currently says "ready" or has a different opId. - if (opId && (requestedOpId !== null || parseStatus !== 'ready')) { - const workerState = await fetchWorkerOperationState(opId); - if ( - workerState - && 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: { - ...merged, - opId: workerState.opId, - }, - opId: workerState.opId, - fromWorker, - }; - } - } - return { - snapshot: { - parseStatus, - parseProgress: state.progress ?? null, - opId, - }, - opId, - fromWorker: false, - }; -} - -async function loadPreferredRow(input: { +async function loadOwnedDocumentRow(input: { documentId: string; - storageUserId: string; allowedUserIds: string[]; -}): Promise { +}): Promise { const rows = (await db .select({ id: documents.id, - userId: documents.userId, - parseState: documents.parseState, + type: documents.type, }) .from(documents) - .where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[]; - - return rows.find((candidate) => candidate.userId === input.storageUserId) ?? rows[0] ?? null; + .where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds))) + .limit(1)) as DocumentRow[]; + return rows[0] ?? null; } -async function writeParseRowState(input: { - documentId: string; - userId: string; - parseState: string; -}): Promise { - await db - .update(documents) - .set({ parseState: input.parseState }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); -} - -async function syncFromDb(input: { - documentId: string; - storageUserId: string; - allowedUserIds: string[]; - signature: string; - preferredOpId?: string | null; - writeSnapshot: (snapshot: ParsedSnapshot) => void; -}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; fromWorker: boolean; signature: string } | null> { - const nextRow = await loadPreferredRow({ - documentId: input.documentId, - storageUserId: input.storageUserId, - allowedUserIds: input.allowedUserIds, - }); - if (!nextRow) return null; - - const nextState = await toSnapshotState(nextRow, input.preferredOpId); - const nextSignature = JSON.stringify(nextState.snapshot); - if (nextSignature !== input.signature) { - input.writeSnapshot(nextState.snapshot); +function workerEventsUrl(baseUrl: string, opId: string, sinceEventId: string | null): string { + const url = new URL(`${baseUrl}/ops/${encodeURIComponent(opId)}/events`); + if (sinceEventId) { + url.searchParams.set('sinceEventId', sinceEventId); } - - return { - snapshot: nextState.snapshot, - opId: nextState.opId, - fromWorker: nextState.fromWorker, - signature: nextSignature, - }; + return url.toString(); } export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { - const { logger, requestId } = createRequestLogger({ + const { logger } = createRequestLogger({ route: '/api/documents/[id]/parsed/events', request: req, }); + try { if (!isS3Configured()) return s3NotConfiguredResponse(); @@ -189,407 +71,65 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string if (!isValidDocumentId(id)) { return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); } - const requestedOpIdRaw = req.nextUrl.searchParams.get('opId'); - const requestedOpId = typeof requestedOpIdRaw === 'string' && requestedOpIdRaw.trim() - ? requestedOpIdRaw.trim() - : null; - const testNamespace = getOpenReaderTestNamespace(req.headers); - const storageUserId = authCtxOrRes.userId; - const storageUserIdHash = hashForLog(storageUserId); - const allowedUserIds = [storageUserId]; + const opId = (req.nextUrl.searchParams.get('opId') || '').trim(); + if (!opId) { + return NextResponse.json({ error: 'opId is required' }, { status: 400 }); + } - const row = await loadPreferredRow({ + const row = await loadOwnedDocumentRow({ documentId: id, - storageUserId, - allowedUserIds, + allowedUserIds: [authCtxOrRes.userId], }); - - if (!row) { - return NextResponse.json({ error: 'Not found' }, { status: 404 }); + if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 }); + if (row.type !== 'pdf') { + return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 }); } - let initialState = await toSnapshotState(row, requestedOpId); - if (!requestedOpId && !initialState.opId && initialState.snapshot.parseStatus !== 'ready' && initialState.snapshot.parseStatus !== 'failed') { - const state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)); - const created = await backfillPendingPdfParseOperation({ - documentId: id, - userId: row.userId, - namespace: testNamespace, - state, - }); - if (created) { - await writeParseRowState({ - documentId: row.id, - userId: row.userId, - parseState: stringifyDocumentParseState(documentParseStateFromWorkerState(created)), - }); - initialState = { - snapshot: { - ...snapshotFromWorkerState(created), - opId: created.opId, - }, - opId: created.opId, - fromWorker: true, - }; - } + const namespace = getOpenReaderTestNamespace(req.headers); + const initialState = await fetchPdfParseOperation(opId); + if (!initialState || !isPdfParseOperationForDocument(initialState, { documentId: id, namespace })) { + return NextResponse.json({ error: 'Operation not found' }, { status: 404 }); } - const workerCfg = getWorkerClientConfigFromEnv(); - const encoder = new TextEncoder(); - const stream = new ReadableStream({ - start(controller) { - let closed = false; - let keepaliveTimer: ReturnType | null = null; - let resyncTimer: ReturnType | null = null; - let workerAbort: AbortController | null = null; - const writeSnapshot = (snapshot: ParsedSnapshot): void => { - controller.enqueue(encoder.encode(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`)); - }; - - const closeStream = (): void => { - if (closed) return; - closed = true; - if (keepaliveTimer) { - clearInterval(keepaliveTimer); - keepaliveTimer = null; - } - if (resyncTimer) { - clearInterval(resyncTimer); - resyncTimer = null; - } - if (workerAbort) { - workerAbort.abort(); - workerAbort = null; - } - try { - controller.close(); - } catch { - // no-op - } - }; - - const runWorkerProxy = async () => { - let current = initialState.snapshot; - let signature = JSON.stringify(current); - let currentOpId = requestedOpId ?? initialState.opId; - let lastEventId: number | null = null; - let loggedMissingOpId = false; - const pinnedRequestedOp = requestedOpId ?? null; - const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot): boolean => { - const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed'; - if (!isTerminal) return false; - if (pinnedRequestedOp && snapshot.opId !== pinnedRequestedOp) return false; - return true; - }; - - writeSnapshot(current); - - if (shouldCloseForTerminalSnapshot(current)) { - closeStream(); - return; - } - - keepaliveTimer = setInterval(() => { - if (closed) return; - controller.enqueue(encoder.encode(': keepalive\n\n')); - }, SSE_KEEPALIVE_MS); - - resyncTimer = setInterval(() => { - if (closed) return; - void syncFromDb({ - documentId: id, - storageUserId, - allowedUserIds, - signature, - preferredOpId: requestedOpId ?? currentOpId, - writeSnapshot, - }).then((next) => { - if (closed) return; - if (!next) { - closeStream(); - return; - } - current = next.snapshot; - signature = next.signature; - if (!requestedOpId && next.opId !== currentOpId) { - currentOpId = next.opId; - lastEventId = null; - if (workerAbort) { - workerAbort.abort(); - workerAbort = null; - } - } - if (shouldCloseForTerminalSnapshot(current)) { - closeStream(); - } - }).catch((error) => { - if (closed) return; - logDegraded(logger, { - event: 'documents.parsed.events.db_resync_failed', - msg: 'SSE DB resync failed', - step: 'db_resync', - context: { - documentId: id, - storageUserIdHash, - requestId, - }, - error, - }); - controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`)); - }); - }, SSE_RESYNC_INTERVAL_MS); - - while (!closed) { - if (!currentOpId) { - await sleep(SSE_RESYNC_INTERVAL_MS); - const next = await syncFromDb({ - documentId: id, - storageUserId, - allowedUserIds, - signature, - preferredOpId: requestedOpId ?? currentOpId, - writeSnapshot, - }); - if (!next) { - closeStream(); - return; - } - current = next.snapshot; - signature = next.signature; - currentOpId = requestedOpId ?? next.opId; - if (!currentOpId) { - if (!loggedMissingOpId) { - loggedMissingOpId = true; - logger.warn({ - event: 'documents.parsed.events.missing_opid_non_terminal', - degraded: true, - step: 'missing_opid_fallback', - documentId: id, - storageUserIdHash, - parseStatus: current.parseStatus, - requestedOpId, - }, 'Parse stream running without opId and non-terminal status'); - } - } else if (loggedMissingOpId) { - loggedMissingOpId = false; - } - if (shouldCloseForTerminalSnapshot(current)) { - closeStream(); - return; - } - continue; - } - - try { - workerAbort = new AbortController(); - const query = lastEventId && lastEventId > 0 - ? `?sinceEventId=${encodeURIComponent(String(lastEventId))}` - : ''; - const response = await fetch( - `${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`, - { - method: 'GET', - headers: { - Authorization: `Bearer ${workerCfg.token}`, - Accept: 'text/event-stream', - ...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}), - }, - cache: 'no-store', - signal: workerAbort.signal, - }, - ); - - if (closed) return; - - if (!response.ok) { - const upstreamResponseBody = await response.text().catch(() => ''); - logger.warn({ - event: 'documents.parsed.events.worker_stream_request_failed', - degraded: true, - step: 'worker_stream_request', - documentId: id, - opId: currentOpId, - status: response.status, - upstreamResponseBody, - error: { - name: 'WorkerStreamRequestFailed', - message: `Worker stream request failed with status ${response.status}`, - }, - }, 'Worker stream request failed'); - await sleep(500); - continue; - } - if (!response.body) { - logger.warn({ - event: 'documents.parsed.events.worker_stream_missing_body', - degraded: true, - step: 'worker_stream_body', - documentId: id, - opId: currentOpId, - error: { - name: 'WorkerStreamMissingBody', - message: 'Worker stream response missing body', - }, - }, 'Worker stream response missing body'); - await sleep(500); - continue; - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - let streamEnded = false; - - while (!closed && !streamEnded) { - const read = await reader.read(); - if (read.done) { - streamEnded = true; - break; - } - - buffer += decoder.decode(read.value, { stream: true }); - - while (true) { - const frameEnd = buffer.indexOf('\n\n'); - if (frameEnd < 0) break; - const frame = buffer.slice(0, frameEnd); - buffer = buffer.slice(frameEnd + 2); - - const eventId = parseSseEventId(frame); - if (eventId && eventId > 0) { - lastEventId = eventId; - } - - const payload = parseSsePayload(frame); - if (!payload) continue; - - let parsed: WorkerOperationEvent | WorkerOperationState; - try { - parsed = JSON.parse(payload) as WorkerOperationEvent | WorkerOperationState; - } catch { - continue; - } - - const workerSnapshot: WorkerOperationState = ( - parsed && typeof parsed === 'object' && 'snapshot' in parsed - ? parsed.snapshot - : parsed as WorkerOperationState - ); - if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; - - const mergedSnapshot = mergeNonReadyParseSnapshot({ - parseStatus: current.parseStatus, - parseProgress: current.parseProgress, - workerState: workerSnapshot, - }); - const nextSnapshot: ParsedSnapshot = { - ...mergedSnapshot, - opId: workerSnapshot.opId, - }; - const nextSignature = JSON.stringify(nextSnapshot); - if (nextSignature !== signature) { - current = nextSnapshot; - signature = nextSignature; - writeSnapshot(current); - } - - if (shouldCloseForTerminalSnapshot(current)) { - closeStream(); - return; - } - } - } - } catch (error) { - if (closed || isAbortLikeError(error)) return; - logDegraded(logger, { - event: 'documents.parsed.events.worker_stream_read_failed', - msg: 'Worker stream read failed; reconnecting', - step: 'worker_stream_read', - context: { - documentId: id, - opId: currentOpId, - requestId, - }, - error, - }); - await sleep(500); - continue; - } - - if (closed) return; - - const next = await syncFromDb({ - documentId: id, - storageUserId, - allowedUserIds, - signature, - preferredOpId: requestedOpId ?? currentOpId, - writeSnapshot, - }); - if (!next) { - closeStream(); - return; - } - current = next.snapshot; - signature = next.signature; - if (!requestedOpId && next.opId !== currentOpId) { - currentOpId = next.opId; - lastEventId = null; - } - if (shouldCloseForTerminalSnapshot(current)) { - closeStream(); - return; - } - await sleep(250); - } - }; - - void runWorkerProxy() - .catch((error) => { - if (closed || isAbortLikeError(error)) return; - logServerError(logger, { - event: 'documents.parsed.events.worker_proxy_crashed', - error, - msg: 'Worker proxy crashed while streaming parse events', - context: { documentId: id }, - normalize: { - code: 'DOCUMENTS_PARSED_EVENTS_WORKER_PROXY_CRASHED', - errorClass: 'upstream', - }, - }); - if (!closed) { - controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`)); - } - }) - .finally(() => { - closeStream(); - }); - - req.signal.addEventListener('abort', () => { - closeStream(); - }, { once: true }); - }, - cancel() { - return; - }, - }); - - return new Response(stream, { + const cfg = getWorkerClientConfigFromEnv(); + const lastEventId = req.headers.get('last-event-id'); + const sinceEventId = req.nextUrl.searchParams.get('sinceEventId') || lastEventId; + const upstream = await fetch(workerEventsUrl(cfg.baseUrl, opId, sinceEventId), { + method: 'GET', headers: { - 'Content-Type': 'text/event-stream; charset=utf-8', - 'Cache-Control': 'no-cache, no-transform', - Connection: 'keep-alive', + Authorization: `Bearer ${cfg.token}`, + Accept: 'text/event-stream', + ...(lastEventId ? { 'Last-Event-ID': lastEventId } : {}), + }, + cache: 'no-store', + signal: req.signal, + }); + + if (!upstream.ok || !upstream.body) { + const detail = await upstream.text().catch(() => ''); + return NextResponse.json( + { error: detail || 'Failed to proxy parse event stream' }, + { status: upstream.status || 502 }, + ); + } + + return new NextResponse(upstream.body, { + status: 200, + headers: { + 'Content-Type': upstream.headers.get('content-type') || 'text/event-stream; charset=utf-8', + 'Cache-Control': upstream.headers.get('cache-control') || 'no-cache, no-transform', + 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no', }, }); } catch (error) { return errorResponse(error, { logger, - event: 'documents.parsed.events.route_failed', - msg: 'Parsed events route failed', - apiErrorMessage: 'Failed to stream parsed PDF progress', - normalize: { code: 'DOCUMENTS_PARSED_EVENTS_ROUTE_FAILED', errorClass: 'upstream' }, + event: 'documents.parsed.events_failed', + msg: 'Failed to proxy parsed PDF events', + apiErrorMessage: 'Failed to proxy parsed PDF events', + normalize: { code: 'DOCUMENTS_PARSED_EVENTS_FAILED', errorClass: 'upstream' }, }); } } diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 9375ee2..e95d538 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -4,39 +4,32 @@ import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; +import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { - isWorkerOperationStateStale, - snapshotFromWorkerState, -} from '@/lib/server/compute/worker-parse-state'; -import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; + createOrReuseCurrentPdfParseOperation, + lookupCurrentPdfParseOperation, +} from '@/lib/server/pdf-parse/operation'; +import { readCurrentParsedPdfArtifact, readParsedPdfArtifactByKey } from '@/lib/server/pdf-parse/artifact'; import { - getParsedDocumentBlob, - getParsedDocumentBlobByKey, - isMissingBlobError, - isValidDocumentId, -} from '@/lib/server/documents/blobstore'; -import { - normalizeDocumentParseStateForCurrentParserVersion, - normalizeParseStatus, - parseDocumentParseState, - stringifyDocumentParseState, -} from '@/lib/server/documents/parse-state'; -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'; + parsedObjectKeyFromWorkerState, + pdfParseSnapshotFromWorkerState, +} from '@/lib/server/pdf-parse/snapshot'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; +import { isS3Configured } from '@/lib/server/storage/s3'; +import { createRequestLogger } from '@/lib/server/logger'; +import { errorResponse } from '@/lib/server/errors/next-response'; 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 } from '@/lib/server/logger'; -import { errorResponse } from '@/lib/server/errors/next-response'; -import type { ParsedPdfDocument } from '@/types/parsed-pdf'; -import { getComputeOpStaleMs } from '@openreader/compute-core'; -import type { PdfLayoutJobResult } from '@openreader/compute-core/api-contracts'; +import type { PdfParseSnapshot } from '@/lib/server/pdf-parse/types'; export const dynamic = 'force-dynamic'; +type DocumentRow = { + id: string; + type: string; +}; + function s3NotConfiguredResponse(): NextResponse { return NextResponse.json( { error: 'Documents storage is not configured. Set S3_* environment variables.' }, @@ -44,57 +37,33 @@ function s3NotConfiguredResponse(): NextResponse { ); } -type ParseRow = { - id: string; - userId: string; - parseState: string | null; - parsedJsonKey: string | null; -}; - -function hasAnyParsedBlocks(doc: ParsedPdfDocument | null): boolean { - if (!doc || !Array.isArray(doc.pages)) return false; - return doc.pages.some((page) => Array.isArray(page.blocks) && page.blocks.length > 0); -} - -function normalizeOpId(value: string | null | undefined): string | null { - const normalized = typeof value === 'string' ? value.trim() : ''; - return normalized || null; -} - -async function loadRows(input: { +async function loadOwnedDocumentRow(input: { documentId: string; allowedUserIds: string[]; -}): Promise { - return (await db +}): Promise { + const rows = (await db .select({ id: documents.id, - userId: documents.userId, - parseState: documents.parseState, - parsedJsonKey: documents.parsedJsonKey, + type: documents.type, }) .from(documents) - .where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[]; + .where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds))) + .limit(1)) as DocumentRow[]; + return rows[0] ?? null; } -function pickPreferredRow(rows: ParseRow[], storageUserId: string): ParseRow | null { - return rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0] ?? null; +function jsonSnapshot(snapshot: PdfParseSnapshot, status = 409): NextResponse { + return NextResponse.json(snapshot, { status }); } -async function writeParseRowState(input: { - documentId: string; - userId: string; - parseState: string; - parsedJsonKey?: string | null; -}): Promise { - await db - .update(documents) - .set({ - parseState: input.parseState, - ...(typeof input.parsedJsonKey === 'string' || input.parsedJsonKey === null - ? { parsedJsonKey: input.parsedJsonKey } - : {}), - }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); +function artifactResponse(bytes: Buffer): NextResponse { + return new NextResponse(new Uint8Array(bytes), { + status: 200, + headers: { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + }, + }); } export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { @@ -102,6 +71,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string route: '/api/documents/[id]/parsed', request: req, }); + try { if (!isS3Configured()) return s3NotConfiguredResponse(); @@ -115,62 +85,48 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); } - const testNamespace = getOpenReaderTestNamespace(req.headers); - const storageUserId = authCtxOrRes.userId; - const allowedUserIds = [storageUserId]; - - const rows = await loadRows({ documentId: id, allowedUserIds }); - const row = pickPreferredRow(rows, storageUserId); - if (!row) { - return NextResponse.json({ error: 'Not found' }, { status: 404 }); + const row = await loadOwnedDocumentRow({ + documentId: id, + allowedUserIds: [authCtxOrRes.userId], + }); + if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 }); + if (row.type !== 'pdf') { + return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 }); } - const state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)); - const effectiveStatus = normalizeParseStatus(state.status); - const effectiveProgress = state.progress ?? null; - const effectiveOpId = normalizeOpId(state.opId); - - if (effectiveStatus !== 'ready') { - return NextResponse.json({ - parseStatus: effectiveStatus, - parseProgress: effectiveProgress, - opId: effectiveOpId, - }, { status: 409 }); + const namespace = getOpenReaderTestNamespace(req.headers); + const artifact = await readCurrentParsedPdfArtifact({ documentId: id, namespace }); + if (artifact) { + return artifactResponse(artifact.bytes); } - try { - const json = row.parsedJsonKey?.trim() - ? await getParsedDocumentBlobByKey(row.parsedJsonKey) - : await getParsedDocumentBlob(id, testNamespace); - let parsedDoc: ParsedPdfDocument | null = null; - try { - parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument; - } catch { - parsedDoc = null; - } - - if (!hasAnyParsedBlocks(parsedDoc)) { - logger.warn({ - event: 'documents.parsed.no_blocks_from_blob', - documentId: id, - userIdHash: hashForLog(row.userId), - parsedJsonKey: row.parsedJsonKey, - }, 'Parsed document blob contained no blocks'); - } - - return new NextResponse(new Uint8Array(json), { - status: 200, - headers: { - 'Content-Type': 'application/json', - 'Cache-Control': 'no-store', - }, + const currentOp = await lookupCurrentPdfParseOperation({ documentId: id, namespace }); + if (!currentOp) { + return jsonSnapshot({ + parseStatus: 'pending', + parseProgress: null, + opId: null, }); - } catch (error) { - if (isMissingBlobError(error)) { - return NextResponse.json({ parseStatus: 'failed', error: 'Parsed document not found' }, { status: 404 }); - } - throw error; } + + if (currentOp.status === 'succeeded') { + const artifactKey = parsedObjectKeyFromWorkerState(currentOp); + if (artifactKey) { + const artifactFromOp = await readParsedPdfArtifactByKey(artifactKey); + if (artifactFromOp) { + return artifactResponse(artifactFromOp.bytes); + } + } + return NextResponse.json( + { + error: 'Current parse operation succeeded without a readable parsed artifact.', + opId: currentOp.opId, + }, + { status: 502 }, + ); + } + + return jsonSnapshot(pdfParseSnapshotFromWorkerState(currentOp)); } catch (error) { return errorResponse(error, { logger, @@ -187,8 +143,8 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string route: '/api/documents/[id]/parsed', request: req, }); + try { - const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); const authCtxOrRes = await requireAuthContext(req); @@ -209,39 +165,34 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string replace = false; } - const testNamespace = getOpenReaderTestNamespace(req.headers); - const storageUserId = authCtxOrRes.userId; - const allowedUserIds = [storageUserId]; - - const rows = await loadRows({ documentId: id, allowedUserIds }); - const row = pickPreferredRow(rows, storageUserId); - if (!row) { - return NextResponse.json({ error: 'Not found' }, { status: 404 }); + const row = await loadOwnedDocumentRow({ + documentId: id, + allowedUserIds: [authCtxOrRes.userId], + }); + if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 }); + if (row.type !== 'pdf') { + return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 }); } - let state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)); - state = await healStaleDocumentParseState({ - documentId: id, - userId: row.userId, - state, - }); + const namespace = getOpenReaderTestNamespace(req.headers); - const existingOpId = normalizeOpId(state.opId); - if (existingOpId) { - const existing = await fetchWorkerOperationState(existingOpId); - if ( - existing - && !isWorkerOperationStateStale(existing, opStaleMs) - && (existing.status === 'queued' || existing.status === 'running') - && !replace - ) { - const snapshot = snapshotFromWorkerState(existing); - return NextResponse.json({ - error: 'Parse operation already in progress', - parseStatus: snapshot.parseStatus, - parseProgress: snapshot.parseProgress, - opId: existing.opId, - }, { status: 409 }); + if (!replace) { + const artifact = await readCurrentParsedPdfArtifact({ documentId: id, namespace }); + if (artifact) { + return jsonSnapshot({ + parseStatus: 'ready', + parseProgress: null, + opId: null, + }, 200); + } + + const currentOp = await lookupCurrentPdfParseOperation({ documentId: id, namespace }); + if (currentOp) { + const snapshot = pdfParseSnapshotFromWorkerState(currentOp); + if (snapshot.parseStatus === 'failed') { + return jsonSnapshot(snapshot); + } + return jsonSnapshot(snapshot, snapshot.parseStatus === 'ready' ? 200 : 202); } } @@ -251,41 +202,20 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname }); } - const forceToken = randomUUID(); - const startedParse = await startPdfParseOperation({ + const workerState = await createOrReuseCurrentPdfParseOperation({ documentId: id, - userId: authCtxOrRes.userId, - namespace: testNamespace, - forceToken, - }); - const snapshot = snapshotFromWorkerState(startedParse.workerState); - await writeParseRowState({ - documentId: row.id, - userId: row.userId, - 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', + namespace, + ...(replace ? { forceToken: randomUUID() } : {}), }); - return NextResponse.json({ - parseStatus: snapshot.parseStatus, - parseProgress: snapshot.parseProgress, - opId: startedParse.workerState.opId, - }, { status: 202 }); + return jsonSnapshot(pdfParseSnapshotFromWorkerState(workerState), 202); } catch (error) { return errorResponse(error, { logger, - event: 'documents.parsed.force_refresh_failed', - msg: 'Failed to force PDF refresh', - apiErrorMessage: 'Failed to force PDF refresh', - normalize: { code: 'DOCUMENTS_PARSED_FORCE_REFRESH_FAILED', errorClass: 'upstream' }, + event: 'documents.parsed.ensure_failed', + msg: 'Failed to ensure parsed PDF operation', + apiErrorMessage: 'Failed to ensure parsed PDF operation', + normalize: { code: 'DOCUMENTS_PARSED_ENSURE_FAILED', errorClass: 'upstream' }, }); } } diff --git a/src/app/api/documents/route.ts b/src/app/api/documents/route.ts index 8002f7a..1cdc2f6 100644 --- a/src/app/api/documents/route.ts +++ b/src/app/api/documents/route.ts @@ -11,11 +11,6 @@ import { deleteDocumentPreviewRows, } from '@/lib/server/documents/previews'; import { deleteDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; -import { - normalizeDocumentParseStateForCurrentParserVersion, - normalizeParseStatus, - parseDocumentParseState, -} from '@/lib/server/documents/parse-state'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import type { BaseDocument, DocumentType } from '@/types/documents'; @@ -78,17 +73,14 @@ export async function GET(req: NextRequest) { const results: BaseDocument[] = rows.map((doc) => { const type = normalizeDocumentType(doc.type, doc.name); - const parseState = type === 'pdf' - ? normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(doc.parseState)) - : null; return { id: doc.id, name: doc.name, size: Number(doc.size), lastModified: Number(doc.lastModified), type, - parseStatus: type === 'pdf' && parseState ? normalizeParseStatus(parseState.status) : null, - parsedJsonKey: doc.parsedJsonKey, + parseStatus: null, + parsedJsonKey: null, scope: 'user', }; }); diff --git a/src/lib/client/api/documents.ts b/src/lib/client/api/documents.ts index f13a235..d01d6e5 100644 --- a/src/lib/client/api/documents.ts +++ b/src/lib/client/api/documents.ts @@ -88,6 +88,27 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort return docs[0] ?? null; } +export class ParsedPdfNotReadyError extends Error { + readonly parseStatus: PdfParseStatus; + readonly parseProgress: PdfParseProgress | null; + readonly opId: string | null; + readonly details: string | null; + + constructor(input: { + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId?: string | null; + details?: string | null; + }) { + super(`Parsed PDF is not ready (${input.parseStatus})`); + this.name = 'ParsedPdfNotReadyError'; + this.parseStatus = input.parseStatus; + this.parseProgress = input.parseProgress; + this.opId = input.opId?.trim() || null; + this.details = input.details?.trim() || null; + } +} + export async function getParsedPdfDocument( id: string, options?: { signal?: AbortSignal }, @@ -102,8 +123,20 @@ export async function getParsedPdfDocument( parseStatus?: string; parseProgress?: PdfParseProgress | null; opId?: string | null; + error?: string; } | null; - throw new Error(data?.parseStatus ? `Parsed PDF is not ready (${data.parseStatus})` : 'Parsed PDF is not ready'); + throw new ParsedPdfNotReadyError({ + parseStatus: data?.parseStatus === 'running' + ? 'running' + : data?.parseStatus === 'ready' + ? 'ready' + : data?.parseStatus === 'failed' + ? 'failed' + : 'pending', + parseProgress: data?.parseProgress ?? null, + opId: data?.opId ?? null, + details: data?.error ?? null, + }); } if (!res.ok) { @@ -117,30 +150,49 @@ export async function getParsedPdfDocument( export function subscribeParsedPdfDocumentEvents( id: string, options: { - opId?: string | null; + opId: string; }, handlers: { onSnapshot: (snapshot: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null; opId?: string | null; + error?: string | null; }) => void; onError?: (error: Event) => void; }, ): () => void { const params = new URLSearchParams(); - if (options.opId) params.set('opId', options.opId); + params.set('opId', options.opId); const query = params.size > 0 ? `?${params.toString()}` : ''; const source = new EventSource(`/api/documents/${encodeURIComponent(id)}/parsed/events${query}`); source.addEventListener('snapshot', (event) => { if (!(event instanceof MessageEvent)) return; try { const payload = JSON.parse(event.data) as { - parseStatus: PdfParseStatus; - parseProgress: PdfParseProgress | null; - opId?: string | null; + snapshot?: { + opId: string; + status: 'queued' | 'running' | 'succeeded' | 'failed'; + progress?: PdfParseProgress | null; + error?: { message?: string } | null; + }; }; - handlers.onSnapshot(payload); + const snapshot = payload?.snapshot; + if (!snapshot?.opId || !snapshot.status) return; + handlers.onSnapshot({ + parseStatus: snapshot.status === 'running' + ? 'running' + : snapshot.status === 'succeeded' + ? 'ready' + : snapshot.status === 'failed' + ? 'failed' + : 'pending', + parseProgress: snapshot.status === 'running' ? (snapshot.progress ?? null) : null, + opId: snapshot.opId, + ...(snapshot.status === 'failed' && snapshot.error?.message + ? { error: snapshot.error.message } + : {}), + }); } catch { // Ignore malformed payloads to avoid breaking active streams. } @@ -153,39 +205,70 @@ export function subscribeParsedPdfDocumentEvents( }; } -export async function forceReparsePdfDocument( +function normalizeParsedPdfOperationResponse( + data: { parseStatus?: string; parseProgress?: PdfParseProgress | null; opId?: string | null; error?: string } | null, +): { + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId: string | null; + error?: string | null; +} { + return { + parseStatus: data?.parseStatus === 'running' + ? 'running' + : data?.parseStatus === 'ready' + ? 'ready' + : data?.parseStatus === 'failed' + ? 'failed' + : 'pending', + parseProgress: data?.parseProgress ?? null, + opId: data?.opId?.trim() || null, + ...(data?.error ? { error: data.error } : {}), + }; +} + +export async function ensureParsedPdfDocumentOperation( id: string, - options?: { signal?: AbortSignal }, -): Promise<{ status: 'pending' | 'running'; opId?: string | null }> { + options?: { signal?: AbortSignal; replace?: boolean }, +): Promise<{ + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId: string | null; + error?: string | null; +}> { const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, { method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ replace: options?.replace === true }), signal: options?.signal, cache: 'no-store', }); - if (res.status === 409) { - const data = (await res.json().catch(() => null)) as { - parseStatus?: string; - opId?: string | null; - error?: string; - } | null; - if (typeof data?.opId === 'string' && data.opId.trim()) { - return { - status: data?.parseStatus === 'running' ? 'running' : 'pending', - opId: data.opId, - }; - } + const data = (await res.json().catch(() => null)) as { + parseStatus?: string; + parseProgress?: PdfParseProgress | null; + opId?: string | null; + error?: string; + } | null; + + if (!res.ok && res.status !== 409) { + throw new Error(data?.error || 'Failed to ensure parsed PDF operation'); } - if (!res.ok) { - const data = (await res.json().catch(() => null)) as { error?: string } | null; - throw new Error(data?.error || 'Failed to force PDF reparse'); - } + return normalizeParsedPdfOperationResponse(data); +} - const data = (await res.json().catch(() => null)) as { parseStatus?: string; opId?: string | null } | null; +export async function forceReparsePdfDocument( + id: string, + options?: { signal?: AbortSignal }, +): Promise<{ status: 'pending' | 'running'; opId?: string | null }> { + const data = await ensureParsedPdfDocumentOperation(id, { + signal: options?.signal, + replace: true, + }); return { - status: data?.parseStatus === 'running' ? 'running' : 'pending', - opId: data?.opId ?? null, + status: data.parseStatus === 'running' ? 'running' : 'pending', + opId: data.opId, }; } diff --git a/src/lib/server/compute/worker-op-state.ts b/src/lib/server/compute/worker-op-state.ts index dc36e1e..feddab5 100644 --- a/src/lib/server/compute/worker-op-state.ts +++ b/src/lib/server/compute/worker-op-state.ts @@ -81,3 +81,86 @@ export async function fetchWorkerOperationState( clearTimeout(timeout); } } + +export async function fetchWorkerOperationStateByKey( + opKey: string | null | undefined, +): Promise | null> { + const normalized = opKey?.trim(); + if (!normalized) return null; + + let cfg: { baseUrl: string; token: string }; + try { + cfg = getWorkerClientConfigFromEnv(); + } catch (error) { + logDegraded(serverLogger, { + event: 'compute.worker_op_lookup.config.invalid', + msg: 'Worker client env missing/invalid', + step: 'read_worker_config', + context: { opKey: normalized }, + error, + }); + return null; + } + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), WORKER_OP_REQUEST_TIMEOUT_MS); + + try { + const res = await fetch(`${cfg.baseUrl}/ops/lookup`, { + method: 'POST', + headers: { + Authorization: `Bearer ${cfg.token}`, + Accept: 'application/json', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ opKey: normalized }), + cache: 'no-store', + signal: controller.signal, + }); + + if (res.status === 404) { + return null; + } + if (!res.ok) { + const upstreamResponseBody = await res.text().catch(() => ''); + logDegraded(serverLogger, { + event: 'compute.worker_op_lookup.fetch.failed', + msg: 'Worker op lookup failed', + step: 'fetch_worker_op_by_key', + context: { + opKey: normalized, + status: res.status, + upstreamResponseBody, + }, + error: { + name: 'WorkerOpLookupFailed', + message: `Worker op lookup failed with status ${res.status}`, + }, + }); + return null; + } + + const parsed = await res.json() as WorkerOperationState; + if (!parsed || typeof parsed !== 'object' || typeof parsed.opId !== 'string' || !parsed.opId.trim()) { + logDegraded(serverLogger, { + event: 'compute.worker_op_lookup.response.invalid', + msg: 'Worker op lookup response invalid', + step: 'validate_worker_op_lookup_response', + context: { opKey: normalized }, + }); + return null; + } + return parsed; + } catch (error) { + logDegraded(serverLogger, { + event: 'compute.worker_op_lookup.fetch.error', + msg: 'Worker op lookup threw', + step: 'fetch_worker_op_by_key', + context: { opKey: normalized }, + error, + }); + return null; + } finally { + clearTimeout(timeout); + } +} diff --git a/src/lib/server/compute/worker-parse-state.ts b/src/lib/server/compute/worker-parse-state.ts deleted file mode 100644 index 6798a22..0000000 --- a/src/lib/server/compute/worker-parse-state.ts +++ /dev/null @@ -1,78 +0,0 @@ -import { PDF_PARSER_VERSION } from '@openreader/compute-core'; -import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; -import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; -import type { DocumentParseState } from '@/lib/server/documents/parse-state'; - -function isInflightWorkerStatus(status: WorkerOperationState['status']): boolean { - return status === 'queued' || status === 'running'; -} - -export function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus { - switch (status) { - case 'queued': - return 'pending'; - case 'running': - return 'running'; - case 'succeeded': - return 'ready'; - case 'failed': - return 'failed'; - default: - return 'pending'; - } -} - -export function snapshotFromWorkerState( - state: WorkerOperationState, -): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } { - const parseStatus = mapWorkerStatusToParseStatus(state.status); - return { - parseStatus, - parseProgress: parseStatus === 'running' ? (state.progress ?? null) : null, - }; -} - -export function documentParseStateFromWorkerState( - state: WorkerOperationState, - nowMs = Date.now(), -): DocumentParseState { - const { parseStatus, parseProgress } = snapshotFromWorkerState(state); - return { - status: parseStatus, - progress: parseStatus === 'pending' || parseStatus === 'running' - ? parseProgress - : null, - updatedAt: nowMs, - parserVersion: PDF_PARSER_VERSION, - ...(typeof state.opId === 'string' && state.opId.trim() ? { opId: state.opId } : {}), - ...(typeof state.jobId === 'string' && state.jobId.trim() ? { jobId: state.jobId } : {}), - ...(parseStatus === 'failed' && state.error?.message ? { error: state.error.message } : {}), - }; -} - -export function isWorkerOperationStateStale( - state: WorkerOperationState, - staleMs: number, - nowMs = Date.now(), -): boolean { - if (!isInflightWorkerStatus(state.status)) return false; - if (!Number.isFinite(staleMs) || staleMs <= 0) return false; - const updatedAt = Number(state.updatedAt ?? 0); - if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false; - return (nowMs - updatedAt) > staleMs; -} - -export function mergeNonReadyParseSnapshot(input: { - parseStatus: PdfParseStatus; - parseProgress: PdfParseProgress | null; - workerState: WorkerOperationState; -}): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } { - const workerSnapshot = snapshotFromWorkerState(input.workerState); - if (workerSnapshot.parseStatus === 'ready') { - return { - parseStatus: input.parseStatus, - parseProgress: input.parseProgress, - }; - } - return workerSnapshot; -} diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index 36316c9..5d53499 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -8,6 +8,7 @@ import { PutObjectCommand, } from '@aws-sdk/client-s3'; import { getSignedUrl } from '@aws-sdk/s3-request-presigner'; +import { PDF_PARSER_VERSION } from '@openreader/compute-core'; import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage/s3'; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; @@ -102,13 +103,26 @@ export function documentKey(id: string, namespace: string | null): string { } export function documentParsedKey(id: string, namespace: string | null): string { + return documentParsedKeyForVersion(id, namespace, PDF_PARSER_VERSION); +} + +function encodeParserVersion(parserVersion: string): string { + const normalized = parserVersion.trim() || PDF_PARSER_VERSION; + return encodeURIComponent(normalized); +} + +export function documentParsedKeyForVersion( + id: string, + namespace: string | null, + parserVersion: string, +): string { if (!isValidDocumentId(id)) { throw new Error(`Invalid document id: ${id}`); } const cfg = getS3Config(); const ns = sanitizeNamespace(namespace); const nsSegment = ns ? `ns/${ns}/` : ''; - return `${cfg.prefix}/documents_v1/parsed_v1/${nsSegment}${id}.json`; + return `${cfg.prefix}/documents_v1/parsed_v2/${nsSegment}${id}/${encodeParserVersion(parserVersion)}.json`; } export function tempDocumentUploadPrefix(userId: string, namespace: string | null): string { @@ -359,9 +373,18 @@ export async function getParsedDocumentBlobByKey(key: string): Promise { } export async function putParsedDocumentBlob(id: string, body: Buffer, namespace: string | null): Promise { + return putParsedDocumentBlobForVersion(id, body, namespace, PDF_PARSER_VERSION); +} + +export async function putParsedDocumentBlobForVersion( + id: string, + body: Buffer, + namespace: string | null, + parserVersion: string, +): Promise { const cfg = getS3Config(); const client = getS3ProxyClient(); - const key = documentParsedKey(id, namespace); + const key = documentParsedKeyForVersion(id, namespace, parserVersion); await client.send( new PutObjectCommand({ Bucket: cfg.bucket, @@ -484,10 +507,14 @@ export async function deleteDocumentBlob(id: string, namespace: string | null): const key = documentKey(id, namespace); const parsedKey = documentParsedKey(id, namespace); const legacyParsedKey = legacyDocumentParsedKey(id, namespace); + const ns = sanitizeNamespace(namespace); + const nsSegment = ns ? `ns/${ns}/` : ''; + const parsedPrefix = `${cfg.prefix}/documents_v1/parsed_v2/${nsSegment}${id}/`; await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: key })); await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: parsedKey })).catch(() => undefined); await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: legacyParsedKey })).catch(() => undefined); + await deleteDocumentPrefix(parsedPrefix).catch(() => undefined); await deleteDocumentPrefix(`${key}/`).catch(() => undefined); } diff --git a/src/lib/server/documents/parse-state-backfill.ts b/src/lib/server/documents/parse-state-backfill.ts deleted file mode 100644 index f058ba1..0000000 --- a/src/lib/server/documents/parse-state-backfill.ts +++ /dev/null @@ -1,37 +0,0 @@ -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'; - -function normalizeOpId(value: string | undefined): string | null { - const normalized = typeof value === 'string' ? value.trim() : ''; - return normalized || null; -} - -export async function backfillPendingPdfParseOperation(input: { - documentId: string; - userId: string; - namespace: string | null; - state: DocumentParseState; -}): Promise | null> { - const parseStatus = normalizeParseStatus(input.state.status); - if (parseStatus === 'ready' || parseStatus === 'failed') return null; - if (normalizeOpId(input.state.opId)) return null; - - const startedParse = await startPdfParseOperation({ - documentId: input.documentId, - userId: input.userId, - namespace: input.namespace, - }); - if (startedParse.parseState.status === 'pending' || startedParse.parseState.status === 'running') { - enqueueParsePdfJob({ - documentId: input.documentId, - userId: input.userId, - namespace: input.namespace, - initialOpId: startedParse.workerState.opId, - initialJobId: startedParse.workerState.jobId, - initialStatus: startedParse.parseState.status, - }); - } - return startedParse.workerState; -} diff --git a/src/lib/server/documents/parse-state-healing.ts b/src/lib/server/documents/parse-state-healing.ts deleted file mode 100644 index d8fcbd4..0000000 --- a/src/lib/server/documents/parse-state-healing.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { and, eq } from 'drizzle-orm'; -import { getComputeOpStaleMs } from '@openreader/compute-core'; -import { db } from '@/db'; -import { documents } from '@/db/schema'; -import { - isDocumentParseStateStale, - parseDocumentParseState, - stringifyDocumentParseState, - type DocumentParseState, -} from '@/lib/server/documents/parse-state'; - -export async function healStaleDocumentParseState(input: { - documentId: string; - userId: string; - state: DocumentParseState; -}): Promise { - const staleMs = getComputeOpStaleMs(); - if (!isDocumentParseStateStale(input.state, staleMs)) return input.state; - - const nextState = parseDocumentParseState(stringifyDocumentParseState({ - status: 'failed', - progress: null, - updatedAt: Date.now(), - error: `Parse state stale for more than ${staleMs}ms; marked failed for retry`, - ...(input.state.opId ? { opId: input.state.opId } : {}), - ...(input.state.jobId ? { jobId: input.state.jobId } : {}), - ...(input.state.parserVersion ? { parserVersion: input.state.parserVersion } : {}), - })); - - await db - .update(documents) - .set({ parseState: stringifyDocumentParseState(nextState) }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); - - return nextState; -} diff --git a/src/lib/server/documents/parse-state.ts b/src/lib/server/documents/parse-state.ts deleted file mode 100644 index 22a94ad..0000000 --- a/src/lib/server/documents/parse-state.ts +++ /dev/null @@ -1,118 +0,0 @@ -import { PDF_PARSER_VERSION } from '@openreader/compute-core'; -import type { ParsedPdfDocument, PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; - -export interface DocumentParseState { - status: PdfParseStatus; - progress?: PdfParseProgress | null; - updatedAt?: number; - error?: string | null; - opId?: string; - jobId?: string; - parserVersion?: string; -} - -export function isInProgressParseStatus(status: PdfParseStatus): status is 'pending' | 'running' { - return status === 'pending' || status === 'running'; -} - -export function isDocumentParseStateStale( - state: DocumentParseState, - staleMs: number, - nowMs = Date.now(), -): boolean { - if (!isInProgressParseStatus(state.status)) return false; - if (!Number.isFinite(staleMs) || staleMs <= 0) return false; - const updatedAt = Number(state.updatedAt ?? 0); - if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false; - return (nowMs - updatedAt) > staleMs; -} - -export function normalizeParseStatus(status: string | null | undefined): PdfParseStatus { - if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') { - return status; - } - return 'pending'; -} - -export function hasCurrentPdfParserVersion(parserVersion: string | null | undefined): boolean { - return typeof parserVersion === 'string' && parserVersion.trim() === PDF_PARSER_VERSION; -} - -export function normalizeDocumentParseStateForCurrentParserVersion( - state: DocumentParseState, - nowMs = Date.now(), -): DocumentParseState { - const status = normalizeParseStatus(state.status); - if (status === 'failed' || hasCurrentPdfParserVersion(state.parserVersion)) { - return state; - } - - return { - status: 'pending', - progress: null, - updatedAt: nowMs, - }; -} - -export function resolveParsedPdfParserVersion(parsed: ParsedPdfDocument | null | undefined): string { - const parserVersion = parsed?.parserVersion?.trim(); - return parserVersion || PDF_PARSER_VERSION; -} - -function normalizeProgress(progress: unknown): PdfParseProgress | null { - if (!progress || typeof progress !== 'object') return null; - const rec = progress as Record; - const totalPages = Number(rec.totalPages ?? 0); - if (!Number.isFinite(totalPages) || totalPages <= 0) return null; - const pagesParsedRaw = Number(rec.pagesParsed ?? 0); - const pagesParsed = Math.max(0, Math.min(totalPages, Number.isFinite(pagesParsedRaw) ? pagesParsedRaw : 0)); - const phase = rec.phase === 'merge' ? 'merge' : 'infer'; - const currentPageRaw = Number(rec.currentPage ?? pagesParsed); - const currentPage = Number.isFinite(currentPageRaw) && currentPageRaw > 0 - ? Math.max(1, Math.min(totalPages, currentPageRaw)) - : undefined; - return { - totalPages, - pagesParsed, - currentPage, - phase, - }; -} - -export function parseDocumentParseState(value: string | null): DocumentParseState { - if (!value) return { status: 'pending', progress: null }; - try { - const parsed = JSON.parse(value) as Record; - const status = normalizeParseStatus(typeof parsed.status === 'string' ? parsed.status : null); - const progress = normalizeProgress(parsed.progress); - const updatedAtRaw = Number(parsed.updatedAt ?? 0); - const updatedAt = Number.isFinite(updatedAtRaw) && updatedAtRaw > 0 ? updatedAtRaw : undefined; - const error = typeof parsed.error === 'string' ? parsed.error : null; - const opId = typeof parsed.opId === 'string' ? parsed.opId : undefined; - const jobId = typeof parsed.jobId === 'string' ? parsed.jobId : undefined; - const parserVersion = typeof parsed.parserVersion === 'string' ? parsed.parserVersion : undefined; - return { - status, - progress, - ...(typeof updatedAt === 'number' ? { updatedAt } : {}), - ...(error ? { error } : {}), - ...(opId ? { opId } : {}), - ...(jobId ? { jobId } : {}), - ...(parserVersion ? { parserVersion } : {}), - }; - } catch { - return { status: 'pending', progress: null }; - } -} - -export function stringifyDocumentParseState(state: DocumentParseState): string { - return JSON.stringify({ - status: normalizeParseStatus(state.status), - progress: state.progress ?? null, - updatedAt: typeof state.updatedAt === 'number' ? state.updatedAt : Date.now(), - ...(state.error ? { error: state.error } : {}), - ...(state.opId ? { opId: state.opId } : {}), - ...(state.jobId ? { jobId: state.jobId } : {}), - ...(state.parserVersion ? { parserVersion: state.parserVersion } : {}), - }); -} diff --git a/src/lib/server/documents/parsed-pdf-reuse.ts b/src/lib/server/documents/parsed-pdf-reuse.ts deleted file mode 100644 index 1bbede4..0000000 --- a/src/lib/server/documents/parsed-pdf-reuse.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { eq } from 'drizzle-orm'; -import { db } from '@/db'; -import { documents } from '@/db/schema'; -import { - normalizeDocumentParseStateForCurrentParserVersion, - parseDocumentParseState, -} from '@/lib/server/documents/parse-state'; - -type ReadyParsedPdfResult = { - parsedJsonKey: string; -}; - -export async function findReusableParsedPdfResult( - documentId: string, -): Promise { - const rows = await db - .select({ - parseState: documents.parseState, - parsedJsonKey: documents.parsedJsonKey, - }) - .from(documents) - .where(eq(documents.id, documentId)); - - for (const row of rows) { - const parsedJsonKey = row.parsedJsonKey?.trim(); - if (!parsedJsonKey) continue; - const parseState = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)); - if (parseState.status !== 'ready') continue; - return { parsedJsonKey }; - } - - return null; -} diff --git a/src/lib/server/documents/pdf-parse-operation.ts b/src/lib/server/documents/pdf-parse-operation.ts deleted file mode 100644 index c356f86..0000000 --- a/src/lib/server/documents/pdf-parse-operation.ts +++ /dev/null @@ -1,30 +0,0 @@ -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 06bafd2..ca2a7ba 100644 --- a/src/lib/server/documents/register-upload.ts +++ b/src/lib/server/documents/register-upload.ts @@ -3,12 +3,7 @@ import { documents } from '@/db/schema'; import { enqueueDocumentPreview, } 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 { PDF_PARSER_VERSION } from '@openreader/compute-core'; import type { BaseDocument, DocumentType } from '@/types/documents'; type RegisterUploadedDocumentInput = { @@ -22,24 +17,8 @@ type RegisterUploadedDocumentInput = { }; export async function registerUploadedDocument(input: RegisterUploadedDocumentInput): Promise { - 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 } - : (startedParse?.parseState ?? { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }), - ) - : null; + const parseState = null; + const parsedJsonKey = null; await db .insert(documents) @@ -84,17 +63,6 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn }, 'Failed to enqueue document preview'); }); - 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', - }); - } - return { id: input.documentId, name: input.name, diff --git a/src/lib/server/jobs/user-pdf-layout-job.ts b/src/lib/server/jobs/user-pdf-layout-job.ts deleted file mode 100644 index 0f44043..0000000 --- a/src/lib/server/jobs/user-pdf-layout-job.ts +++ /dev/null @@ -1,512 +0,0 @@ -import { and, eq, inArray, isNull } from 'drizzle-orm'; -import { db } from '@/db'; -import { documents } from '@/db/schema'; -import { documentKey, getParsedDocumentBlobByKey, putParsedDocumentBlob } from '@/lib/server/documents/blobstore'; -import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse'; -import { serverLogger } from '@/lib/server/logger'; -import { logDegraded, logServerError } from '@/lib/server/errors/logging'; -import { - normalizeDocumentParseStateForCurrentParserVersion, - parseDocumentParseState, - resolveParsedPdfParserVersion, - stringifyDocumentParseState, - type DocumentParseState, -} from '@/lib/server/documents/parse-state'; -import { getCompute } from '@/lib/server/compute'; -import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache'; -import { UNCLAIMED_USER_ID } from '@/lib/server/storage/docstore-legacy'; -import type { - ParsedPdfDocument, - PdfLayoutJobBase, - PdfLayoutJobResult, - PdfLayoutProgress, - WorkerOperationState, -} from '@openreader/compute-core/api-contracts'; -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(); - -const FOLLOWER_WAIT_TIMEOUT_MS = 180_000; -const FOLLOWER_POLL_MS = 1_200; -const PROGRESS_DB_THROTTLE_MS = 10_000; - -type ParseRow = { - userId: string; - parseState: string | null; - parsedJsonKey: string | null; -}; - -function keyFor(input: UserPdfLayoutJobRequest): string { - const forceToken = input.forceToken?.trim(); - if (forceToken) { - return `force:${input.documentId}:${input.namespace || ''}:${forceToken}`; - } - 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)); -} - -function inferNamespaceFromUnclaimedUserId(userId: string): string | null { - const prefix = `${UNCLAIMED_USER_ID}::`; - if (!userId.startsWith(prefix)) return null; - const ns = userId.slice(prefix.length).trim(); - return ns || null; -} - -function rowMatchesScope(row: ParseRow, input: UserPdfLayoutJobRequest): boolean { - const inferredNamespace = inferNamespaceFromUnclaimedUserId(row.userId); - if (inferredNamespace !== null) { - return inferredNamespace === (input.namespace ?? null); - } - - // Regular (non-unclaimed) users are only shared in the non-namespaced path. - if (!input.namespace) return true; - - // In namespaced contexts with regular users, avoid touching other users' - // rows since namespace is not persisted on document rows. - return row.userId === input.userId; -} - -async function loadScopedRows(input: UserPdfLayoutJobRequest): Promise { - const rows = (await db - .select({ - userId: documents.userId, - parseState: documents.parseState, - parsedJsonKey: documents.parsedJsonKey, - }) - .from(documents) - .where(eq(documents.id, input.documentId))) as ParseRow[]; - - return rows.filter((row) => rowMatchesScope(row, input)); -} - -function isReadyRow(row: ParseRow): row is ParseRow & { parsedJsonKey: string } { - if (!row.parsedJsonKey) return false; - return normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)).status === 'ready'; -} - -function userIdsFromRows(rows: ParseRow[]): string[] { - return Array.from(new Set(rows.map((row) => row.userId).filter(Boolean))); -} - -function parseStateMatchCondition(expected: string | null) { - return expected === null ? isNull(documents.parseState) : eq(documents.parseState, expected); -} - -async function loadParsedDocumentForResult( - parsed: ParsedPdfDocument | undefined, - parsedJsonKey: string, -): Promise { - if (parsed) return parsed; - - try { - const json = await getParsedDocumentBlobByKey(parsedJsonKey); - return JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument; - } catch { - return null; - } -} - -async function updateParseStateForUsers(input: { - documentId: string; - userIds: string[]; - parseState: string; - parsedJsonKey?: string | null; -}): Promise { - if (input.userIds.length === 0) return; - - await db - .update(documents) - .set({ - parseState: input.parseState, - ...(typeof input.parsedJsonKey === 'string' || input.parsedJsonKey === null - ? { parsedJsonKey: input.parsedJsonKey } - : {}), - }) - .where( - and( - eq(documents.id, input.documentId), - input.userIds.length === 1 - ? eq(documents.userId, input.userIds[0]) - : inArray(documents.userId, input.userIds), - ), - ); -} - -async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise { - const deadline = Date.now() + FOLLOWER_WAIT_TIMEOUT_MS; - - while (Date.now() < deadline) { - const reusableParsed = await findReusableParsedPdfResult(input.documentId); - if (reusableParsed) { - const readyState: DocumentParseState = { - status: 'ready', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: [input.userId], - parseState: stringifyDocumentParseState(readyState), - parsedJsonKey: reusableParsed.parsedJsonKey, - }); - return; - } - - const rows = await loadScopedRows(input); - const ready = rows.find(isReadyRow); - if (ready) { - const readyState: DocumentParseState = { - status: 'ready', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: [input.userId], - parseState: stringifyDocumentParseState(readyState), - parsedJsonKey: ready.parsedJsonKey, - }); - return; - } - - const statuses = rows.map((row) => normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState))); - const hasInFlight = statuses.some((state) => state.status === 'pending' || state.status === 'running'); - const failedState = statuses.find((state) => state.status === 'failed'); - if (failedState && !hasInFlight) { - const nextFailedState: DocumentParseState = { - status: 'failed', - progress: null, - updatedAt: Date.now(), - ...(failedState.error ? { error: failedState.error } : {}), - ...(failedState.parserVersion ? { parserVersion: failedState.parserVersion } : {}), - }; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: [input.userId], - parseState: stringifyDocumentParseState(nextFailedState), - }); - return; - } - - await sleep(FOLLOWER_POLL_MS); - } -} - -export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise { - const key = keyFor(input); - if (running.has(key)) { - if (!input.forceToken?.trim()) { - await syncCallerToSharedResult(input); - } - return; - } - running.add(key); - - let activeOpId = normalizeOpId(input.initialOpId); - let activeJobId = normalizeOpId(input.initialJobId); - try { - const scopedRows = await loadScopedRows(input); - const scopedUserIds = userIdsFromRows(scopedRows); - - // Non-force jobs can short-circuit by reusing existing ready output from - // any row in the same document scope. - if (!input.forceToken?.trim()) { - const reusableParsed = await findReusableParsedPdfResult(input.documentId); - if (reusableParsed) { - const readyState: DocumentParseState = { - status: 'ready', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: [input.userId], - parseState: stringifyDocumentParseState(readyState), - parsedJsonKey: reusableParsed.parsedJsonKey, - }); - return; - } - - const ready = scopedRows.find(isReadyRow); - if (ready) { - const readyState: DocumentParseState = { - status: 'ready', - progress: null, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - }; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: [input.userId], - parseState: stringifyDocumentParseState(readyState), - parsedJsonKey: ready.parsedJsonKey, - }); - return; - } - } - - const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0]; - if (!coordinator) return; - - 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 inflightState = stringifyDocumentParseState(inflightStateData); - - const claimRows = (await db - .update(documents) - .set({ - parseState: inflightState, - }) - .where( - and( - eq(documents.id, input.documentId), - eq(documents.userId, coordinator.userId), - parseStateMatchCondition(coordinator.parseState), - ), - ) - .returning({ userId: documents.userId })) as Array<{ userId: string }>; - - const claimed = claimRows.some((row) => row.userId === coordinator.userId); - if (!claimed) { - if (!input.forceToken?.trim()) { - await syncCallerToSharedResult(input); - } - return; - } - - const cohortUserIds = Array.from(new Set([...scopedUserIds, input.userId, coordinator.userId])); - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: cohortUserIds, - parseState: inflightState, - }); - - const compute = await getCompute(); - let lastProgressWriteAt = 0; - let lastSnapshotWriteAt = 0; - let lastSnapshotStatus: 'pending' | 'running' | null = null; - let lastSnapshotOpId: string | undefined; - let lastSnapshotJobId: string | undefined; - - const persistRunningState = async (state: DocumentParseState): Promise => { - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: cohortUserIds, - parseState: stringifyDocumentParseState(state), - }); - }; - - const onWorkerSnapshot = async (snapshot: WorkerOperationState): Promise => { - if (snapshot.opId) activeOpId = snapshot.opId; - if (snapshot.jobId) activeJobId = snapshot.jobId; - - const mappedStatus = snapshot.status === 'queued' - ? 'pending' - : snapshot.status === 'running' - ? 'running' - : null; - if (!mappedStatus) return; - - const now = Date.now(); - const forceWrite = ( - mappedStatus !== lastSnapshotStatus - || activeOpId !== lastSnapshotOpId - || activeJobId !== lastSnapshotJobId - ); - if (!forceWrite && (now - lastSnapshotWriteAt) < PROGRESS_DB_THROTTLE_MS) { - return; - } - - const nextState: DocumentParseState = { - status: mappedStatus, - progress: snapshot.progress ?? null, - updatedAt: now, - parserVersion: PDF_PARSER_VERSION, - ...(activeOpId ? { opId: activeOpId } : {}), - ...(activeJobId ? { jobId: activeJobId } : {}), - }; - await persistRunningState(nextState); - lastSnapshotWriteAt = now; - lastSnapshotStatus = mappedStatus; - lastSnapshotOpId = activeOpId; - lastSnapshotJobId = activeJobId; - }; - - const writeProgress = async (progress: PdfLayoutProgress): Promise => { - const now = Date.now(); - if ((now - lastProgressWriteAt) < PROGRESS_DB_THROTTLE_MS && progress.pagesParsed < progress.totalPages) { - return; - } - lastProgressWriteAt = now; - - const runningProgressState: DocumentParseState = { - status: 'running', - progress: { - totalPages: progress.totalPages, - pagesParsed: progress.pagesParsed, - currentPage: progress.currentPage, - phase: progress.phase, - }, - updatedAt: Date.now(), - parserVersion: PDF_PARSER_VERSION, - ...(activeOpId ? { opId: activeOpId } : {}), - ...(activeJobId ? { jobId: activeJobId } : {}), - }; - await persistRunningState(runningProgressState); - }; - - const layout = await compute.parsePdfLayout({ - documentId: input.documentId, - namespace: input.namespace, - documentObjectKey: documentKey(input.documentId, input.namespace), - forceToken: input.forceToken, - onProgress: writeProgress, - onWorkerSnapshot, - }); - - let parsedJsonKey = layout.parsedObjectKey ?? null; - if (!parsedJsonKey) { - if (!layout.parsed) throw new Error('Compute backend did not return parsed result'); - const parsedJson = Buffer.from(JSON.stringify(layout.parsed)); - parsedJsonKey = await putParsedDocumentBlob(input.documentId, parsedJson, input.namespace); - } - - const parsedDoc = await loadParsedDocumentForResult(layout.parsed, parsedJsonKey); - - const finalScopedRows = await loadScopedRows(input); - const finalUserIds = userIdsFromRows(finalScopedRows); - const readyState: DocumentParseState = { - status: 'ready', - progress: null, - updatedAt: Date.now(), - parserVersion: resolveParsedPdfParserVersion(parsedDoc), - ...(activeOpId ? { opId: activeOpId } : {}), - ...(activeJobId ? { jobId: activeJobId } : {}), - }; - const readyUserIds = finalUserIds.length > 0 ? finalUserIds : cohortUserIds; - await updateParseStateForUsers({ - documentId: input.documentId, - userIds: readyUserIds, - parseState: stringifyDocumentParseState(readyState), - parsedJsonKey, - }); - - // Best-effort cache invalidation should not block parse readiness. - for (const userId of readyUserIds) { - void clearTtsSegmentCache({ - userId, - documentId: input.documentId, - readerType: 'pdf', - }).then((cleared) => { - if (cleared.warning) { - logDegraded(serverLogger, { - event: 'documents.parse.cache_invalidation.warning', - msg: 'Parse cache invalidation warning', - step: 'clear_tts_segment_cache', - context: { - documentId: input.documentId, - userId, - warning: cleared.warning, - }, - }); - } - }).catch((cacheError) => { - logDegraded(serverLogger, { - event: 'documents.parse.cache_invalidation.failed', - msg: 'Parse cache invalidation failed', - step: 'clear_tts_segment_cache', - context: { - documentId: input.documentId, - userId, - }, - error: cacheError, - }); - }); - } - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - const parseStatus = 'failed'; - try { - const scopedRows = await loadScopedRows(input); - const scopedUserIds = userIdsFromRows(scopedRows); - const failedState: DocumentParseState = { - status: 'failed', - progress: null, - 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({ - documentId: input.documentId, - userIds: failedUserIds, - parseState: stringifyDocumentParseState(failedState), - }); - } catch (statusError) { - logServerError(serverLogger, { - event: 'documents.parse.status_write.failed', - msg: 'Failed to write parse status', - error: statusError, - context: { - documentId: input.documentId, - parseStatus, - }, - normalize: { code: 'DOCUMENT_PARSE_STATUS_WRITE_FAILED', errorClass: 'db' }, - }); - } - logServerError(serverLogger, { - event: 'documents.parse.job.failed', - msg: 'Parse job failed', - error, - context: { - documentId: input.documentId, - parseStatus, - }, - normalize: { code: 'DOCUMENT_PARSE_JOB_FAILED', errorClass: 'upstream' }, - }); - } finally { - running.delete(key); - } -} - -export function enqueueParsePdfJob(input: UserPdfLayoutJobRequest): void { - Promise.resolve() - .then(() => parsePdfJob(input)) - .catch((error) => { - logServerError(serverLogger, { - event: 'documents.parse.job.uncaught_error', - msg: 'Parse job uncaught error', - error, - context: { documentId: input.documentId }, - normalize: { code: 'DOCUMENT_PARSE_JOB_UNCAUGHT_ERROR', errorClass: 'unknown' }, - }); - }); -} diff --git a/src/lib/server/pdf-parse/artifact.ts b/src/lib/server/pdf-parse/artifact.ts new file mode 100644 index 0000000..2cea275 --- /dev/null +++ b/src/lib/server/pdf-parse/artifact.ts @@ -0,0 +1,39 @@ +import { PDF_PARSER_VERSION } from '@openreader/compute-core'; +import { + documentParsedKeyForVersion, + getParsedDocumentBlobByKey, + isMissingBlobError, +} from '@/lib/server/documents/blobstore'; +import type { ParsedPdfDocument } from '@/types/parsed-pdf'; + +export interface ParsedPdfArtifact { + key: string; + bytes: Buffer; + parsed: ParsedPdfDocument; +} + +function parseArtifact(bytes: Buffer): ParsedPdfDocument { + return JSON.parse(bytes.toString('utf8')) as ParsedPdfDocument; +} + +export async function readParsedPdfArtifactByKey(key: string): Promise { + try { + const bytes = await getParsedDocumentBlobByKey(key); + return { + key, + bytes, + parsed: parseArtifact(bytes), + }; + } catch (error) { + if (isMissingBlobError(error)) return null; + throw error; + } +} + +export async function readCurrentParsedPdfArtifact(input: { + documentId: string; + namespace: string | null; +}): Promise { + const key = documentParsedKeyForVersion(input.documentId, input.namespace, PDF_PARSER_VERSION); + return readParsedPdfArtifactByKey(key); +} diff --git a/src/lib/server/pdf-parse/operation.ts b/src/lib/server/pdf-parse/operation.ts new file mode 100644 index 0000000..f00aa19 --- /dev/null +++ b/src/lib/server/pdf-parse/operation.ts @@ -0,0 +1,65 @@ +import { buildPdfOpKey } from '@/lib/server/compute/worker'; +import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; +import { + fetchWorkerOperationState, + fetchWorkerOperationStateByKey, +} from '@/lib/server/compute/worker-op-state'; +import { documentKey } from '@/lib/server/documents/blobstore'; +import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; + +function currentPdfOperationInput(documentId: string, namespace: string | null, forceToken?: string): { + documentId: string; + namespace: string | null; + documentObjectKey: string; + forceToken?: string; +} { + return { + documentId, + namespace, + documentObjectKey: documentKey(documentId, namespace), + ...(forceToken ? { forceToken } : {}), + }; +} + +export function buildCurrentPdfParseOpKeyPrefix(input: { + documentId: string; + namespace: string | null; +}): string { + return buildPdfOpKey(currentPdfOperationInput(input.documentId, input.namespace)); +} + +export async function lookupCurrentPdfParseOperation(input: { + documentId: string; + namespace: string | null; +}): Promise | null> { + return fetchWorkerOperationStateByKey( + buildCurrentPdfParseOpKeyPrefix(input), + ); +} + +export async function createOrReuseCurrentPdfParseOperation(input: { + documentId: string; + namespace: string | null; + forceToken?: string; +}): Promise> { + return createOrReusePdfWorkerOperation(currentPdfOperationInput( + input.documentId, + input.namespace, + input.forceToken, + )); +} + +export async function fetchPdfParseOperation(opId: string): Promise | null> { + return fetchWorkerOperationState(opId); +} + +export function isPdfParseOperationForDocument( + state: WorkerOperationState, + input: { + documentId: string; + namespace: string | null; + }, +): boolean { + if (state.kind !== 'pdf_layout') return false; + return state.opKey.startsWith(buildCurrentPdfParseOpKeyPrefix(input)); +} diff --git a/src/lib/server/pdf-parse/snapshot.ts b/src/lib/server/pdf-parse/snapshot.ts new file mode 100644 index 0000000..a393c29 --- /dev/null +++ b/src/lib/server/pdf-parse/snapshot.ts @@ -0,0 +1,41 @@ +import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; +import type { PdfParseStatus } from '@/types/parsed-pdf'; +import type { PdfParseSnapshot } from '@/lib/server/pdf-parse/types'; + +function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus { + switch (status) { + case 'queued': + return 'pending'; + case 'running': + return 'running'; + case 'succeeded': + return 'ready'; + case 'failed': + return 'failed'; + default: + return 'pending'; + } +} + +export function parsedObjectKeyFromWorkerState( + state: WorkerOperationState, +): string | null { + const result = state.result; + if (!result || typeof result !== 'object' || !('parsedObjectKey' in result)) return null; + const value = result.parsedObjectKey; + if (typeof value !== 'string') return null; + const normalized = value.trim(); + return normalized || null; +} + +export function pdfParseSnapshotFromWorkerState( + state: WorkerOperationState, +): PdfParseSnapshot { + const parseStatus = mapWorkerStatusToParseStatus(state.status); + return { + parseStatus, + parseProgress: parseStatus === 'running' ? (state.progress ?? null) : null, + opId: state.opId?.trim() || null, + ...(parseStatus === 'failed' && state.error?.message ? { error: state.error.message } : {}), + }; +} diff --git a/src/lib/server/pdf-parse/types.ts b/src/lib/server/pdf-parse/types.ts new file mode 100644 index 0000000..ac30cec --- /dev/null +++ b/src/lib/server/pdf-parse/types.ts @@ -0,0 +1,8 @@ +import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; + +export interface PdfParseSnapshot { + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId: string | null; + error?: string | null; +} diff --git a/tests/unit/pdf-parse-state.vitest.spec.ts b/tests/unit/pdf-parse-state.vitest.spec.ts deleted file mode 100644 index aa3384b..0000000 --- a/tests/unit/pdf-parse-state.vitest.spec.ts +++ /dev/null @@ -1,84 +0,0 @@ -import { describe, expect, test } from 'vitest'; -import { PDF_PARSER_VERSION } from '@openreader/compute-core'; -import { - normalizeDocumentParseStateForCurrentParserVersion, - parseDocumentParseState, - stringifyDocumentParseState, -} from '../../src/lib/server/documents/parse-state'; - -describe('document parse state parser-version handling', () => { - test('preserves parserVersion through stringify/parse', () => { - const state = parseDocumentParseState(stringifyDocumentParseState({ - status: 'ready', - progress: null, - updatedAt: 1234, - parserVersion: PDF_PARSER_VERSION, - })); - - expect(state).toEqual({ - status: 'ready', - progress: null, - updatedAt: 1234, - parserVersion: PDF_PARSER_VERSION, - }); - }); - - test('invalidates ready and inflight states from older parser versions', () => { - expect(normalizeDocumentParseStateForCurrentParserVersion({ - status: 'ready', - progress: null, - updatedAt: 1111, - parserVersion: 'old-parser', - opId: 'op-old', - }, 2222)).toEqual({ - status: 'pending', - progress: null, - updatedAt: 2222, - }); - - expect(normalizeDocumentParseStateForCurrentParserVersion({ - status: 'running', - progress: { - totalPages: 10, - pagesParsed: 3, - currentPage: 4, - phase: 'infer', - }, - updatedAt: 1111, - parserVersion: 'old-parser', - opId: 'op-old', - }, 3333)).toEqual({ - status: 'pending', - progress: null, - updatedAt: 3333, - }); - }); - - test('keeps failed states and current-version ready states intact', () => { - expect(normalizeDocumentParseStateForCurrentParserVersion({ - status: 'failed', - progress: null, - updatedAt: 1111, - error: 'no text layer', - parserVersion: 'old-parser', - })).toEqual({ - status: 'failed', - progress: null, - updatedAt: 1111, - error: 'no text layer', - parserVersion: 'old-parser', - }); - - expect(normalizeDocumentParseStateForCurrentParserVersion({ - status: 'ready', - progress: null, - updatedAt: 1111, - parserVersion: PDF_PARSER_VERSION, - })).toEqual({ - status: 'ready', - progress: null, - updatedAt: 1111, - parserVersion: PDF_PARSER_VERSION, - }); - }); -}); 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 index d39d9f5..c86f679 100644 --- a/tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts +++ b/tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts @@ -4,16 +4,15 @@ 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, + type: 'pdf', }, requireAuthContext: vi.fn(), - backfillPendingPdfParseOperation: vi.fn(), - healStaleDocumentParseState: vi.fn(async ({ state }) => state), + fetchPdfParseOperation: vi.fn(), + isPdfParseOperationForDocument: vi.fn(), + getWorkerClientConfigFromEnv: vi.fn(), })); vi.mock('@/db', () => ({ @@ -26,12 +25,13 @@ vi.mock('@/lib/server/auth/auth', () => ({ requireAuthContext: hoisted.requireAuthContext, })); -vi.mock('@/lib/server/documents/parse-state-backfill', () => ({ - backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation, +vi.mock('@/lib/server/pdf-parse/operation', () => ({ + fetchPdfParseOperation: hoisted.fetchPdfParseOperation, + isPdfParseOperationForDocument: hoisted.isPdfParseOperationForDocument, })); -vi.mock('@/lib/server/documents/parse-state-healing', () => ({ - healStaleDocumentParseState: hoisted.healStaleDocumentParseState, +vi.mock('@/lib/server/compute/worker', () => ({ + getWorkerClientConfigFromEnv: hoisted.getWorkerClientConfigFromEnv, })); vi.mock('@/lib/server/documents/blobstore', () => ({ @@ -55,102 +55,81 @@ vi.mock('@/lib/server/logger', () => ({ }, requestId: 'req-test', })), - hashForLog: vi.fn(() => 'user-hash'), })); -describe('GET /api/documents/[id]/parsed/events legacy backfill', () => { +describe('GET /api/documents/[id]/parsed/events worker proxy', () => { 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 []; - }), + where: vi.fn(() => ({ + limit: vi.fn(async () => [{ ...hoisted.row }]), + })), })), })), }; hoisted.requireAuthContext.mockReset(); hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' }); - hoisted.backfillPendingPdfParseOperation.mockReset(); - hoisted.healStaleDocumentParseState.mockClear(); + hoisted.fetchPdfParseOperation.mockReset(); + hoisted.fetchPdfParseOperation.mockResolvedValue({ + opId: 'op-1', + opKey: 'pdf_layout|v1|parser|doc-1||doc-key|', + kind: 'pdf_layout', + jobId: 'job-1', + status: 'running', + queuedAt: Date.now() - 1000, + updatedAt: Date.now(), + }); + hoisted.isPdfParseOperationForDocument.mockReset(); + hoisted.isPdfParseOperationForDocument.mockReturnValue(true); + hoisted.getWorkerClientConfigFromEnv.mockReset(); + hoisted.getWorkerClientConfigFromEnv.mockReturnValue({ + baseUrl: 'http://worker.local', + token: 'worker-token', + }); + global.fetch = vi.fn(async () => new Response( + 'event: snapshot\ndata: {"eventId":1,"snapshot":{"opId":"op-1","status":"running"}}\n\n', + { + status: 200, + headers: { + 'Content-Type': 'text/event-stream; charset=utf-8', + }, + }, + )) as typeof fetch; + }); - 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; + test('requires an opId', async () => { + const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route'); + const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events'), { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(400); + await expect(response.json()).resolves.toMatchObject({ error: 'opId is required' }); + }); + + test('proxies the worker SSE stream after validating document ownership', async () => { + const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route'); + const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events?opId=op-1'), { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(200); + expect(response.headers.get('content-type')).toContain('text/event-stream'); + const text = await response.text(); + expect(text).toContain('event: snapshot'); + expect(text).toContain('"opId":"op-1"'); + expect(hoisted.fetchPdfParseOperation).toHaveBeenCalledWith('op-1'); + expect(hoisted.isPdfParseOperationForDocument).toHaveBeenCalled(); + expect(global.fetch).toHaveBeenCalledWith( + 'http://worker.local/ops/op-1/events', + expect.anything(), + ); }); 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 75c5cd2..ae1e6d3 100644 --- a/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts +++ b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts @@ -4,19 +4,20 @@ 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, - parsedJsonKey: null as string | null, + type: 'pdf', }, requireAuthContext: vi.fn(), - fetchWorkerOperationState: vi.fn(), - healStaleDocumentParseState: vi.fn(async ({ state }) => state), - startPdfParseOperation: vi.fn(), - enqueueParsePdfJob: vi.fn(), + readCurrentParsedPdfArtifact: vi.fn(), + readParsedPdfArtifactByKey: vi.fn(), + lookupCurrentPdfParseOperation: vi.fn(), + createOrReuseCurrentPdfParseOperation: vi.fn(), + checkJobRate: vi.fn(), + getPdfLayoutRateConfig: vi.fn(), + getResolvedRuntimeConfig: vi.fn(), + buildComputeRateLimitedResponse: vi.fn(), })); vi.mock('@/db', () => ({ @@ -29,29 +30,31 @@ vi.mock('@/lib/server/auth/auth', () => ({ requireAuthContext: hoisted.requireAuthContext, })); -vi.mock('@/lib/server/compute/worker-op-state', () => ({ - fetchWorkerOperationState: hoisted.fetchWorkerOperationState, +vi.mock('@/lib/server/pdf-parse/artifact', () => ({ + readCurrentParsedPdfArtifact: hoisted.readCurrentParsedPdfArtifact, + readParsedPdfArtifactByKey: hoisted.readParsedPdfArtifactByKey, })); -vi.mock('@/lib/server/documents/parse-state-healing', () => ({ - healStaleDocumentParseState: hoisted.healStaleDocumentParseState, +vi.mock('@/lib/server/pdf-parse/operation', () => ({ + lookupCurrentPdfParseOperation: hoisted.lookupCurrentPdfParseOperation, + createOrReuseCurrentPdfParseOperation: hoisted.createOrReuseCurrentPdfParseOperation, })); -vi.mock('@/lib/server/documents/pdf-parse-operation', () => ({ - startPdfParseOperation: hoisted.startPdfParseOperation, +vi.mock('@/lib/server/rate-limit/job-rate-limiter', () => ({ + checkJobRate: hoisted.checkJobRate, + getPdfLayoutRateConfig: hoisted.getPdfLayoutRateConfig, })); -vi.mock('@/lib/server/jobs/user-pdf-layout-job', () => ({ - enqueueParsePdfJob: hoisted.enqueueParsePdfJob, +vi.mock('@/lib/server/rate-limit/problem-response', () => ({ + buildComputeRateLimitedResponse: hoisted.buildComputeRateLimitedResponse, +})); + +vi.mock('@/lib/server/runtime-config', () => ({ + getResolvedRuntimeConfig: hoisted.getResolvedRuntimeConfig, })); vi.mock('@/lib/server/documents/blobstore', () => ({ - documentKey: vi.fn(), - getParsedDocumentBlob: vi.fn(), - getParsedDocumentBlobByKey: vi.fn(), - isMissingBlobError: vi.fn(() => false), isValidDocumentId: vi.fn(() => true), - putParsedDocumentBlob: vi.fn(), })); vi.mock('@/lib/server/storage/s3', () => ({ @@ -71,63 +74,104 @@ vi.mock('@/lib/server/logger', () => ({ }, requestId: 'req-test', })), - hashForLog: vi.fn(() => 'user-hash'), })); -describe('GET /api/documents/[id]/parsed pure data fetch', () => { - beforeEach(async () => { - process.env.BASE_URL = 'http://localhost:3003'; - process.env.AUTH_SECRET = 'test-secret'; - - hoisted.row = { - id: 'doc-1', - userId: 'user-1', - parseState: null, - parsedJsonKey: null, - }; +describe('GET/POST /api/documents/[id]/parsed worker-owned flow', () => { + beforeEach(() => { hoisted.db = { select: vi.fn(() => ({ from: vi.fn(() => ({ - where: vi.fn(async () => ([{ ...hoisted.row }])), - })), - })), - update: vi.fn(() => ({ - set: vi.fn((values: { parseState?: string | null; parsedJsonKey?: string | null }) => ({ - where: vi.fn(async () => { - if (typeof values.parseState !== 'undefined') { - hoisted.row.parseState = values.parseState; - } - if (typeof values.parsedJsonKey !== 'undefined') { - hoisted.row.parsedJsonKey = values.parsedJsonKey; - } - return []; - }), + where: vi.fn(() => ({ + limit: vi.fn(async () => [{ ...hoisted.row }]), + })), })), })), }; hoisted.requireAuthContext.mockReset(); hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' }); - hoisted.fetchWorkerOperationState.mockReset(); - hoisted.fetchWorkerOperationState.mockResolvedValue(null); - hoisted.healStaleDocumentParseState.mockClear(); - hoisted.startPdfParseOperation.mockReset(); - hoisted.enqueueParsePdfJob.mockReset(); + hoisted.readCurrentParsedPdfArtifact.mockReset(); + hoisted.readCurrentParsedPdfArtifact.mockResolvedValue(null); + hoisted.readParsedPdfArtifactByKey.mockReset(); + hoisted.readParsedPdfArtifactByKey.mockResolvedValue(null); + hoisted.lookupCurrentPdfParseOperation.mockReset(); + hoisted.lookupCurrentPdfParseOperation.mockResolvedValue(null); + hoisted.createOrReuseCurrentPdfParseOperation.mockReset(); + hoisted.checkJobRate.mockReset(); + hoisted.checkJobRate.mockResolvedValue({ allowed: true }); + hoisted.getPdfLayoutRateConfig.mockReset(); + hoisted.getPdfLayoutRateConfig.mockReturnValue({}); + hoisted.getResolvedRuntimeConfig.mockReset(); + hoisted.getResolvedRuntimeConfig.mockResolvedValue({}); + hoisted.buildComputeRateLimitedResponse.mockReset(); }); - test('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => { + test('GET returns pending when no current artifact or worker op exists', 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, { + const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed'), { params: Promise.resolve({ id: 'doc-1' }), }); expect(response.status).toBe(409); await expect(response.json()).resolves.toMatchObject({ parseStatus: 'pending', + parseProgress: null, opId: null, }); - expect(hoisted.startPdfParseOperation).not.toHaveBeenCalled(); - expect(hoisted.enqueueParsePdfJob).not.toHaveBeenCalled(); - expect(hoisted.row.parseState).toBeNull(); + }); + + test('GET reads the artifact referenced by a succeeded worker op', async () => { + hoisted.lookupCurrentPdfParseOperation.mockResolvedValue({ + opId: 'op-1', + opKey: 'pdf_layout|v1|parser|doc-1||doc-key|', + kind: 'pdf_layout', + jobId: 'job-1', + status: 'succeeded', + queuedAt: Date.now() - 1000, + updatedAt: Date.now(), + result: { parsedObjectKey: 'parsed-key.json' }, + }); + hoisted.readParsedPdfArtifactByKey.mockResolvedValue({ + key: 'parsed-key.json', + bytes: Buffer.from(JSON.stringify({ documentId: 'doc-1', pages: [] })), + parsed: { documentId: 'doc-1', pages: [] }, + }); + + const { GET } = await import('../../src/app/api/documents/[id]/parsed/route'); + const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed'), { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toMatchObject({ documentId: 'doc-1' }); + expect(hoisted.readParsedPdfArtifactByKey).toHaveBeenCalledWith('parsed-key.json'); + }); + + test('POST creates a worker op when replace is requested', async () => { + hoisted.createOrReuseCurrentPdfParseOperation.mockResolvedValue({ + opId: 'op-force-1', + opKey: 'pdf_layout|v1|parser|doc-1||doc-key|force', + kind: 'pdf_layout', + jobId: 'job-force-1', + status: 'queued', + queuedAt: Date.now(), + updatedAt: Date.now(), + }); + + const { POST } = await import('../../src/app/api/documents/[id]/parsed/route'); + const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', { + method: 'POST', + body: JSON.stringify({ replace: true }), + headers: { 'Content-Type': 'application/json' }, + }); + const response = await POST(request, { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(202); + await expect(response.json()).resolves.toMatchObject({ + parseStatus: 'pending', + opId: 'op-force-1', + }); + expect(hoisted.createOrReuseCurrentPdfParseOperation).toHaveBeenCalled(); }); }); diff --git a/tests/unit/worker-parse-state.vitest.spec.ts b/tests/unit/worker-parse-state.vitest.spec.ts deleted file mode 100644 index 91563b3..0000000 --- a/tests/unit/worker-parse-state.vitest.spec.ts +++ /dev/null @@ -1,122 +0,0 @@ -import { describe, expect, test } from 'vitest'; -import { PDF_PARSER_VERSION } from '@openreader/compute-core'; -import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; -import { - documentParseStateFromWorkerState, - isWorkerOperationStateStale, - snapshotFromWorkerState, -} from '../../src/lib/server/compute/worker-parse-state'; - -function makeWorkerState( - overrides: Partial>, -): WorkerOperationState { - return { - opId: 'op-123', - opKey: `pdf_layout|v1|${PDF_PARSER_VERSION}|doc-1`, - kind: 'pdf_layout', - jobId: 'job-123', - status: 'queued', - queuedAt: 1_700_000_000_000, - updatedAt: 1_700_000_000_000, - ...overrides, - }; -} - -describe('worker parse state mapping', () => { - test('maps queued worker state to pending parse state with op identifiers', () => { - const workerState = makeWorkerState({ - status: 'queued', - progress: { - totalPages: 500, - pagesParsed: 0, - currentPage: 1, - phase: 'infer', - }, - }); - - expect(snapshotFromWorkerState(workerState)).toEqual({ - parseStatus: 'pending', - parseProgress: null, - }); - expect(documentParseStateFromWorkerState(workerState, 1234)).toEqual({ - status: 'pending', - progress: null, - updatedAt: 1234, - parserVersion: PDF_PARSER_VERSION, - opId: 'op-123', - jobId: 'job-123', - }); - }); - - test('maps running worker state to running parse state with progress', () => { - const workerState = makeWorkerState({ - status: 'running', - progress: { - totalPages: 500, - pagesParsed: 120, - currentPage: 121, - phase: 'infer', - }, - }); - - expect(documentParseStateFromWorkerState(workerState, 5678)).toEqual({ - status: 'running', - progress: { - totalPages: 500, - pagesParsed: 120, - currentPage: 121, - phase: 'infer', - }, - updatedAt: 5678, - parserVersion: PDF_PARSER_VERSION, - opId: 'op-123', - jobId: 'job-123', - }); - }); - - test('maps failed worker state to failed parse state and preserves the worker error', () => { - const workerState = makeWorkerState({ - status: 'failed', - error: { - code: 'PDF_PARSE_FAILED', - message: 'layout model crashed', - }, - }); - - expect(documentParseStateFromWorkerState(workerState, 9999)).toEqual({ - status: 'failed', - progress: null, - updatedAt: 9999, - parserVersion: PDF_PARSER_VERSION, - opId: 'op-123', - jobId: 'job-123', - error: 'layout model crashed', - }); - }); - - test('treats old inflight worker states as stale', () => { - const workerState = makeWorkerState({ - status: 'running', - updatedAt: 1_000, - progress: { - totalPages: 500, - pagesParsed: 250, - currentPage: 251, - phase: 'infer', - }, - }); - - expect(isWorkerOperationStateStale(workerState, 5_000, 6_001)).toBe(true); - expect(isWorkerOperationStateStale(workerState, 5_000, 5_999)).toBe(false); - }); - - test('never treats terminal worker states as stale', () => { - const failedState = makeWorkerState({ - status: 'failed', - updatedAt: 1_000, - error: { code: 'PDF_PARSE_FAILED', message: 'crashed' }, - }); - - expect(isWorkerOperationStateStale(failedState, 5_000, 99_999)).toBe(false); - }); -});