1// Persistence on Durable Object SQLite. Every method is2// synchronous, so each call is one atomic write batch on the DO.3import { GRASP_SCHEMA } from "./grasp-state.ts";4import { canonical, expiration, isAddressable, isEphemeral, isPrivate, isReplaceable, tag, tagValues, type Event } from "./event.ts";5import { KIND_PUSH_REGISTRATION } from "./kinds.ts";6import { ftsQuery, searchTerms, type Filter } from "./filter.ts";7import { SITE_SCHEMA, SITE_KINDS } from "./sites.ts";8import { HLL } from "./hll.ts";9import { hexToBytes, type SyncItem } from "./negentropy.ts";10import { archiveCurrent, clearForDelete, clearList, isListKind, listHistory, restoreHistory, LIST_HISTORY_SCHEMA, type ListHistoryRow } from "./list-history.ts";1112export const SCHEMA = `13CREATE TABLE IF NOT EXISTS events (14 seq INTEGER PRIMARY KEY,15 id TEXT NOT NULL UNIQUE,16 pubkey TEXT NOT NULL,17 created_at INTEGER NOT NULL,18 kind INTEGER NOT NULL,19 d TEXT NOT NULL DEFAULT '',20 expires INTEGER NOT NULL DEFAULT 0,21 raw TEXT NOT NULL22);23CREATE INDEX IF NOT EXISTS ev_time ON events(created_at DESC);24CREATE INDEX IF NOT EXISTS ev_pk_kind ON events(pubkey, kind, created_at DESC);25CREATE INDEX IF NOT EXISTS ev_kind ON events(kind, created_at DESC);26CREATE INDEX IF NOT EXISTS ev_expires ON events(expires) WHERE expires > 0;27CREATE TABLE IF NOT EXISTS tags (28 event_id TEXT NOT NULL,29 name TEXT NOT NULL,30 value TEXT NOT NULL31);32CREATE INDEX IF NOT EXISTS tag_lookup ON tags(name, value, event_id);33CREATE INDEX IF NOT EXISTS tag_event ON tags(event_id);34CREATE VIRTUAL TABLE IF NOT EXISTS search USING fts5(content);35CREATE TRIGGER IF NOT EXISTS events_ad AFTER DELETE ON events BEGIN36 DELETE FROM tags WHERE event_id = old.id;37 DELETE FROM search WHERE rowid = old.seq;38 DELETE FROM marmot_principals WHERE event_id = old.id;39END;40CREATE TABLE IF NOT EXISTS marmot_principals (41 event_id TEXT PRIMARY KEY,42 pubkey TEXT NOT NULL43);44CREATE INDEX IF NOT EXISTS marmot_principal_pk ON marmot_principals(pubkey);45CREATE TABLE IF NOT EXISTS vanished (46 pubkey TEXT PRIMARY KEY,47 until INTEGER NOT NULL48);49`;5051export const ERR_DUPLICATE = "duplicate: already have this event";52export const ERR_REPLACED = "duplicate: a newer version exists";53export const ERR_DELETED = "blocked: event was previously deleted by its author";54export const ERR_VANISHED = "blocked: this pubkey asked for its events to be deleted";55export const ERR_TOO_BIG = "blocked: this query is too big, narrow the filter";5657// Access says who is asking. Private kinds only go to their parties unless all.58export interface Access {59 pubkeys: string[];60 all?: boolean;61}6263// Lists are bound as one JSON parameter: DO SQLite allows 100 bindings per statement.64const inList = (col: string) => `${col} IN (SELECT value FROM json_each(?))`;6566// searchable says which kinds go into the full-text index: the ones whose67// content is prose. Reactions, JSON blobs and private kinds are skipped; it68// keeps the index small and every event a few rows cheaper to write.69export const SEARCH_KINDS = new Set([0, 1, 11, 1111, 9802, 30023, 30024, 30818]);70export function searchable(kind: number): boolean {71 return SEARCH_KINDS.has(kind) && !isPrivate(kind);72}7374export class Store {75 private pending: SqlStorageCursor<Record<string, SqlStorageValue>>[] = [];76 // Event ids under a report hold: left out of every read. Settings owns the77 // Set; the relay hands the same instance here.78 hidden: Set<string> = new Set();79 // What the full-text index takes (settings.ts, features.search): every80 // public kind with content, the prose kinds, or nothing. Changing it81 // affects events from then on; the index is not rebuilt.82 searchMode: () => "full" | "prose" | "off" = () => "prose";8384 onSitesChanged: () => void = () => {};8586 private historySQL = <T extends Record<string, SqlStorageValue>>(q: string, ...args: unknown[]) => this.x<T>(q, ...args);8788 constructor(private sql: SqlStorage) {}8990 init() {91 this.sql.exec(SCHEMA);92 this.sql.exec(LIST_HISTORY_SCHEMA);93 try { this.sql.exec(`ALTER TABLE list_history ADD COLUMN expires INTEGER NOT NULL DEFAULT 0`); } catch { /* already migrated */ }94 this.sql.exec(SITE_SCHEMA);95 this.sql.exec(GRASP_SCHEMA);96 }9798 // x runs a statement and remembers its cursor so drain() can report rows99 // read and written (the DO's billing units) once the cursor is consumed.100 private x<T extends Record<string, SqlStorageValue>>(q: string, ...args: unknown[]): SqlStorageCursor<T> {101 const c = this.sql.exec<T>(q, ...args);102 this.pending.push(c as SqlStorageCursor<Record<string, SqlStorageValue>>);103 if (q.startsWith("DELETE FROM events") && c.rowsWritten && this.sql.exec(`SELECT 1 FROM site_outbox LIMIT 1`).toArray().length) this.onSitesChanged();104 return c;105 }106107 drain(): { rowsRead: number; rowsWritten: number } {108 let rowsRead = 0;109 let rowsWritten = 0;110 for (const c of this.pending) {111 rowsRead += c.rowsRead;112 rowsWritten += c.rowsWritten;113 }114 this.pending = [];115 return { rowsRead, rowsWritten };116 }117118 get databaseSize() {119 return this.sql.databaseSize;120 }121122 // save applies the NIP-01/09/40/62 storage rules. It returns "" on success,123 // or the OK reason for a benign rejection. Ephemeral events store nothing.124 save(e: Event, now: number): string {125 if (isEphemeral(e.kind)) return "";126 const has = (q: string, ...args: unknown[]) => this.x(q, ...args).toArray().length > 0;127 if (has(`SELECT 1 FROM events WHERE id=?`, e.id)) return ERR_DUPLICATE;128 if (has(`SELECT 1 FROM vanished WHERE pubkey=? AND until>=?`, e.pubkey, e.created_at)) return ERR_VANISHED;129 // NIP-09 tombstones: the author's own deletions, or a recipient's for gift wraps.130 const deleters = [e.pubkey, ...(e.kind === 1059 ? tagValues(e, "p") : [])];131 if (132 has(133 `SELECT 1 FROM tags t JOIN events d ON d.id=t.event_id WHERE d.kind=5 AND t.name='e' AND t.value=? AND ${inList("d.pubkey")} LIMIT 1`,134 e.id,135 JSON.stringify(deleters),136 )137 )138 return ERR_DELETED;139140 let d = "";141 if (isReplaceable(e.kind)) {142 if (has(`SELECT 1 FROM events WHERE pubkey=? AND kind=? AND (created_at>? OR (created_at=? AND id<?)) LIMIT 1`, e.pubkey, e.kind, e.created_at, e.created_at, e.id))143 return ERR_REPLACED;144 if (isListKind(e.kind)) {145 const current = this.x<{ raw: string }>(`SELECT raw FROM events WHERE pubkey=? AND kind=?`, e.pubkey, e.kind).toArray()[0];146 if (current) archiveCurrent(this.historySQL, JSON.parse(current.raw) as Event, now);147 }148 this.x(`DELETE FROM events WHERE pubkey=? AND kind=?`, e.pubkey, e.kind);149 } else if (isAddressable(e.kind)) {150 d = tag(e, "d");151 if (has(`SELECT 1 FROM events WHERE pubkey=? AND kind=? AND d=? AND (created_at>? OR (created_at=? AND id<?)) LIMIT 1`, e.pubkey, e.kind, d, e.created_at, e.created_at, e.id))152 return ERR_REPLACED;153 if (isListKind(e.kind)) {154 const current = this.x<{ raw: string }>(`SELECT raw FROM events WHERE pubkey=? AND kind=? AND d=?`, e.pubkey, e.kind, d).toArray()[0];155 if (current) archiveCurrent(this.historySQL, JSON.parse(current.raw) as Event, now);156 }157 this.x(`DELETE FROM events WHERE pubkey=? AND kind=? AND d=?`, e.pubkey, e.kind, d);158 } else if (e.kind === 5) {159 for (const t of e.tags) {160 if (t.length < 2) continue;161 if (t[0] === "e") {162 this.x(163 `DELETE FROM events WHERE id=? AND (pubkey=? OR (kind=1059 AND EXISTS (SELECT 1 FROM tags WHERE tags.event_id=events.id AND name='p' AND value=?)))`,164 t[1], e.pubkey, e.pubkey,165 );166 clearForDelete(this.historySQL, e.pubkey, t[1]);167 } else if (t[0] === "a") {168 const parts = t[1].split(":");169 if (parts.length >= 3 && parts[1] === e.pubkey) {170 const addressD = parts.slice(2).join(":");171 this.x(`DELETE FROM events WHERE kind=? AND pubkey=? AND d=? AND created_at<=?`, parseInt(parts[0], 10) || 0, e.pubkey, addressD, e.created_at);172 clearList(this.historySQL, e.pubkey, parseInt(parts[0], 10) || 0, addressD, e.created_at);173 }174 }175 }176 }177178 const { seq } = this.sql179 .exec<{ seq: number }>(`INSERT INTO events(id,pubkey,created_at,kind,d,expires,raw) VALUES(?,?,?,?,?,?,?) RETURNING seq`, e.id, e.pubkey, e.created_at, e.kind, d, expiration(e), canonical(e))180 .one();181 for (const t of e.tags) {182 if (t.length >= 2 && t[0].length === 1) this.x(`INSERT INTO tags(event_id,name,value) VALUES(?,?,?)`, e.id, t[0], t[1]);183 }184 if (SITE_KINDS.includes(e.kind)) this.onSitesChanged();185 const mode = this.searchMode();186 const indexed = mode === "full" ? !isPrivate(e.kind) : mode === "prose" && searchable(e.kind);187 if (e.content !== "" && indexed) this.x(`INSERT INTO search(rowid,content) VALUES(?,?)`, seq, e.content);188 return "";189 }190191 // vanish implements NIP-62.192 vanish(pubkey: string, until: number) {193 this.bytesByAuthor.clear();194 this.x(`INSERT INTO vanished(pubkey,until) VALUES(?,?) ON CONFLICT(pubkey) DO UPDATE SET until=max(until,excluded.until)`, pubkey, until);195 this.x(`DELETE FROM events WHERE pubkey=? AND created_at<=?`, pubkey, until);196 this.x(`DELETE FROM events WHERE kind=1059 AND id IN (SELECT event_id FROM tags WHERE name='p' AND value=?)`, pubkey);197 this.x(`DELETE FROM list_history WHERE owner=?`, pubkey);198 }199200 // sweepExpired deletes NIP-40 expired rows and returns the next expiry, or 0.201 sweepExpired(now: number): number {202 this.bytesByAuthor.clear();203 this.x(`DELETE FROM events WHERE expires > 0 AND expires <= ?`, now);204 const row = this.x<{ next: number | null }>(`SELECT MIN(expires) AS next FROM events WHERE expires > 0`).one();205 return row.next ?? 0;206 }207208 private where(f: Filter, who: Access, now: number): { conds: string[]; args: unknown[] } {209 const conds: string[] = [];210 const args: unknown[] = [];211 if (f.ids) {212 conds.push(inList("id"));213 args.push(JSON.stringify(f.ids));214 }215 if (f.authors) {216 conds.push(inList("pubkey"));217 args.push(JSON.stringify(f.authors));218 }219 if (f.kinds) {220 conds.push(inList("kind"));221 args.push(JSON.stringify(f.kinds));222 }223 if (f.since !== undefined) {224 conds.push("created_at >= ?");225 args.push(f.since);226 }227 if (f.until !== undefined) {228 conds.push("created_at <= ?");229 args.push(f.until);230 }231 for (const [name, vals] of Object.entries(f.tags)) {232 if (vals.length === 0) return { conds: ["0"], args: [] };233 conds.push(`EXISTS (SELECT 1 FROM tags WHERE tags.event_id=events.id AND name=? AND ${inList("value")})`);234 args.push(name, JSON.stringify(vals));235 }236 const terms = searchTerms(f.search);237 if (terms.length > 0) {238 conds.push("seq IN (SELECT rowid FROM search WHERE search MATCH ?)");239 args.push(ftsQuery(terms));240 }241 conds.push("NOT EXISTS (SELECT 1 FROM grasp_pending WHERE grasp_pending.id=events.id)");242 conds.push("(expires = 0 OR expires > ?)");243 args.push(now);244 if (this.hidden.size > 0) {245 conds.push(`id NOT IN (SELECT value FROM json_each(?))`);246 args.push(JSON.stringify([...this.hidden]));247 }248 // Push registrations are private to their author even for internal callers249 // that use all:true. A p tag never grants access to a registration, and an250 // empty pubkey list therefore excludes every registration.251 if (who.pubkeys.length === 0) {252 conds.push(`kind <> ${KIND_PUSH_REGISTRATION}`);253 } else {254 conds.push(`(kind <> ${KIND_PUSH_REGISTRATION} OR ${inList("pubkey")})`);255 args.push(JSON.stringify(who.pubkeys));256 }257 if (!who.all) {258 if (who.pubkeys.length === 0) conds.push("kind NOT IN (4,1059)");259 else {260 conds.push(`(kind NOT IN (4,1059) OR ${inList("pubkey")} OR EXISTS (SELECT 1 FROM tags WHERE tags.event_id=events.id AND name='p' AND ${inList("value")}))`);261 const list = JSON.stringify(who.pubkeys);262 args.push(list, list);263 }264 }265 return { conds, args };266 }267268 // query returns raw event JSON newest first, and whether more exist beyond the limit.269 query(f: Filter, who: Access, maxLimit: number, now: number): { rows: string[]; more: boolean } {270 const { conds, args } = this.where(f, who, now);271 let limit = maxLimit;272 if (f.limit !== undefined && f.limit < limit) limit = f.limit;273 if (limit < 0) limit = 0;274 const cur = this.x<{ raw: string }>(`SELECT raw FROM events WHERE ${conds.join(" AND ")} ORDER BY created_at DESC, id ASC LIMIT ?`, ...args, limit + 1);275 const rows: string[] = [];276 for (const r of cur) {277 if (rows.length === limit) return { rows, more: true };278 rows.push(r.raw);279 }280 return { rows, more: false };281 }282283 count(filters: Filter[], who: Access, now: number): number {284 const ors: string[] = [];285 const args: unknown[] = [];286 for (const f of filters) {287 const w = this.where(f, who, now);288 ors.push("(" + w.conds.join(" AND ") + ")");289 args.push(...w.args);290 }291 return this.x<{ n: number }>(`SELECT count(*) AS n FROM events WHERE ${ors.join(" OR ")}`, ...args).one().n;292 }293294 countHLL(f: Filter, who: Access, offset: number, now: number): { count: number; hll: string } {295 const { conds, args } = this.where(f, who, now);296 const h = new HLL(offset);297 let count = 0;298 for (const r of this.x<{ pubkey: string }>(`SELECT pubkey FROM events WHERE ${conds.join(" AND ")}`, ...args)) {299 h.add(r.pubkey);300 count++;301 }302 return { count, hll: h.hex() };303 }304305 // syncItems lists (created_at, id) for NIP-77; the filter's limit is ignored.306 syncItems(f: Filter, who: Access, max: number, now: number): SyncItem[] | "too big" {307 const { conds, args } = this.where(f, who, now);308 const items: SyncItem[] = [];309 for (const r of this.x<{ created_at: number; id: string }>(`SELECT created_at, id FROM events WHERE ${conds.join(" AND ")} ORDER BY created_at, id LIMIT ?`, ...args, max + 1)) {310 if (items.length === max) return "too big";311 items.push({ timestamp: r.created_at, id: hexToBytes(r.id) });312 }313 return items;314 }315316 stats(): { events: number; bytes: number; oldest: number; newest: number } {317 const row = this.x<{ n: number; oldest: number | null; newest: number | null }>(`SELECT count(*) AS n, min(created_at) AS oldest, max(created_at) AS newest FROM events WHERE kind <> ${KIND_PUSH_REGISTRATION}`).one();318 return { events: row.n, bytes: this.sql.databaseSize, oldest: row.oldest ?? 0, newest: row.newest ?? 0 };319 }320321 // kinds counts events per kind since a timestamp, busiest first, for the dashboard.322 kinds(since: number): { kind: number; n: number }[] {323 return this.x<{ kind: number; n: number }>(`SELECT kind, count(*) AS n FROM events WHERE created_at >= ? AND kind <> ${KIND_PUSH_REGISTRATION} GROUP BY kind ORDER BY n DESC LIMIT 50`, since).toArray();324 }325326 // kindCounts groups events since a timestamp by kind, for the dashboard.327 kindCounts(since: number): { kind: number; n: number }[] {328 return this.x<{ kind: number; n: number }>(`SELECT kind, count(*) AS n FROM events WHERE created_at >= ? AND kind <> ${KIND_PUSH_REGISTRATION} GROUP BY kind ORDER BY n DESC LIMIT 50`, since).toArray();329 }330331 // recent returns the newest events for the dashboard, regardless of visibility.332 // after returns up to `limit` rows with a sequence number past `seq`, in333 // insertion order, for jobs that walk the store with a cursor.334 after(seq: number, f: Filter, limit: number, now: number): { seq: number; raw: string }[] {335 const { conds, args } = this.where(f, { pubkeys: [], all: true }, now);336 return this.x<{ seq: number; raw: string }>(`SELECT seq, raw FROM events WHERE seq>? AND ${conds.join(" AND ")} ORDER BY seq LIMIT ?`, seq, ...args, limit).toArray();337 }338339 recent(limit: number, now: number): string[] {340 return this.x<{ raw: string }>(`SELECT raw FROM events WHERE kind <> ${KIND_PUSH_REGISTRATION} AND (expires = 0 OR expires > ?) ORDER BY created_at DESC, id ASC LIMIT ?`, now, limit).toArray().map((r) => r.raw);341 }342343 deleteEvent(id: string): boolean {344 this.bytesByAuthor.clear();345 return this.x(`DELETE FROM events WHERE id=?`, id).rowsWritten > 0;346 }347348 // authorBytes is how much raw event text one pubkey has stored, for the349 // per-member cap. Cached per author; save adds to it, and anything that350 // deletes clears the cache rather than guessing.351 private bytesByAuthor = new Map<string, number>();352 authorBytes(pubkey: string): number {353 const cached = this.bytesByAuthor.get(pubkey);354 if (cached !== undefined) return cached;355 const n = this.x<{ n: number | null }>(`SELECT COALESCE((SELECT sum(length(raw)) FROM events WHERE pubkey=?),0) + COALESCE((SELECT sum(length(e.raw)) FROM events e JOIN marmot_principals m ON m.event_id=e.id WHERE m.pubkey=?),0) AS n`, pubkey, pubkey).one().n ?? 0;356 this.bytesByAuthor.set(pubkey, n);357 return n;358 }359360 // notePrincipal ties an opaque group event to the authenticated account361 // whose storage allowance admitted it.362 notePrincipal(eventID: string, pubkey: string) {363 this.x(`INSERT OR REPLACE INTO marmot_principals(event_id,pubkey) VALUES(?,?)`, eventID, pubkey);364 this.bytesByAuthor.delete(pubkey);365 }366 // noteSaved keeps the cache right after a save. A replaceable kind may367 // have displaced an older row, so its author is simply recounted next time.368 noteSaved(pubkey: string, bytes: number, replaceable: boolean) {369 if (replaceable) {370 this.bytesByAuthor.delete(pubkey);371 return;372 }373 const cached = this.bytesByAuthor.get(pubkey);374 if (cached !== undefined) this.bytesByAuthor.set(pubkey, cached + bytes);375 }376377 // eraseAuthor deletes everything one pubkey wrote, profile included.378 eraseAuthor(pubkey: string): number {379 const n = this.x<{ n: number }>(`SELECT count(*) AS n FROM events WHERE pubkey=?`, pubkey).one().n;380 if (n > 0) {381 this.x(`DELETE FROM events WHERE pubkey=?`, pubkey);382 this.bytesByAuthor.delete(pubkey);383 }384 this.x(`DELETE FROM list_history WHERE owner=?`, pubkey);385 return n;386 }387388 // purgeAuthor deletes one pubkey's events created before `before`, except389 // the listed kinds. Returns how many went.390 purgeAuthor(pubkey: string, before: number, except: number[]): number {391 const list = JSON.stringify(except);392 const n = this.x<{ n: number }>(`SELECT count(*) AS n FROM events WHERE pubkey=? AND created_at<? AND kind NOT IN (SELECT value FROM json_each(?))`, pubkey, before, list).one().n;393 if (n > 0) {394 this.x(`DELETE FROM events WHERE pubkey=? AND created_at<? AND kind NOT IN (SELECT value FROM json_each(?))`, pubkey, before, list);395 this.bytesByAuthor.delete(pubkey);396 }397 this.x(`DELETE FROM list_history WHERE owner=? AND created_at<?`, pubkey, before);398 return n;399 }400401 // listHistory returns only metadata for versions signed by owner; raw event402 // content stays behind restoreHistory and the same owner check.403 listHistory(owner: string, now: number): Omit<ListHistoryRow, "owner" | "raw">[] {404 return listHistory(this.historySQL, owner, now);405 }406407 // restoreHistory returns a fresh unsigned draft and its current-list diff.408 restoreHistory(owner: string, eventID: string, now: number): ReturnType<typeof restoreHistory> {409 return restoreHistory(this.historySQL, owner, eventID, now);410 }411412 // dumpPage reads a page of raw events by sequence for the JSONL dump.413 dumpPage(afterSeq: number, limit: number): { seq: number; raw: string }[] {414 return this.x<{ seq: number; raw: string }>(`SELECT seq, raw FROM events WHERE seq>? AND kind <> ${KIND_PUSH_REGISTRATION} ORDER BY seq LIMIT ?`, afterSeq, limit).toArray();415 }416417 // kindStats sizes each kind present: how many, how many bytes of raw event,418 // and the span of timestamps. Heaviest first.419 kindStats(): { kind: number; n: number; bytes: number; oldest: number; newest: number }[] {420 return this.x<{ kind: number; n: number; bytes: number; oldest: number; newest: number }>(421 `SELECT kind, count(*) AS n, sum(length(raw)) AS bytes, min(created_at) AS oldest, max(created_at) AS newest FROM events WHERE kind <> ${KIND_PUSH_REGISTRATION} GROUP BY kind ORDER BY bytes DESC`,422 ).toArray();423 }424425 // purge deletes events of one kind, or of every kind not in `except` when426 // kind is null, created before `before`. Returns how many went.427 // exceptPubkey is the relay's own key: what it signed is never swept.428 purge(kind: number | null, before: number, except: number[] = [], exceptPubkey = ""): number {429 this.bytesByAuthor.clear();430 if (kind !== null) {431 const n = this.x<{ n: number }>(`SELECT count(*) AS n FROM events WHERE kind=? AND created_at<? AND pubkey<>?`, kind, before, exceptPubkey).one().n;432 if (n > 0) this.x(`DELETE FROM events WHERE kind=? AND created_at<? AND pubkey<>?`, kind, before, exceptPubkey);433 return n;434 }435 const list = JSON.stringify(except);436 const n = this.x<{ n: number }>(`SELECT count(*) AS n FROM events WHERE created_at<? AND pubkey<>? AND kind NOT IN (SELECT value FROM json_each(?))`, before, exceptPubkey, list).one().n;437 if (n > 0) this.x(`DELETE FROM events WHERE created_at<? AND pubkey<>? AND kind NOT IN (SELECT value FROM json_each(?))`, before, exceptPubkey, list);438 return n;439 }440}