fix(server): merge imported read statuses safely

Preserve target-only statuses and update conflicts only when the source carries newer evidence. Treat timestamp-free source rows as insert-only so they cannot overwrite curated target state.

Fixes #1104

server/src/modules/migration/executor/migration-executor.service.test.ts

@@ -50,7 +50,6 @@ function createImportRepoMock() {
     upsertTag: vi.fn(),
     insertBookTag: vi.fn(() => Promise.resolve()),
     markCoverAsCustom: vi.fn(() => Promise.resolve()),
-    clearUserBookStatuses: vi.fn(() => Promise.resolve()),
     upsertUserBookStatus: vi.fn(() => Promise.resolve()),
     clearReadingProgress: vi.fn(() => Promise.resolve()),
     upsertReadingProgress: vi.fn(() => Promise.resolve()),
@@ -86,7 +85,7 @@ function createImportRepoMock() {
     batchDeleteBookTags: vi.fn(() => Promise.resolve()),
     batchUpsertTags: vi.fn(() => Promise.resolve(new Map<string, number>())),
     batchInsertBookTags: vi.fn(() => Promise.resolve()),
-    batchUpsertUserBookStatuses: vi.fn(() => Promise.resolve()),
+    batchMergeUserBookStatuses: vi.fn(() => Promise.resolve()),
     batchUpsertReadingProgress: vi.fn(() => Promise.resolve()),
     batchUpsertAudiobookProgress: vi.fn(() => Promise.resolve()),
     syncImportedReadingSessions: vi.fn(() => Promise.resolve()),

server/src/modules/migration/executor/migration-import.repository.test.ts

@@ -83,20 +83,6 @@ describe('MigrationImportRepository', () => {
     );
   });
 
-  it('clearUserBookStatuses no-ops for empty targets and deduplicates ids otherwise', async () => {
-    const where = vi.fn().mockResolvedValue(undefined);
-    const deleteFn = vi.fn().mockReturnValue({ where });
-    const repo = new MigrationImportRepository({ delete: deleteFn } as never);
-
-    await repo.clearUserBookStatuses([], [1, 2]);
-    await repo.clearUserBookStatuses([1], []);
-    expect(deleteFn).not.toHaveBeenCalled();
-
-    await repo.clearUserBookStatuses([1, 1, 2], [10, 10, 11]);
-    expect(deleteFn).toHaveBeenCalledTimes(1);
-    expect(where).toHaveBeenCalledTimes(1);
-  });
-
   it('clearUserBookRatings no-ops for empty targets and deduplicates ids otherwise', async () => {
     const where = vi.fn().mockResolvedValue(undefined);
     const deleteFn = vi.fn().mockReturnValue({ where });
@@ -111,14 +97,14 @@ describe('MigrationImportRepository', () => {
     expect(where).toHaveBeenCalledTimes(1);
   });
 
-  it('chunks user/book clear queries for large migration runs', async () => {
+  it('chunks user/book rating clear queries for large migration runs', async () => {
     const where = vi.fn().mockResolvedValue(undefined);
     const deleteFn = vi.fn().mockReturnValue({ where });
     const repo = new MigrationImportRepository({ delete: deleteFn } as never);
     const userIds = Array.from({ length: 501 }, (_, index) => index + 1);
     const bookIds = Array.from({ length: 1001 }, (_, index) => index + 1);
 
-    await repo.clearUserBookStatuses(userIds, bookIds);
+    await repo.clearUserBookRatings(userIds, bookIds);
 
     expect(deleteFn).toHaveBeenCalledTimes(6);
     expect(where).toHaveBeenCalledTimes(6);
@@ -549,7 +535,7 @@ describe('MigrationImportRepository', () => {
     const insert = vi.fn().mockReturnValue({ values });
     const repo = new MigrationImportRepository({ insert } as never);
 
-    await repo.batchUpsertUserBookStatuses([]);
+    await repo.batchMergeUserBookStatuses([]);
     await repo.batchUpsertReadingProgress([]);
     await repo.batchUpsertAudiobookProgress([]);
     await repo.batchInsertBookmarks([]);
@@ -557,7 +543,9 @@ describe('MigrationImportRepository', () => {
     await repo.batchInsertCollectionBooks([]);
     expect(insert).not.toHaveBeenCalled();
 
-    await repo.batchUpsertUserBookStatuses([{ userId: 1, bookId: 2, status: 'read', source: 'manual', updatedAt: new Date() } as never]);
+    await repo.batchMergeUserBookStatuses([
+      { userId: 1, bookId: 2, status: 'read', source: 'manual', sourceUpdatedAt: new Date('2026-01-02T00:00:00.000Z') },
+    ]);
     await repo.batchUpsertReadingProgress([{ userId: 1, bookFileId: 10, percentage: 50, updatedAt: new Date() } as never]);
     await repo.batchUpsertAudiobookProgress([
       { userId: 1, bookId: 2, percentage: 25, currentFileId: 10, positionSeconds: 0, updatedAt: new Date() } as never,
@@ -573,5 +561,23 @@ describe('MigrationImportRepository', () => {
     expect(insert).toHaveBeenCalledWith(schema.annotations);
     expect(insert).toHaveBeenCalledWith(schema.annotationPositions);
     expect(insert).toHaveBeenCalledWith(schema.collectionBooks);
+    expect(onConflictDoUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        target: [schema.userBookStatus.userId, schema.userBookStatus.bookId],
+        setWhere: expect.anything(),
+      }),
+    );
+  });
+
+  it('inserts statuses without source timestamps but does not update conflicts', async () => {
+    const onConflictDoNothing = vi.fn().mockResolvedValue(undefined);
+    const values = vi.fn().mockReturnValue({ onConflictDoNothing });
+    const insert = vi.fn().mockReturnValue({ values });
+    const repo = new MigrationImportRepository({ insert } as never);
+
+    await repo.batchMergeUserBookStatuses([{ userId: 1, bookId: 2, status: 'reading', source: 'manual', sourceUpdatedAt: null }]);
+
+    expect(values).toHaveBeenCalledWith([expect.objectContaining({ userId: 1, bookId: 2, status: 'reading', updatedAt: expect.any(Date) })]);
+    expect(onConflictDoNothing).toHaveBeenCalledOnce();
   });
 });

server/src/modules/migration/executor/migration-import.repository.ts

@@ -20,6 +20,22 @@ type Db = NodePgDatabase<typeof schema>;
 
 const BATCH_CHUNK_SIZE = 500;
 
+type UserBookStatusMergeItem = Omit<typeof schema.userBookStatus.$inferInsert, 'updatedAt'> & {
+  sourceUpdatedAt: Date | null;
+};
+
+function userBookStatusMergeValues(item: UserBookStatusMergeItem, updatedAt: Date): typeof schema.userBookStatus.$inferInsert {
+  return {
+    userId: item.userId,
+    bookId: item.bookId,
+    status: item.status,
+    source: item.source,
+    startedAt: item.startedAt,
+    finishedAt: item.finishedAt,
+    updatedAt,
+  };
+}
+
 function omitKeys<T extends Record<string, unknown>>(obj: T, ...keys: string[]): Partial<T> {
   const keySet = new Set(keys);
   const result = {} as Partial<T>;
@@ -197,19 +213,6 @@ export class MigrationImportRepository {
 
   // --- User book statuses ---
 
-  async clearUserBookStatuses(userIds: number[], bookIds: number[]): Promise<void> {
-    const targetUserIds = uniqueNumbers(userIds);
-    const targetBookIds = uniqueNumbers(bookIds);
-    if (targetUserIds.length === 0 || targetBookIds.length === 0) return;
-    for (const userBatch of chunk(targetUserIds, BATCH_CHUNK_SIZE)) {
-      for (const bookBatch of chunk(targetBookIds, BATCH_CHUNK_SIZE)) {
-        await this.db
-          .delete(schema.userBookStatus)
-          .where(and(inArray(schema.userBookStatus.userId, userBatch), inArray(schema.userBookStatus.bookId, bookBatch)));
-      }
-    }
-  }
-
   async upsertUserBookStatus(values: typeof schema.userBookStatus.$inferInsert): Promise<void> {
     await this.db
       .insert(schema.userBookStatus)
@@ -501,12 +504,16 @@ export class MigrationImportRepository {
     }
   }
 
-  async batchUpsertUserBookStatuses(items: Array<typeof schema.userBookStatus.$inferInsert>): Promise<void> {
+  async batchMergeUserBookStatuses(items: UserBookStatusMergeItem[]): Promise<void> {
     if (items.length === 0) return;
-    for (const batch of chunk(items, BATCH_CHUNK_SIZE)) {
+
+    const timestamped = items.filter((item): item is UserBookStatusMergeItem & { sourceUpdatedAt: Date } => item.sourceUpdatedAt !== null);
+    const untimestamped = items.filter((item) => item.sourceUpdatedAt === null);
+
+    for (const batch of chunk(timestamped, BATCH_CHUNK_SIZE)) {
       await this.db
         .insert(schema.userBookStatus)
-        .values(batch)
+        .values(batch.map((item) => userBookStatusMergeValues(item, item.sourceUpdatedAt)))
         .onConflictDoUpdate({
           target: [schema.userBookStatus.userId, schema.userBookStatus.bookId],
           set: {
@@ -516,8 +523,17 @@ export class MigrationImportRepository {
             finishedAt: sql`excluded.finished_at`,
             updatedAt: sql`excluded.updated_at`,
           },
+          setWhere: sql`excluded.updated_at > ${schema.userBookStatus.updatedAt}`,
         });
     }
+
+    for (const batch of chunk(untimestamped, BATCH_CHUNK_SIZE)) {
+      const insertedAt = new Date();
+      await this.db
+        .insert(schema.userBookStatus)
+        .values(batch.map((item) => userBookStatusMergeValues(item, insertedAt)))
+        .onConflictDoNothing();
+    }
   }
 
   async batchUpsertUserBookRatings(items: Array<typeof schema.userBookRatings.$inferInsert>): Promise<void> {

server/src/modules/migration/executor/user-state.importer.test.ts

@@ -15,8 +15,7 @@ function makeImporter() {
       .mockResolvedValue(
         new Map([[200, [{ id: 500, hash: null, absolutePath: '/target/book.epub', format: 'epub', sortOrder: 0, durationSeconds: null }]]]),
       ),
-    clearUserBookStatuses: vi.fn().mockResolvedValue(undefined),
+    batchMergeUserBookStatuses: vi.fn().mockResolvedValue(undefined),
-    batchUpsertUserBookStatuses: vi.fn().mockResolvedValue(undefined),
     clearUserBookRatings: vi.fn().mockResolvedValue(undefined),
     batchUpsertUserBookRatings: vi.fn().mockResolvedValue(undefined),
     clearReadingProgress: vi.fn().mockResolvedValue(undefined),
@@ -141,12 +140,13 @@ describe('UserStateImporter', () => {
     const ensureRunning = vi.fn().mockResolvedValue(undefined);
     await importer.import(300, planned as never, ensureRunning);
 
-    expect(importRepo.batchUpsertUserBookStatuses).toHaveBeenCalledWith([
+    expect(importRepo.batchMergeUserBookStatuses).toHaveBeenCalledWith([
       expect.objectContaining({
         userId: 10,
         bookId: 200,
         status: 'read',
         source: 'manual',
+        sourceUpdatedAt: new Date('2026-01-02T00:00:00.000Z'),
       }),
     ]);
     expect(importRepo.batchUpsertUserBookRatings).toHaveBeenCalledWith([
@@ -477,7 +477,7 @@ describe('UserStateImporter', () => {
 
     await importer.import(302, planned as never, vi.fn().mockResolvedValue(undefined));
 
-    expect(importRepo.batchUpsertUserBookStatuses).not.toHaveBeenCalled();
+    expect(importRepo.batchMergeUserBookStatuses).not.toHaveBeenCalled();
     expect(importRepo.batchUpsertUserBookRatings).not.toHaveBeenCalled();
     expect(importRepo.batchUpsertReadingProgress).not.toHaveBeenCalled();
     expect(importRepo.batchUpsertAudiobookProgress).not.toHaveBeenCalled();
@@ -975,18 +975,18 @@ describe('UserStateImporter', () => {
 
     await importer.import(306, planned as never, vi.fn().mockResolvedValue(undefined));
 
-    const statusBatch = importRepo.batchUpsertUserBookStatuses.mock.calls[0]?.[0] as Array<{
+    const statusBatch = importRepo.batchMergeUserBookStatuses.mock.calls[0]?.[0] as Array<{
       userId: number;
       bookId: number;
       status: string;
-      updatedAt: Date;
+      sourceUpdatedAt: Date;
     }>;
     expect(statusBatch).toHaveLength(1);
     expect(statusBatch[0]).toMatchObject({
       userId: 10,
       bookId: 200,
       status: 'read',
-      updatedAt: new Date(newerStatusUpdatedAt),
+      sourceUpdatedAt: new Date(newerStatusUpdatedAt),
     });
 
     const ratingBatch = importRepo.batchUpsertUserBookRatings.mock.calls[0]?.[0] as Array<{

server/src/modules/migration/executor/user-state.importer.ts

@@ -55,14 +55,14 @@ function normalizeSourceRating(value: number | null | undefined): number | null
   return null;
 }
 
-type UserBookStatusUpsert = {
+type UserBookStatusMerge = {
   userId: number;
   bookId: number;
   status: ReadStatus;
   source: ReadStatusSource;
   startedAt: Date | null;
   finishedAt: Date | null;
-  updatedAt: Date;
+  sourceUpdatedAt: Date | null;
 };
 
 type UserBookRatingUpsert = {
@@ -181,7 +181,7 @@ export class UserStateImporter {
       return;
     }
 
-    const batch: UserBookStatusUpsert[] = [];
+    const batch: UserBookStatusMerge[] = [];
 
     for (const row of planned.execution.sourceData.userBookStatuses) {
       await ensureRunning();
@@ -200,15 +200,14 @@ export class UserStateImporter {
         source: 'manual',
         startedAt: toDate(row.startedAt),
         finishedAt: toDate(row.finishedAt),
-        updatedAt: toDate(row.updatedAt) ?? new Date(),
+        sourceUpdatedAt: toDate(row.updatedAt),
       });
       counters.imported += 1;
     }
-    const dedupedBatch = dedupeByKey(batch, (item) => `${item.userId}:${item.bookId}`, preferLatestByUpdatedAt);
+    const dedupedBatch = dedupeByKey(batch, (item) => `${item.userId}:${item.bookId}`, preferUserBookStatusEvidence);
 
     await this.importRepo.withTransaction(async (importRepo) => {
-      await importRepo.clearUserBookStatuses([...userMap.values()], [...bookMap.values()]);
+      await importRepo.batchMergeUserBookStatuses(dedupedBatch);
-      await importRepo.batchUpsertUserBookStatuses(dedupedBatch);
     });
     await this.repo.setRunMetric(runId, 'user_state', 'user_book_status', counters);
   }
@@ -897,6 +896,14 @@ function preferLatestByUpdatedAt<T extends { updatedAt: Date }>(current: T, cand
   return candidate;
 }
 
+function preferUserBookStatusEvidence(current: UserBookStatusMerge, candidate: UserBookStatusMerge): UserBookStatusMerge {
+  if (current.sourceUpdatedAt === null) return candidate;
+  if (candidate.sourceUpdatedAt === null) return current;
+  if (candidate.sourceUpdatedAt > current.sourceUpdatedAt) return candidate;
+  if (candidate.sourceUpdatedAt < current.sourceUpdatedAt) return current;
+  return candidate;
+}
+
 function preferReadingProgressRow(current: ReadingProgressUpsert, candidate: ReadingProgressUpsert): ReadingProgressUpsert {
   const currentTs = current.updatedAt.getTime();
   const candidateTs = candidate.updatedAt.getTime();

server/test/migration-audiobookshelf.e2e-spec.ts

@@ -135,6 +135,31 @@ describe('Migration Audiobookshelf live and backup (e2e)', { timeout: 300_000 },
     const backupExport = await adapter.exportData(backupConfig as AudiobookshelfConnectionConfig);
     expect(canonicalizeExport(backupExport)).toEqual(canonicalizeExport(liveExport));
 
+    await ctx.db.insert(schema.userBookStatus).values([
+      {
+        userId: scenario.targetUsers.theo.id,
+        bookId: scenario.targetBooks.glass.bookId,
+        status: 'read',
+        source: 'manual',
+        finishedAt: new Date('2024-01-01T00:00:00.000Z'),
+        updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      },
+      {
+        userId: scenario.targetUsers.maya.id,
+        bookId: scenario.targetBooks.glass.bookId,
+        status: 'want_to_read',
+        source: 'manual',
+        updatedAt: new Date('2099-01-01T00:00:00.000Z'),
+      },
+      {
+        userId: scenario.targetUsers.theo.id,
+        bookId: scenario.targetBooks.map.bookId,
+        status: 'want_to_read',
+        source: 'manual',
+        updatedAt: new Date('2020-01-01T00:00:00.000Z'),
+      },
+    ]);
+
     const started = await apiJson<{ id: number; state: string }>(ctx, {
       method: 'POST',
       url: '/api/v1/migration/runs/live',
@@ -149,6 +174,30 @@ describe('Migration Audiobookshelf live and backup (e2e)', { timeout: 300_000 },
     expect(finished.report.totals.failed).toBe(0);
     expect(finished.report.totals.unresolved).toBeGreaterThanOrEqual(1);
 
+    const mergedStatuses = await ctx.db
+      .select()
+      .from(schema.userBookStatus)
+      .where(inArray(schema.userBookStatus.userId, [scenario.targetUsers.maya.id, scenario.targetUsers.theo.id]));
+    expect(mergedStatuses).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          userId: scenario.targetUsers.theo.id,
+          bookId: scenario.targetBooks.glass.bookId,
+          status: 'read',
+        }),
+        expect.objectContaining({
+          userId: scenario.targetUsers.maya.id,
+          bookId: scenario.targetBooks.glass.bookId,
+          status: 'want_to_read',
+        }),
+        expect.objectContaining({
+          userId: scenario.targetUsers.theo.id,
+          bookId: scenario.targetBooks.map.bookId,
+          status: 'read',
+        }),
+      ]),
+    );
+
     await assertImportedState();
     const countsAfterFirstRun = await loadStateCounts();