lesavka/client/src/sync_probe/runner.rs

//! Runtime entrypoint for the shared-clock upstream sync probe.

use anyhow::{Context, Result, bail};

use crate::app_support;
use crate::handshake;
use crate::sync_probe::capture::SyncProbeCapture;
use crate::sync_probe::config::{ParseOutcome, ProbeConfig, parse_args_outcome_from, usage};
use crate::sync_probe::schedule::PulseSchedule;

#[cfg(not(coverage))]
use lesavka_common::lesavka::relay_client::RelayClient;
#[cfg(not(coverage))]
use tonic::{Request, transport::Channel};

pub async fn run_sync_probe_from_args<I, S>(args: I) -> Result<()>
where
    I: IntoIterator<Item = S>,
    S: Into<String>,
{
    match parse_args_outcome_from(args)? {
        ParseOutcome::Run(config) => run_sync_probe(config).await,
        ParseOutcome::Help => {
            println!("{}", usage());
            Ok(())
        }
    }
}

#[cfg(not(coverage))]
async fn run_sync_probe(config: ProbeConfig) -> Result<()> {
    let caps = handshake::negotiate(config.server.as_str()).await;
    if !caps.camera || !caps.microphone {
        bail!("server does not advertise both camera and microphone support");
    }
    let camera = app_support::camera_config_from_caps(&caps)
        .context("server handshake did not include a complete camera profile")?;

    let schedule = PulseSchedule::new(
        config.warmup,
        config.pulse_period,
        config.pulse_width,
        config.marker_tick_period,
    );
    let capture = SyncProbeCapture::new(camera, schedule, config.duration)?;

    tracing::info!(
        server = %config.server,
        duration_s = config.duration.as_secs(),
        codec = ?camera.codec,
        width = camera.width,
        height = camera.height,
        fps = camera.fps,
        "🧪 A/V sync probe starting"
    );

    let video_channel = connect(config.server.as_str()).await?;
    let audio_channel = connect(config.server.as_str()).await?;
    let video_queue = capture.video_queue();
    let audio_queue = capture.audio_queue();

    let video_task = tokio::spawn(async move {
        let mut client = RelayClient::new(video_channel);
        let outbound = async_stream::stream! {
            loop {
                let next = video_queue.pop_fresh().await;
                if let Some(packet) = next.packet {
                    yield packet;
                    continue;
                }
                break;
            }
        };
        let mut response = client
            .stream_camera(Request::new(outbound))
            .await
            .context("starting sync probe camera stream")?;
        while response.get_mut().message().await.transpose().is_some() {}
        Ok::<(), anyhow::Error>(())
    });

    let audio_task = tokio::spawn(async move {
        let mut client = RelayClient::new(audio_channel);
        let outbound = async_stream::stream! {
            loop {
                let next = audio_queue.pop_fresh().await;
                if let Some(packet) = next.packet {
                    yield packet;
                    continue;
                }
                break;
            }
        };
        let mut response = client
            .stream_microphone(Request::new(outbound))
            .await
            .context("starting sync probe microphone stream")?;
        while response.get_mut().message().await.transpose().is_some() {}
        Ok::<(), anyhow::Error>(())
    });

    let (video_result, audio_result) =
        tokio::try_join!(video_task, audio_task).context("joining sync probe streams")?;
    video_result.context("sync probe camera task failed")?;
    audio_result.context("sync probe microphone task failed")?;

    tracing::info!("🧪 A/V sync probe finished");
    Ok(())
}

#[cfg(not(coverage))]
async fn connect(server_addr: &str) -> Result<Channel> {
    Channel::from_shared(server_addr.to_string())
        .context("invalid relay server address")?
        .tcp_nodelay(true)
        .connect()
        .await
        .with_context(|| format!("connecting to relay at {server_addr}"))
}

#[cfg(coverage)]
async fn run_sync_probe(_config: ProbeConfig) -> Result<()> {
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::parse_args_outcome_from;
    #[cfg(coverage)]
    use super::run_sync_probe_from_args;
    use crate::sync_probe::config::ParseOutcome;

    #[test]
    fn help_passthrough_stays_stable() {
        let outcome = parse_args_outcome_from(["--help"]).expect("help");
        assert_eq!(outcome, ParseOutcome::Help);
    }

    #[cfg(coverage)]
    #[test]
    fn coverage_run_path_accepts_default_probe_args() {
        let rt = tokio::runtime::Runtime::new().expect("runtime");
        rt.block_on(async {
            run_sync_probe_from_args(std::iter::empty::<&str>())
                .await
                .expect("coverage run path");
        });
    }

    #[cfg(coverage)]
    #[test]
    fn coverage_run_path_accepts_custom_probe_args() {
        let rt = tokio::runtime::Runtime::new().expect("runtime");
        rt.block_on(async {
            run_sync_probe_from_args([
                "--server",
                "http://lab:50051",
                "--duration-seconds",
                "2",
                "--warmup-seconds",
                "5",
                "--pulse-period-ms",
                "700",
                "--pulse-width-ms",
                "80",
                "--marker-tick-period",
                "4",
            ])
            .await
            .expect("configured coverage run path");
        });
    }
}