//! Real host↔Core economy integration harness.
//!
//! Spawns OpenFUT Core (axum) on an ephemeral loopback port backed by a
//! disposable temp-file SQLite, seeds a `fifa17`-scoped profile via the real
//! Core HTTP API, then drives the HOST's REAL economy transport
//! (`HttpCoreClient` implementing `CoreEconomy`) + handlers against it — no
//! fakes, no in-memory doubles. It proves the credits / purchasegroup /
//! userMassInfo / match-reward cluster end-to-end and that state survives a
//! Core restart from the same on-disk database.
//!
//! Safety: uses only a temp directory + `127.0.0.1:0` ephemeral ports. Never
//! touches the production Core DB, production ports/containers, or `.105`.

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_adapter_fifa17::fut::store_session::StoreMode;
use openfut_identity::JsonIdentityStore;
use openfut_utas_host::async_bridge::AsyncBridge;
use openfut_utas_host::market_store::MarketStore;
use openfut_utas_host::pile_store::PileStore;
use openfut_utas_host::{
    build_content_pool, handle_credits, handle_match_end, handle_purchasegroup,
    overlay_massinfo_economy, CoreAccess, CoreEconomy, EconomyServices, Fifa17IdentityResolver,
    HttpCoreClient, PassClient, Server,
};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::Arc;

/// Boot a Core instance against `db_url`, serving on an ephemeral port. Returns
/// the serve task handle and its base URL.
async fn start_core(db_url: &str) -> (tokio::task::JoinHandle<()>, String) {
    // Multi-connection pool: the fresh-DB write-lock race is fixed in Core
    // (WAL-establish-once + BEGIN IMMEDIATE write transactions + busy_timeout,
    // core fbb54ea), proven by the concurrency reproduction (800/800 concurrent
    // writes), so a real multi-connection pool is stable here.
    let pool = openfut_core::db::init_pool(db_url, 5)
        .await
        .expect("core pool");
    openfut_core::db::run_migrations(&pool)
        .await
        .expect("core migrations");
    // `data` lives in the sibling core crate; tests run with the host crate as cwd.
    let app = openfut_core::build_app(pool, "../openfut-core/data")
        .await
        .expect("core build_app");
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind ephemeral");
    let addr = listener.local_addr().unwrap();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    (handle, format!("http://{addr}"))
}

/// Boot a Core with the FIFA17 dev CONTENT loaded (so `/collection` renders real
/// card definitions) and, on first boot, the dev inventory SEEDED (a fifa17
/// profile + club with 100k coins + one owned instance per definition). On
/// restart pass `seed=false`: content is reloaded but the durable DB is left as
/// is, proving persistence rather than re-seeding.
async fn start_core_seeded(db_url: &str, seed: bool) -> (tokio::task::JoinHandle<()>, String) {
    use openfut_core::config::Config;
    use openfut_core::seed::{seed_fifa17_dev, FIFA17_GAME};
    use openfut_core::services::card_db::CardDb;

    let data_dir = "../openfut-core/data";
    let pool = openfut_core::db::init_pool(db_url, 5)
        .await
        .expect("core pool");
    openfut_core::db::run_migrations(&pool)
        .await
        .expect("core migrations");
    if seed {
        let mut card_db = CardDb::load(data_dir).expect("card_db load");
        card_db
            .load_game_dev(data_dir, FIFA17_GAME)
            .expect("load fifa17 dev content");
        seed_fifa17_dev(&pool, &card_db)
            .await
            .expect("seed fifa17 dev inventory");
    }
    let cfg = Config {
        listen_addr: "127.0.0.1:0".into(),
        database_url: "sqlite::memory:".into(),
        data_dir: data_dir.into(),
        max_connections: 5,
        dev_content_games: vec![FIFA17_GAME.to_string()],
        content_packs: Vec::new(),
    };
    let app = openfut_core::app::build(pool, cfg)
        .await
        .expect("core app::build");
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind ephemeral");
    let addr = listener.local_addr().unwrap();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    (handle, format!("http://{addr}"))
}

fn wait_ready(base: &str) {
    let http = reqwest::blocking::Client::new();
    // Generous ceiling (~30s): returns on first success, so it only ever waits
    // this long if Core genuinely never comes up. Under heavy parallel test-binary
    // load Core's content load (CardDb::load) can take several seconds to be ready.
    for _ in 0..1500 {
        if let Ok(r) = http.get(format!("{base}/health")).send() {
            if r.status().is_success() {
                return;
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
    panic!("core did not become ready at {base}");
}

fn post(http: &reqwest::blocking::Client, base: &str, path: &str, body: Value) -> Value {
    let resp = http
        .post(format!("{base}{path}"))
        .header("X-OpenFUT-Game", "fifa17")
        .json(&body)
        .send()
        .unwrap_or_else(|e| panic!("POST {path}: {e}"));
    let status = resp.status();
    let v: Value = resp.json().unwrap_or(Value::Null);
    assert!(status.is_success(), "POST {path} -> {status}: {v}");
    v
}

fn pack_ids(pg: &Value) -> Vec<u64> {
    pg["purchase"]
        .as_array()
        .unwrap()
        .iter()
        .map(|p| p["id"].as_u64().unwrap())
        .collect()
}

/// Seed via the real Core HTTP API, then exercise the host handlers + transport.
/// Returns nothing; panics on any mismatch.
fn seed_and_exercise(base: &str) {
    wait_ready(base);
    let http = reqwest::blocking::Client::new();

    // Seed a fifa17 profile + club (auth grants 5000 coins + a starter pack).
    post(&http, base, "/auth/local", json!({ "username": "CAGE" }));

    // The host's REAL transport to Core (no fake).
    let client = HttpCoreClient::new(base, "fifa17");

    // credits reads the authoritative Core balance.
    let credits: Value = serde_json::from_slice(&handle_credits(&client).body).unwrap();
    assert_eq!(credits["currencies"][0]["funds"], 5000, "seeded balance");

    // Match-reward WRITER: win credits +400 via Core grant_reward, end to end.
    let m = handle_match_end(&client, br#"{"endReason":"WIN"}"#);
    assert_eq!(m.status, 200);
    let mb: Value = serde_json::from_slice(&m.body).unwrap();
    assert_eq!(mb["allCoins"], 5400, "match reward credited in Core");
    assert_eq!(client.balance().unwrap(), 5400);

    // Buy a numeric entitlement "70" through the Core economy API (debit 600).
    post(
        &http,
        base,
        "/economy/purchase-entitlement",
        json!({ "cost": 600, "definition_id": "70" }),
    );
    assert_eq!(client.balance().unwrap(), 4800, "debit applied atomically");

    // credits reflects the debit through the same Core state.
    let credits2: Value = serde_json::from_slice(&handle_credits(&client).body).unwrap();
    assert_eq!(credits2["currencies"][0]["funds"], 4800);

    // purchasegroup full-gen shows the owned pack 70 and NO sentinel.
    let pg: Value =
        serde_json::from_slice(&handle_purchasegroup(&client, StoreMode::Sentinel).body).unwrap();
    let ids = pack_ids(&pg);
    assert!(
        ids.contains(&70),
        "owned pack 70 rendered from Core entitlement"
    );
    assert!(!ids.contains(&65534), "no sentinel while a pack is owned");

    // userMassInfo overlay derives coins from the SAME Core state as credits.
    let mut mass = json!({
        "userInfo": { "currencies": [ {"name":"coins","funds":0,"finalFunds":0} ] }
    });
    overlay_massinfo_economy(
        &mut mass,
        client.balance().unwrap(),
        client.entitlements().unwrap().len(),
    );
    assert_eq!(mass["userInfo"]["currencies"][0]["funds"], 4800);
    // Invariant: credits coins == userMassInfo coins == Core balance.
    assert_eq!(
        credits2["currencies"][0]["funds"],
        mass["userInfo"]["currencies"][0]["funds"]
    );
}

/// After a Core restart from the same DB file, all economy state persists.
fn verify_after_restart(base: &str) {
    wait_ready(base);
    let client = HttpCoreClient::new(base, "fifa17");
    assert_eq!(
        client.balance().unwrap(),
        4800,
        "coins persisted across restart"
    );
    let credits: Value = serde_json::from_slice(&handle_credits(&client).body).unwrap();
    assert_eq!(credits["currencies"][0]["funds"], 4800);
    let pg: Value =
        serde_json::from_slice(&handle_purchasegroup(&client, StoreMode::Sentinel).body).unwrap();
    assert!(
        pack_ids(&pg).contains(&70),
        "entitlement persisted across restart"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn economy_end_to_end_and_restart_persistence() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-it-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let db_url = format!("sqlite://{}/econ.db", dir.display());

    // --- Core instance #1: seed + exercise the full cluster ---
    let (h1, base1) = start_core(&db_url).await;
    let b1 = base1.clone();
    let r = tokio::task::spawn_blocking(move || seed_and_exercise(&b1)).await;
    h1.abort();
    r.expect("exercise phase");

    // --- Core instance #2: same on-disk DB, prove persistence ---
    let (h2, base2) = start_core(&db_url).await;
    let b2 = base2.clone();
    let r2 = tokio::task::spawn_blocking(move || verify_after_restart(&b2)).await;
    h2.abort();
    r2.expect("restart phase");

    std::fs::remove_dir_all(&dir).ok();
}

// ─────────────── Full economy sequence through the real host dispatch ────────
//
// Everything below drives the ACTUAL `Server::try_handle_economy` path (the same
// dispatch + async bridge the barrier will route production through), against a
// live in-process Core over the real blocking `HttpCoreClient`. No fakes. The
// sequence runs on a plain OS thread (no ambient Tokio runtime), exactly like the
// thread-per-connection server, so the bridge takes its DIRECT `block_on` path.

/// Facts captured from the write sequence, re-checked after a full restart.
struct SeqResult {
    final_balance: i64,
    sold_listing: String,
    moved_core_id: String,
}

/// Build a real `Server` with economy authority wired: real Core transport, a
/// catalog derived from the seeded content (so the pack pool + shaper resolve),
/// a persistent identity store, and the two durable SQLite stores opened via the
/// bridge. Returns the server, a direct Core client for balance assertions, and
/// the identity resolver (to reverse a wire id for the restart pile check).
fn build_econ_server(
    base: &str,
    dir: &std::path::Path,
) -> (Server, HttpCoreClient, Arc<Fifa17IdentityResolver>, i64) {
    let probe = HttpCoreClient::new(base, "fifa17");
    let owned = probe.all_owned().expect("core collection");
    assert!(!owned.is_empty(), "seed must grant a starter collection");
    // A catalog mapping every seeded definition to a real-looking asset id (in
    // production this is the shipped FIFA catalog; here it is derived so the E2E
    // exercises real shaping without a static fixture).
    let mut entries = String::new();
    let mut seen = std::collections::HashSet::new();
    let mut asset = 20000u32;
    for it in &owned {
        if !seen.insert(it.card_id.clone()) {
            continue;
        }
        if !entries.is_empty() {
            entries.push(',');
        }
        entries.push_str(&format!("\"{}\":{{\"asset_id\":{asset}}}", it.card_id));
        asset += 1;
    }
    let doc = format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}");
    let catalog = Fifa17CardCatalog::from_json_str(&doc).expect("catalog");
    let store = JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).unwrap();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
    let entities = Arc::new(Fifa17Entities::from_maps(
        HashMap::new(),
        HashMap::new(),
        HashMap::new(),
    ));
    let core: Arc<dyn CoreAccess> = Arc::new(HttpCoreClient::new(base, "fifa17"));
    let bridge = Arc::new(AsyncBridge::new().unwrap());
    let market_path = dir.join("market.db").to_string_lossy().into_owned();
    let market = Arc::new(
        bridge
            .block_on(async move { MarketStore::open(&market_path).await })
            .expect("market store"),
    );
    let pile_path = dir.join("pile.db").to_string_lossy().into_owned();
    let piles = Arc::new(
        bridge
            .block_on(async move { PileStore::open(&pile_path).await })
            .expect("pile store"),
    );
    let econ: Arc<dyn CoreEconomy> = Arc::new(HttpCoreClient::new(base, "fifa17"));
    let pool = Arc::new(build_content_pool(core.as_ref(), resolver.as_ref()));
    assert!(
        !pool.is_empty(),
        "content pool derived from real Core content"
    );
    let services = Arc::new(EconomyServices {
        econ,
        market,
        piles,
        bridge,
        pool,
    });
    let server = Server::new(
        core,
        entities,
        resolver.clone(),
        Arc::new(PassClient::new("http://127.0.0.1:9")),
        33068179,
    )
    .with_economy(services);
    // asset 20000 is assigned to the first distinct seeded definition, so wire
    // resourceId 20000 reverse-maps to a real Core card_id (a valid synthetic mint).
    (server, probe, resolver, 20000)
}

/// Drive the whole writer+reader cluster through the real dispatch. Panics on any
/// mismatch. Runs on a plain OS thread (no ambient runtime).
fn economy_sequence(base: &str, dir: &std::path::Path) -> SeqResult {
    wait_ready(base);
    // Core is seeded (start_core_seeded): a fifa17 profile with 100k coins + one
    // owned instance per definition. No /auth/local — the profile already exists.
    let (server, client, resolver, sample_resource) = build_econ_server(base, dir);
    let start = client.balance().unwrap();
    assert!(start >= 5000, "seeded dev balance present ({start})");

    // 1) Store BUY (pack 1 = Bronze, price 400, 5 cards): debit + mint + reveal.
    let buy = server
        .try_handle_economy(
            "PUT",
            "/ut/game/fifa17/store/transaction",
            &[],
            br#"{"packId":1}"#,
            None,
        )
        .expect("store BUY routed");
    assert_eq!(buy.status, 200, "BUY 200");
    let bv: Value = serde_json::from_slice(&buy.body).unwrap();
    let items = bv["createPackResponse"]["itemList"]
        .as_array()
        .expect("itemList")
        .clone();
    assert_eq!(items.len(), 5, "pack 1 awards 5 cards");
    assert_eq!(bv["createPackResponse"]["numberItems"], 5);
    assert!(
        items[0]["id"].as_i64().unwrap() >= 100_000_000,
        "minted wire id above the FIFA floor"
    );
    assert_eq!(
        client.balance().unwrap(),
        start - 400,
        "BUY debited exactly 400"
    );

    // 2) credits reads the SAME Core authority.
    let cr = server
        .try_handle_economy("GET", "/ut/game/fifa17/user/credits", &[], b"", None)
        .unwrap();
    let crv: Value = serde_json::from_slice(&cr.body).unwrap();
    assert_eq!(
        crv["currencies"][0]["funds"],
        start - 400,
        "credits == Core balance"
    );

    // 3) Quick-sell one minted card: reverse-resolve wire -> Core id, credit.
    let sell_wire = items[0]["id"].as_i64().unwrap();
    let before = client.balance().unwrap();
    let qs = server
        .try_handle_economy(
            "DELETE",
            &format!("/ut/game/fifa17/item/{sell_wire}"),
            &[],
            b"",
            None,
        )
        .expect("quick-sell routed");
    assert_eq!(qs.status, 200);
    let qv: Value = serde_json::from_slice(&qs.body).unwrap();
    assert_eq!(qv["items"].as_array().unwrap().len(), 1, "sold exactly one");
    let after_sell = client.balance().unwrap();
    assert!(
        after_sell > before,
        "quick-sell credited ({before}->{after_sell})"
    );
    assert_eq!(
        qv["totalCredits"].as_i64().unwrap(),
        after_sell,
        "totalCredits == absolute Core balance"
    );

    // 4) Match END reward through dispatch (WIN = +400).
    let before_match = client.balance().unwrap();
    let mm = server
        .try_handle_economy(
            "POST",
            "/ut/delete/game/fifa17/match",
            &[],
            br#"{"endReason":"WIN"}"#,
            None,
        )
        .expect("match routed");
    assert_eq!(mm.status, 200);
    assert_eq!(
        client.balance().unwrap(),
        before_match + 400,
        "WIN credited +400 via Core"
    );

    // 5) MARKET buy-now (async handlers via the bridge): list -> query -> buy ->
    //    query -> second buy fails, exactly one debit + one sale.
    let list = server
        .try_handle_economy(
            "POST",
            "/ut/game/fifa17/auctionhouse",
            &[],
            format!(
                r#"{{"itemData":{{"id":777,"resourceId":{sample_resource}}},"buyNowPrice":1000,"startingBid":500}}"#
            )
            .as_bytes(),
            None,
        )
        .expect("market list routed");
    let lv: Value = serde_json::from_slice(&list.body).unwrap();
    let trade_id = lv["id"].as_i64().expect("trade id");
    let trade_path = format!("/ut/game/fifa17/trade/{trade_id}");

    let q1 = server
        .try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None)
        .unwrap();
    let q1v: Value = serde_json::from_slice(&q1.body).unwrap();
    assert_eq!(
        q1v["auctionInfo"].as_array().unwrap().len(),
        1,
        "listing active before buy"
    );

    let before_buy = client.balance().unwrap();
    let buy1 = server
        .try_handle_economy("POST", &trade_path, &[], b"{}", None)
        .expect("market buy routed");
    assert_eq!(buy1.status, 200);
    assert_eq!(
        client.balance().unwrap(),
        before_buy - 1000,
        "buy debited exactly the buy-now price"
    );

    let q2 = server
        .try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None)
        .unwrap();
    let q2v: Value = serde_json::from_slice(&q2.body).unwrap();
    assert_eq!(
        q2v["auctionInfo"].as_array().unwrap().len(),
        0,
        "listing sold: no longer active"
    );

    let after_buy = client.balance().unwrap();
    let buy2 = server
        .try_handle_economy("POST", &trade_path, &[], b"{}", None)
        .unwrap();
    let buy2v: Value = serde_json::from_slice(&buy2.body).unwrap();
    assert_eq!(
        buy2v["auctionInfo"].as_array().unwrap().len(),
        0,
        "second buy sees a closed auction"
    );
    assert_eq!(
        client.balance().unwrap(),
        after_buy,
        "second buy does NOT debit again"
    );

    // 6) MARKET cancel: a cancelled listing cannot be bought.
    let clist = server
        .try_handle_economy(
            "POST",
            "/ut/game/fifa17/auctionhouse",
            &[],
            format!(
                r#"{{"itemData":{{"id":888,"resourceId":{sample_resource}}},"buyNowPrice":1000,"startingBid":500}}"#
            )
            .as_bytes(),
            None,
        )
        .unwrap();
    let cancel_id = serde_json::from_slice::<Value>(&clist.body).unwrap()["id"]
        .as_i64()
        .unwrap();
    server
        .try_handle_economy(
            "DELETE",
            &format!("/ut/delete/game/fifa17/trade/{cancel_id}"),
            &[],
            b"",
            None,
        )
        .expect("market cancel routed");
    let bal_before_cancel_buy = client.balance().unwrap();
    let cancel_buy = server
        .try_handle_economy(
            "POST",
            &format!("/ut/game/fifa17/trade/{cancel_id}"),
            &[],
            b"{}",
            None,
        )
        .unwrap();
    assert_eq!(
        serde_json::from_slice::<Value>(&cancel_buy.body).unwrap()["auctionInfo"]
            .as_array()
            .unwrap()
            .len(),
        0,
        "cancelled listing is not buyable"
    );
    assert_eq!(
        client.balance().unwrap(),
        bal_before_cancel_buy,
        "buying a cancelled listing does not debit"
    );

    // 7) Move a still-owned minted card to the trade pile (durable pile metadata).
    let move_wire = items[1]["id"].as_i64().unwrap();
    let mv = server
        .try_handle_economy(
            "PUT",
            "/ut/game/fifa17/item",
            &[],
            format!(r#"{{"itemData":[{{"id":{move_wire},"pile":"trade"}}]}}"#).as_bytes(),
            None,
        )
        .expect("move routed");
    let mvv: Value = serde_json::from_slice(&mv.body).unwrap();
    assert_eq!(mvv["itemData"][0]["success"], true, "move recorded");
    let moved_core_id = resolver
        .owned_id_for_wire(move_wire)
        .expect("moved item reverses to a Core id");

    SeqResult {
        final_balance: client.balance().unwrap(),
        sold_listing: trade_id.to_string(),
        moved_core_id,
    }
}

/// After a FULL restart from the SAME on-disk state — Core rebooted from its
/// SQLite file, and the durable market/pile stores reopened from their files —
/// coins, the sold listing, and the pile move all persist. The synthetic market
/// buy now mints a REAL Core `card_id` (resourceId reverse-mapped), so Core's
/// content preflight passes on reboot.
fn verify_economy_restart(base: &str, dir: &std::path::Path, seq: &SeqResult) {
    wait_ready(base);
    let client = HttpCoreClient::new(base, "fifa17");
    assert_eq!(
        client.balance().unwrap(),
        seq.final_balance,
        "coins persisted across Core restart"
    );
    let bridge = AsyncBridge::new().unwrap();
    let market_path = dir.join("market.db").to_string_lossy().into_owned();
    let market = bridge
        .block_on(async move { MarketStore::open(&market_path).await })
        .unwrap();
    let sold = seq.sold_listing.clone();
    let m2 = market.clone();
    let listing = bridge
        .block_on(async move { m2.get_listing(&sold).await })
        .expect("sold listing survives reopen");
    assert_eq!(listing.state, "sold", "sold stays sold after reopen");
    let pile_path = dir.join("pile.db").to_string_lossy().into_owned();
    let piles = bridge
        .block_on(async move { PileStore::open(&pile_path).await })
        .unwrap();
    let moved = seq.moved_core_id.clone();
    let p2 = piles.clone();
    let pile = bridge
        .block_on(async move { p2.get(&moved).await })
        .unwrap();
    assert_eq!(
        pile.as_deref(),
        Some("trade"),
        "pile move persisted after reopen"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn economy_full_sequence_through_dispatch_and_restart() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-dispatch-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let db_url = format!("sqlite://{}/econ.db", dir.display());

    // Core #1: run the full write sequence on a plain OS thread (direct bridge
    // path). The join runs in spawn_blocking so Core's runtime keeps serving.
    let (h1, base1) = start_core_seeded(&db_url, true).await;
    let (b1, d1) = (base1.clone(), dir.clone());
    let seq = tokio::task::spawn_blocking(move || {
        let t = std::thread::spawn(move || economy_sequence(&b1, &d1));
        t.join().expect("sequence thread")
    })
    .await
    .expect("write sequence");
    h1.abort();

    // Core #2: same on-disk Core DB + same market/pile files — prove full restart
    // persistence (coins + sold listing + pile). Core content preflight passes
    // because the synthetic mint used a real reverse-mapped card_id.
    let (h2, base2) = start_core_seeded(&db_url, false).await;
    let (b2, d2) = (base2.clone(), dir.clone());
    tokio::task::spawn_blocking(move || {
        let t = std::thread::spawn(move || verify_economy_restart(&b2, &d2, &seq));
        t.join().expect("restart thread")
    })
    .await
    .expect("restart phase");
    h2.abort();

    std::fs::remove_dir_all(&dir).ok();
}