pdf parse: dedupe shared jobs and stop enqueue-on-read

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) {

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,

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,

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<string>();
 
+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<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 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<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 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<void> {
   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();
+    const scopedRows = await loadScopedRows(input);
-    await db
+    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({
+        parseState: runningState,
-          status: 'running',
-          progress: null,
-          updatedAt: now,
-        }),
       })
-      .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<void> => {
-      await db
+      await updateParseStateForUsers({
-        .update(documents)
+        documentId: input.documentId,
-        .set({
+        userIds: cohortUserIds,
-          parseState: stringifyDocumentParseState({
+        parseState: stringifyDocumentParseState({
-            status: 'running',
+          status: 'running',
-            progress: {
+          progress: {
-              totalPages: progress.totalPages,
+            totalPages: progress.totalPages,
-              pagesParsed: progress.pagesParsed,
+            pagesParsed: progress.pagesParsed,
-              currentPage: progress.currentPage,
+            currentPage: progress.currentPage,
-              phase: progress.phase,
+            phase: progress.phase,
-            },
+          },
-            updatedAt: Date.now(),
+          updatedAt: Date.now(),
-          }),
+          opId: claimOpId,
-        })
+        }),
-        .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
+      });
     };
     const layout = await compute.parsePdfLayout({
       documentId: input.documentId,
@@ -69,53 +257,59 @@ export async function parsePdfJob(input: UserPdfLayoutJobRequest): Promise<void>
       parsedJsonKey = await putParsedDocumentBlob(input.documentId, parsedJson, input.namespace);
     }
 
-    await db
+    const finalScopedRows = await loadScopedRows(input);
-      .update(documents)
+    const finalUserIds = userIdsFromRows(finalScopedRows);
-      .set({
+    await updateParseStateForUsers({
-        parseState: stringifyDocumentParseState({
+      documentId: input.documentId,
-          status: 'ready',
+      userIds: finalUserIds.length > 0 ? finalUserIds : cohortUserIds,
-          progress: null,
+      parseState: stringifyDocumentParseState({
-          updatedAt: Date.now(),
+        status: 'ready',
-        }),
+        progress: null,
-        parsedJsonKey,
+        updatedAt: Date.now(),
-      })
+      }),
-      .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
+      parsedJsonKey,
+    });
 
     // Best-effort cache invalidation should not block parse readiness.
-    void clearTtsSegmentCache({
+    for (const userId of (finalUserIds.length > 0 ? finalUserIds : cohortUserIds)) {
-      userId: input.userId,
+      void clearTtsSegmentCache({
-      documentId: input.documentId,
+        userId,
-      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', {
         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
+      const scopedRows = await loadScopedRows(input);
-        .update(documents)
+      const scopedUserIds = userIdsFromRows(scopedRows);
-        .set({
+      await updateParseStateForUsers({
-          parseState: stringifyDocumentParseState({
+        documentId: input.documentId,
-            status: 'failed',
+        userIds: scopedUserIds.length > 0 ? scopedUserIds : [input.userId],
-            progress: null,
+        parseState: stringifyDocumentParseState({
-            updatedAt: Date.now(),
+          status: 'failed',
-            error: message,
+          progress: null,
-          }),
+          updatedAt: Date.now(),
-        })
+          error: message,
-        .where(and(eq(documents.id, input.documentId), eq(documents.userId, input.userId)));
+        }),
+      });
     } catch (statusError) {
       console.error('[parsePdfJob] failed to write parse status', {
         documentId: input.documentId,