feat(upload): implement presigned token-based upload with finalize step
Transition document upload flow to use presigned token-based temporary uploads. Add finalize endpoint to register uploaded documents after blob transfer, decoupling document ID assignment from upload initiation. Update client and server logic to operate on tokens, improve security, and enable canonical ID reuse for identical uploads. Update tests to verify deduplication behavior.
This commit is contained in:
Richard R
parent
58ad6212c9
commit
6c86720926
9 changed files with 665 additions and 256 deletions
|
|
@ -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',
|
||||
|
|
|
|||
224
src/app/api/documents/blob/upload/finalize/route.ts
Normal file
224
src/app/api/documents/blob/upload/finalize/route.ts
Normal file
|
|
@ -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<string, unknown>;
|
||||
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<BaseDocument> {
|
||||
const existingReceipt = await getTempDocumentFinalizeReceipt<FinalizeReceipt>(
|
||||
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' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -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<string, unknown>;
|
||||
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,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
const res = await fetch(`/api/documents/blob/upload/fallback?id=${encodeURIComponent(source.id)}`, {
|
||||
async function uploadDocumentSourceViaProxy(
|
||||
source: UploadSource,
|
||||
token: string,
|
||||
options?: UploadOptions,
|
||||
): Promise<void> {
|
||||
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<string, string> }>;
|
||||
uploads?: Array<{ token: string; url: string; headers?: Record<string, string> }>;
|
||||
};
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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<string, string> }> {
|
||||
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<Buffer> {
|
||||
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<DocumentBlobBody> {
|
||||
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<T>(
|
||||
token: string,
|
||||
userId: string,
|
||||
namespace: string | null,
|
||||
): Promise<T | null> {
|
||||
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<Buffer> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<number> {
|
|||
|
||||
return deleted;
|
||||
}
|
||||
|
||||
export async function deleteExpiredTempDocumentUploads(
|
||||
userId: string,
|
||||
namespace: string | null,
|
||||
olderThanMs: number,
|
||||
): Promise<number> {
|
||||
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;
|
||||
}
|
||||
|
|
|
|||
104
src/lib/server/documents/register-upload.ts
Normal file
104
src/lib/server/documents/register-upload.ts
Normal file
|
|
@ -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<BaseDocument> {
|
||||
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',
|
||||
};
|
||||
}
|
||||
|
|
@ -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');
|
||||
|
|
|
|||
Loading…
Reference in a new issue