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