//! Integration tests for the FIFA 17 UTAS migration host: `/club` served from a
//! fake Core through the real adapter, faithful Python passthrough against a mock
//! upstream, route-classification safety, negatives, and an end-to-end socket run.

use std::collections::HashMap;
use std::io::{BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::club_response::{CoreOwnedItem, ItemIdentityResolver};
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_identity::JsonIdentityStore;
use openfut_utas_host::{
    classify, read_request, CoreAccess, CoreError, CorePage, Fifa17IdentityResolver, PassClient,
    Route, Server,
};
use parking_lot::Mutex;
use serde_json::Value;

// ── Fakes ────────────────────────────────────────────────────────────────────

/// (method, target, body) recorded by the mock upstream.
type Recorded = Arc<Mutex<Vec<(String, String, Vec<u8>)>>>;

struct FakeCore {
    items: Vec<CoreOwnedItem>,
    total: i64,
    calls: AtomicUsize,
    last_params: Mutex<Vec<(String, String)>>,
    panic_if_called: bool,
    return_err: bool,
}

impl FakeCore {
    fn new(items: Vec<CoreOwnedItem>, total: i64) -> Self {
        FakeCore {
            items,
            total,
            calls: AtomicUsize::new(0),
            last_params: Mutex::new(vec![]),
            panic_if_called: false,
            return_err: false,
        }
    }
    fn forbidden() -> Self {
        FakeCore {
            items: vec![],
            total: 0,
            calls: AtomicUsize::new(0),
            last_params: Mutex::new(vec![]),
            panic_if_called: true,
            return_err: false,
        }
    }
    fn erroring() -> Self {
        FakeCore {
            items: vec![],
            total: 0,
            calls: AtomicUsize::new(0),
            last_params: Mutex::new(vec![]),
            panic_if_called: false,
            return_err: true,
        }
    }
    fn calls(&self) -> usize {
        self.calls.load(Ordering::SeqCst)
    }
    fn last(&self) -> Vec<(String, String)> {
        self.last_params.lock().clone()
    }
}

impl CoreAccess for FakeCore {
    fn query_owned(&self, params: &[(&str, String)]) -> Result<CorePage, CoreError> {
        assert!(
            !self.panic_if_called,
            "Core must NOT be called on this path"
        );
        self.calls.fetch_add(1, Ordering::SeqCst);
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        *self.last_params.lock() = params
            .iter()
            .map(|(k, v)| (k.to_string(), v.clone()))
            .collect();
        Ok(CorePage {
            items: self.items.clone(),
            total: self.total,
        })
    }
}

fn entities() -> Fifa17Entities {
    Fifa17Entities::from_maps(
        HashMap::from([(13, "Premier League".to_string())]),
        HashMap::from([(52, "Argentina".to_string())]),
        HashMap::from([(5, "Chelsea".to_string())]),
    )
}

fn item(
    owned: &str,
    card: &str,
    rating: u8,
    pos: &str,
    nation: &str,
    league: &str,
    club: &str,
) -> CoreOwnedItem {
    CoreOwnedItem {
        owned_card_id: owned.into(),
        card_id: card.into(),
        rating,
        position: pos.into(),
        nation: nation.into(),
        league: league.into(),
        club: club.into(),
        attributes: [90, 88, 70, 85, 40, 78],
    }
}

/// A mock Python upstream: records each request, replies 200 + `X-From-Python`.
fn spawn_mock_python() -> (String, Recorded) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let rec = Arc::new(Mutex::new(Vec::new()));
    let rec2 = rec.clone();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let mut s = match stream {
                Ok(s) => s,
                Err(_) => continue,
            };
            let mut r = BufReader::new(s.try_clone().unwrap());
            if let Ok(Some(req)) = read_request(&mut r) {
                rec2.lock()
                    .push((req.method.clone(), req.target.clone(), req.body.clone()));
                let body = br#"{"python":true}"#;
                let head = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-From-Python: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    body.len()
                );
                let _ = s.write_all(head.as_bytes());
                let _ = s.write_all(body);
            }
        }
    });
    (format!("http://{addr}"), rec)
}

/// A unique temp path for a per-server identity store (integration tests run in
/// the same process; each server gets its own store file).
fn unique_store_path() -> std::path::PathBuf {
    static N: AtomicUsize = AtomicUsize::new(0);
    let n = N.fetch_add(1, Ordering::SeqCst);
    std::env::temp_dir().join(format!("ofut-host-it-{}-{n}.json", std::process::id()))
}

fn build_server(
    core: Arc<FakeCore>,
    upstream: &str,
    assets_map: Option<HashMap<String, u32>>,
) -> Server {
    // The map (card id → asset id) becomes a real identity catalog; an absent
    // map is an empty catalog (every item dropped — the honest no-mapping case).
    let cards = assets_map.unwrap_or_default();
    let entries: Vec<String> = cards
        .iter()
        .map(|(id, asset)| format!("\"{id}\":{{\"asset_id\":{asset}}}"))
        .collect();
    let doc = format!(
        "{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{}}}}}",
        entries.join(",")
    );
    let catalog = Fifa17CardCatalog::from_json_str(&doc).unwrap();
    let store = JsonIdentityStore::open(unique_store_path()).unwrap();
    let assets: Arc<dyn ItemIdentityResolver + Send + Sync> =
        Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
    Server::new(
        core,
        Arc::new(entities()),
        assets,
        Arc::new(PassClient::new(upstream)),
    )
}

// ── /club served from Core ─────────────────────────────────────────────────

#[test]
fn club_route_maps_query_and_shapes_core_items() {
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_ch_1",
            86,
            "CDM",
            "Argentina",
            "Premier League",
            "Chelsea",
        )],
        1,
    ));
    let server = build_server(
        core.clone(),
        "http://127.0.0.1:1", // passthrough must not be used
        Some(HashMap::from([("card_ch_1".to_string(), 20801u32)])),
    );

    let resp = server.handle(
        "GET",
        "/ut/game/fifa17/club?year=2017&type=player&count=11&level=gold&nation=52&league=13&team=5&sort=desc&start=10",
        &[],
        b"",
    );
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    let it = &body["itemData"][0];
    assert_eq!(it["resourceId"], 20801, "real asset id from the resolver");
    assert_eq!(it["rating"], 86);
    assert_eq!(it["preferredPosition"], "CDM");
    assert_eq!(it["leagueId"], 13);
    assert_eq!(it["teamid"], 5);
    assert_eq!(it["nation"], 52);

    // Core was queried once with SEMANTIC params (ids resolved to names), and the
    // FIFA UI window (start/count) became semantic offset/limit.
    assert_eq!(core.calls(), 1);
    let p = core.last();
    assert!(
        p.contains(&("quality".into(), "gold".into())),
        "level=gold -> quality {p:?}"
    );
    assert!(
        p.contains(&("league".into(), "Premier League".into())),
        "league id 13 -> name {p:?}"
    );
    assert!(
        p.contains(&("club".into(), "Chelsea".into())),
        "team id 5 -> club name {p:?}"
    );
    assert!(p.contains(&("offset".into(), "10".into())));
    assert!(p.contains(&("limit".into(), "11".into())));
    // No raw FIFA id reached Core.
    for (_, v) in &p {
        if v == "13" || v == "5" || v == "52" {
            panic!("raw FIFA id leaked into Core params: {p:?}");
        }
    }
}

#[test]
fn club_route_with_empty_asset_map_drops_items_not_fakes_them() {
    // The current production reality: no card→asset mapping → empty itemData.
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_pl_001",
            84,
            "ST",
            "England",
            "Premier League",
            "Northgate United",
        )],
        1,
    ));
    let server = build_server(core.clone(), "http://127.0.0.1:1", None);
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=any", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        body["itemData"].as_array().unwrap().len(),
        0,
        "no fabricated ids"
    );
    assert_eq!(
        core.calls(),
        1,
        "Core still queried; drop happens at shaping"
    );
}

#[test]
fn club_route_unknown_id_returns_empty_and_never_calls_core() {
    let core = Arc::new(FakeCore::forbidden());
    let server = build_server(core.clone(), "http://127.0.0.1:1", None);
    // league 9999 is not in the entity map → hard MapError → empty, no Core call.
    let resp = server.handle("GET", "/ut/game/fifa17/club?league=9999", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["itemData"].as_array().unwrap().len(), 0);
    assert_eq!(core.calls(), 0, "unknown id must short-circuit before Core");
}

// ── Passthrough to Python ────────────────────────────────────────────────────

#[test]
fn passthrough_forwards_verbatim_and_never_calls_core() {
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::forbidden()); // proves /club-only for Core
    let server = build_server(core, &upstream, None);

    let resp = server.handle(
        "POST",
        "/ut/game/fifa17/purchased/items",
        &[("X-UT-SID".into(), "sess".into())],
        br#"{"itemData":[{"id":1}]}"#,
    );
    // upstream response returned faithfully
    assert_eq!(resp.status, 200);
    assert!(
        resp.headers
            .iter()
            .any(|(k, v)| k.eq_ignore_ascii_case("x-from-python") && v == "1"),
        "upstream headers preserved: {:?}",
        resp.headers
    );
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["python"], true);

    // upstream received the exact method, target and body
    std::thread::sleep(std::time::Duration::from_millis(50));
    let got = rec.lock().clone();
    assert_eq!(got.len(), 1);
    assert_eq!(got[0].0, "POST");
    assert_eq!(got[0].1, "/ut/game/fifa17/purchased/items");
    assert_eq!(got[0].2, br#"{"itemData":[{"id":1}]}"#);
}

#[test]
fn mutating_route_classifies_to_passthrough_not_rust() {
    // A PUT to the club PATH is NOT the read route; it must go to Python, never
    // execute Rust/Core (guards against double-applying a mutation).
    assert_eq!(classify("PUT", "/ut/game/fifa17/club"), Route::Passthrough);
    assert_eq!(
        classify("POST", "/ut/game/fifa17/squad/0"),
        Route::Passthrough
    );

    let (upstream, _rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::forbidden());
    let server = build_server(core.clone(), &upstream, None);
    let resp = server.handle("PUT", "/ut/game/fifa17/club", &[], br#"{"x":1}"#);
    assert_eq!(resp.status, 200);
    assert_eq!(core.calls(), 0);
}

// ── End-to-end over a socket (read_request + write_response + keep-alive) ─────

#[test]
fn end_to_end_socket_serves_club_and_passthrough() {
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_ch_1",
            86,
            "CDM",
            "Argentina",
            "Premier League",
            "Chelsea",
        )],
        1,
    ));
    let server = build_server(
        core,
        &upstream,
        Some(HashMap::from([("card_ch_1".to_string(), 20801u32)])),
    );
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    std::thread::spawn(move || server.serve_listener(listener));

    // /club → Rust/Core
    let club = raw_get(
        &addr.to_string(),
        "/ut/game/fifa17/club?level=gold&league=13",
    );
    assert!(club.contains("200 OK"), "club status: {club}");
    assert!(
        club.contains("\"resourceId\":20801") || club.contains("\"resourceId\": 20801"),
        "club body: {club}"
    );

    // passthrough → Python
    let pt = raw_get(&addr.to_string(), "/ut/game/fifa17/tradePile?x=1");
    assert!(pt.contains("200 OK"));
    assert!(
        pt.to_lowercase().contains("x-from-python"),
        "upstream header relayed (case-insensitive): {pt}"
    );

    std::thread::sleep(std::time::Duration::from_millis(50));
    let got = rec.lock().clone();
    assert!(
        got.iter()
            .any(|(m, t, _)| m == "GET" && t == "/ut/game/fifa17/tradePile?x=1"),
        "python saw the passthrough with query intact: {got:?}"
    );
    assert!(
        !got.iter().any(|(_, t, _)| t.contains("/club")),
        "python must NOT have seen the /club request: {got:?}"
    );
}

/// Minimal raw HTTP/1.1 GET (Connection: close) returning the whole response text.
fn raw_get(addr: &str, target: &str) -> String {
    let mut s = TcpStream::connect(addr).unwrap();
    let req = format!("GET {target} HTTP/1.1\r\nHost: fifa\r\nConnection: close\r\n\r\n");
    s.write_all(req.as_bytes()).unwrap();
    let mut buf = String::new();
    s.read_to_string(&mut buf).unwrap();
    buf
}

#[test]
fn club_core_error_returns_empty_and_never_falls_back_to_python() {
    // A Core failure on /club must degrade to an empty page, NOT retry on Python
    // (which could double-apply a mutation on other routes; the rule is absolute).
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::erroring());
    let server = build_server(
        core.clone(),
        &upstream,
        Some(HashMap::from([("c".into(), 1u32)])),
    );
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=any", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["itemData"].as_array().unwrap().len(), 0);
    assert_eq!(core.calls(), 1, "Core was attempted once");
    std::thread::sleep(std::time::Duration::from_millis(50));
    assert!(
        rec.lock().is_empty(),
        "Python must NOT be contacted on a /club core error"
    );
}

#[test]
fn host_does_not_refilter_core_results() {
    // Filtering is Core's job. The host must emit whatever Core returns; if it
    // re-applied the filter it would drop items Core already vetted.
    let core = Arc::new(FakeCore::new(
        vec![
            item(
                "oc1",
                "card_a",
                90,
                "ST",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
            item(
                "oc2",
                "card_b",
                60,
                "GK",
                "England",
                "Premier League",
                "Chelsea",
            ),
        ],
        2,
    ));
    let server = build_server(
        core,
        "http://127.0.0.1:1",
        Some(HashMap::from([
            ("card_a".into(), 20801u32),
            ("card_b".into(), 158023u32),
        ])),
    );
    // Query says gold; Core (faked) returns both regardless. Host must emit BOTH.
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=gold", &[], b"");
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        body["itemData"].as_array().unwrap().len(),
        2,
        "host must not second-guess Core's filtering"
    );
}