refactor: stricter domain: voice-connection

2026-04-11 15:08:04 +02:00
parent cea3dccef1
commit 7a0664b3c4
13 changed files with 36 additions and 33 deletions
@@ -0,0 +1,284 @@
+/**
+ * VoiceActivityService - monitors audio levels for local microphone
+ * and remote peer streams, exposing per-user "speaking" state as
+ * reactive Angular signals.
+ *
+ * Usage:
+ * ```ts
+ * const speaking = voiceActivity.isSpeaking(userId);
+ * // speaking() => true when the user's audio level exceeds the threshold
+ *
+ * const volume = voiceActivity.volume(userId);
+ * // volume() => normalised 0-1 audio level
+ * ```
+ *
+ * Internally uses the Web Audio API ({@link AudioContext} +
+ * {@link AnalyserNode}) per tracked stream, with a single
+ * `requestAnimationFrame` poll loop.
+ */
+import {
+  Injectable,
+  signal,
+  computed,
+  inject,
+  OnDestroy,
+  Signal
+} from '@angular/core';
+import { Subscription } from 'rxjs';
+import { VoiceConnectionFacade } from '../facades/voice-connection.facade';
+import { DebuggingService } from '../../../../core/services/debugging.service';
+/* eslint-disable @typescript-eslint/member-ordering, @typescript-eslint/prefer-for-of, max-statements-per-line */
+
+const SPEAKING_THRESHOLD = 0.015;
+const SILENT_FRAME_GRACE = 8;
+const FFT_SIZE = 256;
+
+interface TrackedStream {
+  ctx: AudioContext;
+  sources: MediaStreamAudioSourceNode[];
+  analyser: AnalyserNode;
+  dataArray: Uint8Array<ArrayBuffer>;
+  volumeSignal: ReturnType<typeof signal<number>>;
+  speakingSignal: ReturnType<typeof signal<boolean>>;
+  silentFrames: number;
+  stream: MediaStream;
+}
+
+@Injectable({ providedIn: 'root' })
+export class VoiceActivityService implements OnDestroy {
+  private readonly voiceConnection = inject(VoiceConnectionFacade);
+  private readonly debugging = inject(DebuggingService);
+
+  private readonly tracked = new Map<string, TrackedStream>();
+  private animFrameId: number | null = null;
+  private readonly subs: Subscription[] = [];
+  private readonly _speakingMap = signal<ReadonlyMap<string, boolean>>(new Map());
+
+  readonly speakingMap: Signal<ReadonlyMap<string, boolean>> = this._speakingMap;
+
+  constructor() {
+    this.subs.push(
+      this.voiceConnection.onRemoteStream.subscribe(({ peerId }) => {
+        const voiceStream = this.voiceConnection.getRemoteVoiceStream(peerId);
+
+        if (!voiceStream) {
+          this.untrackStream(peerId);
+          return;
+        }
+
+        this.trackStream(peerId, voiceStream);
+      })
+    );
+
+    this.subs.push(
+      this.voiceConnection.onPeerDisconnected.subscribe((peerId) => {
+        this.untrackStream(peerId);
+      })
+    );
+
+    this.subs.push(
+      this.voiceConnection.onVoiceConnected.subscribe(() => {
+        this.ensureAllRemoteStreamsTracked();
+      })
+    );
+  }
+
+  trackLocalMic(userId: string, stream: MediaStream): void {
+    this.trackStream(userId, stream);
+  }
+
+  untrackLocalMic(userId: string): void {
+    this.untrackStream(userId);
+  }
+
+  isSpeaking(userId: string): Signal<boolean> {
+    const entry = this.tracked.get(userId);
+
+    if (entry)
+      return entry.speakingSignal.asReadonly();
+
+    return computed(() => this._speakingMap().get(userId) ?? false);
+  }
+
+  volume(userId: string): Signal<number> {
+    const entry = this.tracked.get(userId);
+
+    if (entry)
+      return entry.volumeSignal.asReadonly();
+
+    return computed(() => 0);
+  }
+
+  trackStream(id: string, stream: MediaStream): void {
+    const existing = this.tracked.get(id);
+    const audioTracks = stream.getAudioTracks().filter((track) => track.readyState === 'live');
+
+    if (existing && existing.stream === stream)
+      return;
+
+    if (existing)
+      this.disposeEntry(existing);
+
+    if (audioTracks.length === 0) {
+      this.tracked.delete(id);
+      this.publishSpeakingMap();
+
+      if (this.tracked.size === 0)
+        this.stopPolling();
+
+      return;
+    }
+
+    const ctx = new AudioContext();
+
+    if (ctx.state === 'suspended')
+      void ctx.resume();
+
+    const analyser = ctx.createAnalyser();
+    const sources = audioTracks.map((track) => ctx.createMediaStreamSource(new MediaStream([track])));
+
+    analyser.fftSize = FFT_SIZE;
+    sources.forEach((source) => source.connect(analyser));
+
+    const dataArray = new Uint8Array(analyser.fftSize) as Uint8Array<ArrayBuffer>;
+    const volumeSignal = signal(0);
+    const speakingSignal = signal(false);
+
+    this.tracked.set(id, {
+      ctx,
+      sources,
+      analyser,
+      dataArray,
+      volumeSignal,
+      speakingSignal,
+      silentFrames: 0,
+      stream
+    });
+
+    this.ensurePolling();
+  }
+
+  untrackStream(id: string): void {
+    const entry = this.tracked.get(id);
+
+    if (!entry)
+      return;
+
+    if (entry.speakingSignal()) {
+      this.reportSpeakingState(id, false, 0);
+    }
+
+    this.disposeEntry(entry);
+    this.tracked.delete(id);
+    this.publishSpeakingMap();
+
+    if (this.tracked.size === 0)
+      this.stopPolling();
+  }
+
+  private ensureAllRemoteStreamsTracked(): void {
+    const peers = this.voiceConnection.getConnectedPeers();
+
+    for (const peerId of peers) {
+      const stream = this.voiceConnection.getRemoteVoiceStream(peerId);
+
+      if (stream) {
+        this.trackStream(peerId, stream);
+      }
+    }
+  }
+
+  private ensurePolling(): void {
+    if (this.animFrameId !== null)
+      return;
+
+    this.poll();
+  }
+
+  private stopPolling(): void {
+    if (this.animFrameId !== null) {
+      cancelAnimationFrame(this.animFrameId);
+      this.animFrameId = null;
+    }
+  }
+
+  private poll = (): void => {
+    let mapDirty = false;
+
+    this.tracked.forEach((entry, id) => {
+      const { analyser, dataArray, volumeSignal, speakingSignal } = entry;
+
+      analyser.getByteTimeDomainData(dataArray);
+
+      let sumSquares = 0;
+
+      for (let sampleIndex = 0; sampleIndex < dataArray.length; sampleIndex++) {
+        const normalised = (dataArray[sampleIndex] - 128) / 128;
+
+        sumSquares += normalised * normalised;
+      }
+
+      const rms = Math.sqrt(sumSquares / dataArray.length);
+
+      volumeSignal.set(rms);
+
+      const wasSpeaking = speakingSignal();
+
+      if (rms >= SPEAKING_THRESHOLD) {
+        entry.silentFrames = 0;
+
+        if (!wasSpeaking) {
+          speakingSignal.set(true);
+          this.reportSpeakingState(id, true, rms);
+          mapDirty = true;
+        }
+      } else {
+        entry.silentFrames++;
+
+        if (wasSpeaking && entry.silentFrames >= SILENT_FRAME_GRACE) {
+          speakingSignal.set(false);
+          this.reportSpeakingState(id, false, rms);
+          mapDirty = true;
+        }
+      }
+    });
+
+    if (mapDirty)
+      this.publishSpeakingMap();
+
+    this.animFrameId = requestAnimationFrame(this.poll);
+  };
+
+  private publishSpeakingMap(): void {
+    const map = new Map<string, boolean>();
+
+    this.tracked.forEach((entry, id) => {
+      map.set(id, entry.speakingSignal());
+    });
+
+    this._speakingMap.set(map);
+  }
+
+  private reportSpeakingState(peerId: string, isSpeaking: boolean, volume: number): void {
+    this.debugging.recordEvent('webrtc:voice-activity', 'Speaking state changed', {
+      peerId,
+      isSpeaking,
+      volume: Number(volume.toFixed(3))
+    });
+  }
+
+  private disposeEntry(entry: TrackedStream): void {
+    entry.sources.forEach((source) => {
+      try { source.disconnect(); } catch { /* already disconnected */ }
+    });
+
+    try { entry.ctx.close(); } catch { /* already closed */ }
+  }
+
+  ngOnDestroy(): void {
+    this.stopPolling();
+    this.tracked.forEach((entry) => this.disposeEntry(entry));
+    this.tracked.clear();
+    this.subs.forEach((subscription) => subscription.unsubscribe());
+  }
+}