searchevents runs the NIP-50 index in the shape of listrecentevents. PUT /import keeps a JSONL or JSON array in R2 and an import job reads it back a slice at a time, checking signatures, bans and kind rules but not the write rule, and deletes the file when done; it counts as media meanwhile. A pin list (kind 9010) from an owner or moderator, or pinevent and unpinevent, becomes a relay-signed kind 39005, protected like the other group records. A digest switch sends the owner one NIP-17 message a week with what changed.
- commit
- 6725e95d12f644ea8bbcd12cc03c0fc4bbd3884f
- parent
- 9e379115cb53e7df9a6ec3f4db26331959c94d87
- author
- Dami <dami.etoile@gmail.com>
| M | src/ | 30 | +++- |
| M | src/ | 10 | + |
| A | src/ | 144 | +++++++++++++++++ |
| M | src/ | 14 | +- |
| M | src/ | 25 | +++- |
| M | src/ | 39 | ++++- |
| M | src/ | 29 | +++- |
| M | src/ | 2 | +- |
| M | src/ | 15 | ++- |
| M | test/ | 2 | +- |
| A | test/ | 249 | ++++++++++++++++++++++++++++++ |
| M | test/ | 4 | +- |
@@ -30,7 +30,7 @@ export const isModeration = (kind: number) => kind >= 9000 && kind <= 9020;
3030 export const isNIP43Request = (kind: number) => kind === KIND_NIP43_JOIN || kind === KIND_NIP43_LEAVE;3131 export const isGroupManagement = (kind: number) => isModeration(kind) || kind === KIND_JOIN || kind === KIND_LEAVE || isNIP43Request(kind);3232 // The addressable state kinds only the relay writes.33-export const isGroupState = (kind: number) => kind >= 39000 && kind <= 39003;33+export const isGroupState = (kind: number) => kind >= 39000 && kind <= 39005;3434 3535 const HEX64 = /^[0-9a-f]{64}$/;3636 @@ -180,9 +180,33 @@ export async function handleGroupEvent(relay: Relay, e: Event): Promise<Result>
180180 return no("unsupported: this relay is one group; there is nothing to create");181181 case KIND_DELETE_GROUP:182182 return no("unsupported: delete the relay from its page");183- case KIND_PINS:184- return no("unsupported: pins are not supported");183+ case KIND_PINS: {184+ // update-pin-list replaces the whole list: e tags and a tags, in order.185+ const gate = need("deleteEvent");186+ if (gate) return no(gate);187+ const tags = pinTags(e.tags);188+ if (typeof tags === "string") return no(tags);189+ s.setPins(tags);190+ await relay.publishPins();191+ return OK;192+ }185193 default:186194 return no("unsupported: unknown moderation kind " + e.kind);187195 }188196 }197+198+export const MAX_PINS = 20;199+const ADDR = /^\d{1,5}:[0-9a-f]{64}:.*$/;200+201+// pinTags keeps the e and a tags of a pin list, in order, or says why not.202+export function pinTags(tags: string[][]): string[][] | string {203+ const out: string[][] = [];204+ for (const t of tags) {205+ if (t[0] === "e" && HEX64.test(t[1] ?? "")) out.push(["e", t[1]]);206+ else if (t[0] === "a" && ADDR.test(t[1] ?? "")) out.push(["a", t[1]]);207+ }208+ const seen = new Set<string>();209+ const unique = out.filter((t) => !seen.has(t[1]) && seen.add(t[1]));210+ if (unique.length > MAX_PINS) return `invalid: at most ${MAX_PINS} pins`;211+ return unique;212+}@@ -18,6 +18,7 @@ export const KIND_GROUP_METADATA = 39000;
1818 export const KIND_GROUP_ADMINS = 39001;1919 export const KIND_GROUP_MEMBERS = 39002;2020 export const KIND_GROUP_ROLES = 39003;21+export const KIND_GROUP_PINS = 39005;2122 export const KIND_ROLE_DEF = 33534;2223 export const KIND_PROFILE = 0;2324 @@ -99,6 +100,15 @@ export class Identity {
99100 return this.sign(KIND_REMOVE_USER, [["-"], ["h", group], ["p", pubkey]]);100101 }101102 103+ // pins signs the group's pinned list (NIP-29 kind 39005): e and a tags in104+ // the order they should show.105+ pins(group: string, tags: string[][], now = Math.floor(Date.now() / 1000)): Event {106+ const at = Math.max(now, this.lastGroup + 1);107+ this.lastGroup = at;108+ this.storage.put("relay-group-at", at);109+ return this.sign(KIND_GROUP_PINS, [["-"], ["d", group], ...tags], "", at);110+ }111+102112 // group signs the addressable NIP-29 state: metadata, admins, members103113 // (when published) and roles, plus the NIP-43 role definitions and, when104114 // it changed, the relay's own profile. Same strictly increasing clock as@@ -0,0 +1,144 @@
1+// Imports: the reverse of dumps. PUT /import takes a file of events, a2+// JSONL (strfry's export is one) or a JSON array, keeps it in R2 under3+// <name>/imports/, and hands it to a job that reads it back in bounded4+// byte ranges, one round at a time, so a big file never has to fit in a5+// request. Signatures are checked, bans and kind rules apply, the write6+// rule does not: the owner asked for these. The file counts as media while7+// it exists and is deleted when the job is done.8+import { validate, now, type Event } from "./event.ts";9+import { verifyNIP98 } from "./manage.ts";10+import { can } from "./roles.ts";11+import { ERR_DUPLICATE } from "./store.ts";12+import { newJobID, type Job } from "./jobs.ts";13+import type { Relay } from "./relay.ts";14+15+// The door reads the body whole to check the NIP-98 payload hash, so the16+// cap is what an object can hold comfortably next to its own work.17+export const IMPORT_MAX_BYTES = 64 * 1024 * 1024;18+const CHUNK = 1024 * 1024; // bytes read from R2 per round19+const MAX_LINE = 512 * 1024; // a line longer than this is not an event20+21+export function importKey(slug: string, id: string): string {22+ return `${slug}/imports/${id}.jsonl`;23+}24+25+export function importBytes(sql: SqlStorage): number {26+ return sql.exec<{ n: number | null }>(`SELECT sum(bytes) AS n FROM imports`).one().n ?? 0;27+}28+29+// importUpload serves PUT /import for a NIP-98 signer with the storage30+// action. The body is JSONL or a JSON array; arrays are rewritten to JSONL31+// so the job has one shape to read.32+export async function importUpload(relay: Relay, req: Request): Promise<Response> {33+ const json = (b: unknown, status: number) => new Response(JSON.stringify(b), { status, headers: { "content-type": "application/json", "access-control-allow-origin": "*" } });34+ const declared = Number(req.headers.get("content-length"));35+ if (declared > IMPORT_MAX_BYTES) return json({ error: `invalid: at most ${IMPORT_MAX_BYTES / 1024 / 1024} MB per import` }, 413);36+ const body = await req.text();37+ if (body.length > IMPORT_MAX_BYTES) return json({ error: `invalid: at most ${IMPORT_MAX_BYTES / 1024 / 1024} MB per import` }, 413);38+ const auth = verifyNIP98(req.headers.get("authorization") ?? "", req.url, req.method, body);39+ if (typeof auth === "string") return json({ error: auth }, 401);40+ if (!can(relay.settings.roleOf(auth.pubkey), "storage")) return json({ error: "restricted: not the relay owner" }, 403);41+ if (relay.settings.policy.owner === "") return json({ error: "restricted: this relay is unclaimed" }, 403);42+ if (relay.fuelStatus().outOfFuel) return json({ error: "restricted: this relay is out of fuel" }, 403);43+ let text = body.trim();44+ if (text === "") return json({ error: "invalid: the file is empty" }, 400);45+ if (text.startsWith("[")) {46+ let arr: unknown;47+ try {48+ arr = JSON.parse(text);49+ } catch {50+ return json({ error: "invalid: not a JSON array" }, 400);51+ }52+ if (!Array.isArray(arr)) return json({ error: "invalid: not a JSON array" }, 400);53+ text = arr.map((e) => JSON.stringify(e)).join("\n");54+ }55+ const bytes = new TextEncoder().encode(text + "\n");56+ const id = newJobID();57+ const name = (new URL(req.url).searchParams.get("name") ?? "import.jsonl").replace(/[^\w.-]/g, "_").slice(0, 80) || "import.jsonl";58+ await relay.media.put(importKey(relay.slug, id), bytes, { httpMetadata: { contentType: "application/x-ndjson" } });59+ relay.sql.exec(`INSERT INTO imports(id,name,bytes,at) VALUES(?,?,?,?)`, id, name, bytes.length, now());60+ relay.meterBytes(bytes.length, 0);61+ const job = await relay.addChecked({ kind: "import", label: "import", relays: [name], filter: {}, every: 0, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, duplicates: 0, last: null, object: id, size: bytes.length, carry: "" });62+ if (typeof job === "string") {63+ await forgetImport(relay, id);64+ return json({ error: job }, 409);65+ }66+ return json({ job: job.id, bytes: bytes.length, name }, 202);67+}68+69+// forgetImport drops the object and its row, whatever state the job is in.70+export async function forgetImport(relay: Relay, id: string) {71+ await relay.media.delete(importKey(relay.slug, id)).catch(() => {});72+ relay.sql.exec(`DELETE FROM imports WHERE id=?`, id);73+}74+75+const b64 = {76+ enc: (b: Uint8Array) => btoa(String.fromCharCode(...b)),77+ dec: (s: string) => Uint8Array.from(atob(s), (c) => c.charCodeAt(0)),78+};79+80+// runImportRound reads the next byte range, finishes the line carried over81+// from the last round, and stores what checks out. more: call again.82+export async function runImportRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {83+ job.rounds++;84+ const id = job.object ?? "";85+ const size = job.size ?? 0;86+ if (!id || job.cursor >= size) {87+ await forgetImport(relay, id);88+ return { more: false, error: "" };89+ }90+ const obj = await relay.media.get(importKey(relay.slug, id), { range: { offset: job.cursor, length: Math.min(CHUNK, size - job.cursor) } });91+ if (!obj) return { more: false, error: "the import file is gone" };92+ const fresh = new Uint8Array(await obj.arrayBuffer());93+ const carried = job.carry ? b64.dec(job.carry) : new Uint8Array(0);94+ const buf = new Uint8Array(carried.length + fresh.length);95+ buf.set(carried);96+ buf.set(fresh, carried.length);97+ job.cursor += fresh.length;98+ const atEnd = job.cursor >= size;99+ // Split on newlines; the tail after the last one waits for the next round.100+ let start = 0;101+ const dec = new TextDecoder();102+ const t = now();103+ for (let i = 0; i < buf.length; i++) {104+ if (buf[i] !== 10) continue;105+ take(relay, dec.decode(buf.subarray(start, i)), job, t);106+ start = i + 1;107+ }108+ const tail = buf.subarray(start);109+ if (atEnd) {110+ if (tail.length) take(relay, dec.decode(tail), job, t);111+ job.carry = "";112+ await forgetImport(relay, id);113+ return { more: false, error: "" };114+ }115+ if (tail.length > MAX_LINE) {116+ job.skipped++;117+ job.carry = "";118+ } else job.carry = b64.enc(tail);119+ return { more: true, error: "" };120+}121+122+// take stores one line, counting what became of it.123+function take(relay: Relay, line: string, job: Job, t: number) {124+ const s = line.trim();125+ if (s === "" || s === "[" || s === "]") return;126+ let e: Event;127+ try {128+ e = JSON.parse(s.replace(/,$/, "")) as Event;129+ } catch {130+ job.skipped++;131+ return;132+ }133+ if (validate(e) || !relay.settings.kindAllowed(e.kind)) {134+ job.skipped++;135+ return;136+ }137+ const r = relay.accept(e, null);138+ if (r.stored) {139+ job.stored++;140+ relay.broadcast(e);141+ } else if (r.msg === ERR_DUPLICATE) job.duplicates = (job.duplicates ?? 0) + 1;142+ else job.skipped++;143+ void t;144+}@@ -8,9 +8,10 @@ import { hasTag, isPrivate, now, type Event } from "./event.ts";
88 import type { Filter } from "./filter.ts";99 import { Socket, dial, checkPullURL, runPullRound, type PullFilter, type PullJob, type PullResult } from "./pull.ts";1010 import type { Relay } from "./relay.ts";11+import { runImportRound } from "./imports.ts";1112 12-export type JobKind = "pull" | "push";13-export type JobLabel = "pull" | "backfill" | "push" | "replica";13+export type JobKind = "pull" | "push" | "import";14+export type JobLabel = "pull" | "backfill" | "push" | "replica" | "import";1415 export const EVERY = [0, 1, 6, 24] as const; // hours; 0 runs once1516 1617 export interface JobResult {@@ -22,6 +23,7 @@ export interface JobResult {
2223 blobs: number;2324 sent: number;2425 refused: number;26+ duplicates?: number;2527 }2628 2729 export interface Job {@@ -45,6 +47,10 @@ export interface Job {
4547 blobs: number;4648 sent: number;4749 refused: number;50+ duplicates?: number; // import: events the relay already had51+ object?: string; // import: the R2 object's id52+ size?: number; // import: the object's size in bytes53+ carry?: string; // import: the partial last line of the previous round, base644854 last: JobResult | null;4955 }5056 @@ -116,6 +122,7 @@ export function relaysFromList(relay: Relay, pubkey: string): string[] {
116122 // runRound does one round of a job. more: call again for this run.117123 export async function runRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {118124 if (job.kind === "pull") return runPullSourceRound(relay, job);125+ if (job.kind === "import") return runImportRound(relay, job);119126 return runPushRound(relay, job);120127 }121128 @@ -219,11 +226,12 @@ export function startRun(job: Job, t: number) {
219226 job.blobs = 0;220227 job.sent = 0;221228 job.refused = 0;229+ job.duplicates = 0;222230 }223231 224232 // finishRun closes the current run and schedules the next one.225233 export function finishRun(job: Job, error: string, t: number) {226- job.last = { finishedAt: t, error, rounds: job.rounds, stored: job.stored, skipped: job.skipped, blobs: job.blobs, sent: job.sent, refused: job.refused };234+ job.last = { finishedAt: t, error, rounds: job.rounds, stored: job.stored, skipped: job.skipped, blobs: job.blobs, sent: job.sent, refused: job.refused, duplicates: job.duplicates ?? 0 };227235 job.running = false;228236 job.nextRun = job.every > 0 ? t + job.every * 3600 : 0;229237 }@@ -10,6 +10,7 @@ import { descriptor, type Blob } from "./blossom.ts";
1010 import { blockedWords, gateFields, isWriteRule } from "./settings.ts";1111 import { isReplaceable, isProtected, publicFields, dumpFields } from "./settings.ts";1212 import { checkPullURL } from "./pull.ts";13+import { MAX_PINS } from "./groups.ts";1314 import { relaysFromList } from "./jobs.ts";1415 import { notify, notifySettings } from "./notify.ts";1516 import { DUMP_NAME_RE, deleteDump, dumpBytes, listDumps, writeDump } from "./dumps.ts";@@ -52,7 +53,7 @@ const METHODS = [
5253 "supportedmethods", "claim", "stats", "getpolicy", "setpolicy",5354 "banpubkey", "allowpubkey", "unrulepubkey", "listbannedpubkeys", "listallowedpubkeys",5455 "setmember", "removemember", "listpeople",55- "banevent", "allowevent", "listbannedevents", "deleteevent", "listrecentevents",56+ "banevent", "allowevent", "listbannedevents", "deleteevent", "listrecentevents", "searchevents", "pinevent", "unpinevent", "listpins",5657 "allowkind", "disallowkind", "unrulekind", "listallowedkinds", "listblockedkinds",5758 "changerelayname", "changerelaydescription", "changerelayicon",5859 "createinvite", "listinvites", "revokeinvite", "listmembers",@@ -353,6 +354,28 @@ export async function manage(relay: Relay, req: Request): Promise<Response> {
353354 return reply({ result: relay.store.deleteEvent(str(0)) });354355 case "listrecentevents":355356 return reply({ result: relay.store.recent(Math.min(Math.max(num(0) || 50, 1), 500), t).map((r) => JSON.parse(r)) });357+ case "searchevents": {358+ // (query, limit): the NIP-50 index, in the shape of listrecentevents.359+ const q = str(0).trim().slice(0, 200);360+ const limit = Math.min(Math.max(num(1) || 50, 1), 200);361+ if (!q) return reply({ result: relay.store.recent(limit, t).map((r) => JSON.parse(r)) });362+ return reply({ result: relay.store.query({ tags: {}, search: q, limit }, { pubkeys: [], all: true }, limit, t).rows.map((r) => JSON.parse(r)) });363+ }364+ case "pinevent":365+ case "unpinevent": {366+ // (id or address): the group's pin list, at most MAX_PINS, in order.367+ const ref = str(0).trim().toLowerCase();368+ const tag = hex64(ref) ? ["e", ref] : /^\d{1,5}:[0-9a-f]{64}:/.test(ref) ? ["a", ref] : null;369+ if (!tag) return reply({ error: "invalid: give an event id or an address kind:pubkey:d" }, 400);370+ const cur = s.pins().filter((x) => x[1] !== ref);371+ const next = method === "pinevent" ? [...cur, tag] : cur;372+ if (next.length > MAX_PINS) return reply({ error: `invalid: at most ${MAX_PINS} pins` }, 400);373+ s.setPins(next);374+ await relay.publishPins();375+ return reply({ result: next });376+ }377+ case "listpins":378+ return reply({ result: s.pins() });356379 case "allowkind":357380 case "disallowkind":358381 case "unrulekind": {@@ -8,13 +8,15 @@ import { dial } from "./pull.ts";
88 import type { Relay } from "./relay.ts";99 import type { FuelStatus } from "./fuel.ts";1010 11-export type NotifyKind = "reports" | "fuel" | "jobs" | "succession" | "test";11+export type NotifyKind = "reports" | "fuel" | "jobs" | "succession" | "digest" | "test";1212 export interface NotifySettings {1313 reports: boolean;1414 fuel: boolean;1515 jobs: boolean;1616 succession: boolean; // the dead-man's switch warnings and the handover17+ digest: boolean; // one message a week on how the relay is doing1718 }19+export const DIGEST_DAYS = 7;1820 1921 const KIND_DM = 14;2022 export const KIND_WRAP = 1059;@@ -29,7 +31,7 @@ export function notifySettings(raw: unknown, cur: NotifySettings): NotifySetting
2931 if (!raw || typeof raw !== "object") return null;3032 const r = raw as Record<string, unknown>;3133 const pick = (k: keyof NotifySettings) => (typeof r[k] === "boolean" ? (r[k] as boolean) : cur[k]);32- return { reports: pick("reports"), fuel: pick("fuel"), jobs: pick("jobs"), succession: pick("succession") };34+ return { reports: pick("reports"), fuel: pick("fuel"), jobs: pick("jobs"), succession: pick("succession"), digest: pick("digest") };3335 }3436 3537 // notify sends the owner (or `to`, a member the relay may address) a message@@ -115,3 +117,36 @@ export function fuelText(relay: Relay, f: FuelStatus): string {
115117 const state = f.outOfFuel ? "is out of fuel and read-only" : `is past its free allowance with ${sats} sats left`;116118 return `Your relay ${relay.slug} ${state}. Zap it at https://${relay.slug}.${relay.domain}/ to keep it writing.`;117119 }120+121+// digestText is the week on the relay in a few lines, from what the store,122+// the member list, the usage row and the job list already hold. When nothing123+// changed it says so in one line.124+export async function digestText(relay: Relay, since: number, t: number): Promise<string> {125+ const sql = relay.sql;126+ const one = (q: string, ...args: unknown[]) => sql.exec<{ n: number | null }>(q, ...args).one().n ?? 0;127+ const self = relay.identity.pubkey;128+ const joined = one(`SELECT count(*) AS n FROM members WHERE joined_at >= ? AND role <> 'owner'`, since);129+ const left = self ? one(`SELECT count(*) AS n FROM events WHERE kind=8001 AND pubkey=? AND created_at >= ?`, self, since) : 0;130+ const events = one(`SELECT count(*) AS n FROM events WHERE created_at >= ? AND pubkey <> ? AND kind <> 1059`, since, self);131+ const files = one(`SELECT count(*) AS n FROM blobs WHERE uploaded >= ?`, since);132+ const reports = one(`SELECT count(*) AS n FROM reports WHERE status='open'`);133+ const hidden = one(`SELECT count(*) AS n FROM event_rules WHERE rule='hide'`);134+ const jobs = (await relay.jobs()).filter((j) => j.last && j.last.finishedAt >= since);135+ const failed = jobs.filter((j) => j.last && j.last.error).length;136+ const f = relay.fuelStatus();137+ const spent = Math.floor(f.chargedMsats / 1000);138+ const balance = Math.floor(f.balanceMsats / 1000);139+ const sc = await relay.successionStatus();140+ const lines: string[] = [];141+ if (joined || left) lines.push(`People: ${joined} joined${left ? `, ${left} left` : ""}.`);142+ if (events || files) lines.push(`Stored: ${events} events${files ? ` and ${files} files` : ""}.`);143+ if (jobs.length) lines.push(`Jobs: ${jobs.length} ran${failed ? `, ${failed} failed` : ""}.`);144+ if (reports || hidden) lines.push(`Moderation: ${reports} open reports${hidden ? `, ${hidden} events hidden` : ""}.`);145+ if (f.enabled) {146+ const over = f.eventBytes > f.freeEventBytes || f.mediaBytes > f.freeMediaBytes || f.activeMs > f.freeActiveMs || f.rowsWritten > f.freeRowsWritten;147+ if (spent || over || f.outOfFuel) lines.push(`Fuel: ${spent} sats spent this month, ${balance} left${f.outOfFuel ? ", out of fuel" : over ? ", past the free allowance" : ""}.`);148+ }149+ if (sc.succession && sc.handoverAt) lines.push(`Handover to ${sc.succession.heir.slice(0, 8)} in ${Math.max(0, Math.ceil((sc.handoverAt - t) / 86400))} days unless you sign in.`);150+ const days = Math.max(1, Math.round((t - since) / 86400));151+ return lines.length ? [`The last ${days} days on ${relay.slug}:`, ...lines].join("\n") : `Nothing changed on ${relay.slug} in the last ${days} days.`;152+}@@ -18,6 +18,9 @@ import { bridge } from "./bridge.ts";
1818 import { blossom, blobBytes, isBlobPath } from "./blossom.ts";1919 import { nip96 } from "./nip96.ts";2020 import { dumpBytes, dumpDue, dumpDownload, writeDump } from "./dumps.ts";21+import { importBytes, importUpload } from "./imports.ts";22+import { KIND_GROUP_PINS } from "./identity.ts";23+import { DIGEST_DAYS, digestText } from "./notify.ts";2124 import { claimFromProfile, nip05Document } from "./nip05.ts";2225 import { checkInvite, claimInvite, inviteCreator, invitePage, termsPage } from "./invites.ts";2326 import { verifyNIP98 } from "./manage.ts";@@ -231,6 +234,19 @@ export class Relay extends DurableObject<Env> {
231234 this.emit(events, t);232235 }233236 237+ // publishPins signs the group's pin list (39005); an empty list takes the238+ // record down.239+ async publishPins() {240+ if (this.settings.policy.owner === "") return;241+ await this.identity.ensure();242+ const tags = this.settings.pins();243+ if (tags.length === 0) {244+ for (const r of this.sql.exec<{ id: string }>(`SELECT id FROM events WHERE kind=? AND pubkey=?`, KIND_GROUP_PINS, this.identity.pubkey).toArray()) this.store.deleteEvent(r.id);245+ return;246+ }247+ this.emit([this.identity.pins(this.slug, tags)], now());248+ }249+234250 private rolesOf(pubkey: string): string[] {235251 const r = this.settings.roleOf(pubkey);236252 return r && r !== "member" ? [r] : [];@@ -702,7 +718,7 @@ export class Relay extends DurableObject<Env> {
702718 703719 // Files and dumps both live in R2 and both cost media storage.704720 mediaBytes(): number {705- return blobBytes(this.sql) + dumpBytes(this.sql);721+ return blobBytes(this.sql) + dumpBytes(this.sql) + importBytes(this.sql);706722 }707723 708724 fuelStatus() {@@ -857,6 +873,7 @@ export class Relay extends DurableObject<Env> {
857873 }858874 if (url.pathname === "/api/invites/claim" && req.method === "POST") return this.claimInviteRequest(req);859875 if (url.pathname.startsWith("/dumps/") && req.method === "GET") return dumpDownload(this, req);876+ if (url.pathname === "/import" && req.method === "PUT") return importUpload(this, req);860877 if (url.pathname === "/upload" || url.pathname === "/mirror" || url.pathname === "/report" || url.pathname.startsWith("/list/") || isBlobPath(url.pathname)) return blossom(this, req);861878 if (url.pathname === "/.well-known/nostr/nip96.json" || url.pathname === "/nip96" || url.pathname.startsWith("/nip96/")) return nip96(this, req);862879 if (url.pathname === "/fuel" && req.method === "GET") {@@ -1276,6 +1293,16 @@ export class Relay extends DurableObject<Env> {
12761293 }12771294 // Succession: the dead-man's switch, checked once a day.12781295 await this.successionTick(t);1296+ // ---- digest: one message a week on how the relay is doing ----1297+ if (this.settings.policy.notify.digest && this.settings.policy.owner !== "") {1298+ const last = (await this.ctx.storage.get<number>("lastDigest")) ?? 0;1299+ if (last === 0) await this.ctx.storage.put("lastDigest", t);1300+ else if (t - last >= DIGEST_DAYS * 86400 - 3600) {1301+ await this.ctx.storage.put("lastDigest", t);1302+ await notify(this, "digest", await digestText(this, last, t), "the week on " + this.slug);1303+ }1304+ }1305+ // ---- end digest ----12791306 const next = this.store.sweepExpired(t);12801307 // ---- dumps: a scheduled JSONL of everything, into R2 (dumps.ts) ----12811308 if (dumpDue(this, t)) {@@ -42,7 +42,7 @@ export const METHOD_ACTIONS: Record<string, Action> = {
4242 listbannedevents: "read", listrecentevents: "read", listallowedkinds: "read", listblockedkinds: "read", listblobs: "read", listretention: "read", listreports: "read",4343 setmember: "members", allowpubkey: "members", removemember: "members", unrulepubkey: "members",4444 banpubkey: "ban", banevent: "ban", allowevent: "ban", blockip: "ban", unblockip: "ban", listblockedips: "read", setblockedwords: "ban",45- deleteevent: "deleteEvent",45+ deleteevent: "deleteEvent", pinevent: "deleteEvent", unpinevent: "deleteEvent", listpins: "read", searchevents: "read",4646 createinvite: "invites", revokeinvite: "invites",4747 resolvereport: "reports", listeventsneedingmoderation: "reports",4848 setpolicy: "rules", allowkind: "rules", disallowkind: "rules", unrulekind: "rules", setretention: "rules", purgekind: "rules", resetrules: "rules", applypreset: "rules",@@ -84,7 +84,7 @@ export const DEFAULT_POLICY: Policy = {
8484 tags: [],8585 languageTags: [],8686 relayCountries: [],87- notify: { reports: false, fuel: false, jobs: false, succession: false },87+ notify: { reports: false, fuel: false, jobs: false, succession: false, digest: false },8888 succession: null,8989 writes: "open",9090 openKinds: [],@@ -117,6 +117,7 @@ CREATE TABLE IF NOT EXISTS invites (code TEXT PRIMARY KEY, created_by TEXT NOT N
117117 CREATE TABLE IF NOT EXISTS members (pubkey TEXT PRIMARY KEY, role TEXT NOT NULL DEFAULT 'member', name TEXT, note TEXT NOT NULL DEFAULT '', joined_at INTEGER NOT NULL, via TEXT NOT NULL DEFAULT '');118118 CREATE UNIQUE INDEX IF NOT EXISTS members_name ON members(name) WHERE name IS NOT NULL;119119 CREATE TABLE IF NOT EXISTS nip05 (name TEXT PRIMARY KEY, pubkey TEXT NOT NULL, at INTEGER NOT NULL);120+CREATE TABLE IF NOT EXISTS imports (id TEXT PRIMARY KEY, name TEXT NOT NULL, bytes INTEGER NOT NULL, at INTEGER NOT NULL);120121 CREATE TABLE IF NOT EXISTS reports (id TEXT PRIMARY KEY, reporter TEXT NOT NULL, target_pubkey TEXT NOT NULL, target_event TEXT NOT NULL DEFAULT '', type TEXT NOT NULL DEFAULT '', content TEXT NOT NULL DEFAULT '', at INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'open', resolved_by TEXT NOT NULL DEFAULT '', resolved_at INTEGER NOT NULL DEFAULT 0, action TEXT NOT NULL DEFAULT '');121122 CREATE TABLE IF NOT EXISTS blobs (sha256 TEXT PRIMARY KEY, size INTEGER NOT NULL, type TEXT NOT NULL, uploader TEXT NOT NULL, uploaded INTEGER NOT NULL);122123 CREATE INDEX IF NOT EXISTS blobs_uploader ON blobs(uploader, uploaded DESC);@@ -380,6 +381,16 @@ export class Settings {
380381 if (this.policy.writes !== "wot") this.wot.clear();381382 }382383 384+ // ---- pins: the group's pinned events, e and a tags in order ----385+386+ pins(): string[][] {387+ const row = this.sql.exec<{ value: string }>(`SELECT value FROM settings WHERE key='pins'`).toArray()[0];388+ return row ? (JSON.parse(row.value) as string[][]) : [];389+ }390+ setPins(tags: string[][]) {391+ this.sql.exec(`INSERT INTO settings(key,value) VALUES('pins',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, JSON.stringify(tags));392+ }393+383394 isOwner(pubkey: string) {384395 return this.policy.owner !== "" && this.policy.owner === pubkey;385396 }@@ -720,7 +731,7 @@ const RETENTION_ANY = -1;
720731 // its deltas, the NIP-29 put-user and remove-user records, and the group's721732 // metadata, admins, members and roles). Never expired, never purged, only722733 // deleted one at a time.723-export const PROTECTED_KINDS = new Set([0, 3, 10002, 9735, 13534, 8000, 8001, 9000, 9001, 33534, 39000, 39001, 39002, 39003]);734+export const PROTECTED_KINDS = new Set([0, 3, 10002, 9735, 13534, 8000, 8001, 9000, 9001, 33534, 39000, 39001, 39002, 39003, 39005]);724735 export function isProtected(kind: number): boolean {725736 return PROTECTED_KINDS.has(kind);726737 }@@ -240,7 +240,7 @@ describe("NIP-29 moderation", () => {
240240 241241 expect((await c.ok(ev(owner, 9007, "", [h]))).msg).toMatch(/^unsupported:/);242242 expect((await c.ok(ev(owner, 9008, "", [h]))).msg).toMatch(/^unsupported:/);243- expect((await c.ok(ev(owner, 9010, "", [h]))).msg).toMatch(/^unsupported:/);243+ expect((await c.ok(ev(owner, 9010, "", [h]))).ok).toBe(true); // an empty pin list clears the pins244244 expect((await c.ok(ev(owner, 39000, "", [["d", "mod"], ["name", "forged"]]))).msg).toMatch(/written by the relay/);245245 });246246 @@ -0,0 +1,249 @@
1+// Console search, importing a file, NIP-29 pins, and the weekly digest.2+import { SELF, env, runInDurableObject } from "cloudflare:test";3+import { describe, it, expect } from "vitest";4+import { finalizeEvent, generateSecretKey, getPublicKey, type Event } from "nostr-tools/pure";5+import { getToken } from "nostr-tools/nip98";6+import { unwrapEvent } from "nostr-tools/nip59";7+import { sha256 } from "@noble/hashes/sha2.js";8+import type { Relay } from "../src/relay.ts";9+import { bytesToHex } from "../src/negentropy.ts";10+import type { Job } from "../src/jobs.ts";11+12+const now = () => Math.floor(Date.now() / 1000);13+const ev = (sk: Uint8Array, kind: number, content: string, tags: string[][] = [], created_at = now()) => finalizeEvent({ kind, content, tags, created_at }, sk);14+15+async function rpc(host: string, sk: Uint8Array, method: string, ...params: unknown[]) {16+ const url = `http://${host}/`;17+ const payload = { method, params };18+ const token = await getToken(url, "POST", (e) => finalizeEvent(e, sk), true, payload);19+ const resp = await SELF.fetch(url, { method: "POST", headers: { "content-type": "application/nostr+json+rpc", authorization: token }, body: JSON.stringify(payload) });20+ return { status: resp.status, ...(await resp.json<any>()) };21+}22+23+// signedPut sends a raw body with a NIP-98 token whose payload tag is the body's hash.24+async function signedPut(host: string, path: string, sk: Uint8Array | null, body: string) {25+ const url = `http://${host}${path}`;26+ const headers: Record<string, string> = { "content-type": "application/x-ndjson" };27+ if (sk) {28+ const token = ev(sk, 27235, "", [["u", url], ["method", "PUT"], ["payload", bytesToHex(sha256(new TextEncoder().encode(body)))]]);29+ headers.authorization = "Nostr " + btoa(JSON.stringify(token));30+ }31+ const resp = await SELF.fetch(url, { method: "PUT", headers, body });32+ return { status: resp.status, ...(await resp.json<any>()) };33+}34+35+class WS {36+ private queue: any[][] = [];37+ private waiters: ((m: any[]) => void)[] = [];38+ challenge = "";39+ constructor(public ws: WebSocket) {40+ ws.accept();41+ ws.addEventListener("message", (e) => {42+ const m = JSON.parse(e.data as string);43+ const w = this.waiters.shift();44+ if (w) w(m);45+ else this.queue.push(m);46+ });47+ }48+ static async connect(host: string) {49+ const resp = await SELF.fetch(`http://${host}/`, { headers: { upgrade: "websocket" } });50+ const c = new WS(resp.webSocket!);51+ const m = await c.expect("AUTH");52+ c.challenge = m[1];53+ return c;54+ }55+ send(...m: unknown[]) {56+ this.ws.send(JSON.stringify(m));57+ }58+ recv(): Promise<any[]> {59+ const m = this.queue.shift();60+ if (m) return Promise.resolve(m);61+ return new Promise((res) => this.waiters.push(res));62+ }63+ async expect(type: string) {64+ const m = await this.recv();65+ expect(m[0], JSON.stringify(m)).toBe(type);66+ return m;67+ }68+ async ok(e: Event) {69+ this.send("EVENT", e);70+ const m = await this.expect("OK");71+ return { ok: m[2] as boolean, msg: m[3] as string };72+ }73+ async auth(sk: Uint8Array, host: string) {74+ this.send("AUTH", ev(sk, 22242, "", [["relay", "ws://" + host], ["challenge", this.challenge]]));75+ const m = await this.expect("OK");76+ expect(m[2], m[3]).toBe(true);77+ }78+ private n = 0;79+ async req(filter: unknown) {80+ const id = "q" + ++this.n;81+ this.send("REQ", id, filter);82+ const got: Event[] = [];83+ for (;;) {84+ const m = await this.recv();85+ if (m[0] === "EOSE" && m[1] === id) {86+ this.send("CLOSE", id);87+ return got;88+ }89+ if (m[0] === "EVENT" && m[1] === id) got.push(m[2]);90+ else if (m[0] === "CLOSED" && m[1] === id) throw new Error(m[2]);91+ }92+ }93+}94+95+// drive fires the alarm until no job is running or due.96+async function drive(host: string, owner: Uint8Array): Promise<Job[]> {97+ const stub = env.RELAY.getByName(host.split(".")[0]);98+ for (let i = 0; i < 80; i++) {99+ await runInDurableObject(stub, async (r: Relay) => r.alarm());100+ const jobs = (await rpc(host, owner, "listjobs")).result as Job[];101+ if (!jobs.some((j) => j.running || (j.nextRun > 0 && j.nextRun <= now()))) return jobs;102+ await new Promise((r) => setTimeout(r, 25));103+ }104+ throw new Error("jobs did not settle");105+}106+107+describe("search in the console", () => {108+ it("finds notes by words and answers by role", async () => {109+ const host = "seek.bind.ws";110+ const owner = generateSecretKey();111+ const mod = generateSecretKey();112+ const member = generateSecretKey();113+ await rpc(host, owner, "claim");114+ await rpc(host, owner, "setmember", getPublicKey(mod), { role: "moderator" });115+ await rpc(host, owner, "setmember", getPublicKey(member), {});116+ const c = await WS.connect(host);117+ const fox = ev(owner, 1, "the quick brown fox");118+ expect((await c.ok(fox)).ok).toBe(true);119+ expect((await c.ok(ev(owner, 1, "lorem ipsum dolor"))).ok).toBe(true);120+ const hits = (await rpc(host, owner, "searchevents", "fox", 50)).result as Event[];121+ expect(hits.map((e) => e.id)).toEqual([fox.id]);122+ expect((await rpc(host, mod, "searchevents", "ipsum")).result.length).toBe(1);123+ expect((await rpc(host, member, "searchevents", "fox")).status).toBe(403);124+ // An empty query is the recent list.125+ expect((await rpc(host, owner, "searchevents", "", 10)).result.length).toBeGreaterThanOrEqual(2);126+ });127+});128+129+describe("import a file", () => {130+ it("stores what checks out from a JSONL and a JSON array, then forgets the file", async () => {131+ const host = "ingest.bind.ws";132+ const owner = generateSecretKey();133+ const alice = generateSecretKey();134+ const banned = generateSecretKey();135+ await rpc(host, owner, "claim");136+ await rpc(host, owner, "setpolicy", { writes: "owner" });137+ await rpc(host, owner, "banpubkey", getPublicKey(banned), "spam");138+ const a1 = ev(alice, 1, "first", [], now() - 300);139+ const a2 = ev(alice, 1, "second", [], now() - 200);140+ const b1 = ev(banned, 1, "unwanted", [], now() - 100);141+ const forged = { ...ev(alice, 1, "genuine"), content: "tampered" };142+ const jsonl = [a1, a2, b1, forged].map((e) => JSON.stringify(e)).join("\n") + "\nnot json at all\n";143+144+ expect((await signedPut(host, "/import", null, jsonl)).status).toBe(401);145+ expect((await signedPut(host, "/import", alice, jsonl)).status).toBe(403);146+ const r = await signedPut(host, "/import?name=old-relay.jsonl", owner, jsonl);147+ expect(r.status).toBe(202);148+ expect(r.bytes).toBeGreaterThan(0);149+ const stub = env.RELAY.getByName("ingest");150+ await runInDurableObject(stub, async (relay: Relay) => expect(relay.mediaBytes()).toBeGreaterThan(0));151+ expect((await env.MEDIA.list({ prefix: "ingest/imports/" })).objects.length).toBe(1);152+153+ const jobs = await drive(host, owner);154+ const job = jobs.find((j) => j.kind === "import")!;155+ expect(job.relays).toEqual(["old-relay.jsonl"]);156+ expect(job.last?.error).toBe("");157+ expect(job.last?.stored).toBe(2);158+ expect(job.last?.skipped).toBe(3);159+ expect(job.last?.duplicates).toBe(0);160+ expect((await env.MEDIA.list({ prefix: "ingest/imports/" })).objects).toEqual([]);161+ await runInDurableObject(stub, async (relay: Relay) => expect(relay.mediaBytes()).toBe(0));162+ const c = await WS.connect(host);163+ expect((await c.req({ kinds: [1], authors: [getPublicKey(alice)] })).map((e) => e.id).sort()).toEqual([a1.id, a2.id].sort());164+165+ // A JSON array of the same events: all duplicates.166+ const again = await signedPut(host, "/import", owner, JSON.stringify([a1, a2]));167+ expect(again.status).toBe(202);168+ const job2 = (await drive(host, owner)).find((j) => j.kind === "import" && j.id === again.job)!;169+ expect(job2.last?.stored).toBe(0);170+ expect(job2.last?.duplicates).toBe(2);171+ expect((await signedPut(host, "/import", owner, " ")).status).toBe(400);172+ });173+});174+175+describe("pins", () => {176+ it("a moderator's pin list becomes a signed 39005, and unpinning takes it down", async () => {177+ const host = "board.bind.ws";178+ const owner = generateSecretKey();179+ const mod = generateSecretKey();180+ const member = generateSecretKey();181+ await rpc(host, owner, "claim");182+ await rpc(host, owner, "setmember", getPublicKey(mod), { role: "moderator" });183+ await rpc(host, owner, "setmember", getPublicKey(member), {});184+ const c = await WS.connect(host);185+ const note = ev(owner, 1, "read this first");186+ expect((await c.ok(note)).ok).toBe(true);187+ const h = ["h", "board"];188+ expect((await c.ok(ev(member, 9010, "", [h, ["e", note.id]]))).msg).toMatch(/^restricted/);189+ expect((await c.ok(ev(mod, 9010, "", [h, ["e", note.id], ["a", "30023:" + getPublicKey(owner) + ":welcome"]]))).ok).toBe(true);190+ const info: any = await (await SELF.fetch(`http://${host}/`, { headers: { accept: "application/nostr+json" } })).json();191+ const [pins] = await c.req({ kinds: [39005] });192+ expect(pins.pubkey).toBe(info.self);193+ expect(pins.tags).toEqual(expect.arrayContaining([["d", "board"], ["e", note.id], ["a", "30023:" + getPublicKey(owner) + ":welcome"]]));194+ expect((await rpc(host, mod, "listpins")).result).toEqual([["e", note.id], ["a", "30023:" + getPublicKey(owner) + ":welcome"]]);195+ // A client cannot write the record itself.196+ expect((await c.ok(ev(owner, 39005, "", [["d", "board"], ["e", note.id]]))).msg).toMatch(/blocked/);197+ // The console's methods edit the same list.198+ expect((await rpc(host, owner, "unpinevent", "30023:" + getPublicKey(owner) + ":welcome")).result).toEqual([["e", note.id]]);199+ expect((await rpc(host, member, "unpinevent", note.id)).status).toBe(403);200+ expect((await rpc(host, mod, "unpinevent", note.id)).result).toEqual([]);201+ expect(await c.req({ kinds: [39005] })).toEqual([]);202+ // The cap.203+ const many = Array.from({ length: 21 }, (_, i) => ["e", i.toString(16).padStart(64, "0")]);204+ expect((await c.ok(ev(mod, 9010, "", [h, ...many]))).msg).toMatch(/at most 20/);205+ });206+});207+208+describe("weekly digest", () => {209+ it("writes the owner once a week with what changed, and stays quiet otherwise", async () => {210+ const host = "weekly.bind.ws";211+ const owner = generateSecretKey();212+ const pk = getPublicKey(owner);213+ const friend = generateSecretKey();214+ await rpc(host, owner, "claim");215+ await rpc(host, owner, "setpolicy", { notify: { digest: true } });216+ const stub = env.RELAY.getByName("weekly");217+ const wraps = () => runInDurableObject(stub, async (r: Relay) => r.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM events WHERE kind=1059`).one().n);218+ // The first tick with the switch on starts the clock and says nothing.219+ await runInDurableObject(stub, async (r: Relay) => r.alarm());220+ expect(await wraps()).toBe(0);221+ await rpc(host, owner, "setmember", getPublicKey(friend), { name: "friend" });222+ const c = await WS.connect(host);223+ expect((await c.ok(ev(owner, 1, "a week's worth"))).ok).toBe(true);224+ await runInDurableObject(stub, async (_r: Relay, state) => state.storage.put("lastDigest", now() - 8 * 86400));225+ await runInDurableObject(stub, async (r: Relay) => r.alarm());226+ expect(await wraps()).toBe(1);227+ const o = await WS.connect(host);228+ await o.auth(owner, host);229+ const [wrap] = await o.req({ kinds: [1059], "#p": [pk] });230+ const rumor = unwrapEvent(wrap, owner);231+ expect(rumor.kind).toBe(14);232+ expect(rumor.content).toMatch(/People: 1 joined/);233+ expect(rumor.content).toMatch(/Stored: 1 events/);234+ // A day later there is nothing new to say, and it is not a week yet.235+ await runInDurableObject(stub, async (r: Relay) => r.alarm());236+ expect(await wraps()).toBe(1);237+ // A quiet week still gets one line: push last week's changes out of the window first.238+ await runInDurableObject(stub, async (r: Relay, state) => {239+ r.sql.exec(`UPDATE members SET joined_at = joined_at - 30 * 86400`);240+ r.sql.exec(`UPDATE events SET created_at = created_at - 30 * 86400 WHERE kind = 1`);241+ await state.storage.put("lastDigest", now() - 8 * 86400);242+ });243+ await runInDurableObject(stub, async (r: Relay) => r.alarm());244+ expect(await wraps()).toBe(2);245+ const two = await o.req({ kinds: [1059], "#p": [pk] });246+ const texts = two.map((w) => unwrapEvent(w, owner).content);247+ expect(texts.some((t) => /Nothing changed/.test(t))).toBe(true);248+ });249+});@@ -174,7 +174,7 @@ describe("relay-signed notifications", () => {
174174 const stranger = generateSecretKey();175175 await rpc(host, owner, "claim");176176 await rpc(host, owner, "setmember", getPublicKey(member), {});177- expect((await rpc(host, owner, "getpolicy")).result.notify).toEqual({ reports: false, fuel: false, jobs: false, succession: false });177+ expect((await rpc(host, owner, "getpolicy")).result.notify).toEqual({ reports: false, fuel: false, jobs: false, succession: false, digest: false });178178 179179 // Off by default: a report stores nothing.180180 const s = await WS.connect(host);@@ -184,7 +184,7 @@ describe("relay-signed notifications", () => {
184184 expect(await o.live({ kinds: [1059], "#p": [ownerPk] }, "inbox")).toEqual([]);185185 186186 const set = await rpc(host, owner, "setpolicy", { notify: { reports: true } });187- expect(set.result.notify).toEqual({ reports: true, fuel: false, jobs: false, succession: false });187+ expect(set.result.notify).toEqual({ reports: true, fuel: false, jobs: false, succession: false, digest: false });188188 const m = await WS.connect(host);189189 await m.auth(member, host);190190 expect(await m.live({ kinds: [1059] }, "mine")).toEqual([]);