- commit
- 943f2b86b04cb5b48b4f630d994b3454e8982598
- parent
- b3c4b24ad7866b28b68a83a5c670de8b7dc3f510
- author
- Dami <dami.etoile@gmail.com>
| A | helium/ | 48 | ++++++++++++ |
| A | helium/ | 123 | ++++++++++++++++++++++++++++++ |
| A | test/ | 62 | +++++++++++++++ |
| A | test/ | 69 | +++++++++++++++++ |
@@ -0,0 +1,48 @@
1+# 🎈 helium.js - sync for argon2+3+helium.js is a miniature Negentropy (NIP-77) client, the replication layer for argon.js. It reconciles argon's local events for a filter with each relay's, then fetches only what is missing locally and publishes only what the relay lacks. Apps work fully offline and catch up both ways when a connection returns.4+5+```html6+<script src="argon.js"></script>7+<script src="helium.js"></script>8+9+<main nhy-relay="wss://relay.example" nhe-sync='{"authors":["<your pubkey>"]}'></main>10+```11+12+On load and whenever the browser comes back online, the author's events are reconciled with every relay in both directions.13+14+## Attributes15+16+| attribute | default | description |17+|---|---|---|18+| `nhe-sync` | | JSON filter to reconcile. Runs on page load and on `online` |19+| `nhe-mode` | both ways | `pull` only fetches, `push` only publishes |20+21+Relays come from the closest `nhy-relay`, else `nhyCfg.relays`. Leave `limit` out of the filter: relays ignore it for sync.22+23+## Events24+25+| event | cancel to | detail |26+|---|---|---|27+| `nhe:sync` | skip this run | `cfg` (`filter`, `relays`, `mode`), editable |28+| `nhe:synced` | | `cfg`, `result` per relay |29+| `nhe:error` | | `cfg`, `relay`, `error` (for example the relay's `NEG-ERR` reason) |30+31+`result` is `{relay, have, need, pulled, pushed}`: `have` and `need` are the ids only this side or only the relay held.32+33+## Scripting34+35+```js36+let result = await helium.sync({authors:[pk]}, "wss://relay.example", "pull")37+helium.run(element) // run an nhe-sync element now38+```39+40+## Behavior41+42+- Speaks Negentropy protocol v1 as the initiator. Relays that cap frames (tinyrelay and bindws use 60KB) simply take more rounds.43+- A relay ending the exchange with `NEG-CLOSE` (tinyrelay does when the sets already match) counts as done.44+- Fetched events are checked with krypton.js when it is loaded, and stored through argon.45+- Publishing is one event at a time. A `rate-limited: ... retry in Ns` reply waits that long and retries, so a large backlog trickles in at the relay's pace.46+- A connection idle for 60 seconds is closed. There is no fallback for relays without NIP-77: the run reports `nhe:error`.47+48+Requires argon.js. Tested against bindws (bind.ws) and tinyrelay.@@ -0,0 +1,123 @@
1+(()=>{2+ let g = globalThis, d = g.document3+ if (g.helium) return4+ let n = 0, INF = 1/05+ let hex = (b)=>[...b].map((x)=>x.toString(16).padStart(2, "0")).join("")6+ let bytes = (h)=>(h.match(/../g) ?? []).map((x)=>parseInt(x, 16))7+ let varint = (o, v, a = [v % 128])=>{8+ while (v = Math.floor(v / 128)) a.unshift(v % 128 | 128)9+ o.push(...a)10+ }11+ let session = (items)=>{12+ let rt, wt, out13+ let fp = async(lo, hi, sum = Array(32).fill(0))=>{14+ for (let i = lo; i < hi; i++) for (let j = 0, c = 0; j < 32; j++) c = (c >> 8) + sum[j] + items[i].b[j], sum[j] = c & 25515+ varint(sum, hi - lo)16+ return hex(new Uint8Array(await crypto.subtle.digest("SHA-256", new Uint8Array(sum)), 0, 16))17+ }18+ let bound = (t, id = "")=>{19+ varint(out, t == INF ? 0 : t - wt + 1)20+ wt = t21+ varint(out, id.length / 2)22+ out.push(...bytes(id))23+ }24+ let split = async(lo, hi, t, id, c = hi - lo)=>{25+ if (c < 32){26+ bound(t, id)27+ out.push(2)28+ varint(out, c)29+ for (let i = lo; i < hi; i++) out.push(...items[i].b)30+ }31+ else for (let i = 0, cur = lo; i < 16; i++){32+ let f = await fp(cur, cur += (c >> 4) + (i < c % 16)), p = items[cur - 1], q = items[cur], k = 033+ if (cur == hi) bound(t, id)34+ else if (p.t != q.t) bound(q.t)35+ else {36+ while (p.b[k] == q.b[k]) k++37+ bound(q.t, q.id.slice(0, 2 * k + 2))38+ }39+ out.push(1, ...bytes(f))40+ }41+ }42+ let reconcile = async(msg)=>{43+ let b = bytes(msg), p = 1, have = [], need = [], prev = 0, skip, pb44+ let read = (v = 0, x)=>{45+ do v = v * 128 + ((x = b[p++]) & 127)46+ while (x & 128)47+ if (x == null) throw Error("short")48+ return v49+ }50+ if (b[0] != 97) throw Error("v1 only")51+ rt = wt = 052+ out = [97]53+ while (p < b.length){54+ let v = read(), t = rt = v ? rt + v - 1 : INF, len = read(), id = hex(b.slice(p, p += len)), mode = read(), hi = prev, x55+ while ((x = items[hi]) && (x.t < t || x.t == t && x.id < id.padEnd(64, "0"))) hi++56+ if (mode == 1 && hex(b.slice(p, p += 16)) != await fp(prev, hi)){57+ if (skip) bound(...pb), out.push(0)58+ skip = 059+ await split(prev, hi, t, id)60+ } else if (mode == 2){61+ let ids = new Set()62+ for (let c = read(); c--;) ids.add(hex(b.slice(p, p += 32)))63+ for (let i = prev; i < hi; i++) ids.delete(items[i].id) || have.push(items[i].id)64+ need.push(...ids)65+ skip = 166+ } else skip = 167+ prev = hi68+ pb = [t, id]69+ }70+ return {reply:out.length > 1 && hex(out), have, need}71+ }72+ return {start:async()=>(wt = 0, out = [97], await split(0, items.length, INF, ""), hex(out)), reconcile}73+ }74+ let sync = async(filter, url, mode)=>{75+ let items = (await argon.query([filter])).map((e)=>({t:e.created_at, id:e.id, b:bytes(e.id)})).sort((a, b)=>a.t - b.t || (a.id < b.id ? -1 : 1))76+ let s = session(items), id = "nhe" + ++n, have = [], need = [], pulled = 0, pushed = 0, q = [], wake, closed77+ let ws = new WebSocket(url), tx = (...m)=>ws.send(JSON.stringify(m)), t, idle = ()=>(clearTimeout(t), t = setTimeout(()=>ws.close(), 6e4))78+ let rx = async(f, i)=>{79+ while ((i = q.findIndex(f)) < 0) if (closed) throw Error("closed"); else await new Promise((r)=>wake = r)80+ return q.splice(i, 1)[0]81+ }82+ ws.onmessage = (m)=>(idle(), q.push(JSON.parse(m.data)), wake?.())83+ ws.onclose = ()=>(closed = 1, wake?.())84+ try {85+ await new Promise((r)=>ws.onopen = wake = r, idle())86+ tx("NEG-OPEN", id, filter, await s.start())87+ for (let k, m, r; ([k, , m] = await rx((x)=>x[1] == id))[0] != "NEG-CLOSE";){88+ if (k != "NEG-MSG") throw Error(m)89+ r = await s.reconcile(m)90+ have.push(...r.have)91+ need.push(...r.need)92+ tx(...r.reply ? ["NEG-MSG", id, r.reply] : ["NEG-CLOSE", id])93+ if (!r.reply) break94+ }95+ if (mode != "push") for (let i = 0; i < need.length; i += 250){96+ tx("REQ", id, {ids:need.slice(i, i + 250)})97+ for (let x; (x = await rx((x)=>x[1] == id))[0] == "EVENT";)98+ if (need.includes(x[2].id) && await (g.krypton?.check(x[2]) ?? 1) && await argon.put(x[2])) pulled++99+ tx("CLOSE", id)100+ }101+ if (mode != "pull") for (let e of await argon.query([{ids:have}])) for (let k, m; ;){102+ tx("EVENT", e)103+ ;[, , k, m] = await rx((x)=>x[0] == "OK" && x[1] == e.id)104+ if (k && ++pushed || !(m = /retry in (\d+)/.exec(m))) break105+ await new Promise((r)=>setTimeout(r, m[1] * 1e3, idle()))106+ }107+ return {relay:url, have, need, pulled, pushed}108+ } finally {109+ clearTimeout(t)110+ ws.close()111+ }112+ }113+ g.helium = {sync, session}114+ if (!d) return115+ let send = (el, type, detail)=>el.dispatchEvent(new CustomEvent("nhe:" + type, {detail, cancelable:1, bubbles:1}))116+ let run = g.helium.run = (el)=>{117+ let cfg = {filter:JSON.parse(el.getAttribute("nhe-sync")), relays:el.closest("[nhy-relay]")?.getAttribute("nhy-relay").split(/\s+/) ?? g.nhyCfg?.relays ?? [], mode:el.getAttribute("nhe-mode")}118+ if (send(el, "sync", {cfg})) for (let url of cfg.relays) sync(cfg.filter, url, cfg.mode).then((result)=>send(el, "synced", {cfg, result}), (error)=>send(el, "error", {cfg, relay:url, error}))119+ }120+ let all = ()=>d.querySelectorAll("[nhe-sync]").forEach(run)121+ g.addEventListener("online", all)122+ d.addEventListener("DOMContentLoaded", all)123+})()@@ -0,0 +1,62 @@
1+// Reconciles helium's initiator against bindws's Negentropy responder (a port of go-nostr's).2+import { assertEquals } from "jsr:@std/assert@1";3+import { sha256 } from "npm:@noble/hashes@1.8.0/sha2";4+import "../helium/helium.js";5+6+const { helium } = globalThis;7+const peer = new URL("../../bindws/src/negentropy.ts", import.meta.url);8+const hasPeer = await Deno.stat(peer).then(() => true, () => false);9+const { Negentropy, hexToBytes, bytesToHex } = hasPeer ? await import(peer.href) : {};10+11+const rid = () => [...crypto.getRandomValues(new Uint8Array(32))].map((x) => x.toString(16).padStart(2, "0")).join("");12+const item = (t, id = rid()) => ({ t, id, b: hexToBytes(id) });13+const sorted = (xs) => [...xs].sort((a, b) => a.t - b.t || (a.id < b.id ? -1 : 1));14+15+async function reconcile(mine, theirs) {16+ const client = helium.session(sorted(mine));17+ const server = new Negentropy(theirs.map((x) => ({ timestamp: x.t, id: x.b })), (b) => sha256(b));18+ let msg = await client.start(), rounds = 0;19+ const have = new Set(), need = new Set();20+ for (;;) {21+ const { reply } = server.reconcile(hexToBytes(msg));22+ const r = await client.reconcile(bytesToHex(reply));23+ r.have.forEach((x) => have.add(x));24+ r.need.forEach((x) => need.add(x));25+ rounds++;26+ if (!r.reply) break;27+ msg = r.reply;28+ }29+ return { have, need, rounds };30+}31+32+function check(mine, theirs, got) {33+ const m = new Set(mine.map((x) => x.id)), t = new Set(theirs.map((x) => x.id));34+ assertEquals([...got.have].sort(), [...m].filter((x) => !t.has(x)).sort());35+ assertEquals([...got.need].sort(), [...t].filter((x) => !m.has(x)).sort());36+}37+38+Deno.test({ name: "helium reconciles with bindws negentropy", ignore: !hasPeer }, async (t) => {39+ const cases = {40+ "both empty": [0, 0, 0],41+ "only mine": [40, 0, 0],42+ "only theirs": [0, 40, 0],43+ "around the 32 id list threshold": [31, 33, 16],44+ "identical 3000": [0, 0, 3000],45+ "5000 each, mostly shared": [300, 200, 4500],46+ "small diff in 20000": [3, 5, 20000],47+ };48+ for (const [name, [a, b, shared]] of Object.entries(cases)) {49+ await t.step(name, async () => {50+ const common = Array.from({ length: shared }, (_, i) => item(1_700_000_000 + (i % 997)));51+ const mine = [...common, ...Array.from({ length: a }, (_, i) => item(1_700_000_000 + i * 7))];52+ const theirs = [...common, ...Array.from({ length: b }, (_, i) => item(1_700_000_000 + i * 3))];53+ check(mine, theirs, await reconcile(mine, theirs));54+ });55+ }56+ await t.step("shared timestamps force id prefix bounds", async () => {57+ const same = Array.from({ length: 2000 }, () => item(1_700_000_000));58+ const mine = [...same, ...Array.from({ length: 10 }, () => item(1_700_000_000))];59+ const theirs = [...same.slice(50), ...Array.from({ length: 10 }, () => item(1_700_000_000))];60+ check(mine, theirs, await reconcile(mine, theirs));61+ });62+});@@ -0,0 +1,69 @@
1+// Live NIP-77 sync against real relays. Opt in with NOBEL_RELAYS="wss://a ws://b" (space separated).2+// Each relay gets a fresh key: some events only local, some only remote, some on both.3+import "npm:fake-indexeddb@6/auto";4+import { assertEquals } from "jsr:@std/assert@1";5+import "../krypton/krypton.js";6+import "../argon/argon.js";7+import "../helium/helium.js";8+9+const { krypton, argon, helium } = globalThis;10+const relays = (Deno.env.get("NOBEL_RELAYS") ?? "").split(/\s+/).filter(Boolean);11+const SIZE = Number(Deno.env.get("NOBEL_SIZE") ?? 60);12+13+// Publishes in parallel, then retries anything rate limited after the relay's hint.14+async function publish(url, events) {15+ const ws = new WebSocket(url);16+ await new Promise((r, j) => (ws.onopen = r, ws.onerror = j));17+ const pending = new Map();18+ ws.onmessage = (m) => {19+ const [t, id, ok, msg] = JSON.parse(m.data);20+ if (t === "OK") pending.get(id)?.([ok, msg]);21+ };22+ const send = (e) => new Promise((r) => {23+ pending.set(e.id, r);24+ ws.send(JSON.stringify(["EVENT", e]));25+ });26+ // bind.ws allows 120 events a minute per connection: burst, then 18 every 10s.27+ let todo = events, rest = [];28+ while (todo.length) {29+ const results = await Promise.all(todo.map(send));30+ const limited = todo.filter((_, i) => !results[i][0] && /retry in/.test(results[i][1]));31+ const bad = results.find(([ok, msg]) => !ok && !/retry in/.test(msg));32+ if (bad) throw new Error(`relay refused: ${bad[1]}`);33+ rest = [...limited, ...rest];34+ todo = rest.splice(0, 18);35+ if (todo.length) await new Promise((r) => setTimeout(r, 10_000));36+ }37+ ws.close();38+}39+40+for (const url of relays) {41+ Deno.test({ name: `helium syncs with ${url}`, sanitizeResources: false, sanitizeOps: false }, async (t) => {42+ await argon.remove();43+ const sk = krypton.generate(), pk = krypton.pub(sk), now = Math.floor(Date.now() / 1000);44+ const make = (n, tag) => Promise.all(Array.from({ length: n }, (_, i) =>45+ krypton.finish({ kind: 1, created_at: now - (i % 50), tags: [["t", "nobel-test"]], content: `${tag} ${i}` }, sk)));46+ const shared = await make(SIZE, "shared"), local = await make(Math.ceil(SIZE / 3), "local"), remote = await make(Math.ceil(SIZE / 4), "remote");47+ await publish(url, [...shared, ...remote]);48+ for (const e of [...shared, ...local]) await argon.put(e);49+ const filter = { authors: [pk] };50+51+ await t.step("first sync pulls what is missing and pushes what the relay lacks", async () => {52+ const r = await helium.sync(filter, url);53+ assertEquals([r.have.length, r.need.length, r.pushed, r.pulled], [local.length, remote.length, local.length, remote.length]);54+ assertEquals((await argon.query([filter])).length, shared.length + local.length + remote.length);55+ });56+57+ await t.step("second sync finds nothing to do", async () => {58+ const r = await helium.sync(filter, url);59+ assertEquals([r.have.length, r.need.length], [0, 0]);60+ });61+62+ await t.step("pull mode never publishes", async () => {63+ const extra = await make(3, "pull-only");64+ for (const e of extra) await argon.put(e);65+ const r = await helium.sync(filter, url, "pull");66+ assertEquals([r.have.length, r.pushed, r.need.length], [3, 0, 0]);67+ });68+ });69+}