element-call/src/lotus/lotusMicLevel.ts

/*
Copyright 2026 Lotus Guild

SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/

import {
  type LocalTrackPublication,
  type Room as LivekitRoom,
  RoomEvent,
  Track,
} from "livekit-client";
import { Observable, type Subscription, share, switchMap } from "rxjs";
import { distinctUntilChanged, map } from "rxjs/operators";

import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
import { widget } from "../widget";
import { LotusWidgetActions, lotusFlag, lotusSendToHost } from "./lotusWidget";

/**
 * [lotus #146] One local mic sampler shared by the host's mic level meter
 * (`io.lotus.mic_level`) and "talking while muted" (#37, lotusMutedSpeech).
 * It taps a CLONE of the published mic track (the post-processor track when
 * the in-source denoiser is active, so what's measured is what's sent) and
 * reads RMS at ~10 Hz while a mic track is published, muted or not.
 * Local only: nothing here reaches other participants.
 */

const SAMPLE_MS = 100;

export interface LocalMicSample {
  rms: number;
  /** The mic is published but muted (the clone still hears it). */
  muted: boolean;
}

const micPublication = (
  room: LivekitRoom,
): { track: MediaStreamTrack; muted: boolean } | null => {
  const pub: LocalTrackPublication | undefined =
    room.localParticipant.getTrackPublication(Track.Source.Microphone);
  const track = pub?.track?.mediaStreamTrack;
  return track && track.readyState === "live"
    ? { track, muted: pub?.isMuted ?? false }
    : null;
};

/** Pure: the same LiveKit rooms, in the same order (by identity). */
export const sameRooms = <T>(a: readonly T[], b: readonly T[]): boolean =>
  a.length === b.length && a.every((r, i) => r === b[i]);

const samplers = new WeakMap<
  CallViewModel,
  Observable<LocalMicSample | null>
>();

/**
 * RMS samples of the local mic, or `null` while no mic track is published.
 * Shared per call view model, so the meter and the muted-speech detector use
 * one AudioContext between them.
 */
export function observeLocalMicSample$(
  vm: CallViewModel,
): Observable<LocalMicSample | null> {
  const cached = samplers.get(vm);
  if (cached) return cached;
  const sampler$ = vm.allConnections$.pipe(
    map((data) => data.getConnections().map((c) => c.livekitRoom)),
    // [EC #45] allConnections$ re-emits far more often than the set of
    // LiveKit rooms changes; restarting the tap each time closed the
    // AudioContext ~1×/s and emitted null, which reset the muted-speech
    // gate so the "You're muted" nudge could never fire.
    distinctUntilChanged(sameRooms),
    switchMap(
      (rooms) =>
        new Observable<LocalMicSample | null>((subscriber) => {
          let ctx: AudioContext | null = null;
          let clone: MediaStreamTrack | null = null;
          let timer: ReturnType<typeof setInterval> | undefined;
          let tapped: MediaStreamTrack | null = null;
          let muted = false;

          const stopTap = (): void => {
            if (timer !== undefined) clearInterval(timer);
            timer = undefined;
            clone?.stop();
            clone = null;
            void ctx?.close().catch(() => undefined);
            ctx = null;
            tapped = null;
            subscriber.next(null);
          };

          const startTap = (source: MediaStreamTrack): void => {
            try {
              clone = source.clone();
              // Muting disables the published track; the clone must still hear.
              clone.enabled = true;
              ctx = new AudioContext();
              const analyser = ctx.createAnalyser();
              analyser.fftSize = 1024;
              ctx
                .createMediaStreamSource(new MediaStream([clone]))
                .connect(analyser);
              const buf = new Float32Array(analyser.fftSize);
              tapped = source;
              timer = setInterval(() => {
                analyser.getFloatTimeDomainData(buf);
                let sum = 0;
                for (let i = 0; i < buf.length; i += 1) sum += buf[i] * buf[i];
                subscriber.next({ rms: Math.sqrt(sum / buf.length), muted });
              }, SAMPLE_MS);
            } catch {
              stopTap();
            }
          };

          const reconcile = (): void => {
            const pub =
              rooms.map(micPublication).find((p) => p !== null) ?? null;
            muted = pub?.muted ?? false;
            const track = pub?.track ?? null;
            if (track === tapped) return;
            if (tapped) stopTap();
            if (track) startTap(track);
          };

          const events = [
            RoomEvent.TrackMuted,
            RoomEvent.TrackUnmuted,
            RoomEvent.LocalTrackPublished,
            RoomEvent.LocalTrackUnpublished,
            RoomEvent.Reconnected,
            RoomEvent.Disconnected,
          ] as const;
          rooms.forEach((room) =>
            events.forEach((ev) => room.on(ev, reconcile)),
          );
          subscriber.next(null);
          reconcile();
          return (): void => {
            rooms.forEach((room) =>
              events.forEach((ev) => room.off(ev, reconcile)),
            );
            if (tapped) stopTap();
          };
        }),
    ),
    share(),
  );
  samplers.set(vm, sampler$);
  return sampler$;
}

/** RMS at which each bar lights: ≈ −46, −36 and −26 dBFS. */
export const BAR_THRESHOLDS = [0.005, 0.015, 0.05] as const;

/**
 * Quantise RMS to 0–3 bars with a little hysteresis: rises at once, falls one
 * bar per sample, so the meter doesn't flicker between words. Unit-tested.
 */
export class MicLevelQuantizer {
  private bars = 0;

  public reset(): void {
    this.bars = 0;
  }

  public push(rms: number): number {
    const target = BAR_THRESHOLDS.filter((t) => rms >= t).length;
    this.bars = target >= this.bars ? target : this.bars - 1;
    return this.bars;
  }
}

/**
 * Send the host `io.lotus.mic_level { bars }` (0–3) whenever the quantised
 * level changes; 0 while muted or with no mic. At most one message per sample
 * (10 Hz) and none during steady silence. Opt-in with the rest of the host
 * state stream (`lotusCallState=1`). Returns a teardown function.
 */
export function startLotusMicLevel(vm: CallViewModel): () => void {
  if (!lotusFlag("lotusCallState") || !widget) return (): void => undefined;
  const quantizer = new MicLevelQuantizer();
  const sub: Subscription = observeLocalMicSample$(vm)
    .pipe(
      map((sample) => {
        if (!sample || sample.muted) {
          quantizer.reset();
          return 0;
        }
        return quantizer.push(sample.rms);
      }),
      distinctUntilChanged(),
    )
    .subscribe((bars) => {
      lotusSendToHost(LotusWidgetActions.MicLevel, { bars });
    });
  return (): void => sub.unsubscribe();
}