fix(server): import Audiobookshelf finishes dated before their start

Audiobookshelf reports a progress row's creation time as its start, while
clients set the finish date freely, so a backfilled finish can predate it.
One such row broke user_book_status's date check and the whole status batch
with it. Keep the finish date and drop the start, and store the database's
reason on a failed run instead of the failed SQL.

Closes #1678
This commit is contained in:
neon
2026-10-12 05:51:19 +00:00
parent 9a7970f0b3
commit de93564e6e
10 changed files with 299 additions and 12 deletions
+38 -1
View File
@@ -1,6 +1,8 @@
import { DrizzleQueryError } from 'drizzle-orm';
import { DatabaseError } from 'pg';
import { describe, expect, it } from 'vitest';
import { isUniqueViolation } from './db-error.utils';
import { isUniqueViolation, queryFailureReason } from './db-error.utils';
describe('isUniqueViolation', () => {
it('matches the driver error raised straight off the pool', () => {
@@ -25,3 +27,38 @@ describe('isUniqueViolation', () => {
expect(isUniqueViolation(undefined)).toBe(false);
});
});
describe('queryFailureReason', () => {
const constraintMessage = 'new row for relation "user_book_status" violates check constraint "user_book_status_finished_after_started_chk"';
function checkViolation(): DatabaseError {
return Object.assign(new DatabaseError(constraintMessage, 0, 'error'), {
code: '23514',
constraint: 'user_book_status_finished_after_started_chk',
detail: 'Failing row contains (1, 2159, read, manual, ...).',
});
}
it("returns the driver's reason instead of the SQL and parameters Drizzle reports", () => {
const error = new DrizzleQueryError('insert into "user_book_status" ("user_id", "status") values ($1, $2)', [1, 'read'], checkViolation());
expect(error.message).toContain('params: 1,read');
expect(queryFailureReason(error)).toBe(constraintMessage);
});
it('leaves the failing row out of the reason', () => {
expect(queryFailureReason(new DrizzleQueryError('select 1', [], checkViolation()))).not.toContain('Failing row');
});
it('returns null when Drizzle has no driver error to report', () => {
expect(queryFailureReason(new DrizzleQueryError('select 1', []))).toBeNull();
expect(queryFailureReason(new DrizzleQueryError('select 1', [], new Error('')))).toBeNull();
});
it('returns null for errors that are not failed queries, whose own message already explains them', () => {
expect(queryFailureReason(new Error('Audiobookshelf request failed', { cause: new Error('socket hang up') }))).toBeNull();
expect(queryFailureReason(checkViolation())).toBeNull();
expect(queryFailureReason('Failed query')).toBeNull();
expect(queryFailureReason(null)).toBeNull();
});
});
+12
View File
@@ -8,6 +8,8 @@
* translated conflict was expected. Both shapes are checked here so neither can regress.
*/
import { DrizzleQueryError } from 'drizzle-orm';
/** Postgres `unique_violation`. */
const UNIQUE_VIOLATION = '23505';
@@ -18,3 +20,13 @@ export function isUniqueViolation(error: unknown): boolean {
return (error as { cause?: { code?: unknown } }).cause?.code === UNIQUE_VIOLATION;
}
/**
* The driver's reason for a failed query, such as the constraint it violated. A
* `DrizzleQueryError` message is only the SQL and its bound parameters, which reads as if a row
* value were the problem; null for anything else, whose own message already says what happened.
*/
export function queryFailureReason(error: unknown): string | null {
if (!(error instanceof DrizzleQueryError)) return null;
return error.cause?.message || null;
}
@@ -11,6 +11,7 @@ import {
normalizeReadStatus,
pruneUndefined,
toDate,
toReadingDates,
truncateNullableText,
truncateText,
uniqueNumbers,
@@ -274,6 +275,59 @@ describe('migration executor utils', () => {
expect(clampNonNegative(9)).toBe(9);
});
describe('toReadingDates', () => {
const date = (value: string) => new Date(value);
it('keeps the finish date and drops the start when the finish comes first', () => {
expect(toReadingDates('2026-10-12T05:23:33.044Z', '2025-09-07T05:23:32.000Z')).toEqual({
startedAt: null,
finishedAt: date('2025-09-07T05:23:32.000Z'),
});
});
it('drops the start when the finish is a single millisecond earlier', () => {
expect(toReadingDates('2025-02-20T08:45:00.001Z', '2025-02-20T08:45:00.000Z')).toEqual({
startedAt: null,
finishedAt: date('2025-02-20T08:45:00.000Z'),
});
});
it('keeps both dates when the finish matches or follows the start', () => {
// Marking a book finished in Audiobookshelf without listening stamps both dates at once.
expect(toReadingDates('2026-10-12T05:20:33.137Z', '2026-10-12T05:20:33.137Z')).toEqual({
startedAt: date('2026-10-12T05:20:33.137Z'),
finishedAt: date('2026-10-12T05:20:33.137Z'),
});
expect(toReadingDates('2025-01-10T12:30:00.000Z', '2025-01-15T12:30:00.000Z')).toEqual({
startedAt: date('2025-01-10T12:30:00.000Z'),
finishedAt: date('2025-01-15T12:30:00.000Z'),
});
});
it('orders by instant rather than by the text of the timestamp', () => {
// 10:00+05:00 is 05:00Z, so it starts an hour before a 06:00Z finish even though it sorts later as text.
expect(toReadingDates('2025-03-01T10:00:00+05:00', '2025-03-01T06:00:00Z')).toEqual({
startedAt: date('2025-03-01T05:00:00.000Z'),
finishedAt: date('2025-03-01T06:00:00.000Z'),
});
expect(toReadingDates('2025-03-01T06:00:00Z', '2025-03-01T10:00:00+05:00')).toEqual({
startedAt: null,
finishedAt: date('2025-03-01T05:00:00.000Z'),
});
});
it('passes a lone date through whichever side it is on', () => {
expect(toReadingDates('2025-01-10T12:30:00.000Z', null)).toEqual({ startedAt: date('2025-01-10T12:30:00.000Z'), finishedAt: null });
expect(toReadingDates(null, '2025-01-15T12:30:00.000Z')).toEqual({ startedAt: null, finishedAt: date('2025-01-15T12:30:00.000Z') });
expect(toReadingDates(null, null)).toEqual({ startedAt: null, finishedAt: null });
});
it('treats an unparseable date as missing instead of ordering against it', () => {
expect(toReadingDates('not-a-date', '2025-01-15T12:30:00.000Z')).toEqual({ startedAt: null, finishedAt: date('2025-01-15T12:30:00.000Z') });
expect(toReadingDates('2025-01-10T12:30:00.000Z', 'not-a-date')).toEqual({ startedAt: date('2025-01-10T12:30:00.000Z'), finishedAt: null });
});
});
it('truncates text helpers and normalizes read status from status/percentage', () => {
expect(truncateText('abcdef', 4)).toBe('abcd');
expect(truncateText('abc', 4)).toBe('abc');
@@ -342,6 +342,16 @@ export function toDate(value: string | null): Date | null {
return Number.isNaN(date.getTime()) ? null : date;
}
// user_book_status rejects a finish before its start, and sources produce one: Audiobookshelf
// reports a progress row's creation time as its start while clients set the finish date freely.
// The finish date is the one readers care about, so it is kept and the start dropped.
export function toReadingDates(startedAt: string | null, finishedAt: string | null): { startedAt: Date | null; finishedAt: Date | null } {
const started = toDate(startedAt);
const finished = toDate(finishedAt);
if (started && finished && finished.getTime() < started.getTime()) return { startedAt: null, finishedAt: finished };
return { startedAt: started, finishedAt: finished };
}
export function clampPercent(value: number | null): number {
if (value == null || !Number.isFinite(value)) return 0;
return Math.max(0, Math.min(100, value));
@@ -1,3 +1,6 @@
import { DrizzleQueryError } from 'drizzle-orm';
import { DatabaseError } from 'pg';
import { MigrationExecutorService } from './migration-executor.service';
function makeExecutorService() {
@@ -16,6 +19,7 @@ function makeExecutorService() {
const config = { get: vi.fn().mockReturnValue('/app-data') };
const encryption = { decryptConfig: vi.fn((value: unknown) => value) };
const progressGateway = { emitProgress: vi.fn() };
const notificationService = { notify: vi.fn().mockResolvedValue(undefined) };
const service = new MigrationExecutorService(
repo as never,
@@ -25,10 +29,10 @@ function makeExecutorService() {
config as never,
encryption as never,
progressGateway as never,
{ notify: vi.fn().mockResolvedValue(undefined) } as never,
notificationService as never,
);
return { service, repo, sharedOverlays, covers, userState, progressGateway };
return { service, repo, sharedOverlays, covers, userState, progressGateway, notificationService };
}
describe('MigrationExecutorService core orchestration', () => {
@@ -265,6 +269,61 @@ describe('MigrationExecutorService core orchestration', () => {
expect(emittedStates).toEqual(['running', 'running', 'completed']);
});
describe('when a stage fails', () => {
const constraintMessage = 'new row for relation "user_book_status" violates check constraint "user_book_status_finished_after_started_chk"';
function setUpFailingRun(error: unknown) {
const harness = makeExecutorService();
harness.repo.findRunById.mockResolvedValue({
id: 13,
state: 'running',
sourceId: 1,
planArtifactId: 45,
currentStage: 'user_state',
triggeredByUserId: 3,
});
harness.repo.findPlanArtifactById.mockResolvedValue({
id: 45,
plan: { matchedBooks: [], unresolvedBooks: [] },
sourceData: { books: [] },
});
harness.repo.findSourceById.mockResolvedValue({
id: 1,
type: 'audiobookshelf',
connectionConfig: { mode: 'api', baseUrl: 'http://abs.local', apiToken: 'token', allowPrivateNetwork: true },
});
harness.userState.import.mockRejectedValue(error);
return harness;
}
it("records the database's reason rather than the failed SQL and its row values", async () => {
// Issue 1678: the stored message was Drizzle's query text, whose params listed `read`, so the
// report blamed the status value instead of the violated constraint.
const queryError = new DrizzleQueryError(
'insert into "user_book_status" ("user_id", "book_id", "status") values ($1, $2, $3)',
[1, 2159, 'read'],
Object.assign(new DatabaseError(constraintMessage, 0, 'error'), { code: '23514' }),
);
const { service, repo, notificationService } = setUpFailingRun(queryError);
await (service as never).execute(13);
expect(repo.updateRunState).toHaveBeenCalledWith(13, 'failed', expect.objectContaining({ errorMessage: constraintMessage }));
expect(notificationService.notify).toHaveBeenCalledWith(expect.objectContaining({ message: constraintMessage, meta: { runId: 13 } }));
});
it('records a non-query failure by its own message', async () => {
const { service, repo, notificationService } = setUpFailingRun(
new Error('Audiobookshelf request failed', { cause: new Error('socket hang up') }),
);
await (service as never).execute(13);
expect(repo.updateRunState).toHaveBeenCalledWith(13, 'failed', expect.objectContaining({ errorMessage: 'Audiobookshelf request failed' }));
expect(notificationService.notify).toHaveBeenCalledWith(expect.objectContaining({ message: 'Audiobookshelf request failed' }));
});
});
it('execute exits cleanly without failing the run when interrupted', async () => {
const { service, repo, sharedOverlays, progressGateway } = makeExecutorService();
const running = { id: 12, state: 'running', sourceId: 1, planArtifactId: 77, currentStage: 'init' };
@@ -3,6 +3,7 @@ import { ConfigService } from '@nestjs/config';
import type { MigrationProgressEvent } from '@bookorbit/types';
import { NotificationType } from '@bookorbit/types';
import { queryFailureReason } from '../../../common/utils/db-error.utils';
import { NotificationService } from '../../notification/notification.service';
import { asRecord } from '../core/coerce';
import { parseConnectionConfig } from '../core/connection-config';
@@ -141,7 +142,8 @@ export class MigrationExecutorService {
this.logger.warn(
`[migration.execute] [fail] runId=${runId} durationMs=${durationMs} errorClass=${errorClass} error="${message}"${cause} - migration execution failed`,
);
const failedStage = await this.failRun(runId, message);
const reason = queryFailureReason(error) ?? message;
const failedStage = await this.failRun(runId, reason);
await this.emitRunProgress(runId, 'failed', failedStage);
if (run.triggeredByUserId) {
@@ -149,7 +151,7 @@ export class MigrationExecutorService {
.notify({
type: NotificationType.MigrationFailed,
title: 'Migration failed',
message: message.slice(0, 200),
message: reason.slice(0, 200),
scope: { kind: 'user', userId: run.triggeredByUserId },
meta: { runId },
})
@@ -1082,3 +1082,106 @@ describe('UserStateImporter cache invalidation', () => {
expect(notify).not.toHaveBeenCalled();
});
});
describe('UserStateImporter reading dates', () => {
type WrittenStatus = { userId: number; bookId: number; status: string; startedAt: Date | null; finishedAt: Date | null };
async function importStatuses(rows: Array<Record<string, unknown>>) {
const { importer, repo, importRepo } = makeImporter();
const bookMap = new Map(rows.map((row, index) => [row.sourceBookId as string, 100 + index]));
await (importer as any).importUserBookStatuses(
40,
{ execution: { sourceData: { userBookStatuses: rows } } },
new Map([['abs-user', 7]]),
bookMap,
async () => {},
);
expect(importRepo.batchMergeUserBookStatuses).toHaveBeenCalledTimes(1);
const written = importRepo.batchMergeUserBookStatuses.mock.calls[0][0] as WrittenStatus[];
return { repo, written, bookMap };
}
it('imports every status when one book was finished before Audiobookshelf says it was started', async () => {
// The shape of issue 1678: a finish date sent by a client lands before the progress row's
// creation time, which Audiobookshelf reports as the start. One such row used to fail the
// whole batch, so none of the reader's statuses arrived.
const { repo, written, bookMap } = await importStatuses([
{
sourceUserId: 'abs-user',
sourceBookId: 'backfilled',
status: 'read',
percentage: 100,
startedAt: '2026-10-12T05:23:33.044Z',
finishedAt: '2025-09-07T05:23:32.000Z',
updatedAt: '2026-10-12T05:23:33.044Z',
},
{
sourceUserId: 'abs-user',
sourceBookId: 'marked-finished',
status: 'read',
percentage: 100,
startedAt: '2026-10-12T05:20:33.137Z',
finishedAt: '2026-10-12T05:20:33.137Z',
updatedAt: '2026-10-12T05:20:33.138Z',
},
{
sourceUserId: 'abs-user',
sourceBookId: 'listening',
status: 'reading',
percentage: 33.3,
startedAt: '2026-10-12T05:20:33.172Z',
finishedAt: null,
updatedAt: '2026-10-12T05:20:33.172Z',
},
]);
expect(written).toEqual([
expect.objectContaining({
bookId: bookMap.get('backfilled'),
status: 'read',
startedAt: null,
finishedAt: new Date('2025-09-07T05:23:32.000Z'),
}),
expect.objectContaining({
bookId: bookMap.get('marked-finished'),
status: 'read',
startedAt: new Date('2026-10-12T05:20:33.137Z'),
finishedAt: new Date('2026-10-12T05:20:33.137Z'),
}),
expect.objectContaining({
bookId: bookMap.get('listening'),
status: 'reading',
startedAt: new Date('2026-10-12T05:20:33.172Z'),
finishedAt: null,
}),
]);
expect(repo.setRunMetric).toHaveBeenCalledWith(40, 'user_state', 'user_book_status', expect.objectContaining({ processed: 3, imported: 3 }));
});
it('never writes a row the finished-after-started check would reject, and never loses a finish date', async () => {
const candidates = [null, '2024-05-01T00:00:00.000Z', '2024-05-01T00:00:00.001Z', '2025-06-01T12:00:00+02:00', 'not-a-date'];
const rows = candidates.flatMap((startedAt, startIndex) =>
candidates.map((finishedAt, finishIndex) => ({
sourceUserId: 'abs-user',
sourceBookId: `book-${startIndex}-${finishIndex}`,
status: 'read',
percentage: 100,
startedAt,
finishedAt,
updatedAt: '2026-01-01T00:00:00.000Z',
})),
);
const { written } = await importStatuses(rows);
expect(written).toHaveLength(rows.length);
for (const [index, row] of rows.entries()) {
const out = written[index];
const finish = row.finishedAt ? new Date(row.finishedAt) : null;
const expectedFinish = finish && Number.isFinite(finish.getTime()) ? finish : null;
// Mirrors user_book_status_finished_after_started_chk.
if (out.startedAt && out.finishedAt) expect(out.finishedAt.getTime()).toBeGreaterThanOrEqual(out.startedAt.getTime());
expect(out.finishedAt).toEqual(expectedFinish);
}
});
});
@@ -27,6 +27,7 @@ import {
type ProgressMediaKind,
type TargetBookFile,
toDate,
toReadingDates,
truncateNullableText,
truncateText,
uniqueNumbers,
@@ -200,8 +201,7 @@ export class UserStateImporter {
bookId: targetBookId,
status: normalizeReadStatus(row.status, row.percentage),
source: 'manual',
startedAt: toDate(row.startedAt),
finishedAt: toDate(row.finishedAt),
...toReadingDates(row.startedAt, row.finishedAt),
sourceUpdatedAt: toDate(row.updatedAt),
});
counters.imported += 1;
@@ -20,6 +20,7 @@ import {
const execFile = promisify(execFileCallback);
const requireInstaller = createRequire(__filename);
const MAYA_CFI = 'epubcfi(/6/2[chapter]!/4/2/2:5)';
const THEO_MAP_STARTED_AT = '2025-01-10T12:30:00.000Z';
const THEO_MAP_FINISHED_AT = '2025-01-15T12:30:00.000Z';
const THEO_BRASS_FINISHED_AT = '2025-02-20T08:45:00.000Z';
@@ -54,6 +55,7 @@ export interface AudiobookshelfMigrationScenario {
targetBooks: Record<'glass' | 'map' | 'brass' | 'northbound', SeededTargetBook>;
expected: {
mayaCfi: string;
theoMapStartedAt: string;
theoMapFinishedAt: string;
theoBrassFinishedAt: string;
};
@@ -195,7 +197,7 @@ export async function seedAudiobookshelfMigrationScenario(ctx: MigrationAudioboo
ebookProgress: 1,
isFinished: true,
finishedAt: THEO_MAP_FINISHED_AT,
createdAt: '2025-01-10T12:30:00.000Z',
createdAt: THEO_MAP_STARTED_AT,
}),
setProgress(ctx.audiobookshelf, theo.token, sourceItems.brass.id, {
duration: 30,
@@ -203,7 +205,9 @@ export async function seedAudiobookshelfMigrationScenario(ctx: MigrationAudioboo
progress: 1,
isFinished: true,
finishedAt: THEO_BRASS_FINISHED_AT,
createdAt: '2025-02-10T08:45:00.000Z',
// ABS reports a progress row's creation time as its start, while clients set the finish
// date freely, so a backfilled finish lands before the start (issue 1678).
createdAt: '2025-03-05T08:45:00.000Z',
}),
setProgress(ctx.audiobookshelf, lina.token, sourceItems.glass.id, {
duration: 150,
@@ -239,7 +243,12 @@ export async function seedAudiobookshelfMigrationScenario(ctx: MigrationAudioboo
sourceItems,
targetUsers,
targetBooks,
expected: { mayaCfi: MAYA_CFI, theoMapFinishedAt: THEO_MAP_FINISHED_AT, theoBrassFinishedAt: THEO_BRASS_FINISHED_AT },
expected: {
mayaCfi: MAYA_CFI,
theoMapStartedAt: THEO_MAP_STARTED_AT,
theoMapFinishedAt: THEO_MAP_FINISHED_AT,
theoBrassFinishedAt: THEO_BRASS_FINISHED_AT,
},
};
}
@@ -169,7 +169,7 @@ describe('Migration Audiobookshelf live and backup (e2e)', { timeout: 300_000 },
expect(started.statusCode).toBe(201);
expect(started.body.state).toBe('running');
const finished = await waitForMigrationToFinish(ctx, started.body.id);
expect(finished.progress.run.state).toBe('completed');
expect(finished.progress.run).toMatchObject({ state: 'completed', errorMessage: null });
expect(finished.report.run.state).toBe('completed');
expect(finished.report.totals.failed).toBe(0);
expect(finished.report.totals.unresolved).toBeGreaterThanOrEqual(1);
@@ -367,8 +367,9 @@ describe('Migration Audiobookshelf live and backup (e2e)', { timeout: 300_000 },
const theoMap = statusRows.find((row) => row.userId === scenario.targetUsers.theo.id && row.bookId === scenario.targetBooks.map.bookId);
const theoBrass = statusRows.find((row) => row.userId === scenario.targetUsers.theo.id && row.bookId === scenario.targetBooks.brass.bookId);
expect(theoMap).toMatchObject({ status: 'read', source: 'manual' });
expect(theoMap?.startedAt?.toISOString()).toBe(scenario.expected.theoMapStartedAt);
expect(theoMap?.finishedAt?.toISOString()).toBe(scenario.expected.theoMapFinishedAt);
expect(theoBrass).toMatchObject({ status: 'read', source: 'manual' });
expect(theoBrass).toMatchObject({ status: 'read', source: 'manual', startedAt: null });
expect(theoBrass?.finishedAt?.toISOString()).toBe(scenario.expected.theoBrassFinishedAt);
const bookmarkRows = await ctx.db.select().from(schema.bookmarks).where(eq(schema.bookmarks.userId, scenario.targetUsers.maya.id));