openrtc 2.9.8

//! Disposable native publisher for the Live local Draft-14 transport harness.
//! It proves native signed encoded media reaches browser viewers. OBS/WHIP
//! ingest and the Clerk-authorized pairing flow are separate work.

use std::time::Duration;

use anyhow::{anyhow, ensure, Context, Result};
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine as _;
use clap::Parser;
use ed25519_dalek::{Signer, SigningKey, VerifyingKey};
use openrtc::broadcast::draft14::{Draft14RelayAccess, NativeDraft14Publisher};
use openrtc::broadcast::{BroadcastPublisherSigner, Broadcasts};
use openrtc::media::{
    EncodedMediaSample, MediaCodec, MediaKind, MediaPublicationConfig, PublicationId,
};
use serde::Deserialize;

#[derive(Parser)]
struct Args {
    #[arg(long, default_value = "http://localhost:8787")]
    harness_api: String,
    #[arg(long, default_value_t = 10)]
    seconds: u64,
}

#[derive(Deserialize)]
struct GrantResponse {
    grant: String,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct AccessResponse {
    issuer_public_key: String,
    relay: RelayResponse,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct RelayResponse {
    relay_url: String,
    token: String,
    namespace: String,
}

fn key(value: &str) -> Result<[u8; 32]> {
    URL_SAFE_NO_PAD
        .decode(value)
        .context("invalid issuer key")?
        .try_into()
        .map_err(|_| anyhow!("issuer key must be 32 bytes"))
}

fn pcm_tone(frame: u64) -> Vec<u8> {
    const SAMPLE_RATE: f64 = 48_000.0;
    const SAMPLES: usize = 960;
    let mut output = vec![0_u8; SAMPLES * 2];
    for (index, sample) in output.chunks_exact_mut(2).enumerate() {
        let position = frame as usize * SAMPLES + index;
        let value = ((position as f64 * 440.0 * std::f64::consts::TAU / SAMPLE_RATE).sin()
            * 2_400.0) as i16;
        sample.copy_from_slice(&value.to_le_bytes());
    }
    output
}

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();
    ensure!(
        (1..=60).contains(&args.seconds),
        "duration must be 1–60 seconds"
    );
    let api = url::Url::parse(&args.harness_api).context("invalid harness API URL")?;
    ensure!(
        api.scheme() == "http"
            && matches!(api.host_str(), Some("localhost" | "127.0.0.1" | "[::1]")),
        "the disposable grant harness must be loopback HTTP"
    );

    let mut seed = [0_u8; 32];
    getrandom::getrandom(&mut seed).context("native binding key entropy failed")?;
    let binding = SigningKey::from_bytes(&seed);
    seed.fill(0);
    let binding_public_jwk = serde_json::json!({
        "kty": "OKP", "crv": "Ed25519",
        "x": URL_SAFE_NO_PAD.encode(binding.verifying_key().to_bytes())
    });
    let publisher_signer = BroadcastPublisherSigner::generate()?;
    let publication_key = URL_SAFE_NO_PAD.encode(publisher_signer.verifying_key());
    let http = reqwest::Client::new();
    let origin = "http://localhost:5173";
    let publisher: GrantResponse = http
        .post(api.join("/api/broadcast/publisher")?)
        .header("Origin", origin)
        .json(&serde_json::json!({
            "bindingPublicJwk": binding_public_jwk,
            "publicationVerifyingKey": publication_key
        }))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;
    let access: AccessResponse = http
        .post(api.join("/api/broadcast/access")?)
        .header("Origin", origin)
        .json(&serde_json::json!({
            "grant": publisher.grant,
            "bindingPublicJwk": binding_public_jwk,
            "publicationVerifyingKey": publication_key
        }))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;
    let issuer = VerifyingKey::from_bytes(&key(&access.issuer_public_key)?)
        .context("issuer key was rejected")?;
    let now_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_millis() as u64;
    let broadcasts = Broadcasts::default();
    let challenge = broadcasts.prepare_grant_verification(&publisher.grant, &issuer, now_ms)?;
    let proof = binding.sign(&challenge.signing_bytes);
    let grant = broadcasts.complete_grant_verification(
        &publisher.grant,
        &issuer,
        &challenge.handle,
        &proof.to_bytes(),
        now_ms,
    )?;
    let session = broadcasts.open_publisher(grant, &publisher_signer, now_ms)?;
    let relay = Draft14RelayAccess::new(
        &access.relay.relay_url,
        access.relay.token,
        access.relay.namespace,
    )?;
    let carrier = NativeDraft14Publisher::connect(session.clone(), relay).await?;
    let mut publication_bytes = [0_u8; 16];
    getrandom::getrandom(&mut publication_bytes).context("publication ID entropy failed")?;
    let publication = session.begin_publication(
        "main",
        MediaPublicationConfig {
            publication_id: PublicationId(publication_bytes),
            media_generation: 0,
            kind: MediaKind::Audio,
            codec: MediaCodec::Pcm,
            clock_rate: 48_000,
            coded_width: None,
            coded_height: None,
            channels: Some(1),
        },
    )?;
    println!("Native publisher connected; waiting for a browser viewer before the tone starts");
    carrier.flush().await?;
    let frames = args.seconds * 50;
    let mut interval = tokio::time::interval(Duration::from_millis(20));
    for frame in 0..frames {
        interval.tick().await;
        session.publish(
            publication.publication_id,
            EncodedMediaSample {
                timestamp_us: frame * 20_000,
                duration_us: 20_000,
                keyframe: false,
                discardable: false,
                payload: pcm_tone(frame),
            },
        )?;
        carrier.flush().await?;
    }
    session.retire_publication(publication.publication_id)?;
    carrier.flush().await?;
    println!(
        "Native tone delivered {} frames; stats {:?}",
        frames,
        session.stats()
    );
    carrier.close();
    Ok(())
}