//! Cross-language conformance for the move-aware sequence CRDT
//! (`lazily-spec/cell-model.md` § Move-aware sequence order), replaying the
//! canonical compute fixture `lazily-spec/conformance/collections/seqcrdt_convergence.json`.
//!
//! Each element is three independent LWW registers — value, position
//! (fractional-index byte key + peer), deleted — stamped by an HLC. A move is a
//! SINGLE LWW reassignment of position (not delete+reinsert), so concurrent
//! moves of the same element converge to the later stamp without duplication; a
//! concurrent move + value-edit both apply (independent registers). Removal is
//! an LWW tombstone. Order is the lexicographic total order on `(frac, peer)`.
//!
//! Required of every binding — see the Binding Conformance Matrix in
//! `lazily-spec/protocol.md`. Feature-gated because `SeqCrdt` lives behind the
//! `distributed` feature (the CRDT plane).

#![cfg(feature = "distributed")]

mod common;

use std::collections::HashMap;

use common::Expect;
use lazily::{PeerId, SeqCrdt};
use serde_json::Value;

const SPEC_DIR: common::SpecDir = common::SpecDir("collections");

type V = Value;

fn load_fixture(name: &str) -> Value {
    let path = format!("{SPEC_DIR}/{name}");
    let raw = crate::common::spec_read_to_string(&path)
        .unwrap_or_else(|e| panic!("failed to read fixture {path}: {e}"));
    serde_json::from_str(&raw).unwrap_or_else(|e| panic!("failed to parse fixture {path}: {e}"))
}

/// Apply a fixture op to `replica`. `now` comes from the step (or its
/// surrounding `merge` block); the op uses `step.now` if present.
fn apply_op(replica: &mut SeqCrdt<String, V>, op: &Value, now: u64) {
    let kind = op
        .get("op")
        .and_then(|v| v.as_str())
        .expect("seqcrdt op.op");
    match kind {
        "insert_back" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            let val = value_of(op.get("value").unwrap());
            replica.insert_back(id, val, now);
        }
        "insert_front" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            let val = value_of(op.get("value").unwrap());
            replica.insert_front(id, val, now);
        }
        "set_value" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            let val = value_of(op.get("value").unwrap());
            replica.set_value(&id, val, now);
        }
        "move_after" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            let anchor = op
                .get("anchor")
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();
            replica.move_after(&id, &anchor, now);
        }
        "move_before" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            let anchor = op
                .get("anchor")
                .and_then(|v| v.as_str())
                .unwrap()
                .to_string();
            replica.move_before(&id, &anchor, now);
        }
        "remove" => {
            let id = op.get("id").and_then(|v| v.as_str()).unwrap().to_string();
            replica.remove(&id, now);
        }
        other => panic!("unknown seqcrdt op: {other}"),
    }
}

fn value_of(v: &Value) -> Value {
    v.clone()
}

fn assert_order(replica: &SeqCrdt<String, V>, expected: &[Value], msg: &str) {
    let got: Vec<String> = replica.order();
    let want: Vec<String> = expected
        .iter()
        .map(|v| match v {
            Value::String(s) => s.clone(),
            Value::Number(n) => n.to_string(),
            other => panic!("expected order entry is a string/number, got {other}"),
        })
        .collect();
    assert_eq!(
        got, want,
        "{msg}: order mismatch — got {got:?} expected {want:?}"
    );
}

fn run_seqcrdt_fixture(name: &str) {
    let fixture = load_fixture(name);
    // Per-scenario replay ledger (`#lzscenariocoverage`).
    for (i, _id, scenario) in common::scenarios(&format!("{SPEC_DIR}/{name}"), &fixture) {
        let mut replicas: HashMap<String, SeqCrdt<String, V>> = HashMap::new();

        // Seed: either a `replica` (empty single-peer) or a `seed` with inserts.
        if let Some(seed) = scenario.get("seed").and_then(|v| v.as_object()) {
            let peer = seed.get("peer").and_then(|v| v.as_u64()).unwrap();
            let mut a: SeqCrdt<String, V> = SeqCrdt::new(PeerId(peer));
            for ins in seed.get("inserts").and_then(|v| v.as_array()).unwrap() {
                let id = ins.get("id").and_then(|v| v.as_str()).unwrap().to_string();
                let val = value_of(ins.get("value").unwrap());
                let now = ins.get("now").and_then(|v| v.as_u64()).unwrap();
                a.insert_back(id, val, now);
            }
            replicas.insert("a".to_string(), a);
        } else if let Some(rep) = scenario.get("replica") {
            let peer = rep.get("peer").and_then(|v| v.as_u64()).unwrap();
            replicas.insert("a".to_string(), SeqCrdt::new(PeerId(peer)));
        } else {
            panic!("scenario {i}: missing seed or replica");
        }

        for step in scenario
            .get("steps")
            .and_then(|v| v.as_array())
            .expect("seqcrdt steps")
        {
            if let Some(fork_name) = step.get("fork").and_then(|v| v.as_str()) {
                let peer = step.get("peer").and_then(|v| v.as_u64()).unwrap();
                let src = replicas
                    .get("a")
                    .unwrap_or_else(|| panic!("scenario {i}: fork from missing `a`"));
                // Fork = deep copy of the source's entries (original stamps
                // preserved) under a new owning peer id; the forked replica
                // continues from the source's causal state.
                let forked = src.fork(PeerId(peer));
                replicas.insert(fork_name.to_string(), forked);
            } else if let Some(new_name) = step.get("clone").and_then(|v| v.as_str()) {
                let from = step
                    .get("from")
                    .and_then(|v| v.as_str())
                    .unwrap_or_else(|| panic!("scenario {i}: clone missing `from`"));
                // Clone = byte-identical deep copy (same peer, same stamps).
                let cloned = replicas
                    .get(from)
                    .unwrap_or_else(|| panic!("scenario {i}: clone from missing `{from}`"))
                    .clone();
                replicas.insert(new_name.to_string(), cloned);
            } else if let Some(merge) = step.get("merge").and_then(|v| v.as_object()) {
                let into = merge
                    .get("into")
                    .and_then(|v| v.as_str())
                    .unwrap_or_else(|| panic!("scenario {i}: merge missing `into`"));
                let from = merge
                    .get("from")
                    .and_then(|v| v.as_str())
                    .unwrap_or_else(|| panic!("scenario {i}: merge missing `from`"));
                let now = step.get("now").and_then(|v| v.as_u64()).unwrap();
                let from_state = replicas
                    .get(from)
                    .unwrap_or_else(|| panic!("scenario {i}: merge from missing `{from}`"))
                    .clone();
                replicas
                    .get_mut(into)
                    .unwrap_or_else(|| panic!("scenario {i}: merge into missing `{into}`"))
                    .merge(&from_state, now);
            } else if let Some(on) = step.get("on").and_then(|v| v.as_str()) {
                let op = step; // op fields are inlined at the step level
                let now = step.get("now").and_then(|v| v.as_u64()).unwrap();
                apply_op(
                    replicas
                        .get_mut(on)
                        .unwrap_or_else(|| panic!("scenario {i}: `on` target `{on}` missing")),
                    op,
                    now,
                );
            } else if step.get("op").is_some() {
                let now = step.get("now").and_then(|v| v.as_u64()).unwrap();
                apply_op(
                    replicas
                        .get_mut("a")
                        .expect("scenario {i}: default target `a` missing"),
                    step,
                    now,
                );
            } else {
                panic!("scenario {i}: unrecognized step {step}");
            }
        }

        // Assertions. The `expect` block is guarded (`#lzassertunknownkeys`):
        // a key this runner never reads fails the fixture instead of passing.
        let expect = Expect::new(
            format!("{SPEC_DIR}/{name}"),
            format!("scenarios[{i}].expect"),
            scenario.get("expect").unwrap(),
        );
        // Every key is optional per scenario, so each comparison is bound to the
        // key's *presence*: a bare read that a missing key silently skipped would
        // mark the key consumed while asserting nothing
        // (`#lzconsumednotasserted`).
        //
        // `len` and `contains_all` apply to the converged replica — the first in
        // the first `orders_equal` pair when present, else `a`. That lookup goes
        // through `raw()` because it *selects a target*, not a value to compare;
        // `orders_equal` is asserted on its own below.
        let target = expect
            .raw()
            .get("orders_equal")
            .and_then(|v| v.as_array())
            .and_then(|a| a.first())
            .and_then(|v| v.as_array())
            .and_then(|a| a.first())
            .and_then(|v| v.as_str())
            .unwrap_or("a");
        expect.assert_key_if_present("order", |want| {
            assert_order(
                &replicas["a"],
                want.as_array().expect("order"),
                &format!("scenario {i}"),
            );
        });
        // DESCENT (`#lzsubblockkeyset`): the element ids are the child's keys.
        if let Some(get) = expect.sub_if_present("get") {
            for id in get.raw().as_object().expect("get").keys() {
                let got = replicas["a"]
                    .get(&id.to_string())
                    .unwrap_or_else(|| panic!("scenario {i}: get({id}) missing"));
                get.assert_key_at(id.as_str(), got, &format!("scenario {i}.get"));
            }
        }
        expect.assert_key_if_present("len", |want| {
            assert_eq!(
                replicas[target].values().len() as u64,
                want.as_u64().expect("len"),
                "scenario {i}: len mismatch on `{target}`"
            );
        });
        expect.assert_key_if_present("orders_equal", |want| {
            for pair in want.as_array().expect("orders_equal") {
                let a = pair.get(0).and_then(|v| v.as_str()).unwrap();
                let b = pair.get(1).and_then(|v| v.as_str()).unwrap();
                assert_eq!(
                    replicas[a].order(),
                    replicas[b].order(),
                    "scenario {i}: `{a}`/`{b}` orders should converge"
                );
            }
        });
        // DESCENT (`#lzsubblockkeyset`): the replica names are the child's keys.
        if let Some(order_on) = expect.sub_if_present("order_on") {
            let names: Vec<String> = order_on
                .raw()
                .as_object()
                .expect("order_on")
                .keys()
                .cloned()
                .collect();
            for name in &names {
                order_on.assert_key_with(name.as_str(), |order| {
                    assert_order(
                        &replicas[name],
                        order.as_array().unwrap(),
                        &format!("scenario {i} on `{name}`"),
                    );
                });
            }
        }
        // DESCENT twice: replica name, then element id.
        if let Some(get_on) = expect.sub_if_present("get_on") {
            let names: Vec<String> = get_on
                .raw()
                .as_object()
                .expect("get_on")
                .keys()
                .cloned()
                .collect();
            for name in &names {
                let gets = get_on.sub(name);
                let ids: Vec<String> = gets.raw().as_object().unwrap().keys().cloned().collect();
                for id in &ids {
                    let got = replicas[name]
                        .get(&id.to_string())
                        .unwrap_or_else(|| panic!("scenario {i}: get_on({name},{id}) missing"));
                    gets.assert_key_at(
                        id.as_str(),
                        got,
                        &format!("scenario {i}: get_on({name},{id})"),
                    );
                }
                gets.finish();
            }
        }
        expect.assert_key_if_present("contains_all", |want| {
            for id in want.as_array().expect("contains_all") {
                let s = id.as_str().unwrap();
                assert!(
                    replicas[target].contains(&s.to_string()),
                    "scenario {i}: `{target}` should contain `{s}`"
                );
            }
        });
        // DESCENT (`#lzsubblockkeyset`): the replica names are the child's keys.
        if let Some(not_contains_on) = expect.sub_if_present("not_contains_on") {
            let names: Vec<String> = not_contains_on
                .raw()
                .as_object()
                .expect("not_contains_on")
                .keys()
                .cloned()
                .collect();
            for name in &names {
                not_contains_on.assert_key_with(name.as_str(), |ids| {
                    for id in ids.as_array().unwrap() {
                        let s = id.as_str().unwrap();
                        assert!(
                            !replicas[name].contains(&s.to_string()),
                            "scenario {i}: `{name}` should not contain `{s}`"
                        );
                    }
                });
            }
        }
    }
}

#[test]
fn conformance_seqcrdt_convergence() {
    run_seqcrdt_fixture("seqcrdt_convergence.json");
}
