//! Set roots, location hints and a set fetched by its digest (extensions/http-delivery.md §8.5,
//! §8.7; extensions/prior-carriage.md §15).
//!
//! A **set root** is an `https` URI whose path ends in `/`; a set is at the root followed by its
//! digest. A hint is a list of roots a MoQ catalog's `wai_priors` entry (`locations`) or a WAI
//! multivariant playlist (the one `EXT-X-SESSION-DATA` line of [`DATA_ID`], its roots separated
//! by spaces) states; it names where to ask and proves nothing. [`fetch_set`] runs §8.7's steps over the responses its caller
//! supplies and holds bytes only when their SHA-256 is the digest asked for.

use super::dictionary::{check_coded, Asked};
use super::usage::{self, Usage, USAGE_FILE};
use super::{digest_of_hex, hex};
use crate::codecs::CAP_PRIOR_SET;
use crate::pin::{parse_listing, sha256};
use serde_json::Value;

/// The path below an origin's host and port under which it serves its sets (§8.7).
pub const WELL_KNOWN: &str = "/.well-known/wai-sets/";
/// The `DATA-ID` of a playlist's location-hint line (§8.7).
pub const DATA_ID: &str = "science.transaction.wai.sets";
/// The longest set root, in bytes.
pub const MAX_ROOT_BYTES: usize = 2048;
/// The most roots a hint keeps.
pub const MAX_ROOTS: usize = 8;

fn unreserved(c: u8) -> bool {
    c.is_ascii_alphanumeric() || b"-._~".contains(&c)
}

fn sub_delim(c: u8) -> bool {
    b"!$&'()*+,;=".contains(&c)
}

/// Every byte is unreserved, a sub-delimiter, one of `extra`, or starts a `%` and two
/// hexadecimal digits (RFC 3986 §2.1).
fn uri_chars(s: &[u8], extra: &[u8]) -> bool {
    let mut i = 0;
    while i < s.len() {
        let c = s[i];
        if c == b'%' {
            if !(s.get(i + 1).is_some_and(u8::is_ascii_hexdigit) && s.get(i + 2).is_some_and(u8::is_ascii_hexdigit)) {
                return false;
            }
            i += 3;
            continue;
        }
        if !(unreserved(c) || sub_delim(c) || extra.contains(&c)) {
            return false;
        }
        i += 1;
    }
    true
}

/// Whether `s` is a set root: an absolute URI (RFC 3986 §4.3) that begins `https://`, with a
/// non-empty host, no userinfo, no query and no fragment, whose path ends in `/`, at most
/// [`MAX_ROOT_BYTES`] long (http-delivery §8.7).
pub fn is_set_root(s: &str) -> bool {
    if s.len() > MAX_ROOT_BYTES {
        return false;
    }
    let Some(rest) = s.strip_prefix("https://") else { return false };
    if rest.contains('?') || rest.contains('#') {
        return false;
    }
    let Some(slash) = rest.find('/') else { return false };
    let (authority, path) = rest.split_at(slash);
    if authority.contains('@') {
        return false;
    }
    let (host, port) = if let Some(inner) = authority.strip_prefix('[') {
        let Some(close) = inner.find(']') else { return false };
        let literal = &inner[..close];
        if literal.is_empty() || !uri_chars(literal.as_bytes(), b":") || literal.contains('%') {
            return false;
        }
        let after = &inner[close + 1..];
        match after.strip_prefix(':') {
            Some(p) => ("[]", Some(p)),
            None if after.is_empty() => ("[]", None),
            None => return false,
        }
    } else {
        match authority.rfind(':') {
            Some(i) => (&authority[..i], Some(&authority[i + 1..])),
            None => (authority, None),
        }
    };
    if host.is_empty() || (host != "[]" && !uri_chars(host.as_bytes(), b"")) {
        return false;
    }
    if let Some(p) = port {
        if p.is_empty() || !p.bytes().all(|c| c.is_ascii_digit()) {
            return false;
        }
    }
    path.ends_with('/') && uri_chars(path.as_bytes(), b":@/")
}

/// A hint as read: the roots it keeps, in order, and the positions of the elements passed over
/// as not holding a root (`set_location_malformed`). A playlist's elements are its hint line's
/// `VALUE` split at every space.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Hint {
    pub roots: Vec<String>,
    pub malformed: Vec<usize>,
}

impl Hint {
    /// Takes the element at `i`; `false` once the hint is full and the rest are ignored.
    fn take(&mut self, i: usize, root: Option<&str>) -> bool {
        if self.roots.len() == MAX_ROOTS {
            return false;
        }
        match root {
            Some(r) if is_set_root(r) => {
                if !self.roots.iter().any(|k| k == r) {
                    self.roots.push(r.to_string());
                }
            }
            _ => self.malformed.push(i),
        }
        true
    }
}

/// The hint of a `wai_priors` entry: its `locations` member (moq-streaming-format §4.2). A
/// member that is absent, or not an array, gives none.
pub fn read_catalog_hint(entry: &Value) -> Hint {
    let mut h = Hint::default();
    if let Some(a) = entry.get("locations").and_then(Value::as_array) {
        for (i, e) in a.iter().enumerate() {
            if !h.take(i, e.as_str()) {
                break;
            }
        }
    }
    h
}

/// An attribute value of an HLS attribute list: quoted, or not.
#[derive(Debug, PartialEq, Eq)]
enum AttrValue<'a> {
    Quoted(&'a str),
    Bare(&'a str),
}

/// An HLS attribute list (RFC 8216 §4.2), or `None` where it does not read as one: names of
/// `A`–`Z`, `0`–`9` and `-`, each once; a quoted string with no `"`, CR or LF inside.
fn attribute_list(s: &str) -> Option<Vec<(&str, AttrValue<'_>)>> {
    let mut out: Vec<(&str, AttrValue<'_>)> = Vec::new();
    let mut rest = s;
    loop {
        let eq = rest.find('=')?;
        let name = &rest[..eq];
        if name.is_empty() || !name.bytes().all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == b'-') {
            return None;
        }
        if out.iter().any(|(n, _)| *n == name) {
            return None;
        }
        rest = &rest[eq + 1..];
        let value = if let Some(q) = rest.strip_prefix('"') {
            let end = q.find('"')?;
            let v = &q[..end];
            if v.contains('\r') || v.contains('\n') {
                return None;
            }
            rest = &q[end + 1..];
            AttrValue::Quoted(v)
        } else {
            let end = rest.find(',').unwrap_or(rest.len());
            let v = &rest[..end];
            if v.is_empty() || v.contains('"') || v.bytes().any(|c| c.is_ascii_whitespace()) {
                return None;
            }
            rest = &rest[end..];
            AttrValue::Bare(v)
        };
        out.push((name, value));
        if rest.is_empty() {
            return Some(out);
        }
        rest = rest.strip_prefix(',')?;
    }
}

/// The hint of a WAI multivariant playlist: the first `EXT-X-SESSION-DATA` line whose attribute
/// list reads, whose `DATA-ID` is [`DATA_ID`] and that carries no `LANGUAGE`, its `VALUE` split
/// at every space into elements. A later such line is ignored (RFC 8216 §4.3.4.4 admits one);
/// a hint line whose `VALUE` is not a quoted string, or that carries `URI`, gives none.
pub fn read_playlist_hint(text: &str) -> Hint {
    let mut h = Hint::default();
    for line in text.split('\n') {
        let line = line.strip_suffix('\r').unwrap_or(line);
        let Some(attrs) = line.strip_prefix("#EXT-X-SESSION-DATA:") else { continue };
        let Some(list) = attribute_list(attrs) else { continue };
        let get = |n: &str| list.iter().find(|(k, _)| *k == n).map(|(_, v)| v);
        if get("DATA-ID") != Some(&AttrValue::Quoted(DATA_ID)) || get("LANGUAGE").is_some() {
            continue;
        }
        if let (Some(AttrValue::Quoted(v)), None) = (get("VALUE"), get("URI")) {
            for (i, e) in v.split(' ').enumerate() {
                if !h.take(i, Some(e)) {
                    break;
                }
            }
        }
        break;
    }
    h
}

/// Where a set, or a listed file, of digest `sha256` is below `root`.
pub fn set_url(root: &str, sha256: &[u8; 32]) -> String {
    format!("{root}{}", hex(sha256))
}

/// An origin's well-known root, from the origin `https://<host>[:<port>]`; `None` where that
/// does not make a set root.
pub fn well_known_root(origin: &str) -> Option<String> {
    let r = format!("{origin}{WELL_KNOWN}");
    (origin.strip_prefix("https://").is_some_and(|a| !a.contains('/')) && is_set_root(&r)).then_some(r)
}

/// The roots a sink asks, in order, each once: those of its own configuration, the hint's,
/// then the origin's well-known root.
pub fn roots_in_order(own: &[String], hint: &[String], origin_root: Option<&str>) -> Vec<String> {
    let mut out: Vec<String> = Vec::new();
    for r in own.iter().map(String::as_str).chain(hint.iter().map(String::as_str)).chain(origin_root) {
        if !out.iter().any(|k| k == r) {
            out.push(r.to_string());
        }
    }
    out
}

/// A request a sink sends: the URL, and the set it names in `Available-Dictionary` (with `dcz`
/// in `Accept-Encoding`), if any (http-delivery §8.8).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Request<'a> {
    pub url: &'a str,
    pub available_dictionary: Option<[u8; 32]>,
}

/// A response a caller supplies for one request.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Response {
    pub status: u16,
    /// The `Content-Encoding` field's value, `None` where absent.
    pub content_encoding: Option<String>,
    pub body: Vec<u8>,
}

/// One request sent, and what came of it: `held`, or a reason of §8.7 or §8.8.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Attempt {
    pub url: String,
    pub outcome: &'static str,
}

/// A set fetched and held: its bytes in the pin's form (the listing, for `wai.prior.set`) and,
/// for a listing, each file it names with its path and the usage statement it binds as its file
/// [`USAGE_FILE`] (prior-carriage §18.1), if any.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeldSet {
    pub bytes: Vec<u8>,
    pub files: Vec<(String, Vec<u8>)>,
    pub usage: Option<Usage>,
}

/// The requests sent, in order, and the set held or the reason SPEC §3.1 refuses the pin with
/// (`prior_not_held`, `set_malformed`, `part_mismatch`), or `set_usage_malformed` for a listing
/// whose bound statement does not read (prior-carriage §18.1).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Fetch {
    pub attempts: Vec<Attempt>,
    pub result: Result<HeldSet, &'static str>,
}

/// A set a sink holds and names as a dictionary: its digest and bytes.
pub type Dictionary<'a> = (&'a [u8; 32], &'a [u8]);

/// What one response yields for a resource of digest `digest`: its bytes, or the reason; and
/// whether the reason is a check of §8.8 after which the same root is asked without the
/// dictionary.
fn outcome_of(
    digest: &[u8; 32],
    r: Option<Response>,
    dictionary: Option<Dictionary<'_>>,
    limit: u64,
) -> (Result<Vec<u8>, &'static str>, bool) {
    let Some(r) = r else { return (Err("set_location_unavailable"), false) };
    if r.status != 200 {
        return (Err("set_location_unavailable"), false);
    }
    let bytes = match r.content_encoding.as_deref() {
        None => {
            if r.body.len() as u64 > limit {
                return (Err("set_too_large"), false);
            }
            r.body
        }
        Some(c @ ("dcz" | "dcb")) => match check_coded(c, &r.body, Asked { dictionary }, limit) {
            Ok(b) => b,
            Err("set_too_large") => return (Err("set_too_large"), false),
            Err(code) => return (Err(code), dictionary.is_some()),
        },
        Some(_) => return (Err("content_coding_present"), false),
    };
    if sha256(&bytes) != *digest {
        return (Err("set_location_mismatch"), false);
    }
    (Ok(bytes), false)
}

/// The steps of §8.7 for one resource, with §8.8's when the sink names a dictionary: the first
/// root, in `roots`' order, whose response holds it, with that root's position.
fn fetch_one(
    digest: &[u8; 32],
    roots: &[&str],
    limit: u64,
    dictionary: Option<Dictionary<'_>>,
    get: &mut dyn FnMut(&Request<'_>) -> Option<Response>,
    attempts: &mut Vec<Attempt>,
) -> Option<(usize, Vec<u8>)> {
    for (i, root) in roots.iter().enumerate() {
        let url = set_url(root, digest);
        let mut named = dictionary;
        loop {
            let r = get(&Request { url: &url, available_dictionary: named.map(|(d, _)| *d) });
            let (got, again) = outcome_of(digest, r, named, limit);
            attempts.push(Attempt { url: url.clone(), outcome: *got.as_ref().err().unwrap_or(&"held") });
            match got {
                Ok(b) => return Some((i, b)),
                // A coded response refused by §8.8: the same root, asked without the dictionary.
                Err(_) if again => named = None,
                Err(_) => break,
            }
        }
    }
    None
}

/// Fetches the set a pin of digest `sha256_digest` and form `form` names, asking `roots` in
/// order with `get`, reading at most `limit` bytes of each response (http-delivery §8.7). With
/// `dictionary`, a set the sink holds, it names that set in `Available-Dictionary` when it asks
/// for a set in a one-file form (§8.8); it names none for a listing or a listed file. A listing's
/// file at the path [`USAGE_FILE`] is read as its usage statement once every file is held.
pub fn fetch_set(
    sha256_digest: &[u8; 32],
    form: Option<&str>,
    roots: &[String],
    limit: u64,
    dictionary: Option<Dictionary<'_>>,
    get: &mut dyn FnMut(&Request<'_>) -> Option<Response>,
) -> Fetch {
    let mut attempts = Vec::new();
    let order: Vec<&str> = roots.iter().map(String::as_str).collect();
    let listing = form == Some(CAP_PRIOR_SET);
    let named = if listing { None } else { dictionary };
    let Some((at, bytes)) = fetch_one(sha256_digest, &order, limit, named, get, &mut attempts) else {
        return Fetch { attempts, result: Err("prior_not_held") };
    };
    if !listing {
        return Fetch { attempts, result: Ok(HeldSet { bytes, files: Vec::new(), usage: None }) };
    }
    let Ok(lines) = parse_listing(&bytes) else {
        return Fetch { attempts, result: Err("set_malformed") };
    };
    // Each file first at the root that served the listing, then at the others in order.
    let mut file_order = vec![order[at]];
    file_order.extend(order.iter().enumerate().filter(|(i, _)| *i != at).map(|(_, r)| *r));
    let mut files = Vec::new();
    for (path, d) in lines {
        match fetch_one(&d, &file_order, limit, None, get, &mut attempts) {
            Some((_, b)) => files.push((path, b)),
            None => return Fetch { attempts, result: Err("part_mismatch") },
        }
    }
    // The statement the listing binds, at its top-level path only (prior-carriage §18.1).
    let usage = match files.iter().find(|(p, _)| p == USAGE_FILE) {
        Some((_, b)) => match usage::read(b) {
            Ok(u) => Some(u),
            Err(code) => return Fetch { attempts, result: Err(code) },
        },
        None => None,
    };
    Fetch { attempts, result: Ok(HeldSet { bytes, files, usage }) }
}

/// The digest and form a pin's JSON states, read for a fetch: `sha256` as 64 lowercase
/// hexadecimal digits and `form` a string or absent. No other member is read, so a member
/// SPEC §3.1 does not define, whatever it names, never reaches a fetch.
pub fn pin_target(pin: &Value) -> Option<([u8; 32], Option<String>)> {
    let d = digest_of_hex(pin.get("sha256")?.as_str()?)?;
    let form = match pin.get("form") {
        None | Some(Value::Null) => None,
        Some(Value::String(s)) => Some(s.clone()),
        Some(_) => return None,
    };
    Some((d, form))
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn set_roots() {
        for ok in [
            "https://origin.example/.well-known/wai-sets/",
            "https://cdn.example:8443/sets/",
            "https://[2001:db8::1]/s/",
            "https://h.example/a%20b/",
            "https://h.example/",
        ] {
            assert!(is_set_root(ok), "{ok}");
        }
        for bad in [
            "http://origin.example/sets/",
            "HTTPS://origin.example/sets/",
            "https://origin.example/sets",
            "https://origin.example/sets/?x=1",
            "https://origin.example/sets/#f",
            "https://user@origin.example/sets/",
            "https:///sets/",
            "https://origin.example:/sets/",
            "https://origin.example:84a3/sets/",
            "https://origin.example/se ts/",
            "https://origin.example/%zz/",
            "https://origin.example",
            "",
        ] {
            assert!(!is_set_root(bad), "{bad}");
        }
        let long = format!("https://h.example/{}/", "a".repeat(MAX_ROOT_BYTES));
        assert!(!is_set_root(&long));
        let edge = format!("https://h.example/{}/", "a".repeat(MAX_ROOT_BYTES - 19));
        assert_eq!(edge.len(), MAX_ROOT_BYTES);
        assert!(is_set_root(&edge));
    }

    #[test]
    fn catalog_hints() {
        let e = json!({"locations": ["https://a.example/s/", 7, "http://b.example/s/", "https://a.example/s/", "https://c.example/s/"]});
        let h = read_catalog_hint(&e);
        assert_eq!(h.roots, ["https://a.example/s/", "https://c.example/s/"]);
        assert_eq!(h.malformed, [1, 2]);
        assert_eq!(read_catalog_hint(&json!({"locations": "https://a.example/s/"})), Hint::default());
        assert_eq!(read_catalog_hint(&json!({})), Hint::default());
        let many: Vec<String> = (0..12).map(|i| format!("https://m{i}.example/s/")).collect();
        let h = read_catalog_hint(&json!({ "locations": many }));
        assert_eq!(h.roots.len(), MAX_ROOTS);
        assert_eq!(h.roots[7], "https://m7.example/s/");
    }

    #[test]
    fn playlist_hints() {
        let p = "#EXTM3U\r\n#EXT-X-SESSION-DATA:DATA-ID=\"other\",VALUE=\"https://x.example/s/\"\n\
                 #EXT-X-SESSION-DATA:DATA-ID=\"science.transaction.wai.sets\",VALUE=\"https://l.example/s/\",LANGUAGE=\"en\"\n\
                 #EXT-X-SESSION-DATA:DATA-ID=\"science.transaction.wai.sets\",VALUE=\"https://a.example/s/  http://b.example/s/ https://c.example/s/\"\r\n\
                 #EXT-X-SESSION-DATA:DATA-ID=\"science.transaction.wai.sets\",VALUE=\"https://d.example/s/\"\n";
        let h = read_playlist_hint(p);
        assert_eq!(h.roots, ["https://a.example/s/", "https://c.example/s/"]);
        assert_eq!(h.malformed, [1, 2]);
        let uri = "#EXT-X-SESSION-DATA:DATA-ID=\"science.transaction.wai.sets\",VALUE=\"https://a.example/s/\",URI=\"https://b.example/s.json\"\n\
                   #EXT-X-SESSION-DATA:DATA-ID=\"science.transaction.wai.sets\",VALUE=\"https://c.example/s/\"\n";
        assert_eq!(read_playlist_hint(uri), Hint::default());
        assert!(attribute_list("A=\"x\",A=\"y\"").is_none());
        assert!(attribute_list("a=1").is_none());
    }

    #[test]
    fn well_known() {
        assert_eq!(well_known_root("https://origin.example").as_deref(), Some("https://origin.example/.well-known/wai-sets/"));
        assert_eq!(well_known_root("https://origin.example:8443").as_deref(), Some("https://origin.example:8443/.well-known/wai-sets/"));
        assert_eq!(well_known_root("http://origin.example"), None);
        assert_eq!(well_known_root("https://origin.example/x"), None);
    }

    #[test]
    fn a_tampered_mirror_is_passed_over() {
        let set = b"a parameter set".to_vec();
        let d = sha256(&set);
        let roots = vec!["https://m1.example/s/".to_string(), "https://m2.example/s/".to_string()];
        let mut bad = set.clone();
        bad[0] ^= 1;
        let mut get = |r: &Request<'_>| {
            let body = if r.url.starts_with("https://m1.") { bad.clone() } else { set.clone() };
            Some(Response { status: 200, content_encoding: None, body })
        };
        let f = fetch_set(&d, None, &roots, 1 << 20, None, &mut get);
        assert_eq!(f.attempts.iter().map(|a| a.outcome).collect::<Vec<_>>(), ["set_location_mismatch", "held"]);
        assert_eq!(f.result.expect("held").bytes, set);
        let f = fetch_set(&d, None, &roots, 4, None, &mut get);
        assert_eq!(f.result, Err("prior_not_held"));
    }

    #[test]
    fn a_listing_binds_its_statement_at_its_top_level_path() {
        let good: &[u8] = br#"{"kind":"wai-set-usage","licence":"Apache-2.0","terms":null,"content_derived":false,"redistribute":false,"ai_usage":null}"#;
        let bad: &[u8] = br#"{"kind":"wai-set-usage"}"#;
        let run = |path: &str, statement: &[u8]| {
            let mut files = vec![(path.to_string(), statement.to_vec()), ("w.bin".to_string(), b"weights".to_vec())];
            files.sort();
            let listing: String = files.iter().map(|(p, b)| format!("{}  {p}\n", hex(&sha256(b)))).collect();
            let mut bodies: Vec<Vec<u8>> = files.into_iter().map(|(_, b)| b).collect();
            bodies.push(listing.clone().into_bytes());
            let mut get = |r: &Request<'_>| {
                bodies.iter().find(|b| r.url.ends_with(&hex(&sha256(b)))).map(|b| Response { status: 200, content_encoding: None, body: b.clone() })
            };
            let roots = vec!["https://m.example/s/".to_string()];
            fetch_set(&sha256(listing.as_bytes()), Some(CAP_PRIOR_SET), &roots, 1 << 20, None, &mut get).result
        };
        assert!(run(USAGE_FILE, good).expect("held").usage.is_some_and(|u| !u.redistribute));
        assert_eq!(run(USAGE_FILE, bad), Err(usage::USAGE_MALFORMED));
        assert_eq!(run("a/wai-usage.json", bad).expect("held").usage, None);
    }

    #[test]
    fn a_pin_target_reads_two_members() {
        let p = json!({"id": "x", "sha256": "ab".repeat(32), "determinism": "d", "locations": ["https://x.example/s/"]});
        assert_eq!(pin_target(&p), Some(([0xab; 32], None)));
        assert_eq!(pin_target(&json!({"sha256": "AB".repeat(32)})), None);
        assert_eq!(pin_target(&json!({"sha256": "ab".repeat(32), "form": 3})), None);
    }
}
