diff --git a/server/src/common/utils/reading-daily-stats.utils.test.ts b/server/src/common/utils/reading-daily-stats.utils.test.ts index dfef3ba8e..aa8b644c0 100644 --- a/server/src/common/utils/reading-daily-stats.utils.test.ts +++ b/server/src/common/utils/reading-daily-stats.utils.test.ts @@ -4,7 +4,10 @@ import { aggregateReadingSessionDailyStats, getDayRangeForDateKeys, getReadingSessionDayKeys, + mergeReadingDailyStatsSegments, + sortReadingDailyStatsSegments, splitReadingSessionByDay, + type ReadingDailyStatsSegment, } from './reading-daily-stats.utils'; describe('reading daily stats utils', () => { @@ -145,6 +148,46 @@ describe('reading daily stats utils', () => { expect(result).toEqual([{ day: '2026-04-16', readingSeconds: 5400, progressDelta: 2, sessionsCount: 2 }]); }); + it('accumulates batched segments into the same totals as one pass over every session', () => { + const sessions = [ + { startedAt: new Date('2026-04-15T08:00:00.000Z'), endedAt: new Date('2026-04-15T09:00:00.000Z'), durationSeconds: 3600, progressDelta: 3 }, + { startedAt: new Date('2026-04-15T20:00:00.000Z'), endedAt: new Date('2026-04-15T20:30:00.000Z'), durationSeconds: 1800, progressDelta: 1 }, + { startedAt: new Date('2026-04-16T07:00:00.000Z'), endedAt: new Date('2026-04-16T07:15:00.000Z'), durationSeconds: 900, progressDelta: null }, + ]; + + const byDay = new Map(); + for (const session of sessions) { + mergeReadingDailyStatsSegments(byDay, aggregateReadingSessionDailyStats([session], 'UTC')); + } + + expect(sortReadingDailyStatsSegments(byDay)).toEqual(aggregateReadingSessionDailyStats(sessions, 'UTC')); + }); + + it('merges without mutating the segments it was handed', () => { + const incoming: ReadingDailyStatsSegment[] = [{ day: '2026-04-15', readingSeconds: 60, progressDelta: 1, sessionsCount: 1 }]; + const byDay = new Map(); + + mergeReadingDailyStatsSegments(byDay, incoming); + mergeReadingDailyStatsSegments(byDay, [{ day: '2026-04-15', readingSeconds: 40, progressDelta: 2, sessionsCount: 1 }]); + + expect(byDay.get('2026-04-15')).toEqual({ day: '2026-04-15', readingSeconds: 100, progressDelta: 3, sessionsCount: 2 }); + expect(incoming[0]).toEqual({ day: '2026-04-15', readingSeconds: 60, progressDelta: 1, sessionsCount: 1 }); + }); + + it('drops merged days outside the requested set and sorts what is left by day', () => { + const byDay = mergeReadingDailyStatsSegments( + new Map(), + [ + { day: '2026-04-16', readingSeconds: 30, progressDelta: 0, sessionsCount: 1 }, + { day: '2026-04-15', readingSeconds: 60, progressDelta: 0, sessionsCount: 1 }, + { day: '2026-04-14', readingSeconds: 90, progressDelta: 0, sessionsCount: 1 }, + ], + new Set(['2026-04-15', '2026-04-16']), + ); + + expect(sortReadingDailyStatsSegments(byDay).map((segment) => segment.day)).toEqual(['2026-04-15', '2026-04-16']); + }); + it('builds an exclusive UTC range for timezone-local affected days', () => { const range = getDayRangeForDateKeys(['2026-04-15', '2026-04-16'], 'America/New_York'); diff --git a/server/src/common/utils/reading-daily-stats.utils.ts b/server/src/common/utils/reading-daily-stats.utils.ts index d79ad218e..08453f576 100644 --- a/server/src/common/utils/reading-daily-stats.utils.ts +++ b/server/src/common/utils/reading-daily-stats.utils.ts @@ -98,6 +98,34 @@ export function splitReadingSessionByDay(session: ReadingSessionDailyStatsInput, .filter((segment) => segment.readingSeconds > 0); } +/** + * Folds day segments into a running total. Split out from the aggregate below because a + * rebuild that pages through a user's whole history has to accumulate across batches, which + * a single call over one in-memory array cannot express. + */ +export function mergeReadingDailyStatsSegments( + into: Map, + segments: readonly ReadingDailyStatsSegment[], + days?: Set, +): Map { + for (const segment of segments) { + if (days && !days.has(segment.day)) continue; + const existing = into.get(segment.day); + if (existing) { + existing.readingSeconds += segment.readingSeconds; + existing.progressDelta += segment.progressDelta; + existing.sessionsCount += segment.sessionsCount; + } else { + into.set(segment.day, { ...segment }); + } + } + return into; +} + +export function sortReadingDailyStatsSegments(byDay: Map): ReadingDailyStatsSegment[] { + return [...byDay.values()].sort((a, b) => a.day.localeCompare(b.day)); +} + export function aggregateReadingSessionDailyStats( sessions: ReadingSessionDailyStatsInput[], timeZone: string, @@ -105,17 +133,7 @@ export function aggregateReadingSessionDailyStats( ): ReadingDailyStatsSegment[] { const byDay = new Map(); for (const session of sessions) { - for (const segment of splitReadingSessionByDay(session, timeZone)) { - if (days && !days.has(segment.day)) continue; - const existing = byDay.get(segment.day); - if (existing) { - existing.readingSeconds += segment.readingSeconds; - existing.progressDelta += segment.progressDelta; - existing.sessionsCount += segment.sessionsCount; - } else { - byDay.set(segment.day, { ...segment }); - } - } + mergeReadingDailyStatsSegments(byDay, splitReadingSessionByDay(session, timeZone), days); } - return [...byDay.values()].sort((a, b) => a.day.localeCompare(b.day)); + return sortReadingDailyStatsSegments(byDay); } diff --git a/server/src/common/utils/sync-estimate-session.utils.test.ts b/server/src/common/utils/sync-estimate-session.utils.test.ts new file mode 100644 index 000000000..d798f5f88 --- /dev/null +++ b/server/src/common/utils/sync-estimate-session.utils.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from 'vitest'; + +import { syncEstimateSessionId, syncEstimateSessionIdPrefix } from './sync-estimate-session.utils'; + +describe('sync estimate session ids', () => { + it('fits the session_id column, whatever the device calls itself', () => { + const longDeviceId = 'd'.repeat(100); + + // varchar(64) on reading_sessions.session_id, and a device id may be up to 100 characters. + expect(syncEstimateSessionId(longDeviceId, 1_000_000, 1_782_870_480_000).length).toBeLessThanOrEqual(64); + expect(syncEstimateSessionId('device-12', 44, 1_782_870_480_000)).toMatch(/^ks-[0-9a-f]{12}-[0-9a-f]{32}$/); + }); + + it('resolves one interval to one id, so a repeated push is dropped rather than counted again', () => { + expect(syncEstimateSessionId('device-12', 44, 1_782_870_480_000)).toBe(syncEstimateSessionId('device-12', 44, 1_782_870_480_000)); + }); + + it('separates devices, books, and intervals', () => { + const base = syncEstimateSessionId('device-12', 44, 1_782_870_480_000); + + expect(syncEstimateSessionId('device-99', 44, 1_782_870_480_000)).not.toBe(base); + expect(syncEstimateSessionId('device-12', 45, 1_782_870_480_000)).not.toBe(base); + expect(syncEstimateSessionId('device-12', 44, 1_782_870_540_000)).not.toBe(base); + }); + + it('shares the device half across every estimate that device makes, so they can be retired together', () => { + const prefix = syncEstimateSessionIdPrefix('device-12'); + + expect(syncEstimateSessionId('device-12', 44, 1_782_870_480_000).startsWith(prefix)).toBe(true); + expect(syncEstimateSessionId('device-12', 77, 1_782_999_999_000).startsWith(prefix)).toBe(true); + expect(syncEstimateSessionId('device-99', 44, 1_782_870_480_000).startsWith(prefix)).toBe(false); + }); +}); diff --git a/server/src/common/utils/sync-estimate-session.utils.ts b/server/src/common/utils/sync-estimate-session.utils.ts new file mode 100644 index 000000000..078f83f60 --- /dev/null +++ b/server/src/common/utils/sync-estimate-session.utils.ts @@ -0,0 +1,31 @@ +import { createHash } from 'node:crypto'; + +/** + * Identifiers for reading sessions estimated from sync-protocol pushes rather than measured + * from a device's own page timings. + * + * A reading session carries no device column, so the id is the only place the device that + * produced an estimate survives. That matters once the same device starts reporting real page + * timings: it has to be able to find the estimates it is superseding, and a prefix on the id is + * what makes that a bounded, indexed lookup instead of a scan. + */ +const PREFIX = 'ks'; +const DEVICE_TAG_LENGTH = 12; + +// codeql[js/weak-cryptographic-algorithm] - identifier derivation, not security +function digest(value: string, length: number): string { + return createHash('md5').update(value).digest('hex').slice(0, length); +} + +/** Everything an estimate from one device shares, as a literal `LIKE` prefix. */ +export function syncEstimateSessionIdPrefix(deviceId: string): string { + return `${PREFIX}-${digest(deviceId, DEVICE_TAG_LENGTH)}-`; +} + +/** + * Derived rather than random so that two pushes racing for the same file resolve the same + * interval to the same id, and the second is dropped instead of counted again. + */ +export function syncEstimateSessionId(deviceId: string, bookFileId: number, intervalStartMs: number): string { + return `${syncEstimateSessionIdPrefix(deviceId)}${digest(`${bookFileId}:${intervalStartMs}`, 32)}`; +} diff --git a/server/src/modules/koreader/koreader-plugin.repository.test.ts b/server/src/modules/koreader/koreader-plugin.repository.test.ts index c15051c05..17a810d05 100644 --- a/server/src/modules/koreader/koreader-plugin.repository.test.ts +++ b/server/src/modules/koreader/koreader-plugin.repository.test.ts @@ -186,4 +186,17 @@ describe('KoreaderPluginRepository', () => { await expect(repo.listDevicePluginVersions(7)).resolves.toEqual([]); }); }); + describe('hasSweepSince', () => { + it('reports a device that has swept inside the window', async () => { + db.select.mockReturnValue(makeQueryChain([{ deviceId: 'device-12' }])); + + await expect(repo.hasSweepSince(7, 'device-12', new Date('2026-07-26T00:00:00.000Z'))).resolves.toBe(true); + }); + + it('reports a device whose last sweep predates the window, or that never swept at all', async () => { + db.select.mockReturnValue(makeQueryChain([])); + + await expect(repo.hasSweepSince(7, 'device-12', new Date('2026-07-26T00:00:00.000Z'))).resolves.toBe(false); + }); + }); }); diff --git a/server/src/modules/koreader/koreader-plugin.repository.ts b/server/src/modules/koreader/koreader-plugin.repository.ts index e60bb2059..0c8206445 100644 --- a/server/src/modules/koreader/koreader-plugin.repository.ts +++ b/server/src/modules/koreader/koreader-plugin.repository.ts @@ -452,6 +452,28 @@ export class KoreaderPluginRepository { return lastSweepAt; } + /** + * Whether this device has swept since the given moment, which is what separates a device + * currently reporting its own page timings from one that only speaks the sync protocol. + * + * Bounded rather than "ever": a plugin that is removed, or stops working, would otherwise + * leave the device permanently unable to have its reading recorded at all. + */ + async hasSweepSince(userId: number, deviceId: string, since: Date): Promise { + const [row] = await this.db + .select({ deviceId: schema.koreaderDeviceSweeps.deviceId }) + .from(schema.koreaderDeviceSweeps) + .where( + and( + eq(schema.koreaderDeviceSweeps.userId, userId), + eq(schema.koreaderDeviceSweeps.deviceId, deviceId), + gte(schema.koreaderDeviceSweeps.lastSweepAt, since), + ), + ) + .limit(1); + return row != null; + } + async listSweeps(userId: number) { return this.db .select() diff --git a/server/src/modules/koreader/koreader-plugin.service.test.ts b/server/src/modules/koreader/koreader-plugin.service.test.ts index 1739477b4..eb1654270 100644 --- a/server/src/modules/koreader/koreader-plugin.service.test.ts +++ b/server/src/modules/koreader/koreader-plugin.service.test.ts @@ -10,6 +10,9 @@ import type { KoreaderPluginRepository } from './koreader-plugin.repository'; import { KoreaderPluginService } from './koreader-plugin.service'; import type { KoreaderRepository } from './koreader.repository'; import type { KoreaderService } from './koreader.service'; +import { syncEstimateSessionIdPrefix } from '../../common/utils/sync-estimate-session.utils'; +import { buildDeviceSessionIdPrefix } from './koreader-stats.util'; +import type { ReadingSessionService } from '../reading-session/reading-session.service'; const DEVICE_ID = 'abcdef12-3456-7890-abcd-ef1234567890'; const HASH_A = 'a'.repeat(32); @@ -52,6 +55,7 @@ describe('KoreaderPluginService', () => { normalizeNote: (value: string | null | undefined) => string | null; }; let achievementEvents: { emit: ReturnType }; + let readingSessions: { discardSupersededSyncEstimates: ReturnType }; let service: KoreaderPluginService; beforeEach(() => { @@ -90,6 +94,7 @@ describe('KoreaderPluginService', () => { }, }; achievementEvents = { emit: vi.fn() }; + readingSessions = { discardSupersededSyncEstimates: vi.fn().mockResolvedValue({ deleted: 0 }) }; service = new KoreaderPluginService( koreaderRepo as unknown as KoreaderRepository, @@ -98,6 +103,7 @@ describe('KoreaderPluginService', () => { userBookStatusService as unknown as UserBookStatusService, userBookNoteService as unknown as UserBookNoteService, achievementEvents as unknown as AchievementEventsService, + readingSessions as unknown as ReadingSessionService, ); }); @@ -526,5 +532,37 @@ describe('KoreaderPluginService', () => { }); expect(result).toEqual({ ok: true, lastSweepAt: '2026-06-09T10:00:00.000Z', libraryVersion: expect.stringMatching(/^[0-9a-f]{16}$/) }); }); + + it('retires only the estimates a measured session actually overlaps', async () => { + // A sweep is not proof the measured history covers the estimated one: a device may upload + // no page stats at all, or only the window its local statistics database still holds. + const dto = { ...deviceFields(), booksMatched: 3, pageStatsUploaded: 0, annotationsUpserted: 4 } as SweepCompleteDto; + + await service.sweepComplete(makeUser(), dto); + + expect(readingSessions.discardSupersededSyncEstimates).toHaveBeenCalledWith(expect.objectContaining({ id: 7 }), { + estimateSessionIdPrefix: syncEstimateSessionIdPrefix(DEVICE_ID), + measuredSessionIdPrefix: buildDeviceSessionIdPrefix(DEVICE_ID), + }); + }); + + it('runs on every sweep, so a sweep whose cleanup failed is retried by the next one', async () => { + const dto = { ...deviceFields(), booksMatched: 3, pageStatsUploaded: 120, annotationsUpserted: 4 } as SweepCompleteDto; + readingSessions.discardSupersededSyncEstimates.mockRejectedValueOnce(new Error('deadlock detected')); + + await service.sweepComplete(makeUser(), dto); + await service.sweepComplete(makeUser(), dto); + + expect(readingSessions.discardSupersededSyncEstimates).toHaveBeenCalledTimes(2); + }); + + it('still records the sweep when retiring the estimates fails', async () => { + readingSessions.discardSupersededSyncEstimates.mockRejectedValue(new Error('deadlock detected')); + const dto = { ...deviceFields(), booksMatched: 3, pageStatsUploaded: 120, annotationsUpserted: 4 } as SweepCompleteDto; + + await expect(service.sweepComplete(makeUser(), dto)).resolves.toEqual( + expect.objectContaining({ ok: true, lastSweepAt: '2026-06-09T10:00:00.000Z' }), + ); + }); }); }); diff --git a/server/src/modules/koreader/koreader-plugin.service.ts b/server/src/modules/koreader/koreader-plugin.service.ts index 982f86d6b..c7b8c8a3d 100644 --- a/server/src/modules/koreader/koreader-plugin.service.ts +++ b/server/src/modules/koreader/koreader-plugin.service.ts @@ -5,11 +5,14 @@ import type { ReadStatus, UserBookStatus } from '@bookorbit/types'; import type { RequestUser } from '../../common/types/request-user'; import { mapWithConcurrency } from '../../common/utils/batch.utils'; import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; +import { syncEstimateSessionIdPrefix } from '../../common/utils/sync-estimate-session.utils'; import { ACHIEVEMENT_EVENT_BOOK_RATING_CHANGED, AchievementEventsService } from '../achievement/achievement-events.service'; +import { ReadingSessionService } from '../reading-session/reading-session.service'; import { UserBookNoteService, type UserBookNoteDto } from '../user-book-note/user-book-note.service'; import { UserBookStatusService } from '../user-book-status/user-book-status.service'; import type { BookStateDto, BookStatesUploadDto, BulkProgressDto, MatchCheckDto, SweepCompleteDto } from './dto'; import { KoreaderPluginRepository } from './koreader-plugin.repository'; +import { buildDeviceSessionIdPrefix } from './koreader-stats.util'; import { KoreaderRepository } from './koreader.repository'; import { KoreaderService, type BulkProgressEntry } from './koreader.service'; @@ -95,6 +98,7 @@ export class KoreaderPluginService { private readonly userBookStatusService: UserBookStatusService, private readonly userBookNoteService: UserBookNoteService, private readonly achievementEvents: AchievementEventsService, + private readonly readingSessions: ReadingSessionService, ) {} async matchCheck(user: RequestUser, dto: MatchCheckDto): Promise { @@ -264,6 +268,21 @@ export class KoreaderPluginService { }); await this.koreaderRepo.restoreDevice(user.id, dto.deviceId); + // This device reports KOReader's own page timings, and a sweep backfills them from a zero + // watermark the first time, so reading that had only been estimated from sync pushes now + // arrives measured. Only the estimates a measured session actually overlaps are retired, + // and it runs on every sweep so a failure here is retried rather than left double counted. + await this.readingSessions + .discardSupersededSyncEstimates(user, { + estimateSessionIdPrefix: syncEstimateSessionIdPrefix(dto.deviceId), + measuredSessionIdPrefix: buildDeviceSessionIdPrefix(dto.deviceId), + }) + .catch((error: unknown) => { + this.logger.warn( + `[${SWEEP_EVENT}] [fail] userId=${user.id} deviceId=${dto.deviceId.slice(0, 8)} durationMs=${Date.now() - startedAtMs} errorClass=${error instanceof Error ? error.constructor.name : 'UnknownError'} error="${sanitizeLogValue(error instanceof Error ? error.message : 'unknown error')}" - discarding superseded sync estimates failed`, + ); + }); + const accessibleLibraryIds = await this.koreaderRepo.getAccessibleLibraryIds(user.id); const libraryVersion = await this.computeLibraryVersion(user.id, accessibleLibraryIds); diff --git a/server/src/modules/koreader/koreader-stats.util.ts b/server/src/modules/koreader/koreader-stats.util.ts index f4a19885f..5a42b7ee1 100644 --- a/server/src/modules/koreader/koreader-stats.util.ts +++ b/server/src/modules/koreader/koreader-stats.util.ts @@ -20,8 +20,13 @@ export interface DerivedKoreaderSession { endProgress: number | null; } +/** Everything one device's derived sessions share, across every book it has read. */ +export function buildDeviceSessionIdPrefix(deviceId: string): string { + return `kor:${deviceId.slice(0, 8)}:`; +} + export function buildSessionIdPrefix(deviceId: string, bookFileId: number): string { - return `kor:${deviceId.slice(0, 8)}:${bookFileId}:`; + return `${buildDeviceSessionIdPrefix(deviceId)}${bookFileId}:`; } export function buildSessionId(deviceId: string, bookFileId: number, clusterStartEpoch: number): string { diff --git a/server/src/modules/koreader/koreader.controller.test.ts b/server/src/modules/koreader/koreader.controller.test.ts index 80dc5dac9..8e5b2e611 100644 --- a/server/src/modules/koreader/koreader.controller.test.ts +++ b/server/src/modules/koreader/koreader.controller.test.ts @@ -76,7 +76,7 @@ describe('KoreaderController', () => { const progressDto = { document: 'abc', percentage: 0.5 } as never; await controller.saveProgress(user, progressDto); - expect(koreaderService.saveProgress).toHaveBeenCalledWith(7, progressDto); + expect(koreaderService.saveProgress).toHaveBeenCalledWith(user, progressDto); await expect(controller.getProgress(user, 'abc')).resolves.toEqual({ document: 'abc', percentage: 0.5 }); expect(koreaderService.getProgress).toHaveBeenCalledWith(7, 'abc'); diff --git a/server/src/modules/koreader/koreader.controller.ts b/server/src/modules/koreader/koreader.controller.ts index 82936955f..7852987ad 100644 --- a/server/src/modules/koreader/koreader.controller.ts +++ b/server/src/modules/koreader/koreader.controller.ts @@ -67,7 +67,7 @@ export class KoreaderController { @UseGuards(KoreaderAuthGuard) @Put('syncs/progress') async saveProgress(@CurrentUser() user: RequestUser, @Body() dto: KoreaderSaveProgressDto) { - return this.koreaderService.saveProgress(user.id, dto); + return this.koreaderService.saveProgress(user, dto); } @Public() diff --git a/server/src/modules/koreader/koreader.module.ts b/server/src/modules/koreader/koreader.module.ts index 160506630..c1a26ea72 100644 --- a/server/src/modules/koreader/koreader.module.ts +++ b/server/src/modules/koreader/koreader.module.ts @@ -10,6 +10,7 @@ import { BrowseCountsModule } from '../browse-counts/browse-counts.module'; import { DashboardModule } from '../dashboard/dashboard.module'; import { OpdsModule } from '../opds/opds.module'; import { PositionConverterModule } from '../position-converter/position-converter.module'; +import { ReadingSessionModule } from '../reading-session/reading-session.module'; import { RecommendationModule } from '../recommendation/recommendation.module'; import { UserModule } from '../user/user.module'; import { UserBookNoteModule } from '../user-book-note/user-book-note.module'; @@ -48,6 +49,7 @@ import { KoreaderStatsService } from './koreader-stats.service'; DashboardModule, OpdsModule, PositionConverterModule, + ReadingSessionModule, RecommendationModule, ], controllers: [KoreaderController, KoreaderPluginController, KoreaderCatalogController], diff --git a/server/src/modules/koreader/koreader.service.test.ts b/server/src/modules/koreader/koreader.service.test.ts index 6c26ffb0d..bf3fae40c 100644 --- a/server/src/modules/koreader/koreader.service.test.ts +++ b/server/src/modules/koreader/koreader.service.test.ts @@ -21,9 +21,15 @@ import { KoreaderChapterExtractorService } from './koreader-chapter-extractor.se import { KoreaderChapterService } from './koreader-chapter.service'; import type { KoreaderPackageService } from './koreader-package.service'; import { KoreaderPluginRepository } from './koreader-plugin.repository'; +import { syncEstimateSessionId } from '../../common/utils/sync-estimate-session.utils'; +import type { ReadingSessionService } from '../reading-session/reading-session.service'; import { KoreaderRepository } from './koreader.repository'; import { KoreaderService } from './koreader.service'; +function syncUser(id: number, timezone?: string) { + return { id, settings: timezone ? { timezone } : {} } as never; +} + function md5Hex(value: string): string { return `md5:${value}:hex:0123456789abcdef0123456789abcdef`; } @@ -94,6 +100,7 @@ describe('KoreaderService', () => { emit: ReturnType; }; let mockPluginRepo: { + hasSweepSince: ReturnType; listSweeps: ReturnType; getPluginTotals: ReturnType; }; @@ -107,6 +114,9 @@ describe('KoreaderService', () => { let mockPackageService: { getVersionInfo: ReturnType; }; + let mockReadingSessions: { + recordSyncedSession: ReturnType; + }; beforeEach(() => { vi.restoreAllMocks(); @@ -193,7 +203,12 @@ describe('KoreaderService', () => { getVersionInfo: vi.fn().mockResolvedValue({ pluginVersion: 'unknown', serverVersion: '1.0.0' }), }; + mockReadingSessions = { + recordSyncedSession: vi.fn().mockResolvedValue({ kind: 'saved' }), + }; + mockPluginRepo = { + hasSweepSince: vi.fn().mockResolvedValue(false), listSweeps: vi.fn().mockResolvedValue([]), getPluginTotals: vi.fn().mockResolvedValue({ matchedBooks: 0, @@ -228,6 +243,7 @@ describe('KoreaderService', () => { mockPositionConverter as never, mockBookService as never, mockPackageService as unknown as KoreaderPackageService, + mockReadingSessions as unknown as ReadingSessionService, ); }); @@ -381,7 +397,7 @@ describe('KoreaderService', () => { mockChapterService.parseChapterIndexFromProgress.mockReturnValue(6); mockChapterExtractor.extractAndStoreChapters.mockRejectedValueOnce(new Error('extract failed')); - const result = await service.saveProgress(12, { + const result = await service.saveProgress(syncUser(12), { document: 'abcdef1234567890fedcba', percentage: 0.5, progress: '/body/DocFragment[7]', @@ -439,7 +455,7 @@ describe('KoreaderService', () => { mockRepo.resolveBookFileByHash.mockResolvedValue(null); await expect( - service.saveProgress(12, { + service.saveProgress(syncUser(12), { document: 'missing-document', percentage: 0.2, }), @@ -456,7 +472,7 @@ describe('KoreaderService', () => { mockRepo.resolveBookFileByHash.mockResolvedValue(null); await expect( - service.saveProgress(12, { + service.saveProgress(syncUser(12), { document: 'no-access-document', percentage: 0.2, }), @@ -468,7 +484,7 @@ describe('KoreaderService', () => { it('uses the default device and generated device id when the payload leaves them empty', async () => { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 88, bookId: 99, libraryId: 4, format: 'epub' }); - await service.saveProgress(12, { + await service.saveProgress(syncUser(12), { document: 'default-device-document', percentage: 0.25, device: '', @@ -486,10 +502,169 @@ describe('KoreaderService', () => { }); }); + describe('reading sessions estimated from sync progress', () => { + const NOW = new Date('2026-07-01T02:00:00.000Z'); + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(NOW); + mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 44, bookId: 55, libraryId: 3, format: 'epub' }); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function lastPushedAt(minutesAgo: number) { + return new Date(NOW.getTime() - minutesAgo * 60_000); + } + + async function push(percentage: number, previous: { percentage: number; minutesAgo: number } | null, deviceId = 'device-12') { + mockRepo.getLatestDeviceProgress.mockResolvedValue( + previous ? { percentage: previous.percentage, deviceId, updatedAt: lastPushedAt(previous.minutesAgo) } : null, + ); + await service.saveProgress(syncUser(12, 'America/Halifax'), { + document: 'abcdef1234567890fedcba', + percentage, + progress: '/body/DocFragment[7]', + device: 'Kobo Sage', + device_id: deviceId, + }); + } + + it('records the interval between two advancing pushes as a session', async () => { + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + + expect(mockReadingSessions.recordSyncedSession).toHaveBeenCalledWith( + expect.objectContaining({ + userId: 12, + bookFileId: 44, + source: 'koreader', + durationSeconds: 12 * 60, + startedAt: lastPushedAt(12), + endedAt: NOW, + endProgress: 50, + timeZone: 'America/Halifax', + }), + ); + const call = mockReadingSessions.recordSyncedSession.mock.calls[0]![0] as { progressDelta: number }; + expect(call.progressDelta).toBeCloseTo(8, 5); + }); + + it('caps a long silence rather than recording it as unbroken reading', async () => { + await push(0.5, { percentage: 0.42, minutesAgo: 9 * 60 }); + + expect(mockReadingSessions.recordSyncedSession).toHaveBeenCalledWith( + expect.objectContaining({ durationSeconds: 30 * 60, startedAt: new Date(NOW.getTime() - 30 * 60_000), endedAt: NOW }), + ); + }); + + it('leaves a device that is still sweeping alone, since its page stats already measure the same reading', async () => { + mockPluginRepo.hasSweepSince.mockResolvedValue(true); + + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + + // Thirty days of silence, so a plugin that is removed or breaks does not retire the + // device from every statistic for good. + expect(mockPluginRepo.hasSweepSince).toHaveBeenCalledWith(12, 'device-12', new Date(NOW.getTime() - 30 * 24 * 60 * 60 * 1000)); + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('estimates again once a device has gone quiet long enough to have lost the plugin', async () => { + mockPluginRepo.hasSweepSince.mockResolvedValue(false); + + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + + expect(mockReadingSessions.recordSyncedSession).toHaveBeenCalled(); + }); + + it('ignores a push that did not move the position forward', async () => { + await push(0.42, { percentage: 0.42, minutesAgo: 12 }); + await push(0.3, { percentage: 0.42, minutesAgo: 12 }); + + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('ignores a gap too short to be a sitting', async () => { + await push(0.5, { percentage: 0.42, minutesAgo: 0.5 }); + + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('records nothing on a first sync, where there is no interval to measure', async () => { + await push(0.5, null); + + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('records nothing while the push is held behind a pending reset', async () => { + mockRepo.getProgressReset.mockResolvedValue(new Date('2026-06-30T00:00:00.000Z')); + + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + + expect(mockRepo.upsertDeviceProgress).toHaveBeenCalled(); + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('measures the pushing device rather than whichever device synced last', async () => { + // A second device syncing a stale position in between owns the newest row, so neither its + // clock nor its position may be borrowed: the interval and the distance are both this + // device's own. Scoring against the stale row would claim 45 percent read in five minutes. + mockRepo.getLatestDeviceProgress.mockResolvedValue({ percentage: 0.05, deviceId: 'other-device', updatedAt: lastPushedAt(1) }); + mockRepo.getDeviceProgressForDevice.mockResolvedValue({ percentage: 0.45, deviceId: 'device-12', updatedAt: lastPushedAt(5) }); + + await service.saveProgress(syncUser(12), { + document: 'abcdef1234567890fedcba', + percentage: 0.5, + device: 'Kobo Sage', + device_id: 'device-12', + }); + + expect(mockRepo.getDeviceProgressForDevice).toHaveBeenCalledWith(44, 12, 'device-12'); + const recorded = mockReadingSessions.recordSyncedSession.mock.calls[0]![0] as { durationSeconds: number; progressDelta: number }; + expect(recorded.durationSeconds).toBe(5 * 60); + expect(recorded.progressDelta).toBeCloseTo(5, 5); + }); + + it("ignores a push that only looks forward next to another device's stale position", async () => { + // Own position 50, other device replayed 5, this push repeats 50: no reading happened. + mockRepo.getLatestDeviceProgress.mockResolvedValue({ percentage: 0.05, deviceId: 'other-device', updatedAt: lastPushedAt(1) }); + mockRepo.getDeviceProgressForDevice.mockResolvedValue({ percentage: 0.5, deviceId: 'device-12', updatedAt: lastPushedAt(5) }); + + await service.saveProgress(syncUser(12), { + document: 'abcdef1234567890fedcba', + percentage: 0.5, + device: 'Kobo Sage', + device_id: 'device-12', + }); + + expect(mockReadingSessions.recordSyncedSession).not.toHaveBeenCalled(); + }); + + it('derives one session id per interval, so a repeated push is not counted twice', async () => { + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + await push(0.5, { percentage: 0.42, minutesAgo: 12 }); + + const [first, second] = mockReadingSessions.recordSyncedSession.mock.calls.map((call) => (call[0] as { sessionId: string }).sessionId); + expect(first).toBe(second); + // Derived through the shared helper, whose device half is what a later plugin sweep + // matches on to retire the estimates it supersedes. + expect(first).toBe(syncEstimateSessionId('device-12', 44, lastPushedAt(12).getTime())); + }); + + it('keeps the position write when the session cannot be stored', async () => { + mockReadingSessions.recordSyncedSession.mockRejectedValue(new Error('daily stats deadlock')); + + await expect(push(0.5, { percentage: 0.42, minutesAgo: 12 })).resolves.toBeUndefined(); + + expect(mockRepo.upsertReadingProgress).toHaveBeenCalled(); + }); + }); + describe('shared progress position routing', () => { async function syncFormat(format: string | null, progress?: string) { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 44, bookId: 55, libraryId: 3, format }); - await service.saveProgress(12, { document: 'abcdef1234567890fedcba', percentage: 0.5, progress }); + await service.saveProgress(syncUser(12), { document: 'abcdef1234567890fedcba', percentage: 0.5, progress }); return mockRepo.upsertReadingProgress.mock.calls[0]![0] as { cfi: string | null; pageNumber: number | null; xpointer: string | null }; } @@ -738,7 +913,12 @@ describe('KoreaderService', () => { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 10, bookId: 20, libraryId: 1, format: 'epub' }); mockRepo.getProgressReset.mockResolvedValue(resetAt); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.42, progress: '/body/DocFragment[6]/body', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.42, + progress: '/body/DocFragment[6]/body', + device_id: 'device-1', + }); expect(mockRepo.upsertDeviceProgress).toHaveBeenCalledTimes(1); expect(mockRepo.upsertReadingProgress).not.toHaveBeenCalled(); @@ -749,7 +929,12 @@ describe('KoreaderService', () => { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 10, bookId: 20, libraryId: 1, format: 'epub' }); mockRepo.getProgressReset.mockResolvedValue(resetAt); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.004, progress: '/body/DocFragment[1]/body', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.004, + progress: '/body/DocFragment[1]/body', + device_id: 'device-1', + }); expect(mockRepo.recordResetConvergence).toHaveBeenCalledWith(10, 7, 'device-1'); expect(mockRepo.upsertReadingProgress).toHaveBeenCalledTimes(1); @@ -761,7 +946,7 @@ describe('KoreaderService', () => { // Page 1 of a hundred-page comic reports 1%, which no threshold meaning "the start of a // long book" could accept, and that device would be held forever. - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.01, progress: '1', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { document: 'doc-hash', percentage: 0.01, progress: '1', device_id: 'device-1' }); expect(mockRepo.recordResetConvergence).toHaveBeenCalledWith(10, 7, 'device-1'); }); @@ -770,7 +955,7 @@ describe('KoreaderService', () => { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 10, bookId: 20, libraryId: 1, format: 'cbz' }); mockRepo.getProgressReset.mockResolvedValue(resetAt); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.008, progress: '9', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { document: 'doc-hash', percentage: 0.008, progress: '9', device_id: 'device-1' }); expect(mockRepo.recordResetConvergence).not.toHaveBeenCalled(); expect(mockRepo.upsertReadingProgress).not.toHaveBeenCalled(); @@ -781,7 +966,12 @@ describe('KoreaderService', () => { mockRepo.getProgressReset.mockResolvedValue(resetAt); mockRepo.getConvergedResetDeviceIds.mockResolvedValue(new Set(['device-1'])); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.5, progress: '/body/DocFragment[9]/body', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.5, + progress: '/body/DocFragment[9]/body', + device_id: 'device-1', + }); // The pull is anonymous, so a marker left alive past its purpose is answered to this // device too, and it would be sent back to the start on every sync. @@ -793,7 +983,7 @@ describe('KoreaderService', () => { mockRepo.getProgressReset.mockResolvedValue(resetAt); mockRepo.getConvergedResetDeviceIds.mockResolvedValue(new Set(['device-1'])); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0, progress: '/body/DocFragment[1]/body', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { document: 'doc-hash', percentage: 0, progress: '/body/DocFragment[1]/body', device_id: 'device-1' }); expect(mockRepo.clearProgressReset).not.toHaveBeenCalled(); }); @@ -804,7 +994,12 @@ describe('KoreaderService', () => { // Every position in a one-spine EPUB is inside DocFragment[1], so the fragment alone // would call this device converged wherever it happens to be sitting. - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.5, progress: '/body/DocFragment[1]/body/p[80]', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.5, + progress: '/body/DocFragment[1]/body/p[80]', + device_id: 'device-1', + }); expect(mockRepo.recordResetConvergence).not.toHaveBeenCalled(); expect(mockRepo.upsertReadingProgress).not.toHaveBeenCalled(); @@ -815,7 +1010,12 @@ describe('KoreaderService', () => { mockRepo.getProgressReset.mockResolvedValue(resetAt); mockRepo.getConvergedResetDeviceIds.mockResolvedValue(new Set(['device-1'])); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.42, progress: '/body/DocFragment[6]/body', device_id: 'device-2' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.42, + progress: '/body/DocFragment[6]/body', + device_id: 'device-2', + }); expect(mockRepo.upsertReadingProgress).not.toHaveBeenCalled(); }); @@ -825,7 +1025,12 @@ describe('KoreaderService', () => { mockRepo.getProgressReset.mockResolvedValue(resetAt); mockRepo.getConvergedResetDeviceIds.mockResolvedValue(new Set(['device-1'])); - await service.saveProgress(7, { document: 'doc-hash', percentage: 0.55, progress: '/body/DocFragment[9]/body', device_id: 'device-1' }); + await service.saveProgress(syncUser(7), { + document: 'doc-hash', + percentage: 0.55, + progress: '/body/DocFragment[9]/body', + device_id: 'device-1', + }); expect(mockRepo.upsertReadingProgress).toHaveBeenCalledTimes(1); }); diff --git a/server/src/modules/koreader/koreader.service.ts b/server/src/modules/koreader/koreader.service.ts index 834a04bb0..23cb0e738 100644 --- a/server/src/modules/koreader/koreader.service.ts +++ b/server/src/modules/koreader/koreader.service.ts @@ -4,9 +4,12 @@ import { createHash } from 'crypto'; import { DEFAULT_KOREADER_DEVICE_PATTERN, type KoreaderBookSyncInfo, type KoreaderDeviceInfo, type KoreaderSyncStatus } from '@bookorbit/types'; import { StatsCache } from '../../common/cache/stats-cache'; +import type { RequestUser } from '../../common/types/request-user'; import { mapWithConcurrency } from '../../common/utils/batch.utils'; import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; +import { syncEstimateSessionId } from '../../common/utils/sync-estimate-session.utils'; import { isSemverNewer } from '../../common/utils/semver.utils'; +import { resolveTimeZone } from '../../common/utils/timezone.utils'; import { KoreaderRepository, type DeviceProgressUpsert } from './koreader.repository'; import { KoreaderChapterService } from './koreader-chapter.service'; import { KoreaderChapterExtractorService } from './koreader-chapter-extractor.service'; @@ -16,10 +19,12 @@ import { KoreaderPluginRepository } from './koreader-plugin.repository'; import { isPagedReadingFormat, parseKoreaderPageNumber } from './koreader-progress-position.util'; import { BookService } from '../book/book.service'; import { PositionConverterService } from '../position-converter/position-converter.service'; +import { ReadingSessionService } from '../reading-session/reading-session.service'; import { AchievementEventsService, ACHIEVEMENT_EVENT_BOOK_PROGRESS_CHANGED } from '../achievement/achievement-events.service'; const BCRYPT_ROUNDS = 12; const SYNC_EVENT = 'koreader.sync'; +const SYNC_SESSION_EVENT = 'koreader.sync_session'; const CREDENTIALS_EVENT = 'koreader.credentials'; const DEVICE_REMOVE_EVENT = 'koreader.device_remove'; const DEVICE_RETIRE_EVENT = 'koreader.device_retire'; @@ -42,6 +47,20 @@ const RESET_REFLOWABLE_POSITION = '/body/DocFragment[1]/body'; * would stay held forever. */ const RESET_CONVERGED_PERCENTAGE_FALLBACK = 0.01; +/** + * Bounds on a reading session estimated from the gap between two sync pushes. Below the floor + * the gap is a page turn rather than a sitting; above the ceiling it is mostly idle time, and + * recording it whole would credit a night's sleep as reading. + */ +const MIN_SYNC_SESSION_SECONDS = 60; +const MAX_SYNC_SESSION_SECONDS = 30 * 60; +/** + * How long a device is still treated as reporting its own page timings after its last sweep. + * Bounded rather than permanent: a plugin that is removed, or quietly stops working, would + * otherwise leave that device unable to have its reading recorded ever again. An estimate that + * does turn out to overlap a later sweep's measured session is retired by that sweep. + */ +const PLUGIN_SWEEP_SILENCE_MS = 30 * 24 * 60 * 60 * 1000; /** `format` routes the incoming position to the cfi or the pageNumber column. */ export interface KoreaderProgressBookFile { @@ -51,6 +70,20 @@ export interface KoreaderProgressBookFile { format: string | null; } +/** What one applied sync push changed, for callers that have to reason about the move it made. */ +export interface KoreaderProgressApplyResult { + /** Whether shared reading progress moved; false when the push was held behind a reset. */ + shared: boolean; + /** BookOrbit-scale position held before this push, across every device of this user. */ + previousPercentage: number | null; + /** BookOrbit-scale position this push wrote. */ + percentage: number; + /** When this same device last pushed a position for this file, before this push. */ + deviceLastPushedAt: Date | null; + /** BookOrbit-scale position this same device last reported for this file, before this push. */ + deviceLastPercentage: number | null; +} + export interface BulkProgressEntry { bookFile: KoreaderProgressBookFile; percentage: number; @@ -72,6 +105,7 @@ export class KoreaderService { private readonly positionConverter: PositionConverterService, private readonly bookService: BookService, private readonly packageService: KoreaderPackageService, + private readonly readingSessions: ReadingSessionService, ) {} async createCredentials(userId: number, username: string, password: string) { @@ -147,9 +181,10 @@ export class KoreaderService { } async saveProgress( - userId: number, + user: RequestUser, data: { document: string; percentage: number; progress?: string; device?: string; device_id?: string; timestamp?: number }, ) { + const userId = user.id; const startedAt = Date.now(); const device = data.device || DEFAULT_DEVICE; const deviceId = data.device_id || createHash('md5').update(`${device}:${userId}`).digest('hex').slice(0, 16); // codeql[js/weak-cryptographic-algorithm] - non-security device identifier @@ -166,7 +201,7 @@ export class KoreaderService { throw new NotFoundException('Book not found for the given document hash'); } - await this.applyProgressForResolvedFile(userId, bookFile, { + const applied = await this.applyProgressForResolvedFile(userId, bookFile, { percentage: data.percentage, progress: data.progress, device, @@ -174,6 +209,8 @@ export class KoreaderService { timestamp: data.timestamp, }); + await this.recordSyncedReadingSession(user, bookFile, deviceId, applied); + this.logger.log( `[${SYNC_EVENT}] [end] userId=${userId} bookFileId=${bookFile.id} device=${device} durationMs=${Date.now() - startedAt} percentage=${data.percentage} - save progress completed`, ); @@ -186,10 +223,25 @@ export class KoreaderService { bookFile: KoreaderProgressBookFile, data: { percentage: number; progress?: string; device: string; deviceId: string; timestamp?: number }, options?: { skipSharedProgress?: boolean }, - ) { + ): Promise { const chapterIndex = this.chapterService.parseChapterIndexFromProgress(data.progress ?? null); const previousDeviceProgress = await this.repo.getLatestDeviceProgress(bookFile.id, userId); + // No row at all means no row for this device either, and the newest one is this device's own + // whenever a single device syncs the book, so the second lookup is the multi-device case only. + const ownDeviceProgress = !previousDeviceProgress + ? null + : previousDeviceProgress.deviceId === data.deviceId + ? previousDeviceProgress + : await this.repo.getDeviceProgressForDevice(bookFile.id, userId, data.deviceId); + + const result: KoreaderProgressApplyResult = { + shared: false, + previousPercentage: previousDeviceProgress?.percentage != null ? toBookorbitPercentage(previousDeviceProgress.percentage) : null, + percentage: toBookorbitPercentage(data.percentage), + deviceLastPushedAt: ownDeviceProgress?.updatedAt ?? null, + deviceLastPercentage: ownDeviceProgress?.percentage != null ? toBookorbitPercentage(ownDeviceProgress.percentage) : null, + }; this.chapterExtractor.extractAndStoreChapters(bookFile.id).catch(() => {}); @@ -208,12 +260,90 @@ export class KoreaderService { // visible rather than letting activity accrue against a device the user cannot see. await this.repo.restoreDevice(userId, data.deviceId); - if (options?.skipSharedProgress) return; + if (options?.skipSharedProgress) return result; - if (!(await this.resolveResetHold(userId, bookFile, data))) return; + if (!(await this.resolveResetHold(userId, bookFile, data))) return result; - const previousPercentage = previousDeviceProgress?.percentage != null ? toBookorbitPercentage(previousDeviceProgress.percentage) : null; - await this.applySharedProgress(userId, bookFile, data, previousPercentage); + await this.applySharedProgress(userId, bookFile, data, result.previousPercentage); + result.shared = true; + return result; + } + + /** + * Turns an advancing sync push into a reading session, for devices that can offer nothing + * better. + * + * The kosync protocol carries a position and nothing else: no page timings, no session + * boundaries, no idle detection. So the only evidence of reading time a plain KOReader + * install ever produces is the interval between two of its own pushes, and without this the + * reading log, the streak and every daily statistic show that device as having read nothing + * at all. + * + * The interval is capped rather than trusted. A device that goes quiet overnight and pushes + * again in the morning has not been reading the whole time, and the cap is what keeps a long + * idle gap from being recorded as a long read. + * + * Both the interval and the distance read come from this device's own previous row, so a + * second device syncing the same book in between cannot lend it either one. + * + * A device that has swept recently is excluded. It uploads KOReader's own page timings, which + * produce real sessions for the same reading, and estimating from its pushes as well would + * count that reading twice. The exclusion lapses once the device stops sweeping, so a removed + * or broken plugin does not silently retire the device from every statistic. + */ + private async recordSyncedReadingSession( + user: RequestUser, + bookFile: KoreaderProgressBookFile, + deviceId: string, + applied: KoreaderProgressApplyResult, + ): Promise { + if (!applied.shared || applied.deviceLastPushedAt === null || applied.deviceLastPercentage === null) return; + + // Measured against this device's own last position, not the newest one across the shelf. + // Another device syncing a stale position in between makes it the newest row, and reading + // on from there would be scored as the whole distance between the two devices. + const progressDelta = applied.percentage - applied.deviceLastPercentage; + if (progressDelta <= 0) return; + + const endedAt = new Date(); + const elapsedSeconds = Math.floor((endedAt.getTime() - applied.deviceLastPushedAt.getTime()) / 1000); + if (elapsedSeconds < MIN_SYNC_SESSION_SECONDS) return; + + const durationSeconds = Math.min(elapsedSeconds, MAX_SYNC_SESSION_SECONDS); + const startedAt = new Date(endedAt.getTime() - durationSeconds * 1000); + + const operationStartedAt = Date.now(); + + try { + if (await this.pluginRepo.hasSweepSince(user.id, deviceId, new Date(endedAt.getTime() - PLUGIN_SWEEP_SILENCE_MS))) return; + + const sessionId = syncEstimateSessionId(deviceId, bookFile.id, applied.deviceLastPushedAt.getTime()); + + const result = await this.readingSessions.recordSyncedSession({ + userId: user.id, + bookFileId: bookFile.id, + sessionId, + startedAt, + endedAt, + durationSeconds, + progressDelta, + endProgress: applied.percentage, + source: 'koreader', + timeZone: resolveTimeZone((user.settings as { timezone?: unknown } | undefined)?.timezone, 'UTC'), + }); + + this.logger.log( + `[${SYNC_SESSION_EVENT}] [end] userId=${user.id} bookFileId=${bookFile.id} deviceId="${sanitizeLogValue(deviceId)}" durationMs=${Date.now() - operationStartedAt} durationSeconds=${durationSeconds} elapsedSeconds=${elapsedSeconds} outcome=${result.kind}${result.kind === 'skipped' ? ` reason=${result.reason}` : ''} - reading session estimated from sync progress`, + ); + } catch (error) { + // A device retries the whole push on failure, so a session that cannot be stored must not + // take the position write down with it. + const errorClass = error instanceof Error ? error.constructor.name : 'UnknownError'; + const message = sanitizeLogValue(error instanceof Error ? error.message : 'unknown error'); + this.logger.warn( + `[${SYNC_SESSION_EVENT}] [fail] userId=${user.id} bookFileId=${bookFile.id} deviceId="${sanitizeLogValue(deviceId)}" durationMs=${Date.now() - operationStartedAt} errorClass=${errorClass} error="${message}" - estimating a reading session from sync progress failed`, + ); + } } /** diff --git a/server/src/modules/reading-session/reading-session.repository.test.ts b/server/src/modules/reading-session/reading-session.repository.test.ts index d4733fe69..0e0127129 100644 --- a/server/src/modules/reading-session/reading-session.repository.test.ts +++ b/server/src/modules/reading-session/reading-session.repository.test.ts @@ -696,3 +696,94 @@ describe('ReadingSessionRepository - deleteSessionByBook', () => { expect(transaction).toHaveBeenCalledOnce(); }); }); + +describe('ReadingSessionRepository - deleteSupersededSyncEstimates', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + function makeHarness( + supersededPages: Array< + Array<{ id: number; startedAt: Date; endedAt: Date; durationSeconds: number; progressDelta: number | null; libraryId: number }> + >, + survivingByLibrary: Array> = [], + ) { + const pages = [...supersededPages]; + const remaining = [...survivingByLibrary]; + + // One chain serves both callers: the paged lookup takes .orderBy().limit(), the recompute + // awaits the .where() itself, so the chain is a thenable that also carries .orderBy. + const where = vi.fn(() => ({ + orderBy: vi.fn().mockReturnValue({ limit: vi.fn(() => Promise.resolve(pages.shift() ?? [])) }), + then: (resolve: (value: unknown) => unknown, reject?: (reason: unknown) => unknown) => + Promise.resolve(remaining.shift() ?? []).then(resolve, reject), + })); + const select = vi.fn().mockReturnValue({ from: vi.fn().mockReturnValue({ innerJoin: vi.fn().mockReturnValue({ where }) }) }); + + const deleteWhere = vi.fn().mockResolvedValue(undefined); + const deleteFn = vi.fn().mockReturnValue({ where: deleteWhere }); + const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) }); + const insert = vi.fn((table: unknown) => { + if (table === userReadingDailyStats) return { values: dailyValues }; + throw new Error('Unexpected table in insert'); + }); + + const tx = { execute: vi.fn().mockResolvedValue(undefined), select, delete: deleteFn, insert }; + const transaction = vi.fn(async (cb: (trx: typeof tx) => Promise) => cb(tx)); + const repo = new ReadingSessionRepository({ transaction } as never); + + return { repo, tx, deleteFn, dailyValues, transaction, select }; + } + + const estimate = (id: number, day: string, libraryId = 3) => ({ + id, + startedAt: new Date(`${day}T10:00:00.000Z`), + endedAt: new Date(`${day}T10:30:00.000Z`), + durationSeconds: 1800, + progressDelta: 3, + libraryId, + }); + + it('keeps estimates that no measured session overlaps', async () => { + // A sweep that uploaded nothing, or only part of the history, must not erase the rest. + const { repo, deleteFn, dailyValues } = makeHarness([[]]); + + await expect(repo.deleteSupersededSyncEstimates(7, 'ks-abcdef123456-', 'kor:device01:')).resolves.toEqual({ deleted: 0 }); + expect(deleteFn).not.toHaveBeenCalled(); + expect(dailyValues).not.toHaveBeenCalled(); + }); + + it('deletes the overlapped estimates and rebuilds the days they were counted on', async () => { + const { repo, deleteFn, dailyValues } = makeHarness( + [[estimate(5, '2026-04-15')]], + // What survives that day: the measured session the plugin derived for the same reading. + [[{ startedAt: new Date('2026-04-15T10:02:00.000Z'), endedAt: new Date('2026-04-15T10:26:00.000Z'), durationSeconds: 1440, progressDelta: 3 }]], + ); + + await expect(repo.deleteSupersededSyncEstimates(7, 'ks-abcdef123456-', 'kor:device01:')).resolves.toEqual({ deleted: 1 }); + + expect(deleteFn).toHaveBeenCalledWith(readingSessions); + expect(dailyValues).toHaveBeenCalledWith([ + expect.objectContaining({ day: '2026-04-15', readingSeconds: 1440, progressDelta: 3, sessionsCount: 1 }), + ]); + }); + + it('pages the selection rather than reading a whole history into memory', async () => { + const pageSize = 500; + const firstPage = Array.from({ length: pageSize }, (_, index) => estimate(index + 1, '2026-04-15')); + const { repo, deleteFn } = makeHarness([firstPage, [estimate(pageSize + 1, '2026-04-16')]], [[]]); + + await expect(repo.deleteSupersededSyncEstimates(7, 'ks-abcdef123456-', 'kor:device01:')).resolves.toEqual({ deleted: pageSize + 1 }); + + // One delete per page, rather than one delete of everything gathered up front. + expect(deleteFn.mock.calls.filter((call) => call[0] === readingSessions)).toHaveLength(2); + }); + + it('does the whole retirement in one transaction', async () => { + const { repo, transaction } = makeHarness([[]]); + + await repo.deleteSupersededSyncEstimates(7, 'ks-abcdef123456-', 'kor:device01:'); + + expect(transaction).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/modules/reading-session/reading-session.repository.ts b/server/src/modules/reading-session/reading-session.repository.ts index bf980585c..e8b978bf5 100644 --- a/server/src/modules/reading-session/reading-session.repository.ts +++ b/server/src/modules/reading-session/reading-session.repository.ts @@ -1,5 +1,5 @@ import { Inject, Injectable } from '@nestjs/common'; -import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, lt, lte, max, min, sql } from 'drizzle-orm'; +import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, like, lt, lte, max, min, sql } from 'drizzle-orm'; import type { NodePgDatabase } from 'drizzle-orm/node-postgres'; import type { @@ -26,6 +26,7 @@ type Db = NodePgDatabase; type Tx = Parameters[0]>[0]; const MIN_READING_SESSION_SECONDS = 10; +const ESTIMATE_CLEANUP_PAGE_SIZE = 500; export type SaveReadingSessionResult = | { kind: 'saved' } @@ -380,6 +381,97 @@ export class ReadingSessionRepository { }); } + /** + * Drops the estimates a device made for reading it has since reported real page timings for. + * + * Scoped by overlap rather than by device, because a sweep is not proof that the measured + * history covers the estimated one. A device may upload no page stats at all, or only the + * window its local statistics database still holds, and deleting everything on the strength + * of "this device now has the plugin" would erase reading nothing replaced. + * + * Run on every sweep, not only the first. It is idempotent, it costs one indexed lookup when + * there is nothing to retire, and a sweep whose cleanup failed is retried by the next one + * rather than leaving the double count in place forever. + */ + async deleteSupersededSyncEstimates( + userId: number, + estimateSessionIdPrefix: string, + measuredSessionIdPrefix: string, + timeZone = 'UTC', + ): Promise<{ deleted: number }> { + return this.db.transaction(async (tx) => { + const daysByLibrary = new Map>(); + let deleted = 0; + let cursor = 0; + + for (;;) { + const rows = await tx + .select({ + id: readingSessions.id, + startedAt: readingSessions.startedAt, + endedAt: readingSessions.endedAt, + durationSeconds: readingSessions.durationSeconds, + progressDelta: readingSessions.progressDelta, + libraryId: books.libraryId, + }) + .from(readingSessions) + .innerJoin(books, eq(books.id, readingSessions.bookId)) + .where( + and( + eq(readingSessions.userId, userId), + like(readingSessions.sessionId, `${estimateSessionIdPrefix}%`), + gt(readingSessions.id, cursor), + sql`exists ( + select 1 + from reading_sessions measured + where measured.user_id = ${userId} + and measured.session_id like ${`${measuredSessionIdPrefix}%`} + and measured.started_at < ${readingSessions.endedAt} + and measured.ended_at > ${readingSessions.startedAt} + )`, + ), + ) + .orderBy(readingSessions.id) + .limit(ESTIMATE_CLEANUP_PAGE_SIZE); + + if (rows.length === 0) break; + cursor = rows[rows.length - 1]!.id; + + for (const row of rows) { + const days = daysByLibrary.get(row.libraryId) ?? new Set(); + for (const day of getReadingSessionDayKeys( + { startedAt: row.startedAt, endedAt: row.endedAt, durationSeconds: row.durationSeconds, progressDelta: row.progressDelta ?? null }, + timeZone, + )) { + days.add(day); + } + daysByLibrary.set(row.libraryId, days); + } + + // Deleted rows sit behind the cursor, so paging is unaffected by removing them. + await tx.delete(readingSessions).where( + and( + eq(readingSessions.userId, userId), + inArray( + readingSessions.id, + rows.map((row) => row.id), + ), + ), + ); + deleted += rows.length; + + if (rows.length < ESTIMATE_CLEANUP_PAGE_SIZE) break; + } + + // Ascending so concurrent writers take the per-library locks in one order. + for (const libraryId of [...daysByLibrary.keys()].sort((a, b) => a - b)) { + await this.recomputeDailyStats(tx, userId, libraryId, [...daysByLibrary.get(libraryId)!], timeZone); + } + + return { deleted }; + }); + } + private async upsertDailyStats( tx: Tx, params: { diff --git a/server/src/modules/reading-session/reading-session.service.test.ts b/server/src/modules/reading-session/reading-session.service.test.ts index 2ee961e24..ff426d89e 100644 --- a/server/src/modules/reading-session/reading-session.service.test.ts +++ b/server/src/modules/reading-session/reading-session.service.test.ts @@ -497,3 +497,67 @@ describe('ReadingSessionService - deleteSessionByBook', () => { expect(bookAccessOrder).toBeLessThan(repoOrder); }); }); + +describe('ReadingSessionService.recordSyncedSession', () => { + const repo = { + saveSession: + vi.fn< + ( + ...args: [number, number, string, Date, Date, number, number | null, number | null, ReadingSessionSource, string] + ) => Promise + >(), + }; + const bookService = { verifyFileAccess: vi.fn(), autoUpdateReadStatusForProgress: vi.fn() }; + const achievementEvents = { emit: vi.fn() }; + let service: ReadingSessionService; + + const params = { + userId: 12, + bookFileId: 44, + sessionId: 'kosync-abc', + startedAt: new Date('2026-07-01T01:48:00.000Z'), + endedAt: new Date('2026-07-01T02:00:00.000Z'), + durationSeconds: 720, + progressDelta: 8, + endProgress: 50, + source: 'koreader' as ReadingSessionSource, + timeZone: 'America/Halifax', + }; + + beforeEach(() => { + vi.clearAllMocks(); + service = new ReadingSessionService( + repo as unknown as ReadingSessionRepository, + bookService as unknown as BookService, + achievementEvents as never, + ); + }); + + it('stores the session and announces it for achievement and streak evaluation', async () => { + repo.saveSession.mockResolvedValue({ kind: 'saved' }); + + await expect(service.recordSyncedSession(params)).resolves.toEqual({ kind: 'saved' }); + + expect(repo.saveSession).toHaveBeenCalledWith(12, 44, 'kosync-abc', params.startedAt, params.endedAt, 720, 8, 50, 'koreader', 'America/Halifax'); + expect(achievementEvents.emit).toHaveBeenCalledWith( + 'reading-session.saved', + expect.objectContaining({ userId: 12, bookFileId: 44, durationSeconds: 720, timezone: 'America/Halifax' }), + ); + }); + + it('stays quiet when the session was not stored', async () => { + repo.saveSession.mockResolvedValue({ kind: 'skipped', reason: 'duplicate_session_id' }); + + await expect(service.recordSyncedSession(params)).resolves.toEqual({ kind: 'skipped', reason: 'duplicate_session_id' }); + expect(achievementEvents.emit).not.toHaveBeenCalled(); + }); + + it('does not re-check access or move the read status, which the caller has already done', async () => { + repo.saveSession.mockResolvedValue({ kind: 'saved' }); + + await service.recordSyncedSession(params); + + expect(bookService.verifyFileAccess).not.toHaveBeenCalled(); + expect(bookService.autoUpdateReadStatusForProgress).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/modules/reading-session/reading-session.service.ts b/server/src/modules/reading-session/reading-session.service.ts index 2400ac941..f4eca04cd 100644 --- a/server/src/modules/reading-session/reading-session.service.ts +++ b/server/src/modules/reading-session/reading-session.service.ts @@ -11,7 +11,7 @@ import { AchievementEventsService, ACHIEVEMENT_EVENT_READING_SESSION_SAVED } fro import type { CreateManualReadingSessionDto } from './dto/create-manual-reading-session.dto'; import type { ListBookReadingSessionsDto } from './dto/list-book-reading-sessions.dto'; import type { SaveReadingSessionDto } from './dto/save-reading-session.dto'; -import { ReadingSessionRepository } from './reading-session.repository'; +import { ReadingSessionRepository, type SaveReadingSessionResult } from './reading-session.repository'; @Injectable() export class ReadingSessionService { @@ -99,6 +99,82 @@ export class ReadingSessionService { } } + /** + * Stores a session a device reported over a sync protocol rather than through the reader. + * + * Deliberately not `save()`: the caller resolved the file through the user's own accessible + * libraries and has already applied the position and the status change that came with it, so + * repeating the access check here would re-run one just done and the status update would run + * twice for a single push. + */ + async recordSyncedSession(params: { + userId: number; + bookFileId: number; + sessionId: string; + startedAt: Date; + endedAt: Date; + durationSeconds: number; + progressDelta: number | null; + endProgress: number | null; + source: ReadingSessionSource; + timeZone: string; + }): Promise { + const result = await this.repo.saveSession( + params.userId, + params.bookFileId, + params.sessionId, + params.startedAt, + params.endedAt, + params.durationSeconds, + params.progressDelta, + params.endProgress, + params.source, + params.timeZone, + ); + + if (result.kind === 'saved') { + this.achievementEvents.emit(ACHIEVEMENT_EVENT_READING_SESSION_SAVED, { + userId: params.userId, + bookFileId: params.bookFileId, + durationSeconds: params.durationSeconds, + startedAt: params.startedAt, + endedAt: params.endedAt, + progressDelta: params.progressDelta, + endProgress: params.endProgress, + timezone: params.timeZone, + }); + } + + return result; + } + + /** + * Retires the estimates a device made for reading the same device has since reported real + * page timings for. Prefixes are supplied by the caller, which owns both id formats. + */ + async discardSupersededSyncEstimates( + user: RequestUser, + prefixes: { estimateSessionIdPrefix: string; measuredSessionIdPrefix: string }, + ): Promise<{ deleted: number }> { + const event = 'reading_session.discard_sync_estimates'; + const startedAtMs = Date.now(); + + const result = await this.repo.deleteSupersededSyncEstimates( + user.id, + prefixes.estimateSessionIdPrefix, + prefixes.measuredSessionIdPrefix, + this.resolveUserTimeZone(user), + ); + + if (result.deleted > 0) { + this.logger.log( + `[${event}] [end] userId=${user.id} durationMs=${Date.now() - startedAtMs} deleted=${result.deleted} - estimated sessions superseded by measured page timings`, + ); + } + + return result; + } + async createManualSession(bookId: number, dto: CreateManualReadingSessionDto, user: RequestUser): Promise { const event = 'book.reading_session.create_manual'; const startedAtMs = Date.now(); diff --git a/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.test.ts b/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.test.ts new file mode 100644 index 000000000..2eefeee3c --- /dev/null +++ b/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.test.ts @@ -0,0 +1,121 @@ +import { Logger } from '@nestjs/common'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { UserReadingStatsTimeZoneBackfillService } from './user-reading-stats-timezone-backfill.service'; + +describe('UserReadingStatsTimeZoneBackfillService', () => { + const userStatistics = { + listUserIdsWithReadingHistory: vi.fn(), + getUserTimeZone: vi.fn(), + rebuildDailyStatsForUser: vi.fn(), + }; + const appSettings = { + getValue: vi.fn(), + setValue: vi.fn(), + }; + let service: UserReadingStatsTimeZoneBackfillService; + + beforeEach(() => { + vi.resetAllMocks(); + vi.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + vi.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined); + appSettings.getValue.mockResolvedValue(null); + appSettings.setValue.mockResolvedValue(undefined); + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 0, inserted: 0, libraries: 0 }); + userStatistics.getUserTimeZone.mockResolvedValue('UTC'); + service = new UserReadingStatsTimeZoneBackfillService(userStatistics as never, appSettings as never); + }); + + it('rebuilds every reader in their own timezone', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5, 8]); + userStatistics.getUserTimeZone.mockResolvedValueOnce('America/Halifax').mockResolvedValueOnce('UTC'); + + await expect(service.run()).resolves.toEqual({ skipped: false, users: 2, failed: 0 }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'America/Halifax'); + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(8, 'UTC'); + }); + + it('reads each timezone at the moment of rebuilding, not from the roster it started with', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5]); + // The user corrects their own timezone while the pass is walking the roster. + userStatistics.getUserTimeZone.mockResolvedValue('America/Halifax'); + + await service.run(); + + const listOrder = userStatistics.listUserIdsWithReadingHistory.mock.invocationCallOrder[0]; + const readOrder = userStatistics.getUserTimeZone.mock.invocationCallOrder[0]; + expect(listOrder).toBeLessThan(readOrder); + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'America/Halifax'); + }); + + it('skips a reader deleted between the listing and the rebuild', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5]); + userStatistics.getUserTimeZone.mockResolvedValue(null); + + await expect(service.run()).resolves.toEqual({ skipped: false, users: 1, failed: 0 }); + expect(userStatistics.rebuildDailyStatsForUser).not.toHaveBeenCalled(); + }); + + it('records the repair so a restart does not redo it', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5]); + + await service.run(); + + expect(appSettings.setValue).toHaveBeenCalledWith('reading_stats_timezone_backfill_v1', expect.any(String)); + }); + + it('does nothing on a later start, without listing readers at all', async () => { + appSettings.getValue.mockResolvedValue('2026-08-25T00:00:00.000Z'); + + await expect(service.run()).resolves.toEqual({ skipped: true, users: 0, failed: 0 }); + + expect(userStatistics.listUserIdsWithReadingHistory).not.toHaveBeenCalled(); + expect(userStatistics.rebuildDailyStatsForUser).not.toHaveBeenCalled(); + }); + + it('finishes the remaining readers when one of them fails', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5, 8, 9]); + userStatistics.rebuildDailyStatsForUser.mockRejectedValueOnce(new Error('deadlock detected')); + + await expect(service.run()).resolves.toEqual({ skipped: false, users: 3, failed: 1 }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledTimes(3); + }); + + it('leaves a partial repair unrecorded, so the next start retries it', async () => { + userStatistics.listUserIdsWithReadingHistory.mockResolvedValue([5]); + userStatistics.rebuildDailyStatsForUser.mockRejectedValue(new Error('deadlock detected')); + + await service.run(); + + expect(appSettings.setValue).not.toHaveBeenCalled(); + }); + + it('does not delay startup or crash it when the pass fails outright', async () => { + userStatistics.listUserIdsWithReadingHistory.mockRejectedValue(new Error('database unavailable')); + + expect(service.onApplicationBootstrap()).toBeUndefined(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(appSettings.setValue).not.toHaveBeenCalled(); + }); + + it('measures a failed pass from when it started, not from when it failed', async () => { + // The clock has to be read before run() is handed to catch(), or the failure log reports the + // moment it failed minus itself and every outright failure looks instant. + const error = vi.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined); + userStatistics.listUserIdsWithReadingHistory.mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 25)); + throw new Error('database unavailable'); + }); + + service.onApplicationBootstrap(); + await new Promise((resolve) => setTimeout(resolve, 80)); + + const message = String(error.mock.calls.at(-1)?.[0] ?? ''); + expect(message).toContain('[fail]'); + expect(Number(/durationMs=(\d+)/.exec(message)?.[1] ?? -1)).toBeGreaterThanOrEqual(20); + }); +}); diff --git a/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.ts b/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.ts new file mode 100644 index 000000000..bda571cc2 --- /dev/null +++ b/server/src/modules/user-statistics/user-reading-stats-timezone-backfill.service.ts @@ -0,0 +1,93 @@ +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; + +import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; +import { AppSettingsService } from '../app-settings/app-settings.service'; +import { UserStatisticsService } from './user-statistics.service'; + +/** + * Daily reading stats were aggregated in UTC before per-user timezones reached this pipeline, + * so a row's day boundary is wherever UTC put it rather than where the reader was. An evening + * session west of UTC landed on the following day, which starts a streak late and leaves the + * real reading day empty. + * + * Correcting the setting rebuilds that history from then on, but only for someone whose zone + * was wrong. Anyone already configured correctly has nothing left to change, and the rolling + * aggregation only ever revisits the last couple of days, so their history would stay wrong + * forever. This repairs it once, for everyone, from the sessions themselves. + */ +const BACKFILL_SETTING_KEY = 'reading_stats_timezone_backfill_v1'; +const EVENT = 'user_statistics.timezone_backfill'; + +@Injectable() +export class UserReadingStatsTimeZoneBackfillService implements OnApplicationBootstrap { + private readonly logger = new Logger(UserReadingStatsTimeZoneBackfillService.name); + + constructor( + private readonly userStatistics: UserStatisticsService, + private readonly appSettings: AppSettingsService, + ) {} + + onApplicationBootstrap(): void { + // Detached: this walks every reader's whole history, and nothing it produces is needed to + // serve a request. It contends with the hourly aggregation only through the per-library + // advisory lock both take, and both derive the same rows from the same sessions. + const startedAt = Date.now(); + void this.run().catch((error: unknown) => this.logFailure(error, startedAt)); + } + + async run(): Promise<{ skipped: boolean; users: number; failed: number }> { + if (await this.appSettings.getValue(BACKFILL_SETTING_KEY)) { + return { skipped: true, users: 0, failed: 0 }; + } + + const startedAt = Date.now(); + const userIds = await this.userStatistics.listUserIdsWithReadingHistory(); + if (userIds.length === 0) { + await this.appSettings.setValue(BACKFILL_SETTING_KEY, new Date().toISOString()); + return { skipped: false, users: 0, failed: 0 }; + } + + this.logger.log(`[${EVENT}] [start] users=${userIds.length} - rebuilding daily reading stats in each reader's own timezone`); + + let failed = 0; + for (const userId of userIds) { + let timeZone = 'UTC'; + try { + // Read here rather than with the roster: a user correcting their own timezone while + // this walks the list would otherwise have that correct rebuild overwritten with the + // value captured before they changed it. + const current = await this.userStatistics.getUserTimeZone(userId); + if (current === null) continue; + timeZone = current; + await this.userStatistics.rebuildDailyStatsForUser(userId, timeZone); + } catch (error) { + failed++; + const errorClass = error instanceof Error ? error.constructor.name : 'UnknownError'; + const message = sanitizeLogValue(error instanceof Error ? error.message : 'unknown error'); + this.logger.warn( + `[${EVENT}] [fail] userId=${userId} timeZone="${sanitizeLogValue(timeZone)}" errorClass=${errorClass} error="${message}" - rebuilding one reader's daily stats failed`, + ); + } + } + + // Marked done only on a clean pass. A partial repair that recorded itself as finished would + // leave the users it missed with no second chance, and repeating it costs only time. + if (failed === 0) { + await this.appSettings.setValue(BACKFILL_SETTING_KEY, new Date().toISOString()); + } + + this.logger.log( + `[${EVENT}] [end] durationMs=${Date.now() - startedAt} users=${userIds.length} failed=${failed} completed=${failed === 0} - daily reading stats rebuilt${failed > 0 ? ', retrying the whole pass on the next start' : ''}`, + ); + + return { skipped: false, users: userIds.length, failed }; + } + + private logFailure(error: unknown, startedAt: number): void { + const errorClass = error instanceof Error ? error.constructor.name : 'UnknownError'; + const message = sanitizeLogValue(error instanceof Error ? error.message : 'unknown error'); + this.logger.error( + `[${EVENT}] [fail] durationMs=${Date.now() - startedAt} errorClass=${errorClass} error="${message}" - daily reading stats backfill failed`, + ); + } +} diff --git a/server/src/modules/user-statistics/user-statistics.module.test.ts b/server/src/modules/user-statistics/user-statistics.module.test.ts index 0c90c3f2d..971a73c34 100644 --- a/server/src/modules/user-statistics/user-statistics.module.test.ts +++ b/server/src/modules/user-statistics/user-statistics.module.test.ts @@ -2,6 +2,7 @@ import 'reflect-metadata'; import { MODULE_METADATA } from '@nestjs/common/constants'; +import { UserReadingStatsTimeZoneBackfillService } from './user-reading-stats-timezone-backfill.service'; import { UserStatisticsAggregationJob } from './user-statistics-aggregation.job'; import { UserStatisticsController } from './user-statistics.controller'; import { UserStatisticsModule } from './user-statistics.module'; @@ -15,6 +16,8 @@ describe('UserStatisticsModule', () => { UserStatisticsService, UserStatisticsRepository, UserStatisticsAggregationJob, + UserReadingStatsTimeZoneBackfillService, ]); + expect(Reflect.getMetadata(MODULE_METADATA.EXPORTS, UserStatisticsModule)).toEqual([UserStatisticsService]); }); }); diff --git a/server/src/modules/user-statistics/user-statistics.module.ts b/server/src/modules/user-statistics/user-statistics.module.ts index 9b7644e64..27ce83f99 100644 --- a/server/src/modules/user-statistics/user-statistics.module.ts +++ b/server/src/modules/user-statistics/user-statistics.module.ts @@ -1,12 +1,17 @@ import { Module } from '@nestjs/common'; +import { AppSettingsModule } from '../app-settings/app-settings.module'; +import { UserReadingStatsTimeZoneBackfillService } from './user-reading-stats-timezone-backfill.service'; + import { UserStatisticsAggregationJob } from './user-statistics-aggregation.job'; import { UserStatisticsController } from './user-statistics.controller'; import { UserStatisticsRepository } from './user-statistics.repository'; import { UserStatisticsService } from './user-statistics.service'; @Module({ + imports: [AppSettingsModule], controllers: [UserStatisticsController], - providers: [UserStatisticsService, UserStatisticsRepository, UserStatisticsAggregationJob], + providers: [UserStatisticsService, UserStatisticsRepository, UserStatisticsAggregationJob, UserReadingStatsTimeZoneBackfillService], + exports: [UserStatisticsService], }) export class UserStatisticsModule {} diff --git a/server/src/modules/user-statistics/user-statistics.repository.test.ts b/server/src/modules/user-statistics/user-statistics.repository.test.ts index 80beb6e95..8405d3cf2 100644 --- a/server/src/modules/user-statistics/user-statistics.repository.test.ts +++ b/server/src/modules/user-statistics/user-statistics.repository.test.ts @@ -477,4 +477,146 @@ describe('UserStatisticsRepository', () => { }), ]); }); + it('takes the shared advisory locks in one global order', async () => { + // rebuildDailyStatsForUser locks ascending inside its own transaction. Walking the group map + // in query order can invert that for a user holding two libraries, and Postgres resolves the + // cycle by aborting one side: the rebuild logs and moves on, this pass loses the whole hour. + const txSelectQueue = [ + [ + { userId: 5, libraryId: 9, settings: { timezone: 'UTC' } }, + { userId: 5, libraryId: 3, settings: { timezone: 'UTC' } }, + ], + [], + ]; + const tx = { + execute: vi.fn().mockResolvedValue({ rowCount: 0 }), + select: vi.fn((fields?: Record) => makeChain(txSelectQueue.shift() ?? [], fields)), + insert: vi.fn().mockReturnValue({ values: vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) }) }), + }; + const db = { transaction: vi.fn(async (callback: (trx: typeof tx) => Promise) => callback(tx)) }; + const repo = new UserStatisticsRepository(db as never); + const lock = vi.spyOn(repo as any, 'lockDailyStats').mockResolvedValue(undefined); + + await repo.recomputeRecentDailyStats(2); + + expect(lock.mock.calls.map((call) => [call[1], call[2]])).toEqual([ + [5, 3], + [5, 9], + ]); + }); + + describe('rebuildDailyStatsForUser', () => { + function makeRebuildTx(distinctResults: unknown[], sessionPages: unknown[], deletedRowCount = 0) { + const distinct = [...distinctResults]; + const pages = [...sessionPages]; + const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) }); + const tx = { + // One pair per library: the advisory lock, then the delete whose count is reported. + execute: vi.fn().mockResolvedValue({ rowCount: deletedRowCount }), + selectDistinct: vi.fn((fields?: Record) => makeChain(distinct.shift() ?? [], fields)), + select: vi.fn((fields?: Record) => makeChain(pages.shift() ?? [], fields)), + insert: vi.fn().mockReturnValue({ values: dailyValues }), + }; + const db = { transaction: vi.fn(async (callback: (trx: typeof tx) => Promise) => callback(tx)) }; + return { tx, db, dailyValues }; + } + + it('re-attributes history to the local day of the new timezone', async () => { + // The reported failure: a session at 00:49 UTC is the previous evening in Halifax, so a + // UTC-built row starts the streak a day late and leaves the real reading day empty. + const { db, tx, dailyValues } = makeRebuildTx( + [[{ libraryId: 3 }], [{ libraryId: 3 }]], + [ + [ + { + id: 11, + startedAt: new Date('2026-07-01T00:49:37.000Z'), + endedAt: new Date('2026-07-01T02:00:06.000Z'), + durationSeconds: 3941, + progressDelta: 4, + }, + ], + ], + 6, + ); + const repo = new UserStatisticsRepository(db as never); + + await expect(repo.rebuildDailyStatsForUser(5, 'America/Halifax')).resolves.toEqual({ deleted: 6, inserted: 1, libraries: 1 }); + + expect(dailyValues).toHaveBeenCalledWith([ + expect.objectContaining({ userId: 5, libraryId: 3, day: '2026-06-30', readingSeconds: 3941, progressDelta: 4, sessionsCount: 1 }), + ]); + expect(tx.execute).toHaveBeenCalledTimes(2); + }); + + it('rebuilds libraries that only still hold rows, so stale days are dropped rather than kept', async () => { + const { db, tx, dailyValues } = makeRebuildTx([[{ libraryId: 9 }, { libraryId: 2 }], [{ libraryId: 2 }]], [[], []], 3); + const repo = new UserStatisticsRepository(db as never); + + // Libraries are locked in ascending order so concurrent writers cannot deadlock on them. + await expect(repo.rebuildDailyStatsForUser(5, 'UTC')).resolves.toEqual({ deleted: 6, inserted: 0, libraries: 2 }); + expect(tx.execute).toHaveBeenCalledTimes(4); + expect(dailyValues).not.toHaveBeenCalled(); + }); + + it('accumulates across pages instead of stopping at the first batch', async () => { + const pageSize = 5_000; + const firstPage = Array.from({ length: pageSize }, (_, index) => ({ + id: index + 1, + startedAt: new Date('2026-04-15T08:00:00.000Z'), + endedAt: new Date('2026-04-15T08:01:00.000Z'), + durationSeconds: 60, + progressDelta: null, + })); + const secondPage = [ + { + id: pageSize + 1, + startedAt: new Date('2026-04-15T09:00:00.000Z'), + endedAt: new Date('2026-04-15T09:00:30.000Z'), + durationSeconds: 30, + progressDelta: null, + }, + ]; + const { db, tx, dailyValues } = makeRebuildTx([[{ libraryId: 1 }], []], [firstPage, secondPage]); + const repo = new UserStatisticsRepository(db as never); + + await expect(repo.rebuildDailyStatsForUser(5, 'UTC')).resolves.toEqual({ deleted: 0, inserted: 1, libraries: 1 }); + + expect(tx.select).toHaveBeenCalledTimes(2); + expect(dailyValues).toHaveBeenCalledWith([ + expect.objectContaining({ day: '2026-04-15', readingSeconds: pageSize * 60 + 30, sessionsCount: pageSize + 1 }), + ]); + }); + + it('does nothing for a user with no reading history at all', async () => { + const { db, tx, dailyValues } = makeRebuildTx([[], []], []); + const repo = new UserStatisticsRepository(db as never); + + await expect(repo.rebuildDailyStatsForUser(5, 'Europe/Berlin')).resolves.toEqual({ deleted: 0, inserted: 0, libraries: 0 }); + expect(tx.execute).not.toHaveBeenCalled(); + expect(dailyValues).not.toHaveBeenCalled(); + }); + + it('falls back to UTC rather than trusting an unusable timezone', async () => { + const { db, dailyValues } = makeRebuildTx( + [[{ libraryId: 1 }], []], + [ + [ + { + id: 1, + startedAt: new Date('2026-07-01T00:49:37.000Z'), + endedAt: new Date('2026-07-01T01:00:00.000Z'), + durationSeconds: 623, + progressDelta: null, + }, + ], + ], + ); + const repo = new UserStatisticsRepository(db as never); + + await repo.rebuildDailyStatsForUser(5, 'Not/AZone'); + + expect(dailyValues).toHaveBeenCalledWith([expect.objectContaining({ day: '2026-07-01' })]); + }); + }); }); diff --git a/server/src/modules/user-statistics/user-statistics.repository.ts b/server/src/modules/user-statistics/user-statistics.repository.ts index c915df4cc..2c3d6cb76 100644 --- a/server/src/modules/user-statistics/user-statistics.repository.ts +++ b/server/src/modules/user-statistics/user-statistics.repository.ts @@ -18,6 +18,8 @@ import { aggregateReadingSessionDailyStats, getDayRangeForDateKeys, getReadingSessionDayKeys, + mergeReadingDailyStatsSegments, + sortReadingDailyStatsSegments, type ReadingDailyStatsSegment, } from '../../common/utils/reading-daily-stats.utils'; import { resolveTimeZone } from '../../common/utils/timezone.utils'; @@ -59,6 +61,9 @@ type SessionTimelineConflictRow = { endedAt: Date; }; const RECENT_DAILY_AGGREGATION_DAYS = 2; +/** A full-history rebuild pages sessions rather than loading a long reader's whole timeline at once. */ +const DAILY_STATS_REBUILD_PAGE_SIZE = 5_000; +const DAILY_STATS_INSERT_CHUNK_SIZE = 1_000; @Injectable() export class UserStatisticsRepository { @@ -849,10 +854,15 @@ export class UserStatisticsRepository { groups.set(`${group.userId}:${group.libraryId}`, group); } + // Sorted so this pass and a concurrent per-user rebuild take the shared advisory locks in + // one order. Map insertion order follows the query results, which can invert against the + // rebuild's ascending order and deadlock a user holding more than one library. + const orderedGroups = [...groups.values()].sort((a, b) => a.userId - b.userId || a.libraryId - b.libraryId); + let deleted = 0; let inserted = 0; - for (const group of groups.values()) { + for (const group of orderedGroups) { await this.lockDailyStats(tx, group.userId, group.libraryId); const deleteResult = await tx.execute(sql` @@ -895,4 +905,115 @@ export class UserStatisticsRepository { return { deleted, inserted, since: sinceDay }; }); } + + /** Everyone who has reading history to rebuild. */ + async listUserIdsWithReadingHistory(): Promise { + const [sessionUsers, statsUsers] = await Promise.all([ + this.db.selectDistinct({ userId: readingSessions.userId }).from(readingSessions), + this.db.selectDistinct({ userId: userReadingDailyStats.userId }).from(userReadingDailyStats), + ]); + + return [...new Set([...sessionUsers, ...statsUsers].map((row) => row.userId))].sort((a, b) => a - b); + } + + /** + * The timezone one user's history should be rebuilt under, read at the moment of rebuilding. + * + * Deliberately not carried along from a listing taken earlier. A user who corrects their + * timezone while a bulk rebuild is walking the roster would otherwise have their own correct + * rebuild overwritten by the stale value that listing captured. + */ + async getUserTimeZone(userId: number): Promise { + const [row] = await this.db.select({ settings: users.settings }).from(users).where(eq(users.id, userId)).limit(1); + if (!row) return null; + return resolveTimeZone((row.settings as { timezone?: unknown } | null)?.timezone, 'UTC'); + } + + /** + * Rebuilds every daily-stat row one user owns, under a single timezone. + * + * The rolling recompute above is enough while a timezone holds still, because it only ever + * revisits days that are still being written to. A user who corrects theirs invalidates all + * of their history at once: rows are keyed by local day, so every day boundary moves, and a + * streak broken by the old zone stays broken until the rows that built it are rebuilt. + * + * Sessions are paged rather than read whole. The day map is bounded by how long the user has + * been reading; their session list is not. + */ + async rebuildDailyStatsForUser(userId: number, timeZone: string): Promise<{ deleted: number; inserted: number; libraries: number }> { + const resolvedTimeZone = resolveTimeZone(timeZone, 'UTC'); + + return this.db.transaction(async (tx) => { + const [statLibraries, sessionLibraries] = await Promise.all([ + tx.selectDistinct({ libraryId: userReadingDailyStats.libraryId }).from(userReadingDailyStats).where(eq(userReadingDailyStats.userId, userId)), + tx + .selectDistinct({ libraryId: books.libraryId }) + .from(readingSessions) + .innerJoin(books, eq(books.id, readingSessions.bookId)) + .where(eq(readingSessions.userId, userId)), + ]); + + // Libraries the user no longer has sessions in still hold rows, and dropping them is the + // point of a rebuild. Sorted so concurrent writers take the per-library locks in one order. + const libraryIds = [...new Set([...statLibraries, ...sessionLibraries].map((row) => row.libraryId))].sort((a, b) => a - b); + + let deleted = 0; + let inserted = 0; + + for (const libraryId of libraryIds) { + await this.lockDailyStats(tx, userId, libraryId); + + const deleteResult = await tx.execute(sql` + delete from user_reading_daily_stats + where user_id = ${userId} + and library_id = ${libraryId} + `); + deleted += Number((deleteResult as { rowCount?: number }).rowCount ?? 0); + + const byDay = new Map(); + let cursor = 0; + for (;;) { + const rows = await tx + .select({ + id: readingSessions.id, + startedAt: readingSessions.startedAt, + endedAt: readingSessions.endedAt, + durationSeconds: readingSessions.durationSeconds, + progressDelta: readingSessions.progressDelta, + }) + .from(readingSessions) + .innerJoin(books, eq(books.id, readingSessions.bookId)) + .where(and(eq(readingSessions.userId, userId), eq(books.libraryId, libraryId), gt(readingSessions.id, cursor))) + .orderBy(readingSessions.id) + .limit(DAILY_STATS_REBUILD_PAGE_SIZE); + + if (rows.length === 0) break; + cursor = rows[rows.length - 1]!.id; + + mergeReadingDailyStatsSegments( + byDay, + aggregateReadingSessionDailyStats( + rows.map((row) => ({ + startedAt: row.startedAt, + endedAt: row.endedAt, + durationSeconds: row.durationSeconds, + progressDelta: row.progressDelta ?? null, + })), + resolvedTimeZone, + ), + ); + + if (rows.length < DAILY_STATS_REBUILD_PAGE_SIZE) break; + } + + const segments = sortReadingDailyStatsSegments(byDay); + for (let offset = 0; offset < segments.length; offset += DAILY_STATS_INSERT_CHUNK_SIZE) { + await this.insertDailyStatsSegments(tx, userId, libraryId, segments.slice(offset, offset + DAILY_STATS_INSERT_CHUNK_SIZE)); + } + inserted += segments.length; + } + + return { deleted, inserted, libraries: libraryIds.length }; + }); + } } 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 40c4befaf..7b56cd0cd 100644 --- a/server/src/modules/user-statistics/user-statistics.service.test.ts +++ b/server/src/modules/user-statistics/user-statistics.service.test.ts @@ -686,4 +686,27 @@ describe('UserStatisticsService', () => { await service.getProgressFunnel(userB, query); expect(repo.getProgressFunnelInRange).toHaveBeenCalledTimes(3); }); + it('drops only the rebuilt user from the statistics cache', async () => { + const repo = { + 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 userA = { id: 1, isSuperuser: false } as any; + const userB = { id: 2, isSuperuser: false } as any; + const query = { libraryIds: [1] }; + + await service.getSummary(userA, query); + await service.getSummary(userB, query); + expect(repo.getSummary).toHaveBeenCalledTimes(2); + + await expect(service.rebuildDailyStatsForUser(1, 'America/Halifax')).resolves.toEqual({ deleted: 4, inserted: 2, libraries: 1 }); + expect(repo.rebuildDailyStatsForUser).toHaveBeenCalledWith(1, 'America/Halifax'); + + await service.getSummary(userA, query); + expect(repo.getSummary).toHaveBeenCalledTimes(3); + + await service.getSummary(userB, query); + expect(repo.getSummary).toHaveBeenCalledTimes(3); + }); }); diff --git a/server/src/modules/user-statistics/user-statistics.service.ts b/server/src/modules/user-statistics/user-statistics.service.ts index 77b1fe938..f55a4b531 100644 --- a/server/src/modules/user-statistics/user-statistics.service.ts +++ b/server/src/modules/user-statistics/user-statistics.service.ts @@ -618,4 +618,18 @@ export class UserStatisticsService { this.cache.clear(); return result; } + + listUserIdsWithReadingHistory() { + return this.repo.listUserIdsWithReadingHistory(); + } + + getUserTimeZone(userId: number) { + return this.repo.getUserTimeZone(userId); + } + + async rebuildDailyStatsForUser(userId: number, timeZone: string) { + const result = await this.repo.rebuildDailyStatsForUser(userId, timeZone); + this.cache.clearForScope(String(userId)); + return result; + } } diff --git a/server/src/modules/user/user.module.ts b/server/src/modules/user/user.module.ts index c5fae9696..64478c393 100644 --- a/server/src/modules/user/user.module.ts +++ b/server/src/modules/user/user.module.ts @@ -1,6 +1,7 @@ import { Module } from '@nestjs/common'; import { AppSettingsModule } from '../app-settings/app-settings.module'; +import { UserStatisticsModule } from '../user-statistics/user-statistics.module'; import { ContentFilterRepository } from './content-filter.repository'; import { OidcIdentityRepository } from './oidc-identity.repository'; import { UserAvatarStorageService } from './user-avatar-storage.service'; @@ -10,7 +11,7 @@ import { UserRepository } from './user.repository'; import { UserService } from './user.service'; @Module({ - imports: [AppSettingsModule], + imports: [AppSettingsModule, UserStatisticsModule], controllers: [UserController], providers: [UserService, UserRepository, UserAvatarService, UserAvatarStorageService, OidcIdentityRepository, ContentFilterRepository], exports: [UserService, UserRepository, UserAvatarService, UserAvatarStorageService, OidcIdentityRepository, ContentFilterRepository], diff --git a/server/src/modules/user/user.service.test.ts b/server/src/modules/user/user.service.test.ts index 412c3ef53..59c35899d 100644 --- a/server/src/modules/user/user.service.test.ts +++ b/server/src/modules/user/user.service.test.ts @@ -1,7 +1,7 @@ vi.mock('bcryptjs', () => ({ hash: vi.fn() })); vi.mock('crypto', () => ({ randomBytes: vi.fn() })); -import { BadRequestException, ConflictException, ForbiddenException, NotFoundException } from '@nestjs/common'; +import { BadRequestException, ConflictException, ForbiddenException, Logger, NotFoundException } from '@nestjs/common'; import { hash } from 'bcryptjs'; import { randomBytes } from 'crypto'; import { Permission } from '@bookorbit/types'; @@ -53,12 +53,17 @@ describe('UserService', () => { const appSettingsService = { getDefaultLibraryAccessLibraryIds: vi.fn(), }; + const userStatistics = { + rebuildDailyStatsForUser: vi.fn(), + }; let service: UserService; beforeEach(() => { vi.resetAllMocks(); - service = new UserService(userRepo as any, config as any, contentFilterRepo as any, appSettingsService as any); + vi.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + vi.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined); + service = new UserService(userRepo as any, config as any, contentFilterRepo as any, appSettingsService as any, userStatistics as any); mockHash.mockResolvedValue('hashed-secret'); mockRandomBytes.mockReturnValue(Buffer.from('abcd', 'hex')); @@ -279,6 +284,100 @@ describe('UserService', () => { expect(userRepo.update).toHaveBeenCalledWith(5, { settings }); }); + it('rebuilds daily reading stats when the timezone actually changes', async () => { + userRepo.findSettingsById.mockResolvedValue({ timezone: 'UTC' }); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'America/Halifax' } }); + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 12, inserted: 9, libraries: 1 }); + + await service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'America/Halifax'); + }); + + it('rebuilds again when the same timezone is submitted, which is the only retry a user has', async () => { + // The setting saves even when the rebuild fails, so a retry looks like no change at all. + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 0, inserted: 0, libraries: 0 }); + userRepo.findSettingsById.mockResolvedValue({ timezone: 'America/Halifax' }); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'America/Halifax' } }); + + await service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'America/Halifax'); + }); + + it('recovers from a failed rebuild when the user saves the timezone a second time', async () => { + userRepo.findSettingsById.mockResolvedValue({ timezone: 'UTC' }); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'America/Halifax' } }); + userStatistics.rebuildDailyStatsForUser.mockRejectedValueOnce(new Error('deadlock detected')); + + await service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } }); + + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 4, inserted: 3, libraries: 1 }); + userRepo.findSettingsById.mockResolvedValue({ timezone: 'America/Halifax' }); + + await service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledTimes(2); + }); + + it('answers the settings write without waiting for the rebuild to finish', async () => { + // The rebuild walks the reader's whole history and its result is never returned, so holding + // the response open for it only costs a long library its settings save. + userRepo.findSettingsById.mockResolvedValue({ timezone: 'UTC' }); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'America/Halifax' } }); + + let finishRebuild!: () => void; + userStatistics.rebuildDailyStatsForUser.mockReturnValue( + new Promise((resolve) => { + finishRebuild = () => resolve({ deleted: 0, inserted: 0, libraries: 0 }); + }), + ); + + await expect(service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } })).resolves.toEqual({ + id: 5, + settings: { timezone: 'America/Halifax' }, + }); + + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'America/Halifax'); + finishRebuild(); + }); + + it('leaves stats alone for a settings write that does not touch the timezone', async () => { + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'America/Halifax', theme: 'dark' } }); + + await service.updateMySettings(5, { settings: { theme: 'dark' } }); + + // Reading the stored settings is only worth a query when this write can replace the zone. + expect(userRepo.findSettingsById).not.toHaveBeenCalled(); + expect(userStatistics.rebuildDailyStatsForUser).not.toHaveBeenCalled(); + }); + + it('treats first-time and unusable timezones as a move away from the UTC default', async () => { + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 0, inserted: 0, libraries: 0 }); + userRepo.findSettingsById.mockResolvedValue({}); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'Europe/Berlin' } }); + + await service.updateMySettings(5, { settings: { timezone: 'Europe/Berlin' } }); + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'Europe/Berlin'); + + userStatistics.rebuildDailyStatsForUser.mockClear(); + userStatistics.rebuildDailyStatsForUser.mockResolvedValue({ deleted: 0, inserted: 0, libraries: 0 }); + userRepo.findSettingsById.mockResolvedValue({ timezone: 'Europe/Berlin' }); + userRepo.update.mockResolvedValue({ id: 5, settings: { timezone: 'Not/AZone' } }); + + await service.updateMySettings(5, { settings: { timezone: 'Not/AZone' } }); + expect(userStatistics.rebuildDailyStatsForUser).toHaveBeenCalledWith(5, 'UTC'); + }); + + it('still saves the setting when the stats rebuild fails', async () => { + userRepo.findSettingsById.mockResolvedValue({ timezone: 'UTC' }); + const updated = { id: 5, settings: { timezone: 'America/Halifax' } }; + userRepo.update.mockResolvedValue(updated); + userStatistics.rebuildDailyStatsForUser.mockRejectedValue(new Error('deadlock detected')); + + await expect(service.updateMySettings(5, { settings: { timezone: 'America/Halifax' } })).resolves.toEqual(updated); + }); + it('caches an explicit achievement preference when settings are updated', async () => { userRepo.update.mockResolvedValue({ id: 5, settings: { achievementPreferences: { enabled: false } } }); @@ -612,12 +711,15 @@ describe('UserService.updateSeriesCollapsePreferences', () => { const appSettingsService = { getDefaultLibraryAccessLibraryIds: vi.fn(), }; + const userStatistics = { + rebuildDailyStatsForUser: vi.fn(), + }; beforeEach(() => { vi.resetAllMocks(); config.get.mockReturnValue('http://localhost:5173'); appSettingsService.getDefaultLibraryAccessLibraryIds.mockResolvedValue([]); - service = new UserService(userRepo as any, config as any, contentFilterRepo as any, appSettingsService as any); + service = new UserService(userRepo as any, config as any, contentFilterRepo as any, appSettingsService as any, userStatistics as any); }); it('throws NotFoundException when user does not exist', async () => { diff --git a/server/src/modules/user/user.service.ts b/server/src/modules/user/user.service.ts index 05159635c..6580a8ee3 100644 --- a/server/src/modules/user/user.service.ts +++ b/server/src/modules/user/user.service.ts @@ -1,4 +1,4 @@ -import { BadRequestException, ConflictException, ForbiddenException, Injectable, NotFoundException } from '@nestjs/common'; +import { BadRequestException, ConflictException, ForbiddenException, Injectable, Logger, NotFoundException } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { hash } from 'bcryptjs'; import { randomBytes } from 'crypto'; @@ -6,6 +6,8 @@ import { Permission } from '@bookorbit/types'; import type { ProvisioningMethod, UserAttentionResponse, UserListSummary, UserSettings } from '@bookorbit/types'; import type { RequestUser } from '../../common/types/request-user'; +import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; +import { resolveTimeZone } from '../../common/utils/timezone.utils'; import { ContentFilterRepository } from './content-filter.repository'; import { CreateUserDto } from './dto/create-user.dto'; import { CreateSharedUserDto } from './dto/create-shared-user.dto'; @@ -17,12 +19,14 @@ import { UpdateMeSettingsDto } from './dto/update-me-settings.dto'; import { UpdateSeriesCollapsePreferencesDto } from './dto/update-series-collapse-preferences.dto'; import { UserRepository, type UserListQuery } from './user.repository'; import { AppSettingsService } from '../app-settings/app-settings.service'; +import { UserStatisticsService } from '../user-statistics/user-statistics.service'; /** The band is a to-do list, not a second roster. */ const ATTENTION_BAND_LIMIT = 8; @Injectable() export class UserService { + private readonly logger = new Logger(UserService.name); private readonly achievementEnabledCache = new Map(); constructor( @@ -30,6 +34,7 @@ export class UserService { private readonly config: ConfigService, private readonly contentFilterRepo: ContentFilterRepository, private readonly appSettingsService: AppSettingsService, + private readonly userStatistics: UserStatisticsService, ) {} findByUsername(username: string) { @@ -222,12 +227,67 @@ export class UserService { } async updateMySettings(userId: number, dto: UpdateMeSettingsDto) { + // Settings are merged server-side, so the stored row is the only place the outgoing + // timezone still exists. Read only when this write is the one that can replace it. + const touchesTimeZone = Object.prototype.hasOwnProperty.call(dto.settings, 'timezone'); + const previousSettings = touchesTimeZone ? await this.userRepo.findSettingsById(userId) : null; + const user = await this.userRepo.update(userId, { settings: dto.settings }); if (!user) throw new NotFoundException('User not found'); this.updateAchievementEnabledCache(userId, dto.settings); + + if (touchesTimeZone) { + void this.rebuildReadingStatsForSubmittedTimeZone(userId, previousSettings, user.settings as Record | null); + } return user; } + /** + * Daily reading stats are stored per local day, so the timezone that produced a row is baked + * into it. The hourly aggregation only revisits the last couple of days, which leaves every + * older row attributed to the zone the user has just corrected, and a streak broken at the + * old day boundary stays broken. The rebuild is what makes the new setting retroactive. + * + * Submitting a timezone rebuilds even when it matches the stored one. The rebuild is + * idempotent, and skipping the unchanged case would make a failure permanent: the setting is + * saved either way, so saving it again is the only retry a user has, and that retry is + * exactly the case where nothing appears to have changed. + * + * A failure does not fail the save. The setting itself is already stored, and refusing the + * write would leave the user with neither the setting nor a way to ask for it again. + * + * Detached for the same reason the bootstrap backfill is: this walks the reader's whole + * history, and the response neither returns its result nor depends on it, so awaiting it + * would hold the request open for a long library and nothing else. + */ + private async rebuildReadingStatsForSubmittedTimeZone( + userId: number, + previousSettings: Record | null, + nextSettings: Record | null, + ): Promise { + const previousTimeZone = resolveTimeZone(previousSettings?.['timezone'], 'UTC'); + const nextTimeZone = resolveTimeZone(nextSettings?.['timezone'], 'UTC'); + + const event = 'user.reading_stats_rebuild'; + const startedAt = Date.now(); + this.logger.log( + `[${event}] [start] userId=${userId} previousTimeZone="${sanitizeLogValue(previousTimeZone)}" timeZone="${sanitizeLogValue(nextTimeZone)}" - rebuilding daily reading stats after a timezone change`, + ); + + try { + const result = await this.userStatistics.rebuildDailyStatsForUser(userId, nextTimeZone); + this.logger.log( + `[${event}] [end] userId=${userId} timeZone="${sanitizeLogValue(nextTimeZone)}" durationMs=${Date.now() - startedAt} libraries=${result.libraries} deleted=${result.deleted} inserted=${result.inserted} - daily reading stats rebuilt`, + ); + } catch (error) { + const errorClass = error instanceof Error ? error.constructor.name : 'UnknownError'; + const message = sanitizeLogValue(error instanceof Error ? error.message : 'unknown error'); + this.logger.warn( + `[${event}] [fail] userId=${userId} timeZone="${sanitizeLogValue(nextTimeZone)}" durationMs=${Date.now() - startedAt} errorClass=${errorClass} error="${message}" - daily reading stats rebuild failed`, + ); + } + } + async isAchievementEnabled(userId: number): Promise { const cached = this.achievementEnabledCache.get(userId); if (cached && cached.expiresAt > Date.now()) return cached.enabled;