diff --git a/compute/worker/src/pdf-artifact-persistence.ts b/compute/worker/src/pdf-artifact-persistence.ts new file mode 100644 index 0000000..fff9f00 --- /dev/null +++ b/compute/worker/src/pdf-artifact-persistence.ts @@ -0,0 +1,18 @@ +export async function persistParsedPdfWhileSourceExists(input: { + sourceObjectKey: string; + sourceExists: (key: string) => Promise; + putParsedObject: () => Promise; + deleteParsedObject: (key: string) => Promise; +}): Promise { + if (!await input.sourceExists(input.sourceObjectKey)) { + throw new Error('PDF source object was deleted before parsed output could be persisted'); + } + + const parsedObjectKey = await input.putParsedObject(); + if (!await input.sourceExists(input.sourceObjectKey)) { + await input.deleteParsedObject(parsedObjectKey); + throw new Error('PDF source object was deleted while parsed output was being persisted'); + } + + return parsedObjectKey; +} diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index a00121d..442df53 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -49,7 +49,13 @@ import type { WorkerOperationState, PdfLayoutProgress, } from '@openreader/compute-core/api-contracts'; -import { GetObjectCommand, PutObjectCommand, S3Client } from '@aws-sdk/client-s3'; +import { + DeleteObjectCommand, + GetObjectCommand, + HeadObjectCommand, + PutObjectCommand, + S3Client, +} from '@aws-sdk/client-s3'; import { JetStreamOperationEventStream, OP_EVENTS_SUBJECT_WILDCARD, @@ -64,6 +70,7 @@ import { } from './orphan-recovery'; import { buildInferProgressForPageParsed, buildInferProgressForPageStart } from './pdf-progress'; import { buildQueueWaitTiming, decideRetryAction } from './worker-loop-policy'; +import { persistParsedPdfWhileSourceExists } from './pdf-artifact-persistence'; const JOBS_STREAM_NAME = 'compute_jobs'; const WHISPER_JOBS_SUBJECT = 'jobs.whisper'; @@ -701,6 +708,42 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti return toArrayBuffer(new Uint8Array(bytes)); }; + const objectExists = async (key: string): Promise => { + if (!s3) { + throw new Error('S3 access is disabled for this worker app instance'); + } + const safeKey = ensureSafeKey(key); + try { + await s3.send(new HeadObjectCommand({ + Bucket: s3Bucket, + Key: safeKey, + })); + return true; + } catch (error) { + const maybe = error as { name?: string; Code?: string; $metadata?: { httpStatusCode?: number } }; + if ( + maybe.$metadata?.httpStatusCode === 404 + || maybe.name === 'NotFound' + || maybe.name === 'NoSuchKey' + || maybe.Code === 'NotFound' + || maybe.Code === 'NoSuchKey' + ) { + return false; + } + throw error; + } + }; + + const deleteObjectByKey = async (key: string): Promise => { + if (!s3) { + throw new Error('S3 access is disabled for this worker app instance'); + } + await s3.send(new DeleteObjectCommand({ + Bucket: s3Bucket, + Key: ensureSafeKey(key), + })); + }; + const putParsedObject = async (documentId: string, namespace: string | null, parsed: unknown): Promise => { if (!s3) { throw new Error('S3 access is disabled for this worker app instance'); @@ -1175,7 +1218,12 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti phase: 'merge', }); } - const parsedObjectKey = await putParsedObject(parsed.documentId, parsed.namespace, result.parsed); + const parsedObjectKey = await persistParsedPdfWhileSourceExists({ + sourceObjectKey: parsed.documentObjectKey, + sourceExists: objectExists, + putParsedObject: () => putParsedObject(parsed.documentId, parsed.namespace, result.parsed), + deleteParsedObject: deleteObjectByKey, + }); return { parsedObjectKey, timing: { diff --git a/compute/worker/tests/unit/pdf-artifact-persistence.test.ts b/compute/worker/tests/unit/pdf-artifact-persistence.test.ts new file mode 100644 index 0000000..9378435 --- /dev/null +++ b/compute/worker/tests/unit/pdf-artifact-persistence.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, test, vi } from 'vitest'; +import { persistParsedPdfWhileSourceExists } from '../../src/pdf-artifact-persistence'; + +describe('PDF artifact persistence', () => { + test('does not write parsed output after the source was deleted', async () => { + const putParsedObject = vi.fn(async () => 'parsed.json'); + + await expect(persistParsedPdfWhileSourceExists({ + sourceObjectKey: 'document.pdf', + sourceExists: async () => false, + putParsedObject, + deleteParsedObject: async () => undefined, + })).rejects.toThrow('before parsed output'); + + expect(putParsedObject).not.toHaveBeenCalled(); + }); + + test('removes parsed output when the source disappears during the write', async () => { + const sourceExists = vi.fn() + .mockResolvedValueOnce(true) + .mockResolvedValueOnce(false); + const deleteParsedObject = vi.fn(async () => undefined); + + await expect(persistParsedPdfWhileSourceExists({ + sourceObjectKey: 'document.pdf', + sourceExists, + putParsedObject: async () => 'parsed.json', + deleteParsedObject, + })).rejects.toThrow('while parsed output'); + + expect(deleteParsedObject).toHaveBeenCalledWith('parsed.json'); + }); +}); diff --git a/src/app/api/documents/blob/get/fallback/route.ts b/src/app/api/documents/blob/get/fallback/route.ts index cd74281..0db67e0 100644 --- a/src/app/api/documents/blob/get/fallback/route.ts +++ b/src/app/api/documents/blob/get/fallback/route.ts @@ -9,7 +9,7 @@ import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace' import { isS3Configured } from '@/lib/server/storage/s3'; import { errorToLog, serverLogger } from '@/lib/server/logger'; import { errorResponse } from '@/lib/server/errors/next-response'; -import { deleteDocumentTtsSegmentCache } from '@/lib/server/tts/segments-cache'; +import { deleteOwnedDocument } from '@/lib/server/documents/delete-owned'; export const dynamic = 'force-dynamic'; @@ -81,14 +81,11 @@ export async function GET(req: NextRequest) { }); } catch (error) { if (isMissingBlobError(error)) { - await deleteDocumentTtsSegmentCache({ + await deleteOwnedDocument({ userId: doc.userId, documentId: id, namespace: testNamespace, }); - await db - .delete(documents) - .where(and(eq(documents.id, id), eq(documents.userId, doc.userId))); return NextResponse.json({ error: 'Not found' }, { status: 404 }); } throw error; diff --git a/src/app/api/documents/blob/preview/fallback/route.ts b/src/app/api/documents/blob/preview/fallback/route.ts index 83e17d4..004aa74 100644 --- a/src/app/api/documents/blob/preview/fallback/route.ts +++ b/src/app/api/documents/blob/preview/fallback/route.ts @@ -22,7 +22,7 @@ import { import { extractRawTextSnippet } from '@/lib/server/documents/text-snippets'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; -import { deleteDocumentTtsSegmentCache } from '@/lib/server/tts/segments-cache'; +import { deleteOwnedDocument } from '@/lib/server/documents/delete-owned'; export const dynamic = 'force-dynamic'; @@ -109,14 +109,11 @@ export async function GET(req: NextRequest) { ); } catch (error) { if (isMissingDocumentBlobError(error)) { - await deleteDocumentTtsSegmentCache({ + await deleteOwnedDocument({ userId: doc.userId, documentId: id, namespace: testNamespace, }); - await db - .delete(documents) - .where(and(eq(documents.id, id), eq(documents.userId, doc.userId))); return NextResponse.json({ error: 'Not found' }, { status: 404 }); } throw error; diff --git a/src/app/api/documents/blob/upload/finalize/route.ts b/src/app/api/documents/blob/upload/finalize/route.ts index b1f5f6e..76c6e24 100644 --- a/src/app/api/documents/blob/upload/finalize/route.ts +++ b/src/app/api/documents/blob/upload/finalize/route.ts @@ -24,6 +24,7 @@ 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'; +import { withDocumentMutationLock } from '@/lib/server/documents/mutation-lock'; export const dynamic = 'force-dynamic'; @@ -149,47 +150,49 @@ async function finalizeOne(input: { : input.upload.name; const documentId = createHash('sha256').update(finalizedBody).digest('hex'); - try { - await headDocumentBlob(documentId, input.namespace); - } catch (error) { - if (!isMissingBlobError(error)) throw error; - if (!isDocxUpload) { - try { - await copyTempDocumentBlobToDocument( - input.upload.token, - input.userId, - documentId, - input.namespace, - finalizedContentType, - { ifNoneMatch: true }, - ); - } catch (copyError) { - if (!isPreconditionFailed(copyError)) throw copyError; - } - } else { - try { - await putDocumentBlob( - documentId, - finalizedBody, - finalizedContentType, - input.namespace, - { ifNoneMatch: true }, - ); - } catch (putError) { - if (!isPreconditionFailed(putError)) throw putError; + const stored = await withDocumentMutationLock(documentId, async () => { + try { + await headDocumentBlob(documentId, input.namespace); + } catch (error) { + if (!isMissingBlobError(error)) throw error; + if (!isDocxUpload) { + try { + await copyTempDocumentBlobToDocument( + input.upload.token, + input.userId, + documentId, + input.namespace, + finalizedContentType, + { ifNoneMatch: true }, + ); + } catch (copyError) { + if (!isPreconditionFailed(copyError)) throw copyError; + } + } else { + try { + await putDocumentBlob( + documentId, + finalizedBody, + finalizedContentType, + input.namespace, + { ifNoneMatch: true }, + ); + } catch (putError) { + if (!isPreconditionFailed(putError)) throw putError; + } } } - } - const canonicalHead = await headDocumentBlob(documentId, input.namespace); - const stored = await registerUploadedDocument({ - documentId, - userId: input.userId, - namespace: input.namespace, - name: finalizedName, - type: finalizedType, - size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength, - lastModified: input.upload.lastModified, + const canonicalHead = await headDocumentBlob(documentId, input.namespace); + return registerUploadedDocument({ + documentId, + userId: input.userId, + namespace: input.namespace, + name: finalizedName, + type: finalizedType, + size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength, + lastModified: input.upload.lastModified, + }); }); await putTempDocumentFinalizeReceipt( diff --git a/src/app/api/documents/route.ts b/src/app/api/documents/route.ts index 87c89a7..a414d4d 100644 --- a/src/app/api/documents/route.ts +++ b/src/app/api/documents/route.ts @@ -1,19 +1,15 @@ import { NextRequest, NextResponse } from 'next/server'; -import { and, count, eq, inArray } from 'drizzle-orm'; +import { and, inArray } from 'drizzle-orm'; import { db } from '@/db'; import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; 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, -} from '@/lib/server/documents/previews'; -import { deleteDocumentBlob, isMissingBlobError, isValidDocumentId } from '@/lib/server/documents/blobstore'; +import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace'; import { isS3Configured } from '@/lib/server/storage/s3'; -import { deleteDocumentTtsSegmentCache } from '@/lib/server/tts/segments-cache'; +import { deleteOwnedDocument } from '@/lib/server/documents/delete-owned'; import type { BaseDocument, DocumentType } from '@/types/documents'; export const dynamic = 'force-dynamic'; @@ -145,63 +141,18 @@ export async function DELETE(req: NextRequest) { userId: string; }>; - // TTS audio is user-scoped even when the underlying document blob is - // shared. Remove it before deleting ownership metadata so failures remain - // retryable and cannot create untraceable S3 objects. + let deleted = 0; for (const row of ownedRows) { - await deleteDocumentTtsSegmentCache({ + if (await deleteOwnedDocument({ userId: row.userId, documentId: row.id, namespace: testNamespace, - }); - } - - const deletedRows = (await db - .delete(documents) - .where(and(inArray(documents.userId, targetUserIds), inArray(documents.id, targetIds))) - .returning({ id: documents.id })) as Array<{ id: string }>; - - const uniqueIds = Array.from(new Set(deletedRows.map((row) => row.id))); - for (const id of uniqueIds) { - const [ref] = await db.select({ count: count() }).from(documents).where(eq(documents.id, id)); - const refCount = Number(ref?.count ?? 0); - if (refCount > 0) continue; - - try { - await deleteDocumentBlob(id, testNamespace); - } catch (error) { - if (!isMissingBlobError(error)) { - serverLogger.warn({ - event: 'documents.delete.blob_cleanup_failed', - degraded: true, - step: 'delete_document_blob', - documentId: id, - error: errorToLog(error), - }, 'Failed to delete document blob during cleanup'); - } + })) { + deleted += 1; } - - await cleanupDocumentPreviewArtifacts(id, testNamespace).catch((error) => { - serverLogger.warn({ - event: 'documents.delete.preview_artifacts_cleanup_failed', - degraded: true, - step: 'delete_preview_artifacts', - documentId: id, - error: errorToLog(error), - }, 'Failed to cleanup preview artifacts'); - }); - await deleteDocumentPreviewRows(id, testNamespace).catch((error) => { - serverLogger.warn({ - event: 'documents.delete.preview_rows_cleanup_failed', - degraded: true, - step: 'delete_preview_rows', - documentId: id, - error: errorToLog(error), - }, 'Failed to cleanup preview rows'); - }); } - return NextResponse.json({ success: true, deleted: deletedRows.length }); + return NextResponse.json({ success: true, deleted }); } catch (error) { serverLogger.error({ event: 'documents.delete.failed', diff --git a/src/app/api/tts/segments/ensure/route.ts b/src/app/api/tts/segments/ensure/route.ts index 4dfb6d9..28d7a0a 100644 --- a/src/app/api/tts/segments/ensure/route.ts +++ b/src/app/api/tts/segments/ensure/route.ts @@ -1,11 +1,12 @@ import { NextRequest, NextResponse } from 'next/server'; import { and, eq, inArray } from 'drizzle-orm'; import { db } from '@/db'; -import { ttsSegmentEntries, ttsSegmentVariants } from '@/db/schema'; +import { documents, ttsSegmentEntries, ttsSegmentVariants } from '@/db/schema'; import { isS3Configured, getS3Config } from '@/lib/server/storage/s3'; import { generateTTSBuffer } from '@/lib/server/tts/generate'; import { getTtsSegmentAudioObject, + deleteTtsSegmentAudioObjects, putTtsSegmentAudioObject, } from '@/lib/server/tts/segments-blobstore'; import { @@ -300,6 +301,28 @@ export async function POST(request: NextRequest) { const existingById = new Map(rows.map((row) => [row.segmentId, row])); const manifest: TTSSegmentManifestItem[] = []; + const documentOwnershipExists = async (): Promise => { + const [row] = await db + .select({ id: documents.id }) + .from(documents) + .where(and( + eq(documents.id, parsed.documentId), + eq(documents.userId, scope.storageUserId), + )) + .limit(1); + return Boolean(row); + }; + const assertDocumentOwnership = async (): Promise => { + if (!await documentOwnershipExists()) { + throw new Error('Document ownership was removed during TTS generation'); + } + }; + const deleteAbandonedSegmentMetadata = async (): Promise => { + await db.delete(ttsSegmentEntries).where(and( + eq(ttsSegmentEntries.userId, scope.storageUserId), + eq(ttsSegmentEntries.documentId, parsed.documentId), + )); + }; const upsertSegmentEntry = async (input: { segmentEntryId: string; segmentIndex: number; @@ -308,6 +331,7 @@ export async function POST(request: NextRequest) { textHash: string; textLength: number; }): Promise => { + await assertDocumentOwnership(); await db .insert(ttsSegmentEntries) .values({ @@ -337,6 +361,10 @@ export async function POST(request: NextRequest) { updatedAt: nowMs, }, }); + if (!await documentOwnershipExists()) { + await deleteAbandonedSegmentMetadata(); + throw new Error('Document ownership was removed during TTS metadata persistence'); + } }; for (const segment of normalized) { @@ -657,7 +685,13 @@ export async function POST(request: NextRequest) { failedStage = 's3.put_audio'; const putStartedAt = Date.now(); + await assertDocumentOwnership(); await putTtsSegmentAudioObject(audioKey, ttsBuffer); + if (!await documentOwnershipExists()) { + await deleteTtsSegmentAudioObjects([audioKey]); + await deleteAbandonedSegmentMetadata(); + throw new Error('Document ownership was removed during TTS audio persistence'); + } stageTimings.putAudioMs = Date.now() - putStartedAt; let persistedBuffer = ttsBuffer; diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index 3f94bf0..5b4a85a 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -506,11 +506,14 @@ export async function deleteDocumentBlob(id: string, namespace: string | null): const nsSegment = ns ? `ns/${ns}/` : ''; const parsedPrefix = `${cfg.prefix}/documents_v1/parsed_v2/${nsSegment}${id}/`; - await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: key })); - await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: parsedKey })); - await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: legacyParsedKey })); await deleteDocumentPrefix(parsedPrefix); await deleteDocumentPrefix(`${key}/`); + await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: parsedKey })); + await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: legacyParsedKey })); + // Delete the source after the initial derived-artifact cleanup, then sweep + // parsed output once more to catch a worker that finished during deletion. + await client.send(new DeleteObjectCommand({ Bucket: cfg.bucket, Key: key })); + await deleteDocumentPrefix(parsedPrefix); } export async function deleteTempDocumentUpload(token: string, userId: string, namespace: string | null): Promise { diff --git a/src/lib/server/documents/delete-owned.ts b/src/lib/server/documents/delete-owned.ts new file mode 100644 index 0000000..c7deba9 --- /dev/null +++ b/src/lib/server/documents/delete-owned.ts @@ -0,0 +1,111 @@ +import { and, eq, ne } from 'drizzle-orm'; +import { db } from '@/db'; +import { documents } from '@/db/schema'; +import { deleteDocumentBlob } from '@/lib/server/documents/blobstore'; +import { + cleanupDocumentPreviewArtifacts, + deleteDocumentPreviewRows, +} from '@/lib/server/documents/previews'; +import { deleteDocumentTtsSegmentCache } from '@/lib/server/tts/segments-cache'; +import { withDocumentMutationLock } from '@/lib/server/documents/mutation-lock'; + +type DocumentRow = typeof documents.$inferSelect; + +async function restoreDocumentOwnership(row: DocumentRow): Promise { + await db.insert(documents).values(row).onConflictDoNothing(); +} + +async function removeOwnershipAndCheckLastOwner(input: { + userId: string; + documentId: string; +}): Promise<{ removed: DocumentRow | null; isLastOwner: boolean }> { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const database = db as any; + const runAsync = async (tx: typeof database) => { + await tx + .select({ id: documents.id }) + .from(documents) + .where(eq(documents.id, input.documentId)) + .for('update'); + + const [removed] = await tx + .delete(documents) + .where(and( + eq(documents.id, input.documentId), + eq(documents.userId, input.userId), + )) + .returning(); + if (!removed) return { removed: null, isLastOwner: false }; + + const otherOwners = await tx + .select({ id: documents.id }) + .from(documents) + .where(and( + eq(documents.id, input.documentId), + ne(documents.userId, input.userId), + )) + .limit(1); + return { removed, isLastOwner: otherOwners.length === 0 }; + }; + + if (process.env.POSTGRES_URL) { + return database.transaction(runAsync); + } + + return database.transaction((tx: typeof database) => { + const removed = tx + .delete(documents) + .where(and( + eq(documents.id, input.documentId), + eq(documents.userId, input.userId), + )) + .returning() + .all()[0] ?? null; + if (!removed) return { removed: null, isLastOwner: false }; + + const otherOwners = tx + .select({ id: documents.id }) + .from(documents) + .where(and( + eq(documents.id, input.documentId), + ne(documents.userId, input.userId), + )) + .limit(1) + .all(); + return { removed, isLastOwner: otherOwners.length === 0 }; + }, { behavior: 'immediate' }); +} + +export async function deleteOwnedDocument(input: { + userId: string; + documentId: string; + namespace: string | null; +}): Promise { + return withDocumentMutationLock(input.documentId, async () => { + const { removed, isLastOwner } = await removeOwnershipAndCheckLastOwner(input); + if (!removed) return false; + + try { + await deleteDocumentTtsSegmentCache(input); + + if (isLastOwner) { + await cleanupDocumentPreviewArtifacts(input.documentId, input.namespace); + await deleteDocumentPreviewRows(input.documentId, input.namespace); + + const [newOwner] = await db + .select({ id: documents.id }) + .from(documents) + .where(eq(documents.id, input.documentId)) + .limit(1); + if (!newOwner) { + await deleteDocumentBlob(input.documentId, input.namespace); + } + } + } catch (error) { + await restoreDocumentOwnership(removed); + throw error; + } + + return true; + }); +} diff --git a/src/lib/server/documents/mutation-lock.ts b/src/lib/server/documents/mutation-lock.ts new file mode 100644 index 0000000..15856e2 --- /dev/null +++ b/src/lib/server/documents/mutation-lock.ts @@ -0,0 +1,40 @@ +import { sql } from 'drizzle-orm'; +import { db } from '@/db'; + +const localTails = new Map>(); + +async function withLocalDocumentLock(documentId: string, fn: () => Promise): Promise { + const previous = localTails.get(documentId) ?? Promise.resolve(); + let release!: () => void; + const current = new Promise((resolve) => { + release = resolve; + }); + localTails.set(documentId, current); + + await previous; + try { + return await fn(); + } finally { + release(); + if (localTails.get(documentId) === current) { + localTails.delete(documentId); + } + } +} + +export async function withDocumentMutationLock( + documentId: string, + fn: () => Promise, +): Promise { + return withLocalDocumentLock(documentId, async () => { + if (!process.env.POSTGRES_URL) { + return fn(); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (db as any).transaction(async (tx: any) => { + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${documentId}, 0))`); + return fn(); + }); + }); +} diff --git a/src/lib/server/user/claim-data.ts b/src/lib/server/user/claim-data.ts index 24673ec..e36a0db 100644 --- a/src/lib/server/user/claim-data.ts +++ b/src/lib/server/user/claim-data.ts @@ -12,6 +12,7 @@ import { import { isS3Configured } from '../storage/s3'; import { logDegraded } from '../errors/logging'; import { hashForLog, serverLogger } from '../logger'; +import { deleteOwnedDocument } from '../documents/delete-owned'; type AudiobookRow = { id: string; @@ -106,7 +107,7 @@ export async function claimAnonymousData(userId: string, unclaimedUserId: string ]); const [documentsClaimed, audiobooksClaimed, preferencesClaimed, progressClaimed] = await Promise.all([ - transferUserDocuments(unclaimedUserId, userId), + transferUserDocuments(unclaimedUserId, userId, { namespace }), transferUserAudiobooks(unclaimedUserId, userId, namespace), transferUserPreferences(unclaimedUserId, userId), transferUserProgress(unclaimedUserId, userId), @@ -154,7 +155,7 @@ export async function transferUserDocuments( fromUserId: string, toUserId: string, // eslint-disable-next-line @typescript-eslint/no-explicit-any - options?: { db?: any }, + options?: { db?: any; namespace?: string | null }, ): Promise { if (!fromUserId || !toUserId) return 0; if (fromUserId === toUserId) return 0; @@ -164,6 +165,21 @@ export async function transferUserDocuments( const rows = await database.select().from(documents).where(eq(documents.userId, fromUserId)); if (rows.length === 0) return 0; + if (!options?.db && isS3Configured()) { + for (const row of rows) { + await db + .insert(documents) + .values({ ...row, userId: toUserId }) + .onConflictDoNothing(); + await deleteOwnedDocument({ + userId: fromUserId, + documentId: row.id, + namespace: options?.namespace ?? null, + }); + } + return rows.length; + } + await database .insert(documents) // eslint-disable-next-line @typescript-eslint/no-explicit-any diff --git a/src/lib/server/user/data-cleanup.ts b/src/lib/server/user/data-cleanup.ts index 3787420..b90bffe 100644 --- a/src/lib/server/user/data-cleanup.ts +++ b/src/lib/server/user/data-cleanup.ts @@ -21,6 +21,7 @@ import { audiobookPrefix, deleteAudiobookPrefix } from '@/lib/server/audiobooks/ import { deleteTtsSegmentPrefix } from '@/lib/server/tts/segments-blobstore'; import { hashForLog, serverLogger } from '@/lib/server/logger'; import { logDegraded } from '@/lib/server/errors/logging'; +import { withDocumentMutationLock } from '@/lib/server/documents/mutation-lock'; type DocumentRow = typeof documents.$inferSelect; type AudiobookRow = { id: string }; @@ -114,68 +115,72 @@ export async function deleteUserStorageData( }; for (const doc of userDocs) { - const { removedDoc, isLastOwner } = await removeOwnershipAndCheckLastOwner(doc); - if (!removedDoc) continue; - removedDocs.push(removedDoc); + await withDocumentMutationLock(doc.id, async () => { + const { removedDoc, isLastOwner } = await removeOwnershipAndCheckLastOwner(doc); + if (!removedDoc) return; + removedDocs.push(removedDoc); - if (s3Enabled && isLastOwner) { - try { - await deleteDocumentBlob(doc.id, namespace); - docsDeleted++; - } catch (error) { - failures.push(error); - logDegraded(serverLogger, { - event: 'user.data_cleanup.document_blob_delete.failed', - msg: 'Failed to delete document blob', - step: 'delete_document_blob', - context: { - documentId: doc.id, - userIdHash: hashForLog(userId), - }, - error, - }); + if (s3Enabled && isLastOwner) { + try { + await deleteDocumentPreviewArtifacts(doc.id, namespace); + } catch (error) { + failures.push(error); + logDegraded(serverLogger, { + event: 'user.data_cleanup.document_preview_delete.failed', + msg: 'Failed to delete preview artifacts', + step: 'delete_document_preview_artifacts', + context: { + documentId: doc.id, + userIdHash: hashForLog(userId), + }, + error, + }); + } } - try { - await deleteDocumentPreviewArtifacts(doc.id, namespace); - } catch (error) { - failures.push(error); - logDegraded(serverLogger, { - event: 'user.data_cleanup.document_preview_delete.failed', - msg: 'Failed to delete preview artifacts', - step: 'delete_document_preview_artifacts', - context: { - documentId: doc.id, - userIdHash: hashForLog(userId), - }, - error, - }); + // Preview metadata is global, so only the canonical final-owner pass may + // remove it. + if (namespace === null && isLastOwner) { + try { + await deleteDocumentPreviewRows(doc.id, namespace); + } catch (error) { + failures.push(error); + logDegraded(serverLogger, { + event: 'user.data_cleanup.document_preview_rows_delete.failed', + msg: 'Failed to delete preview rows', + step: 'delete_document_preview_rows', + context: { + documentId: doc.id, + userIdHash: hashForLog(userId), + }, + error, + }); + } } - } - // Preview metadata is global, so only the canonical final-owner pass may - // remove it. - if (namespace === null && isLastOwner) { - try { - await deleteDocumentPreviewRows(doc.id, namespace); - } catch (error) { - failures.push(error); - logDegraded(serverLogger, { - event: 'user.data_cleanup.document_preview_rows_delete.failed', - msg: 'Failed to delete preview rows', - step: 'delete_document_preview_rows', - context: { - documentId: doc.id, - userIdHash: hashForLog(userId), - }, - error, - }); + if (s3Enabled && isLastOwner) { + try { + await deleteDocumentBlob(doc.id, namespace); + docsDeleted++; + } catch (error) { + failures.push(error); + logDegraded(serverLogger, { + event: 'user.data_cleanup.document_blob_delete.failed', + msg: 'Failed to delete document blob', + step: 'delete_document_blob', + context: { + documentId: doc.id, + userIdHash: hashForLog(userId), + }, + error, + }); + } } - } - if (namespace !== null) { - await restoreRemovedDocs(); - } + if (namespace !== null) { + await restoreRemovedDocs(); + } + }); } // --- Audiobooks --- diff --git a/tests/unit/document-delete-cleanup.vitest.spec.ts b/tests/unit/document-delete-cleanup.vitest.spec.ts new file mode 100644 index 0000000..60feca0 --- /dev/null +++ b/tests/unit/document-delete-cleanup.vitest.spec.ts @@ -0,0 +1,135 @@ +import { beforeEach, describe, expect, test, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + selectResults: [] as unknown[][], + deleteResults: [] as unknown[][], + insertValues: vi.fn(() => ({ onConflictDoNothing: vi.fn(async () => undefined) })), + deleteDocumentBlob: vi.fn(async () => undefined), + cleanupDocumentPreviewArtifacts: vi.fn(async () => undefined), + deleteDocumentPreviewRows: vi.fn(async () => undefined), + deleteDocumentTtsSegmentCache: vi.fn(async () => undefined), +})); + +function resultBuilder(result: unknown[]) { + const limited = { + all: () => result, + then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) => + Promise.resolve(result).then(resolve, reject), + }; + return { + all: () => result, + limit: () => limited, + then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) => + Promise.resolve(result).then(resolve, reject), + }; +} + +vi.mock('@/db', () => { + const database = { + select: vi.fn(() => ({ + from: vi.fn(() => ({ + where: vi.fn(() => resultBuilder(mocks.selectResults.shift() ?? [])), + })), + })), + delete: vi.fn(() => ({ + where: vi.fn(() => ({ + returning: () => resultBuilder(mocks.deleteResults.shift() ?? []), + })), + })), + insert: vi.fn(() => ({ + values: mocks.insertValues, + })), + transaction: vi.fn((callback: (tx: unknown) => unknown) => callback(database)), + }; + return { db: database }; +}); + +vi.mock('@/lib/server/documents/blobstore', () => ({ + deleteDocumentBlob: mocks.deleteDocumentBlob, +})); + +vi.mock('@/lib/server/documents/previews', () => ({ + cleanupDocumentPreviewArtifacts: mocks.cleanupDocumentPreviewArtifacts, + deleteDocumentPreviewRows: mocks.deleteDocumentPreviewRows, +})); + +vi.mock('@/lib/server/tts/segments-cache', () => ({ + deleteDocumentTtsSegmentCache: mocks.deleteDocumentTtsSegmentCache, +})); + +import { deleteOwnedDocument } from '../../src/lib/server/documents/delete-owned'; + +describe('owned document cleanup', () => { + beforeEach(() => { + mocks.selectResults = []; + mocks.deleteResults = []; + mocks.insertValues.mockReset(); + mocks.insertValues.mockReturnValue({ onConflictDoNothing: vi.fn(async () => undefined) }); + mocks.deleteDocumentBlob.mockReset(); + mocks.deleteDocumentBlob.mockResolvedValue(undefined); + mocks.cleanupDocumentPreviewArtifacts.mockReset(); + mocks.cleanupDocumentPreviewArtifacts.mockResolvedValue(undefined); + mocks.deleteDocumentPreviewRows.mockReset(); + mocks.deleteDocumentPreviewRows.mockResolvedValue(undefined); + mocks.deleteDocumentTtsSegmentCache.mockReset(); + mocks.deleteDocumentTtsSegmentCache.mockResolvedValue(undefined); + }); + + test('deletes only user-scoped TTS when another owner remains', async () => { + mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]]; + mocks.selectResults = [[{ id: 'doc-1' }]]; + + await expect(deleteOwnedDocument({ + userId: 'user-1', + documentId: 'doc-1', + namespace: null, + })).resolves.toBe(true); + + expect(mocks.deleteDocumentTtsSegmentCache).toHaveBeenCalledOnce(); + expect(mocks.deleteDocumentBlob).not.toHaveBeenCalled(); + expect(mocks.cleanupDocumentPreviewArtifacts).not.toHaveBeenCalled(); + }); + + test('deletes shared artifacts only for the final owner', async () => { + mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]]; + mocks.selectResults = [[], []]; + + await deleteOwnedDocument({ + userId: 'user-1', + documentId: 'doc-1', + namespace: null, + }); + + expect(mocks.cleanupDocumentPreviewArtifacts).toHaveBeenCalledWith('doc-1', null); + expect(mocks.deleteDocumentPreviewRows).toHaveBeenCalledWith('doc-1', null); + expect(mocks.deleteDocumentBlob).toHaveBeenCalledWith('doc-1', null); + }); + + test('restores ownership when cleanup fails', async () => { + const removed = { id: 'doc-1', userId: 'user-1' }; + mocks.deleteResults = [[removed]]; + mocks.selectResults = [[], []]; + mocks.deleteDocumentBlob.mockRejectedValueOnce(new Error('storage unavailable')); + + await expect(deleteOwnedDocument({ + userId: 'user-1', + documentId: 'doc-1', + namespace: null, + })).rejects.toThrow('storage unavailable'); + + expect(mocks.insertValues).toHaveBeenCalledWith(removed); + }); + + test('keeps shared artifacts when a new owner appears before deletion', async () => { + mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]]; + mocks.selectResults = [[], [{ id: 'doc-1' }]]; + + await deleteOwnedDocument({ + userId: 'user-1', + documentId: 'doc-1', + namespace: null, + }); + + expect(mocks.deleteDocumentBlob).not.toHaveBeenCalled(); + }); +}); diff --git a/tests/unit/storage-prefix-cleanup.vitest.spec.ts b/tests/unit/storage-prefix-cleanup.vitest.spec.ts index 0bc77fc..6267527 100644 --- a/tests/unit/storage-prefix-cleanup.vitest.spec.ts +++ b/tests/unit/storage-prefix-cleanup.vitest.spec.ts @@ -11,7 +11,7 @@ vi.mock('@/lib/server/storage/s3', () => ({ })); import { deleteAudiobookPrefix } from '../../src/lib/server/audiobooks/blobstore'; -import { deleteDocumentPrefix } from '../../src/lib/server/documents/blobstore'; +import { deleteDocumentBlob, deleteDocumentPrefix } from '../../src/lib/server/documents/blobstore'; describe('storage prefix cleanup', () => { beforeEach(() => { @@ -47,4 +47,42 @@ describe('storage prefix cleanup', () => { await expect(removePrefix('prefix/')).rejects.toThrow('Failed deleting 1'); }); + + test('deletes the source after derived artifacts and then sweeps late parsed output', async () => { + mocks.send + .mockResolvedValueOnce({ Contents: [], IsTruncated: false }) + .mockResolvedValueOnce({ Contents: [], IsTruncated: false }) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({ Contents: [], IsTruncated: false }); + + const documentId = 'a'.repeat(64); + await deleteDocumentBlob(documentId, null); + + const commands = mocks.send.mock.calls.map(([command]) => command); + const sourceDeleteIndex = commands.findIndex((command) => + command.constructor.name === 'DeleteObjectCommand' + && command.input.Key === `openreader-test/documents_v1/${documentId}`); + const finalCommand = commands.at(-1); + + expect(sourceDeleteIndex).toBeGreaterThan(1); + expect(finalCommand?.constructor.name).toBe('ListObjectsV2Command'); + expect(finalCommand?.input.Prefix).toBe( + `openreader-test/documents_v1/parsed_v2/${documentId}/`, + ); + }); + + test('keeps the source document when derived-artifact cleanup fails', async () => { + mocks.send.mockRejectedValueOnce(new Error('list failed')); + + const documentId = 'b'.repeat(64); + await expect(deleteDocumentBlob(documentId, null)).rejects.toThrow('list failed'); + + const sourceDelete = mocks.send.mock.calls + .map(([command]) => command) + .find((command) => command.constructor.name === 'DeleteObjectCommand' + && command.input.Key === `openreader-test/documents_v1/${documentId}`); + expect(sourceDelete).toBeUndefined(); + }); }); diff --git a/tests/unit/user-data-cleanup.vitest.spec.ts b/tests/unit/user-data-cleanup.vitest.spec.ts index f862813..05cee91 100644 --- a/tests/unit/user-data-cleanup.vitest.spec.ts +++ b/tests/unit/user-data-cleanup.vitest.spec.ts @@ -52,6 +52,7 @@ vi.mock('@/db', () => { insert: vi.fn(() => ({ values: mocks.insertValues, })), + execute: vi.fn(async () => undefined), transaction: vi.fn((callback: (tx: unknown) => unknown) => callback(database)), }; return { db: database };