//! SCTE 104 into the room's cues, end to end: an automation system of our own opens an
//! API connection to the gateway's injector (SCTE 104 §9.1), initialises it, keeps it
//! alive, and sends splice and time-signal requests. Each is answered as §9.6 says; each
//! cue is signed when its splice instant comes, after its injection record, and goes out
//! on the injector principal's events track at the relay, where a subscriber reads it.
//! Every event verifies under the principal's key, chains to the one before, and every
//! cue is the translation of the message its injection record carries; the cues make the
//! breaks live-ads §4.3 derives.

mod common;

use common::*;
use ed25519_dalek::{SigningKey, Verifier};
use joule_moq::draft22::{Parameters, StreamHeader, UniEvent};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_ingest::room::{build, Descriptor};
use joule_moq_ingest::scte::scte104::{self, op, Single};
use joule_moq_ingest::scte::{EVENT_DOMAIN, KIND_CUE, KIND_META};
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::time::timeout;

const INJECTOR_SEED: [u8; 32] = [0x71; 32];

async fn world() -> World {
    // The relay's door holds the room's media tracks to their receipts; an events track
    // carries none (live-rooms §6.1), so the door names the media kinds.
    let door = |rc: &mut joule_moq_relay::RelayConfig, room: &[u8; 32], gk: &str| {
        rc.receipts.door = ["audio", "video"]
            .iter()
            .map(|k| joule_moq_relay::config::DoorRule { prefix: vec![hex::encode(room), "*".into(), k.to_string()], trust: vec![gk.to_string()], trust_field: None, hold_ms: 10_000 })
            .collect();
    };
    World::start_with(door, |c| {
        // The room's authority is World's, seed 0x51; its descriptor is the one handed in.
        let authority = SigningKey::from_bytes(&[0x51; 32]);
        let d = Descriptor::read(c.rooms[0].descriptor.as_ref().unwrap()).unwrap();
        let key = SigningKey::from_bytes(&INJECTOR_SEED);
        let s = now_us() / 1_000_000;
        let g = build::grant(&d, &authority, &key.verifying_key().to_bytes(), 0, s - 60, s + 3600);
        c.injector.listen = Some("127.0.0.1:0".parse().unwrap());
        c.injector.room = hex::encode(d.room_id());
        c.injector.key_seed = Some(INJECTOR_SEED);
        // The principal's chain, kept across a restart of this gateway (a key given in code needs a
        // place for it).
        c.injector.state_path = Some(std::env::temp_dir().join(format!("wai-ingest-injector-scte104-{}-{}.chain.json", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos())));
        c.injector.grant = Some(g.to_bytes());
    })
    .await
}

/// An automation system's API connection.
struct As {
    s: TcpStream,
}

impl As {
    async fn connect(w: &World) -> As {
        As { s: TcpStream::connect(w.gateway.injector_addr.unwrap()).await.unwrap() }
    }
    async fn send(&mut self, b: &[u8]) {
        self.s.write_all(b).await.unwrap();
    }
    async fn reply(&mut self) -> Single {
        let mut head = [0u8; 4];
        timeout(STEP, self.s.read_exact(&mut head)).await.expect("a response in time").unwrap();
        let n = u16::from_be_bytes([head[2], head[3]]) as usize;
        let mut m = head.to_vec();
        m.resize(n, 0);
        self.s.read_exact(&mut m[4..]).await.unwrap();
        Single::read(&m).unwrap()
    }
    async fn single(&mut self, opid: u16, data: Vec<u8>) -> Single {
        let m = Single { opid, result: 0xffff, result_extension: 0xffff, protocol_version: 0, as_index: 3, message_number: 0, dpi_pid_index: 1, data };
        self.send(&m.to_bytes()).await;
        self.reply().await
    }
}

/// A `multiple_operation_message()` of `ops`, processed on arrival.
fn message(n: u8, ops: &[(u16, Vec<u8>)]) -> Vec<u8> {
    let mut o = vec![0xff, 0xff, 0, 0, 0, 3, n, 0, 1, 0, 0, ops.len() as u8];
    for (id, d) in ops {
        o.extend_from_slice(&id.to_be_bytes());
        o.extend_from_slice(&(d.len() as u16).to_be_bytes());
        o.extend_from_slice(d);
    }
    let len = o.len() as u16;
    o[2..4].copy_from_slice(&len.to_be_bytes());
    o
}

/// `splice_request_data()` (Table 9-5).
fn splice(ty: u8, event: u32, preroll_ms: u16, tenths: u16, auto: bool) -> Vec<u8> {
    let mut d = vec![ty];
    d.extend_from_slice(&event.to_be_bytes());
    d.extend_from_slice(&[0, 9]);
    d.extend_from_slice(&preroll_ms.to_be_bytes());
    d.extend_from_slice(&tenths.to_be_bytes());
    d.extend_from_slice(&[1, 2, auto as u8, 0]);
    d
}

/// `insert_segmentation_descriptor_request_data()` (Table 9-29), ending after
/// `device_restrictions`.
fn segmentation(event: u32, type_id: u8, seconds: u16) -> Vec<u8> {
    let mut d = event.to_be_bytes().to_vec();
    d.push(0);
    d.extend_from_slice(&seconds.to_be_bytes());
    d.extend_from_slice(&[0x0c, 4, 0xde, 0xad, 0xbe, 0xef, type_id, 1, 1, 0, 1, 0, 0, 0, 0]);
    d
}

/// The objects a subscriber read, by (group, object).
type Objects = Arc<Mutex<BTreeMap<(u64, u64), Vec<u8>>>>;

/// Every object of the injector's events track at the relay, by (group, object).
async fn watch_events(w: &World, principal: &[u8; 32]) -> (Objects, Vec<Box<dyn std::any::Any + Send>>) {
    let (c, ep) = relay_client(w).await;
    let c = Arc::new(c);
    let ns = TrackNamespace::from_strs([hex::encode(w.room).as_str(), hex::encode(principal).as_str(), "events"]);
    let (_, tx, rx) = c.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE the events track");
    let objects: Objects = Arc::default();
    let (c2, o2) = (c.clone(), objects.clone());
    tokio::spawn(async move {
        loop {
            let Some(mut r) = c2.next_stream(Duration::from_secs(1)).await else {
                if Arc::strong_count(&o2) == 1 {
                    return;
                }
                continue;
            };
            let o2 = o2.clone();
            tokio::spawn(async move {
                let Ok(Ok(h)) = timeout(STEP, r.start()).await else { return };
                while let Ok(Ok(Some(e))) = timeout(STEP, r.next()).await {
                    if let (StreamHeader::Subgroup(sh), UniEvent::SubgroupObject(o)) = (&h, e) {
                        o2.lock().unwrap().insert((sh.group_id, o.object_id), o.payload);
                    }
                }
            });
        }
    });
    (objects, vec![Box::new(c), Box::new(ep), Box::new(tx), Box::new(rx)])
}

fn b64(s: &str) -> Vec<u8> {
    use base64::Engine;
    base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(s).unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn splices_and_time_signals_become_signed_cues() {
    timeout(Duration::from_secs(90), run()).await.expect("the test finishes in time");
}

async fn run() {
    let w = world().await;
    let injector = w.gateway.injector.clone().expect("the injector runs");
    let principal = injector.principal();
    let (objects, _held) = watch_events(&w, &principal).await;
    let mut a = As::connect(&w).await;

    // Before init_request, a request is not taken.
    a.send(&message(1, &[(op::SPLICE_NULL, vec![])])).await;
    let r = a.reply().await;
    assert_eq!((r.opid, r.result), (op::INJECT_RESPONSE, 124));
    let r = a.single(op::INIT_REQUEST, vec![]).await;
    assert_eq!((r.opid, r.result, r.as_index, r.dpi_pid_index), (op::INIT_RESPONSE, 100, 3, 1));
    // A second automation system is answered "Injector is already in use".
    let mut b = As::connect(&w).await;
    assert_eq!(b.single(op::INIT_REQUEST, vec![]).await.result, 110);
    let r = a.single(op::ALIVE_REQUEST, vec![0; 8]).await;
    assert_eq!((r.opid, r.result, r.data.len()), (op::ALIVE_RESPONSE, 100, 8));

    // 1. An immediate break of 1.5 s with auto-return: an out now, its in 1.5 s later.
    a.send(&message(2, &[(op::SPLICE, splice(2, 0x10, 0, 15, true)), (op::INSERT_AVAIL, vec![1, 0, 0, 1, 0x35])])).await;
    let r = a.reply().await;
    assert_eq!((r.opid, r.result, r.message_number, r.data.clone()), (op::INJECT_RESPONSE, 100, 2, vec![2]));
    // 2. A splice with a pre-roll of 1 s (below 4 000 ms: issued, answered 122), then
    //    cancelled before it comes: never signed.
    a.send(&message(3, &[(op::SPLICE, splice(1, 0x20, 1000, 0, false))])).await;
    assert_eq!(a.reply().await.result, 122);
    a.send(&message(4, &[(op::SPLICE, splice(5, 0x20, 0, 0, false))])).await;
    assert_eq!(a.reply().await.result, 100);
    // Refusals carry their result codes.
    a.send(&message(5, &[(0x7777, vec![])])).await;
    let r = a.reply().await;
    assert_eq!((r.result, r.result_extension), (125, 0x7777));
    a.send(&message(6, &[(op::INSERT_TIER, vec![0, 1])])).await;
    assert_eq!(a.reply().await.result, 115);
    // The first message completes once its in is signed: two cues.
    let done = timeout(STEP, a.reply()).await.unwrap();
    assert_eq!((done.opid, done.message_number, done.data.clone()), (op::INJECT_COMPLETE_RESPONSE, 2, vec![2, 2]));
    // 3. A time signal with a Provider Placement Opportunity Start of 1 s, and its End.
    a.send(&message(7, &[(op::TIME_SIGNAL, vec![0, 0]), (op::INSERT_SEGMENTATION, segmentation(0x30, 0x34, 1))])).await;
    assert_eq!(a.reply().await.result, 100);
    assert_eq!(a.reply().await.opid, op::INJECT_COMPLETE_RESPONSE);
    tokio::time::sleep(Duration::from_millis(1_100)).await;
    a.send(&message(8, &[(op::TIME_SIGNAL, vec![0, 0]), (op::INSERT_SEGMENTATION, segmentation(0x30, 0x35, 0))])).await;
    assert_eq!(a.reply().await.result, 100);
    assert_eq!(a.reply().await.opid, op::INJECT_COMPLETE_RESPONSE);

    // The injector's chain: four cues, each after its injection record.
    let events = injector.events();
    assert_eq!(events.len(), 8, "four cues and their injection records");
    let key = ed25519_dalek::VerifyingKey::from_bytes(&principal).unwrap();
    let t0 = w.descriptor.t0_us;
    let params = scte104::Params { t0_us: t0, leap_s: 18, fps_num: 30_000, fps_den: 1_001 };
    let mut prev = [0u8; 32];
    let mut cues = Vec::new();
    for (i, e) in events.iter().enumerate() {
        let body = &e[..e.len() - 64];
        let sig = ed25519_dalek::Signature::from_bytes(e[e.len() - 64..].try_into().unwrap());
        key.verify(&[EVENT_DOMAIN, body].concat(), &sig).expect("every event verifies under the injector's key");
        assert_eq!(&e[5..37], &w.room[..]);
        assert_eq!(u64::from_le_bytes(e[69..77].try_into().unwrap()), i as u64, "seq");
        assert_eq!(&e[77..109], &prev, "each event chains to the one before");
        prev = joule_moq_ingest::room::sha256(&[body]);
        let t = u64::from_le_bytes(e[109..117].try_into().unwrap());
        let kind = e[118];
        let payload = &e[121..e.len() - 64];
        if i % 2 == 0 {
            assert_eq!(kind, KIND_META);
            assert_eq!(&payload[..4], b"WIJ1");
            let received = u64::from_le_bytes(payload[5..13].try_into().unwrap());
            let index = payload[22] as usize;
            let n = u16::from_le_bytes([payload[23], payload[24]]) as usize;
            let msg = &payload[25..25 + n];
            let tr = scte104::translate(msg, received, &params).expect("the recorded message translates");
            let want = tr.cues()[index];
            let next = &events[i + 1];
            assert_eq!(next[118], KIND_CUE);
            assert_eq!(&next[121..next.len() - 64], &want.payload()[..], "the cue is its message's translation");
            assert_eq!(u64::from_le_bytes(next[109..117].try_into().unwrap()), want.t);
            assert_eq!(t, want.t);
            cues.push((want.kind, want.cue_id, want.t, want.duration_ms, want.section.is_some()));
        }
    }
    let kinds: Vec<(u8, u32)> = cues.iter().map(|c| (c.0, c.1)).collect();
    assert_eq!(kinds, vec![(1, 0x10), (2, 0x10), (1, 0x30), (2, 0x30)], "out and in of each break, the cancelled splice absent");
    assert_eq!(cues[1].2 - cues[0].2, 1_500_000, "the auto-return in, 1.5 s after the out");
    assert!(!cues[1].4, "an auto-return in carries no section");
    assert_eq!(cues[2].3, 1_000, "the placement opportunity's planned duration");
    let counts = injector.counts();
    assert_eq!(counts.get("withdrawn"), Some(&1));
    assert_eq!(counts.get("opid_unknown"), Some(&1));

    // The events track at the relay: an MSF event timeline, every record one of the chain.
    assert!(until(STEP, || objects.lock().unwrap().values().filter(|o| o.as_slice() != b"[]").count() >= 8).await, "every event reached the relay");
    let mut seen = Vec::new();
    for o in objects.lock().unwrap().values() {
        let v: serde_json::Value = serde_json::from_slice(o).expect("a JSON array");
        for r in v.as_array().unwrap() {
            let e = b64(r["data"]["wev1"].as_str().unwrap());
            let t = u64::from_le_bytes(e[109..117].try_into().unwrap());
            assert_eq!(r["t"].as_u64().unwrap(), (t0 + t) / 1000, "the record's index is its wallclock in ms");
            if !seen.contains(&e) {
                seen.push(e);
            }
        }
    }
    assert_eq!(seen, events, "the relay carries the chain, in order");
    drop(b);
    w.stop().await;
}