Migrations/bookorbit
1
0
Fork 0
mirror of https://github.com/bookorbit/bookorbit.git synced 2026-11-03 00:11:37 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(kobo): reconcile counter estimates on their counter interval (#1796)

Browse Source 
fix(kobo): reconcile counter estimates on their counter interval (#1796)

Stop counting offline Kobo reading twice. Record when each counter
estimate's interval began, split measured sessions across intervals by
wall-clock overlap, and subtract measured seconds instead of dropping the
estimate, so both arrival orders converge.

Closes #1667

Co-authored-by: neon <18559658+neonsolstice@users.noreply.github.com>
This commit is contained in:
François-Xavier Payet
2026-10-17 01:33:05 +02:00
committed by GitHub
parent d0da80e751
commit 7f12e8cd6d
11 changed files with 24707 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/e2e/suite-registry.mjs
+19
View File
@@ -1080,6 +1080,25 @@ export const E2E_SUITES = Object.freeze({
...SHARED_DB_AND_HELPER_PATHS,
],
},
"kobo-reading-time-reconciliation": {
id: "kobo-reading-time-reconciliation",
name: "Kobo Reading Time Reconciliation",
timeout: 45,
lane: "full",
description: "Kobo counter-derived estimate and measured session reconciliation suite",
vitestTarget: "test/kobo-reading-time-reconciliation.e2e-spec.ts",
junitOutput: `${TEST_RESULTS_DIR}/kobo-reading-time-reconciliation-e2e-junit.xml`,
prepareDedicatedDatabase: true,
useDedicatedDatabase: true,
changedPaths: [
"server/src/modules/kobo/**",
"server/src/modules/reading-session/**",
"server/src/common/utils/reading-daily-stats.utils.ts",
"server/test/kobo-reading-time-reconciliation.e2e-spec.ts",
"server/test/e2e/reader-state-isolation/**",
...SHARED_DB_AND_HELPER_PATHS,
],
},
"kobo-cover-identity": {
id: "kobo-cover-identity",
name: "Kobo Cover Identity",
server/src/db/migrations/0117_add_reading_session_sync_interval.sql
+1
View File
@@ -0,0 +1 @@
ALTER TABLE "reading_sessions" ADD COLUMN "sync_interval_started_at" timestamp with time zone;
server/src/db/migrations/meta/0117_snapshot.json
+23855
View File
File diff suppressed because it is too large Load Diff
server/src/db/migrations/meta/_journal.json
+7
View File
@@ -820,6 +820,13 @@
"when": 1792185442568,
"tag": "0116_regex_metadata",
"breakpoints": true
},
{
"idx": 117,
"version": "7",
"when": 1792191772317,
"tag": "0117_add_reading_session_sync_interval",
"breakpoints": true
}
]
}
server/src/db/schema/reader.ts
+2
View File
@@ -308,6 +308,8 @@ export const readingSessions = pgTable(
// Nullable: CBX with no percentage tracking may omit these.
progressDelta: real('progress_delta'),
endProgress: real('end_progress'),
// Counter-derived sync estimates only: the previous counter push, where the interval the estimate stands for begins.
syncIntervalStartedAt: timestamp('sync_interval_started_at', { withTimezone: true }),
},
(t) => [
uniqueIndex('rs_user_session_id_uidx').on(t.userId, t.sessionId),
server/src/modules/kobo/kobo-statistics-session.util.ts
+1 -1
View File
@@ -1,6 +1,6 @@
/**
* Identifiers for reading sessions derived from the reading counter a Kobo reports alongside
* its bookmark, rather than measured from the analytics events it no longer sends.
* its bookmark, rather than measured from the analytics events only some devices still send.
*
* A reading session carries no column saying where it came from beyond `source`, and `source`
* is `kobo` for both kinds. The prefix is what lets the derivation find, and exclude itself
server/src/modules/kobo/services/kobo-reading-state.service.ts
+9 -6
View File
@@ -312,16 +312,19 @@ export class KoboReadingStateService {
/**
* Turns the reading time a device reports alongside its bookmark into a reading session.
*
* Current Kobo firmware no longer emits the `LeaveContent` analytics events the reading log
* was built on, so `POST /v1/analytics/event` never arrives and every Kobo statistic reads
* zero however much the device is read. What the device does still send, on every state push,
* is its own `SpentReadingMinutes` - the counter it keeps as `content.TimeSpentReading` on
* device, and the only reading time these devices offer at all.
* Some Kobo devices no longer emit the `LeaveContent` analytics events the reading log was
* built on, so `POST /v1/analytics/event` never arrives and every Kobo statistic reads zero
* however much the device is read. What every device does still send, on every state push, is
* its own `SpentReadingMinutes` - the counter it keeps as `content.TimeSpentReading` on device,
* and the only reading time those devices offer at all. Others still send both: a Libra Colour
* on firmware 4.45.23792 does, and after offline reading it pushes the counter and the events
* together with its next sync, long after the reading.
*
* Each device has an independent durable cursor. Session insertion, daily aggregation, and
* cursor advancement share one transaction so a failed write is retried from the same counter.
* Measured analytics sessions and counter-derived sessions are serialized by the same lock and
* reconciled by overlap, regardless of which endpoint arrives first.
* reconciled on the counter interval each estimate stands for, regardless of which endpoint
* arrives first.
*
* Failures are swallowed. A device retries the entire push when the response is not a clean
* acknowledgement, so a session that cannot be stored must not take the bookmark down with it.
server/src/modules/reading-session/reading-session-cumulative.repository.test.ts
+209 -52
View File
@@ -1,3 +1,6 @@
import type { SQL } from 'drizzle-orm';
import { PgDialect } from 'drizzle-orm/pg-core';
import { bookFiles, readingSessionSyncCursors, readingSessions, userReadingDailyStats } from '../../db/schema';
import { ReadingSessionRepository, type RecordCumulativeReadingSessionParams } from './reading-session.repository';
@@ -10,19 +13,19 @@ function makeLimitChain(rows: unknown[]) {
function makeHarness(
options: {
cursor?: { counter: number; generation: number; lastModified: Date } | null;
measured?: boolean;
measuredSeconds?: number;
file?: { bookId: number; libraryId: number } | null;
inserted?: boolean;
} = {},
) {
const cursorChain = makeLimitChain(options.cursor ? [options.cursor] : []);
const measuredChain = makeLimitChain(options.measured ? [{ id: 88 }] : []);
const measuredWhere = vi.fn().mockResolvedValue([{ seconds: options.measuredSeconds ?? 0 }]);
const fileChain = makeLimitChain(options.file === null ? [] : [options.file ?? { bookId: 46, libraryId: 9 }]);
const select = vi.fn().mockImplementation(() => ({
from: vi.fn().mockImplementation((table: unknown) => {
if (table === readingSessionSyncCursors) return { where: cursorChain.where };
if (table === readingSessions) return { where: measuredChain.where };
if (table === readingSessions) return { where: measuredWhere };
if (table === bookFiles) return { innerJoin: vi.fn().mockReturnValue({ where: fileChain.where }) };
throw new Error('Unexpected table in select');
}),
@@ -47,7 +50,11 @@ function makeHarness(
const transaction = vi.fn(async (callback: (trx: typeof tx) => Promise<unknown>) => callback(tx));
const repo = new ReadingSessionRepository({ transaction } as never);
return { repo, tx, transaction, cursorInsertValues, cursorUpdateSet, sessionValues, dailyValues };
return { repo, tx, transaction, cursorInsertValues, cursorUpdateSet, sessionValues, dailyValues, measuredWhere };
}
function boundParams(where: ReturnType<typeof vi.fn>): unknown[] {
return new PgDialect().sqlToQuery(where.mock.calls[0]![0] as SQL).params;
}
const params = (overrides: Partial<RecordCumulativeReadingSessionParams> = {}): RecordCumulativeReadingSessionParams => ({
@@ -85,6 +92,7 @@ describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
kind: 'saved',
durationSeconds: 3600,
progressDelta: 7,
sessionId: 'kst:30:53:2:860',
});
@@ -97,6 +105,7 @@ describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
sourceDeviceKey: '30',
sessionId: 'kst:30:53:2:860',
durationSeconds: 3600,
syncIntervalStartedAt: new Date('2026-08-19T10:00:00.000Z'),
}),
);
expect(cursorUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ counter: 860, generation: 2 }));
@@ -117,10 +126,10 @@ describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
expect(sessionValues).not.toHaveBeenCalled();
});
it('advances without an estimate when overlapping measured analytics already exist', async () => {
it('advances without an estimate when measured analytics already cover the counter delta', async () => {
const { repo, cursorUpdateSet, sessionValues } = makeHarness({
cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
measured: true,
measuredSeconds: 3590,
});
await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
@@ -132,6 +141,41 @@ describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
expect(sessionValues).not.toHaveBeenCalled();
});
it('keeps the part of the counter delta the device did not measure', async () => {
const { repo, sessionValues, dailyValues } = makeHarness({
cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
measuredSeconds: 1500,
});
await expect(repo.recordCumulativeSyncedSession(params())).resolves.toEqual({
kind: 'saved',
durationSeconds: 2100,
progressDelta: 4.08,
sessionId: 'kst:30:53:0:860',
});
expect(sessionValues).toHaveBeenCalledWith(
expect.objectContaining({
startedAt: new Date('2026-08-20T18:55:00.000Z'),
endedAt: new Date('2026-08-20T19:30:00.000Z'),
durationSeconds: 2100,
progressDelta: 4.08,
}),
);
expect(dailyValues).toHaveBeenCalledWith([expect.objectContaining({ readingSeconds: 2100, progressDelta: 4.08 })]);
});
it('measures against the whole counter interval rather than the minutes before the push', async () => {
const { repo, measuredWhere } = makeHarness({
cursor: { counter: 800, generation: 0, lastModified: new Date('2026-08-19T10:00:00.000Z') },
});
await repo.recordCumulativeSyncedSession(params());
expect(boundParams(measuredWhere)).toEqual(expect.arrayContaining(['2026-08-19T10:00:00.000Z', '2026-08-20T19:30:00.000Z']));
expect(boundParams(measuredWhere)).not.toContain('2026-08-20T18:30:00.000Z');
});
it('starts a new generation when a newer device counter resets', async () => {
const { repo, cursorUpdateSet, sessionValues } = makeHarness({
cursor: { counter: 860, generation: 4, lastModified: new Date('2026-08-19T10:00:00.000Z') },
@@ -167,42 +211,42 @@ describe('ReadingSessionRepository.recordCumulativeSyncedSession', () => {
});
describe('ReadingSessionRepository measured-session reconciliation', () => {
it('stores the device key and removes an overlapping estimate in the same transaction', async () => {
type EstimateRow = { id: number; startedAt: Date; endedAt: Date; durationSeconds: number; progressDelta: number | null };
type IntervalEstimateRow = EstimateRow & { syncIntervalStartedAt: Date };
type SurvivingRow = Omit<EstimateRow, 'id'>;
const estimateLookup = (rows: EstimateRow[]) => ({
from: vi.fn().mockReturnValue({
where: vi.fn().mockReturnValue({ orderBy: vi.fn().mockReturnValue({ limit: vi.fn().mockResolvedValue(rows) }) }),
}),
});
const intervalLookup = (rows: IntervalEstimateRow[]) => ({
from: vi.fn().mockReturnValue({ where: vi.fn().mockReturnValue({ orderBy: vi.fn().mockResolvedValue(rows) }) }),
});
const survivingLookup = (rows: SurvivingRow[]) => ({
from: vi.fn().mockReturnValue({ innerJoin: vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(rows) }) }),
});
const existingLookup = (rows: { id: number; durationSeconds: number; progressDelta: number | null }[]) => ({
from: vi.fn().mockReturnValue({ where: vi.fn().mockReturnValue({ limit: vi.fn().mockResolvedValue(rows) }) }),
});
const measured = {
startedAt: new Date('2026-08-20T19:20:00.000Z'),
endedAt: new Date('2026-08-20T19:30:00.000Z'),
durationSeconds: 600,
progressDelta: 2,
};
function makeMeasuredHarness(selects: unknown[], options: { inserted?: boolean } = {}) {
const outerLimit = vi.fn().mockResolvedValue([{ bookId: 46, libraryId: 9 }]);
const outerSelect = vi.fn().mockReturnValue({
from: vi.fn().mockReturnValue({ innerJoin: vi.fn().mockReturnValue({ where: vi.fn().mockReturnValue({ limit: outerLimit }) }) }),
});
const estimate = {
id: 20,
startedAt: new Date('2026-08-20T18:30:00.000Z'),
endedAt: new Date('2026-08-20T19:30:00.000Z'),
durationSeconds: 3600,
progressDelta: 7,
};
const estimateSelect = {
from: vi.fn().mockReturnValue({
where: vi.fn().mockReturnValue({ orderBy: vi.fn().mockReturnValue({ limit: vi.fn().mockResolvedValue([estimate]) }) }),
}),
};
const survivingSelect = {
from: vi.fn().mockReturnValue({
innerJoin: vi.fn().mockReturnValue({
where: vi.fn().mockResolvedValue([
{
startedAt: new Date('2026-08-20T19:20:00.000Z'),
endedAt: new Date('2026-08-20T19:30:00.000Z'),
durationSeconds: 600,
progressDelta: 2,
},
]),
}),
}),
};
const txSelect = vi.fn().mockReturnValueOnce(estimateSelect).mockReturnValueOnce(survivingSelect);
const txSelect = vi.fn();
for (const select of selects) txSelect.mockReturnValueOnce(select);
const sessionValues = vi.fn().mockReturnValue({
onConflictDoNothing: vi.fn().mockReturnValue({ returning: vi.fn().mockResolvedValue([{ id: 21 }]) }),
onConflictDoNothing: vi.fn().mockReturnValue({ returning: vi.fn().mockResolvedValue(options.inserted === false ? [] : [{ id: 21 }]) }),
});
const dailyValues = vi.fn().mockReturnValue({ onConflictDoUpdate: vi.fn().mockResolvedValue(undefined) });
const insert = vi.fn().mockImplementation((table: unknown) => {
@@ -210,32 +254,145 @@ describe('ReadingSessionRepository measured-session reconciliation', () => {
if (table === userReadingDailyStats) return { values: dailyValues };
throw new Error('Unexpected table in insert');
});
const updateSet = vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) });
const update = vi.fn().mockReturnValue({ set: updateSet });
const deleteFn = vi.fn().mockReturnValue({ where: vi.fn().mockResolvedValue(undefined) });
const tx = { execute: vi.fn().mockResolvedValue(undefined), insert, select: txSelect, delete: deleteFn };
const tx = { execute: vi.fn().mockResolvedValue(undefined), insert, update, select: txSelect, delete: deleteFn };
const transaction = vi.fn(async (callback: (trx: typeof tx) => Promise<unknown>) => callback(tx));
const repo = new ReadingSessionRepository({ select: outerSelect, transaction } as never);
await expect(
repo.saveSession(
7,
53,
'measured-event',
new Date('2026-08-20T19:20:00.000Z'),
new Date('2026-08-20T19:30:00.000Z'),
600,
2,
29,
'kobo',
'UTC',
{ sourceDeviceKey: '30', estimateSessionIdPrefix: 'kst:30:' },
),
).resolves.toEqual({ kind: 'saved' });
const saveMeasured = () =>
repo.saveSession(7, 53, 'measured-event', measured.startedAt, measured.endedAt, measured.durationSeconds, 2, 29, 'kobo', 'UTC', {
sourceDeviceKey: '30',
estimateSessionIdPrefix: 'kst:30:',
});
return { saveMeasured, transaction, txSelect, sessionValues, dailyValues, updateSet, deleteFn };
}
it('stores the device key and removes an overlapping estimate that predates interval tracking', async () => {
const legacyEstimate = {
id: 20,
startedAt: new Date('2026-08-20T18:30:00.000Z'),
endedAt: new Date('2026-08-20T19:30:00.000Z'),
durationSeconds: 3600,
progressDelta: 7,
};
const { saveMeasured, transaction, sessionValues, dailyValues, updateSet, deleteFn } = makeMeasuredHarness([
estimateLookup([legacyEstimate]),
intervalLookup([]),
survivingLookup([measured]),
]);
await expect(saveMeasured()).resolves.toEqual({ kind: 'saved' });
expect(transaction).toHaveBeenCalledOnce();
expect(sessionValues).toHaveBeenCalledWith(expect.objectContaining({ sessionId: 'measured-event', sourceDeviceKey: '30' }));
expect(deleteFn).toHaveBeenCalledWith(readingSessions);
expect(updateSet).not.toHaveBeenCalled();
expect(dailyValues).toHaveBeenLastCalledWith([
expect.objectContaining({ day: '2026-08-20', readingSeconds: 600, progressDelta: 2, sessionsCount: 1 }),
]);
});
it('removes the estimate of a later push once the measured session covers its counter interval', async () => {
const lateEstimate = {
id: 20,
startedAt: new Date('2026-08-21T06:50:00.000Z'),
endedAt: new Date('2026-08-21T07:00:00.000Z'),
durationSeconds: 600,
progressDelta: 2,
syncIntervalStartedAt: new Date('2026-08-20T07:00:00.000Z'),
};
const { saveMeasured, dailyValues, updateSet, deleteFn } = makeMeasuredHarness([
estimateLookup([]),
intervalLookup([lateEstimate]),
survivingLookup([measured]),
]);
await expect(saveMeasured()).resolves.toEqual({ kind: 'saved' });
expect(deleteFn).toHaveBeenCalledWith(readingSessions);
expect(deleteFn).toHaveBeenCalledWith(userReadingDailyStats);
expect(updateSet).not.toHaveBeenCalled();
expect(dailyValues).toHaveBeenLastCalledWith([
expect.objectContaining({ day: '2026-08-20', readingSeconds: 600, progressDelta: 2, sessionsCount: 1 }),
]);
});
it('shortens an estimate the measured session covers only in part', async () => {
const lateEstimate = {
id: 20,
startedAt: new Date('2026-08-21T06:00:00.000Z'),
endedAt: new Date('2026-08-21T07:00:00.000Z'),
durationSeconds: 3600,
progressDelta: 6,
syncIntervalStartedAt: new Date('2026-08-20T07:00:00.000Z'),
};
const shortened = {
startedAt: new Date('2026-08-21T06:10:00.000Z'),
endedAt: new Date('2026-08-21T07:00:00.000Z'),
durationSeconds: 3000,
progressDelta: 5,
};
const { saveMeasured, dailyValues, updateSet, deleteFn } = makeMeasuredHarness([
estimateLookup([]),
intervalLookup([lateEstimate]),
survivingLookup([measured, shortened]),
]);
await expect(saveMeasured()).resolves.toEqual({ kind: 'saved' });
expect(updateSet).toHaveBeenCalledWith({ startedAt: shortened.startedAt, durationSeconds: 3000, progressDelta: 5 });
expect(deleteFn).not.toHaveBeenCalledWith(readingSessions);
expect(dailyValues).toHaveBeenLastCalledWith([
expect.objectContaining({ day: '2026-08-21', readingSeconds: 3000, progressDelta: 5, sessionsCount: 1 }),
]);
});
it('splits a measured session across the estimates whose counter intervals it overlaps', async () => {
const pushedMidSession = new Date('2026-08-20T19:25:00.000Z');
const before = {
id: 20,
startedAt: new Date('2026-08-20T19:05:00.000Z'),
endedAt: pushedMidSession,
durationSeconds: 1200,
progressDelta: null,
syncIntervalStartedAt: new Date('2026-08-20T07:00:00.000Z'),
};
const after = {
id: 22,
startedAt: new Date('2026-08-21T06:40:00.000Z'),
endedAt: new Date('2026-08-21T07:00:00.000Z'),
durationSeconds: 1200,
progressDelta: null,
syncIntervalStartedAt: pushedMidSession,
};
const { saveMeasured, updateSet, deleteFn } = makeMeasuredHarness([
estimateLookup([]),
intervalLookup([before, after]),
survivingLookup([measured]),
]);
await expect(saveMeasured()).resolves.toEqual({ kind: 'saved' });
expect(updateSet).toHaveBeenCalledTimes(2);
expect(updateSet).toHaveBeenNthCalledWith(1, { startedAt: new Date('2026-08-20T19:10:00.000Z'), durationSeconds: 900, progressDelta: null });
expect(updateSet).toHaveBeenNthCalledWith(2, { startedAt: new Date('2026-08-21T06:45:00.000Z'), durationSeconds: 900, progressDelta: null });
expect(deleteFn).not.toHaveBeenCalledWith(readingSessions);
});
it('takes nothing off an estimate when the measured session is a replay', async () => {
const { saveMeasured, txSelect, updateSet, deleteFn } = makeMeasuredHarness(
[existingLookup([{ id: 21, durationSeconds: 600, progressDelta: 2 }]), estimateLookup([])],
{ inserted: false },
);
await expect(saveMeasured()).resolves.toEqual({ kind: 'skipped', reason: 'duplicate_session_id' });
expect(txSelect).toHaveBeenCalledTimes(2);
expect(updateSet).toHaveBeenCalledOnce();
expect(updateSet).toHaveBeenCalledWith({ sourceDeviceKey: '30' });
expect(deleteFn).not.toHaveBeenCalled();
});
});
server/src/modules/reading-session/reading-session.repository.ts
+124 -39
View File
@@ -1,5 +1,5 @@
import { Inject, Injectable } from '@nestjs/common';
import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, isNull, like, lt, lte, max, min, notLike, sql } from 'drizzle-orm';
import { and, asc, count, desc, eq, gt, gte, inArray, isNotNull, isNull, like, lt, lte, max, min, notLike, sql, type SQL } from 'drizzle-orm';
import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
import type {
@@ -36,6 +36,32 @@ function toReadingSessionType(value: string | null | undefined): ReadingSessionT
}
const ESTIMATE_CLEANUP_PAGE_SIZE = 500;
const CLEANUP_DAILY_STATS_MAX_SPAN_DAYS = 31;
// Cumulative device counters tick in whole minutes, so less than one tick left is rounding, not reading.
const SYNC_COUNTER_RESOLUTION_SECONDS = 60;
// A device can push its counter while a book is open, on waking or on a manual sync, so a counter
// interval can begin or end inside a measured session. The session's seconds are split by how much
// of its wall-clock span falls in each interval rather than given whole to the one holding its start.
// `measuredSecondsInWindow` and `measuredShareOfWindow` must apply the same rule.
function measuredSecondsInWindow(windowStart: Date, windowEnd: Date): SQL<number> {
return sql<number>`case when ${readingSessions.endedAt} > ${readingSessions.startedAt} then round(
${readingSessions.durationSeconds} * greatest(0, extract(epoch from (
least(${readingSessions.endedAt}, ${windowEnd}::timestamptz) - greatest(${readingSessions.startedAt}, ${windowStart}::timestamptz)
))) / extract(epoch from (${readingSessions.endedAt} - ${readingSessions.startedAt}))
) else ${readingSessions.durationSeconds} end`;
}
function measuredShareOfWindow(session: { seconds: number; startedAt: Date; endedAt: Date }, windowStart: Date, windowEnd: Date): number {
const spanMs = session.endedAt.getTime() - session.startedAt.getTime();
if (spanMs <= 0) return session.seconds;
const overlapMs = Math.min(session.endedAt.getTime(), windowEnd.getTime()) - Math.max(session.startedAt.getTime(), windowStart.getTime());
return Math.round((session.seconds * Math.max(0, overlapMs)) / spanMs);
}
function scaleProgressDelta(progressDelta: number | null, keptSeconds: number, totalSeconds: number): number | null {
if (progressDelta === null || keptSeconds === totalSeconds) return progressDelta;
return Math.round(((progressDelta * keptSeconds) / totalSeconds) * 100) / 100;
}
function groupDateKeysByMaxSpan(days: Iterable<string>): string[][] {
const sorted = [...new Set(days)].sort();
@@ -70,7 +96,7 @@ export type SaveReadingSessionResult =
export type RecordCumulativeReadingSessionResult =
| { kind: 'baseline' | 'unchanged' | 'reset' | 'stale' }
| { kind: 'saved'; durationSeconds: number; sessionId: string }
| { kind: 'saved'; durationSeconds: number; progressDelta: number | null; sessionId: string }
| {
kind: 'skipped';
reason: 'book_file_not_found' | 'duplicate_session_id' | 'measured_session_present';
@@ -236,7 +262,7 @@ export class ReadingSessionRepository {
}
if (sync?.estimateSessionIdPrefix) {
await this.deleteOverlappingSyncEstimates(tx, {
await this.reconcileSyncEstimates(tx, {
userId,
bookId,
libraryId,
@@ -246,6 +272,7 @@ export class ReadingSessionRepository {
startedAt,
endedAt,
timeZone,
newlyMeasuredSeconds: result.kind === 'saved' ? durationSeconds : result.kind === 'superseded' ? result.addedSeconds : 0,
});
}
@@ -300,11 +327,14 @@ export class ReadingSessionRepository {
return { kind: 'reset' };
}
const durationSeconds = (params.counter - cursor.counter) * 60;
const startedAt = new Date(params.endedAt.getTime() - durationSeconds * 1000);
// The counter covers everything read since the previous push, however long ago that reading
// was: a device that read offline only pushes with its next sync. The share of measured
// sessions that falls in that interval is subtracted rather than taken as proof the whole
// delta is covered, so reading the device did not measure is kept.
const counterSeconds = (params.counter - cursor.counter) * 60;
const [measured] = await tx
.select({ id: readingSessions.id })
.select({ seconds: sql<number>`coalesce(sum(${measuredSecondsInWindow(cursor.lastModified, params.endedAt)}), 0)::int` })
.from(readingSessions)
.where(
and(
@@ -314,16 +344,19 @@ export class ReadingSessionRepository {
eq(readingSessions.sourceDeviceKey, params.sourceDeviceKey),
notLike(readingSessions.sessionId, `${params.sessionIdPrefix}%`),
lt(readingSessions.startedAt, params.endedAt),
gt(readingSessions.endedAt, startedAt),
gt(readingSessions.endedAt, cursor.lastModified),
),
)
.limit(1);
);
if (measured) {
const durationSeconds = counterSeconds - (measured?.seconds ?? 0);
if (durationSeconds < SYNC_COUNTER_RESOLUTION_SECONDS) {
await this.updateSyncCursor(tx, params, cursor.generation);
return { kind: 'skipped', reason: 'measured_session_present' };
}
const startedAt = new Date(params.endedAt.getTime() - durationSeconds * 1000);
const progressDelta = scaleProgressDelta(params.progressDelta, durationSeconds, counterSeconds);
if (params.bookFileId === null) return { kind: 'skipped', reason: 'book_file_not_found' };
const sessionId = params.buildSessionId(params.bookFileId, cursor.generation, params.counter);
@@ -349,8 +382,9 @@ export class ReadingSessionRepository {
startedAt,
endedAt: params.endedAt,
durationSeconds,
progressDelta: params.progressDelta,
progressDelta,
endProgress: params.endProgress,
syncIntervalStartedAt: cursor.lastModified,
})
.onConflictDoNothing({ target: [readingSessions.userId, readingSessions.sessionId] })
.returning({ id: readingSessions.id });
@@ -364,11 +398,11 @@ export class ReadingSessionRepository {
startedAt,
endedAt: params.endedAt,
durationSeconds,
progressDelta: params.progressDelta,
progressDelta,
timeZone: params.timeZone,
});
return { kind: 'saved', durationSeconds, sessionId };
return { kind: 'saved', durationSeconds, progressDelta, sessionId };
});
}
@@ -823,7 +857,20 @@ export class ReadingSessionRepository {
await tx.execute(sql`select pg_advisory_xact_lock(${userId}::int, hashtext(${`${sourceDeviceKey}:${bookId}`})::int)`);
}
private async deleteOverlappingSyncEstimates(
/**
* Reconciles the counter-derived estimates of a device with a session it also measured.
*
* An estimate stands for whatever was read between two counter pushes, and it is dated on the
* minutes leading up to the push. That is only when the reading happened if the device pushed
* right after it; after offline reading the push comes with the next sync. So the measured
* session is matched to every estimate whose counter interval it overlaps, and each takes the
* share of its seconds that falls in that interval. They are taken once, when first stored: a
* replay takes nothing, and the shares never add up to more than the session. What is left
* stays as reading the device did not measure, unless it is below the counter resolution.
*
* Estimates written before the interval was recorded keep the overlap rule they were created under.
*/
private async reconcileSyncEstimates(
tx: Tx,
params: {
userId: number;
@@ -835,28 +882,37 @@ export class ReadingSessionRepository {
startedAt: Date;
endedAt: Date;
timeZone: string;
newlyMeasuredSeconds: number;
},
): Promise<void> {
const affectedDays = new Set<string>();
const estimateColumns = {
id: readingSessions.id,
startedAt: readingSessions.startedAt,
endedAt: readingSessions.endedAt,
durationSeconds: readingSessions.durationSeconds,
progressDelta: readingSessions.progressDelta,
};
const sameDeviceEstimate = and(
eq(readingSessions.userId, params.userId),
eq(readingSessions.bookId, params.bookId),
eq(readingSessions.source, params.source),
eq(readingSessions.sourceDeviceKey, params.sourceDeviceKey),
like(readingSessions.sessionId, `${params.estimateSessionIdPrefix}%`),
);
const collectAffectedDays = (row: { startedAt: Date; endedAt: Date; durationSeconds: number; progressDelta: number | null }) => {
for (const day of getReadingSessionDayKeys(row, params.timeZone)) affectedDays.add(day);
};
let cursor = 0;
for (;;) {
const rows = await tx
.select({
id: readingSessions.id,
startedAt: readingSessions.startedAt,
endedAt: readingSessions.endedAt,
durationSeconds: readingSessions.durationSeconds,
progressDelta: readingSessions.progressDelta,
})
.select(estimateColumns)
.from(readingSessions)
.where(
and(
eq(readingSessions.userId, params.userId),
eq(readingSessions.bookId, params.bookId),
eq(readingSessions.source, params.source),
eq(readingSessions.sourceDeviceKey, params.sourceDeviceKey),
like(readingSessions.sessionId, `${params.estimateSessionIdPrefix}%`),
sameDeviceEstimate,
isNull(readingSessions.syncIntervalStartedAt),
lt(readingSessions.startedAt, params.endedAt),
gt(readingSessions.endedAt, params.startedAt),
gt(readingSessions.id, cursor),
@@ -868,19 +924,7 @@ export class ReadingSessionRepository {
if (rows.length === 0) break;
cursor = rows[rows.length - 1]!.id;
for (const row of rows) {
for (const day of getReadingSessionDayKeys(
{
startedAt: row.startedAt,
endedAt: row.endedAt,
durationSeconds: row.durationSeconds,
progressDelta: row.progressDelta ?? null,
},
params.timeZone,
)) {
affectedDays.add(day);
}
}
for (const row of rows) collectAffectedDays(row);
await tx.delete(readingSessions).where(
and(
@@ -895,6 +939,47 @@ export class ReadingSessionRepository {
if (rows.length < ESTIMATE_CLEANUP_PAGE_SIZE) break;
}
if (params.newlyMeasuredSeconds > 0) {
const measured = { seconds: params.newlyMeasuredSeconds, startedAt: params.startedAt, endedAt: params.endedAt };
const overlapping = await tx
.select({ ...estimateColumns, syncIntervalStartedAt: readingSessions.syncIntervalStartedAt })
.from(readingSessions)
.where(
and(
sameDeviceEstimate,
isNotNull(readingSessions.syncIntervalStartedAt),
lt(readingSessions.syncIntervalStartedAt, params.endedAt),
gt(readingSessions.endedAt, params.startedAt),
),
)
.orderBy(asc(readingSessions.endedAt), asc(readingSessions.id));
for (const estimate of overlapping) {
const share = measuredShareOfWindow(measured, estimate.syncIntervalStartedAt!, estimate.endedAt);
if (share <= 0) continue;
const ownEstimate = and(eq(readingSessions.userId, params.userId), eq(readingSessions.id, estimate.id));
const remainingSeconds = estimate.durationSeconds - share;
collectAffectedDays(estimate);
if (remainingSeconds < SYNC_COUNTER_RESOLUTION_SECONDS) {
await tx.delete(readingSessions).where(ownEstimate);
} else {
const shortened = {
startedAt: new Date(estimate.endedAt.getTime() - remainingSeconds * 1000),
endedAt: estimate.endedAt,
durationSeconds: remainingSeconds,
progressDelta: scaleProgressDelta(estimate.progressDelta, remainingSeconds, estimate.durationSeconds),
};
await tx
.update(readingSessions)
.set({ startedAt: shortened.startedAt, durationSeconds: shortened.durationSeconds, progressDelta: shortened.progressDelta })
.where(ownEstimate);
collectAffectedDays(shortened);
}
}
}
if (affectedDays.size > 0) {
await this.recomputeDailyStats(tx, params.userId, params.libraryId, [...affectedDays], params.timeZone);
}
server/src/modules/reading-session/reading-session.service.ts
+1 -1
View File
@@ -127,7 +127,7 @@ export class ReadingSessionService {
durationSeconds: result.durationSeconds,
startedAt,
endedAt: params.endedAt,
progressDelta: params.progressDelta,
progressDelta: result.progressDelta,
endProgress: params.endProgress,
timezone: params.timeZone,
});
server/test/kobo-reading-time-reconciliation.e2e-spec.ts
+479
View File
@@ -0,0 +1,479 @@
import { randomUUID } from 'crypto';
import { and, asc, eq, inArray } from 'drizzle-orm';
import * as schema from '../src/db/schema';
import { createEpubFixture } from './e2e/reader-state-isolation/reader-state-isolation-fixture-builder';
import {
authHeader,
closeReaderStateIsolationE2EContext,
createLibraryWithFolder,
createReaderStateIsolationE2EContext,
locateBookByAbsolutePath,
triggerAndWaitForLibraryScan,
type CreatedLibrary,
type ReaderStateIsolationE2EContext,
} from './e2e/reader-state-isolation/reader-state-isolation-harness';
type KoboDevice = { id: number; token: string };
type Book = { bookId: number; bookFileId: number; entitlementId: string };
type MeasuredSession = { id: string; startedAt: string; endedAt: string; seconds: number };
const BOOK_COUNT = 13;
const RUN = randomUUID().slice(0, 8);
/** Day `n` of a calendar that starts on 2026-03-01, so every scenario owns its own dates. */
function day(n: number): string {
return new Date(Date.UTC(2026, 2, n)).toISOString().slice(0, 10);
}
function at(n: number, time: string): string {
return `${day(n)}T${time}.000Z`;
}
function measured(name: string, n: number, from: string, to: string): MeasuredSession {
const startedAt = at(n, from);
const endedAt = at(n, to);
return { id: `${RUN}-${name}`, startedAt, endedAt, seconds: (Date.parse(endedAt) - Date.parse(startedAt)) / 1000 };
}
describe('Kobo reading time reconciliation (e2e)', { timeout: 180_000 }, () => {
let ctx!: ReaderStateIsolationE2EContext;
let library!: CreatedLibrary;
let userId!: number;
let books!: Book[];
let deviceA!: KoboDevice;
let deviceB!: KoboDevice;
async function createDevice(name: string): Promise<KoboDevice> {
const response = await ctx.app.inject({
method: 'POST',
url: '/api/v1/kobo/devices',
headers: authHeader(ctx.adminToken),
payload: { name },
});
expect([200, 201]).toContain(response.statusCode);
return response.json() as KoboDevice;
}
async function drainLibrary(device: KoboDevice): Promise<void> {
let syncToken: string | undefined;
for (let page = 0; page < 10; page += 1) {
const response = await ctx.app.inject({
method: 'GET',
url: `/api/v1/kobo/${device.token}/v1/library/sync`,
headers: syncToken ? { 'x-kobo-synctoken': syncToken } : undefined,
});
expect(response.statusCode).toBe(200);
if (response.headers['x-kobo-sync'] !== 'continue') return;
syncToken = response.headers['x-kobo-synctoken'] as string;
}
throw new Error(`Kobo device ${device.id} did not finish syncing within ten pages`);
}
/** A state push carrying the device reading-time counter, as sent when a book is closed or a sync starts. */
async function pushCounter(device: KoboDevice, book: Book, minutes: number, lastModified: string, progressPercent = 10): Promise<void> {
const response = await ctx.app.inject({
method: 'PUT',
url: `/api/v1/kobo/${device.token}/v1/library/${book.entitlementId}/state`,
payload: {
ReadingStates: [
{
EntitlementId: book.entitlementId,
LastModified: lastModified,
PriorityTimestamp: lastModified,
CurrentBookmark: { LastModified: lastModified, ProgressPercent: progressPercent },
Statistics: { LastModified: lastModified, SpentReadingMinutes: minutes },
StatusInfo: { LastModified: lastModified, Status: 'Reading' },
},
],
},
});
expect(response.statusCode).toBe(200);
}
/** The analytics batch a device that still measures its sessions sends with a library sync. */
async function sendMeasured(device: KoboDevice, book: Book, sessions: MeasuredSession[]): Promise<void> {
const response = await ctx.app.inject({
method: 'POST',
url: `/api/v1/kobo/${device.token}/v1/analytics/event`,
payload: {
Events: sessions.flatMap((session) => [
{ Id: `${session.id}-open`, EventType: 'OpenContent', Timestamp: session.startedAt, Attributes: { volumeid: book.entitlementId } },
{
Id: session.id,
EventType: 'LeaveContent',
Timestamp: session.endedAt,
Metrics: { SecondsRead: session.seconds },
Attributes: { volumeid: book.entitlementId },
},
]),
},
});
expect(response.statusCode).toBe(200);
}
async function storedSessions(book: Book) {
const rows = await ctx.db
.select({
sessionId: schema.readingSessions.sessionId,
sourceDeviceKey: schema.readingSessions.sourceDeviceKey,
startedAt: schema.readingSessions.startedAt,
endedAt: schema.readingSessions.endedAt,
durationSeconds: schema.readingSessions.durationSeconds,
progressDelta: schema.readingSessions.progressDelta,
syncIntervalStartedAt: schema.readingSessions.syncIntervalStartedAt,
})
.from(schema.readingSessions)
.where(and(eq(schema.readingSessions.userId, userId), eq(schema.readingSessions.bookId, book.bookId)))
.orderBy(asc(schema.readingSessions.startedAt), asc(schema.readingSessions.id));
return rows.map((row) => ({
sessionId: row.sessionId,
device: row.sourceDeviceKey,
startedAt: row.startedAt.toISOString(),
endedAt: row.endedAt.toISOString(),
seconds: row.durationSeconds,
progressDelta: row.progressDelta,
intervalStartedAt: row.syncIntervalStartedAt?.toISOString() ?? null,
}));
}
function estimate(
device: KoboDevice,
book: Book,
counter: number,
endedAt: string,
seconds: number,
intervalStartedAt: string,
extra: { generation?: number; progressDelta?: number | null } = {},
) {
return {
sessionId: `kst:${device.id}:${book.bookFileId}:${extra.generation ?? 0}:${counter}`,
device: String(device.id),
startedAt: new Date(Date.parse(endedAt) - seconds * 1000).toISOString(),
endedAt,
seconds,
progressDelta: extra.progressDelta === undefined ? 0 : extra.progressDelta,
intervalStartedAt,
};
}
function stored(device: KoboDevice, session: MeasuredSession) {
return {
sessionId: session.id,
device: String(device.id),
startedAt: session.startedAt,
endedAt: session.endedAt,
seconds: session.seconds,
progressDelta: null,
intervalStartedAt: null,
};
}
/** Reading seconds per day over the given days of the scenario calendar; days without a row are absent. */
async function dailySeconds(...days: number[]): Promise<Record<string, number>> {
const rows = await ctx.db
.select({ day: schema.userReadingDailyStats.day, readingSeconds: schema.userReadingDailyStats.readingSeconds })
.from(schema.userReadingDailyStats)
.where(
and(
eq(schema.userReadingDailyStats.userId, userId),
eq(schema.userReadingDailyStats.libraryId, library.libraryId),
inArray(schema.userReadingDailyStats.day, days.map(day)),
),
);
return Object.fromEntries(rows.map((row) => [row.day, row.readingSeconds]));
}
beforeAll(async () => {
ctx = await createReaderStateIsolationE2EContext();
library = await createLibraryWithFolder(ctx, { name: `kobo-reading-time-${RUN}` });
const paths = await Promise.all(
Array.from({ length: BOOK_COUNT }, (_, index) =>
createEpubFixture(library.folderPath, `kobo-reading-time-${index + 1}.epub`, {
title: `Kobo Reading Time ${index + 1}`,
uid: `urn:uuid:${randomUUID()}`,
}),
),
);
await triggerAndWaitForLibraryScan(ctx, library.libraryId);
const located = await Promise.all(paths.map((path) => locateBookByAbsolutePath(ctx, path)));
const [user] = await ctx.db.select({ id: schema.users.id }).from(schema.users).where(eq(schema.users.username, 'reader-state-e2e-admin'));
userId = user!.id;
const collectionResponse = await ctx.app.inject({
method: 'POST',
url: '/api/v1/collections',
headers: authHeader(ctx.adminToken),
payload: { name: `Kobo Reading Time ${RUN}`, icon: 'book', syncToKobo: true },
});
expect([200, 201]).toContain(collectionResponse.statusCode);
const collectionId = (collectionResponse.json() as { id: number }).id;
const addBooksResponse = await ctx.app.inject({
method: 'POST',
url: `/api/v1/collections/${collectionId}/books`,
headers: authHeader(ctx.adminToken),
payload: { bookIds: located.map((book) => book.bookId) },
});
expect([200, 201]).toContain(addBooksResponse.statusCode);
deviceA = await createDevice(`Kobo A ${RUN}`);
deviceB = await createDevice(`Kobo B ${RUN}`);
await drainLibrary(deviceA);
await drainLibrary(deviceB);
const entitlements = await ctx.db
.select({ bookId: schema.koboBookEntitlements.bookId, entitlementId: schema.koboBookEntitlements.entitlementId })
.from(schema.koboBookEntitlements)
.where(eq(schema.koboBookEntitlements.userId, userId));
books = located.map((book) => ({
bookId: book.bookId,
bookFileId: book.bookFileId,
entitlementId: entitlements.find((entitlement) => entitlement.bookId === book.bookId)!.entitlementId,
}));
}, 180_000);
afterAll(async () => {
if (ctx) await closeReaderStateIsolationE2EContext(ctx);
});
describe('reading done offline, well before the sync that reports it', () => {
const evening = [
measured('offline-1', 2, '20:16:11', '20:26:36'),
measured('offline-2', 2, '20:31:36', '20:54:35'),
measured('offline-3', 2, '20:54:55', '21:09:57'),
];
it('retires the estimate when the counter push arrives before the measured sessions', async () => {
const book = books[0]!;
await pushCounter(deviceA, book, 9, at(1, '20:52:44'));
await pushCounter(deviceA, book, 57, at(3, '07:41:58'));
expect(await storedSessions(book)).toEqual([estimate(deviceA, book, 57, at(3, '07:41:58'), 2880, at(1, '20:52:44'))]);
expect(await dailySeconds(1, 2, 3)).toEqual({ [day(3)]: 2880 });
await sendMeasured(deviceA, book, evening);
expect(await storedSessions(book)).toEqual(evening.map((session) => stored(deviceA, session)));
expect(await dailySeconds(1, 2, 3)).toEqual({ [day(2)]: 2906 });
});
it('derives no estimate when the measured sessions arrive before the counter push', async () => {
const book = books[1]!;
const sessions = evening.map((session) => measured(`${session.id}-first`, 6, session.startedAt.slice(11, 19), session.endedAt.slice(11, 19)));
await pushCounter(deviceA, book, 9, at(5, '20:52:44'));
await sendMeasured(deviceA, book, sessions);
await pushCounter(deviceA, book, 57, at(7, '07:41:58'));
expect(await storedSessions(book)).toEqual(sessions.map((session) => stored(deviceA, session)));
expect(await dailySeconds(5, 6, 7)).toEqual({ [day(6)]: 2906 });
});
});
it('matches each delayed measured session to its own counter interval when several estimates exist', async () => {
const book = books[2]!;
await pushCounter(deviceA, book, 0, at(9, '07:00:00'));
await pushCounter(deviceA, book, 30, at(10, '07:00:00'));
await pushCounter(deviceA, book, 60, at(11, '07:00:00'));
await pushCounter(deviceA, book, 80, at(12, '07:00:00'));
const unmeasured = estimate(deviceA, book, 80, at(12, '07:00:00'), 1200, at(11, '07:00:00'));
expect(await storedSessions(book)).toEqual([
estimate(deviceA, book, 30, at(10, '07:00:00'), 1800, at(9, '07:00:00')),
estimate(deviceA, book, 60, at(11, '07:00:00'), 1800, at(10, '07:00:00')),
unmeasured,
]);
const delayed = [
measured('delayed-1', 9, '20:00:00', '20:20:00'),
measured('delayed-2', 9, '20:30:00', '20:40:00'),
measured('delayed-3', 10, '20:00:00', '20:15:00'),
measured('delayed-4', 10, '20:20:00', '20:35:00'),
];
await sendMeasured(deviceA, book, delayed);
expect(await storedSessions(book)).toEqual([...delayed.map((session) => stored(deviceA, session)), unmeasured]);
expect(await dailySeconds(9, 10, 11, 12)).toEqual({ [day(9)]: 1800, [day(10)]: 1800, [day(12)]: 1200 });
});
describe('a counter delta the measured sessions cover only in part', () => {
it('keeps the unmeasured remainder and takes nothing more off it on a replay', async () => {
const book = books[3]!;
const partial = [measured('partial', 15, '20:00:00', '20:15:00')];
await pushCounter(deviceA, book, 0, at(14, '07:00:00'), 10);
await pushCounter(deviceA, book, 40, at(16, '07:00:00'), 30);
expect(await storedSessions(book)).toEqual([estimate(deviceA, book, 40, at(16, '07:00:00'), 2400, at(14, '07:00:00'), { progressDelta: 20 })]);
await sendMeasured(deviceA, book, partial);
const remainder = estimate(deviceA, book, 40, at(16, '07:00:00'), 1500, at(14, '07:00:00'), { progressDelta: 12.5 });
expect(await storedSessions(book)).toEqual([stored(deviceA, partial[0]!), remainder]);
expect(await dailySeconds(14, 15, 16)).toEqual({ [day(15)]: 900, [day(16)]: 1500 });
await sendMeasured(deviceA, book, partial);
expect(await storedSessions(book)).toEqual([stored(deviceA, partial[0]!), remainder]);
expect(await dailySeconds(14, 15, 16)).toEqual({ [day(15)]: 900, [day(16)]: 1500 });
});
it('reaches the same remainder when the measured session arrives first', async () => {
const book = books[4]!;
const partial = [measured('partial-first', 19, '20:00:00', '20:15:00')];
await pushCounter(deviceA, book, 0, at(18, '07:00:00'), 10);
await sendMeasured(deviceA, book, partial);
await pushCounter(deviceA, book, 40, at(20, '07:00:00'), 30);
expect(await storedSessions(book)).toEqual([
stored(deviceA, partial[0]!),
estimate(deviceA, book, 40, at(20, '07:00:00'), 1500, at(18, '07:00:00'), { progressDelta: 12.5 }),
]);
expect(await dailySeconds(18, 19, 20)).toEqual({ [day(19)]: 900, [day(20)]: 1500 });
});
});
it('keeps measured sessions from before a counter reset away from the estimates after it', async () => {
const book = books[5]!;
const beforeReset = measured('before-reset', 22, '20:00:00', '20:10:00');
const beforeResetLate = measured('before-reset-late', 22, '21:00:00', '21:05:00');
await pushCounter(deviceA, book, 100, at(22, '07:00:00'));
await sendMeasured(deviceA, book, [beforeReset]);
await pushCounter(deviceA, book, 5, at(23, '07:00:00'));
expect(await storedSessions(book)).toEqual([stored(deviceA, beforeReset)]);
await pushCounter(deviceA, book, 25, at(24, '07:00:00'));
await sendMeasured(deviceA, book, [beforeResetLate]);
expect(await storedSessions(book)).toEqual([
stored(deviceA, beforeReset),
stored(deviceA, beforeResetLate),
estimate(deviceA, book, 25, at(24, '07:00:00'), 1200, at(23, '07:00:00'), { generation: 1 }),
]);
expect(await dailySeconds(22, 23, 24)).toEqual({ [day(22)]: 900, [day(24)]: 1200 });
});
it('leaves the estimate of another device alone', async () => {
const book = books[6]!;
const onDeviceA = measured('device-a', 27, '20:00:00', '20:20:00');
await pushCounter(deviceA, book, 0, at(27, '07:00:00'));
await pushCounter(deviceB, book, 0, at(27, '07:01:00'));
await pushCounter(deviceA, book, 20, at(28, '07:00:00'));
await pushCounter(deviceB, book, 20, at(28, '07:05:00'));
const otherDevice = estimate(deviceB, book, 20, at(28, '07:05:00'), 1200, at(27, '07:01:00'));
expect(await storedSessions(book)).toEqual([estimate(deviceA, book, 20, at(28, '07:00:00'), 1200, at(27, '07:00:00')), otherDevice]);
await sendMeasured(deviceA, book, [onDeviceA]);
expect(await storedSessions(book)).toEqual([stored(deviceA, onDeviceA), otherDevice]);
expect(await dailySeconds(27, 28)).toEqual({ [day(27)]: 1200, [day(28)]: 1200 });
});
it('recomputes the daily totals of an estimate that spans midnight as it shrinks and disappears', async () => {
const book = books[7]!;
const first = measured('midnight-1', 33, '12:00:00', '12:30:00');
const second = measured('midnight-2', 33, '13:00:00', '13:30:00');
await pushCounter(deviceA, book, 0, at(32, '07:00:00'));
await pushCounter(deviceA, book, 60, at(34, '00:20:00'));
expect(await dailySeconds(32, 33, 34)).toEqual({ [day(33)]: 2400, [day(34)]: 1200 });
await sendMeasured(deviceA, book, [first]);
expect(await storedSessions(book)).toEqual([stored(deviceA, first), estimate(deviceA, book, 60, at(34, '00:20:00'), 1800, at(32, '07:00:00'))]);
expect(await dailySeconds(32, 33, 34)).toEqual({ [day(33)]: 2400, [day(34)]: 1200 });
await sendMeasured(deviceA, book, [second]);
expect(await storedSessions(book)).toEqual([stored(deviceA, first), stored(deviceA, second)]);
expect(await dailySeconds(32, 33, 34)).toEqual({ [day(33)]: 3600 });
});
describe('a counter push made while the book is open', () => {
it('retires the estimate when the device syncs on waking, seconds after the book reopens', async () => {
const book = books[8]!;
const session = measured('wake-sync', 37, '20:00:00', '20:30:00');
await pushCounter(deviceA, book, 100, at(36, '07:00:00'));
await pushCounter(deviceA, book, 100, at(37, '20:00:05'));
await pushCounter(deviceA, book, 130, at(37, '20:30:10'));
expect(await storedSessions(book)).toEqual([estimate(deviceA, book, 130, at(37, '20:30:10'), 1800, at(37, '20:00:05'))]);
await sendMeasured(deviceA, book, [session]);
expect(await storedSessions(book)).toEqual([stored(deviceA, session)]);
expect(await dailySeconds(36, 37, 38)).toEqual({ [day(37)]: 1800 });
});
it('derives no estimate after a wake-up sync when the measured session arrives first', async () => {
const book = books[9]!;
const session = measured('wake-sync-first', 40, '20:00:00', '20:30:00');
await pushCounter(deviceA, book, 100, at(39, '07:00:00'));
await pushCounter(deviceA, book, 100, at(40, '20:00:05'));
await sendMeasured(deviceA, book, [session]);
await pushCounter(deviceA, book, 130, at(40, '20:30:10'));
expect(await storedSessions(book)).toEqual([stored(deviceA, session)]);
expect(await dailySeconds(39, 40, 41)).toEqual({ [day(40)]: 1800 });
});
it('shares the session between the estimates on both sides of a push made while reading', async () => {
const book = books[10]!;
const session = measured('mid-session', 43, '20:00:00', '20:30:00');
await pushCounter(deviceA, book, 100, at(42, '07:00:00'));
await pushCounter(deviceA, book, 110, at(43, '20:10:00'));
await pushCounter(deviceA, book, 130, at(43, '20:30:05'));
expect(await storedSessions(book)).toEqual([
estimate(deviceA, book, 110, at(43, '20:10:00'), 600, at(42, '07:00:00')),
estimate(deviceA, book, 130, at(43, '20:30:05'), 1200, at(43, '20:10:00')),
]);
await sendMeasured(deviceA, book, [session]);
expect(await storedSessions(book)).toEqual([stored(deviceA, session)]);
expect(await dailySeconds(42, 43, 44)).toEqual({ [day(43)]: 1800 });
});
it('retires the estimate of reading that carried on after the sync that reported the evening', async () => {
const book = books[11]!;
const evening = measured('carried-on-evening', 46, '20:16:11', '21:09:57');
const morning = measured('carried-on-morning', 47, '07:41:50', '08:10:00');
await pushCounter(deviceA, book, 9, at(45, '20:52:44'));
await pushCounter(deviceA, book, 63, at(47, '07:41:58'));
await sendMeasured(deviceA, book, [evening]);
await pushCounter(deviceA, book, 91, at(47, '12:00:00'));
expect(await storedSessions(book)).toEqual([
stored(deviceA, evening),
estimate(deviceA, book, 91, at(47, '12:00:00'), 1680, at(47, '07:41:58')),
]);
await sendMeasured(deviceA, book, [morning]);
expect(await storedSessions(book)).toEqual([stored(deviceA, evening), stored(deviceA, morning)]);
expect(await dailySeconds(45, 46, 47)).toEqual({ [day(46)]: 3226, [day(47)]: 1690 });
});
it('keeps the unmeasured reading of an earlier interval that the session only grazes', async () => {
const book = books[12]!;
const session = measured('grazing', 50, '19:59:50', '20:20:00');
await pushCounter(deviceA, book, 100, at(49, '07:00:00'));
await pushCounter(deviceA, book, 105, at(50, '19:59:55'));
await pushCounter(deviceA, book, 125, at(51, '08:00:00'));
await sendMeasured(deviceA, book, [session]);
// Five of the session's 1210 seconds fall before the push, so the earlier estimate gives up only those.
expect(await storedSessions(book)).toEqual([
estimate(deviceA, book, 105, at(50, '19:59:55'), 295, at(49, '07:00:00')),
stored(deviceA, session),
]);
expect(await dailySeconds(49, 50, 51)).toEqual({ [day(50)]: 1505 });
});
});
});