//! 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,
    pass_url: &str,
) -> (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(pass_url)),
        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, "http://127.0.0.1:9");
    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"
    );

    // 6b) Owned-pack (70) open + GET /purchased reveal (Part 8 + 0B). Seed an
    // unopened pack-70 entitlement via the Core economy API (cost 0), open it,
    // and prove the reveal screen (GET /purchased) shows the freshly opened items
    // and is idempotent on repeat (presentation state, not a second grant).
    let http = reqwest::blocking::Client::new();
    post(
        &http,
        base,
        "/economy/purchase-entitlement",
        json!({ "cost": 0, "definition_id": "70" }),
    );
    let reveal_before = {
        let r = server
            .try_handle_economy("GET", "/ut/game/fifa17/purchased", &[], b"", None)
            .expect("reveal routed");
        serde_json::from_slice::<Value>(&r.body).unwrap()["itemData"]
            .as_array()
            .map(|a| a.len())
            .unwrap_or(0)
    };
    let open = server
        .try_handle_economy(
            "POST",
            "/ut/game/fifa17/purchased",
            &[],
            br#"{"packId":70}"#,
            None,
        )
        .expect("pack open routed");
    assert_eq!(open.status, 200, "pack-70 open 200");
    let ov: Value = serde_json::from_slice(&open.body).unwrap();
    assert_eq!(ov["packId"], 70, "open echoes the pack id");
    let reveal = server
        .try_handle_economy("GET", "/ut/game/fifa17/purchased", &[], b"", None)
        .expect("reveal routed");
    let reveal_items = serde_json::from_slice::<Value>(&reveal.body).unwrap()["itemData"]
        .as_array()
        .expect("reveal itemData")
        .len();
    assert!(
        reveal_items > reveal_before,
        "GET /purchased reveals the opened items ({reveal_before} -> {reveal_items})"
    );
    // Idempotent: a repeat GET does not re-grant or clear (same reveal).
    let reveal2 = server
        .try_handle_economy("GET", "/ut/game/fifa17/purchased", &[], b"", None)
        .unwrap();
    let reveal2_items = serde_json::from_slice::<Value>(&reveal2.body).unwrap()["itemData"]
        .as_array()
        .unwrap()
        .len();
    assert_eq!(
        reveal2_items, reveal_items,
        "repeated GET /purchased is idempotent"
    );
    // 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();
}

// ─────────────── Production constructor (Server::from_config) E2E ────────────
//
// Proves the PRODUCTION wiring path, not just manual `with_economy`: build the
// Server from a real (disposable) HostConfig — which itself opens the durable
// market/pile stores + the runtime bridge + the content pool — and drive the
// economy through it, then restart from the same config/files and prove
// persistence.

/// Write a catalog FILE mapping every seeded Core definition to a deterministic
/// asset id (asset 20000 = the first, so wire resourceId 20000 reverse-maps).
fn write_catalog_file(base: &str, path: &std::path::Path) {
    let owned = HttpCoreClient::new(base, "fifa17")
        .all_owned()
        .expect("core collection");
    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;
    }
    std::fs::write(
        path,
        format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}"),
    )
    .unwrap();
}

fn from_config(base: &str, dir: &std::path::Path) -> openfut_utas_host::config::HostConfig {
    let catalog = dir.join("catalog.json");
    write_catalog_file(base, &catalog);
    openfut_utas_host::config::HostConfig {
        listen_addr: "127.0.0.1:0".into(),
        python_upstream: "http://127.0.0.1:9".into(), // unused by economy dispatch
        core_url: base.to_string(),
        tables_dir: "../fifa17-recon/data/tables".into(),
        catalog_path: catalog.to_string_lossy().into_owned(),
        identity_store_path: dir.join("identity.json").to_string_lossy().into_owned(),
        persona_id: 33_068_179,
        market_db_path: dir.join("market.db").to_string_lossy().into_owned(),
        pile_db_path: dir.join("pile.db").to_string_lossy().into_owned(),
        clientdata_path: dir.join("clientdata.json").to_string_lossy().into_owned(),
    }
}

fn exercise_from_config(base: &str, dir: &std::path::Path) -> i64 {
    wait_ready(base);
    let cfg = from_config(base, dir);
    // PRODUCTION constructor — economy services are attached by from_config, NOT
    // injected by the test.
    let server = Server::from_config(&cfg).expect("from_config builds economy services");
    let client = HttpCoreClient::new(base, "fifa17");
    let start = client.balance().unwrap();

    // credits + purchasegroup readers.
    let cr = server
        .try_handle_economy("GET", "/ut/game/fifa17/user/credits", &[], b"", None)
        .expect("credits routed");
    assert_eq!(
        serde_json::from_slice::<Value>(&cr.body).unwrap()["currencies"][0]["funds"],
        start
    );
    let pg = server
        .try_handle_economy("GET", "/ut/game/fifa17/store/purchasegroup", &[], b"", None)
        .expect("purchasegroup routed");
    assert_eq!(pg.status, 200);

    // Store BUY (writer) through the production-built server.
    let buy = server
        .try_handle_economy(
            "PUT",
            "/ut/game/fifa17/store/transaction",
            &[],
            br#"{"packId":1}"#,
            None,
        )
        .expect("buy routed");
    assert_eq!(buy.status, 200);
    assert_eq!(
        client.balance().unwrap(),
        start - 400,
        "BUY debited via from_config server"
    );

    // Market list -> query -> buy through the production-built server.
    let list = server
        .try_handle_economy(
            "POST",
            "/ut/game/fifa17/auctionhouse",
            &[],
            br#"{"itemData":{"id":555,"resourceId":20000},"buyNowPrice":1000,"startingBid":500}"#,
            None,
        )
        .expect("list routed");
    let trade_id = serde_json::from_slice::<Value>(&list.body).unwrap()["id"]
        .as_i64()
        .unwrap();
    let q = server
        .try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None)
        .unwrap();
    assert_eq!(
        serde_json::from_slice::<Value>(&q.body).unwrap()["auctionInfo"]
            .as_array()
            .unwrap()
            .len(),
        1
    );
    let before_buy = client.balance().unwrap();
    server
        .try_handle_economy(
            "POST",
            &format!("/ut/game/fifa17/trade/{trade_id}"),
            &[],
            b"{}",
            None,
        )
        .expect("market buy routed");
    assert_eq!(
        client.balance().unwrap(),
        before_buy - 1000,
        "market buy debited"
    );
    client.balance().unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn from_config_constructs_and_serves_economy() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-fromcfg-{}-{}",
        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());

    let (h1, base1) = start_core_seeded(&db_url, true).await;
    let (b1, d1) = (base1.clone(), dir.clone());
    let final_balance = tokio::task::spawn_blocking(move || {
        let t = std::thread::spawn(move || exercise_from_config(&b1, &d1));
        t.join().expect("from_config thread")
    })
    .await
    .expect("from_config exercise");
    h1.abort();

    // Restart Core + rebuild the Server from the SAME config/files: balance
    // persists, and a fresh from_config server serves it.
    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 || {
            wait_ready(&b2);
            let cfg = from_config(&b2, &d2);
            let server = Server::from_config(&cfg).expect("from_config rebuild");
            let cr = server
                .try_handle_economy("GET", "/ut/game/fifa17/user/credits", &[], b"", None)
                .unwrap();
            assert_eq!(
                serde_json::from_slice::<Value>(&cr.body).unwrap()["currencies"][0]["funds"],
                final_balance,
                "balance persists across a from_config restart"
            );
        });
        t.join().expect("restart thread")
    })
    .await
    .expect("from_config restart");
    h2.abort();

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

// ─────────────── Post-barrier authority proofs (NEVER BOTH / no fallback / ────
//                 stale reader), through the REAL handle_with_ip dispatch ──────

/// A mock Python UTAS upstream that COUNTS every request it receives and always
/// answers with a distinctive marker body carrying coins=111. If an economy
/// route ever reaches Python, this counter moves and/or the marker leaks.
struct MockPython {
    calls: Arc<std::sync::atomic::AtomicUsize>,
    url: String,
}

fn start_mock_python() -> MockPython {
    use std::io::{Read, Write};
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let c2 = calls.clone();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(mut s) = stream else { continue };
            c2.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            let mut buf = [0u8; 8192];
            let _ = s.read(&mut buf);
            let body = br#"{"__python__":true,"credits":111,"currencies":[{"name":"coins","funds":111,"finalFunds":111}],"userInfo":{"currencies":[{"name":"coins","funds":111,"finalFunds":111}]},"purchase":[]}"#;
            let head = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                body.len()
            );
            let _ = s.write_all(head.as_bytes());
            let _ = s.write_all(body);
        }
    });
    MockPython {
        calls,
        url: format!("http://{addr}"),
    }
}

/// The pure economy routes (userMassInfo excluded — it is the documented hybrid
/// that proxies the Python envelope but Rust-overlays the economy fields).
fn pure_economy_routes() -> Vec<(&'static str, String, Vec<u8>)> {
    vec![
        ("GET", "/ut/game/fifa17/user/credits".into(), b"".to_vec()),
        (
            "GET",
            "/ut/game/fifa17/store/purchasegroup".into(),
            b"".to_vec(),
        ),
        (
            "PUT",
            "/ut/game/fifa17/store/transaction".into(),
            br#"{"packId":1}"#.to_vec(),
        ),
        (
            "POST",
            "/ut/game/fifa17/purchased".into(),
            br#"{"packId":70}"#.to_vec(),
        ),
        ("GET", "/ut/game/fifa17/purchased".into(), b"".to_vec()),
        (
            "DELETE",
            "/ut/game/fifa17/item/100000001".into(),
            b"".to_vec(),
        ),
        (
            "POST",
            "/ut/delete/game/fifa17/item".into(),
            br#"{"itemData":[{"id":100000001}]}"#.to_vec(),
        ),
        (
            "PUT",
            "/ut/game/fifa17/item".into(),
            br#"{"itemData":[{"id":100000001,"pile":"trade"}]}"#.to_vec(),
        ),
        (
            "POST",
            "/ut/delete/game/fifa17/match".into(),
            br#"{"endReason":"WIN"}"#.to_vec(),
        ),
        (
            "POST",
            "/ut/game/fifa17/auctionhouse".into(),
            br#"{"itemData":{"id":555,"resourceId":20000},"buyNowPrice":1000,"startingBid":500}"#
                .to_vec(),
        ),
        ("GET", "/ut/game/fifa17/tradePile".into(), b"".to_vec()),
        (
            "POST",
            "/ut/game/fifa17/trade/900000001".into(),
            b"{}".to_vec(),
        ),
        (
            "DELETE",
            "/ut/delete/game/fifa17/trade/900000001".into(),
            b"".to_vec(),
        ),
        // ── Retail v2 Store family (the S2 live-failure shapes). These MUST be
        //    Rust-owned exactly like their v1 forms. ──
        (
            "PUT",
            "/ut/v2/game/fifa17/store/transaction/0".into(),
            br#"{"packId":1}"#.to_vec(),
        ),
        (
            "GET",
            "/ut/v2/game/fifa17/store/purchasegroup".into(),
            b"".to_vec(),
        ),
        (
            "POST",
            "/ut/v2/game/fifa17/purchased".into(),
            br#"{"packId":70}"#.to_vec(),
        ),
        ("GET", "/ut/v2/game/fifa17/purchased".into(), b"".to_vec()),
        // ── Round-2 retail shapes: the confirmed BUY uses POST /purchased/items,
        //    reveal GET /purchased/items; hub tile polls lowercase tradepile + /counts. ──
        (
            "POST",
            "/ut/game/fifa17/purchased/items".into(),
            br#"{"packId":1}"#.to_vec(),
        ),
        (
            "GET",
            "/ut/game/fifa17/purchased/items".into(),
            b"".to_vec(),
        ),
        ("GET", "/ut/game/fifa17/tradepile".into(), b"".to_vec()),
        (
            "GET",
            "/ut/game/fifa17/tradePile/counts".into(),
            b"".to_vec(),
        ),
    ]
}

fn barrier_checks(base: &str, dir: &std::path::Path, mock: &MockPython) {
    wait_ready(base);
    let (server, client, _r, _sample) = build_econ_server(base, dir, &mock.url);
    let core_coins = client.balance().unwrap();
    assert_ne!(
        core_coins, 111,
        "Core must diverge from the Python marker (111)"
    );

    // ── STALE READER: readers show Core values, never the Python 111 ──
    let cr = server.handle_with_ip("GET", "/ut/game/fifa17/user/credits", &[], b"", None);
    let crv: Value = serde_json::from_slice(&cr.body).unwrap();
    assert!(
        crv.get("__python__").is_none(),
        "credits is Rust, not the Python body"
    );
    assert_eq!(
        crv["currencies"][0]["funds"], core_coins,
        "credits coins = Core, not 111"
    );
    // userMassInfo is the hybrid: Python envelope proxied, economy Rust-overlaid.
    let mi = server.handle_with_ip("GET", "/ut/game/fifa17/userMassInfo", &[], b"", None);
    let miv: Value = serde_json::from_slice(&mi.body).unwrap();
    assert_eq!(
        miv["userInfo"]["currencies"][0]["funds"], core_coins,
        "userMassInfo coins overlaid to Core (stale Python 111 not visible)"
    );

    // ── PART 7 REPRO: the exact S2 live-failure shape (retail v2 Store BUY,
    //    `PUT /ut/v2/game/fifa17/store/transaction/0`) is now Rust-owned — it
    //    debits Core and returns a `createPackResponse`, NOT the Python
    //    `{"state":"TRANSACTIONCANCEL"}` no-op the rejected candidate produced. ──
    let before_buy = client.balance().unwrap();
    let calls_before_buy = mock.calls.load(std::sync::atomic::Ordering::SeqCst);
    let buy = server.handle_with_ip(
        "PUT",
        "/ut/v2/game/fifa17/store/transaction/0",
        &[],
        br#"{"packId":1}"#,
        None,
    );
    assert_eq!(buy.status, 200, "v2 Store BUY handled by Rust (200)");
    let buyv: Value = serde_json::from_slice(&buy.body).unwrap();
    assert!(
        buyv.get("createPackResponse").is_some(),
        "v2 Store BUY returns a Rust createPackResponse, not the Python no-op: {buyv}"
    );
    assert_ne!(
        buyv.get("state").and_then(|s| s.as_str()),
        Some("TRANSACTIONCANCEL"),
        "v2 Store BUY must NOT be the Python TRANSACTIONCANCEL fallback"
    );
    assert_eq!(
        mock.calls.load(std::sync::atomic::Ordering::SeqCst),
        calls_before_buy,
        "v2 Store BUY never reached the Python proxy"
    );
    let after_buy = client.balance().unwrap();
    assert!(
        after_buy < before_buy,
        "v2 Store BUY debited Core coins ({before_buy} -> {after_buy})"
    );

    // ── NEVER BOTH (Core up): pure economy routes reach Rust, never Python ──
    let before = mock.calls.load(std::sync::atomic::Ordering::SeqCst);
    for (m, p, b) in pure_economy_routes() {
        let r = server.handle_with_ip(m, &p, &[], &b, None);
        assert!(
            !r.body.windows(10).any(|w| w == b"__python__"),
            "{m} {p} must be Rust-owned (no Python marker in body)"
        );
    }
    assert_eq!(
        mock.calls.load(std::sync::atomic::Ordering::SeqCst),
        before,
        "NEVER BOTH: no pure economy route reached the Python proxy"
    );

    // ── NO FALLBACK: a server pointed at a DEAD Core still fails closed and
    //    never proxies to Python. Built without probing Core (empty catalog +
    //    empty pool), so no live Core is needed to construct it. ──
    let dead_dir = dir.join("dead");
    std::fs::create_dir_all(&dead_dir).unwrap();
    let dead = build_dead_core_server(&dead_dir, &mock.url);
    let before_down = mock.calls.load(std::sync::atomic::Ordering::SeqCst);
    let credits_down = dead.handle_with_ip("GET", "/ut/game/fifa17/user/credits", &[], b"", None);
    assert_eq!(
        credits_down.status, 503,
        "credits fails closed against a dead Core"
    );
    let match_down = dead.handle_with_ip(
        "POST",
        "/ut/delete/game/fifa17/match",
        &[],
        br#"{"endReason":"WIN"}"#,
        None,
    );
    assert_eq!(
        match_down.status, 503,
        "match fails closed against a dead Core"
    );
    for (m, p, b) in pure_economy_routes() {
        let _ = dead.handle_with_ip(m, &p, &[], &b, None);
    }
    assert_eq!(
        mock.calls.load(std::sync::atomic::Ordering::SeqCst),
        before_down,
        "NO FALLBACK: economy routes never proxy to Python even against a dead Core"
    );
}

/// A `Server` whose Core (read + economy) points at a definitely-dead loopback
/// port, wired WITHOUT probing Core: an empty catalog + empty content pool. Used
/// to prove economy routes fail closed (503) and never fall back to Python.
fn build_dead_core_server(dir: &std::path::Path, pass_url: &str) -> Server {
    // A closed loopback port: bind then drop, so connects are refused.
    let dead_addr = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        l.local_addr().unwrap()
    };
    let dead_url = format!("http://{dead_addr}");
    let catalog =
        Fifa17CardCatalog::from_json_str(r#"{"schema_version":1,"game":"fifa17","cards":{}}"#)
            .unwrap();
    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(dead_url.clone(), "fifa17"));
    let bridge = Arc::new(AsyncBridge::new().unwrap());
    let mp = dir.join("market.db").to_string_lossy().into_owned();
    let market = Arc::new(
        bridge
            .block_on(async move { MarketStore::open(&mp).await })
            .unwrap(),
    );
    let pp = dir.join("pile.db").to_string_lossy().into_owned();
    let piles = Arc::new(
        bridge
            .block_on(async move { PileStore::open(&pp).await })
            .unwrap(),
    );
    let econ: Arc<dyn CoreEconomy> = Arc::new(HttpCoreClient::new(dead_url, "fifa17"));
    let services = Arc::new(EconomyServices {
        econ,
        market,
        piles,
        bridge,
        pool: Arc::new(Vec::new()),
    });
    Server::new(
        core,
        entities,
        resolver,
        Arc::new(PassClient::new(pass_url)),
        33_068_179,
    )
    .with_economy(services)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn barrier_never_both_no_fallback_and_stale_reader() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-barrier-{}-{}",
        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());
    let (h, base) = start_core_seeded(&db_url, true).await;
    let mock = start_mock_python();
    let (b, d) = (base.clone(), dir.clone());
    tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || barrier_checks(&b, &d, &mock))
            .join()
            .expect("barrier checks thread")
    })
    .await
    .expect("barrier phase");
    h.abort();
    std::fs::remove_dir_all(&dir).ok();
}

// ─────────────── Retail v2 Store E2E + v1/v2 route equivalence ───────────────
//
// Proves the S2 fix end-to-end through the REAL dispatch: the Store flow driven
// over the retail `/ut/v2/game/<sku>/…` paths is Rust-owned (Python proxy count
// 0), mutates Core, and behaves IDENTICALLY to the v1 paths for the same op.

fn v2_store_flow(base: &str, dir: &std::path::Path) {
    let mock = start_mock_python();
    let (server, client, _r, _s) = build_econ_server(base, dir, &mock.url);
    let calls0 = mock.calls.load(std::sync::atomic::Ordering::SeqCst);

    // GET purchasegroup via v2 → Rust catalogue (non-empty).
    let pg = server.handle_with_ip(
        "GET",
        "/ut/v2/game/fifa17/store/purchasegroup",
        &[],
        b"",
        None,
    );
    assert_eq!(pg.status, 200, "v2 purchasegroup handled by Rust");
    let pgv: Value = serde_json::from_slice(&pg.body).unwrap();
    assert!(
        pgv.get("purchase")
            .and_then(|p| p.as_array())
            .is_some_and(|a| !a.is_empty()),
        "v2 purchasegroup returns a Rust catalogue: {pgv}"
    );

    // Same pack (id 1) via v1 then v2 → IDENTICAL debit + item count (Part 6).
    let bal0 = client.balance().unwrap();
    let v1 = server.handle_with_ip(
        "PUT",
        "/ut/game/fifa17/store/transaction",
        &[],
        br#"{"packId":1}"#,
        None,
    );
    assert_eq!(v1.status, 200);
    let bal1 = client.balance().unwrap();
    let v1v: Value = serde_json::from_slice(&v1.body).unwrap();
    let v1_items = v1v["createPackResponse"]["itemList"]
        .as_array()
        .map_or(0, |a| a.len());
    let v1_debit = bal0 - bal1;

    let v2 = server.handle_with_ip(
        "PUT",
        "/ut/v2/game/fifa17/store/transaction/0",
        &[],
        br#"{"packId":1}"#,
        None,
    );
    assert_eq!(v2.status, 200);
    let bal2 = client.balance().unwrap();
    let v2v: Value = serde_json::from_slice(&v2.body).unwrap();
    let v2_items = v2v["createPackResponse"]["itemList"]
        .as_array()
        .map_or(0, |a| a.len());
    let v2_debit = bal1 - bal2;

    assert!(v1_items > 0, "v1 BUY minted items");
    assert_eq!(v1_items, v2_items, "v1/v2 BUY yield identical item counts");
    assert_eq!(
        v1_debit, v2_debit,
        "v1/v2 BUY debit identically ({v1_debit} vs {v2_debit})"
    );

    // GET purchased via v2 → Rust reveal shape; the just-bought items are in the pile.
    let reveal = server.handle_with_ip("GET", "/ut/v2/game/fifa17/purchased", &[], b"", None);
    assert_eq!(reveal.status, 200, "v2 GET /purchased handled by Rust");
    let rv: Value = serde_json::from_slice(&reveal.body).unwrap();
    assert!(
        rv.get("itemData").and_then(|d| d.as_array()).is_some(),
        "v2 reveal is a Rust itemData array: {rv}"
    );

    // NEVER any Python proxy for the whole v2 Store flow.
    assert_eq!(
        mock.calls.load(std::sync::atomic::Ordering::SeqCst),
        calls0,
        "v2 Store flow never reached the Python proxy"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn retail_v2_store_flow_matches_v1_through_dispatch() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-v2-{}-{}",
        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());
    let (h, base) = start_core_seeded(&db_url, true).await;
    let (b, d) = (base.clone(), dir.clone());
    tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || v2_store_flow(&b, &d))
            .join()
            .expect("v2 store flow thread")
    })
    .await
    .expect("v2 store phase");
    h.abort();
    std::fs::remove_dir_all(&dir).ok();
}

// ─────────────── Retail /purchased/items BUY sequence (round-2 S2 regression) ─
//
// The rejected candidate 47ced22 sent the confirmed retail Store BUY
// (POST /ut/game/fifa17/purchased/items) to Python and left Core coins unchanged.
// This replays the exact live sequence through real dispatch and asserts the BUY
// debits Core, the reveal shows the minted items, and Python proxy count is 0.

fn retail_purchased_items_flow(base: &str, dir: &std::path::Path) {
    let mock = start_mock_python();
    let (server, client, _r, _s) = build_econ_server(base, dir, &mock.url);
    let calls0 = mock.calls.load(std::sync::atomic::Ordering::SeqCst);

    // Store screen catalogue (v1 /all) — Rust.
    let pg = server.handle_with_ip(
        "GET",
        "/ut/game/fifa17/store/purchasegroup/all",
        &[],
        b"",
        None,
    );
    assert_eq!(pg.status, 200, "purchasegroup/all Rust-owned");

    // THE CONFIRMED RETAIL BUY: POST /ut/game/fifa17/purchased/items must debit Core.
    let bal0 = client.balance().unwrap();
    let buy = server.handle_with_ip(
        "POST",
        "/ut/game/fifa17/purchased/items",
        &[],
        br#"{"packId":1}"#,
        None,
    );
    assert_eq!(buy.status, 200, "purchased/items BUY handled by Rust (200)");
    assert!(
        !buy.body.windows(10).any(|w| w == b"__python__"),
        "purchased/items BUY is Rust-owned (no Python marker)"
    );
    let bal1 = client.balance().unwrap();
    assert!(
        bal1 < bal0,
        "purchased/items BUY debited Core ({bal0} -> {bal1}) — the round-2 S2 was NO debit"
    );

    // Reveal poll: GET /ut/game/fifa17/purchased/items — Rust, shows the minted items.
    let reveal = server.handle_with_ip("GET", "/ut/game/fifa17/purchased/items", &[], b"", None);
    assert_eq!(reveal.status, 200, "purchased/items reveal Rust-owned");
    let rv: Value = serde_json::from_slice(&reveal.body).unwrap();
    let revealed = rv["itemData"].as_array().map_or(0, |a| a.len());
    assert!(revealed > 0, "reveal shows the freshly-minted items: {rv}");

    // Repeat reveal is idempotent (no re-grant, no extra debit).
    let bal2 = client.balance().unwrap();
    let _ = server.handle_with_ip("GET", "/ut/game/fifa17/purchased/items", &[], b"", None);
    assert_eq!(
        client.balance().unwrap(),
        bal2,
        "repeat reveal does not mutate coins"
    );

    // NEVER any Python proxy across the whole /purchased/items sequence.
    assert_eq!(
        mock.calls.load(std::sync::atomic::Ordering::SeqCst),
        calls0,
        "retail /purchased/items sequence never reached the Python proxy"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn retail_purchased_items_buy_debits_core_through_dispatch() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-pi-{}-{}",
        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());
    let (h, base) = start_core_seeded(&db_url, true).await;
    let (b, d) = (base.clone(), dir.clone());
    tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || retail_purchased_items_flow(&b, &d))
            .join()
            .expect("purchased/items flow thread")
    })
    .await
    .expect("purchased/items phase");
    h.abort();
    std::fs::remove_dir_all(&dir).ok();
}