feat(pdf): enable opId tracking and propagation for layout parsing operations
Integrate opId handling throughout PDF layout parsing by updating client hooks, API routes, and event streams to support operation ID propagation. Add server utility for worker operation creation to centralize opId management. Enhance polling and SSE subscription utilities to accept and return opId, improving traceability and coordination of parse operations across the client and server.
This commit is contained in:
Richard R
parent
1ec9632d4b
commit
7732426963
5 changed files with 422 additions and 157 deletions
|
|
@ -146,6 +146,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
const [parsedDocument, setParsedDocument] = useState<ParsedPdfDocument | null>(null);
|
||||
const [parseStatus, setParseStatus] = useState<PdfParseStatus | null>(null);
|
||||
const [parseProgress, setParseProgress] = useState<PdfParseProgress | null>(null);
|
||||
const [, setActiveParseOpId] = useState<string | null>(null);
|
||||
const [documentSettings, setDocumentSettings] = useState<DocumentSettings>(DEFAULT_DOCUMENT_SETTINGS);
|
||||
const [parsedOverlayEnabled, setParsedOverlayEnabled] = useState(false);
|
||||
const [isAudioCombining] = useState(false);
|
||||
|
|
@ -173,6 +174,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
documentId: string,
|
||||
initialStatus: PdfParseStatus | null,
|
||||
signal: AbortSignal,
|
||||
initialOpId?: string | null,
|
||||
): Promise<void> => {
|
||||
// Legacy PDFs may have null parseStatus; treat as pending so opening the
|
||||
// document backfills parse output via the parsed endpoint polling path.
|
||||
|
|
@ -182,22 +184,30 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
const delayMs = 1200;
|
||||
const retryFailed = effectiveInitialStatus === 'failed';
|
||||
let attempt = 0;
|
||||
let effectiveOpId = initialOpId?.trim() || null;
|
||||
while (!signal.aborted) {
|
||||
if (signal.aborted) return;
|
||||
const result = await getParsedPdfDocument(documentId, {
|
||||
signal,
|
||||
retryFailed: retryFailed && attempt === 0,
|
||||
...(effectiveOpId ? { opId: effectiveOpId } : {}),
|
||||
});
|
||||
if (result.status === 'ready') {
|
||||
setParsedDocument(result.parsed);
|
||||
setParseStatus('ready');
|
||||
setParseProgress(null);
|
||||
setActiveParseOpId(null);
|
||||
return;
|
||||
}
|
||||
if ('opId' in result && typeof result.opId === 'string' && result.opId.trim()) {
|
||||
effectiveOpId = result.opId.trim();
|
||||
setActiveParseOpId(effectiveOpId);
|
||||
}
|
||||
setParseStatus(result.status);
|
||||
setParseProgress(result.parseProgress ?? null);
|
||||
if (result.status === 'failed') {
|
||||
setParsedDocument(null);
|
||||
setActiveParseOpId(null);
|
||||
return;
|
||||
}
|
||||
attempt += 1;
|
||||
|
|
@ -216,24 +226,36 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
}
|
||||
}, []);
|
||||
|
||||
const startParsedPolling = useCallback((documentId: string) => {
|
||||
const startParsedPolling = useCallback((documentId: string, initialOpId?: string | null) => {
|
||||
parsePollAbortRef.current?.abort();
|
||||
parseSseCloseRef.current?.();
|
||||
parseSseCloseRef.current = null;
|
||||
setParseProgress(null);
|
||||
setActiveParseOpId(initialOpId?.trim() || null);
|
||||
const controller = new AbortController();
|
||||
parsePollAbortRef.current = controller;
|
||||
|
||||
const closeSse = subscribeParsedPdfDocumentEvents(documentId, {
|
||||
opId: initialOpId?.trim() || null,
|
||||
}, {
|
||||
onSnapshot: (snapshot) => {
|
||||
if (controller.signal.aborted) return;
|
||||
if (typeof snapshot.opId === 'string' && snapshot.opId.trim()) {
|
||||
setActiveParseOpId(snapshot.opId.trim());
|
||||
}
|
||||
setParseStatus(snapshot.parseStatus);
|
||||
setParseProgress(snapshot.parseProgress);
|
||||
if (snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed') {
|
||||
if (snapshot.parseStatus === 'failed') {
|
||||
setParsedDocument(null);
|
||||
setActiveParseOpId(null);
|
||||
} else {
|
||||
void fetchParsedDocument(documentId, 'ready', controller.signal);
|
||||
void fetchParsedDocument(
|
||||
documentId,
|
||||
'ready',
|
||||
controller.signal,
|
||||
typeof snapshot.opId === 'string' ? snapshot.opId : (initialOpId ?? null),
|
||||
);
|
||||
}
|
||||
closeSse();
|
||||
parseSseCloseRef.current = null;
|
||||
|
|
@ -263,7 +285,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
parsePollAbortRef.current = null;
|
||||
}
|
||||
}, { once: true });
|
||||
}, [fetchParsedDocument]);
|
||||
}, [fetchParsedDocument, setActiveParseOpId]);
|
||||
|
||||
useEffect(() => {
|
||||
pdfDocumentRef.current = pdfDocument;
|
||||
|
|
@ -449,6 +471,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
setParsedDocument(null);
|
||||
setParseStatus(null);
|
||||
setParseProgress(null);
|
||||
setActiveParseOpId(null);
|
||||
setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);
|
||||
|
||||
const meta = await getDocumentMetadata(id, { signal: controller.signal });
|
||||
|
|
@ -461,7 +484,8 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
const initialParseStatus = (meta.parseStatus ?? null) as PdfParseStatus | null;
|
||||
setParseStatus(initialParseStatus);
|
||||
setParseProgress(null);
|
||||
startParsedPolling(id);
|
||||
setActiveParseOpId(null);
|
||||
startParsedPolling(id, null);
|
||||
void fetchDocumentSettings(id, controller.signal);
|
||||
}
|
||||
|
||||
|
|
@ -513,11 +537,12 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
const forceReparseParsedPdf = useCallback(async (): Promise<void> => {
|
||||
if (!currDocId) return;
|
||||
try {
|
||||
await forceReparsePdfDocument(currDocId);
|
||||
const forced = await forceReparsePdfDocument(currDocId);
|
||||
setParsedDocument(null);
|
||||
setParseStatus('pending');
|
||||
setParseStatus(forced.status);
|
||||
setParseProgress(null);
|
||||
startParsedPolling(currDocId);
|
||||
setActiveParseOpId(forced.opId ?? null);
|
||||
startParsedPolling(currDocId, forced.opId ?? null);
|
||||
} catch (error) {
|
||||
console.error('Failed to force PDF reparse:', error);
|
||||
}
|
||||
|
|
@ -548,6 +573,7 @@ export function usePdfDocument(): PdfDocumentState {
|
|||
setParsedDocument(null);
|
||||
setParseStatus(null);
|
||||
setParseProgress(null);
|
||||
setActiveParseOpId(null);
|
||||
setDocumentSettings(DEFAULT_DOCUMENT_SETTINGS);
|
||||
pageTextCacheRef.current.clear();
|
||||
stop();
|
||||
|
|
|
|||
|
|
@ -31,11 +31,13 @@ type ParseRow = {
|
|||
type ParsedSnapshot = {
|
||||
parseStatus: PdfParseStatus;
|
||||
parseProgress: PdfParseProgress | null;
|
||||
opId?: string | null;
|
||||
};
|
||||
|
||||
type SnapshotState = {
|
||||
snapshot: ParsedSnapshot;
|
||||
opId: string | null;
|
||||
fromWorker: boolean;
|
||||
};
|
||||
|
||||
function s3NotConfiguredResponse(): NextResponse {
|
||||
|
|
@ -49,20 +51,28 @@ function sleep(ms: number): Promise<void> {
|
|||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function toSnapshotState(row: ParseRow): Promise<SnapshotState> {
|
||||
async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise<SnapshotState> {
|
||||
const state = await healStaleDocumentParseState({
|
||||
documentId: row.id,
|
||||
userId: row.userId,
|
||||
state: parseDocumentParseState(row.parseState),
|
||||
});
|
||||
const parseStatus = normalizeParseStatus(state.status);
|
||||
const opId = typeof state.opId === 'string' && state.opId.trim() ? state.opId.trim() : null;
|
||||
if (opId && parseStatus !== 'ready') {
|
||||
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) {
|
||||
return {
|
||||
snapshot: snapshotFromWorkerState(workerState),
|
||||
opId,
|
||||
snapshot: {
|
||||
...snapshotFromWorkerState(workerState),
|
||||
opId: workerState.opId,
|
||||
},
|
||||
opId: workerState.opId,
|
||||
fromWorker: true,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -70,8 +80,10 @@ async function toSnapshotState(row: ParseRow): Promise<SnapshotState> {
|
|||
snapshot: {
|
||||
parseStatus,
|
||||
parseProgress: state.progress ?? null,
|
||||
opId,
|
||||
},
|
||||
opId,
|
||||
fromWorker: false,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -97,8 +109,9 @@ async function syncFromDb(input: {
|
|||
storageUserId: string;
|
||||
allowedUserIds: string[];
|
||||
signature: string;
|
||||
preferredOpId?: string | null;
|
||||
writeSnapshot: (snapshot: ParsedSnapshot) => void;
|
||||
}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; signature: string } | null> {
|
||||
}): Promise<{ snapshot: ParsedSnapshot; opId: string | null; fromWorker: boolean; signature: string } | null> {
|
||||
const nextRow = await loadPreferredRow({
|
||||
documentId: input.documentId,
|
||||
storageUserId: input.storageUserId,
|
||||
|
|
@ -106,7 +119,7 @@ async function syncFromDb(input: {
|
|||
});
|
||||
if (!nextRow) return null;
|
||||
|
||||
const nextState = await toSnapshotState(nextRow);
|
||||
const nextState = await toSnapshotState(nextRow, input.preferredOpId);
|
||||
const nextSignature = JSON.stringify(nextState.snapshot);
|
||||
if (nextSignature !== input.signature) {
|
||||
input.writeSnapshot(nextState.snapshot);
|
||||
|
|
@ -115,6 +128,7 @@ async function syncFromDb(input: {
|
|||
return {
|
||||
snapshot: nextState.snapshot,
|
||||
opId: nextState.opId,
|
||||
fromWorker: nextState.fromWorker,
|
||||
signature: nextSignature,
|
||||
};
|
||||
}
|
||||
|
|
@ -131,6 +145,10 @@ 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 unclaimedUserId = getUnclaimedUserIdForNamespace(testNamespace);
|
||||
|
|
@ -147,16 +165,9 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
return NextResponse.json({ error: 'Not found' }, { status: 404 });
|
||||
}
|
||||
|
||||
const initialState = await toSnapshotState(row);
|
||||
const initialState = await toSnapshotState(row, requestedOpId);
|
||||
const workerCfg = getWorkerClientConfigFromEnv();
|
||||
const encoder = new TextEncoder();
|
||||
console.info('[parsed/events] stream open', {
|
||||
documentId: id,
|
||||
storageUserId,
|
||||
initialParseStatus: initialState.snapshot.parseStatus,
|
||||
initialOpId: initialState.opId,
|
||||
});
|
||||
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
let closed = false;
|
||||
|
|
@ -193,12 +204,23 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
const runWorkerProxy = async () => {
|
||||
let current = initialState.snapshot;
|
||||
let signature = JSON.stringify(current);
|
||||
let currentOpId = initialState.opId;
|
||||
let currentOpId = requestedOpId ?? initialState.opId;
|
||||
let currentFromWorker = initialState.fromWorker;
|
||||
let lastEventId: number | null = null;
|
||||
let loggedMissingOpId = false;
|
||||
const pinnedRequestedOp = requestedOpId ?? null;
|
||||
const shouldCloseForTerminalSnapshot = (snapshot: ParsedSnapshot, fromWorker: boolean): boolean => {
|
||||
const isTerminal = snapshot.parseStatus === 'ready' || snapshot.parseStatus === 'failed';
|
||||
if (!isTerminal) return false;
|
||||
// If caller pinned an opId, keep streaming until worker confirms that
|
||||
// op state. DB fallback can report stale terminal status for other rows.
|
||||
if (pinnedRequestedOp && snapshot.opId === pinnedRequestedOp && !fromWorker) return false;
|
||||
return true;
|
||||
};
|
||||
|
||||
writeSnapshot(current);
|
||||
|
||||
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
|
||||
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) {
|
||||
closeStream();
|
||||
return;
|
||||
}
|
||||
|
|
@ -215,6 +237,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
storageUserId,
|
||||
allowedUserIds,
|
||||
signature,
|
||||
preferredOpId: requestedOpId ?? currentOpId,
|
||||
writeSnapshot,
|
||||
}).then((next) => {
|
||||
if (closed) return;
|
||||
|
|
@ -224,7 +247,8 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
}
|
||||
current = next.snapshot;
|
||||
signature = next.signature;
|
||||
if (next.opId !== currentOpId) {
|
||||
currentFromWorker = next.fromWorker;
|
||||
if (!requestedOpId && next.opId !== currentOpId) {
|
||||
currentOpId = next.opId;
|
||||
lastEventId = null;
|
||||
if (workerAbort) {
|
||||
|
|
@ -232,7 +256,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
workerAbort = null;
|
||||
}
|
||||
}
|
||||
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
|
||||
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) {
|
||||
closeStream();
|
||||
}
|
||||
}).catch((error) => {
|
||||
|
|
@ -254,6 +278,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
storageUserId,
|
||||
allowedUserIds,
|
||||
signature,
|
||||
preferredOpId: requestedOpId ?? currentOpId,
|
||||
writeSnapshot,
|
||||
});
|
||||
if (!next) {
|
||||
|
|
@ -262,14 +287,22 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
}
|
||||
current = next.snapshot;
|
||||
signature = next.signature;
|
||||
currentOpId = next.opId;
|
||||
currentFromWorker = next.fromWorker;
|
||||
currentOpId = requestedOpId ?? next.opId;
|
||||
if (!currentOpId) {
|
||||
console.warn('[parsed/events] waiting for worker opId', {
|
||||
documentId: id,
|
||||
parseStatus: current.parseStatus,
|
||||
});
|
||||
if (!loggedMissingOpId) {
|
||||
loggedMissingOpId = true;
|
||||
console.warn('[parsed/events] missing worker opId while parse is non-terminal', {
|
||||
documentId: id,
|
||||
storageUserId,
|
||||
parseStatus: current.parseStatus,
|
||||
requestedOpId,
|
||||
});
|
||||
}
|
||||
} else if (loggedMissingOpId) {
|
||||
loggedMissingOpId = false;
|
||||
}
|
||||
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
|
||||
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) {
|
||||
closeStream();
|
||||
return;
|
||||
}
|
||||
|
|
@ -358,15 +391,19 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
);
|
||||
if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue;
|
||||
|
||||
const nextSnapshot = snapshotFromWorkerState(workerSnapshot);
|
||||
const nextSnapshot: ParsedSnapshot = {
|
||||
...snapshotFromWorkerState(workerSnapshot),
|
||||
opId: workerSnapshot.opId,
|
||||
};
|
||||
const nextSignature = JSON.stringify(nextSnapshot);
|
||||
if (nextSignature !== signature) {
|
||||
current = nextSnapshot;
|
||||
signature = nextSignature;
|
||||
currentFromWorker = true;
|
||||
writeSnapshot(current);
|
||||
}
|
||||
|
||||
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
|
||||
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) {
|
||||
closeStream();
|
||||
return;
|
||||
}
|
||||
|
|
@ -380,6 +417,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
storageUserId,
|
||||
allowedUserIds,
|
||||
signature,
|
||||
preferredOpId: requestedOpId ?? currentOpId,
|
||||
writeSnapshot,
|
||||
});
|
||||
if (!next) {
|
||||
|
|
@ -388,11 +426,12 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
}
|
||||
current = next.snapshot;
|
||||
signature = next.signature;
|
||||
if (next.opId !== currentOpId) {
|
||||
currentFromWorker = next.fromWorker;
|
||||
if (!requestedOpId && next.opId !== currentOpId) {
|
||||
currentOpId = next.opId;
|
||||
lastEventId = null;
|
||||
}
|
||||
if (current.parseStatus === 'ready' || current.parseStatus === 'failed') {
|
||||
if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) {
|
||||
closeStream();
|
||||
return;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,15 +4,17 @@ import { and, eq, inArray } from 'drizzle-orm';
|
|||
import { db } from '@/db';
|
||||
import { documents } from '@/db/schema';
|
||||
import { requireAuthContext } from '@/lib/server/auth/auth';
|
||||
import { mergeNonReadyParseSnapshot } from '@/lib/server/compute/worker-parse-state';
|
||||
import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create';
|
||||
import { snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state';
|
||||
import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state';
|
||||
import {
|
||||
documentKey,
|
||||
getParsedDocumentBlob,
|
||||
getParsedDocumentBlobByKey,
|
||||
isMissingBlobError,
|
||||
isValidDocumentId,
|
||||
putParsedDocumentBlob,
|
||||
} from '@/lib/server/documents/blobstore';
|
||||
import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job';
|
||||
import {
|
||||
normalizeParseStatus,
|
||||
parseDocumentParseState,
|
||||
|
|
@ -22,7 +24,7 @@ import { healStaleDocumentParseState } from '@/lib/server/documents/parse-state-
|
|||
import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace';
|
||||
import { isS3Configured } from '@/lib/server/storage/s3';
|
||||
import type { ParsedPdfDocument } from '@/types/parsed-pdf';
|
||||
import type { PdfLayoutJobResult } from '@openreader/compute-core/api-contracts';
|
||||
import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts';
|
||||
|
||||
export const dynamic = 'force-dynamic';
|
||||
|
||||
|
|
@ -33,11 +35,146 @@ 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: {
|
||||
documentId: string;
|
||||
allowedUserIds: string[];
|
||||
}): Promise<ParseRow[]> {
|
||||
return (await db
|
||||
.select({
|
||||
id: documents.id,
|
||||
userId: documents.userId,
|
||||
parseState: documents.parseState,
|
||||
parsedJsonKey: documents.parsedJsonKey,
|
||||
})
|
||||
.from(documents)
|
||||
.where(and(eq(documents.id, input.documentId), inArray(documents.userId, input.allowedUserIds)))) as ParseRow[];
|
||||
}
|
||||
|
||||
function pickPreferredRow(rows: ParseRow[], storageUserId: string): ParseRow | null {
|
||||
return rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0] ?? null;
|
||||
}
|
||||
|
||||
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)));
|
||||
}
|
||||
|
||||
async function finalizeFromWorkerState(input: {
|
||||
workerState: WorkerOperationState<PdfLayoutJobResult>;
|
||||
row: ParseRow;
|
||||
namespace: string | null;
|
||||
}): Promise<NextResponse> {
|
||||
const snapshot = snapshotFromWorkerState(input.workerState);
|
||||
|
||||
if (snapshot.parseStatus === 'pending' || snapshot.parseStatus === 'running') {
|
||||
return NextResponse.json({
|
||||
parseStatus: snapshot.parseStatus,
|
||||
parseProgress: snapshot.parseProgress,
|
||||
opId: input.workerState.opId,
|
||||
}, { status: 202 });
|
||||
}
|
||||
|
||||
if (snapshot.parseStatus === 'failed') {
|
||||
await writeParseRowState({
|
||||
documentId: input.row.id,
|
||||
userId: input.row.userId,
|
||||
parseState: stringifyDocumentParseState({
|
||||
status: 'failed',
|
||||
progress: null,
|
||||
updatedAt: Date.now(),
|
||||
error: input.workerState.error?.message ?? 'Worker parse failed',
|
||||
}),
|
||||
});
|
||||
|
||||
return NextResponse.json({
|
||||
parseStatus: 'failed',
|
||||
parseProgress: null,
|
||||
opId: input.workerState.opId,
|
||||
error: input.workerState.error?.message ?? 'Worker parse failed',
|
||||
}, { status: 202 });
|
||||
}
|
||||
|
||||
let parsedJsonKey: string | null = null;
|
||||
if (input.workerState.result && 'parsedObjectKey' in input.workerState.result) {
|
||||
parsedJsonKey = typeof input.workerState.result.parsedObjectKey === 'string'
|
||||
? input.workerState.result.parsedObjectKey
|
||||
: null;
|
||||
}
|
||||
|
||||
if (!parsedJsonKey && input.workerState.result && 'parsed' in input.workerState.result && input.workerState.result.parsed) {
|
||||
const parsedJson = Buffer.from(JSON.stringify(input.workerState.result.parsed));
|
||||
parsedJsonKey = await putParsedDocumentBlob(input.row.id, parsedJson, input.namespace);
|
||||
}
|
||||
|
||||
if (!parsedJsonKey) {
|
||||
return NextResponse.json({ error: 'Worker completed without parsed output' }, { status: 500 });
|
||||
}
|
||||
|
||||
await writeParseRowState({
|
||||
documentId: input.row.id,
|
||||
userId: input.row.userId,
|
||||
parseState: stringifyDocumentParseState({
|
||||
status: 'ready',
|
||||
progress: null,
|
||||
updatedAt: Date.now(),
|
||||
}),
|
||||
parsedJsonKey,
|
||||
});
|
||||
|
||||
const json = await getParsedDocumentBlobByKey(parsedJsonKey);
|
||||
let parsedDoc: ParsedPdfDocument | null = null;
|
||||
try {
|
||||
parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument;
|
||||
} catch {
|
||||
parsedDoc = null;
|
||||
}
|
||||
|
||||
if (!hasAnyParsedBlocks(parsedDoc)) {
|
||||
console.warn('[documents/parsed] parsed doc has no blocks', {
|
||||
documentId: input.row.id,
|
||||
userId: input.row.userId,
|
||||
parsedJsonKey,
|
||||
});
|
||||
}
|
||||
|
||||
return new NextResponse(new Uint8Array(json), {
|
||||
status: 200,
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Cache-Control': 'no-store',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string }> }) {
|
||||
try {
|
||||
if (!isS3Configured()) return s3NotConfiguredResponse();
|
||||
|
|
@ -48,6 +185,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
const params = await ctx.params;
|
||||
const id = (params.id || '').trim().toLowerCase();
|
||||
const retryFailed = req.nextUrl.searchParams.get('retry') === '1';
|
||||
const requestedOpId = normalizeOpId(req.nextUrl.searchParams.get('opId'));
|
||||
if (!isValidDocumentId(id)) {
|
||||
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
|
||||
}
|
||||
|
|
@ -57,78 +195,69 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
const storageUserId = authCtxOrRes.userId ?? unclaimedUserId;
|
||||
const allowedUserIds = authCtxOrRes.authEnabled ? [storageUserId, unclaimedUserId] : [unclaimedUserId];
|
||||
|
||||
const rows = (await db
|
||||
.select({
|
||||
id: documents.id,
|
||||
userId: documents.userId,
|
||||
parseState: documents.parseState,
|
||||
parsedJsonKey: documents.parsedJsonKey,
|
||||
})
|
||||
.from(documents)
|
||||
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
|
||||
id: string;
|
||||
userId: string;
|
||||
parseState: string | null;
|
||||
parsedJsonKey: string | null;
|
||||
}>;
|
||||
|
||||
const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0];
|
||||
const rows = await loadRows({ documentId: id, allowedUserIds });
|
||||
const row = pickPreferredRow(rows, storageUserId);
|
||||
if (!row) {
|
||||
return NextResponse.json({ error: 'Not found' }, { status: 404 });
|
||||
}
|
||||
|
||||
if (requestedOpId) {
|
||||
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(requestedOpId);
|
||||
if (workerState && workerState.opId === requestedOpId) {
|
||||
return finalizeFromWorkerState({
|
||||
workerState,
|
||||
row,
|
||||
namespace: testNamespace,
|
||||
});
|
||||
}
|
||||
console.warn('[documents/parsed:get] requested opId unavailable', {
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
opId: requestedOpId,
|
||||
});
|
||||
}
|
||||
|
||||
let state = parseDocumentParseState(row.parseState);
|
||||
state = await healStaleDocumentParseState({
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
state,
|
||||
});
|
||||
let effectiveStatus = normalizeParseStatus(state.status);
|
||||
let effectiveProgress = state.progress ?? null;
|
||||
const opId = typeof state.opId === 'string' ? state.opId.trim() : '';
|
||||
if (opId && effectiveStatus !== 'ready') {
|
||||
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(opId);
|
||||
if (workerState && workerState.opId === opId) {
|
||||
const merged = mergeNonReadyParseSnapshot({
|
||||
parseStatus: effectiveStatus,
|
||||
parseProgress: effectiveProgress,
|
||||
|
||||
const effectiveStatus = normalizeParseStatus(state.status);
|
||||
const effectiveProgress = state.progress ?? null;
|
||||
const effectiveOpId = normalizeOpId(state.opId);
|
||||
|
||||
if (effectiveOpId && effectiveStatus !== 'ready') {
|
||||
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(effectiveOpId);
|
||||
if (workerState && workerState.opId === effectiveOpId) {
|
||||
return finalizeFromWorkerState({
|
||||
workerState,
|
||||
});
|
||||
effectiveStatus = merged.parseStatus;
|
||||
effectiveProgress = merged.parseProgress;
|
||||
} else {
|
||||
console.warn('[documents/parsed:get] worker state unavailable for op', {
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
opId,
|
||||
parseStatus: effectiveStatus,
|
||||
row,
|
||||
namespace: testNamespace,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (effectiveStatus === 'failed' && retryFailed) {
|
||||
await db
|
||||
.update(documents)
|
||||
.set({
|
||||
parseState: stringifyDocumentParseState({
|
||||
status: 'pending',
|
||||
progress: null,
|
||||
updatedAt: Date.now(),
|
||||
}),
|
||||
})
|
||||
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
|
||||
enqueueParsePdfJob({
|
||||
const created = await createOrReusePdfWorkerOperation({
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
namespace: testNamespace,
|
||||
documentObjectKey: documentKey(id, testNamespace),
|
||||
});
|
||||
return NextResponse.json({ parseStatus: 'pending', parseProgress: null }, { status: 202 });
|
||||
const snapshot = snapshotFromWorkerState(created);
|
||||
return NextResponse.json({
|
||||
parseStatus: snapshot.parseStatus,
|
||||
parseProgress: snapshot.parseProgress,
|
||||
opId: created.opId,
|
||||
}, { status: 202 });
|
||||
}
|
||||
|
||||
if (effectiveStatus !== 'ready') {
|
||||
return NextResponse.json({
|
||||
parseStatus: effectiveStatus,
|
||||
parseProgress: effectiveProgress,
|
||||
opId: effectiveOpId,
|
||||
}, { status: 202 });
|
||||
}
|
||||
|
||||
|
|
@ -183,27 +312,21 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
return NextResponse.json({ error: 'Invalid document id' }, { status: 400 });
|
||||
}
|
||||
|
||||
let replace = false;
|
||||
try {
|
||||
const body = (await req.json()) as { replace?: unknown };
|
||||
replace = body?.replace === true;
|
||||
} catch {
|
||||
replace = false;
|
||||
}
|
||||
|
||||
const testNamespace = getOpenReaderTestNamespace(req.headers);
|
||||
const unclaimedUserId = getUnclaimedUserIdForNamespace(testNamespace);
|
||||
const storageUserId = authCtxOrRes.userId ?? unclaimedUserId;
|
||||
const allowedUserIds = authCtxOrRes.authEnabled ? [storageUserId, unclaimedUserId] : [unclaimedUserId];
|
||||
|
||||
const rows = (await db
|
||||
.select({
|
||||
id: documents.id,
|
||||
userId: documents.userId,
|
||||
parseState: documents.parseState,
|
||||
parsedJsonKey: documents.parsedJsonKey,
|
||||
})
|
||||
.from(documents)
|
||||
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
|
||||
id: string;
|
||||
userId: string;
|
||||
parseState: string | null;
|
||||
parsedJsonKey: string | null;
|
||||
}>;
|
||||
|
||||
const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0];
|
||||
const rows = await loadRows({ documentId: id, allowedUserIds });
|
||||
const row = pickPreferredRow(rows, storageUserId);
|
||||
if (!row) {
|
||||
return NextResponse.json({ error: 'Not found' }, { status: 404 });
|
||||
}
|
||||
|
|
@ -214,58 +337,37 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
|
|||
userId: row.userId,
|
||||
state,
|
||||
});
|
||||
let effectiveStatus = normalizeParseStatus(state.status);
|
||||
let effectiveProgress = state.progress ?? null;
|
||||
const opId = typeof state.opId === 'string' ? state.opId.trim() : '';
|
||||
if (opId && effectiveStatus !== 'ready') {
|
||||
const workerState = await fetchWorkerOperationState<PdfLayoutJobResult>(opId);
|
||||
if (workerState && workerState.opId === opId) {
|
||||
const merged = mergeNonReadyParseSnapshot({
|
||||
parseStatus: effectiveStatus,
|
||||
parseProgress: effectiveProgress,
|
||||
workerState,
|
||||
});
|
||||
effectiveStatus = merged.parseStatus;
|
||||
effectiveProgress = merged.parseProgress;
|
||||
} else {
|
||||
console.warn('[documents/parsed:post] worker state unavailable for op', {
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
opId,
|
||||
parseStatus: effectiveStatus,
|
||||
});
|
||||
|
||||
const existingOpId = normalizeOpId(state.opId);
|
||||
if (existingOpId) {
|
||||
const existing = await fetchWorkerOperationState<PdfLayoutJobResult>(existingOpId);
|
||||
if (existing && (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 (effectiveStatus !== 'running') {
|
||||
await db
|
||||
.update(documents)
|
||||
.set({
|
||||
parseState: stringifyDocumentParseState({
|
||||
status: 'pending',
|
||||
progress: null,
|
||||
updatedAt: Date.now(),
|
||||
}),
|
||||
})
|
||||
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
|
||||
}
|
||||
|
||||
enqueueParsePdfJob({
|
||||
const created = await createOrReusePdfWorkerOperation({
|
||||
documentId: id,
|
||||
userId: row.userId,
|
||||
namespace: testNamespace,
|
||||
documentObjectKey: documentKey(id, testNamespace),
|
||||
forceToken: randomUUID(),
|
||||
});
|
||||
|
||||
return NextResponse.json(
|
||||
{
|
||||
parseStatus: effectiveStatus === 'running' ? 'running' : 'pending',
|
||||
parseProgress: effectiveStatus === 'running' ? effectiveProgress : null,
|
||||
},
|
||||
{ status: 202 },
|
||||
);
|
||||
const snapshot = snapshotFromWorkerState(created);
|
||||
|
||||
return NextResponse.json({
|
||||
parseStatus: snapshot.parseStatus,
|
||||
parseProgress: snapshot.parseProgress,
|
||||
opId: created.opId,
|
||||
}, { status: 202 });
|
||||
} catch (error) {
|
||||
console.error('Error forcing parsed PDF refresh:', error);
|
||||
return NextResponse.json({ error: 'Failed to force parsed PDF refresh' }, { status: 500 });
|
||||
return NextResponse.json({ error: 'Failed to force PDF refresh' }, { status: 500 });
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -80,24 +80,31 @@ export async function getDocumentMetadata(id: string, options?: { signal?: Abort
|
|||
|
||||
export async function getParsedPdfDocument(
|
||||
id: string,
|
||||
options?: { signal?: AbortSignal; retryFailed?: boolean },
|
||||
options?: { signal?: AbortSignal; retryFailed?: boolean; opId?: string },
|
||||
): Promise<
|
||||
| { status: 'ready'; parsed: ParsedPdfDocument }
|
||||
| { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null }
|
||||
| { status: 'pending' | 'running' | 'failed'; parseProgress?: PdfParseProgress | null; opId?: string | null }
|
||||
> {
|
||||
const query = options?.retryFailed ? '?retry=1' : '';
|
||||
const params = new URLSearchParams();
|
||||
if (options?.retryFailed) params.set('retry', '1');
|
||||
if (options?.opId) params.set('opId', options.opId);
|
||||
const query = params.size > 0 ? `?${params.toString()}` : '';
|
||||
const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed${query}`, {
|
||||
signal: options?.signal,
|
||||
cache: 'no-store',
|
||||
});
|
||||
|
||||
if (res.status === 202) {
|
||||
const data = (await res.json().catch(() => null)) as { parseStatus?: string; parseProgress?: PdfParseProgress | null } | null;
|
||||
const data = (await res.json().catch(() => null)) as {
|
||||
parseStatus?: string;
|
||||
parseProgress?: PdfParseProgress | null;
|
||||
opId?: string | null;
|
||||
} | null;
|
||||
const parseStatus = data?.parseStatus;
|
||||
if (parseStatus === 'pending' || parseStatus === 'running' || parseStatus === 'failed') {
|
||||
return { status: parseStatus, parseProgress: data?.parseProgress ?? null };
|
||||
return { status: parseStatus, parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null };
|
||||
}
|
||||
return { status: 'pending', parseProgress: data?.parseProgress ?? null };
|
||||
return { status: 'pending', parseProgress: data?.parseProgress ?? null, opId: data?.opId ?? null };
|
||||
}
|
||||
|
||||
if (!res.ok) {
|
||||
|
|
@ -111,16 +118,30 @@ export async function getParsedPdfDocument(
|
|||
|
||||
export function subscribeParsedPdfDocumentEvents(
|
||||
id: string,
|
||||
options: {
|
||||
opId?: string | null;
|
||||
},
|
||||
handlers: {
|
||||
onSnapshot: (snapshot: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null }) => void;
|
||||
onSnapshot: (snapshot: {
|
||||
parseStatus: PdfParseStatus;
|
||||
parseProgress: PdfParseProgress | null;
|
||||
opId?: string | null;
|
||||
}) => void;
|
||||
onError?: (error: Event) => void;
|
||||
},
|
||||
): () => void {
|
||||
const source = new EventSource(`/api/documents/${encodeURIComponent(id)}/parsed/events`);
|
||||
const params = new URLSearchParams();
|
||||
if (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 };
|
||||
const payload = JSON.parse(event.data) as {
|
||||
parseStatus: PdfParseStatus;
|
||||
parseProgress: PdfParseProgress | null;
|
||||
opId?: string | null;
|
||||
};
|
||||
handlers.onSnapshot(payload);
|
||||
} catch {
|
||||
// Ignore malformed payloads to avoid breaking active streams.
|
||||
|
|
@ -137,20 +158,37 @@ export function subscribeParsedPdfDocumentEvents(
|
|||
export async function forceReparsePdfDocument(
|
||||
id: string,
|
||||
options?: { signal?: AbortSignal },
|
||||
): Promise<'pending' | 'running'> {
|
||||
): Promise<{ status: 'pending' | 'running'; opId?: string | null }> {
|
||||
const res = await fetch(`/api/documents/${encodeURIComponent(id)}/parsed`, {
|
||||
method: 'POST',
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (!res.ok) {
|
||||
const data = (await res.json().catch(() => null)) as { error?: string } | null;
|
||||
throw new Error(data?.error || 'Failed to force PDF reparse');
|
||||
}
|
||||
|
||||
const data = (await res.json().catch(() => null)) as { parseStatus?: string } | null;
|
||||
return data?.parseStatus === 'running' ? 'running' : 'pending';
|
||||
const data = (await res.json().catch(() => null)) as { parseStatus?: string; opId?: string | null } | null;
|
||||
return {
|
||||
status: data?.parseStatus === 'running' ? 'running' : 'pending',
|
||||
opId: data?.opId ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
type DocumentSettingsResponse = {
|
||||
|
|
|
|||
60
src/lib/server/compute/worker-op-create.ts
Normal file
60
src/lib/server/compute/worker-op-create.ts
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
import { getWorkerClientConfigFromEnv, buildPdfOpKey } from '@/lib/server/compute/worker';
|
||||
import type { PdfLayoutInput } from '@/lib/server/compute/types';
|
||||
import type {
|
||||
PdfLayoutJobResult,
|
||||
WorkerOperationState,
|
||||
} from '@openreader/compute-core/api-contracts';
|
||||
|
||||
type CreatePdfWorkerOpInput =
|
||||
Pick<PdfLayoutInput, 'documentId' | 'namespace' | 'forceToken'>
|
||||
& { documentObjectKey: string };
|
||||
|
||||
const CREATE_OP_TIMEOUT_MS = 10_000;
|
||||
|
||||
export async function createOrReusePdfWorkerOperation(
|
||||
input: CreatePdfWorkerOpInput,
|
||||
): Promise<WorkerOperationState<PdfLayoutJobResult>> {
|
||||
const cfg = getWorkerClientConfigFromEnv();
|
||||
const opKey = buildPdfOpKey({
|
||||
documentId: input.documentId,
|
||||
namespace: input.namespace,
|
||||
documentObjectKey: input.documentObjectKey,
|
||||
forceToken: input.forceToken,
|
||||
});
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), CREATE_OP_TIMEOUT_MS);
|
||||
try {
|
||||
const res = await fetch(`${cfg.baseUrl}/ops`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
Authorization: `Bearer ${cfg.token}`,
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({
|
||||
kind: 'pdf_layout',
|
||||
opKey,
|
||||
payload: {
|
||||
documentId: input.documentId,
|
||||
namespace: input.namespace,
|
||||
documentObjectKey: input.documentObjectKey,
|
||||
},
|
||||
}),
|
||||
cache: 'no-store',
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
if (!res.ok) {
|
||||
const detail = await res.text().catch(() => '');
|
||||
throw new Error(`Worker op create failed: ${res.status}${detail ? ` ${detail}` : ''}`);
|
||||
}
|
||||
|
||||
const parsed = await res.json() as WorkerOperationState<PdfLayoutJobResult>;
|
||||
if (!parsed || typeof parsed !== 'object' || typeof parsed.opId !== 'string' || !parsed.opId.trim()) {
|
||||
throw new Error('Worker op create returned invalid response');
|
||||
}
|
||||
return parsed;
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue