captoys/diigo-cache/src/cache.ts

/**
 * @file the cache Durable Objects' logic: a private and a public page cache.
 * The runtime classes are made at the edge (src/index.ts).
 */

import { DurableObject, RpcTarget } from "cloudflare:workers";
import type { PagedGet } from "./diigo.js";
import { publicOnly, toRows } from "./items.js";
import { makeReader, makeWriter, SCHEMA_VERSION } from "./store.js";
import type { Reader, Writer } from "./store.js";
import type { Item, ItemRow } from "./items.js";

export type { Item };

/** 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;
	/** when the page was stored (absent on a fresh fetch) */
	etchedAt?: string;
};

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;
	/** the page numbering `nextPage` counts in: 2 since diigo's 0-based
	 * page_num is translated; absent on jobs that sent it unchanged */
	numbering?: 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;
}

/** does a stored fill job count pages the way this code does? */
function fillCountsAsNow(status: FillStatus): boolean {
	return fillPageSize(status) === PAGE_SIZE && status.numbering === SCHEMA_VERSION;
}

/** 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; {@link DiigoCacheLogic} and
 * {@link PublicCacheLogic} are independent and differ only in what they
 * `keep` of a fetched page (the public one drops private rows before
 * storing, {@link publicOnly}). Pages are the unit of caching: a hit is
 * served from storage with no diigo call; a miss asks the webapp instance
 * in the cache's wiring and stores the page.
 *
 * The webapp 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. The {@link Clock} is required: time comes from
 * the edge that constructs the DO.
 */
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) */
	#reader: Reader;
	#writer: Writer;

	constructor(ctx: DurableObjectState, env: CacheEnv, clock: Clock) {
		super(ctx, env);
		this.#clock = clock;
		this.#reader = makeReader(ctx.storage.sql);
		this.#writer = makeWriter(ctx.storage);
		// tables exist (and old storage is migrated) before any request, so
		// a read-only viewer never meets a missing table
		void ctx.blockConcurrencyWhile(async () => {
			await null; // let the subclass constructor finish: `kind` is its field
			const from = this.#writer.migrate();
			if (from < SCHEMA_VERSION) console.log({ event: "schema", kind: this.kind, from, to: SCHEMA_VERSION });
		});
	}

	/** 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" && fillCountsAsNow(current)) 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,
			numbering: SCHEMA_VERSION,
			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;
	}

	/**
	 * drop the cached pages holding anything diigo updated in the last
	 * `days` days, and every page before them (newer still; and new
	 * updates push items back), so the next read or fill refetches them.
	 * Item rows stay. Returns how many pages were dropped.
	 */
	async invalidateRecent(days: number): Promise<{ pages: number }> {
		if (!Number.isInteger(days) || days < 1 || days > 3650) throw new Error("days: expected a whole number from 1 to 3650");
		const cutoff = Math.floor(this.#clock.now() / 1000) - days * 24 * 60 * 60;
		const deepest = this.#reader.deepestPageSince(cutoff);
		if (!deepest) {
			console.log({ event: "cache", kind: this.kind, outcome: "invalidated", days, pages: 0 });
			return { pages: 0 };
		}
		const pages = this.#reader.pagesThrough(deepest);
		this.#writer.dropPagesThrough(deepest);
		console.log({ event: "cache", kind: this.kind, outcome: "invalidated", days, pages });
		return { pages };
	}

	/** 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 or numbering counts its pages
		// differently: resuming it would skip or repeat items, so it fails
		if (status?.state === "running" && !fillCountsAsNow(status)) {
			const error =
				fillPageSize(status) !== PAGE_SIZE
					? `page size changed (from ${fillPageSize(status)} to ${PAGE_SIZE}); start the fill again`
					: "page numbering changed (diigo counts pages from 0); 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 {
		const total = this.#reader.knownTotal();
		return total === null ? null : Math.max(1, Math.ceil(total / PAGE_SIZE));
	}

	/**
	 * 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");

		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. If `total` ever
		// overstated what diigo returns, the true last page would look short
		// too. 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");
		// diigo's page_num counts from 0; ours from 1
		const queried = joined.query({ sort: "updated", count, page_num: pageNum - 1 });
		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.#writer.etch(pageNum, count, total, rows, 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 stored = this.#reader.page(pageNum, count);
		return stored && { ...stored, cached: true };
	}

	/**
	 * a read-only view of this cache, for pages anyone may read (the public
	 * page): stored pages and fill progress, from the reader facet only —
	 * it cannot fetch from diigo or change storage
	 */
	viewer(): PageViewer {
		return new PageViewer(this.#reader, () => this.fillStatus());
	}
}

/** what anyone may see of a cache: stored pages and fill progress */
export class PageViewer extends RpcTarget {
	#reader: Reader;
	#fillStatus: () => Promise<FillStatus | undefined>;

	constructor(reader: Reader, fillStatus: () => Promise<FillStatus | undefined>) {
		super();
		this.#reader = reader;
		this.#fillStatus = fillStatus;
	}

	/** a stored page, or null if it hasn't been fetched (at this size) */
	page(pageNum: number, count: number): CachePage | null {
		const stored = this.#reader.page(pageNum, count);
		return stored && { ...stored, cached: true };
	}

	/** diigo's total, as last stored, or null if nothing is stored */
	knownTotal(): number | null {
		return this.#reader.knownTotal();
	}

	/** the cache's fill job, if any */
	fillStatus(): Promise<FillStatus | undefined> {
		return this.#fillStatus();
	}
}

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