diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 39738f2..d48771c 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -3,7 +3,6 @@ import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; @@ -96,13 +95,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } const initial = await toSnapshot(row); - if (initial.parseStatus === 'pending') { - enqueueParsePdfJob({ - documentId: id, - userId: row.userId, - namespace: testNamespace, - }); - } const encoder = new TextEncoder(); @@ -132,13 +124,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string if (!nextRow) break; const next = await toSnapshot(nextRow); - if (next.parseStatus === 'pending') { - enqueueParsePdfJob({ - documentId: id, - userId: nextRow.userId, - namespace: testNamespace, - }); - } const nextSignature = JSON.stringify(next); if (nextSignature !== signature) { diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 98b5c25..57bc25f 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -102,13 +102,6 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string } if (effectiveStatus !== 'ready') { - if (effectiveStatus === 'pending') { - enqueueParsePdfJob({ - documentId: id, - userId: row.userId, - namespace: testNamespace, - }); - } return NextResponse.json({ parseStatus: effectiveStatus, parseProgress: state.progress ?? null, diff --git a/src/app/api/documents/docx-to-pdf/upload/route.ts b/src/app/api/documents/docx-to-pdf/upload/route.ts index 4a65228..2aeeb7e 100644 --- a/src/app/api/documents/docx-to-pdf/upload/route.ts +++ b/src/app/api/documents/docx-to-pdf/upload/route.ts @@ -10,6 +10,8 @@ import { db } from '@/db'; import { documents } from '@/db/schema'; import { safeDocumentName } from '@/lib/server/documents/utils'; import { enqueueDocumentPreview } from '@/lib/server/documents/previews'; +import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; +import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state'; import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; import { putDocumentBlob } from '@/lib/server/documents/blobstore'; @@ -136,6 +138,8 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, + parseState: stringifyDocumentParseState({ status: 'pending', progress: null, updatedAt: Date.now() }), + parsedJsonKey: null, }) .onConflictDoUpdate({ target: [documents.id, documents.userId], @@ -145,6 +149,8 @@ export async function POST(req: NextRequest) { size: pdfContent.length, lastModified, filePath: id, + parseState: stringifyDocumentParseState({ status: 'pending', progress: null, updatedAt: Date.now() }), + parsedJsonKey: null, }, }); @@ -159,6 +165,12 @@ export async function POST(req: NextRequest) { console.error(`Failed to enqueue preview for converted DOCX ${id}:`, error); }); + enqueueParsePdfJob({ + documentId: id, + userId: storageUserId, + namespace: testNamespace, + }); + return NextResponse.json({ stored: { id, diff --git a/src/lib/server/jobs/user-pdf-layout-job.ts b/src/lib/server/jobs/user-pdf-layout-job.ts index 141d029..deec00e 100644 --- a/src/lib/server/jobs/user-pdf-layout-job.ts +++ b/src/lib/server/jobs/user-pdf-layout-job.ts @@ -1,10 +1,12 @@ -import { and, eq } from 'drizzle-orm'; +import { randomUUID } from 'node:crypto'; +import { and, eq, inArray, isNull } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { documentKey, putParsedDocumentBlob } from '@/lib/server/documents/blobstore'; -import { stringifyDocumentParseState } from '@/lib/server/documents/parse-state'; +import { parseDocumentParseState, stringifyDocumentParseState } 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 { PdfLayoutJobBase, PdfLayoutProgress } from '@openreader/compute-core/api-contracts'; type UserPdfLayoutJobRequest = PdfLayoutJobBase & { @@ -14,45 +16,231 @@ type UserPdfLayoutJobRequest = PdfLayoutJobBase & { const running = new Set(); +const FOLLOWER_WAIT_TIMEOUT_MS = 180_000; +const FOLLOWER_POLL_MS = 1_200; + +type ParseRow = { + userId: string; + parseState: string | null; + parsedJsonKey: string | null; +}; + function keyFor(input: UserPdfLayoutJobRequest): string { - return `${input.userId}:${input.documentId}:${input.namespace || ''}`; + const forceToken = input.forceToken?.trim(); + if (forceToken) { + return `force:${input.documentId}:${input.namespace || ''}:${forceToken}`; + } + return `shared:${input.documentId}:${input.namespace || ''}`; +} + +function sleep(ms: number): Promise { + 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 { + 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 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 updateParseStateForUsers(input: { + documentId: string; + userIds: string[]; + parseState: string; + parsedJsonKey?: string | null; +}): Promise { + 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 { + const deadline = Date.now() + FOLLOWER_WAIT_TIMEOUT_MS; + + while (Date.now() < deadline) { + const rows = await loadScopedRows(input); + const ready = rows.find(isReadyRow); + if (ready) { + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: [input.userId], + parseState: stringifyDocumentParseState({ + status: 'ready', + progress: null, + updatedAt: Date.now(), + }), + parsedJsonKey: ready.parsedJsonKey, + }); + return; + } + + const statuses = rows.map((row) => 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) { + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: [input.userId], + parseState: stringifyDocumentParseState({ + status: 'failed', + progress: null, + updatedAt: Date.now(), + ...(failedState.error ? { error: failedState.error } : {}), + }), + }); + return; + } + + await sleep(FOLLOWER_POLL_MS); + } } export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise { const key = keyFor(input); - if (running.has(key)) return; + if (running.has(key)) { + if (!input.forceToken?.trim()) { + await syncCallerToSharedResult(input); + } + return; + } running.add(key); try { - const now = Date.now(); - await db + 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 ready = scopedRows.find(isReadyRow); + if (ready) { + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: [input.userId], + parseState: stringifyDocumentParseState({ + status: 'ready', + progress: null, + updatedAt: Date.now(), + }), + parsedJsonKey: ready.parsedJsonKey, + }); + return; + } + } + + const coordinator = [...scopedRows].sort((a, b) => a.userId.localeCompare(b.userId))[0]; + if (!coordinator) return; + + const claimOpId = randomUUID(); + const runningState = stringifyDocumentParseState({ + status: 'running', + progress: null, + updatedAt: Date.now(), + opId: claimOpId, + }); + + const claimRows = (await db .update(documents) .set({ - parseState: stringifyDocumentParseState({ - status: 'running', - progress: null, - updatedAt: now, - }), + parseState: runningState, }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); + .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: runningState, + }); const compute = await getCompute(); const writeProgress = async (progress: PdfLayoutProgress): Promise => { - await db - .update(documents) - .set({ - parseState: stringifyDocumentParseState({ - status: 'running', - progress: { - totalPages: progress.totalPages, - pagesParsed: progress.pagesParsed, - currentPage: progress.currentPage, - phase: progress.phase, - }, - updatedAt: Date.now(), - }), - }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: cohortUserIds, + parseState: stringifyDocumentParseState({ + status: 'running', + progress: { + totalPages: progress.totalPages, + pagesParsed: progress.pagesParsed, + currentPage: progress.currentPage, + phase: progress.phase, + }, + updatedAt: Date.now(), + opId: claimOpId, + }), + }); }; const layout = await compute.parsePdfLayout({ documentId: input.documentId, @@ -69,53 +257,59 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise parsedJsonKey = await putParsedDocumentBlob(input.documentId, parsedJson, input.namespace); } - await db - .update(documents) - .set({ - parseState: stringifyDocumentParseState({ - status: 'ready', - progress: null, - updatedAt: Date.now(), - }), - parsedJsonKey, - }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); + const finalScopedRows = await loadScopedRows(input); + const finalUserIds = userIdsFromRows(finalScopedRows); + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: finalUserIds.length > 0 ? finalUserIds : cohortUserIds, + parseState: stringifyDocumentParseState({ + status: 'ready', + progress: null, + updatedAt: Date.now(), + }), + parsedJsonKey, + }); // Best-effort cache invalidation should not block parse readiness. - void clearTtsSegmentCache({ - userId: input.userId, - documentId: input.documentId, - readerType: 'pdf', - }).then((cleared) => { - if (cleared.warning) { - console.warn('[parsePdfJob] cache invalidation warning', { - documentId: input.documentId, - warning: cleared.warning, - }); - } - }).catch((cacheError) => { - console.warn('[parsePdfJob] cache invalidation failed', { + for (const userId of (finalUserIds.length > 0 ? finalUserIds : cohortUserIds)) { + void clearTtsSegmentCache({ + userId, documentId: input.documentId, - error: cacheError instanceof Error ? cacheError.message : String(cacheError), + readerType: 'pdf', + }).then((cleared) => { + if (cleared.warning) { + console.warn('[parsePdfJob] cache invalidation warning', { + documentId: input.documentId, + userId, + warning: cleared.warning, + }); + } + }).catch((cacheError) => { + console.warn('[parsePdfJob] cache invalidation failed', { + documentId: input.documentId, + userId, + error: cacheError instanceof Error ? cacheError.message : String(cacheError), + }); }); - }); + } } catch (error) { const message = error instanceof Error ? error.message : String(error); const stack = error instanceof Error ? error.stack : undefined; const cause = error instanceof Error ? error.cause : undefined; const parseStatus = 'failed'; try { - await db - .update(documents) - .set({ - parseState: stringifyDocumentParseState({ - status: 'failed', - progress: null, - updatedAt: Date.now(), - error: message, - }), - }) - .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId))); + const scopedRows = await loadScopedRows(input); + const scopedUserIds = userIdsFromRows(scopedRows); + await updateParseStateForUsers({ + documentId: input.documentId, + userIds: scopedUserIds.length > 0 ? scopedUserIds : [input.userId], + parseState: stringifyDocumentParseState({ + status: 'failed', + progress: null, + updatedAt: Date.now(), + error: message, + }), + }); } catch (statusError) { console.error('[parsePdfJob] failed to write parse status', { documentId: input.documentId,