From 7344fcbc51d4b6b4f0f0a124dd7c08e30b232bef Mon Sep 17 00:00:00 2001 From: Richard R Date: Thu, 4 Jun 2026 12:47:37 -0600 Subject: [PATCH] fix(pdf): prevent redundant parse jobs and improve docx-to-pdf upload reuse Avoid enqueuing duplicate parse jobs for PDFs that are already parsed or in terminal states. Update docx-to-pdf upload to reuse existing parsed PDF results when possible, skipping unnecessary parse operations and job enqueuing. Adjust parse-state-backfill to only enqueue jobs for pending or running states. Refine client upload logic to better infer document type from MIME when file names are missing. Enhance usePdfDocument to handle terminal parse states more robustly. This change ensures idempotent parse job creation and reduces unnecessary worker load, while improving document upload and parse state handling. --- src/app/(app)/pdf/[id]/usePdfDocument.ts | 31 +++++++++---- .../api/documents/docx-to-pdf/upload/route.ts | 46 ++++++++++++------- src/lib/client/api/documents.ts | 19 ++++++-- src/lib/server/compute/worker-parse-state.ts | 2 +- .../server/documents/parse-state-backfill.ts | 18 ++++---- ...arsed-route-legacy-backfill.vitest.spec.ts | 21 ++++++--- 6 files changed, 90 insertions(+), 47 deletions(-) diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index ffa2be8..c916bb1 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -213,31 +213,42 @@ export function usePdfDocument(): PdfDocumentState { setActiveParseOpId(initialOpId?.trim() || null); const controller = new AbortController(); parseStreamAbortRef.current = controller; + let isResolvingTerminalState = false; const closeSse = subscribeParsedPdfDocumentEvents(documentId, { opId: initialOpId?.trim() || null, }, { onSnapshot: (snapshot) => { if (controller.signal.aborted) return; + if (isResolvingTerminalState) return; if (typeof snapshot.opId === 'string' && snapshot.opId.trim()) { setActiveParseOpId(snapshot.opId.trim()); } setParseStatus(snapshot.parseStatus); setParseProgress(snapshot.parseProgress); if (snapshot.parseStatus === 'ready') { - closeSse(); - parseSseCloseRef.current = null; - if (parseStreamAbortRef.current === controller) { - parseStreamAbortRef.current = null; - } - void loadParsedDocumentOnce(documentId, controller.signal).catch((error) => { - if (error instanceof DOMException && error.name === 'AbortError') return; - console.error('Failed to load parsed PDF after ready status:', error); - resetParsedDocumentState(); - }); + isResolvingTerminalState = true; + void (async () => { + try { + await loadParsedDocumentOnce(documentId, controller.signal); + } catch (error) { + if (error instanceof DOMException && error.name === 'AbortError') return; + console.error('Failed to load parsed PDF after ready status:', error); + resetParsedDocumentState(); + } finally { + if (parseSseCloseRef.current === closeSse) { + closeSse(); + parseSseCloseRef.current = null; + } + if (parseStreamAbortRef.current === controller) { + parseStreamAbortRef.current = null; + } + } + })(); return; } if (snapshot.parseStatus === 'failed') { + isResolvingTerminalState = true; closeSse(); parseSseCloseRef.current = null; if (parseStreamAbortRef.current === controller) { diff --git a/src/app/api/documents/docx-to-pdf/upload/route.ts b/src/app/api/documents/docx-to-pdf/upload/route.ts index 834b4ab..51d1bf1 100644 --- a/src/app/api/documents/docx-to-pdf/upload/route.ts +++ b/src/app/api/documents/docx-to-pdf/upload/route.ts @@ -8,6 +8,7 @@ import { pathToFileURL } from 'url'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { db } from '@/db'; import { documents } from '@/db/schema'; +import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse'; import { safeDocumentName } from '@/lib/server/documents/utils'; import { enqueueDocumentPreview } from '@/lib/server/documents/previews'; import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation'; @@ -18,6 +19,7 @@ import { isS3Configured } from '@/lib/server/storage/s3'; import { putDocumentBlob } from '@/lib/server/documents/blobstore'; import { errorToLog, serverLogger } from '@/lib/server/logger'; import { errorResponse } from '@/lib/server/errors/next-response'; +import { PDF_PARSER_VERSION } from '@openreader/compute-core'; const DOCSTORE_DIR = path.join(process.cwd(), 'docstore'); const TEMP_DIR = path.join(DOCSTORE_DIR, 'tmp'); @@ -130,11 +132,19 @@ export async function POST(req: NextRequest) { const derivedName = safeDocumentName(`${path.parse(file.name).name}.pdf`, `${id}.pdf`); const lastModified = Number.isFinite(file.lastModified) ? file.lastModified : Date.now(); - const startedParse = await startPdfParseOperation({ - documentId: id, - userId: storageUserId, - namespace: testNamespace, - }); + const reusableParsedPdf = await findReusableParsedPdfResult(id); + const startedParse = reusableParsedPdf + ? null + : await startPdfParseOperation({ + documentId: id, + userId: storageUserId, + namespace: testNamespace, + }); + const parseState = stringifyDocumentParseState( + reusableParsedPdf + ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } + : startedParse!.parseState, + ); await db .insert(documents) @@ -146,8 +156,8 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, - parseState: stringifyDocumentParseState(startedParse.parseState), - parsedJsonKey: null, + parseState, + parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, }) .onConflictDoUpdate({ target: [documents.id, documents.userId], @@ -157,8 +167,8 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, - parseState: stringifyDocumentParseState(startedParse.parseState), - parsedJsonKey: null, + parseState, + parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, }, }); @@ -179,14 +189,16 @@ export async function POST(req: NextRequest) { }, 'Failed to enqueue preview for converted DOCX'); }); - enqueueParsePdfJob({ - documentId: id, - userId: storageUserId, - namespace: testNamespace, - initialOpId: startedParse.workerState.opId, - initialJobId: startedParse.workerState.jobId, - initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending', - }); + if (startedParse) { + enqueueParsePdfJob({ + documentId: id, + userId: storageUserId, + namespace: testNamespace, + initialOpId: startedParse.workerState.opId, + initialJobId: startedParse.workerState.jobId, + initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending', + }); + } return NextResponse.json({ stored: { diff --git a/src/lib/client/api/documents.ts b/src/lib/client/api/documents.ts index d6a80ce..9a70cf5 100644 --- a/src/lib/client/api/documents.ts +++ b/src/lib/client/api/documents.ts @@ -47,6 +47,14 @@ function documentTypeForName(name: string): DocumentType { return 'html'; } +function documentTypeForMime(contentType: string): DocumentType | null { + const normalized = contentType.trim().toLowerCase(); + if (normalized === 'application/pdf') return 'pdf'; + if (normalized === 'application/epub+zip') return 'epub'; + if (normalized === 'application/vnd.openxmlformats-officedocument.wordprocessingml.document') return 'docx'; + return null; +} + export function mimeTypeForDoc(doc: Pick): string { if (doc.type === 'pdf') return 'application/pdf'; if (doc.type === 'epub') return 'application/epub+zip'; @@ -314,11 +322,14 @@ export async function uploadDocuments(files: File[], options?: UploadOptions): P const sources: UploadSource[] = []; for (const file of files) { - const type = documentTypeForName(file.name); - const name = file.name || `upload.${type}`; - const contentType = file.type || mimeTypeForDoc({ name, type }); + const name = file.name || ''; + const type = name + ? documentTypeForName(name) + : (documentTypeForMime(file.type) ?? 'html'); + const resolvedName = name || `upload.${type}`; + const contentType = file.type || mimeTypeForDoc({ name: resolvedName, type }); sources.push({ - name, + name: resolvedName, type, size: file.size, lastModified: Number.isFinite(file.lastModified) ? file.lastModified : Date.now(), diff --git a/src/lib/server/compute/worker-parse-state.ts b/src/lib/server/compute/worker-parse-state.ts index 4b954b7..6798a22 100644 --- a/src/lib/server/compute/worker-parse-state.ts +++ b/src/lib/server/compute/worker-parse-state.ts @@ -68,7 +68,7 @@ export function mergeNonReadyParseSnapshot(input: { workerState: WorkerOperationState; }): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } { const workerSnapshot = snapshotFromWorkerState(input.workerState); - if (workerSnapshot.parseStatus === 'ready' || workerSnapshot.parseStatus === 'failed') { + if (workerSnapshot.parseStatus === 'ready') { return { parseStatus: input.parseStatus, parseProgress: input.parseProgress, diff --git a/src/lib/server/documents/parse-state-backfill.ts b/src/lib/server/documents/parse-state-backfill.ts index b972553..f058ba1 100644 --- a/src/lib/server/documents/parse-state-backfill.ts +++ b/src/lib/server/documents/parse-state-backfill.ts @@ -23,13 +23,15 @@ export async function backfillPendingPdfParseOperation(input: { userId: input.userId, namespace: input.namespace, }); - 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', - }); + 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/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts index 1c48cb6..75c5cd2 100644 --- a/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts +++ b/tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts @@ -14,8 +14,9 @@ const hoisted = vi.hoisted(() => ({ }, requireAuthContext: vi.fn(), fetchWorkerOperationState: vi.fn(), - backfillPendingPdfParseOperation: vi.fn(), healStaleDocumentParseState: vi.fn(async ({ state }) => state), + startPdfParseOperation: vi.fn(), + enqueueParsePdfJob: vi.fn(), })); vi.mock('@/db', () => ({ @@ -32,14 +33,18 @@ 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/pdf-parse-operation', () => ({ + startPdfParseOperation: hoisted.startPdfParseOperation, +})); + +vi.mock('@/lib/server/jobs/user-pdf-layout-job', () => ({ + enqueueParsePdfJob: hoisted.enqueueParsePdfJob, +})); + vi.mock('@/lib/server/documents/blobstore', () => ({ documentKey: vi.fn(), getParsedDocumentBlob: vi.fn(), @@ -104,8 +109,9 @@ describe('GET /api/documents/[id]/parsed pure data fetch', () => { hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' }); hoisted.fetchWorkerOperationState.mockReset(); hoisted.fetchWorkerOperationState.mockResolvedValue(null); - hoisted.backfillPendingPdfParseOperation.mockReset(); hoisted.healStaleDocumentParseState.mockClear(); + hoisted.startPdfParseOperation.mockReset(); + hoisted.enqueueParsePdfJob.mockReset(); }); test('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => { @@ -120,7 +126,8 @@ describe('GET /api/documents/[id]/parsed pure data fetch', () => { parseStatus: 'pending', opId: null, }); - expect(hoisted.backfillPendingPdfParseOperation).not.toHaveBeenCalled(); + expect(hoisted.startPdfParseOperation).not.toHaveBeenCalled(); + expect(hoisted.enqueueParsePdfJob).not.toHaveBeenCalled(); expect(hoisted.row.parseState).toBeNull(); }); });