element-call/src/lotus/lotusCallState.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 { combineLatest, of, type Subscription } from "rxjs";
import {
  distinctUntilChanged,
  map,
  switchMap,
  throttleTime,
} from "rxjs/operators";

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

interface ParticipantState {
  /** EC media id (`${userId}:${deviceId}`), stable per participant device. */
  id: string;
  /** Matrix user id this media belongs to. */
  userId: string;
  speaking: boolean;
  audioEnabled: boolean;
  videoEnabled: boolean;
}

/**
 * [lotus #20] Field-wise equality for `ParticipantState[]`, used in place of
 * `JSON.stringify` comparison: cheaper (no serialisation of every
 * participant on every emission) and just as correct, since array order here
 * is stable (it mirrors `members` from `userMedia$`).
 */
function participantsEqual(
  a: ParticipantState[],
  b: ParticipantState[],
): boolean {
  return (
    a.length === b.length &&
    a.every(
      (p, i) =>
        p.id === b[i].id &&
        p.userId === b[i].userId &&
        p.speaking === b[i].speaking &&
        p.audioEnabled === b[i].audioEnabled &&
        p.videoEnabled === b[i].videoEnabled,
    )
  );
}

/**
 * Stream per-participant speaking / mute / camera state to the Lotus host
 * (cinny) over the widget API, so the host can drive speaking rings, mute
 * badges and PiP from real events instead of scraping Element Call's rendered
 * DOM (`useCallSpeakers.ts`).
 *
 * Opt-in: does nothing unless the host set `lotusCallState=1` on the widget
 * URL. Returns a teardown function.
 */
export function startLotusCallState(vm: CallViewModel): () => void {
  if (!lotusFlag("lotusCallState")) return () => undefined;
  // [lotus] Standalone (non-widget) mode has no host to send state to;
  // skip building the whole stream pipeline, mirroring lotusFocus.ts /
  // lotusDecorations.ts.
  if (!widget) return () => undefined;

  const sub: Subscription = vm.userMedia$
    .pipe(
      switchMap((members) =>
        members.length === 0
          ? of([] as ParticipantState[])
          : combineLatest(
              members.map((m) =>
                combineLatest([
                  m.speaking$,
                  m.audioEnabled$,
                  m.videoEnabled$,
                ]).pipe(
                  map(
                    ([
                      speaking,
                      audioEnabled,
                      videoEnabled,
                    ]): ParticipantState => ({
                      id: m.id,
                      userId: m.userId,
                      speaking,
                      audioEnabled,
                      videoEnabled,
                    }),
                  ),
                ),
              ),
            ),
      ),
      // `speaking` flips rapidly; drop no-op repeats BEFORE throttling so
      // the throttle window isn't spent re-emitting an unchanged value.
      // Field-wise (cheaper than JSON.stringify, and correct: array order is
      // stable since it mirrors `members` from userMedia$).
      distinctUntilChanged(participantsEqual),
      // [lotus #20] `speaking` is the field that flips constantly in an active
      // conversation; mute/camera toggles are rare and user-intentional and
      // would ideally stay prompt, but a single combined stream is far
      // simpler than splitting it, and the leading+trailing 250ms window
      // previously allowed ~8 sends/sec (each re-serialising every
      // participant) in a busy call. Trailing-only + a longer window caps
      // that to 2/sec while still reflecting mute/camera changes within
      // 500ms.
      throttleTime(500, undefined, { leading: false, trailing: true }),
    )
    .subscribe((participants) => {
      lotusSendToHost(LotusWidgetActions.CallState, { participants });
    });

  return () => sub.unsubscribe();
}