//! Differential tests: the Rust adapter against the Python Blaze responder.
//!
//! `openfut-protocol-blaze` proves the *codec* matches. This proves the layer
//! that decides **what to say**: for each inbound frame, the exact frames that
//! go back and their order.
//!
//! Every vector in `fixtures/blaze_transactions.jsonl` was produced by calling
//! the real `blaze_responder_v3b.dispatch()`. Transactions replay in file order
//! against a shared session per connection, so ordering-dependent behaviour is
//! exercised rather than assumed — preAuth captures the locale that later ALOC
//! fields echo, and login sets the auth code getAuthToken returns afterwards.
//!
//! Comparison is byte-for-byte, including frame count and order. A missing
//! post-login notification or a reply where the oracle stays silent is a
//! failure here, which is the whole point.
//!
//! Regenerate after any oracle change:  python3 fixtures/generate.py

use std::collections::HashMap;

use openfut_adapter_fifa17::blaze::{Adapter, AdapterConfig, Endpoints, Identity, Session};
use openfut_protocol_blaze::fire2::Frame;
use openfut_protocol_blaze::heat2;
use serde_json::Value as J;

fn records() -> Vec<J> {
    let path = format!(
        "{}/fixtures/blaze_transactions.jsonl",
        env!("CARGO_MANIFEST_DIR")
    );
    let text = std::fs::read_to_string(&path)
        .unwrap_or_else(|e| panic!("cannot read {path}: {e}\nrun: python3 fixtures/generate.py"));
    text.lines()
        .filter(|l| !l.trim().is_empty())
        .map(|l| serde_json::from_str(l).expect("fixture line is valid JSON"))
        .collect()
}

fn unhex(s: &str) -> Vec<u8> {
    (0..s.len())
        .step_by(2)
        .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("valid hex"))
        .collect()
}

fn hex(b: &[u8]) -> String {
    b.iter().map(|x| format!("{x:02x}")).collect()
}

fn st(j: &J, k: &str) -> String {
    j[k].as_str()
        .unwrap_or_else(|| panic!("{k} is a string"))
        .to_string()
}

fn n(j: &J, k: &str) -> i64 {
    j[k].as_i64().unwrap_or_else(|| panic!("{k} is a number"))
}

/// Rebuild the exact configuration the fixtures were generated under.
///
/// The generator uses deliberately non-loopback addresses, so an adapter that
/// hardcoded one instead of reading its config fails loudly here rather than
/// coincidentally matching a default.
fn config_from(record: &J) -> (AdapterConfig, i64) {
    let id = &record["identity"];
    let identity = Identity {
        persona_id: n(id, "persona_id"),
        persona_name: st(id, "persona_name"),
        user_id: n(id, "user_id"),
        ext_id: n(id, "ext_id"),
        email: st(id, "email"),
        namespace: st(id, "namespace"),
        client_platform: n(id, "client_platform"),
        persona_status: n(id, "persona_status"),
        user_session_type: n(id, "user_session_type"),
        account_locale: n(id, "account_locale_int"),
        locale: st(id, "locale"),
        content_id: st(id, "content_id"),
        entitlement_tag: st(id, "entitlement_tag"),
        entitlement_group: st(id, "entitlement_group"),
        title_id: st(id, "title_id"),
        client_id: st(id, "client_id"),
        platform: st(id, "platform"),
    };
    let endpoints = Endpoints {
        advertise: st(record, "advertise"),
        bind: st(record, "bind"),
        pow_content_host: st(record, "pow_content_host"),
        pow_host: st(record, "pow_host"),
        ..Endpoints::default()
    };
    let cfg = AdapterConfig {
        identity,
        endpoints,
        server_version: st(&record["identity"], "server_version"),
    };
    (cfg, n(record, "now"))
}

struct Replay {
    adapter: Adapter,
    now: i64,
    sessions: HashMap<String, Session>,
    records: Vec<J>,
}

fn setup() -> Replay {
    let records = records();
    let cfg_rec = records
        .iter()
        .find(|r| r["kind"] == "config")
        .expect("fixture carries a config record")
        .clone();
    let (cfg, now) = config_from(&cfg_rec);

    let mut sessions = HashMap::new();
    for r in &records {
        if r["kind"] == "session" {
            // The session key is injected, not generated: it appears verbatim
            // in three responses, so a self-minted one could never match.
            sessions.insert(
                st(r, "id"),
                Session::new(st(r, "session_key"), n(r, "account_locale")),
            );
        }
    }

    Replay {
        adapter: Adapter::new(cfg),
        now,
        sessions,
        records,
    }
}

/// The headline test: replay every transaction and require identical frames.
#[test]
fn dispatch_matches_python_oracle_byte_for_byte() {
    let mut rp = setup();
    let records = rp.records.clone();
    let mut checked = 0usize;

    for rec in records.iter().filter(|r| r["kind"] == "tx") {
        let name = st(rec, "name");
        let sid = st(rec, "session");
        let request = unhex(&st(rec, "request_hex"));
        let expected: Vec<String> = rec["responses"]
            .as_array()
            .expect("responses array")
            .iter()
            .map(|f| f.as_str().unwrap().to_string())
            .collect();

        let (frame, used) = Frame::parse(&request)
            .unwrap_or_else(|e| panic!("{name}: fixture request does not parse: {e}"));
        assert_eq!(used, request.len(), "{name}: trailing bytes in request");

        let body = if frame.payload.is_empty() {
            heat2::Struct::new()
        } else {
            heat2::decode(&frame.payload)
                .unwrap_or_else(|e| panic!("{name}: request body is not valid TDF: {e}"))
        };

        let session = rp.sessions.get_mut(&sid).expect("session declared");
        let out = rp.adapter.dispatch(&frame.header, &body, session, rp.now);

        assert_eq!(
            out.len(),
            expected.len(),
            "\n{name}: produced {} frame(s), oracle produced {}",
            out.len(),
            expected.len()
        );
        for (idx, (got, want)) in out.iter().zip(expected.iter()).enumerate() {
            let got_hex = hex(&got.encode());
            if &got_hex != want {
                // Narrow the failure to header vs body before dumping bytes.
                let want_bytes = unhex(want);
                let got_bytes = got.encode();
                assert_eq!(
                    hex(&got_bytes[..16.min(got_bytes.len())]),
                    hex(&want_bytes[..16.min(want_bytes.len())]),
                    "\n{name} frame {idx}: HEADER differs"
                );
                panic!(
                    "\n{name} frame {idx}: BODY differs\n  got  {} bytes\n  want {} bytes",
                    got_bytes.len().saturating_sub(16),
                    want_bytes.len().saturating_sub(16)
                );
            }
        }
        checked += 1;
    }

    assert!(
        checked >= 40,
        "expected the full script, replayed {checked}"
    );
}

/// Frame counts and ordering are part of the contract, so assert them
/// separately from bytes — a rewrite that answered correctly but dropped a
/// notification would otherwise fail with an unhelpful byte diff.
#[test]
fn frame_counts_and_ordering_match() {
    let mut rp = setup();
    let records = rp.records.clone();

    for rec in records.iter().filter(|r| r["kind"] == "tx") {
        let name = st(rec, "name");
        let request = unhex(&st(rec, "request_hex"));
        let expected = rec["responses"].as_array().unwrap();

        let (frame, _) = Frame::parse(&request).unwrap();
        let body = if frame.payload.is_empty() {
            heat2::Struct::new()
        } else {
            heat2::decode(&frame.payload).unwrap()
        };
        let session = rp.sessions.get_mut(&st(rec, "session")).unwrap();
        let out = rp.adapter.dispatch(&frame.header, &body, session, rp.now);

        assert_eq!(out.len(), expected.len(), "{name}: frame count");

        for (got, want_hex) in out.iter().zip(expected.iter()) {
            let want = Frame::parse(&unhex(want_hex.as_str().unwrap())).unwrap().0;
            assert_eq!(
                got.header.component, want.header.component,
                "{name}: component"
            );
            assert_eq!(got.header.command, want.header.command, "{name}: command");
            assert_eq!(got.header.msg_type, want.header.msg_type, "{name}: msgType");
            assert_eq!(got.header.msg_num, want.header.msg_num, "{name}: msgNum");
            assert_eq!(
                got.header.user_index, want.header.user_index,
                "{name}: userIndex"
            );
        }
    }
}

/// The login burst is the sequence most likely to be silently wrong, so pin it
/// explicitly rather than relying on it being buried in the byte comparison.
#[test]
fn login_emits_reply_then_exactly_three_pushes() {
    let rp = setup();
    let login = rp
        .records
        .iter()
        .find(|r| r["kind"] == "tx" && r["name"] == "login")
        .expect("login transaction present");

    let frames: Vec<Frame> = login["responses"]
        .as_array()
        .unwrap()
        .iter()
        .map(|h| Frame::parse(&unhex(h.as_str().unwrap())).unwrap().0)
        .collect();

    assert_eq!(frames.len(), 4, "reply + three UserSessions pushes");
    assert_eq!(
        frames[0].header.msg_type,
        openfut_protocol_blaze::fire2::MsgType::Reply
    );
    let notify_ids: Vec<u16> = frames[1..].iter().map(|f| f.header.command).collect();
    assert_eq!(notify_ids, vec![0x0008, 0x0001, 0x0002]);
}

/// The generator uses non-loopback addresses, so any loopback literal left in a
/// response means the adapter hardcoded something it should have read from
/// config — the exact regression the client/server split was meant to prevent.
///
/// Two keys are genuine literals in the oracle, not substitution failures.
/// Both are OAuth redirect targets the client never actually dials (the flow is
/// forged), so the loopback is inert; they are allowlisted by key rather than
/// by pattern so a third one cannot slip in unnoticed.
const ALLOWED_LOOPBACK_KEYS: [&str; 2] = ["identityRedirectUri", "redirect_uri"];

#[test]
fn no_response_hardcodes_a_loopback_address() {
    let mut rp = setup();
    let records = rp.records.clone();
    let advertise = "198.51.100.7";

    for rec in records.iter().filter(|r| r["kind"] == "tx") {
        let name = st(rec, "name");
        let request = unhex(&st(rec, "request_hex"));
        let (frame, _) = Frame::parse(&request).unwrap();
        let body = if frame.payload.is_empty() {
            heat2::Struct::new()
        } else {
            heat2::decode(&frame.payload).unwrap()
        };
        let session = rp.sessions.get_mut(&st(rec, "session")).unwrap();
        let out = rp.adapter.dispatch(&frame.header, &body, session, rp.now);

        for f in &out {
            let text = String::from_utf8_lossy(&f.payload);
            for (at, _) in text.match_indices("127.0.0.1") {
                // TDF strings are length-prefixed and NUL-terminated, so the
                // owning key sits shortly before the value. Look back far
                // enough to name it, and require it to be allowlisted.
                let start = at.saturating_sub(80);
                let context = &text[start..text.len().min(at + 64)];
                assert!(
                    ALLOWED_LOOPBACK_KEYS.iter().any(|k| context.contains(k)),
                    "\n{name}: unexpected loopback literal, context {context:?}"
                );
            }
        }
        // The advertised address must actually appear somewhere in the config
        // responses, or substitution silently did nothing.
        if name.starts_with("fetch_config") || name == "preauth" {
            let text = String::from_utf8_lossy(&out[0].payload);
            assert!(
                text.contains(advertise),
                "{name}: advertised address missing from the config payload"
            );
        }
    }
}

/// Session state must survive across RPCs on one connection, and must NOT leak
/// between connections.
#[test]
fn session_state_is_per_connection() {
    let mut rp = setup();
    let records = rp.records.clone();
    for rec in records.iter().filter(|r| r["kind"] == "tx") {
        let request = unhex(&st(rec, "request_hex"));
        let (frame, _) = Frame::parse(&request).unwrap();
        let body = if frame.payload.is_empty() {
            heat2::Struct::new()
        } else {
            heat2::decode(&frame.payload).unwrap()
        };
        let session = rp.sessions.get_mut(&st(rec, "session")).unwrap();
        rp.adapter.dispatch(&frame.header, &body, session, rp.now);
    }

    // "main" logged in with an auth code and an enUS preAuth.
    let main = &rp.sessions["main"];
    assert!(main.logged_in);
    assert_eq!(main.auth_code, "OPENFUT-TEST-AUTHCODE");
    assert_eq!(main.account_locale, 0x656E_5553);

    // "locale" ran a deDE preAuth and a login carrying no AUTH member.
    let loc = &rp.sessions["locale"];
    assert_eq!(loc.account_locale, 0x6465_4445, "deDE locale captured");
    assert_eq!(loc.service_name, "fifa-2017-pc-de");
    assert!(loc.auth_code.is_empty());

    // "fallbacks" never logged in.
    assert!(!rp.sessions["fallbacks"].logged_in);
}