From e6ea5bd77e7e55ca233b96f25755ce2666dcdf12 Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Wed, 7 Oct 2026 01:28:10 +0000 Subject: [PATCH] fix(stats): invalidate reading caches after committed changes 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 --- server/src/common/cache/stats-cache.test.ts | 50 ++ server/src/common/cache/stats-cache.ts | 7 +- server/src/modules/book/book.service.test.ts | 21 +- server/src/modules/book/book.service.ts | 3 + .../dashboard-widget.service.test.ts | 26 +- .../dashboard/dashboard-widget.service.ts | 48 +- .../src/modules/dashboard/dashboard.module.ts | 3 +- .../hardcover-import.service.test.ts | 43 +- .../hardcover/hardcover-import.service.ts | 118 ++-- server/src/modules/library/library.module.ts | 2 + .../modules/library/library.service.test.ts | 18 + server/src/modules/library/library.service.ts | 3 + .../modules/maintenance/maintenance.module.ts | 3 +- .../missing-resources.service.test.ts | 35 +- .../maintenance/missing-resources.service.ts | 14 +- .../migration-executor.service.test.ts | 11 +- .../executor/user-state.importer.test.ts | 60 +- .../migration/executor/user-state.importer.ts | 3 + .../src/modules/migration/migration.module.ts | 2 + .../reading-state/reading-state.module.ts | 3 +- .../reading-state.service.test.ts | 37 +- .../reading-state/reading-state.service.ts | 3 + .../reading-attempt-events.module.ts | 8 + .../reading-attempt-events.service.test.ts | 135 +++++ .../reading-attempt-events.service.ts | 71 +++ .../reading-attempt.repository.ts | 5 +- .../reading-attempt.service.test.ts | 132 +--- .../reading-attempt.service.ts | 142 +++-- .../user-book-status.module.ts | 5 +- .../user-book-status.service.test.ts | 2 + .../user-book-status.service.ts | 25 +- .../user-statistics/user-statistics.module.ts | 3 +- .../user-statistics.service.test.ts | 69 +-- .../user-statistics.service.ts | 55 +- .../reading-attempt-repository.fake.ts | 116 ++++ .../reading-attempt-cache.integration.test.ts | 569 ++++++++++++++++++ server/test/reading-attempts.e2e-spec.ts | 89 +++ 37 files changed, 1628 insertions(+), 311 deletions(-) create mode 100644 server/src/modules/user-book-status/reading-attempt-events.module.ts create mode 100644 server/src/modules/user-book-status/reading-attempt-events.service.test.ts create mode 100644 server/src/modules/user-book-status/reading-attempt-events.service.ts create mode 100644 server/test/helpers/reading-attempt-repository.fake.ts create mode 100644 server/test/reading-attempt-cache.integration.test.ts diff --git a/server/src/common/cache/stats-cache.test.ts b/server/src/common/cache/stats-cache.test.ts index 8b07cbaaa..b990358b4 100644 --- a/server/src/common/cache/stats-cache.test.ts +++ b/server/src/common/cache/stats-cache.test.ts @@ -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((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((resolve) => { + finishOld = resolve; + }), + ); + cache.clearForScope('1'); + const fresh = vi.fn( + () => + new Promise((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'); + }); +}); diff --git a/server/src/common/cache/stats-cache.ts b/server/src/common/cache/stats-cache.ts index d5d823aa7..272be2683 100644 --- a/server/src/common/cache/stats-cache.ts +++ b/server/src/common/cache/stats-cache.ts @@ -10,6 +10,7 @@ interface CacheEntry { } export class StatsCache { + private generation = 0; private readonly entries = new Map(); private readonly inFlight = new Map>(); private readonly scopeIndex = new Map>(); @@ -36,11 +37,12 @@ export class StatsCache { return inFlight as Promise; } + 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); @@ -89,6 +91,7 @@ export class StatsCache { } clear(): void { + this.generation += 1; this.entries.clear(); this.inFlight.clear(); this.scopeIndex.clear(); diff --git a/server/src/modules/book/book.service.test.ts b/server/src/modules/book/book.service.test.ts index ceadfe32f..7e0ea98db 100644 --- a/server/src/modules/book/book.service.test.ts +++ b/server/src/modules/book/book.service.test.ts @@ -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) => 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(); }); diff --git a/server/src/modules/book/book.service.ts b/server/src/modules/book/book.service.ts index 3e1e6a98a..e21db3637 100644 --- a/server/src/modules/book/book.service.ts +++ b/server/src/modules/book/book.service.ts @@ -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('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. diff --git a/server/src/modules/dashboard/dashboard-widget.service.test.ts b/server/src/modules/dashboard/dashboard-widget.service.test.ts index e58d73bbe..916ca9be4 100644 --- a/server/src/modules/dashboard/dashboard-widget.service.test.ts +++ b/server/src/modules/dashboard/dashboard-widget.service.test.ts @@ -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 { 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 }); + }); +}); diff --git a/server/src/modules/dashboard/dashboard-widget.service.ts b/server/src/modules/dashboard/dashboard-widget.service.ts index 8bb742e3e..78211a3d0 100644 --- a/server/src/modules/dashboard/dashboard-widget.service.ts +++ b/server/src/modules/dashboard/dashboard-widget.service.ts @@ -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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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)); diff --git a/server/src/modules/dashboard/dashboard.module.ts b/server/src/modules/dashboard/dashboard.module.ts index 4511c5b29..d49dab86c 100644 --- a/server/src/modules/dashboard/dashboard.module.ts +++ b/server/src/modules/dashboard/dashboard.module.ts @@ -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], diff --git a/server/src/modules/hardcover/hardcover-import.service.test.ts b/server/src/modules/hardcover/hardcover-import.service.test.ts index 43d2f38f6..4b4c81041 100644 --- a/server/src/modules/hardcover/hardcover-import.service.test.ts +++ b/server/src/modules/hardcover/hardcover-import.service.test.ts @@ -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) => 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); + }); }); diff --git a/server/src/modules/hardcover/hardcover-import.service.ts b/server/src/modules/hardcover/hardcover-import.service.ts index 2b319fa71..1391fa794 100644 --- a/server/src/modules/hardcover/hardcover-import.service.ts +++ b/server/src/modules/hardcover/hardcover-import.service.ts @@ -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, diff --git a/server/src/modules/library/library.module.ts b/server/src/modules/library/library.module.ts index 21e72329f..96e0c15e6 100644 --- a/server/src/modules/library/library.module.ts +++ b/server/src/modules/library/library.module.ts @@ -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), diff --git a/server/src/modules/library/library.service.test.ts b/server/src/modules/library/library.service.test.ts index a979615e8..1a18e343e 100644 --- a/server/src/modules/library/library.service.test.ts +++ b/server/src/modules/library/library.service.test.ts @@ -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([]); diff --git a/server/src/modules/library/library.service.ts b/server/src/modules/library/library.service.ts index cf8358134..5c5977f90 100644 --- a/server/src/modules/library/library.service.ts +++ b/server/src/modules/library/library.service.ts @@ -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('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( diff --git a/server/src/modules/maintenance/maintenance.module.ts b/server/src/modules/maintenance/maintenance.module.ts index b485c13a1..fe192f6c8 100644 --- a/server/src/modules/maintenance/maintenance.module.ts +++ b/server/src/modules/maintenance/maintenance.module.ts @@ -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], }) diff --git a/server/src/modules/maintenance/missing-resources.service.test.ts b/server/src/modules/maintenance/missing-resources.service.test.ts index 79247975f..134ff1a24 100644 --- a/server/src/modules/maintenance/missing-resources.service.test.ts +++ b/server/src/modules/maintenance/missing-resources.service.test.ts @@ -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 { @@ -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'); diff --git a/server/src/modules/maintenance/missing-resources.service.ts b/server/src/modules/maintenance/missing-resources.service.ts index 6e9491744..0fa8be08a 100644 --- a/server/src/modules/maintenance/missing-resources.service.ts +++ b/server/src/modules/maintenance/missing-resources.service.ts @@ -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('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`, diff --git a/server/src/modules/migration/executor/migration-executor.service.test.ts b/server/src/modules/migration/executor/migration-executor.service.test.ts index 545a15a76..5f71d6bab 100644 --- a/server/src/modules/migration/executor/migration-executor.service.test.ts +++ b/server/src/modules/migration/executor/migration-executor.service.test.ts @@ -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: { diff --git a/server/src/modules/migration/executor/user-state.importer.test.ts b/server/src/modules/migration/executor/user-state.importer.test.ts index b42fd2ed4..9348792e2 100644 --- a/server/src/modules/migration/executor/user-state.importer.test.ts +++ b/server/src/modules/migration/executor/user-state.importer.test.ts @@ -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(); + }); +}); diff --git a/server/src/modules/migration/executor/user-state.importer.ts b/server/src/modules/migration/executor/user-state.importer.ts index d1002f971..f9be1f1ab 100644 --- a/server/src/modules/migration/executor/user-state.importer.ts +++ b/server/src/modules/migration/executor/user-state.importer.ts @@ -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 { @@ -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); } diff --git a/server/src/modules/migration/migration.module.ts b/server/src/modules/migration/migration.module.ts index 35979c94f..7cd52d0df 100644 --- a/server/src/modules/migration/migration.module.ts +++ b/server/src/modules/migration/migration.module.ts @@ -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, diff --git a/server/src/modules/reading-state/reading-state.module.ts b/server/src/modules/reading-state/reading-state.module.ts index f2e17d7e7..159559493 100644 --- a/server/src/modules/reading-state/reading-state.module.ts +++ b/server/src/modules/reading-state/reading-state.module.ts @@ -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], }) diff --git a/server/src/modules/reading-state/reading-state.service.test.ts b/server/src/modules/reading-state/reading-state.service.test.ts index 570bc18e9..8d9b7fbcf 100644 --- a/server/src/modules/reading-state/reading-state.service.test.ts +++ b/server/src/modules/reading-state/reading-state.service.test.ts @@ -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((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(); + }); }); diff --git a/server/src/modules/reading-state/reading-state.service.ts b/server/src/modules/reading-state/reading-state.service.ts index 552394050..68e92315c 100644 --- a/server/src/modules/reading-state/reading-state.service.ts +++ b/server/src/modules/reading-state/reading-state.service.ts @@ -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 { @@ -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`, ); diff --git a/server/src/modules/user-book-status/reading-attempt-events.module.ts b/server/src/modules/user-book-status/reading-attempt-events.module.ts new file mode 100644 index 000000000..532ac2f04 --- /dev/null +++ b/server/src/modules/user-book-status/reading-attempt-events.module.ts @@ -0,0 +1,8 @@ +import { Module } from '@nestjs/common'; +import { ReadingAttemptEventsService } from './reading-attempt-events.service'; + +@Module({ + providers: [ReadingAttemptEventsService], + exports: [ReadingAttemptEventsService], +}) +export class ReadingAttemptEventsModule {} diff --git a/server/src/modules/user-book-status/reading-attempt-events.service.test.ts b/server/src/modules/user-book-status/reading-attempt-events.service.test.ts new file mode 100644 index 000000000..e1acac092 --- /dev/null +++ b/server/src/modules/user-book-status/reading-attempt-events.service.test.ts @@ -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((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +describe('ReadingAttemptEventsService', () => { + let events: ReadingAttemptEventsService; + let listener: ReturnType; + + 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; + await events.coalesceChanges(() => { + late = gate.promise.then(() => events.notifyChanged(1)); + }); + gate.resolve(); + await late; + expect(listener).toHaveBeenCalledExactlyOnceWith({ userId: 1 }); + }); +}); diff --git a/server/src/modules/user-book-status/reading-attempt-events.service.ts b/server/src/modules/user-book-status/reading-attempt-events.service.ts new file mode 100644 index 000000000..23ab7d697 --- /dev/null +++ b/server/src/modules/user-book-status/reading-attempt-events.service.ts @@ -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; 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(); + private readonly batches = new Set(); + + 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(operation: () => T | Promise): Promise { + 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`, + ); + } + } + } +} diff --git a/server/src/modules/user-book-status/reading-attempt.repository.ts b/server/src/modules/user-book-status/reading-attempt.repository.ts index 056b6446c..fae565ce4 100644 --- a/server/src/modules/user-book-status/reading-attempt.repository.ts +++ b/server/src/modules/user-book-status/reading-attempt.repository.ts @@ -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; } diff --git a/server/src/modules/user-book-status/reading-attempt.service.test.ts b/server/src/modules/user-book-status/reading-attempt.service.test.ts index 8ada28e65..504411711 100644 --- a/server/src/modules/user-book-status/reading-attempt.service.test.ts +++ b/server/src/modules/user-book-status/reading-attempt.service.test.ts @@ -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, 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) => 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 & { - 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 & { - 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) => { - 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; @@ -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 () => { diff --git a/server/src/modules/user-book-status/reading-attempt.service.ts b/server/src/modules/user-book-status/reading-attempt.service.ts index 9b901c819..abe75da72 100644 --- a/server/src/modules/user-book-status/reading-attempt.service.ts +++ b/server/src/modules/user-book-status/reading-attempt.service.ts @@ -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(operation: () => T | Promise): Promise { + return this.events.coalesceChanges(operation); + } async applyManualStatus( userId: number, @@ -62,7 +70,8 @@ export class ReadingAttemptService { today: string, options: ManualStatusOptions = {}, ): Promise { - return this.repo.transaction(async (tx) => { + let changed = false; + const result = await this.repo.transaction(async (tx) => { let active: Awaited> | 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[4], ): Promise>>> => { + 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 { - return this.repo.transaction(async (tx) => { + let changed = false; + const result = await this.repo.transaction(async (tx) => { let active: Awaited> | 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 { @@ -360,15 +391,16 @@ export class ReadingAttemptService { input: { provider: 'hardcover'; externalId: string; startedOn: string | null; endedOn: string | null }, ): Promise { 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 { - 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 { - 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 { + 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>, 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 { diff --git a/server/src/modules/user-book-status/user-book-status.module.ts b/server/src/modules/user-book-status/user-book-status.module.ts index 11e54fab9..4e3ecb239 100644 --- a/server/src/modules/user-book-status/user-book-status.module.ts +++ b/server/src/modules/user-book-status/user-book-status.module.ts @@ -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 {} diff --git a/server/src/modules/user-book-status/user-book-status.service.test.ts b/server/src/modules/user-book-status/user-book-status.service.test.ts index 6f75a5a7e..3537888bb 100644 --- a/server/src/modules/user-book-status/user-book-status.service.test.ts +++ b/server/src/modules/user-book-status/user-book-status.service.test.ts @@ -665,6 +665,7 @@ describe('kobo status projection', () => { describe('with reading attempts enabled', () => { const mockAttempts = { applyManualStatus: vi.fn(), + coalesceChanges: vi.fn((operation: () => Promise) => 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) => operation()), }; let attemptService: UserBookStatusService; diff --git a/server/src/modules/user-book-status/user-book-status.service.ts b/server/src/modules/user-book-status/user-book-status.service.ts index 8a0f0f3a6..216030804 100644 --- a/server/src/modules/user-book-status/user-book-status.service.ts +++ b/server/src/modules/user-book-status/user-book-status.service.ts @@ -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; } diff --git a/server/src/modules/user-statistics/user-statistics.module.ts b/server/src/modules/user-statistics/user-statistics.module.ts index 27ce83f99..50a84866e 100644 --- a/server/src/modules/user-statistics/user-statistics.module.ts +++ b/server/src/modules/user-statistics/user-statistics.module.ts @@ -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], diff --git a/server/src/modules/user-statistics/user-statistics.service.test.ts b/server/src/modules/user-statistics/user-statistics.service.test.ts index 3795f7b75..3aadbd6d7 100644 --- a/server/src/modules/user-statistics/user-statistics.service.test.ts +++ b/server/src/modules/user-statistics/user-statistics.service.test.ts @@ -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] }; diff --git a/server/src/modules/user-statistics/user-statistics.service.ts b/server/src/modules/user-statistics/user-statistics.service.ts index 50055f358..f42d63108 100644 --- a/server/src/modules/user-statistics/user-statistics.service.ts +++ b/server/src/modules/user-statistics/user-statistics.service.ts @@ -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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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 { 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(); @@ -1161,12 +1204,14 @@ export class UserStatisticsService { async getSessionArchetypes(user: RequestUser, query: UserDailyReadingQueryDto): Promise { 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 { 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 }>(); @@ -1189,12 +1234,14 @@ export class UserStatisticsService { async getReadingPace(user: RequestUser, query: UserDailyReadingQueryDto): Promise { 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 { 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)); } diff --git a/server/test/helpers/reading-attempt-repository.fake.ts b/server/test/helpers/reading-attempt-repository.fake.ts new file mode 100644 index 000000000..b778a272b --- /dev/null +++ b/server/test/helpers/reading-attempt-repository.fake.ts @@ -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(); + const projections: Array<{ + status: ReadStatus; + source: 'manual' | 'auto'; + startedAt: Date | null; + finishedAt: Date | null; + }> = []; + + function assertValidAttempt(candidate: Pick, 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) => 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 & { + 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 & { + 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) => { + 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 }; +} diff --git a/server/test/reading-attempt-cache.integration.test.ts b/server/test/reading-attempt-cache.integration.test.ts new file mode 100644 index 000000000..b9c69d350 --- /dev/null +++ b/server/test/reading-attempt-cache.integration.test.ts @@ -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; + let service: ReadingAttemptService; + let dashboard: DashboardWidgetService; + let events: ReadingAttemptEventsService; + let statistics: UserStatisticsService; + let listener: ReturnType; + let countCompletedBooks: Mock; + 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( + (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((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((resolve) => { + release = resolve; + }); + const started = new Promise((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((resolve) => { + release = resolve; + }); + const entered = new Promise((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 }); + }); +}); diff --git a/server/test/reading-attempts.e2e-spec.ts b/server/test/reading-attempts.e2e-spec.ts index 8d11a7e18..b525450c9 100644 --- a/server/test/reading-attempts.e2e-spec.ts +++ b/server/test/reading-attempts.e2e-spec.ts @@ -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); + }); });