From d657d79bc95062a8fc5e128e795e23bcefd0bf04 Mon Sep 17 00:00:00 2001 From: Adam Soltys Date: Tue, 6 Oct 2026 11:22:06 -0700 Subject: Cordn group chats: MLS groups delivered through a coordinator (Amethyst 1.17 interop) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/contextvm.js speaks ContextVM (kind 25910 inside ephemeral gift wraps) to a coordinator; src/cordn.js is the protocol (KeyPackages with hex-text identities, the 0xC04D metadata extension, exporter-keyed ChaCha20 sealing, unsigned nostr-shaped envelopes with the sender as MLS AAD, cordn1 group refs, a record engine like marmot.js); src/cordn-client.js mirrors marmot-client.js with the same interface, so the Messages group screens drive both — New group gains a White Noise / Cordn toggle. Verified against @cordn/cli (ts-mls) on the public coordinator both ways, from the engine (tools/cordn-interop-test.js) and from the real app (tools/cordn-ui-test.js). mls.js: the OpenMLS required_capabilities rule for GroupContextExtensions now applies only to groups that carry one. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01YRC6aKhwAud231ot9FPmhi --- src/cordn-client.js | 365 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 365 insertions(+) create mode 100644 src/cordn-client.js (limited to 'src/cordn-client.js') diff --git a/src/cordn-client.js b/src/cordn-client.js new file mode 100644 index 0000000..d09488d --- /dev/null +++ b/src/cordn-client.js @@ -0,0 +1,365 @@ +// Cordn in the app: this device's standing with a coordinator (its MLS +// signature key and a published KeyPackage), its groups, and the traffic +// with the coordinator. Protocol rules live in cordn.js, the transport in +// contextvm.js; the messages feature hears about everything through +// `on(type, group, data)`, the same way it hears from marmot-client.js, and +// drives both through one interface. +// +// Like Marmot, MLS state is per DEVICE and lives in IndexedDB: a second +// device of the same account is a second leaf with its own KeyPackage. + +import * as C from './cordn.js'; +import { cvmClient } from './contextvm.js'; +import { marmotStore } from './marmot-store.js'; +import { finalizeEvent, nip44 } from './nostr.js'; +import { dlog } from './debug.js'; + +const DAY = 86400; +const KP_REFRESH = 7 * DAY; // republish well inside the 84-day lifetime; the coordinator keeps one last-resort package per identity +const POLL_FOCUSED = 4000; // a Cordn group on screen +const POLL_IDLE = 45_000; // groups held, none on screen +const POLL_EMPTY = 180_000; // no Cordn groups at all: only Welcomes to look for +const now = () => Math.floor(Date.now() / 1000); +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +// `store` is injectable for tests outside a browser (no IndexedDB there). +export function cordnClient({ scope, pubkey, identity, on, coordinator = C.DEFAULT_COORDINATOR, subClosed, store = marmotStore(scope + '|cordn') }) { + const groups = new Map(); // gid -> record + let dev = null, loaded = null, stopped = false, cvm = null, cvmSigner = null; + let kpBusy = null, pollTimer = 0, polling = null, focused = null, rotateTimer = 0; + const dirty = new Set(); + let flushTimer = 0; + + const emit = (type, g, data) => { try { on(type, g, data); } catch (e) { dlog('cordn: listener', e); } }; + + // ---- storage ---- + + function load() { + if (!loaded) + loaded = store.all().then((all) => { + dev = all.get('dev') || null; + for (const [k, v] of all) if (k.startsWith('g:') && v && v.tip) groups.set(v.id, v); + }).catch((e) => { dlog('cordn: store', e); }); + return loaded; + } + const touch = (g) => { dirty.add(g.id); clearTimeout(flushTimer); flushTimer = setTimeout(flush, 400); }; + async function flush() { + clearTimeout(flushTimer); + const ids = [...dirty]; + dirty.clear(); + await Promise.all(ids.map((id) => groups.has(id) ? store.put('g:' + id, groups.get(id)) : null)).catch((e) => dlog('cordn: save', e)); + } + // a ratchet that moved is on disk before anything of ours leaves + const persist = (g) => { dirty.delete(g.id); return store.put('g:' + g.id, g); }; + const saveDev = () => store.put('dev', dev); + + // ---- this device and its coordinator ---- + + async function signer({ interactive = false } = {}) { + const id = await identity(); + if (!id || !id.signer || id.pubkey !== pubkey) return null; + if (id.signer instanceof Uint8Array) { + const sk = id.signer; + return { raw: sk, sign: async (e) => finalizeEvent(e, sk), decrypt: async (peer, ct) => nip44.decrypt(ct, nip44.getConversationKey(sk, peer)) }; + } + // Every coordinator call is signed, and every reply decrypted, by the + // signer: an interactive one (a trip through another app per call) is + // never used unasked. + if (id.signer.interactive && !interactive) return null; + if (!id.signer.decryptFrom) return null; + return { raw: null, sign: (e) => id.signer.signEvent(e), decrypt: (peer, ct) => id.signer.decryptFrom(peer, ct) }; + } + + // The transport, once a signer can back it. Rebuilt if the signer changes. + async function transport(opts) { + const s = await signer(opts); + if (!s) return null; + const key = s.raw ? Array.from(s.raw.slice(0, 8)).join(',') : 'adapter'; + if (cvm && cvmSigner === key) return cvm; + if (cvm) cvm.stop(); + cvmSigner = key; + cvm = cvmClient({ server: coordinator.pubkey, relays: coordinator.relays, pubkey, sign: s.sign, decrypt: s.decrypt, + subClosed: () => { if (!stopped && subClosed) subClosed(); } }); + return cvm; + } + // The public coordinator is slow and sometimes silent (2 of 12 probes went + // unanswered in 15 s on 2026-10-06), so a call is tried twice. A repeated + // msg_post carries the same bytes, which every reader drops as a duplicate; + // kp_take is never repeated (a second take would spend a second package). + const CALL_MS = 12_000; + const call = async (name, args, opts) => { + const t = await transport(opts); + if (!t) throw new Error('no signer'); + const tries = name === 'kp_take' ? 1 : 2; + let err; + for (let i = 0; i < tries; i++) { + try { return await t.call(name, args, { timeoutMs: CALL_MS }); } + catch (e) { err = e; if (!/did not answer|no relay took/.test(e.message)) throw e; } + } + throw err; + }; + + // A device record and a current last-resort KeyPackage on the coordinator. + function ready(opts = {}) { + if (!kpBusy) kpBusy = ensure(opts).catch((e) => { dlog('cordn: keypackage', e); return false; }).finally(() => { kpBusy = null; }); + return kpBusy; + } + async function ensure(opts) { + await load(); + const cur = dev && dev.kps && dev.kps.find((k) => k.published && !k.spent); + if (cur && now() - cur.published < KP_REFRESH) return true; + const t = await transport(opts); + if (!t) return false; + if (!dev) { dev = { sig: C.newSigner(), kps: [] }; await saveDev(); } + return publishKeyPackage(t); + } + async function publishKeyPackage(t) { + const kp = C.newKeyPackage(dev.sig, pubkey); + // the private half is on disk before the public half is findable + dev.kps.push(kp); + await saveDev(); + try { + await call('kp_publish', { kp_ref: kp.ref, kp_64: C.keyPackageBase64(kp) }); + } catch (e) { + dev.kps = dev.kps.filter((k) => k !== kp); + await saveDev(); + throw e; + } + kp.published = now(); + // older packages give way; the ones a Welcome may still name stay a while + const old = dev.kps.filter((k) => k !== kp && k.published); + for (const k of old) if (now() - k.at > 2 * DAY) { dev.kps = dev.kps.filter((x) => x !== k); call('kp_remove', { kp_refs: [k.ref] }).catch(() => {}); } + await saveDev(); + return true; + } + + // Can this person be invited? Their KeyPackage on the coordinator, or null. + const reach = new Map(); // pk -> { at, p } + function lookup(pk) { + const hit = reach.get(pk); + if (hit && Date.now() - hit.at < 5 * 60_000) return hit.p; + const p = (async () => { + const r = await call('kp_list', {}); + const mine = (r.keyPackages || []).filter((k) => k.pk === pk); + return mine.length ? { pubkey: pk, refs: mine.map((k) => k.kp_ref) } : null; + })().catch(() => null); + reach.set(pk, { at: Date.now(), p }); + return p; + } + // Take their package off the coordinator for an invitation (a regular one + // is consumed by this; a last-resort one stays). + async function takeKeyPackage(pk) { + const r = await call('kp_take', { id: pk }, { interactive: true }); + if (!r.keyPackage) { const e = new Error('not reachable'); e.missing = [pk]; throw e; } + return C.readKeyPackage(r.keyPackage.event ? keyPackageOfEvent(r.keyPackage.event) : r.keyPackage.kp_64, pk); + } + // The coordinator serves the publication event verbatim; the package is its kp_64 argument. + function keyPackageOfEvent(ev) { + try { return JSON.parse(ev.content).params.arguments.kp_64; } catch { throw new Error('bad key package record'); } + } + + // ---- the stream ---- + + function heard(g, out) { + for (const o of out) { + if (o.id) emit('message', g, o); + else if (o.epoch !== undefined) emit('state', g); + else if (o.removed) emit('removed', g); + else if (o.invalid) dlog('cordn: skipped record', g.id, o.invalid); + } + } + + // Welcomes waiting for us, then every live group's records past its cursor. + async function poll() { + if (polling) return polling; + polling = (async () => { + const t = await transport(); + if (!t || stopped) return; + await load(); + try { + const w = await call('welcome_take', {}); + for (const welcome of w.welcomes || []) await acceptWelcome(welcome).catch((e) => dlog('cordn: welcome', e)); + } catch (e) { dlog('cordn: welcomes', e); } + const live = [...groups.values()].filter((g) => !g.removed); + if (!live.length || stopped) return; + let r; + try { r = await call('msg_fetch_many', { groups: live.map((g) => ({ gid: g.id, ...(g.cursor ? { after: g.cursor } : {}) })) }); } catch (e) { dlog('cordn: fetch', e.message); return; } + const byGid = new Map(); + for (const m of r.messages || []) (byGid.get(m.gid) || byGid.set(m.gid, []).get(m.gid)).push(m); + for (const [gid, recs] of byGid) { + const g = groups.get(gid); + if (!g) continue; + recs.sort((a, b) => a.cursor - b.cursor); + for (const rec of recs) { + let out; + try { out = C.receive(g, rec); } catch (e) { dlog('cordn: receive', e); continue; } + heard(g, out); + } + touch(g); + } + })().catch((e) => dlog('cordn: poll', e)).finally(() => { polling = null; schedule(); }); + return polling; + } + function schedule() { + clearTimeout(pollTimer); + if (stopped) return; + const live = [...groups.values()].some((g) => !g.removed); + pollTimer = setTimeout(poll, focused ? POLL_FOCUSED : live ? POLL_IDLE : POLL_EMPTY); + } + // Which group is on screen: that one is read every few seconds. + function focus(gid) { + gid = gid || null; + if (gid === focused) return; + focused = gid; + if (focused) { clearTimeout(pollTimer); poll(); } else schedule(); + } + + async function acceptWelcome(welcome) { + if (!dev) return null; + let j; + try { j = C.join(dev.sig, pubkey, welcome, dev.kps, { cursor: welcome.after || 0 }); } catch (e) { dlog('cordn: welcome', e.message); return null; } + const old = groups.get(j.g.id); + // a retained copy is never silently replaced; a group we were removed from can be rejoined + if (old && !old.removed) return null; + if (old) j.g.log = old.log; + groups.set(j.g.id, j.g); + await persist(j.g); + emit('joined', j.g); + // A regular package that got us in is spent; a last-resort one is not, + // but its keys have been seen: rotate either, a little later (more + // Welcomes may name it while we were away). + if (j.used) { j.used.spent = now(); await saveDev(); } + clearTimeout(rotateTimer); + rotateTimer = setTimeout(() => { if (!stopped) transport().then((t) => t && publishKeyPackage(t)).catch(() => {}); }, 45_000); + return j.g; + } + + // ---- changes of ours ---- + + // Post-then-confirm: the commit is group state once the coordinator has + // it (its own cursor says so) and no peer's commit got in first. The + // stream is read up to our record so a race is settled in order. + async function commit(g, proposals, { interactive = true } = {}) { + const t = await transport({ interactive }); + if (!t) throw new Error('no signer'); + await poll().catch(() => {}); + if (g.removed) throw new Error('not in this group'); + const p = C.stage(g, proposals); + await persist(g); + let posted; + try { posted = await call('msg_post', { gid: g.id, msg_64: p.msg_64 }, { interactive }); } + catch (e) { C.drop(g); await persist(g); throw e; } + await settle(g, t, posted.cursor); + if (g.pending === p) { C.drop(g); await persist(g); throw new Error('the coordinator did not echo the change'); } + if (g.tip !== p.state) throw new Error('another change got there first'); + return p; + } + // Read the stream up to a cursor of ours, so what happened before it is applied first. + async function settle(g, t, cursor) { + for (let i = 0; i < 4 && g.cursor < cursor && !stopped; i++) { + if (i) await sleep(500 * i); + let r; + try { r = await call('msg_fetch_many', { groups: [{ gid: g.id, ...(g.cursor ? { after: g.cursor } : {}) }] }); } catch { continue; } + const recs = (r.messages || []).filter((m) => m.gid === g.id).sort((a, b) => a.cursor - b.cursor); + for (const rec of recs) { let out; try { out = C.receive(g, rec); } catch { continue; } heard(g, out); } + } + await persist(g); + } + + async function deliverWelcomes(g, t, welcome, cands, after) { + if (!welcome) return; + const welcome_64 = C.welcomeBase64(welcome); + for (const c of cands) { + try { await call('welcome_store', { target_pk: c.pubkey, kp_ref: c.ref, welcome_64, ...(after ? { after } : {}) }, { interactive: true }); } + catch (e) { dlog('cordn: welcome_store', e); } + } + } + + async function candidates(pks) { + const out = []; + const missing = []; + for (const pk of pks) { try { out.push(await takeKeyPackage(pk)); } catch (e) { if (e.missing) missing.push(pk); else throw e; } } + if (missing.length) { const e = new Error('not reachable'); e.missing = missing; throw e; } + return out; + } + + async function create({ name = '', icon = '', members = [] }) { + if (!(await ready({ interactive: true }))) throw new Error('no signer'); + const t = await transport({ interactive: true }); + const cands = await candidates(members); + const { g, pending } = C.found(dev.sig, pubkey, { name, icon, invitees: cands }); + groups.set(g.id, g); + await persist(g); + if (pending) { + let posted; + try { posted = await call('msg_post', { gid: g.id, msg_64: pending.msg_64 }, { interactive: true }); } + catch (e) { groups.delete(g.id); store.del('g:' + g.id).catch(() => {}); throw e; } + // nobody else is in the group yet: the first commit cannot be raced + g.cursor = posted.cursor; + const out = C.confirm(g); + await persist(g); + heard(g, out); + await deliverWelcomes(g, t, pending.welcome, cands, posted.cursor); + } + emit('state', g); + focus(g.id); + return g; + } + + // `inner` comes from event(): the app shows it before it has left. + async function send(g, inner) { + const t = await transport({ interactive: true }); + if (!t) throw new Error('no signer'); + const { msg_64 } = C.send(g, inner); + await persist(g); + try { await call('msg_post', { gid: g.id, msg_64 }, { interactive: true }); return true; } + catch (e) { dlog('cordn: post', e); g.log = g.log.filter((m) => m.id !== inner.id); touch(g); return false; } + } + + async function add(g, pks) { + const cands = await candidates(pks); + const p = await commit(g, C.invite(cands)); + await deliverWelcomes(g, await transport({ interactive: true }), p.welcome, cands, g.cursor); + } + const kick = (g, pk) => commit(g, C.remove(g, pk)); + const rename = (g, name) => commit(g, [C.setMetadata(g, { name })]); + + // Cordn has no self-removal yet (their client's "leave" only forgets the + // group locally); ours does the same, and a member can still remove us. + async function leave(g) { forget(g); } + function forget(g) { + groups.delete(g.id); + dirty.delete(g.id); + store.del('g:' + g.id).catch(() => {}); + if (focused === g.id) focus(null); + emit('gone', g); + } + function accept(g) { g.fresh = false; touch(g); emit('state', g); } + + async function start() { + await load(); + if (stopped) return; + emit('loaded', null); + poll(); + } + function resubscribe() { if (cvm) cvm.listen(); poll(); } + function stop() { + stopped = true; + clearTimeout(pollTimer); clearTimeout(rotateTimer); + if (cvm) cvm.stop(); + flush(); + } + + return { + pubkey, groups, load, start, stop, resubscribe, ready, lookup, accept, focus, poll, + create, send, add, kick, rename, leave, forget, + event: (kind, content, tags) => C.appEvent(pubkey, kind, content, tags), + view: (g) => C.groupView(g.tip), + peer: () => null, + accounts: (g) => C.accounts(g.tip), + isAdmin: (g) => C.isAdmin(g) && !g.removed, + groupRef: (g) => C.encodeGroupRef({ gid: g.id, coordinator: coordinator.pubkey, relays: coordinator.relays }), + isSetUp: () => !!dev, + coordinator, + }; +} -- cgit v1.2.3