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.

compute/worker/src/pdf-artifact-persistence.ts

@@ -0,0 +1,18 @@
+export async function persistParsedPdfWhileSourceExists(input: {
+  sourceObjectKey: string;
+  sourceExists: (key: string) => Promise<boolean>;
+  putParsedObject: () => Promise<string>;
+  deleteParsedObject: (key: string) => Promise<void>;
+}): Promise<string> {
+  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;
+}

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<boolean> => {
+    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<void> => {
+    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<string> => {
     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: {

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');
+  });
+});

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;

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;

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(

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',

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<boolean> => {
+      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<void> => {
+      if (!await documentOwnershipExists()) {
+        throw new Error('Document ownership was removed during TTS generation');
+      }
+    };
+    const deleteAbandonedSegmentMetadata = async (): Promise<void> => {
+      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<void> => {
+      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;

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<void> {

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<void> {
+  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<boolean> {
+  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;
+  });
+}

src/lib/server/documents/mutation-lock.ts

@@ -0,0 +1,40 @@
+import { sql } from 'drizzle-orm';
+import { db } from '@/db';
+
+const localTails = new Map<string, Promise<void>>();
+
+async function withLocalDocumentLock<T>(documentId: string, fn: () => Promise<T>): Promise<T> {
+  const previous = localTails.get(documentId) ?? Promise.resolve();
+  let release!: () => void;
+  const current = new Promise<void>((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<T>(
+  documentId: string,
+  fn: () => Promise<T>,
+): Promise<T> {
+  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();
+    });
+  });
+}

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<number> {
   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

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 ---

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();
+  });
+});

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();
+  });
 });

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 };