refactor(pdf-parse): migrate PDF parse state to worker-owned model
This change removes legacy server-side PDF parse state management and transitions
to a fully worker-owned model for PDF parsing operations. Key updates include:
- Deletes all code related to server-managed parse state, including:
- parse-state.ts, parse-state-backfill.ts, parse-state-healing.ts,
parsed-pdf-reuse.ts, pdf-parse-operation.ts, and related job logic
- Removes the user-pdf-layout-job queue and associated job logic
- Refactors API routes for parsed PDF documents and events to use the new
worker-owned PDF parse operation flow under src/lib/server/pdf-parse/
- Updates S3 parsed PDF artifact keying to include parser version for
deduplication and compatibility
- Refactors client API and hooks to handle new error and progress reporting
- Removes all tests for the legacy parse state and job system, updating remaining
tests to mock the new worker-owned flow
BREAKING CHANGE: Server no longer manages per-user PDF parse state or jobs.
All PDF parsing is now managed by the compute worker and new artifact keying.
Legacy parse state and jobs are no longer supported. Existing parsed PDFs may
need to be reprocessed for compatibility with the new model.
This commit is contained in:
Richard R
parent
833f38059c
commit
7a86ce6a94
24 changed files with 827 additions and 1980 deletions
|
|
@ -30,6 +30,7 @@ import {
|
|||
getComputeOpStaleMs,
|
||||
getAvailableCpuCores,
|
||||
getOnnxThreadsPerJob,
|
||||
PDF_PARSER_VERSION,
|
||||
withIdleTimeoutAndHardCap,
|
||||
withTimeout,
|
||||
} from '@openreader/compute-core';
|
||||
|
|
@ -124,6 +125,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[]>;
|
||||
}
|
||||
|
||||
|
|
@ -243,13 +245,18 @@ function sanitizeNamespace(namespace: string | null): string | null {
|
|||
return SAFE_NAMESPACE_REGEX.test(namespace) ? namespace : null;
|
||||
}
|
||||
|
||||
function encodeParserVersion(parserVersion: string): string {
|
||||
const normalized = parserVersion.trim() || 'unknown-parser';
|
||||
return encodeURIComponent(normalized);
|
||||
}
|
||||
|
||||
function documentParsedKey(id: string, namespace: string | null, prefix: string): string {
|
||||
if (!DOCUMENT_ID_REGEX.test(id)) {
|
||||
throw new Error(`Invalid document id: ${id}`);
|
||||
}
|
||||
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 +432,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 +912,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) {
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
@ -205,18 +207,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;
|
||||
|
|
@ -281,6 +283,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 +543,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 +581,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
setCurrDocText,
|
||||
setPdfDocument,
|
||||
fetchDocumentSettings,
|
||||
startParsedEventStream,
|
||||
resolveParsedDocumentState,
|
||||
]);
|
||||
|
||||
const updateDocumentSettings = useCallback(async (settings: DocumentSettings): Promise<void> => {
|
||||
|
|
@ -553,7 +608,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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,34 @@ 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);
|
||||
}
|
||||
return jsonSnapshot(snapshot, snapshot.parseStatus === 'ready' ? 200 : 202);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -251,41 +202,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' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
@ -78,17 +73,14 @@ export async function GET(req: NextRequest) {
|
|||
|
||||
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,
|
||||
parseStatus: null,
|
||||
parsedJsonKey: null,
|
||||
scope: 'user',
|
||||
};
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -81,3 +81,86 @@ 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;
|
||||
}
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -8,6 +8,7 @@ import {
|
|||
PutObjectCommand,
|
||||
} from '@aws-sdk/client-s3';
|
||||
import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
|
||||
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
|
||||
import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage/s3';
|
||||
|
||||
const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i;
|
||||
|
|
@ -102,13 +103,26 @@ 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);
|
||||
}
|
||||
|
||||
function encodeParserVersion(parserVersion: string): string {
|
||||
const normalized = parserVersion.trim() || PDF_PARSER_VERSION;
|
||||
return encodeURIComponent(normalized);
|
||||
}
|
||||
|
||||
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 +373,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 +507,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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -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 } : {}),
|
||||
});
|
||||
}
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -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),
|
||||
};
|
||||
}
|
||||
|
|
@ -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,24 +17,8 @@ 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;
|
||||
const parseState = null;
|
||||
const parsedJsonKey = null;
|
||||
|
||||
await db
|
||||
.insert(documents)
|
||||
|
|
@ -84,17 +63,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,
|
||||
|
|
|
|||
|
|
@ -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
39
src/lib/server/pdf-parse/artifact.ts
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
import { PDF_PARSER_VERSION } from '@openreader/compute-core';
|
||||
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
65
src/lib/server/pdf-parse/operation.ts
Normal 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));
|
||||
}
|
||||
41
src/lib/server/pdf-parse/snapshot.ts
Normal file
41
src/lib/server/pdf-parse/snapshot.ts
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
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';
|
||||
default:
|
||||
return 'pending';
|
||||
}
|
||||
}
|
||||
|
||||
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
8
src/lib/server/pdf-parse/types.ts
Normal 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;
|
||||
}
|
||||
|
|
@ -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,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -4,16 +4,15 @@ 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,
|
||||
type: 'pdf',
|
||||
},
|
||||
requireAuthContext: vi.fn(),
|
||||
backfillPendingPdfParseOperation: vi.fn(),
|
||||
healStaleDocumentParseState: vi.fn(async ({ state }) => state),
|
||||
fetchPdfParseOperation: vi.fn(),
|
||||
isPdfParseOperationForDocument: vi.fn(),
|
||||
getWorkerClientConfigFromEnv: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('@/db', () => ({
|
||||
|
|
@ -26,12 +25,13 @@ vi.mock('@/lib/server/auth/auth', () => ({
|
|||
requireAuthContext: hoisted.requireAuthContext,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/documents/parse-state-backfill', () => ({
|
||||
backfillPendingPdfParseOperation: hoisted.backfillPendingPdfParseOperation,
|
||||
vi.mock('@/lib/server/pdf-parse/operation', () => ({
|
||||
fetchPdfParseOperation: hoisted.fetchPdfParseOperation,
|
||||
isPdfParseOperationForDocument: hoisted.isPdfParseOperationForDocument,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/documents/parse-state-healing', () => ({
|
||||
healStaleDocumentParseState: hoisted.healStaleDocumentParseState,
|
||||
vi.mock('@/lib/server/compute/worker', () => ({
|
||||
getWorkerClientConfigFromEnv: hoisted.getWorkerClientConfigFromEnv,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/documents/blobstore', () => ({
|
||||
|
|
@ -55,102 +55,81 @@ vi.mock('@/lib/server/logger', () => ({
|
|||
},
|
||||
requestId: 'req-test',
|
||||
})),
|
||||
hashForLog: vi.fn(() => 'user-hash'),
|
||||
}));
|
||||
|
||||
describe('GET /api/documents/[id]/parsed/events legacy backfill', () => {
|
||||
describe('GET /api/documents/[id]/parsed/events worker proxy', () => {
|
||||
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 [];
|
||||
}),
|
||||
where: vi.fn(() => ({
|
||||
limit: vi.fn(async () => [{ ...hoisted.row }]),
|
||||
})),
|
||||
})),
|
||||
})),
|
||||
};
|
||||
hoisted.requireAuthContext.mockReset();
|
||||
hoisted.requireAuthContext.mockResolvedValue({ userId: 'user-1' });
|
||||
hoisted.backfillPendingPdfParseOperation.mockReset();
|
||||
hoisted.healStaleDocumentParseState.mockClear();
|
||||
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;
|
||||
});
|
||||
|
||||
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;
|
||||
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(),
|
||||
);
|
||||
});
|
||||
|
||||
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"');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -4,19 +4,20 @@ 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,
|
||||
type: 'pdf',
|
||||
},
|
||||
requireAuthContext: vi.fn(),
|
||||
fetchWorkerOperationState: vi.fn(),
|
||||
healStaleDocumentParseState: vi.fn(async ({ state }) => state),
|
||||
startPdfParseOperation: vi.fn(),
|
||||
enqueueParsePdfJob: 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', () => ({
|
||||
|
|
@ -29,29 +30,31 @@ vi.mock('@/lib/server/auth/auth', () => ({
|
|||
requireAuthContext: hoisted.requireAuthContext,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/compute/worker-op-state', () => ({
|
||||
fetchWorkerOperationState: hoisted.fetchWorkerOperationState,
|
||||
vi.mock('@/lib/server/pdf-parse/artifact', () => ({
|
||||
readCurrentParsedPdfArtifact: hoisted.readCurrentParsedPdfArtifact,
|
||||
readParsedPdfArtifactByKey: hoisted.readParsedPdfArtifactByKey,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/documents/parse-state-healing', () => ({
|
||||
healStaleDocumentParseState: hoisted.healStaleDocumentParseState,
|
||||
vi.mock('@/lib/server/pdf-parse/operation', () => ({
|
||||
lookupCurrentPdfParseOperation: hoisted.lookupCurrentPdfParseOperation,
|
||||
createOrReuseCurrentPdfParseOperation: hoisted.createOrReuseCurrentPdfParseOperation,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/documents/pdf-parse-operation', () => ({
|
||||
startPdfParseOperation: hoisted.startPdfParseOperation,
|
||||
vi.mock('@/lib/server/rate-limit/job-rate-limiter', () => ({
|
||||
checkJobRate: hoisted.checkJobRate,
|
||||
getPdfLayoutRateConfig: hoisted.getPdfLayoutRateConfig,
|
||||
}));
|
||||
|
||||
vi.mock('@/lib/server/jobs/user-pdf-layout-job', () => ({
|
||||
enqueueParsePdfJob: hoisted.enqueueParsePdfJob,
|
||||
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', () => ({
|
||||
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', () => ({
|
||||
|
|
@ -71,63 +74,104 @@ vi.mock('@/lib/server/logger', () => ({
|
|||
},
|
||||
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,
|
||||
};
|
||||
describe('GET/POST /api/documents/[id]/parsed worker-owned flow', () => {
|
||||
beforeEach(() => {
|
||||
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 [];
|
||||
}),
|
||||
where: vi.fn(() => ({
|
||||
limit: vi.fn(async () => [{ ...hoisted.row }]),
|
||||
})),
|
||||
})),
|
||||
})),
|
||||
};
|
||||
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();
|
||||
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('returns non-ready status without creating a worker op for legacy pending PDFs without opId', async () => {
|
||||
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 request = new NextRequest('http://localhost/api/documents/doc-1/parsed');
|
||||
const response = await GET(request, {
|
||||
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,
|
||||
});
|
||||
expect(hoisted.startPdfParseOperation).not.toHaveBeenCalled();
|
||||
expect(hoisted.enqueueParsePdfJob).not.toHaveBeenCalled();
|
||||
expect(hoisted.row.parseState).toBeNull();
|
||||
});
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue