//! Enhanced RTMP from a local encoder, used purely as a test client. Each case runs the
//! encoder on a generated test pattern and tone, publishes to the gateway with a stream key
//! from `/rtmp`, and a subscriber at the relay receives every group, verifies every
//! receipt under the gateway's key, writes each track's frames out as an elementary stream,
//! and decodes it with the encoder's tool, a conformant decoder: the count of decoded frames
//! is the count of objects the relay carried, which is the count the encoder made.
//!
//! The encoder sends no media before the gateway's `NetStream.Publish.Start`, so a proxy
//! between the two holds that answer until the subscriber is in place: every group is seen.
//!
//! These tests need the encoder binary (`WAI_TEST_FFMPEG`, or `ffmpeg` and `ffprobe` on the
//! PATH) and, for the hardware cases, the platform's media engine. They are ignored by
//! default and run where both are present: `cargo test --test ertmp_ffmpeg -- --ignored`.
//! A case whose encoder is missing fails; it is never passed over.

mod common;

use common::rtmp::stream_key;
use common::*;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::room::media;
use std::process::Command;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::timeout;

fn tool(name: &str) -> String {
    match std::env::var("WAI_TEST_FFMPEG") {
        Ok(p) if name == "ffmpeg" => p,
        Ok(p) => std::path::Path::new(&p).with_file_name(name).to_string_lossy().into_owned(),
        Err(_) => name.to_string(),
    }
}

/// A proxy from a listener of its own to the gateway's RTMP port; it holds the gateway's
/// bytes once the publish is admitted until `ready` is set.
async fn proxy(w: &World, ready: std::sync::Arc<std::sync::atomic::AtomicBool>) -> u16 {
    let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = l.local_addr().unwrap().port();
    let up = w.gateway.rtmp_addr.unwrap();
    let gw = w.gateway.gateway.clone();
    tokio::spawn(async move {
        let (c, _) = l.accept().await.unwrap();
        let s = TcpStream::connect(up).await.unwrap();
        let (mut cr, mut cw) = c.into_split();
        let (mut sr, mut sw) = s.into_split();
        tokio::spawn(async move {
            let mut b = vec![0u8; 65536];
            while let Ok(n) = cr.read(&mut b).await {
                if n == 0 || sw.write_all(&b[..n]).await.is_err() {
                    break;
                }
            }
            let _ = sw.shutdown().await;
        });
        let mut b = vec![0u8; 65536];
        while let Ok(n) = sr.read(&mut b).await {
            if n == 0 {
                break;
            }
            if gw.rtmp_publishing() {
                while !ready.load(std::sync::atomic::Ordering::Relaxed) {
                    tokio::time::sleep(Duration::from_millis(5)).await;
                }
            }
            if cw.write_all(&b[..n]).await.is_err() {
                break;
            }
        }
    });
    port
}

struct Case {
    name: &'static str,
    /// The encoder's arguments between its inputs and its output.
    args: &'static [&'static str],
    video: Codec,
    audio: Option<Codec>,
    /// The elementary stream's extension, as the decoder reads it.
    vformat: &'static str,
    aformat: Option<&'static str>,
}

const FRAMES: usize = 90;

async fn run(c: &Case) {
    let w = World::start(|g| g.rtmp.listen = Some("127.0.0.1:0".parse().unwrap())).await;
    let auth = w.auth(&w.publisher, 0);
    let key = stream_key(&w, &auth).await;
    let ready = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
    let port = proxy(&w, ready.clone()).await;
    let url = format!("rtmp://127.0.0.1:{port}/live/{key}");
    let mut args: Vec<String> = ["-hide_banner", "-loglevel", "error", "-re", "-f", "lavfi", "-i", "testsrc2=size=640x360:rate=30", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000"]
        .iter()
        .map(|s| s.to_string())
        .collect();
    args.extend(["-frames:v".into(), FRAMES.to_string(), "-t".into(), "3".into()]);
    args.extend(c.args.iter().map(|s| s.to_string()));
    if c.audio.is_none() {
        args.push("-an".into());
    }
    args.extend(["-f".into(), "flv".into(), url]);
    let mut child = tokio::process::Command::new(tool("ffmpeg")).args(&args).spawn().unwrap_or_else(|e| panic!("{}: the encoder ({}): {e}", c.name, tool("ffmpeg")));
    // Subscribe once the publish is admitted, then let the encoder start.
    assert!(until(STEP, || w.gateway.gateway.rtmp_publishing()).await, "{}: the publish is admitted", c.name);
    let kinds: Vec<u8> = if c.audio.is_some() { vec![media::AUDIO, media::VIDEO] } else { vec![media::VIDEO] };
    let watcher = watch_kinds(&w, &kinds).await;
    ready.store(true, std::sync::atomic::Ordering::Relaxed);
    let st = timeout(Duration::from_secs(60), child.wait()).await.expect("the encoder finishes").unwrap();
    assert!(st.success(), "{}: the encoder's status {st}", c.name);
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == kinds.len()).await, "{}: every track ends", c.name);
    settle(&watcher, Duration::from_millis(400)).await;
    let dir = std::env::temp_dir().join(format!("ertmp-ffmpeg-{}-{}", c.name, std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let (frames, groups) = verified(&watcher, video, &w.gateway_key, c.video);
    assert_eq!(frames.len(), FRAMES, "{}: every frame the encoder made reached the subscriber", c.name);
    assert!(groups >= 3, "{}: {groups} groups", c.name);
    let env = envelopes(&watcher, video);
    let es = elementary(c.video, &env.iter().map(|e| e.payload.clone()).collect::<Vec<_>>());
    let vpath = dir.join(format!("video.{}", c.vformat));
    std::fs::write(&vpath, es).unwrap();
    assert_eq!(decoded(&vpath, c.vformat, "v"), FRAMES, "{}: the decoder decodes every frame", c.name);
    if let Some(ac) = c.audio {
        let audio = watcher.tracks.iter().find(|t| t.kind == media::AUDIO).unwrap();
        let (af_, _) = verified(&watcher, audio, &w.gateway_key, ac);
        assert!(!af_.is_empty(), "{}: audio arrived", c.name);
        // An Opus packet is not a stream a decoder's command line reads without a
        // container, so its packets are verified and counted, not decoded here.
        if let Some(af) = c.aformat {
            let env = envelopes(&watcher, audio);
            let es = elementary(ac, &env.iter().map(|e| e.payload.clone()).collect::<Vec<_>>());
            let apath = dir.join(format!("audio.{af}"));
            std::fs::write(&apath, es).unwrap();
            assert_eq!(decoded(&apath, af, "a"), env.len(), "{}: the decoder decodes every audio frame", c.name);
        }
    }
    assert_eq!(w.relay.relay().stats().refused_at_door, 0, "{}: the relay's door verified every object", c.name);
    assert!(until(STEP, || !w.gateway.gateway.records().is_empty()).await);
    let rec = w.gateway.gateway.records().remove(0);
    rec.verify().unwrap();
    let counts = rec.record.detail["counts"].as_object().unwrap();
    eprintln!("{}: what the gateway read and did not carry: {counts:?}", c.name);
    assert!(counts.keys().all(|k| k == "metadata" || k == "packet_not_carried"), "{}: no frame refused: {counts:?}", c.name);
    eprintln!("{}: {} video objects in {groups} groups, catalog codec from the configuration; decoded", c.name, frames.len());
    let _ = std::fs::remove_dir_all(&dir);
    w.stop().await;
}

/// The objects' payloads as one elementary stream the decoder reads.
fn elementary(codec: Codec, payloads: &[Vec<u8>]) -> Vec<u8> {
    match codec {
        // Annex B access units and ADTS frames concatenate.
        Codec::Avc | Codec::Hevc | Codec::Aac => payloads.concat(),
        // SPEC §5's AV1 framing: 24 bytes, then the temporal unit (a low-overhead stream).
        Codec::Av1 => payloads.iter().flat_map(|p| p[24..].to_vec()).collect(),
        // One-frame IVF files: the first's file header, then each frame header and frame.
        Codec::Vp8 | Codec::Vp9 => {
            let mut o = payloads[0][..32].to_vec();
            o[24..28].copy_from_slice(&(payloads.len() as u32).to_le_bytes());
            for p in payloads {
                o.extend_from_slice(&p[32..]);
            }
            o
        }
        // A FLAC stream: the first object's, then the frames of the others.
        Codec::Flac => {
            let mut o = payloads[0].clone();
            for p in &payloads[1..] {
                o.extend_from_slice(&p[42..]);
            }
            o
        }
        _ => payloads.concat(),
    }
}

/// The frames the decoder decodes from `path` read as `format`, by the stream type `s`.
fn decoded(path: &std::path::Path, format: &str, s: &str) -> usize {
    let fmt = format;
    let out = Command::new(tool("ffprobe"))
        .args(["-v", "error", "-f", fmt, "-count_frames", "-select_streams", &format!("{s}:0"), "-show_entries", "stream=nb_read_frames", "-of", "csv=p=0"])
        .arg(path)
        .output()
        .expect("the decoder runs");
    assert!(out.status.success(), "decoding {}: {}", path.display(), String::from_utf8_lossy(&out.stderr));
    String::from_utf8_lossy(&out.stdout).trim().trim_end_matches(',').parse().unwrap_or_else(|_| panic!("a frame count: {:?}", String::from_utf8_lossy(&out.stdout)))
}

macro_rules! case {
    ($fn:ident, $c:expr) => {
        #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
        #[ignore = "needs a local encoder (WAI_TEST_FFMPEG or ffmpeg on the PATH); run with --ignored"]
        async fn $fn() {
            timeout(Duration::from_secs(120), run(&$c)).await.expect("in time");
        }
    };
}

case!(legacy_avc_with_b_frames_and_aac, Case { name: "x264-aac", args: &["-c:v", "libx264", "-preset", "veryfast", "-g", "30", "-bf", "2", "-c:a", "aac", "-b:a", "96k"], video: Codec::Avc, audio: Some(Codec::Aac), vformat: "h264", aformat: Some("aac") });
case!(hevc_on_the_media_engine_and_aac, Case { name: "hw-hevc-aac", args: &["-c:v", "hevc_videotoolbox", "-g", "30", "-b:v", "2M", "-c:a", "aac", "-b:a", "96k"], video: Codec::Hevc, audio: Some(Codec::Aac), vformat: "hevc", aformat: Some("aac") });
case!(h264_on_the_media_engine, Case { name: "hw-h264", args: &["-c:v", "h264_videotoolbox", "-g", "30", "-b:v", "2M"], video: Codec::Avc, audio: None, vformat: "h264", aformat: None });
case!(hevc_in_software_and_opus, Case { name: "x265-opus", args: &["-c:v", "libx265", "-preset", "ultrafast", "-x265-params", "keyint=30:min-keyint=30:log-level=error", "-c:a", "libopus", "-b:a", "96k"], video: Codec::Hevc, audio: Some(Codec::Opus), vformat: "hevc", aformat: None });
case!(av1_and_flac, Case { name: "svtav1-flac", args: &["-c:v", "libsvtav1", "-preset", "12", "-g", "30", "-svtav1-params", "keyint=30:scd=0", "-c:a", "flac"], video: Codec::Av1, audio: Some(Codec::Flac), vformat: "obu", aformat: Some("flac") });
case!(vp9_and_opus, Case { name: "vp9-opus", args: &["-c:v", "libvpx-vp9", "-deadline", "realtime", "-cpu-used", "8", "-g", "30", "-b:v", "1M", "-c:a", "libopus"], video: Codec::Vp9, audio: Some(Codec::Opus), vformat: "ivf", aformat: None });