/** * @file the cache Durable Objects' logic: `DiigoCacheLogic` (public+private * items) and `PublicCacheLogic` (public-only items, widely shared). Logic * only: each requires its {@link Clock}; the worker's entry module * (src/index.ts) is the edge that makes the live one and exports the * runtime-constructed `DiigoCache`/`PublicCache`. They are independent * and work the same way ({@link PagedCache}): pages are the unit of * caching; a hit is answered from `ctx.storage.sqlite` with no diigo call; * a miss asks the webapp instance in the cache's wiring and etches the * page. The only difference is what each keeps: the public cache drops * private rows *before* etching ({@link publicOnly}). */ import { DurableObject } from "cloudflare:workers"; import type { PagedGet } from "./diigo.js"; import { publicOnly, toRows } from "./items.js"; import type { Item, ItemRow } from "./items.js"; export type { Item }; /** schema migration steps; the sqlite-backed class starts empty, so V1 is * the initial shape. `page_members` keeps each page's fetch-time item * order: a page-hit replays that order exactly, independent of later * re-sorting as items shift ranks. */ const SCHEMA_V1 = ` create table if not exists items ( item_id text primary key, url text not null, title text not null, private integer not null, json text not null, updated text not null ); create table if not exists pages ( page_num integer primary key, count integer not null, total integer, etched_at text not null ); create table if not exists page_members ( page_num integer not null, pos integer not null, item_id text not null, primary key (page_num, pos) ); `; /** one cached page, raw: `items` are the dump rows (JSON strings), ready * to re-join with newlines */ export type CachePage = { total: number; items: string[]; /** true when answered from storage (no diigo call) */ cached: boolean; }; const MS_PER_CALL = 500; /** * items per diigo page — the one size the worker asks for everywhere * (fill, public pages, dumps), so a cache never mixes sizes. Pages are * cached by (page number, size): changing this turns every cached page * into a miss (items survive; pages refetch), and fails any fill job * counting in the old size. 100 is diigo's cap: asked for 128, it * returned 100 (observed 2026-10-07). */ export const PAGE_SIZE = 100; /** the size fill jobs counted in before FillStatus recorded it */ const LEGACY_FILL_PAGE_SIZE = 24; /** the pause before each fill step's alarm: a step already paces its * pages at MS_PER_CALL; this keeps steps from running back to back */ const FILL_STEP_DELAY_MS = 1000; /** * a fill job's progress (goal 1e; tests/fill.test.ts). Stored in the DO * (new state, additive), so it survives eviction; advanced by alarms. */ export type FillStatus = { state: "running" | "done" | "stopped" | "failed"; /** the next diigo page to fetch (pages already cached are skipped) */ nextPage: number; /** diigo's page count, known after page 1 */ pages: number | null; /** the page size `nextPage` and `pages` count in; absent on jobs * saved before this field existed (they counted in 24s) */ pageSize?: number; /** pages fetched from diigo by this job (cache hits not counted) */ fetched: number; startedAt: string; updatedAt: string; /** why it failed: safe to show and log (no ids, no header values) */ error?: string; }; /** the page size a fill job counts in (old jobs didn't record it) */ function fillPageSize(status: FillStatus): number { return status.pageSize ?? LEGACY_FILL_PAGE_SIZE; } /** release an RPC stub, if it is one (plain objects are left alone) */ function disposeStub(stub: unknown): void { const dispose = (stub as { [Symbol.dispose]?: () => void } | null)?.[Symbol.dispose]; if (typeof dispose === "function") dispose.call(stub); } /** the clock authority a cache uses: reading time, and waiting. Passed * in (ocap discipline), so tests can interpose a fixed clock. */ export type Clock = { /** epoch milliseconds */ now(): number; sleep(ms: number): Promise; }; /** a cache's wiring: the webapp instance it asks on a miss, plus (private * cache only) the public companion's id, recorded once published */ type Wiring = { webappId: string; publicId?: string | undefined }; /** the diigo-cache DO's bindings (structural; `at` mirrors webapp's * WebAppFactory) */ export type CacheEnv = { WEBAPP: { at(id: string): Promise; }; }; /** * the shared page-cache behavior. The webapp-instance id in the wiring * authorizes diigo reads (the cookie rides in that DO's storage); the id * this DO is addressed by authorizes cache reads. Neither is logged. * `setup()` persists the wiring (webapp's configure pattern); `page(n)` is * the read path; subclasses choose what to `keep` of a fetched page. */ abstract class PagedCache extends DurableObject { #step = 0; #pause?: Promise; #clock: Clock; /** `clock` is required: time comes from the edge that constructs us * (src/index.ts in the worker, tests/test-worker.ts in tests) */ constructor(ctx: DurableObjectState, env: CacheEnv, clock: Clock) { super(ctx, env); this.#clock = clock; } /** now, as an ISO timestamp */ #iso(): string { return new Date(this.#clock.now()).toISOString(); } /** one-shot wiring. Idempotent; different values are refused (mint a * new cache instead). */ async setup(init: Wiring): Promise { const stored = await this.ctx.storage.get("init"); if (stored) { if (stored.webappId !== init.webappId || stored.publicId !== init.publicId) { throw new Error("already wired to different ids; mint a new cache instead"); } return; } await this.ctx.storage.put("init", init); } /** the stored wiring */ async wiring(): Promise { return this.ctx.storage.get("init"); } /** which rows of a fetched page this cache stores */ protected abstract keep(rows: ItemRow[]): ItemRow[]; /** which cache this is, for logs (never the id: that's the bearer key) */ protected abstract readonly kind: "private" | "public"; /** * start (or rejoin) the fill job: walk every diigo page into this * cache, `pagesPerStep` per alarm (default 20, 1–50). Starting while a * job runs returns that job unchanged; a finished, stopped, or failed * job is replaced by a fresh one (cached pages are skipped anyway). */ async startFill(opts: { pagesPerStep?: number } = {}): Promise { const current = await this.fillStatus(); if (current?.state === "running" && fillPageSize(current) === PAGE_SIZE) return current; const step = Math.trunc(opts.pagesPerStep ?? 20); if (!(step >= 1 && step <= 50)) throw new Error("pagesPerStep: expected 1–50"); const now = this.#iso(); const status: FillStatus = { state: "running", nextPage: 1, pages: this.#knownPages(), pageSize: PAGE_SIZE, fetched: 0, startedAt: now, updatedAt: now, }; await this.ctx.storage.put({ fill: status, fillStep: step }); await this.ctx.storage.setAlarm(this.#clock.now() + FILL_STEP_DELAY_MS); console.log({ event: "fill", kind: this.kind, state: status.state, pagesPerStep: step }); return status; } /** the fill job's progress, or undefined if none was ever started; * the page count comes from any cached page until the job learns it */ async fillStatus(): Promise { const fill = await this.ctx.storage.get("fill"); return fill && fill.pages === null ? { ...fill, pages: this.#knownPages() } : fill; } /** stop the fill job after its current step; cached pages stay */ async stopFill(): Promise { const current = await this.fillStatus(); if (!current) throw new Error("no fill job to stop"); if (current.state !== "running") return current; const stopped: FillStatus = { ...current, state: "stopped", updatedAt: this.#iso() }; await this.ctx.storage.put("fill", stopped); await this.ctx.storage.deleteAlarm(); console.log({ event: "fill", kind: this.kind, state: "stopped", nextPage: stopped.nextPage, pages: stopped.pages }); return stopped; } /** one fill step: up to `fillStep` pages, progress saved after each, * then the next step is scheduled (or the job ends) */ async alarm(): Promise { const step = (await this.ctx.storage.get("fillStep")) ?? 20; let status = await this.fillStatus(); // a job saved under another page size counts its pages differently: // resuming it would skip or repeat items, so it fails instead if (status?.state === "running" && fillPageSize(status) !== PAGE_SIZE) { const error = `page size changed (from ${fillPageSize(status)} to ${PAGE_SIZE}); start the fill again`; status = { ...status, state: "failed", error, updatedAt: this.#iso() }; console.error({ event: "fill", kind: this.kind, state: "failed", error }); await this.ctx.storage.put("fill", status); return; } for (let i = 0; i < step && status?.state === "running"; i += 1) { if (status.pages !== null && status.nextPage > status.pages) { status = { ...status, state: "done" }; break; } try { const page = await this.page(status.nextPage, PAGE_SIZE); const fetched = status.fetched + (page.cached ? 0 : 1); // a page whose own total says the data ended before it (diigo // answers total 0 past the end) ends the walk; the count is the // last page with items. Judged by total, not by items: the // public cache may keep none of a page that isn't the end. if (page.total <= (status.nextPage - 1) * PAGE_SIZE) { status = { ...status, state: "done", pages: status.nextPage - 1, fetched, updatedAt: this.#iso() }; break; } status = { ...status, nextPage: status.nextPage + 1, pages: Math.ceil(page.total / PAGE_SIZE), fetched, updatedAt: this.#iso(), }; } catch (err) { status = { ...status, state: "failed", error: String(err), updatedAt: this.#iso() }; console.error({ event: "fill", kind: this.kind, state: "failed", page: status.nextPage, error: status.error }); } // a stop that landed mid-step wins if ((await this.fillStatus())?.state === "stopped") return; await this.ctx.storage.put("fill", status); } if (!status) return; if (status.state === "running" && status.pages !== null && status.nextPage > status.pages) { status = { ...status, state: "done", updatedAt: this.#iso() }; } await this.ctx.storage.put("fill", status); console.log({ event: "fill", kind: this.kind, state: status.state, nextPage: status.nextPage, pages: status.pages, fetched: status.fetched }); if (status.state === "running") await this.ctx.storage.setAlarm(this.#clock.now() + FILL_STEP_DELAY_MS); } /** diigo's page count, if any cached page recorded `total` (pages past * the end record 0, so they don't count) */ #knownPages(): number | null { this.ctx.storage.sql.exec(SCHEMA_V1); const row = this.ctx.storage.sql .exec<{ total: number | null }>("select total from pages where total > 0 order by etched_at desc limit 1") .toArray(); return row.length > 0 && row[0].total !== null ? Math.max(1, Math.ceil(row[0].total / PAGE_SIZE)) : null; } /** * read page `pageNum` (`count` per page, diigo's updated-desc order): * storage first; on miss, fetch that one page through the webapp * instance (every page carries `total`) and etch. Cache provenance * rides in the result (`cached`). */ async page(pageNum: number, count: number): Promise { const init = await this.wiring(); if (!init) throw new Error("not wired: this cache id was never set up"); this.ctx.storage.sql.exec(SCHEMA_V1); const hit = await this.#pageFromStorage(pageNum, count); if (hit) { console.log({ event: "cache", kind: this.kind, page: pageNum, outcome: "hit", items: hit.items.length }); return hit; } // a miss calls diigo: one call path at a time, min spacing // MS_PER_CALL (webapp's Gate pattern; per activation). Hits above // never wait. let before; do { before = this.#step; if (this.#pause) await this.#pause; } while (before !== this.#step); this.#step += 1; this.#pause = this.#clock.sleep(MS_PER_CALL); // a request ahead of us at the gate may have etched this page const late = await this.#pageFromStorage(pageNum, count); if (late) { console.log({ event: "cache", kind: this.kind, page: pageNum, outcome: "hit", items: late.items.length }); return late; } // RPC stubs (the app and each pipelined step) are disposed once the // page is in hand: undisposed stubs keep this DO from being evicted const data = await this.#fetchPage(init.webappId, pageNum, count); // diigo may cap the page size below what we asked: a short page that // isn't the last would silently skip items. But diigo's `total` can // also overstate what it returns, making the true last page look // short. The next page tells them apart: empty means the data ends // here; anything else means items would be skipped — refuse loudly. const lastPage = Math.max(1, Math.ceil(data.total / count)); if (data.items.length < count && pageNum < lastPage) { await this.#clock.sleep(MS_PER_CALL); const next = await this.#fetchPage(init.webappId, pageNum + 1, count); if (next.items.length > 0) { const error = `diigo returned ${data.items.length} items for page ${pageNum} of ${lastPage} (asked for ${count}), then ${next.items.length} for page ${pageNum + 1}: page size not honored?`; console.error({ event: "cache", kind: this.kind, page: pageNum, outcome: "short-page", fetched: data.items.length, asked: count, nextFetched: next.items.length }); throw new Error(error); } const actual = (pageNum - 1) * count + data.items.length; console.log({ event: "cache", kind: this.kind, page: pageNum, outcome: "total-overstated", total: data.total, actual }); } const kept = this.keep(toRows(data.items, console)); // fetched vs kept: the public cache's drop of private rows shows here console.log({ event: "cache", kind: this.kind, page: pageNum, outcome: "miss", fetched: data.items.length, kept: kept.length }); return this.#etch(pageNum, count, data.total, kept); } /** one diigo page, raw, through the webapp instance; the RPC stubs (the * app and each pipelined step) are disposed once the page is in hand — * undisposed stubs keep this DO from being evicted */ async #fetchPage(webappId: string, pageNum: number, count: number): Promise<{ total: number; items: Item[] }> { const app = await this.env.WEBAPP.at(webappId); const joined = app.pathjoin("load_user_items"); const queried = joined.query({ sort: "updated", count, page_num: pageNum }); try { return JSON.parse(await queried.get()) as { total: number; items: Item[] }; } finally { for (const stub of [queried, joined, app]) disposeStub(stub); } } /** etch a page (one transaction): upsert items, mark the page, keep * the fetch-time order; returns the cache-page shape */ #etch(pageNum: number, count: number, total: number, rows: ItemRow[]): CachePage { this.ctx.storage.transactionSync(() => { this.ctx.storage.sql.exec("delete from page_members where page_num = ?", pageNum); let pos = 0; for (const r of rows) { pos += 1; this.ctx.storage.sql.exec( "insert into items (item_id, url, title, private, json, updated) values (?, ?, ?, ?, ?, ?) on conflict(item_id) do update set url = excluded.url, title = excluded.title, private = excluded.private, json = excluded.json, updated = excluded.updated", r.itemId, r.url, r.title, r.private ? 1 : 0, r.json, r.updated, ); this.ctx.storage.sql.exec( "insert into page_members (page_num, pos, item_id) values (?, ?, ?)", pageNum, pos, r.itemId, ); } this.ctx.storage.sql.exec( "insert into pages (page_num, count, total, etched_at) values (?, ?, ?, ?) on conflict(page_num) do update set count = excluded.count, total = excluded.total, etched_at = excluded.etched_at", pageNum, count, total, this.#iso(), ); }); return { total, items: rows.map((r) => r.json), cached: false }; } /** storage-backed page read; null = cache miss. Replay order comes * from `page_members`. */ async #pageFromStorage(pageNum: number, count: number): Promise { const page = this.ctx.storage.sql .exec<{ total: number | null }>( "select total from pages where page_num = ? and count = ?", pageNum, count, ) .toArray(); if (page.length === 0) return null; const rows = this.ctx.storage.sql .exec<{ json: string }>( "select items.json from page_members join items on items.item_id = page_members.item_id where page_members.page_num = ? order by page_members.pos", pageNum, ) .toArray(); return { total: page[0].total ?? -1, items: rows.map((r) => r.json), cached: true }; } } /** * the private cache's logic (exported to the runtime as `DiigoCache` by * src/index.ts): public+private items, addressed by its cache id (a * webkey, held by the user). */ export class DiigoCacheLogic extends PagedCache { protected readonly kind = "private"; protected keep(rows: ItemRow[]): ItemRow[] { return rows; } /** one-way addition of the public companion (private → public * publishable later; once set, immutable). The webapp id already in * the wiring is what makes this work without re-minting. */ async addPublic(publicId: string): Promise { const stored = await this.ctx.storage.get("init"); if (!stored) throw new Error("not wired"); if (stored.publicId) { if (stored.publicId !== publicId) { throw new Error("already has a public cache"); } return; } await this.ctx.storage.put("init", { ...stored, publicId }); } } /** * the public cache: shared items only — {@link publicOnly} drops private * rows before they are stored, so this DO's storage holds nothing * private. Widely shared: addressed by diigo username in the worker's * route path (`/public/`). It does hold the webapp id (the same * cookie session the private cache uses), so it *could* read private * items; it keeps none. Public page n is diigo page n, filtered: it may * hold fewer than `count` items. Doubles as the username→instance index * (the idFromName("index") instance holds the kv mapping and is never * wired). */ export class PublicCacheLogic extends PagedCache { protected readonly kind = "public"; protected keep(rows: ItemRow[]): ItemRow[] { return publicOnly(rows); } /** register/replace the username → this-instance mapping (called on * the idFromName("index") instance) */ async register(username: string, publicId: string): Promise { this.ctx.storage.kv.put(`user:${username}`, publicId); } /** resolve a username to a public-cache instance id (index role); * undefined = not registered */ async lookup(username: string): Promise { return this.ctx.storage.kv.get(`user:${username}`); } }