feat(statistics): KOReader-compatible reading stats with cross-device sync (#4606)
Supersedes #3156. Adds a reading-statistics system whose canonical data model is KOReader's own (book + page_stat_data), so stats round-trip losslessly between Readest and KOReader. - Storage: a cross-platform Turso statistics.db in KOReader's exact schema (web/Workers, desktop, iOS, Android) — replacing #3156's Node-only better-sqlite3 + statistics.json. - Tracking: per-page reading events (time-on-page, idle-capped) flushed on page-change/idle/hide/close — the KOReader model — not session aggregates. - Sync: legacy /api/sync extended with a stats type backed by self-contained Supabase tables (stat_books, stat_pages); union/longer-duration-wins merge keyed on book_hash. apps/readest.koplugin syncs through the same endpoint. - Scale & robustness: per-tab singleton connection (avoids OPFS lock conflicts) + explicit WAL checkpoint; transactional bulk apply; chunked resumable push; client-driven paged pull with trailing-ms completion; paginated/scoped server merge. Verified: 5668 unit tests, 155 koplugin busted tests, biome+tsgo + luacheck all green; web OPFS DB verified live. Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Huang Xin
@@ -51,6 +51,65 @@ const migrations: Record<SchemaType, MigrationEntry[]> = {
|
||||
// rows rather than UPDATE — Tantivy 0.25→0.26 has a known WASM-only
|
||||
// UPDATE regression (see fts-tests.ts:306 FIXME). MVP indexing is
|
||||
// write-once per book so this is naturally satisfied.
|
||||
statistics: [
|
||||
{
|
||||
name: '2026061501_statistics_koreader_schema',
|
||||
sql: `
|
||||
CREATE TABLE IF NOT EXISTS book (
|
||||
id integer PRIMARY KEY autoincrement,
|
||||
title text, authors text, notes integer, last_open integer,
|
||||
highlights integer, pages integer, series text, language text,
|
||||
md5 text, total_read_time integer, total_read_pages integer
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS book_title_authors_md5 ON book(title, authors, md5);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS page_stat_data (
|
||||
id_book integer,
|
||||
page integer NOT NULL DEFAULT 0,
|
||||
start_time integer NOT NULL DEFAULT 0,
|
||||
duration integer NOT NULL DEFAULT 0,
|
||||
total_pages integer NOT NULL DEFAULT 0,
|
||||
UNIQUE (id_book, page, start_time)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS page_stat_data_start_time ON page_stat_data(start_time);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS numbers (number INTEGER PRIMARY KEY);
|
||||
|
||||
INSERT OR IGNORE INTO numbers(number)
|
||||
SELECT h.n * 100 + t.n * 10 + o.n + 1
|
||||
FROM (SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL SELECT 8 UNION ALL SELECT 9) o,
|
||||
(SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL SELECT 8 UNION ALL SELECT 9) t,
|
||||
(SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL SELECT 8 UNION ALL SELECT 9) h;
|
||||
|
||||
CREATE VIEW IF NOT EXISTS page_stat AS
|
||||
SELECT id_book, first_page + idx - 1 AS page, start_time, duration / (last_page - first_page + 1) AS duration
|
||||
FROM (
|
||||
SELECT id_book, page, total_pages, pages, start_time, duration,
|
||||
((page - 1) * pages) / total_pages + 1 AS first_page,
|
||||
max(((page - 1) * pages) / total_pages + 1, (page * pages) / total_pages) AS last_page,
|
||||
idx
|
||||
FROM page_stat_data
|
||||
JOIN book ON book.id = id_book
|
||||
JOIN (SELECT number as idx FROM numbers) AS N ON idx <= (last_page - first_page + 1)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS readest_page_ext (
|
||||
book_hash text NOT NULL, page integer NOT NULL, start_time integer NOT NULL,
|
||||
ext text, PRIMARY KEY (book_hash, page, start_time)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS readest_book_ext (
|
||||
book_hash text PRIMARY KEY, ext text
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS readest_stat_sync_state (
|
||||
key text PRIMARY KEY, value integer NOT NULL DEFAULT 0
|
||||
);
|
||||
`,
|
||||
},
|
||||
],
|
||||
reedy: [
|
||||
{
|
||||
name: '2026052601_reedy_init',
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
import type { AppService } from '@/types/system';
|
||||
import type { DatabaseService, DatabaseRow } from '@/types/database';
|
||||
import type { PageStatEvent, StatBook } from '@/types/statistics';
|
||||
|
||||
interface BookRow extends DatabaseRow {
|
||||
id: number;
|
||||
title: string;
|
||||
authors: string;
|
||||
md5: string;
|
||||
total_read_time: number;
|
||||
total_read_pages: number;
|
||||
last_open: number;
|
||||
pages: number;
|
||||
}
|
||||
|
||||
type CursorKey = 'push' | 'pull';
|
||||
|
||||
/**
|
||||
* Per-tab singleton open promise. OPFS permits only ONE access handle per file
|
||||
* across the whole origin, so a second `connect()` to statistics.db throws
|
||||
* `NoModificationAllowedError`. Every ReadingStatsTracker instance (and split-
|
||||
* view books) must therefore share a single connection — we memoise the open
|
||||
* and never thrash it.
|
||||
*/
|
||||
let sharedDb: Promise<StatisticsDb> | null = null;
|
||||
let lifecycleBound = false;
|
||||
|
||||
function bindLifecycle(): void {
|
||||
if (lifecycleBound || typeof document === 'undefined') return;
|
||||
lifecycleBound = true;
|
||||
// Fold the WAL into the main db when the tab is backgrounded/closed — the
|
||||
// most reliable point for best-effort async OPFS work before teardown.
|
||||
document.addEventListener('visibilitychange', () => {
|
||||
if (document.visibilityState === 'hidden' && sharedDb) {
|
||||
void sharedDb.then((s) => s.checkpoint()).catch(() => {});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Typed wrapper over the KOReader-compatible `statistics.db`. All identity is
|
||||
* keyed on `book_hash` (= Book.hash); the local autoincrement `id_book` never
|
||||
* leaves this class.
|
||||
*/
|
||||
export class StatisticsDb {
|
||||
private constructor(private readonly db: DatabaseService) {}
|
||||
|
||||
/** Production entry point — opens + migrates statistics.db (per-tab singleton). */
|
||||
static async open(appService: AppService): Promise<StatisticsDb> {
|
||||
bindLifecycle();
|
||||
if (!sharedDb) {
|
||||
sharedDb = (async () => {
|
||||
const db = await appService.openDatabase('statistics', 'statistics.db', 'Data');
|
||||
return new StatisticsDb(db);
|
||||
})();
|
||||
}
|
||||
return sharedDb;
|
||||
}
|
||||
|
||||
/** Test/advanced entry point — wrap an already-migrated DatabaseService. */
|
||||
static from(db: DatabaseService): StatisticsDb {
|
||||
return new StatisticsDb(db);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold the WAL into the main db file and truncate it. The Turso engine does
|
||||
* NOT implement `PRAGMA wal_autocheckpoint` (so there's no auto threshold to
|
||||
* rely on), but `wal_checkpoint(TRUNCATE)` works — verified folding a 688 KB
|
||||
* WAL back to 0 B. We call this when the tab is hidden so the WAL stays bounded.
|
||||
*/
|
||||
async checkpoint(): Promise<void> {
|
||||
await this.db.execute('PRAGMA wal_checkpoint(TRUNCATE)');
|
||||
}
|
||||
|
||||
/** Checkpoint, close the underlying connection, and reset the singleton. */
|
||||
async close(): Promise<void> {
|
||||
try {
|
||||
await this.checkpoint();
|
||||
} catch {
|
||||
// best-effort — a checkpoint failure must not block close
|
||||
}
|
||||
await this.db.close();
|
||||
sharedDb = null;
|
||||
}
|
||||
|
||||
async upsertBook(book: StatBook): Promise<number> {
|
||||
const existing = await this.db.select<BookRow>(`SELECT id FROM book WHERE md5 = ? LIMIT 1`, [
|
||||
book.bookMd5,
|
||||
]);
|
||||
if (existing[0]) {
|
||||
// md5 is the identity; keep the latest title/authors (LWW, matches server stat_books).
|
||||
await this.db.execute(`UPDATE book SET title = ?, authors = ? WHERE id = ?`, [
|
||||
book.title,
|
||||
book.authors,
|
||||
existing[0].id,
|
||||
]);
|
||||
return existing[0].id;
|
||||
}
|
||||
await this.db.execute(
|
||||
`INSERT INTO book (title, authors, md5, notes, last_open, highlights, pages, total_read_time, total_read_pages)
|
||||
VALUES (?, ?, ?, 0, 0, 0, 0, 0, 0)
|
||||
ON CONFLICT(title, authors, md5) DO NOTHING`,
|
||||
[book.title, book.authors, book.bookMd5],
|
||||
);
|
||||
const rows = await this.db.select<BookRow>(`SELECT id FROM book WHERE md5 = ? LIMIT 1`, [
|
||||
book.bookMd5,
|
||||
]);
|
||||
return rows[0]!.id;
|
||||
}
|
||||
|
||||
async insertPageEvent(
|
||||
idBook: number,
|
||||
e: { page: number; startTime: number; duration: number; totalPages: number },
|
||||
): Promise<void> {
|
||||
await this.db.execute(
|
||||
`INSERT INTO page_stat_data (id_book, page, start_time, duration, total_pages)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id_book, page, start_time)
|
||||
DO UPDATE SET duration = max(duration, excluded.duration), total_pages = excluded.total_pages`,
|
||||
[idBook, e.page, e.startTime, e.duration, e.totalPages],
|
||||
);
|
||||
}
|
||||
|
||||
async recomputeBookTotals(idBook: number): Promise<void> {
|
||||
await this.db.execute(
|
||||
`UPDATE book SET
|
||||
total_read_time = COALESCE((SELECT SUM(duration) FROM page_stat_data WHERE id_book = ?), 0),
|
||||
total_read_pages = COALESCE((SELECT COUNT(DISTINCT page) FROM page_stat_data WHERE id_book = ?), 0),
|
||||
last_open = COALESCE((SELECT MAX(start_time + duration) FROM page_stat_data WHERE id_book = ?), last_open),
|
||||
pages = COALESCE((SELECT total_pages FROM page_stat_data WHERE id_book = ? ORDER BY start_time DESC LIMIT 1), pages)
|
||||
WHERE id = ?`,
|
||||
[idBook, idBook, idBook, idBook, idBook],
|
||||
);
|
||||
}
|
||||
|
||||
async getBookByMd5(md5: string): Promise<BookRow | null> {
|
||||
const rows = await this.db.select<BookRow>(`SELECT * FROM book WHERE md5 = ? LIMIT 1`, [md5]);
|
||||
return rows[0] ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure a book row exists for an event whose metadata record isn't in the
|
||||
* current batch (paged pull can deliver an event before its `stat_books`
|
||||
* record). Unlike `upsertBook`, this NEVER overwrites an existing real title
|
||||
* with the hash placeholder — the real record, arriving in any page, wins.
|
||||
*/
|
||||
private async ensureBookId(bookMd5: string): Promise<number> {
|
||||
const existing = await this.db.select<BookRow>(`SELECT id FROM book WHERE md5 = ? LIMIT 1`, [
|
||||
bookMd5,
|
||||
]);
|
||||
if (existing[0]) return existing[0].id;
|
||||
return this.upsertBook({ bookMd5, title: bookMd5, authors: '' });
|
||||
}
|
||||
|
||||
/** Events with start_time > cursor, joined to their md5, for pushing. */
|
||||
async getEventsForPush(
|
||||
sinceStartTime: number,
|
||||
): Promise<{ events: PageStatEvent[]; books: StatBook[] }> {
|
||||
const rows = await this.db.select<DatabaseRow>(
|
||||
`SELECT b.md5 AS bookMd5, b.title AS title, b.authors AS authors,
|
||||
p.page AS page, p.start_time AS startTime, p.duration AS duration, p.total_pages AS totalPages
|
||||
FROM page_stat_data p JOIN book b ON b.id = p.id_book
|
||||
WHERE p.start_time > ?
|
||||
ORDER BY p.start_time ASC`,
|
||||
[sinceStartTime],
|
||||
);
|
||||
const events: PageStatEvent[] = rows.map((r) => ({
|
||||
bookMd5: String(r['bookMd5']),
|
||||
page: Number(r['page']),
|
||||
startTime: Number(r['startTime']),
|
||||
duration: Number(r['duration']),
|
||||
totalPages: Number(r['totalPages']),
|
||||
}));
|
||||
const bookMap = new Map<string, StatBook>();
|
||||
for (const r of rows) {
|
||||
const md5 = String(r['bookMd5']);
|
||||
if (!bookMap.has(md5)) {
|
||||
bookMap.set(md5, {
|
||||
bookMd5: md5,
|
||||
title: String(r['title']),
|
||||
authors: String(r['authors']),
|
||||
});
|
||||
}
|
||||
}
|
||||
return { events, books: [...bookMap.values()] };
|
||||
}
|
||||
|
||||
async applyRemoteEvents(books: StatBook[], events: PageStatEvent[]): Promise<void> {
|
||||
if (books.length === 0 && events.length === 0) return;
|
||||
// One transaction for the whole pulled batch: a single commit instead of
|
||||
// O(rows) fsyncs, and the apply is atomic (a failed pull leaves no partial
|
||||
// state). Critical when a fresh device backfills tens of thousands of rows.
|
||||
await this.db.execute('BEGIN');
|
||||
try {
|
||||
const idByMd5 = new Map<string, number>();
|
||||
for (const b of books) idByMd5.set(b.bookMd5, await this.upsertBook(b));
|
||||
// Books referenced only by events (no metadata record) get a placeholder row.
|
||||
const touched = new Set<number>();
|
||||
for (const e of events) {
|
||||
let id = idByMd5.get(e.bookMd5);
|
||||
if (id === undefined) {
|
||||
id = await this.ensureBookId(e.bookMd5);
|
||||
idByMd5.set(e.bookMd5, id);
|
||||
}
|
||||
await this.insertPageEvent(id, e);
|
||||
touched.add(id);
|
||||
}
|
||||
for (const id of touched) await this.recomputeBookTotals(id);
|
||||
await this.db.execute('COMMIT');
|
||||
} catch (err) {
|
||||
await this.db.execute('ROLLBACK');
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async getCursor(key: CursorKey): Promise<number> {
|
||||
const rows = await this.db.select<{ value: number }>(
|
||||
`SELECT value FROM readest_stat_sync_state WHERE key = ?`,
|
||||
[key],
|
||||
);
|
||||
return rows[0]?.value ?? 0;
|
||||
}
|
||||
|
||||
async setCursor(key: CursorKey, value: number): Promise<void> {
|
||||
await this.db.execute(
|
||||
`INSERT INTO readest_stat_sync_state (key, value) VALUES (?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
|
||||
[key, value],
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import type { StatisticsDb } from './statisticsDb';
|
||||
import type { SyncClient, StatPageRecord, StatBookRecord } from '@/libs/sync';
|
||||
import type { PageStatEvent, StatBook } from '@/types/statistics';
|
||||
|
||||
type PushClient = Pick<SyncClient, 'pushChanges'>;
|
||||
type PullClient = Pick<SyncClient, 'pullChanges'>;
|
||||
|
||||
const toWirePage = (e: PageStatEvent): StatPageRecord => ({
|
||||
book_hash: e.bookMd5,
|
||||
page: e.page,
|
||||
start_time: e.startTime,
|
||||
duration: e.duration,
|
||||
total_pages: e.totalPages,
|
||||
});
|
||||
|
||||
const toWireBook = (b: StatBook): StatBookRecord => ({
|
||||
book_hash: b.bookMd5,
|
||||
title: b.title,
|
||||
authors: b.authors,
|
||||
});
|
||||
|
||||
/** Events per push request — bounds request size for a large offline backlog. */
|
||||
const PUSH_CHUNK = 500;
|
||||
/** Page events per pull request — bounds the receiving device's memory. */
|
||||
const PULL_PAGE = 1000;
|
||||
|
||||
/**
|
||||
* Push local events newer than the push cursor, in bounded chunks. The cursor
|
||||
* advances per successful chunk, so an interrupted push (e.g. a 1000-event
|
||||
* backlog over flaky network) resumes from the last chunk rather than restarting.
|
||||
*/
|
||||
export async function pushStats(stats: StatisticsDb, client: PushClient): Promise<void> {
|
||||
const cursor = await stats.getCursor('push');
|
||||
const { events, books } = await stats.getEventsForPush(cursor);
|
||||
if (events.length === 0) return;
|
||||
const bookByHash = new Map(books.map((b) => [b.bookMd5, b]));
|
||||
let i = 0;
|
||||
while (i < events.length) {
|
||||
let end = Math.min(i + PUSH_CHUNK, events.length);
|
||||
// Never split a start_time across chunks — advancing the push cursor past it
|
||||
// would drop the remaining same-second events (e.g. split-view) on resume.
|
||||
const lastStart = events[end - 1]!.startTime;
|
||||
while (end < events.length && events[end]!.startTime === lastStart) end++;
|
||||
const chunk = events.slice(i, end);
|
||||
const seen = new Set<string>();
|
||||
const chunkBooks: StatBookRecord[] = [];
|
||||
for (const e of chunk) {
|
||||
if (seen.has(e.bookMd5)) continue;
|
||||
seen.add(e.bookMd5);
|
||||
const b = bookByHash.get(e.bookMd5);
|
||||
if (b) chunkBooks.push(toWireBook(b));
|
||||
}
|
||||
await client.pushChanges({ statBooks: chunkBooks, statPages: chunk.map(toWirePage) });
|
||||
await stats.setCursor('push', chunk[chunk.length - 1]!.startTime);
|
||||
i = end;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pull events since the pull cursor in bounded pages, applying each before
|
||||
* fetching the next so memory stays flat and a fresh-device backfill is
|
||||
* resumable (the cursor persists between pages). The server completes the
|
||||
* trailing millisecond of each page, so a strict `> cursor` next pull never
|
||||
* skips rows that share an `updated_at`.
|
||||
*/
|
||||
export async function pullStats(stats: StatisticsDb, client: PullClient): Promise<void> {
|
||||
for (;;) {
|
||||
const since = await stats.getCursor('pull');
|
||||
const res = await client.pullChanges(since, 'stats', undefined, undefined, PULL_PAGE);
|
||||
const wireBooks = (res.statBooks ?? []) as StatBookRecord[];
|
||||
const wirePages = (res.statPages ?? []) as StatPageRecord[];
|
||||
if (wireBooks.length === 0 && wirePages.length === 0) break;
|
||||
const books: StatBook[] = wireBooks.map((b) => ({
|
||||
bookMd5: b.book_hash,
|
||||
title: b.title,
|
||||
authors: b.authors,
|
||||
}));
|
||||
const events: PageStatEvent[] = wirePages.map((p) => ({
|
||||
bookMd5: p.book_hash,
|
||||
page: p.page,
|
||||
startTime: p.start_time,
|
||||
duration: p.duration,
|
||||
totalPages: p.total_pages,
|
||||
}));
|
||||
await stats.applyRemoteEvents(books, events);
|
||||
// Advance the cursor to the newest page-event updated_at_ms. Stop when a
|
||||
// page yields no further page-event progress (covers the empty-pages and
|
||||
// the all-books-no-pages cases, and guards against a stalled cursor).
|
||||
const newest = wirePages.reduce((m, p) => Math.max(m, p.updated_at_ms ?? 0), since);
|
||||
if (newest <= since) break;
|
||||
await stats.setCursor('pull', newest);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import type { StatsTrackingConfig } from '@/types/statistics';
|
||||
|
||||
interface PendingEvent {
|
||||
page: number;
|
||||
startTime: number; // Unix seconds
|
||||
totalPages: number;
|
||||
}
|
||||
|
||||
export interface FlushedEvent {
|
||||
page: number;
|
||||
startTime: number;
|
||||
duration: number;
|
||||
totalPages: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pure page-dwell tracker. The React layer feeds it `nowSeconds` and the
|
||||
* current `(page, totalPages)`; it returns zero-or-one immutable events to
|
||||
* persist. Events end on page-change / idle / hide / close. Each returned
|
||||
* event is final — its start_time and duration never change afterwards, which
|
||||
* lets sync use a simple start_time high-water cursor.
|
||||
*/
|
||||
export class TrackerCore {
|
||||
private pending: PendingEvent | null = null;
|
||||
|
||||
constructor(private readonly cfg: StatsTrackingConfig) {}
|
||||
|
||||
/** Notify the current page at `now`. Returns events flushed by leaving the prior page. */
|
||||
onPage(page: number, totalPages: number, now: number): FlushedEvent[] {
|
||||
if (this.pending && this.pending.page === page) {
|
||||
// Same page (e.g. resume after idle, or a no-op relocate): keep dwelling.
|
||||
return [];
|
||||
}
|
||||
const flushed = this.flush(now);
|
||||
this.pending = { page, startTime: now, totalPages };
|
||||
return flushed;
|
||||
}
|
||||
|
||||
onIdle(now: number): FlushedEvent[] {
|
||||
return this.flush(now); // flush + pause (pending cleared)
|
||||
}
|
||||
|
||||
onHide(now: number): FlushedEvent[] {
|
||||
return this.flush(now);
|
||||
}
|
||||
|
||||
onClose(now: number): FlushedEvent[] {
|
||||
return this.flush(now);
|
||||
}
|
||||
|
||||
private flush(now: number): FlushedEvent[] {
|
||||
const p = this.pending;
|
||||
this.pending = null;
|
||||
if (!p) return [];
|
||||
const raw = now - p.startTime;
|
||||
const duration = Math.min(Math.max(raw, 0), this.cfg.maxEventSeconds);
|
||||
if (duration < this.cfg.minEventSeconds) return [];
|
||||
return [{ page: p.page, startTime: p.startTime, duration, totalPages: p.totalPages }];
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user