From a17d287e1d9f4277ee1544821bd81fe4397a873c Mon Sep 17 00:00:00 2001 From: Dan Connolly Date: Wed, 7 Oct 2026 09:24:33 -0500 Subject: [PATCH] diigo-cache: SQL in .sql files, behind a read-only facet No SQL text in the cache logic any more. Statements live in two files, parsed by sqlStatements (`-- name:` sections, as in finquick's brcal/gcbook.js) and imported as text (wrangler Text rule for *.sql): - cache-read.sql: selects only, checked when the module loads - cache-write.sql: schema, the v2 renumbering, etch, drops store.ts turns them into two frozen facets with domain methods: makeReader(sql) runs only the read statements (page, knownTotal, deepestPageSince, pagesThrough); makeWriter(storage) only the write ones (migrate, etch, dropPagesThrough), each change one transaction. The cache migrates in its constructor under blockConcurrencyWhile, as Cloudflare recommends, so readers never meet missing tables; the migration test plants v1 storage and evicts to get a fresh activation. A viewer() facet (PageViewer: stored pages and fill progress, from the reader alone) is added for the public page. Pure refactor: stored state and behavior unchanged. Generated-by: Claude Opus 5.5 --- diigo-cache/src/cache-read.sql | 21 +++ diigo-cache/src/cache-write.sql | 51 +++++++ diigo-cache/src/cache.ts | 171 ++++++++--------------- diigo-cache/src/sql-statements.ts | 18 +++ diigo-cache/src/sql.d.ts | 5 + diigo-cache/src/store.ts | 101 +++++++++++++ diigo-cache/tests/caches.test.ts | 11 +- diigo-cache/tests/sql-statements.test.ts | 15 ++ diigo-cache/wrangler.jsonc | 2 +- diigo-cache/wrangler.tests.jsonc | 2 +- 10 files changed, 275 insertions(+), 122 deletions(-) create mode 100644 diigo-cache/src/cache-read.sql create mode 100644 diigo-cache/src/cache-write.sql create mode 100644 diigo-cache/src/sql-statements.ts create mode 100644 diigo-cache/src/sql.d.ts create mode 100644 diigo-cache/src/store.ts create mode 100644 diigo-cache/tests/sql-statements.test.ts diff --git a/diigo-cache/src/cache-read.sql b/diigo-cache/src/cache-read.sql new file mode 100644 index 0000000..f361f55 --- /dev/null +++ b/diigo-cache/src/cache-read.sql @@ -0,0 +1,21 @@ +-- pageHeader: +select total, etched_at from pages where page_num = ? and count = ?; + +-- pageItems: +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; + +-- knownTotal: +select total from pages where total > 0 order by etched_at desc limit 1; + +-- deepestPageSince: +select max(page_members.page_num) as last +from page_members +join items on items.item_id = page_members.item_id +where cast(items.updated as integer) >= ?; + +-- pagesThrough: +select count(*) as n from pages where page_num <= ?; diff --git a/diigo-cache/src/cache-write.sql b/diigo-cache/src/cache-write.sql new file mode 100644 index 0000000..5af7a0f --- /dev/null +++ b/diigo-cache/src/cache-write.sql @@ -0,0 +1,51 @@ +-- schemaV1: +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) +); + +-- renumberV2: +update page_members set page_num = -(page_num + 1); +update page_members set page_num = -page_num; +update pages set page_num = -(page_num + 1); +update pages set page_num = -page_num; + +-- clearPageMembers: +delete from page_members where page_num = ?; + +-- upsertItem: +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; + +-- insertPageMember: +insert into page_members (page_num, pos, item_id) values (?, ?, ?); + +-- upsertPage: +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; + +-- dropPageMembersThrough: +delete from page_members where page_num <= ?; + +-- dropPagesThrough: +delete from pages where page_num <= ?; diff --git a/diigo-cache/src/cache.ts b/diigo-cache/src/cache.ts index 6d3c277..648ad74 100644 --- a/diigo-cache/src/cache.ts +++ b/diigo-cache/src/cache.ts @@ -11,56 +11,15 @@ * private rows *before* etching ({@link publicOnly}). */ -import { DurableObject } from "cloudflare:workers"; +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 }; -/** 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) -); -`; - -/** - * V2: renumber stored pages +1. Pages are 1-based here, but diigo's - * `page_num` counts from 0, and V1 sent our page number unchanged — so a - * V1 "page n" holds what is really page n+1 (and the newest page was never - * fetched). Via negation, so no row's new number collides with an old one. - */ -const SCHEMA_V2_RENUMBER = ` -update page_members set page_num = -(page_num + 1); -update page_members set page_num = -page_num; -update pages set page_num = -(page_num + 1); -update pages set page_num = -page_num; -`; - -/** the schema version a cache's storage is at (kv key "schema") */ -const SCHEMA_VERSION = 2; - /** one cached page, raw: `items` are the dump rows (JSON strings), ready * to re-join with newlines */ export type CachePage = { @@ -68,6 +27,8 @@ export type CachePage = { 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; @@ -164,9 +125,21 @@ abstract class PagedCache extends DurableObject { /** `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 */ @@ -253,25 +226,14 @@ abstract class PagedCache extends DurableObject { */ 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"); - this.#schema(); const cutoff = Math.floor(this.#clock.now() / 1000) - days * 24 * 60 * 60; - const sql = this.ctx.storage.sql; - const deepest = sql - .exec<{ last: number | null }>( - "select max(page_members.page_num) as last from page_members join items on items.item_id = page_members.item_id where cast(items.updated as integer) >= ?", - cutoff, - ) - .toArray()[0]?.last; + const deepest = this.#reader.deepestPageSince(cutoff); if (!deepest) { console.log({ event: "cache", kind: this.kind, outcome: "invalidated", days, pages: 0 }); return { pages: 0 }; } - let pages = 0; - this.ctx.storage.transactionSync(() => { - pages = Number(sql.exec<{ n: number }>("select count(*) as n from pages where page_num <= ?", deepest).toArray()[0].n); - sql.exec("delete from page_members where page_num <= ?", deepest); - sql.exec("delete from pages where page_num <= ?", deepest); - }); + const pages = this.#reader.pagesThrough(deepest); + this.#writer.dropPagesThrough(deepest); console.log({ event: "cache", kind: this.kind, outcome: "invalidated", days, pages }); return { pages }; } @@ -333,31 +295,11 @@ abstract class PagedCache extends DurableObject { if (status.state === "running") await this.ctx.storage.setAlarm(this.#clock.now() + FILL_STEP_DELAY_MS); } - #schemaReady = false; - - /** create tables, and migrate stored state once per activation */ - #schema(): void { - if (this.#schemaReady) return; - this.ctx.storage.sql.exec(SCHEMA_V1); - const version = this.ctx.storage.kv.get("schema") ?? 1; - if (version < 2) { - this.ctx.storage.transactionSync(() => { - this.ctx.storage.sql.exec(SCHEMA_V2_RENUMBER); - this.ctx.storage.kv.put("schema", SCHEMA_VERSION); - }); - console.log({ event: "schema", kind: this.kind, from: version, to: SCHEMA_VERSION }); - } - this.#schemaReady = true; - } - /** 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.#schema(); - 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; + const total = this.#reader.knownTotal(); + return total === null ? null : Math.max(1, Math.ceil(total / PAGE_SIZE)); } /** @@ -369,7 +311,6 @@ abstract class PagedCache extends DurableObject { 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.#schema(); const hit = await this.#pageFromStorage(pageNum, count); if (hit) { @@ -438,45 +379,47 @@ abstract class PagedCache extends DurableObject { /** 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(), - ); - }); + 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 { - 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 }; + 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; + + constructor(reader: Reader, fillStatus: () => Promise) { + 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 }; + } + + /** the cache's fill job, if any */ + fillStatus(): Promise { + return this.#fillStatus(); } } diff --git a/diigo-cache/src/sql-statements.ts b/diigo-cache/src/sql-statements.ts new file mode 100644 index 0000000..31f2748 --- /dev/null +++ b/diigo-cache/src/sql-statements.ts @@ -0,0 +1,18 @@ +/** + * named statements from a .sql file: each `-- name:` line starts one, + * running to the next (a trailing `;` dropped). As `sqlStatements` in + * finquick's brcal/src/gcbook.js. + */ +export function sqlStatements(text: string): Record { + const byName: Record = {}; + const chop = (lo: number, hi: number) => text.substring(lo, hi).replace(/;\s*$/, "").trim(); + let pos = 0; + let name: string | undefined; + for (const m of text.matchAll(/^-- (?[^:\n]+):\n/gm)) { + if (name) byName[name] = chop(pos, m.index); + pos = m.index + m[0].length; + name = m.groups?.name; + } + if (name) byName[name] = chop(pos, text.length); + return byName; +} diff --git a/diigo-cache/src/sql.d.ts b/diigo-cache/src/sql.d.ts new file mode 100644 index 0000000..ec1d3df --- /dev/null +++ b/diigo-cache/src/sql.d.ts @@ -0,0 +1,5 @@ +/** .sql files import as their text (wrangler.jsonc "rules": Text) */ +declare module "*.sql" { + const text: string; + export default text; +} diff --git a/diigo-cache/src/store.ts b/diigo-cache/src/store.ts new file mode 100644 index 0000000..a2de7ff --- /dev/null +++ b/diigo-cache/src/store.ts @@ -0,0 +1,101 @@ +/** + * @file a cache DO's SQL storage, as two facets: {@link makeReader} runs + * only the statements in cache-read.sql (all `select`s, checked at load), + * {@link makeWriter} only those in cache-write.sql. No SQL text lives in + * the cache logic; whoever holds only a reader cannot change storage. + */ + +import readText from "./cache-read.sql"; +import writeText from "./cache-write.sql"; +import type { ItemRow } from "./items.js"; +import { sqlStatements } from "./sql-statements.js"; + +const READ = sqlStatements(readText); +const WRITE = sqlStatements(writeText); +for (const [name, statement] of Object.entries(READ)) { + if (!/^(select|with)\b/i.test(statement)) throw new Error(`cache-read.sql: ${name} is not a select`); +} + +/** the SQL surface the facets need (a DO's `ctx.storage.sql`) */ +type Sql = Pick; + +/** one cached page, raw: `items` are the dump rows (JSON strings) */ +export type StoredPage = { total: number; items: string[]; etchedAt: string }; + +/** read-only access to a cache's pages and items */ +export function makeReader(sql: Sql) { + const run = >(name: string, ...params: SqlStorageValue[]): T[] => + sql.exec(READ[name], ...params).toArray(); + return Object.freeze({ + /** a cached page in its fetch-time order, or null if not cached at this size */ + page(pageNum: number, count: number): StoredPage | null { + const [header] = run<{ total: number | null; etched_at: string }>("pageHeader", pageNum, count); + if (!header) return null; + const rows = run<{ json: string }>("pageItems", pageNum); + return { total: header.total ?? -1, items: rows.map((r) => r.json), etchedAt: header.etched_at }; + }, + /** diigo's total, from the most recently cached page that had one */ + knownTotal(): number | null { + return run<{ total: number }>("knownTotal")[0]?.total ?? null; + }, + /** the deepest cached page holding an item updated at or after `cutoff` (epoch seconds) */ + deepestPageSince(cutoff: number): number | null { + return run<{ last: number | null }>("deepestPageSince", cutoff)[0]?.last ?? null; + }, + /** how many cached pages are numbered `pageNum` or lower */ + pagesThrough(pageNum: number): number { + return Number(run<{ n: number }>("pagesThrough", pageNum)[0]?.n ?? 0); + }, + }); +} +export type Reader = ReturnType; + +/** the schema version storage is at (kv key "schema") */ +export const SCHEMA_VERSION = 2; + +/** write access to a cache's pages and items, each change one transaction */ +export function makeWriter(storage: { + sql: Sql; + kv: Pick; + transactionSync(fn: () => T): T; +}) { + const run = (name: string, ...params: SqlStorageValue[]) => { + storage.sql.exec(WRITE[name], ...params); + }; + return Object.freeze({ + /** create tables, and migrate older storage; returns the version + * migrated from (= SCHEMA_VERSION when nothing was to do) */ + migrate(): number { + run("schemaV1"); + const version = storage.kv.get("schema") ?? 1; + if (version < 2) { + // V1 sent our 1-based page number as diigo's 0-based page_num, + // so a stored "page n" is really page n+1 + storage.transactionSync(() => { + run("renumberV2"); + storage.kv.put("schema", SCHEMA_VERSION); + }); + } + return version; + }, + /** store a fetched page: upsert its items, record their order */ + etch(pageNum: number, count: number, total: number, rows: ItemRow[], at: string): void { + storage.transactionSync(() => { + run("clearPageMembers", pageNum); + rows.forEach((r, i) => { + run("upsertItem", r.itemId, r.url, r.title, r.private ? 1 : 0, r.json, r.updated); + run("insertPageMember", pageNum, i + 1, r.itemId); + }); + run("upsertPage", pageNum, count, total, at); + }); + }, + /** drop pages numbered `pageNum` or lower (their items stay) */ + dropPagesThrough(pageNum: number): void { + storage.transactionSync(() => { + run("dropPageMembersThrough", pageNum); + run("dropPagesThrough", pageNum); + }); + }, + }); +} +export type Writer = ReturnType; diff --git a/diigo-cache/tests/caches.test.ts b/diigo-cache/tests/caches.test.ts index c8d93d4..b7ad35a 100644 --- a/diigo-cache/tests/caches.test.ts +++ b/diigo-cache/tests/caches.test.ts @@ -133,19 +133,18 @@ describe("page numbering (schema v2)", () => { it("migrates v1 storage: a stored 'page n' was really page n+1, so it is renumbered, not refetched", async () => { const stub = testEnv.CACHE_DO.get(testEnv.CACHE_DO.newUniqueId()); await stub.setup({ webappId: WEBAPP_ID }); - // the shape v1 left: tables, a page 1 holding what diigo's - // page_num=1 returned (really page 2), and no schema marker + // the shape v1 left: a page 1 holding what diigo's page_num=1 + // returned (really page 2), and no schema marker; then a fresh + // activation, whose constructor migrates await runInDurableObject(stub, (_obj, state) => { const sql = state.storage.sql; - sql.exec(`create table 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)`); - sql.exec("create table pages (page_num integer primary key, count integer not null, total integer, etched_at text not null)"); - sql.exec("create table page_members (page_num integer not null, pos integer not null, item_id text not null, primary key (page_num, pos))"); sql.exec("insert into items values ('a', 'https://a.example/', 'old page one, first', 0, '{\"title\":\"A\"}', '1')"); sql.exec("insert into items values ('b', 'https://b.example/', 'old page one, second', 0, '{\"title\":\"B\"}', '1')"); sql.exec("insert into pages values (1, 100, 250, '2026-10-07T00:00:00.000Z')"); sql.exec("insert into page_members values (1, 1, 'a'), (1, 2, 'b')"); + state.storage.kv.delete("schema"); }); + await evictDurableObject(stub); const before = await testEnv.WEBAPP.calls(); const two = await stub.page(2, PAGE_SIZE); expect(two.cached).toBe(true); diff --git a/diigo-cache/tests/sql-statements.test.ts b/diigo-cache/tests/sql-statements.test.ts new file mode 100644 index 0000000..80a3e22 --- /dev/null +++ b/diigo-cache/tests/sql-statements.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, it } from "vitest"; +import { sqlStatements } from "../src/sql-statements.js"; + +describe("sqlStatements", () => { + it("splits a .sql file into named statements at each `-- name:` line, dropping the trailing ;", () => { + const text = "-- first:\nselect 1;\n\n-- second:\nselect *\nfrom t\nwhere a = ?;\n"; + expect(sqlStatements(text)).toEqual({ first: "select 1", second: "select *\nfrom t\nwhere a = ?" }); + }); + + it("keeps a multi-statement entry whole (inner semicolons stay)", () => { + expect(sqlStatements("-- both:\nupdate a set x = 1;\nupdate b set y = 2;\n")).toEqual({ + both: "update a set x = 1;\nupdate b set y = 2", + }); + }); +}); diff --git a/diigo-cache/wrangler.jsonc b/diigo-cache/wrangler.jsonc index b3b4d05..4e1baed 100644 --- a/diigo-cache/wrangler.jsonc +++ b/diigo-cache/wrangler.jsonc @@ -6,7 +6,7 @@ "observability": { "enabled": true }, // capnweb's browser build, served as text at /_capnweb.js (the worker // itself imports capnweb's workerd build as code) - "rules": [{ "type": "Text", "globs": ["**/capnweb/dist/index.js"], "fallthrough": true }], + "rules": [{ "type": "Text", "globs": ["**/capnweb/dist/index.js", "**/*.sql"], "fallthrough": true }], "services": [ { "binding": "WEBAPP", diff --git a/diigo-cache/wrangler.tests.jsonc b/diigo-cache/wrangler.tests.jsonc index 1a6823a..993c1a5 100644 --- a/diigo-cache/wrangler.tests.jsonc +++ b/diigo-cache/wrangler.tests.jsonc @@ -6,7 +6,7 @@ "name": "diigo-cache-tests", "main": "./tests/test-worker.ts", "compatibility_date": "2026-09-29", - "rules": [{ "type": "Text", "globs": ["**/capnweb/dist/index.js"], "fallthrough": true }], + "rules": [{ "type": "Text", "globs": ["**/capnweb/dist/index.js", "**/*.sql"], "fallthrough": true }], "services": [ { "binding": "WEBAPP", "service": "fake-webapp", "entrypoint": "WebAppFactory" } ],