diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/features/messages.js | 125 | ||||
| -rw-r--r-- | src/negentropy.js | 197 | ||||
| -rw-r--r-- | src/nostr.js | 94 |
3 files changed, 399 insertions, 17 deletions
diff --git a/src/features/messages.js b/src/features/messages.js index a40f7d5..ad5c32c 100644 --- a/src/features/messages.js +++ b/src/features/messages.js @@ -17,6 +17,7 @@ import { npubOf, neventOf, parseNostrPubkey, parseNostrRef, generateSecretKey, getPublicKey, finalizeEvent, nip44, PROFILE_RELAYS, openWrapsOffthread, unwrapDMsOffthread, verifyEventsAsync, setSyncConfig, DEFAULT_SYNC_RELAYS, } from '../nostr.js'; +import { reconcileOn, syncOn, negentropyOff } from '../nostr.js'; // NIP-77: fetch only what we lack import { channelKey, channelStream, channelEpoch, channelIsPrivate, controlKey, guestbookKey, openWrap, wrapRumor, rumorWithId, foldControl, foldGuestbook, observeAuthor, eventMs, msTags, makeEdition, @@ -688,23 +689,57 @@ export function messagesFeature(ctx) { healSince = 0; } const healing = (wrap) => !!healSince && wrap.created_at >= healSince && !healed.has(wrap.id) && !!healed.add(wrap.id); - function rememberWrap(id) { - const k = wrapKey(id); - if (!wrapLog) wrapLog = []; - if (!seenWraps.has(k)) wrapLog.push(k); - seenWraps.add(k); + // The wraps this device holds — full id and time — for reconciliation + // (NIP-77) with a relay that speaks it: a subscribe then fetches only the + // wraps we lack instead of the relay's newest two hundred (half a + // megabyte, every boot and every resubscribe), and the live subscription + // opens with limit 0, nothing re-served. The prefix log above stays the + // "skip this one" memory; a wrap it knows but this list doesn't (logged + // before the list existed) is added as the relay re-serves it, once. + const WRAP_ITEMS_MAX = 600; // the relay reconciles its newest 400: that many held, with room (46 KB of a phone's near-full localStorage) + let wrapItems = null; // id -> created_at, oldest first + function loadWrapItems() { + if (wrapItems) return; + wrapItems = new Map(); + try { for (const [id, ts] of st().wrapItems || []) if (/^[0-9a-f]{64}$/.test(id)) wrapItems.set(id, ts | 0); } catch {} + } + function noteWrapItem(id, ts) { + if (!id || !ts) return; + loadWrapItems(); + if (wrapItems.has(id)) return; + wrapItems.set(id, ts); + while (wrapItems.size > WRAP_ITEMS_MAX) wrapItems.delete(wrapItems.keys().next().value); + scheduleWrapSave(); + } + const wrapItemsHeld = () => { loadWrapItems(); return [...wrapItems].map(([id, ts]) => [ts, id]); }; + function scheduleWrapSave() { clearTimeout(wrapsTimer); wrapsTimer = setTimeout(() => { try { // another tab may have logged wraps since this one loaded: keep theirs const s = st(); - const mine = new Set(wrapLog); - s.wraps = [...(s.wraps || []).filter((k) => !mine.has(k)), ...wrapLog].slice(-WRAPS_MAX); - wrapLog = s.wraps; + if (wrapLog) { + const mine = new Set(wrapLog); + s.wraps = [...(s.wraps || []).filter((k) => !mine.has(k)), ...wrapLog].slice(-WRAPS_MAX); + wrapLog = s.wraps; + } + if (wrapItems) { + const all = new Map(s.wrapItems || []); + for (const [id, ts] of wrapItems) all.set(id, ts); + s.wrapItems = [...all].slice(-WRAP_ITEMS_MAX); + } save(s); } catch {} }, 1500); } + function rememberWrap(id, ts) { + const k = wrapKey(id); + if (!wrapLog) wrapLog = []; + if (!seenWraps.has(k)) wrapLog.push(k); + seenWraps.add(k); + if (ts) noteWrapItem(id, ts); + scheduleWrapSave(); + } let dmStarted = false; let allUnsubs = []; let dmUnsubs = []; // the DM wrap subscriptions alone, so they can be rebuilt @@ -3055,13 +3090,14 @@ export function messagesFeature(ctx) { async function handleInboxWrap(wrap) { let heal = false; if (seenWraps.has(wrap.id) || seenWraps.has(wrapKey(wrap.id))) { + noteWrapItem(wrap.id, wrap.created_at); // known by its prefix alone: now held in full if (!healing(wrap)) return; heal = true; } seenWraps.add(wrap.id); const r = await openInboxWrap(wrap, heal); if (r === 'stale') return; - if (r === true) { rememberWrap(wrap.id); return; } + if (r === true) { rememberWrap(wrap.id, wrap.created_at); return; } if (pendingWraps.size >= PENDING_MAX) pendingWraps.delete(pendingWraps.keys().next().value); pendingWraps.set(wrap.id, { wrap, heal, tries: r !== 'later' && decryptorsComplete() ? 1 : 0 }); scheduleDrain(8000); @@ -3101,9 +3137,9 @@ export function messagesFeature(ctx) { if (complete && !decryptorsComplete()) break; const r = await openInboxWrap(p.wrap, p.heal); if (r === 'stale') break; // the account changed under this pass - if (r === true) { pendingWraps.delete(id); rememberWrap(id); } + if (r === true) { pendingWraps.delete(id); rememberWrap(id, p.wrap && p.wrap.created_at); } else if (r === 'later') break; // the signer went quiet — nothing learned, try the rest later - else if (complete && ++p.tries >= PENDING_TRIES) { pendingWraps.delete(id); rememberWrap(id); } + else if (complete && ++p.tries >= PENDING_TRIES) { pendingWraps.delete(id); rememberWrap(id, p.wrap && p.wrap.created_at); } if (performance.now() - chunkStart > 8) { await new Promise((r) => setTimeout(r)); chunkStart = performance.now(); @@ -3160,12 +3196,42 @@ export function messagesFeature(ctx) { } bumpMsgRev(); if (swept) save(s); - const dmSub = (relays) => { - const u = subscribeOn(relays, { kinds: [1059], '#p': pks, limit: 400 }, (wrap) => { - handleInboxWrap(wrap).catch(() => {}); - }, { onclose: noteSubClosed }); + const onWrap = (wrap) => { handleInboxWrap(wrap).catch(() => {}); }; + const plainDmSub = (relays) => { + const u = subscribeOn(relays, { kinds: [1059], '#p': pks, limit: 400 }, onWrap, { onclose: noteSubClosed }); allUnsubs.push(u); dmUnsubs.push(u); }; + // Relay by relay. One that speaks negentropy gets the live subscription + // with limit 0 — opened first, so nothing slips between — and a + // reconciliation of the wraps we hold against its newest four hundred, + // the ones we lack fetched by id. One that doesn't (nos.lol has it + // disabled; remembered for a day) is asked the old way, its newest + // wraps re-served and skipped by the prefix log. So is every relay + // while the one-time heal is running: that pass wants them re-served. + const dmSub = (relays) => { + for (const relay of relays) { + if (healSince || negentropyOff(relay)) { plainDmSub([relay]); continue; } + const u = subscribeOn([relay], { kinds: [1059], '#p': pks, limit: 0 }, onWrap, { onclose: noteSubClosed }); + allUnsubs.push(u); dmUnsubs.push(u); + reconcileWraps(relay).then((ok) => { + if (ok || !dmStarted || !dmUnsubs.includes(u)) return; + // it doesn't speak it after all: the plain subscription, with its backlog + try { u(); } catch {} + allUnsubs = allUnsubs.filter((x) => x !== u); dmUnsubs = dmUnsubs.filter((x) => x !== u); + plainDmSub([relay]); + }).catch(() => {}); + } + }; + async function reconcileWraps(relay) { + const r = await reconcileOn(relay, { kinds: [1059], '#p': pks, limit: 400 }, wrapItemsHeld()); + if (!r) return false; + dlog('dm: ' + relay + ' reconciled: ' + r.need.length + ' wraps to fetch, ' + r.have.length + ' held it lacks'); + for (let i = 0; i < Math.min(r.need.length, 600) && dmStarted; i += 100) { + const evs = await queryOn([relay], { ids: r.need.slice(i, i + 100) }, 6000); + for (const w of evs) onWrap(w); + } + return true; + } dmSub(DM_RELAYS); // Senders deliver to the relays our kind-10050 advertises — a list other // clients may have published with relays beyond our defaults. Read those @@ -15092,6 +15158,28 @@ export function messagesFeature(ctx) { let notif = null; // { status, items } const NOTIF_V = 2; let notifAt = 0, notifUnsub = null; + // Every notification event seen, id and time, for reconciliation (NIP-77) + // with the relays that speak it: a refresh then fetches only what's new, + // not the newest hundred and twenty again — a hundred kilobytes a visit. + // All of them, not the rows kept: a zap receipt folded into its twin, or + // a hidden author's reply, would otherwise come back on every refresh. + const NOTIF_SEEN_MAX = 300; // the relay reconciles its newest 120 + let notifHeld = null; // id -> created_at, oldest first + function notifHeldItems() { + if (!notifHeld) { + notifHeld = new Map(); + try { const s = st(); if (s.notifHeldPk === identityPk()) for (const [id, ts] of s.notifHeld || []) notifHeld.set(id, ts | 0); } catch {} + for (const x of notifNow().items) if (x && x.id && x.ts && !notifHeld.has(x.id)) notifHeld.set(x.id, x.ts); + } + return [...notifHeld].map(([id, ts]) => [ts, id]); + } + function noteNotifHeld(evs) { + notifHeldItems(); + let grew = false; + for (const ev of evs || []) if (ev && ev.id && !notifHeld.has(ev.id)) { notifHeld.set(ev.id, ev.created_at | 0); grew = true; } + while (notifHeld.size > NOTIF_SEEN_MAX) notifHeld.delete(notifHeld.keys().next().value); + return grew; + } const notifNoteAsking = new Set(); const notifSeen = () => st().notifSeen || 0; // Notifications are about the IDENTITY, not the wallet: the stored list @@ -15151,6 +15239,7 @@ export function messagesFeature(ctx) { for (const id of wanted) if (notifNotes.has(id)) notes[id] = savedNotifNote(notifNotes.get(id)); const s2 = st(); s2.notifs = notif.items; s2.notifsPk = identityPk(); s2.notifsV = NOTIF_V; + if (notifHeld) { s2.notifHeld = [...notifHeld]; s2.notifHeldPk = identityPk(); } s2.notifNoteCache = notes; s2.notifNotesPk = identityPk(); save(s2); } @@ -15263,9 +15352,11 @@ export function messagesFeature(ctx) { notifAt = Date.now(); const g = session; try { - const evs = await queryOn(zapRelays(), { kinds: NOTIF_KINDS, '#p': my, limit: 120 }, 5000); + const evs = await syncOn(zapRelays(), { kinds: NOTIF_KINDS, '#p': my, limit: 120 }, notifHeldItems(), { maxWait: 5000 }); if (g !== session) return; // asked for the previous account + const grew = noteNotifHeld(evs); if (mergeNotifs(evs)) scheduleRepaint(); + if (grew) persistNotifs(); } catch {} finally { if (g === session) { if (notif) notif.status = 'ready'; @@ -15817,7 +15908,7 @@ export function messagesFeature(ctx) { noteCountsReady.clear(); // what happened to THEIR posts stays with them — the next identity // starts its list empty and asks the relays under its own key - stopNotifWatch(); notif = null; notifAt = 0; notifNotes.clear(); + stopNotifWatch(); notif = null; notifHeld = null; notifAt = 0; notifNotes.clear(); session++; zapWho.clear(); whoOpenIds.clear(); clearTimeout(zapSaveT); zapSaveT = null; zapSeed = null; diff --git a/src/negentropy.js b/src/negentropy.js new file mode 100644 index 0000000..83f65ae --- /dev/null +++ b/src/negentropy.js @@ -0,0 +1,197 @@ +// Negentropy (NIP-77): set reconciliation with a relay. The client holds a +// set of (created_at, id) pairs — the events it already has for a filter — +// and the relay holds its own. Instead of the relay re-sending every event +// the filter matches (a DM catch-up re-served 200 gift wraps, half a +// megabyte, on every subscribe; a notification refresh a hundred kilobytes), +// the two sides exchange fingerprints of ranges of their sets, split the +// ranges that differ, and end with the ids each side lacks — a few hundred +// bytes, usually in one round. The events themselves are then fetched by id. +// +// Protocol version 1 (0x61), as strfry speaks it. A port of Doug Hoyte's +// reference client (github.com/hoytech/negentropy, MIT, © 2023), trimmed to +// the initiator's side plus what a test needs to play the relay. +// Hashing is WebCrypto SHA-256, so everything here is async. + +const PROTOCOL_VERSION = 0x61; +const ID_SIZE = 32; +const FINGERPRINT_SIZE = 16; +const MODE = { skip: 0, fingerprint: 1, idList: 2 }; +const MAX_TS = Number.MAX_SAFE_INTEGER; // "infinity": the bound past every item + +// ---- bytes ---------------------------------------------------------------- +const hexOf = (a) => { let s = ''; for (let i = 0; i < a.length; i++) s += (a[i] < 16 ? '0' : '') + a[i].toString(16); return s; }; +const bytesOf = (h) => { if (h instanceof Uint8Array) return h; if (h.length % 2) throw new Error('odd hex'); const a = new Uint8Array(h.length / 2); for (let i = 0; i < a.length; i++) a[i] = parseInt(h.substr(i * 2, 2), 16); return a; }; +const cmpBytes = (a, b) => { const n = Math.min(a.length, b.length); for (let i = 0; i < n; i++) { if (a[i] !== b[i]) return a[i] < b[i] ? -1 : 1; } return a.length === b.length ? 0 : a.length < b.length ? -1 : 1; }; +const cmpItem = (a, b) => (a.ts === b.ts ? cmpBytes(a.id, b.id) : a.ts < b.ts ? -1 : 1); + +class Out { + constructor() { this.parts = []; this.length = 0; } + push(a) { if (Array.isArray(a)) a = Uint8Array.from(a); this.parts.push(a); this.length += a.length; } + bytes() { const o = new Uint8Array(this.length); let p = 0; for (const a of this.parts) { o.set(a, p); p += a.length; } return o; } +} +class In { + constructor(a) { this.a = a; this.p = 0; } + get left() { return this.a.length - this.p; } + byte() { if (this.p >= this.a.length) throw new Error('negentropy: message ends early'); return this.a[this.p++]; } + take(n) { if (this.p + n > this.a.length) throw new Error('negentropy: message ends early'); const s = this.a.subarray(this.p, this.p + n); this.p += n; return s; } +} +// varints, big-endian 7-bit groups with a continuation bit (the protocol's) +const putVar = (o, n) => { if (n === 0) return o.push([0]); const g = []; while (n > 0) { g.push(n % 128); n = Math.floor(n / 128); } g.reverse(); for (let i = 0; i < g.length - 1; i++) g[i] |= 128; o.push(g); }; +const getVar = (i) => { let n = 0; for (;;) { const b = i.byte(); n = n * 128 + (b & 127); if (!(b & 128)) return n; } }; + +// ---- the fingerprint of a range: the ids summed mod 2^256, hashed with the count +const sha256 = async (bytes) => new Uint8Array(await crypto.subtle.digest('SHA-256', bytes)); +function addInto(acc, id) { + let carry = 0; + for (let i = 0; i < ID_SIZE; i++) { const s = acc[i] + id[i] + carry; acc[i] = s & 255; carry = s >> 8; } +} +async function fingerprint(items, begin, end) { + const acc = new Uint8Array(ID_SIZE); + for (let i = begin; i < end; i++) addInto(acc, items[i].id); + const o = new Out(); o.push(acc); putVar(o, end - begin); + return (await sha256(o.bytes())).subarray(0, FINGERPRINT_SIZE); +} + +// ---- the set ---------------------------------------------------------------- +// `items`: [[created_at, id hex], ...] in any order; duplicates are dropped. +export class NegSet { + constructor(items = []) { + const seen = new Set(); + this.items = []; + for (const [ts, id] of items) { + if (typeof id !== 'string' || id.length !== 64 || seen.has(id)) continue; + seen.add(id); + this.items.push({ ts: Math.max(0, Math.floor(ts || 0)), id: bytesOf(id) }); + } + this.items.sort(cmpItem); + } + get size() { return this.items.length; } + // the first index in [begin, end) at or past the bound + lowerBound(begin, end, bound) { + let lo = begin, hi = end; + while (lo < hi) { const m = (lo + hi) >> 1; if (cmpItem(this.items[m], bound) < 0) lo = m + 1; else hi = m; } + return lo; + } +} + +// ---- one side of a reconciliation ------------------------------------------ +export class Negentropy { + constructor(set, { frameSizeLimit = 0 } = {}) { + if (frameSizeLimit && frameSizeLimit < 4096) throw new Error('negentropy: frame size limit too small'); + this.set = set; + this.frameSizeLimit = frameSizeLimit; + this.initiator = false; + this.lastTsIn = 0; this.lastTsOut = 0; + } + // the client's opening message (hex) + async initiate() { + this.initiator = true; + const o = new Out(); o.push([PROTOCOL_VERSION]); + await this.splitRange(0, this.set.size, { ts: MAX_TS, id: new Uint8Array(0) }, o); + return hexOf(o.bytes()); + } + // One message in, one out. For the initiator the answer is null once the + // sets are reconciled; `have` are ids we hold that the other side lacks, + // `need` the ones it holds that we lack. The non-initiator (a relay, or + // a test playing one) always answers and collects nothing. + async reconcile(hex) { + const q = new In(bytesOf(hex)); + const have = [], need = []; + this.lastTsIn = this.lastTsOut = 0; + const full = new Out(); full.push([PROTOCOL_VERSION]); + const version = q.byte(); + if (version < 0x60 || version > 0x6f) throw new Error('negentropy: not a negentropy message'); + if (version !== PROTOCOL_VERSION) { + if (this.initiator) throw new Error('negentropy: unsupported protocol version ' + (version - 0x60)); + return { next: hexOf(full.bytes()), have, need }; + } + const items = this.set.items, size = this.set.size; + let prevBound = { ts: 0, id: new Uint8Array(0) }, prevIndex = 0, skip = false; + while (q.left) { + const o = new Out(); + const doSkip = () => { if (skip) { skip = false; this.putBound(o, prevBound); putVar(o, MODE.skip); } }; + const bound = this.getBound(q); + const mode = getVar(q); + const lower = prevIndex; + let upper = this.set.lowerBound(prevIndex, size, bound); + if (mode === MODE.skip) skip = true; + else if (mode === MODE.fingerprint) { + const theirs = q.take(FINGERPRINT_SIZE); + const ours = await fingerprint(items, lower, upper); + if (cmpBytes(theirs, ours) !== 0) { doSkip(); await this.splitRange(lower, upper, bound, o); } + else skip = true; + } else if (mode === MODE.idList) { + const n = getVar(q); + const theirs = new Map(); + for (let i = 0; i < n; i++) { const id = q.take(ID_SIZE); if (this.initiator) theirs.set(hexOf(id), id); } + if (this.initiator) { + skip = true; + for (let i = lower; i < upper; i++) { + const k = hexOf(items[i].id); + if (theirs.has(k)) theirs.delete(k); else have.push(k); + } + for (const k of theirs.keys()) need.push(k); + } else { + doSkip(); + const ids = new Out(); let count = 0, endBound = bound; + for (let i = lower; i < upper; i++) { + if (this.overFrame(full.length + ids.length)) { endBound = items[i]; upper = i; break; } + ids.push(items[i].id); count++; + } + this.putBound(o, endBound); putVar(o, MODE.idList); putVar(o, count); o.push(ids.bytes()); + full.push(o.bytes()); o.parts = []; o.length = 0; + } + } else throw new Error('negentropy: unexpected mode ' + mode); + if (this.overFrame(full.length + o.length)) { + // the frame is full: one fingerprint for everything left + const rest = await fingerprint(items, upper, size); + this.putBound(full, { ts: MAX_TS, id: new Uint8Array(0) }); putVar(full, MODE.fingerprint); full.push(rest); + break; + } + full.push(o.bytes()); + prevIndex = upper; prevBound = bound; + } + return { next: full.length === 1 && this.initiator ? null : hexOf(full.bytes()), have, need }; + } + // a range that differs: sixteen fingerprinted buckets, or the ids outright when small + async splitRange(lower, upper, upperBound, o) { + const items = this.set.items, n = upper - lower, buckets = 16; + if (n < buckets * 2) { + this.putBound(o, upperBound); putVar(o, MODE.idList); putVar(o, n); + for (let i = lower; i < upper; i++) o.push(items[i].id); + return; + } + const per = Math.floor(n / buckets), extra = n % buckets; + let cur = lower; + for (let b = 0; b < buckets; b++) { + const size = per + (b < extra ? 1 : 0); + const fp = await fingerprint(items, cur, cur + size); + cur += size; + const next = cur === upper ? upperBound : minimalBound(items[cur - 1], items[cur]); + this.putBound(o, next); putVar(o, MODE.fingerprint); o.push(fp); + } + } + overFrame(n) { return !!this.frameSizeLimit && n > this.frameSizeLimit - 200; } + // bounds: a timestamp as a delta from the last one (+1; 0 = infinity), then a prefix of an id + getBound(q) { + let ts = getVar(q); + ts = ts === 0 ? MAX_TS : ts - 1; + if (this.lastTsIn === MAX_TS || ts === MAX_TS) { this.lastTsIn = MAX_TS; ts = MAX_TS; } + else { ts += this.lastTsIn; this.lastTsIn = ts; } + const len = getVar(q); + if (len > ID_SIZE) throw new Error('negentropy: bound key too long'); + return { ts, id: q.take(len) }; + } + putBound(o, b) { + if (b.ts === MAX_TS) { this.lastTsOut = MAX_TS; putVar(o, 0); } + else { putVar(o, b.ts - this.lastTsOut + 1); this.lastTsOut = b.ts; } + putVar(o, b.id.length); o.push(b.id); + } +} +// the shortest bound that separates two neighbouring items +function minimalBound(prev, cur) { + if (cur.ts !== prev.ts) return { ts: cur.ts, id: new Uint8Array(0) }; + let shared = 0; + while (shared < ID_SIZE && cur.id[shared] === prev.id[shared]) shared++; + return { ts: cur.ts, id: cur.id.subarray(0, shared + 1) }; +} diff --git a/src/nostr.js b/src/nostr.js index b76e865..8dcab01 100644 --- a/src/nostr.js +++ b/src/nostr.js @@ -18,6 +18,7 @@ import { normalizeURL } from 'nostr-tools/utils'; import { wrapEvent as nip17WrapEvent } from 'nostr-tools/nip17'; import { SimplePool } from 'nostr-tools/pool'; import { dlog } from './debug.js'; +import { NegSet, Negentropy } from './negentropy.js'; import { randomBytes, bytesToHex } from '@noble/hashes/utils'; import { base64urlnopad } from '@scure/base'; @@ -365,6 +366,99 @@ export async function queryOn(relays, filter, maxWait = 1500) { } // Several filters in ONE REQ per relay (NIP-01 allows a list) — one // subscription against the relay's REQ budget instead of one per filter. +// ---- negentropy (NIP-77): fetch only what we lack --------------------------- +// A relay that speaks it (strfry with it enabled: relay.coinos.io does; +// nos.lol, damus and primal answer "negentropy disabled") compares the ids +// it holds for a filter with the ids we hold and tells us the difference — +// a few hundred bytes — instead of re-sending the newest N events every +// time we ask. The events we lack are then fetched by id. A relay that +// doesn't speak it is remembered for a day, so the plain query goes +// straight out there next time rather than after a wasted socket. +const NEG_OFF_KEY = 'coinos-neg-off'; +const NEG_OFF_MS = 24 * 60 * 60_000; +const negOff = new Map(); // normalized url -> until +try { + for (const [u, until] of Object.entries(JSON.parse(localStorage.getItem(NEG_OFF_KEY) || '{}'))) if (until > Date.now()) negOff.set(u, until); +} catch {} +function negRefused(url) { + negOff.set(relayKey(url), Date.now() + NEG_OFF_MS); + try { localStorage.setItem(NEG_OFF_KEY, JSON.stringify(Object.fromEntries([...negOff].slice(-100)))); } catch {} +} +export const negentropyOff = (url) => (negOff.get(relayKey(url)) || 0) > Date.now(); +// Reconcile `items` ([[created_at, id], ...]: the events we hold) with one +// relay under `filter`. Resolves { need, have } — the ids it holds that we +// don't, and ours it lacks — or null when it can't be done there: the relay +// doesn't speak it (a NOTICE, a CLOSED, a socket shut on the opening +// message, or silence — remembered for a day), refuses this query (NEG-ERR +// blocked/closed — not remembered), is sick, or we're offline. Its own +// socket: the pool's parser has no ear for NEG-* frames. +export async function reconcileOn(url, filter, items, { timeout = 8000 } = {}) { + const n = relayKey(url); + if (sick.has(n) || negentropyOff(url) || offline()) return null; + let ws; + try { ws = new WebSocket(url); } catch { return null; } + const neg = new Negentropy(new NegSet(items)); + const id = 'neg' + Math.random().toString(36).slice(2, 10); + return new Promise((resolve) => { + let done = false, heard = false; + const need = [], have = []; + const finish = (r, refused) => { + if (done) return; + done = true; clearTimeout(t); + if (refused) negRefused(url); + try { if (ws.readyState === 1) ws.send(JSON.stringify(['NEG-CLOSE', id])); } catch {} + try { ws.close(); } catch {} + resolve(r); + }; + const t = setTimeout(() => finish(null, !heard), timeout); + ws.onerror = () => finish(null, false); + ws.onclose = () => finish(null, !heard && !offline() && !hiddenTab()); + ws.onopen = async () => { + try { ws.send(JSON.stringify(['NEG-OPEN', id, filter, await neg.initiate()])); } catch { finish(null, false); } + }; + ws.onmessage = async (e) => { + let m; try { m = JSON.parse(e.data); } catch { return; } + if (!Array.isArray(m)) return; + if (m[0] === 'NEG-MSG' && m[1] === id) { + heard = true; + try { + const r = await neg.reconcile(m[2]); + need.push(...r.need); have.push(...r.have); + if (r.next) ws.send(JSON.stringify(['NEG-MSG', id, r.next])); + else finish({ need, have }, false); + } catch (err) { dlog('nostr: negentropy with ' + url + ' failed: ' + (err?.message || err)); finish(null, false); } + } else if (m[0] === 'NEG-ERR' && m[1] === id) { dlog('nostr: negentropy refused by ' + url + ': ' + m[2]); finish(null, false); } + else if (m[0] === 'NOTICE' && /negentropy|NEG-OPEN|bad msg/i.test(String(m[1] || ''))) finish(null, true); + else if (m[0] === 'CLOSED' && m[1] === id) finish(null, true); + }; + }); +} +// What `filter` matches on these relays, each event once, verified: by +// reconciliation where a relay speaks it (only the events we don't hold +// come, by id) and by the plain query where it doesn't. `items` are the +// events held, [[created_at, id], ...]. +export async function syncOn(relays, filter, items, { maxWait = 5000 } = {}) { + probeRelays(relays); + const live = readRelays(relays); + if (!live.length) return []; + const plain = [], byRelay = new Map(), seen = new Set(); + await Promise.all(live.map(async (url) => { + const r = await reconcileOn(url, filter, items).catch(() => null); + if (!r) { plain.push(url); return; } + for (const id of r.need) { + if (seen.has(id)) continue; // one copy is enough: the first relay that has it + seen.add(id); + if (!byRelay.has(url)) byRelay.set(url, []); + byRelay.get(url).push(id); + } + })); + const jobs = []; + if (plain.length) jobs.push(queryOn(plain, filter, maxWait)); + for (const [url, ids] of byRelay) for (let i = 0; i < Math.min(ids.length, 1000); i += 200) jobs.push(queryOn([url], { ids: ids.slice(i, i + 200) }, maxWait)); + const out = new Map(); + for (const evs of await Promise.all(jobs)) for (const e of evs || []) if (e && e.id && !out.has(e.id)) out.set(e.id, e); + return [...out.values()]; +} export async function queryManyOn(relays, filters, maxWait = 1500) { probeRelays(relays); try { |