//! Delivery telemetry (`extensions/delivery-telemetry.md`): unsigned, advisory
//! observations about WAI delivery, their mapping onto CMCD version 2 and CMSD as RFC
//! 9651 Structured Field Values, and the session metrics a log of them gives.
//!
//! - [`sf`]: the RFC 9651 subset the reports use;
//! - [`observation`]: an observation, read strictly and written in one order;
//! - [`cmcd`]: an observation as a CMCD report, its header shards, and reading one back;
//! - [`cmsd`]: CMSD-Static and CMSD-Dynamic for a WAI object;
//! - [`qoe`]: one session's metrics.
//!
//! An observation is never evidence of delivery, presentation, quality or energy: no
//! signature covers it, and it carries no joule figure.

pub mod cmcd;
pub mod cmsd;
pub mod observation;
pub mod qoe;
pub mod sf;

/// Why an observation, a report or a log is refused.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TelemetryError {
    Invalid(String),
}

impl std::fmt::Display for TelemetryError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Invalid(s) => f.write_str(s),
        }
    }
}

impl std::error::Error for TelemetryError {}

#[cfg(test)]
mod tests {
    use serde_json::{json, Value};

    use super::cmcd::{cmcd, cmcd_headers, from_cmcd, is_custom_key, Report};
    use super::cmsd::{encode_dynamic, CmsdDynamic, CmsdStatic};
    use super::observation::Observation;
    use super::qoe::session_metrics;

    const SID: &str = "0123456789abcdef0123456789abcdef";
    const H: &str = "aaf90792d6ff5318e5f306fbb000dfcbe73637e244aad1a2eff6dae537cd6b5f";

    fn obs(event: &str, sn: u64, ts: Option<u64>, f: impl FnOnce(&mut Value)) -> Observation {
        let mut v = json!({
            "v": 1, "sid": SID, "sn": sn, "ts_ms": ts, "event": event, "cid": null, "st": "v", "sf": "o",
            "object": null, "present": null, "switch": null, "layer_change": null, "capable": null, "cohort": null,
            "metrics": null, "claim": null,
        });
        f(&mut v);
        Observation::from_json(&v).unwrap_or_else(|e| panic!("{event}: {e}"))
    }

    fn object(v: &mut Value, members: Value) {
        let mut o = json!({
            "ot": null, "url": null, "content_hash": null, "capability": null, "rendition": null, "layer": null,
            "layers_total": null, "layer_cap": null, "counted": null, "transferred": null, "d_ms": null, "dl_ms": null,
            "ttfb_ms": null, "ttlb_ms": null, "rc": null, "path": null, "receipt": null,
        });
        for (k, x) in members.as_object().unwrap() {
            o[k] = x.clone();
        }
        v["object"] = o;
    }

    #[test]
    fn a_request_and_its_shards() {
        let o = obs("request", 3, None, |v| {
            v["cid"] = json!("abc");
            v["cohort"] = json!("wai");
            object(v, json!({"ot": "v", "d_ms": 4000, "dl_ms": 1200, "counted": 1234, "content_hash": H, "path": "primary", "layer": 1, "layers_total": 3, "layer_cap": 2}));
        });
        let want = format!(
            "cid=\"abc\",d=4000,dl=1200,ot=v,science.transaction.wai-b=\"1234\",science.transaction.wai-ch=\"{H}\",\
             science.transaction.wai-coh=\"wai\",science.transaction.wai-l=\"1\",science.transaction.wai-lc=\"2\",\
             science.transaction.wai-lt=\"3\",science.transaction.wai-p=\"p\",sf=o,sid=\"{SID}\",sn=3,st=v,v=2"
        );
        assert_eq!(cmcd(&o).unwrap(), Report::Request(want.clone()));
        let h = cmcd_headers(&o).unwrap();
        assert_eq!(h["CMCD-Session"], format!("cid=\"abc\",science.transaction.wai-coh=\"wai\",sf=o,sid=\"{SID}\",st=v,v=2"));
        assert_eq!(h["CMCD-Request"], "dl=1200,science.transaction.wai-lc=\"2\",sn=3");
        assert!(h["CMCD-Object"].starts_with("d=4000,ot=v,science.transaction.wai-b=\"1234\""));
        assert!(!h.contains_key("CMCD-Status"));
        assert_eq!(from_cmcd(&want).unwrap().len(), 16);
    }

    #[test]
    fn a_delivery_only_request_carries_the_delivery_class_alone() {
        use super::cmcd::{check_delivery_only, cmcd_delivery_only, cmcd_delivery_only_headers, key_class, KeyClass};
        let o = obs("request", 3, None, |v| {
            v["cid"] = json!("abc");
            v["cohort"] = json!("wai");
            object(v, json!({"ot": "v", "d_ms": 4000, "dl_ms": 1200, "counted": 1234, "content_hash": H, "path": "primary", "layer": 1, "layers_total": 3, "layer_cap": 2}));
        });
        let d = cmcd_delivery_only(&o).unwrap();
        assert_eq!(d, format!("d=4000,dl=1200,ot=v,science.transaction.wai-ch=\"{H}\",science.transaction.wai-lc=\"2\",st=v,v=2"));
        let h = cmcd_delivery_only_headers(&o).unwrap();
        assert_eq!(h["CMCD-Session"], "st=v,v=2");
        assert_eq!(h["CMCD-Request"], "dl=1200,science.transaction.wai-lc=\"2\"");
        assert_eq!(check_delivery_only(&d), Ok(()));
        let Report::Request(full) = cmcd(&o).unwrap() else { panic!("a request") };
        assert_eq!(check_delivery_only(&full), Err("cid".to_string()));
        assert_eq!(check_delivery_only("bl=1200,v=2"), Err("bl".to_string()), "a key the mapping does not write has no class");
        assert_eq!(key_class("sid"), Some(KeyClass::Measurement));
        assert_eq!(key_class("science.transaction.wai-ch"), Some(KeyClass::Delivery));
        assert_eq!(key_class("pt"), None);
        let start = obs("start", 4, Some(1), |_| {});
        assert!(cmcd_delivery_only(&start).is_err(), "an Event-mode observation has no delivery-only report");
    }

    #[test]
    fn event_reports() {
        let start = obs("start", 4, Some(1_700_000_000_000), |_| {});
        assert_eq!(cmcd(&start).unwrap(), Report::Event(format!("e=ps,sf=o,sid=\"{SID}\",sn=4,st=v,sta=s,ts=1700000000000,v=2")));
        let rr = obs("response", 5, Some(9), |v| object(v, json!({"rc": 200, "ttfb_ms": 30, "ttlb_ms": 80, "url": "https://x.example/a.wai", "counted": 99})));
        assert_eq!(
            cmcd(&rr).unwrap(),
            Report::Event(format!("e=rr,rc=200,science.transaction.wai-b=\"99\",sf=o,sid=\"{SID}\",sn=5,st=v,ts=9,ttfb=30,ttlb=80,url=\"https://x.example/a.wai\",v=2"))
        );
        let sw = obs("switch", 6, Some(10), |v| v["switch"] = json!({"unit": 3, "from": "primary", "to": "fallback", "reason": "missing-capability"}));
        assert_eq!(
            cmcd(&sw).unwrap(),
            Report::Event(format!("cen=\"wai-switch\",e=ce,science.transaction.wai-p=\"f\",science.transaction.wai-sw=\"missing-capability\",science.transaction.wai-u=\"3\",sf=o,sid=\"{SID}\",sn=6,st=v,ts=10,v=2"))
        );
        let stall = obs("stall-start", 7, Some(11), |v| v["metrics"] = json!({"msd_ms": null, "bsa": 2, "bsda_ms": 300, "dfa": 1}));
        assert_eq!(cmcd(&stall).unwrap(), Report::Event(format!("bsa=(2),bsda=(300),dfa=1,e=ps,sf=o,sid=\"{SID}\",sn=7,st=v,sta=r,ts=11,v=2")));
        // Only the first presentation is reported; header shards are Request mode only.
        let later = obs("present", 8, None, |v| v["present"] = json!({"unit": 4, "layers": 2, "first": false, "late_ms": null, "decode_us": 900}));
        assert_eq!(cmcd(&later).unwrap(), Report::None);
        assert!(cmcd_headers(&start).is_err());
    }

    #[test]
    fn observations_are_read_strictly() {
        let base = json!({
            "v": 1, "sid": SID, "sn": 0, "ts_ms": 1, "event": "start", "cid": null, "st": "v", "sf": "o",
            "object": null, "present": null, "switch": null, "layer_change": null, "capable": null, "cohort": null,
            "metrics": null, "claim": null,
        });
        assert!(Observation::from_json(&base).is_ok());
        assert_eq!(Observation::from_json(&base).unwrap().to_json(), base);
        let mutate = |f: &dyn Fn(&mut Value)| {
            let mut v = base.clone();
            f(&mut v);
            Observation::from_json(&v)
        };
        for (what, f) in [
            ("an unknown member", &(|v: &mut Value| v["x"] = json!(1)) as &dyn Fn(&mut Value)),
            ("a missing member", &|v: &mut Value| { v.as_object_mut().unwrap().remove("claim"); }),
            ("v 2", &|v: &mut Value| v["v"] = json!(2)),
            ("an uppercase sid", &|v: &mut Value| v["sid"] = json!(SID.to_uppercase())),
            ("an event mode report with no time", &|v: &mut Value| v["ts_ms"] = json!(null)),
            ("an integer past 2^53 − 1", &|v: &mut Value| v["sn"] = json!(1u64 << 53)),
            ("a claim before the end", &|v: &mut Value| v["claim"] = json!(H)),
            ("an object on a start", &|v: &mut Value| v["object"] = json!({})),
            ("an unknown event", &|v: &mut Value| v["event"] = json!("pause")),
        ] {
            assert!(mutate(f).is_err(), "{what}");
        }
        let mut req = base.clone();
        req["event"] = json!("request");
        object(&mut req, json!({"rc": 200}));
        assert!(Observation::from_json(&req).is_err(), "a response key on a request");
    }

    #[test]
    fn integers_a_report_cannot_carry_are_refused() {
        // 10^15 is past an RFC 9651 Integer: an observation that holds it is refused,
        // so every observation read is one cmcd() can write.
        let at = |k: &str, x: u64| {
            let mut v = obs("start", 0, Some(1), |_| {}).to_json();
            v[k] = json!(x);
            Observation::from_json(&v)
        };
        assert!(at("ts_ms", 999_999_999_999_999).is_ok());
        assert!(at("ts_ms", 1_000_000_000_000_000).is_err());
        assert!(at("sn", 1_000_000_000_000_000).is_err());
        // A log whose counted bytes pass 64 bits is refused, not wrapped.
        let mut log = vec![obs("start", 0, Some(1), |_| {})];
        for i in 1..=2049u64 {
            log.push(obs("response", i, Some(1), |v| object(v, json!({"counted": (1u64 << 53) - 1}))));
        }
        assert!(session_metrics(&log).is_err());
        log.truncate(2048);
        assert!(session_metrics(&log).is_ok());
    }

    #[test]
    fn short_keys_do_not_panic() {
        // One- and two-character keys have no inner hyphen: neither CMCD keys nor
        // custom keys, refused, never a panic.
        for k in ["q", "x", "-", "a-", "zz"] {
            assert!(!is_custom_key(k), "{k}");
            assert!(from_cmcd(k).is_err(), "{k}");
        }
        assert!(from_cmcd("v=2,x=\"1\"").is_err());
        assert!(is_custom_key("a-b"));
    }

    /// Every parser of the module, on inputs drawn from a fixed seed: structured-field
    /// text built from the characters the grammar uses, and the corpus's JSON documents
    /// with bytes changed. Each returns, `Ok` or `Err`; none panics.
    #[test]
    fn parsers_never_panic_on_drawn_inputs() {
        struct Rng(u64);
        impl Rng {
            fn next(&mut self) -> u64 {
                self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
                let mut z = self.0;
                z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
                z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
                z ^ (z >> 31)
            }
            fn below(&mut self, n: u64) -> u64 {
                self.next() % n
            }
        }
        let mut r = Rng(0x7e1e_0001);
        const PIECES: &[&str] = &[
            "v", "=", ",", ";", " ", "(", ")", "\"", "\\", "?", "?1", "?0", "-", ".", "*", "2", "0", "999999999999999", "9999999999999999",
            "-999999999999999", "sid", "science.transaction.wai-b", "a-b", "x", "q", "ot", "v=2", "\t", "é", "\u{0}", "1.5", "e=ps", ":", "/",
            "bsa=(1 2)", "\"s\"", "A", "~",
        ];
        for _ in 0..20_000 {
            let n = r.below(12) as usize;
            let s: String = (0..n).map(|_| PIECES[r.below(PIECES.len() as u64) as usize]).collect();
            let _ = super::sf::parse_dict(&s);
            let _ = super::sf::parse_list(&s);
            let _ = from_cmcd(&s);
            let _ = super::cmcd::is_custom_key(&s);
        }
        // JSON documents with bytes changed, read as observations and as CMSD.
        let base = obs("request", 3, None, |v| object(v, json!({"ot": "v", "counted": 12, "path": "primary"}))).to_json().to_string();
        let cmsd = r#"{"at": 1, "d": 2, "ht": null, "nor": "n", "nrr": null, "ot": "v", "sf": "o", "st": "v", "su": true, "content_hash": null, "layer": 1, "layers_total": 2, "receipt": null, "op": "crop"}"#;
        let dynamic = r#"{"server": "s", "du": true, "etp": 1, "mb": null, "rd": 2, "rtt": null, "drop": 3}"#;
        let swaps: &[&[u8]] = &[b"0", b"1", b"-", b"\"", b"null", b"true", b"[", b"{", b"}", b"99999999999999999999", b"1e3", b"-1", b"\"x\""];
        for doc in [base.as_str(), cmsd, dynamic] {
            for _ in 0..4000 {
                let mut b = doc.as_bytes().to_vec();
                for _ in 0..1 + r.below(3) {
                    let at = r.below(b.len() as u64) as usize;
                    let len = (r.below(4) as usize).min(b.len() - at);
                    let with = swaps[r.below(swaps.len() as u64) as usize];
                    b.splice(at..at + len, with.iter().copied());
                }
                if let Ok(v) = serde_json::from_slice::<Value>(&b) {
                    let _ = Observation::from_json(&v);
                    let _ = CmsdStatic::from_json(&v);
                    let _ = CmsdDynamic::from_json(&v);
                    if let Ok(o) = Observation::from_json(&v) {
                        let _ = cmcd(&o);
                        let _ = cmcd_headers(&o);
                    }
                }
            }
        }
    }

    #[test]
    fn custom_keys_and_reading_reports() {
        assert!(is_custom_key("science.transaction.wai-b") && is_custom_key("com.example-x"));
        assert!(!is_custom_key("wai") && !is_custom_key("-x") && !is_custom_key("x-") && !is_custom_key("Abc-d"));
        assert!(from_cmcd("v=2,science.transaction.wai-b=12").is_err(), "a custom key carries a String or a Token");
        assert!(from_cmcd("v=2,zz=1").is_err(), "neither a CMCD key nor a custom key");
        assert!(from_cmcd("v=2,sid=\"s\",com.example-x=tok").is_ok());
    }

    #[test]
    fn cmsd_static_and_dynamic() {
        let s = CmsdStatic::from_json(&json!({
            "at": 1_700_000_000_000u64, "d": 4000, "ht": null, "nor": "next.wai", "nrr": null, "ot": "v", "sf": "o", "st": "v",
            "su": true, "content_hash": H, "layer": 1, "layers_total": 3, "receipt": null, "op": "crop",
        }))
        .unwrap();
        assert_eq!(
            s.encode().unwrap(),
            format!("at=1700000000000,d=4000,nor=\"next.wai\",ot=v,science.transaction.wai-ch=\"{H}\",science.transaction.wai-l=\"1\",science.transaction.wai-lt=\"3\",science.transaction.wai-op=crop,sf=o,st=v,su")
        );
        let d = [
            CmsdDynamic { server: "edge-1".into(), du: true, etp: Some(25000), mb: None, rd: Some(5), rtt: Some(40), drop: Some(2) },
            CmsdDynamic { server: "origin".into(), du: false, etp: None, mb: None, rd: Some(12), rtt: None, drop: None },
        ];
        assert_eq!(encode_dynamic(&d).unwrap(), "\"edge-1\";du;etp=25000;rd=5;rtt=40;science.transaction.wai-drop=\"2\",\"origin\";rd=12");
    }

    #[test]
    fn a_session_log() {
        let log = vec![
            obs("start", 0, Some(1000), |_| {}),
            obs("response", 1, Some(1100), |v| object(v, json!({"counted": 500}))),
            obs("present", 2, Some(1250), |v| v["present"] = json!({"unit": 0, "layers": 1, "first": true, "late_ms": null, "decode_us": null})),
            obs("present", 3, Some(1300), |v| v["present"] = json!({"unit": 0, "layers": 2, "first": false, "late_ms": null, "decode_us": null})),
            obs("stall-start", 4, Some(2000), |_| {}),
            obs("stall-end", 5, Some(2400), |_| {}),
            obs("present", 6, Some(2450), |v| v["present"] = json!({"unit": 1, "layers": 1, "first": false, "late_ms": 40, "decode_us": null})),
            obs("stall-start", 7, Some(3000), |_| {}),
            obs("end", 8, Some(3100), |v| v["claim"] = json!(H)),
        ];
        let m = session_metrics(&log).unwrap();
        assert_eq!(
            (m.started, m.vst_ms, m.exit_before_start, m.start_failed, m.stalls, m.stall_ms, m.engaged_ms, m.units, m.counted),
            (true, Some(250), false, false, 2, 500, 1850, 2, 500)
        );
        assert_eq!(m.claim.as_deref(), Some(H));
        // An abandon before any presentation is an exit before start; an end is a start
        // failure.
        let quit = vec![obs("start", 0, Some(1), |_| {}), obs("abandon", 1, Some(9), |_| {})];
        let m = session_metrics(&quit).unwrap();
        assert!(m.exit_before_start && !m.start_failed);
        let fail = vec![obs("start", 0, Some(1), |_| {}), obs("end", 1, Some(9), |_| {})];
        assert!(session_metrics(&fail).unwrap().start_failed);
        // Out of order, or after the end, is refused.
        assert!(session_metrics(&[obs("start", 2, Some(1), |_| {}), obs("end", 1, Some(2), |_| {})]).is_err());
        assert!(session_metrics(&[obs("end", 0, Some(1), |_| {}), obs("start", 1, Some(2), |_| {})]).is_err());
    }
}
