captoys/diigo-cache/src/cache.ts

/**
 * @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<void>;
};


/** 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<PagedGet>;
	};
};

/**
 * 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<CacheEnv> {
	#step = 0;
	#pause?: Promise<void>;
	#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<void> {
		const stored = await this.ctx.storage.get<Wiring>("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<Wiring | undefined> {
		return this.ctx.storage.get<Wiring>("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<FillStatus> {
		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<FillStatus | undefined> {
		const fill = await this.ctx.storage.get<FillStatus>("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<FillStatus> {
		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<void> {
		const step = (await this.ctx.storage.get<number>("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);
				status = {
					...status,
					nextPage: status.nextPage + 1,
					pages: Math.max(1, Math.ceil(page.total / PAGE_SIZE)),
					fetched: status.fetched + (page.cached ? 0 : 1),
					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` */
	#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 is not null 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<CachePage> {
		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);

		// one diigo call path at a time, min spacing MS_PER_CALL (webapp's
		// Gate pattern; per activation)
		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);

		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;
		}

		// 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<CachePage | null> {
		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<void> {
		const stored = await this.ctx.storage.get<Wiring>("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/<username>`). 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<void> {
		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<string | undefined> {
		return this.ctx.storage.kv.get<string>(`user:${username}`);
	}
}