diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 2649ae5..e1adc9d 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -8,8 +8,15 @@ import { isAbortLikeError } from '@/lib/server/compute/abort-like-error'; import { isWorkerOperationStateStale, 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 { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; +import { + 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'; +import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import { createRequestLogger, hashForLog } from '@/lib/server/logger'; import { errorResponse } from '@/lib/server/errors/next-response'; @@ -113,6 +120,17 @@ async function loadPreferredRow(input: { return rows.find((candidate) => candidate.userId === input.storageUserId) ?? 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; @@ -163,6 +181,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string const requestedOpId = typeof requestedOpIdRaw === 'string' && requestedOpIdRaw.trim() ? requestedOpIdRaw.trim() : null; + const testNamespace = getOpenReaderTestNamespace(req.headers); const storageUserId = authCtxOrRes.userId; const storageUserIdHash = hashForLog(storageUserId); @@ -178,7 +197,31 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string return NextResponse.json({ error: 'Not found' }, { status: 404 }); } - const initialState = await toSnapshotState(row, requestedOpId); + let initialState = await toSnapshotState(row, requestedOpId); + if (!requestedOpId && !initialState.opId && initialState.snapshot.parseStatus !== 'ready' && initialState.snapshot.parseStatus !== 'failed') { + const state = 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 workerCfg = getWorkerClientConfigFromEnv(); const encoder = new TextEncoder(); const stream = new ReadableStream({ diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 0a783f2..650760c 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -24,6 +24,7 @@ import { 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 { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { checkJobRate, recordJobEvent, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter'; @@ -272,6 +273,28 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } } + if (!effectiveOpId && (effectiveStatus === 'pending' || effectiveStatus === 'running')) { + const created = await backfillPendingPdfParseOperation({ + documentId: id, + userId: row.userId, + namespace: testNamespace, + state, + }); + if (created) { + const snapshot = snapshotFromWorkerState(created); + await writeParseRowState({ + documentId: row.id, + userId: row.userId, + parseState: stringifyDocumentParseState(documentParseStateFromWorkerState(created)), + }); + return NextResponse.json({ + parseStatus: snapshot.parseStatus, + parseProgress: snapshot.parseProgress, + opId: created.opId, + }, { status: 202 }); + } + } + if (effectiveStatus === 'failed' && retryFailed) { const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); const rateDecision = await checkJobRate(authCtxOrRes.userId, 'pdf_layout', rateConfig); diff --git a/src/lib/server/documents/parse-state-backfill.ts b/src/lib/server/documents/parse-state-backfill.ts new file mode 100644 index 0000000..97f9cae --- /dev/null +++ b/src/lib/server/documents/parse-state-backfill.ts @@ -0,0 +1,32 @@ +import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; +import { documentKey } from '@/lib/server/documents/blobstore'; +import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; +import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter'; +import { 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 created = await createOrReusePdfWorkerOperation({ + documentId: input.documentId, + namespace: input.namespace, + documentObjectKey: documentKey(input.documentId, input.namespace), + }); + + const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); + await recordJobEvent(input.userId, 'pdf_layout', created.opId, rateConfig); + return created; +} diff --git a/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts new file mode 100644 index 0000000..15b4bf1 --- /dev/null +++ b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts @@ -0,0 +1,147 @@ +import { beforeEach, describe, expect, test, vi } from 'vitest'; +import { NextRequest } from 'next/server'; + +const hoisted = vi.hoisted(() => ({ + db: null as { + select: ReturnType; + update: ReturnType; + } | null, + row: { + id: 'doc-1', + userId: 'user-1', + parseState: null as string | null, + parsedJsonKey: null as string | null, + }, + requireAuthContext: vi.fn(), + fetchWorkerOperationState: vi.fn(), + backfillPendingPdfParseOperation: vi.fn(), + healStaleDocumentParseState: vi.fn(async ({ state }) => state), +})); + +vi.mock('@/db', () => ({ + get db() { + return hoisted.db; + }, +})); + +vi.mock('@/lib/server/auth/auth', () => ({ + requireAuthContext: hoisted.requireAuthContext, +})); + +vi.mock('@/lib/server/compute/worker-op-state', () => ({ + fetchWorkerOperationState: hoisted.fetchWorkerOperationState, +})); + +vi.mock('@/lib/server/documents/parse-state-backfill', () => ({ + backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation, +})); + +vi.mock('@/lib/server/documents/parse-state-healing', () => ({ + healStaleDocumentParseState: hoisted.healStaleDocumentParseState, +})); + +vi.mock('@/lib/server/documents/blobstore', () => ({ + 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', () => ({ + isS3Configured: vi.fn(() => true), +})); + +vi.mock('@/lib/server/testing/test-namespace', () => ({ + getOpenReaderTestNamespace: vi.fn(() => null), +})); + +vi.mock('@/lib/server/logger', () => ({ + createRequestLogger: vi.fn(() => ({ + logger: { + warn: vi.fn(), + info: vi.fn(), + error: vi.fn(), + }, + requestId: 'req-test', + })), + hashForLog: vi.fn(() => 'user-hash'), +})); + +describe('GET /api/documents/[id]/parsed legacy backfill', () => { + 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, + }; + 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 []; + }), + })), + })), + }; + hoisted.requireAuthContext.mockReset(); + hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' }); + hoisted.fetchWorkerOperationState.mockReset(); + hoisted.fetchWorkerOperationState.mockResolvedValue(null); + hoisted.backfillPendingPdfParseOperation.mockReset(); + hoisted.healStaleDocumentParseState.mockClear(); + }); + + test('creates a worker op for legacy pending PDFs without opId', async () => { + hoisted.backfillPendingPdfParseOperation.mockResolvedValue({ + opId: 'op-legacy-1', + opKey: 'pdf_layout|v1|doc-1||doc-1|', + jobId: 'job-legacy-1', + kind: 'pdf_layout', + status: 'queued', + queuedAt: Date.now(), + progress: null, + result: undefined, + error: undefined, + updatedAt: Date.now(), + }); + + const { GET } = await import('../../src/app/api/documents/[id]/parsed/route'); + const request = new NextRequest('http://localhost/api/documents/doc-1/parsed'); + const response = await GET(request, { + params: Promise.resolve({ id: 'doc-1' }), + }); + + expect(response.status).toBe(202); + await expect(response.json()).resolves.toMatchObject({ + parseStatus: 'pending', + opId: 'op-legacy-1', + }); + 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"'); + }); +});