Toju/toju-app/src/app/domains/voice-connection/application/services/voice-activity.service.ts

/**
 * 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/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>();
  /** The id the local microphone is tracked under, so it can follow a device switch. */
  private localMicUserId: string | null = null;
  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.localMicUserId = userId;
    this.trackStream(userId, stream);
  }

  untrackLocalMic(userId: string): void {
    if (this.localMicUserId === userId) {
      this.localMicUserId = null;
    }

    this.untrackStream(userId);
  }

  /**
   * Point the local speaking indicator at a replacement microphone stream,
   * so switching devices mid-call does not leave it watching a dead track.
   */
  refreshLocalMicStream(stream: MediaStream): void {
    if (!this.localMicUserId) {
      return;
    }

    this.trackStream(this.localMicUserId, stream);
  }

  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());
  }
}