1import {selfId, sign, verify, sha256HexBytes} from './identity'
2import {Nostr, peerTopic, rootTopic} from './nostr'
3import {Peer, type Signal} from './peer'
4import {BlobReceiver, decodeSamples, encodeSamples} from './blob'
5import type {Params, Points} from '../types'
7// ---------------------------------------------------------------------------
8// Shared state. Everyone sees the same figure: the parameter selections, the
9// region, and the samples. Only the CENTRAL peer (oldest in the room) runs the
10// numbl engine; commands from any viewer are forwarded to it, it computes, and
11// it broadcasts the result.
12//
13// The state travels in two parts:
14// - the "view" (params/region/status/movie): a small JSON message, broadcast
15// on every change;
16// - the samples: a Float32 blob (up to ~800 KB at n=100k), streamed in
17// chunks and announced by a signed header carrying its SHA-256. The view
18// references the blob by `samplesId`.
19// Per-connection sends are serialized (a promise chain), so on the ordered
20// data channel a peer always receives a blob before the view that points at
21// it.
22// ---------------------------------------------------------------------------
24export type EngineStatus = 'none' | 'starting' | 'ready' | 'error'
26export interface ViewState {
27 params: Params
28 region: Points | null
29 /** The central peer is computing new samples. */
30 busy: boolean
31 /** Status of the numbl engine on the central peer. */
32 engine: EngineStatus
33 engineError: string | null
34 /** Shared sampling movie: index of the point being sampled, null = off. */
35 movieStep: number | null
36 /** Identifies the sample blob this view belongs to. */
37 samplesId: number
38 samplesN: number
39}
41export type Command =
42 | {op: 'resample'; n: number; local: boolean}
43 | {op: 'newRegion'; n: number; convex: boolean; local: boolean}
44 | {op: 'movie'; play: boolean}
46const initialView = (): ViewState => ({
47 params: {n: 10000, convex: true, local: false},
48 region: null,
49 busy: false,
50 engine: 'none',
51 engineError: null,
52 movieStep: null,
53 samplesId: 0,
54 samplesN: 0
55})
57/** What the network needs from the numbl engine (implemented in ../engine).
58 * Kept abstract so this layer stays independent of the runtime. */
59export interface EngineLike {
60 start(): Promise<{region: Points; samples: Points; n: number; convex: boolean}>
61 resample(req: {params: Params; region: Points}): Promise<{samples: Points; n: number}>
62 newRegion(req: {params: Params}): Promise<{region: Points; samples: Points; n: number; convex: boolean}>
63 dispose(): void
64}
66// ---------------------------------------------------------------------------
67// Wire protocol (over the WebRTC data channel). Every JSON message is a signed
68// envelope; binary frames are the chunks of the blob most recently announced
69// by a 'blob' header on the same channel (integrity via the header's SHA-256).
70// ---------------------------------------------------------------------------
72type Message =
73 | {t: 'hello'; connectedAt: number; engineFailed: boolean}
74 | {t: 'command'; cmd: Command; forwarded?: boolean}
75 | {t: 'view'; view: ViewState; version: number}
76 | {t: 'blob'; id: number; bytes: number; hash: string}
78interface Envelope {
79 data: string
80 from: string
81 sig: string
82}
84// ---------------------------------------------------------------------------
86export interface RosterEntry {
87 peerId: string
88 connectedAt: number
89 isSelf: boolean
90 isCentral: boolean
91 engineFailed: boolean
92}
94export interface Snapshot {
95 selfId: string
96 connectedAt: number
97 centralId: string | null
98 amCentral: boolean
99 roster: RosterEntry[]
100 view: ViewState
101 version: number
102 samples: Points | null
103 /** True when `samples` is the set the current view refers to. */
104 samplesSynced: boolean
105}
107const ANNOUNCE_INTERVAL_MS = 5000
108const ROOM_ID = 'default'
109// A connection attempt that hasn't opened after this long is torn down and
110// retried on the peer's next announcement. Signaling events are ephemeral, so
111// an offer published before the other side was listening is simply lost —
112// without a retry the pair would deadlock forever.
113const CONNECT_RETRY_MS = 15000
114// Wait for discovery before assuming we're the first (and therefore central)
115// peer, to avoid booting an engine just to resign seconds later.
116const INITIAL_ELECTION_DELAY_MS = 4000
118// Shared sampling movie (mirrors the original figure's constants).
119const MOVIE_STEP_MS = 750
120const MOVIE_LAST_INDEX = 41
122interface Connection {
123 peer: Peer
124 /** When this connection attempt started (local clock), for retry pacing. */
125 createdAt: number
126 connectedAt: number | null // self-reported timestamp from the remote peer
127 engineFailed: boolean
128 /** Serializes everything we send on this channel (JSON + blob chunks). */
129 sendChain: Promise<void>
130 /** Serializes inbound processing too: envelope verification is async, and a
131 * blob header must finish verifying (arming `recv`) before the binary
132 * chunks right behind it are handled. */
133 recvChain: Promise<void>
134 /** Blob transfer in progress from this peer, if any. */
135 recv: BlobReceiver | null
136}
138export class Network {
139 private nostr = new Nostr()
140 private root = ''
141 private connections = new Map<string, Connection>()
143 private connectedAt = Date.now()
144 private view: ViewState = initialView()
145 private version = 0
147 // The current sample set (kept by every peer, so any of them can serve it
148 // if it becomes central).
149 private samples: Points | null = null
150 private samplesBuf: ArrayBuffer | null = null
151 private samplesHash: string | null = null
152 private samplesLocalId = 0
154 // Engine (central peer only).
155 private engine: EngineLike | null = null
156 private selfEngineFailed = false
157 private pendingCmd: Command | null = null
158 private computing = false
159 private movieTimer: ReturnType<typeof setInterval> | null = null
160 private electionArmed = false
162 private snapshot!: Snapshot
163 private listeners = new Set<() => void>()
165 constructor(private engineFactory: () => EngineLike) {
166 this.rebuildSnapshot()
167 void this.start()
169 setTimeout(() => {
170 this.electionArmed = true
171 this.recompute()
172 }, INITIAL_ELECTION_DELAY_MS)
174 window.addEventListener('online', () => {
175 // Regaining a connection counts as a reconnect: new timestamp.
176 this.connectedAt = Date.now()
177 this.broadcast({
178 t: 'hello',
179 connectedAt: this.connectedAt,
180 engineFailed: this.selfEngineFailed
181 })
182 this.recompute()
183 })
184 }
186 private async start() {
187 this.root = await rootTopic(ROOM_ID)
189 // Receive WebRTC signaling addressed to us.
190 const selfSignalTopic = await peerTopic(this.root, selfId)
191 this.nostr.subscribe(selfSignalTopic, (content, from) => {
192 if (from === selfId) return
193 let signal: Signal
194 try {
195 signal = JSON.parse(content)
196 } catch {
197 return
198 }
199 this.handleSignal(from, signal)
200 })
202 // Discover peers via announcements on the root topic.
203 this.nostr.subscribe(this.root, (content, from) => {
204 if (from === selfId) return
205 let ann: {peerId?: string}
206 try {
207 ann = JSON.parse(content)
208 } catch {
209 return
210 }
211 if (ann.peerId && ann.peerId === from) this.maybeConnect(from)
212 })
214 const announce = () =>
215 void this.nostr.publish(this.root, JSON.stringify({peerId: selfId}))
216 announce()
217 setInterval(announce, ANNOUNCE_INTERVAL_MS)
218 }
220 // ---- connection setup -------------------------------------------------
222 private maybeConnect(peerId: string) {
223 if (peerId === selfId) return
224 const existing = this.connections.get(peerId)
225 if (existing) {
226 const stalled =
227 !existing.peer.isConnected &&
228 Date.now() - existing.createdAt > CONNECT_RETRY_MS
229 if (!stalled) return
230 // destroy() fires the close handler, which removes it from the map.
231 existing.peer.destroy()
232 this.connections.delete(peerId)
233 }
234 // Deterministic initiator: the peer with the smaller ID makes the offer.
235 const initiator = selfId < peerId
236 this.createPeer(peerId, initiator)
237 }
239 private createPeer(peerId: string, initiator: boolean): Connection {
240 const peer = new Peer(initiator)
241 const conn: Connection = {
242 peer,
243 createdAt: Date.now(),
244 connectedAt: null,
245 engineFailed: false,
246 sendChain: Promise.resolve(),
247 recvChain: Promise.resolve(),
248 recv: null
249 }
250 this.connections.set(peerId, conn)
252 peer.setHandlers({
253 signal: signal => {
254 void this.sendSignal(peerId, signal)
255 },
256 connect: () => {
257 // Tell the new peer our self-reported connect time; if we're central,
258 // sync it immediately (blob first, then the view that references it —
259 // the chain keeps that order on the wire).
260 this.sendTo(peerId, {
261 t: 'hello',
262 connectedAt: this.connectedAt,
263 engineFailed: this.selfEngineFailed
264 })
265 if (this.snapshot.amCentral) {
266 this.sendBlobTo(conn)
267 this.sendViewTo(conn)
268 }
269 this.recompute()
270 },
271 data: raw => {
272 conn.recvChain = conn.recvChain
273 .then(() => this.handleData(peerId, raw))
274 .catch(() => {})
275 },
276 binary: chunk => {
277 conn.recvChain = conn.recvChain
278 .then(() => this.handleBinary(peerId, chunk))
279 .catch(() => {})
280 },
281 close: () => {
282 if (this.connections.get(peerId)?.peer === peer) {
283 this.connections.delete(peerId)
284 this.recompute()
285 }
286 }
287 })
289 return conn
290 }
292 private async sendSignal(peerId: string, signal: Signal) {
293 const topic = await peerTopic(this.root, peerId)
294 void this.nostr.publish(topic, JSON.stringify(signal))
295 }
297 private handleSignal(from: string, signal: Signal) {
298 let conn = this.connections.get(from)
299 if (!conn) {
300 if (signal.type !== 'offer') return // nothing to attach it to yet
301 conn = this.createPeer(from, false)
302 }
303 void conn.peer.signal(signal)
304 }
306 // ---- sending ------------------------------------------------------------
308 private async envelope(msg: Message): Promise<string> {
309 const data = JSON.stringify(msg)
310 const sig = await sign(data)
311 const env: Envelope = {data, from: selfId, sig}
312 return JSON.stringify(env)
313 }
315 /** Queue a send task on a connection; tasks run strictly in order. */
316 private chain(conn: Connection, task: () => Promise<void> | void) {
317 conn.sendChain = conn.sendChain.then(task).catch(() => {})
318 }
320 private sendTo(peerId: string, msg: Message) {
321 const conn = this.connections.get(peerId)
322 if (!conn) return
323 const payload = this.envelope(msg)
324 this.chain(conn, async () => conn.peer.send(await payload))
325 }
327 private broadcast(msg: Message) {
328 const payload = this.envelope(msg) // sign once, share across connections
329 for (const conn of this.connections.values()) {
330 this.chain(conn, async () => conn.peer.send(await payload))
331 }
332 }
334 private broadcastView() {
335 this.version++
336 this.broadcast({t: 'view', view: this.view, version: this.version})
337 this.rebuildSnapshot()
338 }
340 private sendViewTo(conn: Connection) {
341 const payload = this.envelope({t: 'view', view: this.view, version: this.version})
342 this.chain(conn, async () => conn.peer.send(await payload))
343 }
345 /** Stream the current sample blob to one connection: signed header, then the
346 * raw chunks. */
347 private sendBlobTo(conn: Connection) {
348 const buf = this.samplesBuf
349 const hash = this.samplesHash
350 const id = this.samplesLocalId
351 if (!buf || !hash) return
352 const header = this.envelope({t: 'blob', id, bytes: buf.byteLength, hash})
353 this.chain(conn, async () => {
354 conn.peer.send(await header)
355 await conn.peer.sendBinary(buf)
356 })
357 }
359 /** Adopt a fresh sample set (central only) and stream it to everyone. The
360 * caller broadcasts the updated view afterwards. */
361 private async setSamples(samples: Points) {
362 this.samples = samples
363 this.samplesBuf = encodeSamples(samples)
364 this.samplesHash = await sha256HexBytes(this.samplesBuf)
365 this.samplesLocalId = ++this.view.samplesId
366 this.view.samplesN = Math.min(samples.x.length, samples.y.length)
367 for (const conn of this.connections.values()) this.sendBlobTo(conn)
368 }
370 // ---- receiving ----------------------------------------------------------
372 private async handleData(from: string, raw: string) {
373 let env: Envelope
374 try {
375 env = JSON.parse(raw)
376 } catch {
377 return
378 }
379 // The envelope must be signed by the peer we received it from.
380 if (env.from !== from) return
381 if (!(await verify(env.data, env.sig, env.from))) return
383 let msg: Message
384 try {
385 msg = JSON.parse(env.data)
386 } catch {
387 return
388 }
390 switch (msg.t) {
391 case 'hello': {
392 const conn = this.connections.get(from)
393 if (conn) {
394 conn.connectedAt = msg.connectedAt
395 conn.engineFailed = !!msg.engineFailed
396 this.recompute()
397 }
398 break
399 }
400 case 'command': {
401 if (this.snapshot.amCentral) {
402 this.acceptCommand(msg.cmd)
403 } else if (!msg.forwarded) {
404 // Not central; forward once toward the central peer.
405 const central = this.snapshot.centralId
406 if (central && this.connections.has(central)) {
407 this.sendTo(central, {...msg, forwarded: true})
408 }
409 }
410 break
411 }
412 case 'view': {
413 // Only trust state from the current central peer.
414 if (from === this.snapshot.centralId && !this.snapshot.amCentral) {
415 this.view = msg.view
416 this.version = msg.version
417 this.rebuildSnapshot()
418 }
419 break
420 }
421 case 'blob': {
422 if (from !== this.snapshot.centralId || this.snapshot.amCentral) break
423 const conn = this.connections.get(from)
424 if (conn) conn.recv = new BlobReceiver(msg.id, msg.bytes, msg.hash)
425 break
426 }
427 }
428 }
430 private handleBinary(from: string, chunk: ArrayBuffer) {
431 const conn = this.connections.get(from)
432 if (!conn?.recv) return
433 const buf = conn.recv.append(chunk)
434 if (!buf) return
435 const recv = conn.recv
436 conn.recv = null
437 void this.adoptBlob(recv, buf)
438 }
440 private async adoptBlob(recv: BlobReceiver, buf: ArrayBuffer) {
441 if ((await sha256HexBytes(buf)) !== recv.hash) return
442 if (this.samples && recv.id <= this.samplesLocalId) return // stale
443 this.samples = decodeSamples(buf)
444 this.samplesBuf = buf
445 this.samplesHash = recv.hash
446 this.samplesLocalId = recv.id
447 this.rebuildSnapshot()
448 }
450 // ---- central-peer election -------------------------------------------
452 /** All peers we know about, with a reported connect time, plus ourselves. */
453 private participants(): {
454 peerId: string
455 connectedAt: number
456 engineFailed: boolean
457 }[] {
458 const list = [
459 {peerId: selfId, connectedAt: this.connectedAt, engineFailed: this.selfEngineFailed}
460 ]
461 for (const [peerId, conn] of this.connections) {
462 if (conn.peer.isConnected && conn.connectedAt !== null) {
463 list.push({
464 peerId,
465 connectedAt: conn.connectedAt,
466 engineFailed: conn.engineFailed
467 })
468 }
469 }
470 return list
471 }
473 /** The oldest peer (smallest connect time; ties broken by peer ID) whose
474 * engine hasn't failed is central. If every engine failed, fall back to the
475 * oldest overall so the room still has an authority for its state. */
476 private centralId(): string | null {
477 const list = this.participants()
478 if (list.length === 0) return null
479 const healthy = list.filter(p => !p.engineFailed)
480 const pool = healthy.length > 0 ? healthy : list
481 return pool.reduce((oldest, p) =>
482 p.connectedAt < oldest.connectedAt ||
483 (p.connectedAt === oldest.connectedAt && p.peerId < oldest.peerId)
484 ? p
485 : oldest
486 ).peerId
487 }
489 private recompute() {
490 const wasCentral = this.snapshot?.amCentral ?? false
491 this.rebuildSnapshot()
492 if (!this.electionArmed) return
493 const isCentral = this.snapshot.amCentral
494 if (isCentral && !this.engine && !this.selfEngineFailed) {
495 void this.becomeCentral()
496 } else if (!isCentral && wasCentral) {
497 this.resignCentral()
498 }
499 }
501 // ---- central role: engine lifecycle ------------------------------------
503 private async becomeCentral() {
504 this.stopMovieTicker()
505 this.view = {
506 ...this.view,
507 busy: false,
508 movieStep: null,
509 engine: 'starting',
510 engineError: null
511 }
512 this.broadcastView()
514 const engine = this.engineFactory()
515 this.engine = engine
516 try {
517 const init = await engine.start()
518 if (this.engine !== engine) return // resigned while booting
519 const held = this.samples
520 if (!held || !this.view.region) {
521 // Fresh room — or we never received the previous central's samples
522 // (it died mid-transfer), in which case the inherited region can't be
523 // served. Adopt the script's initial region + samples.
524 this.view.params = {n: init.n, convex: init.convex, local: false}
525 this.view.region = init.region
526 await this.setSamples(init.samples)
527 } else {
528 // Reconcile the inherited view with the blob we actually hold (they
529 // can differ if the old central died between blob and view). Relabel
530 // our blob as the view's current one — ids must never regress.
531 this.samplesLocalId = this.view.samplesId
532 this.view.samplesN = Math.min(held.x.length, held.y.length)
533 }
534 this.view = {...this.view, engine: 'ready', engineError: null}
535 this.selfEngineFailed = false
536 this.broadcastView()
537 void this.processCompute()
538 } catch (err) {
539 if (this.engine !== engine) return
540 this.engineFailure(err instanceof Error ? err.message : String(err))
541 }
542 }
544 private resignCentral() {
545 this.stopMovieTicker()
546 this.engine?.dispose()
547 this.engine = null
548 this.pendingCmd = null
549 this.computing = false
550 // Our sample lineage is no longer authoritative; make sure the real
551 // central's next blob is accepted even if its ids overlap ours (e.g. two
552 // solo-started rooms merging). The samples stay visible until replaced.
553 this.samplesLocalId = 0
554 }
556 /** The engine could not boot or compute. Step down: announce the failure so
557 * the room elects the next-oldest healthy peer, whose last-known state
558 * becomes the source of truth. */
559 private engineFailure(message: string) {
560 this.engine?.dispose()
561 this.engine = null
562 this.pendingCmd = null
563 this.computing = false
564 this.stopMovieTicker()
565 this.selfEngineFailed = true
566 this.view = {
567 ...this.view,
568 busy: false,
569 movieStep: null,
570 engine: 'error',
571 engineError: message
572 }
573 // Still central at this instant, so receivers accept this view; the hello
574 // that follows triggers the re-election.
575 this.broadcastView()
576 this.broadcast({
577 t: 'hello',
578 connectedAt: this.connectedAt,
579 engineFailed: true
580 })
581 this.recompute()
582 }
584 // ---- central role: commands --------------------------------------------
586 private acceptCommand(cmd: Command) {
587 if (cmd.op === 'movie') {
588 this.handleMovie(cmd.play)
589 return
590 }
591 if (!this.engine || this.view.engine !== 'ready') return
592 // Latest wins: a newer request supersedes one still waiting its turn.
593 this.pendingCmd = cmd
594 void this.processCompute()
595 }
597 private async processCompute() {
598 if (this.computing) return
599 const cmd = this.pendingCmd
600 const engine = this.engine
601 if (!cmd || !engine || cmd.op === 'movie') return
602 this.pendingCmd = null
603 if (cmd.op === 'resample' && !this.view.region) return
605 this.computing = true
606 this.stopMovieTicker()
607 this.view = {...this.view, busy: true, movieStep: null}
608 this.broadcastView()
610 try {
611 if (cmd.op === 'resample') {
612 const params: Params = {...this.view.params, n: cmd.n, local: cmd.local}
613 const r = await engine.resample({params, region: this.view.region!})
614 if (this.engine !== engine) return
615 this.view.params = {...params, n: r.n}
616 await this.setSamples(r.samples)
617 } else {
618 const params: Params = {n: cmd.n, convex: cmd.convex, local: cmd.local}
619 const r = await engine.newRegion({params})
620 if (this.engine !== engine) return
621 this.view.params = {...params, n: r.n, convex: r.convex}
622 this.view.region = r.region
623 await this.setSamples(r.samples)
624 }
625 this.view = {...this.view, busy: false}
626 this.broadcastView()
627 } catch (err) {
628 if (this.engine === engine) {
629 this.engineFailure(err instanceof Error ? err.message : String(err))
630 }
631 return
632 } finally {
633 this.computing = false
634 }
635 if (this.pendingCmd) void this.processCompute()
636 }
638 // ---- central role: the shared movie --------------------------------------
640 private handleMovie(play: boolean) {
641 if (!play) {
642 this.stopMovieTicker()
643 if (this.view.movieStep !== null) {
644 this.view = {...this.view, movieStep: null}
645 this.broadcastView()
646 }
647 return
648 }
649 if (this.view.busy || this.view.movieStep !== null || this.view.samplesN < 3) {
650 return
651 }
652 const lastIndex = Math.min(this.view.samplesN - 1, MOVIE_LAST_INDEX)
653 this.view = {...this.view, movieStep: 2}
654 this.broadcastView()
655 this.movieTimer = setInterval(() => {
656 const next = (this.view.movieStep ?? lastIndex) + 1
657 if (next > lastIndex) {
658 this.stopMovieTicker()
659 this.view = {...this.view, movieStep: null}
660 } else {
661 this.view = {...this.view, movieStep: next}
662 }
663 this.broadcastView()
664 }, MOVIE_STEP_MS)
665 }
667 private stopMovieTicker() {
668 if (this.movieTimer !== null) clearInterval(this.movieTimer)
669 this.movieTimer = null
670 }
672 // ---- public API -------------------------------------------------------
674 dispatch(cmd: Command) {
675 if (this.snapshot.amCentral) {
676 this.acceptCommand(cmd)
677 } else {
678 const central = this.snapshot.centralId
679 if (central && this.connections.has(central)) {
680 this.sendTo(central, {t: 'command', cmd})
681 }
682 }
683 }
685 getSnapshot = (): Snapshot => this.snapshot
687 subscribe = (listener: () => void): (() => void) => {
688 this.listeners.add(listener)
689 return () => this.listeners.delete(listener)
690 }
692 private rebuildSnapshot() {
693 const centralId = this.centralId()
694 const roster: RosterEntry[] = this.participants()
695 .map(p => ({
696 peerId: p.peerId,
697 connectedAt: p.connectedAt,
698 isSelf: p.peerId === selfId,
699 isCentral: p.peerId === centralId,
700 engineFailed: p.engineFailed
701 }))
702 .sort((a, b) => a.connectedAt - b.connectedAt)
704 this.snapshot = {
705 selfId,
706 connectedAt: this.connectedAt,
707 centralId,
708 amCentral: centralId === selfId,
709 roster,
710 view: this.view,
711 version: this.version,
712 samples: this.samples,
713 samplesSynced: this.samples !== null && this.samplesLocalId === this.view.samplesId
714 }
715 for (const l of this.listeners) l()
716 }
717}