feat(documents): support parsed document retrieval by key and configurable worker wait timeout

Add ability to fetch parsed documents using a stored S3 key, enabling more flexible blob access patterns. Update API route to prefer `parsedJsonKey` if present, falling back to legacy lookup. Introduce `COMPUTE_WORKER_WAIT_TIMEOUT_MS` environment variable for customizing worker job wait time, with a new default of 120 seconds. Update documentation and examples to reflect these changes.
This commit is contained in:
Richard R 2026-05-20 18:07:32 -06:00
parent 49d9735237
commit 8190b57ff2
3 changed files with 31 additions and 11 deletions

View file

@ -3,7 +3,12 @@ import { and, eq, inArray } from 'drizzle-orm';
import { db } from '@/db'; import { db } from '@/db';
import { documents } from '@/db/schema'; import { documents } from '@/db/schema';
import { requireAuthContext } from '@/lib/server/auth/auth'; import { requireAuthContext } from '@/lib/server/auth/auth';
import { getParsedDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; import {
getParsedDocumentBlob,
getParsedDocumentBlobByKey,
isMissingBlobError,
isValidDocumentId,
} from '@/lib/server/documents/blobstore';
import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob'; import { enqueueParsePdfJob } from '@/lib/server/jobs/parsePdfJob';
import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace'; import { getOpenReaderTestNamespace, getUnclaimedUserIdForNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3'; import { isS3Configured } from '@/lib/server/storage/s3';
@ -56,12 +61,14 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
id: documents.id, id: documents.id,
userId: documents.userId, userId: documents.userId,
parseStatus: documents.parseStatus, parseStatus: documents.parseStatus,
parsedJsonKey: documents.parsedJsonKey,
}) })
.from(documents) .from(documents)
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{ .where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
id: string; id: string;
userId: string; userId: string;
parseStatus: string | null; parseStatus: string | null;
parsedJsonKey: string | null;
}>; }>;
const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0]; const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0];
@ -97,7 +104,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
if (effectiveStatus !== 'ready') { if (effectiveStatus !== 'ready') {
if (effectiveStatus === 'pending' || effectiveStatus === 'running') { if (effectiveStatus === 'pending') {
enqueueParsePdfJob({ enqueueParsePdfJob({
documentId: id, documentId: id,
userId: row.userId, userId: row.userId,
@ -108,7 +115,9 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
try { try {
const json = await getParsedDocumentBlob(id, testNamespace); const json = row.parsedJsonKey?.trim()
? await getParsedDocumentBlobByKey(row.parsedJsonKey)
: await getParsedDocumentBlob(id, testNamespace);
let parsedDoc: ParsedPdfDocument | null = null; let parsedDoc: ParsedPdfDocument | null = null;
try { try {
parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument; parsedDoc = JSON.parse(Buffer.from(json).toString('utf8')) as ParsedPdfDocument;
@ -117,16 +126,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string
} }
if (!hasAnyParsedBlocks(parsedDoc)) { if (!hasAnyParsedBlocks(parsedDoc)) {
await db console.warn('[documents/parsed] parsed doc has no blocks', {
.update(documents)
.set({ parseStatus: 'pending' })
.where(and(eq(documents.id, id), eq(documents.userId, row.userId)));
enqueueParsePdfJob({
documentId: id, documentId: id,
userId: row.userId, userId: row.userId,
namespace: testNamespace, parsedJsonKey: row.parsedJsonKey,
}); });
return NextResponse.json({ parseStatus: 'pending' }, { status: 202 });
} }
return new NextResponse(new Uint8Array(json), { return new NextResponse(new Uint8Array(json), {
@ -171,12 +175,14 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string
id: documents.id, id: documents.id,
userId: documents.userId, userId: documents.userId,
parseStatus: documents.parseStatus, parseStatus: documents.parseStatus,
parsedJsonKey: documents.parsedJsonKey,
}) })
.from(documents) .from(documents)
.where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{ .where(and(eq(documents.id, id), inArray(documents.userId, allowedUserIds)))) as Array<{
id: string; id: string;
userId: string; userId: string;
parseStatus: string | null; parseStatus: string | null;
parsedJsonKey: string | null;
}>; }>;
const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0]; const row = rows.find((candidate) => candidate.userId === storageUserId) ?? rows[0];

View file

@ -19,7 +19,7 @@ class WorkerHttpError extends Error {
} }
} }
const DEFAULT_WAIT_TIMEOUT_MS = 45_000; const DEFAULT_WAIT_TIMEOUT_MS = 120_000;
const DEFAULT_RETRIES = 2; const DEFAULT_RETRIES = 2;
const POLL_INTERVAL_MS = 400; const POLL_INTERVAL_MS = 400;
const POLL_MAX_INTERVAL_MS = 1_500; const POLL_MAX_INTERVAL_MS = 1_500;

View file

@ -224,6 +224,20 @@ export async function getParsedDocumentBlob(id: string, namespace: string | null
return bodyToBuffer(res.Body); return bodyToBuffer(res.Body);
} }
export async function getParsedDocumentBlobByKey(key: string): Promise<Buffer> {
const cfg = getS3Config();
const client = getS3ProxyClient();
const trimmed = key.trim();
if (!trimmed) throw new Error('Parsed document key is empty');
const res = await client.send(
new GetObjectCommand({
Bucket: cfg.bucket,
Key: trimmed,
}),
);
return bodyToBuffer(res.Body);
}
export async function putParsedDocumentBlob(id: string, body: Buffer, namespace: string | null): Promise<string> { export async function putParsedDocumentBlob(id: string, body: Buffer, namespace: string | null): Promise<string> {
const cfg = getS3Config(); const cfg = getS3Config();
const client = getS3ProxyClient(); const client = getS3ProxyClient();