//! Amortisation, bootstrap and generations (extensions/prior-carriage.md §17).
//!
//! A set is worth holding for a sink when what it saves that sink over the titles the sink is
//! expected to fetch within a declared span exceeds what holding it costs that sink. The cost is
//! charged per sink: nothing when the sink holds the set, the coded bytes when it holds the set
//! a successor is coded against (§16), the set's bytes otherwise. A sink holds at most a stated
//! number of generations; [`plan`] keeps those whose net saving is largest and gives every
//! other title its classical rendition. Integer arithmetic only.

use serde_json::{json, Map, Value};
use std::collections::BTreeMap;

/// The reason a plan that does not read is refused with.
pub const PLAN_MALFORMED: &str = "plan_malformed";

const MAX_INT: u64 = (1 << 53) - 1;

fn uint(v: &Value) -> Option<u64> {
    v.as_u64().filter(|n| *n <= MAX_INT)
}

/// `YYYY-MM-DDTHH:MM:SSZ`, digits where the pattern has them.
fn is_instant(s: &str) -> bool {
    let b = s.as_bytes();
    b.len() == 20
        && b.iter().enumerate().all(|(i, c)| match i {
            4 | 7 => *c == b'-',
            10 => *c == b'T',
            13 | 16 => *c == b':',
            19 => *c == b'Z',
            _ => c.is_ascii_digit(),
        })
}

/// A generation as a plan states it.
#[derive(Debug, Clone)]
struct Generation {
    id: String,
    epoch: u64,
    set_bytes: u64,
    /// The generation this one's set is coded against, and the coded bytes.
    delta: Option<(String, u64)>,
}

#[derive(Debug, Clone)]
struct Title {
    generation: String,
    pinned: u64,
    classical: u64,
}

fn ids_unique<'a>(ids: impl Iterator<Item = &'a str>) -> bool {
    let mut seen = std::collections::BTreeSet::new();
    ids.into_iter().all(|i| !i.is_empty() && seen.insert(i))
}

fn read(doc: &Value) -> Option<(u64, Vec<Generation>, BTreeMap<String, Title>, Vec<&Value>)> {
    let span = doc.get("span")?;
    let (from, to) = (span.get("from")?.as_str()?, span.get("to")?.as_str()?);
    if !(is_instant(from) && is_instant(to) && from < to) {
        return None;
    }
    let cap = uint(doc.get("cap")?).filter(|c| *c >= 1)?;
    let mut gens: Vec<Generation> = Vec::new();
    for g in doc.get("generations")?.as_array()? {
        let id = g.get("id")?.as_str()?.to_string();
        let epoch = uint(g.get("epoch")?)?;
        let set_bytes = uint(g.get("set_bytes")?).filter(|n| *n >= 1)?;
        let delta = match g.get("delta")? {
            Value::Null => None,
            d => {
                let from = d.get("from")?.as_str()?.to_string();
                let bytes = uint(d.get("bytes")?).filter(|n| *n >= 1)?;
                // Coded against an earlier generation only.
                gens.iter().find(|e| e.id == from)?;
                Some((from, bytes))
            }
        };
        if gens.last().is_some_and(|l| l.epoch >= epoch) {
            return None;
        }
        gens.push(Generation { id, epoch, set_bytes, delta });
    }
    if gens.is_empty() || !ids_unique(gens.iter().map(|g| g.id.as_str())) {
        return None;
    }
    let mut titles = BTreeMap::new();
    for t in doc.get("titles")?.as_array()? {
        let id = t.get("id")?.as_str()?.to_string();
        let generation = t.get("generation")?.as_str()?.to_string();
        gens.iter().find(|g| g.id == generation)?;
        let title = Title { generation, pinned: uint(t.get("pinned_bytes")?)?, classical: uint(t.get("classical_bytes")?)? };
        if id.is_empty() || titles.insert(id, title).is_some() {
            return None;
        }
    }
    let sinks: Vec<&Value> = doc.get("sinks")?.as_array()?.iter().collect();
    for s in &sinks {
        let holds = s.get("holds")?.as_array()?;
        let held: Vec<&str> = holds.iter().map(Value::as_str).collect::<Option<_>>()?;
        if !ids_unique(held.iter().copied()) || !held.iter().all(|h| gens.iter().any(|g| g.id == *h)) {
            return None;
        }
        for (t, n) in s.get("demand")?.as_object()? {
            titles.get(t)?;
            uint(n)?;
        }
    }
    if sinks.is_empty() {
        return None;
    }
    Some((cap, gens, titles, sinks))
}

/// The plan for every sink a document states (§17), or [`PLAN_MALFORMED`].
pub fn plan(doc: &Value) -> Result<Value, &'static str> {
    let (cap, gens, titles, sinks) = read(doc).ok_or(PLAN_MALFORMED)?;
    let mut out = Vec::new();
    for s in sinks {
        let held: Vec<&str> = s["holds"].as_array().into_iter().flatten().filter_map(Value::as_str).collect();
        let demand: BTreeMap<&str, u64> =
            s["demand"].as_object().into_iter().flatten().map(|(k, v)| (k.as_str(), v.as_u64().unwrap_or(0))).collect();
        // Each generation the demand touches, or the sink holds: saving, cost, net.
        let mut rows: Vec<(usize, u128, u128, &'static str, i128)> = Vec::new();
        for (i, g) in gens.iter().enumerate() {
            let touched = demand.iter().any(|(t, n)| *n > 0 && titles[*t].generation == g.id);
            let is_held = held.contains(&g.id.as_str());
            if !touched && !is_held {
                continue;
            }
            let saving: u128 = demand
                .iter()
                .filter(|(t, n)| **n > 0 && titles[**t].generation == g.id && titles[**t].classical > titles[**t].pinned)
                .map(|(t, n)| *n as u128 * (titles[*t].classical - titles[*t].pinned) as u128)
                .sum();
            let (cost, via) = if is_held {
                (0, "held")
            } else {
                match &g.delta {
                    Some((from, bytes)) if held.contains(&from.as_str()) => (*bytes as u128, "delta"),
                    _ => (g.set_bytes as u128, "set"),
                }
            };
            rows.push((i, saving, cost, via, saving as i128 - cost as i128));
        }
        // Keep those worth holding, at most `cap`, the largest net first and then the later
        // generation; then, with room left, generations already held, the later first.
        let mut worth: Vec<&(usize, u128, u128, &str, i128)> = rows.iter().filter(|r| r.4 > 0).collect();
        worth.sort_by(|a, b| b.4.cmp(&a.4).then(b.0.cmp(&a.0)));
        let mut kept: Vec<usize> = worth.iter().take(cap as usize).map(|r| r.0).collect();
        let mut rest: Vec<&(usize, u128, u128, &str, i128)> = rows.iter().filter(|r| r.3 == "held" && !kept.contains(&r.0)).collect();
        rest.sort_by(|a, b| b.0.cmp(&a.0));
        for r in rest {
            if kept.len() < cap as usize {
                kept.push(r.0);
            }
        }
        let mut fetch = Vec::new();
        let mut bytes: u128 = 0;
        for r in &rows {
            if kept.contains(&r.0) && r.3 != "held" {
                fetch.push(json!([gens[r.0].id, r.3, r.2 as u64]));
                bytes += r.2;
            }
        }
        let mut per_title = Map::new();
        let mut classical_bytes: u128 = 0;
        for (t, n) in &demand {
            if *n == 0 {
                continue;
            }
            let title = &titles[*t];
            let g = gens.iter().position(|g| g.id == title.generation).unwrap_or(usize::MAX);
            let pinned = kept.contains(&g) && title.classical > title.pinned;
            per_title.insert(t.to_string(), json!(if pinned { "pinned" } else { "classical" }));
            bytes += *n as u128 * (if pinned { title.pinned } else { title.classical }) as u128;
            classical_bytes += *n as u128 * title.classical as u128;
        }
        out.push(json!({
            "generations": rows.iter().map(|r| json!({
                "id": gens[r.0].id, "saving": r.1 as u64, "cost": r.2 as u64, "via": r.3, "net": r.4 as i64,
            })).collect::<Vec<_>>(),
            "hold": kept.iter().map(|i| gens[*i].id.clone()).collect::<Vec<_>>(),
            "fetch": fetch,
            "evict": held.iter().filter(|h| !kept.iter().any(|k| gens[*k].id == **h)).collect::<Vec<_>>(),
            "titles": per_title,
            "bytes": bytes as u64,
            "classical_bytes": classical_bytes as u64,
        }));
    }
    Ok(Value::Array(out))
}

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

    fn doc(cap: u64, holds: &[&str], demand: Value) -> Value {
        json!({
            "span": { "from": "2026-11-01T00:00:00Z", "to": "2026-12-01T00:00:00Z" }, "cap": cap,
            "generations": [
                { "id": "g1", "epoch": 1, "set_bytes": 6000, "delta": null },
                { "id": "g2", "epoch": 2, "set_bytes": 6000, "delta": { "from": "g1", "bytes": 1500 } },
            ],
            "titles": [
                { "id": "a", "generation": "g1", "pinned_bytes": 1000, "classical_bytes": 3500 },
                { "id": "b", "generation": "g2", "pinned_bytes": 1000, "classical_bytes": 3000 },
            ],
            "sinks": [{ "holds": holds, "demand": demand }],
        })
    }

    #[test]
    fn worth_holding_is_strictly_more_than_the_cost() {
        let p = plan(&doc(2, &[], json!({ "a": 3 }))).expect("reads");
        assert_eq!(p[0]["fetch"], json!([["g1", "set", 6000]]));
        assert_eq!(p[0]["bytes"], json!(9000));
        // 2500 × 2 = 5000 < 6000: classical.
        let p = plan(&doc(2, &[], json!({ "a": 2 }))).expect("reads");
        assert_eq!(p[0]["titles"]["a"], json!("classical"));
        assert_eq!(p[0]["hold"], json!([]));
    }

    #[test]
    fn a_successor_costs_its_delta_where_the_predecessor_is_held() {
        let p = plan(&doc(2, &["g1"], json!({ "b": 1 }))).expect("reads");
        assert_eq!(p[0]["fetch"], json!([["g2", "delta", 1500]]));
        assert_eq!(p[0]["hold"], json!(["g2", "g1"]));
        let p = plan(&doc(1, &["g1"], json!({ "b": 1 }))).expect("reads");
        assert_eq!(p[0]["evict"], json!(["g1"]));
        let p = plan(&doc(2, &[], json!({ "b": 1 }))).expect("reads");
        assert_eq!(p[0]["titles"]["b"], json!("classical"));
    }

    #[test]
    fn malformed_plans() {
        let mut d = doc(2, &[], json!({ "a": 1 }));
        d["cap"] = json!(0);
        assert_eq!(plan(&d), Err(PLAN_MALFORMED));
        let mut d = doc(2, &[], json!({ "z": 1 }));
        assert_eq!(plan(&d), Err(PLAN_MALFORMED));
        d = doc(2, &[], json!({ "a": 1 }));
        d["span"]["to"] = json!("2026-10-01T00:00:00Z");
        assert_eq!(plan(&d), Err(PLAN_MALFORMED));
        d = doc(2, &[], json!({ "a": 1 }));
        d["generations"][0]["delta"] = json!({ "from": "g2", "bytes": 10 });
        assert_eq!(plan(&d), Err(PLAN_MALFORMED));
    }
}
