openrtc 2.9.1

//! Native Draft-14 publication mechanics for a non-peer broadcast session.
//!
//! The shared `BroadcastSession` owns admission, signing, generations, media
//! budgets, and terminal decisions. This mechanism only sends its exact
//! generation-bound actions to the same Draft-14 track used by browser viewers.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use anyhow::{anyhow, bail, ensure, Context, Result};
use bytes::Bytes;
use moq_transport::coding::TrackNamespace;
use moq_transport::serve;
use tokio::sync::{Mutex, Notify};
use url::Url;

use super::{BroadcastAdapterAction, BroadcastAdapterObservation, BroadcastRole, BroadcastSession};

const TRACK_NAME: &str = "openrtc-broadcast-1";
const MAX_RETAINED_OBJECTS: usize = 128;
const MAX_RETAINED_BYTES: usize = super::MAX_BROADCAST_MEDIA_BYTES as usize;
const WRITE_TIMEOUT: Duration = Duration::from_secs(2);
const FIRST_VIEWER_TIMEOUT: Duration = Duration::from_secs(60);

/// A short-lived relay allocation supplied by the app's authorized grant
/// service. Secrets stay in the native process and never enter creator pages.
pub struct Draft14RelayAccess {
    url: Url,
    token: String,
    namespace: String,
}

impl Draft14RelayAccess {
    pub fn new(relay_url: &str, token: String, namespace: String) -> Result<Self> {
        let url = Url::parse(relay_url).context("broadcast relay URL is invalid")?;
        ensure!(
            url.scheme() == "https"
                && url.username().is_empty()
                && url.password().is_none()
                && url.query().is_none()
                && url.fragment().is_none(),
            "broadcast relay URL must be HTTPS without credentials or query"
        );
        ensure!(
            (16..=16 * 1024).contains(&token.len())
                && token
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte)),
            "broadcast relay token is invalid"
        );
        ensure!(
            !namespace.is_empty()
                && namespace.len() <= 512
                && namespace
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || b"._/-".contains(&byte))
                && namespace
                    .split('/')
                    .all(|segment| !segment.is_empty() && segment != "." && segment != ".."),
            "broadcast relay namespace is invalid"
        );
        Ok(Self {
            url,
            token,
            namespace,
        })
    }

    fn connection_url(&self) -> Result<Url> {
        let mut url = self.url.clone();
        url.path_segments_mut()
            .map_err(|_| anyhow!("broadcast relay URL has no path segments"))?
            .pop_if_empty()
            .push(&self.token);
        Ok(url)
    }

    fn is_loopback(&self) -> bool {
        self.url
            .host_str()
            .is_some_and(|host| matches!(host, "localhost" | "127.0.0.1" | "[::1]"))
    }
}

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

    #[test]
    fn relay_access_keeps_token_in_path_and_rejects_unsafe_inputs() {
        let access = Draft14RelayAccess::new(
            "https://127.0.0.1:4443/moq",
            "test-token-123456".to_string(),
            "live/one".to_string(),
        )
        .unwrap();
        assert_eq!(
            access.connection_url().unwrap().as_str(),
            "https://127.0.0.1:4443/moq/test-token-123456"
        );
        assert!(access.is_loopback());
        assert!(Draft14RelayAccess::new(
            "https://example.com/moq?token=secret",
            "test-token-123456".to_string(),
            "live/one".to_string(),
        )
        .is_err());
        assert!(Draft14RelayAccess::new(
            "https://example.com/moq",
            "test-token-123456".to_string(),
            "live/../other".to_string(),
        )
        .is_err());
    }
}

struct TrackWriter {
    writer: serve::SubgroupWriter,
    retained_sizes: VecDeque<usize>,
    retained_bytes: usize,
}

impl TrackWriter {
    async fn write(&mut self, object: Vec<u8>) -> Result<()> {
        ensure!(
            object.len() <= MAX_RETAINED_BYTES,
            "broadcast relay object exceeds its bound"
        );
        let object = Bytes::from(object);
        tokio::time::timeout(WRITE_TIMEOUT, async {
            loop {
                while self.retained_sizes.len() > self.writer.len() {
                    if let Some(bytes) = self.retained_sizes.pop_front() {
                        self.retained_bytes = self.retained_bytes.saturating_sub(bytes);
                    }
                }
                if self.retained_sizes.len() < MAX_RETAINED_OBJECTS
                    && self.retained_bytes.saturating_add(object.len()) <= MAX_RETAINED_BYTES
                {
                    self.writer
                        .write(object.clone())
                        .map_err(|_| anyhow!("broadcast relay object write failed"))?;
                    self.retained_sizes.push_back(object.len());
                    self.retained_bytes = self.retained_bytes.saturating_add(object.len());
                    return Ok(());
                }
                tokio::time::sleep(Duration::from_millis(1)).await;
            }
        })
        .await
        .map_err(|_| anyhow!("broadcast relay backpressure timed out"))?
    }
}

/// Runs only the physical publisher mechanics for one already-verified
/// `BroadcastSession`. The caller invokes `flush` after publication changes.
pub struct NativeDraft14Publisher {
    session: BroadcastSession,
    source_slot: String,
    broadcast_generation: u64,
    grant_generation: u64,
    writer: Arc<Mutex<Option<TrackWriter>>>,
    writer_ready: Arc<Notify>,
    retired: Arc<AtomicBool>,
    wait_for_first_viewer: bool,
    _endpoint: moq_native_ietf::quic::Client,
    tasks: Vec<tokio::task::JoinHandle<()>>,
}

impl NativeDraft14Publisher {
    pub async fn connect(session: BroadcastSession, access: Draft14RelayAccess) -> Result<Self> {
        Self::connect_with_viewer_policy(session, access, true).await
    }

    /// For continuously running sources such as OBS, let signed media before
    /// the first track request expire. The session repeats publication control
    /// and H.264 keyframes restart decoding when a viewer attaches.
    pub async fn connect_live(
        session: BroadcastSession,
        access: Draft14RelayAccess,
    ) -> Result<Self> {
        Self::connect_with_viewer_policy(session, access, false).await
    }

    async fn connect_with_viewer_policy(
        session: BroadcastSession,
        access: Draft14RelayAccess,
        wait_for_first_viewer: bool,
    ) -> Result<Self> {
        ensure!(
            matches!(
                session.role(),
                BroadcastRole::Owner | BroadcastRole::Publisher | BroadcastRole::Processor
            ),
            "broadcast grant cannot publish"
        );
        let source_slots = session.source_slots();
        ensure!(
            source_slots.len() == 1,
            "native demo publisher requires one source slot"
        );
        let source_slot = source_slots[0].clone();
        let actions = session.take_actions(1);
        let Some(BroadcastAdapterAction::Open {
            broadcast_generation,
            grant_generation,
            ..
        }) = actions.into_iter().next()
        else {
            bail!("broadcast session has no opening action");
        };
        let result = Self::connect_open(
            session.clone(),
            access,
            source_slot,
            broadcast_generation,
            grant_generation,
            wait_for_first_viewer,
        )
        .await;
        if result.is_err() {
            let _ = session.observe(BroadcastAdapterObservation::Failed {
                broadcast_generation,
                grant_generation,
                retryable: false,
                code: "relay-open-failed".to_string(),
            });
        }
        result
    }

    async fn connect_open(
        session: BroadcastSession,
        access: Draft14RelayAccess,
        source_slot: String,
        broadcast_generation: u64,
        grant_generation: u64,
        wait_for_first_viewer: bool,
    ) -> Result<Self> {
        let connection_url = access.connection_url()?;
        let tls_args = moq_native_ietf::tls::Args {
            disable_verify: access.is_loopback(),
            ..Default::default()
        };
        let tls = tls_args
            .load()
            .context("broadcast relay TLS configuration failed")?;
        let quic = moq_native_ietf::quic::Args {
            tls: tls_args,
            ..Default::default()
        }
        .load()
        .or_else(|_| {
            moq_native_ietf::quic::Config::new(
                "0.0.0.0:0".parse().expect("valid fallback bind"),
                None,
                tls,
            )
        })
        .context("broadcast relay QUIC configuration failed")?;
        let endpoint = moq_native_ietf::quic::Endpoint::new(quic)
            .context("broadcast relay endpoint creation failed")?;
        let endpoint_client = endpoint.client;
        let (transport, _, connection) = endpoint_client
            .connect(&connection_url, None)
            .await
            .map_err(|_| anyhow!("broadcast relay connection failed"))?;
        let (wire_session, mut publisher, _) =
            moq_transport::session::Session::connect(transport, None, connection)
                .await
                .context("broadcast relay Draft-14 setup failed")?;

        let writer = Arc::new(Mutex::new(None));
        let writer_ready = Arc::new(Notify::new());
        let retired = Arc::new(AtomicBool::new(false));
        let run_retired = retired.clone();
        let run_session = session.clone();
        let run_ready = writer_ready.clone();
        let run_task = tokio::spawn(async move {
            let _ = wire_session.run().await;
            if !run_retired.swap(true, Ordering::AcqRel) {
                run_ready.notify_waiters();
                let _ = run_session.observe(BroadcastAdapterObservation::Failed {
                    broadcast_generation,
                    grant_generation,
                    retryable: false,
                    code: "relay-session-closed".to_string(),
                });
            }
        });

        let track_namespace = TrackNamespace::from_utf8_path(&format!(
            "{}/source/{source_slot}",
            access.namespace.trim_end_matches('/')
        ));
        let (_tracks_writer, mut requests, tracks_reader) =
            serve::Tracks::new(track_namespace).produce();
        let announce_writer = writer.clone();
        let announce_ready = writer_ready.clone();
        let announce_retired = retired.clone();
        let announce_session = session.clone();
        let announce_task = tokio::spawn(async move {
            let announce = publisher.announce(tracks_reader);
            tokio::pin!(announce);
            loop {
                tokio::select! {
                    result = &mut announce => {
                        if !announce_retired.swap(true, Ordering::AcqRel) {
                            announce_ready.notify_waiters();
                            let _ = announce_session.observe(BroadcastAdapterObservation::Failed {
                                broadcast_generation,
                                grant_generation,
                                retryable: false,
                                code: if result.is_ok() { "relay-publication-closed" } else { "relay-publication-failed" }.to_string(),
                            });
                        }
                        return;
                    }
                    requested = requests.next() => {
                        let Some(track) = requested else { return; };
                        if track.name != TRACK_NAME {
                            let _ = track.close(serve::ServeError::NotFound);
                            continue;
                        }
                        if let Ok(subgroup) = track.subgroups().and_then(|mut groups| groups.append(0)) {
                            *announce_writer.lock().await = Some(TrackWriter {
                                writer: subgroup,
                                retained_sizes: VecDeque::new(),
                                retained_bytes: 0,
                            });
                            announce_ready.notify_one();
                        }
                    }
                }
            }
        });
        let mechanism = Self {
            session,
            source_slot,
            broadcast_generation,
            grant_generation,
            writer,
            writer_ready,
            retired,
            wait_for_first_viewer,
            _endpoint: endpoint_client,
            tasks: vec![run_task, announce_task],
        };
        mechanism
            .session
            .observe(BroadcastAdapterObservation::Opened {
                broadcast_generation,
                grant_generation,
            })
            .context("broadcast opening observation rejected")?;
        Ok(mechanism)
    }

    pub async fn flush(&self) -> Result<()> {
        for action in self.session.take_actions(super::MAX_BROADCAST_ACTIONS) {
            match action {
                BroadcastAdapterAction::Publish {
                    broadcast_generation,
                    grant_generation,
                    source_slot,
                    object,
                    ..
                } if broadcast_generation == self.broadcast_generation
                    && grant_generation == self.grant_generation
                    && source_slot == self.source_slot =>
                {
                    self.publish(object).await?;
                }
                BroadcastAdapterAction::Close {
                    broadcast_generation,
                    grant_generation,
                    reason: _,
                } if broadcast_generation == self.broadcast_generation
                    && grant_generation == self.grant_generation =>
                {
                    self.retired.store(true, Ordering::Release);
                    return Ok(());
                }
                _ => bail!("broadcast adapter received an unexpected action"),
            }
        }
        Ok(())
    }

    async fn publish(&self, object: Vec<u8>) -> Result<()> {
        let result = tokio::time::timeout(FIRST_VIEWER_TIMEOUT, async {
            loop {
                let notified = self.writer_ready.notified();
                if self.retired.load(Ordering::Acquire) {
                    bail!("broadcast relay session has closed");
                }
                let mut writer = self.writer.lock().await;
                if let Some(writer) = writer.as_mut() {
                    return writer.write(object).await;
                }
                if !self.wait_for_first_viewer {
                    return Ok(());
                }
                drop(writer);
                notified.await;
            }
        })
        .await
        .unwrap_or_else(|_| Err(anyhow!("broadcast relay has no viewer track request")));
        if let Err(error) = result {
            let _ = self.session.observe(BroadcastAdapterObservation::Failed {
                broadcast_generation: self.broadcast_generation,
                grant_generation: self.grant_generation,
                retryable: false,
                code: "relay-publish-failed".to_string(),
            });
            return Err(error);
        }
        Ok(())
    }

    pub fn close(mut self) {
        self.retired.store(true, Ordering::Release);
        for task in self.tasks.drain(..) {
            task.abort();
        }
        self.session.close();
    }
}

impl Drop for NativeDraft14Publisher {
    fn drop(&mut self) {
        self.retired.store(true, Ordering::Release);
        for task in self.tasks.drain(..) {
            task.abort();
        }
        self.session.close();
    }
}