From f555f080fe8c03f17515407b2aa5dd2892684b8e Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Mon, 31 Aug 2026 02:07:33 +0000 Subject: [PATCH] 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 --- .../migration-executor.service.test.ts | 3 +- .../migration-import.repository.test.ts | 42 +++++++++------- .../executor/migration-import.repository.ts | 48 ++++++++++++------ .../executor/user-state.importer.test.ts | 14 +++--- .../migration/executor/user-state.importer.ts | 21 +++++--- .../test/migration-audiobookshelf.e2e-spec.ts | 49 +++++++++++++++++++ 6 files changed, 127 insertions(+), 50 deletions(-) diff --git a/server/src/modules/migration/executor/migration-executor.service.test.ts b/server/src/modules/migration/executor/migration-executor.service.test.ts index 4d2701739..fcc776752 100644 --- a/server/src/modules/migration/executor/migration-executor.service.test.ts +++ b/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())), 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()), diff --git a/server/src/modules/migration/executor/migration-import.repository.test.ts b/server/src/modules/migration/executor/migration-import.repository.test.ts index a7be57bf4..edd63f4df 100644 --- a/server/src/modules/migration/executor/migration-import.repository.test.ts +++ b/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(); }); }); diff --git a/server/src/modules/migration/executor/migration-import.repository.ts b/server/src/modules/migration/executor/migration-import.repository.ts index 4bb67c6a7..45cc88e75 100644 --- a/server/src/modules/migration/executor/migration-import.repository.ts +++ b/server/src/modules/migration/executor/migration-import.repository.ts @@ -20,6 +20,22 @@ type Db = NodePgDatabase; const BATCH_CHUNK_SIZE = 500; +type UserBookStatusMergeItem = Omit & { + 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>(obj: T, ...keys: string[]): Partial { const keySet = new Set(keys); const result = {} as Partial; @@ -197,19 +213,6 @@ export class MigrationImportRepository { // --- User book statuses --- - async clearUserBookStatuses(userIds: number[], bookIds: number[]): Promise { - 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 { await this.db .insert(schema.userBookStatus) @@ -501,12 +504,16 @@ export class MigrationImportRepository { } } - async batchUpsertUserBookStatuses(items: Array): Promise { + async batchMergeUserBookStatuses(items: UserBookStatusMergeItem[]): Promise { 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): Promise { diff --git a/server/src/modules/migration/executor/user-state.importer.test.ts b/server/src/modules/migration/executor/user-state.importer.test.ts index 9f8b149f1..b42fd2ed4 100644 --- a/server/src/modules/migration/executor/user-state.importer.test.ts +++ b/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), - batchUpsertUserBookStatuses: vi.fn().mockResolvedValue(undefined), + batchMergeUserBookStatuses: 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<{ diff --git a/server/src/modules/migration/executor/user-state.importer.ts b/server/src/modules/migration/executor/user-state.importer.ts index 07a60c45b..d1002f971 100644 --- a/server/src/modules/migration/executor/user-state.importer.ts +++ b/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.batchUpsertUserBookStatuses(dedupedBatch); + await importRepo.batchMergeUserBookStatuses(dedupedBatch); }); await this.repo.setRunMetric(runId, 'user_state', 'user_book_status', counters); } @@ -897,6 +896,14 @@ function preferLatestByUpdatedAt(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(); diff --git a/server/test/migration-audiobookshelf.e2e-spec.ts b/server/test/migration-audiobookshelf.e2e-spec.ts index 07f2cd4a9..bd4c7bda1 100644 --- a/server/test/migration-audiobookshelf.e2e-spec.ts +++ b/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();