arcly-stream 0.8.0

//! Conference rooms: a thin multi-party model over the bus.
//!
//! A conference is N publishers who each also subscribe to the others. Rather
//! than a new routing layer, a [`Room`] maps onto primitives the bus already
//! provides: a room is an **application**, each participant is a **stream** in it
//! (`<room>/<participant>`), and join/leave are ordinary
//! [`PublishStarted`](crate::bus::StreamEventKind::PublishStarted) /
//! [`PublishEnded`](crate::bus::StreamEventKind::PublishEnded) events.
//!
//! So the existing WHIP (publish a participant's track) and WHEP (subscribe to a
//! peer) paths compose into a conference: a participant WHIP-publishes to
//! [`participant_key`](Room::participant_key), enumerates the others with
//! [`peers`](Room::peers), and opens a WHEP egress to each. Per-participant
//! adaptive bitrate and the keyframe/feedback loop come for free from the kernel
//! WebRTC plane; this type only adds the room membership view.
//!
//! Simulcast/ABR layer siblings (`<participant>~<rid>`) are *not* participants —
//! [`peers`](Room::peers) filters them out via
//! [`StreamKey::layer_base`](crate::StreamKey::layer_base).

use crate::bus::{EventBus, PlaybackRegistry, StreamEvent};
use crate::{AppName, Result, StreamId, StreamKey};
use std::collections::HashMap;
use tokio::sync::broadcast;

/// A conference room: an application whose streams are its participants.
#[derive(Debug, Clone)]
pub struct Room {
    app: AppName,
}

impl Room {
    /// Open a handle to the room named `name` (its application name).
    pub fn new(name: impl Into<AppName>) -> Self {
        Self { app: name.into() }
    }

    /// The room's name (its application name).
    pub fn name(&self) -> &AppName {
        &self.app
    }

    /// The stream key a `participant` publishes their track to: `<room>/<participant>`.
    pub fn participant_key(&self, participant: &str) -> StreamKey {
        StreamKey::new(self.app.as_str(), participant)
    }

    /// The participants currently publishing in the room, excluding `me` and any
    /// adaptive-bitrate layer siblings (only base streams are participants).
    ///
    /// This is the set a participant opens a WHEP egress to. Pass an empty `me`
    /// to list everyone.
    pub fn peers(&self, playback: &dyn PlaybackRegistry, me: &str) -> Result<Vec<StreamId>> {
        let mut out: Vec<StreamId> = playback
            .list_streams(&self.app)?
            .into_iter()
            .filter(|id| {
                let s = id.as_str();
                // Base streams only (drop `~rid`/`~rung` layer siblings) and not me.
                !s.contains('~') && s != me
            })
            .collect();
        out.sort_by(|a, b| a.as_str().cmp(b.as_str()));
        Ok(out)
    }

    /// Subscribe to the room's lifecycle feed — participant join
    /// ([`PublishStarted`]) and leave ([`PublishEnded`]) arrive here, so a server
    /// can renegotiate each peer's subscriptions as the room changes.
    ///
    /// [`PublishStarted`]: crate::bus::StreamEventKind::PublishStarted
    /// [`PublishEnded`]: crate::bus::StreamEventKind::PublishEnded
    pub fn events(&self, bus: &dyn EventBus) -> Result<broadcast::Receiver<StreamEvent>> {
        bus.subscribe_events(&self.app)
    }
}

/// Tracks the room's **dominant speaker** from per-participant RFC 6464 audio
/// levels (decode each ingress packet's level with [`audio_level`]).
///
/// Each participant's loudness is smoothed with an exponential moving average so
/// a single loud packet doesn't win; the speaker only changes when a challenger
/// beats the incumbent by `switch_margin` loudness points, which prevents rapid
/// flapping between two similar voices. Loudness is `127 - level_dbov` while
/// voice is active (louder = higher), and `0` otherwise.
///
/// [`audio_level`]: crate::protocol::rtp::audio_level
#[derive(Debug, Clone)]
pub struct DominantSpeaker {
    scores: HashMap<String, f32>,
    current: Option<String>,
    switch_margin: f32,
}

impl DominantSpeaker {
    /// New tracker; `switch_margin` is the loudness lead (0–127 scale) a
    /// challenger needs over the current speaker to take over (e.g. `8.0`).
    pub fn new(switch_margin: f32) -> Self {
        Self {
            scores: HashMap::new(),
            current: None,
            switch_margin,
        }
    }

    /// Feed one audio-level observation for `participant`. Returns `Some(name)`
    /// when the dominant speaker *changes* as a result, else `None`.
    pub fn observe(&mut self, participant: &str, level_dbov: u8, voice: bool) -> Option<&str> {
        let loudness = if voice {
            (127u8.saturating_sub(level_dbov)) as f32
        } else {
            0.0
        };
        // EMA per participant; decays an idle speaker toward zero over time.
        let score = self.scores.entry(participant.to_string()).or_insert(0.0);
        *score = *score * 0.9 + loudness * 0.1;

        // Loudest participant right now.
        let (best, best_score) = self
            .scores
            .iter()
            .max_by(|a, b| a.1.partial_cmp(b.1).unwrap_or(std::cmp::Ordering::Equal))
            .map(|(k, v)| (k.clone(), *v))?;

        let take_over = match &self.current {
            Some(cur) if cur == &best => false,
            Some(cur) => {
                let cur_score = self.scores.get(cur).copied().unwrap_or(0.0);
                best_score > cur_score + self.switch_margin
            }
            None => best_score > 0.0,
        };
        if take_over {
            self.current = Some(best);
            self.current.as_deref()
        } else {
            None
        }
    }

    /// The current dominant speaker, if any.
    pub fn current(&self) -> Option<&str> {
        self.current.as_deref()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn peers_lists_base_participants_excluding_self_and_layers() {
        use crate::inbound::IngestContext;

        let engine = crate::Engine::builder()
            .application(crate::AppSpec::new("standup").gop_cache(2))
            .build();
        let ctx = IngestContext::new(engine.clone());
        let room = Room::new("standup");

        // Three participants; "bob" also publishes a simulcast layer sibling.
        for id in ["alice", "bob", "carol", "bob~h"] {
            let s = ctx.open_publish(room.participant_key(id)).await.unwrap();
            std::mem::forget(s); // keep live for the listing
        }

        // From alice's view: the other base participants, no self, no `bob~h`.
        let peers = room.peers(engine.as_ref(), "alice").unwrap();
        let ids: Vec<&str> = peers.iter().map(|i| i.as_str()).collect();
        assert_eq!(ids, vec!["bob", "carol"]);
    }

    #[test]
    fn participant_key_is_room_scoped() {
        let room = Room::new("standup");
        let k = room.participant_key("alice");
        assert_eq!(k.app.as_str(), "standup");
        assert_eq!(k.stream_id.as_str(), "alice");
    }

    #[test]
    fn dominant_speaker_picks_loudest_and_holds_with_hysteresis() {
        let mut ds = DominantSpeaker::new(8.0);
        // Alice speaks loudly (low dBov) → becomes dominant.
        let mut became = None;
        for _ in 0..20 {
            if let Some(s) = ds.observe("alice", 10, true) {
                became = Some(s.to_string());
            }
            ds.observe("bob", 90, false);
        }
        assert_eq!(ds.current(), Some("alice"));
        assert_eq!(became.as_deref(), Some("alice"));

        // Bob now slightly louder than alice but within the margin → no switch.
        for _ in 0..3 {
            assert_eq!(ds.observe("bob", 8, true), None);
            ds.observe("alice", 12, true);
        }
        assert_eq!(
            ds.current(),
            Some("alice"),
            "within-margin lead keeps alice"
        );

        // Bob becomes decisively louder for a sustained period → takes over.
        // A change can be reported on *any* observe (the moment bob overtakes
        // alice may land on either participant's update), so check both.
        let mut switched = false;
        for _ in 0..40 {
            switched |= ds.observe("bob", 0, true).is_some();
            switched |= ds.observe("alice", 100, false).is_some();
        }
        assert!(switched && ds.current() == Some("bob"));
    }
}