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);