From f41a58ac5859031334ddf15e329b667544a9cbe3 Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Tue, 1 Sep 2026 05:05:52 +0000 Subject: [PATCH] fix(stats): stop estimating sessions from KOReader progress Keep KOSync limited to position synchronization and rely on measured page timings from the BookOrbit plugin for KOReader reading sessions. Remove legacy estimates at startup and rebuild affected daily totals. Fixes #1233 --- .../utils/sync-estimate-session.utils.test.ts | 33 ---- .../utils/sync-estimate-session.utils.ts | 31 ---- .../koreader-plugin.repository.test.ts | 13 -- .../koreader/koreader-plugin.repository.ts | 22 --- .../koreader/koreader-plugin.service.test.ts | 39 ---- .../koreader/koreader-plugin.service.ts | 19 -- ...ader-sync-estimate-cleanup.service.test.ts | 57 ++++++ .../koreader-sync-estimate-cleanup.service.ts | 57 ++++++ .../modules/koreader/koreader.module.test.ts | 2 + .../src/modules/koreader/koreader.module.ts | 2 + .../modules/koreader/koreader.service.test.ts | 171 ++---------------- .../src/modules/koreader/koreader.service.ts | 117 +----------- .../reading-session.repository.test.ts | 152 +++++++++------- .../reading-session.repository.ts | 158 ++++++++-------- .../reading-session.service.test.ts | 64 ------- .../reading-session.service.ts | 77 +------- .../koreader-progress-position.e2e-spec.ts | 69 ++++++- 17 files changed, 361 insertions(+), 722 deletions(-) delete mode 100644 server/src/common/utils/sync-estimate-session.utils.test.ts delete mode 100644 server/src/common/utils/sync-estimate-session.utils.ts create mode 100644 server/src/modules/koreader/koreader-sync-estimate-cleanup.service.test.ts create mode 100644 server/src/modules/koreader/koreader-sync-estimate-cleanup.service.ts diff --git a/server/src/common/utils/sync-estimate-session.utils.test.ts b/server/src/common/utils/sync-estimate-session.utils.test.ts deleted file mode 100644 index d798f5f88..000000000 --- a/server/src/common/utils/sync-estimate-session.utils.test.ts +++ /dev/null @@ -1,33 +0,0 @@ -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 deleted file mode 100644 index 078f83f60..000000000 --- a/server/src/common/utils/sync-estimate-session.utils.ts +++ /dev/null @@ -1,31 +0,0 @@ -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 17a810d05..c15051c05 100644 --- a/server/src/modules/koreader/koreader-plugin.repository.test.ts +++ b/server/src/modules/koreader/koreader-plugin.repository.test.ts @@ -186,17 +186,4 @@ 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 0c8206445..e60bb2059 100644 --- a/server/src/modules/koreader/koreader-plugin.repository.ts +++ b/server/src/modules/koreader/koreader-plugin.repository.ts @@ -452,28 +452,6 @@ 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 eb1654270..2ad9a5dae 100644 --- a/server/src/modules/koreader/koreader-plugin.service.test.ts +++ b/server/src/modules/koreader/koreader-plugin.service.test.ts @@ -10,9 +10,6 @@ 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); @@ -55,7 +52,6 @@ describe('KoreaderPluginService', () => { normalizeNote: (value: string | null | undefined) => string | null; }; let achievementEvents: { emit: ReturnType }; - let readingSessions: { discardSupersededSyncEstimates: ReturnType }; let service: KoreaderPluginService; beforeEach(() => { @@ -94,8 +90,6 @@ describe('KoreaderPluginService', () => { }, }; achievementEvents = { emit: vi.fn() }; - readingSessions = { discardSupersededSyncEstimates: vi.fn().mockResolvedValue({ deleted: 0 }) }; - service = new KoreaderPluginService( koreaderRepo as unknown as KoreaderRepository, pluginRepo as unknown as KoreaderPluginRepository, @@ -103,7 +97,6 @@ describe('KoreaderPluginService', () => { userBookStatusService as unknown as UserBookStatusService, userBookNoteService as unknown as UserBookNoteService, achievementEvents as unknown as AchievementEventsService, - readingSessions as unknown as ReadingSessionService, ); }); @@ -532,37 +525,5 @@ 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 c7b8c8a3d..982f86d6b 100644 --- a/server/src/modules/koreader/koreader-plugin.service.ts +++ b/server/src/modules/koreader/koreader-plugin.service.ts @@ -5,14 +5,11 @@ 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'; @@ -98,7 +95,6 @@ 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 { @@ -268,21 +264,6 @@ 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-sync-estimate-cleanup.service.test.ts b/server/src/modules/koreader/koreader-sync-estimate-cleanup.service.test.ts new file mode 100644 index 000000000..b0646f66f --- /dev/null +++ b/server/src/modules/koreader/koreader-sync-estimate-cleanup.service.test.ts @@ -0,0 +1,57 @@ +import { Logger } from '@nestjs/common'; + +import { KoreaderSyncEstimateCleanupService } from './koreader-sync-estimate-cleanup.service'; + +describe('KoreaderSyncEstimateCleanupService', () => { + const readingSessions = { deleteLegacyKoreaderSyncEstimatesBatch: vi.fn() }; + const appSettings = { getValue: vi.fn(), setValue: vi.fn() }; + let service: KoreaderSyncEstimateCleanupService; + + beforeEach(() => { + vi.resetAllMocks(); + vi.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined); + vi.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined); + appSettings.getValue.mockResolvedValue(null); + appSettings.setValue.mockResolvedValue(undefined); + service = new KoreaderSyncEstimateCleanupService(readingSessions as never, appSettings as never); + }); + + it('deletes every bounded batch before recording completion', async () => { + readingSessions.deleteLegacyKoreaderSyncEstimatesBatch.mockResolvedValueOnce({ deleted: 500 }).mockResolvedValueOnce({ deleted: 3 }); + + await expect(service.run()).resolves.toEqual({ skipped: false, deleted: 503 }); + + expect(readingSessions.deleteLegacyKoreaderSyncEstimatesBatch).toHaveBeenNthCalledWith(1, 500); + expect(readingSessions.deleteLegacyKoreaderSyncEstimatesBatch).toHaveBeenNthCalledWith(2, 500); + expect(appSettings.setValue).toHaveBeenCalledWith('koreader_sync_estimate_cleanup_v1', expect.any(String)); + }); + + it('does not repeat a completed cleanup', async () => { + appSettings.getValue.mockResolvedValue('2026-08-31T00:00:00.000Z'); + + await expect(service.run()).resolves.toEqual({ skipped: true, deleted: 0 }); + + expect(readingSessions.deleteLegacyKoreaderSyncEstimatesBatch).not.toHaveBeenCalled(); + expect(appSettings.setValue).not.toHaveBeenCalled(); + }); + + it('leaves completion unrecorded when a batch fails so startup can retry', async () => { + readingSessions.deleteLegacyKoreaderSyncEstimatesBatch + .mockResolvedValueOnce({ deleted: 500 }) + .mockRejectedValueOnce(new Error('database unavailable')); + + await expect(service.run()).rejects.toThrow('database unavailable'); + + expect(readingSessions.deleteLegacyKoreaderSyncEstimatesBatch).toHaveBeenCalledTimes(2); + expect(appSettings.setValue).not.toHaveBeenCalled(); + }); + + it('runs detached without failing application bootstrap', async () => { + readingSessions.deleteLegacyKoreaderSyncEstimatesBatch.mockRejectedValue(new Error('database unavailable')); + + expect(service.onApplicationBootstrap()).toBeUndefined(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(appSettings.setValue).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/modules/koreader/koreader-sync-estimate-cleanup.service.ts b/server/src/modules/koreader/koreader-sync-estimate-cleanup.service.ts new file mode 100644 index 000000000..0aca5aa6b --- /dev/null +++ b/server/src/modules/koreader/koreader-sync-estimate-cleanup.service.ts @@ -0,0 +1,57 @@ +import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common'; + +import { sanitizeLogValue } from '../../common/utils/log-sanitize.utils'; +import { AppSettingsService } from '../app-settings/app-settings.service'; +import { ReadingSessionService } from '../reading-session/reading-session.service'; + +const BATCH_SIZE = 500; +const CLEANUP_SETTING_KEY = 'koreader_sync_estimate_cleanup_v1'; +const EVENT = 'koreader.sync_estimate_cleanup'; + +@Injectable() +export class KoreaderSyncEstimateCleanupService implements OnApplicationBootstrap { + private readonly logger = new Logger(KoreaderSyncEstimateCleanupService.name); + + constructor( + private readonly readingSessions: ReadingSessionService, + private readonly appSettings: AppSettingsService, + ) {} + + onApplicationBootstrap(): void { + const startedAt = Date.now(); + void this.run().catch((error: unknown) => this.logFailure(error, startedAt)); + } + + async run(): Promise<{ skipped: boolean; deleted: number }> { + if (await this.appSettings.getValue(CLEANUP_SETTING_KEY)) { + return { skipped: true, deleted: 0 }; + } + + const startedAt = Date.now(); + let deleted = 0; + + this.logger.log(`[${EVENT}] [start] batchSize=${BATCH_SIZE} - removing sessions estimated from KOReader progress sync`); + + for (;;) { + const batch = await this.readingSessions.deleteLegacyKoreaderSyncEstimatesBatch(BATCH_SIZE); + deleted += batch.deleted; + if (batch.deleted < BATCH_SIZE) break; + } + + await this.appSettings.setValue(CLEANUP_SETTING_KEY, new Date().toISOString()); + + this.logger.log( + `[${EVENT}] [end] durationMs=${Date.now() - startedAt} deleted=${deleted} - sessions estimated from KOReader progress sync removed`, + ); + + return { skipped: false, deleted }; + } + + 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}" - removing sessions estimated from KOReader progress sync failed`, + ); + } +} diff --git a/server/src/modules/koreader/koreader.module.test.ts b/server/src/modules/koreader/koreader.module.test.ts index c83abb5e1..f3701cd67 100644 --- a/server/src/modules/koreader/koreader.module.test.ts +++ b/server/src/modules/koreader/koreader.module.test.ts @@ -19,6 +19,7 @@ import { KoreaderPluginService } from './koreader-plugin.service'; import { KoreaderRepository } from './koreader.repository'; import { KoreaderService } from './koreader.service'; import { KoreaderStatsService } from './koreader-stats.service'; +import { KoreaderSyncEstimateCleanupService } from './koreader-sync-estimate-cleanup.service'; describe('KoreaderModule', () => { it('registers expected controllers, providers, and exports', () => { @@ -39,6 +40,7 @@ describe('KoreaderModule', () => { KoreaderBookmarkExchangeService, KoreaderBookmarkRepository, KoreaderStatsService, + KoreaderSyncEstimateCleanupService, ]); expect(Reflect.getMetadata('exports', KoreaderModule)).toEqual([KoreaderService, KoreaderRepository]); }); diff --git a/server/src/modules/koreader/koreader.module.ts b/server/src/modules/koreader/koreader.module.ts index c1a26ea72..b8c80869e 100644 --- a/server/src/modules/koreader/koreader.module.ts +++ b/server/src/modules/koreader/koreader.module.ts @@ -33,6 +33,7 @@ import { KoreaderPluginService } from './koreader-plugin.service'; import { KoreaderRepository } from './koreader.repository'; import { KoreaderService } from './koreader.service'; import { KoreaderStatsService } from './koreader-stats.service'; +import { KoreaderSyncEstimateCleanupService } from './koreader-sync-estimate-cleanup.service'; @Module({ imports: [ @@ -69,6 +70,7 @@ import { KoreaderStatsService } from './koreader-stats.service'; KoreaderBookmarkExchangeService, KoreaderBookmarkRepository, KoreaderStatsService, + KoreaderSyncEstimateCleanupService, ], exports: [KoreaderService, KoreaderRepository], }) diff --git a/server/src/modules/koreader/koreader.service.test.ts b/server/src/modules/koreader/koreader.service.test.ts index bf3fae40c..e966d3115 100644 --- a/server/src/modules/koreader/koreader.service.test.ts +++ b/server/src/modules/koreader/koreader.service.test.ts @@ -21,8 +21,6 @@ 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'; @@ -100,7 +98,6 @@ describe('KoreaderService', () => { emit: ReturnType; }; let mockPluginRepo: { - hasSweepSince: ReturnType; listSweeps: ReturnType; getPluginTotals: ReturnType; }; @@ -114,10 +111,6 @@ describe('KoreaderService', () => { let mockPackageService: { getVersionInfo: ReturnType; }; - let mockReadingSessions: { - recordSyncedSession: ReturnType; - }; - beforeEach(() => { vi.restoreAllMocks(); vi.clearAllMocks(); @@ -203,12 +196,7 @@ 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, @@ -243,7 +231,6 @@ describe('KoreaderService', () => { mockPositionConverter as never, mockBookService as never, mockPackageService as unknown as KoreaderPackageService, - mockReadingSessions as unknown as ReadingSessionService, ); }); @@ -502,162 +489,26 @@ 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); + describe('reading sessions from sync progress', () => { + it('updates progress without consulting per-device history or creating a reading session', async () => { mockRepo.resolveBookFileByHash.mockResolvedValue({ id: 44, bookId: 55, libraryId: 3, format: 'epub' }); - }); + mockRepo.getLatestDeviceProgress.mockResolvedValue({ + percentage: 0.42, + deviceId: 'device-12', + updatedAt: new Date('2026-07-01T01:48:00.000Z'), + }); - 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'), { + await service.saveProgress(syncUser(12), { document: 'abcdef1234567890fedcba', - percentage, + percentage: 0.5, progress: '/body/DocFragment[7]', device: 'Kobo Sage', - device_id: deviceId, + device_id: 'device-12', }); - } - - 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(); + expect(mockRepo.getDeviceProgressForDevice).not.toHaveBeenCalled(); }); }); diff --git a/server/src/modules/koreader/koreader.service.ts b/server/src/modules/koreader/koreader.service.ts index 23cb0e738..dbd245141 100644 --- a/server/src/modules/koreader/koreader.service.ts +++ b/server/src/modules/koreader/koreader.service.ts @@ -7,9 +7,7 @@ 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'; @@ -19,12 +17,10 @@ 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'; @@ -47,20 +43,6 @@ 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 { @@ -76,12 +58,6 @@ export interface KoreaderProgressApplyResult { 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 { @@ -105,7 +81,6 @@ 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) { @@ -201,7 +176,7 @@ export class KoreaderService { throw new NotFoundException('Book not found for the given document hash'); } - const applied = await this.applyProgressForResolvedFile(userId, bookFile, { + await this.applyProgressForResolvedFile(userId, bookFile, { percentage: data.percentage, progress: data.progress, device, @@ -209,8 +184,6 @@ 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`, ); @@ -227,20 +200,9 @@ export class KoreaderService { 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(() => {}); @@ -269,83 +231,6 @@ export class KoreaderService { 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`, - ); - } - } - /** * Decides whether a pushed position may move shared progress while a reset is outstanding. * 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 8cf731151..b78345ee9 100644 --- a/server/src/modules/reading-session/reading-session.repository.test.ts +++ b/server/src/modules/reading-session/reading-session.repository.test.ts @@ -697,29 +697,42 @@ describe('ReadingSessionRepository - deleteSessionByBook', () => { }); }); -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> = [], +describe('ReadingSessionRepository - deleteLegacyKoreaderSyncEstimatesBatch', () => { + function makeCleanupHarness( + rows: Array<{ + id: number; + userId: number; + libraryId: number; + startedAt: Date; + endedAt: Date; + durationSeconds: number; + progressDelta: number | null; + userSettings: Record; + }>, ) { - 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 initialLimit = vi.fn().mockResolvedValue(rows); + const initialSelect = { + from: vi.fn().mockReturnValue({ + innerJoin: vi.fn().mockReturnValue({ + innerJoin: vi.fn().mockReturnValue({ + where: vi.fn().mockReturnValue({ orderBy: vi.fn().mockReturnValue({ limit: initialLimit }) }), + }), + }), + }), + }; + const survivingSessions = [ + { + startedAt: new Date('2026-07-01T10:00:00.000Z'), + endedAt: new Date('2026-07-01T10:05:00.000Z'), + durationSeconds: 300, + progressDelta: 1, + }, + ]; + const recomputeWhere = vi.fn().mockResolvedValue(survivingSessions); + const recomputeSelect = { + from: vi.fn().mockReturnValue({ innerJoin: vi.fn().mockReturnValue({ where: recomputeWhere }) }), + }; + const select = vi.fn().mockReturnValue(recomputeSelect).mockReturnValueOnce(initialSelect); const deleteWhere = vi.fn().mockResolvedValue(undefined); const deleteFn = vi.fn().mockReturnValue({ where: deleteWhere }); const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) }); @@ -727,63 +740,68 @@ describe('ReadingSessionRepository - deleteSupersededSyncEstimates', () => { if (table === userReadingDailyStats) return { values: dailyValues }; throw new Error('Unexpected table in insert'); }); + const tx = { select, delete: deleteFn, insert, execute: vi.fn().mockResolvedValue(undefined) }; + const transaction = vi.fn(async (callback: (trx: typeof tx) => Promise) => callback(tx)); - 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 }; + return { + repo: new ReadingSessionRepository({ transaction } as never), + deleteFn, + dailyValues, + initialLimit, + recomputeWhere, + transaction, + }; } - 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('deletes one bounded batch and rebuilds the affected daily totals', async () => { + const { repo, deleteFn, dailyValues, initialLimit } = makeCleanupHarness([ + { + id: 9, + userId: 7, + libraryId: 3, + startedAt: new Date('2026-07-01T09:45:00.000Z'), + endedAt: new Date('2026-07-01T10:00:00.000Z'), + durationSeconds: 900, + progressDelta: 4, + userSettings: { timezone: 'UTC' }, + }, + ]); - 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 }); + await expect(repo.deleteLegacyKoreaderSyncEstimatesBatch(500)).resolves.toEqual({ deleted: 1 }); + expect(initialLimit).toHaveBeenCalledWith(500); expect(deleteFn).toHaveBeenCalledWith(readingSessions); + expect(deleteFn).toHaveBeenCalledWith(userReadingDailyStats); expect(dailyValues).toHaveBeenCalledWith([ - expect.objectContaining({ day: '2026-04-15', readingSeconds: 1440, progressDelta: 3, sessionsCount: 1 }), + expect.objectContaining({ userId: 7, libraryId: 3, day: '2026-07-01', readingSeconds: 300, 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')]], [[]]); + it('does not rewrite daily totals when no legacy estimates remain', async () => { + const { repo, deleteFn, dailyValues, transaction } = makeCleanupHarness([]); - 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:'); + await expect(repo.deleteLegacyKoreaderSyncEstimatesBatch(500)).resolves.toEqual({ deleted: 0 }); + expect(deleteFn).not.toHaveBeenCalled(); + expect(dailyValues).not.toHaveBeenCalled(); expect(transaction).toHaveBeenCalledOnce(); }); + + it('recomputes sparse history in bounded date ranges', async () => { + const row = (id: number, day: string) => ({ + id, + userId: 7, + libraryId: 3, + startedAt: new Date(`${day}T09:45:00.000Z`), + endedAt: new Date(`${day}T10:00:00.000Z`), + durationSeconds: 900, + progressDelta: 4, + userSettings: { timezone: 'UTC' }, + }); + const { repo, recomputeWhere } = makeCleanupHarness([row(9, '2024-01-01'), row(10, '2026-07-01')]); + + await repo.deleteLegacyKoreaderSyncEstimatesBatch(500); + + expect(recomputeWhere).toHaveBeenCalledTimes(2); + }); }); diff --git a/server/src/modules/reading-session/reading-session.repository.ts b/server/src/modules/reading-session/reading-session.repository.ts index a5b0984b5..5fd5fc6f2 100644 --- a/server/src/modules/reading-session/reading-session.repository.ts +++ b/server/src/modules/reading-session/reading-session.repository.ts @@ -17,7 +17,7 @@ import { splitReadingSessionByDay, type ReadingDailyStatsSegment, } from '../../common/utils/reading-daily-stats.utils'; -import { toDateKeyInTimeZone } from '../../common/utils/timezone.utils'; +import { resolveTimeZone, toDateKeyInTimeZone } from '../../common/utils/timezone.utils'; import { DB } from '../../db'; import * as schema from '../../db/schema'; import { bookFiles, books, readingSessionSyncCursors, readingSessions, userReadingDailyStats } from '../../db/schema'; @@ -27,6 +27,30 @@ type Tx = Parameters[0]>[0]; const MIN_READING_SESSION_SECONDS = 10; const ESTIMATE_CLEANUP_PAGE_SIZE = 500; +const CLEANUP_DAILY_STATS_MAX_SPAN_DAYS = 31; + +function groupDateKeysByMaxSpan(days: Iterable): string[][] { + const sorted = [...new Set(days)].sort(); + const groups: string[][] = []; + let group: string[] = []; + let firstDayNumber = 0; + + for (const day of sorted) { + const dayNumber = Math.floor(Date.parse(`${day}T00:00:00.000Z`) / 86_400_000); + if (group.length === 0 || dayNumber - firstDayNumber < CLEANUP_DAILY_STATS_MAX_SPAN_DAYS) { + if (group.length === 0) firstDayNumber = dayNumber; + group.push(day); + continue; + } + + groups.push(group); + group = [day]; + firstDayNumber = dayNumber; + } + + if (group.length > 0) groups.push(group); + return groups; +} export type SaveReadingSessionResult = | { kind: 'saved' } @@ -578,94 +602,66 @@ 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 }> { + async deleteLegacyKoreaderSyncEstimatesBatch(limit: number): Promise<{ deleted: number }> { return this.db.transaction(async (tx) => { - const daysByLibrary = new Map>(); - let deleted = 0; - let cursor = 0; + // The exact id shape was reserved by the removed estimator. Plugin-measured sessions use + // the `kor:` namespace, so this cannot match supported KOReader telemetry. + const rows = await tx + .select({ + id: readingSessions.id, + userId: readingSessions.userId, + libraryId: books.libraryId, + startedAt: readingSessions.startedAt, + endedAt: readingSessions.endedAt, + durationSeconds: readingSessions.durationSeconds, + progressDelta: readingSessions.progressDelta, + userSettings: schema.users.settings, + }) + .from(readingSessions) + .innerJoin(books, eq(books.id, readingSessions.bookId)) + .innerJoin(schema.users, eq(schema.users.id, readingSessions.userId)) + .where(and(eq(readingSessions.source, 'koreader'), sql`${readingSessions.sessionId} ~ ${'^ks-[0-9a-f]{12}-[0-9a-f]{32}$'}`)) + .orderBy(readingSessions.id) + .limit(limit); - 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) return { deleted: 0 }; - 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); + const affected = new Map }>(); + for (const row of rows) { + const timeZone = resolveTimeZone((row.userSettings as { timezone?: unknown } | null)?.timezone, 'UTC'); + const key = `${row.userId}:${row.libraryId}`; + const group = affected.get(key) ?? { userId: row.userId, libraryId: row.libraryId, timeZone, days: new Set() }; + for (const day of getReadingSessionDayKeys( + { + startedAt: row.startedAt, + endedAt: row.endedAt, + durationSeconds: row.durationSeconds, + progressDelta: row.progressDelta ?? null, + }, + timeZone, + )) { + group.days.add(day); } - - // 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; + affected.set(key, group); } - // 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); + await tx.delete(readingSessions).where( + inArray( + readingSessions.id, + rows.map((row) => row.id), + ), + ); + + const groups = [...affected.values()].sort((left, right) => left.userId - right.userId || left.libraryId - right.libraryId); + for (const group of groups) { + // Keep each recompute's database range bounded even when one batch contains sparse + // sessions from years apart. + for (const days of groupDateKeysByMaxSpan(group.days)) { + await this.recomputeDailyStats(tx, group.userId, group.libraryId, days, group.timeZone); + } } - return { deleted }; + return { deleted: rows.length }; }); } 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 ff426d89e..2ee961e24 100644 --- a/server/src/modules/reading-session/reading-session.service.test.ts +++ b/server/src/modules/reading-session/reading-session.service.test.ts @@ -497,67 +497,3 @@ 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 14aa02afa..ee02a88dd 100644 --- a/server/src/modules/reading-session/reading-session.service.ts +++ b/server/src/modules/reading-session/reading-session.service.ts @@ -16,7 +16,6 @@ import { type ReadingSessionSyncOptions, type RecordCumulativeReadingSessionParams, type RecordCumulativeReadingSessionResult, - type SaveReadingSessionResult, } from './reading-session.repository'; @Injectable() @@ -112,55 +111,6 @@ 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; - } - async recordCumulativeSyncedSession(params: RecordCumulativeReadingSessionParams): Promise { const result = await this.repo.recordCumulativeSyncedSession(params); @@ -181,31 +131,8 @@ export class ReadingSessionService { 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 deleteLegacyKoreaderSyncEstimatesBatch(limit: number): Promise<{ deleted: number }> { + return this.repo.deleteLegacyKoreaderSyncEstimatesBatch(limit); } async createManualSession(bookId: number, dto: CreateManualReadingSessionDto, user: RequestUser): Promise { diff --git a/server/test/koreader-progress-position.e2e-spec.ts b/server/test/koreader-progress-position.e2e-spec.ts index eeda86764..5f037be93 100644 --- a/server/test/koreader-progress-position.e2e-spec.ts +++ b/server/test/koreader-progress-position.e2e-spec.ts @@ -1,7 +1,8 @@ import { randomUUID } from 'crypto'; -import { eq } from 'drizzle-orm'; +import { and, eq } from 'drizzle-orm'; import * as schema from '../src/db/schema'; +import { ReadingSessionService } from '../src/modules/reading-session/reading-session.service'; import { createCbzFixture, createEpubFixture } from './e2e/reader-state-isolation/reader-state-isolation-fixture-builder'; import { authHeader, @@ -18,6 +19,7 @@ import { const KOREADER_USERNAME = `progress-position-device-${randomUUID().slice(0, 8)}`; const KOREADER_PASSWORD = 'ProgressPositionPass123'; const XPOINTER = '/body/DocFragment[8]/body/p[12]/text().0'; +const DEVICE_ID = 'progress-regression-device'; /** * KOReader reports a paged document's position as a page number and a reflowable one's as an @@ -47,7 +49,13 @@ describe('KOReader progress position routing (e2e)', { timeout: 180_000 }, () => method: 'PUT', url: '/api/v1/koreader/syncs/progress', headers: deviceHeaders(), - payload: { document: hash, percentage, ...(progress === undefined ? {} : { progress }) }, + payload: { + document: hash, + percentage, + device: 'Regression device', + device_id: DEVICE_ID, + ...(progress === undefined ? {} : { progress }), + }, }); expect(response.statusCode).toBe(200); } @@ -159,6 +167,8 @@ describe('KOReader progress position routing (e2e)', { timeout: 180_000 }, () => document: epubHash, percentage: 0.35, progress: XPOINTER, + device: 'Regression device', + device_id: DEVICE_ID, metadata: { filename: 'progress-position-book.epub', title: 'Progress Position Book', authors: 'Test Author' }, }, }); @@ -194,6 +204,61 @@ describe('KOReader progress position routing (e2e)', { timeout: 180_000 }, () => expect(stored?.percentage).toBeCloseTo(80, 5); }); + it('never turns KOSync progress pushes into reading sessions', async () => { + await syncFromDevice(epubHash, 0.4, XPOINTER); + await ctx.db + .update(schema.koreaderDeviceProgress) + .set({ updatedAt: new Date(Date.now() - 12 * 60 * 1000) }) + .where(and(eq(schema.koreaderDeviceProgress.bookFileId, epub.bookFileId), eq(schema.koreaderDeviceProgress.deviceId, DEVICE_ID))); + + await syncFromDevice(epubHash, 0.5, XPOINTER); + + const sessions = await ctx.db + .select({ id: schema.readingSessions.id }) + .from(schema.readingSessions) + .where(and(eq(schema.readingSessions.bookFileId, epub.bookFileId), eq(schema.readingSessions.source, 'koreader'))); + expect(sessions).toEqual([]); + }); + + it('removes only historical KOSync estimates during repair', async () => { + const [koreaderUser] = await ctx.db + .select({ userId: schema.koreaderUsers.userId }) + .from(schema.koreaderUsers) + .where(eq(schema.koreaderUsers.username, KOREADER_USERNAME)); + expect(koreaderUser).toBeDefined(); + + const legacySessionId = `ks-${'a'.repeat(12)}-${'b'.repeat(32)}`; + const measuredSessionId = 'kor:cleanup-supported'; + const webSessionId = 'web-cleanup-supported'; + const startedAt = new Date('2026-08-31T10:00:00.000Z'); + const endedAt = new Date('2026-08-31T10:10:00.000Z'); + await ctx.db.insert(schema.readingSessions).values( + [ + { sessionId: legacySessionId, source: 'koreader' as const }, + { sessionId: measuredSessionId, source: 'koreader' as const }, + { sessionId: webSessionId, source: 'web' as const }, + ].map((session) => ({ + ...session, + userId: koreaderUser!.userId, + bookId: epub.bookId, + bookFileId: epub.bookFileId, + startedAt, + endedAt, + durationSeconds: 600, + progressDelta: 2, + endProgress: 50, + })), + ); + + await expect(ctx.app.get(ReadingSessionService).deleteLegacyKoreaderSyncEstimatesBatch(500)).resolves.toEqual({ deleted: 1 }); + + const remaining = await ctx.db + .select({ sessionId: schema.readingSessions.sessionId }) + .from(schema.readingSessions) + .where(and(eq(schema.readingSessions.userId, koreaderUser!.userId), eq(schema.readingSessions.bookFileId, epub.bookFileId))); + expect(remaining.map((row) => row.sessionId).sort()).toEqual([measuredSessionId, webSessionId].sort()); + }); + it('does not route progress when the document hash matches different books', async () => { const comicProgressBefore = await storedProgress(comic.bookFileId); const epubProgressBefore = await storedProgress(epub.bookFileId);