1// Pull: bounded history import uses NIP-77 when available and ordinary2// NIP-01 queries otherwise. Query windows survive between alarm rounds;3// their completion is best effort, since a relay may silently cap results.4// The alarm calls rounds until nothing is missing (see jobs.ts), so a pull5// survives the object sleeping, and running it again only fetches what is6// new. An optional filter narrows the reconciliation: authors for a7// backfill of the owner's own history, kinds, since.8import { sha256 } from "@noble/hashes/sha2.js";9import { validate, now, type Event } from "./event.ts";10import type { Filter } from "./filter.ts";11import { Negentropy, bytesToHex, hexToBytes } from "./negentropy.ts";12import { ERR_DUPLICATE } from "./store.ts";13import { validName } from "./names.ts";14import type { Blob } from "./blossom.ts";15import type { Relay } from "./relay.ts";1617export interface PullFilter {18 authors?: string[];19 kinds?: number[];20 since?: number;21}2223export interface PullJob {24 url: string;25 startedAt: number;26 rounds: number;27 stored: number;28 skipped: number;29 blobs: number;30 failures: number;31 filter?: PullFilter;32 progress?: PullProgress;33}34export interface PullProgress {35 mode: "negentropy" | "query";36 status: "pending" | "running" | "complete" | "best-effort" | "partial" | "refused" | "failed";37 windows?: { since: number; until: number }[];38 pages: number;39 failures: number;40 error: string;41 warning: string;42 partial: boolean;43}44export interface PullSource extends PullProgress {45 url: string;46 stored: number;47 skipped: number;48 blobs: number;49}50export const newPullProgress = (): PullProgress => ({ mode: "negentropy", status: "pending", pages: 0, failures: 0, error: "", warning: "", partial: false });51export interface PullSocket {52 send(...msg: unknown[]): void;53 recv(timeout?: number): Promise<unknown[]>;54 close(): void;55}56export type PullConnect = (relay: Relay, url: string) => Promise<PullSocket>;57const connectPull: PullConnect = async (relay, url) => new Socket(await dial(relay, url));58export interface PullResult extends PullJob {59 finishedAt: number;60 error: string;61}6263const ROUND_EVENTS = 500;64const ROUND_BLOBS = 20;65const IDS_PER_REQ = 100;66const MAX_ITEMS = 100_000;67const MESSAGE_TIMEOUT_MS = 20_000;68export const QUERY_PAGE = 500;69const QUERY_PAGES = 2048;70const ROUND_MESSAGES = 4096;71const ROUND_MS = 25_000;7273// localName is the relay name when the URL is one of ours, else "".74export function localName(url: URL, domain: string): string {75 const host = url.hostname.toLowerCase();76 const d = domain.toLowerCase();77 let name = "";78 if (host.endsWith("." + d)) name = host.slice(0, -(d.length + 1));79 else if (host.endsWith(".localhost")) name = host.slice(0, -".localhost".length);80 return validName(name) && !name.includes(".") ? name : "";81}8283// checkPullURL returns "" or a reason the URL cannot be pulled from.84export function checkPullURL(raw: string, self: string, domain: string): string {85 let u: URL;86 try {87 u = new URL(raw);88 } catch {89 return "invalid: not a URL";90 }91 if (u.protocol !== "wss:" && u.protocol !== "ws:") return "invalid: the URL must start with wss:// or ws://";92 if (localName(u, domain) === self) return "invalid: a relay cannot pull from itself";93 return "";94}9596// dial opens a websocket to a relay. Our own relays are reached through97// their object directly, which also works in wrangler dev and tests.98export async function dial(relay: Relay, raw: string): Promise<WebSocket> {99 const u = new URL(raw);100 const local = localName(u, relay.domain);101 const headers: Record<string, string> = { upgrade: "websocket" };102 let resp: Response;103 if (local) resp = await relay.relays.getByName(local).fetch("https://" + u.host + "/", { headers: { ...headers, "x-relay-name": local } });104 else resp = await fetch(u.href.replace(/^ws/, "http"), { headers });105 const ws = resp.webSocket;106 if (!ws) throw new Error(`relay did not accept a websocket (${resp.status})`);107 ws.accept();108 return ws;109}110111// Socket turns a websocket into a queue of parsed messages.112export class Socket {113 private queue: { message: unknown[]; size: number }[] = [];114 private waiters: { res: (m: unknown[]) => void; rej: (e: Error) => void }[] = [];115 private closed: Error | null = null;116 private queuedBytes = 0;117118 constructor(private ws: WebSocket) {119 ws.addEventListener("message", (e) => {120 const raw = typeof e.data === "string" ? e.data : "";121 if (raw.length > 4 * 1024 * 1024 || this.queue.length >= 2048 || this.queuedBytes + raw.length > 16 * 1024 * 1024) {122 end("relay response exceeded the import buffer limit");123 this.close();124 return;125 }126 let m: unknown;127 try {128 m = JSON.parse(raw);129 } catch {130 return;131 }132 if (!Array.isArray(m)) return;133 const w = this.waiters.shift();134 if (w) w.res(m);135 else { this.queue.push({ message: m, size: raw.length }); this.queuedBytes += raw.length; }136 });137 const end = (why: string) => {138 this.closed = new Error(why);139 for (const w of this.waiters.splice(0)) w.rej(this.closed);140 };141 ws.addEventListener("close", (e) => end("connection closed" + (e.reason ? ": " + e.reason : "")));142 ws.addEventListener("error", () => end("connection failed"));143 }144145 send(...msg: unknown[]) {146 this.ws.send(JSON.stringify(msg));147 }148149 recv(timeout = MESSAGE_TIMEOUT_MS): Promise<unknown[]> {150 const m = this.queue.shift();151 if (m) { this.queuedBytes -= m.size; return Promise.resolve(m.message); }152 if (this.closed) return Promise.reject(this.closed);153 return new Promise((res, rej) => {154 const timer = setTimeout(() => {155 this.waiters = this.waiters.filter((w) => w.res !== ok);156 rej(new Error("the relay stopped answering"));157 }, Math.max(1, Math.min(MESSAGE_TIMEOUT_MS, timeout)));158 const ok = (m: unknown[]) => {159 clearTimeout(timer);160 res(m);161 };162 this.waiters.push({ res: ok, rej: (e) => (clearTimeout(timer), rej(e)) });163 });164 }165166 close() {167 try {168 this.ws.close(1000, "done");169 } catch {170 /* already closed */171 }172 }173}174175// runPullRound does one round of the job. more: call again. error: the176// round failed and nothing more was learned.177export async function runPullRound(relay: Relay, job: PullJob, connect: PullConnect = connectPull): Promise<{ more: boolean; error: string }> {178 job.rounds++;179 const progress = job.progress ??= newPullProgress();180 progress.status = "running";181 let sock: PullSocket | null = null;182 try {183 sock = await connect(relay, job.url);184 if (progress.mode === "query") return await queryRound(relay, sock, job);185 if (progress.pages >= QUERY_PAGES) {186 progress.status = "partial";187 progress.warning = "Import stopped at its round budget; narrow the filter and run again.";188 return { more: false, error: "" };189 }190 let need: string[];191 try {192 need = await reconcile(relay, sock, job.filter ?? {});193 } catch (err) {194 progress.mode = "query";195 progress.warning = "NIP-77 unavailable: " + reason(err);196 progress.windows = [{ since: job.filter?.since ?? 0, until: job.startedAt }];197 return { more: true, error: "" };198 }199 progress.pages++;200 const storedBefore = job.stored;201 const deadline = Date.now() + ROUND_MS;202 const want = need.slice(0, ROUND_EVENTS);203 for (let i = 0; i < want.length; i += IDS_PER_REQ) await fetchEvents(relay, sock, want.slice(i, i + IDS_PER_REQ), job, deadline);204 let more = need.length > want.length;205 if (want.length && job.stored === storedBefore) {206 progress.partial = true;207 progress.warning = "The source advertises missing events that it did not supply or this relay cannot admit.";208 more = false;209 }210 if (!more && !job.filter) {211 // Files come along on a whole-relay pull; a filtered pull is about events.212 const local = localName(new URL(job.url), relay.domain);213 if (local) more = await copyBlobs(relay, local, job);214 }215 if (!more) progress.status = progress.partial ? "partial" : "complete";216 return { more, error: "" };217 } catch (err) {218 const error = reason(err);219 progress.error = error;220 return { more: false, error };221 } finally {222 sock?.close();223 }224}225226const reason = (err: unknown) => (err instanceof Error ? err.message : String(err)).slice(0, 300);227228// queryRound stores one bounded time window. Full windows divide rather229// than skipping timestamps; a full single second remains explicitly partial.230async function queryRound(relay: Relay, sock: PullSocket, job: PullJob): Promise<{ more: boolean; error: string }> {231 const p = job.progress!;232 const windows = p.windows ??= [{ since: job.filter?.since ?? 0, until: job.startedAt }];233 if (!windows.length) return { more: false, error: "" };234 if (p.pages >= QUERY_PAGES) {235 p.status = "partial";236 p.warning = "Import stopped at the query page budget; narrow the time or author filter and run again.";237 return { more: false, error: "" };238 }239 const window = windows[windows.length - 1];240 const sub = "history";241 const wire = { ...job.filter, ...window, limit: QUERY_PAGE };242 sock.send("REQ", sub, wire);243 const ids = new Set<string>();244 let received = 0;245 let invalid = false;246 let auth = false;247 const deadline = Date.now() + ROUND_MS;248 for (let messages = 0; ; messages++) {249 if (messages >= ROUND_MESSAGES || Date.now() >= deadline) throw new Error(auth ? "auth-required: source needs authentication; no user key is shared for imports" : "source did not complete its query within the import limits");250 let m: unknown[];251 try { m = await sock.recv(deadline - Date.now()); }252 catch (err) { if (auth) throw new Error("auth-required: source needs authentication; no user key is shared for imports"); throw err; }253 if (m[0] === "AUTH") auth = true;254 if (m[0] === "EOSE" && m[1] === sub) break;255 if (m[0] === "CLOSED" && m[1] === sub) throw new Error("query refused: " + String(m[2]).slice(0, 200));256 if (m[0] !== "EVENT" || m[1] !== sub) continue;257 received++;258 if (received > QUERY_PAGE) throw new Error("source exceeded the requested event limit");259 const event = m[2];260 if (validate(event)) { job.skipped++; invalid = true; continue; }261 const e = event as Event;262 if (!matchesPull(e, job.filter) || e.created_at < window.since || e.created_at > window.until || ids.has(e.id)) {263 job.skipped++; invalid = true; continue;264 }265 ids.add(e.id);266 storePulled(relay, e, job);267 }268 sock.send("CLOSE", sub);269 windows.pop();270 p.pages++;271 if (invalid) { p.partial = true; p.warning = "Source returned invalid, duplicate or out-of-filter events; coverage is incomplete."; }272 if (received >= QUERY_PAGE) {273 if (window.since >= window.until) {274 p.partial = true;275 p.warning = "A one-second window reached the result limit; events sharing that timestamp may be missing.";276 } else {277 const middle = window.since + Math.floor((window.until - window.since) / 2);278 windows.push({ since: window.since, until: middle }, { since: middle + 1, until: window.until });279 }280 }281 if (!windows.length) {282 p.status = p.partial ? "partial" : "best-effort";283 if (!p.partial) p.warning = "Query scan finished; the source may silently cap or omit events, so complete history is not guaranteed.";284 }285 return { more: windows.length > 0, error: "" };286}287288// matchesPull checks source-supplied events against the requested filter.289function matchesPull(e: Event, f?: PullFilter): boolean {290 return !f || ((!f.authors?.length || f.authors.includes(e.pubkey)) && (!f.kinds?.length || f.kinds.includes(e.kind)) && (f.since === undefined || e.created_at >= f.since));291}292293// storePulled applies normal host admission and counts each accepted event.294function storePulled(relay: Relay, e: Event, job: PullJob) {295 if (!relay.settings.kindAllowed(e.kind)) { job.skipped++; markRejected(job); return; }296 const r = relay.accept(e, null);297 if (r.stored) { job.stored++; relay.broadcast(e, false); }298 else if (r.msg !== ERR_DUPLICATE) { job.skipped++; markRejected(job); }299}300301// markRejected preserves admission gaps in the source result.302function markRejected(job: PullJob) {303 if (!job.progress) return;304 job.progress.partial = true;305 job.progress.warning = "Some source events could not be admitted by this relay; coverage is incomplete.";306}307308// reconcile runs NIP-77 as the initiator over the filter and returns the309// ids the other side has that we do not.310async function reconcile(relay: Relay, sock: PullSocket, filter: PullFilter): Promise<string[]> {311 const f: Filter = { tags: {} };312 const wire: Record<string, unknown> = {};313 if (filter.authors?.length) f.authors = wire.authors = filter.authors;314 if (filter.kinds?.length) f.kinds = wire.kinds = filter.kinds;315 if (filter.since) f.since = wire.since = filter.since;316 const items = relay.store.syncItems(f, { pubkeys: [], all: true }, MAX_ITEMS, now());317 if (items === "too big") throw new Error("this relay holds too many events to sync");318 const neg = new Negentropy(items, sha256);319 const id = "pull";320 sock.send("NEG-OPEN", id, wire, bytesToHex(neg.initiate()));321 const need: string[] = [];322 const deadline = Date.now() + ROUND_MS;323 for (let messages = 0; ; messages++) {324 if (messages >= ROUND_MESSAGES || Date.now() >= deadline) throw new Error("sync exceeded its round budget");325 const m = await sock.recv(deadline - Date.now());326 if (m[0] === "NEG-ERR" && m[1] === id) throw new Error("sync refused: " + String(m[2]));327 if (m[0] !== "NEG-MSG" || m[1] !== id) continue; // AUTH challenges, notices328 const r = neg.reconcile(hexToBytes(String(m[2])));329 need.push(...r.need);330 if (need.length > MAX_ITEMS) throw new Error("sync result exceeded the import limit");331 if (r.reply === null) break;332 sock.send("NEG-MSG", id, bytesToHex(r.reply));333 }334 sock.send("NEG-CLOSE", id);335 return need;336}337338// fetchEvents asks for a batch by id and stores what checks out. Signatures339// are verified, bans and kind rules apply, the write policy does not: the340// owner asked for these.341async function fetchEvents(relay: Relay, sock: PullSocket, ids: string[], job: PullJob, deadline: number) {342 const sub = "pull-" + job.stored + "-" + job.skipped;343 sock.send("REQ", sub, { ids });344 const seen = new Set<string>();345 for (let messages = 0; ; messages++) {346 if (messages >= ROUND_MESSAGES || Date.now() >= deadline) throw new Error("source exceeded its event response budget");347 const m = await sock.recv(deadline - Date.now());348 if (m[0] === "EOSE" && m[1] === sub) break;349 if (m[0] === "CLOSED" && m[1] === sub) throw new Error("query refused: " + String(m[2]));350 if (m[0] !== "EVENT" || m[1] !== sub) continue;351 const e = m[2] as Event;352 if (validate(e) || !ids.includes(e.id) || !matchesPull(e, job.filter) || seen.has(e.id)) {353 job.skipped++;354 markRejected(job);355 continue;356 }357 seen.add(e.id);358 storePulled(relay, e, job);359 }360 sock.send("CLOSE", sub);361 if (seen.size < ids.length && job.progress) {362 job.progress.partial = true;363 job.progress.warning = "The source did not supply all events advertised by its sync response.";364 }365}366367// copyBlobs brings over a batch of the other relay's files. Returns true368// when more remain.369async function copyBlobs(relay: Relay, srcName: string, job: PullJob): Promise<boolean> {370 // The source applies its read rule to the copier, who has proved no key371 // (the pull never authenticates), so files come across from an open relay372 // only, the same as its events.373 const rows: Blob[] | string = await relay.relays.getByName(srcName).listBlobs([]);374 if (typeof rows === "string") throw new Error("files refused: " + rows);375 let n = 0;376 for (const b of rows) {377 if (relay.sql.exec(`SELECT 1 FROM blobs WHERE sha256=?`, b.sha256).toArray().length) continue;378 if (n === ROUND_BLOBS) return true;379 const obj = await relay.media.get(`${srcName}/${b.sha256}`);380 if (!obj) continue;381 await relay.media.put(`${relay.slug}/${b.sha256}`, obj.body, { httpMetadata: { contentType: b.type } });382 relay.sql.exec(`INSERT OR IGNORE INTO blobs(sha256,size,type,uploader,uploaded) VALUES(?,?,?,?,?)`, b.sha256, b.size, b.type, b.uploader, b.uploaded);383 relay.meterBytes(b.size, 0);384 job.blobs++;385 n++;386 }387 return false;388}