4a0be26Initial commit: P2P counter app (nostr discovery + WebRTC mesh)Jeremy Magland 1import {selfId, sign, verify} from './identity'
2import {Nostr, peerTopic, rootTopic} from './nostr'
3import {Peer, type Signal} from './peer'
5// ---------------------------------------------------------------------------
6// Application state + commands. For this first version the shared state is just
7// a single counter. Commands are dispatched to the central peer, which applies
8// them to the authoritative state and broadcasts the result.
9// ---------------------------------------------------------------------------
11export interface AppState {
12 counter: number
13}
15export type Command = {op: 'increment'} | {op: 'decrement'}
17const initialState = (): AppState => ({counter: 0})
19const applyCommand = (state: AppState, cmd: Command): AppState => {
20 switch (cmd.op) {
21 case 'increment':
22 return {counter: state.counter + 1}
23 case 'decrement':
24 return {counter: state.counter - 1}
25 default:
26 return state
27 }
28}
30// ---------------------------------------------------------------------------
31// Wire protocol (over the WebRTC data channel). Every message is a signed
32// envelope: `data` is the exact JSON string that was signed, `from` is the
33// sender's peer ID (public key), and `sig` is the schnorr signature.
34// ---------------------------------------------------------------------------
36type Message =
37 | {t: 'hello'; connectedAt: number}
38 | {t: 'command'; cmd: Command; forwarded?: boolean}
39 | {t: 'state'; state: AppState; version: number}
41interface Envelope {
42 data: string
43 from: string
44 sig: string
45}
47// ---------------------------------------------------------------------------
49export interface RosterEntry {
50 peerId: string
51 connectedAt: number
52 isSelf: boolean
53 isCentral: boolean
54}
56export interface Snapshot {
57 selfId: string
58 connectedAt: number
59 centralId: string | null
60 amCentral: boolean
61 roster: RosterEntry[]
62 state: AppState
63 version: number
64}
66const ANNOUNCE_INTERVAL_MS = 5000
67const ROOM_ID = 'default'
5263936p2p: retry stalled connection attemptsJeremy Magland 68// A connection attempt that hasn't opened after this long is torn down and
69// retried on the peer's next announcement. Signaling events are ephemeral, so
70// an offer published before the other side was listening is simply lost —
71// without a retry the pair would deadlock forever.
72const CONNECT_RETRY_MS = 15000
74interface Connection {
75 peer: Peer
5263936p2p: retry stalled connection attemptsJeremy Magland 76 /** When this connection attempt started (local clock), for retry pacing. */
77 createdAt: number
4a0be26Initial commit: P2P counter app (nostr discovery + WebRTC mesh)Jeremy Magland 78 connectedAt: number | null // self-reported timestamp from the remote peer
79}
81export class Network {
82 private nostr = new Nostr()
83 private root = ''
84 private connections = new Map<string, Connection>()
86 private connectedAt = Date.now()
87 private state: AppState = initialState()
88 private version = 0
90 private snapshot!: Snapshot
91 private listeners = new Set<() => void>()
93 constructor() {
94 this.rebuildSnapshot()
95 void this.start()
97 window.addEventListener('online', () => {
98 // Regaining a connection counts as a reconnect: new timestamp.
99 this.connectedAt = Date.now()
100 this.broadcastHello()
101 this.recompute()
102 })
103 }
105 private async start() {
106 this.root = await rootTopic(ROOM_ID)
108 // Receive WebRTC signaling addressed to us.
109 const selfSignalTopic = await peerTopic(this.root, selfId)
110 this.nostr.subscribe(selfSignalTopic, (content, from) => {
111 if (from === selfId) return
112 let signal: Signal
113 try {
114 signal = JSON.parse(content)
115 } catch {
116 return
117 }
118 this.handleSignal(from, signal)
119 })
121 // Discover peers via announcements on the root topic.
122 this.nostr.subscribe(this.root, (content, from) => {
123 if (from === selfId) return
124 let ann: {peerId?: string}
125 try {
126 ann = JSON.parse(content)
127 } catch {
128 return
129 }
130 if (ann.peerId && ann.peerId === from) this.maybeConnect(from)
131 })
133 const announce = () =>
134 void this.nostr.publish(this.root, JSON.stringify({peerId: selfId}))
135 announce()
136 setInterval(announce, ANNOUNCE_INTERVAL_MS)
137 }
139 // ---- connection setup -------------------------------------------------
141 private maybeConnect(peerId: string) {
143 const existing = this.connections.get(peerId)
144 if (existing) {
145 const stalled =
146 !existing.peer.isConnected &&
147 Date.now() - existing.createdAt > CONNECT_RETRY_MS
148 if (!stalled) return
149 // destroy() fires the close handler, which removes it from the map.
150 existing.peer.destroy()
151 this.connections.delete(peerId)
152 }
4a0be26Initial commit: P2P counter app (nostr discovery + WebRTC mesh)Jeremy Magland 153 // Deterministic initiator: the peer with the smaller ID makes the offer.
154 const initiator = selfId < peerId
155 this.createPeer(peerId, initiator)
156 }
158 private createPeer(peerId: string, initiator: boolean): Connection {
159 const peer = new Peer(initiator)
5263936p2p: retry stalled connection attemptsJeremy Magland 160 const conn: Connection = {peer, createdAt: Date.now(), connectedAt: null}
4a0be26Initial commit: P2P counter app (nostr discovery + WebRTC mesh)Jeremy Magland 161 this.connections.set(peerId, conn)
163 peer.setHandlers({
164 signal: signal => {
165 void this.sendSignal(peerId, signal)
166 },
167 connect: () => {
168 // Tell the new peer our self-reported connect time.
169 void this.sendTo(peerId, {t: 'hello', connectedAt: this.connectedAt})
170 // If we're central, sync the newcomer immediately.
171 if (this.amCentral()) void this.broadcastState()
172 this.recompute()
173 },
174 data: raw => void this.handleData(peerId, raw),
175 close: () => {
176 if (this.connections.get(peerId)?.peer === peer) {
177 this.connections.delete(peerId)
178 this.recompute()
179 }
180 }
181 })
183 return conn
184 }
186 private async sendSignal(peerId: string, signal: Signal) {
187 const topic = await peerTopic(this.root, peerId)
188 void this.nostr.publish(topic, JSON.stringify(signal))
189 }
191 private handleSignal(from: string, signal: Signal) {
192 let conn = this.connections.get(from)
193 if (!conn) {
194 if (signal.type !== 'offer') return // nothing to attach it to yet
195 conn = this.createPeer(from, false)
196 }
197 void conn.peer.signal(signal)
198 }
200 // ---- messaging --------------------------------------------------------
202 private async sendTo(peerId: string, msg: Message) {
203 const conn = this.connections.get(peerId)
204 if (!conn) return
205 const data = JSON.stringify(msg)
206 const sig = await sign(data)
207 const env: Envelope = {data, from: selfId, sig}
208 conn.peer.send(JSON.stringify(env))
209 }
211 private async broadcast(msg: Message) {
212 const data = JSON.stringify(msg)
213 const sig = await sign(data)
214 const env: Envelope = {data, from: selfId, sig}
215 const payload = JSON.stringify(env)
216 for (const conn of this.connections.values()) conn.peer.send(payload)
217 }
219 private broadcastHello() {
220 void this.broadcast({t: 'hello', connectedAt: this.connectedAt})
221 }
223 private async broadcastState() {
224 await this.broadcast({t: 'state', state: this.state, version: this.version})
225 }
227 private async handleData(from: string, raw: string) {
228 let env: Envelope
229 try {
230 env = JSON.parse(raw)
231 } catch {
232 return
233 }
234 // The envelope must be signed by the peer we received it from.
235 if (env.from !== from) return
236 if (!(await verify(env.data, env.sig, env.from))) return
238 let msg: Message
239 try {
240 msg = JSON.parse(env.data)
241 } catch {
242 return
243 }
245 switch (msg.t) {
246 case 'hello': {
247 const conn = this.connections.get(from)
248 if (conn) {
249 conn.connectedAt = msg.connectedAt
250 this.recompute()
251 }
252 break
253 }
254 case 'command': {
255 if (this.amCentral()) {
256 this.state = applyCommand(this.state, msg.cmd)
257 this.version++
258 await this.broadcastState()
259 this.rebuildSnapshot()
260 } else if (!msg.forwarded) {
261 // Not central; forward once toward the central peer.
262 const central = this.centralId()
263 if (central && this.connections.has(central)) {
264 void this.sendTo(central, {...msg, forwarded: true})
265 }
266 }
267 break
268 }
269 case 'state': {
270 // Only trust state from the current central peer.
271 if (from === this.centralId()) {
272 this.state = msg.state
273 this.version = msg.version
274 this.rebuildSnapshot()
275 }
276 break
277 }
278 }
279 }
281 // ---- central-peer election -------------------------------------------
283 /** All peers we know about, with a reported connect time, plus ourselves. */
284 private participants(): {peerId: string; connectedAt: number}[] {
285 const list = [{peerId: selfId, connectedAt: this.connectedAt}]
286 for (const [peerId, conn] of this.connections) {
287 if (conn.peer.isConnected && conn.connectedAt !== null) {
288 list.push({peerId, connectedAt: conn.connectedAt})
289 }
290 }
291 return list
292 }
294 /** The oldest peer (smallest connect time; ties broken by peer ID) is central. */
295 private centralId(): string | null {
296 const list = this.participants()
297 if (list.length === 0) return null
298 return list.reduce((oldest, p) =>
299 p.connectedAt < oldest.connectedAt ||
300 (p.connectedAt === oldest.connectedAt && p.peerId < oldest.peerId)
301 ? p
302 : oldest
303 ).peerId
304 }
306 private amCentral(): boolean {
307 return this.centralId() === selfId
308 }
310 private recompute() {
311 const wasCentral = this.snapshot.amCentral
312 this.rebuildSnapshot()
313 // If we just became central, our last-known state is now the source of
314 // truth — push it to everyone.
315 if (!wasCentral && this.snapshot.amCentral) void this.broadcastState()
316 }
318 // ---- public API -------------------------------------------------------
320 dispatch(cmd: Command) {
321 if (this.amCentral()) {
322 this.state = applyCommand(this.state, cmd)
323 this.version++
324 void this.broadcastState()
325 this.rebuildSnapshot()
326 } else {
327 const central = this.centralId()
328 if (central && this.connections.has(central)) {
329 void this.sendTo(central, {t: 'command', cmd})
330 }
331 }
332 }
334 getSnapshot = (): Snapshot => this.snapshot
336 subscribe = (listener: () => void): (() => void) => {
337 this.listeners.add(listener)
338 return () => this.listeners.delete(listener)
339 }
341 private rebuildSnapshot() {
342 const centralId = this.centralId()
343 const roster: RosterEntry[] = this.participants()
344 .map(p => ({
345 peerId: p.peerId,
346 connectedAt: p.connectedAt,
347 isSelf: p.peerId === selfId,
348 isCentral: p.peerId === centralId
349 }))
350 .sort((a, b) => a.connectedAt - b.connectedAt)
352 this.snapshot = {
353 selfId,
354 connectedAt: this.connectedAt,
355 centralId,
356 amCentral: centralId === selfId,
357 roster,
358 state: this.state,
359 version: this.version
360 }
361 for (const l of this.listeners) l()
362 }
363}