From c0de2156fe9074864a27e8c4d39df144f97c12a8 Mon Sep 17 00:00:00 2001 From: Richard R Date: Sat, 6 Jun 2026 18:09:47 -0600 Subject: [PATCH] refactor(documents): centralize owned document deletion and add mutation locking Introduce `deleteOwnedDocument` utility to encapsulate all logic for removing a user's ownership of a document, including TTS segment cache cleanup, preview artifact removal, and S3 blob deletion for last-owner cases. Add `withDocumentMutationLock` to serialize concurrent mutations on the same document, using advisory locks in Postgres and a local queue fallback. Refactor all API routes and user data flows to use these utilities, ensuring transactional safety and preventing race conditions during document deletion or transfer. Update tests for new flows and add coverage for document cleanup sequencing and locking behavior. --- .../worker/src/pdf-artifact-persistence.ts | 18 +++ compute/worker/src/runtime.ts | 52 ++++++- .../unit/pdf-artifact-persistence.test.ts | 33 +++++ .../api/documents/blob/get/fallback/route.ts | 7 +- .../documents/blob/preview/fallback/route.ts | 7 +- .../documents/blob/upload/finalize/route.ts | 79 +++++----- src/app/api/documents/route.ts | 65 ++------- src/app/api/tts/segments/ensure/route.ts | 36 ++++- src/lib/server/documents/blobstore.ts | 9 +- src/lib/server/documents/delete-owned.ts | 111 ++++++++++++++ src/lib/server/documents/mutation-lock.ts | 40 ++++++ src/lib/server/user/claim-data.ts | 20 ++- src/lib/server/user/data-cleanup.ts | 115 ++++++++------- .../document-delete-cleanup.vitest.spec.ts | 135 ++++++++++++++++++ .../storage-prefix-cleanup.vitest.spec.ts | 40 +++++- tests/unit/user-data-cleanup.vitest.spec.ts | 1 + 16 files changed, 599 insertions(+), 169 deletions(-) create mode 100644 compute/worker/src/pdf-artifact-persistence.ts create mode 100644 compute/worker/tests/unit/pdf-artifact-persistence.test.ts create mode 100644 src/lib/server/documents/delete-owned.ts create mode 100644 src/lib/server/documents/mutation-lock.ts create mode 100644 tests/unit/document-delete-cleanup.vitest.spec.ts 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 };