mirror of
https://github.com/bookorbit/bookorbit.git
synced 2026-11-03 00:11:37 -06:00
fix(server): deduplicate migration user-state upsert batches (#81)
* fix(server): deduplicate migration user-state upsert batches Prevent migration failures when source data contains duplicate user-state rows for the same target keys. Keep the latest row per upsert key and prefer richer reading progress signals on timestamp ties, then cover the behavior with focused unit and e2e tests. Closes #71 * ci: raise commitlint body line limit to 500 chars Allow longer commit body lines so convention checks do not fail on detailed but still concise commit explanations in this repository.
This commit is contained in:
neon
GitHub
@@ -3,7 +3,7 @@ module.exports = {
|
||||
rules: {
|
||||
"type-enum": [2, "always", ["feat", "fix", "db", "perf", "refactor", "style", "docs", "test", "build", "ci", "chore", "security", "revert"]],
|
||||
"header-max-length": [2, "always", 100],
|
||||
"body-max-line-length": [2, "always", 250],
|
||||
"body-max-line-length": [2, "always", 500],
|
||||
"subject-full-stop": [2, "never", "."],
|
||||
"scope-enum": [0],
|
||||
"scope-empty": [0],
|
||||
|
||||
@@ -511,4 +511,126 @@ describe('UserStateImporter', () => {
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('deduplicates conflicting upsert keys before writing user-state batches', async () => {
|
||||
const { importer, importRepo } = makeImporter();
|
||||
importRepo.fetchTargetBookPrimaryFiles.mockResolvedValue({
|
||||
primaryFilesByBookId: new Map([[200, 500]]),
|
||||
audiobookPrimaryFilesByBookId: new Map([[200, 500]]),
|
||||
});
|
||||
|
||||
const newerStatusUpdatedAt = '2026-01-11T00:00:00.000Z';
|
||||
const tiedProgressUpdatedAt = '2026-01-12T00:00:00.000Z';
|
||||
const planned = {
|
||||
plan: {
|
||||
userMappings: [{ sourceUserId: 'u1', targetUserId: 10 }],
|
||||
pathMappings: [],
|
||||
},
|
||||
execution: {
|
||||
matchedBooks: [{ sourceBookId: 'b-source', targetBookId: 200 }],
|
||||
sourceData: {
|
||||
availableDomains: {
|
||||
userBookStatuses: true,
|
||||
readingProgress: true,
|
||||
bookmarks: false,
|
||||
annotations: false,
|
||||
shelves: false,
|
||||
},
|
||||
books: [{ sourceBookId: 'b-source', files: [] }],
|
||||
userBookStatuses: [
|
||||
{
|
||||
sourceUserId: 'u1',
|
||||
sourceBookId: 'b-source',
|
||||
status: 'reading',
|
||||
percentage: 20,
|
||||
startedAt: null,
|
||||
finishedAt: null,
|
||||
updatedAt: '2026-01-10T00:00:00.000Z',
|
||||
},
|
||||
{
|
||||
sourceUserId: 'u1',
|
||||
sourceBookId: 'b-source',
|
||||
status: 'completed',
|
||||
percentage: 100,
|
||||
startedAt: null,
|
||||
finishedAt: null,
|
||||
updatedAt: newerStatusUpdatedAt,
|
||||
},
|
||||
],
|
||||
userFileProgress: [
|
||||
{
|
||||
sourceUserId: 'u1',
|
||||
sourceBookId: 'b-source',
|
||||
sourceFileId: null,
|
||||
percentage: 20,
|
||||
cfi: null,
|
||||
href: null,
|
||||
pageNumber: 12,
|
||||
positionSeconds: 1,
|
||||
updatedAt: tiedProgressUpdatedAt,
|
||||
},
|
||||
{
|
||||
sourceUserId: 'u1',
|
||||
sourceBookId: 'b-source',
|
||||
sourceFileId: null,
|
||||
percentage: 20,
|
||||
cfi: '/6/8',
|
||||
href: null,
|
||||
pageNumber: null,
|
||||
positionSeconds: 5,
|
||||
updatedAt: tiedProgressUpdatedAt,
|
||||
},
|
||||
],
|
||||
bookmarks: [],
|
||||
annotations: [],
|
||||
shelves: [],
|
||||
shelfBooks: [],
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
await importer.import(306, planned as never, vi.fn().mockResolvedValue(undefined));
|
||||
|
||||
const statusBatch = importRepo.batchUpsertUserBookStatuses.mock.calls[0]?.[0] as Array<{
|
||||
userId: number;
|
||||
bookId: number;
|
||||
status: string;
|
||||
updatedAt: Date;
|
||||
}>;
|
||||
expect(statusBatch).toHaveLength(1);
|
||||
expect(statusBatch[0]).toMatchObject({
|
||||
userId: 10,
|
||||
bookId: 200,
|
||||
status: 'read',
|
||||
updatedAt: new Date(newerStatusUpdatedAt),
|
||||
});
|
||||
|
||||
const readingBatch = importRepo.batchUpsertReadingProgress.mock.calls[0]?.[0] as Array<{
|
||||
userId: number;
|
||||
bookFileId: number;
|
||||
cfi: string | null;
|
||||
pageNumber: number | null;
|
||||
}>;
|
||||
expect(readingBatch).toHaveLength(1);
|
||||
expect(readingBatch[0]).toMatchObject({
|
||||
userId: 10,
|
||||
bookFileId: 500,
|
||||
cfi: '/6/8',
|
||||
pageNumber: null,
|
||||
});
|
||||
|
||||
const audioBatch = importRepo.batchUpsertAudiobookProgress.mock.calls[0]?.[0] as Array<{
|
||||
userId: number;
|
||||
bookId: number;
|
||||
currentFileId: number;
|
||||
positionSeconds: number;
|
||||
}>;
|
||||
expect(audioBatch).toHaveLength(1);
|
||||
expect(audioBatch[0]).toMatchObject({
|
||||
userId: 10,
|
||||
bookId: 200,
|
||||
currentFileId: 500,
|
||||
positionSeconds: 5,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -29,6 +29,35 @@ function hasMeaningfulProgressSignal(row: SourceUserFileProgress, percentage: nu
|
||||
return percentage > 0 || hasLocator || hasPageNumber || hasPosition;
|
||||
}
|
||||
|
||||
type UserBookStatusUpsert = {
|
||||
userId: number;
|
||||
bookId: number;
|
||||
status: ReadStatus;
|
||||
source: ReadStatusSource;
|
||||
startedAt: Date | null;
|
||||
finishedAt: Date | null;
|
||||
updatedAt: Date;
|
||||
};
|
||||
|
||||
type ReadingProgressUpsert = {
|
||||
bookFileId: number;
|
||||
userId: number;
|
||||
percentage: number;
|
||||
cfi: string | null;
|
||||
pageNumber: number | null;
|
||||
positionSeconds: number | null;
|
||||
updatedAt: Date;
|
||||
};
|
||||
|
||||
type AudiobookProgressUpsert = {
|
||||
userId: number;
|
||||
bookId: number;
|
||||
percentage: number;
|
||||
currentFileId: number;
|
||||
positionSeconds: number;
|
||||
updatedAt: Date;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class UserStateImporter {
|
||||
constructor(
|
||||
@@ -82,15 +111,7 @@ export class UserStateImporter {
|
||||
return;
|
||||
}
|
||||
|
||||
const batch: Array<{
|
||||
userId: number;
|
||||
bookId: number;
|
||||
status: ReadStatus;
|
||||
source: ReadStatusSource;
|
||||
startedAt: Date | null;
|
||||
finishedAt: Date | null;
|
||||
updatedAt: Date;
|
||||
}> = [];
|
||||
const batch: UserBookStatusUpsert[] = [];
|
||||
|
||||
for (const row of planned.execution.sourceData.userBookStatuses) {
|
||||
await ensureRunning();
|
||||
@@ -113,10 +134,11 @@ export class UserStateImporter {
|
||||
});
|
||||
counters.imported += 1;
|
||||
}
|
||||
const dedupedBatch = dedupeByKey(batch, (item) => `${item.userId}:${item.bookId}`, preferLatestByUpdatedAt);
|
||||
|
||||
await this.importRepo.withTransaction(async (importRepo) => {
|
||||
await importRepo.clearUserBookStatuses([...userMap.values()], [...bookMap.values()]);
|
||||
await importRepo.batchUpsertUserBookStatuses(batch);
|
||||
await importRepo.batchUpsertUserBookStatuses(dedupedBatch);
|
||||
});
|
||||
await this.repo.setRunMetric(runId, 'user_state', 'user_book_status', counters);
|
||||
}
|
||||
@@ -136,15 +158,7 @@ export class UserStateImporter {
|
||||
return;
|
||||
}
|
||||
|
||||
const batch: Array<{
|
||||
bookFileId: number;
|
||||
userId: number;
|
||||
percentage: number;
|
||||
cfi: string | null;
|
||||
pageNumber: number | null;
|
||||
positionSeconds: number | null;
|
||||
updatedAt: Date;
|
||||
}> = [];
|
||||
const batch: ReadingProgressUpsert[] = [];
|
||||
|
||||
for (const row of planned.execution.sourceData.userFileProgress) {
|
||||
await ensureRunning();
|
||||
@@ -185,10 +199,11 @@ export class UserStateImporter {
|
||||
});
|
||||
counters.imported += 1;
|
||||
}
|
||||
const dedupedBatch = dedupeByKey(batch, (item) => `${item.bookFileId}:${item.userId}`, preferReadingProgressRow);
|
||||
|
||||
await this.importRepo.withTransaction(async (importRepo) => {
|
||||
await importRepo.clearReadingProgress([...userMap.values()], [...primaryFilesByBookId.values(), ...sourceFileToTargetFile.values()]);
|
||||
await importRepo.batchUpsertReadingProgress(batch);
|
||||
await importRepo.batchUpsertReadingProgress(dedupedBatch);
|
||||
});
|
||||
await this.repo.setRunMetric(runId, 'user_state', 'reading_progress', counters);
|
||||
}
|
||||
@@ -213,14 +228,7 @@ export class UserStateImporter {
|
||||
return targetBookId ? audiobookPrimaryFilesByBookId.has(targetBookId) : false;
|
||||
});
|
||||
|
||||
const batch: Array<{
|
||||
userId: number;
|
||||
bookId: number;
|
||||
percentage: number;
|
||||
currentFileId: number;
|
||||
positionSeconds: number;
|
||||
updatedAt: Date;
|
||||
}> = [];
|
||||
const batch: AudiobookProgressUpsert[] = [];
|
||||
|
||||
for (const row of sourceAudioRows) {
|
||||
await ensureRunning();
|
||||
@@ -258,9 +266,11 @@ export class UserStateImporter {
|
||||
counters.imported += 1;
|
||||
}
|
||||
|
||||
const dedupedBatch = dedupeByKey(batch, (item) => `${item.userId}:${item.bookId}`, preferLatestByUpdatedAt);
|
||||
|
||||
await this.importRepo.withTransaction(async (importRepo) => {
|
||||
await importRepo.clearAudiobookProgress([...userMap.values()], [...audiobookPrimaryFilesByBookId.keys()]);
|
||||
await importRepo.batchUpsertAudiobookProgress(batch);
|
||||
await importRepo.batchUpsertAudiobookProgress(dedupedBatch);
|
||||
});
|
||||
await this.repo.setRunMetric(runId, 'user_state', 'audiobook_progress', counters);
|
||||
}
|
||||
@@ -520,6 +530,50 @@ function nextImportedCollectionName(baseName: string, usedNames: Set<string>): s
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function dedupeByKey<T>(items: T[], toKey: (item: T) => string, pickPreferred: (current: T, candidate: T) => T): T[] {
|
||||
const byKey = new Map<string, T>();
|
||||
for (const item of items) {
|
||||
const key = toKey(item);
|
||||
const existing = byKey.get(key);
|
||||
if (!existing) {
|
||||
byKey.set(key, item);
|
||||
continue;
|
||||
}
|
||||
byKey.set(key, pickPreferred(existing, item));
|
||||
}
|
||||
return [...byKey.values()];
|
||||
}
|
||||
|
||||
function preferLatestByUpdatedAt<T extends { updatedAt: Date }>(current: T, candidate: T): T {
|
||||
const currentTs = current.updatedAt.getTime();
|
||||
const candidateTs = candidate.updatedAt.getTime();
|
||||
if (candidateTs > currentTs) return candidate;
|
||||
if (candidateTs < currentTs) return current;
|
||||
return candidate;
|
||||
}
|
||||
|
||||
function preferReadingProgressRow(current: ReadingProgressUpsert, candidate: ReadingProgressUpsert): ReadingProgressUpsert {
|
||||
const currentTs = current.updatedAt.getTime();
|
||||
const candidateTs = candidate.updatedAt.getTime();
|
||||
if (candidateTs > currentTs) return candidate;
|
||||
if (candidateTs < currentTs) return current;
|
||||
|
||||
const currentScore = readingProgressSignalScore(current);
|
||||
const candidateScore = readingProgressSignalScore(candidate);
|
||||
if (candidateScore > currentScore) return candidate;
|
||||
if (candidateScore < currentScore) return current;
|
||||
return candidate;
|
||||
}
|
||||
|
||||
function readingProgressSignalScore(row: ReadingProgressUpsert): number {
|
||||
let score = 0;
|
||||
if ((row.cfi?.trim().length ?? 0) > 0) score += 8;
|
||||
if (typeof row.pageNumber === 'number' && Number.isFinite(row.pageNumber) && row.pageNumber > 0) score += 2;
|
||||
if (typeof row.positionSeconds === 'number' && Number.isFinite(row.positionSeconds) && row.positionSeconds > 0) score += 2;
|
||||
if (row.percentage > 0) score += 1;
|
||||
return score;
|
||||
}
|
||||
|
||||
function resolveBookmarkPositionSeconds(row: SourceBookmark, sourceBook: SourceBook | undefined): number | null {
|
||||
if (row.positionSeconds == null || !Number.isFinite(row.positionSeconds)) return null;
|
||||
const positionSeconds = Math.max(0, row.positionSeconds);
|
||||
|
||||
@@ -619,6 +619,34 @@ export async function seedCanonicalScenario(ctx: MigrationBookloreE2EContext): P
|
||||
};
|
||||
}
|
||||
|
||||
export async function seedCanonicalScenarioWithDuplicateUserStateRows(ctx: MigrationBookloreE2EContext): Promise<CanonicalMigrationScenario> {
|
||||
const scenario = await seedCanonicalScenario(ctx);
|
||||
|
||||
await withBookloreConnection(ctx, async (conn) => {
|
||||
await insertRows(
|
||||
conn,
|
||||
'user_book_progress',
|
||||
['user_id', 'book_id', 'status', 'percentage', 'started_at', 'finished_at', 'updated_at'],
|
||||
[
|
||||
[1, 101, 'reading', 64, '2024-01-02 12:00:00', null, '2024-01-11 14:00:00'],
|
||||
[2, 103, 'completed', 100, '2024-02-01 12:00:00', '2024-02-05 12:00:00', '2024-02-05 12:00:00'],
|
||||
],
|
||||
);
|
||||
|
||||
await insertRows(
|
||||
conn,
|
||||
'user_book_file_progress',
|
||||
['user_id', 'book_id', 'book_file_id', 'percentage', 'cfi', 'position_href', 'page_number', 'position_seconds', 'updated_at'],
|
||||
[
|
||||
[1, 101, 1001, 44.4, 'epubcfi(/6/18!/4/2:10)', null, null, null, '2024-01-11 14:00:00'],
|
||||
[2, 103, 1004, 77, null, null, null, 45, '2024-02-05 12:30:00'],
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
return scenario;
|
||||
}
|
||||
|
||||
export async function seedCompatibilityScenario(ctx: MigrationBookloreE2EContext): Promise<CompatibilityScenario> {
|
||||
const targetUser = await createUser(ctx, {
|
||||
username: 'migration-booklore-variant-source',
|
||||
|
||||
@@ -18,6 +18,7 @@ import {
|
||||
} from './e2e/migration-booklore/migration-booklore-harness';
|
||||
import {
|
||||
seedCanonicalScenario,
|
||||
seedCanonicalScenarioWithDuplicateUserStateRows,
|
||||
seedCompatibilityScenario,
|
||||
seedMissingRequiredTablesScenario,
|
||||
seedWarningsOnlyScenario,
|
||||
@@ -60,7 +61,7 @@ describe('Migration Booklore API to DB (e2e)', { timeout: 240_000 }, () => {
|
||||
token: ctx.adminToken,
|
||||
});
|
||||
expect(supportedTypes.statusCode).toBe(200);
|
||||
expect(supportedTypes.body).toEqual(['booklore']);
|
||||
expect(supportedTypes.body).toEqual(expect.arrayContaining(['booklore', 'grimmory']));
|
||||
|
||||
const testedSource = await apiJson<{
|
||||
ok: boolean;
|
||||
@@ -698,6 +699,126 @@ describe('Migration Booklore API to DB (e2e)', { timeout: 240_000 }, () => {
|
||||
await assertNoIntegrityViolations(ctx.db);
|
||||
});
|
||||
|
||||
it('handles duplicate source user-state rows without failing migration and keeps latest values', async () => {
|
||||
const scenario = await seedCanonicalScenarioWithDuplicateUserStateRows(ctx);
|
||||
|
||||
const createdSource = await apiJson<{ id: number }>(ctx, {
|
||||
method: 'POST',
|
||||
url: '/api/v1/migration/sources',
|
||||
token: ctx.adminToken,
|
||||
payload: {
|
||||
type: 'booklore',
|
||||
name: 'Canonical With Duplicate User State',
|
||||
connectionConfig: scenario.connectionConfig,
|
||||
},
|
||||
});
|
||||
expect(createdSource.statusCode).toBe(201);
|
||||
|
||||
const validatedSource = await apiJson<{ ok: boolean }>(ctx, {
|
||||
method: 'POST',
|
||||
url: `/api/v1/migration/sources/${createdSource.body.id}/validate`,
|
||||
token: ctx.adminToken,
|
||||
});
|
||||
expect(validatedSource.statusCode).toBe(200);
|
||||
expect(validatedSource.body.ok).toBe(true);
|
||||
|
||||
const profile = await apiJson<{ id: number }>(ctx, {
|
||||
method: 'POST',
|
||||
url: '/api/v1/migration/profiles',
|
||||
token: ctx.adminToken,
|
||||
payload: {
|
||||
sourceId: createdSource.body.id,
|
||||
name: 'Duplicate User State Profile',
|
||||
userMappings: [
|
||||
{ sourceUserId: '1', targetUserId: scenario.targetUsers.alice.id },
|
||||
{ sourceUserId: '2', targetUserId: scenario.targetUsers.bob.id },
|
||||
],
|
||||
pathMappings: scenario.pathMappings,
|
||||
},
|
||||
});
|
||||
expect(profile.statusCode).toBe(201);
|
||||
|
||||
const validatedMappings = await apiJson<{ persistedProfileId: number | null }>(ctx, {
|
||||
method: 'POST',
|
||||
url: `/api/v1/migration/sources/${createdSource.body.id}/path-mappings/validate`,
|
||||
token: ctx.adminToken,
|
||||
payload: {
|
||||
pathMappings: scenario.pathMappings,
|
||||
},
|
||||
});
|
||||
expect(validatedMappings.statusCode).toBe(201);
|
||||
expect(validatedMappings.body.persistedProfileId).toBe(profile.body.id);
|
||||
|
||||
const dryRun = await apiJson<{
|
||||
id: number;
|
||||
plan: { duplicateBookMatches: Array<{ targetBookId: number; sourceBookIds: string[] }> };
|
||||
}>(ctx, {
|
||||
method: 'POST',
|
||||
url: '/api/v1/migration/plans/dry-run',
|
||||
token: ctx.adminToken,
|
||||
payload: { profileId: profile.body.id },
|
||||
});
|
||||
expect(dryRun.statusCode).toBe(201);
|
||||
expect(dryRun.body.plan.duplicateBookMatches).toHaveLength(1);
|
||||
|
||||
const resolvedPlan = await apiJson<{
|
||||
id: number;
|
||||
plan: { duplicateBookMatches: unknown[] };
|
||||
summary: { status: string };
|
||||
}>(ctx, {
|
||||
method: 'POST',
|
||||
url: `/api/v1/migration/plans/${dryRun.body.id}/resolve-duplicates`,
|
||||
token: ctx.adminToken,
|
||||
payload: {
|
||||
resolutions: [{ targetBookId: scenario.books.duplicate.bookId, selectedSourceBookId: scenario.sourceBookIds.duplicatePreferred }],
|
||||
},
|
||||
});
|
||||
expect(resolvedPlan.statusCode).toBe(200);
|
||||
expect(resolvedPlan.body.plan.duplicateBookMatches).toEqual([]);
|
||||
expect(resolvedPlan.body.summary.status).toBe('ready_for_live_run');
|
||||
|
||||
const startedRun = await apiJson<{ id: number; state: string }>(ctx, {
|
||||
method: 'POST',
|
||||
url: '/api/v1/migration/runs/live',
|
||||
token: ctx.adminToken,
|
||||
payload: { planArtifactId: resolvedPlan.body.id },
|
||||
});
|
||||
expect(startedRun.statusCode).toBe(201);
|
||||
expect(startedRun.body.state).toBe('running');
|
||||
|
||||
const finished = await waitForMigrationToFinish(ctx, startedRun.body.id);
|
||||
expect(finished.progress.run.state).toBe('completed');
|
||||
|
||||
const statusRows = await ctx.db.select().from(schema.userBookStatus);
|
||||
expect(statusRows.find((row) => row.userId === scenario.targetUsers.alice.id && row.bookId === scenario.books.isbn.bookId)).toMatchObject({
|
||||
status: 'reading',
|
||||
});
|
||||
expect(statusRows.find((row) => row.userId === scenario.targetUsers.bob.id && row.bookId === scenario.books.audio.bookId)).toMatchObject({
|
||||
status: 'read',
|
||||
});
|
||||
|
||||
const progressRows = await ctx.db.select().from(schema.readingProgress);
|
||||
expect(
|
||||
progressRows.find((row) => row.userId === scenario.targetUsers.alice.id && row.bookFileId === scenario.books.isbn.primaryFileId),
|
||||
).toMatchObject({
|
||||
percentage: 44.4,
|
||||
cfi: 'epubcfi(/6/18!/4/2:10)',
|
||||
});
|
||||
expect(
|
||||
progressRows.find((row) => row.userId === scenario.targetUsers.bob.id && row.bookFileId === scenario.books.audio.fileIds[1]),
|
||||
).toMatchObject({
|
||||
percentage: 77,
|
||||
positionSeconds: 45,
|
||||
});
|
||||
|
||||
const audioProgressRows = await ctx.db.select().from(schema.audiobookProgress);
|
||||
expect(audioProgressRows.find((row) => row.userId === scenario.targetUsers.bob.id && row.bookId === scenario.books.audio.bookId)).toMatchObject({
|
||||
percentage: 77,
|
||||
currentFileId: scenario.books.audio.fileIds[1],
|
||||
positionSeconds: 45,
|
||||
});
|
||||
});
|
||||
|
||||
it('supports alternate Booklore schema variants and join-through file progress', async () => {
|
||||
const scenario = await seedCompatibilityScenario(ctx);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user