test(api): add regression tests and backfill handler for legacy PDF parse state

Introduce a backfill handler to restore missing PDF parse operations for
documents in legacy 'pending' or 'running' states lacking an opId.
Update parsed route handlers to invoke this logic, ensuring correct
operation association for affected documents. Add regression tests to
validate backfill behavior for legacy scenarios.

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<void> {
+  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<Uint8Array>({

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);

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<WorkerOperationState<PdfLayoutJobResult> | 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;
+}

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<typeof vi.fn>;
+    update: ReturnType<typeof vi.fn>;
+  } | 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"');
+  });
+});