diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 9660453..077b7e0 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -4,6 +4,7 @@ import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; +import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing'; @@ -76,12 +77,22 @@ async function toSnapshotState(row: ParseRow): Promise { 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 workerState = await fetchWorkerOperationState(opId); + if (workerState && workerState.opId === opId) { + return { + snapshot: snapshotFromWorkerState(workerState), + opId, + }; + } + } return { snapshot: { parseStatus, parseProgress: state.progress ?? null, }, - opId: typeof state.opId === 'string' && state.opId.trim() ? state.opId.trim() : null, + opId, }; } diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 57bc25f..5bcb2dc 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -4,6 +4,7 @@ import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; +import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { getParsedDocumentBlob, getParsedDocumentBlobByKey, @@ -20,6 +21,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, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -35,6 +37,21 @@ function hasAnyParsedBlocks(doc: ParsedPdfDocument | null): boolean { return doc.pages.some((page) => Array.isArray(page.blocks) && page.blocks.length > 0); } +function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']) { + switch (status) { + case 'queued': + return 'pending' as const; + case 'running': + return 'running' as const; + case 'succeeded': + return 'ready' as const; + case 'failed': + return 'failed' as const; + default: + return 'pending' as const; + } +} + export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) { try { if (!isS3Configured()) return s3NotConfiguredResponse(); @@ -80,7 +97,20 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string userId: row.userId, state, }); - const effectiveStatus = normalizeParseStatus(state.status); + 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 workerStatus = mapWorkerStatusToParseStatus(workerState.status); + // Keep DB/blob as source of truth for "ready"; prefer worker only for active/failed states. + if (workerStatus === 'pending' || workerStatus === 'running' || workerStatus === 'failed') { + effectiveStatus = workerStatus; + effectiveProgress = workerStatus === 'running' ? (workerState.progress ?? null) : null; + } + } + } if (effectiveStatus === 'failed' && retryFailed) { await db @@ -104,7 +134,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string if (effectiveStatus !== 'ready') { return NextResponse.json({ parseStatus: effectiveStatus, - parseProgress: state.progress ?? null, + parseProgress: effectiveProgress, }, { status: 202 }); } @@ -190,7 +220,19 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string userId: row.userId, state, }); - const effectiveStatus = normalizeParseStatus(state.status); + 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 workerStatus = mapWorkerStatusToParseStatus(workerState.status); + if (workerStatus === 'pending' || workerStatus === 'running' || workerStatus === 'failed') { + effectiveStatus = workerStatus; + effectiveProgress = workerStatus === 'running' ? (workerState.progress ?? null) : null; + } + } + } if (effectiveStatus !== 'running') { await db @@ -215,7 +257,7 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json( { parseStatus: effectiveStatus === 'running' ? 'running' : 'pending', - parseProgress: effectiveStatus === 'running' ? (state.progress ?? null) : null, + parseProgress: effectiveStatus === 'running' ? effectiveProgress : null, }, { status: 202 }, ); diff --git a/src/lib/server/compute/worker-op-state.ts b/src/lib/server/compute/worker-op-state.ts new file mode 100644 index 0000000..fcf4ce8 --- /dev/null +++ b/src/lib/server/compute/worker-op-state.ts @@ -0,0 +1,43 @@ +import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; +import type { WorkerOperationState } from '@openreader/compute-core/api-contracts'; + +const WORKER_OP_REQUEST_TIMEOUT_MS = 2_500; + +export async function fetchWorkerOperationState( + opId: string | null | undefined, +): Promise | null> { + const normalized = opId?.trim(); + if (!normalized) return null; + + let cfg: { baseUrl: string; token: string }; + try { + cfg = getWorkerClientConfigFromEnv(); + } catch { + return null; + } + + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), WORKER_OP_REQUEST_TIMEOUT_MS); + + try { + const res = await fetch(`${cfg.baseUrl}/ops/${encodeURIComponent(normalized)}`, { + method: 'GET', + headers: { + Authorization: `Bearer ${cfg.token}`, + Accept: 'application/json', + }, + cache: 'no-store', + signal: controller.signal, + }); + + if (!res.ok) return null; + const parsed = await res.json() as WorkerOperationState; + if (!parsed || typeof parsed !== 'object' || parsed.opId !== normalized) return null; + return parsed; + } catch { + return null; + } finally { + clearTimeout(timeout); + } +} +