diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 4966e17..1720f63 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -146,6 +146,7 @@ export function usePdfDocument(): PdfDocumentState { const [parsedDocument, setParsedDocument] = useState(null); const [parseStatus, setParseStatus] = useState(null); const [parseProgress, setParseProgress] = useState(null); + const [, setActiveParseOpId] = useState(null); const [documentSettings, setDocumentSettings] = useState(DEFAULT_DOCUMENT_SETTINGS); const [parsedOverlayEnabled, setParsedOverlayEnabled] = useState(false); const [isAudioCombining] = useState(false); @@ -173,6 +174,7 @@ export function usePdfDocument(): PdfDocumentState { documentId: string, initialStatus: PdfParseStatus | null, signal: AbortSignal, + initialOpId?: string | null, ): Promise => { // Legacy PDFs may have null parseStatus; treat as pending so opening the // document backfills parse output via the parsed endpoint polling path. @@ -182,22 +184,30 @@ export function usePdfDocument(): PdfDocumentState { const delayMs = 1200; const retryFailed = effectiveInitialStatus === 'failed'; let attempt = 0; + let effectiveOpId = initialOpId?.trim() || null; while (!signal.aborted) { if (signal.aborted) return; const result = await getParsedPdfDocument(documentId, { signal, retryFailed: retryFailed && attempt === 0, + ...(effectiveOpId ? { opId: effectiveOpId } : {}), }); if (result.status === 'ready') { setParsedDocument(result.parsed); setParseStatus('ready'); setParseProgress(null); + setActiveParseOpId(null); return; } + if ('opId' in result && typeof result.opId === 'string' && result.opId.trim()) { + effectiveOpId = result.opId.trim(); + setActiveParseOpId(effectiveOpId); + } setParseStatus(result.status); setParseProgress(result.parseProgress ?? null); if (result.status === 'failed') { setParsedDocument(null); + setActiveParseOpId(null); return; } attempt += 1; @@ -216,24 +226,36 @@ export function usePdfDocument(): PdfDocumentState { } }, []); - const startParsedPolling = useCallback((documentId: string) => { + const startParsedPolling = useCallback((documentId: string, initialOpId?: string | null) => { parsePollAbortRef.current?.abort(); parseSseCloseRef.current?.(); parseSseCloseRef.current = null; setParseProgress(null); + setActiveParseOpId(initialOpId?.trim() || null); const controller = new AbortController(); parsePollAbortRef.current = controller; const closeSse = subscribeParsedPdfDocumentEvents(documentId, { + opId: initialOpId?.trim() || null, + }, { onSnapshot: (snapshot) => { if (controller.signal.aborted) return; + if (typeof snapshot.opId === 'string' && snapshot.opId.trim()) { + setActiveParseOpId(snapshot.opId.trim()); + } setParseStatus(snapshot.parseStatus); setParseProgress(snapshot.parseProgress); if (snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed') { if (snapshot.parseStatus === 'failed') { setParsedDocument(null); + setActiveParseOpId(null); } else { - void fetchParsedDocument(documentId, 'ready', controller.signal); + void fetchParsedDocument( + documentId, + 'ready', + controller.signal, + typeof snapshot.opId === 'string' ? snapshot.opId : (initialOpId ?? null), + ); } closeSse(); parseSseCloseRef.current = null; @@ -263,7 +285,7 @@ export function usePdfDocument(): PdfDocumentState { parsePollAbortRef.current = null; } }, { once: true }); - }, [fetchParsedDocument]); + }, [fetchParsedDocument, setActiveParseOpId]); useEffect(() => { pdfDocumentRef.current = pdfDocument; @@ -449,6 +471,7 @@ export function usePdfDocument(): PdfDocumentState { setParsedDocument(null); setParseStatus(null); setParseProgress(null); + setActiveParseOpId(null); setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS); const meta = await getDocumentMetadata(id, { signal: controller.signal }); @@ -461,7 +484,8 @@ export function usePdfDocument(): PdfDocumentState { const initialParseStatus = (meta.parseStatus ?? null) as PdfParseStatus | null; setParseStatus(initialParseStatus); setParseProgress(null); - startParsedPolling(id); + setActiveParseOpId(null); + startParsedPolling(id, null); void fetchDocumentSettings(id, controller.signal); } @@ -513,11 +537,12 @@ export function usePdfDocument(): PdfDocumentState { const forceReparseParsedPdf = useCallback(async (): Promise => { if (!currDocId) return; try { - await forceReparsePdfDocument(currDocId); + const forced = await forceReparsePdfDocument(currDocId); setParsedDocument(null); - setParseStatus('pending'); + setParseStatus(forced.status); setParseProgress(null); - startParsedPolling(currDocId); + setActiveParseOpId(forced.opId ?? null); + startParsedPolling(currDocId, forced.opId ?? null); } catch (error) { console.error('Failed to force PDF reparse:', error); } @@ -548,6 +573,7 @@ export function usePdfDocument(): PdfDocumentState { setParsedDocument(null); setParseStatus(null); setParseProgress(null); + setActiveParseOpId(null); setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS); pageTextCacheRef.current.clear(); stop(); diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 28bc7b8..cbdbc7c 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -31,11 +31,13 @@ type ParseRow = { type ParsedSnapshot = { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null; + opId?: string | null; }; type SnapshotState = { snapshot: ParsedSnapshot; opId: string | null; + fromWorker: boolean; }; function s3NotConfiguredResponse(): NextResponse { @@ -49,20 +51,28 @@ function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -async function toSnapshotState(row: ParseRow): Promise { +async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise { const state = await healStaleDocumentParseState({ documentId: row.id, userId: row.userId, state: parseDocumentParseState(row.parseState), }); const parseStatus = normalizeParseStatus(state.status); - const opId = typeof state.opId === 'string' && state.opId.trim() ? state.opId.trim() : null; - if (opId && parseStatus !== 'ready') { + 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) { return { - snapshot: snapshotFromWorkerState(workerState), - opId, + snapshot: { + ...snapshotFromWorkerState(workerState), + opId: workerState.opId, + }, + opId: workerState.opId, + fromWorker: true, }; } } @@ -70,8 +80,10 @@ async function toSnapshotState(row: ParseRow): Promise { snapshot: { parseStatus, parseProgress: state.progress ?? null, + opId, }, opId, + fromWorker: false, }; } @@ -97,8 +109,9 @@ async function syncFromDb(input: { storageUserId: string; allowedUserIds: string[]; signature: string; + preferredOpId?: string | null; writeSnapshot: (snapshot: ParsedSnapshot) => void; -}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; signature: string } | null> { +}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; fromWorker: boolean; signature: string } | null> { const nextRow = await loadPreferredRow({ documentId: input.documentId, storageUserId: input.storageUserId, @@ -106,7 +119,7 @@ async function syncFromDb(input: { }); if (!nextRow) return null; - const nextState = await toSnapshotState(nextRow); + const nextState = await toSnapshotState(nextRow, input.preferredOpId); const nextSignature = JSON.stringify(nextState.snapshot); if (nextSignature !== input.signature) { input.writeSnapshot(nextState.snapshot); @@ -115,6 +128,7 @@ async function syncFromDb(input: { return { snapshot: nextState.snapshot, opId: nextState.opId, + fromWorker: nextState.fromWorker, signature: nextSignature, }; } @@ -131,6 +145,10 @@ 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 unclaimedUserId = getUnclaimedUserIdForNamespace(testNamespace); @@ -147,16 +165,9 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Not found' }, { status: 404 }); } - const initialState = await toSnapshotState(row); + const initialState = await toSnapshotState(row, requestedOpId); const workerCfg = getWorkerClientConfigFromEnv(); const encoder = new TextEncoder(); - console.info('[parsed/events] stream open', { - documentId: id, - storageUserId, - initialParseStatus: initialState.snapshot.parseStatus, - initialOpId: initialState.opId, - }); - const stream = new ReadableStream({ start(controller) { let closed = false; @@ -193,12 +204,23 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string const runWorkerProxy = async () => { let current = initialState.snapshot; let signature = JSON.stringify(current); - let currentOpId = initialState.opId; + let currentOpId = requestedOpId ?? initialState.opId; + let currentFromWorker = initialState.fromWorker; let lastEventId: number | null = null; + let loggedMissingOpId = false; + const pinnedRequestedOp = requestedOpId ?? null; + const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot, fromWorker: boolean): boolean => { + const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed'; + if (!isTerminal) return false; + // If caller pinned an opId, keep streaming until worker confirms that + // op state. DB fallback can report stale terminal status for other rows. + if (pinnedRequestedOp && snapshot.opId === pinnedRequestedOp && !fromWorker) return false; + return true; + }; writeSnapshot(current); - if (current.parseStatus === 'ready' || current.parseStatus === 'failed') { + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { closeStream(); return; } @@ -215,6 +237,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string storageUserId, allowedUserIds, signature, + preferredOpId: requestedOpId ?? currentOpId, writeSnapshot, }).then((next) => { if (closed) return; @@ -224,7 +247,8 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - if (next.opId !== currentOpId) { + currentFromWorker = next.fromWorker; + if (!requestedOpId && next.opId !== currentOpId) { currentOpId = next.opId; lastEventId = null; if (workerAbort) { @@ -232,7 +256,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string workerAbort = null; } } - if (current.parseStatus === 'ready' || current.parseStatus === 'failed') { + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { closeStream(); } }).catch((error) => { @@ -254,6 +278,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string storageUserId, allowedUserIds, signature, + preferredOpId: requestedOpId ?? currentOpId, writeSnapshot, }); if (!next) { @@ -262,14 +287,22 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - currentOpId = next.opId; + currentFromWorker = next.fromWorker; + currentOpId = requestedOpId ?? next.opId; if (!currentOpId) { - console.warn('[parsed/events] waiting for worker opId', { - documentId: id, - parseStatus: current.parseStatus, - }); + if (!loggedMissingOpId) { + loggedMissingOpId = true; + console.warn('[parsed/events] missing worker opId while parse is non-terminal', { + documentId: id, + storageUserId, + parseStatus: current.parseStatus, + requestedOpId, + }); + } + } else if (loggedMissingOpId) { + loggedMissingOpId = false; } - if (current.parseStatus === 'ready' || current.parseStatus === 'failed') { + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { closeStream(); return; } @@ -358,15 +391,19 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string ); if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; - const nextSnapshot = snapshotFromWorkerState(workerSnapshot); + const nextSnapshot: ParsedSnapshot = { + ...snapshotFromWorkerState(workerSnapshot), + opId: workerSnapshot.opId, + }; const nextSignature = JSON.stringify(nextSnapshot); if (nextSignature !== signature) { current = nextSnapshot; signature = nextSignature; + currentFromWorker = true; writeSnapshot(current); } - if (current.parseStatus === 'ready' || current.parseStatus === 'failed') { + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { closeStream(); return; } @@ -380,6 +417,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string storageUserId, allowedUserIds, signature, + preferredOpId: requestedOpId ?? currentOpId, writeSnapshot, }); if (!next) { @@ -388,11 +426,12 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } current = next.snapshot; signature = next.signature; - if (next.opId !== currentOpId) { + currentFromWorker = next.fromWorker; + if (!requestedOpId && next.opId !== currentOpId) { currentOpId = next.opId; lastEventId = null; } - if (current.parseStatus === 'ready' || current.parseStatus === 'failed') { + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { closeStream(); return; } diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 8713c95..2dccd90 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -4,15 +4,17 @@ import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { mergeNonReadyParseSnapshot } from '@/lib/server/compute/worker-parse-state'; +import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; +import { snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { + documentKey, getParsedDocumentBlob, getParsedDocumentBlobByKey, isMissingBlobError, isValidDocumentId, + putParsedDocumentBlob, } from '@/lib/server/documents/blobstore'; -import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; import { normalizeParseStatus, parseDocumentParseState, @@ -22,7 +24,7 @@ import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state- import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import type { ParsedPdfDocument } from '@/types/parsed-pdf'; -import type { PdfLayoutJobResult } from '@openreader/compute-core/api-contracts'; +import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -33,11 +35,146 @@ 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: { + documentId: string; + allowedUserIds: string[]; +}): Promise { + return (await db + .select({ + id: documents.id, + userId: documents.userId, + parseState: documents.parseState, + parsedJsonKey: documents.parsedJsonKey, + }) + .from(documents) + .where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[]; +} + +function pickPreferredRow(rows: ParseRow[], storageUserId: string): ParseRow | null { + return rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0] ?? null; +} + +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))); +} + +async function finalizeFromWorkerState(input: { + workerState: WorkerOperationState; + row: ParseRow; + namespace: string | null; +}): Promise { + const snapshot = snapshotFromWorkerState(input.workerState); + + if (snapshot.parseStatus === 'pending' || snapshot.parseStatus === 'running') { + return NextResponse.json({ + parseStatus: snapshot.parseStatus, + parseProgress: snapshot.parseProgress, + opId: input.workerState.opId, + }, { status: 202 }); + } + + if (snapshot.parseStatus === 'failed') { + await writeParseRowState({ + documentId: input.row.id, + userId: input.row.userId, + parseState: stringifyDocumentParseState({ + status: 'failed', + progress: null, + updatedAt: Date.now(), + error: input.workerState.error?.message ?? 'Worker parse failed', + }), + }); + + return NextResponse.json({ + parseStatus: 'failed', + parseProgress: null, + opId: input.workerState.opId, + error: input.workerState.error?.message ?? 'Worker parse failed', + }, { status: 202 }); + } + + let parsedJsonKey: string | null = null; + if (input.workerState.result && 'parsedObjectKey' in input.workerState.result) { + parsedJsonKey = typeof input.workerState.result.parsedObjectKey === 'string' + ? input.workerState.result.parsedObjectKey + : null; + } + + if (!parsedJsonKey && input.workerState.result && 'parsed' in input.workerState.result && input.workerState.result.parsed) { + const parsedJson = Buffer.from(JSON.stringify(input.workerState.result.parsed)); + parsedJsonKey = await putParsedDocumentBlob(input.row.id, parsedJson, input.namespace); + } + + if (!parsedJsonKey) { + return NextResponse.json({ error: 'Worker completed without parsed output' }, { status: 500 }); + } + + await writeParseRowState({ + documentId: input.row.id, + userId: input.row.userId, + parseState: stringifyDocumentParseState({ + status: 'ready', + progress: null, + updatedAt: Date.now(), + }), + parsedJsonKey, + }); + + const json = await getParsedDocumentBlobByKey(parsedJsonKey); + let parsedDoc: ParsedPdfDocument | null = null; + try { + parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument; + } catch { + parsedDoc = null; + } + + if (!hasAnyParsedBlocks(parsedDoc)) { + console.warn('[documents/parsed] parsed doc has no blocks', { + documentId: input.row.id, + userId: input.row.userId, + parsedJsonKey, + }); + } + + return new NextResponse(new Uint8Array(json), { + status: 200, + headers: { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + }, + }); +} + export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { try { if (!isS3Configured()) return s3NotConfiguredResponse(); @@ -48,6 +185,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string const params = await ctx.params; const id = (params.id || '').trim().toLowerCase(); const retryFailed = req.nextUrl.searchParams.get('retry') === '1'; + const requestedOpId = normalizeOpId(req.nextUrl.searchParams.get('opId')); if (!isValidDocumentId(id)) { return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); } @@ -57,78 +195,69 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string const storageUserId = authCtxOrRes.userId ?? unclaimedUserId; const allowedUserIds = authCtxOrRes.authEnabled ? [storageUserId, unclaimedUserId] : [unclaimedUserId]; - const rows = (await db - .select({ - id: documents.id, - userId: documents.userId, - parseState: documents.parseState, - parsedJsonKey: documents.parsedJsonKey, - }) - .from(documents) - .where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{ - id: string; - userId: string; - parseState: string | null; - parsedJsonKey: string | null; - }>; - - const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0]; + const rows = await loadRows({ documentId: id, allowedUserIds }); + const row = pickPreferredRow(rows, storageUserId); if (!row) { return NextResponse.json({ error: 'Not found' }, { status: 404 }); } + if (requestedOpId) { + const workerState = await fetchWorkerOperationState(requestedOpId); + if (workerState && workerState.opId === requestedOpId) { + return finalizeFromWorkerState({ + workerState, + row, + namespace: testNamespace, + }); + } + console.warn('[documents/parsed:get] requested opId unavailable', { + documentId: id, + userId: row.userId, + opId: requestedOpId, + }); + } + let state = parseDocumentParseState(row.parseState); state = await healStaleDocumentParseState({ documentId: id, userId: row.userId, state, }); - let effectiveStatus = normalizeParseStatus(state.status); - let effectiveProgress = state.progress ?? null; - const opId = typeof state.opId === 'string' ? state.opId.trim() : ''; - if (opId && effectiveStatus !== 'ready') { - const workerState = await fetchWorkerOperationState(opId); - if (workerState && workerState.opId === opId) { - const merged = mergeNonReadyParseSnapshot({ - parseStatus: effectiveStatus, - parseProgress: effectiveProgress, + + const effectiveStatus = normalizeParseStatus(state.status); + const effectiveProgress = state.progress ?? null; + const effectiveOpId = normalizeOpId(state.opId); + + if (effectiveOpId && effectiveStatus !== 'ready') { + const workerState = await fetchWorkerOperationState(effectiveOpId); + if (workerState && workerState.opId === effectiveOpId) { + return finalizeFromWorkerState({ workerState, - }); - effectiveStatus = merged.parseStatus; - effectiveProgress = merged.parseProgress; - } else { - console.warn('[documents/parsed:get] worker state unavailable for op', { - documentId: id, - userId: row.userId, - opId, - parseStatus: effectiveStatus, + row, + namespace: testNamespace, }); } } if (effectiveStatus === 'failed' && retryFailed) { - await db - .update(documents) - .set({ - parseState: stringifyDocumentParseState({ - status: 'pending', - progress: null, - updatedAt: Date.now(), - }), - }) - .where(and(eq(documents.id, id), eq(documents.userId, row.userId))); - enqueueParsePdfJob({ + const created = await createOrReusePdfWorkerOperation({ documentId: id, - userId: row.userId, namespace: testNamespace, + documentObjectKey: documentKey(id, testNamespace), }); - return NextResponse.json({ parseStatus: 'pending', parseProgress: null }, { status: 202 }); + const snapshot = snapshotFromWorkerState(created); + return NextResponse.json({ + parseStatus: snapshot.parseStatus, + parseProgress: snapshot.parseProgress, + opId: created.opId, + }, { status: 202 }); } if (effectiveStatus !== 'ready') { return NextResponse.json({ parseStatus: effectiveStatus, parseProgress: effectiveProgress, + opId: effectiveOpId, }, { status: 202 }); } @@ -183,27 +312,21 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); } + let replace = false; + try { + const body = (await req.json()) as { replace?: unknown }; + replace = body?.replace === true; + } catch { + replace = false; + } + const testNamespace = getOpenReaderTestNamespace(req.headers); const unclaimedUserId = getUnclaimedUserIdForNamespace(testNamespace); const storageUserId = authCtxOrRes.userId ?? unclaimedUserId; const allowedUserIds = authCtxOrRes.authEnabled ? [storageUserId, unclaimedUserId] : [unclaimedUserId]; - const rows = (await db - .select({ - id: documents.id, - userId: documents.userId, - parseState: documents.parseState, - parsedJsonKey: documents.parsedJsonKey, - }) - .from(documents) - .where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{ - id: string; - userId: string; - parseState: string | null; - parsedJsonKey: string | null; - }>; - - const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0]; + const rows = await loadRows({ documentId: id, allowedUserIds }); + const row = pickPreferredRow(rows, storageUserId); if (!row) { return NextResponse.json({ error: 'Not found' }, { status: 404 }); } @@ -214,58 +337,37 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string userId: row.userId, state, }); - let effectiveStatus = normalizeParseStatus(state.status); - let effectiveProgress = state.progress ?? null; - const opId = typeof state.opId === 'string' ? state.opId.trim() : ''; - if (opId && effectiveStatus !== 'ready') { - const workerState = await fetchWorkerOperationState(opId); - if (workerState && workerState.opId === opId) { - const merged = mergeNonReadyParseSnapshot({ - parseStatus: effectiveStatus, - parseProgress: effectiveProgress, - workerState, - }); - effectiveStatus = merged.parseStatus; - effectiveProgress = merged.parseProgress; - } else { - console.warn('[documents/parsed:post] worker state unavailable for op', { - documentId: id, - userId: row.userId, - opId, - parseStatus: effectiveStatus, - }); + + const existingOpId = normalizeOpId(state.opId); + if (existingOpId) { + const existing = await fetchWorkerOperationState(existingOpId); + if (existing && (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 (effectiveStatus !== 'running') { - await db - .update(documents) - .set({ - parseState: stringifyDocumentParseState({ - status: 'pending', - progress: null, - updatedAt: Date.now(), - }), - }) - .where(and(eq(documents.id, id), eq(documents.userId, row.userId))); - } - - enqueueParsePdfJob({ + const created = await createOrReusePdfWorkerOperation({ documentId: id, - userId: row.userId, namespace: testNamespace, + documentObjectKey: documentKey(id, testNamespace), forceToken: randomUUID(), }); - return NextResponse.json( - { - parseStatus: effectiveStatus === 'running' ? 'running' : 'pending', - parseProgress: effectiveStatus === 'running' ? effectiveProgress : null, - }, - { status: 202 }, - ); + const snapshot = snapshotFromWorkerState(created); + + return NextResponse.json({ + parseStatus: snapshot.parseStatus, + parseProgress: snapshot.parseProgress, + opId: created.opId, + }, { status: 202 }); } catch (error) { console.error('Error forcing parsed PDF refresh:', error); - return NextResponse.json({ error: 'Failed to force parsed PDF refresh' }, { status: 500 }); + return NextResponse.json({ error: 'Failed to force PDF refresh' }, { status: 500 }); } } diff --git a/src/lib/client/api/documents.ts b/src/lib/client/api/documents.ts index 5e73292..7eb4570 100644 --- a/src/lib/client/api/documents.ts +++ b/src/lib/client/api/documents.ts @@ -80,24 +80,31 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort export async function getParsedPdfDocument( id: string, - options?: { signal?: AbortSignal; retryFailed?: boolean }, + options?: { signal?: AbortSignal; retryFailed?: boolean; opId?: string }, ): Promise< | { status: 'ready'; parsed: ParsedPdfDocument } - | { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null } + | { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null; opId?: string | null } > { - const query = options?.retryFailed ? '?retry=1' : ''; + const params = new URLSearchParams(); + if (options?.retryFailed) params.set('retry', '1'); + if (options?.opId) params.set('opId', options.opId); + const query = params.size > 0 ? `?${params.toString()}` : ''; const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed${query}`, { signal: options?.signal, cache: 'no-store', }); if (res.status === 202) { - const data = (await res.json().catch(() => null)) as { parseStatus?: string; parseProgress?: PdfParseProgress | null } | null; + const data = (await res.json().catch(() => null)) as { + parseStatus?: string; + parseProgress?: PdfParseProgress | null; + opId?: string | null; + } | null; const parseStatus = data?.parseStatus; if (parseStatus === 'pending' || parseStatus === 'running' || parseStatus === 'failed') { - return { status: parseStatus, parseProgress: data?.parseProgress ?? null }; + return { status: parseStatus, parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null }; } - return { status: 'pending', parseProgress: data?.parseProgress ?? null }; + return { status: 'pending', parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null }; } if (!res.ok) { @@ -111,16 +118,30 @@ export async function getParsedPdfDocument( export function subscribeParsedPdfDocumentEvents( id: string, + options: { + opId?: string | null; + }, handlers: { - onSnapshot: (snapshot: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null }) => void; + onSnapshot: (snapshot: { + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId?: string | null; + }) => void; onError?: (error: Event) => void; }, ): () => void { - const source = new EventSource(`/api/documents/${encodeURIComponent(id)}/parsed/events`); + const params = new URLSearchParams(); + if (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 }; + const payload = JSON.parse(event.data) as { + parseStatus: PdfParseStatus; + parseProgress: PdfParseProgress | null; + opId?: string | null; + }; handlers.onSnapshot(payload); } catch { // Ignore malformed payloads to avoid breaking active streams. @@ -137,20 +158,37 @@ export function subscribeParsedPdfDocumentEvents( export async function forceReparsePdfDocument( id: string, options?: { signal?: AbortSignal }, -): Promise<'pending' | 'running'> { +): Promise<{ status: 'pending' | 'running'; opId?: string | null }> { const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, { method: 'POST', 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, + }; + } + } + if (!res.ok) { const data = (await res.json().catch(() => null)) as { error?: string } | null; throw new Error(data?.error || 'Failed to force PDF reparse'); } - const data = (await res.json().catch(() => null)) as { parseStatus?: string } | null; - return data?.parseStatus === 'running' ? 'running' : 'pending'; + const data = (await res.json().catch(() => null)) as { parseStatus?: string; opId?: string | null } | null; + return { + status: data?.parseStatus === 'running' ? 'running' : 'pending', + opId: data?.opId ?? null, + }; } type DocumentSettingsResponse = { diff --git a/src/lib/server/compute/worker-op-create.ts b/src/lib/server/compute/worker-op-create.ts new file mode 100644 index 0000000..433b68e --- /dev/null +++ b/src/lib/server/compute/worker-op-create.ts @@ -0,0 +1,60 @@ +import { getWorkerClientConfigFromEnv, buildPdfOpKey } from '@/lib/server/compute/worker'; +import type { PdfLayoutInput } from '@/lib/server/compute/types'; +import type { + PdfLayoutJobResult, + WorkerOperationState, +} from '@openreader/compute-core/api-contracts'; + +type CreatePdfWorkerOpInput = + Pick + & { documentObjectKey: string }; + +const CREATE_OP_TIMEOUT_MS = 10_000; + +export async function createOrReusePdfWorkerOperation( + input: CreatePdfWorkerOpInput, +): Promise> { + const cfg = getWorkerClientConfigFromEnv(); + const opKey = buildPdfOpKey({ + documentId: input.documentId, + namespace: input.namespace, + documentObjectKey: input.documentObjectKey, + forceToken: input.forceToken, + }); + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), CREATE_OP_TIMEOUT_MS); + try { + const res = await fetch(`${cfg.baseUrl}/ops`, { + method: 'POST', + headers: { + Authorization: `Bearer ${cfg.token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + kind: 'pdf_layout', + opKey, + payload: { + documentId: input.documentId, + namespace: input.namespace, + documentObjectKey: input.documentObjectKey, + }, + }), + cache: 'no-store', + signal: controller.signal, + }); + + if (!res.ok) { + const detail = await res.text().catch(() => ''); + throw new Error(`Worker op create failed: ${res.status}${detail ? ` ${detail}` : ''}`); + } + + const parsed = await res.json() as WorkerOperationState; + if (!parsed || typeof parsed !== 'object' || typeof parsed.opId !== 'string' || !parsed.opId.trim()) { + throw new Error('Worker op create returned invalid response'); + } + return parsed; + } finally { + clearTimeout(timeout); + } +}