diff --git a/src/app/api/documents/blob/upload/fallback/route.ts b/src/app/api/documents/blob/upload/fallback/route.ts index 7fc8208..f6d7f15 100644 --- a/src/app/api/documents/blob/upload/fallback/route.ts +++ b/src/app/api/documents/blob/upload/fallback/route.ts @@ -1,6 +1,6 @@ import { NextRequest, NextResponse } from 'next/server'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { isValidDocumentId, putDocumentBlob } from '@/lib/server/documents/blobstore'; +import { isValidTempUploadToken, putTempDocumentBlob } from '@/lib/server/documents/blobstore'; import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; import { isS3Configured } from '@/lib/server/storage/s3'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; @@ -26,11 +26,12 @@ export async function PUT(req: NextRequest) { const ctxOrRes = await requireAuthContext(req); if (ctxOrRes instanceof Response) return ctxOrRes; + if (!ctxOrRes.userId) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); const url = new URL(req.url); - const id = (url.searchParams.get('id') || '').trim().toLowerCase(); - if (!isValidDocumentId(id)) { - return NextResponse.json({ error: 'Invalid document id' }, { status: 400 }); + const token = (url.searchParams.get('token') || '').trim().toLowerCase(); + if (!isValidTempUploadToken(token)) { + return NextResponse.json({ error: 'Invalid upload token' }, { status: 400 }); } const contentType = (req.headers.get('content-type') || 'application/octet-stream').trim() || 'application/octet-stream'; @@ -86,7 +87,7 @@ export async function PUT(req: NextRequest) { const namespace = getOpenReaderTestNamespace(req.headers); try { - await putDocumentBlob(id, body, contentType, namespace); + await putTempDocumentBlob(token, ctxOrRes.userId, body, contentType, namespace); } catch (error) { if (!isPreconditionFailed(error)) { throw error; @@ -97,12 +98,12 @@ export async function PUT(req: NextRequest) { event: 'documents.blob.upload.fallback.proxy_used', degraded: true, fallbackPath: 'upload_proxy', - documentId: id, + uploadToken: token, contentType, bytes: body.byteLength, }, 'Document upload fallback proxy used'); - return NextResponse.json({ success: true, id }); + return NextResponse.json({ success: true, token }); } catch (error) { serverLogger.error({ event: 'documents.blob.upload.fallback.failed', diff --git a/src/app/api/documents/blob/upload/finalize/route.ts b/src/app/api/documents/blob/upload/finalize/route.ts new file mode 100644 index 0000000..ea82ce4 --- /dev/null +++ b/src/app/api/documents/blob/upload/finalize/route.ts @@ -0,0 +1,224 @@ +import { createHash } from 'node:crypto'; +import { NextRequest, NextResponse } from 'next/server'; +import { requireAuthContext } from '@/lib/server/auth/auth'; +import { + TEMP_DOCUMENT_UPLOAD_TTL_MS, + copyTempDocumentBlobToDocument, + deleteTempDocumentUpload, + getTempDocumentBlob, + getTempDocumentFinalizeReceipt, + headDocumentBlob, + headTempDocumentBlob, + isMissingBlobError, + isValidTempUploadToken, + putTempDocumentFinalizeReceipt, +} from '@/lib/server/documents/blobstore'; +import { registerUploadedDocument } from '@/lib/server/documents/register-upload'; +import { safeDocumentName, toDocumentTypeFromName } from '@/lib/server/documents/utils'; +import { errorResponse } from '@/lib/server/errors/next-response'; +import { errorToLog, serverLogger } from '@/lib/server/logger'; +import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; +import { isS3Configured } from '@/lib/server/storage/s3'; +import type { BaseDocument, DocumentType } from '@/types/documents'; + +export const dynamic = 'force-dynamic'; + +type FinalizeUpload = { + token: string; + name: string; + type: DocumentType; + lastModified: number; +}; + +type FinalizeReceipt = { + stored: BaseDocument; +}; + +function s3NotConfiguredResponse(): NextResponse { + return NextResponse.json( + { error: 'Documents storage is not configured. Set S3_* environment variables.' }, + { status: 503 }, + ); +} + +function normalizeDocumentType(rawType: unknown, safeName: string): DocumentType { + if (rawType === 'pdf' || rawType === 'epub' || rawType === 'docx' || rawType === 'html') { + return rawType; + } + return toDocumentTypeFromName(safeName); +} + +function normalizeLastModified(value: unknown): number { + return Number.isFinite(value) && Number(value) > 0 ? Number(value) : Date.now(); +} + +function parseFinalizePayload(body: unknown): FinalizeUpload[] { + if (!body || typeof body !== 'object') return []; + const rawUploads = (body as { uploads?: unknown }).uploads; + if (!Array.isArray(rawUploads)) return []; + + const uploads: FinalizeUpload[] = []; + for (const rawUpload of rawUploads) { + if (!rawUpload || typeof rawUpload !== 'object') continue; + const rec = rawUpload as Record; + const token = typeof rec.token === 'string' ? rec.token.trim().toLowerCase() : ''; + if (!isValidTempUploadToken(token)) continue; + const fallbackName = `upload-${token}.txt`; + const name = safeDocumentName(typeof rec.name === 'string' ? rec.name : '', fallbackName); + uploads.push({ + token, + name, + type: normalizeDocumentType(rec.type, name), + lastModified: normalizeLastModified(rec.lastModified), + }); + } + return uploads; +} + +async function loadTempUpload(input: { + token: string; + userId: string; + namespace: string | null; +}): Promise<{ contentType: string; size: number; lastModified: number; body: Buffer }> { + const RETRIES = 3; + const RETRY_DELAY_MS = 500; + + let lastError: unknown = null; + for (let attempt = 0; attempt < RETRIES; attempt += 1) { + try { + const head = await headTempDocumentBlob(input.token, input.userId, input.namespace); + const body = await getTempDocumentBlob(input.token, input.userId, input.namespace); + return { + contentType: head.contentType ?? 'application/octet-stream', + size: head.contentLength > 0 ? head.contentLength : body.byteLength, + lastModified: head.lastModified ?? Date.now(), + body, + }; + } catch (error) { + lastError = error; + if (isMissingBlobError(error) && attempt < RETRIES - 1) { + await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY_MS)); + continue; + } + throw error; + } + } + + throw lastError instanceof Error ? lastError : new Error('Temporary upload is unavailable'); +} + +async function finalizeOne(input: { + upload: FinalizeUpload; + userId: string; + namespace: string | null; +}): Promise { + const existingReceipt = await getTempDocumentFinalizeReceipt( + input.upload.token, + input.userId, + input.namespace, + ); + if (existingReceipt?.stored) { + return existingReceipt.stored; + } + + const temp = await loadTempUpload({ + token: input.upload.token, + userId: input.userId, + namespace: input.namespace, + }); + + if (Date.now() - temp.lastModified > TEMP_DOCUMENT_UPLOAD_TTL_MS) { + await deleteTempDocumentUpload(input.upload.token, input.userId, input.namespace).catch(() => undefined); + throw new Error('Temporary upload expired before finalize'); + } + + const documentId = createHash('sha256').update(temp.body).digest('hex'); + + try { + await headDocumentBlob(documentId, input.namespace); + } catch (error) { + if (!isMissingBlobError(error)) throw error; + await copyTempDocumentBlobToDocument( + input.upload.token, + input.userId, + documentId, + input.namespace, + temp.contentType, + ); + } + + const canonicalHead = await headDocumentBlob(documentId, input.namespace); + const stored = await registerUploadedDocument({ + documentId, + userId: input.userId, + namespace: input.namespace, + name: input.upload.name, + type: input.upload.type, + size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : temp.size, + lastModified: input.upload.lastModified, + }); + + await putTempDocumentFinalizeReceipt( + input.upload.token, + input.userId, + input.namespace, + Buffer.from(JSON.stringify({ stored }), 'utf8'), + ); + + await deleteTempDocumentUpload(input.upload.token, input.userId, input.namespace).catch((error) => { + serverLogger.warn({ + event: 'documents.blob.upload.finalize.temp_delete_failed', + degraded: true, + fallbackPath: 'leave_temp_upload', + documentId, + token: input.upload.token, + error: errorToLog(error), + }, 'Failed to delete temp upload after finalize'); + }); + + return stored; +} + +export async function POST(req: NextRequest) { + try { + if (!isS3Configured()) return s3NotConfiguredResponse(); + + const ctxOrRes = await requireAuthContext(req); + if (ctxOrRes instanceof Response) return ctxOrRes; + if (!ctxOrRes.userId) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); + const userId = ctxOrRes.userId; + + const namespace = getOpenReaderTestNamespace(req.headers); + const uploads = parseFinalizePayload(await req.json().catch(() => null)); + if (uploads.length === 0) { + return NextResponse.json({ error: 'No valid uploads provided' }, { status: 400 }); + } + + const stored: BaseDocument[] = []; + for (const upload of uploads) { + stored.push(await finalizeOne({ + upload, + userId, + namespace, + })); + } + + return NextResponse.json({ stored }); + } catch (error) { + if (error instanceof Error && error.message === 'Temporary upload expired before finalize') { + return NextResponse.json({ error: error.message }, { status: 410 }); + } + if (isMissingBlobError(error)) { + return NextResponse.json({ error: 'Temporary upload missing. Upload bytes again and retry finalize.' }, { status: 409 }); + } + + serverLogger.error({ + event: 'documents.blob.upload.finalize.failed', + error: errorToLog(error), + }, 'Failed to finalize uploaded documents'); + return errorResponse(error, { + apiErrorMessage: 'Failed to finalize uploaded documents', + normalize: { code: 'DOCUMENTS_BLOB_UPLOAD_FINALIZE_FAILED', errorClass: 'storage' }, + }); + } +} diff --git a/src/app/api/documents/blob/upload/presign/route.ts b/src/app/api/documents/blob/upload/presign/route.ts index 6d91ee4..183074c 100644 --- a/src/app/api/documents/blob/upload/presign/route.ts +++ b/src/app/api/documents/blob/upload/presign/route.ts @@ -1,6 +1,11 @@ +import { randomUUID } from 'node:crypto'; import { NextRequest, NextResponse } from 'next/server'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { isValidDocumentId, presignPut } from '@/lib/server/documents/blobstore'; +import { + TEMP_DOCUMENT_UPLOAD_TTL_MS, + deleteExpiredTempDocumentUploads, + presignTempPut, +} from '@/lib/server/documents/blobstore'; import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; @@ -10,7 +15,6 @@ import { errorResponse } from '@/lib/server/errors/next-response'; export const dynamic = 'force-dynamic'; type PresignUpload = { - id: string; contentType: string; size: number; }; @@ -24,14 +28,12 @@ function parseUploads(body: unknown): PresignUpload[] { for (const raw of rawUploads) { if (!raw || typeof raw !== 'object') continue; const rec = raw as Record; - const id = typeof rec.id === 'string' ? rec.id.trim().toLowerCase() : ''; - if (!isValidDocumentId(id)) continue; const contentType = typeof rec.contentType === 'string' && rec.contentType.trim() ? rec.contentType.trim() : 'application/octet-stream'; const size = Number.isFinite(rec.size) && Number(rec.size) >= 0 ? Number(rec.size) : 0; - uploads.push({ id, contentType, size }); + uploads.push({ contentType, size }); } return uploads; } @@ -47,6 +49,8 @@ export async function POST(req: NextRequest) { const ctxOrRes = await requireAuthContext(req); if (ctxOrRes instanceof Response) return ctxOrRes; + if (!ctxOrRes.userId) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); + const userId = ctxOrRes.userId; const body = await req.json().catch(() => null); const uploads = parseUploads(body); @@ -68,13 +72,16 @@ export async function POST(req: NextRequest) { } const namespace = getOpenReaderTestNamespace(req.headers); + await deleteExpiredTempDocumentUploads(userId, namespace, Date.now() - TEMP_DOCUMENT_UPLOAD_TTL_MS) + .catch(() => undefined); const signed = await Promise.all( uploads.map(async (upload) => { - const res = await presignPut(upload.id, upload.contentType, namespace, { + const token = randomUUID(); + const res = await presignTempPut(token, userId, upload.contentType, namespace, { contentLength: upload.size, }); return { - id: upload.id, + token, url: res.url, headers: res.headers, }; diff --git a/src/app/api/documents/route.ts b/src/app/api/documents/route.ts index 7ea9f40..8002f7a 100644 --- a/src/app/api/documents/route.ts +++ b/src/app/api/documents/route.ts @@ -1,43 +1,27 @@ -import { randomUUID } from 'node:crypto'; import { NextRequest, NextResponse } from 'next/server'; import { and, count, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; -import { safeDocumentName, toDocumentTypeFromName } from '@/lib/server/documents/utils'; +import { toDocumentTypeFromName } from '@/lib/server/documents/utils'; import { errorToLog, serverLogger } from '@/lib/server/logger'; import { errorResponse } from '@/lib/server/errors/next-response'; import { cleanupDocumentPreviewArtifacts, deleteDocumentPreviewRows, - enqueueDocumentPreview, } from '@/lib/server/documents/previews'; -import { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; -import { recordJobEvent, getPdfLayoutRateConfig } from '@/lib/server/rate-limit/job-rate-limiter'; -import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; -import { deleteDocumentBlob, headDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; +import { deleteDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; import { normalizeDocumentParseStateForCurrentParserVersion, normalizeParseStatus, parseDocumentParseState, - stringifyDocumentParseState, } from '@/lib/server/documents/parse-state'; -import { findReusableParsedPdfResult } from '@/lib/server/documents/parsed-pdf-reuse'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; -import { PDF_PARSER_VERSION } from '@openreader/compute-core'; import type { BaseDocument, DocumentType } from '@/types/documents'; export const dynamic = 'force-dynamic'; -type RegisterDocument = { - id: string; - name: string; - type: DocumentType; - size: number; - lastModified: number; -}; - function s3NotConfiguredResponse(): NextResponse { return NextResponse.json( { error: 'Documents storage is not configured. Set S3_* environment variables.' }, @@ -52,186 +36,6 @@ function normalizeDocumentType(rawType: unknown, safeName: string): DocumentType return toDocumentTypeFromName(safeName); } -function normalizeLastModified(value: unknown): number { - return Number.isFinite(value) && Number(value) > 0 ? Number(value) : Date.now(); -} - -function parseDocumentPayload(body: unknown): RegisterDocument[] { - if (!body || typeof body !== 'object') return []; - const rawDocs = (body as { documents?: unknown }).documents; - if (!Array.isArray(rawDocs)) return []; - - const docs: RegisterDocument[] = []; - for (const rawDoc of rawDocs) { - if (!rawDoc || typeof rawDoc !== 'object') continue; - const rec = rawDoc as Record; - const id = typeof rec.id === 'string' ? rec.id.trim().toLowerCase() : ''; - if (!isValidDocumentId(id)) continue; - const fallbackName = `${id}.${typeof rec.type === 'string' ? rec.type : 'txt'}`; - const name = safeDocumentName(typeof rec.name === 'string' ? rec.name : '', fallbackName); - const type = normalizeDocumentType(rec.type, name); - const lastModified = normalizeLastModified(rec.lastModified); - const size = Number.isFinite(rec.size) && Number(rec.size) >= 0 ? Number(rec.size) : 0; - docs.push({ id, name, type, size, lastModified }); - } - return docs; -} - -export async function POST(req: NextRequest) { - try { - if (!isS3Configured()) return s3NotConfiguredResponse(); - - const ctxOrRes = await requireAuthContext(req); - if (ctxOrRes instanceof Response) return ctxOrRes; - if (!ctxOrRes.userId) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }); - - const testNamespace = getOpenReaderTestNamespace(req.headers); - const storageUserId = ctxOrRes.userId; - - const body = await req.json().catch(() => null); - const documentsData = parseDocumentPayload(body); - if (documentsData.length === 0) { - return NextResponse.json({ error: 'No valid documents provided' }, { status: 400 }); - } - - const stored: BaseDocument[] = []; - - // Resolve the parse rate-limit config once (only when a PDF is present). - const pdfRateConfig = documentsData.some((doc) => doc.type === 'pdf') - ? getPdfLayoutRateConfig(await getResolvedRuntimeConfig()) - : null; - - for (const doc of documentsData) { - let headSize = doc.size; - const reusableParsedPdf = doc.type === 'pdf' - ? await findReusableParsedPdfResult(doc.id) - : null; - - // Retry HEAD check to handle S3 read-after-write propagation delays. - // The client uploads bytes directly to S3 via presigned URL, then - // immediately calls this endpoint. On serverless platforms the HEAD - // request may reach S3 before the PUT is visible. - const HEAD_RETRIES = 3; - const HEAD_RETRY_DELAY_MS = 500; - let headError: unknown = null; - for (let attempt = 0; attempt < HEAD_RETRIES; attempt++) { - headError = null; - try { - const head = await headDocumentBlob(doc.id, testNamespace); - if (head.contentLength > 0) headSize = head.contentLength; - break; - } catch (error) { - if (isMissingBlobError(error)) { - headError = error; - if (attempt < HEAD_RETRIES - 1) { - await new Promise((r) => setTimeout(r, HEAD_RETRY_DELAY_MS)); - continue; - } - } else { - throw error; - } - } - } - if (headError && isMissingBlobError(headError)) { - return NextResponse.json( - { - error: `Blob missing for document ${doc.id}. Upload bytes first using /api/documents/blob/upload/presign.`, - }, - { status: 409 }, - ); - } - - await db - .insert(documents) - .values({ - id: doc.id, - userId: storageUserId, - name: doc.name, - type: doc.type, - size: headSize, - lastModified: doc.lastModified, - filePath: doc.id, - parseState: doc.type === 'pdf' - ? stringifyDocumentParseState( - reusableParsedPdf - ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } - : { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }, - ) - : null, - parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, - }) - .onConflictDoUpdate({ - target: [documents.id, documents.userId], - set: { - name: doc.name, - type: doc.type, - size: headSize, - lastModified: doc.lastModified, - filePath: doc.id, - parseState: doc.type === 'pdf' - ? stringifyDocumentParseState( - reusableParsedPdf - ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } - : { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }, - ) - : null, - parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, - }, - }); - - stored.push({ - id: doc.id, - name: doc.name, - type: doc.type, - size: headSize, - lastModified: doc.lastModified, - scope: 'user', - }); - - await enqueueDocumentPreview( - { - id: doc.id, - type: doc.type, - lastModified: doc.lastModified, - }, - testNamespace, - ).catch((error) => { - serverLogger.warn({ - event: 'documents.preview.enqueue.failed', - degraded: true, - fallbackPath: 'skip_preview_enqueue', - documentId: doc.id, - error: errorToLog(error), - }, 'Failed to enqueue document preview'); - }); - - if (doc.type === 'pdf' && !reusableParsedPdf) { - // Account for upload-driven parse load in the same ledger the explicit - // re-parse limiter reads. We record (not reject) here so a legitimate - // bulk upload always parses; the recorded load still throttles - // subsequent loopable re-parse spam via /parsed. - await recordJobEvent(ctxOrRes.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig ?? { enabled: false, windows: [] }); - enqueueParsePdfJob({ - documentId: doc.id, - userId: storageUserId, - namespace: testNamespace, - }); - } - } - - return NextResponse.json({ success: true, stored }); - } catch (error) { - serverLogger.error({ - event: 'documents.register.failed', - error: errorToLog(error), - }, 'Failed to register documents'); - return errorResponse(error, { - apiErrorMessage: 'Failed to register documents', - normalize: { code: 'DOCUMENTS_REGISTER_FAILED', errorClass: 'db' }, - }); - } -} - export async function GET(req: NextRequest) { try { if (!isS3Configured()) return s3NotConfiguredResponse(); diff --git a/src/lib/client/api/documents.ts b/src/lib/client/api/documents.ts index b943354..360914f 100644 --- a/src/lib/client/api/documents.ts +++ b/src/lib/client/api/documents.ts @@ -1,10 +1,8 @@ -import { sha256HexFromArrayBuffer } from '@/lib/client/sha256'; import type { BaseDocument, DocumentType } from '@/types/documents'; import type { ParsedPdfDocument, PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import type { DocumentSettings } from '@/types/document-settings'; export type UploadSource = { - id: string; name: string; type: DocumentType; size: number; @@ -23,8 +21,12 @@ function toUploadBody(body: UploadSource['body']): BodyInit { return body as unknown as BodyInit; } -async function uploadDocumentSourceViaProxy(source: UploadSource, options?: UploadOptions): Promise { - const res = await fetch(`/api/documents/blob/upload/fallback?id=${encodeURIComponent(source.id)}`, { +async function uploadDocumentSourceViaProxy( + source: UploadSource, + token: string, + options?: UploadOptions, +): Promise { + const res = await fetch(`/api/documents/blob/upload/fallback?token=${encodeURIComponent(token)}`, { method: 'PUT', headers: { 'Content-Type': source.contentType || 'application/octet-stream' }, body: toUploadBody(source.body), @@ -241,7 +243,6 @@ export async function uploadDocumentSources(sources: UploadSource[], options?: U headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ uploads: sources.map((source) => ({ - id: source.id, contentType: source.contentType, size: source.size, })), @@ -255,14 +256,18 @@ export async function uploadDocumentSources(sources: UploadSource[], options?: U } const presigned = (await presignRes.json()) as { - uploads?: Array<{ id: string; url: string; headers?: Record }>; + uploads?: Array<{ token: string; url: string; headers?: Record }>; }; - const byId = new Map((presigned.uploads || []).map((upload) => [upload.id, upload])); + const uploads = presigned.uploads || []; + if (uploads.length !== sources.length) { + throw new Error('Upload preparation returned an unexpected number of temp uploads'); + } - for (const source of sources) { - const upload = byId.get(source.id); - if (!upload?.url) { - throw new Error(`Missing presigned upload for document ${source.id}`); + for (let index = 0; index < sources.length; index += 1) { + const source = sources[index]; + const upload = uploads[index]; + if (!upload?.url || !upload.token) { + throw new Error(`Missing presigned upload for document ${source.name}`); } let putError: unknown = null; @@ -285,7 +290,7 @@ export async function uploadDocumentSources(sources: UploadSource[], options?: U } try { - await uploadDocumentSourceViaProxy(source, options); + await uploadDocumentSourceViaProxy(source, upload.token, options); } catch (proxyError) { const directMessage = putError instanceof Error ? putError.message : 'unknown direct upload error'; const proxyMessage = proxyError instanceof Error ? proxyError.message : 'unknown proxy upload error'; @@ -293,27 +298,26 @@ export async function uploadDocumentSources(sources: UploadSource[], options?: U } } - const registerRes = await fetch('/api/documents', { + const finalizeRes = await fetch('/api/documents/blob/upload/finalize', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ - documents: sources.map((source) => ({ - id: source.id, + uploads: sources.map((source, index) => ({ + token: uploads[index]?.token, name: source.name, type: source.type, - size: source.size, lastModified: source.lastModified, })), }), signal: options?.signal, }); - if (!registerRes.ok) { - const data = (await registerRes.json().catch(() => null)) as { error?: string } | null; - throw new Error(data?.error || 'Failed to register uploaded documents'); + if (!finalizeRes.ok) { + const data = (await finalizeRes.json().catch(() => null)) as { error?: string } | null; + throw new Error(data?.error || 'Failed to finalize uploaded documents'); } - const data = (await registerRes.json()) as { stored: BaseDocument[] }; + const data = (await finalizeRes.json()) as { stored: BaseDocument[] }; return data.stored || []; } @@ -322,13 +326,10 @@ export async function uploadDocuments(files: File[], options?: UploadOptions): P const sources: UploadSource[] = []; for (const file of files) { - const bytes = await file.arrayBuffer(); - const id = await sha256HexFromArrayBuffer(bytes); const type = documentTypeForName(file.name); - const name = file.name || `${id}.${type}`; + const name = file.name || `upload.${type}`; const contentType = file.type || mimeTypeForDoc({ name, type }); sources.push({ - id, name, type, size: file.size, diff --git a/src/lib/client/dexie.ts b/src/lib/client/dexie.ts index 40e99f6..19bc09b 100644 --- a/src/lib/client/dexie.ts +++ b/src/lib/client/dexie.ts @@ -971,11 +971,9 @@ export async function syncDocumentsToServer( for (const doc of pdfDocs) { const bytes = new Uint8Array(doc.data); - const id = await sha256HexFromBytes(bytes); uploads.push({ oldId: doc.id, source: { - id, name: doc.name, type: 'pdf', size: bytes.byteLength, @@ -992,11 +990,9 @@ export async function syncDocumentsToServer( for (const doc of epubDocs) { const bytes = new Uint8Array(doc.data); - const id = await sha256HexFromBytes(bytes); uploads.push({ oldId: doc.id, source: { - id, name: doc.name, type: 'epub', size: bytes.byteLength, @@ -1013,11 +1009,9 @@ export async function syncDocumentsToServer( for (const doc of htmlDocs) { const encoded = textEncoder.encode(doc.data); - const id = await sha256HexFromBytes(encoded); uploads.push({ oldId: doc.id, source: { - id, name: doc.name, type: 'html', size: encoded.byteLength, @@ -1036,11 +1030,13 @@ export async function syncDocumentsToServer( onProgress(50, 'Uploading to server...'); } - await uploadDocumentSources(uploads.map((entry) => entry.source), { signal }); + const stored = await uploadDocumentSources(uploads.map((entry) => entry.source), { signal }); - for (const entry of uploads) { - if (entry.oldId === entry.source.id) continue; - await applyDocumentIdMapping(entry.oldId, entry.source.id); + for (let index = 0; index < uploads.length; index += 1) { + const entry = uploads[index]; + const nextId = stored[index]?.id; + if (!nextId || entry.oldId === nextId) continue; + await applyDocumentIdMapping(entry.oldId, nextId); } if (onProgress) { @@ -1064,11 +1060,9 @@ export async function syncSelectedDocumentsToServer( const data = await getPdfDocument(doc.id); if (data) { const bytes = new Uint8Array(data.data); - const id = await sha256HexFromBytes(bytes); uploads.push({ oldId: data.id, source: { - id, name: data.name, type: 'pdf', size: bytes.byteLength, @@ -1082,11 +1076,9 @@ export async function syncSelectedDocumentsToServer( const data = await getEpubDocument(doc.id); if (data) { const bytes = new Uint8Array(data.data); - const id = await sha256HexFromBytes(bytes); uploads.push({ oldId: data.id, source: { - id, name: data.name, type: 'epub', size: bytes.byteLength, @@ -1100,11 +1092,9 @@ export async function syncSelectedDocumentsToServer( const data = await getHtmlDocument(doc.id); if (data) { const bytes = textEncoder.encode(data.data); - const id = await sha256HexFromBytes(bytes); uploads.push({ oldId: data.id, source: { - id, name: data.name, type: 'html', size: bytes.byteLength, @@ -1121,11 +1111,13 @@ export async function syncSelectedDocumentsToServer( } if (onProgress) onProgress(50, 'Uploading to server...'); - await uploadDocumentSources(uploads.map((entry) => entry.source), { signal }); + const stored = await uploadDocumentSources(uploads.map((entry) => entry.source), { signal }); - for (const entry of uploads) { - if (entry.oldId === entry.source.id) continue; - await applyDocumentIdMapping(entry.oldId, entry.source.id); + for (let index = 0; index < uploads.length; index += 1) { + const entry = uploads[index]; + const nextId = stored[index]?.id; + if (!nextId || entry.oldId === nextId) continue; + await applyDocumentIdMapping(entry.oldId, nextId); } if (onProgress) { diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index fb84f00..b8fa988 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -1,4 +1,5 @@ import { + CopyObjectCommand, DeleteObjectCommand, DeleteObjectsCommand, GetObjectCommand, @@ -11,6 +12,9 @@ import { getS3Client, getS3Config, getS3ProxyClient } from '@/lib/server/storage const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; +const TEMP_UPLOAD_TOKEN_REGEX = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +const TEMP_UPLOAD_USER_ID_REGEX = /^[a-zA-Z0-9._:-]{1,256}$/; +export const TEMP_DOCUMENT_UPLOAD_TTL_MS = 24 * 60 * 60 * 1000; function sanitizeNamespace(namespace: string | null): string | null { if (!namespace) return null; @@ -69,6 +73,17 @@ export function isValidDocumentId(id: string): boolean { return DOCUMENT_ID_REGEX.test(id); } +export function isValidTempUploadToken(token: string): boolean { + return TEMP_UPLOAD_TOKEN_REGEX.test(token); +} + +function sanitizeTempUploadUserId(userId: string): string { + if (!TEMP_UPLOAD_USER_ID_REGEX.test(userId)) { + throw new Error(`Invalid temp upload user id: ${userId}`); + } + return encodeURIComponent(userId); +} + export function documentKey(id: string, namespace: string | null): string { if (!isValidDocumentId(id)) { throw new Error(`Invalid document id: ${id}`); @@ -90,6 +105,27 @@ export function documentParsedKey(id: string, namespace: string | null): string return `${cfg.prefix}/documents_v1/parsed_v1/${nsSegment}${id}.json`; } +export function tempDocumentUploadPrefix(userId: string, namespace: string | null): string { + const cfg = getS3Config(); + const ns = sanitizeNamespace(namespace); + const nsSegment = ns ? `ns/${ns}/` : ''; + return `${cfg.prefix}/document_uploads_temp_v1/${nsSegment}users/${sanitizeTempUploadUserId(userId)}/`; +} + +export function tempDocumentUploadKey(token: string, userId: string, namespace: string | null): string { + if (!isValidTempUploadToken(token)) { + throw new Error(`Invalid temp upload token: ${token}`); + } + return `${tempDocumentUploadPrefix(userId, namespace)}${token}.bin`; +} + +export function tempDocumentUploadReceiptKey(token: string, userId: string, namespace: string | null): string { + if (!isValidTempUploadToken(token)) { + throw new Error(`Invalid temp upload token: ${token}`); + } + return `${tempDocumentUploadPrefix(userId, namespace)}${token}.receipt.json`; +} + function legacyDocumentParsedKey(id: string, namespace: string | null): string { if (!isValidDocumentId(id)) { throw new Error(`Invalid document id: ${id}`); @@ -138,6 +174,40 @@ export async function presignPut( }; } +export async function presignTempPut( + token: string, + userId: string, + contentType: string, + namespace: string | null, + options?: { contentLength?: number }, +): Promise<{ url: string; headers: Record }> { + const cfg = getS3Config(); + const client = getS3Client(); + const key = tempDocumentUploadKey(token, userId, namespace); + const normalizedType = (contentType || 'application/octet-stream').trim() || 'application/octet-stream'; + const contentLength = + typeof options?.contentLength === 'number' && Number.isFinite(options.contentLength) && options.contentLength > 0 + ? Math.floor(options.contentLength) + : undefined; + + const command = new PutObjectCommand({ + Bucket: cfg.bucket, + Key: key, + ContentType: normalizedType, + ServerSideEncryption: 'AES256', + ...(contentLength !== undefined ? { ContentLength: contentLength } : {}), + }); + const url = await getSignedUrl(client, command, { expiresIn: 60 * 5 }); + + return { + url, + headers: { + 'Content-Type': normalizedType, + 'x-amz-server-side-encryption': 'AES256', + }, + }; +} + export async function headDocumentBlob( id: string, namespace: string | null, @@ -153,6 +223,23 @@ export async function headDocumentBlob( }; } +export async function headTempDocumentBlob( + token: string, + userId: string, + namespace: string | null, +): Promise<{ contentLength: number; contentType: string | null; eTag: string | null; lastModified: number | null }> { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const key = tempDocumentUploadKey(token, userId, namespace); + const res = await client.send(new HeadObjectCommand({ Bucket: cfg.bucket, Key: key })); + return { + contentLength: Number(res.ContentLength ?? 0), + contentType: res.ContentType ?? null, + eTag: res.ETag ?? null, + lastModified: res.LastModified?.getTime() ?? null, + }; +} + export async function getDocumentRange( id: string, start: number, @@ -185,6 +272,23 @@ export async function getDocumentBlob(id: string, namespace: string | null): Pro return bodyToBuffer(res.Body); } +export async function getTempDocumentBlob( + token: string, + userId: string, + namespace: string | null, +): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const key = tempDocumentUploadKey(token, userId, namespace); + const res = await client.send( + new GetObjectCommand({ + Bucket: cfg.bucket, + Key: key, + }), + ); + return bodyToBuffer(res.Body); +} + export async function getDocumentBlobStream(id: string, namespace: string | null): Promise { const cfg = getS3Config(); const client = getS3ProxyClient(); @@ -198,6 +302,29 @@ export async function getDocumentBlobStream(id: string, namespace: string | null return res.Body as DocumentBlobBody; } +export async function getTempDocumentFinalizeReceipt( + token: string, + userId: string, + namespace: string | null, +): Promise { + try { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const key = tempDocumentUploadReceiptKey(token, userId, namespace); + const res = await client.send( + new GetObjectCommand({ + Bucket: cfg.bucket, + Key: key, + }), + ); + const body = await bodyToBuffer(res.Body); + return JSON.parse(body.toString('utf8')) as T; + } catch (error) { + if (isMissingBlobError(error)) return null; + throw error; + } +} + export async function getParsedDocumentBlob(id: string, namespace: string | null): Promise { const cfg = getS3Config(); const client = getS3ProxyClient(); @@ -241,6 +368,26 @@ export async function putParsedDocumentBlob(id: string, body: Buffer, namespace: return key; } +export async function putTempDocumentFinalizeReceipt( + token: string, + userId: string, + namespace: string | null, + body: Buffer, +): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const key = tempDocumentUploadReceiptKey(token, userId, namespace); + await client.send( + new PutObjectCommand({ + Bucket: cfg.bucket, + Key: key, + Body: body, + ContentType: 'application/json', + ServerSideEncryption: 'AES256', + }), + ); +} + export async function presignGet( id: string, namespace: string | null, @@ -264,6 +411,7 @@ export async function putDocumentBlob( body: Buffer, contentType: string, namespace: string | null, + options?: { ifNoneMatch?: boolean }, ): Promise { const cfg = getS3Config(); const client = getS3ProxyClient(); @@ -275,6 +423,49 @@ export async function putDocumentBlob( Body: body, ContentType: contentType, ServerSideEncryption: 'AES256', + ...(options?.ifNoneMatch ? { IfNoneMatch: '*' } : {}), + }), + ); +} + +export async function putTempDocumentBlob( + token: string, + userId: string, + body: Buffer, + contentType: string, + namespace: string | null, +): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const key = tempDocumentUploadKey(token, userId, namespace); + await client.send( + new PutObjectCommand({ + Bucket: cfg.bucket, + Key: key, + Body: body, + ContentType: contentType, + ServerSideEncryption: 'AES256', + }), + ); +} + +export async function copyTempDocumentBlobToDocument( + token: string, + userId: string, + documentId: string, + namespace: string | null, + contentType: string, +): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + await client.send( + new CopyObjectCommand({ + Bucket: cfg.bucket, + Key: documentKey(documentId, namespace), + CopySource: `${cfg.bucket}/${tempDocumentUploadKey(token, userId, namespace)}`, + ContentType: contentType, + MetadataDirective: 'REPLACE', + ServerSideEncryption: 'AES256', }), ); } @@ -292,6 +483,18 @@ export async function deleteDocumentBlob(id: string, namespace: string | null): await deleteDocumentPrefix(`${key}/`).catch(() => undefined); } +export async function deleteTempDocumentUpload(token: string, userId: string, namespace: string | null): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: tempDocumentUploadKey(token, userId, namespace) })); +} + +export async function deleteTempDocumentFinalizeReceipt(token: string, userId: string, namespace: string | null): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: tempDocumentUploadReceiptKey(token, userId, namespace) })); +} + export function isMissingBlobError(error: unknown): boolean { if (!error || typeof error !== 'object') return false; const maybe = error as { name?: string; Code?: string; $metadata?: { httpStatusCode?: number } }; @@ -339,3 +542,53 @@ export async function deleteDocumentPrefix(prefix: string): Promise { return deleted; } + +export async function deleteExpiredTempDocumentUploads( + userId: string, + namespace: string | null, + olderThanMs: number, +): Promise { + const cfg = getS3Config(); + const client = getS3ProxyClient(); + const prefix = tempDocumentUploadPrefix(userId, namespace); + let continuationToken: string | undefined; + const keys: string[] = []; + + do { + const listRes = await client.send( + new ListObjectsV2Command({ + Bucket: cfg.bucket, + Prefix: prefix, + ContinuationToken: continuationToken, + }), + ); + + for (const item of listRes.Contents ?? []) { + const key = item.Key; + const lastModified = item.LastModified?.getTime() ?? 0; + if (!key || lastModified <= 0 || lastModified >= olderThanMs) continue; + keys.push(key); + } + + continuationToken = listRes.IsTruncated ? listRes.NextContinuationToken : undefined; + } while (continuationToken); + + if (keys.length === 0) return 0; + + let deleted = 0; + for (let i = 0; i < keys.length; i += 1000) { + const batch = keys.slice(i, i + 1000); + const deleteRes = await client.send( + new DeleteObjectsCommand({ + Bucket: cfg.bucket, + Delete: { + Objects: batch.map((Key) => ({ Key })), + Quiet: true, + }, + }), + ); + deleted += deleteRes.Deleted?.length ?? 0; + } + + return deleted; +} diff --git a/src/lib/server/documents/register-upload.ts b/src/lib/server/documents/register-upload.ts new file mode 100644 index 0000000..6fe134e --- /dev/null +++ b/src/lib/server/documents/register-upload.ts @@ -0,0 +1,104 @@ +import { randomUUID } from 'node:crypto'; +import { db } from '@/db'; +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 { enqueueParsePdfJob } from '@/lib/server/jobs/user-pdf-layout-job'; +import { errorToLog, serverLogger } from '@/lib/server/logger'; +import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config'; +import { getPdfLayoutRateConfig, recordJobEvent } from '@/lib/server/rate-limit/job-rate-limiter'; +import { PDF_PARSER_VERSION } from '@openreader/compute-core'; +import type { BaseDocument, DocumentType } from '@/types/documents'; + +type RegisterUploadedDocumentInput = { + documentId: string; + userId: string; + namespace: string | null; + name: string; + type: DocumentType; + size: number; + lastModified: number; +}; + +export async function registerUploadedDocument(input: RegisterUploadedDocumentInput): Promise { + const reusableParsedPdf = input.type === 'pdf' + ? await findReusableParsedPdfResult(input.documentId) + : null; + + await db + .insert(documents) + .values({ + id: input.documentId, + userId: input.userId, + name: input.name, + type: input.type, + size: input.size, + lastModified: input.lastModified, + filePath: input.documentId, + parseState: input.type === 'pdf' + ? stringifyDocumentParseState( + reusableParsedPdf + ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } + : { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }, + ) + : null, + parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, + }) + .onConflictDoUpdate({ + target: [documents.id, documents.userId], + set: { + name: input.name, + type: input.type, + size: input.size, + lastModified: input.lastModified, + filePath: input.documentId, + parseState: input.type === 'pdf' + ? stringifyDocumentParseState( + reusableParsedPdf + ? { status: 'ready', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION } + : { status: 'pending', progress: null, updatedAt: Date.now(), parserVersion: PDF_PARSER_VERSION }, + ) + : null, + parsedJsonKey: reusableParsedPdf?.parsedJsonKey ?? null, + }, + }); + + await enqueueDocumentPreview( + { + id: input.documentId, + type: input.type, + lastModified: input.lastModified, + }, + input.namespace, + ).catch((error) => { + serverLogger.warn({ + event: 'documents.preview.enqueue.failed', + degraded: true, + fallbackPath: 'skip_preview_enqueue', + documentId: input.documentId, + error: errorToLog(error), + }, 'Failed to enqueue document preview'); + }); + + if (input.type === 'pdf' && !reusableParsedPdf) { + const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig()); + await recordJobEvent(input.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig); + enqueueParsePdfJob({ + documentId: input.documentId, + userId: input.userId, + namespace: input.namespace, + }); + } + + return { + id: input.documentId, + name: input.name, + type: input.type, + size: input.size, + lastModified: input.lastModified, + scope: 'user', + }; +} diff --git a/tests/upload.spec.ts b/tests/upload.spec.ts index 2008f72..7a6ef51 100644 --- a/tests/upload.spec.ts +++ b/tests/upload.spec.ts @@ -30,6 +30,29 @@ test.describe('Document Upload Tests', () => { await expectDocumentListed(page, 'sample.txt'); }); + test('reuses the same canonical id for identical uploads', async ({ page }) => { + await uploadFile(page, 'sample.pdf'); + await uploadFile(page, 'sample.pdf'); + + const result = await page.evaluate(async () => { + const res = await fetch('/api/documents', { cache: 'no-store' }); + if (!res.ok) { + return { ok: false as const, reason: `status:${res.status}` }; + } + const data = await res.json() as { documents?: Array<{ id: string; name: string }> }; + const matching = (data.documents || []).filter((doc) => doc.name === 'sample.pdf'); + const uniqueIds = Array.from(new Set(matching.map((doc) => doc.id))); + return { ok: true as const, matchingCount: matching.length, uniqueIds }; + }); + + expect(result.ok).toBeTruthy(); + if (!result.ok) { + throw new Error(`Failed to inspect uploaded documents: ${result.reason}`); + } + expect(result.matchingCount).toBe(1); + expect(result.uniqueIds).toHaveLength(1); + }); + test('hashes text/HTML docs using UTF-8 encoded stored string', async ({ page }) => { await uploadFile(page, 'sample.txt'); await expectDocumentListed(page, 'sample.txt');