1// NIP-65 delivery: a small, durable, opt-in fanout queue. This is deliberately2// separate from NIP-9a callbacks: targets are Nostr relays and receive EVENT.3import { expiration, isPrivate, now, tagValues, type Event } from "./event.ts";4import { dial, Socket, checkPullURL } from "./pull.ts";5import type { Relay } from "./relay.ts";67export const DELIVERY_SCHEMA = `8CREATE TABLE IF NOT EXISTS delivery_queue (9 event_id TEXT NOT NULL, target TEXT NOT NULL, author TEXT NOT NULL,10 due INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,11 status TEXT NOT NULL DEFAULT 'pending', error TEXT NOT NULL DEFAULT '',12 updated_at INTEGER NOT NULL, PRIMARY KEY(event_id,target)13);14CREATE INDEX IF NOT EXISTS delivery_due ON delivery_queue(status,due);15`;16const MAX_QUEUE = 512, MAX_ATTEMPTS = 4, TIMEOUT = 5000, BATCH = 4;17const rows = <T extends Record<string, any>>(r: Relay, q: string, ...a: any[]): T[] => { const c = r.sql.exec<T>(q, ...a); const out = c.toArray(); r.meterPush(c.rowsRead, c.rowsWritten); return out; };18const exec = (r: Relay, q: string, ...a: any[]) => { const c = r.sql.exec(q, ...a); r.meterPush(c.rowsRead, c.rowsWritten); return c; };1920function list(r: Relay, pk: string): { read: string[]; write: string[] } {21 const row = r.store.query({ kinds: [10002], authors: [pk], tags: {} }, { pubkeys: [], all: true }, 1, now()).rows[0];22 const out = { read: [] as string[], write: [] as string[] };23 if (!row) return out;24 try {25 for (const t of (JSON.parse(row) as Event).tags) {26 if (t[0] !== "r" || !t[1]) continue;27 let u: URL; try { u = new URL(t[1]); } catch { continue; }28 if ((u.protocol !== "wss:" && u.protocol !== "ws:") || u.username || u.password || u.hash) continue;29 // NIP-65 is user supplied egress. Refuse obvious loopback/private30 // destinations; public DNS names remain the interoperable path.31 const h = u.hostname.toLowerCase();32 if (!/^(?:[a-z0-9](?:[a-z0-9-]*[a-z0-9])?\.)+[a-z][a-z0-9-]*$/.test(h) || /\.(?:localhost|local|internal|lan|home|test|invalid|onion)$/.test(h)) continue;33 const url = u.toString().replace(/\/$/, "");34 const marker = !t[2] ? "both" : t[2] === "read" ? "read" : t[2] === "write" ? "write" : "";35 if (!marker) continue;36 if ((marker === "read" || marker === "both") && !out.read.includes(url)) out.read.push(url);37 if ((marker === "write" || marker === "both") && !out.write.includes(url)) out.write.push(url);38 }39 } catch { /* malformed historical list */ }40 return out;41}4243function targets(r: Relay, e: Event): string[] {44 const max = r.settings.policy.delivery?.maxTargets ?? 8;45 const out: string[] = [];46 const add = (u: string) => {47 if (out.length >= max || out.includes(u)) return;48 if (checkPullURL(u, r.slug, r.domain)) return;49 out.push(u);50 };51 for (const u of list(r, e.pubkey).write) add(u);52 for (const pk of tagValues(e, "p").slice(0, 32)) for (const u of list(r, pk).read) add(u);53 return out;54}5556export function queueDelivery(r: Relay, e: Event): boolean {57 const p = r.settings.policy.delivery;58 if (!p?.enabled || r.settings.policy.reads === "members" || r.settings.leaseExpired(now()) || e.pubkey === r.identity.pubkey || r.settings.isBanned(e.pubkey) || !r.settings.isAllowed(e.pubkey) || isPrivate(e.kind) || e.tags.some((t: string[]) => t[0] === "-")) return false;59 const ts = targets(r, e);60 if (!ts.length) return false;61 let n = rows<{ n: number }>(r, `SELECT count(*) n FROM delivery_queue WHERE status='pending'`)[0]?.n ?? 0;62 let added = false;63 for (const target of ts) {64 if (n >= MAX_QUEUE) break;65 exec(r, `INSERT OR IGNORE INTO delivery_queue(event_id,target,author,due,attempts,status,error,updated_at) VALUES(?,?,?,?,0,'pending','',?)`, e.id, target, e.pubkey, now(), now());66 if (rows<{ n: number }>(r, `SELECT changes() AS n`)[0]?.n) { n++; added = true; }67 }68 return added;69}7071async function send(r: Relay, target: string, e: Event): Promise<{ ok: boolean; error: string }> {72 let s: Socket | null = null;73 try {74 s = new Socket(await dial(r, target));75 s.send("EVENT", e);76 const end = Date.now() + TIMEOUT;77 while (Date.now() < end) {78 const m = await s.recv(end - Date.now());79 if (m[0] !== "OK" || m[1] !== e.id) continue;80 if (m[2] === true || String(m[3] ?? "").startsWith("duplicate:")) return { ok: true, error: "" };81 return { ok: false, error: String(m[3] ?? "rejected").slice(0, 300) };82 }83 return { ok: false, error: "timeout" };84 } catch (e) { return { ok: false, error: (e instanceof Error ? e.message : String(e)).slice(0, 300) }; }85 finally { s?.close(); }86}8788export async function deliveryTick(r: Relay): Promise<number> {89 if (!r.settings.policy.delivery?.enabled) { exec(r, `DELETE FROM delivery_queue`); return 0; }90 const t = now();91 exec(r, `DELETE FROM delivery_queue WHERE status<>'pending' AND updated_at<?`, t - 7 * 86400);92 exec(r, `DELETE FROM delivery_queue WHERE rowid IN (SELECT rowid FROM delivery_queue WHERE status<>'pending' ORDER BY updated_at ASC LIMIT max(0,(SELECT count(*) FROM delivery_queue WHERE status<>'pending')-1024))`);93 const jobs = rows<{ event_id: string; target: string; attempts: number }>(r, `SELECT event_id,target,attempts FROM delivery_queue WHERE status='pending' AND due<=? ORDER BY due LIMIT ?`, t, BATCH);94 for (const j of jobs) {95 const author = rows<{ author: string }>(r, `SELECT author FROM delivery_queue WHERE event_id=? AND target=?`, j.event_id, j.target)[0]?.author ?? "";96 const ev = rows<{ raw: string }>(r, `SELECT raw FROM events WHERE id=? AND pubkey=? AND kind NOT IN (4,1059,21059,24133)`, j.event_id, author)[0];97 if (!ev) { exec(r, `UPDATE delivery_queue SET status='rejected',error='event unavailable',updated_at=? WHERE event_id=? AND target=?`, t, j.event_id, j.target); continue; }98 const event = JSON.parse(ev.raw) as Event;99 if (r.settings.policy.reads === "members" || r.settings.isBanned(author) || !r.settings.isAllowed(author) || isPrivate(event.kind) || event.tags.some((t) => t[0] === "-") || (expiration(event) > 0 && expiration(event) <= now()) || !targets(r, event).includes(j.target)) { exec(r, `UPDATE delivery_queue SET status='rejected',error='routing no longer permitted',updated_at=? WHERE event_id=? AND target=?`, t, j.event_id, j.target); continue; }100 exec(r, `UPDATE delivery_queue SET attempts=attempts+1,due=?,updated_at=? WHERE event_id=? AND target=?`, t + 60, t, j.event_id, j.target);101 const result = await send(r, j.target, event);102 const attempts = j.attempts + 1;103 if (result.ok || attempts >= MAX_ATTEMPTS) {104 exec(r, `UPDATE delivery_queue SET status=?,error=?,updated_at=? WHERE event_id=? AND target=?`, result.ok ? "accepted" : "rejected", result.error, now(), j.event_id, j.target);105 } else exec(r, `UPDATE delivery_queue SET due=?,error=?,updated_at=? WHERE event_id=? AND target=?`, now() + [30, 120, 600][Math.min(attempts - 1, 2)], result.error, now(), j.event_id, j.target);106 }107 const next = rows<{ next: number | null }>(r, `SELECT min(due) next FROM delivery_queue WHERE status='pending'`)[0]?.next ?? 0;108 return next;109}110111export function deliveryStatus(r: Relay) {112 return rows<{ event_id: string; target: string; status: string; attempts: number; error: string; updated_at: number }>(r, `SELECT event_id,target,status,attempts,error,updated_at FROM delivery_queue ORDER BY updated_at DESC LIMIT 100`);113}