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
This commit is contained in:
dckc committed
1 parent
6dd7a37532
commit
a17d287e1d
10 files changed
+275
-122
No files matched your search
+57
-114
@@ -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<CacheEnv> {
|
||||
|
||||
/** `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<CacheEnv> {
|
||||
*/
|
||||
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<CacheEnv> {
|
||||
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<number>("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<CacheEnv> {
|
||||
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.#schema();
|
||||
|
||||
const hit = await this.#pageFromStorage(pageNum, count);
|
||||
if (hit) {
|
||||
@@ -438,45 +379,47 @@ abstract class PagedCache extends DurableObject<CacheEnv> {
|
||||
/** 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<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 };
|
||||
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 };
|
||||
}
|
||||
|
||||
/** the cache's fill job, if any */
|
||||
fillStatus(): Promise<FillStatus | undefined> {
|
||||
return this.#fillStatus();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user