Migrations/bookorbit
1
0
Fork 0
mirror of https://github.com/bookorbit/bookorbit.git synced 2026-11-03 00:11:37 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(stats): invalidate reading caches after committed changes

Browse Source 
Refresh dashboard and statistics caches immediately after reading history,
projection, reset, import, and shared deletion changes. Coalesce bulk
notifications while keeping committed changes visible to concurrent reads,
and isolate listener failures from successful writes.

Skip unchanged Hardcover attempt updates and protect cache generations
against stale in-flight loads. Cover date edits, projection-only resumes,
rollback, bulk failures, user scope, and cache races with regression tests.

Ref #1695
This commit is contained in:
neon
2026-10-07 01:28:10 +00:00
parent 58756565fc
commit e6ea5bd77e
37 changed files with 1628 additions and 311 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/common/cache/stats-cache.test.ts
Vendored
+50
View File
@@ -438,3 +438,53 @@ describe('StatsCache', () => {
});
});
});
describe('StatsCache invalidation races', () => {
it('does not repopulate globally cleared caches from an older in-flight load', async () => {
const cache = new StatsCache({ ttlMs: 1000, maxEntries: 10 });
let finish!: (value: string) => void;
const old = cache.get(
'1',
'goal',
() =>
new Promise<string>((resolve) => {
finish = resolve;
}),
);
cache.clear();
finish('old');
expect(await old).toBe('old');
const fresh = vi.fn().mockResolvedValue('new');
expect(await cache.get('1', 'goal', fresh)).toBe('new');
expect(fresh).toHaveBeenCalledOnce();
});
it('keeps a newer in-flight load deduplicated when the invalidated load finishes', async () => {
const cache = new StatsCache({ ttlMs: 1000, maxEntries: 10 });
let finishOld!: (value: string) => void;
let finishNew!: (value: string) => void;
const old = cache.get(
'1',
'goal',
() =>
new Promise<string>((resolve) => {
finishOld = resolve;
}),
);
cache.clearForScope('1');
const fresh = vi.fn(
() =>
new Promise<string>((resolve) => {
finishNew = resolve;
}),
);
const next = cache.get('1', 'goal', fresh);
finishOld('old');
await old;
const shared = cache.get('1', 'goal', fresh);
expect(fresh).toHaveBeenCalledOnce();
finishNew('new');
expect(await next).toBe('new');
expect(await shared).toBe('new');
});
});
server/src/common/cache/stats-cache.ts
Vendored
+5 -2
View File
@@ -10,6 +10,7 @@ interface CacheEntry {
}
export class StatsCache {
private generation = 0;
private readonly entries = new Map<string, CacheEntry>();
private readonly inFlight = new Map<string, Promise<unknown>>();
private readonly scopeIndex = new Map<string, Set<string>>();
@@ -36,11 +37,12 @@ export class StatsCache {
return inFlight as Promise<T>;
}
const generation = this.generation;
const gen = this.scopeGen.get(scope) ?? 0;
const pending = load()
.then((value) => {
if ((this.scopeGen.get(scope) ?? 0) === gen) {
if (this.generation === generation && (this.scopeGen.get(scope) ?? 0) === gen) {
const ts = Date.now();
this.entries.set(fullKey, { value, expiresAt: ts + this.options.ttlMs, lastAccessedAt: ts });
if (!this.scopeIndex.has(scope)) {
@@ -52,7 +54,7 @@ export class StatsCache {
return value;
})
.finally(() => {
this.inFlight.delete(fullKey);
if (this.inFlight.get(fullKey) === pending) this.inFlight.delete(fullKey);
});
this.inFlight.set(fullKey, pending as Promise<unknown>);
@@ -89,6 +91,7 @@ export class StatsCache {
}
clear(): void {
this.generation += 1;
this.entries.clear();
this.inFlight.clear();
this.scopeIndex.clear();
server/src/modules/book/book.service.test.ts
+19 -2
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import { BadRequestException, ConflictException, ForbiddenException, InternalServerErrorException, Logger, NotFoundException } from '@nestjs/common';
import type { MockedFunction } from 'vitest';
import { rm, stat, rename } from 'fs/promises';
@@ -277,6 +278,7 @@ function makeService(overrides: { bookMetadataLockService?: unknown; appDataPath
bookRepo.withTransaction.mockImplementation(async (callback: (tx: unknown) => Promise<unknown>) => callback({}));
const selfWriteRegistry = new SelfWriteRegistry();
const readingEvents = new ReadingAttemptEventsService();
const service = new BookService(
bookRepo as never,
@@ -299,10 +301,17 @@ function makeService(overrides: { bookMetadataLockService?: unknown; appDataPath
fileWriteService as never,
fileRenameService as never,
achievementEvents as never,
undefined,
undefined,
undefined,
undefined,
undefined,
readingEvents,
);
return {
service,
readingEvents,
bookRepo,
libraryService,
queryBuilder,
@@ -2840,7 +2849,7 @@ describe('BookService', () => {
});
it('deletes books through the scan-state-aware repository transaction', async () => {
const { service, bookRepo } = makeService();
const { service, bookRepo, readingEvents } = makeService();
const user = makeUser();
bookRepo.findLibraryIdsByBookIds.mockResolvedValue([{ id: 3, libraryId: 7 }]);
@@ -2848,13 +2857,19 @@ describe('BookService', () => {
bookRepo.findAllFilesByBookIds.mockResolvedValue([]);
bookRepo.deleteByIdsAndInvalidateScanState.mockResolvedValue(undefined);
const notify = vi.spyOn(readingEvents, 'notifyChanged');
bookRepo.deleteByIdsAndInvalidateScanState.mockImplementation(() => {
expect(notify).not.toHaveBeenCalled();
return Promise.resolve();
});
await service.deleteBooks([3], user);
expect(notify).toHaveBeenCalledExactlyOnceWith(null);
expect(bookRepo.deleteByIdsAndInvalidateScanState).toHaveBeenCalledWith([3]);
});
it('does not commit book deletion when scan state invalidation fails', async () => {
const { service, bookRepo } = makeService();
const { service, bookRepo, readingEvents } = makeService();
const user = makeUser();
const warnSpy = vi.spyOn((service as unknown as { logger: { warn: (message: string) => void } }).logger, 'warn').mockImplementation();
@@ -2863,7 +2878,9 @@ describe('BookService', () => {
bookRepo.findAllFilesByBookIds.mockResolvedValue([]);
bookRepo.deleteByIdsAndInvalidateScanState.mockRejectedValue(new Error('db down'));
const notify = vi.spyOn(readingEvents, 'notifyChanged');
await expect(service.deleteBooks([3], user)).rejects.toThrow('db down');
expect(notify).not.toHaveBeenCalled();
expect(warnSpy).toHaveBeenCalled();
expect(mockRm).not.toHaveBeenCalled();
});
server/src/modules/book/book.service.ts
+3
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import {
BadRequestException,
ForbiddenException,
@@ -306,6 +307,7 @@ export class BookService {
@Optional() private readonly audiobookEbookProgressSync?: AudiobookEbookProgressSyncService,
@Optional() private readonly audiolessEpubService?: AudiolessEpubService,
@Optional() private readonly coverReconciler?: CoverSlotReconciler,
@Optional() private readonly readingAttemptEvents?: ReadingAttemptEventsService,
) {
this.appDataPath = this.config.get<string>('storage.appDataPath')!;
}
@@ -1667,6 +1669,7 @@ export class BookService {
}
await this.bookRepo.deleteByIdsAndInvalidateScanState(deletedBookIds);
this.readingAttemptEvents?.notifyChanged(null);
});
// Cover cleanup cannot resurrect a book and must not turn a committed deletion into a
// failure. Leave covers alone until the database transaction has succeeded.
server/src/modules/dashboard/dashboard-widget.service.test.ts
+24 -2
View File
@@ -4,6 +4,7 @@ import type { RequestUser } from '../../common/types/request-user';
import { pickAnnotationIndex } from './dashboard-widget.calculations';
import { DashboardWidgetService } from './dashboard-widget.service';
import { EMPTY_CONTENT_FILTER_RULES } from '@bookorbit/types';
import { READING_ATTEMPT_CHANGED, ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
function makeUser(overrides: Partial<RequestUser> = {}): RequestUser {
return {
@@ -46,8 +47,9 @@ function makeService() {
findAccessibleLibraryIds: vi.fn(),
};
const service = new DashboardWidgetService(widgetRepo as never, libraryService as never);
return { service, widgetRepo, libraryService };
const events = new ReadingAttemptEventsService();
const service = new DashboardWidgetService(widgetRepo as never, libraryService as never, events);
return { service, widgetRepo, libraryService, events };
}
describe('DashboardWidgetService', () => {
@@ -829,3 +831,23 @@ describe('DashboardWidgetService', () => {
});
});
});
describe('DashboardWidgetService reading-attempt subscription', () => {
it('registers through lifecycle initialization and stops invalidating on destruction', async () => {
const { service, widgetRepo, libraryService, events } = makeService();
libraryService.findAccessibleLibraryIds.mockResolvedValue([1]);
widgetRepo.countCompletedBooks.mockResolvedValueOnce(0).mockResolvedValueOnce(1).mockResolvedValueOnce(2);
service.onModuleInit();
const user = makeUser();
expect((await service.getReadingGoal(user)).completedBooks).toBe(0);
events.notifyChanged(user.id);
expect((await service.getReadingGoal(user)).completedBooks).toBe(1);
const unrelated = vi.fn();
events.on(READING_ATTEMPT_CHANGED, unrelated);
service.onModuleDestroy();
events.notifyChanged(user.id);
expect((await service.getReadingGoal(user)).completedBooks).toBe(1);
expect(widgetRepo.countCompletedBooks).toHaveBeenCalledTimes(2);
expect(unrelated).toHaveBeenCalledExactlyOnceWith({ userId: user.id });
});
});
server/src/modules/dashboard/dashboard-widget.service.ts
+44 -4
View File
@@ -1,4 +1,4 @@
import { Injectable, Logger } from '@nestjs/common';
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import type {
CurrentlyReadingWidgetData,
@@ -27,6 +27,11 @@ import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils';
import { addDateKeyDays } from '../../common/utils/reading-daily-stats.utils';
import { toTimeZoneStartOfDay } from '../../common/utils/timezone.utils';
import { LibraryService } from '../library/library.service';
import {
READING_ATTEMPT_CHANGED,
ReadingAttemptEventsService,
type ReadingAttemptChangedPayload,
} from '../user-book-status/reading-attempt-events.service';
import {
buildDaysSeries,
computeChallengeResult,
@@ -54,16 +59,38 @@ const RHYTHM_WINDOW_DAYS = 14;
const PROJECTION_RECENT_DAYS = 30;
@Injectable()
export class DashboardWidgetService {
export class DashboardWidgetService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(DashboardWidgetService.name);
private readonly liveCache = new StatsCache({ ttlMs: DASHBOARD_LIVE_TTL_MS, maxEntries: DASHBOARD_CACHE_MAX_ENTRIES });
private readonly staleCache = new StatsCache({ ttlMs: DASHBOARD_STALE_TTL_MS, maxEntries: DASHBOARD_CACHE_MAX_ENTRIES });
private readonly liveCache = new StatsCache({
ttlMs: DASHBOARD_LIVE_TTL_MS,
maxEntries: DASHBOARD_CACHE_MAX_ENTRIES,
});
private readonly staleCache = new StatsCache({
ttlMs: DASHBOARD_STALE_TTL_MS,
maxEntries: DASHBOARD_CACHE_MAX_ENTRIES,
});
constructor(
private readonly widgetRepo: DashboardWidgetRepository,
private readonly libraryService: LibraryService,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {}
private readonly onReadingAttemptChanged = ({ userId }: ReadingAttemptChangedPayload): void => {
if (userId === null) {
this.liveCache.clear();
this.staleCache.clear();
} else this.clearCacheForUser(userId);
};
onModuleInit(): void {
this.readingAttemptEvents.on(READING_ATTEMPT_CHANGED, this.onReadingAttemptChanged);
}
onModuleDestroy(): void {
this.readingAttemptEvents.removeListener(READING_ATTEMPT_CHANGED, this.onReadingAttemptChanged);
}
private getContentFilters(user: RequestUser) {
return user.isSuperuser ? undefined : user.contentFilters;
}
@@ -103,6 +130,7 @@ export class DashboardWidgetService {
const accessibleLibraryIds = await this.getLibraryIds(user);
// Keyed by the local year alone: attempt end dates are already local days, so the zone only
// decides which year it is.
this.readingAttemptEvents.flushPendingChanges(user.id);
const completedBooks = await this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), `reading-goal-completed:${year}`, async () => {
const contentFilters = this.getContentFilters(user);
return this.widgetRepo.countCompletedBooks(user.id, accessibleLibraryIds, `${year}-01-01`, `${year + 1}-01-01`, contentFilters);
@@ -113,6 +141,7 @@ export class DashboardWidgetService {
async getCurrentlyReading(user: RequestUser): Promise<CurrentlyReadingWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.liveCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), 'currently-reading', async () => {
const contentFilters = this.getContentFilters(user);
return this.widgetRepo.getCurrentlyReadingBooks(user.id, accessibleLibraryIds, contentFilters);
@@ -122,6 +151,7 @@ export class DashboardWidgetService {
async getReadingStreak(user: RequestUser): Promise<ReadingStreakWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.liveCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('reading-streak', clock), async () => {
const contentFilters = this.getContentFilters(user);
return this.widgetRepo.getReadingStreak(user.id, accessibleLibraryIds, clock.today, contentFilters);
@@ -131,6 +161,7 @@ export class DashboardWidgetService {
async getLibraryOverview(user: RequestUser): Promise<LibraryOverviewWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('library-overview', clock), async () => {
const contentFilters = this.getContentFilters(user);
const yearStart = toTimeZoneStartOfDay(`${clock.year}-01-01`, clock.timeZone);
@@ -141,6 +172,7 @@ export class DashboardWidgetService {
async getHighlightOfTheDay(user: RequestUser): Promise<HighlightOfTheDayWidgetData | null> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.liveCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('highlight-of-the-day', clock), async () => {
const contentFilters = this.getContentFilters(user);
const total = await this.widgetRepo.getAnnotationCount(user.id, accessibleLibraryIds, contentFilters);
@@ -153,6 +185,7 @@ export class DashboardWidgetService {
async getHighlights(user: RequestUser): Promise<HighlightsWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.liveCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('highlights', clock), async () => {
const contentFilters = this.getContentFilters(user);
const total = await this.widgetRepo.getAnnotationCount(user.id, accessibleLibraryIds, contentFilters);
@@ -180,6 +213,7 @@ export class DashboardWidgetService {
async getMonthlyChallenge(user: RequestUser): Promise<MonthlyChallengeWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('monthly-challenge', clock), async () => {
const contentFilters = this.getContentFilters(user);
const { year, month } = clock;
@@ -219,6 +253,7 @@ export class DashboardWidgetService {
async getYearProjection(user: RequestUser): Promise<YearProjectionWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('year-projection', clock), async () => {
const contentFilters = this.getContentFilters(user);
const yearStartDay = `${clock.year}-01-01`;
@@ -252,6 +287,7 @@ export class DashboardWidgetService {
async getNeglectedGems(user: RequestUser): Promise<NeglectedGemsWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), 'neglected-gems', async () => {
const contentFilters = this.getContentFilters(user);
return this.widgetRepo.getNeglectedGems(user.id, accessibleLibraryIds, new Date(), contentFilters);
@@ -261,6 +297,7 @@ export class DashboardWidgetService {
async getReadingDna(user: RequestUser): Promise<ReadingDnaWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('reading-dna', clock), async () => {
const contentFilters = this.getContentFilters(user);
const since = new Date();
@@ -272,6 +309,7 @@ export class DashboardWidgetService {
async getLongWait(user: RequestUser): Promise<LongWaitWidgetData | null> {
const accessibleLibraryIds = await this.getLibraryIds(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), 'long-wait', async () => {
const contentFilters = this.getContentFilters(user);
return this.widgetRepo.getLongWait(user.id, accessibleLibraryIds, new Date(), contentFilters);
@@ -280,6 +318,7 @@ export class DashboardWidgetService {
async getDiversityScore(user: RequestUser): Promise<DiversityScoreWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.staleCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), 'diversity-score', async () => {
const contentFilters = this.getContentFilters(user);
const data = await this.widgetRepo.getDiversityData(user.id, accessibleLibraryIds, contentFilters);
@@ -297,6 +336,7 @@ export class DashboardWidgetService {
async getReadingRhythm(user: RequestUser): Promise<ReadingRhythmWidgetData> {
const accessibleLibraryIds = await this.getLibraryIds(user);
const clock = this.readerClock(user);
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.liveCache.get(this.cacheOwnerKey(user, accessibleLibraryIds), this.calendarCacheKey('reading-rhythm', clock), async () => {
const contentFilters = this.getContentFilters(user);
const since = addDateKeyDays(clock.today, -(RHYTHM_WINDOW_DAYS - 1));
server/src/modules/dashboard/dashboard.module.ts
+2 -1
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsModule } from '../user-book-status/reading-attempt-events.module';
import { Module } from '@nestjs/common';
import { BookModule } from '../book/book.module';
@@ -11,7 +12,7 @@ import { DashboardWidgetRepository } from './dashboard-widget.repository';
import { DashboardWidgetService } from './dashboard-widget.service';
@Module({
imports: [BookModule, BookCoverStoreModule, LibraryModule, SmartScopeModule],
imports: [BookModule, BookCoverStoreModule, LibraryModule, SmartScopeModule, ReadingAttemptEventsModule],
controllers: [DashboardController],
providers: [DashboardService, DashboardRepository, DashboardWidgetService, DashboardWidgetRepository],
exports: [DashboardService, DashboardWidgetService],
server/src/modules/hardcover/hardcover-import.service.test.ts
+42 -1
View File
@@ -1,3 +1,5 @@
import { READING_ATTEMPT_CHANGED, ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import type { ReadingAttemptService } from '../user-book-status/reading-attempt.service';
import { BadRequestException, Logger } from '@nestjs/common';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
@@ -32,13 +34,14 @@ const user = {
contentFilters: { includeTagIds: [], excludeTagIds: [], includeGenreIds: [], excludeGenreIds: [] },
};
function makeService() {
function makeService(attempts?: ReadingAttemptService) {
return new HardcoverImportService(
mockRepo as never,
mockClient as never,
mockSettingsService as never,
mockLibraryService as never,
mockUserBookStatusService as never,
attempts,
);
}
@@ -902,4 +905,42 @@ describe('HardcoverImportService', () => {
await expect(makeService().previewImport(user as never)).rejects.toBe('settings unavailable');
});
it('coalesces Hardcover reads and projections across the whole import, including partial failures', async () => {
const events = new ReadingAttemptEventsService();
const listener = vi.fn();
events.on(READING_ATTEMPT_CHANGED, listener);
const coalesce = vi.spyOn(events, 'coalesceChanges');
const attempts = {
coalesceChanges: (operation: () => Promise<unknown>) => events.coalesceChanges(operation),
importExternalRead: vi.fn(() => {
events.notifyChanged(7);
return Promise.resolve();
}),
};
const service = makeService(attempts as never);
const preview = await service.previewImport(user as never);
vi.spyOn(service as any, 'buildPreview').mockResolvedValue({
...preview,
rows: Array.from({ length: 121 }, (_, index) => ({
...preview.rows[0],
hardcoverUserBookId: 1000 + index,
localBookId: 42 + index,
hardcoverReads: [
{ id: index * 2 + 1, startedAt: null, finishedAt: '2026-01-15' },
{ id: index * 2 + 2, startedAt: null, finishedAt: '2026-01-16' },
],
})),
});
mockUserBookStatusService.updateManual.mockImplementation(() => {
events.notifyChanged(7);
return Promise.resolve();
});
mockRepo.upsertBookState.mockRejectedValueOnce(new Error('State write failed'));
const result = await service.applyImport(user as never);
expect(result).toMatchObject({ applied: 120, failed: 1 });
expect(attempts.importExternalRead).toHaveBeenCalledTimes(242);
expect(coalesce).toHaveBeenCalledOnce();
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 7 });
expect(listener.mock.calls.every(([payload]) => payload.userId === 7)).toBe(true);
});
});
server/src/modules/hardcover/hardcover-import.service.ts
+61 -57
View File
@@ -173,67 +173,71 @@ export class HardcoverImportService {
let progressApplied = 0;
let failed = 0;
for (const row of preview.rows) {
if (!canApplyRow(row, selectedIds) || row.localBookId == null || row.importedStatus == null) continue;
const applyRows = async () => {
for (const row of preview.rows) {
if (!canApplyRow(row, selectedIds) || row.localBookId == null || row.importedStatus == null) continue;
try {
if (this.readingAttempts) {
for (const read of [...(row.hardcoverReads ?? [])].sort((left, right) => left.id - right.id)) {
await this.readingAttempts.importExternalRead(user.id, row.localBookId, {
provider: 'hardcover',
externalId: String(read.id),
startedOn: read.startedAt?.slice(0, 10) ?? null,
endedOn: read.finishedAt?.slice(0, 10) ?? null,
});
try {
if (this.readingAttempts) {
for (const read of [...(row.hardcoverReads ?? [])].sort((left, right) => left.id - right.id)) {
await this.readingAttempts.importExternalRead(user.id, row.localBookId, {
provider: 'hardcover',
externalId: String(read.id),
startedOn: read.startedAt?.slice(0, 10) ?? null,
endedOn: read.finishedAt?.slice(0, 10) ?? null,
});
}
}
// want_to_read cannot carry lifecycle dates, and an unfinished status cannot carry a finish
// date. The imported reads above already hold the real dates either way.
const clearsLifecycle = row.importedStatus === 'want_to_read';
const isActiveStatus = row.importedStatus === 'reading' || row.importedStatus === 'on_hold';
await this.userBookStatusService.updateManual(user.id, row.localBookId, {
status: row.importedStatus,
...(clearsLifecycle
? {}
: {
startedAt: toDate(row.importedStartedAt),
finishedAt: isActiveStatus ? null : toDate(row.importedFinishedAt),
}),
});
const progressImported = await this.applyProgressIfRequested(user.id, row, importProgress, startedAt);
await this.repo.upsertBookState({
userId: user.id,
bookId: row.localBookId,
hardcoverBookId: row.hardcoverBookId,
hardcoverEditionId: row.hardcoverEditionId,
hardcoverUserBookId: row.hardcoverUserBookId,
hardcoverReadId: row.hardcoverReadId,
matchMethod: row.matchMethod,
matchError: null,
syncError: null,
lastSyncedAt: new Date(),
lastSyncedStatus: row.importedStatus,
lastSyncedProgress: progressImported ? row.importedProgressPercent : null,
lastSyncedStartedAt: row.importedStartedAt,
lastSyncedFinishedAt: row.importedFinishedAt,
});
if (progressImported) {
progressApplied++;
}
applied++;
} catch (err) {
failed++;
const errorClass = err instanceof Error ? err.constructor.name : 'Error';
const error = sanitizeLogValue(err instanceof Error ? err.message : String(err));
this.logger.warn(
`[hardcover.import_status] [fail] userId=${user.id} bookId=${row.localBookId} hardcoverBookId=${row.hardcoverBookId} durationMs=${Date.now() - startedAt} errorClass=${errorClass} error="${error}" - import row failed`,
);
}
// want_to_read cannot carry lifecycle dates, and an unfinished status cannot carry a finish
// date. The imported reads above already hold the real dates either way.
const clearsLifecycle = row.importedStatus === 'want_to_read';
const isActiveStatus = row.importedStatus === 'reading' || row.importedStatus === 'on_hold';
await this.userBookStatusService.updateManual(user.id, row.localBookId, {
status: row.importedStatus,
...(clearsLifecycle
? {}
: {
startedAt: toDate(row.importedStartedAt),
finishedAt: isActiveStatus ? null : toDate(row.importedFinishedAt),
}),
});
const progressImported = await this.applyProgressIfRequested(user.id, row, importProgress, startedAt);
await this.repo.upsertBookState({
userId: user.id,
bookId: row.localBookId,
hardcoverBookId: row.hardcoverBookId,
hardcoverEditionId: row.hardcoverEditionId,
hardcoverUserBookId: row.hardcoverUserBookId,
hardcoverReadId: row.hardcoverReadId,
matchMethod: row.matchMethod,
matchError: null,
syncError: null,
lastSyncedAt: new Date(),
lastSyncedStatus: row.importedStatus,
lastSyncedProgress: progressImported ? row.importedProgressPercent : null,
lastSyncedStartedAt: row.importedStartedAt,
lastSyncedFinishedAt: row.importedFinishedAt,
});
if (progressImported) {
progressApplied++;
}
applied++;
} catch (err) {
failed++;
const errorClass = err instanceof Error ? err.constructor.name : 'Error';
const error = sanitizeLogValue(err instanceof Error ? err.message : String(err));
this.logger.warn(
`[hardcover.import_status] [fail] userId=${user.id} bookId=${row.localBookId} hardcoverBookId=${row.hardcoverBookId} durationMs=${Date.now() - startedAt} errorClass=${errorClass} error="${error}" - import row failed`,
);
}
}
};
if (this.readingAttempts) await this.readingAttempts.coalesceChanges(applyRows);
else await applyRows();
const result: HardcoverImportApplyResult = {
...preview.summary,
server/src/modules/library/library.module.ts
+2
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsModule } from '../user-book-status/reading-attempt-events.module';
import { Module, forwardRef } from '@nestjs/common';
import { AchievementModule } from '../achievement/achievement.module';
@@ -16,6 +17,7 @@ import { LibraryAddedAtService } from './library-added-at.service';
@Module({
imports: [
ReadingAttemptEventsModule,
ScannerModule,
AchievementModule,
forwardRef(() => BookModule),
server/src/modules/library/library.service.test.ts
+18
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
vi.mock('fs/promises', () => ({
readdir: vi.fn(),
realpath: vi.fn(),
@@ -96,10 +97,12 @@ describe('LibraryService', () => {
};
let service: LibraryService;
let readingEvents: ReadingAttemptEventsService;
beforeEach(() => {
vi.resetAllMocks();
config.get.mockReturnValue('/books');
readingEvents = new ReadingAttemptEventsService();
service = new LibraryService(
libraryRepo as any,
config as any,
@@ -109,6 +112,7 @@ describe('LibraryService', () => {
achievementEvents as any,
pathPolicy as any,
scanScheduler as any,
readingEvents,
);
libraryRepo.findPodcastIds.mockResolvedValue([]);
@@ -623,7 +627,13 @@ describe('LibraryService', () => {
libraryRepo.findById.mockResolvedValue([{ id: 4, name: 'L' }]);
libraryRepo.findBookIdsByLibrary.mockResolvedValue([{ id: 101 }, { id: 102 }]);
const notify = vi.spyOn(readingEvents, 'notifyChanged');
libraryRepo.delete.mockImplementation(() => {
expect(notify).not.toHaveBeenCalled();
return Promise.resolve();
});
await service.remove(4);
expect(notify).toHaveBeenCalledExactlyOnceWith(null);
expect(fileWatcherService.stopWatcher).toHaveBeenCalledWith(4);
expect(libraryRepo.delete).toHaveBeenCalledWith(4);
@@ -632,6 +642,14 @@ describe('LibraryService', () => {
expect(mockRm).toHaveBeenCalledWith('/books/covers/102', { recursive: true, force: true });
});
it('does not invalidate reading caches when deleting the library fails', async () => {
libraryRepo.findBookIdsByLibrary.mockResolvedValue([]);
libraryRepo.delete.mockRejectedValueOnce(new Error('Delete failed'));
const notify = vi.spyOn(readingEvents, 'notifyChanged');
await expect(service.remove(1)).rejects.toThrow('Delete failed');
expect(notify).not.toHaveBeenCalled();
});
it('remove deletes downloaded podcast files before deleting the library', async () => {
libraryRepo.findById.mockResolvedValue([{ id: 8, name: 'Podcasts', type: 'podcasts' }]);
libraryRepo.findBookIdsByLibrary.mockResolvedValue([]);
server/src/modules/library/library.service.ts
+3
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import {
BadRequestException,
ConflictException,
@@ -97,6 +98,7 @@ export class LibraryService {
private readonly achievementEvents: AchievementEventsService,
private readonly pathPolicy: PathPolicyService,
private readonly scanScheduler: LibraryScanSchedulerService,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {
this.appDataPath = this.config.get<string>('storage.appDataPath')!;
}
@@ -348,6 +350,7 @@ export class LibraryService {
const removedPodcastFiles = existing.type === 'podcasts' ? await this.removePodcastFiles(id) : 0;
if (existing.type === 'podcasts') await this.removePodcastAppDataFiles(id);
await this.libraryRepo.delete(id);
this.readingAttemptEvents.notifyChanged(null);
this.scanScheduler.removeSchedule(id);
await this.cleanupCoverDirectories(bookRows.map(({ id: bookId }) => bookId));
this.logger.log(
server/src/modules/maintenance/maintenance.module.ts
+2 -1
View File
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
import { BookModule } from '../book/book.module';
import { BookCoverStoreModule } from '../book-cover-store/book-cover-store.module';
import { ReadingAttemptEventsModule } from '../user-book-status/reading-attempt-events.module';
import { LibraryModule } from '../library/library.module';
import { CoverSweepStore } from './cover-sweep.store';
import { MissingResourcesController } from './missing-resources.controller';
@@ -9,7 +10,7 @@ import { MissingResourcesRepository } from './missing-resources.repository';
import { MissingResourcesService } from './missing-resources.service';
@Module({
imports: [BookModule, BookCoverStoreModule, LibraryModule],
imports: [BookModule, BookCoverStoreModule, LibraryModule, ReadingAttemptEventsModule],
controllers: [MissingResourcesController],
providers: [MissingResourcesService, MissingResourcesRepository, CoverSweepStore],
})
server/src/modules/maintenance/missing-resources.service.test.ts
+34 -1
View File
@@ -5,6 +5,7 @@ import { join } from 'path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { CoverSweepStore } from './cover-sweep.store';
import { ReadingAttemptEventsService, READING_ATTEMPT_CHANGED } from '../user-book-status/reading-attempt-events.service';
import { MissingResourcesService } from './missing-resources.service';
const user = { id: 7, isSuperuser: true } as never;
@@ -71,6 +72,7 @@ describe('MissingResourcesService', () => {
const libraryService = { findAccessibleLibraryIds: vi.fn().mockResolvedValue([1, 2]) };
const bookService = { deleteBooks: vi.fn().mockImplementation((ids: number[]) => Promise.resolve({ total: ids.length, books: [], omitted: 0 })) };
const config = { get: vi.fn().mockReturnValue(appDataPath) };
const events = new ReadingAttemptEventsService();
const service = new MissingResourcesService(
repo as never,
store,
@@ -78,8 +80,9 @@ describe('MissingResourcesService', () => {
bookService as never,
config as never,
coverStore as never,
events,
);
return { service, repo, libraryService, bookService, coverStore };
return { service, repo, libraryService, bookService, coverStore, events };
}
async function runSweep(service: MissingResourcesService): Promise<void> {
@@ -251,6 +254,36 @@ describe('MissingResourcesService', () => {
expect(result).toMatchObject({ cleaned: 2, remaining: 0 });
});
it('coalesces global cache invalidation across every bounded deletion batch', async () => {
const ids = Array.from({ length: 1043 }, (_, index) => index + 1);
const { service, events, bookService } = setup({ findMissingBookIds: vi.fn().mockResolvedValue(ids) });
const listener = vi.fn();
events.on(READING_ATTEMPT_CHANGED, listener);
bookService.deleteBooks.mockImplementation((batch: number[]) => {
events.notifyChanged(null);
expect(listener).not.toHaveBeenCalled();
return Promise.resolve({ total: batch.length, books: [], omitted: 0 });
});
expect(await service.cleanMissingBooks(user, { all: true })).toMatchObject({ cleaned: 1043 });
expect(bookService.deleteBooks.mock.calls.map(([batch]) => batch.length)).toEqual([500, 500, 43]);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: null });
});
it('flushes committed deletions when a later cleanup batch fails', async () => {
const ids = Array.from({ length: 501 }, (_, index) => index + 1);
const { service, events, bookService } = setup({ findMissingBookIds: vi.fn().mockResolvedValue(ids) });
const listener = vi.fn();
events.on(READING_ATTEMPT_CHANGED, listener);
bookService.deleteBooks.mockImplementation((batch: number[]) => {
if (batch[0] === 501) return Promise.reject(new Error('Deletion failed'));
events.notifyChanged(null);
expect(listener).not.toHaveBeenCalled();
return Promise.resolve({ total: batch.length, books: [], omitted: 0 });
});
await expect(service.cleanMissingBooks(user, { all: true })).rejects.toThrow('Deletion failed');
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: null });
});
it('rejects a request with neither ids nor all', async () => {
const { service } = setup();
await expect(service.cleanMissingBooks(user, {})).rejects.toThrow('Either bookIds or all must be provided');
server/src/modules/maintenance/missing-resources.service.ts
+9 -5
View File
@@ -20,6 +20,7 @@ import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils';
import { BookService } from '../book/book.service';
import type { BookCoverSlotRow } from '../book-cover-store/book-cover-store.repository';
import { BookCoverStore } from '../book-cover-store/book-cover-store.service';
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import { LibraryService } from '../library/library.service';
import { CoverSweepStore, type SweepRecord } from './cover-sweep.store';
import {
@@ -49,6 +50,7 @@ export class MissingResourcesService {
private readonly bookService: BookService,
config: ConfigService,
private readonly coverStore: BookCoverStore,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {
this.coversRoot = join(config.get<string>('storage.appDataPath')!, 'covers');
}
@@ -140,11 +142,13 @@ export class MissingResourcesService {
this.logger.log(`[${event}] [start] userId=${user.id} requested=${requestedIds.length} targets=${targets.length} - missing book cleanup started`);
try {
let cleaned = 0;
for (let index = 0; index < targets.length; index += DELETE_BATCH_SIZE) {
const batch = targets.slice(index, index + DELETE_BATCH_SIZE);
const result = await this.bookService.deleteBooks(batch, user);
cleaned += result.total;
}
await this.readingAttemptEvents.coalesceChanges(async () => {
for (let index = 0; index < targets.length; index += DELETE_BATCH_SIZE) {
const batch = targets.slice(index, index + DELETE_BATCH_SIZE);
const result = await this.bookService.deleteBooks(batch, user);
cleaned += result.total;
}
});
const remaining = await this.repo.countMissingBooks(libraryIds);
this.logger.log(
`[${event}] [end] userId=${user.id} requested=${requestedIds.length} cleaned=${cleaned} remaining=${remaining} durationMs=${Date.now() - startedAt} - missing book cleanup completed`,
server/src/modules/migration/executor/migration-executor.service.test.ts
+6 -5
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../../user-book-status/reading-attempt-events.service';
import { mkdir, mkdtemp, rm, writeFile } from 'fs/promises';
import { tmpdir } from 'os';
import { join } from 'path';
@@ -106,7 +107,7 @@ describe('UserStateImporter audiobook progress import', () => {
const importRepo = createImportRepoMock();
importRepo.upsertAudiobookProgress.mockResolvedValue(undefined);
const importer = new UserStateImporter(repo as never, importRepo as never);
const importer = new UserStateImporter(repo as never, importRepo as never, new ReadingAttemptEventsService());
const planned = {
execution: {
@@ -196,7 +197,7 @@ describe('UserStateImporter audiobook progress import', () => {
const importRepo = createImportRepoMock();
importRepo.upsertAudiobookProgress.mockResolvedValue(undefined);
const importer = new UserStateImporter(repo as never, importRepo as never);
const importer = new UserStateImporter(repo as never, importRepo as never, new ReadingAttemptEventsService());
const planned = {
execution: {
@@ -386,7 +387,7 @@ describe('UserStateImporter collection import', () => {
importRepo.insertCollection.mockResolvedValue({ id: 701 });
importRepo.upsertCollectionBook.mockResolvedValue(true);
const importer = new UserStateImporter(repo as never, importRepo as never);
const importer = new UserStateImporter(repo as never, importRepo as never, new ReadingAttemptEventsService());
const planned = {
execution: {
@@ -465,7 +466,7 @@ describe('UserStateImporter progress and bookmark edge cases', () => {
it('marks user-state stages skipped when unavailable', async () => {
const repo = createRepoMock();
const importRepo = createImportRepoMock();
const importer = new UserStateImporter(repo as never, importRepo as never);
const importer = new UserStateImporter(repo as never, importRepo as never, new ReadingAttemptEventsService());
const planned = {
plan: {
@@ -508,7 +509,7 @@ describe('UserStateImporter progress and bookmark edge cases', () => {
it('derives bookmark position from track index when source file id is missing', async () => {
const repo = createRepoMock();
const importRepo = createImportRepoMock();
const importer = new UserStateImporter(repo as never, importRepo as never);
const importer = new UserStateImporter(repo as never, importRepo as never, new ReadingAttemptEventsService());
const planned = {
execution: {
server/src/modules/migration/executor/user-state.importer.test.ts
+58 -2
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../../user-book-status/reading-attempt-events.service';
import { UserStateImporter } from './user-state.importer';
function makeImporter() {
@@ -33,8 +34,9 @@ function makeImporter() {
batchInsertCollectionBooks: vi.fn().mockResolvedValue(undefined),
};
const importer = new UserStateImporter(repo as never, importRepo as never);
return { importer, repo, importRepo };
const events = new ReadingAttemptEventsService();
const importer = new UserStateImporter(repo as never, importRepo as never, events);
return { importer, repo, importRepo, events };
}
describe('UserStateImporter', () => {
@@ -1026,3 +1028,57 @@ describe('UserStateImporter', () => {
expect(audioBatch).toEqual([]);
});
});
describe('UserStateImporter cache invalidation', () => {
function plan() {
return {
execution: {
sourceData: {
userBookStatuses: [
{ sourceUserId: 'reader', sourceBookId: 'a', status: 'read' },
{ sourceUserId: 'reader', sourceBookId: 'b', status: 'read' },
],
},
},
};
}
it('invalidates each affected reader once after the status transaction commits', async () => {
const { importer, importRepo, events } = makeImporter();
const notify = vi.spyOn(events, 'notifyChanged');
importRepo.withTransaction.mockImplementation(async (callback) => {
await callback(importRepo);
expect(notify).not.toHaveBeenCalled();
});
await (importer as any).importUserBookStatuses(
1,
plan(),
new Map([['reader', 7]]),
new Map([
['a', 10],
['b', 11],
]),
async () => {},
);
expect(notify).toHaveBeenCalledExactlyOnceWith(7);
});
it('keeps caches when the status transaction rolls back', async () => {
const { importer, importRepo, events } = makeImporter();
const notify = vi.spyOn(events, 'notifyChanged');
importRepo.withTransaction.mockRejectedValueOnce(new Error('Rollback'));
await expect(
(importer as any).importUserBookStatuses(
1,
plan(),
new Map([['reader', 7]]),
new Map([
['a', 10],
['b', 11],
]),
async () => {},
),
).rejects.toThrow('Rollback');
expect(notify).not.toHaveBeenCalled();
});
});
server/src/modules/migration/executor/user-state.importer.ts
+3
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../../user-book-status/reading-attempt-events.service';
import { createHash } from 'node:crypto';
import { Injectable } from '@nestjs/common';
@@ -124,6 +125,7 @@ export class UserStateImporter {
constructor(
private readonly repo: MigrationRepository,
private readonly importRepo: MigrationImportRepository,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {}
async import(runId: number, planned: PlannerResult, ensureRunning: RunStateCheck): Promise<void> {
@@ -209,6 +211,7 @@ export class UserStateImporter {
await this.importRepo.withTransaction(async (importRepo) => {
await importRepo.batchMergeUserBookStatuses(dedupedBatch);
});
for (const userId of new Set(dedupedBatch.map((item) => item.userId))) this.readingAttemptEvents.notifyChanged(userId);
await this.repo.setRunMetric(runId, 'user_state', 'user_book_status', counters);
}
server/src/modules/migration/migration.module.ts
+2
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsModule } from '../user-book-status/reading-attempt-events.module';
import { Module, forwardRef } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { JwtModule } from '@nestjs/jwt';
@@ -37,6 +38,7 @@ import { MigrationReportingService } from './reporting/migration-reporting.servi
@Module({
imports: [
ReadingAttemptEventsModule,
AuthModule,
forwardRef(() => NotificationModule),
ScannerModule,
server/src/modules/reading-state/reading-state.module.ts
+2 -1
View File
@@ -1,12 +1,13 @@
import { Module } from '@nestjs/common';
import { BookModule } from '../book/book.module';
import { UserBookStatusModule } from '../user-book-status/user-book-status.module';
import { ReadingStateController } from './reading-state.controller';
import { ReadingStateRepository } from './reading-state.repository';
import { ReadingStateService } from './reading-state.service';
@Module({
imports: [BookModule],
imports: [BookModule, UserBookStatusModule],
controllers: [ReadingStateController],
providers: [ReadingStateService, ReadingStateRepository],
})
server/src/modules/reading-state/reading-state.service.test.ts
+35 -2
View File
@@ -3,6 +3,7 @@ import { EMPTY_CONTENT_FILTER_RULES } from '@bookorbit/types';
import type { RequestUser } from '../../common/types/request-user';
import { BookService } from '../book/book.service';
import { READING_ATTEMPT_CHANGED, ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import { ReadingStateRepository, type ResetReadingStateResult } from './reading-state.repository';
import { ReadingStateService } from './reading-state.service';
@@ -51,11 +52,13 @@ function makeService() {
const repo = {
resetBookReadingState: vi.fn().mockResolvedValue(makeResult()),
};
const events = new ReadingAttemptEventsService();
return {
service: new ReadingStateService(bookService as unknown as BookService, repo as unknown as ReadingStateRepository),
service: new ReadingStateService(bookService as unknown as BookService, repo as unknown as ReadingStateRepository, events),
bookService,
repo,
events,
};
}
@@ -86,7 +89,9 @@ describe('ReadingStateService', () => {
});
it('propagates access failures, skips the repository, and emits a sanitized failure log', async () => {
const { service, bookService, repo } = makeService();
const { service, bookService, repo, events } = makeService();
const changed = vi.fn();
events.on(READING_ATTEMPT_CHANGED, changed);
const warnSpy = vi.spyOn(Logger.prototype, 'warn');
const error = new ForbiddenException('cannot reset "another user"\nstate');
bookService.verifyBookAccess.mockRejectedValue(error);
@@ -94,6 +99,34 @@ describe('ReadingStateService', () => {
await expect(service.resetBookReadingState(42, makeUser())).rejects.toThrow(error);
expect(repo.resetBookReadingState).not.toHaveBeenCalled();
expect(changed).not.toHaveBeenCalled();
expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('error="cannot reset \\"another user\\" state" - reset reading state failed'));
});
it('notifies dashboard caches only after the reading-state reset commits', async () => {
const { service, repo, events } = makeService();
const changed = vi.fn();
events.on(READING_ATTEMPT_CHANGED, changed);
let completeReset!: (result: ResetReadingStateResult) => void;
repo.resetBookReadingState.mockReturnValueOnce(
new Promise<ResetReadingStateResult>((resolve) => {
completeReset = resolve;
}),
);
const pending = service.resetBookReadingState(42, makeUser());
await vi.waitFor(() => expect(repo.resetBookReadingState).toHaveBeenCalledOnce());
expect(changed).not.toHaveBeenCalled();
completeReset(makeResult());
await pending;
expect(changed).toHaveBeenCalledExactlyOnceWith({ userId: 7 });
});
it('does not invalidate dashboard caches when the reset transaction fails', async () => {
const { service, repo, events } = makeService();
const changed = vi.fn();
events.on(READING_ATTEMPT_CHANGED, changed);
repo.resetBookReadingState.mockRejectedValueOnce(new Error('Reset rolled back'));
await expect(service.resetBookReadingState(42, makeUser())).rejects.toThrow('Reset rolled back');
expect(changed).not.toHaveBeenCalled();
});
});
server/src/modules/reading-state/reading-state.service.ts
+3
View File
@@ -5,6 +5,7 @@ import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils';
import { resolveTimeZone } from '../../common/utils/timezone.utils';
import type { RequestUser } from '../../common/types/request-user';
import { BookService } from '../book/book.service';
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import { ReadingStateRepository } from './reading-state.repository';
@Injectable()
@@ -14,6 +15,7 @@ export class ReadingStateService {
constructor(
private readonly bookService: BookService,
private readonly repo: ReadingStateRepository,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {}
async resetBookReadingState(bookId: number, user: RequestUser): Promise<ResetBookReadingStateResponse> {
@@ -25,6 +27,7 @@ export class ReadingStateService {
await this.bookService.verifyBookAccess(bookId, user);
const timeZone = resolveTimeZone((user.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC');
const result = await this.repo.resetBookReadingState(user.id, bookId, timeZone);
this.readingAttemptEvents.notifyChanged(user.id);
this.logger.log(
`[${event}] [end] bookId=${bookId} userId=${user.id} durationMs=${Date.now() - startedAt} sessionsDeleted=${result.sessionsDeleted} progressDeleted=${result.progressDeleted} audioProgressDeleted=${result.audioProgressDeleted} koreaderDeviceProgressDeleted=${result.koreaderDeviceProgressDeleted} koreaderPageStatsDeleted=${result.koreaderPageStatsDeleted} koboStateReset=${result.koboStateReset} - reset reading state completed`,
);
server/src/modules/user-book-status/reading-attempt-events.module.ts
+8
View File
@@ -0,0 +1,8 @@
import { Module } from '@nestjs/common';
import { ReadingAttemptEventsService } from './reading-attempt-events.service';
@Module({
providers: [ReadingAttemptEventsService],
exports: [ReadingAttemptEventsService],
})
export class ReadingAttemptEventsModule {}
server/src/modules/user-book-status/reading-attempt-events.service.test.ts
+135
View File
@@ -0,0 +1,135 @@
import { Logger } from '@nestjs/common';
import { Test } from '@nestjs/testing';
import { READING_ATTEMPT_CHANGED, ReadingAttemptEventsService } from './reading-attempt-events.service';
function deferred() {
let resolve!: () => void;
const promise = new Promise<void>((done) => {
resolve = done;
});
return { promise, resolve };
}
describe('ReadingAttemptEventsService', () => {
let events: ReadingAttemptEventsService;
let listener: ReturnType<typeof vi.fn>;
beforeEach(async () => {
const module = await Test.createTestingModule({ providers: [ReadingAttemptEventsService] }).compile();
events = module.get(ReadingAttemptEventsService);
listener = vi.fn();
events.on(READING_ATTEMPT_CHANGED, listener);
});
it('notifies a single mutation synchronously', () => {
events.notifyChanged(1);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
});
it('routes direct emit calls through batching and error isolation', async () => {
const warn = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => {});
events.prependListener(READING_ATTEMPT_CHANGED, () => {
throw new Error('Listener failed');
});
await events.coalesceChanges(() => {
expect(events.emit(READING_ATTEMPT_CHANGED, { userId: 1 })).toBe(true);
events.emit(READING_ATTEMPT_CHANGED, { userId: 1 });
expect(listener).not.toHaveBeenCalled();
expect(warn).not.toHaveBeenCalled();
});
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
expect(warn).toHaveBeenCalledOnce();
expect(() => events.emit(READING_ATTEMPT_CHANGED, { userId: 11 })).not.toThrow();
expect(listener).toHaveBeenLastCalledWith({ userId: 11 });
warn.mockRestore();
});
it('coalesces repeated and nested changes by user', async () => {
await events.coalesceChanges(async () => {
for (let i = 0; i < 10000; i++) events.notifyChanged(1);
await events.coalesceChanges(() => {
events.notifyChanged(1);
events.notifyChanged(11);
});
expect(listener).not.toHaveBeenCalled();
});
expect(listener.mock.calls).toEqual([[{ userId: 1 }], [{ userId: 11 }]]);
});
it('flushes committed changes even if later work fails, preserving the original error', async () => {
const warn = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => {});
events.prependListener(READING_ATTEMPT_CHANGED, () => {
throw new Error('Listener failed');
});
const original = new Error('Write failed');
await expect(
events.coalesceChanges(() => {
events.notifyChanged(1);
throw original;
}),
).rejects.toBe(original);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
expect(warn).toHaveBeenCalledOnce();
warn.mockRestore();
});
it('isolates a throwing listener and preserves once listeners', () => {
const warn = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => {});
const once = vi.fn();
events.prependListener(READING_ATTEMPT_CHANGED, () => {
throw new Error('Listener failed');
});
events.once(READING_ATTEMPT_CHANGED, once);
expect(() => events.notifyChanged(1)).not.toThrow();
events.notifyChanged(1);
expect(listener).toHaveBeenCalledTimes(2);
expect(once).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
warn.mockRestore();
});
it('flushes changes for a reader during a batch and flushes later writes again at the end', async () => {
await events.coalesceChanges(() => {
events.notifyChanged(1);
events.notifyChanged(11);
events.flushPendingChanges(1);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
events.flushPendingChanges(1);
expect(listener).toHaveBeenCalledOnce();
events.notifyChanged(1);
});
expect(listener.mock.calls).toEqual([[{ userId: 1 }], [{ userId: 11 }], [{ userId: 1 }]]);
});
it('keeps unrelated requests independent of a running batch for the same user', async () => {
const gate = deferred();
const batch = events.coalesceChanges(async () => {
events.notifyChanged(1);
await gate.promise;
});
events.notifyChanged(1);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
gate.resolve();
await batch;
expect(listener).toHaveBeenCalledTimes(2);
});
it('flushes shared deletions during a batch for any reader', async () => {
await events.coalesceChanges(() => {
events.notifyChanged(null);
events.flushPendingChanges(11);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: null });
});
expect(listener).toHaveBeenCalledOnce();
});
it('does not strand changes from work that outlives its batch context', async () => {
const gate = deferred();
let late!: Promise<void>;
await events.coalesceChanges(() => {
late = gate.promise.then(() => events.notifyChanged(1));
});
gate.resolve();
await late;
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
});
});
server/src/modules/user-book-status/reading-attempt-events.service.ts
+71
View File
@@ -0,0 +1,71 @@
import { Injectable, Logger } from '@nestjs/common';
import { AsyncLocalStorage } from 'node:async_hooks';
import { EventEmitter } from 'node:events';
import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils';
export const READING_ATTEMPT_CHANGED = 'reading-attempt.changed';
export interface ReadingAttemptChangedPayload {
userId: number | null;
}
type ChangeBatch = { pending: Set<number | null>; closed: boolean };
// Notifications invalidate only this process; other replicas fall back to their cache TTLs.
@Injectable()
export class ReadingAttemptEventsService extends EventEmitter {
private readonly logger = new Logger(ReadingAttemptEventsService.name);
private readonly context = new AsyncLocalStorage<ChangeBatch>();
private readonly batches = new Set<ChangeBatch>();
override emit(eventName: string | symbol, ...args: unknown[]): boolean {
if (eventName !== READING_ATTEMPT_CHANGED) return super.emit(eventName, ...args);
const payload = args[0] as ReadingAttemptChangedPayload;
this.notifyChanged(payload.userId);
return this.listenerCount(eventName) > 0;
}
notifyChanged(userId: number | null): void {
const batch = this.context.getStore();
if (batch && !batch.closed) batch.pending.add(userId);
else this.publish(userId);
}
async coalesceChanges<T>(operation: () => T | Promise<T>): Promise<T> {
const parent = this.context.getStore();
if (parent && !parent.closed) return operation();
const batch: ChangeBatch = { pending: new Set(), closed: false };
this.batches.add(batch);
try {
return await this.context.run(batch, operation);
} finally {
batch.closed = true;
this.batches.delete(batch);
for (const userId of batch.pending) this.publish(userId);
}
}
// A read during a bulk operation must observe all changes that have already committed.
flushPendingChanges(userId: number): void {
let changed = false;
let allChanged = false;
for (const batch of this.batches) {
changed = batch.pending.delete(userId) || changed;
allChanged = batch.pending.delete(null) || allChanged;
}
if (allChanged) this.publish(null);
else if (changed) this.publish(userId);
}
private publish(userId: number | null): void {
for (const listener of this.rawListeners(READING_ATTEMPT_CHANGED)) {
try {
listener.call(this, { userId } satisfies ReadingAttemptChangedPayload);
} catch (error) {
this.logger.warn(
`[reading_attempt.notify] [fail] userId=${userId ?? 'all'} durationMs=0 errorClass=${error instanceof Error ? error.name : 'Error'} error="${sanitizeLogValue(error instanceof Error ? error.message : String(error))}" - cache listener failed`,
);
}
}
}
}
server/src/modules/user-book-status/reading-attempt.repository.ts
+3 -2
View File
@@ -233,8 +233,8 @@ export class ReadingAttemptRepository {
return { items, total: totals[0]?.total ?? 0 };
}
async findOwned(userId: number, bookId: number, attemptId: number) {
const [row] = await this.db
async findOwned(userId: number, bookId: number, attemptId: number, tx?: Tx) {
const query = (tx ?? this.db)
.select()
.from(readingAttempts)
.where(
@@ -246,6 +246,7 @@ export class ReadingAttemptRepository {
),
)
.limit(1);
const [row] = await (tx ? query.for('update') : query);
return row ?? null;
}
server/src/modules/user-book-status/reading-attempt.service.test.ts
+4 -128
View File
@@ -1,132 +1,8 @@
import { BadRequestException, NotFoundException } from '@nestjs/common';
import type { ReadingAttemptOrigin, ReadingAttemptOutcome, ReadStatus } from '@bookorbit/types';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { beforeEach, describe, expect, it } from 'vitest';
import { ReadingAttemptService } from './reading-attempt.service';
type Row = {
id: number;
userId: number;
bookId: number;
startedOn: string | null;
endedOn: string | null;
outcome: ReadingAttemptOutcome | null;
origin: ReadingAttemptOrigin;
externalProvider: string | null;
externalId: string | null;
deletedAt: Date | null;
createdAt: Date;
updatedAt: Date;
};
function makeFakeRepo() {
const rows: Row[] = [];
let nextId = 1;
const transactionContext = {};
const projections: Array<{
status: ReadStatus;
source: 'manual' | 'auto';
startedAt: Date | null;
finishedAt: Date | null;
}> = [];
function assertValidAttempt(candidate: Pick<Row, 'userId' | 'bookId' | 'startedOn' | 'endedOn' | 'outcome' | 'deletedAt'>, excludedId?: number) {
if (candidate.endedOn !== null && candidate.outcome === null) {
throw new Error('reading_attempts_closed_has_outcome_chk');
}
if (candidate.startedOn !== null && candidate.endedOn !== null && candidate.endedOn < candidate.startedOn) {
throw new Error('reading_attempts_end_after_start_chk');
}
if (candidate.outcome === null && candidate.deletedAt === null) {
const duplicateActive = rows.some(
(row) =>
row.id !== excludedId &&
row.userId === candidate.userId &&
row.bookId === candidate.bookId &&
row.outcome === null &&
row.deletedAt === null,
);
if (duplicateActive) throw new Error('reading_attempts_one_active_uidx');
}
}
const repo = {
transaction: vi.fn((callback: (tx: object) => Promise<unknown>) => callback(transactionContext)),
findActive: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve(rows.find((row) => row.userId === userId && row.bookId === bookId && row.outcome === null && row.deletedAt === null)),
),
findLatest: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve([...rows].reverse().find((row) => row.userId === userId && row.bookId === bookId && row.deletedAt === null)),
),
hasCompleted: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve(rows.some((row) => row.userId === userId && row.bookId === bookId && row.outcome === 'completed' && row.deletedAt === null)),
),
create: vi.fn(
(
_tx: object,
values: Omit<Row, 'id' | 'deletedAt' | 'createdAt' | 'updatedAt' | 'externalProvider' | 'externalId'> & {
externalProvider?: string | null;
externalId?: string | null;
},
) => {
const now = new Date('2026-07-12T12:00:00.000Z');
const row: Row = {
...values,
id: nextId++,
externalProvider: values.externalProvider ?? null,
externalId: values.externalId ?? null,
deletedAt: null,
createdAt: now,
updatedAt: now,
};
assertValidAttempt(row);
rows.push(row);
return Promise.resolve(row);
},
),
createActive: vi.fn(
(
_tx: object,
values: Omit<Row, 'id' | 'deletedAt' | 'createdAt' | 'updatedAt' | 'externalProvider' | 'externalId' | 'endedOn' | 'outcome'> & {
externalProvider?: string | null;
externalId?: string | null;
},
) => {
const existing = rows.find(
(row) => row.userId === values.userId && row.bookId === values.bookId && row.outcome === null && row.deletedAt === null,
);
if (existing) return Promise.resolve(existing);
return repo.create(_tx, { ...values, endedOn: null, outcome: null });
},
),
update: vi.fn((_tx: object, userId: number, bookId: number, id: number, patch: Partial<Row>) => {
const row = rows.find((item) => item.id === id && item.userId === userId && item.bookId === bookId && item.deletedAt === null);
if (!row) return Promise.resolve(null);
assertValidAttempt({ ...row, ...patch }, row.id);
Object.assign(row, patch, { updatedAt: new Date('2026-07-12T12:00:00.000Z') });
return Promise.resolve(row);
}),
project: vi.fn((_tx: object, _userId: number, _bookId: number, projection: (typeof projections)[number]) => {
projections.push(projection);
return Promise.resolve();
}),
findStatus: vi.fn(() => Promise.resolve(null)),
findByExternal: vi.fn((_tx: object, userId: number, provider: string, externalId: string) =>
Promise.resolve(rows.find((row) => row.userId === userId && row.externalProvider === provider && row.externalId === externalId)),
),
findOwned: vi.fn((userId: number, bookId: number, id: number) =>
Promise.resolve(rows.find((row) => row.id === id && row.userId === userId && row.bookId === bookId && row.deletedAt === null)),
),
softDelete: vi.fn((userId: number, bookId: number, id: number) => {
const row = rows.find((item) => item.id === id && item.userId === userId && item.bookId === bookId && item.deletedAt === null);
if (!row) return Promise.resolve(false);
row.deletedAt = new Date();
return Promise.resolve(true);
}),
list: vi.fn(() => Promise.resolve({ items: [], total: 0 })),
};
return { repo, rows, projections, transactionContext };
}
import { ReadingAttemptEventsService } from './reading-attempt-events.service';
import { makeFakeRepo } from '../../../test/helpers/reading-attempt-repository.fake';
describe('ReadingAttemptService', () => {
let fake: ReturnType<typeof makeFakeRepo>;
@@ -134,7 +10,7 @@ describe('ReadingAttemptService', () => {
beforeEach(() => {
fake = makeFakeRepo();
service = new ReadingAttemptService(fake.repo as never);
service = new ReadingAttemptService(fake.repo as never, new ReadingAttemptEventsService());
});
it('creates a placeholder completion before a legacy manual reread', async () => {
server/src/modules/user-book-status/reading-attempt.service.ts
+98 -44
View File
@@ -9,7 +9,8 @@ import type {
UserBookStatus,
} from '@bookorbit/types';
import { ReadingAttemptRepository } from './reading-attempt.repository';
import { ReadingAttemptRepository, type AttemptProjection } from './reading-attempt.repository';
import { ReadingAttemptEventsService } from './reading-attempt-events.service';
import { READING_DATE_ERROR_CODES } from './user-book-status.constants';
import { hasReachedProgressThreshold } from '../../common/utils/progress-threshold.utils';
@@ -51,7 +52,14 @@ type ManualStatusOptions = {
@Injectable()
export class ReadingAttemptService {
constructor(private readonly repo: ReadingAttemptRepository) {}
constructor(
private readonly repo: ReadingAttemptRepository,
private readonly events: ReadingAttemptEventsService,
) {}
coalesceChanges<T>(operation: () => T | Promise<T>): Promise<T> {
return this.events.coalesceChanges(operation);
}
async applyManualStatus(
userId: number,
@@ -62,7 +70,8 @@ export class ReadingAttemptService {
today: string,
options: ManualStatusOptions = {},
): Promise<UserBookStatus> {
return this.repo.transaction(async (tx) => {
let changed = false;
const result = await this.repo.transaction<UserBookStatus>(async (tx) => {
let active: Awaited<ReturnType<ReadingAttemptRepository['findActive']>> | null = await this.repo.findActive(tx, userId, bookId);
let latest = active ?? (await this.repo.findLatest(tx, userId, bookId));
const hasStartedOn = startedOn !== undefined;
@@ -75,6 +84,8 @@ export class ReadingAttemptService {
attemptId: number,
patch: Parameters<ReadingAttemptRepository['update']>[4],
): Promise<NonNullable<Awaited<ReturnType<ReadingAttemptRepository['update']>>>> => {
const original = active?.id === attemptId ? active : latest;
changed ||= Object.entries(patch).some(([key, value]) => original?.[key as keyof typeof original] !== value);
const updated = await this.repo.update(tx, userId, bookId, attemptId, patch);
if (!updated) throw new InternalServerErrorException('Reading attempt disappeared during update');
return updated;
@@ -86,6 +97,7 @@ export class ReadingAttemptService {
if (endedOn !== undefined && endedOn !== null) {
const resolvedStartedOn = startedOn ?? null;
this.validateDates(resolvedStartedOn, endedOn);
changed = true;
latest = await this.repo.create(tx, {
userId,
bookId,
@@ -96,6 +108,7 @@ export class ReadingAttemptService {
});
projectedStatus = 'read';
} else if (startedOn !== undefined && startedOn !== null) {
changed = true;
active = await this.repo.createActive(tx, {
userId,
bookId,
@@ -144,6 +157,7 @@ export class ReadingAttemptService {
}
if (status === 'rereading' && !(await this.repo.hasCompleted(tx, userId, bookId))) {
changed = true;
await this.repo.create(tx, {
userId,
bookId,
@@ -156,6 +170,7 @@ export class ReadingAttemptService {
if (isActiveStatus) {
if (!active) {
changed = true;
active = await this.repo.createActive(tx, {
userId,
bookId,
@@ -184,6 +199,7 @@ export class ReadingAttemptService {
const resolvedStartedOn = startedOn ?? null;
const resolvedEndedOn = hasEndedOn ? (endedOn ?? null) : notBeforeStart(today, resolvedStartedOn);
this.validateDates(resolvedStartedOn, resolvedEndedOn);
changed = true;
latest = await this.repo.create(tx, {
userId,
bookId,
@@ -228,14 +244,17 @@ export class ReadingAttemptService {
}
}
const current = await this.repo.findStatus(tx, userId, bookId);
const projectionTarget = active ?? latest;
const projected = projectionDatesFor(projectedStatus, projectionTarget);
await this.repo.project(tx, userId, bookId, {
const projection: AttemptProjection = {
status: projectedStatus,
source: 'manual',
startedAt: dateToUtcDate(projected.startedOn),
finishedAt: dateToUtcDate(projected.endedOn),
});
};
changed ||= this.hasProjectionChanged(current, projection);
await this.repo.project(tx, userId, bookId, projection);
return {
status: projectedStatus,
source: 'manual',
@@ -244,6 +263,8 @@ export class ReadingAttemptService {
updatedAt: new Date().toISOString(),
};
});
if (changed) this.events.notifyChanged(userId);
return result;
}
async recordActivity(input: {
@@ -256,7 +277,8 @@ export class ReadingAttemptService {
strongRereadEvidence: boolean;
meaningfulActivity: boolean;
}): Promise<UserBookStatus | null> {
return this.repo.transaction(async (tx) => {
let changed = false;
const result = await this.repo.transaction<UserBookStatus | null>(async (tx) => {
let active: Awaited<ReturnType<ReadingAttemptRepository['findActive']>> | null = await this.repo.findActive(tx, input.userId, input.bookId);
let latest = active ?? (await this.repo.findLatest(tx, input.userId, input.bookId));
const hasCompleted = await this.repo.hasCompleted(tx, input.userId, input.bookId);
@@ -272,6 +294,7 @@ export class ReadingAttemptService {
outcome: 'completed',
origin: input.origin,
});
changed = true;
}
if (!active && isFinished && hasCompleted && input.strongRereadEvidence) {
latest = await this.repo.create(tx, {
@@ -282,6 +305,7 @@ export class ReadingAttemptService {
outcome: 'completed',
origin: input.origin,
});
changed = true;
}
if (!active && !isFinished && (input.progress > 0 || input.strongRereadEvidence || input.meaningfulActivity)) {
active = await this.repo.createActive(tx, {
@@ -290,6 +314,7 @@ export class ReadingAttemptService {
startedOn: input.occurredOn,
origin: input.origin,
});
changed = true;
}
let projectionTarget = active ?? latest;
@@ -299,6 +324,7 @@ export class ReadingAttemptService {
endedOn: notBeforeStart(input.occurredOn, active.startedOn),
outcome: 'completed',
});
changed = true;
status = 'read';
} else if (active) {
status = hasCompleted ? 'rereading' : 'reading';
@@ -308,12 +334,15 @@ export class ReadingAttemptService {
return null;
}
await this.repo.project(tx, input.userId, input.bookId, {
const current = await this.repo.findStatus(tx, input.userId, input.bookId);
const projection: AttemptProjection = {
status,
source: 'auto',
startedAt: dateToUtcDate(projectionTarget?.startedOn ?? null),
finishedAt: projectionTarget?.outcome === 'completed' ? dateToUtcDate(projectionTarget.endedOn) : null,
});
};
changed ||= this.hasProjectionChanged(current, projection);
await this.repo.project(tx, input.userId, input.bookId, projection);
return {
status,
source: 'auto',
@@ -322,6 +351,8 @@ export class ReadingAttemptService {
updatedAt: new Date().toISOString(),
};
});
if (changed) this.events.notifyChanged(input.userId);
return result;
}
async list(userId: number, bookId: number, page = 1, pageSize = 20): Promise<ReadingAttemptListResponse> {
@@ -360,15 +391,16 @@ export class ReadingAttemptService {
input: { provider: 'hardcover'; externalId: string; startedOn: string | null; endedOn: string | null },
): Promise<void> {
this.validateDates(input.startedOn, input.endedOn);
await this.repo.transaction(async (tx) => {
const changed = await this.repo.transaction(async (tx) => {
const existing = await this.repo.findByExternal(tx, userId, input.provider, input.externalId);
if (existing?.deletedAt) return;
if (existing?.deletedAt) return false;
if (existing) {
await this.repo.update(tx, userId, bookId, existing.id, {
const patch = {
...(existing.startedOn === null && input.startedOn !== null ? { startedOn: input.startedOn } : {}),
...(existing.endedOn === null && input.endedOn !== null ? { endedOn: input.endedOn, outcome: 'completed' } : {}),
});
return;
...(existing.endedOn === null && input.endedOn !== null ? { endedOn: input.endedOn, outcome: 'completed' as const } : {}),
};
if (Object.keys(patch).length === 0) return false;
return (await this.repo.update(tx, userId, bookId, existing.id, patch)) !== null;
}
const active = input.endedOn === null ? await this.repo.findActive(tx, userId, bookId) : null;
if (input.endedOn === null && !active) {
@@ -392,20 +424,26 @@ export class ReadingAttemptService {
externalId: input.externalId,
});
}
return true;
});
if (changed) this.events.notifyChanged(userId);
}
async update(userId: number, bookId: number, attemptId: number, patch: ReadingAttemptPatch): Promise<ReadingAttempt> {
const existing = await this.repo.findOwned(userId, bookId, attemptId);
if (!existing) throw new NotFoundException('Reading attempt not found');
const startedOn = patch.startedOn === undefined ? existing.startedOn : patch.startedOn;
const endedOn = patch.endedOn === undefined ? existing.endedOn : patch.endedOn;
const outcome = patch.outcome === undefined ? existing.outcome : patch.outcome;
this.validateDates(startedOn, endedOn);
if (endedOn && outcome === null) throw new BadRequestException('A closed attempt requires an outcome');
const row = await this.repo.transaction((tx) => this.repo.update(tx, userId, bookId, attemptId, patch));
if (!row) throw new NotFoundException('Reading attempt not found');
await this.rebuildProjection(userId, bookId);
const { row, changed } = await this.repo.transaction(async (tx) => {
const existing = await this.repo.findOwned(userId, bookId, attemptId, tx);
if (!existing) throw new NotFoundException('Reading attempt not found');
const startedOn = patch.startedOn === undefined ? existing.startedOn : patch.startedOn;
const endedOn = patch.endedOn === undefined ? existing.endedOn : patch.endedOn;
const outcome = patch.outcome === undefined ? existing.outcome : patch.outcome;
this.validateDates(startedOn, endedOn);
if (endedOn && outcome === null) throw new BadRequestException('A closed attempt requires an outcome');
const changed = Object.entries(patch).some(([key, value]) => existing[key as keyof typeof existing] !== value);
const row = await this.repo.update(tx, userId, bookId, attemptId, patch);
if (!row) throw new NotFoundException('Reading attempt not found');
return { row, changed };
});
await this.rebuildProjection(userId, bookId, changed);
return this.toDto({ ...row, totalSessions: 0, totalSeconds: 0 });
}
@@ -414,27 +452,43 @@ export class ReadingAttemptService {
await this.rebuildProjection(userId, bookId);
}
private async rebuildProjection(userId: number, bookId: number): Promise<void> {
await this.repo.transaction(async (tx) => {
const active = await this.repo.findActive(tx, userId, bookId);
const latest = active ?? (await this.repo.findLatest(tx, userId, bookId));
const current = await this.repo.findStatus(tx, userId, bookId);
const hasCompleted = await this.repo.hasCompleted(tx, userId, bookId);
const target = active ?? latest;
let status: ReadStatus = 'unread';
if (active) status = current?.status === 'on_hold' ? 'on_hold' : hasCompleted ? 'rereading' : 'reading';
else if (current?.status === 'want_to_read' || current?.status === 'unread') status = current.status;
else if (latest?.outcome === 'completed') status = 'read';
else if (latest?.outcome === 'skimmed') status = 'skimmed';
else if (latest?.outcome === 'abandoned') status = 'abandoned';
const projected = projectionDatesFor(status, target);
await this.repo.project(tx, userId, bookId, {
status,
source: 'manual',
startedAt: dateToUtcDate(projected.startedOn),
finishedAt: dateToUtcDate(projected.endedOn),
private async rebuildProjection(userId: number, bookId: number, changed = true): Promise<void> {
try {
await this.repo.transaction(async (tx) => {
const active = await this.repo.findActive(tx, userId, bookId);
const latest = active ?? (await this.repo.findLatest(tx, userId, bookId));
const current = await this.repo.findStatus(tx, userId, bookId);
const hasCompleted = await this.repo.hasCompleted(tx, userId, bookId);
const target = active ?? latest;
let status: ReadStatus = 'unread';
if (active) status = current?.status === 'on_hold' ? 'on_hold' : hasCompleted ? 'rereading' : 'reading';
else if (current?.status === 'want_to_read' || current?.status === 'unread') status = current.status;
else if (latest?.outcome === 'completed') status = 'read';
else if (latest?.outcome === 'skimmed') status = 'skimmed';
else if (latest?.outcome === 'abandoned') status = 'abandoned';
const projected = projectionDatesFor(status, target);
const projection: AttemptProjection = {
status,
source: 'manual',
startedAt: dateToUtcDate(projected.startedOn),
finishedAt: dateToUtcDate(projected.endedOn),
};
changed ||= this.hasProjectionChanged(current, projection);
await this.repo.project(tx, userId, bookId, projection);
});
});
} finally {
// The history mutation already committed even if rebuilding its projection fails.
if (changed) this.events.notifyChanged(userId);
}
}
private hasProjectionChanged(current: Awaited<ReturnType<ReadingAttemptRepository['findStatus']>>, projection: AttemptProjection): boolean {
return (
current?.status !== projection.status ||
current?.source !== projection.source ||
(current?.startedAt?.toISOString() ?? null) !== (projection.startedAt?.toISOString() ?? null) ||
(current?.finishedAt?.toISOString() ?? null) !== (projection.finishedAt?.toISOString() ?? null)
);
}
private validateDates(startedOn: string | null, endedOn: string | null): void {
server/src/modules/user-book-status/user-book-status.module.ts
+3 -2
View File
@@ -7,9 +7,10 @@ import { UserBookStatusService } from './user-book-status.service';
import { ReadingAttemptRepository } from './reading-attempt.repository';
import { ReadingAttemptService } from './reading-attempt.service';
import { ReadingAttemptBackfillService } from './reading-attempt-backfill.service';
import { ReadingAttemptEventsModule } from './reading-attempt-events.module';
@Module({
imports: [AchievementModule],
imports: [AchievementModule, ReadingAttemptEventsModule],
providers: [
UserBookStatusService,
UserBookStatusRepository,
@@ -18,6 +19,6 @@ import { ReadingAttemptBackfillService } from './reading-attempt-backfill.servic
ReadingAttemptRepository,
ReadingAttemptBackfillService,
],
exports: [UserBookStatusService, ReadingAttemptService],
exports: [UserBookStatusService, ReadingAttemptService, ReadingAttemptEventsModule],
})
export class UserBookStatusModule {}
server/src/modules/user-book-status/user-book-status.service.test.ts
+2
View File
@@ -665,6 +665,7 @@ describe('kobo status projection', () => {
describe('with reading attempts enabled', () => {
const mockAttempts = {
applyManualStatus: vi.fn(),
coalesceChanges: vi.fn((operation: () => Promise<unknown>) => operation()),
};
let attemptService: UserBookStatusService;
@@ -781,6 +782,7 @@ describe('reading dates are filed on the reader local day', () => {
const mockAttempts = {
recordActivity: vi.fn(),
applyManualStatus: vi.fn(),
coalesceChanges: vi.fn((operation: () => Promise<unknown>) => operation()),
};
let attemptService: UserBookStatusService;
server/src/modules/user-book-status/user-book-status.service.ts
+15 -10
View File
@@ -171,16 +171,21 @@ export class UserBookStatusService {
};
if (this.attempts) {
const batchSize = 50;
const timeZone = await this.repo.findUserTimeZone(userId);
for (let offset = 0; offset < bookIds.length; offset += batchSize) {
const batch = bookIds.slice(offset, offset + batchSize);
const results = await Promise.all(batch.map((bookId) => this.applyManualStatus(userId, bookId, { status }, undefined, timeZone)));
results.forEach((result, index) => {
if (result.statusChanged) recordChange(result.state.status, batch[index]!);
});
}
await this.projectGroupsToKobo(userId, changedByStatus);
await this.attempts.coalesceChanges(async () => {
const batchSize = 50;
const timeZone = await this.repo.findUserTimeZone(userId);
for (let offset = 0; offset < bookIds.length; offset += batchSize) {
const batch = bookIds.slice(offset, offset + batchSize);
const settled = await Promise.allSettled(batch.map((bookId) => this.applyManualStatus(userId, bookId, { status }, undefined, timeZone)));
const failure = settled.find((result) => result.status === 'rejected');
if (failure?.status === 'rejected') throw failure.reason;
const results = settled.filter((result) => result.status === 'fulfilled').map((result) => result.value);
results.forEach((result, index) => {
if (result.statusChanged) recordChange(result.state.status, batch[index]!);
});
}
await this.projectGroupsToKobo(userId, changedByStatus);
});
return;
}
server/src/modules/user-statistics/user-statistics.module.ts
+2 -1
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsModule } from '../user-book-status/reading-attempt-events.module';
import { Module } from '@nestjs/common';
import { AppSettingsModule } from '../app-settings/app-settings.module';
@@ -9,7 +10,7 @@ import { UserStatisticsRepository } from './user-statistics.repository';
import { UserStatisticsService } from './user-statistics.service';
@Module({
imports: [AppSettingsModule],
imports: [AppSettingsModule, ReadingAttemptEventsModule],
controllers: [UserStatisticsController],
providers: [UserStatisticsService, UserStatisticsRepository, UserStatisticsAggregationJob, UserReadingStatsTimeZoneBackfillService],
exports: [UserStatisticsService],
server/src/modules/user-statistics/user-statistics.service.test.ts
+35 -34
View File
@@ -1,3 +1,4 @@
import { ReadingAttemptEventsService } from '../user-book-status/reading-attempt-events.service';
import { UserStatisticsService } from './user-statistics.service';
describe('UserStatisticsService', () => {
@@ -59,7 +60,7 @@ describe('UserStatisticsService', () => {
}),
getProgressFunnelInRange: vi.fn().mockResolvedValue(funnel),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getActivityOverview(
{
id: 7,
@@ -115,7 +116,7 @@ describe('UserStatisticsService', () => {
},
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getActivityDay({ id: 7, isSuperuser: false, settings: { timezone: 'UTC' } } as any, '2026-04-08', {
libraryIds: [1],
});
@@ -132,7 +133,7 @@ describe('UserStatisticsService', () => {
});
it('rejects future Activity calendar years', async () => {
const service = new UserStatisticsService({} as any);
const service = new UserStatisticsService({} as any, new ReadingAttemptEventsService());
await expect(service.getActivityCalendar({ id: 7, settings: { timezone: 'UTC' } } as any, 2027, { libraryIds: [] })).rejects.toThrow(
'Invalid activity year',
);
@@ -144,7 +145,7 @@ describe('UserStatisticsService', () => {
getActivityAvailableYears: vi.fn().mockResolvedValue([2024, 2026]),
getActivitySessionPage: vi.fn().mockResolvedValue([]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(
service.getActivityCalendar({ id: 7, isSuperuser: false, settings: { timezone: 'UTC' } } as any, 2023, { libraryIds: [2] }),
@@ -168,7 +169,7 @@ describe('UserStatisticsService', () => {
}),
getProgressFunnelInRange: vi.fn().mockResolvedValue({ started: 0, reached25: 0, reached50: 0, reached75: 0, completed: 0 }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getActivityOverview({ id: 7, isSuperuser: false, settings: { timezone: 'UTC' } } as any, {
libraryIds: [2, 99],
@@ -187,7 +188,7 @@ describe('UserStatisticsService', () => {
{ id: 4, startedAt: new Date('2026-03-30T18:00:00Z'), endedAt: new Date('2026-03-30T19:30:00Z'), durationSeconds: 5400 },
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getActivitySessionPatterns({ id: 7, isSuperuser: false, settings: { timezone: 'America/Denver' } } as any, {
libraryIds: [3],
@@ -215,7 +216,7 @@ describe('UserStatisticsService', () => {
{ firstStartedAt: new Date('2026-01-01T20:00:00Z'), completedAt: new Date('2026-03-20T20:00:00Z'), format: 'PDF' },
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getActivityCompletionSpeed({ id: 7, isSuperuser: false, settings: { timezone: 'UTC' } } as any, {
libraryIds: [1],
@@ -255,7 +256,7 @@ describe('UserStatisticsService', () => {
bands: [{ band: '15_to_30', medianProgressDelta: 2.5, sampleCount: 3 }],
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const user = { id: 7, isSuperuser: false, settings: { timezone: 'UTC' } } as any;
const genre = await service.getActivityGenreTime(user, { libraryIds: [1] });
@@ -285,7 +286,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getSummary({ id: 123, isSuperuser: false } as any, { libraryIds: [1, 2] });
@@ -310,7 +311,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getReadingHeatmap({ id: 123, isSuperuser: false } as any, { days: 3, libraryIds: [] });
expect(repo.getDailyReadingStats).toHaveBeenCalledWith(123, false, [], 3);
@@ -352,7 +353,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getReadingSourceDistribution({ id: 123, isSuperuser: false } as any, { libraryIds: [] });
expect(repo.getDailyReadingSecondsBySource).toHaveBeenCalledWith(123, false, [], 365);
@@ -370,7 +371,7 @@ describe('UserStatisticsService', () => {
it('returns an empty source distribution and honours the days override', async () => {
const repo = { getDailyReadingSecondsBySource: vi.fn().mockResolvedValue([]) };
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getReadingSourceDistribution({ id: 7, isSuperuser: true } as any, { days: 30, libraryIds: [1] });
@@ -385,7 +386,7 @@ describe('UserStatisticsService', () => {
.mockResolvedValueOnce([{ day: '2026-04-06', source: 'web', readingSeconds: 60 }])
.mockResolvedValueOnce([{ day: '2026-04-06', source: 'watchos', readingSeconds: 90 }]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const user = { id: 7, isSuperuser: false } as any;
const first = await service.getReadingSourceDistribution(user, { libraryIds: [] });
@@ -410,7 +411,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getPeakReadingHours({ id: 123, isSuperuser: false } as any, { libraryIds: [] });
expect(repo.getPeakReadingHours).toHaveBeenCalledWith(123, false, [], 365, 'UTC');
@@ -446,7 +447,7 @@ describe('UserStatisticsService', () => {
getPeakReadingHours: vi.fn().mockResolvedValue([]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await service.getPeakReadingHours({ id: 123, isSuperuser: false, settings: { timezone: 'Australia/Brisbane' } } as any, {
days: 30,
libraryIds: [2],
@@ -463,7 +464,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getFavoriteReadingDays({ id: 123, isSuperuser: false } as any, { libraryIds: [] });
expect(repo.getFavoriteReadingDays).toHaveBeenCalledWith(123, false, [], 365);
@@ -492,7 +493,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getCompletionTimeline({ id: 123, isSuperuser: false } as any, { days: 120, libraryIds: [] });
expect(repo.getCompletionTimeline).toHaveBeenCalledWith(123, false, [], 120);
@@ -513,7 +514,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getGoalTrajectory({ id: 123, isSuperuser: false } as any, { days: 120, goalBooks: 12, libraryIds: [] });
expect(repo.getMonthlyCompletions).toHaveBeenCalledWith(123, false, [], 120);
@@ -538,7 +539,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getProgressFunnel({ id: 123, isSuperuser: false } as any, { libraryIds: [] });
expect(repo.getProgressFunnelInRange).toHaveBeenCalledTimes(1);
@@ -558,7 +559,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const user = { id: 7, isSuperuser: false } as any;
const query = { libraryIds: [2, 1], days: 365 };
@@ -574,7 +575,7 @@ describe('UserStatisticsService', () => {
getCompletionLatencyDays: vi.fn().mockResolvedValue([2, 10, 20, 60, 120, 400]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getCompletionLatency({ id: 123, isSuperuser: false } as any, { libraryIds: [] });
expect(repo.getCompletionLatencyDays).toHaveBeenCalledWith(123, false, [], 1825);
@@ -602,7 +603,7 @@ describe('UserStatisticsService', () => {
recomputeRecentDailyStats: vi.fn().mockResolvedValue({ deleted: 1, inserted: 2, since: '2026-04-07' }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const user = { id: 9, isSuperuser: false } as any;
const query = { libraryIds: [1], days: 365 };
@@ -624,7 +625,7 @@ describe('UserStatisticsService', () => {
.mockResolvedValueOnce({ started: 18, reached25: 13, reached50: 9, reached75: 6, completed: 4 }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getProgressFunnel({ id: 123, isSuperuser: false } as any, { libraryIds: [1], days: 365, comparePrevious: true });
expect(repo.getProgressFunnelInRange).toHaveBeenCalledTimes(2);
@@ -651,7 +652,7 @@ describe('UserStatisticsService', () => {
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.getSessionTimeline({ id: 123, isSuperuser: false } as any, { libraryIds: [1] });
expect(result.year).toBe(2026);
@@ -701,7 +702,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const result = await service.updateSessionTimelineSession(
{ id: 123, isSuperuser: false } as any,
101,
@@ -750,7 +751,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(
service.updateSessionTimelineSession(
{ id: 123, isSuperuser: false } as any,
@@ -781,7 +782,7 @@ describe('UserStatisticsService', () => {
moveSessionTimelineSessionAtomic: vi.fn(),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(
service.updateSessionTimelineSession(
{ id: 123, isSuperuser: false } as any,
@@ -802,7 +803,7 @@ describe('UserStatisticsService', () => {
moveSessionTimelineSessionAtomic: vi.fn(),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(
service.updateSessionTimelineSession(
{ id: 123, isSuperuser: false } as any,
@@ -859,7 +860,7 @@ describe('UserStatisticsService', () => {
}),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(
service.updateSessionTimelineSession(
{ id: 123, isSuperuser: false } as any,
@@ -878,7 +879,7 @@ describe('UserStatisticsService', () => {
getDailyReadingStats: vi.fn().mockResolvedValue([{ day: '2026-04-08', readingSeconds: 10, progressDelta: 1.234567, eventsCount: 1 }]),
getCompletionLatencyDays: vi.fn().mockResolvedValue([]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
await expect(service.getDailyReading({ id: 11, isSuperuser: false } as any, { days: 1, libraryIds: [] })).resolves.toEqual([
{ day: '2026-04-08', readingSeconds: 10, progressDelta: 1.2346, eventsCount: 1 },
@@ -912,7 +913,7 @@ describe('UserStatisticsService', () => {
{ bookId: 2, title: 'Single Session', startedAt: new Date('2026-04-02T00:00:00.000Z'), endProgress: 90 },
]),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const survival = await service.getReadingSurvival({ id: 11, isSuperuser: false } as any, { days: 365, libraryIds: [] });
expect(survival.find((point) => point.threshold === 50)).toEqual({
@@ -943,7 +944,7 @@ describe('UserStatisticsService', () => {
getReadingPacePoints: vi.fn().mockResolvedValue([{ durationSeconds: 240, progressDelta: 1.4, bucket: 'bookorbit', format: 'EPUB' }]),
getAuthorGenreChord: vi.fn().mockResolvedValue({ nodes: [{ name: 'A' }, { name: 'G' }], links: [{ source: 'A', target: 'G', value: 10 }] }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const user = { id: 9, isSuperuser: false } as any;
await expect(service.getSessionArchetypes(user, { libraryIds: [1] })).resolves.toEqual([{ hour: 9, durationMinutes: 20, dayOfWeek: 2 }]);
@@ -990,7 +991,7 @@ describe('UserStatisticsService', () => {
getSessionTimelineSessionById: vi.fn().mockResolvedValue(existing),
moveSessionTimelineSessionAtomic: vi.fn().mockResolvedValue({ updated, conflict: null }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const userA = { id: 10, isSuperuser: false } as any;
const userB = { id: 20, isSuperuser: false } as any;
const query = { libraryIds: [1] };
@@ -1017,7 +1018,7 @@ describe('UserStatisticsService', () => {
getSummary: vi.fn().mockResolvedValue({ trackedBooks: 1, startedBooks: 1, inProgressBooks: 1, completedBooks: 0, meanProgressPercent: 10 }),
rebuildDailyStatsForUser: vi.fn().mockResolvedValue({ deleted: 4, inserted: 2, libraries: 1 }),
};
const service = new UserStatisticsService(repo as any);
const service = new UserStatisticsService(repo as any, new ReadingAttemptEventsService());
const userA = { id: 1, isSuperuser: false } as any;
const userB = { id: 2, isSuperuser: false } as any;
const query = { libraryIds: [1] };
server/src/modules/user-statistics/user-statistics.service.ts
+51 -4
View File
@@ -1,4 +1,9 @@
import { BadRequestException, ConflictException, Injectable, NotFoundException } from '@nestjs/common';
import {
READING_ATTEMPT_CHANGED,
ReadingAttemptEventsService,
type ReadingAttemptChangedPayload,
} from '../user-book-status/reading-attempt-events.service';
import { BadRequestException, ConflictException, Injectable, NotFoundException, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import type {
ActivityCalendarResponse,
@@ -76,10 +81,29 @@ const ACTIVITY_DETAILS_WINDOW_DAYS = 365;
const ACTIVITY_DETAILS_LONG_WINDOW_DAYS = 1825;
@Injectable()
export class UserStatisticsService {
private readonly cache = new StatsCache({ ttlMs: USER_STATS_CACHE_TTL_MS, maxEntries: USER_STATS_CACHE_MAX_ENTRIES });
export class UserStatisticsService implements OnModuleInit, OnModuleDestroy {
private readonly cache = new StatsCache({
ttlMs: USER_STATS_CACHE_TTL_MS,
maxEntries: USER_STATS_CACHE_MAX_ENTRIES,
});
constructor(private readonly repo: UserStatisticsRepository) {}
constructor(
private readonly repo: UserStatisticsRepository,
private readonly readingAttemptEvents: ReadingAttemptEventsService,
) {}
private readonly onReadingAttemptChanged = ({ userId }: ReadingAttemptChangedPayload): void => {
if (userId === null) this.cache.clear();
else this.invalidateUser(userId);
};
onModuleInit(): void {
this.readingAttemptEvents.on(READING_ATTEMPT_CHANGED, this.onReadingAttemptChanged);
}
onModuleDestroy(): void {
this.readingAttemptEvents.removeListener(READING_ATTEMPT_CHANGED, this.onReadingAttemptChanged);
}
invalidateUser(userId: number): void {
this.cache.clearForScope(String(userId));
@@ -266,6 +290,7 @@ export class UserStatisticsService {
today,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const rangeStartDay = addDateKeyDays(today, -(ACTIVITY_OVERVIEW_DAYS - 1));
@@ -469,6 +494,7 @@ export class UserStatisticsService {
year,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const availableYears = [...new Set([...(await this.repo.getActivityAvailableYears(user.id, libraryIds, timeZone)), currentYear])].sort(
@@ -542,6 +568,7 @@ export class UserStatisticsService {
today,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const range = getDayRangeForDateKeys([startDay, today], timeZone)!;
@@ -603,6 +630,7 @@ export class UserStatisticsService {
format: selectedFormat ?? '',
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const range = getDayRangeForDateKeys([startDay, today], timeZone)!;
@@ -657,6 +685,7 @@ export class UserStatisticsService {
today,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const range = getDayRangeForDateKeys([startDay, today], timeZone)!;
@@ -700,6 +729,7 @@ export class UserStatisticsService {
media: selectedMedia ?? '',
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const libraryIds = await this.repo.resolveActivityLibraryIds(user.id, user.isSuperuser, query.libraryIds);
const range = getDayRangeForDateKeys([startDay, today], timeZone)!;
@@ -726,6 +756,7 @@ export class UserStatisticsService {
async getSummary(user: RequestUser, query: UserStatisticsFilterQueryDto): Promise<UserStatisticsSummary> {
const key = this.buildUserCacheKey('summary', user, { libraries: this.normalizeLibraryIds(query.libraryIds) });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const summary = await this.repo.getSummary(user.id, user.isSuperuser, query.libraryIds);
return {
@@ -738,6 +769,7 @@ export class UserStatisticsService {
async getDailyReading(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserDailyReadingStat[]> {
const days = query.days ?? 365;
const key = this.buildUserCacheKey('daily-reading', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const items = await this.repo.getDailyReadingStats(user.id, user.isSuperuser, query.libraryIds, days);
return items.map((item) => ({
@@ -750,6 +782,7 @@ export class UserStatisticsService {
async getReadingHeatmap(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserDailyReadingStat[]> {
const days = query.days ?? HEATMAP_DEFAULT_DAYS;
const key = this.buildUserCacheKey('reading-heatmap', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const items = await this.repo.getDailyReadingStats(user.id, user.isSuperuser, query.libraryIds, days);
const bySourceRows = await this.repo.getDailyReadingSecondsBySource(user.id, user.isSuperuser, query.libraryIds, days);
@@ -786,6 +819,7 @@ export class UserStatisticsService {
async getReadingSourceDistribution(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserReadingSourceDistribution> {
const days = query.days ?? BEHAVIOR_DEFAULT_DAYS;
const key = this.buildUserCacheKey('source-distribution', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getDailyReadingSecondsBySource(user.id, user.isSuperuser, query.libraryIds, days);
const totals = emptySourceBucketRecord();
@@ -805,6 +839,7 @@ export class UserStatisticsService {
const days = query.days ?? BEHAVIOR_DEFAULT_DAYS;
const timeZone = resolveTimeZone((user.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC');
const key = this.buildUserCacheKey('peak-hours', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days, timeZone });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getPeakReadingHours(user.id, user.isSuperuser, query.libraryIds, days, timeZone);
@@ -840,6 +875,7 @@ export class UserStatisticsService {
async getFavoriteReadingDays(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserFavoriteDayStat[]> {
const days = query.days ?? BEHAVIOR_DEFAULT_DAYS;
const key = this.buildUserCacheKey('favorite-days', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getFavoriteReadingDays(user.id, user.isSuperuser, query.libraryIds, days);
const byDay = new Map<
@@ -910,6 +946,7 @@ export class UserStatisticsService {
week,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getSessionTimelineItems(
user.id,
@@ -984,6 +1021,7 @@ export class UserStatisticsService {
async getCompletionTimeline(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserCompletionTimelinePoint[]> {
const days = query.days ?? COMPLETION_TIMELINE_DEFAULT_DAYS;
const key = this.buildUserCacheKey('completion-timeline', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getCompletionTimeline(user.id, user.isSuperuser, query.libraryIds, days);
const byMonth = new Map(rows.map((row) => [`${row.year}-${row.month}`, row.count]));
@@ -1013,6 +1051,7 @@ export class UserStatisticsService {
days,
goalBooks,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getMonthlyCompletions(user.id, user.isSuperuser, query.libraryIds, days);
const byMonth = new Map(rows.map((row) => [`${row.year}-${row.month}`, row.count]));
@@ -1051,6 +1090,7 @@ export class UserStatisticsService {
comparePrevious: comparePrevious ? 1 : 0,
});
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const currentSince = this.sinceDateForDays(days);
const currentUntilExclusive = new Date(this.startOfUtcDay(new Date()));
@@ -1076,6 +1116,7 @@ export class UserStatisticsService {
async getCompletionLatency(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserCompletionLatencyDistribution> {
const days = query.days ?? COMPLETION_LATENCY_DEFAULT_DAYS;
const key = this.buildUserCacheKey('completion-latency', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const values = await this.repo.getCompletionLatencyDays(user.id, user.isSuperuser, query.libraryIds, days);
const sorted = [...values].sort((a, b) => a - b);
@@ -1111,6 +1152,7 @@ export class UserStatisticsService {
async getReadingSurvival(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserReadingSurvivalPoint[]> {
const days = query.days ?? READING_SURVIVAL_DEFAULT_DAYS;
const key = this.buildUserCacheKey('reading-survival', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const values = await this.repo.getReadingSurvivalMaxProgress(user.id, user.isSuperuser, query.libraryIds, days);
const total = values.length;
@@ -1129,6 +1171,7 @@ export class UserStatisticsService {
async getCompletionRace(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserCompletionRaceBook[]> {
const days = query.days ?? COMPLETION_RACE_DEFAULT_DAYS;
const key = this.buildUserCacheKey('completion-race', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getCompletionRaceRawSessions(user.id, user.isSuperuser, query.libraryIds, days);
const byBook = new Map<number, { title: string; sessions: { startedAt: Date; endProgress: number }[] }>();
@@ -1161,12 +1204,14 @@ export class UserStatisticsService {
async getSessionArchetypes(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserSessionArchetypePoint[]> {
const days = query.days ?? SESSION_ARCHETYPES_DEFAULT_DAYS;
const key = this.buildUserCacheKey('session-archetypes', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, () => this.repo.getSessionArchetypePoints(user.id, user.isSuperuser, query.libraryIds, days));
}
async getGenreReadingTime(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserGenreReadingTimeItem[]> {
const days = query.days ?? GENRE_READING_TIME_DEFAULT_DAYS;
const key = this.buildUserCacheKey('genre-reading-time', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, async () => {
const rows = await this.repo.getGenreReadingTime(user.id, user.isSuperuser, query.libraryIds, days);
const byGenre = new Map<string, { readingSeconds: number; bySource: Record<ReadingSessionSourceBucket, number> }>();
@@ -1189,12 +1234,14 @@ export class UserStatisticsService {
async getReadingPace(user: RequestUser, query: UserDailyReadingQueryDto): Promise<UserReadingPacePoint[]> {
const days = query.days ?? READING_PACE_DEFAULT_DAYS;
const key = this.buildUserCacheKey('reading-pace', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, () => this.repo.getReadingPacePoints(user.id, user.isSuperuser, query.libraryIds, days));
}
async getAuthorGenreChord(user: RequestUser, query: UserDailyReadingQueryDto): Promise<ChordDiagramData> {
const days = query.days ?? 1825;
const key = this.buildUserCacheKey('author-genre-chord', user, { libraries: this.normalizeLibraryIds(query.libraryIds), days });
this.readingAttemptEvents.flushPendingChanges(user.id);
return this.cache.get(String(user.id), key, () => this.repo.getAuthorGenreChord(user.id, user.isSuperuser, query.libraryIds, days));
}
server/test/helpers/reading-attempt-repository.fake.ts
+116
View File
@@ -0,0 +1,116 @@
import { vi } from 'vitest';
import type { ReadStatus } from '@bookorbit/types';
import { readingAttempts } from '../../src/db/schema';
type Row = typeof readingAttempts.$inferSelect;
export function makeFakeRepo() {
const rows: Row[] = [];
let nextId = 1;
const transactionContext = {};
const statuses = new Map<string, { status: ReadStatus; source: string; startedAt: Date | null; finishedAt: Date | null }>();
const projections: Array<{
status: ReadStatus;
source: 'manual' | 'auto';
startedAt: Date | null;
finishedAt: Date | null;
}> = [];
function assertValidAttempt(candidate: Pick<Row, 'userId' | 'bookId' | 'startedOn' | 'endedOn' | 'outcome' | 'deletedAt'>, excludedId?: number) {
if (candidate.endedOn !== null && candidate.outcome === null) {
throw new Error('reading_attempts_closed_has_outcome_chk');
}
if (candidate.startedOn !== null && candidate.endedOn !== null && candidate.endedOn < candidate.startedOn) {
throw new Error('reading_attempts_end_after_start_chk');
}
if (candidate.outcome === null && candidate.deletedAt === null) {
const duplicateActive = rows.some(
(row) =>
row.id !== excludedId &&
row.userId === candidate.userId &&
row.bookId === candidate.bookId &&
row.outcome === null &&
row.deletedAt === null,
);
if (duplicateActive) throw new Error('reading_attempts_one_active_uidx');
}
}
const repo = {
transaction: vi.fn((callback: (tx: object) => Promise<unknown>) => callback(transactionContext)),
findActive: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve(rows.find((row) => row.userId === userId && row.bookId === bookId && row.outcome === null && row.deletedAt === null)),
),
findLatest: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve([...rows].reverse().find((row) => row.userId === userId && row.bookId === bookId && row.deletedAt === null)),
),
hasCompleted: vi.fn((_tx: object, userId: number, bookId: number) =>
Promise.resolve(rows.some((row) => row.userId === userId && row.bookId === bookId && row.outcome === 'completed' && row.deletedAt === null)),
),
create: vi.fn(
(
_tx: object,
values: Omit<Row, 'id' | 'deletedAt' | 'createdAt' | 'updatedAt' | 'externalProvider' | 'externalId'> & {
externalProvider?: string | null;
externalId?: string | null;
},
) => {
const now = new Date('2026-07-12T12:00:00.000Z');
const row: Row = {
...values,
id: nextId++,
externalProvider: values.externalProvider ?? null,
externalId: values.externalId ?? null,
deletedAt: null,
createdAt: now,
updatedAt: now,
};
assertValidAttempt(row);
rows.push(row);
return Promise.resolve(row);
},
),
createActive: vi.fn(
(
_tx: object,
values: Omit<Row, 'id' | 'deletedAt' | 'createdAt' | 'updatedAt' | 'externalProvider' | 'externalId' | 'endedOn' | 'outcome'> & {
externalProvider?: string | null;
externalId?: string | null;
},
) => {
const existing = rows.find(
(row) => row.userId === values.userId && row.bookId === values.bookId && row.outcome === null && row.deletedAt === null,
);
if (existing) return Promise.resolve(existing);
return repo.create(_tx, { ...values, endedOn: null, outcome: null });
},
),
update: vi.fn((_tx: object, userId: number, bookId: number, id: number, patch: Partial<Row>) => {
const row = rows.find((item) => item.id === id && item.userId === userId && item.bookId === bookId && item.deletedAt === null);
if (!row) return Promise.resolve(null);
assertValidAttempt({ ...row, ...patch }, row.id);
Object.assign(row, patch, { updatedAt: new Date('2026-07-12T12:00:00.000Z') });
return Promise.resolve(row);
}),
project: vi.fn((_tx: object, userId: number, bookId: number, projection: (typeof projections)[number]) => {
projections.push(projection);
statuses.set(`${userId}:${bookId}`, projection);
return Promise.resolve();
}),
findStatus: vi.fn((_tx: object, userId: number, bookId: number) => Promise.resolve(statuses.get(`${userId}:${bookId}`) ?? null)),
findByExternal: vi.fn((_tx: object, userId: number, provider: string, externalId: string) =>
Promise.resolve(rows.find((row) => row.userId === userId && row.externalProvider === provider && row.externalId === externalId)),
),
findOwned: vi.fn((userId: number, bookId: number, id: number) =>
Promise.resolve(rows.find((row) => row.id === id && row.userId === userId && row.bookId === bookId && row.deletedAt === null)),
),
softDelete: vi.fn((userId: number, bookId: number, id: number) => {
const row = rows.find((item) => item.id === id && item.userId === userId && item.bookId === bookId && item.deletedAt === null);
if (!row) return Promise.resolve(false);
row.deletedAt = new Date();
return Promise.resolve(true);
}),
list: vi.fn(() => Promise.resolve({ items: [], total: 0 })),
};
return { repo, rows, projections, transactionContext };
}
server/test/reading-attempt-cache.integration.test.ts
+569
View File
@@ -0,0 +1,569 @@
import { BadRequestException, NotFoundException } from '@nestjs/common';
import { Test, type TestingModule } from '@nestjs/testing';
import { EMPTY_CONTENT_FILTER_RULES } from '@bookorbit/types';
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest';
import { ReadingAttemptService } from '../src/modules/user-book-status/reading-attempt.service';
import { ReadingAttemptRepository } from '../src/modules/user-book-status/reading-attempt.repository';
import { READING_ATTEMPT_CHANGED, ReadingAttemptEventsService } from '../src/modules/user-book-status/reading-attempt-events.service';
import { DashboardWidgetService } from '../src/modules/dashboard/dashboard-widget.service';
import { DashboardWidgetRepository } from '../src/modules/dashboard/dashboard-widget.repository';
import { LibraryService } from '../src/modules/library/library.service';
import type { RequestUser } from '../src/common/types/request-user';
import { UserBookStatusService } from '../src/modules/user-book-status/user-book-status.service';
import { UserBookStatusRepository } from '../src/modules/user-book-status/user-book-status.repository';
import { KoboStatusProjectionRepository } from '../src/modules/user-book-status/kobo-status-projection.repository';
import { AchievementEventsService } from '../src/modules/achievement/achievement-events.service';
import { UserStatisticsService } from '../src/modules/user-statistics/user-statistics.service';
import { UserStatisticsRepository } from '../src/modules/user-statistics/user-statistics.repository';
import { makeFakeRepo } from './helpers/reading-attempt-repository.fake';
describe('Reading attempt cache integration', () => {
let module: TestingModule;
let fake: ReturnType<typeof makeFakeRepo>;
let service: ReadingAttemptService;
let dashboard: DashboardWidgetService;
let events: ReadingAttemptEventsService;
let statistics: UserStatisticsService;
let listener: ReturnType<typeof vi.fn>;
let countCompletedBooks: Mock<DashboardWidgetRepository['countCompletedBooks']>;
const user = { id: 1, settings: { timezone: 'UTC' }, contentFilters: EMPTY_CONTENT_FILTER_RULES } as RequestUser;
beforeEach(async () => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-10-06T12:00:00Z'));
fake = makeFakeRepo();
countCompletedBooks = vi.fn<DashboardWidgetRepository['countCompletedBooks']>(
(userId: number, _libraries: number[], fromDay: string, untilDay: string) =>
Promise.resolve(
fake.rows.filter(
(row) =>
row.userId === userId &&
row.deletedAt === null &&
row.outcome === 'completed' &&
row.endedOn !== null &&
row.endedOn >= fromDay &&
row.endedOn < untilDay,
).length,
),
);
module = await Test.createTestingModule({
providers: [
ReadingAttemptService,
UserBookStatusService,
{
provide: UserBookStatusRepository,
useValue: { findOne: vi.fn().mockResolvedValue(null), findUserTimeZone: vi.fn().mockResolvedValue('UTC') },
},
{ provide: AchievementEventsService, useValue: { emit: vi.fn() } },
{ provide: KoboStatusProjectionRepository, useValue: { isEnabled: vi.fn().mockResolvedValue(false) } },
ReadingAttemptEventsService,
DashboardWidgetService,
UserStatisticsService,
{
provide: UserStatisticsRepository,
useValue: {
resolveActivityLibraryIds: vi.fn().mockResolvedValue([1, 2]),
getActivitySessionPage: vi.fn().mockResolvedValue([]),
getActivityActiveDays: vi.fn().mockResolvedValue([]),
getActivityAvailableYears: vi.fn().mockResolvedValue([2025, 2026]),
getActivityCompletionTimeline: vi.fn((userId: number) =>
Promise.resolve(
[2025, 2026].map((year) => ({
year,
month: 1,
count: fake.rows.filter(
(row) => row.userId === userId && row.deletedAt === null && row.outcome === 'completed' && row.endedOn?.startsWith(String(year)),
).length,
})),
),
),
getProgressFunnelInRange: vi.fn().mockResolvedValue({ started: 0, reached25: 0, reached50: 0, reached75: 0, completed: 0 }),
getActivityPaceSummary: vi
.fn()
.mockResolvedValue({ overall: { eligibleSessions: 0, medianDurationSeconds: null, medianProgressDelta: null }, byMedia: [] }),
},
},
{ provide: ReadingAttemptRepository, useValue: fake.repo },
{ provide: DashboardWidgetRepository, useValue: { countCompletedBooks, getCurrentlyReadingBooks: vi.fn().mockResolvedValue({ books: [] }) } },
{ provide: LibraryService, useValue: { findAccessibleLibraryIds: vi.fn().mockResolvedValue([1, 2]) } },
],
}).compile();
await module.init();
service = module.get(ReadingAttemptService);
dashboard = module.get(DashboardWidgetService);
statistics = module.get(UserStatisticsService);
events = module.get(ReadingAttemptEventsService);
listener = vi.fn();
events.on(READING_ATTEMPT_CHANGED, listener);
});
afterEach(async () => {
await module?.close();
vi.useRealTimers();
});
async function expectCount(count: number, reader = user) {
expect(await dashboard.getReadingGoal(reader)).toMatchObject({ completedBooks: count, year: 2026 });
}
it('updates the cached goal immediately when a completion moves across years', async () => {
await expectCount(0);
const attempt = await service.createHistorical(1, 10, { endedOn: '2025-12-15', outcome: 'completed' });
await expectCount(0);
await service.update(1, 10, attempt.id, { endedOn: '2026-01-15' });
await expectCount(1);
await service.update(1, 10, attempt.id, { endedOn: '2025-12-15' });
await expectCount(0);
expect(countCompletedBooks).toHaveBeenCalledTimes(4);
});
it('refreshes after date-only edits when the book stays read', async () => {
await service.applyManualStatus(1, 10, 'read', undefined, '2026-01-15', '2026-10-06');
await expectCount(1);
await service.applyManualStatus(1, 10, 'read', undefined, '2025-12-15', '2026-10-06', { statusWasExplicit: false });
await expectCount(0);
});
it('refreshes after clearing and restoring a completion date', async () => {
const attempt = await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await expectCount(1);
await service.update(1, 10, attempt.id, { endedOn: null });
await expectCount(0);
await service.update(1, 10, attempt.id, { endedOn: '2026-02-01' });
await expectCount(1);
});
it('refreshes after changing an outcome and deleting a completion', async () => {
const attempt = await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await expectCount(1);
await service.update(1, 10, attempt.id, { outcome: 'skimmed' });
await expectCount(0);
await service.update(1, 10, attempt.id, { outcome: 'completed' });
await expectCount(1);
await service.delete(1, 10, attempt.id);
await expectCount(0);
});
it.each(['bookorbit', 'kobo', 'koreader'] as const)('refreshes after an automatic %s completion', async (origin) => {
await service.applyManualStatus(1, 10, 'reading', undefined, undefined, '2026-10-06');
await expectCount(0);
await service.recordActivity({
userId: 1,
bookId: 10,
occurredOn: '2026-10-06',
origin,
progress: 100,
finishThreshold: 98,
strongRereadEvidence: false,
meaningfulActivity: true,
});
await expectCount(1);
});
it('refreshes after importing an external completion or filling its missing date', async () => {
await expectCount(0);
await service.importExternalRead(1, 10, { provider: 'hardcover', externalId: 'dated', startedOn: null, endedOn: '2026-01-15' });
await expectCount(1);
await service.importExternalRead(1, 20, { provider: 'hardcover', externalId: 'undated', startedOn: null, endedOn: null });
await expectCount(1);
await service.importExternalRead(1, 20, { provider: 'hardcover', externalId: 'undated', startedOn: null, endedOn: '2026-02-01' });
await expectCount(2);
});
it('invalidates every library scope for the changed reader without evicting other readers', async () => {
const selected = { ...user, settings: { dashboardConfig: { libraryIds: [1] } } } as RequestUser;
const other = { ...user, id: 11 };
await expectCount(0);
await expectCount(0, selected);
await expectCount(0, other);
await dashboard.getCurrentlyReading(user);
await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await expectCount(1);
await expectCount(1, selected);
await expectCount(0, other);
await dashboard.getCurrentlyReading(user);
expect(countCompletedBooks).toHaveBeenCalledTimes(5);
expect(module.get(DashboardWidgetRepository).getCurrentlyReadingBooks).toHaveBeenCalledTimes(2);
});
it('does not repopulate the cache from a query started before the mutation committed', async () => {
let resolveOld!: (count: number) => void;
countCompletedBooks.mockImplementationOnce(
() =>
new Promise<number>((resolve) => {
resolveOld = resolve;
}),
);
const pending = dashboard.getReadingGoal(user);
await vi.waitFor(() => expect(countCompletedBooks).toHaveBeenCalledOnce());
await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await expectCount(1);
resolveOld(0);
expect((await pending).completedBooks).toBe(0);
await expectCount(1);
expect(countCompletedBooks).toHaveBeenCalledTimes(2);
});
it('emits only after the transaction commits', async () => {
const commit = vi.fn();
fake.repo.transaction.mockImplementation(async (callback) => {
const result = await callback(fake.transactionContext);
expect(listener).not.toHaveBeenCalled();
commit();
return result;
});
await service.applyManualStatus(1, 10, 'read', undefined, '2026-01-15', '2026-10-06');
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
expect(commit.mock.invocationCallOrder[0]).toBeLessThan(listener.mock.invocationCallOrder[0]);
});
it('keeps cached results on rejected mutations', async () => {
await expectCount(0);
await expect(service.createHistorical(1, 10, { startedOn: '2026-02-01', endedOn: '2026-01-15', outcome: 'completed' })).rejects.toBeInstanceOf(
BadRequestException,
);
await expect(service.update(1, 10, 999, { endedOn: '2026-01-15' })).rejects.toBeInstanceOf(NotFoundException);
await expect(service.delete(1, 10, 999)).rejects.toBeInstanceOf(NotFoundException);
fake.repo.transaction.mockRejectedValueOnce(new Error('Transaction rolled back'));
await expect(service.applyManualStatus(1, 10, 'read', undefined, '2026-01-15', '2026-10-06')).rejects.toThrow('Transaction rolled back');
await expectCount(0);
expect(listener).not.toHaveBeenCalled();
expect(countCompletedBooks).toHaveBeenCalledOnce();
});
it('invalidates committed history even when rebuilding its status projection fails', async () => {
await expectCount(0);
fake.repo.project.mockRejectedValueOnce(new Error('Projection failed'));
await expect(service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' })).rejects.toThrow('Projection failed');
await expectCount(1);
});
it.each(['bookorbit', 'kobo', 'koreader'] as const)('refreshes currently-reading immediately on a projection-only %s resume', async (origin) => {
await service.applyManualStatus(1, 10, 'on_hold', undefined, undefined, '2026-10-06');
const getBooks = module.get(DashboardWidgetRepository).getCurrentlyReadingBooks;
vi.mocked(getBooks).mockImplementation(async () => {
const status = await fake.repo.findStatus(fake.transactionContext, 1, 10);
return {
books:
status?.status === 'reading'
? [
{
bookId: 10,
title: 'Book',
authors: [],
progress: 25,
hasCover: false,
fileId: null,
fileFormat: null,
readFileId: null,
readFileFormat: null,
readAlongFileId: null,
hasAudio: false,
},
]
: [],
};
});
expect((await dashboard.getCurrentlyReading(user)).books).toEqual([]);
await expectCount(0);
listener.mockClear();
fake.repo.create.mockClear();
fake.repo.update.mockClear();
await service.recordActivity({
userId: 1,
bookId: 10,
occurredOn: '2026-10-06',
origin,
progress: 25,
finishThreshold: 98,
strongRereadEvidence: false,
meaningfulActivity: true,
});
expect(fake.repo.create).not.toHaveBeenCalled();
expect(fake.repo.update).not.toHaveBeenCalled();
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
expect((await dashboard.getCurrentlyReading(user)).books).toEqual([expect.objectContaining({ bookId: 10 })]);
await expectCount(0);
expect(getBooks).toHaveBeenCalledTimes(2);
});
it('does not notify a projection-only resume when its transaction rolls back', async () => {
await service.applyManualStatus(1, 10, 'on_hold', undefined, undefined, '2026-10-06');
await dashboard.getCurrentlyReading(user);
listener.mockClear();
fake.repo.project.mockRejectedValueOnce(new Error('Projection failed'));
await expect(
service.recordActivity({
userId: 1,
bookId: 10,
occurredOn: '2026-10-06',
origin: 'kobo',
progress: 25,
finishThreshold: 98,
strongRereadEvidence: false,
meaningfulActivity: true,
}),
).rejects.toThrow('Projection failed');
expect(listener).not.toHaveBeenCalled();
await dashboard.getCurrentlyReading(user);
expect(module.get(DashboardWidgetRepository).getCurrentlyReadingBooks).toHaveBeenCalledOnce();
});
it('flushes cache notifications with explicit owner ids on both cache hits and misses', async () => {
const flush = vi.spyOn(events, 'flushPendingChanges');
await expectCount(0);
await expectCount(0);
await dashboard.getCurrentlyReading(user);
await dashboard.getCurrentlyReading(user);
await statistics.getActivityOverview({ ...user, id: 11 }, {});
await statistics.getActivityOverview({ ...user, id: 11 }, {});
expect(flush.mock.calls.slice(0, 4)).toEqual([[1], [1], [1], [1]]);
expect(flush.mock.calls.slice(4).every(([id]) => id === 11)).toBe(true);
expect(flush).toHaveBeenCalledWith(11);
});
it('preserves the cache on automatic activity that changes neither attempts nor projection', async () => {
await service.applyManualStatus(1, 10, 'reading', undefined, undefined, '2026-10-06');
const activity = {
userId: 1,
bookId: 10,
occurredOn: '2026-10-06',
origin: 'bookorbit' as const,
progress: 25,
finishThreshold: 98,
strongRereadEvidence: false,
meaningfulActivity: true,
};
await service.recordActivity(activity);
await expectCount(0);
listener.mockClear();
await service.recordActivity(activity);
await expectCount(0);
expect(listener).not.toHaveBeenCalled();
expect(countCompletedBooks).toHaveBeenCalledOnce();
});
it('preserves the cache when an external import makes no changes or would resurrect a deletion', async () => {
const input = { provider: 'hardcover' as const, externalId: 'external-read', startedOn: null, endedOn: '2026-01-15' };
await service.importExternalRead(1, 10, input);
await expectCount(1);
listener.mockClear();
await service.importExternalRead(1, 10, input);
await expectCount(1);
expect(listener).not.toHaveBeenCalled();
expect(fake.repo.update).not.toHaveBeenCalled();
await service.delete(1, 10, fake.rows[0].id);
await expectCount(0);
listener.mockClear();
await service.importExternalRead(1, 10, input);
await expectCount(0);
expect(listener).not.toHaveBeenCalled();
});
it('refreshes statistics completion totals and goal charts immediately across years and users', async () => {
const other = { ...user, id: 11 };
const overview = (reader = user) => statistics.getActivityOverview(reader, { libraryIds: [1] });
expect((await overview()).goal.completedBooks).toBe(0);
expect((await overview(other)).goal.completedBooks).toBe(0);
await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
const current = await overview();
expect(current.goal.completedBooks).toBe(1);
expect(current.goal.points[0].actualCumulative).toBe(1);
await service.update(1, 10, fake.rows[0].id, { endedOn: '2025-12-15' });
expect((await overview()).goal.completedBooks).toBe(0);
expect((await overview(other)).goal.completedBooks).toBe(0);
const repo = module.get(UserStatisticsRepository);
expect(repo.getActivityCompletionTimeline).toHaveBeenCalledTimes(4);
});
it('makes a committed batch mutation visible to dashboard and statistics reads before the batch finishes', async () => {
await expectCount(0);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(0);
await service.coalesceChanges(async () => {
await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
expect(listener).not.toHaveBeenCalled();
await expectCount(1);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(1);
await service.update(1, 10, fake.rows[0].id, { endedOn: '2025-12-15' });
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(0);
await expectCount(0);
});
});
it('preserves caches for no-op manual and historical updates while retaining timestamp writes', async () => {
await service.applyManualStatus(1, 10, 'read', undefined, '2026-01-15', '2026-10-06');
await expectCount(1);
listener.mockClear();
await service.applyManualStatus(1, 10, 'read', undefined, undefined, '2026-10-06');
await service.applyManualStatus(1, 10, 'read', undefined, '2026-01-15', '2026-10-06');
await service.update(1, 10, fake.rows[0].id, {});
await service.update(1, 10, fake.rows[0].id, { endedOn: '2026-01-15' });
await expectCount(1);
expect(listener).not.toHaveBeenCalled();
expect(countCompletedBooks).toHaveBeenCalledOnce();
expect(fake.repo.update).toHaveBeenCalled();
});
it('keeps invalidating both subscribers when another listener throws after a committed mutation', async () => {
const { Logger } = await import('@nestjs/common');
const warn = vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => {});
events.prependListener(READING_ATTEMPT_CHANGED, () => {
throw new Error('Listener failed');
});
await expectCount(0);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(0);
await expect(service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' })).resolves.toMatchObject({ endedOn: '2026-01-15' });
await expectCount(1);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(1);
fake.repo.project.mockRejectedValueOnce(new Error('Projection failed'));
await expect(service.update(1, 10, fake.rows[0].id, { endedOn: '2025-12-15' })).rejects.toThrow('Projection failed');
await expectCount(0);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(0);
warn.mockRestore();
});
it('refreshes every reader after shared book or library history is deleted', async () => {
const other = { ...user, id: 11 };
await service.createHistorical(1, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await service.createHistorical(11, 10, { endedOn: '2026-01-15', outcome: 'completed' });
await expectCount(1);
await expectCount(1, other);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(1);
expect((await statistics.getActivityOverview(other, {})).goal.completedBooks).toBe(1);
fake.rows.splice(0);
events.notifyChanged(null);
await expectCount(0);
await expectCount(0, other);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(0);
expect((await statistics.getActivityOverview(other, {})).goal.completedBooks).toBe(0);
});
it('coalesces large bulk status changes across the whole operation and skips unchanged reruns', async () => {
const statuses = module.get(UserBookStatusService);
await expectCount(0);
await statuses.bulkSetManual(
1,
Array.from({ length: 123 }, (_, index) => index + 1),
'read',
);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
await expectCount(123);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(123);
listener.mockClear();
await statuses.bulkSetManual(
1,
Array.from({ length: 123 }, (_, index) => index + 1),
'read',
);
expect(listener).not.toHaveBeenCalled();
await expectCount(123);
expect(countCompletedBooks).toHaveBeenCalledTimes(2);
});
it('serves fresh reads between bounded write batches and flushes later commits at operation end', async () => {
const statuses = module.get(UserBookStatusService);
await expectCount(0);
let release!: () => void;
let entered!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const started = new Promise<void>((resolve) => {
entered = resolve;
});
const create = fake.repo.create.getMockImplementation()!;
fake.repo.create.mockImplementation(async (tx, values) => {
if (values.bookId === 51) {
entered();
await gate;
}
return create(tx, values);
});
const operation = statuses.bulkSetManual(
1,
Array.from({ length: 51 }, (_, index) => index + 1),
'read',
);
await started;
expect(listener).not.toHaveBeenCalled();
await expectCount(50);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(50);
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
release();
await operation;
expect(listener).toHaveBeenCalledTimes(2);
await expectCount(51);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(51);
});
it('flushes earlier committed batches when a later write batch fails', async () => {
const statuses = module.get(UserBookStatusService);
await expectCount(0);
const create = fake.repo.create.getMockImplementation()!;
fake.repo.create.mockImplementation((tx, values) => {
if (values.bookId === 51) return Promise.reject(new Error('Later batch failed'));
return create(tx, values);
});
await expect(
statuses.bulkSetManual(
1,
Array.from({ length: 51 }, (_, index) => index + 1),
'read',
),
).rejects.toThrow('Later batch failed');
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
await expectCount(50);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(50);
});
it('waits for successful siblings before flushing a failed bulk batch', async () => {
const statuses = module.get(UserBookStatusService);
await expectCount(0);
let release!: () => void;
let started!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const entered = new Promise<void>((resolve) => {
started = resolve;
});
const create = fake.repo.create.getMockImplementation()!;
fake.repo.create.mockImplementation(async (tx, values) => {
if (values.bookId === 1) throw new Error('Write failed');
started();
await gate;
return create(tx, values);
});
const outcome = statuses.bulkSetManual(1, [1, 2], 'read').catch((error: unknown) => error);
await entered;
expect(listener).not.toHaveBeenCalled();
release();
expect(await outcome).toMatchObject({ message: 'Write failed' });
expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
await expectCount(1);
expect((await statistics.getActivityOverview(user, {})).goal.completedBooks).toBe(1);
});
it('removes the statistics subscription without removing other subscribers', () => {
statistics.onModuleDestroy();
const invalidate = vi.spyOn(statistics, 'invalidateUser');
const dashboardClear = vi.spyOn(dashboard, 'clearCacheForUser');
events.notifyChanged(1);
expect(invalidate).not.toHaveBeenCalled();
expect(dashboardClear).toHaveBeenCalledExactlyOnceWith(1);
});
it('removes its listener when the dashboard module shuts down', () => {
const unrelated = vi.fn();
events.on(READING_ATTEMPT_CHANGED, unrelated);
dashboard.onModuleDestroy();
const clear = vi.spyOn(dashboard, 'clearCacheForUser');
events.notifyChanged(1);
expect(clear).not.toHaveBeenCalled();
expect(unrelated).toHaveBeenCalledExactlyOnceWith({ userId: 1 });
});
});
server/test/reading-attempts.e2e-spec.ts
+89
View File
@@ -970,4 +970,93 @@ describe('Reading attempts main-flow simulation (docker e2e)', { timeout: TIMEOU
const monthsLastYear = overview.completion.months.filter((month) => month.year === thisYear - 1);
expect(monthsLastYear.reduce((sum, month) => sum + month.count, 0)).toBe(1);
});
it('refreshes cached dashboard and statistics counts immediately after reading history and status date edits', async () => {
const book = await createBook();
let dashboardBaseline: number | undefined;
let statisticsBaseline: number | undefined;
async function readingGoal() {
const response = await ctx.app.inject({
method: 'GET',
url: '/api/v1/dashboard/widgets/reading-goal',
headers: auth(ctx.adminToken),
});
expect(response.statusCode).toBe(200);
const goal = response.json() as { completedBooks: number; year: number };
const statistics = await ctx.app.inject({ method: 'GET', url: '/api/v1/user-statistics/activity-overview', headers: auth(ctx.adminToken) });
expect(statistics.statusCode).toBe(200);
const completed = statistics.json().goal.completedBooks as number;
dashboardBaseline ??= goal.completedBooks;
statisticsBaseline ??= completed;
expect(completed - statisticsBaseline).toBe(goal.completedBooks - dashboardBaseline);
return goal;
}
const baseline = await readingGoal();
const thisYearEnd = `${baseline.year}-01-15`;
const lastYearEnd = `${baseline.year - 1}-12-15`;
const created = await ctx.app.inject({
method: 'POST',
url: `/api/v1/books/${book.bookId}/reading-attempts`,
headers: auth(ctx.adminToken),
payload: { endedOn: thisYearEnd, outcome: 'completed' },
});
expect(created.statusCode).toBe(201);
const attempt = created.json() as { id: number };
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks + 1);
const backdated = await ctx.app.inject({
method: 'PATCH',
url: `/api/v1/books/${book.bookId}/reading-attempts/${attempt.id}`,
headers: auth(ctx.adminToken),
payload: { endedOn: lastYearEnd },
});
expect(backdated.statusCode).toBe(200);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks);
expect((await patchStatus(book.bookId, { finishedAt: thisYearEnd })).statusCode).toBe(200);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks + 1);
expect((await patchStatus(book.bookId, { finishedAt: null })).statusCode).toBe(200);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks);
expect((await patchStatus(book.bookId, { finishedAt: thisYearEnd })).statusCode).toBe(200);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks + 1);
const deleted = await ctx.app.inject({
method: 'DELETE',
url: `/api/v1/books/${book.bookId}/reading-attempts/${attempt.id}`,
headers: auth(ctx.adminToken),
});
expect(deleted.statusCode).toBe(204);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks);
expect((await patchStatus(book.bookId, { status: 'read', finishedAt: thisYearEnd })).statusCode).toBe(200);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks + 1);
const reset = await ctx.app.inject({ method: 'POST', url: `/api/v1/books/${book.bookId}/reset-reading-state`, headers: auth(ctx.adminToken) });
expect(reset.statusCode).toBe(201);
expect((await readingGoal()).completedBooks).toBe(baseline.completedBooks);
});
it('refreshes currently-reading after automatic activity resumes a manually paused book', async () => {
const book = await createBook();
await ctx.db.insert(schema.bookMetadata).values({ bookId: book.bookId, title: 'Paused book' });
expect((await patchStatus(book.bookId, { status: 'on_hold', startedAt: '2026-01-01' })).statusCode).toBe(200);
const current = async () => {
const response = await ctx.app.inject({ method: 'GET', url: '/api/v1/dashboard/widgets/currently-reading', headers: auth(ctx.adminToken) });
expect(response.statusCode).toBe(200);
return response.json().books as Array<{ bookId: number }>;
};
expect((await current()).some((row) => row.bookId === book.bookId)).toBe(false);
const before = await listAttempts(book.bookId);
await ctx.app.get(ReadingAttemptService).recordActivity({
userId: adminUserId,
bookId: book.bookId,
occurredOn: '2026-01-02',
origin: 'kobo',
progress: 25,
finishThreshold: 98,
strongRereadEvidence: false,
meaningfulActivity: true,
});
const after = await listAttempts(book.bookId);
expect(after).toEqual(before);
expect((await current()).some((row) => row.bookId === book.bookId)).toBe(true);
});
});