danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(pdf): implement granular PDF parse progress tracking and migrate to parseState

Browse source 
Introduce detailed progress reporting for PDF layout parsing, exposing phase and page-level updates to clients. Replace legacy parseStatus with a structured parseState field in the database schema, updating all relevant backend and API logic. Add SSE endpoint for real-time parse progress updates. Update client hooks and UI to reflect granular progress and improve user feedback during document parsing. Includes migration scripts and new parse-state utility module.
This commit is contained in:
Richard R 2026-05-21 10:27:16 -06:00
parent a7b955e1ca
commit 4497f610c0
26 changed files with 4226 additions and 99 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
compute/core/src/contracts.ts
View file

@ -61,6 +61,15 @@ export interface WorkerJobTiming {
computeMs?: number;
}
export type PdfLayoutProgressPhase = 'infer' | 'merge';
export interface PdfLayoutProgress {
totalPages: number;
pagesParsed: number;
currentPage?: number;
phase: PdfLayoutProgressPhase;
}
export interface WorkerJobStatusResponse<Result> {
status: WorkerJobState;
result?: Result;
@ -96,4 +105,5 @@ export interface WorkerOperationState<Result = unknown> {
result?: Result;
error?: WorkerJobErrorShape;
timing?: WorkerJobTiming;
progress?: PdfLayoutProgress;
}

6
compute/core/src/local-runtime.ts
View file

@ -25,10 +25,16 @@ export async function runWhisperAlignmentFromAudioBuffer(input: {
export async function runPdfLayoutFromPdfBuffer(input: {
documentId: string;
pdfBytes: ArrayBuffer;
onPageParsed?: (input: {
pageNumber: number;
totalPages: number;
pageMs: number;
}) => void | Promise<void>;
}) {
const parsed = await parsePdf({
documentId: input.documentId,
pdfBytes: input.pdfBytes,
onPageParsed: input.onPageParsed,
});
return { parsed };
}

14
compute/core/src/pdf-layout/parsePdf.ts
View file

@ -11,6 +11,11 @@ import { renderPage } from './renderPage';
interface ParsePdfInput {
documentId: string;
pdfBytes: ArrayBuffer;
onPageParsed?: (input: {
pageNumber: number;
totalPages: number;
pageMs: number;
}) => void | Promise<void>;
}
const LAYOUT_RENDER_SCALE = 1.5;
@ -80,6 +85,7 @@ export async function parsePdf(input: ParsePdfInput): Promise<ParsedPdfDocument>
let sawText = false;
for (let pageNumber = 1; pageNumber <= pdf.numPages; pageNumber += 1) {
const pageStartedAt = Date.now();
const page = await pdf.getPage(pageNumber);
const viewport = page.getViewport({ scale: 1.0 });
const textContent = await page.getTextContent();
@ -143,6 +149,14 @@ export async function parsePdf(input: ParsePdfInput): Promise<ParsedPdfDocument>
height: viewport.height,
blocks,
});
if (input.onPageParsed) {
await input.onPageParsed({
pageNumber,
totalPages: pdf.numPages,
pageMs: Date.now() - pageStartedAt,
});
}
}
if (!sawText) {

82
compute/worker/src/server.ts
View file

@ -37,6 +37,7 @@ import {
type WorkerOperationKind,
type WorkerOperationRequest,
type WorkerOperationState,
type PdfLayoutProgress,
} from '@openreader/compute-core/contracts';
import { GetObjectCommand, PutObjectCommand, S3Client } from '@aws-sdk/client-s3';
@ -75,6 +76,7 @@ interface StoredJobState<Result> {
result?: Result;
error?: WorkerJobErrorShape;
timing?: WorkerJobTiming;
progress?: PdfLayoutProgress;
}
interface OpIndexEntry {
@ -772,6 +774,7 @@ async function main(): Promise<void> {
const runLayout = async (
payload: PdfLayoutJobRequest,
queueWaitMs: number,
hooks?: { onProgress?: (progress: PdfLayoutProgress) => Promise<void> },
): Promise<PdfLayoutJobResult> => {
const parsed = layoutSchema.parse(payload);
@ -779,17 +782,38 @@ async function main(): Promise<void> {
const pdfBytes = await readObjectByKey(parsed.documentObjectKey);
const s3FetchMs = Date.now() - s3FetchStartedAt;
let lastTotalPages = 0;
let lastPagesParsed = 0;
const computeStartedAt = Date.now();
const result = await withTimeout(
runPdfLayoutFromPdfBuffer({
documentId: parsed.documentId,
pdfBytes,
onPageParsed: async ({ pageNumber, totalPages }) => {
lastTotalPages = totalPages;
lastPagesParsed = pageNumber;
if (!hooks?.onProgress) return;
await hooks.onProgress({
totalPages,
pagesParsed: pageNumber,
currentPage: pageNumber,
phase: 'infer',
});
},
}),
pdfTimeoutMs,
'pdf layout job',
);
const computeMs = Date.now() - computeStartedAt;
if (hooks?.onProgress && lastTotalPages > 0) {
await hooks.onProgress({
totalPages: lastTotalPages,
pagesParsed: lastPagesParsed,
currentPage: lastPagesParsed || undefined,
phase: 'merge',
});
}
const parsedObjectKey = await putParsedObject(parsed.documentId, parsed.namespace, result.parsed);
return {
parsedObjectKey,
@ -804,11 +828,16 @@ async function main(): Promise<void> {
async function processMessage<TPayload, TResult>(input: {
msg: JsMsg;
codec: JsonCodec<QueuedJob<TPayload>>;
run: (payload: TPayload, queueWaitMs: number) => Promise<TResult>;
run: (
payload: TPayload,
queueWaitMs: number,
hooks?: { onProgress?: (progress: PdfLayoutProgress) => Promise<void> },
) => Promise<TResult>;
workerLabel: string;
}): Promise<void> {
let decoded: QueuedJob<TPayload> | null = null;
let heartbeat: NodeJS.Timeout | null = null;
let latestProgress: PdfLayoutProgress | undefined;
try {
decoded = input.codec.decode(input.msg.data);
const startedAt = Date.now();
@ -824,6 +853,7 @@ async function main(): Promise<void> {
startedAt,
updatedAt: startedAt,
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
};
await putOpState(runningState);
@ -837,11 +867,11 @@ async function main(): Promise<void> {
startedAt,
updatedAt: startedAt,
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
});
heartbeat = setInterval(() => {
const now = Date.now();
const heartbeatState: WorkerOperationState<WhisperAlignJobResult | PdfLayoutJobResult> = {
const persistRunningState = async (updatedAt: number): Promise<void> => {
const runningOpState: WorkerOperationState<WhisperAlignJobResult | PdfLayoutJobResult> = {
opId: decoded!.opId,
opKey: decoded!.opKey,
kind: decoded!.kind,
@ -849,18 +879,13 @@ async function main(): Promise<void> {
status: 'running',
queuedAt: decoded!.queuedAt,
startedAt,
updatedAt: now,
updatedAt,
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
};
void putOpState(heartbeatState).catch((stateError) => {
app.log.error({
worker: input.workerLabel,
opId: decoded?.opId,
jobId: decoded?.jobId,
error: toErrorMessage(stateError),
}, 'failed to persist running heartbeat op state');
});
void putJobState({
await putOpState(runningOpState);
await putJobState({
jobId: decoded!.jobId,
opId: decoded!.opId,
opKey: decoded!.opKey,
@ -868,19 +893,30 @@ async function main(): Promise<void> {
status: 'running',
timestamp: decoded!.queuedAt,
startedAt,
updatedAt: now,
updatedAt,
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
}).catch((stateError) => {
...(latestProgress ? { progress: latestProgress } : {}),
});
};
heartbeat = setInterval(() => {
const now = Date.now();
void persistRunningState(now).catch((stateError) => {
app.log.error({
worker: input.workerLabel,
opId: decoded?.opId,
jobId: decoded?.jobId,
error: toErrorMessage(stateError),
}, 'failed to persist running heartbeat job state');
}, 'failed to persist running heartbeat state');
});
}, RUNNING_HEARTBEAT_MS);
const result = await input.run(decoded.payload, queueWaitMs ?? 0);
const result = await input.run(decoded.payload, queueWaitMs ?? 0, {
onProgress: async (progress) => {
latestProgress = progress;
await persistRunningState(Date.now());
},
});
const resultTiming = result && typeof result === 'object' && 'timing' in result
? (result as { timing?: WorkerJobTiming }).timing
: undefined;
@ -897,6 +933,7 @@ async function main(): Promise<void> {
updatedAt: now,
result: result as WhisperAlignJobResult | PdfLayoutJobResult,
...(resultTiming ? { timing: resultTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
};
await putOpState(succeededState);
@ -911,6 +948,7 @@ async function main(): Promise<void> {
updatedAt: now,
result: result as WhisperAlignJobResult | PdfLayoutJobResult,
...(resultTiming ? { timing: resultTiming } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
});
input.msg.ack();
@ -941,6 +979,7 @@ async function main(): Promise<void> {
updatedAt: now,
...(status === 'failed' ? { error: { message } } : {}),
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
};
await putOpState(opState).catch((stateError) => {
@ -962,6 +1001,7 @@ async function main(): Promise<void> {
updatedAt: now,
...(status === 'failed' ? { error: { message } } : {}),
...(typeof queueWaitMs === 'number' ? { timing: { queueWaitMs } } : {}),
...(latestProgress ? { progress: latestProgress } : {}),
}).catch((stateError) => {
app.log.error({
worker: input.workerLabel,
@ -1001,7 +1041,11 @@ async function main(): Promise<void> {
async function createWorkerLoop<TPayload, TResult>(input: {
consumer: Consumer;
codec: JsonCodec<QueuedJob<TPayload>>;
run: (payload: TPayload, queueWaitMs: number) => Promise<TResult>;
run: (
payload: TPayload,
queueWaitMs: number,
hooks?: { onProgress?: (progress: PdfLayoutProgress) => Promise<void> },
) => Promise<TResult>;
workerLabel: string;
}): Promise<void> {
while (!stopping) {

2
drizzle/postgres/0007_pdf_parse_progress.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE "documents" ADD COLUMN "parse_state" text;--> statement-breakpoint
ALTER TABLE "documents" DROP COLUMN "parse_status";

1831
drizzle/postgres/meta/0007_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

7
drizzle/postgres/meta/_journal.json
View file

@ -50,6 +50,13 @@
"when": 1779102950715,
"tag": "0006_preview-defaults-cleanup",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1779378127846,
"tag": "0007_pdf_parse_progress",
"breakpoints": true
}
]
}

2
drizzle/sqlite/0007_pdf_parse_progress.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE `documents` ADD `parse_state` text;--> statement-breakpoint
ALTER TABLE `documents` DROP COLUMN `parse_status`;

1693
drizzle/sqlite/meta/0007_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

7
drizzle/sqlite/meta/_journal.json
View file

@ -50,6 +50,13 @@
"when": 1779102950385,
"tag": "0006_preview-defaults-cleanup",
"breakpoints": true
},
{
"idx": 7,
"version": "6",
"when": 1779378125905,
"tag": "0007_pdf_parse_progress",
"breakpoints": true
}
]
}

112
src/app/(app)/pdf/[id]/page.tsx
View file

@ -42,6 +42,7 @@ export default function PDFViewerPage() {
currDocPage,
currDocPages,
parseStatus,
parseProgress,
documentSettings,
updateDocumentSettings,
parsedOverlayEnabled,
@ -240,30 +241,113 @@ export default function PDFViewerPage() {
}
const renderPdfStatusLoader = () => {
const totalPages = parseProgress?.totalPages ?? 0;
const pagesParsed = parseProgress?.pagesParsed ?? 0;
const progressPercent = totalPages > 0
? Math.max(0, Math.min(100, (pagesParsed / totalPages) * 100))
: 0;
const hasMeasuredProgress = totalPages > 0;
const isMerging = parseProgress?.phase === 'merge';
let statusText = 'Loading PDF...';
let statusSubText = 'Initializing document renderer';
if (!isLoading) {
if (parseState === 'pending') {
statusText = 'Preparing PDF layout...';
statusText = parseProgress
? `Page ${Math.max(0, parseProgress.pagesParsed)} / ${parseProgress.totalPages} parsed`
: 'Preparing PDF layout...';
statusSubText = parseProgress?.phase === 'merge'
? 'Finalizing stitched block structure'
: 'Queueing parser and preparing page extraction';
} else if (parseState === 'running') {
statusText = 'Parsing PDF layout blocks...';
statusText = parseProgress
? `Page ${Math.max(0, parseProgress.pagesParsed)} / ${parseProgress.totalPages} parsed`
: 'Parsing PDF layout blocks...';
statusSubText = parseProgress?.phase === 'merge'
? 'Merging cross-page sections'
: 'Inferring reading order and text regions';
} else if (parseState === 'failed') {
statusText = 'PDF parsing failed. Retry to continue.';
statusSubText = 'The parser could not build a usable layout map';
}
}
const stageOneComplete = !isLoading;
const stageTwoActive = parseState === 'running' && !isMerging;
const stageTwoComplete = (parseState === 'running' && isMerging) || parseState === 'ready';
const stageThreeActive = parseState === 'running' && isMerging;
const stageThreeComplete = parseState === 'ready';
return (
<div className="h-full w-full flex flex-col items-center justify-center gap-4 bg-base">
<LoadingSpinner className="w-8 h-8 text-accent" />
<p className="text-sm text-muted animate-pulse">{statusText}</p>
{!isLoading && parseState === 'failed' ? (
<button
type="button"
onClick={() => forceReparseParsedPdf()}
className="inline-flex items-center rounded-md border border-offbase bg-offbase px-2.5 py-1 text-xs text-foreground hover:text-accent transition-colors"
>
Retry Parse
</button>
) : null}
<div className="h-full w-full bg-base">
<div className="mx-auto flex h-full max-w-2xl items-center px-4 py-8">
<div className="w-full space-y-3">
<div className="rounded-lg border border-offbase bg-offbase p-4 sm:p-5">
<div className="flex items-start justify-between gap-3">
<div className="min-w-0">
<p className="text-accent font-semibold text-[11px] uppercase tracking-wide">PDF Layout Parse</p>
<p className="mt-1 text-sm font-medium text-foreground">{statusText}</p>
<p className="mt-0.5 text-xs text-muted">{statusSubText}</p>
</div>
<div className="shrink-0 inline-flex items-center gap-1.5 rounded-md border border-offbase bg-base px-2 py-1">
<LoadingSpinner className="h-3.5 w-3.5 text-accent" />
<span className="text-[10px] font-semibold uppercase tracking-wide text-muted">
{parseState === 'failed' ? 'blocked' : (isMerging ? 'merge' : 'infer')}
</span>
</div>
</div>
<div className="mt-3">
<div className="mb-1.5 flex items-center justify-between text-xs">
<span className="text-muted">Progress</span>
<span className="font-medium text-foreground">
{hasMeasuredProgress ? `${Math.round(progressPercent)}%` : 'Starting'}
</span>
</div>
<div className="w-full bg-background rounded-full overflow-hidden h-1.5">
<div
className="h-full bg-accent transition-all duration-300 ease-out"
style={{ width: `${hasMeasuredProgress ? progressPercent : 7}%` }}
/>
</div>
<div className="mt-2 flex flex-wrap items-center gap-2 text-[11px] text-muted">
<span className="font-medium">{hasMeasuredProgress ? `Page ${pagesParsed}/${totalPages}` : 'Awaiting first page'}</span>
<span>•</span>
<span>{isMerging ? 'Cross-page merge' : 'Page inference'}</span>
</div>
</div>
</div>
<div className="rounded-lg border border-offbase bg-offbase px-4 py-3">
<div className="flex flex-wrap items-center gap-2 text-[11px]">
<span className={`inline-flex items-center gap-1 rounded-full px-2 py-1 ${stageOneComplete ? 'bg-accent/15 text-foreground' : 'bg-background text-muted'}`}>
<span className={`h-1.5 w-1.5 rounded-full ${stageOneComplete ? 'bg-accent' : 'bg-muted'}`} />
Prepare
</span>
<span className={`inline-flex items-center gap-1 rounded-full px-2 py-1 ${stageTwoActive || stageTwoComplete ? 'bg-accent/15 text-foreground' : 'bg-background text-muted'}`}>
<span className={`h-1.5 w-1.5 rounded-full ${(stageTwoActive || stageTwoComplete) ? 'bg-accent' : 'bg-muted'} ${stageTwoActive ? 'animate-pulse' : ''}`} />
Infer
</span>
<span className={`inline-flex items-center gap-1 rounded-full px-2 py-1 ${stageThreeActive || stageThreeComplete ? 'bg-accent/15 text-foreground' : 'bg-background text-muted'}`}>
<span className={`h-1.5 w-1.5 rounded-full ${(stageThreeActive || stageThreeComplete) ? 'bg-accent' : 'bg-muted'} ${stageThreeActive ? 'animate-pulse' : ''}`} />
Merge
</span>
</div>
</div>
{!isLoading && parseState === 'failed' ? (
<div className="flex justify-start">
<button
type="button"
onClick={() => forceReparseParsedPdf()}
className="inline-flex items-center rounded-md border border-offbase bg-offbase px-2.5 py-1 text-xs text-foreground hover:text-accent transition-colors"
>
Retry Parse
</button>
</div>
) : null}
</div>
</div>
</div>
);
};

69
src/app/(app)/pdf/[id]/usePdfDocument.ts
View file

@ -24,6 +24,7 @@ import {
getDocumentSettings,
getParsedPdfDocument,
putDocumentSettings,
subscribeParsedPdfDocumentEvents,
} from '@/lib/client/api/documents';
import { createPdfAudiobookSourceAdapter } from '@/lib/client/audiobooks/adapters/pdf';
import { regenerateAudiobookChapter, runAudiobookGeneration } from '@/lib/client/audiobooks/pipeline';
@ -44,7 +45,7 @@ import {
type DocumentSettings,
} from '@/types/document-settings';
import { mergeDocumentSettings } from '@/lib/shared/document-settings';
import type { ParsedPdfDocument, ParsedPdfPage, PdfParseStatus } from '@/types/parsed-pdf';
import type { ParsedPdfDocument, ParsedPdfPage, PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
import type {
TTSSentenceAlignment,
@ -68,6 +69,7 @@ export interface PdfDocumentState {
pdfDocument: PDFDocumentProxy | undefined;
parsedDocument: ParsedPdfDocument | null;
parseStatus: PdfParseStatus | null;
parseProgress: PdfParseProgress | null;
documentSettings: DocumentSettings;
updateDocumentSettings: (settings: DocumentSettings) => Promise<void>;
parsedOverlayEnabled: boolean;
@ -143,6 +145,7 @@ export function usePdfDocument(): PdfDocumentState {
const [pdfDocument, setPdfDocument] = useState<PDFDocumentProxy>();
const [parsedDocument, setParsedDocument] = useState<ParsedPdfDocument | null>(null);
const [parseStatus, setParseStatus] = useState<PdfParseStatus | null>(null);
const [parseProgress, setParseProgress] = useState<PdfParseProgress | null>(null);
const [documentSettings, setDocumentSettings] = useState<DocumentSettings>(DEFAULT_DOCUMENT_SETTINGS);
const [parsedOverlayEnabled, setParsedOverlayEnabled] = useState(false);
const [isAudioCombining] = useState(false);
@ -164,6 +167,7 @@ export function usePdfDocument(): PdfDocumentState {
const docLoadSeqRef = useRef(0);
const docLoadAbortRef = useRef<AbortController | null>(null);
const parsePollAbortRef = useRef<AbortController | null>(null);
const parseSseCloseRef = useRef<(() => void) | null>(null);
const fetchParsedDocument = useCallback(async (
documentId: string,
@ -174,11 +178,11 @@ export function usePdfDocument(): PdfDocumentState {
// document backfills parse output via the parsed endpoint polling path.
const effectiveInitialStatus: PdfParseStatus = initialStatus ?? 'pending';
setParseStatus(effectiveInitialStatus);
const maxAttempts = 25;
setParseProgress(null);
const delayMs = 1200;
const retryFailed = effectiveInitialStatus === 'failed';
for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
let attempt = 0;
while (!signal.aborted) {
if (signal.aborted) return;
const result = await getParsedPdfDocument(documentId, {
signal,
@ -187,13 +191,16 @@ export function usePdfDocument(): PdfDocumentState {
if (result.status === 'ready') {
setParsedDocument(result.parsed);
setParseStatus('ready');
setParseProgress(null);
return;
}
setParseStatus(result.status);
setParseProgress(result.parseProgress ?? null);
if (result.status === 'failed') {
setParsedDocument(null);
return;
}
attempt += 1;
await new Promise((resolve) => setTimeout(resolve, delayMs));
}
}, []);
@ -211,13 +218,54 @@ export function usePdfDocument(): PdfDocumentState {
const startParsedPolling = useCallback((documentId: string, initialStatus: PdfParseStatus | null) => {
parsePollAbortRef.current?.abort();
parseSseCloseRef.current?.();
parseSseCloseRef.current = null;
setParseProgress(null);
const controller = new AbortController();
parsePollAbortRef.current = controller;
void fetchParsedDocument(documentId, initialStatus, controller.signal).finally(() => {
const closeSse = subscribeParsedPdfDocumentEvents(documentId, {
onSnapshot: (snapshot) => {
if (controller.signal.aborted) return;
setParseStatus(snapshot.parseStatus);
setParseProgress(snapshot.parseProgress);
if (snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed') {
if (snapshot.parseStatus === 'failed') {
setParsedDocument(null);
} else {
void fetchParsedDocument(documentId, 'ready', controller.signal);
}
closeSse();
parseSseCloseRef.current = null;
if (parsePollAbortRef.current === controller) {
parsePollAbortRef.current = null;
}
return;
}
},
onError: () => {
// Fall back to parsed polling if browser SSE disconnects.
if (controller.signal.aborted) return;
closeSse();
parseSseCloseRef.current = null;
void fetchParsedDocument(documentId, initialStatus, controller.signal).finally(() => {
if (parsePollAbortRef.current === controller) {
parsePollAbortRef.current = null;
}
});
},
});
parseSseCloseRef.current = closeSse;
controller.signal.addEventListener('abort', () => {
closeSse();
if (parseSseCloseRef.current === closeSse) {
parseSseCloseRef.current = null;
}
if (parsePollAbortRef.current === controller) {
parsePollAbortRef.current = null;
}
});
}, { once: true });
}, [fetchParsedDocument]);
useEffect(() => {
@ -392,6 +440,8 @@ export function usePdfDocument(): PdfDocumentState {
loadSeqRef.current += 1;
parsePollAbortRef.current?.abort();
parsePollAbortRef.current = null;
parseSseCloseRef.current?.();
parseSseCloseRef.current = null;
pageTextCacheRef.current.clear();
setPdfDocument(undefined);
setCurrDocPages(undefined);
@ -401,6 +451,7 @@ export function usePdfDocument(): PdfDocumentState {
setCurrDocData(undefined);
setParsedDocument(null);
setParseStatus(null);
setParseProgress(null);
setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);
const meta = await getDocumentMetadata(id, { signal: controller.signal });
@ -465,6 +516,7 @@ export function usePdfDocument(): PdfDocumentState {
await forceReparsePdfDocument(currDocId);
setParsedDocument(null);
setParseStatus('pending');
setParseProgress(null);
startParsedPolling(currDocId, 'pending');
} catch (error) {
console.error('Failed to force PDF reparse:', error);
@ -485,6 +537,8 @@ export function usePdfDocument(): PdfDocumentState {
docLoadAbortRef.current = null;
parsePollAbortRef.current?.abort();
parsePollAbortRef.current = null;
parseSseCloseRef.current?.();
parseSseCloseRef.current = null;
setCurrDocId(undefined);
setCurrDocName(undefined);
setCurrDocData(undefined);
@ -493,6 +547,7 @@ export function usePdfDocument(): PdfDocumentState {
setPdfDocument(undefined);
setParsedDocument(null);
setParseStatus(null);
setParseProgress(null);
setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);
pageTextCacheRef.current.clear();
stop();
@ -595,6 +650,7 @@ export function usePdfDocument(): PdfDocumentState {
currDocText,
parsedDocument,
parseStatus,
parseProgress,
documentSettings,
updateDocumentSettings,
parsedOverlayEnabled,
@ -621,6 +677,7 @@ export function usePdfDocument(): PdfDocumentState {
currDocText,
parsedDocument,
parseStatus,
parseProgress,
documentSettings,
updateDocumentSettings,
parsedOverlayEnabled,

191
src/app/api/documents/[id]/parsed/events/route.ts Normal file
View file

@ -0,0 +1,191 @@
import { NextRequest, NextResponse } from 'next/server';
import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob';
import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
import { isValidDocumentId } from '@/lib/server/documents/blobstore';
import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state';
export const dynamic = 'force-dynamic';
const SSE_POLL_INTERVAL_MS = 1200;
type ParseRow = {
id: string;
userId: string;
parseState: string | null;
};
type ParsedSnapshot = {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
};
function s3NotConfiguredResponse(): NextResponse {
return NextResponse.json(
{ error: 'Documents storage is not configured. Set S3_* environment variables.' },
{ status: 503 },
);
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function toSnapshot(row: ParseRow): ParsedSnapshot {
const state = parseDocumentParseState(row.parseState);
const parseStatus = normalizeParseStatus(state.status);
return {
parseStatus,
parseProgress: state.progress ?? null,
};
}
async function loadPreferredRow(input: {
documentId: string;
storageUserId: string;
allowedUserIds: string[];
}): Promise<ParseRow | null> {
const rows = (await db
.select({
id: documents.id,
userId: documents.userId,
parseState: documents.parseState,
})
.from(documents)
.where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[];
return rows.find((candidate) => candidate.userId === input.storageUserId) ?? rows[0] ?? null;
}
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
try {
if (!isS3Configured()) return s3NotConfiguredResponse();
const authCtxOrRes = await requireAuthContext(req);
if (authCtxOrRes instanceof Response) return authCtxOrRes;
const params = await ctx.params;
const id = (params.id || '').trim().toLowerCase();
if (!isValidDocumentId(id)) {
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
}
const testNamespace = getOpenReaderTestNamespace(req.headers);
const unclaimedUserId = getUnclaimedUserIdForNamespace(testNamespace);
const storageUserId = authCtxOrRes.userId ?? unclaimedUserId;
const allowedUserIds = authCtxOrRes.authEnabled ? [storageUserId, unclaimedUserId] : [unclaimedUserId];
const row = await loadPreferredRow({
documentId: id,
storageUserId,
allowedUserIds,
});
if (!row) {
return NextResponse.json({ error: 'Not found' }, { status: 404 });
}
const initial = toSnapshot(row);
if (initial.parseStatus === 'pending') {
enqueueParsePdfJob({
documentId: id,
userId: row.userId,
namespace: testNamespace,
});
}
const encoder = new TextEncoder();
const stream = new ReadableStream<Uint8Array>({
start(controller) {
let closed = false;
const writeSnapshot = (snapshot: ParsedSnapshot): void => {
controller.enqueue(encoder.encode(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`));
};
const run = async () => {
let current = initial;
writeSnapshot(current);
let signature = JSON.stringify(current);
while (!closed) {
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') break;
await sleep(SSE_POLL_INTERVAL_MS);
if (closed) break;
const nextRow = await loadPreferredRow({
documentId: id,
storageUserId,
allowedUserIds,
});
if (!nextRow) break;
const next = toSnapshot(nextRow);
if (next.parseStatus === 'pending') {
enqueueParsePdfJob({
documentId: id,
userId: nextRow.userId,
namespace: testNamespace,
});
}
const nextSignature = JSON.stringify(next);
if (nextSignature !== signature) {
current = next;
signature = nextSignature;
writeSnapshot(current);
}
}
};
void run()
.catch((error) => {
if (!closed) {
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
}
})
.finally(() => {
if (!closed) {
closed = true;
try {
controller.close();
} catch {
// no-op
}
}
});
req.signal.addEventListener('abort', () => {
if (closed) return;
closed = true;
try {
controller.close();
} catch {
// no-op
}
}, { once: true });
},
cancel() {
return;
},
});
return new Response(stream, {
headers: {
'Content-Type': 'text/event-stream; charset=utf-8',
'Cache-Control': 'no-cache, no-transform',
Connection: 'keep-alive',
'X-Accel-Buffering': 'no',
},
});
} catch (error) {
console.error('Error streaming parsed PDF progress:', error);
return NextResponse.json({ error: 'Failed to stream parsed PDF progress' }, { status: 500 });
}
}

68
src/app/api/documents/[id]/parsed/route.ts
View file

@ -10,6 +10,11 @@ import {
isValidDocumentId,
} from '@/lib/server/documents/blobstore';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob';
import {
normalizeParseStatus,
parseDocumentParseState,
stringifyDocumentParseState,
} 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';
@ -28,15 +33,6 @@ function hasAnyParsedBlocks(doc: ParsedPdfDocument | null): boolean {
return doc.pages.some((page) => Array.isArray(page.blocks) && page.blocks.length > 0);
}
function normalizeParseStatus(
status: string | null,
): 'pending' | 'running' | 'ready' | 'failed' {
if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') {
return status;
}
return 'pending';
}
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
try {
if (!isS3Configured()) return s3NotConfiguredResponse();
@ -60,14 +56,14 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
.select({
id: documents.id,
userId: documents.userId,
parseStatus: documents.parseStatus,
parseState: documents.parseState,
parsedJsonKey: documents.parsedJsonKey,
})
.from(documents)
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
id: string;
userId: string;
parseStatus: string | null;
parseState: string | null;
parsedJsonKey: string | null;
}>;
@ -76,31 +72,26 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
return NextResponse.json({ error: 'Not found' }, { status: 404 });
}
const effectiveStatus = normalizeParseStatus(row.parseStatus);
if (row.parseStatus !== effectiveStatus) {
await db
.update(documents)
.set({ parseStatus: 'pending' })
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
enqueueParsePdfJob({
documentId: id,
userId: row.userId,
namespace: testNamespace,
});
return NextResponse.json({ parseStatus: 'pending' }, { status: 202 });
}
const state = parseDocumentParseState(row.parseState);
const effectiveStatus = normalizeParseStatus(state.status);
if (effectiveStatus === 'failed' && retryFailed) {
await db
.update(documents)
.set({ parseStatus: 'pending' })
.set({
parseState: stringifyDocumentParseState({
status: 'pending',
progress: null,
updatedAt: Date.now(),
}),
})
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
enqueueParsePdfJob({
documentId: id,
userId: row.userId,
namespace: testNamespace,
});
return NextResponse.json({ parseStatus: 'pending' }, { status: 202 });
return NextResponse.json({ parseStatus: 'pending', parseProgress: null }, { status: 202 });
}
if (effectiveStatus !== 'ready') {
@ -111,7 +102,10 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
namespace: testNamespace,
});
}
return NextResponse.json({ parseStatus: effectiveStatus }, { status: 202 });
return NextResponse.json({
parseStatus: effectiveStatus,
parseProgress: state.progress ?? null,
}, { status: 202 });
}
try {
@ -174,14 +168,14 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
.select({
id: documents.id,
userId: documents.userId,
parseStatus: documents.parseStatus,
parseState: documents.parseState,
parsedJsonKey: documents.parsedJsonKey,
})
.from(documents)
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
id: string;
userId: string;
parseStatus: string | null;
parseState: string | null;
parsedJsonKey: string | null;
}>;
@ -190,12 +184,19 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
return NextResponse.json({ error: 'Not found' }, { status: 404 });
}
const effectiveStatus = normalizeParseStatus(row.parseStatus);
const state = parseDocumentParseState(row.parseState);
const effectiveStatus = normalizeParseStatus(state.status);
if (effectiveStatus !== 'running') {
await db
.update(documents)
.set({ parseStatus: 'pending' })
.set({
parseState: stringifyDocumentParseState({
status: 'pending',
progress: null,
updatedAt: Date.now(),
}),
})
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
}
@ -206,7 +207,10 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
});
return NextResponse.json(
{ parseStatus: effectiveStatus === 'running' ? 'running' : 'pending' },
{
parseStatus: effectiveStatus === 'running' ? 'running' : 'pending',
parseProgress: effectiveStatus === 'running' ? (state.progress ?? null) : null,
},
{ status: 202 },
);
} catch (error) {

27
src/app/api/documents/route.ts
View file

@ -11,6 +11,11 @@ import {
} from '@/lib/server/documents/previews';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob';
import { deleteDocumentBlob, headDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore';
import {
normalizeParseStatus,
parseDocumentParseState,
stringifyDocumentParseState,
} from '@/lib/server/documents/parse-state';
import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import type { BaseDocument, DocumentType } from '@/types/documents';
@ -25,16 +30,6 @@ type RegisterDocument = {
lastModified: number;
};
function normalizeParseStatus(
status: string | null,
): 'pending' | 'running' | 'ready' | 'failed' | null {
if (status === null) return null;
if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') {
return status;
}
return 'pending';
}
function s3NotConfiguredResponse(): NextResponse {
return NextResponse.json(
{ error: 'Documents storage is not configured. Set S3_* environment variables.' },
@ -140,7 +135,9 @@ export async function POST(req: NextRequest) {
size: headSize,
lastModified: doc.lastModified,
filePath: doc.id,
parseStatus: doc.type === 'pdf' ? 'pending' : null,
parseState: doc.type === 'pdf'
? stringifyDocumentParseState({ status: 'pending', progress: null, updatedAt: Date.now() })
: null,
parsedJsonKey: null,
})
.onConflictDoUpdate({
@ -151,7 +148,9 @@ export async function POST(req: NextRequest) {
size: headSize,
lastModified: doc.lastModified,
filePath: doc.id,
parseStatus: doc.type === 'pdf' ? 'pending' : null,
parseState: doc.type === 'pdf'
? stringifyDocumentParseState({ status: 'pending', progress: null, updatedAt: Date.now() })
: null,
parsedJsonKey: null,
},
});
@ -229,7 +228,7 @@ export async function GET(req: NextRequest) {
size: number;
lastModified: number;
filePath: string;
parseStatus: string | null;
parseState: string | null;
parsedJsonKey: string | null;
}>;
@ -255,7 +254,7 @@ export async function GET(req: NextRequest) {
size: Number(doc.size),
lastModified: Number(doc.lastModified),
type,
parseStatus: normalizeParseStatus(doc.parseStatus),
parseStatus: type === 'pdf' ? normalizeParseStatus(parseDocumentParseState(doc.parseState).status) : null,
parsedJsonKey: doc.parsedJsonKey,
scope: doc.userId === unclaimedUserId ? 'unclaimed' : 'user',
};

2
src/db/schema_postgres.ts
View file

@ -12,7 +12,7 @@ export const documents = pgTable('documents', {
size: bigint('size', { mode: 'number' }).notNull(),
lastModified: bigint('last_modified', { mode: 'number' }).notNull(),
filePath: text('file_path').notNull(),
parseStatus: text('parse_status'),
parseState: text('parse_state'),
parsedJsonKey: text('parsed_json_key'),
createdAt: bigint('created_at', { mode: 'number' }).default(PG_NOW_MS),
}, (table) => [

2
src/db/schema_sqlite.ts
View file

@ -12,7 +12,7 @@ export const documents = sqliteTable('documents', {
size: integer('size').notNull(),
lastModified: integer('last_modified').notNull(),
filePath: text('file_path').notNull(),
parseStatus: text('parse_status'),
parseState: text('parse_state'),
parsedJsonKey: text('parsed_json_key'),
createdAt: integer('created_at').default(SQLITE_NOW_MS),
}, (table) => [

38
src/lib/client/api/documents.ts
View file

@ -1,6 +1,6 @@
import { sha256HexFromArrayBuffer } from '@/lib/client/sha256';
import type { BaseDocument, DocumentType } from '@/types/documents';
import type { ParsedPdfDocument } from '@/types/parsed-pdf';
import type { ParsedPdfDocument, PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
import type { DocumentSettings } from '@/types/document-settings';
export type UploadSource = {
@ -81,7 +81,10 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort
export async function getParsedPdfDocument(
id: string,
options?: { signal?: AbortSignal; retryFailed?: boolean },
): Promise<{ status: 'ready'; parsed: ParsedPdfDocument } | { status: 'pending' | 'running' | 'failed' }> {
): Promise<
| { status: 'ready'; parsed: ParsedPdfDocument }
| { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null }
> {
const query = options?.retryFailed ? '?retry=1' : '';
const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed${query}`, {
signal: options?.signal,
@ -89,12 +92,12 @@ export async function getParsedPdfDocument(
});
if (res.status === 202) {
const data = (await res.json().catch(() => null)) as { parseStatus?: string } | null;
const data = (await res.json().catch(() => null)) as { parseStatus?: string; parseProgress?: PdfParseProgress | null } | null;
const parseStatus = data?.parseStatus;
if (parseStatus === 'pending' || parseStatus === 'running' || parseStatus === 'failed') {
return { status: parseStatus };
return { status: parseStatus, parseProgress: data?.parseProgress ?? null };
}
return { status: 'pending' };
return { status: 'pending', parseProgress: data?.parseProgress ?? null };
}
if (!res.ok) {
@ -106,6 +109,31 @@ export async function getParsedPdfDocument(
return { status: 'ready', parsed };
}
export function subscribeParsedPdfDocumentEvents(
id: string,
handlers: {
onSnapshot: (snapshot: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null }) => void;
onError?: (error: Event) => void;
},
): () => void {
const source = new EventSource(`/api/documents/${encodeURIComponent(id)}/parsed/events`);
source.addEventListener('snapshot', (event) => {
if (!(event instanceof MessageEvent)) return;
try {
const payload = JSON.parse(event.data) as { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null };
handlers.onSnapshot(payload);
} catch {
// Ignore malformed payloads to avoid breaking active streams.
}
});
source.addEventListener('error', (event) => {
handlers.onError?.(event);
});
return () => {
source.close();
};
}
export async function forceReparsePdfDocument(
id: string,
options?: { signal?: AbortSignal },

13
src/lib/server/compute/local.ts
View file

@ -35,6 +35,17 @@ export class LocalComputeBackend implements ComputeBackend {
if (!pdfBytes) {
throw new Error('Local compute PDF layout requires pdfBytes or (documentId + namespace)');
}
return { parsed: (await runPdfLayoutFromPdfBuffer({ documentId: input.documentId, pdfBytes })).parsed };
return {
parsed: (await runPdfLayoutFromPdfBuffer({
documentId: input.documentId,
pdfBytes,
onPageParsed: (page) => input.onProgress?.({
totalPages: page.totalPages,
pagesParsed: page.pageNumber,
currentPage: page.pageNumber,
phase: 'infer',
}),
})).parsed,
};
}
}

8
src/lib/server/compute/types.ts
View file

@ -1,4 +1,9 @@
import type { TTSAudioBuffer, TTSSentenceAlignment, ParsedPdfDocument } from '@openreader/compute-core/contracts';
import type {
TTSAudioBuffer,
TTSSentenceAlignment,
ParsedPdfDocument,
PdfLayoutProgress,
} from '@openreader/compute-core/contracts';
export type ComputeMode = 'local' | 'worker';
@ -19,6 +24,7 @@ export interface PdfLayoutInput {
namespace?: string | null;
documentObjectKey?: string;
pdfBytes?: ArrayBuffer;
onProgress?: (progress: PdfLayoutProgress) => void | Promise<void>;
}
export type PdfLayoutResult =

2
src/lib/server/compute/worker-contract.ts
View file

@ -1,6 +1,8 @@
export {
type PdfLayoutJobRequest,
type PdfLayoutJobResult,
type PdfLayoutProgress,
type PdfLayoutProgressPhase,
type PdfLayoutOperationRequest,
type WhisperAlignJobRequest,
type WhisperAlignJobResult,

10
src/lib/server/compute/worker.ts
View file

@ -375,6 +375,11 @@ export class WorkerComputeBackend implements ComputeBackend {
opKeyHash,
documentId: input.documentId,
attempt,
onSnapshot: (snapshot) => {
if (snapshot.progress) {
void input.onProgress?.(snapshot.progress);
}
},
});
if (final.status !== 'succeeded' || !final.result) {
@ -482,7 +487,9 @@ export class WorkerComputeBackend implements ComputeBackend {
private async waitForOperation<Result>(
opId: string,
context: Record<string, unknown> = {},
context: Record<string, unknown> & {
onSnapshot?: (snapshot: WorkerOperationState<Result>) => void
} = {},
): Promise<WorkerOperationState<Result>> {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), this.waitTimeoutMs);
@ -577,6 +584,7 @@ export class WorkerComputeBackend implements ComputeBackend {
eventCount += 1;
latest = snapshot;
context.onSnapshot?.(snapshot);
if (snapshot.status !== lastStatus) {
lastStatus = snapshot.status;
logWorker('info', 'sse.wait.status', {

72
src/lib/server/documents/parse-state.ts Normal file
View file

@ -0,0 +1,72 @@
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
export interface DocumentParseState {
status: PdfParseStatus;
progress?: PdfParseProgress | null;
updatedAt?: number;
error?: string | null;
opId?: string;
jobId?: string;
}
export function normalizeParseStatus(status: string | null | undefined): PdfParseStatus {
if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') {
return status;
}
return 'pending';
}
function normalizeProgress(progress: unknown): PdfParseProgress | null {
if (!progress || typeof progress !== 'object') return null;
const rec = progress as Record<string, unknown>;
const totalPages = Number(rec.totalPages ?? 0);
if (!Number.isFinite(totalPages) || totalPages <= 0) return null;
const pagesParsedRaw = Number(rec.pagesParsed ?? 0);
const pagesParsed = Math.max(0, Math.min(totalPages, Number.isFinite(pagesParsedRaw) ? pagesParsedRaw : 0));
const phase = rec.phase === 'merge' ? 'merge' : 'infer';
const currentPageRaw = Number(rec.currentPage ?? pagesParsed);
const currentPage = Number.isFinite(currentPageRaw) && currentPageRaw > 0
? Math.max(1, Math.min(totalPages, currentPageRaw))
: undefined;
return {
totalPages,
pagesParsed,
currentPage,
phase,
};
}
export function parseDocumentParseState(value: string | null): DocumentParseState {
if (!value) return { status: 'pending', progress: null };
try {
const parsed = JSON.parse(value) as Record<string, unknown>;
const status = normalizeParseStatus(typeof parsed.status === 'string' ? parsed.status : null);
const progress = normalizeProgress(parsed.progress);
const updatedAtRaw = Number(parsed.updatedAt ?? 0);
const updatedAt = Number.isFinite(updatedAtRaw) && updatedAtRaw > 0 ? updatedAtRaw : undefined;
const error = typeof parsed.error === 'string' ? parsed.error : null;
const opId = typeof parsed.opId === 'string' ? parsed.opId : undefined;
const jobId = typeof parsed.jobId === 'string' ? parsed.jobId : undefined;
return {
status,
progress,
...(typeof updatedAt === 'number' ? { updatedAt } : {}),
...(error ? { error } : {}),
...(opId ? { opId } : {}),
...(jobId ? { jobId } : {}),
};
} catch {
return { status: 'pending', progress: null };
}
}
export function stringifyDocumentParseState(state: DocumentParseState): string {
return JSON.stringify({
status: normalizeParseStatus(state.status),
progress: state.progress ?? null,
updatedAt: typeof state.updatedAt === 'number' ? state.updatedAt : Date.now(),
...(state.error ? { error: state.error } : {}),
...(state.opId ? { opId: state.opId } : {}),
...(state.jobId ? { jobId: state.jobId } : {}),
});
}

47
src/lib/server/jobs/parsePdfJob.ts
View file

@ -2,8 +2,10 @@ import { and, eq } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { documentKey, putParsedDocumentBlob } from '@/lib/server/documents/blobstore';
import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state';
import { getCompute } from '@/lib/server/compute';
import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache';
import type { PdfLayoutProgress } from '@openreader/compute-core/contracts';
interface ParsePdfJobInput {
documentId: string;
@ -23,16 +25,41 @@ export async function parsePdfJob(input: ParsePdfJobInput): Promise<void> {
running.add(key);
try {
const now = Date.now();
await db
.update(documents)
.set({ parseStatus: 'running' })
.set({
parseState: stringifyDocumentParseState({
status: 'running',
progress: null,
updatedAt: now,
}),
})
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
const compute = await getCompute();
const writeProgress = async (progress: PdfLayoutProgress): Promise<void> => {
await db
.update(documents)
.set({
parseState: stringifyDocumentParseState({
status: 'running',
progress: {
totalPages: progress.totalPages,
pagesParsed: progress.pagesParsed,
currentPage: progress.currentPage,
phase: progress.phase,
},
updatedAt: Date.now(),
}),
})
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
};
const layout = await compute.parsePdfLayout({
documentId: input.documentId,
namespace: input.namespace,
documentObjectKey: documentKey(input.documentId, input.namespace),
onProgress: writeProgress,
});
let parsedJsonKey = layout.parsedObjectKey ?? null;
@ -56,7 +83,14 @@ export async function parsePdfJob(input: ParsePdfJobInput): Promise<void> {
await db
.update(documents)
.set({ parseStatus: 'ready', parsedJsonKey })
.set({
parseState: stringifyDocumentParseState({
status: 'ready',
progress: null,
updatedAt: Date.now(),
}),
parsedJsonKey,
})
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
@ -66,7 +100,14 @@ export async function parsePdfJob(input: ParsePdfJobInput): Promise<void> {
try {
await db
.update(documents)
.set({ parseStatus })
.set({
parseState: stringifyDocumentParseState({
status: 'failed',
progress: null,
updatedAt: Date.now(),
error: message,
}),
})
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
} catch (statusError) {
console.error('[parsePdfJob] failed to write parse status', {

8
src/types/parsed-pdf.ts
View file

@ -77,3 +77,11 @@ export interface ParsedPdfDocument {
}
export type PdfParseStatus = 'pending' | 'running' | 'ready' | 'failed';
export type PdfParsePhase = 'infer' | 'merge';
export interface PdfParseProgress {
totalPages: number;
pagesParsed: number;
currentPage?: number;
phase: PdfParsePhase;
}

2
tests/unit/transfer-user-documents.spec.ts
View file

@ -20,7 +20,7 @@ test.describe('transferUserDocuments', () => {
size INTEGER NOT NULL,
last_modified INTEGER NOT NULL,
file_path TEXT NOT NULL,
parse_status TEXT,
parse_state TEXT,
parsed_json_key TEXT,
created_at INTEGER,
PRIMARY KEY (id, user_id)