1// Bounded GRASP-02/03 reconciliation. Durable windows retain history gaps;2// each connection follows EOSE with a live tail and overlaps the next wake.3import { sha256 } from "@noble/hashes/sha2.js";4import { bytesToHex } from "./negentropy.ts";5import { Socket, localName, type PullSocket } from "./pull.ts";6import { validate, now, isPrivate, isEphemeral, type Event } from "./event.ts";7import { match, type Filter } from "./filter.ts";8import { featureOn } from "./settings.ts";9import { announcements, hostedAnnouncements, graspEvents, graspVisible } from "./grasp-state.ts";10import { repositoryCoordinate, recursiveMaintainers, relatedRepositoryCoordinates } from "./grasp-policy.ts";11import { gitSource } from "./grasp-git-sync.ts";12import { KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR, KIND_GIT_PR_UPDATE, KIND_REPO, KIND_REPO_STATE } from "./kinds.ts";13import type { Relay } from "./relay.ts";1415export const GRASP_SYNC_PAGE = 256;16const MAX_JOBS = 512, MAX_SOURCES = 16, MAX_WINDOWS = 128, LIVE_MS = 1500, ROUND_MS = 10_000;17const UTF8 = new TextEncoder();18type Window = { since: number; until: number };19type Scope = { repo: string; mode: "repository" | "thread" | "metadata"; filters: Filter[] };20type Job = { id: string; source: string; scope: Scope };21type Row = { id: string; source: string; scope: string; windows: string; live_since: number; due: number; failures: number; error: string };22export type GraspSyncConnect = (relay: Relay, source: string) => Promise<PullSocket>;23const enabled = (relay: Relay) => featureOn(relay.settings.policy, "grasp02") && relay.settings.policy.reads === "open" && !relay.settings.isUnclaimed() && !relay.fuelStatus().outOfFuel;2425export function eventSource(relay: Relay, raw: string): string | null {26 try {27 const u = new URL(raw);28 if (u.protocol !== "wss:") return null;29 u.protocol = "https:";30 const safe = gitSource(relay, u.href);31 return safe ? safe.replace(/^https:/u, "wss:") : null;32 } catch { return null; }33}3435const productionConnect: GraspSyncConnect = async (relay, source) => {36 const safe = eventSource(relay, source);37 if (!safe) throw new Error("source-not-admitted");38 const url = new URL(safe.replace(/^wss:/u, "https:"));39 const local = localName(url, relay.domain);40 const abort = new AbortController();41 const timer = setTimeout(() => abort.abort(), ROUND_MS);42 try {43 const init: RequestInit = { headers: { upgrade: "websocket", ...(local ? { "x-relay-name": local } : {}) }, redirect: "manual", signal: abort.signal };44 const response = local ? await relay.relays.getByName(local).fetch(url.href, init) : await relay.fetcher(url.href, init);45 if (response.status !== 101 || !response.webSocket) { await response.body?.cancel(); throw new Error("websocket-refused"); }46 response.webSocket.accept();47 return new Socket(response.webSocket);48 } finally { clearTimeout(timer); }49};5051function jobs(relay: Relay): { jobs: Job[]; partial: boolean } {52 const result = new Map<string, Job>();53 let partial = false;54 const sources = (raw: string[]) => {55 const values = [...new Set(raw.map(s => eventSource(relay, s)).filter((s): s is string => s !== null))].sort();56 if (values.length > MAX_SOURCES) partial = true;57 return values.slice(0, MAX_SOURCES);58 };59 const add = (scope: Scope, urls: string[]) => {60 for (const source of sources(urls)) {61 const id = bytesToHex(sha256(UTF8.encode(JSON.stringify({ source, scope }))));62 if (result.has(id)) continue;63 if (result.size >= MAX_JOBS) { partial = true; return; }64 result.set(id, { id, source, scope });65 }66 };67 const all = announcements(relay);68 const roots = [KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR, KIND_GIT_PR_UPDATE].flatMap(k => { const events = graspEvents(relay, k); if (events.length >= 1025) partial = true; return events; }).filter(e => graspVisible(relay, e.id));69 const rootCoordinates = new Map(roots.map(e => [e.id, relatedRepositoryCoordinates(e)]));70 for (const repo of hostedAnnouncements(relay)) {71 const coordinate = repositoryCoordinate(repo);72 const maintainers = recursiveMaintainers(repo, all);73 if (!maintainers) { partial = true; continue; }74 add({ repo: coordinate, mode: "repository", filters: [75 { kinds: [KIND_REPO, KIND_REPO_STATE], authors: [...maintainers].sort(), tags: { d: [repo.identifier] } },76 { tags: { a: [coordinate] } }, { tags: { A: [coordinate] } },77 { tags: { e: [repo.id] } }, { tags: { E: [repo.id] } },78 ] }, repo.relays);79 const threads = roots.filter(e => rootCoordinates.get(e.id)?.includes(coordinate) || e.tags.some(t => (t[0] === "e" || t[0] === "E") && rootCoordinates.get(t[1])?.includes(coordinate))).sort((a, b) => a.id.localeCompare(b.id));80 for (let start = 0; start < threads.length; start += 50) {81 const batch = threads.slice(start, start + 50), ids = batch.map(e => e.id);82 const filters: Filter[] = [{ tags: { e: ids } }, { tags: { E: ids } }];83 const mentioned = [...new Set(batch.flatMap(e => e.tags.filter(t => t[0] === "e" || t[0] === "E").map(t => t[1]).filter(id => /^[0-9a-f]{64}$/u.test(id))))];84 if (mentioned.length > 50) partial = true;85 if (mentioned.length) filters.push({ ids: mentioned.slice(0, 50).sort(), tags: {} });86 const outboxes: string[] = [];87 if (featureOn(relay.settings.policy, "grasp03")) {88 const authors = new Set(batch.map(e => e.pubkey));89 for (const f of filters.slice(0, 2)) {90 const rows = relay.store.query(f, { pubkeys: [], all: true }, GRASP_SYNC_PAGE, now()).rows;91 if (rows.length >= GRASP_SYNC_PAGE) partial = true;92 for (const raw of rows) { const e = JSON.parse(raw) as Event; if (!isPrivate(e.kind) && graspVisible(relay, e.id)) authors.add(e.pubkey); }93 }94 if (authors.size > 64) partial = true;95 for (const author of [...authors].sort().slice(0, 64)) {96 const lists = relay.store.query({ kinds: [10002], authors: [author], tags: {} }, { pubkeys: [], all: true }, 1, now()).rows;97 const outbox = lists.flatMap(raw => (JSON.parse(raw) as Event).tags.filter(t => t[0] === "r" && (!t[2] || t[2] === "write")).map(t => t[1]));98 outboxes.push(...outbox);99 add({ repo: coordinate, mode: "metadata", filters: [{ kinds: [0, 10002, 10317], authors: [author], tags: {} }] }, [...repo.relays, ...outbox]);100 }101 }102 add({ repo: coordinate, mode: "thread", filters }, [...repo.relays, ...outboxes]);103 }104 }105 return { jobs: [...result.values()], partial };106}107108const wire = (f: Filter, range: Partial<Window>) => ({ ...(f.ids ? { ids: f.ids } : {}), ...(f.kinds ? { kinds: f.kinds } : {}), ...(f.authors ? { authors: f.authors } : {}), ...Object.fromEntries(Object.entries(f.tags).map(([k, v]) => ["#" + k, v])), ...range, limit: GRASP_SYNC_PAGE });109110async function round(relay: Relay, job: Job, row: Row, connect: GraspSyncConnect) {111 const windows: Window[] = JSON.parse(row.windows);112 let warning = "", liveSince = row.live_since;113 const socket = await connect(relay, job.source);114 const deadline = Date.now() + ROUND_MS;115 let messages = 0, bytes = 0;116 const send = (...m: unknown[]) => { relay.meterBytes(0, UTF8.encode(JSON.stringify(m)).length); socket.send(...m); };117 const recv = async (end: number) => {118 if (++messages > 1024 || Date.now() >= deadline) throw new Error("round-limit");119 const m = await socket.recv(Math.max(1, Math.min(end, deadline) - Date.now()));120 const n = UTF8.encode(JSON.stringify(m)).length;121 relay.meterBytes(n, 0); bytes += n;122 if (bytes > 4 * 1024 * 1024 || n > 512 * 1024) throw new Error("byte-limit");123 return m;124 };125 const accept = async (raw: unknown, range: Partial<Window>) => {126 if (validate(raw)) { warning = "invalid-event"; return; }127 const e = raw as Event;128 if (!job.scope.filters.some(f => match({ ...f, ...range }, e))) { warning = "out-of-scope-event"; return; }129 if (isPrivate(e.kind) || isEphemeral(e.kind) || e.tags.some(t => t[0] === "-")) { warning = "private-event-skipped"; return; }130 if (!enabled(relay)) throw new Error("sync-disabled");131 const result = await relay.acceptAny(e, { host: `${relay.slug}.${relay.domain}`, ip: "grasp-sync", challenge: "", authed: [], subs: {} });132 if (result.stored) relay.broadcast(e, false);133 if (!result.ok) warning = "admission-gap";134 };135 try {136 if (windows.length) {137 const range = windows[windows.length - 1], sub = "grasp-history";138 let received = 0;139 send("REQ", sub, ...job.scope.filters.map(f => wire(f, range)));140 while (true) {141 const m = await recv(deadline);142 if (m[1] !== sub) continue;143 if (m[0] === "EOSE") break;144 if (m[0] === "CLOSED") throw new Error("history-refused");145 if (m[0] !== "EVENT") continue;146 if (++received > GRASP_SYNC_PAGE * job.scope.filters.length) throw new Error("page-limit");147 await accept(m[2], range);148 }149 send("CLOSE", sub);150 if (received >= GRASP_SYNC_PAGE) {151 if (range.since >= range.until) warning = "partial-single-second";152 else if (windows.length >= MAX_WINDOWS) warning = "partial-window-limit";153 else {154 windows.pop();155 const mid = range.since + Math.floor((range.until - range.since) / 2);156 windows.push({ since: range.since, until: mid }, { since: mid + 1, until: range.until });157 }158 } else if (!warning) windows.pop(); // Invalid/refused coverage remains repairable.159 }160 const historyWarning = warning;161 warning = "";162 const sub = "grasp-live", began = now(), range = { since: row.live_since };163 const end = Math.min(deadline, Date.now() + LIVE_MS);164 let eose = false, received = 0;165 send("REQ", sub, ...job.scope.filters.map(f => wire(f, range)));166 while (Date.now() < end) {167 let m: unknown[];168 try { m = await recv(end); } catch (error) { if (!eose || messages > 1024 || bytes > 4 * 1024 * 1024 || (error instanceof Error && error.message === "byte-limit")) throw error; break; }169 if (m[1] !== sub) continue;170 if (m[0] === "CLOSED") throw new Error("live-refused");171 if (m[0] === "EOSE") { eose = true; continue; }172 if (m[0] !== "EVENT") continue;173 if (++received > GRASP_SYNC_PAGE) { warning = "partial-live-limit"; break; }174 await accept(m[2], range);175 }176 send("CLOSE", sub);177 if (!eose) throw new Error("live-missing-eose");178 if (received >= GRASP_SYNC_PAGE || warning) {179 if (windows.length < MAX_WINDOWS) windows.push({ since: row.live_since, until: began });180 else warning = "partial-window-limit";181 // Keep the inclusive live cursor when coverage was uncertain.182 } else liveSince = began;183 return { windows, liveSince, warning: warning || historyWarning };184 } finally { socket.close(); }185}186187// One fair, bounded source round per wake; failed jobs reserve their retry188// before I/O. Source/filter changes remove stale jobs, including outboxes.189export async function graspSyncTick(relay: Relay, connect: GraspSyncConnect = productionConnect): Promise<number> {190 if (!enabled(relay)) return 0;191 if (!relay.repositoryAccess.owned) throw new Error("event synchronization requires repository authority");192 const current = jobs(relay), ids = new Set(current.jobs.map(j => j.id));193 relay.sql.exec(`INSERT INTO grasp_sync_status(id,partial) VALUES(1,?) ON CONFLICT(id) DO UPDATE SET partial=excluded.partial WHERE partial!=excluded.partial`, current.partial ? 1 : 0);194 const existing = new Set(relay.sql.exec<{ id: string }>(`SELECT id FROM grasp_event_sync`).toArray().map(row => row.id));195 for (const id of existing) if (!ids.has(id)) relay.sql.exec(`DELETE FROM grasp_event_sync WHERE id=?`, id);196 for (const j of current.jobs) if (!existing.has(j.id)) relay.sql.exec(`INSERT OR IGNORE INTO grasp_event_sync(id,source,scope,windows,live_since) VALUES(?,?,?,?,?)`, j.id, j.source, JSON.stringify(j.scope), JSON.stringify([{ since: 0, until: now() }]), now());197 const row = relay.sql.exec<Row>(`SELECT * FROM grasp_event_sync ORDER BY due,id LIMIT 1`).toArray()[0];198 if (!row) return 0;199 if (row.due > now()) return row.due;200 relay.sql.exec(`UPDATE grasp_event_sync SET due=?,failures=failures+1 WHERE id=?`, now() + Math.min(300, 30 * 2 ** Math.min(row.failures, 4)), row.id);201 try {202 const result = await round(relay, current.jobs.find(j => j.id === row.id)!, row, connect);203 relay.sql.exec(`UPDATE grasp_event_sync SET windows=?,live_since=?,due=?,failures=0,error=? WHERE id=?`, JSON.stringify(result.windows), result.liveSince, now() + (result.warning ? 300 : result.windows.length ? 1 : 60), result.warning, row.id);204 } catch (error) {205 relay.sql.exec(`UPDATE grasp_event_sync SET error=? WHERE id=?`, (error instanceof Error ? error.message : "sync-failed").slice(0, 120), row.id);206 }207 const next = relay.sql.exec<{ due: number | null }>(`SELECT min(due) AS due FROM grasp_event_sync`).one().due;208 return next === null ? 0 : Math.max(now() + 1, next);209}