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