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'
69interface Connection {
70 peer: Peer
71 connectedAt: number | null // self-reported timestamp from the remote peer
72}
74export class Network {
75 private nostr = new Nostr()
76 private root = ''
77 private connections = new Map<string, Connection>()
79 private connectedAt = Date.now()
80 private state: AppState = initialState()
81 private version = 0
83 private snapshot!: Snapshot
84 private listeners = new Set<() => void>()
86 constructor() {
87 this.rebuildSnapshot()
88 void this.start()
90 window.addEventListener('online', () => {
91 // Regaining a connection counts as a reconnect: new timestamp.
92 this.connectedAt = Date.now()
93 this.broadcastHello()
94 this.recompute()
95 })
96 }
98 private async start() {
99 this.root = await rootTopic(ROOM_ID)
101 // Receive WebRTC signaling addressed to us.
102 const selfSignalTopic = await peerTopic(this.root, selfId)
103 this.nostr.subscribe(selfSignalTopic, (content, from) => {
104 if (from === selfId) return
105 let signal: Signal
106 try {
107 signal = JSON.parse(content)
108 } catch {
109 return
110 }
111 this.handleSignal(from, signal)
112 })
114 // Discover peers via announcements on the root topic.
115 this.nostr.subscribe(this.root, (content, from) => {
116 if (from === selfId) return
117 let ann: {peerId?: string}
118 try {
119 ann = JSON.parse(content)
120 } catch {
121 return
122 }
123 if (ann.peerId && ann.peerId === from) this.maybeConnect(from)
124 })
126 const announce = () =>
127 void this.nostr.publish(this.root, JSON.stringify({peerId: selfId}))
128 announce()
129 setInterval(announce, ANNOUNCE_INTERVAL_MS)
130 }
132 // ---- connection setup -------------------------------------------------
134 private maybeConnect(peerId: string) {
135 if (peerId === selfId || this.connections.has(peerId)) return
136 // Deterministic initiator: the peer with the smaller ID makes the offer.
137 const initiator = selfId < peerId
138 this.createPeer(peerId, initiator)
139 }
141 private createPeer(peerId: string, initiator: boolean): Connection {
142 const peer = new Peer(initiator)
143 const conn: Connection = {peer, connectedAt: null}
144 this.connections.set(peerId, conn)
146 peer.setHandlers({
147 signal: signal => {
148 void this.sendSignal(peerId, signal)
149 },
150 connect: () => {
151 // Tell the new peer our self-reported connect time.
152 void this.sendTo(peerId, {t: 'hello', connectedAt: this.connectedAt})
153 // If we're central, sync the newcomer immediately.
154 if (this.amCentral()) void this.broadcastState()
155 this.recompute()
156 },
157 data: raw => void this.handleData(peerId, raw),
158 close: () => {
159 if (this.connections.get(peerId)?.peer === peer) {
160 this.connections.delete(peerId)
161 this.recompute()
162 }
163 }
164 })
166 return conn
167 }
169 private async sendSignal(peerId: string, signal: Signal) {
170 const topic = await peerTopic(this.root, peerId)
171 void this.nostr.publish(topic, JSON.stringify(signal))
172 }
174 private handleSignal(from: string, signal: Signal) {
175 let conn = this.connections.get(from)
176 if (!conn) {
177 if (signal.type !== 'offer') return // nothing to attach it to yet
178 conn = this.createPeer(from, false)
179 }
180 void conn.peer.signal(signal)
181 }
183 // ---- messaging --------------------------------------------------------
185 private async sendTo(peerId: string, msg: Message) {
186 const conn = this.connections.get(peerId)
187 if (!conn) return
188 const data = JSON.stringify(msg)
189 const sig = await sign(data)
190 const env: Envelope = {data, from: selfId, sig}
191 conn.peer.send(JSON.stringify(env))
192 }
194 private async broadcast(msg: Message) {
195 const data = JSON.stringify(msg)
196 const sig = await sign(data)
197 const env: Envelope = {data, from: selfId, sig}
198 const payload = JSON.stringify(env)
199 for (const conn of this.connections.values()) conn.peer.send(payload)
200 }
202 private broadcastHello() {
203 void this.broadcast({t: 'hello', connectedAt: this.connectedAt})
204 }
206 private async broadcastState() {
207 await this.broadcast({t: 'state', state: this.state, version: this.version})
208 }
210 private async handleData(from: string, raw: string) {
211 let env: Envelope
212 try {
213 env = JSON.parse(raw)
214 } catch {
215 return
216 }
217 // The envelope must be signed by the peer we received it from.
218 if (env.from !== from) return
219 if (!(await verify(env.data, env.sig, env.from))) return
221 let msg: Message
222 try {
223 msg = JSON.parse(env.data)
224 } catch {
225 return
226 }
228 switch (msg.t) {
229 case 'hello': {
230 const conn = this.connections.get(from)
231 if (conn) {
232 conn.connectedAt = msg.connectedAt
233 this.recompute()
234 }
235 break
236 }
237 case 'command': {
238 if (this.amCentral()) {
239 this.state = applyCommand(this.state, msg.cmd)
240 this.version++
241 await this.broadcastState()
242 this.rebuildSnapshot()
243 } else if (!msg.forwarded) {
244 // Not central; forward once toward the central peer.
245 const central = this.centralId()
246 if (central && this.connections.has(central)) {
247 void this.sendTo(central, {...msg, forwarded: true})
248 }
249 }
250 break
251 }
252 case 'state': {
253 // Only trust state from the current central peer.
254 if (from === this.centralId()) {
255 this.state = msg.state
256 this.version = msg.version
257 this.rebuildSnapshot()
258 }
259 break
260 }
261 }
262 }
264 // ---- central-peer election -------------------------------------------
266 /** All peers we know about, with a reported connect time, plus ourselves. */
267 private participants(): {peerId: string; connectedAt: number}[] {
268 const list = [{peerId: selfId, connectedAt: this.connectedAt}]
269 for (const [peerId, conn] of this.connections) {
270 if (conn.peer.isConnected && conn.connectedAt !== null) {
271 list.push({peerId, connectedAt: conn.connectedAt})
272 }
273 }
274 return list
275 }
277 /** The oldest peer (smallest connect time; ties broken by peer ID) is central. */
278 private centralId(): string | null {
279 const list = this.participants()
280 if (list.length === 0) return null
281 return list.reduce((oldest, p) =>
282 p.connectedAt < oldest.connectedAt ||
283 (p.connectedAt === oldest.connectedAt && p.peerId < oldest.peerId)
284 ? p
285 : oldest
286 ).peerId
287 }
289 private amCentral(): boolean {
290 return this.centralId() === selfId
291 }
293 private recompute() {
294 const wasCentral = this.snapshot.amCentral
295 this.rebuildSnapshot()
296 // If we just became central, our last-known state is now the source of
297 // truth — push it to everyone.
298 if (!wasCentral && this.snapshot.amCentral) void this.broadcastState()
299 }
301 // ---- public API -------------------------------------------------------
303 dispatch(cmd: Command) {
304 if (this.amCentral()) {
305 this.state = applyCommand(this.state, cmd)
306 this.version++
307 void this.broadcastState()
308 this.rebuildSnapshot()
309 } else {
310 const central = this.centralId()
311 if (central && this.connections.has(central)) {
312 void this.sendTo(central, {t: 'command', cmd})
313 }
314 }
315 }
317 getSnapshot = (): Snapshot => this.snapshot
319 subscribe = (listener: () => void): (() => void) => {
320 this.listeners.add(listener)
321 return () => this.listeners.delete(listener)
322 }
324 private rebuildSnapshot() {
325 const centralId = this.centralId()
326 const roster: RosterEntry[] = this.participants()
327 .map(p => ({
328 peerId: p.peerId,
329 connectedAt: p.connectedAt,
330 isSelf: p.peerId === selfId,
331 isCentral: p.peerId === centralId
332 }))
333 .sort((a, b) => a.connectedAt - b.connectedAt)
335 this.snapshot = {
336 selfId,
337 connectedAt: this.connectedAt,
338 centralId,
339 amCentral: centralId === selfId,
340 roster,
341 state: this.state,
342 version: this.version
343 }
344 for (const l of this.listeners) l()
345 }
346}