1import * as fs from 'node:fs'
2import * as path from 'node:path'
3import wrtc from '@roamhq/wrtc'
4import {selfId} from './identity.js'
5import {Nostr, peerTopic, roomTopic} from './nostr.js'
6import {Peer, type Signal} from './peer.js'
7import {WavWriter} from './wav.js'
9// The recorder's network layer: commonroom's protocol (presence announcements,
10// per-peer signaling topics, deterministic initiator, control data channel)
11// with all the browser UI/media-capture machinery replaced by audio sinks and
12// file writers. It joins a room as an ordinary — visible — participant that
13// reports itself fully muted, receives every other participant's audio, and
14// writes:
15//
16// audio/<name>-<peer8>-segN.wav one file per participant per connection
17// events.jsonl every join/left/chat/mute/segment event
18// chat.txt human-readable chat + join/left log
19// manifest.json session summary: participants + segments
20//
21// All files are written incrementally (manifest every segment boundary and
22// every 30 s), so a crash loses at most ~1 s of audio.
24export const MAX_PARTICIPANTS = 8
26const ANNOUNCE_INTERVAL_MS = 5000
27const PRESENCE_TTL_MS = 15000
28const CONNECT_RETRY_MS = 15000
29const MANIFEST_INTERVAL_MS = 30000
31/** Pad with silence when the sink falls this far behind wall clock, so a
32 * file's sample position always tracks elapsed time (within ~1 s). */
33const PAD_THRESHOLD_FRAC = 1.0 // seconds
34const PAD_MARGIN_FRAC = 0.1 // stay this far behind wall clock when padding
36interface Announcement {
37 peerId: string
38 name: string
39}
41type PeerMsg = {t: 'signal'; signal: Signal} | {t: 'room-full'}
43type ControlMsg =
44 | {
45 t: 'hello'
46 name: string
47 audioMuted: boolean
48 videoMuted: boolean
49 joinedAt: number
50 settings: unknown[]
51 }
52 | {t: 'set'; key: string; value: unknown; rev: number; by: string}
53 | {t: 'mute'; audio: boolean; video: boolean}
54 | {t: 'chat'; text: string}
55 | {t: 'bye'}
57interface AudioSinkData {
58 samples: Int16Array
59 sampleRate: number
60 bitsPerSample?: number
61 channelCount?: number
62 numberOfFrames?: number
63}
65interface Segment {
66 file: string
67 peerId: string
68 name: string
69 startedAt: string
70 endedAt: string | null
71 durationSec: number
72 sampleRate: number
73 channels: number
74}
76interface Conn {
77 peer: Peer
78 createdAt: number
79 name: string | null
80 connected: boolean
81 audioMuted: boolean
82 videoMuted: boolean
83 sink: InstanceType<typeof wrtc.nonstandard.RTCAudioSink> | null
84 writer: WavWriter | null
85 /** Wall-clock ms when the current segment's first audio arrived. */
86 segStartMs: number
87 segment: Segment | null
88}
90export interface RecorderOptions {
91 room: string
92 name: string
93 outDir: string
94 /** Chat line sent to each participant when we connect to them (so everyone
95 * in the room sees, once, that recording is happening). null = none. */
96 notice: string | null
97 onLog: (line: string) => void
98 /** Unrecoverable situation (e.g. the room is full). */
99 onFatal: (message: string) => void
100}
102const sanitize = (name: string): string => {
103 const s = name.replace(/[^a-zA-Z0-9_-]+/g, '_').replace(/^_+|_+$/g, '')
104 return (s || 'peer').slice(0, 24)
105}
107const iso = (ms: number): string => new Date(ms).toISOString()
109const stamp = (ms: number): string => {
110 const d = new Date(ms)
111 const p = (n: number, w = 2) => String(n).padStart(w, '0')
112 return (
113 `${d.getFullYear()}-${p(d.getMonth() + 1)}-${p(d.getDate())} ` +
114 `${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}`
115 )
116}
118export class Recorder {
119 private nostr = new Nostr()
120 private root = ''
121 private presence = new Map<string, {name: string; lastSeen: number}>()
122 private conns = new Map<string, Conn>()
123 private unsubs: (() => void)[] = []
124 private timers: ReturnType<typeof setInterval>[] = []
125 private startedAtMs = 0
126 private stopped = false
128 /** Last known display name per peer (for the manifest). */
129 private names = new Map<string, string>()
130 /** Peers currently counted present (got their hello, not yet left). */
131 private present = new Set<string>()
132 /** Peers we've ever logged join/left for (reconnects get a fresh line). */
133 private seenEver = new Set<string>()
134 /** peerId -> epoch ms until which we won't reconnect: an announcement
135 * published just before a peer's bye can arrive just after it (relay
136 * latency) and would otherwise trigger an instant, pointless reconnect. */
137 private byeCooldown = new Map<string, number>()
138 /** Per-peer segment counter, surviving reconnects. */
139 private segCounts = new Map<string, number>()
140 private segments: Segment[] = []
142 // Outgoing placeholder tracks, shared across all connections (like the
143 // browser's single localStream): a silent mic and a camera that never
144 // produces a frame — the shape of a fully muted participant.
145 private audioSource = new wrtc.nonstandard.RTCAudioSource()
146 private videoSource = new wrtc.nonstandard.RTCVideoSource()
147 private audioTrack = this.audioSource.createTrack()
148 private videoTrack = this.videoSource.createTrack()
150 private audioDir: string
151 private eventsPath: string
152 private chatPath: string
153 private manifestPath: string
155 constructor(private opts: RecorderOptions) {
156 this.audioDir = path.join(opts.outDir, 'audio')
157 this.eventsPath = path.join(opts.outDir, 'events.jsonl')
158 this.chatPath = path.join(opts.outDir, 'chat.txt')
159 this.manifestPath = path.join(opts.outDir, 'manifest.json')
160 }
162 async start() {
163 fs.mkdirSync(this.audioDir, {recursive: true})
164 this.startedAtMs = Date.now()
165 this.event({type: 'start', room: this.opts.room, peerId: selfId, name: this.opts.name})
166 this.chatLine(`* recording started (room: ${this.opts.room})`)
167 this.opts.onLog(`joined room "${this.opts.room}" as "${this.opts.name}" (peer ${selfId.slice(0, 8)})`)
168 this.opts.onLog(`writing to ${this.opts.outDir}`)
170 this.root = await roomTopic(this.opts.room)
171 const selfTopic = await peerTopic(this.root, selfId)
173 this.unsubs.push(
174 this.nostr.subscribe(selfTopic, (content, from) => {
175 if (from === selfId || this.stopped) return
176 let msg: PeerMsg
177 try {
178 msg = JSON.parse(content)
179 } catch {
180 return
181 }
182 this.handlePeerMsg(from, msg)
183 })
184 )
186 this.unsubs.push(
187 this.nostr.subscribe(this.root, (content, from) => {
188 if (from === selfId || this.stopped) return
189 let ann: Partial<Announcement>
190 try {
191 ann = JSON.parse(content)
192 } catch {
193 return
194 }
195 if (ann.peerId !== from || typeof ann.name !== 'string') return
196 const annName = ann.name.slice(0, 40)
197 this.presence.set(from, {name: annName, lastSeen: Date.now()})
198 this.names.set(from, annName)
199 this.maybeConnect(from)
200 })
201 )
203 void this.announce()
204 this.timers.push(setInterval(() => void this.announce(), ANNOUNCE_INTERVAL_MS))
205 this.timers.push(setInterval(() => this.sweepPresence(), ANNOUNCE_INTERVAL_MS))
206 this.timers.push(setInterval(() => this.writeManifest(), MANIFEST_INTERVAL_MS))
207 this.writeManifest()
208 }
210 // ---- presence and the mesh ----------------------------------------------
212 private async announce() {
213 if (this.stopped) return
214 const ann: Announcement = {peerId: selfId, name: this.opts.name}
215 void this.nostr.publish(this.root, JSON.stringify(ann))
216 }
218 private sweepPresence() {
219 const cutoff = Date.now() - PRESENCE_TTL_MS
220 for (const [peerId, p] of this.presence) {
221 if (p.lastSeen < cutoff) this.presence.delete(peerId)
222 }
223 }
225 private async sendToPeer(peerId: string, msg: PeerMsg) {
226 const topic = await peerTopic(this.root, peerId)
227 void this.nostr.publish(topic, JSON.stringify(msg))
228 }
230 private atCapacity(): boolean {
231 return this.conns.size >= MAX_PARTICIPANTS - 1
232 }
234 private maybeConnect(peerId: string) {
235 if (this.stopped || peerId === selfId) return
236 const cooldown = this.byeCooldown.get(peerId)
237 if (cooldown !== undefined) {
238 if (Date.now() < cooldown) return
239 this.byeCooldown.delete(peerId)
240 }
241 const existing = this.conns.get(peerId)
242 if (existing) {
243 const stalled =
244 !existing.connected &&
245 Date.now() - existing.createdAt > CONNECT_RETRY_MS
246 if (!stalled) return
247 this.conns.delete(peerId) // deleted first so the close handler no-ops
248 this.closeConn(peerId, existing)
249 }
250 if (this.atCapacity()) {
251 void this.sendToPeer(peerId, {t: 'room-full'})
252 return
253 }
254 this.createPeer(peerId, selfId < peerId)
255 }
257 private createPeer(peerId: string, initiator: boolean): Conn {
258 const peer = new Peer(initiator, this.audioTrack, this.videoTrack)
259 const conn: Conn = {
260 peer,
261 createdAt: Date.now(),
262 name: null,
263 connected: false,
264 audioMuted: true,
265 videoMuted: true,
266 sink: null,
267 writer: null,
268 segStartMs: 0,
269 segment: null
270 }
271 this.conns.set(peerId, conn)
273 peer.setHandlers({
274 signal: signal => {
275 void this.sendToPeer(peerId, {t: 'signal', signal})
276 },
277 track: track => {
278 if (track.kind !== 'audio' || conn.sink) return
279 this.attachSink(peerId, conn, track)
280 },
281 connect: () => {
282 if (conn.connected) return // connectionState can flap during ICE settling
283 conn.connected = true
284 peer.send(
285 JSON.stringify({
286 t: 'hello',
287 name: this.opts.name,
288 audioMuted: true,
289 videoMuted: true,
290 joinedAt: this.startedAtMs,
291 settings: []
292 } satisfies ControlMsg)
293 )
294 if (this.opts.notice) {
295 peer.send(
296 JSON.stringify({t: 'chat', text: this.opts.notice} satisfies ControlMsg)
297 )
298 }
299 },
300 data: raw => this.handleControl(peerId, conn, raw),
301 close: () => {
302 if (this.conns.get(peerId) === conn) {
303 this.conns.delete(peerId)
304 this.closeConn(peerId, conn)
305 if (this.present.delete(peerId)) {
306 const name = this.displayName(peerId, conn)
307 this.event({type: 'left', peerId, name})
308 this.chatLine(`* ${name} left`)
309 this.opts.onLog(`${name} left`)
310 }
311 }
312 }
313 })
315 return conn
316 }
318 private handlePeerMsg(from: string, msg: PeerMsg) {
319 switch (msg.t) {
320 case 'signal': {
321 let conn = this.conns.get(from)
322 if (!conn) {
323 // An offer can arrive before we've seen the peer's announcement.
324 if (msg.signal?.type !== 'offer') return
325 if (this.atCapacity()) {
326 void this.sendToPeer(from, {t: 'room-full'})
327 return
328 }
329 conn = this.createPeer(from, false)
330 }
331 void conn.peer.signal(msg.signal)
332 return
333 }
334 case 'room-full': {
335 // Only fatal while we have no foothold — once connected, we're in.
336 if (this.conns.size === 0) {
337 this.opts.onFatal(
338 `The room is full (up to ${MAX_PARTICIPANTS} participants) — nothing recorded.`
339 )
340 }
341 return
342 }
343 }
344 }
346 // ---- control channel ----------------------------------------------------
348 private displayName(peerId: string, conn: Conn | null): string {
349 return (
350 this.presence.get(peerId)?.name ??
351 conn?.name ??
352 this.names.get(peerId) ??
353 peerId.slice(0, 8)
354 )
355 }
357 private handleControl(peerId: string, conn: Conn, raw: string) {
358 if (this.conns.get(peerId) !== conn) return
359 let msg: ControlMsg
360 try {
361 msg = JSON.parse(raw)
362 } catch {
363 return
364 }
365 switch (msg.t) {
366 case 'hello': {
367 if (typeof msg.name === 'string' && msg.name) {
368 conn.name = msg.name.slice(0, 40)
369 this.names.set(peerId, conn.name)
370 }
371 conn.audioMuted = msg.audioMuted !== false
372 conn.videoMuted = msg.videoMuted !== false
373 if (!this.present.has(peerId)) {
374 this.present.add(peerId)
375 const name = this.displayName(peerId, conn)
376 const joinedAt = typeof msg.joinedAt === 'number' ? msg.joinedAt : 0
377 const alreadyHere =
378 joinedAt <= this.startedAtMs && !this.seenEver.has(peerId)
379 this.seenEver.add(peerId)
380 this.event({type: 'join', peerId, name, alreadyHere})
381 this.chatLine(`* ${name} ${alreadyHere ? 'was already here' : 'joined'}`)
382 this.opts.onLog(`${name} ${alreadyHere ? 'was already here' : 'joined'} (mic ${conn.audioMuted ? 'muted' : 'on'})`)
383 }
384 return
385 }
386 case 'mute': {
387 if (typeof msg.audio !== 'boolean' || typeof msg.video !== 'boolean') {
388 return
389 }
390 if (conn.audioMuted !== msg.audio) {
391 this.opts.onLog(
392 `${this.displayName(peerId, conn)} ${msg.audio ? 'muted' : 'unmuted'} their mic`
393 )
394 }
395 conn.audioMuted = msg.audio
396 conn.videoMuted = msg.video
397 this.event({
398 type: 'mute',
399 peerId,
400 name: this.displayName(peerId, conn),
401 audio: msg.audio,
402 video: msg.video
403 })
404 return
405 }
406 case 'chat': {
407 if (typeof msg.text !== 'string') return
408 const text = msg.text.slice(0, 2000)
409 if (!text.trim()) return
410 const name = this.displayName(peerId, conn)
411 this.event({type: 'chat', peerId, name, text})
412 this.chatLine(`${name}: ${text}`)
413 this.opts.onLog(`${name}: ${text}`)
414 return
415 }
416 case 'set': // room settings don't matter to the recorder
417 return
418 case 'bye': {
419 this.presence.delete(peerId)
420 this.byeCooldown.set(peerId, Date.now() + 3000)
421 conn.peer.destroy() // its close handler finalizes the segment
422 return
423 }
424 }
425 }
427 // ---- audio capture ------------------------------------------------------
429 private attachSink(peerId: string, conn: Conn, track: MediaStreamTrack) {
430 const sink = new wrtc.nonstandard.RTCAudioSink(track)
431 conn.sink = sink
432 sink.ondata = (data: AudioSinkData) => {
433 if (this.stopped || this.conns.get(peerId) !== conn) return
434 const channels = data.channelCount ?? 1
435 const rate = data.sampleRate
436 if (!rate || !data.samples?.length) return
438 // A decoder format change (rare) starts a fresh segment.
439 if (
440 conn.writer &&
441 (conn.writer.sampleRate !== rate || conn.writer.channels !== channels)
442 ) {
443 this.endSegment(peerId, conn)
444 }
446 const now = Date.now()
447 if (!conn.writer) {
448 // Before the first RTP packet the sink delivers all-zero frames (at a
449 // provisional sample rate, even) — don't open a file until there is
450 // actual audio. A participant who never unmutes produces no file.
451 if (!data.samples.some(s => s !== 0)) return
452 const n = (this.segCounts.get(peerId) ?? 0) + 1
453 this.segCounts.set(peerId, n)
454 const name = this.displayName(peerId, conn)
455 const file = path.join(
456 'audio',
457 `${sanitize(name)}-${peerId.slice(0, 8)}-seg${n}.wav`
458 )
459 conn.writer = new WavWriter(
460 path.join(this.opts.outDir, file),
461 rate,
462 channels
463 )
464 conn.segStartMs = now
465 conn.segment = {
466 file,
467 peerId,
468 name,
469 startedAt: iso(now),
470 endedAt: null,
471 durationSec: 0,
472 sampleRate: rate,
473 channels
474 }
475 this.event({type: 'segment-start', peerId, name, file, sampleRate: rate, channels})
476 this.opts.onLog(`recording ${name} -> ${file}`)
477 } else {
478 // If the sink stalled (network gap, DTX), pad with silence so sample
479 // position keeps tracking wall-clock time.
480 const expected = Math.floor(((now - conn.segStartMs) / 1000) * rate)
481 const deficit = expected - conn.writer.framesWritten
482 if (deficit > rate * PAD_THRESHOLD_FRAC) {
483 conn.writer.appendSilence(deficit - Math.floor(rate * PAD_MARGIN_FRAC))
484 }
485 }
486 conn.writer.append(data.samples)
487 }
488 }
490 private endSegment(peerId: string, conn: Conn) {
491 if (!conn.writer || !conn.segment) return
492 conn.writer.finalize()
493 conn.segment.endedAt = iso(Date.now())
494 conn.segment.durationSec = Math.round(conn.writer.durationSec * 100) / 100
495 this.segments.push(conn.segment)
496 this.event({
497 type: 'segment-end',
498 peerId,
499 name: conn.segment.name,
500 file: conn.segment.file,
501 durationSec: conn.segment.durationSec
502 })
503 this.opts.onLog(
504 `closed ${conn.segment.file} (${conn.segment.durationSec.toFixed(1)}s)`
505 )
506 conn.writer = null
507 conn.segment = null
508 this.writeManifest()
509 }
511 /** Tear down a conn's media capture and finalize its segment. */
512 private closeConn(peerId: string, conn: Conn) {
513 try {
514 conn.sink?.stop()
515 } catch {
516 /* ignore */
517 }
518 conn.sink = null
519 this.endSegment(peerId, conn)
520 conn.peer.destroy()
521 }
523 // ---- output files -------------------------------------------------------
525 private event(ev: Record<string, unknown>) {
526 const line = JSON.stringify({time: iso(Date.now()), ...ev})
527 try {
528 fs.appendFileSync(this.eventsPath, line + '\n')
529 } catch {
530 /* ignore */
531 }
532 }
534 private chatLine(text: string) {
535 try {
536 fs.appendFileSync(this.chatPath, `[${stamp(Date.now())}] ${text}\n`)
537 } catch {
538 /* ignore */
539 }
540 }
542 private writeManifest() {
543 const active = [...this.conns.values()]
544 .filter(c => c.segment && c.writer)
545 .map(c => ({
546 ...c.segment!,
547 durationSec: Math.round(c.writer!.durationSec * 100) / 100
548 }))
549 const manifest = {
550 room: this.opts.room,
551 recorder: {peerId: selfId, name: this.opts.name},
552 startedAt: iso(this.startedAtMs),
553 endedAt: this.stopped ? iso(Date.now()) : null,
554 participants: Object.fromEntries(this.names),
555 segments: [...this.segments, ...active]
556 }
557 try {
558 const tmp = this.manifestPath + '.tmp'
559 fs.writeFileSync(tmp, JSON.stringify(manifest, null, 2) + '\n')
560 fs.renameSync(tmp, this.manifestPath)
561 } catch {
562 /* ignore */
563 }
564 }
566 // ---- shutdown -----------------------------------------------------------
568 stop(): {segments: number; participants: number} {
569 if (this.stopped) return {segments: this.segments.length, participants: this.names.size}
570 this.stopped = true
571 const bye = JSON.stringify({t: 'bye'} satisfies ControlMsg)
572 for (const conn of this.conns.values()) conn.peer.send(bye)
573 const conns = [...this.conns.entries()]
574 this.conns.clear()
575 for (const [peerId, conn] of conns) this.closeConn(peerId, conn)
576 for (const u of this.unsubs.splice(0)) u()
577 for (const t of this.timers.splice(0)) clearInterval(t)
578 this.nostr.close()
579 try {
580 this.audioTrack.stop()
581 this.videoTrack.stop()
582 } catch {
583 /* ignore */
584 }
585 this.event({type: 'stop'})
586 this.chatLine('* recording stopped')
587 this.writeManifest()
588 return {segments: this.segments.length, participants: this.names.size}
589 }
590}