1// NIP-9a relay push: bounded registrations, durable work and authenticated delivery.2//3// The queue keeps references rather than event payloads. A deleted event or4// registration therefore disappears at the next validation pass, and a5// callback never receives an event that the author can no longer read.6import { canonical, type Event } from "./event.ts";7import { match, parseFilter, type Filter } from "./filter.ts";8import { callbackOrigin } from "./push-policy.ts";910export const PUSH_KIND = KIND_PUSH_REGISTRATION;11export const PUSH_SCHEMA = `12CREATE TABLE IF NOT EXISTS push_queue (13 registration_id TEXT NOT NULL,14 event_id TEXT NOT NULL,15 due INTEGER NOT NULL,16 attempts INTEGER NOT NULL DEFAULT 0,17 expires INTEGER NOT NULL,18 PRIMARY KEY (registration_id,event_id)19);20CREATE INDEX IF NOT EXISTS push_queue_due ON push_queue(due,expires);21CREATE TABLE IF NOT EXISTS push_delivered (22 registration_id TEXT NOT NULL,23 event_id TEXT NOT NULL,24 delivered_at INTEGER NOT NULL,25 PRIMARY KEY (registration_id,event_id)26);27CREATE INDEX IF NOT EXISTS push_delivered_at ON push_delivered(delivered_at);28CREATE TRIGGER IF NOT EXISTS push_event_deleted AFTER DELETE ON events BEGIN29 DELETE FROM push_queue WHERE registration_id=old.id OR event_id=old.id;30 DELETE FROM push_delivered WHERE registration_id=old.id;31END;32CREATE TRIGGER IF NOT EXISTS push_member_removed AFTER DELETE ON members BEGIN33 DELETE FROM events WHERE kind=30390 AND pubkey=old.pubkey;34END;35CREATE TRIGGER IF NOT EXISTS push_policy_changed AFTER UPDATE OF value ON settings36WHEN new.key='policy' AND (37 json_extract(new.value,'$.features.push') IS NOT 1 OR38 json_extract(new.value,'$.owner') IS NOT json_extract(old.value,'$.owner') OR39 json_extract(new.value,'$.pushCallbacks') IS NOT json_extract(old.value,'$.pushCallbacks') OR40 json_extract(new.value,'$.reads') IS NOT json_extract(old.value,'$.reads') OR41 json_extract(new.value,'$.writes') IS NOT json_extract(old.value,'$.writes')42) BEGIN43 DELETE FROM push_queue;44END;45`;4647const MAX_REGISTRATIONS = 32;48const MAX_AUTHOR_REGISTRATIONS = 4;49const MAX_FILTERS = 8;50const MAX_REGISTRATION_BYTES = 8192;51const MAX_PENDING = 256;52// A NIP event can occupy up to the relay's one MiB message ceiling. The53// larger bound keeps include_event lossless while retaining a hard callback54// allocation limit and room for JSON envelope overhead.55const MAX_EVENT_TEXT = 1024 * 1024;56const MAX_PAYLOAD = 4 * MAX_EVENT_TEXT + 4096;57const MAX_DEDUP = 2048;58const MAX_ATTEMPTS = 4;59const RETRY_SECONDS = [30, 120, 600];60const QUEUE_TTL = 24 * 60 * 60;61const DEDUP_TTL = 7 * 24 * 60 * 60;62const REQUEST_TIMEOUT = 5000;6364// All authorization uses the relay's current policy and event read path.65import type { Relay, ConnState } from "./relay.ts";66import { KIND_PUSH_REGISTRATION } from "./kinds.ts";67export type PushRelay = Relay;68export type PushConn = Pick<ConnState, "authed">;6970// rows consumes each SQL cursor before recording its billing units.71function rows<T extends Record<string, SqlStorageValue>>(r: Relay, query: string, ...args: SqlStorageValue[]): T[] {72 const cursor = r.sql.exec<T>(query, ...args);73 const result = cursor.toArray();74 r.meterPush(cursor.rowsRead, cursor.rowsWritten);75 return result;76}77const utf8Size = (s: string) => new TextEncoder().encode(s).byteLength;7879const tagValues = (e: Event, name: string) => e.tags.filter((t) => t[0] === name && t.length > 1).map((t) => t[1]);80const tag = (e: Event, name: string) => tagValues(e, name)[0] ?? "";81const featureEnabled = (r: PushRelay) => r.settings.policy.features.push === true;82const ownRelay = (r: PushRelay) => {83 try { return new URL(r.relayURL(r.slug + "." + r.domain)); }84 catch { return null; }85};8687function originAllowed(r: PushRelay, raw: string): string {88 const origin = callbackOrigin(raw);89 if (!origin) return "";90 const host = r.pushCallbackOrigins ?? [];91 const owner = r.settings.policy.pushCallbacks ?? [];92 return host.includes(origin) && owner.includes(origin) ? (() => { try { return new URL(raw).href; } catch { return ""; } })() : "";93}9495function parseRegistration(e: Event, r: PushRelay): { relay: string; filters: Filter[]; ignores: Filter[]; callback: string; include: boolean } | string {96 if (e.kind !== PUSH_KIND) return "invalid: not a push registration";97 if (utf8Size(canonical(e)) > MAX_REGISTRATION_BYTES) return "invalid: push registration is too large";98 const d = tagValues(e, "d");99 const relays = tagValues(e, "relay");100 const callbacks = tagValues(e, "callback");101 if (d.length !== 1 || !d[0] || d[0].length > 256) return "invalid: push registration needs one d tag";102 if (!relays.length || relays.length > 4 || callbacks.length !== 1) return "invalid: push registration needs relay and callback tags";103 const self = ownRelay(r);104 const relayURL = relays.map((raw) => { try { return new URL(raw); } catch { return null; } }).find((u) => u && /^wss?:$/.test(u.protocol) && u.toString() === self?.toString())?.toString() ?? "";105 if (!self || !relayURL) return "invalid: registration does not name this relay";106 const callback = originAllowed(r, callbacks[0]);107 if (!callback) return "restricted: callback origin is not approved by this relay";108 const read = (name: string): Filter[] | string => {109 const vals = tagValues(e, name);110 if (vals.length > MAX_FILTERS) return "invalid: too many push filters";111 const out: Filter[] = [];112 for (const raw of vals) {113 let parsed: unknown;114 try { parsed = JSON.parse(raw); } catch { return "invalid: push filter is not JSON"; }115 if (!parsed || typeof parsed !== "object" || Array.isArray(parsed) || Object.keys(parsed as Record<string, unknown>).some((k) => !(["ids", "authors", "kinds", "since", "until"].includes(k) || (k.length === 2 && k[0] === "#")))) return "invalid: unknown push filter field";116 const f = parseFilter(parsed);117 if (typeof f === "string") return "invalid: bad push filter: " + f;118 if (f.kinds?.some((n) => n < 0 || n > 65535) || (f.since !== undefined && f.since < 0) || (f.until !== undefined && f.until < 0)) return "invalid: push filter value out of range";119 out.push(f);120 }121 return out;122 };123 const filters = read("filter");124 const ignores = read("ignore");125 if (typeof filters === "string") return filters;126 if (typeof ignores === "string") return ignores;127 if (!filters.length) return "invalid: push registration needs a filter";128 return { relay: relayURL, filters, ignores, callback, include: e.tags.some((t) => t[0] === "include_event") };129}130131132// checkPush validates the opt-in registration. Its caller still applies the133// ordinary write gate, while this gate requires the author's AUTH identity.134export function checkPush(r: PushRelay, e: Event, conn: PushConn | null): string {135 if (e.kind !== PUSH_KIND) return "";136 if (!featureEnabled(r)) return "unsupported: relay push is switched off";137 if (!conn?.authed.includes(e.pubkey)) return "auth-required: push registration must be authenticated as its author";138 if (!r.settings.isAllowed(e.pubkey)) return "restricted: push registrations are limited to relay members";139 const parsed = parseRegistration(e, r);140 if (typeof parsed === "string") return parsed;141 const total = rows<{ n: number }>(r, `SELECT count(*) AS n FROM events WHERE kind=?`, PUSH_KIND)[0].n;142 if (total >= MAX_REGISTRATIONS && !rows(r, `SELECT 1 FROM events WHERE pubkey=? AND kind=? AND d=?`, e.pubkey, PUSH_KIND, tag(e, "d")).length) return "restricted: relay push registration limit reached";143 const mine = rows<{ n: number }>(r, `SELECT count(*) AS n FROM events WHERE kind=? AND pubkey=?`, PUSH_KIND, e.pubkey)[0].n;144 if (mine >= MAX_AUTHOR_REGISTRATIONS && !rows(r, `SELECT 1 FROM events WHERE pubkey=? AND kind=? AND d=?`, e.pubkey, PUSH_KIND, tag(e, "d")).length) return "restricted: author push registration limit reached";145 return "";146}147148// authorized checks current membership and both read/write policy. Callback149// destinations never inherit the owner's broad export privileges.150function authorized(r: Relay, pubkey: string): boolean {151 return !r.settings.isUnclaimed() && !r.settings.leaseExpired(Math.floor(Date.now() / 1000)) && r.settings.isAllowed(pubkey) && !r.settings.isBanned(pubkey) && !r.settings.mayRead([pubkey]) && !r.settings.mayWrite(pubkey);152}153154function visible(r: Relay, id: string, author: string, t: number): Event | null {155 if (r.settings.isEventBanned(id)) return null;156 // Imports and the HTTP bridge may hold events larger than websocket157 // messages. Do not materialize those rows for callback fanout.158 if (!rows(r, `SELECT 1 FROM events WHERE id=? AND length(raw)<=?`, id, MAX_EVENT_TEXT).length) return null;159 const raw = r.store.query({ ids: [id], tags: {} }, { pubkeys: [author] }, 1, t).rows[0];160 return raw ? JSON.parse(raw) as Event : null;161}162163// queuePush persists bounded references without any callback I/O. A true164// return asks the caller to keep alarm scheduling alive beyond its response.165export function queuePush(r: Relay, e: Event): boolean {166 if (!featureEnabled(r) || e.kind === PUSH_KIND) return false;167 const t = Math.floor(Date.now() / 1000);168 const registrations = rows<{ id: string; pubkey: string }>(r, `SELECT id,pubkey FROM events WHERE kind=? ORDER BY created_at DESC LIMIT ?`, PUSH_KIND, MAX_REGISTRATIONS);169 let added = false;170 for (const row of registrations) {171 if (!authorized(r, row.pubkey)) continue;172 const reg = visible(r, row.id, row.pubkey, t);173 if (!reg || !visible(r, e.id, row.pubkey, t)) continue;174 const p = parseRegistration(reg, r);175 if (typeof p === "string" || !p.filters.some((f) => match(f, e)) || p.ignores.some((f) => match(f, e))) continue;176 if (rows(r, `SELECT 1 FROM push_delivered WHERE registration_id=? AND event_id=?`, row.id, e.id).length) continue;177 if (rows<{ n: number }>(r, `SELECT count(*) AS n FROM push_queue`)[0].n >= MAX_PENDING) break;178 rows(r, `INSERT OR IGNORE INTO push_queue(registration_id,event_id,due,attempts,expires) VALUES(?,?,?,?,?)`, row.id, e.id, t, 0, t + QUEUE_TTL);179 added = true;180 }181 return added;182}183184function payload(e: Event, relayURL: string, include: boolean): string {185 return JSON.stringify({ id: e.id, relay: relayURL, ...(include ? { event: e } : {}) });186}187188function finish(r: Relay, registration: string, event: string, t: number) {189 rows(r, `DELETE FROM push_queue WHERE registration_id=? AND event_id=?`, registration, event);190 rows(r, `INSERT OR REPLACE INTO push_delivered(registration_id,event_id,delivered_at) VALUES(?,?,?)`, registration, event, t);191 rows(r, `DELETE FROM push_delivered WHERE rowid IN (SELECT rowid FROM push_delivered ORDER BY delivered_at ASC,rowid ASC LIMIT max(0,(SELECT count(*) FROM push_delivered)-?))`, MAX_DEDUP);192}193194// pushTick runs outside the repository admission lock, so slow callbacks195// cannot block event acceptance. Each attempt is checked again after prior I/O.196export async function pushTick(r: Relay): Promise<number> {197 const started = Date.now();198 let t = Math.floor(started / 1000);199 rows(r, `DELETE FROM push_queue WHERE expires<=?`, t);200 rows(r, `DELETE FROM push_delivered WHERE delivered_at<?`, t - DEDUP_TTL);201 if (!featureEnabled(r)) {202 rows(r, `DELETE FROM push_queue`);203 return 0;204 }205 const jobs = rows<{ registration_id: string; event_id: string; attempts: number }>(r, `SELECT registration_id,event_id,attempts FROM push_queue WHERE due<=? ORDER BY due,rowid LIMIT 4`, t);206 for (const job of jobs) {207 t = Math.floor(Date.now() / 1000);208 // Rows can disappear during the previous callback (deletion, teardown,209 // member removal). An alarm restart cannot exceed the reserved attempt cap.210 if (r.repositoryAccess.busy) return t + 1;211 const current = rows<{ attempts: number }>(r, `SELECT attempts FROM push_queue WHERE registration_id=? AND event_id=? AND expires>? AND due<=?`, job.registration_id, job.event_id, t, t)[0];212 if (!current) continue;213 job.attempts = current.attempts;214 if (!featureEnabled(r) || job.attempts >= MAX_ATTEMPTS) { finish(r, job.registration_id, job.event_id, t); continue; }215 r.tally();216 if (r.fuelStatus().outOfFuel) {217 rows(r, `UPDATE push_queue SET due=max(due,?)`, t + 60);218 return t + 60;219 }220 const owner = rows<{ pubkey: string }>(r, `SELECT pubkey FROM events WHERE id=? AND kind=?`, job.registration_id, PUSH_KIND)[0]?.pubkey;221 const reg = owner && authorized(r, owner) ? visible(r, job.registration_id, owner, t) : null;222 const e = owner ? visible(r, job.event_id, owner, t) : null;223 const p = reg ? parseRegistration(reg, r) : "missing registration";224 if (!reg || !e || typeof p === "string" || !p.filters.some((f) => match(f, e)) || p.ignores.some((f) => match(f, e))) {225 finish(r, job.registration_id, job.event_id, t); continue;226 }227 const body = payload(e, p.relay, p.include);228 const bytes = utf8Size(body);229 if (bytes > MAX_PAYLOAD) { finish(r, job.registration_id, job.event_id, t); continue; }230 // Reserve before external I/O. A crash consumes the attempt and waits231 // longer than the request timeout before another alarm may retry it.232 rows(r, `UPDATE push_queue SET attempts=attempts+1,due=? WHERE registration_id=? AND event_id=?`, t + 60, job.registration_id, job.event_id);233 const ctrl = new AbortController();234 const timer = setTimeout(() => ctrl.abort(), REQUEST_TIMEOUT);235 let response: Response | null = null;236 const requestStarted = Date.now();237 r.meterBytes(0, bytes);238 try { response = await r.fetcher(p.callback, { method: "POST", redirect: "manual", headers: { "content-type": "application/json" }, body, signal: ctrl.signal }); }239 catch { response = null; }240 finally { clearTimeout(timer); r.meterPush(0, 0, Date.now() - requestStarted); }241 if (response?.body) await response.body.cancel().catch(() => {});242 // An obsolete in-flight response must not mutate a replacement row.243 if (!rows(r, `SELECT 1 FROM push_queue WHERE registration_id=? AND event_id=?`, job.registration_id, job.event_id).length) continue;244 if (response?.status === 404) {245 r.store.deleteEvent(job.registration_id);246 } else if ((response && response.status >= 200 && response.status < 300) || job.attempts + 1 >= MAX_ATTEMPTS || (response && response.status < 500 && response.status !== 429)) {247 finish(r, job.registration_id, job.event_id, t);248 } else {249 rows(r, `UPDATE push_queue SET due=? WHERE registration_id=? AND event_id=?`, Math.floor(Date.now() / 1000) + RETRY_SECONDS[job.attempts], job.registration_id, job.event_id);250 }251 }252 r.tally();253 const next = rows<{ next: number | null }>(r, `SELECT MIN(due) AS next FROM push_queue`)[0].next;254 return next ? Math.max(next, Math.floor(Date.now() / 1000) + 1) : 0;255}256257// nextPush includes work enqueued by views or discovery after the push round.258export function nextPush(r: Relay): number {259 return rows<{ next: number | null }>(r, `SELECT MIN(due) AS next FROM push_queue`)[0].next ?? 0;260}