1// The relay: one Durable Object per name, holding its SQLite database and2// its live websockets (hibernating while idle). Protocol handling mirrors3// relay.go; policy is per relay and owner-managed (see manage.ts).4import { callbackOrigins } from "./push-policy.ts";5import { pushTick, queuePush, nextPush, PUSH_SCHEMA } from "./push.ts";6import { BACKUP_SCHEMA, backupBytes } from "./backups.ts";7import { GIT_SQLITE_SCHEMA } from "./git-sqlite.ts";8import { GIT_CATALOG_SCHEMA } from "./git-catalog.ts";9import { DELIVERY_SCHEMA, deliveryTick, queueDelivery } from "./delivery.ts";10import { graspTick, isGitPath, graspCORS } from "./grasp.ts";11import { holdGrasp, graspVisible } from "./grasp-state.ts";12import { queueMirrors } from "./site-mirror.ts";13import { syncSiteIndex, forgetSites, serveSite } from "./sites.ts";14import { DurableObject } from "cloudflare:workers";15import { sha256 } from "@noble/hashes/sha2.js";16import { difficulty, expiration, hasTag, isPrivate, now, tagValues, validate, canonical, type Event } from "./event.ts";17import { match, parseFilter, type Filter } from "./filter.ts";18import { hllOffset } from "./hll.ts";19import { Negentropy, bytesToHex, hexToBytes } from "./negentropy.ts";20import { discovery } from "./nip66.ts";21import { Audit } from "./audit.ts";22import { ERR_DUPLICATE, ERR_TOO_BIG, Store, type Access } from "./store.ts";23import { Settings, isReplaceable, isProtected, featureOn } from "./settings.ts";24import { blobBytes, type Blob } from "./blossom.ts";25import { DIGEST_DAYS, digestText, notify, fuelLow, fuelText } from "./notify.ts";26import { Succession } from "./succession.ts";27import { isGated, isManagementRequest, route } from "./routes.ts";28import { guestPass, readGate, writeGate } from "./gates.ts";29import { Fuel, fuelConfig, acceptReceipt, type Fetcher, type LnurlParams } from "./fuel.ts";30import { Identity } from "./identity.ts";31import { Bucket } from "./ratelimit.ts";32import { dumpBytes, dumpDue, writeDump } from "./dumps.ts";33import { importBytes } from "./imports.ts";34import { claimFromProfile } from "./nip05.ts";35import type { PullFilter, PullJob, PullResult } from "./pull.ts";36import { leaseDays, leaseNames, validName } from "./names.ts";37import { MAX_JOBS, MAX_STANDING, checkJob, finishRun, newJobID, pruneFinished, pullView, runRound, startRun, type Job, type JobSpec } from "./jobs.ts";38import { Hostnames, forgetDomains } from "./domains.ts";39import { hostOf } from "./edge.ts";40import { groupFacts, handleGroupEvent, isGroupManagement, isNIP43Request } from "./groups.ts";41import { markView, notePresence, viewsTick } from "./views.ts";42import { KIND_VANISH, KIND_AUTH, KIND_REPORT, KIND_NOSTR_CONNECT, KIND_GROUP_MEMBERS, KIND_GROUP_PINS, KIND_RELAY_DISCOVERY, KIND_MARMOT_GROUP, KIND_PUSH_REGISTRATION } from "./kinds.ts";43import { marmotPrincipal } from "./marmot.ts";44import { RepositoryAccess } from "./repository-access.ts";4546export interface Env {47 RELAY: DurableObjectNamespace<Relay>;48 MEDIA: R2Bucket;49 DOMAIN: string;50 DEV_RELAY: string;51 PUSH_CALLBACK_ORIGINS?: string; // JSON array of operator-trusted public HTTPS origins52 LEASE_DAYS?: string; // how long a temporary relay lives; 14 by default53 // Cloudflare's rate limit bindings; without them (celld) edge.ts keeps54 // token buckets in memory instead.55 LEASE_LIMIT_IP?: RateLimit; // leases per address per minute56 LEASE_LIMIT_ALL?: RateLimit; // leases per minute, everyone together57 // The header the client's address arrives in; cf-connecting-ip if unset,58 // no address at all if empty (edge.ts, clientIP).59 CLIENT_IP_HEADER?: string;60 // Custom domains (see domains.ts): hostname to relay name, and the61 // Cloudflare for SaaS credentials. Without the token the feature is off.62 HOSTS?: KVNamespace;63 CF_API_TOKEN?: string;64 ZONE_ID?: string;65 CNAME_TARGET?: string; // what owners point their CNAME at; customers.<DOMAIN> if unset66 // Fuel (see fuel.ts); all optional.67 LIGHTNING_ADDRESS?: string;68 SERVICE_PUBKEY?: string;69 FREE_EVENTS_MB?: string;70 FREE_MEDIA_MB?: string;71 FREE_ACTIVE_HOURS?: string;72 FREE_ROWS_WRITTEN?: string;73 SATS_PER_GB_MONTH_EVENTS?: string;74 SATS_PER_GB_MONTH_MEDIA?: string;75 SATS_PER_ACTIVE_HOUR?: string;76 SATS_PER_MILLION_ROWS?: string;77 // Metrics (optional): a Workers Analytics Engine dataset that takes one78 // data point per usage flush, next to the meter log line (flushUsage).79 METRICS?: AnalyticsEngineDataset;80}8182export const MAX_MESSAGE = 1024 * 1024;83const MAX_SYNC = 100_000;84// How long a leased relay keeps everything, so a claim inherits a bounded85// window until the owner resets the rules.86const LEASE_RETENTION_DAYS = 14;87// Forks are owner actions, not rate limited at the apex, so one an hour.88const FORK_INTERVAL = 3600;8990// ConnState is everything about a websocket that must survive hibernation.91export interface ConnState {92 challenge: string;93 host: string; // Host header, for NIP-42 relay-tag checks94 ip: string; // the client's address as the worker saw it; "unknown" outside Cloudflare95 authed: string[];96 subs: Record<string, Filter[]>;97 // Open NIP-77 syncs by id, each with its filter: enough to rebuild the98 // session's item list after a wake. Absent on sockets from before.99 syncs?: Record<string, Filter>;100}101102const FLUSH_BYTES = 256 * 1024;103// An address gets this many times a connection's per-minute allowance, so a104// few tabs from one place are fine and a swarm of sockets is not.105export const IP_LIMIT_MULTIPLE = 4;106107// clientIP is the address the worker stamped on the request (index.ts).108function clientIP(req: Request): string {109 return req.headers.get("x-relay-ip") || "unknown";110}111// The legacy fuel estimate caps entrypoint gaps at this window.112const IDLE_GRACE_MS = 10_000;113114export class Relay extends DurableObject<Env> {115 store: Store;116 settings: Settings;117 fuel: Fuel;118 identity: Identity;119 audit: Audit;120 slug = "";121 private states = new Map<WebSocket, ConnState>();122 private syncs = new Map<WebSocket, Map<string, Negentropy>>();123 private buckets = new Map<WebSocket, { events: Bucket; reqs: Bucket }>();124 // Per-address buckets, in memory only: they reset when the object sleeps.125 private ipBuckets = new Map<string, { events: Bucket; reqs: Bucket }>();126 // Usage counters, flushed to the usage table in batches. The last three127 // are for the meter log line only: events accepted, events refused with128 // an OK false, and REQ or COUNT messages served.129 private meter = { bytesIn: 0, bytesOut: 0, rowsRead: 0, rowsWritten: 0, activeMs: 0, accepted: 0, refused: 0, reqs: 0 };130 // When the object last did work, for the active-time meter (ms). Zero after a wake.131 private lastActive = 0;132 // Outbound HTTP to the lightning provider and to Cloudflare; tests substitute a fake.133 fetcher: Fetcher = (u, i) => fetch(u, i);134 // The heir and the dead-man's switch (succession.ts).135 succession = new Succession(this);136 // A job round in flight, so overlapping alarms do not run two.137 private working = false;138 readonly repositoryAccess = new RepositoryAccess();139 // Presence (views.ts): who wrote lately, in memory; the sockets are asked140 // directly. A broadcast is at most one per PRESENCE_THROTTLE_S.141 presenceActive = new Map<string, number>();142 presenceAt = 0;143 presenceTimer: ReturnType<typeof setTimeout> | null = null;144 // Write-triggered views wait a moment so a burst republishes once.145 viewDirty = new Set<string>();146 viewTimer: ReturnType<typeof setTimeout> | null = null;147 // The custom hostnames client, null when the host has not enabled them.148 hostnames: Hostnames | null;149150 constructor(ctx: DurableObjectState, env: Env) {151 super(ctx, env);152 this.store = new Store(ctx.storage.sql);153 this.store.onSitesChanged = () => this.queueSites();154 this.settings = new Settings(ctx.storage.sql);155 this.fuel = new Fuel(ctx.storage.sql, fuelConfig(env as unknown as Record<string, unknown>));156 this.identity = new Identity(ctx.storage);157 this.audit = new Audit(ctx.storage.sql, () => this.slug);158 this.hostnames = env.CF_API_TOKEN && env.ZONE_ID && env.HOSTS ? new Hostnames(env.CF_API_TOKEN, env.ZONE_ID, (u, i) => this.fetcher(u, i)) : null;159 ctx.blockConcurrencyWhile(async () => {160 this.store.init();161 this.settings.load();162 this.sql.exec(PUSH_SCHEMA);163 this.sql.exec(DELIVERY_SCHEMA);164 this.sql.exec(BACKUP_SCHEMA);165 this.sql.exec(GIT_SQLITE_SCHEMA);166 this.sql.exec(GIT_CATALOG_SCHEMA);167 this.store.hidden = this.settings.hiddenEvents;168 this.store.searchMode = () => this.settings.policy.features.search;169 this.fuel.init();170 this.slug = (await ctx.storage.get<string>("slug")) ?? "";171 this.fuel.lnurl = (await ctx.storage.get<LnurlParams>("lnurl")) ?? null;172 await this.succession.load();173 await this.identity.load();174 });175 }176177 get sql(): SqlStorage {178 return this.ctx.storage.sql;179 }180181 // storage is the object's key-value store, for the modules that keep a182 // little state beside the tables.183 get storage(): DurableObjectStorage {184 return this.ctx.storage;185 }186187 get media(): R2Bucket {188 return this.env.MEDIA;189 }190191 get domain(): string {192 return this.env.DOMAIN;193 }194195 // hosts is the custom domain map the worker routes by; absent on a host196 // without the namespace, where hostnames is null too.197 get hosts(): KVNamespace | undefined {198 return this.env.HOSTS;199 }200201 get cnameTarget(): string {202 return this.env.CNAME_TARGET || "customers." + this.env.DOMAIN;203 }204205 get pushCallbackOrigins(): string[] {206 try { return callbackOrigins(JSON.parse(this.env.PUSH_CALLBACK_ORIGINS || "[]")) ?? []; }207 catch { return []; }208 }209210 // meterPush accounts for bounded callback work, including failed attempts.211 meterPush(rowsRead: number, rowsWritten: number, activeMs = 0) {212 this.meter.rowsRead += rowsRead;213 this.meter.rowsWritten += rowsWritten;214 this.meter.activeMs += activeMs;215 }216217 // relays reaches the other objects on this host, for pulls between them.218 get relays(): DurableObjectNamespace<Relay> {219 return this.env.RELAY;220 }221222 // meterBytes lets HTTP paths (bridge, media) add to the traffic counters.223 meterBytes(bytesIn: number, bytesOut: number) {224 this.meter.bytesIn += bytesIn;225 this.meter.bytesOut += bytesOut;226 }227228 // touch estimates awake time from entrypoint gaps, with one window on wake.229 // It does not measure long handlers or Cloudflare's billable duration:230 // idle objects eligible for hibernation incur no provider duration charge.231 private touch() {232 const t = Date.now();233 this.meter.activeMs += this.lastActive === 0 ? IDLE_GRACE_MS : Math.min(t - this.lastActive, IDLE_GRACE_MS);234 this.lastActive = t;235 }236237 // ---- relay identity and the NIP-43 roster ----238239 // publishMembership signs everything the relay vouches for from its member240 // list and settings, in one place so the NIP-43 roster and the NIP-29 group241 // can never disagree: the roster (13534) and, for an add or a removal, its242 // 8000/8001 delta; the NIP-29 put-user or remove-user record for each243 // change given (a change without `added` is a role change); the group's244 // metadata, admins, roles and, when the directory is public, members245 // (39000-39003); the NIP-43 role definitions (33534); and the relay's own246 // profile (kind 0) when its name, description or icon changed.247 async publishMembership(...changes: { pubkey: string; added?: boolean }[]) {248 if (this.settings.policy.owner === "") return;249 await this.identity.ensure();250 const t = now();251 const events: Event[] = [];252 for (const c of changes) {253 if (c.added !== undefined) events.push(this.identity.delta(c.added, c.pubkey));254 if (c.added === false) events.push(this.identity.removeUser(this.slug, c.pubkey));255 else events.push(this.identity.putUser(this.slug, c.pubkey, this.rolesOf(c.pubkey)));256 }257 const f = groupFacts(this);258 // A members list that is no longer public is taken down.259 if (f.members === null) {260 for (const r of this.sql.exec<{ id: string }>(`SELECT id FROM events WHERE kind=? AND pubkey=?`, KIND_GROUP_MEMBERS, this.identity.pubkey).toArray()) this.store.deleteEvent(r.id);261 }262 events.push(this.identity.roster(this.settings.members(), t));263 events.push(...this.identity.group(f, t));264 this.emit(events, t);265 markView(this, "profiles");266 await this.publishDiscovery();267 }268269 // publishDiscovery signs the NIP-66 record the relay keeps about itself270 // (nip66.ts), under its primary name, and only when what it would say has271 // changed: the rules, the kind rules, the identity fields and the fuel272 // state all show in it. Called with the membership records and daily273 // from the alarm, so a change made by a method that publishes nothing274 // else still reaches the record within a day.275 async publishDiscovery() {276 if (this.settings.policy.owner === "") return;277 await this.identity.ensure();278 if (!featureOn(this.settings.policy, "discovery")) {279 // Switched off: the record comes down and the fingerprint with it.280 for (const r of this.sql.exec<{ id: string }>(`SELECT id FROM events WHERE kind=? AND pubkey=?`, KIND_RELAY_DISCOVERY, this.identity.pubkey).toArray()) this.store.deleteEvent(r.id);281 await this.ctx.storage.delete("nip66-fp");282 return;283 }284 const d = discovery(this, this.slug + "." + this.domain);285 const fp = JSON.stringify([d.tags, d.content]);286 const row = this.sql.exec<{ created_at: number }>(`SELECT created_at FROM events WHERE kind=? AND pubkey=? ORDER BY created_at DESC LIMIT 1`, KIND_RELAY_DISCOVERY, this.identity.pubkey).toArray()[0];287 if (row && fp === (await this.ctx.storage.get<string>("nip66-fp"))) return;288 const t = now();289 this.emit([this.identity.sign(KIND_RELAY_DISCOVERY, d.tags, d.content, Math.max(t, (row?.created_at ?? 0) + 1))], t);290 await this.ctx.storage.put("nip66-fp", fp);291 }292293 // ---- views (views.ts) ----294295 // rowsWritten is the running count of rows written since the last flush,296 // for a view to say what its fold cost.297 rowsWritten(): number {298 this.tally();299 return this.meter.rowsWritten;300 }301302 // authedNow lists every pubkey with an authenticated socket open.303 authedNow(): string[] {304 const out = new Set<string>();305 for (const ws of this.ctx.getWebSockets()) for (const pk of this.state(ws).authed) out.add(pk);306 return [...out];307 }308309 // publishPins signs the group's pin list (39005); an empty list takes the310 // record down.311 async publishPins() {312 if (this.settings.policy.owner === "") return;313 await this.identity.ensure();314 const tags = this.settings.pins();315 if (tags.length === 0) {316 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);317 return;318 }319 this.emit([this.identity.pins(this.slug, tags)], now());320 }321322 private rolesOf(pubkey: string): string[] {323 const r = this.settings.roleOf(pubkey);324 return r && r !== "member" ? [r] : [];325 }326327 // emit stores relay-signed events and broadcasts the ones that were new.328 emit(events: Event[], t: number) {329 for (const e of events) {330 const err = this.store.save(e, t);331 if (!err) this.broadcast(e);332 }333 }334335 // setMember edits or adds a person and publishes the roster if membership changed.336 async setMember(pubkey: string, patch: { name?: string | null; note?: string; via?: string; invitedBy?: string; keepDays?: number; maxBytes?: number }, force = false): Promise<string> {337 const was = this.settings.isAllowed(pubkey);338 const err = this.settings.upsertMember(pubkey, patch, now(), force);339 if (err) return err;340 if (!was) await this.publishMembership({ pubkey, added: true });341 return "";342 }343344 async removeMember(pubkey: string): Promise<boolean> {345 if (!this.settings.removeMember(pubkey)) return false;346 if (this.settings.policy.reads === "members") this.evict(pubkey, "restricted: this relay only serves its members", false);347 await this.publishMembership({ pubkey, added: false });348 return true;349 }350351 // removeSubtree removes a member and everyone they invited, plain members352 // only (see Settings.subtree), and publishes membership once for all of353 // them. Returns who went.354 async removeSubtree(pubkey: string): Promise<string[]> {355 const gone: string[] = [];356 for (const pk of this.settings.subtree(pubkey)) {357 if (!this.settings.removeMember(pk)) continue;358 if (this.settings.policy.reads === "members") this.evict(pk, "restricted: this relay only serves its members", false);359 gone.push(pk);360 }361 if (gone.length) await this.publishMembership(...gone.map((pk) => ({ pubkey: pk, added: false })));362 return gone;363 }364365 // ban refuses a pubkey from now on; with erase, everything they wrote and366 // uploaded goes too, their profile included.367 async ban(pubkey: string, reason: string, erase = false): Promise<void> {368 const was = this.settings.isAllowed(pubkey);369 this.settings.setBan(pubkey, true, reason, now());370 this.evict(pubkey, "blocked: you are banned from this relay", true);371 if (erase) {372 this.store.eraseAuthor(pubkey);373 for (const b of this.sql.exec<{ sha256: string }>(`SELECT sha256 FROM blobs WHERE uploader=?`, pubkey).toArray()) await this.deleteBlob(b.sha256);374 }375 if (was) await this.publishMembership({ pubkey, added: false });376 }377378 // evict closes the door on a pubkey: subscriptions are ended and, for a379 // ban, the socket is closed; hibernating sockets are covered too.380 evict(pubkey: string, reason: string, close: boolean) {381 for (const ws of this.ctx.getWebSockets()) {382 const s = this.state(ws);383 if (!s.authed.includes(pubkey)) continue;384 for (const id of Object.keys(s.subs)) this.send(ws, "CLOSED", id, reason);385 s.subs = {};386 this.syncs.delete(ws);387 this.persist(ws, s);388 if (close) {389 try {390 ws.close(4403, reason.slice(0, 120));391 } catch {392 /* already closed */393 }394 }395 }396 }397398 // enforceReads runs after the read rule changed: every open subscription399 // is judged again and the ones the rule no longer admits are closed, so a400 // relay that turned members-only stops feeding the sockets it let in401 // while it was open. Sync sessions go with them.402 enforceReads() {403 for (const ws of this.ctx.getWebSockets()) {404 const s = this.state(ws);405 let changed = false;406 for (const [id, filters] of Object.entries(s.subs)) {407 const { reason } = readGate(this, s, filters);408 if (!reason) continue;409 this.send(ws, "CLOSED", id, reason);410 delete s.subs[id];411 changed = true;412 }413 if (this.settings.mayRead(s.authed)) {414 this.syncs.delete(ws);415 if (s.syncs && Object.keys(s.syncs).length) {416 delete s.syncs;417 changed = true;418 }419 }420 if (changed) this.persist(ws, s);421 }422 }423424 // virtualConn represents an HTTP caller as if it had authenticated over the socket.425 virtualConn(host: string, pubkey: string, ip = "unknown"): ConnState {426 void this.succession.seen(pubkey);427 return { challenge: "", host, authed: [pubkey], subs: {}, ip };428 }429430 // ---- addresses ----431432 // blockIP refuses an address from now on and drops its open sockets.433 blockIP(ip: string, reason: string) {434 this.settings.setIPBlock(ip, true, reason, now());435 for (const ws of this.ctx.getWebSockets()) {436 if (this.state(ws).ip !== ip) continue;437 try {438 ws.close(4403, "blocked: this address is blocked from this relay");439 } catch {440 /* already closed */441 }442 }443 }444445 // ipLimit spends one token from the address's bucket. "" allows.446 ipLimit(ip: string, which: "events" | "reqs"): string {447 if (ip === "unknown") return "";448 let b = this.ipBuckets.get(ip);449 if (!b) {450 if (this.ipBuckets.size >= 10_000) this.ipBuckets.clear();451 b = { events: new Bucket(this.settings.policy.eventsPerMinute * IP_LIMIT_MULTIPLE), reqs: new Bucket(this.settings.policy.reqsPerMinute * IP_LIMIT_MULTIPLE) };452 this.ipBuckets.set(ip, b);453 }454 const p = this.settings.policy;455 return b[which].take((which === "events" ? p.eventsPerMinute : p.reqsPerMinute) * IP_LIMIT_MULTIPLE);456 }457458 isMember(pubkey: string): boolean {459 return this.settings.isAllowed(pubkey);460 }461462 // mayUpload applies the write policy to Blossom uploads. "" allows.463 mayUpload(pubkey: string, host: string): string {464 if (!featureOn(this.settings.policy, "files")) return "restricted: files are switched off on this relay";465 if (this.settings.isUnclaimed()) return "restricted: this relay is unclaimed";466 if (this.settings.leaseExpired(now())) return "restricted: this temporary relay has expired";467 if (this.settings.isBanned(pubkey)) return "blocked: this pubkey is banned from this relay";468 const why = this.settings.mayWrite(pubkey);469 if (why) return why.replace("publish here", "upload here").replace("accepts events", "accepts uploads");470 if (this.fuelStatus().outOfFuel) return "restricted: this relay is out of fuel; zap it at https://" + host + "/ to top up";471 return "";472 }473474 // teardown deletes everything this relay holds and returns the name to475 // unclaimed. Every socket is closed. Nothing is recoverable afterwards.476 async teardown(): Promise<void> {477 if (!this.repositoryAccess.owned) return this.repositoryAccess.run("teardown", () => this.teardown(), () => { throw new Error("relay operation in progress; retry teardown"); });478 await forgetSites(this);479 await forgetDomains(this);480 for (const ws of this.ctx.getWebSockets()) {481 try {482 ws.close(4410, "relay deleted");483 } catch {484 /* already closed */485 }486 }487 let cursor: string | undefined;488 do {489 const list = await this.media.list({ prefix: `${this.slug}/`, cursor });490 if (list.objects.length) await this.media.delete(list.objects.map((o) => o.key));491 cursor = list.truncated ? list.cursor : undefined;492 } while (cursor);493 await this.ctx.storage.deleteAlarm();494 await this.ctx.storage.deleteAll();495 this.states.clear();496 this.syncs.clear();497 this.buckets.clear();498 this.ipBuckets.clear();499 this.meter = { bytesIn: 0, bytesOut: 0, rowsRead: 0, rowsWritten: 0, activeMs: 0, accepted: 0, refused: 0, reqs: 0 };500 this.store = new Store(this.sql);501 this.store.onSitesChanged = () => this.queueSites();502 this.settings = new Settings(this.sql);503 this.fuel = new Fuel(this.sql, this.fuel.cfg);504 this.identity = new Identity(this.ctx.storage);505 this.audit = new Audit(this.sql, () => this.slug);506 this.store.init();507 this.settings.load();508 this.sql.exec(PUSH_SCHEMA);509 this.sql.exec(DELIVERY_SCHEMA);510 this.sql.exec(BACKUP_SCHEMA);511 this.sql.exec(GIT_SQLITE_SCHEMA);512 this.sql.exec(GIT_CATALOG_SCHEMA);513 this.store.hidden = this.settings.hiddenEvents;514 this.store.searchMode = () => this.settings.policy.features.search;515 this.fuel.init();516 await this.ctx.storage.put("slug", this.slug);517 }518519 private sitesSync: Promise<boolean> | null = null;520 syncSites(): Promise<boolean> {521 if (!this.sitesSync) this.sitesSync = syncSiteIndex(this).finally(() => { this.sitesSync = null; });522 return this.sitesSync;523 }524 private sitesQueued = false;525 private queueSites() {526 if (this.sitesQueued) return;527 this.sitesQueued = true;528 this.ctx.waitUntil(Promise.resolve().then(async () => {529 this.sitesQueued = false;530 try {531 await this.syncSites();532 } catch (err) {533 console.log("site index sync failed: " + String(err));534 }535 await this.ensureAlarm(now() + 1);536 }));537 }538539 async deleteBlob(sha: string) {540 await this.media.delete(`${this.slug}/${sha}`);541 this.sql.exec(`DELETE FROM blobs WHERE sha256=?`, sha);542 }543544 // listBlobs is for another relay on this host pulling our files. The545 // caller says which pubkeys it has proved; the read rule decides.546 listBlobs(pubkeys: string[]): Blob[] | string {547 const gate = this.settings.mayRead(pubkeys);548 if (gate) return gate;549 return this.sql.exec<Blob>(`SELECT * FROM blobs ORDER BY uploaded`).toArray();550 }551552 // ---- owner presence and succession ----553554 // ---- temporary leases ----555556 // lease turns an unclaimed relay into a temporary one: open to everyone,557 // everything kept for a bounded window, wiped at `until` unless claimed.558 // holder "" lets anyone claim; a pubkey reserves the claim. Returns "" or559 // a reason.560 async lease(name: string, host: string, until: number, holder: string): Promise<string> {561 return this.repositoryAccess.run("control", async () => {562 if (!this.settings.isUnclaimed()) return "taken";563 if (this.slug !== name) {564 this.slug = name;565 await this.ctx.storage.put("slug", name);566 }567 const day = new Date(until * 1000).toISOString().slice(0, 10);568 this.settings.update({569 lease: { until, holder },570 writes: "open",571 reads: "open",572 name: "",573 description: `Temporary relay. Anyone can read and write here until ${day}; then everything on it is deleted. To keep it, claim it at https://${host}/ (sign once). Or claim a new name and pull from this one.`,574 });575 this.settings.setRetention(null, LEASE_RETENTION_DAYS);576 await this.ensureAlarm(until);577 return "";578 }, () => "restricted: relay operation in progress; retry");579 }580581 // ---- forking: a new name, this relay's events pulled into it ----582583 // forkRelay leases a new name, reserved for `holder`, and has it pull584 // this relay. `people` copies the plain members along. Nothing new in585 // the protocol: a lease, a pull job, a claim. One fork an hour.586 async forkRelay(host: string, opts: { name?: string; holder: string; filter?: PullFilter; people?: boolean }): Promise<{ name: string; url: string; console: string; holder: string; expires_at: number } | string> {587 const t = now();588 const last = (await this.ctx.storage.get<number>("lastFork")) ?? 0;589 if (t - last < FORK_INTERVAL) return `restricted: one fork an hour; the last was ${Math.ceil((t - last) / 60)} minutes ago`;590 const suffix = host.startsWith(this.slug + ".") ? host.slice(this.slug.length) : "." + this.env.DOMAIN;591 const secure = this.relayURL(host).startsWith("wss");592 const until = t + leaseDays(this.env) * 86400;593 if (opts.name && (!validName(opts.name) || opts.name === this.slug)) return "invalid: that is not a usable name";594 const candidates = opts.name ? [opts.name] : [...leaseNames()];595 for (const name of candidates) {596 const newHost = name + suffix;597 const stub = this.relays.getByName(name);598 const err = await stub.lease(name, newHost, until, opts.holder);599 if (err) {600 if (opts.name) return "restricted: that name is taken";601 continue;602 }603 const people = opts.people ? this.settings.members().filter((m) => m.role === "member").map((m) => ({ pubkey: m.pubkey, name: m.name, note: m.note })) : [];604 const adopted = await stub.adoptFrom(this.relayURL(host), opts.filter ?? {}, people, host);605 if (adopted) return adopted;606 await this.ctx.storage.put("lastFork", t);607 return { name, url: (secure ? "wss://" : "ws://") + newHost, console: (secure ? "https://" : "http://") + newHost + "/", holder: opts.holder, expires_at: until };608 }609 return "error: no free name found, try again";610 }611612 // adoptFrom is the forked side, reached over RPC only: a fresh lease613 // takes the people and starts pulling from the source.614 async adoptFrom(sourceURL: string, filter: PullFilter, people: { pubkey: string; name: string | null; note: string }[], sourceHost: string): Promise<string> {615 return this.repositoryAccess.run("control", async () => {616 if (!this.settings.isLeased()) return "restricted: only a fresh lease can adopt";617 if (this.store.stats().events > 0) return "restricted: this lease already holds events";618 const day = new Date((this.settings.policy.lease?.until ?? 0) * 1000).toISOString().slice(0, 10);619 this.settings.update({ description: `Forked from ${sourceHost}. Temporary until ${day} unless claimed; then everything on it is deleted. Claim it to keep it.` });620 const t = now();621 for (const m of people) this.settings.upsertMember(m.pubkey, { name: m.name, note: m.note, via: "forked" }, t, true);622 const r = await this.addChecked({ kind: "pull", label: "pull", relays: [sourceURL], 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, last: null });623 return typeof r === "string" ? r : "";624 }, () => "restricted: relay operation in progress; retry");625 }626627 // ---- jobs: pulls, backfills, rebroadcasts, once or standing ----628629 // jobs is the persisted list. A job left from before the list existed630 // is folded in on first read.631 async jobs(): Promise<Job[]> {632 let list = (await this.ctx.storage.get<Job[]>("jobs")) ?? [];633 const old = await this.ctx.storage.get<PullJob>("pull");634 if (old) {635 list = [...list, { id: newJobID(), kind: "pull", label: "pull", relays: [old.url], filter: {}, every: 0, createdAt: old.startedAt, nextRun: now(), running: false, startedAt: old.startedAt, rounds: old.rounds, failures: 0, relayIndex: 0, cursor: 0, stored: old.stored, skipped: old.skipped, blobs: old.blobs, sent: 0, refused: 0, last: null }];636 await this.ctx.storage.delete("pull");637 await this.ctx.storage.delete("lastPull");638 await this.ctx.storage.put("jobs", list);639 }640 return list;641 }642643 private async saveJobs(list: Job[]) {644 await this.ctx.storage.put("jobs", list);645 }646647 // addJob validates, records and wakes the alarm. Returns the job or a reason.648 async addJob(raw: unknown): Promise<Job | string> {649 const spec = checkJob(raw, this);650 if (typeof spec === "string") return spec;651 return this.addChecked(spec);652 }653654 async addChecked(spec: JobSpec): Promise<Job | string> {655 if (this.fuelStatus().outOfFuel) return "restricted: this relay is out of fuel";656 let list = pruneFinished(await this.jobs());657 if (spec.every > 0 && list.filter((j) => j.every > 0).length >= MAX_STANDING) return `restricted: at most ${MAX_STANDING} standing jobs`;658 if (list.length >= MAX_JOBS) return `restricted: at most ${MAX_JOBS} jobs; remove one first`;659 if (spec.kind === "pull" && list.some((j) => j.kind === "pull" && j.every === 0 && (j.running || j.nextRun > 0) && j.relays.join() === spec.relays.join())) return "restricted: a pull from there is already running";660 const job: Job = { ...spec, id: newJobID(), createdAt: now(), nextRun: now() };661 list = [...list, job];662 await this.saveJobs(list);663 await this.ctx.storage.setAlarm(Date.now() + 50);664 return job;665 }666667 async removeJob(id: string): Promise<boolean> {668 const list = await this.jobs();669 const rest = list.filter((j) => j.id !== id);670 if (rest.length === list.length) return false;671 await this.saveJobs(rest);672 return true;673 }674675 // runJob makes a job due now.676 async runJob(id: string): Promise<boolean> {677 const list = await this.jobs();678 const job = list.find((j) => j.id === id);679 if (!job) return false;680 if (!job.running) job.nextRun = now();681 await this.saveJobs(list);682 await this.ctx.storage.setAlarm(Date.now() + 50);683 return true;684 }685686 // pullStart and pullStatus are the first console's view: one pull at a time.687 async pullStart(url: string): Promise<string> {688 const r = await this.addChecked({ kind: "pull", label: "pull", relays: [url], 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, last: null });689 return typeof r === "string" ? r : "";690 }691692 async pullStatus(): Promise<{ running: PullJob | null; last: PullResult | null }> {693 return pullView(await this.jobs());694 }695696 // jobsTick runs one round of the job that is due. Returns whether another697 // round is due soon. A few failed rounds in a row end a run with the698 // reason; a standing job tries again at its next interval.699 private async jobsTick(): Promise<boolean> {700 if (this.working) return true;701 const list = await this.jobs();702 const t = now();703 const job = list.find((j) => j.running) ?? list.find((j) => j.nextRun > 0 && j.nextRun <= t);704 if (!job) return false;705 if (!job.running) startRun(job, t);706 this.working = true;707 let r: { more: boolean; error: string };708 try {709 r = await runRound(this, job);710 } finally {711 this.working = false;712 }713 // The job may have been removed while the round ran.714 const fresh = await this.jobs();715 const slot = fresh.findIndex((j) => j.id === job.id);716 if (slot < 0) return fresh.some((j) => j.running || (j.nextRun > 0 && j.nextRun <= now()));717 // A finished run is worth a word to the owner, when they asked for one.718 const finish = (error: string) => {719 finishRun(job, error, now());720 error = job.last?.error ?? error;721 const where = (job.kind === "push" ? "to " : "from ") + job.relays.join(", ");722 const outcome = error ? `failed after ${job.rounds} rounds: ${error}` : job.kind === "push" ? `finished: ${job.sent} events sent${job.refused ? ", " + job.refused + " refused" : ""}` : `finished: ${job.stored} events${job.blobs ? " and " + job.blobs + " files" : ""}${job.skipped ? ", " + job.skipped + " skipped" : ""}`;723 void notify(this, "jobs", `${job.label} ${where} ${outcome}.`, "jobs on " + this.slug);724 };725 if (r.error) {726 job.failures++;727 if (job.failures >= 3) finish(r.error);728 } else {729 job.failures = 0;730 if (!r.more) finish("");731 }732 fresh[slot] = job;733 await this.saveJobs(fresh);734 return fresh.some((j) => j.running || (j.nextRun > 0 && j.nextRun <= now()));735 }736737 // nextJobRun is the earliest standing run, or 0.738 private async nextJobRun(): Promise<number> {739 let at = 0;740 for (const j of await this.jobs()) if (j.nextRun > 0 && (at === 0 || j.nextRun < at)) at = j.nextRun;741 return at;742 }743744 // ---- fuel plumbing ----745746 tally() {747 const d = this.store.drain();748 this.meter.rowsRead += d.rowsRead;749 this.meter.rowsWritten += d.rowsWritten;750 if (this.meter.bytesIn + this.meter.bytesOut > FLUSH_BYTES || this.meter.rowsRead + this.meter.rowsWritten > 5000) this.flushUsage();751 }752753 // flushUsage moves the counters to the usage table, and reports the same754 // numbers outward: one JSON line on the console, which Workers Logs and755 // any OTLP logs destination carry with the relay's name in it, and one756 // Analytics Engine data point when the host bound a dataset. Both are757 // deltas since the last flush; a destination sums them.758 flushUsage() {759 const m = this.meter;760 if (m.bytesIn + m.bytesOut + m.rowsRead + m.rowsWritten + m.activeMs === 0) return;761 this.meter = { bytesIn: 0, bytesOut: 0, rowsRead: 0, rowsWritten: 0, activeMs: 0, accepted: 0, refused: 0, reqs: 0 };762 this.fuel.record(now(), m);763 this.store.drain();764 const connections = this.connections();765 console.log(JSON.stringify({ msg: "meter", relay: this.slug, ...m, connections }));766 this.env.METRICS?.writeDataPoint({767 indexes: [this.slug],768 blobs: ["meter"],769 doubles: [m.bytesIn, m.bytesOut, m.rowsRead, m.rowsWritten, m.activeMs, m.accepted, m.refused, m.reqs, connections],770 });771 }772773 // Events and SQL Git share the physical SQLite meter, including indexes.774 eventBytes(): number {775 return this.store.databaseSize;776 }777778 // Files and archives remain in R2. SQL Git is counted above.779 mediaBytes(): number {780 return blobBytes(this.sql) + dumpBytes(this.sql) + importBytes(this.sql) + backupBytes(this);781 }782783 fuelStatus() {784 this.flushUsage();785 return this.fuel.status(now(), this.eventBytes(), this.mediaBytes());786 }787788 // ensureAlarm keeps a daily tick scheduled for storage charging and usage789 // flushes, without pre-empting a sooner NIP-40 expiry.790 async ensureAlarm(latest = now() + 86400) {791 const lease = this.settings.policy.lease;792 if (this.settings.isLeased() && lease && lease.until < latest) latest = lease.until;793 const cur = await this.ctx.storage.getAlarm();794 if (cur === null || cur > latest * 1000) await this.ctx.storage.setAlarm(latest * 1000);795 }796797 async fetch(req: Request): Promise<Response> {798 const url = new URL(req.url);799 const completed = (response: Response) => featureOn(this.settings.policy, "grasp") ? graspCORS(response, req.method === "OPTIONS") : response;800 const name = req.headers.get("x-relay-name");801 const initializes = Boolean(name && name !== this.slug) || Boolean(this.settings.policy.owner && !this.identity.pubkey);802 this.touch();803 if (initializes) {804 const initialized = await this.repositoryAccess.run("control", async () => {805 if (name && name !== this.slug) {806 this.slug = name;807 await this.ctx.storage.put("slug", name);808 }809 // Relays claimed before identities existed get theirs on first contact.810 if (this.settings.policy.owner && !this.identity.pubkey) await this.publishMembership();811 return true;812 }, () => false);813 if (!initialized) return completed(new Response(JSON.stringify({ error: "restricted: relay operation in progress; retry" }), { status: 429, headers: { "content-type": "application/json", "access-control-allow-origin": "*", "retry-after": "1" } }));814 }815 // The local custom-domain record wins over a cached edge mapping, so816 // changing a relay domain into a site never exposes the old relay door.817 const domainSite = this.settings.policy.customHosts?.find((h) => h.host === url.hostname)?.site;818 const site = domainSite || req.headers.get("x-relay-site");819 // A blocked address gets no socket and no gated door (routes.ts).820 const upgrade = req.headers.get("upgrade")?.toLowerCase() === "websocket";821 if ((site || upgrade || isGated(url, req)) && this.settings.isIPBlocked(clientIP(req))) {822 const msg = "blocked: this address is blocked from this relay";823 return completed(new Response(JSON.stringify({ error: msg }), { status: 403, headers: { "content-type": "application/json", "x-reason": msg, "access-control-allow-origin": "*" } }));824 }825 if (site) return serveSite(this, req, site);826 if (upgrade) return this.acceptWebSocket(req);827 if (isGitPath(url)) return route(this, req, url);828 // Management parses and verifies the request before acquiring authority.829 if (isManagementRequest(req)) return completed(await route(this, req, url));830 if (req.method !== "GET" && req.method !== "HEAD" && req.method !== "OPTIONS") {831 return completed(await this.repositoryAccess.run("control", () => route(this, req, url), () => new Response(JSON.stringify({ error: "restricted: relay operation in progress; retry" }), { status: 429, headers: { "content-type": "application/json", "access-control-allow-origin": "*", "retry-after": "1" } })));832 }833 return completed(await route(this, req, url));834 }835836 connections(): number {837 return this.ctx.getWebSockets().length;838 }839840 // relayURL is how this relay is addressed for a given Host header.841 relayURL(host: string): string {842 // Everything but local development is behind TLS, custom domains included.843 const h = host.replace(/:\d+$/, "");844 const local = h === "localhost" || h.endsWith(".localhost") || h === "127.0.0.1" || h === "[::1]";845 return (local ? "ws://" : "wss://") + host;846 }847848 // webURL is the same address for a browser.849 webURL(host: string): string {850 return this.relayURL(host).replace(/^ws/, "http");851 }852853 // maxMessage is the socket message cap in bytes: the owner's rule, under the platform's ceiling.854 maxMessage(): number {855 return Math.min(this.settings.policy.maxMessageKB * 1024, MAX_MESSAGE);856 }857858 // ---- websockets ----859860 private acceptWebSocket(req: Request): Response {861 const pair = new WebSocketPair();862 const [client, server] = [pair[0], pair[1]];863 const state: ConnState = { challenge: bytesToHex(crypto.getRandomValues(new Uint8Array(16))), host: new URL(req.url).host, authed: [], subs: {}, ip: clientIP(req) };864 server.serializeAttachment(state);865 this.ctx.acceptWebSocket(server);866 this.states.set(server, state);867 // NIP-42: offer a challenge up front.868 server.send(JSON.stringify(["AUTH", state.challenge]));869 return new Response(null, { status: 101, webSocket: client });870 }871872 private state(ws: WebSocket): ConnState {873 let s = this.states.get(ws);874 if (!s) {875 s = (ws.deserializeAttachment() as ConnState) ?? { challenge: "", host: "", authed: [], subs: {}, ip: "unknown" };876 if (!s.ip) s.ip = "unknown"; // sockets from before addresses were kept877 this.states.set(ws, s);878 }879 return s;880 }881882 // persist writes the connection state back to the socket so it survives883 // hibernation. Returns false if it no longer fits the attachment limit.884 private persist(ws: WebSocket, s: ConnState): boolean {885 try {886 ws.serializeAttachment(s);887 return true;888 } catch {889 return false;890 }891 }892893 private send(ws: WebSocket, ...msg: unknown[]) {894 this.raw(ws, JSON.stringify(msg));895 }896897 private raw(ws: WebSocket, text: string) {898 this.meter.bytesOut += text.length;899 try {900 ws.send(text);901 } catch {902 // closed; the close handler cleans up903 }904 }905906 async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {907 this.touch();908 if (typeof message !== "string") {909 this.send(ws, "NOTICE", "error: binary frames are not supported");910 return;911 }912 this.meter.bytesIn += message.length;913 if (message.length > this.maxMessage()) {914 this.send(ws, "NOTICE", "error: message too large");915 return;916 }917 let arr: unknown[];918 try {919 arr = JSON.parse(message);920 if (!Array.isArray(arr) || arr.length < 1) throw new Error();921 } catch {922 this.send(ws, "NOTICE", "error: could not parse message");923 return;924 }925 const s = this.state(ws);926 try {927 await this.dispatch(ws, s, arr);928 } finally {929 this.tally();930 }931 }932933 private bucket(ws: WebSocket) {934 let b = this.buckets.get(ws);935 if (!b) {936 b = { events: new Bucket(this.settings.policy.eventsPerMinute), reqs: new Bucket(this.settings.policy.reqsPerMinute) };937 this.buckets.set(ws, b);938 }939 return b;940 }941942 private async dispatch(ws: WebSocket, s: ConnState, arr: unknown[]) {943 const typ = arr[0];944 const p = this.settings.policy;945 switch (typ) {946 case "EVENT": {947 if (arr.length < 2) return this.send(ws, "NOTICE", "error: EVENT needs an event");948 const limited = this.bucket(ws).events.take(p.eventsPerMinute) || this.ipLimit(s.ip, "events");949 if (limited) {950 const id = (arr[1] as { id?: unknown })?.id;951 return this.send(ws, "OK", typeof id === "string" ? id : "", false, limited);952 }953 return this.handleEvent(ws, s, arr[1]);954 }955 case "REQ":956 case "COUNT": {957 if (arr.length < 3) return this.send(ws, "NOTICE", `error: ${typ} needs an id and at least one filter`);958 const limited = this.bucket(ws).reqs.take(p.reqsPerMinute) || this.ipLimit(s.ip, "reqs");959 if (limited) return this.send(ws, "CLOSED", typeof arr[1] === "string" ? arr[1] : "", limited);960 this.meter.reqs++;961 return this.handleReq(ws, s, typ, arr[1], arr.slice(2));962 }963 case "CLOSE":964 if (typeof arr[1] === "string" && s.subs[arr[1]]) {965 delete s.subs[arr[1]];966 this.persist(ws, s);967 }968 return;969 case "AUTH":970 if (arr.length < 2) return this.send(ws, "NOTICE", "error: AUTH needs an event");971 return this.handleAuth(ws, s, arr[1]);972 case "NEG-OPEN":973 case "NEG-MSG":974 case "NEG-CLOSE":975 if (arr.length < 2) return this.send(ws, "NOTICE", `error: ${typ} needs a subscription id`);976 return this.handleSync(ws, s, typ, arr[1], arr.slice(2));977 default:978 this.send(ws, "NOTICE", "error: unknown message type " + String(typ));979 }980 }981982 async webSocketClose(ws: WebSocket) {983 this.states.delete(ws);984 this.syncs.delete(ws);985 this.buckets.delete(ws);986 this.flushUsage();987 notePresence(this);988 }989990 async webSocketError(ws: WebSocket) {991 this.states.delete(ws);992 this.syncs.delete(ws);993 this.buckets.delete(ws);994 }995996 // ---- EVENT ----997998 private async handleEvent(ws: WebSocket, s: ConnState, raw: unknown) {999 const reason = validate(raw);1000 const e = raw as Event;1001 if (reason) {1002 this.send(ws, "OK", typeof e?.id === "string" ? e.id : "", false, reason);1003 return;1004 }1005 const r = await this.acceptAny(e, s);1006 if (r.ok) this.meter.accepted++;1007 else this.meter.refused++;1008 this.send(ws, "OK", e.id, r.ok, r.msg);1009 if (r.stored) this.broadcast(e);1010 }10111012 // acceptAny routes the special kinds (zap receipts, NIP-56 reports) and1013 // falls back to the ordinary write path. Callers broadcast when stored.1014 async acceptAny(e: Event, conn: ConnState): Promise<{ ok: boolean; msg: string; stored: boolean }> {1015 if (e.kind === 9735) {1016 const receipt = await this.repositoryAccess.runFuel(1017 () => acceptReceipt(this, e, conn.host),1018 () => null,1019 );1020 if (receipt) return receipt;1021 }1022 return this.repositoryAccess.run("event", async () => {1023 if (e.kind === KIND_REPORT) return this.acceptReport(e, conn.host);1024 if (isGroupManagement(e.kind) && (hasTag(e, "h") || isNIP43Request(e.kind))) return this.acceptGroup(e, conn);1025 return this.accept(e, conn);1026 }, () => ({ ok: false, msg: "restricted: relay operation in progress; retry the event", stored: false }));1027 }10281029 // acceptGroup handles NIP-29 joins, leaves and moderation: the common gate,1030 // then role checks instead of the write policy, then stored like any event.1031 private async acceptGroup(e: Event, conn: ConnState): Promise<{ ok: boolean; msg: string; stored: boolean }> {1032 const t = now();1033 const reason = writeGate(this, e, conn, t);1034 if (reason) return { ok: false, msg: reason, stored: false };1035 if (this.sql.exec(`SELECT 1 FROM events WHERE id=?`, e.id).toArray().length) return { ok: true, msg: ERR_DUPLICATE, stored: false };1036 const r = await handleGroupEvent(this, e);1037 if (!r.ok || !r.stored) return r;1038 const err = this.store.save(e, t);1039 if (err) return { ok: false, msg: err, stored: false };1040 return { ok: true, msg: "", stored: true };1041 }10421043 // acceptReport files a NIP-56 report in the moderation queue. It is never1044 // stored as an event or served: reports are for the owner, not the feed.1045 private acceptReport(e: Event, host = ""): { ok: boolean; msg: string; stored: boolean } {1046 if (this.settings.policy.owner === "") return { ok: false, msg: this.settings.isLeased() ? "restricted: this temporary relay has no owner to report to" : "restricted: this relay is unclaimed", stored: false };1047 if (this.settings.isBanned(e.pubkey)) return { ok: false, msg: "blocked: this pubkey is banned from this relay", stored: false };1048 const p = tagValues(e, "p")[0] ?? "";1049 const et = e.tags.find((t) => t[0] === "e");1050 const pt = e.tags.find((t) => t[0] === "p");1051 const type = et?.[2] ?? pt?.[2] ?? "";1052 if (!/^[0-9a-f]{64}$/.test(p)) return { ok: false, msg: "invalid: report needs a p tag", stored: false };1053 this.sql.exec(1054 `INSERT OR IGNORE INTO reports(id,reporter,target_pubkey,target_event,type,content,at) VALUES(?,?,?,?,?,?,?)`,1055 e.id, e.pubkey, p, et?.[1] ?? "", type.slice(0, 32), e.content.slice(0, 2000), e.created_at,1056 );1057 // Enough distinct reporters hide an event until a moderator looks.1058 const th = this.settings.policy.reportThreshold;1059 const target = et?.[1] ?? "";1060 if (th > 0 && /^[0-9a-f]{64}$/.test(target) && !this.settings.isEventHidden(target) && !this.settings.isEventBanned(target)) {1061 const n = this.sql.exec<{ n: number }>(`SELECT count(DISTINCT reporter) AS n FROM reports WHERE target_event=? AND status='open'`, target).one().n;1062 if (n >= th && this.sql.exec(`SELECT 1 FROM events WHERE id=?`, target).toArray().length) this.settings.setEvent(target, "hide", `${n} reports`, e.created_at);1063 }1064 const where = host ? `https://${host}/#people` : "the People tab";1065 void notify(this, "reports", `New report on ${this.slug}: ${type || "report"} about ${p.slice(0, 8)}${et ? " (event " + et[1].slice(0, 8) + ")" : ""} by ${e.pubkey.slice(0, 8)}.${e.content ? " " + e.content.slice(0, 300) : ""} Review it at ${where}`, "a report on " + this.slug);1066 return { ok: true, msg: "info: report received", stored: false };1067 }10681069 // accept runs the write-side rules and stores the event. conn is null for1070 // host-side publishes (the dashboard), which skip client policy.1071 accept(e: Event, conn: ConnState | null): { ok: boolean; msg: string; stored: boolean } {1072 if (!this.repositoryAccess.owned) return this.repositoryAccess.sync("event", () => this.accept(e, conn), () => ({ ok: false, msg: "restricted: relay operation in progress; retry the event", stored: false }));1073 const p = this.settings.policy;1074 const t = now();1075 const no = (msg: string) => ({ ok: false, msg, stored: false });1076 const gate = writeGate(this, e, conn, t);1077 if (gate) return no(gate);1078 if (e.kind === KIND_NOSTR_CONNECT && featureOn(p, "signer")) return { ok: true, msg: "", stored: true }; // see gate1079 const exp = expiration(e);1080 if (conn) {1081 // The owner's own profile and lists always land, whatever the kind1082 // rules say: a relay that refuses its owner's relay list cannot be1083 // wired into it.1084 if (!this.settings.kindAllowed(e.kind) && !(isReplaceable(e.kind) && this.settings.isOwner(e.pubkey))) return no("blocked: this relay does not accept kind " + e.kind);1085 const keep = this.settings.retentionDays(e.kind);1086 if (keep > 0 && e.created_at < t - keep * 86400) return no(`blocked: this relay keeps kind ${e.kind} for ${keep} days and this event is older`);1087 // The owner's per-member rules: a keep-for window on top of the kind1088 // rules, and a cap on stored bytes. Neither touches the owner.1089 const principal = e.kind === KIND_MARMOT_GROUP ? marmotPrincipal(this, e, conn) : e.pubkey;1090 const lim = this.settings.limitsOf(principal);1091 if (lim) {1092 if (lim.keep > 0 && !isProtected(e.kind) && !isReplaceable(e.kind) && e.created_at < t - lim.keep * 86400) return no(`blocked: this relay keeps your events for ${lim.keep} days and this event is older`);1093 if (lim.cap > 0 && this.store.authorBytes(principal) + canonical(e).length > lim.cap) return no(`restricted: you have reached your storage cap of ${Math.max(1, Math.round(lim.cap / 1024))} KB on this relay`);1094 }1095 const why = this.settings.mayWrite(principal);1096 if (why && !guestPass(this, e)) return no(why);1097 if (p.minPow > 0) {1098 const d = difficulty(e);1099 if (d.difficulty < p.minPow) return no(`pow: difficulty ${d.difficulty} is less than ${p.minPow}`);1100 if (d.target > 0 && d.target < p.minPow) return no(`pow: committed target ${d.target} is less than ${p.minPow}`);1101 }1102 }1103 if (e.kind === KIND_VANISH) {1104 const host = conn?.host ?? "";1105 const targets = tagValues(e, "relay").some((v) => v === "ALL_RELAYS" || (host !== "" && hostOf(v) === hostOf("ws://" + host)));1106 if (!targets) return no("blocked: this relay is not named in the vanish request");1107 this.store.vanish(e.pubkey, e.created_at);1108 return { ok: true, msg: "", stored: false };1109 }1110 const err = this.store.save(e, t);1111 if (err === ERR_DUPLICATE) return { ok: true, msg: err, stored: false };1112 if (err) return no(err);1113 const held = holdGrasp(this, e, conn?.host ?? "");1114 if (held) void this.ensureAlarm(now() + 1);1115 if (e.kind === 3) this.settings.noteContacts(e.pubkey);1116 if (e.kind === KIND_MARMOT_GROUP && conn) {1117 const principal = marmotPrincipal(this, e, conn);1118 if (principal && principal !== e.pubkey) this.store.notePrincipal(e.id, principal);1119 }1120 this.store.noteSaved(e.pubkey, canonical(e).length, isReplaceable(e.kind));1121 if (exp > 0) this.scheduleSweep(exp);1122 else this.ensureAlarm();1123 if (e.kind === 0 && conn) claimFromProfile(this.settings, e.content, e.pubkey, conn.host, t);1124 if (conn) void this.succession.seen(e.pubkey);1125 if (conn) notePresence(this, e.pubkey);1126 if (e.kind === 30023) markView(this, "articles");1127 return { ok: true, msg: held ? "purgatory: won't be served until git data arrives" : "", stored: !held };1128 }11291130 private scheduleSweep(at: number) {1131 this.ctx.storage.getAlarm().then((cur) => {1132 if (cur === null || cur > at * 1000) return this.ctx.storage.setAlarm(at * 1000 + 500);1133 });1134 }11351136 async alarm() {1137 await pushTick(this);1138 return this.repositoryAccess.run("alarm", async () => {1139 this.touch();1140 const t = now();1141 // A lease that has run out is wiped whole: the name is free again.1142 if (this.settings.leaseExpired(t)) {1143 await this.teardown();1144 return;1145 }1146 const deliveryAt = await deliveryTick(this);1147 const graspAt = await graspTick(this);1148 await this.syncSites();1149 await queueMirrors(this);1150 if (await this.jobsTick()) {1151 await this.ctx.storage.setAlarm(Date.now() + 250);1152 return;1153 }1154 this.flushUsage();1155 this.fuel.chargeStorage(t, this.eventBytes(), this.mediaBytes());1156 this.store.drain();1157 this.sweepRetention(t);1158 await viewsTick(this, t);1159 await this.publishDiscovery();1160 // Fuel notice: once when it turns low, then once a day while it stays low.1161 if (this.settings.policy.notify.fuel) {1162 const low = fuelLow(this.fuelStatus());1163 const last = (await this.ctx.storage.get<number>("fuel-low-at")) ?? 0;1164 if (low && t - last >= 86400) {1165 await this.ctx.storage.put("fuel-low-at", t);1166 await notify(this, "fuel", fuelText(this, this.fuelStatus()), "fuel on " + this.slug);1167 } else if (!low && last) await this.ctx.storage.delete("fuel-low-at");1168 }1169 // Succession: the dead-man's switch, checked once a day.1170 await this.succession.tick(t);1171 // ---- digest: one message a week on how the relay is doing ----1172 if (this.settings.policy.notify.digest && this.settings.policy.owner !== "") {1173 const last = (await this.ctx.storage.get<number>("lastDigest")) ?? 0;1174 if (last === 0) await this.ctx.storage.put("lastDigest", t);1175 else if (t - last >= DIGEST_DAYS * 86400 - 3600) {1176 await this.ctx.storage.put("lastDigest", t);1177 await notify(this, "digest", await digestText(this, last, t), "the week on " + this.slug);1178 }1179 }1180 // ---- end digest ----1181 const next = this.store.sweepExpired(t);1182 if (await this.syncSites()) {1183 await this.ctx.storage.setAlarm(Date.now() + 250);1184 return;1185 }1186 // ---- dumps: a scheduled JSONL of everything, into R2 (dumps.ts) ----1187 if (dumpDue(this, t)) {1188 try {1189 await writeDump(this, t);1190 } catch (err) {1191 console.log("dump failed: " + (err instanceof Error ? err.message : String(err)));1192 }1193 }1194 // ---- end dumps ----1195 let at = t + 86400;1196 if (this.sql.exec(`SELECT 1 FROM site_mirror_queue LIMIT 1`).toArray().length) at = Math.min(at, t + 60);1197 if (next > 0 && next < at) at = next;1198 const lease = this.settings.policy.lease;1199 if (this.settings.isLeased() && lease && lease.until < at) at = lease.until;1200 const run = await this.nextJobRun();1201 if (run > 0 && run < at) at = Math.max(run, t + 1);1202 if (graspAt && graspAt < at) at = graspAt;1203 const pushAt = nextPush(this);1204 if (pushAt) at = Math.min(at, Math.max(pushAt, now() + 1));1205 if (deliveryAt) at = Math.min(at, Math.max(deliveryAt, now() + 1));1206 await this.ctx.storage.setAlarm(at * 1000 + 500);1207 }, async () => { await this.ctx.storage.setAlarm(Date.now() + 1000); });1208 }12091210 // sweepRetention applies the owner's keep-for rules. Returns how many1211 // events were shed.1212 sweepRetention(t: number): number {1213 const rules = this.settings.listRetention();1214 let gone = 0;1215 const own = rules.filter((r) => r.kind !== null).map((r) => r.kind as number);1216 for (const r of rules) {1217 if (r.kind !== null) gone += this.store.purge(r.kind, t - r.days * 86400, [], this.identity.pubkey);1218 else gone += this.store.purge(null, t - r.days * 86400, [...own, ...this.store.kindStats().map((k) => k.kind).filter((k) => isReplaceable(k) || isProtected(k) || k === 1059)], this.identity.pubkey);1219 }1220 // Per-member keep-for rules, same pass, same exceptions as the catch-all.1221 const limited = this.settings.limited();1222 if (limited.length) {1223 const keepKinds = this.store.kindStats().map((k) => k.kind).filter((k) => isReplaceable(k) || isProtected(k));1224 for (const m of limited) gone += this.store.purgeAuthor(m.pubkey, t - m.keep * 86400, keepKinds);1225 }1226 return gone;1227 }12281229 // canSee decides whether a private kind reaches a socket through the1230 // filter that matched it. A socket that has AUTHed as the sender or a1231 // recipient always sees it. NIP-46 traffic is also delivered to a filter1232 // that names a party's key outright, in "#p" or "authors", under any read1233 // rule and without AUTH: the payload is encrypted end to end, and signers1234 // and their clients hold no key this relay knows. Amber probes a relay1235 // this way before it will list it, and Amethyst's bunker listens under a1236 // transport key it never AUTHs with. A bare subscription to the kind1237 // still sees nothing.1238 private canSee(s: ConnState, e: Event, f: Filter): boolean {1239 if (e.kind === KIND_PUSH_REGISTRATION) return s.authed.includes(e.pubkey);1240 if (!isPrivate(e.kind)) return true;1241 const parties = [e.pubkey, ...tagValues(e, "p")];1242 if (parties.some((p) => s.authed.includes(p))) return true;1243 if (e.kind !== KIND_NOSTR_CONNECT) return false;1244 return [...(f.authors ?? []), ...(f.tags.p ?? [])].some((k) => parties.includes(k));1245 }12461247 broadcast(e: Event, route = true) {1248 if (!graspVisible(this, e.id)) return;1249 if (route && queueDelivery(this, e)) this.ctx.waitUntil(this.ensureAlarm(now() + 1));1250 if (queuePush(this, e)) this.ctx.waitUntil(this.ensureAlarm(now() + 1));1251 const raw = canonical(e);1252 for (const ws of this.ctx.getWebSockets()) {1253 const s = this.state(ws);1254 for (const [id, filters] of Object.entries(s.subs)) {1255 if (!readGate(this, s, filters).reason && filters.some((f) => match(f, e) && this.canSee(s, e, f))) {1256 this.raw(ws, `["EVENT",${JSON.stringify(id)},${raw}]`);1257 break;1258 }1259 }1260 }1261 }12621263 // ---- REQ / COUNT ----12641265 private parseFilters(rawFilters: unknown[]): Filter[] | string {1266 const filters: Filter[] = [];1267 for (const rf of rawFilters) {1268 const f = parseFilter(rf);1269 if (typeof f === "string") return "invalid: bad filter: " + f;1270 filters.push(f);1271 }1272 return filters;1273 }12741275 private handleReq(ws: WebSocket, s: ConnState, verb: string, rawID: unknown, rawFilters: unknown[]) {1276 if (typeof rawID !== "string" || rawID === "" || rawID.length > 64) return this.send(ws, "NOTICE", "error: bad subscription id");1277 const id = rawID;1278 const filters = this.parseFilters(rawFilters);1279 if (typeof filters === "string") return this.send(ws, "CLOSED", id, filters);1280 const { reason, authHint } = readGate(this, s, filters);1281 if (reason) return this.send(ws, "CLOSED", id, reason);1282 const who: Access = { pubkeys: s.authed };1283 const t = now();1284 const p = this.settings.policy;12851286 if (verb === "COUNT") {1287 if (!featureOn(p, "count")) return this.send(ws, "CLOSED", id, "unsupported: COUNT is switched off on this relay");1288 const result: Record<string, unknown> = {};1289 if (filters.length === 1 && hllOffset(filters[0]) >= 0) {1290 const r = this.store.countHLL(filters[0], who, hllOffset(filters[0]), t);1291 result.count = r.count;1292 result.hll = r.hll;1293 } else result.count = this.store.count(filters, who, t);1294 return this.send(ws, "COUNT", id, result);1295 }12961297 if (!s.subs[id] && Object.keys(s.subs).length >= p.maxSubs) return this.send(ws, "CLOSED", id, "error: too many subscriptions");1298 s.subs[id] = filters;1299 if (!this.persist(ws, s)) {1300 delete s.subs[id];1301 this.persist(ws, s);1302 return this.send(ws, "CLOSED", id, "error: subscription filters are too large");1303 }1304 const seen = new Set<string>();1305 let complete = true;1306 for (const f of filters) {1307 const r = this.store.query(f, who, p.maxLimit, t);1308 complete = complete && !r.more;1309 for (const raw of r.rows) {1310 const probe = JSON.parse(raw) as { id: string };1311 if (seen.has(probe.id)) continue;1312 seen.add(probe.id);1313 this.raw(ws, `["EVENT",${JSON.stringify(id)},${raw}]`);1314 }1315 }1316 const hints = [complete ? "finish" : "more"];1317 if (authHint) hints.push("auth");1318 this.send(ws, "EOSE", id, hints);1319 }13201321 // ---- AUTH (NIP-42) ----13221323 private handleAuth(ws: WebSocket, s: ConnState, raw: unknown) {1324 const reason = validate(raw);1325 const e = raw as Event;1326 if (reason) return this.send(ws, "OK", typeof e?.id === "string" ? e.id : "", false, reason);1327 const t = now();1328 const fail = (msg: string) => this.send(ws, "OK", e.id, false, msg);1329 if (e.kind !== KIND_AUTH) return fail("invalid: auth event must be kind 22242");1330 if (Math.abs(t - e.created_at) > 600) return fail("invalid: auth event created_at is too far from the current time");1331 if (tagValues(e, "challenge")[0] !== s.challenge) return fail("invalid: auth challenge does not match");1332 if (hostOf(tagValues(e, "relay")[0] ?? "") !== hostOf("ws://" + s.host)) return fail("invalid: auth relay tag does not name this relay");1333 if (!s.authed.includes(e.pubkey)) s.authed.push(e.pubkey);1334 void this.succession.seen(e.pubkey);1335 notePresence(this);1336 this.persist(ws, s);1337 this.send(ws, "OK", e.id, true, "");1338 }13391340 // ---- NIP-77 ----13411342 private handleSync(ws: WebSocket, s: ConnState, verb: string, rawID: unknown, rest: unknown[]) {1343 if (typeof rawID !== "string" || rawID === "" || rawID.length > 64) return this.send(ws, "NOTICE", "error: bad subscription id");1344 const id = rawID;1345 let sessions = this.syncs.get(ws);1346 if (!sessions) {1347 sessions = new Map();1348 this.syncs.set(ws, sessions);1349 }1350 let sess = sessions.get(id);1351 let msgHex = "";1352 // The responder's side of a sync is the sorted item list and nothing1353 // else, and the list comes from the filter. So the filter rides on the1354 // socket's attachment with the subscriptions, and a session whose1355 // Negentropy object went with the object's memory is rebuilt from it1356 // when the next message arrives, rather than ended with closed:.1357 const remember = (f: Filter | null) => {1358 const next = { ...(s.syncs ?? {}) };1359 if (f) next[id] = f;1360 else delete next[id];1361 if (Object.keys(next).length) s.syncs = next;1362 else delete s.syncs;1363 // Past the attachment limit the filter stays in memory only, and the1364 // session ends when the object sleeps, as every session once did.1365 if (!this.persist(ws, s) && f) {1366 delete s.syncs![id];1367 if (Object.keys(s.syncs!).length === 0) delete s.syncs;1368 }1369 };1370 if (verb === "NEG-CLOSE") {1371 sessions.delete(id);1372 if (s.syncs?.[id]) remember(null);1373 return;1374 }1375 if (verb === "NEG-OPEN") {1376 if (!featureOn(this.settings.policy, "sync")) return this.send(ws, "NEG-ERR", id, "unsupported: sync is switched off on this relay");1377 if (rest.length < 2) return this.send(ws, "NEG-ERR", id, "error: NEG-OPEN needs a filter and a message");1378 const f = parseFilter(rest[0]);1379 if (typeof f === "string") return this.send(ws, "NEG-ERR", id, "invalid: bad filter: " + f);1380 const { reason } = readGate(this, s, [f]);1381 if (reason) return this.send(ws, "NEG-ERR", id, reason);1382 const open = new Set([...sessions.keys(), ...Object.keys(s.syncs ?? {})]);1383 if (!open.has(id) && open.size >= this.settings.policy.maxSubs) return this.send(ws, "NEG-ERR", id, "blocked: too many open syncs");1384 const items = this.store.syncItems(f, { pubkeys: s.authed }, MAX_SYNC, now());1385 if (items === "too big") return this.send(ws, "NEG-ERR", id, ERR_TOO_BIG, MAX_SYNC);1386 sess = new Negentropy(items, sha256);1387 sessions.set(id, sess);1388 remember(f);1389 msgHex = typeof rest[1] === "string" ? rest[1] : "";1390 } else {1391 const f = s.syncs?.[id];1392 if (!sess && f) {1393 // A wake since NEG-OPEN. The read rule is asked again with what the1394 // socket has proved by now, and the items are read afresh, so the1395 // session carries on from the store as it is.1396 const { reason } = readGate(this, s, [f]);1397 if (reason) {1398 remember(null);1399 return this.send(ws, "NEG-ERR", id, reason);1400 }1401 const items = this.store.syncItems(f, { pubkeys: s.authed }, MAX_SYNC, now());1402 if (items === "too big") {1403 remember(null);1404 return this.send(ws, "NEG-ERR", id, ERR_TOO_BIG, MAX_SYNC);1405 }1406 sess = new Negentropy(items, sha256);1407 sessions.set(id, sess);1408 }1409 if (!sess) return this.send(ws, "NEG-ERR", id, "closed: no such sync, send NEG-OPEN first");1410 msgHex = typeof rest[0] === "string" ? rest[0] : "";1411 }1412 if (!/^([0-9a-f]{2})*$/i.test(msgHex)) return this.send(ws, "NEG-ERR", id, "invalid: message is not hex");1413 try {1414 const { reply } = sess.reconcile(hexToBytes(msgHex));1415 this.send(ws, "NEG-MSG", id, bytesToHex(reply ?? new Uint8Array()));1416 } catch (err) {1417 sessions.delete(id);1418 if (s.syncs?.[id]) remember(null);1419 this.send(ws, "NEG-ERR", id, "invalid: " + (err instanceof Error ? err.message : String(err)));1420 }1421 }1422}