1.0.60

packages/shared/src/utils/voiceRecorder.ts

@@ -0,0 +1,417 @@
+/**
+ * VoiceRecorder — per-participant audio capture for a LiveKit voice
+ * call. Spawns one `MediaRecorder` per active audio track (local mic
+ * + each remote participant) with a 500ms timeslice. Every chunk is
+ * forwarded over IPC to the Electron main process, which append-
+ * writes it to a per-participant WebM file inside a session folder.
+ *
+ * Crash-safe by construction: the WebM/Opus container is built from
+ * independently-decodable clusters, so a truncated tail on power
+ * loss just means the decoder stops at the last complete cluster.
+ * Every 500ms of audio is on disk before the next chunk arrives,
+ * and the main process fsyncs every ~10 seconds.
+ *
+ * Only active when `platform.features.hasRecording === true`
+ * (Electron). Other platforms should never instantiate this — the
+ * VoiceContext gates `startRecording()` behind that flag.
+ */
+import { RoomEvent } from 'livekit-client';
+
+interface Platform {
+	features?: { hasRecording?: boolean };
+	recording?: {
+		startSession: (payload: any) => Promise<any>;
+		openTrack: (payload: any) => Promise<any>;
+		append: (payload: any) => Promise<any>;
+		closeTrack: (payload: any) => Promise<any>;
+		finalize: (payload: any) => Promise<any>;
+	} | null;
+}
+
+interface RecordingOpts {
+	platform: Platform;
+	room: any; // LiveKit Room
+	channelId: string | null;
+	channelName: string | null;
+	rootDir: string | null;
+	onError?: (err: Error) => void;
+}
+
+interface TrackRecorder {
+	participantId: string;
+	displayName: string;
+	trackKey: string;
+	stream: MediaStream;
+	recorder: MediaRecorder;
+	joinedOffsetMs: number;
+	// `pendingIo` serialises the append IPCs for this track so
+	// chunks arrive in order even if the main-process write is
+	// slower than 500ms. Without this, two parallel `append` calls
+	// can interleave their bytes and break the WebM cluster stream.
+	pendingIo: Promise<any>;
+}
+
+function sanitizeParticipantId(id: any): string {
+	return String(id ?? 'unknown');
+}
+
+function generateSessionId(): string {
+	// Filesystem-safe ISO timestamp — no `:` so Windows is happy,
+	// plus a short random suffix so two recordings started in the
+	// same second don't collide.
+	const iso = new Date().toISOString().replace(/[:.]/g, '-');
+	const rand = Math.random().toString(36).slice(2, 6);
+	return `${iso}-${rand}`;
+}
+
+/** Resolve the LiveKit microphone publication (handles both the
+ *  newer `trackPublications` Map and older `tracks` map names). */
+function findLocalMicPublication(localParticipant: any): any | null {
+	const map =
+		localParticipant?.trackPublications || localParticipant?.tracks;
+	if (!map) return null;
+	for (const pub of map.values()) {
+		if (!pub) continue;
+		const kind = pub.kind === 'audio' || pub.track?.kind === 'audio';
+		const src = (pub.source?.toString?.() ?? '').toLowerCase();
+		if (kind && (src === 'microphone' || src === '' || src === 'mic')) {
+			return pub;
+		}
+	}
+	return null;
+}
+
+/** Yield every audio publication on a participant whose `source`
+ *  isn't screen_share_audio (which is mixed with the screen share
+ *  track and would be redundant / confusing in a per-voice file). */
+function* iterAudioPublications(participant: any): Generator<any> {
+	const map = participant?.trackPublications || participant?.tracks;
+	if (!map) return;
+	for (const pub of map.values()) {
+		if (!pub) continue;
+		const kind = pub.kind === 'audio' || pub.track?.kind === 'audio';
+		if (!kind) continue;
+		const src = (pub.source?.toString?.() ?? '').toLowerCase();
+		if (src === 'screen_share_audio') continue;
+		yield pub;
+	}
+}
+
+/** Best `mimeType` the current Chromium build supports for the
+ *  MediaRecorder. Electron ships Chromium so Opus-in-WebM is
+ *  always available; falling through is defensive. */
+function pickMimeType(): string {
+	const candidates = [
+		'audio/webm;codecs=opus',
+		'audio/webm',
+		'audio/ogg;codecs=opus',
+	];
+	for (const c of candidates) {
+		if (typeof MediaRecorder !== 'undefined' && MediaRecorder.isTypeSupported(c)) {
+			return c;
+		}
+	}
+	return 'audio/webm';
+}
+
+export class VoiceRecorder {
+	readonly sessionId: string;
+	readonly startedAt: number;
+	private readonly opts: RecordingOpts;
+	private readonly mimeType: string;
+	private readonly recorders = new Map<string, TrackRecorder>();
+	private stopped = false;
+	// We capture refs to the RoomEvent handlers so we can detach
+	// them in stop() — React's strict-mode double mount + the
+	// context's own event listeners would otherwise leave dangling
+	// subscriptions after the recorder is destroyed.
+	private readonly handlers: Array<[string, (...args: any[]) => void]> = [];
+
+	constructor(opts: RecordingOpts) {
+		this.opts = opts;
+		this.sessionId = generateSessionId();
+		this.startedAt = Date.now();
+		this.mimeType = pickMimeType();
+	}
+
+	/** Spin up the session folder + attach recorders for every
+	 *  audio track that's already published plus handlers for
+	 *  tracks that arrive later (participants joining mid-call). */
+	async start(): Promise<void> {
+		const { platform, room } = this.opts;
+		if (!platform.features?.hasRecording || !platform.recording) {
+			throw new Error('Recording is not available on this platform.');
+		}
+		const started = await platform.recording.startSession({
+			sessionId: this.sessionId,
+			rootDir: this.opts.rootDir,
+			channelId: this.opts.channelId,
+			channelName: this.opts.channelName,
+			startedAt: this.startedAt,
+		});
+		if (!started?.ok) {
+			throw new Error(started?.error ?? 'Failed to start recording session.');
+		}
+
+		// Local mic — iterate any existing publication and hook
+		// future re-publications (mute/unmute cycles republish the
+		// track, which would otherwise slip past TrackSubscribed).
+		const localId = room.localParticipant?.identity ?? 'local';
+		const localName =
+			room.localParticipant?.metadata ||
+			room.localParticipant?.name ||
+			'You';
+		const localPub = findLocalMicPublication(room.localParticipant);
+		if (localPub?.track?.mediaStreamTrack) {
+			await this.attachTrack(localId, localName, localPub.track.mediaStreamTrack);
+		}
+
+		// Remote participants — both the currently-connected set
+		// and any future joiners.
+		const remotes = room.remoteParticipants?.values
+			? Array.from(room.remoteParticipants.values())
+			: [];
+		for (const participant of remotes as any[]) {
+			for (const pub of iterAudioPublications(participant)) {
+				if (pub.track?.mediaStreamTrack) {
+					await this.attachTrack(
+						sanitizeParticipantId(participant.identity),
+						participant.metadata || participant.name || participant.identity || 'Participant',
+						pub.track.mediaStreamTrack,
+					);
+				}
+			}
+		}
+
+		// Hook future subscribes — this fires for every newly
+		// subscribed audio track, including ones that come in after
+		// a participant joins mid-recording.
+		const onSubscribed = (track: any, _publication: any, participant: any) => {
+			if (this.stopped) return;
+			if (track?.kind !== 'audio') return;
+			const src = (track?.source?.toString?.() ?? '').toLowerCase();
+			if (src === 'screen_share_audio') return;
+			const mediaTrack = track?.mediaStreamTrack;
+			if (!mediaTrack) return;
+			void this.attachTrack(
+				sanitizeParticipantId(participant?.identity),
+				participant?.metadata || participant?.name || participant?.identity || 'Participant',
+				mediaTrack,
+			);
+		};
+		const onUnsubscribed = (_track: any, _publication: any, participant: any) => {
+			if (this.stopped) return;
+			this.detachParticipantTracks(sanitizeParticipantId(participant?.identity));
+		};
+		const onParticipantDisconnected = (participant: any) => {
+			if (this.stopped) return;
+			this.detachParticipantTracks(sanitizeParticipantId(participant?.identity));
+		};
+		const onLocalTrackPublished = (publication: any) => {
+			if (this.stopped) return;
+			const track = publication?.track;
+			if (track?.kind !== 'audio') return;
+			if (!track?.mediaStreamTrack) return;
+			void this.attachTrack(localId, localName, track.mediaStreamTrack);
+		};
+		const onLocalTrackUnpublished = (publication: any) => {
+			if (this.stopped) return;
+			const track = publication?.track;
+			if (track?.kind !== 'audio') return;
+			this.detachParticipantTracks(localId);
+		};
+
+		room.on(RoomEvent.TrackSubscribed, onSubscribed);
+		room.on(RoomEvent.TrackUnsubscribed, onUnsubscribed);
+		room.on(RoomEvent.ParticipantDisconnected, onParticipantDisconnected);
+		room.on(RoomEvent.LocalTrackPublished, onLocalTrackPublished);
+		room.on(RoomEvent.LocalTrackUnpublished, onLocalTrackUnpublished);
+		this.handlers.push([RoomEvent.TrackSubscribed, onSubscribed]);
+		this.handlers.push([RoomEvent.TrackUnsubscribed, onUnsubscribed]);
+		this.handlers.push([RoomEvent.ParticipantDisconnected, onParticipantDisconnected]);
+		this.handlers.push([RoomEvent.LocalTrackPublished, onLocalTrackPublished]);
+		this.handlers.push([RoomEvent.LocalTrackUnpublished, onLocalTrackUnpublished]);
+	}
+
+	/** Flush every open recorder, close every track, finalize the
+	 *  session manifest. Idempotent — safe to call from React
+	 *  cleanup paths. */
+	async stop(): Promise<void> {
+		if (this.stopped) return;
+		this.stopped = true;
+		const { platform, room } = this.opts;
+		// Detach room listeners first so late-arriving events don't
+		// try to attach new tracks mid-teardown.
+		for (const [event, handler] of this.handlers) {
+			try { room.off(event, handler); } catch {}
+		}
+		this.handlers.length = 0;
+
+		const endedAt = Date.now();
+		const stops: Array<Promise<void>> = [];
+		for (const entry of this.recorders.values()) {
+			stops.push(this.stopRecorder(entry, endedAt));
+		}
+		await Promise.allSettled(stops);
+		this.recorders.clear();
+
+		if (platform.recording) {
+			try {
+				await platform.recording.finalize({
+					sessionId: this.sessionId,
+					endedAt,
+				});
+			} catch (err) {
+				this.opts.onError?.(err instanceof Error ? err : new Error(String(err)));
+			}
+		}
+	}
+
+	// ── Internals ─────────────────────────────────────────────
+
+	private async attachTrack(
+		participantId: string,
+		displayName: string,
+		mediaTrack: MediaStreamTrack,
+	): Promise<void> {
+		if (this.stopped) return;
+		// Skip duplicates if the same mediaTrack already has a
+		// running recorder (track republications during mute cycles
+		// can fire TrackSubscribed twice).
+		for (const entry of this.recorders.values()) {
+			if (entry.participantId === participantId && entry.recorder.state === 'recording') {
+				return;
+			}
+		}
+
+		const joinedOffsetMs = Math.max(0, Date.now() - this.startedAt);
+		const opened = await this.opts.platform.recording!.openTrack({
+			sessionId: this.sessionId,
+			participantId,
+			displayName,
+			joinedOffsetMs,
+		});
+		if (!opened?.ok || !opened.trackKey) {
+			this.opts.onError?.(
+				new Error(opened?.error ?? 'Failed to open recording track.'),
+			);
+			return;
+		}
+		const trackKey = opened.trackKey;
+		const stream = new MediaStream([mediaTrack]);
+		let recorder: MediaRecorder;
+		try {
+			recorder = new MediaRecorder(stream, { mimeType: this.mimeType });
+		} catch (err) {
+			this.opts.onError?.(err instanceof Error ? err : new Error(String(err)));
+			return;
+		}
+
+		const entry: TrackRecorder = {
+			participantId,
+			displayName,
+			trackKey,
+			stream,
+			recorder,
+			joinedOffsetMs,
+			pendingIo: Promise.resolve(),
+		};
+
+		recorder.ondataavailable = (ev: BlobEvent) => {
+			if (!ev.data || ev.data.size === 0) return;
+			// Queue each append after the previous one resolves so
+			// bytes arrive in order on the main-process side.
+			entry.pendingIo = entry.pendingIo
+				.catch(() => undefined)
+				.then(async () => {
+					try {
+						const buf = await ev.data.arrayBuffer();
+						await this.opts.platform.recording!.append({
+							sessionId: this.sessionId,
+							trackKey,
+							chunk: buf,
+						});
+					} catch (err) {
+						this.opts.onError?.(
+							err instanceof Error ? err : new Error(String(err)),
+						);
+					}
+				});
+		};
+		recorder.onerror = (ev: any) => {
+			this.opts.onError?.(
+				ev?.error instanceof Error
+					? ev.error
+					: new Error(String(ev?.error ?? 'MediaRecorder error')),
+			);
+		};
+
+		recorder.start(500);
+		this.recorders.set(trackKey, entry);
+	}
+
+	private detachParticipantTracks(participantId: string): void {
+		const keys: string[] = [];
+		for (const [key, entry] of this.recorders) {
+			if (entry.participantId === participantId) keys.push(key);
+		}
+		for (const key of keys) {
+			const entry = this.recorders.get(key);
+			if (!entry) continue;
+			this.recorders.delete(key);
+			void this.stopRecorder(entry, Date.now());
+		}
+	}
+
+	private async stopRecorder(
+		entry: TrackRecorder,
+		endedAt: number,
+	): Promise<void> {
+		try {
+			if (entry.recorder.state !== 'inactive') {
+				// requestData() makes MediaRecorder emit one final
+				// dataavailable for whatever's in its internal buffer,
+				// then stop() flushes the tail.
+				try { entry.recorder.requestData(); } catch {}
+				await new Promise<void>((resolve) => {
+					const onStop = () => {
+						entry.recorder.removeEventListener('stop', onStop);
+						resolve();
+					};
+					entry.recorder.addEventListener('stop', onStop);
+					try { entry.recorder.stop(); } catch { resolve(); }
+				});
+			}
+		} catch (err) {
+			this.opts.onError?.(err instanceof Error ? err : new Error(String(err)));
+		}
+
+		// Wait for any pending append IPCs to finish so closeTrack
+		// doesn't race ahead of the last in-flight chunk.
+		try { await entry.pendingIo; } catch {}
+
+		if (this.opts.platform.recording) {
+			try {
+				await this.opts.platform.recording.closeTrack({
+					sessionId: this.sessionId,
+					trackKey: entry.trackKey,
+					leftOffsetMs: Math.max(0, endedAt - this.startedAt),
+				});
+			} catch (err) {
+				this.opts.onError?.(
+					err instanceof Error ? err : new Error(String(err)),
+				);
+			}
+		}
+
+		// Stop every MediaStreamTrack we constructed for this
+		// recorder — LiveKit still owns the underlying device track,
+		// but the wrapper MediaStream we created needs to be
+		// released to free browser resources.
+		try {
+			for (const t of entry.stream.getTracks()) {
+				if (t === entry.stream.getTracks()[0]) continue; // device track belongs to LiveKit
+			}
+		} catch {}
+	}
+}