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

Merge pull request #101 from richardr1126/clean/job-worker

Browse source 
Simplify PDF parsing to worker-owned SSE and drop stale DB parse state
This commit is contained in:
Richard R 2026-06-04 20:51:28 -06:00 committed by GitHub
commit 2898e7e5cc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
41 changed files with 4897 additions and 2158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
compute/core/src/api-contracts/index.ts
View file

@ -17,6 +17,8 @@ export type {
export const ALIGN_QUEUE_NAME = 'whisper-align';
export const PDF_LAYOUT_QUEUE_NAME = 'pdf-layout';
export { PDF_PARSER_VERSION } from '../pdf/parser-version';
export { encodeParserVersion } from '../pdf/parser-version-key';
export interface WhisperAlignJobBase {
text: string;

1
compute/core/src/index.ts
View file

@ -17,6 +17,7 @@ export {
export { renderPage } from './pdf/render';
export { mergeTextWithRegions } from './pdf/merge';
export { PDF_PARSER_VERSION } from './pdf/parser-version';
export { encodeParserVersion } from './pdf/parser-version-key';
export { stitchCrossPageBlocks } from './pdf/stitch';
export { normalizeTextItemsForLayout } from './pdf/normalize-text';
export { mapWordsToSentenceOffsets, type WhisperWord } from './whisper/alignment-map';

9
compute/core/src/pdf/parser-version-key.ts Normal file
View file

@ -0,0 +1,9 @@
import { PDF_PARSER_VERSION } from './parser-version';
export function encodeParserVersion(
parserVersion: string,
defaultVersion = PDF_PARSER_VERSION,
): string {
const normalized = parserVersion.trim() || defaultVersion;
return encodeURIComponent(normalized);
}

39
compute/worker/src/runtime.ts
View file

@ -30,6 +30,8 @@ import {
getComputeOpStaleMs,
getAvailableCpuCores,
getOnnxThreadsPerJob,
PDF_PARSER_VERSION,
encodeParserVersion,
withIdleTimeoutAndHardCap,
withTimeout,
} from '@openreader/compute-core';
@ -124,6 +126,7 @@ interface OperationEventStreamLike {
interface OperationStateStoreLike {
getOpState(opId: string): Promise<StreamedOperationState | null>;
getOpStateRecord?(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>;
getOpIndex?(opKey: string): Promise<{ opId: string } | null>;
listOpStates?(): Promise<StreamedOperationState[]>;
}
@ -249,7 +252,7 @@ function documentParsedKey(id: string, namespace: string | null, prefix: string)
}
const ns = sanitizeNamespace(namespace);
const nsSegment = ns ? `ns/${ns}/` : '';
return `${prefix}/documents_v1/parsed_v1/${nsSegment}${id}.json`;
return `${prefix}/documents_v1/parsed_v2/${nsSegment}${id}/${encodeParserVersion(PDF_PARSER_VERSION)}.json`;
}
function buildS3Client(): S3Client {
@ -425,6 +428,10 @@ const operationCreateSchema = z.discriminatedUnion('kind', [
}),
]);
const operationLookupSchema = z.object({
opKey: z.string().trim().min(1).max(1024),
});
async function ensureJetStreamResources(
jsm: JetStreamManager,
whisperTimeoutMs: number,
@ -901,6 +908,36 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
return op;
});
app.post('/ops/lookup', async (request, reply) => {
const parsed = operationLookupSchema.safeParse(request.body);
if (!parsed.success) {
reply.code(400);
return {
error: 'Invalid request body',
issues: parsed.error.issues,
};
}
await ensureOrphanedOpRecovery();
if (typeof operationStateStore.getOpIndex !== 'function') {
reply.code(501);
return { error: 'Operation lookup by opKey is not supported by this state store' };
}
const indexEntry = await operationStateStore.getOpIndex(parsed.data.opKey);
if (!indexEntry?.opId) {
reply.code(404);
return { error: 'Operation not found' };
}
const state = await operationStateStore.getOpState(indexEntry.opId);
if (!state) {
reply.code(404);
return { error: 'Operation not found' };
}
return state;
});
app.get('/ops/:opId', async (request, reply) => {
const params = z.object({ opId: z.string().trim().min(1) }).safeParse(request.params);
if (!params.success) {

2
drizzle/postgres/0009_drop_pdf_parse.sql Normal file
View file

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

1894
drizzle/postgres/meta/0009_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

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

@ -64,6 +64,13 @@
"when": 1780162695652,
"tag": "0008_user_job_events",
"breakpoints": true
},
{
"idx": 9,
"version": "7",
"when": 1780625663880,
"tag": "0009_drop_pdf_parse",
"breakpoints": true
}
]
}

2
drizzle/sqlite/0009_drop_pdf_parse.sql Normal file
View file

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

1737
drizzle/sqlite/meta/0009_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

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

@ -64,6 +64,13 @@
"when": 1780162695101,
"tag": "0008_user_job_events",
"breakpoints": true
},
{
"idx": 9,
"version": "6",
"when": 1780625663601,
"tag": "0009_drop_pdf_parse",
"breakpoints": true
}
]
}

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

@ -19,10 +19,12 @@ import {
import type { PDFDocumentProxy } from 'pdfjs-dist';
import {
ensureParsedPdfDocumentOperation,
forceReparsePdfDocument,
getDocumentMetadata,
getDocumentSettings,
getParsedPdfDocument,
ParsedPdfNotReadyError,
putDocumentSettings,
subscribeParsedPdfDocumentEvents,
} from '@/lib/client/api/documents';
@ -117,6 +119,10 @@ export interface PdfDocumentState {
isAudioCombining: boolean;
}
function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/**
* Main PDF route hook.
*/
@ -205,18 +211,18 @@ export function usePdfDocument(): PdfDocumentState {
}
}, []);
const startParsedEventStream = useCallback((documentId: string, initialOpId?: string | null) => {
const startParsedEventStream = useCallback((documentId: string, initialOpId: string) => {
parseStreamAbortRef.current?.abort();
parseSseCloseRef.current?.();
parseSseCloseRef.current = null;
setParseProgress(null);
setActiveParseOpId(initialOpId?.trim() || null);
setActiveParseOpId(initialOpId.trim() || null);
const controller = new AbortController();
parseStreamAbortRef.current = controller;
let isResolvingTerminalState = false;
const closeSse = subscribeParsedPdfDocumentEvents(documentId, {
opId: initialOpId?.trim() || null,
opId: initialOpId.trim(),
}, {
onSnapshot: (snapshot) => {
if (controller.signal.aborted) return;
@ -229,13 +235,20 @@ export function usePdfDocument(): PdfDocumentState {
if (snapshot.parseStatus === 'ready') {
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 {
let loaded = false;
let retryMs = 500;
while (!controller.signal.aborted && !loaded) {
try {
await loadParsedDocumentOnce(documentId, controller.signal);
loaded = true;
} catch (error) {
if (error instanceof DOMException && error.name === 'AbortError') return;
console.warn('Parsed PDF reported ready before artifact was readable; retrying:', error);
await delay(retryMs);
retryMs = Math.min(retryMs * 2, 2_000);
}
}
if (loaded) {
if (parseSseCloseRef.current === closeSse) {
closeSse();
parseSseCloseRef.current = null;
@ -281,6 +294,60 @@ export function usePdfDocument(): PdfDocumentState {
}, { once: true });
}, [loadParsedDocumentOnce, resetParsedDocumentState, setActiveParseOpId]);
const resolveParsedDocumentState = useCallback(async (
documentId: string,
signal: AbortSignal,
): Promise<void> => {
try {
await loadParsedDocumentOnce(documentId, signal);
} catch (error) {
if (signal.aborted) return;
if (!(error instanceof ParsedPdfNotReadyError)) {
throw error;
}
resetParsedDocumentState();
setParseStatus(error.parseStatus);
setParseProgress(error.parseProgress);
setActiveParseOpId(error.opId);
if (error.parseStatus === 'failed') {
return;
}
let nextOpId = error.opId;
let nextStatus: PdfParseStatus = error.parseStatus;
let nextProgress = error.parseProgress;
if (!nextOpId) {
const ensured = await ensureParsedPdfDocumentOperation(documentId, { signal });
if (signal.aborted) return;
nextOpId = ensured.opId;
nextStatus = ensured.parseStatus;
nextProgress = ensured.parseProgress;
setParseStatus(nextStatus);
setParseProgress(nextProgress);
setActiveParseOpId(nextOpId);
}
if (nextStatus === 'ready') {
await loadParsedDocumentOnce(documentId, signal);
return;
}
if (nextStatus === 'failed' || !nextOpId) {
return;
}
startParsedEventStream(documentId, nextOpId);
}
}, [
loadParsedDocumentOnce,
resetParsedDocumentState,
setActiveParseOpId,
startParsedEventStream,
]);
useEffect(() => {
pdfDocumentRef.current = pdfDocument;
}, [pdfDocument]);
@ -487,12 +554,11 @@ export function usePdfDocument(): PdfDocumentState {
return false;
}
if (meta.type === 'pdf') {
const initialParseStatus = (meta.parseStatus ?? null) as PdfParseStatus | null;
setParseStatus(initialParseStatus);
setParseProgress(null);
setActiveParseOpId(null);
startParsedEventStream(id, null);
void fetchDocumentSettings(id, controller.signal);
void resolveParsedDocumentState(id, controller.signal).catch((error) => {
if (controller.signal.aborted) return;
console.error('Failed to resolve parsed PDF state:', error);
});
}
const doc = await ensureCachedDocument(meta, { signal: controller.signal });
@ -526,7 +592,7 @@ export function usePdfDocument(): PdfDocumentState {
setCurrDocText,
setPdfDocument,
fetchDocumentSettings,
startParsedEventStream,
resolveParsedDocumentState,
]);
const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise<void> => {
@ -553,7 +619,9 @@ export function usePdfDocument(): PdfDocumentState {
setParseStatus(forced.status);
setParseProgress(null);
setActiveParseOpId(forced.opId ?? null);
startParsedEventStream(currDocId, forced.opId ?? null);
if (forced.opId) {
startParsedEventStream(currDocId, forced.opId);
}
} catch (error) {
console.error('Failed to force PDF reparse:', error);
}

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

@ -4,56 +4,23 @@ import { db } from '@/db';
import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth';
import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker';
import { isAbortLikeError } from '@/lib/server/compute/abort-like-error';
import {
isWorkerOperationStateStale,
mergeNonReadyParseSnapshot,
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 {
normalizeDocumentParseStateForCurrentParserVersion,
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';
fetchPdfParseOperation,
isPdfParseOperationForDocument,
} from '@/lib/server/pdf-parse/operation';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import { createRequestLogger, hashForLog } from '@/lib/server/logger';
import { createRequestLogger } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response';
import { logDegraded, logServerError } from '@/lib/server/errors/logging';
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
import { parseSseEventId, parseSsePayload } from '@openreader/compute-core';
import { getComputeOpStaleMs } from '@openreader/compute-core';
import type { PdfLayoutJobResult, WorkerOperationEvent, WorkerOperationState } from '@openreader/compute-core/api-contracts';
export const dynamic = 'force-dynamic';
export const runtime = 'nodejs';
export const maxDuration = 300;
const SSE_KEEPALIVE_MS = 15_000;
const SSE_RESYNC_INTERVAL_MS = 30_000;
type ParseRow = {
type DocumentRow = {
id: string;
userId: string;
parseState: string | null;
};
type ParsedSnapshot = {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId?: string | null;
};
type SnapshotState = {
snapshot: ParsedSnapshot;
opId: string | null;
fromWorker: boolean;
type: string;
};
function s3NotConfiguredResponse(): NextResponse {
@ -63,120 +30,35 @@ function s3NotConfiguredResponse(): NextResponse {
);
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise<SnapshotState> {
const opStaleMs = getComputeOpStaleMs();
const state = await healStaleDocumentParseState({
documentId: row.id,
userId: row.userId,
state: normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)),
});
const parseStatus = normalizeParseStatus(state.status);
const dbOpId = typeof state.opId === 'string' && state.opId.trim() ? state.opId.trim() : null;
const requestedOpId = preferredOpId?.trim() || null;
const opId = requestedOpId ?? dbOpId;
// When a caller pins an opId, that op is the live source of truth even if
// the per-user document row currently says "ready" or has a different opId.
if (opId && (requestedOpId !== null || parseStatus !== 'ready')) {
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(opId);
if (
workerState
&& workerState.opId === opId
&& !isWorkerOperationStateStale(workerState, opStaleMs)
) {
const merged = mergeNonReadyParseSnapshot({
parseStatus,
parseProgress: state.progress ?? null,
workerState,
});
const workerSnapshot = snapshotFromWorkerState(workerState);
const fromWorker = workerSnapshot.parseStatus === 'pending' || workerSnapshot.parseStatus === 'running';
return {
snapshot: {
...merged,
opId: workerState.opId,
},
opId: workerState.opId,
fromWorker,
};
}
}
return {
snapshot: {
parseStatus,
parseProgress: state.progress ?? null,
opId,
},
opId,
fromWorker: false,
};
}
async function loadPreferredRow(input: {
async function loadOwnedDocumentRow(input: {
documentId: string;
storageUserId: string;
allowedUserIds: string[];
}): Promise<ParseRow | null> {
}): Promise<DocumentRow | null> {
const rows = (await db
.select({
id: documents.id,
userId: documents.userId,
parseState: documents.parseState,
type: documents.type,
})
.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;
.where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))
.limit(1)) as DocumentRow[];
return 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;
allowedUserIds: string[];
signature: string;
preferredOpId?: string | null;
writeSnapshot: (snapshot: ParsedSnapshot) => void;
}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; fromWorker: boolean; signature: string } | null> {
const nextRow = await loadPreferredRow({
documentId: input.documentId,
storageUserId: input.storageUserId,
allowedUserIds: input.allowedUserIds,
});
if (!nextRow) return null;
const nextState = await toSnapshotState(nextRow, input.preferredOpId);
const nextSignature = JSON.stringify(nextState.snapshot);
if (nextSignature !== input.signature) {
input.writeSnapshot(nextState.snapshot);
function workerEventsUrl(baseUrl: string, opId: string, sinceEventId: string | null): string {
const url = new URL(`${baseUrl}/ops/${encodeURIComponent(opId)}/events`);
if (sinceEventId) {
url.searchParams.set('sinceEventId', sinceEventId);
}
return {
snapshot: nextState.snapshot,
opId: nextState.opId,
fromWorker: nextState.fromWorker,
signature: nextSignature,
};
return url.toString();
}
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
const { logger, requestId } = createRequestLogger({
const { logger } = createRequestLogger({
route: '/api/documents/[id]/parsed/events',
request: req,
});
try {
if (!isS3Configured()) return s3NotConfiguredResponse();
@ -189,407 +71,65 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
if (!isValidDocumentId(id)) {
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
}
const requestedOpIdRaw = req.nextUrl.searchParams.get('opId');
const requestedOpId = typeof requestedOpIdRaw === 'string' && requestedOpIdRaw.trim()
? requestedOpIdRaw.trim()
: null;
const testNamespace = getOpenReaderTestNamespace(req.headers);
const storageUserId = authCtxOrRes.userId;
const storageUserIdHash = hashForLog(storageUserId);
const allowedUserIds = [storageUserId];
const opId = (req.nextUrl.searchParams.get('opId') || '').trim();
if (!opId) {
return NextResponse.json({ error: 'opId is required' }, { status: 400 });
}
const row = await loadPreferredRow({
const row = await loadOwnedDocumentRow({
documentId: id,
storageUserId,
allowedUserIds,
allowedUserIds: [authCtxOrRes.userId],
});
if (!row) {
return NextResponse.json({ error: 'Not found' }, { status: 404 });
if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 });
if (row.type !== 'pdf') {
return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 });
}
let initialState = await toSnapshotState(row, requestedOpId);
if (!requestedOpId && !initialState.opId && initialState.snapshot.parseStatus !== 'ready' && initialState.snapshot.parseStatus !== 'failed') {
const state = normalizeDocumentParseStateForCurrentParserVersion(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 namespace = getOpenReaderTestNamespace(req.headers);
const initialState = await fetchPdfParseOperation(opId);
if (!initialState || !isPdfParseOperationForDocument(initialState, { documentId: id, namespace })) {
return NextResponse.json({ error: 'Operation not found' }, { status: 404 });
}
const workerCfg = getWorkerClientConfigFromEnv();
const encoder = new TextEncoder();
const stream = new ReadableStream<Uint8Array>({
start(controller) {
let closed = false;
let keepaliveTimer: ReturnType<typeof setInterval> | null = null;
let resyncTimer: ReturnType<typeof setInterval> | null = null;
let workerAbort: AbortController | null = null;
const writeSnapshot = (snapshot: ParsedSnapshot): void => {
controller.enqueue(encoder.encode(`event: snapshot\ndata: ${JSON.stringify(snapshot)}\n\n`));
};
const closeStream = (): void => {
if (closed) return;
closed = true;
if (keepaliveTimer) {
clearInterval(keepaliveTimer);
keepaliveTimer = null;
}
if (resyncTimer) {
clearInterval(resyncTimer);
resyncTimer = null;
}
if (workerAbort) {
workerAbort.abort();
workerAbort = null;
}
try {
controller.close();
} catch {
// no-op
}
};
const runWorkerProxy = async () => {
let current = initialState.snapshot;
let signature = JSON.stringify(current);
let currentOpId = requestedOpId ?? initialState.opId;
let lastEventId: number | null = null;
let loggedMissingOpId = false;
const pinnedRequestedOp = requestedOpId ?? null;
const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot): boolean => {
const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed';
if (!isTerminal) return false;
if (pinnedRequestedOp && snapshot.opId !== pinnedRequestedOp) return false;
return true;
};
writeSnapshot(current);
if (shouldCloseForTerminalSnapshot(current)) {
closeStream();
return;
}
keepaliveTimer = setInterval(() => {
if (closed) return;
controller.enqueue(encoder.encode(': keepalive\n\n'));
}, SSE_KEEPALIVE_MS);
resyncTimer = setInterval(() => {
if (closed) return;
void syncFromDb({
documentId: id,
storageUserId,
allowedUserIds,
signature,
preferredOpId: requestedOpId ?? currentOpId,
writeSnapshot,
}).then((next) => {
if (closed) return;
if (!next) {
closeStream();
return;
}
current = next.snapshot;
signature = next.signature;
if (!requestedOpId && next.opId !== currentOpId) {
currentOpId = next.opId;
lastEventId = null;
if (workerAbort) {
workerAbort.abort();
workerAbort = null;
}
}
if (shouldCloseForTerminalSnapshot(current)) {
closeStream();
}
}).catch((error) => {
if (closed) return;
logDegraded(logger, {
event: 'documents.parsed.events.db_resync_failed',
msg: 'SSE DB resync failed',
step: 'db_resync',
context: {
documentId: id,
storageUserIdHash,
requestId,
},
error,
});
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
});
}, SSE_RESYNC_INTERVAL_MS);
while (!closed) {
if (!currentOpId) {
await sleep(SSE_RESYNC_INTERVAL_MS);
const next = await syncFromDb({
documentId: id,
storageUserId,
allowedUserIds,
signature,
preferredOpId: requestedOpId ?? currentOpId,
writeSnapshot,
});
if (!next) {
closeStream();
return;
}
current = next.snapshot;
signature = next.signature;
currentOpId = requestedOpId ?? next.opId;
if (!currentOpId) {
if (!loggedMissingOpId) {
loggedMissingOpId = true;
logger.warn({
event: 'documents.parsed.events.missing_opid_non_terminal',
degraded: true,
step: 'missing_opid_fallback',
documentId: id,
storageUserIdHash,
parseStatus: current.parseStatus,
requestedOpId,
}, 'Parse stream running without opId and non-terminal status');
}
} else if (loggedMissingOpId) {
loggedMissingOpId = false;
}
if (shouldCloseForTerminalSnapshot(current)) {
closeStream();
return;
}
continue;
}
try {
workerAbort = new AbortController();
const query = lastEventId && lastEventId > 0
? `?sinceEventId=${encodeURIComponent(String(lastEventId))}`
: '';
const response = await fetch(
`${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`,
{
method: 'GET',
headers: {
Authorization: `Bearer ${workerCfg.token}`,
Accept: 'text/event-stream',
...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}),
},
cache: 'no-store',
signal: workerAbort.signal,
},
);
if (closed) return;
if (!response.ok) {
const upstreamResponseBody = await response.text().catch(() => '');
logger.warn({
event: 'documents.parsed.events.worker_stream_request_failed',
degraded: true,
step: 'worker_stream_request',
documentId: id,
opId: currentOpId,
status: response.status,
upstreamResponseBody,
error: {
name: 'WorkerStreamRequestFailed',
message: `Worker stream request failed with status ${response.status}`,
},
}, 'Worker stream request failed');
await sleep(500);
continue;
}
if (!response.body) {
logger.warn({
event: 'documents.parsed.events.worker_stream_missing_body',
degraded: true,
step: 'worker_stream_body',
documentId: id,
opId: currentOpId,
error: {
name: 'WorkerStreamMissingBody',
message: 'Worker stream response missing body',
},
}, 'Worker stream response missing body');
await sleep(500);
continue;
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
let streamEnded = false;
while (!closed && !streamEnded) {
const read = await reader.read();
if (read.done) {
streamEnded = true;
break;
}
buffer += decoder.decode(read.value, { stream: true });
while (true) {
const frameEnd = buffer.indexOf('\n\n');
if (frameEnd < 0) break;
const frame = buffer.slice(0, frameEnd);
buffer = buffer.slice(frameEnd + 2);
const eventId = parseSseEventId(frame);
if (eventId && eventId > 0) {
lastEventId = eventId;
}
const payload = parseSsePayload(frame);
if (!payload) continue;
let parsed: WorkerOperationEvent<PdfLayoutJobResult> | WorkerOperationState<PdfLayoutJobResult>;
try {
parsed = JSON.parse(payload) as WorkerOperationEvent<PdfLayoutJobResult> | WorkerOperationState<PdfLayoutJobResult>;
} catch {
continue;
}
const workerSnapshot: WorkerOperationState<PdfLayoutJobResult> = (
parsed && typeof parsed === 'object' && 'snapshot' in parsed
? parsed.snapshot
: parsed as WorkerOperationState<PdfLayoutJobResult>
);
if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue;
const mergedSnapshot = mergeNonReadyParseSnapshot({
parseStatus: current.parseStatus,
parseProgress: current.parseProgress,
workerState: workerSnapshot,
});
const nextSnapshot: ParsedSnapshot = {
...mergedSnapshot,
opId: workerSnapshot.opId,
};
const nextSignature = JSON.stringify(nextSnapshot);
if (nextSignature !== signature) {
current = nextSnapshot;
signature = nextSignature;
writeSnapshot(current);
}
if (shouldCloseForTerminalSnapshot(current)) {
closeStream();
return;
}
}
}
} catch (error) {
if (closed || isAbortLikeError(error)) return;
logDegraded(logger, {
event: 'documents.parsed.events.worker_stream_read_failed',
msg: 'Worker stream read failed; reconnecting',
step: 'worker_stream_read',
context: {
documentId: id,
opId: currentOpId,
requestId,
},
error,
});
await sleep(500);
continue;
}
if (closed) return;
const next = await syncFromDb({
documentId: id,
storageUserId,
allowedUserIds,
signature,
preferredOpId: requestedOpId ?? currentOpId,
writeSnapshot,
});
if (!next) {
closeStream();
return;
}
current = next.snapshot;
signature = next.signature;
if (!requestedOpId && next.opId !== currentOpId) {
currentOpId = next.opId;
lastEventId = null;
}
if (shouldCloseForTerminalSnapshot(current)) {
closeStream();
return;
}
await sleep(250);
}
};
void runWorkerProxy()
.catch((error) => {
if (closed || isAbortLikeError(error)) return;
logServerError(logger, {
event: 'documents.parsed.events.worker_proxy_crashed',
error,
msg: 'Worker proxy crashed while streaming parse events',
context: { documentId: id },
normalize: {
code: 'DOCUMENTS_PARSED_EVENTS_WORKER_PROXY_CRASHED',
errorClass: 'upstream',
},
});
if (!closed) {
controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(error) })}\n\n`));
}
})
.finally(() => {
closeStream();
});
req.signal.addEventListener('abort', () => {
closeStream();
}, { once: true });
},
cancel() {
return;
},
});
return new Response(stream, {
const cfg = getWorkerClientConfigFromEnv();
const lastEventId = req.headers.get('last-event-id');
const sinceEventId = req.nextUrl.searchParams.get('sinceEventId') || lastEventId;
const upstream = await fetch(workerEventsUrl(cfg.baseUrl, opId, sinceEventId), {
method: 'GET',
headers: {
'Content-Type': 'text/event-stream; charset=utf-8',
'Cache-Control': 'no-cache, no-transform',
Connection: 'keep-alive',
Authorization: `Bearer ${cfg.token}`,
Accept: 'text/event-stream',
...(lastEventId ? { 'Last-Event-ID': lastEventId } : {}),
},
cache: 'no-store',
signal: req.signal,
});
if (!upstream.ok || !upstream.body) {
const detail = await upstream.text().catch(() => '');
return NextResponse.json(
{ error: detail || 'Failed to proxy parse event stream' },
{ status: upstream.status || 502 },
);
}
return new NextResponse(upstream.body, {
status: 200,
headers: {
'Content-Type': upstream.headers.get('content-type') || 'text/event-stream; charset=utf-8',
'Cache-Control': upstream.headers.get('cache-control') || 'no-cache, no-transform',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no',
},
});
} catch (error) {
return errorResponse(error, {
logger,
event: 'documents.parsed.events.route_failed',
msg: 'Parsed events route failed',
apiErrorMessage: 'Failed to stream parsed PDF progress',
normalize: { code: 'DOCUMENTS_PARSED_EVENTS_ROUTE_FAILED', errorClass: 'upstream' },
event: 'documents.parsed.events_failed',
msg: 'Failed to proxy parsed PDF events',
apiErrorMessage: 'Failed to proxy parsed PDF events',
normalize: { code: 'DOCUMENTS_PARSED_EVENTS_FAILED', errorClass: 'upstream' },
});
}
}

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

@ -4,39 +4,32 @@ import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth';
import { isValidDocumentId } from '@/lib/server/documents/blobstore';
import {
isWorkerOperationStateStale,
snapshotFromWorkerState,
} from '@/lib/server/compute/worker-parse-state';
import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state';
createOrReuseCurrentPdfParseOperation,
lookupCurrentPdfParseOperation,
} from '@/lib/server/pdf-parse/operation';
import { readCurrentParsedPdfArtifact, readParsedPdfArtifactByKey } from '@/lib/server/pdf-parse/artifact';
import {
getParsedDocumentBlob,
getParsedDocumentBlobByKey,
isMissingBlobError,
isValidDocumentId,
} from '@/lib/server/documents/blobstore';
import {
normalizeDocumentParseStateForCurrentParserVersion,
normalizeParseStatus,
parseDocumentParseState,
stringifyDocumentParseState,
} from '@/lib/server/documents/parse-state';
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-healing';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
parsedObjectKeyFromWorkerState,
pdfParseSnapshotFromWorkerState,
} from '@/lib/server/pdf-parse/snapshot';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import { createRequestLogger } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response';
import { checkJobRate, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter';
import { buildComputeRateLimitedResponse } from '@/lib/server/rate-limit/problem-response';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import { isS3Configured } from '@/lib/server/storage/s3';
import { createRequestLogger, hashForLog } from '@/lib/server/logger';
import { errorResponse } from '@/lib/server/errors/next-response';
import type { ParsedPdfDocument } from '@/types/parsed-pdf';
import { getComputeOpStaleMs } from '@openreader/compute-core';
import type { PdfLayoutJobResult } from '@openreader/compute-core/api-contracts';
import type { PdfParseSnapshot } from '@/lib/server/pdf-parse/types';
export const dynamic = 'force-dynamic';
type DocumentRow = {
id: string;
type: string;
};
function s3NotConfiguredResponse(): NextResponse {
return NextResponse.json(
{ error: 'Documents storage is not configured. Set S3_* environment variables.' },
@ -44,57 +37,33 @@ function s3NotConfiguredResponse(): NextResponse {
);
}
type ParseRow = {
id: string;
userId: string;
parseState: string | null;
parsedJsonKey: string | null;
};
function hasAnyParsedBlocks(doc: ParsedPdfDocument | null): boolean {
if (!doc || !Array.isArray(doc.pages)) return false;
return doc.pages.some((page) => Array.isArray(page.blocks) && page.blocks.length > 0);
}
function normalizeOpId(value: string | null | undefined): string | null {
const normalized = typeof value === 'string' ? value.trim() : '';
return normalized || null;
}
async function loadRows(input: {
async function loadOwnedDocumentRow(input: {
documentId: string;
allowedUserIds: string[];
}): Promise<ParseRow[]> {
return (await db
}): Promise<DocumentRow | null> {
const rows = (await db
.select({
id: documents.id,
userId: documents.userId,
parseState: documents.parseState,
parsedJsonKey: documents.parsedJsonKey,
type: documents.type,
})
.from(documents)
.where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[];
.where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))
.limit(1)) as DocumentRow[];
return rows[0] ?? null;
}
function pickPreferredRow(rows: ParseRow[], storageUserId: string): ParseRow | null {
return rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0] ?? null;
function jsonSnapshot(snapshot: PdfParseSnapshot, status = 409): NextResponse {
return NextResponse.json(snapshot, { status });
}
async function writeParseRowState(input: {
documentId: string;
userId: string;
parseState: string;
parsedJsonKey?: string | null;
}): Promise<void> {
await db
.update(documents)
.set({
parseState: input.parseState,
...(typeof input.parsedJsonKey === 'string' || input.parsedJsonKey === null
? { parsedJsonKey: input.parsedJsonKey }
: {}),
})
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
function artifactResponse(bytes: Buffer): NextResponse {
return new NextResponse(new Uint8Array(bytes), {
status: 200,
headers: {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
},
});
}
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
@ -102,6 +71,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
route: '/api/documents/[id]/parsed',
request: req,
});
try {
if (!isS3Configured()) return s3NotConfiguredResponse();
@ -115,62 +85,48 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
}
const testNamespace = getOpenReaderTestNamespace(req.headers);
const storageUserId = authCtxOrRes.userId;
const allowedUserIds = [storageUserId];
const rows = await loadRows({ documentId: id, allowedUserIds });
const row = pickPreferredRow(rows, storageUserId);
if (!row) {
return NextResponse.json({ error: 'Not found' }, { status: 404 });
const row = await loadOwnedDocumentRow({
documentId: id,
allowedUserIds: [authCtxOrRes.userId],
});
if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 });
if (row.type !== 'pdf') {
return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 });
}
const state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState));
const effectiveStatus = normalizeParseStatus(state.status);
const effectiveProgress = state.progress ?? null;
const effectiveOpId = normalizeOpId(state.opId);
if (effectiveStatus !== 'ready') {
return NextResponse.json({
parseStatus: effectiveStatus,
parseProgress: effectiveProgress,
opId: effectiveOpId,
}, { status: 409 });
const namespace = getOpenReaderTestNamespace(req.headers);
const artifact = await readCurrentParsedPdfArtifact({ documentId: id, namespace });
if (artifact) {
return artifactResponse(artifact.bytes);
}
try {
const json = row.parsedJsonKey?.trim()
? await getParsedDocumentBlobByKey(row.parsedJsonKey)
: await getParsedDocumentBlob(id, testNamespace);
let parsedDoc: ParsedPdfDocument | null = null;
try {
parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument;
} catch {
parsedDoc = null;
}
if (!hasAnyParsedBlocks(parsedDoc)) {
logger.warn({
event: 'documents.parsed.no_blocks_from_blob',
documentId: id,
userIdHash: hashForLog(row.userId),
parsedJsonKey: row.parsedJsonKey,
}, 'Parsed document blob contained no blocks');
}
return new NextResponse(new Uint8Array(json), {
status: 200,
headers: {
'Content-Type': 'application/json',
'Cache-Control': 'no-store',
},
const currentOp = await lookupCurrentPdfParseOperation({ documentId: id, namespace });
if (!currentOp) {
return jsonSnapshot({
parseStatus: 'pending',
parseProgress: null,
opId: null,
});
} catch (error) {
if (isMissingBlobError(error)) {
return NextResponse.json({ parseStatus: 'failed', error: 'Parsed document not found' }, { status: 404 });
}
throw error;
}
if (currentOp.status === 'succeeded') {
const artifactKey = parsedObjectKeyFromWorkerState(currentOp);
if (artifactKey) {
const artifactFromOp = await readParsedPdfArtifactByKey(artifactKey);
if (artifactFromOp) {
return artifactResponse(artifactFromOp.bytes);
}
}
return NextResponse.json(
{
error: 'Current parse operation succeeded without a readable parsed artifact.',
opId: currentOp.opId,
},
{ status: 502 },
);
}
return jsonSnapshot(pdfParseSnapshotFromWorkerState(currentOp));
} catch (error) {
return errorResponse(error, {
logger,
@ -187,8 +143,8 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
route: '/api/documents/[id]/parsed',
request: req,
});
try {
const opStaleMs = getComputeOpStaleMs();
if (!isS3Configured()) return s3NotConfiguredResponse();
const authCtxOrRes = await requireAuthContext(req);
@ -209,39 +165,45 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
replace = false;
}
const testNamespace = getOpenReaderTestNamespace(req.headers);
const storageUserId = authCtxOrRes.userId;
const allowedUserIds = [storageUserId];
const rows = await loadRows({ documentId: id, allowedUserIds });
const row = pickPreferredRow(rows, storageUserId);
if (!row) {
return NextResponse.json({ error: 'Not found' }, { status: 404 });
const row = await loadOwnedDocumentRow({
documentId: id,
allowedUserIds: [authCtxOrRes.userId],
});
if (!row) return NextResponse.json({ error: 'Not found' }, { status: 404 });
if (row.type !== 'pdf') {
return NextResponse.json({ error: 'Document is not a PDF' }, { status: 400 });
}
let state = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState));
state = await healStaleDocumentParseState({
documentId: id,
userId: row.userId,
state,
});
const namespace = getOpenReaderTestNamespace(req.headers);
const existingOpId = normalizeOpId(state.opId);
if (existingOpId) {
const existing = await fetchWorkerOperationState<PdfLayoutJobResult>(existingOpId);
if (
existing
&& !isWorkerOperationStateStale(existing, opStaleMs)
&& (existing.status === 'queued' || existing.status === 'running')
&& !replace
) {
const snapshot = snapshotFromWorkerState(existing);
return NextResponse.json({
error: 'Parse operation already in progress',
parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress,
opId: existing.opId,
}, { status: 409 });
if (!replace) {
const artifact = await readCurrentParsedPdfArtifact({ documentId: id, namespace });
if (artifact) {
return jsonSnapshot({
parseStatus: 'ready',
parseProgress: null,
opId: null,
}, 200);
}
const currentOp = await lookupCurrentPdfParseOperation({ documentId: id, namespace });
if (currentOp) {
const snapshot = pdfParseSnapshotFromWorkerState(currentOp);
if (snapshot.parseStatus === 'failed') {
return jsonSnapshot(snapshot);
}
if (snapshot.parseStatus === 'ready') {
const artifactKey = parsedObjectKeyFromWorkerState(currentOp);
if (artifactKey && await readParsedPdfArtifactByKey(artifactKey)) {
return jsonSnapshot(snapshot, 200);
}
return jsonSnapshot({
parseStatus: 'running',
parseProgress: null,
opId: snapshot.opId,
}, 202);
}
return jsonSnapshot(snapshot, 202);
}
}
@ -251,41 +213,20 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
return buildComputeRateLimitedResponse({ decision: rateDecision, pathname: req.nextUrl.pathname });
}
const forceToken = randomUUID();
const startedParse = await startPdfParseOperation({
const workerState = await createOrReuseCurrentPdfParseOperation({
documentId: id,
userId: authCtxOrRes.userId,
namespace: testNamespace,
forceToken,
});
const snapshot = snapshotFromWorkerState(startedParse.workerState);
await writeParseRowState({
documentId: row.id,
userId: row.userId,
parseState: stringifyDocumentParseState(startedParse.parseState),
});
enqueueParsePdfJob({
documentId: id,
userId: row.userId,
namespace: testNamespace,
forceToken,
initialOpId: startedParse.workerState.opId,
initialJobId: startedParse.workerState.jobId,
initialStatus: startedParse.parseState.status === 'running' ? 'running' : 'pending',
namespace,
...(replace ? { forceToken: randomUUID() } : {}),
});
return NextResponse.json({
parseStatus: snapshot.parseStatus,
parseProgress: snapshot.parseProgress,
opId: startedParse.workerState.opId,
}, { status: 202 });
return jsonSnapshot(pdfParseSnapshotFromWorkerState(workerState), 202);
} catch (error) {
return errorResponse(error, {
logger,
event: 'documents.parsed.force_refresh_failed',
msg: 'Failed to force PDF refresh',
apiErrorMessage: 'Failed to force PDF refresh',
normalize: { code: 'DOCUMENTS_PARSED_FORCE_REFRESH_FAILED', errorClass: 'upstream' },
event: 'documents.parsed.ensure_failed',
msg: 'Failed to ensure parsed PDF operation',
apiErrorMessage: 'Failed to ensure parsed PDF operation',
normalize: { code: 'DOCUMENTS_PARSED_ENSURE_FAILED', errorClass: 'upstream' },
});
}
}

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

@ -11,11 +11,6 @@ import {
deleteDocumentPreviewRows,
} from '@/lib/server/documents/previews';
import { deleteDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore';
import {
normalizeDocumentParseStateForCurrentParserVersion,
normalizeParseStatus,
parseDocumentParseState,
} from '@/lib/server/documents/parse-state';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import type { BaseDocument, DocumentType } from '@/types/documents';
@ -72,23 +67,16 @@ export async function GET(req: NextRequest) {
size: number;
lastModified: number;
filePath: string;
parseState: string | null;
parsedJsonKey: string | null;
}>;
const results: BaseDocument[] = rows.map((doc) => {
const type = normalizeDocumentType(doc.type, doc.name);
const parseState = type === 'pdf'
? normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(doc.parseState))
: null;
return {
id: doc.id,
name: doc.name,
size: Number(doc.size),
lastModified: Number(doc.lastModified),
type,
parseStatus: type === 'pdf' && parseState ? normalizeParseStatus(parseState.status) : null,
parsedJsonKey: doc.parsedJsonKey,
scope: 'user',
};
});

14
src/components/doclist/views/GalleryView.tsx
View file

@ -28,14 +28,6 @@ function formatDateTime(value: number | undefined): string {
});
}
function formatParseStatus(status: DocumentListDocument['parseStatus']): string {
if (!status) return 'N/A';
if (status === 'pending') return 'Pending';
if (status === 'running') return 'Running';
if (status === 'ready') return 'Ready';
return 'Failed';
}
function KindIcon({ doc, className }: { doc: DocumentListDocument; className?: string }) {
if (doc.type === 'pdf') return <PDFIcon className={className ?? 'w-4 h-4 text-danger'} />;
if (doc.type === 'epub') return <EPUBIcon className={className ?? 'w-4 h-4 text-accent'} />;
@ -244,12 +236,6 @@ export function GalleryView({
</dd>
</>
)}
{activeDoc.type === 'pdf' && (
<>
<dt className="text-soft">Parse status</dt>
<dd className="text-foreground text-right">{formatParseStatus(activeDoc.parseStatus)}</dd>
</>
)}
</dl>
</div>
) : (

2
src/db/schema_postgres.ts
View file

@ -12,8 +12,6 @@ export const documents = pgTable('documents', {
size: bigint('size', { mode: 'number' }).notNull(),
lastModified: bigint('last_modified', { mode: 'number' }).notNull(),
filePath: text('file_path').notNull(),
parseState: text('parse_state'),
parsedJsonKey: text('parsed_json_key'),
createdAt: bigint('created_at', { mode: 'number' }).default(PG_NOW_MS),
}, (table) => [
primaryKey({ columns: [table.id, table.userId] }),

2
src/db/schema_sqlite.ts
View file

@ -12,8 +12,6 @@ export const documents = sqliteTable('documents', {
size: integer('size').notNull(),
lastModified: integer('last_modified').notNull(),
filePath: text('file_path').notNull(),
parseState: text('parse_state'),
parsedJsonKey: text('parsed_json_key'),
createdAt: integer('created_at').default(SQLITE_NOW_MS),
}, (table) => [
primaryKey({ columns: [table.id, table.userId] }),

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

@ -88,6 +88,27 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort
return docs[0] ?? null;
}
export class ParsedPdfNotReadyError extends Error {
readonly parseStatus: PdfParseStatus;
readonly parseProgress: PdfParseProgress | null;
readonly opId: string | null;
readonly details: string | null;
constructor(input: {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId?: string | null;
details?: string | null;
}) {
super(`Parsed PDF is not ready (${input.parseStatus})`);
this.name = 'ParsedPdfNotReadyError';
this.parseStatus = input.parseStatus;
this.parseProgress = input.parseProgress;
this.opId = input.opId?.trim() || null;
this.details = input.details?.trim() || null;
}
}
export async function getParsedPdfDocument(
id: string,
options?: { signal?: AbortSignal },
@ -102,8 +123,20 @@ export async function getParsedPdfDocument(
parseStatus?: string;
parseProgress?: PdfParseProgress | null;
opId?: string | null;
error?: string;
} | null;
throw new Error(data?.parseStatus ? `Parsed PDF is not ready (${data.parseStatus})` : 'Parsed PDF is not ready');
throw new ParsedPdfNotReadyError({
parseStatus: data?.parseStatus === 'running'
? 'running'
: data?.parseStatus === 'ready'
? 'ready'
: data?.parseStatus === 'failed'
? 'failed'
: 'pending',
parseProgress: data?.parseProgress ?? null,
opId: data?.opId ?? null,
details: data?.error ?? null,
});
}
if (!res.ok) {
@ -117,30 +150,49 @@ export async function getParsedPdfDocument(
export function subscribeParsedPdfDocumentEvents(
id: string,
options: {
opId?: string | null;
opId: string;
},
handlers: {
onSnapshot: (snapshot: {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId?: string | null;
error?: string | null;
}) => void;
onError?: (error: Event) => void;
},
): () => void {
const params = new URLSearchParams();
if (options.opId) params.set('opId', options.opId);
params.set('opId', options.opId);
const query = params.size > 0 ? `?${params.toString()}` : '';
const source = new EventSource(`/api/documents/${encodeURIComponent(id)}/parsed/events${query}`);
source.addEventListener('snapshot', (event) => {
if (!(event instanceof MessageEvent)) return;
try {
const payload = JSON.parse(event.data) as {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId?: string | null;
snapshot?: {
opId: string;
status: 'queued' | 'running' | 'succeeded' | 'failed';
progress?: PdfParseProgress | null;
error?: { message?: string } | null;
};
};
handlers.onSnapshot(payload);
const snapshot = payload?.snapshot;
if (!snapshot?.opId || !snapshot.status) return;
handlers.onSnapshot({
parseStatus: snapshot.status === 'running'
? 'running'
: snapshot.status === 'succeeded'
? 'ready'
: snapshot.status === 'failed'
? 'failed'
: 'pending',
parseProgress: snapshot.status === 'running' ? (snapshot.progress ?? null) : null,
opId: snapshot.opId,
...(snapshot.status === 'failed' && snapshot.error?.message
? { error: snapshot.error.message }
: {}),
});
} catch {
// Ignore malformed payloads to avoid breaking active streams.
}
@ -153,39 +205,70 @@ export function subscribeParsedPdfDocumentEvents(
};
}
export async function forceReparsePdfDocument(
function normalizeParsedPdfOperationResponse(
data: { parseStatus?: string; parseProgress?: PdfParseProgress | null; opId?: string | null; error?: string } | null,
): {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId: string | null;
error?: string | null;
} {
return {
parseStatus: data?.parseStatus === 'running'
? 'running'
: data?.parseStatus === 'ready'
? 'ready'
: data?.parseStatus === 'failed'
? 'failed'
: 'pending',
parseProgress: data?.parseProgress ?? null,
opId: data?.opId?.trim() || null,
...(data?.error ? { error: data.error } : {}),
};
}
export async function ensureParsedPdfDocumentOperation(
id: string,
options?: { signal?: AbortSignal },
): Promise<{ status: 'pending' | 'running'; opId?: string | null }> {
options?: { signal?: AbortSignal; replace?: boolean },
): Promise<{
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId: string | null;
error?: string | null;
}> {
const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ replace: options?.replace === true }),
signal: options?.signal,
cache: 'no-store',
});
if (res.status === 409) {
const data = (await res.json().catch(() => null)) as {
parseStatus?: string;
opId?: string | null;
error?: string;
} | null;
if (typeof data?.opId === 'string' && data.opId.trim()) {
return {
status: data?.parseStatus === 'running' ? 'running' : 'pending',
opId: data.opId,
};
}
const data = (await res.json().catch(() => null)) as {
parseStatus?: string;
parseProgress?: PdfParseProgress | null;
opId?: string | null;
error?: string;
} | null;
if (!res.ok && res.status !== 409) {
throw new Error(data?.error || 'Failed to ensure parsed PDF operation');
}
if (!res.ok) {
const data = (await res.json().catch(() => null)) as { error?: string } | null;
throw new Error(data?.error || 'Failed to force PDF reparse');
}
return normalizeParsedPdfOperationResponse(data);
}
const data = (await res.json().catch(() => null)) as { parseStatus?: string; opId?: string | null } | null;
export async function forceReparsePdfDocument(
id: string,
options?: { signal?: AbortSignal },
): Promise<{ status: 'pending' | 'running'; opId?: string | null }> {
const data = await ensureParsedPdfDocumentOperation(id, {
signal: options?.signal,
replace: true,
});
return {
status: data?.parseStatus === 'running' ? 'running' : 'pending',
opId: data?.opId ?? null,
status: data.parseStatus === 'running' ? 'running' : 'pending',
opId: data.opId,
};
}

96
src/lib/server/compute/worker-op-state.ts
View file

@ -81,3 +81,99 @@ export async function fetchWorkerOperationState<Result>(
clearTimeout(timeout);
}
}
export async function fetchWorkerOperationStateByKey<Result>(
opKey: string | null | undefined,
): Promise<WorkerOperationState<Result> | null> {
const normalized = opKey?.trim();
if (!normalized) return null;
let cfg: { baseUrl: string; token: string };
try {
cfg = getWorkerClientConfigFromEnv();
} catch (error) {
logDegraded(serverLogger, {
event: 'compute.worker_op_lookup.config.invalid',
msg: 'Worker client env missing/invalid',
step: 'read_worker_config',
context: { opKey: normalized },
error,
});
return null;
}
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), WORKER_OP_REQUEST_TIMEOUT_MS);
try {
const res = await fetch(`${cfg.baseUrl}/ops/lookup`, {
method: 'POST',
headers: {
Authorization: `Bearer ${cfg.token}`,
Accept: 'application/json',
'Content-Type': 'application/json',
},
body: JSON.stringify({ opKey: normalized }),
cache: 'no-store',
signal: controller.signal,
});
if (res.status === 404) {
return null;
}
if (!res.ok) {
const upstreamResponseBody = await res.text().catch(() => '');
logDegraded(serverLogger, {
event: 'compute.worker_op_lookup.fetch.failed',
msg: 'Worker op lookup failed',
step: 'fetch_worker_op_by_key',
context: {
opKey: normalized,
status: res.status,
upstreamResponseBody,
},
error: {
name: 'WorkerOpLookupFailed',
message: `Worker op lookup failed with status ${res.status}`,
},
});
return null;
}
const parsed = await res.json() as WorkerOperationState<Result>;
if (!parsed || typeof parsed !== 'object' || typeof parsed.opId !== 'string' || !parsed.opId.trim()) {
logDegraded(serverLogger, {
event: 'compute.worker_op_lookup.response.invalid',
msg: 'Worker op lookup response invalid',
step: 'validate_worker_op_lookup_response',
context: { opKey: normalized },
});
return null;
}
if (typeof parsed.opKey !== 'string' || parsed.opKey.trim() !== normalized) {
logDegraded(serverLogger, {
event: 'compute.worker_op_lookup.response.mismatch',
msg: 'Worker op lookup response did not match requested op key',
step: 'validate_worker_op_lookup_response',
context: {
opKey: normalized,
responseOpId: parsed.opId,
responseOpKey: typeof parsed.opKey === 'string' ? parsed.opKey : null,
},
});
return null;
}
return parsed;
} catch (error) {
logDegraded(serverLogger, {
event: 'compute.worker_op_lookup.fetch.error',
msg: 'Worker op lookup threw',
step: 'fetch_worker_op_by_key',
context: { opKey: normalized },
error,
});
return null;
} finally {
clearTimeout(timeout);
}
}

78
src/lib/server/compute/worker-parse-state.ts
View file

@ -1,78 +0,0 @@
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
import type { DocumentParseState } from '@/lib/server/documents/parse-state';
function isInflightWorkerStatus(status: WorkerOperationState['status']): boolean {
return status === 'queued' || status === 'running';
}
export function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus {
switch (status) {
case 'queued':
return 'pending';
case 'running':
return 'running';
case 'succeeded':
return 'ready';
case 'failed':
return 'failed';
default:
return 'pending';
}
}
export function snapshotFromWorkerState(
state: WorkerOperationState<PdfLayoutJobResult>,
): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } {
const parseStatus = mapWorkerStatusToParseStatus(state.status);
return {
parseStatus,
parseProgress: parseStatus === 'running' ? (state.progress ?? null) : null,
};
}
export function documentParseStateFromWorkerState(
state: WorkerOperationState<PdfLayoutJobResult>,
nowMs = Date.now(),
): DocumentParseState {
const { parseStatus, parseProgress } = snapshotFromWorkerState(state);
return {
status: parseStatus,
progress: parseStatus === 'pending' || parseStatus === 'running'
? parseProgress
: null,
updatedAt: nowMs,
parserVersion: PDF_PARSER_VERSION,
...(typeof state.opId === 'string' && state.opId.trim() ? { opId: state.opId } : {}),
...(typeof state.jobId === 'string' && state.jobId.trim() ? { jobId: state.jobId } : {}),
...(parseStatus === 'failed' && state.error?.message ? { error: state.error.message } : {}),
};
}
export function isWorkerOperationStateStale(
state: WorkerOperationState<PdfLayoutJobResult>,
staleMs: number,
nowMs = Date.now(),
): boolean {
if (!isInflightWorkerStatus(state.status)) return false;
if (!Number.isFinite(staleMs) || staleMs <= 0) return false;
const updatedAt = Number(state.updatedAt ?? 0);
if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false;
return (nowMs - updatedAt) > staleMs;
}
export function mergeNonReadyParseSnapshot(input: {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
workerState: WorkerOperationState<PdfLayoutJobResult>;
}): { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null } {
const workerSnapshot = snapshotFromWorkerState(input.workerState);
if (workerSnapshot.parseStatus === 'ready') {
return {
parseStatus: input.parseStatus,
parseProgress: input.parseProgress,
};
}
return workerSnapshot;
}

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

@ -1,7 +1,8 @@
import { createHash, randomUUID } from 'node:crypto';
import type { ComputeBackend, PdfLayoutInput, WhisperAlignInput, WhisperAlignResult } from '@/lib/server/compute/types';
import { parseSseEventId, parseSsePayload } from '@openreader/compute-core';
import { getWorkerClientWaitTimeoutMs, PDF_PARSER_VERSION } from '@openreader/compute-core';
import { getWorkerClientWaitTimeoutMs } from '@openreader/compute-core';
import { PDF_PARSER_VERSION } from '@openreader/compute-core/api-contracts';
import { errorToLog, serverLogger } from '@/lib/server/logger';
import { logDegraded, logServerError } from '@/lib/server/errors/logging';
import type {

26
src/lib/server/documents/blobstore.ts
View file

@ -8,6 +8,7 @@ import {
PutObjectCommand,
} from '@aws-sdk/client-s3';
import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
import { PDF_PARSER_VERSION, encodeParserVersion } from '@openreader/compute-core/api-contracts';
import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage/s3';
const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i;
@ -102,13 +103,21 @@ export function documentKey(id: string, namespace: string | null): string {
}
export function documentParsedKey(id: string, namespace: string | null): string {
return documentParsedKeyForVersion(id, namespace, PDF_PARSER_VERSION);
}
export function documentParsedKeyForVersion(
id: string,
namespace: string | null,
parserVersion: string,
): string {
if (!isValidDocumentId(id)) {
throw new Error(`Invalid document id: ${id}`);
}
const cfg = getS3Config();
const ns = sanitizeNamespace(namespace);
const nsSegment = ns ? `ns/${ns}/` : '';
return `${cfg.prefix}/documents_v1/parsed_v1/${nsSegment}${id}.json`;
return `${cfg.prefix}/documents_v1/parsed_v2/${nsSegment}${id}/${encodeParserVersion(parserVersion)}.json`;
}
export function tempDocumentUploadPrefix(userId: string, namespace: string | null): string {
@ -359,9 +368,18 @@ export async function getParsedDocumentBlobByKey(key: string): Promise<Buffer> {
}
export async function putParsedDocumentBlob(id: string, body: Buffer, namespace: string | null): Promise<string> {
return putParsedDocumentBlobForVersion(id, body, namespace, PDF_PARSER_VERSION);
}
export async function putParsedDocumentBlobForVersion(
id: string,
body: Buffer,
namespace: string | null,
parserVersion: string,
): Promise<string> {
const cfg = getS3Config();
const client = getS3ProxyClient();
const key = documentParsedKey(id, namespace);
const key = documentParsedKeyForVersion(id, namespace, parserVersion);
await client.send(
new PutObjectCommand({
Bucket: cfg.bucket,
@ -484,10 +502,14 @@ export async function deleteDocumentBlob(id: string, namespace: string | null):
const key = documentKey(id, namespace);
const parsedKey = documentParsedKey(id, namespace);
const legacyParsedKey = legacyDocumentParsedKey(id, namespace);
const ns = sanitizeNamespace(namespace);
const nsSegment = ns ? `ns/${ns}/` : '';
const parsedPrefix = `${cfg.prefix}/documents_v1/parsed_v2/${nsSegment}${id}/`;
await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: key }));
await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: parsedKey })).catch(() => undefined);
await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: legacyParsedKey })).catch(() => undefined);
await deleteDocumentPrefix(parsedPrefix).catch(() => undefined);
await deleteDocumentPrefix(`${key}/`).catch(() => undefined);
}

37
src/lib/server/documents/parse-state-backfill.ts
View file

@ -1,37 +0,0 @@
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
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 startedParse = await startPdfParseOperation({
documentId: input.documentId,
userId: input.userId,
namespace: input.namespace,
});
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;
}

36
src/lib/server/documents/parse-state-healing.ts
View file

@ -1,36 +0,0 @@
import { and, eq } from 'drizzle-orm';
import { getComputeOpStaleMs } from '@openreader/compute-core';
import { db } from '@/db';
import { documents } from '@/db/schema';
import {
isDocumentParseStateStale,
parseDocumentParseState,
stringifyDocumentParseState,
type DocumentParseState,
} from '@/lib/server/documents/parse-state';
export async function healStaleDocumentParseState(input: {
documentId: string;
userId: string;
state: DocumentParseState;
}): Promise<DocumentParseState> {
const staleMs = getComputeOpStaleMs();
if (!isDocumentParseStateStale(input.state, staleMs)) return input.state;
const nextState = parseDocumentParseState(stringifyDocumentParseState({
status: 'failed',
progress: null,
updatedAt: Date.now(),
error: `Parse state stale for more than ${staleMs}ms; marked failed for retry`,
...(input.state.opId ? { opId: input.state.opId } : {}),
...(input.state.jobId ? { jobId: input.state.jobId } : {}),
...(input.state.parserVersion ? { parserVersion: input.state.parserVersion } : {}),
}));
await db
.update(documents)
.set({ parseState: stringifyDocumentParseState(nextState) })
.where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
return nextState;
}

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

@ -1,118 +0,0 @@
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import type { ParsedPdfDocument, PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
export interface DocumentParseState {
status: PdfParseStatus;
progress?: PdfParseProgress | null;
updatedAt?: number;
error?: string | null;
opId?: string;
jobId?: string;
parserVersion?: string;
}
export function isInProgressParseStatus(status: PdfParseStatus): status is 'pending' | 'running' {
return status === 'pending' || status === 'running';
}
export function isDocumentParseStateStale(
state: DocumentParseState,
staleMs: number,
nowMs = Date.now(),
): boolean {
if (!isInProgressParseStatus(state.status)) return false;
if (!Number.isFinite(staleMs) || staleMs <= 0) return false;
const updatedAt = Number(state.updatedAt ?? 0);
if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false;
return (nowMs - updatedAt) > staleMs;
}
export function normalizeParseStatus(status: string | null | undefined): PdfParseStatus {
if (status === 'pending' || status === 'running' || status === 'ready' || status === 'failed') {
return status;
}
return 'pending';
}
export function hasCurrentPdfParserVersion(parserVersion: string | null | undefined): boolean {
return typeof parserVersion === 'string' && parserVersion.trim() === PDF_PARSER_VERSION;
}
export function normalizeDocumentParseStateForCurrentParserVersion(
state: DocumentParseState,
nowMs = Date.now(),
): DocumentParseState {
const status = normalizeParseStatus(state.status);
if (status === 'failed' || hasCurrentPdfParserVersion(state.parserVersion)) {
return state;
}
return {
status: 'pending',
progress: null,
updatedAt: nowMs,
};
}
export function resolveParsedPdfParserVersion(parsed: ParsedPdfDocument | null | undefined): string {
const parserVersion = parsed?.parserVersion?.trim();
return parserVersion || PDF_PARSER_VERSION;
}
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;
const parserVersion = typeof parsed.parserVersion === 'string' ? parsed.parserVersion : undefined;
return {
status,
progress,
...(typeof updatedAt === 'number' ? { updatedAt } : {}),
...(error ? { error } : {}),
...(opId ? { opId } : {}),
...(jobId ? { jobId } : {}),
...(parserVersion ? { parserVersion } : {}),
};
} 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 } : {}),
...(state.parserVersion ? { parserVersion: state.parserVersion } : {}),
});
}

33
src/lib/server/documents/parsed-pdf-reuse.ts
View file

@ -1,33 +0,0 @@
import { eq } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import {
normalizeDocumentParseStateForCurrentParserVersion,
parseDocumentParseState,
} from '@/lib/server/documents/parse-state';
type ReadyParsedPdfResult = {
parsedJsonKey: string;
};
export async function findReusableParsedPdfResult(
documentId: string,
): Promise<ReadyParsedPdfResult | null> {
const rows = await db
.select({
parseState: documents.parseState,
parsedJsonKey: documents.parsedJsonKey,
})
.from(documents)
.where(eq(documents.id, documentId));
for (const row of rows) {
const parsedJsonKey = row.parsedJsonKey?.trim();
if (!parsedJsonKey) continue;
const parseState = normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState));
if (parseState.status !== 'ready') continue;
return { parsedJsonKey };
}
return null;
}

30
src/lib/server/documents/pdf-parse-operation.ts
View file

@ -1,30 +0,0 @@
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create';
import { documentParseStateFromWorkerState } from '@/lib/server/compute/worker-parse-state';
import { documentKey } from '@/lib/server/documents/blobstore';
import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
import type { DocumentParseState } from '@/lib/server/documents/parse-state';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
export async function startPdfParseOperation(input: {
documentId: string;
userId: string;
namespace: string | null;
forceToken?: string;
}): Promise<{
workerState: WorkerOperationState<PdfLayoutJobResult>;
parseState: DocumentParseState;
}> {
const workerState = await createOrReusePdfWorkerOperation({
documentId: input.documentId,
namespace: input.namespace,
documentObjectKey: documentKey(input.documentId, input.namespace),
...(input.forceToken ? { forceToken: input.forceToken } : {}),
});
const rateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
await recordJobEvent(input.userId, 'pdf_layout', workerState.opId, rateConfig);
return {
workerState,
parseState: documentParseStateFromWorkerState(workerState),
};
}

39
src/lib/server/documents/register-upload.ts
View file

@ -3,12 +3,7 @@ import { documents } from '@/db/schema';
import {
enqueueDocumentPreview,
} from '@/lib/server/documents/previews';
import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse';
import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state';
import { startPdfParseOperation } from '@/lib/server/documents/pdf-parse-operation';
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
import { errorToLog, serverLogger } from '@/lib/server/logger';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import type { BaseDocument, DocumentType } from '@/types/documents';
type RegisterUploadedDocumentInput = {
@ -22,25 +17,6 @@ type RegisterUploadedDocumentInput = {
};
export async function registerUploadedDocument(input: RegisterUploadedDocumentInput): Promise<BaseDocument> {
const reusableParsedPdf = input.type === 'pdf'
? await findReusableParsedPdfResult(input.documentId)
: null;
const startedParse = input.type === 'pdf' && !reusableParsedPdf
? await startPdfParseOperation({
documentId: input.documentId,
userId: input.userId,
namespace: input.namespace,
})
: null;
const parsedJsonKey = reusableParsedPdf?.parsedJsonKey ?? null;
const parseState = input.type === 'pdf'
? stringifyDocumentParseState(
reusableParsedPdf
? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }
: (startedParse?.parseState ?? { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }),
)
: null;
await db
.insert(documents)
.values({
@ -51,8 +27,6 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn
size: input.size,
lastModified: input.lastModified,
filePath: input.documentId,
parseState,
parsedJsonKey,
})
.onConflictDoUpdate({
target: [documents.id, documents.userId],
@ -62,8 +36,6 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn
size: input.size,
lastModified: input.lastModified,
filePath: input.documentId,
parseState,
parsedJsonKey,
},
});
@ -84,17 +56,6 @@ export async function registerUploadedDocument(input: RegisterUploadedDocumentIn
}, 'Failed to enqueue document preview');
});
if (startedParse) {
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',
});
}
return {
id: input.documentId,
name: input.name,

512
src/lib/server/jobs/user-pdf-layout-job.ts
View file

@ -1,512 +0,0 @@
import { and, eq, inArray, isNull } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { documentKey, getParsedDocumentBlobByKey, putParsedDocumentBlob } from '@/lib/server/documents/blobstore';
import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse';
import { serverLogger } from '@/lib/server/logger';
import { logDegraded, logServerError } from '@/lib/server/errors/logging';
import {
normalizeDocumentParseStateForCurrentParserVersion,
parseDocumentParseState,
resolveParsedPdfParserVersion,
stringifyDocumentParseState,
type DocumentParseState,
} from '@/lib/server/documents/parse-state';
import { getCompute } from '@/lib/server/compute';
import { clearTtsSegmentCache } from '@/lib/server/tts/segments-cache';
import { UNCLAIMED_USER_ID } from '@/lib/server/storage/docstore-legacy';
import type {
ParsedPdfDocument,
PdfLayoutJobBase,
PdfLayoutJobResult,
PdfLayoutProgress,
WorkerOperationState,
} from '@openreader/compute-core/api-contracts';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
type UserPdfLayoutJobRequest = PdfLayoutJobBase & {
userId: string;
forceToken?: string;
initialOpId?: string;
initialJobId?: string;
initialStatus?: 'pending' | 'running';
};
const running = new Set<string>();
const FOLLOWER_WAIT_TIMEOUT_MS = 180_000;
const FOLLOWER_POLL_MS = 1_200;
const PROGRESS_DB_THROTTLE_MS = 10_000;
type ParseRow = {
userId: string;
parseState: string | null;
parsedJsonKey: string | null;
};
function keyFor(input: UserPdfLayoutJobRequest): string {
const forceToken = input.forceToken?.trim();
if (forceToken) {
return `force:${input.documentId}:${input.namespace || ''}:${forceToken}`;
}
return `shared:${input.documentId}`;
}
function normalizeOpId(value: string | undefined): string | undefined {
const normalized = value?.trim();
return normalized ? normalized : undefined;
}
function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function inferNamespaceFromUnclaimedUserId(userId: string): string | null {
const prefix = `${UNCLAIMED_USER_ID}::`;
if (!userId.startsWith(prefix)) return null;
const ns = userId.slice(prefix.length).trim();
return ns || null;
}
function rowMatchesScope(row: ParseRow, input: UserPdfLayoutJobRequest): boolean {
const inferredNamespace = inferNamespaceFromUnclaimedUserId(row.userId);
if (inferredNamespace !== null) {
return inferredNamespace === (input.namespace ?? null);
}
// Regular (non-unclaimed) users are only shared in the non-namespaced path.
if (!input.namespace) return true;
// In namespaced contexts with regular users, avoid touching other users'
// rows since namespace is not persisted on document rows.
return row.userId === input.userId;
}
async function loadScopedRows(input: UserPdfLayoutJobRequest): Promise<ParseRow[]> {
const rows = (await db
.select({
userId: documents.userId,
parseState: documents.parseState,
parsedJsonKey: documents.parsedJsonKey,
})
.from(documents)
.where(eq(documents.id, input.documentId))) as ParseRow[];
return rows.filter((row) => rowMatchesScope(row, input));
}
function isReadyRow(row: ParseRow): row is ParseRow & { parsedJsonKey: string } {
if (!row.parsedJsonKey) return false;
return normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)).status === 'ready';
}
function userIdsFromRows(rows: ParseRow[]): string[] {
return Array.from(new Set(rows.map((row) => row.userId).filter(Boolean)));
}
function parseStateMatchCondition(expected: string | null) {
return expected === null ? isNull(documents.parseState) : eq(documents.parseState, expected);
}
async function loadParsedDocumentForResult(
parsed: ParsedPdfDocument | undefined,
parsedJsonKey: string,
): Promise<ParsedPdfDocument | null> {
if (parsed) return parsed;
try {
const json = await getParsedDocumentBlobByKey(parsedJsonKey);
return JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument;
} catch {
return null;
}
}
async function updateParseStateForUsers(input: {
documentId: string;
userIds: string[];
parseState: string;
parsedJsonKey?: string | null;
}): Promise<void> {
if (input.userIds.length === 0) return;
await db
.update(documents)
.set({
parseState: input.parseState,
...(typeof input.parsedJsonKey === 'string' || input.parsedJsonKey === null
? { parsedJsonKey: input.parsedJsonKey }
: {}),
})
.where(
and(
eq(documents.id, input.documentId),
input.userIds.length === 1
? eq(documents.userId, input.userIds[0])
: inArray(documents.userId, input.userIds),
),
);
}
async function syncCallerToSharedResult(input: UserPdfLayoutJobRequest): Promise<void> {
const deadline = Date.now() + FOLLOWER_WAIT_TIMEOUT_MS;
while (Date.now() < deadline) {
const reusableParsed = await findReusableParsedPdfResult(input.documentId);
if (reusableParsed) {
const readyState: DocumentParseState = {
status: 'ready',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
};
await updateParseStateForUsers({
documentId: input.documentId,
userIds: [input.userId],
parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: reusableParsed.parsedJsonKey,
});
return;
}
const rows = await loadScopedRows(input);
const ready = rows.find(isReadyRow);
if (ready) {
const readyState: DocumentParseState = {
status: 'ready',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
};
await updateParseStateForUsers({
documentId: input.documentId,
userIds: [input.userId],
parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: ready.parsedJsonKey,
});
return;
}
const statuses = rows.map((row) => normalizeDocumentParseStateForCurrentParserVersion(parseDocumentParseState(row.parseState)));
const hasInFlight = statuses.some((state) => state.status === 'pending' || state.status === 'running');
const failedState = statuses.find((state) => state.status === 'failed');
if (failedState && !hasInFlight) {
const nextFailedState: DocumentParseState = {
status: 'failed',
progress: null,
updatedAt: Date.now(),
...(failedState.error ? { error: failedState.error } : {}),
...(failedState.parserVersion ? { parserVersion: failedState.parserVersion } : {}),
};
await updateParseStateForUsers({
documentId: input.documentId,
userIds: [input.userId],
parseState: stringifyDocumentParseState(nextFailedState),
});
return;
}
await sleep(FOLLOWER_POLL_MS);
}
}
export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void> {
const key = keyFor(input);
if (running.has(key)) {
if (!input.forceToken?.trim()) {
await syncCallerToSharedResult(input);
}
return;
}
running.add(key);
let activeOpId = normalizeOpId(input.initialOpId);
let activeJobId = normalizeOpId(input.initialJobId);
try {
const scopedRows = await loadScopedRows(input);
const scopedUserIds = userIdsFromRows(scopedRows);
// Non-force jobs can short-circuit by reusing existing ready output from
// any row in the same document scope.
if (!input.forceToken?.trim()) {
const reusableParsed = await findReusableParsedPdfResult(input.documentId);
if (reusableParsed) {
const readyState: DocumentParseState = {
status: 'ready',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
};
await updateParseStateForUsers({
documentId: input.documentId,
userIds: [input.userId],
parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: reusableParsed.parsedJsonKey,
});
return;
}
const ready = scopedRows.find(isReadyRow);
if (ready) {
const readyState: DocumentParseState = {
status: 'ready',
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
};
await updateParseStateForUsers({
documentId: input.documentId,
userIds: [input.userId],
parseState: stringifyDocumentParseState(readyState),
parsedJsonKey: ready.parsedJsonKey,
});
return;
}
}
const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0];
if (!coordinator) return;
const initialInflightStatus = input.initialStatus === 'running' ? 'running' : 'pending';
const inflightStateData: DocumentParseState = {
status: initialInflightStatus,
progress: null,
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
};
const inflightState = stringifyDocumentParseState(inflightStateData);
const claimRows = (await db
.update(documents)
.set({
parseState: inflightState,
})
.where(
and(
eq(documents.id, input.documentId),
eq(documents.userId, coordinator.userId),
parseStateMatchCondition(coordinator.parseState),
),
)
.returning({ userId: documents.userId })) as Array<{ userId: string }>;
const claimed = claimRows.some((row) => row.userId === coordinator.userId);
if (!claimed) {
if (!input.forceToken?.trim()) {
await syncCallerToSharedResult(input);
}
return;
}
const cohortUserIds = Array.from(new Set([...scopedUserIds, input.userId, coordinator.userId]));
await updateParseStateForUsers({
documentId: input.documentId,
userIds: cohortUserIds,
parseState: inflightState,
});
const compute = await getCompute();
let lastProgressWriteAt = 0;
let lastSnapshotWriteAt = 0;
let lastSnapshotStatus: 'pending' | 'running' | null = null;
let lastSnapshotOpId: string | undefined;
let lastSnapshotJobId: string | undefined;
const persistRunningState = async (state: DocumentParseState): Promise<void> => {
await updateParseStateForUsers({
documentId: input.documentId,
userIds: cohortUserIds,
parseState: stringifyDocumentParseState(state),
});
};
const onWorkerSnapshot = async (snapshot: WorkerOperationState<PdfLayoutJobResult>): Promise<void> => {
if (snapshot.opId) activeOpId = snapshot.opId;
if (snapshot.jobId) activeJobId = snapshot.jobId;
const mappedStatus = snapshot.status === 'queued'
? 'pending'
: snapshot.status === 'running'
? 'running'
: null;
if (!mappedStatus) return;
const now = Date.now();
const forceWrite = (
mappedStatus !== lastSnapshotStatus
|| activeOpId !== lastSnapshotOpId
|| activeJobId !== lastSnapshotJobId
);
if (!forceWrite && (now - lastSnapshotWriteAt) < PROGRESS_DB_THROTTLE_MS) {
return;
}
const nextState: DocumentParseState = {
status: mappedStatus,
progress: snapshot.progress ?? null,
updatedAt: now,
parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
};
await persistRunningState(nextState);
lastSnapshotWriteAt = now;
lastSnapshotStatus = mappedStatus;
lastSnapshotOpId = activeOpId;
lastSnapshotJobId = activeJobId;
};
const writeProgress = async (progress: PdfLayoutProgress): Promise<void> => {
const now = Date.now();
if ((now - lastProgressWriteAt) < PROGRESS_DB_THROTTLE_MS && progress.pagesParsed < progress.totalPages) {
return;
}
lastProgressWriteAt = now;
const runningProgressState: DocumentParseState = {
status: 'running',
progress: {
totalPages: progress.totalPages,
pagesParsed: progress.pagesParsed,
currentPage: progress.currentPage,
phase: progress.phase,
},
updatedAt: Date.now(),
parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
};
await persistRunningState(runningProgressState);
};
const layout = await compute.parsePdfLayout({
documentId: input.documentId,
namespace: input.namespace,
documentObjectKey: documentKey(input.documentId, input.namespace),
forceToken: input.forceToken,
onProgress: writeProgress,
onWorkerSnapshot,
});
let parsedJsonKey = layout.parsedObjectKey ?? null;
if (!parsedJsonKey) {
if (!layout.parsed) throw new Error('Compute backend did not return parsed result');
const parsedJson = Buffer.from(JSON.stringify(layout.parsed));
parsedJsonKey = await putParsedDocumentBlob(input.documentId, parsedJson, input.namespace);
}
const parsedDoc = await loadParsedDocumentForResult(layout.parsed, parsedJsonKey);
const finalScopedRows = await loadScopedRows(input);
const finalUserIds = userIdsFromRows(finalScopedRows);
const readyState: DocumentParseState = {
status: 'ready',
progress: null,
updatedAt: Date.now(),
parserVersion: resolveParsedPdfParserVersion(parsedDoc),
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
};
const readyUserIds = finalUserIds.length > 0 ? finalUserIds : cohortUserIds;
await updateParseStateForUsers({
documentId: input.documentId,
userIds: readyUserIds,
parseState: stringifyDocumentParseState(readyState),
parsedJsonKey,
});
// Best-effort cache invalidation should not block parse readiness.
for (const userId of readyUserIds) {
void clearTtsSegmentCache({
userId,
documentId: input.documentId,
readerType: 'pdf',
}).then((cleared) => {
if (cleared.warning) {
logDegraded(serverLogger, {
event: 'documents.parse.cache_invalidation.warning',
msg: 'Parse cache invalidation warning',
step: 'clear_tts_segment_cache',
context: {
documentId: input.documentId,
userId,
warning: cleared.warning,
},
});
}
}).catch((cacheError) => {
logDegraded(serverLogger, {
event: 'documents.parse.cache_invalidation.failed',
msg: 'Parse cache invalidation failed',
step: 'clear_tts_segment_cache',
context: {
documentId: input.documentId,
userId,
},
error: cacheError,
});
});
}
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
const parseStatus = 'failed';
try {
const scopedRows = await loadScopedRows(input);
const scopedUserIds = userIdsFromRows(scopedRows);
const failedState: DocumentParseState = {
status: 'failed',
progress: null,
updatedAt: Date.now(),
error: message,
parserVersion: PDF_PARSER_VERSION,
...(activeOpId ? { opId: activeOpId } : {}),
...(activeJobId ? { jobId: activeJobId } : {}),
};
const failedUserIds = scopedUserIds.length > 0 ? scopedUserIds : [input.userId];
await updateParseStateForUsers({
documentId: input.documentId,
userIds: failedUserIds,
parseState: stringifyDocumentParseState(failedState),
});
} catch (statusError) {
logServerError(serverLogger, {
event: 'documents.parse.status_write.failed',
msg: 'Failed to write parse status',
error: statusError,
context: {
documentId: input.documentId,
parseStatus,
},
normalize: { code: 'DOCUMENT_PARSE_STATUS_WRITE_FAILED', errorClass: 'db' },
});
}
logServerError(serverLogger, {
event: 'documents.parse.job.failed',
msg: 'Parse job failed',
error,
context: {
documentId: input.documentId,
parseStatus,
},
normalize: { code: 'DOCUMENT_PARSE_JOB_FAILED', errorClass: 'upstream' },
});
} finally {
running.delete(key);
}
}
export function enqueueParsePdfJob(input: UserPdfLayoutJobRequest): void {
Promise.resolve()
.then(() => parsePdfJob(input))
.catch((error) => {
logServerError(serverLogger, {
event: 'documents.parse.job.uncaught_error',
msg: 'Parse job uncaught error',
error,
context: { documentId: input.documentId },
normalize: { code: 'DOCUMENT_PARSE_JOB_UNCAUGHT_ERROR', errorClass: 'unknown' },
});
});
}

39
src/lib/server/pdf-parse/artifact.ts Normal file
View file

@ -0,0 +1,39 @@
import { PDF_PARSER_VERSION } from '@openreader/compute-core/api-contracts';
import {
documentParsedKeyForVersion,
getParsedDocumentBlobByKey,
isMissingBlobError,
} from '@/lib/server/documents/blobstore';
import type { ParsedPdfDocument } from '@/types/parsed-pdf';
export interface ParsedPdfArtifact {
key: string;
bytes: Buffer;
parsed: ParsedPdfDocument;
}
function parseArtifact(bytes: Buffer): ParsedPdfDocument {
return JSON.parse(bytes.toString('utf8')) as ParsedPdfDocument;
}
export async function readParsedPdfArtifactByKey(key: string): Promise<ParsedPdfArtifact | null> {
try {
const bytes = await getParsedDocumentBlobByKey(key);
return {
key,
bytes,
parsed: parseArtifact(bytes),
};
} catch (error) {
if (isMissingBlobError(error)) return null;
throw error;
}
}
export async function readCurrentParsedPdfArtifact(input: {
documentId: string;
namespace: string | null;
}): Promise<ParsedPdfArtifact | null> {
const key = documentParsedKeyForVersion(input.documentId, input.namespace, PDF_PARSER_VERSION);
return readParsedPdfArtifactByKey(key);
}

65
src/lib/server/pdf-parse/operation.ts Normal file
View file

@ -0,0 +1,65 @@
import { buildPdfOpKey } from '@/lib/server/compute/worker';
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create';
import {
fetchWorkerOperationState,
fetchWorkerOperationStateByKey,
} from '@/lib/server/compute/worker-op-state';
import { documentKey } from '@/lib/server/documents/blobstore';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
function currentPdfOperationInput(documentId: string, namespace: string | null, forceToken?: string): {
documentId: string;
namespace: string | null;
documentObjectKey: string;
forceToken?: string;
} {
return {
documentId,
namespace,
documentObjectKey: documentKey(documentId, namespace),
...(forceToken ? { forceToken } : {}),
};
}
export function buildCurrentPdfParseOpKeyPrefix(input: {
documentId: string;
namespace: string | null;
}): string {
return buildPdfOpKey(currentPdfOperationInput(input.documentId, input.namespace));
}
export async function lookupCurrentPdfParseOperation(input: {
documentId: string;
namespace: string | null;
}): Promise<WorkerOperationState<PdfLayoutJobResult> | null> {
return fetchWorkerOperationStateByKey<PdfLayoutJobResult>(
buildCurrentPdfParseOpKeyPrefix(input),
);
}
export async function createOrReuseCurrentPdfParseOperation(input: {
documentId: string;
namespace: string | null;
forceToken?: string;
}): Promise<WorkerOperationState<PdfLayoutJobResult>> {
return createOrReusePdfWorkerOperation(currentPdfOperationInput(
input.documentId,
input.namespace,
input.forceToken,
));
}
export async function fetchPdfParseOperation(opId: string): Promise<WorkerOperationState<PdfLayoutJobResult> | null> {
return fetchWorkerOperationState<PdfLayoutJobResult>(opId);
}
export function isPdfParseOperationForDocument(
state: WorkerOperationState<PdfLayoutJobResult>,
input: {
documentId: string;
namespace: string | null;
},
): boolean {
if (state.kind !== 'pdf_layout') return false;
return state.opKey.startsWith(buildCurrentPdfParseOpKeyPrefix(input));
}

42
src/lib/server/pdf-parse/snapshot.ts Normal file
View file

@ -0,0 +1,42 @@
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
import type { PdfParseStatus } from '@/types/parsed-pdf';
import type { PdfParseSnapshot } from '@/lib/server/pdf-parse/types';
function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus {
switch (status) {
case 'queued':
return 'pending';
case 'running':
return 'running';
case 'succeeded':
return 'ready';
case 'failed':
return 'failed';
}
const exhaustive: never = status;
return exhaustive;
}
export function parsedObjectKeyFromWorkerState(
state: WorkerOperationState<PdfLayoutJobResult>,
): string | null {
const result = state.result;
if (!result || typeof result !== 'object' || !('parsedObjectKey' in result)) return null;
const value = result.parsedObjectKey;
if (typeof value !== 'string') return null;
const normalized = value.trim();
return normalized || null;
}
export function pdfParseSnapshotFromWorkerState(
state: WorkerOperationState<PdfLayoutJobResult>,
): PdfParseSnapshot {
const parseStatus = mapWorkerStatusToParseStatus(state.status);
return {
parseStatus,
parseProgress: parseStatus === 'running' ? (state.progress ?? null) : null,
opId: state.opId?.trim() || null,
...(parseStatus === 'failed' && state.error?.message ? { error: state.error.message } : {}),
};
}

8
src/lib/server/pdf-parse/types.ts Normal file
View file

@ -0,0 +1,8 @@
import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf';
export interface PdfParseSnapshot {
parseStatus: PdfParseStatus;
parseProgress: PdfParseProgress | null;
opId: string | null;
error?: string | null;
}

2
src/types/documents.ts
View file

@ -7,8 +7,6 @@ export interface BaseDocument {
lastModified: number;
recentlyOpenedAt?: number;
type: DocumentType;
parseStatus?: 'pending' | 'running' | 'ready' | 'failed' | null;
parsedJsonKey?: string | null;
scope?: 'user';
folderId?: string;
}

169
tests/unit/pdf-parse-client-lifecycle.vitest.spec.ts Normal file
View file

@ -0,0 +1,169 @@
import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
type SnapshotListener = (event: Event) => void;
class MockMessageEvent extends Event {
data: string;
constructor(type: string, init: { data: string }) {
super(type);
this.data = init.data;
}
}
class MockEventSource {
static instances: MockEventSource[] = [];
readonly url: string;
readonly listeners = new Map<string, SnapshotListener[]>();
closed = false;
constructor(url: string) {
this.url = url;
MockEventSource.instances.push(this);
}
addEventListener(type: string, listener: SnapshotListener): void {
const arr = this.listeners.get(type) ?? [];
arr.push(listener);
this.listeners.set(type, arr);
}
close(): void {
this.closed = true;
}
emit(type: string, payload: unknown): void {
const event = new MockMessageEvent(type, {
data: JSON.stringify(payload),
});
for (const listener of this.listeners.get(type) ?? []) {
listener(event);
}
}
}
describe('PDF parse client lifecycle', () => {
const originalFetch = global.fetch;
const originalEventSource = global.EventSource;
const originalMessageEvent = global.MessageEvent;
beforeEach(() => {
MockEventSource.instances = [];
const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
const url = String(input);
const method = init?.method ?? 'GET';
if (url.endsWith('/api/documents/doc-1/parsed') && method === 'GET') {
fetchMock.getCount = (fetchMock.getCount ?? 0) + 1;
if (fetchMock.getCount === 1) {
return new Response(JSON.stringify({
parseStatus: 'pending',
parseProgress: null,
opId: null,
}), {
status: 409,
headers: { 'Content-Type': 'application/json' },
});
}
return new Response(JSON.stringify({
documentId: 'doc-1',
pages: [],
}), {
status: 200,
headers: { 'Content-Type': 'application/json' },
});
}
if (url.endsWith('/api/documents/doc-1/parsed') && method === 'POST') {
return new Response(JSON.stringify({
parseStatus: 'pending',
parseProgress: null,
opId: 'op-1',
}), {
status: 202,
headers: { 'Content-Type': 'application/json' },
});
}
throw new Error(`Unexpected fetch: ${method} ${url}`);
}) as typeof fetch & { getCount?: number };
global.fetch = fetchMock;
global.EventSource = MockEventSource as unknown as typeof EventSource;
global.MessageEvent = MockMessageEvent as unknown as typeof MessageEvent;
});
afterEach(() => {
global.fetch = originalFetch;
global.EventSource = originalEventSource;
global.MessageEvent = originalMessageEvent;
});
test('follows not-ready -> ensure op -> SSE snapshots -> ready artifact', async () => {
const {
ParsedPdfNotReadyError,
ensureParsedPdfDocumentOperation,
getParsedPdfDocument,
subscribeParsedPdfDocumentEvents,
} = await import('../../src/lib/client/api/documents');
let initialError: unknown;
try {
await getParsedPdfDocument('doc-1');
} catch (error) {
initialError = error;
}
expect(initialError).toBeInstanceOf(ParsedPdfNotReadyError);
expect((initialError as InstanceType<typeof ParsedPdfNotReadyError>).parseStatus).toBe('pending');
const ensured = await ensureParsedPdfDocumentOperation('doc-1');
expect(ensured).toMatchObject({
parseStatus: 'pending',
opId: 'op-1',
});
const snapshots: Array<{ parseStatus: string; opId?: string | null }> = [];
const unsubscribe = subscribeParsedPdfDocumentEvents('doc-1', { opId: 'op-1' }, {
onSnapshot: (snapshot) => {
snapshots.push({
parseStatus: snapshot.parseStatus,
opId: snapshot.opId,
});
},
});
expect(MockEventSource.instances).toHaveLength(1);
expect(MockEventSource.instances[0]?.url).toBe('/api/documents/doc-1/parsed/events?opId=op-1');
MockEventSource.instances[0]?.emit('snapshot', {
eventId: 1,
snapshot: {
opId: 'op-1',
status: 'running',
progress: { totalPages: 5, pagesParsed: 2, currentPage: 3, phase: 'infer' },
},
});
MockEventSource.instances[0]?.emit('snapshot', {
eventId: 2,
snapshot: {
opId: 'op-1',
status: 'succeeded',
},
});
expect(snapshots).toEqual([
{ parseStatus: 'running', opId: 'op-1' },
{ parseStatus: 'ready', opId: 'op-1' },
]);
unsubscribe();
expect(MockEventSource.instances[0]?.closed).toBe(true);
const parsed = await getParsedPdfDocument('doc-1');
expect(parsed).toMatchObject({ documentId: 'doc-1' });
});
});

84
tests/unit/pdf-parse-state.vitest.spec.ts
View file

@ -1,84 +0,0 @@
import { describe, expect, test } from 'vitest';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import {
normalizeDocumentParseStateForCurrentParserVersion,
parseDocumentParseState,
stringifyDocumentParseState,
} from '../../src/lib/server/documents/parse-state';
describe('document parse state parser-version handling', () => {
test('preserves parserVersion through stringify/parse', () => {
const state = parseDocumentParseState(stringifyDocumentParseState({
status: 'ready',
progress: null,
updatedAt: 1234,
parserVersion: PDF_PARSER_VERSION,
}));
expect(state).toEqual({
status: 'ready',
progress: null,
updatedAt: 1234,
parserVersion: PDF_PARSER_VERSION,
});
});
test('invalidates ready and inflight states from older parser versions', () => {
expect(normalizeDocumentParseStateForCurrentParserVersion({
status: 'ready',
progress: null,
updatedAt: 1111,
parserVersion: 'old-parser',
opId: 'op-old',
}, 2222)).toEqual({
status: 'pending',
progress: null,
updatedAt: 2222,
});
expect(normalizeDocumentParseStateForCurrentParserVersion({
status: 'running',
progress: {
totalPages: 10,
pagesParsed: 3,
currentPage: 4,
phase: 'infer',
},
updatedAt: 1111,
parserVersion: 'old-parser',
opId: 'op-old',
}, 3333)).toEqual({
status: 'pending',
progress: null,
updatedAt: 3333,
});
});
test('keeps failed states and current-version ready states intact', () => {
expect(normalizeDocumentParseStateForCurrentParserVersion({
status: 'failed',
progress: null,
updatedAt: 1111,
error: 'no text layer',
parserVersion: 'old-parser',
})).toEqual({
status: 'failed',
progress: null,
updatedAt: 1111,
error: 'no text layer',
parserVersion: 'old-parser',
});
expect(normalizeDocumentParseStateForCurrentParserVersion({
status: 'ready',
progress: null,
updatedAt: 1111,
parserVersion: PDF_PARSER_VERSION,
})).toEqual({
status: 'ready',
progress: null,
updatedAt: 1111,
parserVersion: PDF_PARSER_VERSION,
});
});
});

156
tests/unit/pdf-parsed-events-route-legacy-backfill.vitest.spec.ts
View file

@ -1,156 +0,0 @@
import { afterEach, 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,
},
requireAuthContext: 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/documents/parse-state-backfill', () => ({
backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation,
}));
vi.mock('@/lib/server/documents/parse-state-healing', () => ({
healStaleDocumentParseState: hoisted.healStaleDocumentParseState,
}));
vi.mock('@/lib/server/documents/blobstore', () => ({
isValidDocumentId: vi.fn(() => true),
}));
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/events legacy backfill', () => {
const originalFetch = global.fetch;
beforeEach(() => {
process.env.COMPUTE_WORKER_URL = 'http://localhost:4010';
process.env.COMPUTE_WORKER_TOKEN = 'worker-test-token';
hoisted.row = {
id: 'doc-1',
userId: 'user-1',
parseState: 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 }) => ({
where: vi.fn(async () => {
if (typeof values.parseState !== 'undefined') {
hoisted.row.parseState = values.parseState;
}
return [];
}),
})),
})),
};
hoisted.requireAuthContext.mockReset();
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
hoisted.backfillPendingPdfParseOperation.mockReset();
hoisted.healStaleDocumentParseState.mockClear();
global.fetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => new Promise((_, reject) => {
const signal = init?.signal;
if (signal) {
signal.addEventListener('abort', () => {
reject(new DOMException('Aborted', 'AbortError'));
}, { once: true });
}
})) as typeof fetch;
});
afterEach(() => {
global.fetch = originalFetch;
});
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/events/route');
const controller = new AbortController();
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed/events', {
signal: controller.signal,
});
const response = await GET(request, {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(200);
const reader = response.body?.getReader();
expect(reader).toBeDefined();
const first = await reader!.read();
const chunk = new TextDecoder().decode(first.value);
expect(chunk).toContain('event: snapshot');
expect(chunk).toContain('"parseStatus":"pending"');
expect(chunk).toContain('"opId":"op-legacy-1"');
controller.abort();
await reader!.cancel().catch(() => {});
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"');
});
});

149
tests/unit/pdf-parsed-events-route-worker-proxy.vitest.spec.ts Normal file
View file

@ -0,0 +1,149 @@
import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
import { NextRequest } from 'next/server';
const hoisted = vi.hoisted(() => ({
db: null as {
select: ReturnType<typeof vi.fn>;
} | null,
row: {
id: 'doc-1',
type: 'pdf',
},
requireAuthContext: vi.fn(),
fetchPdfParseOperation: vi.fn(),
isPdfParseOperationForDocument: vi.fn(),
getWorkerClientConfigFromEnv: vi.fn(),
}));
vi.mock('@/db', () => ({
get db() {
return hoisted.db;
},
}));
vi.mock('@/lib/server/auth/auth', () => ({
requireAuthContext: hoisted.requireAuthContext,
}));
vi.mock('@/lib/server/pdf-parse/operation', () => ({
fetchPdfParseOperation: hoisted.fetchPdfParseOperation,
isPdfParseOperationForDocument: hoisted.isPdfParseOperationForDocument,
}));
vi.mock('@/lib/server/compute/worker', () => ({
getWorkerClientConfigFromEnv: hoisted.getWorkerClientConfigFromEnv,
}));
vi.mock('@/lib/server/documents/blobstore', () => ({
isValidDocumentId: vi.fn(() => true),
}));
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',
})),
}));
describe('GET /api/documents/[id]/parsed/events worker event proxy', () => {
const originalFetch = global.fetch;
beforeEach(() => {
hoisted.db = {
select: vi.fn(() => ({
from: vi.fn(() => ({
where: vi.fn(() => ({
limit: vi.fn(async () => [{ ...hoisted.row }]),
})),
})),
})),
};
hoisted.requireAuthContext.mockReset();
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
hoisted.fetchPdfParseOperation.mockReset();
hoisted.fetchPdfParseOperation.mockResolvedValue({
opId: 'op-1',
opKey: 'pdf_layout|v1|parser|doc-1||doc-key|',
kind: 'pdf_layout',
jobId: 'job-1',
status: 'running',
queuedAt: Date.now() - 1000,
updatedAt: Date.now(),
});
hoisted.isPdfParseOperationForDocument.mockReset();
hoisted.isPdfParseOperationForDocument.mockReturnValue(true);
hoisted.getWorkerClientConfigFromEnv.mockReset();
hoisted.getWorkerClientConfigFromEnv.mockReturnValue({
baseUrl: 'http://worker.local',
token: 'worker-token',
});
global.fetch = vi.fn(async () => new Response(
'event: snapshot\ndata: {"eventId":1,"snapshot":{"opId":"op-1","status":"running"}}\n\n',
{
status: 200,
headers: {
'Content-Type': 'text/event-stream; charset=utf-8',
},
},
)) as typeof fetch;
});
test('requires an opId', async () => {
const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route');
const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events'), {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(400);
await expect(response.json()).resolves.toMatchObject({ error: 'opId is required' });
});
test('proxies the worker SSE stream after validating document ownership', async () => {
const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route');
const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events?opId=op-1'), {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(200);
expect(response.headers.get('content-type')).toContain('text/event-stream');
const text = await response.text();
expect(text).toContain('event: snapshot');
expect(text).toContain('"opId":"op-1"');
expect(hoisted.fetchPdfParseOperation).toHaveBeenCalledWith('op-1');
expect(hoisted.isPdfParseOperationForDocument).toHaveBeenCalled();
expect(global.fetch).toHaveBeenCalledWith(
'http://worker.local/ops/op-1/events',
expect.anything(),
);
});
test('denies proxying when the op does not belong to the document', async () => {
hoisted.isPdfParseOperationForDocument.mockReturnValue(false);
const { GET } = await import('../../src/app/api/documents/[id]/parsed/events/route');
const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed/events?opId=op-1'), {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(404);
await expect(response.json()).resolves.toMatchObject({ error: 'Operation not found' });
expect(hoisted.fetchPdfParseOperation).toHaveBeenCalledWith('op-1');
expect(global.fetch).not.toHaveBeenCalled();
});
afterEach(() => {
global.fetch = originalFetch;
});
});

133
tests/unit/pdf-parsed-route-legacy-backfill.vitest.spec.ts
View file

@ -1,133 +0,0 @@
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(),
healStaleDocumentParseState: vi.fn(async ({ state }) => state),
startPdfParseOperation: vi.fn(),
enqueueParsePdfJob: vi.fn(),
}));
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-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(),
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 pure data fetch', () => {
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.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 () => {
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(409);
await expect(response.json()).resolves.toMatchObject({
parseStatus: 'pending',
opId: null,
});
expect(hoisted.startPdfParseOperation).not.toHaveBeenCalled();
expect(hoisted.enqueueParsePdfJob).not.toHaveBeenCalled();
expect(hoisted.row.parseState).toBeNull();
});
});

228
tests/unit/pdf-parsed-route-worker-flow.vitest.spec.ts Normal file
View file

@ -0,0 +1,228 @@
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>;
} | null,
row: {
id: 'doc-1',
type: 'pdf',
},
requireAuthContext: vi.fn(),
readCurrentParsedPdfArtifact: vi.fn(),
readParsedPdfArtifactByKey: vi.fn(),
lookupCurrentPdfParseOperation: vi.fn(),
createOrReuseCurrentPdfParseOperation: vi.fn(),
checkJobRate: vi.fn(),
getPdfLayoutRateConfig: vi.fn(),
getResolvedRuntimeConfig: vi.fn(),
buildComputeRateLimitedResponse: vi.fn(),
}));
vi.mock('@/db', () => ({
get db() {
return hoisted.db;
},
}));
vi.mock('@/lib/server/auth/auth', () => ({
requireAuthContext: hoisted.requireAuthContext,
}));
vi.mock('@/lib/server/pdf-parse/artifact', () => ({
readCurrentParsedPdfArtifact: hoisted.readCurrentParsedPdfArtifact,
readParsedPdfArtifactByKey: hoisted.readParsedPdfArtifactByKey,
}));
vi.mock('@/lib/server/pdf-parse/operation', () => ({
lookupCurrentPdfParseOperation: hoisted.lookupCurrentPdfParseOperation,
createOrReuseCurrentPdfParseOperation: hoisted.createOrReuseCurrentPdfParseOperation,
}));
vi.mock('@/lib/server/rate-limit/job-rate-limiter', () => ({
checkJobRate: hoisted.checkJobRate,
getPdfLayoutRateConfig: hoisted.getPdfLayoutRateConfig,
}));
vi.mock('@/lib/server/rate-limit/problem-response', () => ({
buildComputeRateLimitedResponse: hoisted.buildComputeRateLimitedResponse,
}));
vi.mock('@/lib/server/runtime-config', () => ({
getResolvedRuntimeConfig: hoisted.getResolvedRuntimeConfig,
}));
vi.mock('@/lib/server/documents/blobstore', () => ({
isValidDocumentId: vi.fn(() => true),
}));
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',
})),
}));
describe('GET/POST /api/documents/[id]/parsed worker flow', () => {
beforeEach(() => {
hoisted.db = {
select: vi.fn(() => ({
from: vi.fn(() => ({
where: vi.fn(() => ({
limit: vi.fn(async () => [{ ...hoisted.row }]),
})),
})),
})),
};
hoisted.requireAuthContext.mockReset();
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
hoisted.readCurrentParsedPdfArtifact.mockReset();
hoisted.readCurrentParsedPdfArtifact.mockResolvedValue(null);
hoisted.readParsedPdfArtifactByKey.mockReset();
hoisted.readParsedPdfArtifactByKey.mockResolvedValue(null);
hoisted.lookupCurrentPdfParseOperation.mockReset();
hoisted.lookupCurrentPdfParseOperation.mockResolvedValue(null);
hoisted.createOrReuseCurrentPdfParseOperation.mockReset();
hoisted.checkJobRate.mockReset();
hoisted.checkJobRate.mockResolvedValue({ allowed: true });
hoisted.getPdfLayoutRateConfig.mockReset();
hoisted.getPdfLayoutRateConfig.mockReturnValue({});
hoisted.getResolvedRuntimeConfig.mockReset();
hoisted.getResolvedRuntimeConfig.mockResolvedValue({});
hoisted.buildComputeRateLimitedResponse.mockReset();
});
test('GET returns pending when no current artifact or worker op exists', async () => {
const { GET } = await import('../../src/app/api/documents/[id]/parsed/route');
const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed'), {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(409);
await expect(response.json()).resolves.toMatchObject({
parseStatus: 'pending',
parseProgress: null,
opId: null,
});
});
test('GET reads the artifact referenced by a succeeded worker op', async () => {
hoisted.lookupCurrentPdfParseOperation.mockResolvedValue({
opId: 'op-1',
opKey: 'pdf_layout|v1|parser|doc-1||doc-key|',
kind: 'pdf_layout',
jobId: 'job-1',
status: 'succeeded',
queuedAt: Date.now() - 1000,
updatedAt: Date.now(),
result: { parsedObjectKey: 'parsed-key.json' },
});
hoisted.readParsedPdfArtifactByKey.mockResolvedValue({
key: 'parsed-key.json',
bytes: Buffer.from(JSON.stringify({ documentId: 'doc-1', pages: [] })),
parsed: { documentId: 'doc-1', pages: [] },
});
const { GET } = await import('../../src/app/api/documents/[id]/parsed/route');
const response = await GET(new NextRequest('http://localhost/api/documents/doc-1/parsed'), {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(200);
await expect(response.json()).resolves.toMatchObject({ documentId: 'doc-1' });
expect(hoisted.readParsedPdfArtifactByKey).toHaveBeenCalledWith('parsed-key.json');
});
test('POST creates a worker op when replace is requested', async () => {
hoisted.createOrReuseCurrentPdfParseOperation.mockResolvedValue({
opId: 'op-force-1',
opKey: 'pdf_layout|v1|parser|doc-1||doc-key|force',
kind: 'pdf_layout',
jobId: 'job-force-1',
status: 'queued',
queuedAt: Date.now(),
updatedAt: Date.now(),
});
const { POST } = await import('../../src/app/api/documents/[id]/parsed/route');
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', {
method: 'POST',
body: JSON.stringify({ replace: true }),
headers: { 'Content-Type': 'application/json' },
});
const response = await POST(request, {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(202);
await expect(response.json()).resolves.toMatchObject({
parseStatus: 'pending',
opId: 'op-force-1',
});
expect(hoisted.createOrReuseCurrentPdfParseOperation).toHaveBeenCalled();
});
test('POST does not report ready for a succeeded op until its artifact is readable', async () => {
hoisted.lookupCurrentPdfParseOperation.mockResolvedValue({
opId: 'op-ready-1',
opKey: 'pdf_layout|v1|parser|doc-1||doc-key|',
kind: 'pdf_layout',
jobId: 'job-ready-1',
status: 'succeeded',
queuedAt: Date.now() - 1000,
updatedAt: Date.now(),
result: { parsedObjectKey: 'missing-parsed-key.json' },
});
hoisted.readParsedPdfArtifactByKey.mockResolvedValue(null);
const { POST } = await import('../../src/app/api/documents/[id]/parsed/route');
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', {
method: 'POST',
body: JSON.stringify({ replace: false }),
headers: { 'Content-Type': 'application/json' },
});
const response = await POST(request, {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response.status).toBe(202);
await expect(response.json()).resolves.toMatchObject({
parseStatus: 'running',
opId: 'op-ready-1',
});
expect(hoisted.readParsedPdfArtifactByKey).toHaveBeenCalledWith('missing-parsed-key.json');
expect(hoisted.createOrReuseCurrentPdfParseOperation).not.toHaveBeenCalled();
});
test('POST returns the rate-limited response without creating a worker op', async () => {
const sentinel = new Response('rate limited', { status: 429 });
hoisted.checkJobRate.mockResolvedValue({ allowed: false });
hoisted.buildComputeRateLimitedResponse.mockReturnValue(sentinel);
const { POST } = await import('../../src/app/api/documents/[id]/parsed/route');
const request = new NextRequest('http://localhost/api/documents/doc-1/parsed', {
method: 'POST',
body: JSON.stringify({ replace: true }),
headers: { 'Content-Type': 'application/json' },
});
const response = await POST(request, {
params: Promise.resolve({ id: 'doc-1' }),
});
expect(response).toBe(sentinel);
expect(hoisted.createOrReuseCurrentPdfParseOperation).not.toHaveBeenCalled();
});
});

122
tests/unit/worker-parse-state.vitest.spec.ts
View file

@ -1,122 +0,0 @@
import { describe, expect, test } from 'vitest';
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
import {
documentParseStateFromWorkerState,
isWorkerOperationStateStale,
snapshotFromWorkerState,
} from '../../src/lib/server/compute/worker-parse-state';
function makeWorkerState(
overrides: Partial<WorkerOperationState<PdfLayoutJobResult>>,
): WorkerOperationState<PdfLayoutJobResult> {
return {
opId: 'op-123',
opKey: `pdf_layout|v1|${PDF_PARSER_VERSION}|doc-1`,
kind: 'pdf_layout',
jobId: 'job-123',
status: 'queued',
queuedAt: 1_700_000_000_000,
updatedAt: 1_700_000_000_000,
...overrides,
};
}
describe('worker parse state mapping', () => {
test('maps queued worker state to pending parse state with op identifiers', () => {
const workerState = makeWorkerState({
status: 'queued',
progress: {
totalPages: 500,
pagesParsed: 0,
currentPage: 1,
phase: 'infer',
},
});
expect(snapshotFromWorkerState(workerState)).toEqual({
parseStatus: 'pending',
parseProgress: null,
});
expect(documentParseStateFromWorkerState(workerState, 1234)).toEqual({
status: 'pending',
progress: null,
updatedAt: 1234,
parserVersion: PDF_PARSER_VERSION,
opId: 'op-123',
jobId: 'job-123',
});
});
test('maps running worker state to running parse state with progress', () => {
const workerState = makeWorkerState({
status: 'running',
progress: {
totalPages: 500,
pagesParsed: 120,
currentPage: 121,
phase: 'infer',
},
});
expect(documentParseStateFromWorkerState(workerState, 5678)).toEqual({
status: 'running',
progress: {
totalPages: 500,
pagesParsed: 120,
currentPage: 121,
phase: 'infer',
},
updatedAt: 5678,
parserVersion: PDF_PARSER_VERSION,
opId: 'op-123',
jobId: 'job-123',
});
});
test('maps failed worker state to failed parse state and preserves the worker error', () => {
const workerState = makeWorkerState({
status: 'failed',
error: {
code: 'PDF_PARSE_FAILED',
message: 'layout model crashed',
},
});
expect(documentParseStateFromWorkerState(workerState, 9999)).toEqual({
status: 'failed',
progress: null,
updatedAt: 9999,
parserVersion: PDF_PARSER_VERSION,
opId: 'op-123',
jobId: 'job-123',
error: 'layout model crashed',
});
});
test('treats old inflight worker states as stale', () => {
const workerState = makeWorkerState({
status: 'running',
updatedAt: 1_000,
progress: {
totalPages: 500,
pagesParsed: 250,
currentPage: 251,
phase: 'infer',
},
});
expect(isWorkerOperationStateStale(workerState, 5_000, 6_001)).toBe(true);
expect(isWorkerOperationStateStale(workerState, 5_000, 5_999)).toBe(false);
});
test('never treats terminal worker states as stale', () => {
const failedState = makeWorkerState({
status: 'failed',
updatedAt: 1_000,
error: { code: 'PDF_PARSE_FAILED', message: 'crashed' },
});
expect(isWorkerOperationStateStale(failedState, 5_000, 99_999)).toBe(false);
});
});