diigo-cache: bookmark cache in Durable Objects (goal 1c)
The dump worker becomes a page cache: a hit is served from DO SQLite storage with no diigo call; a miss asks the webapp instance in the cache's wiring and etches the page. - two independent caches, one behavior (PagedCache): DiigoCache keeps every item; PublicCache keeps shared items only, dropping private rows before they are stored; served at /public/<username> (HTML page form, or JSON) - create flow on one port: /_connect relays webapp's mint form, /_mint proxies the mint, /_wire wires a private or public cache; /_publish adds a public cache later - ids are webkeys: kept out of URLs (POST bodies), shown and logged only as a hash fingerprint; saved to the password manager under the holder's petname (never sent to the server) - items keyed on diigo's link_id; username from u_name; unkeyable items and every failure path are logged; each page read logs hit/miss - tests run the real worker + DOs in the vitest pool against a fake webapp (tests/fake-webapp.js) Generated-by: opencode (GLM) Generated-by: Claude Opus 5.5
This commit is contained in:
dckc committed
1 parent
c5e433cc9c
commit
6f409fc83a
15 files changed
+1664
-211
No files matched your search
@@ -0,0 +1,252 @@
|
||||
/**
|
||||
* @file the cache Durable Objects: `DiigoCache` (public+private items) and
|
||||
* `PublicCache` (public-only items, widely shared). 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;
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((r) => setTimeout(r, ms));
|
||||
}
|
||||
|
||||
/** 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>;
|
||||
|
||||
/** 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";
|
||||
|
||||
/**
|
||||
* read page `pageNum` (`count` per page, diigo's updated-desc order):
|
||||
* storage first; on miss, fetch through the webapp instance (etching
|
||||
* page 1 alongside — it 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 = 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;
|
||||
}
|
||||
|
||||
const app = await this.env.WEBAPP.at(init.webappId);
|
||||
const fetchPage = (n: number): Promise<{ total: number; items: Item[] }> =>
|
||||
app
|
||||
.pathjoin("load_user_items")
|
||||
.query({ sort: "updated", count, page_num: n })
|
||||
.get()
|
||||
.then((t) => JSON.parse(t)) as Promise<{ total: number; items: Item[] }>;
|
||||
|
||||
// a miss on page n>1 etches page 1 first (it carries `total`)
|
||||
const first = await fetchPage(1);
|
||||
const data = pageNum === 1 ? first : await fetchPage(pageNum);
|
||||
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);
|
||||
}
|
||||
|
||||
/** 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, new Date().toISOString(),
|
||||
);
|
||||
});
|
||||
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 (export name `DiigoCache`): public+private items,
|
||||
* addressed by its cache id (a webkey, held by the user).
|
||||
*/
|
||||
export class DiigoCache 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 PublicCache 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}`);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user