mirror of
https://github.com/bookorbit/bookorbit.git
synced 2026-11-03 00:11:37 -06:00
feat(reader): make KOReader plugin sync non-blocking and resumable
Lifecycle handlers previously blocked on network work, so closing or suspending a book stalled the device on slow or offline links. Persist snapshots first and drain them through a lifecycle outbox with connected-state gates and phase-aware retries, so the UI stays responsive and pending work survives restarts. Move persistence behind a state manager with atomic flushes, and chunk sweeps, bulk downloads, and annotation exchange through a background queue with checkpoints and transfer policies, so large libraries make progress incrementally instead of restarting from scratch after an interruption. Server side, add bulk manifest and page stats endpoints with batched queries and catalog query filtering so the plugin can reconcile many books per round trip rather than one request per book.
This commit is contained in:
neon
@@ -0,0 +1,182 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { and, eq, gt, inArray, sql } from 'drizzle-orm';
|
||||
|
||||
import * as schema from '../src/db/schema';
|
||||
import {
|
||||
authHeader,
|
||||
closeReaderStateIsolationE2EContext,
|
||||
createLibraryWithFolder,
|
||||
createReaderStateIsolationE2EContext,
|
||||
createUserAndLogin,
|
||||
grantLibraryAccess,
|
||||
type CreatedLibrary,
|
||||
type ReaderStateIsolationE2EContext,
|
||||
type TestUserSession,
|
||||
} from './e2e/reader-state-isolation/reader-state-isolation-harness';
|
||||
import { Permission, type KoreaderCatalogManifestPage } from '@bookorbit/types';
|
||||
|
||||
const KOREADER_USERNAME = `manifest-device-${randomUUID().slice(0, 8)}`;
|
||||
const KOREADER_PASSWORD = 'ManifestDevicePass123';
|
||||
// Above the retired MAX_PAGES ceiling of 20,000 so enumeration is proven past it.
|
||||
const BOOK_COUNT = 25_000;
|
||||
const DECOY_BOOK_COUNT = 5_000;
|
||||
const INSERT_CHUNK = 1_000;
|
||||
const PAGE_SIZE = 200;
|
||||
|
||||
type PlanNode = { 'Node Type'?: string; 'Relation Name'?: string; Plans?: PlanNode[] };
|
||||
|
||||
/** Flattens an EXPLAIN (FORMAT JSON) tree down to the nodes that touch a relation. */
|
||||
function collectScanNodes(queryPlan: unknown): PlanNode[] {
|
||||
const roots = Array.isArray(queryPlan) ? queryPlan : [queryPlan];
|
||||
const found: PlanNode[] = [];
|
||||
const pending = roots.map((entry) => (entry as { Plan?: PlanNode })?.Plan).filter((node): node is PlanNode => !!node);
|
||||
while (pending.length > 0) {
|
||||
const node = pending.pop()!;
|
||||
if (node['Relation Name']) found.push(node);
|
||||
if (node.Plans) pending.push(...node.Plans);
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
describe('KOReader bulk manifest at scale (e2e)', { timeout: 600_000 }, () => {
|
||||
let ctx!: ReaderStateIsolationE2EContext;
|
||||
let library!: CreatedLibrary;
|
||||
let decoyLibrary!: CreatedLibrary;
|
||||
let reader!: TestUserSession;
|
||||
|
||||
async function seedBooks(target: CreatedLibrary, count: number, label: string): Promise<void> {
|
||||
for (let offset = 0; offset < count; offset += INSERT_CHUNK) {
|
||||
const size = Math.min(INSERT_CHUNK, count - offset);
|
||||
const bookRows = await ctx.db
|
||||
.insert(schema.books)
|
||||
.values(
|
||||
Array.from({ length: size }, (_, index) => ({
|
||||
libraryId: target.libraryId,
|
||||
libraryFolderId: target.libraryFolderId,
|
||||
folderPath: `${target.folderPath}/${label}-${offset + index}`,
|
||||
status: 'present' as const,
|
||||
})),
|
||||
)
|
||||
.returning({ id: schema.books.id });
|
||||
|
||||
await ctx.db.insert(schema.bookMetadata).values(
|
||||
bookRows.map((row, index) => ({
|
||||
bookId: row.id,
|
||||
title: `${label} Book ${offset + index}`,
|
||||
})),
|
||||
);
|
||||
|
||||
await ctx.db.insert(schema.bookFiles).values(
|
||||
bookRows.map((row, index) => ({
|
||||
bookId: row.id,
|
||||
libraryFolderId: target.libraryFolderId,
|
||||
absolutePath: `${target.folderPath}/${label}-${offset + index}/book.epub`,
|
||||
relPath: `${label}-${offset + index}/book.epub`,
|
||||
ino: BigInt(row.id),
|
||||
format: 'epub',
|
||||
sizeBytes: 1024 + index,
|
||||
fileHash: `${label.slice(0, 4)}${String(offset + index).padStart(28, '0')}`,
|
||||
})),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function fetchManifestPage(cursor: string | null): Promise<KoreaderCatalogManifestPage> {
|
||||
const params = new URLSearchParams({ size: String(PAGE_SIZE) });
|
||||
if (cursor) params.set('cursor', cursor);
|
||||
const response = await ctx.app.inject({
|
||||
method: 'GET',
|
||||
url: `/api/v1/koreader/plugin/catalog/manifest?${params.toString()}`,
|
||||
headers: { 'x-auth-user': KOREADER_USERNAME, 'x-auth-key': KOREADER_PASSWORD },
|
||||
});
|
||||
expect(response.statusCode).toBe(200);
|
||||
return response.json() as KoreaderCatalogManifestPage;
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
ctx = await createReaderStateIsolationE2EContext();
|
||||
// Lower ids belong to a library the reader cannot see, so the keyset has to skip
|
||||
// 5,000 inaccessible rows before its first accessible one.
|
||||
decoyLibrary = await createLibraryWithFolder(ctx, { name: `manifest-decoy-${randomUUID()}` });
|
||||
library = await createLibraryWithFolder(ctx, { name: `manifest-scale-${randomUUID()}` });
|
||||
await seedBooks(decoyLibrary, DECOY_BOOK_COUNT, 'decoy');
|
||||
await seedBooks(library, BOOK_COUNT, 'scale');
|
||||
|
||||
reader = await createUserAndLogin(ctx, { permissions: [Permission.KoreaderSync] });
|
||||
await grantLibraryAccess(ctx, reader.userId, library.libraryId);
|
||||
|
||||
const credentials = await ctx.app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/v1/koreader/credentials',
|
||||
headers: authHeader(reader.accessToken),
|
||||
payload: { username: KOREADER_USERNAME, password: KOREADER_PASSWORD },
|
||||
});
|
||||
expect([200, 201]).toContain(credentials.statusCode);
|
||||
|
||||
await ctx.db.execute(sql`analyze books`);
|
||||
await ctx.db.execute(sql`analyze book_metadata`);
|
||||
await ctx.db.execute(sql`analyze book_files`);
|
||||
}, 600_000);
|
||||
|
||||
afterAll(async () => {
|
||||
if (ctx) await closeReaderStateIsolationE2EContext(ctx);
|
||||
});
|
||||
|
||||
it('enumerates every accessible book past the retired 20,000 ceiling without gaps or duplicates', async () => {
|
||||
const seen: number[] = [];
|
||||
let cursor: string | null = null;
|
||||
let pages = 0;
|
||||
|
||||
for (;;) {
|
||||
const page: KoreaderCatalogManifestPage = await fetchManifestPage(cursor);
|
||||
expect(page.restartRequired).toBe(false);
|
||||
pages += 1;
|
||||
for (const item of page.items) seen.push(item.id);
|
||||
if (!page.hasNext) break;
|
||||
expect(page.nextCursor).toEqual(expect.any(String));
|
||||
cursor = page.nextCursor;
|
||||
expect(pages).toBeLessThanOrEqual(Math.ceil(BOOK_COUNT / PAGE_SIZE) + 1);
|
||||
}
|
||||
|
||||
expect(seen).toHaveLength(BOOK_COUNT);
|
||||
expect(new Set(seen).size).toBe(BOOK_COUNT);
|
||||
for (let index = 1; index < seen.length; index += 1) {
|
||||
expect(seen[index]!).toBeGreaterThan(seen[index - 1]!);
|
||||
}
|
||||
|
||||
const [row] = await ctx.db
|
||||
.select({ inaccessible: sql<number>`count(*)::int` })
|
||||
.from(schema.books)
|
||||
.where(and(eq(schema.books.libraryId, decoyLibrary.libraryId), inArray(schema.books.id, seen.slice(0, 1))));
|
||||
expect(row?.inaccessible).toBe(0);
|
||||
});
|
||||
|
||||
it('serves a late keyset page from an index rather than a sequential scan', async () => {
|
||||
const [midpoint] = await ctx.db
|
||||
.select({ id: schema.books.id })
|
||||
.from(schema.books)
|
||||
.where(eq(schema.books.libraryId, library.libraryId))
|
||||
.orderBy(schema.books.id)
|
||||
.limit(1)
|
||||
.offset(BOOK_COUNT / 2);
|
||||
expect(midpoint?.id).toBeDefined();
|
||||
|
||||
const explained = await ctx.db.execute<{ 'QUERY PLAN': unknown }>(sql`
|
||||
explain (format json, analyze, buffers)
|
||||
select ${schema.books.id} from ${schema.books}
|
||||
left join ${schema.bookMetadata} on ${schema.bookMetadata.bookId} = ${schema.books.id}
|
||||
where ${inArray(schema.books.libraryId, [library.libraryId])}
|
||||
and ${eq(schema.books.status, 'present')}
|
||||
and ${gt(schema.books.id, midpoint!.id)}
|
||||
order by ${schema.books.id}
|
||||
limit ${PAGE_SIZE + 1}
|
||||
`);
|
||||
|
||||
const scans = collectScanNodes(explained.rows[0]?.['QUERY PLAN']);
|
||||
const booksScan = scans.find((node) => node['Relation Name'] === 'books');
|
||||
expect(booksScan, `no scan node for books in plan ${JSON.stringify(scans)}`).toBeDefined();
|
||||
// A sequential scan over books here would mean the scope filters and the id keyset
|
||||
// have no usable index, which is the regression this fixture exists to catch.
|
||||
expect(booksScan!['Node Type']).toMatch(/^Index (Only )?Scan$/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,206 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { and, asc, eq, like } from 'drizzle-orm';
|
||||
|
||||
import * as schema from '../src/db/schema';
|
||||
import {
|
||||
KOREADER_SESSION_GAP_SECONDS,
|
||||
buildSessionIdPrefix,
|
||||
deriveKoreaderSessions,
|
||||
type KoreaderPageEvent,
|
||||
} from '../src/modules/koreader/koreader-stats.util';
|
||||
import { createEpubFixture } from './e2e/reader-state-isolation/reader-state-isolation-fixture-builder';
|
||||
import {
|
||||
authHeader,
|
||||
closeReaderStateIsolationE2EContext,
|
||||
createLibraryWithFolder,
|
||||
createReaderStateIsolationE2EContext,
|
||||
locateBookByAbsolutePath,
|
||||
triggerAndWaitForLibraryScan,
|
||||
type CreatedLibrary,
|
||||
type LocatedBookFile,
|
||||
type ReaderStateIsolationE2EContext,
|
||||
} from './e2e/reader-state-isolation/reader-state-isolation-harness';
|
||||
|
||||
const KOREADER_USERNAME = `page-stats-device-${randomUUID().slice(0, 8)}`;
|
||||
const KOREADER_PASSWORD = 'PageStatsDevicePass123';
|
||||
const DEVICE_ID = 'e2e-stats-0001';
|
||||
const BASE_EPOCH = 1_780_000_000;
|
||||
const TOTAL_PAGES = 300;
|
||||
|
||||
/** A contiguous run of one-minute page reads starting at `startEpoch`. */
|
||||
function cluster(startEpoch: number, pages: number, firstPage: number): KoreaderPageEvent[] {
|
||||
return Array.from({ length: pages }, (_, index) => ({
|
||||
page: firstPage + index,
|
||||
startTime: startEpoch + index * 60,
|
||||
durationSeconds: 55,
|
||||
totalPages: TOTAL_PAGES,
|
||||
}));
|
||||
}
|
||||
|
||||
describe('KOReader page stats derivation (e2e)', { timeout: 180_000 }, () => {
|
||||
let ctx!: ReaderStateIsolationE2EContext;
|
||||
let library!: CreatedLibrary;
|
||||
let epub!: LocatedBookFile;
|
||||
let fileHash!: string;
|
||||
|
||||
function deviceHeaders(): Record<string, string> {
|
||||
return { 'x-auth-user': KOREADER_USERNAME, 'x-auth-key': KOREADER_PASSWORD };
|
||||
}
|
||||
|
||||
async function uploadPageStats(events: KoreaderPageEvent[]) {
|
||||
const response = await ctx.app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/v1/koreader/plugin/page-stats',
|
||||
headers: deviceHeaders(),
|
||||
payload: {
|
||||
deviceId: DEVICE_ID,
|
||||
deviceModel: 'E2E',
|
||||
pluginVersion: '0.4.0',
|
||||
books: [{ hash: fileHash, events }],
|
||||
},
|
||||
});
|
||||
expect(response.statusCode).toBe(201);
|
||||
return response.json() as { results: { hash: string; accepted: number; duplicates: number }[]; unmatched: string[] };
|
||||
}
|
||||
|
||||
async function storedSessions() {
|
||||
return ctx.db
|
||||
.select({
|
||||
id: schema.readingSessions.id,
|
||||
sessionId: schema.readingSessions.sessionId,
|
||||
startedAt: schema.readingSessions.startedAt,
|
||||
endedAt: schema.readingSessions.endedAt,
|
||||
durationSeconds: schema.readingSessions.durationSeconds,
|
||||
progressDelta: schema.readingSessions.progressDelta,
|
||||
endProgress: schema.readingSessions.endProgress,
|
||||
})
|
||||
.from(schema.readingSessions)
|
||||
.where(
|
||||
and(
|
||||
eq(schema.readingSessions.bookFileId, epub.bookFileId),
|
||||
like(schema.readingSessions.sessionId, `${buildSessionIdPrefix(DEVICE_ID, epub.bookFileId)}%`),
|
||||
),
|
||||
)
|
||||
.orderBy(asc(schema.readingSessions.startedAt));
|
||||
}
|
||||
|
||||
function expectedSessions(events: KoreaderPageEvent[]) {
|
||||
return deriveKoreaderSessions(events, DEVICE_ID, epub.bookFileId)
|
||||
.slice()
|
||||
.sort((left, right) => left.startedAt.getTime() - right.startedAt.getTime())
|
||||
.map((session) => ({
|
||||
sessionId: session.sessionId,
|
||||
startedAt: session.startedAt,
|
||||
endedAt: session.endedAt,
|
||||
durationSeconds: session.durationSeconds,
|
||||
progressDelta: session.progressDelta,
|
||||
endProgress: session.endProgress,
|
||||
}));
|
||||
}
|
||||
|
||||
function comparable(rows: Awaited<ReturnType<typeof storedSessions>>) {
|
||||
return rows.map((row) => ({
|
||||
sessionId: row.sessionId,
|
||||
startedAt: row.startedAt,
|
||||
endedAt: row.endedAt,
|
||||
durationSeconds: row.durationSeconds,
|
||||
progressDelta: row.progressDelta,
|
||||
endProgress: row.endProgress,
|
||||
}));
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
ctx = await createReaderStateIsolationE2EContext();
|
||||
library = await createLibraryWithFolder(ctx, { name: `koreader-page-stats-${randomUUID()}` });
|
||||
const epubPath = await createEpubFixture(library.folderPath, 'page-stats-book.epub', {
|
||||
title: `Page Stats Book ${randomUUID()}`,
|
||||
uid: `urn:uuid:${randomUUID()}`,
|
||||
});
|
||||
await triggerAndWaitForLibraryScan(ctx, library.libraryId);
|
||||
epub = await locateBookByAbsolutePath(ctx, epubPath);
|
||||
|
||||
const [fileRow] = await ctx.db
|
||||
.select({ fileHash: schema.bookFiles.fileHash })
|
||||
.from(schema.bookFiles)
|
||||
.where(eq(schema.bookFiles.id, epub.bookFileId));
|
||||
expect(fileRow?.fileHash).toBeTruthy();
|
||||
fileHash = fileRow!.fileHash!;
|
||||
|
||||
const credentials = await ctx.app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/v1/koreader/credentials',
|
||||
headers: authHeader(ctx.adminToken),
|
||||
payload: { username: KOREADER_USERNAME, password: KOREADER_PASSWORD },
|
||||
});
|
||||
expect([200, 201]).toContain(credentials.statusCode);
|
||||
}, 180_000);
|
||||
|
||||
afterAll(async () => {
|
||||
if (ctx) await closeReaderStateIsolationE2EContext(ctx);
|
||||
});
|
||||
|
||||
it('derives the same sessions as a full-history derivation across many batches', async () => {
|
||||
// Ten clusters an hour apart, uploaded in 400-event batches.
|
||||
const history = Array.from({ length: 10 }, (_, index) => cluster(BASE_EPOCH + index * 4 * KOREADER_SESSION_GAP_SECONDS, 20, index * 20)).flat();
|
||||
|
||||
for (let offset = 0; offset < history.length; offset += 400) {
|
||||
const batch = history.slice(offset, offset + 400);
|
||||
const result = await uploadPageStats(batch);
|
||||
expect(result.results[0]!.accepted).toBe(batch.length);
|
||||
}
|
||||
|
||||
expect(comparable(await storedSessions())).toEqual(expectedSessions(history));
|
||||
});
|
||||
|
||||
it('merges two stored sessions when a later batch fills the gap between them', async () => {
|
||||
const early = cluster(BASE_EPOCH + 5_000_000, 10, 0);
|
||||
const late = cluster(BASE_EPOCH + 5_000_000 + 3 * KOREADER_SESSION_GAP_SECONDS, 10, 40);
|
||||
await uploadPageStats([...early, ...late]);
|
||||
|
||||
const before = await storedSessions();
|
||||
const separated = before.filter((row) => row.startedAt.getTime() >= (BASE_EPOCH + 5_000_000) * 1000);
|
||||
expect(separated).toHaveLength(2);
|
||||
|
||||
// Chains to the earlier cluster's end and reaches within the gap of the later one,
|
||||
// so the pair becomes a single session under a new id taken from the earlier start.
|
||||
const bridge = cluster(early.at(-1)!.startTime + 1_760, 22, 10);
|
||||
expect(bridge[0]!.startTime - (early.at(-1)!.startTime + early.at(-1)!.durationSeconds)).toBeLessThanOrEqual(KOREADER_SESSION_GAP_SECONDS);
|
||||
expect(late[0]!.startTime - (bridge.at(-1)!.startTime + bridge.at(-1)!.durationSeconds)).toBeLessThanOrEqual(KOREADER_SESSION_GAP_SECONDS);
|
||||
await uploadPageStats(bridge);
|
||||
|
||||
const after = await storedSessions();
|
||||
const merged = after.filter((row) => row.startedAt.getTime() >= (BASE_EPOCH + 5_000_000) * 1000);
|
||||
expect(merged).toHaveLength(1);
|
||||
expect(comparable(merged)).toEqual(expectedSessions([...early, ...bridge, ...late]));
|
||||
// The superseded rows are gone, not merely orphaned.
|
||||
expect(after.some((row) => row.sessionId === separated[1]!.sessionId)).toBe(false);
|
||||
});
|
||||
|
||||
it('leaves sessions outside the affected window untouched', async () => {
|
||||
const untouched = (await storedSessions())[0]!;
|
||||
const distant = cluster(BASE_EPOCH + 9_000_000, 5, 100);
|
||||
|
||||
await uploadPageStats(distant);
|
||||
|
||||
const after = await storedSessions();
|
||||
expect(after.find((row) => row.sessionId === untouched.sessionId)).toEqual(untouched);
|
||||
});
|
||||
|
||||
it('makes a replayed batch a no-op after an interrupted upload', async () => {
|
||||
const resumed = cluster(BASE_EPOCH + 11_000_000, 12, 150);
|
||||
await uploadPageStats(resumed.slice(0, 6));
|
||||
const partial = await storedSessions();
|
||||
|
||||
const replay = await uploadPageStats(resumed);
|
||||
expect(replay.results[0]!.duplicates).toBe(6);
|
||||
|
||||
const complete = await storedSessions();
|
||||
expect(complete).toHaveLength(partial.length);
|
||||
const finished = complete.find((row) => row.startedAt.getTime() === resumed[0]!.startTime * 1000)!;
|
||||
expect(finished.endedAt.getTime()).toBe((resumed.at(-1)!.startTime + resumed.at(-1)!.durationSeconds) * 1000);
|
||||
|
||||
const noop = await uploadPageStats(resumed);
|
||||
expect(noop.results[0]!.accepted).toBe(0);
|
||||
expect(await storedSessions()).toEqual(complete);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user