//! Real host↔Core economy FAILURE-INJECTION proofs.
//!
//! Each case drives the REAL host dispatch (`Server::try_handle_economy`)
//! against a live in-process Core, but with a chosen dependency armed to fail at
//! a chosen point, proving the economy cluster is fail-closed and leaves NO
//! silent Core-vs-host state divergence:
//!
//! * A wrapping [`CoreEconomy`] double (`FaultEconomy`) forwards to the real
//!   blocking `HttpCoreClient` but can be armed to fail one op with a
//!   `CoreError` — the transport-failure seam.
//! * A wrapping [`ExternalIdentityStore`] double (`FaultIdentity`) forwards to
//!   the real `JsonIdentityStore` but can be armed to fail wire-id allocation.
//! * The durable listing/pile stores are the REAL SQLite stores; their narrow,
//!   inert-by-default `StoreFault` switch (host `market_store`/`pile_store`) is
//!   armed to fail a market reserve / complete-sale / pile write. That seam is
//!   the only way to fault those concrete stores, which are wired straight into
//!   the handlers with no trait to substitute.
//!
//! Execution contract mirrors `economy_integration.rs`: all economy work runs on
//! plain `std::thread`s off any Tokio runtime, so the blocking Core client and
//! the async bridge behave exactly as in the thread-per-connection server.
//!
//! Safety: temp dir + `127.0.0.1:0` only. Never touches production/`.105`.

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::club_response::ItemIdentityResolver;
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_identity::{ExternalIdentityStore, IdError, 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, CoreAccess, CoreEconomy, CoreError, EconomyEntitlement, EconomyGrantItem,
    EconomyPurchase, EconomySale, EconomySaleReceipt, EconomyServices, Fifa17IdentityResolver,
    HttpCoreClient, PassClient, Server, WireResponse,
};
use parking_lot::Mutex;
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

const GAME: &str = "fifa17";
const OWNED_KIND: &str = "owned-item";
const OWNED_FLOOR: i64 = 100_000_001;

// ───────────────────────────── Fault doubles ────────────────────────────────

/// A `CoreEconomy` that forwards every op to the real `HttpCoreClient` but can
/// be armed to fail a named op's next N invocations with a `CoreError`.
struct FaultEconomy {
    inner: HttpCoreClient,
    fail: Mutex<HashMap<&'static str, u32>>,
}

impl FaultEconomy {
    fn new(base: &str) -> Self {
        FaultEconomy {
            inner: HttpCoreClient::new(base, GAME),
            fail: Mutex::new(HashMap::new()),
        }
    }
    fn arm(&self, op: &'static str, times: u32) {
        self.fail.lock().insert(op, times);
    }
    fn trip(&self, op: &'static str) -> bool {
        let mut g = self.fail.lock();
        match g.get_mut(op) {
            Some(n) if *n > 0 => {
                *n -= 1;
                true
            }
            _ => false,
        }
    }
    fn injected() -> CoreError {
        CoreError::Http("injected core fault".into())
    }
}

impl CoreEconomy for FaultEconomy {
    fn balance(&self) -> Result<i64, CoreError> {
        if self.trip("balance") {
            return Err(Self::injected());
        }
        self.inner.balance()
    }
    fn entitlements(&self) -> Result<Vec<EconomyEntitlement>, CoreError> {
        if self.trip("entitlements") {
            return Err(Self::injected());
        }
        self.inner.entitlements()
    }
    fn purchase_entitlement(
        &self,
        cost: i64,
        definition_id: &str,
    ) -> Result<EconomyPurchase, CoreError> {
        if self.trip("purchase_entitlement") {
            return Err(Self::injected());
        }
        self.inner.purchase_entitlement(cost, definition_id)
    }
    fn redeem_entitlement(
        &self,
        entitlement_id: &str,
        items: &[EconomyGrantItem],
    ) -> Result<String, CoreError> {
        if self.trip("redeem_entitlement") {
            return Err(Self::injected());
        }
        self.inner.redeem_entitlement(entitlement_id, items)
    }
    fn sell_item(&self, item_id: &str, price: i64) -> Result<i64, CoreError> {
        if self.trip("sell_item") {
            return Err(Self::injected());
        }
        self.inner.sell_item(item_id, price)
    }
    fn grant_reward(&self, amount: i64) -> Result<i64, CoreError> {
        if self.trip("grant_reward") {
            return Err(Self::injected());
        }
        self.inner.grant_reward(amount)
    }
    fn purchase_item(&self, cost: i64, item_id: &str, card_id: &str) -> Result<i64, CoreError> {
        if self.trip("purchase_item") {
            return Err(Self::injected());
        }
        self.inner.purchase_item(cost, item_id, card_id)
    }
    fn purchase_items(&self, cost: i64, items: &[EconomyGrantItem]) -> Result<i64, CoreError> {
        if self.trip("purchase_items") {
            return Err(Self::injected());
        }
        self.inner.purchase_items(cost, items)
    }
    fn settle_sale(&self, sale: &EconomySale<'_>) -> Result<EconomySaleReceipt, CoreError> {
        if self.trip("settle_sale") {
            return Err(Self::injected());
        }
        self.inner.settle_sale(sale)
    }
}

/// An `ExternalIdentityStore` that forwards to a real `JsonIdentityStore` but can
/// be armed to fail wire-id allocation (`resolve_or_allocate`).
struct FaultIdentity {
    inner: Arc<JsonIdentityStore>,
    fail_alloc: Mutex<u32>,
}

impl FaultIdentity {
    fn new(inner: Arc<JsonIdentityStore>) -> Self {
        FaultIdentity {
            inner,
            fail_alloc: Mutex::new(0),
        }
    }
    fn arm_alloc(&self, times: u32) {
        *self.fail_alloc.lock() = times;
    }
}

impl ExternalIdentityStore for FaultIdentity {
    fn resolve_or_allocate(
        &self,
        game: &str,
        kind: &str,
        core_id: &str,
        base_floor: i64,
    ) -> Result<i64, IdError> {
        {
            let mut n = self.fail_alloc.lock();
            if *n > 0 {
                *n -= 1;
                return Err(IdError::Corrupt("injected identity alloc fault".into()));
            }
        }
        self.inner
            .resolve_or_allocate(game, kind, core_id, base_floor)
    }
    fn external_for(&self, game: &str, kind: &str, core_id: &str) -> Result<Option<i64>, IdError> {
        self.inner.external_for(game, kind, core_id)
    }
    fn core_for(
        &self,
        game: &str,
        kind: &str,
        external_id: i64,
    ) -> Result<Option<String>, IdError> {
        self.inner.core_for(game, kind, external_id)
    }
}

// ───────────────────────────── Core boot (seeded) ───────────────────────────

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();
    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}");
}

// ───────────────────────────── Harness ──────────────────────────────────────

struct FailHarness {
    server: Server,
    client: HttpCoreClient,
    resolver: Arc<Fifa17IdentityResolver>,
    ident: Arc<JsonIdentityStore>,
    fault_ident: Arc<FaultIdentity>,
    econ: Arc<FaultEconomy>,
    market: Arc<MarketStore>,
    piles: Arc<PileStore>,
    bridge: Arc<AsyncBridge>,
    core: Arc<dyn CoreAccess>,
    entities: Arc<Fifa17Entities>,
}

fn catalog_from_core(core: &dyn CoreAccess) -> Fifa17CardCatalog {
    let owned = core.all_owned().expect("core collection");
    assert!(!owned.is_empty(), "seed must grant a starter collection");
    let mut entries = String::new();
    let mut seen = 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}}}}}");
    Fifa17CardCatalog::from_json_str(&doc).expect("catalog")
}

fn build_fail_harness(base: &str, dir: &std::path::Path) -> FailHarness {
    let core: Arc<dyn CoreAccess> = Arc::new(HttpCoreClient::new(base, GAME));
    let catalog = catalog_from_core(core.as_ref());
    let ident = Arc::new(
        JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).expect("identity"),
    );
    let fault_ident = Arc::new(FaultIdentity::new(ident.clone()));
    let dyn_store: Arc<dyn ExternalIdentityStore> = fault_ident.clone();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, dyn_store));
    let entities = Arc::new(Fifa17Entities::from_maps(
        HashMap::new(),
        HashMap::new(),
        HashMap::new(),
    ));
    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::new(FaultEconomy::new(base));
    let econ_dyn: Arc<dyn CoreEconomy> = econ.clone();
    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 {
        // Production default: the sold experiment is OFF.
        sold_experiment: openfut_utas_host::sold_experiment::SoldExperiment::OFF,
        econ: econ_dyn,
        market: market.clone(),
        piles: piles.clone(),
        bridge: bridge.clone(),
        pool,
    });
    let server = Server::new(
        core.clone(),
        entities.clone(),
        resolver.clone(),
        Arc::new(PassClient::new("http://127.0.0.1:9")),
        33068179,
    )
    .with_economy(services);
    FailHarness {
        server,
        client: HttpCoreClient::new(base, GAME),
        resolver,
        ident,
        fault_ident,
        econ,
        market,
        piles,
        bridge,
        core,
        entities,
    }
}

impl FailHarness {
    /// A sibling `Server` whose content pool is EMPTY (fail-closed generator),
    /// sharing the same Core/identity/stores. Used to prove the empty-pool
    /// generator path never consumes an entitlement.
    fn empty_pool_server(&self) -> Server {
        let services = Arc::new(EconomyServices {
            // Production default: the sold experiment is OFF.
            sold_experiment: openfut_utas_host::sold_experiment::SoldExperiment::OFF,
            econ: {
                let e: Arc<dyn CoreEconomy> = self.econ.clone();
                e
            },
            market: self.market.clone(),
            piles: self.piles.clone(),
            bridge: self.bridge.clone(),
            pool: Arc::new(Vec::new()),
        });
        Server::new(
            self.core.clone(),
            self.entities.clone(),
            self.resolver.clone(),
            Arc::new(PassClient::new("http://127.0.0.1:9")),
            33068179,
        )
        .with_economy(services)
    }

    fn dispatch(&self, method: &str, path: &str, body: &[u8]) -> WireResponse {
        self.server
            .try_handle_economy(method, path, &[], body, None)
            .expect("economy route matched")
    }

    fn listing_state(&self, listing_id: &str) -> Option<String> {
        let m = self.market.clone();
        let id = listing_id.to_string();
        self.bridge
            .block_on(async move { m.get_listing(&id).await })
            .ok()
            .map(|l| l.state)
    }

    fn owns(&self, core_id: &str) -> bool {
        self.client
            .all_owned()
            .unwrap()
            .iter()
            .any(|it| it.owned_card_id == core_id)
    }

    fn owned_ids(&self) -> HashSet<String> {
        self.client
            .all_owned()
            .unwrap()
            .into_iter()
            .map(|it| it.owned_card_id)
            .collect()
    }

    fn unopened_70(&self) -> usize {
        self.client
            .entitlements()
            .unwrap()
            .iter()
            .filter(|e| e.definition_id == "70")
            .count()
    }

    /// Mint one owned card via a real Store BUY (pack 1). Returns (wire, core).
    fn mint_one(&self) -> (i64, String) {
        set_balance(&self.client, 50_000);
        let resp = self.dispatch(
            "PUT",
            "/ut/game/fifa17/store/transaction",
            br#"{"packId":1}"#,
        );
        assert_eq!(resp.status, 200, "mint buy 200");
        let items = bj(&resp)["createPackResponse"]["itemList"]
            .as_array()
            .expect("itemList")
            .clone();
        let wire = items[0]["id"].as_i64().expect("wire id");
        let core = self.resolver.owned_id_for_wire(wire).expect("reverse");
        (wire, core)
    }
}

fn bj(r: &WireResponse) -> Value {
    serde_json::from_slice(&r.body).unwrap_or(Value::Null)
}

fn set_balance(client: &HttpCoreClient, target: i64) {
    let cur = client.balance().unwrap();
    if cur > target {
        client
            .purchase_entitlement(cur - target, "econ-test-drain")
            .expect("drain");
    } else if cur < target {
        client.grant_reward(target - cur).expect("top up");
    }
    assert_eq!(client.balance().unwrap(), target, "balance set");
}

// ───────────────────────────── Cases ────────────────────────────────────────

/// BUY Core failure → no debit, no grant.
fn case_buy_core_failure(h: &FailHarness) -> String {
    set_balance(&h.client, 5_000);
    let owned_before = h.owned_ids();
    h.econ.arm("purchase_items", 1);
    let resp = h.dispatch(
        "PUT",
        "/ut/game/fifa17/store/transaction",
        br#"{"packId":1}"#,
    );
    assert_eq!(resp.status, 503, "BUY Core failure is fail-closed 503");
    assert_eq!(
        h.client.balance().unwrap(),
        5_000,
        "no debit on BUY failure"
    );
    assert_eq!(h.owned_ids(), owned_before, "no grant on BUY failure");
    "BUY core-fail: 503, no debit, no grant".into()
}

/// OWNED-PACK OPEN Core-redeem failure → entitlement + inventory coherent
/// (all-or-nothing): entitlement NOT consumed, no items granted.
fn case_open_redeem_failure(h: &FailHarness) -> String {
    h.client.purchase_entitlement(0, "70").expect("seed 70");
    let owned_before = h.owned_ids();
    assert_eq!(h.unopened_70(), 1, "one unopened 70 before");
    h.econ.arm("redeem_entitlement", 1);
    let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#);
    assert_eq!(resp.status, 503, "redeem failure is fail-closed 503");
    assert_eq!(h.unopened_70(), 1, "entitlement survives a failed redeem");
    assert_eq!(
        h.owned_ids(),
        owned_before,
        "no items granted on failed redeem"
    );
    // Clean up the surviving entitlement so later cases start fresh.
    let ok = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#);
    assert_eq!(
        ok.status, 200,
        "the same entitlement opens once Core recovers"
    );
    "OPEN redeem-fail: 503, ent survives, no grant; recovers on retry".into()
}

/// OWNED-PACK OPEN generator failure (empty content pool) → nothing minted,
/// entitlement NOT consumed.
fn case_open_generator_failure(h: &FailHarness) -> String {
    h.client.purchase_entitlement(0, "70").expect("seed 70");
    let owned_before = h.owned_ids();
    let empty = h.empty_pool_server();
    let resp = empty
        .try_handle_economy(
            "POST",
            "/ut/game/fifa17/purchased",
            &[],
            br#"{"packId":70}"#,
            None,
        )
        .expect("routed");
    assert_eq!(resp.status, 503, "empty-pool generator is fail-closed 503");
    assert_eq!(
        h.unopened_70(),
        1,
        "entitlement not consumed when generator draws nothing"
    );
    assert_eq!(
        h.owned_ids(),
        owned_before,
        "no items minted by empty generator"
    );
    // Consume it via the real (populated) pool to reset.
    assert_eq!(
        h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#)
            .status,
        200
    );
    "OPEN generator-fail: 503, ent not consumed, no mint".into()
}

/// OWNED-PACK OPEN pile-persist failure → Core is coherent (entitlement consumed
/// once, +11 owned); the reveal simply omits the un-piled items (presentation
/// only). Wire-id allocation stays monotonic.
fn case_open_pile_persist_failure(h: &FailHarness) -> String {
    h.client.purchase_entitlement(0, "70").expect("seed 70");
    let owned_before = h.owned_ids();
    h.piles.fault().arm("set", 32); // fail every purchased-pile write for this open
    let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#);
    h.piles.fault().arm("set", 0); // clear any residual armed count for later cases
    assert_eq!(
        resp.status, 200,
        "pile-write failure is non-fatal to the open"
    );
    assert_eq!(h.unopened_70(), 0, "entitlement consumed exactly once");
    let new: Vec<String> = h.owned_ids().difference(&owned_before).cloned().collect();
    assert_eq!(
        new.len(),
        11,
        "inventory granted once (11 cards) despite pile failure"
    );
    // Reveal is presentation-only: the un-piled new items are simply not shown.
    let reveal = h.dispatch("GET", "/ut/game/fifa17/purchased", b"");
    let shown_new = bj(&reveal)["itemData"]
        .as_array()
        .map(|a| {
            a.iter()
                .filter(|x| {
                    x["id"]
                        .as_i64()
                        .and_then(|w| h.resolver.owned_id_for_wire(w))
                        .map(|c| new.contains(&c))
                        .unwrap_or(false)
                })
                .count()
        })
        .unwrap_or(0);
    assert_eq!(
        shown_new, 0,
        "un-piled items are omitted from the reveal (coherent)"
    );
    let next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR);
    assert!(next > OWNED_FLOOR, "watermark advanced monotonically");
    "OPEN pile-fail: 200, ent consumed once, +11 owned, reveal omits un-piled (coherent)".into()
}

/// OWNED-PACK OPEN identity-alloc failure → Core coherent (consumed + 11), the
/// reveal omits the not-yet-resolvable items; on recovery every item resolves to
/// a UNIQUE wire id above the floor (a burned id is never reused — monotonic).
fn case_open_identity_failure(h: &FailHarness) -> String {
    h.client.purchase_entitlement(0, "70").expect("seed 70");
    let owned_before = h.owned_ids();
    let before_next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR);
    h.fault_ident.arm_alloc(64); // fail every wire-id allocation for this open
    let resp = h.dispatch("POST", "/ut/game/fifa17/purchased", br#"{"packId":70}"#);
    assert_eq!(
        resp.status, 200,
        "identity failure is non-fatal to the redeem"
    );
    assert_eq!(h.unopened_70(), 0, "entitlement consumed exactly once");
    let new: Vec<String> = h.owned_ids().difference(&owned_before).cloned().collect();
    assert_eq!(new.len(), 11, "inventory granted once (11 cards)");
    // Recovery: disarm identity, then resolve each new item DIRECTLY (no reliance
    // on the pile/reveal path). A previously-faulted allocation left nothing
    // persisted, so each now receives a fresh monotonic wire id.
    h.fault_ident.arm_alloc(0);
    let owned_now = h.client.all_owned().unwrap();
    let mut wires = HashSet::new();
    for item in owned_now
        .iter()
        .filter(|it| new.contains(&it.owned_card_id))
    {
        let id = h.resolver.resolve(item).expect("resolves after recovery");
        let w = id.item_id as i64;
        assert!(w >= OWNED_FLOOR, "wire id ≥ floor");
        assert!(wires.insert(w), "each recovered wire id is unique");
    }
    let after_next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR);
    assert!(
        after_next > before_next,
        "watermark strictly advanced: no id reused ({before_next} -> {after_next})"
    );
    assert!(
        wires.iter().all(|&w| w < after_next),
        "every issued id is below the next watermark (monotonic)"
    );
    "OPEN identity-fail: 200, ent consumed once, +11 owned; recovered ids unique + monotonic".into()
}

/// QUICK-SELL Core failure → item remains, coins unchanged.
fn case_quicksell_core_failure(h: &FailHarness) -> String {
    let (wire, core) = h.mint_one();
    let before = h.client.balance().unwrap();
    h.econ.arm("sell_item", 1);
    let resp = h.dispatch("DELETE", &format!("/ut/game/fifa17/item/{wire}"), b"");
    assert_eq!(
        resp.status, 503,
        "quick-sell Core failure is fail-closed 503"
    );
    assert!(h.owns(&core), "item still owned after failed quick-sell");
    assert_eq!(
        h.client.balance().unwrap(),
        before,
        "coins unchanged after failed quick-sell"
    );
    "QUICK-SELL core-fail: 503, item remains, coins unchanged".into()
}

/// MOVE pile-store write failure → Core ownership and the pile do not silently
/// disagree: Core still owns the item (move never touches Core), and the pile is
/// simply not updated (verdict success=false), so reveal/pile stay coherent.
fn case_move_pile_failure(h: &FailHarness) -> String {
    let (wire, core) = h.mint_one();
    h.piles.fault().arm("set", 1);
    let resp = h.dispatch(
        "PUT",
        "/ut/game/fifa17/item",
        format!(r#"{{"itemData":[{{"id":{wire},"pile":"trade"}}]}}"#).as_bytes(),
    );
    assert_eq!(
        resp.status, 200,
        "move responds 200 with a per-item verdict"
    );
    assert_eq!(
        bj(&resp)["itemData"][0]["success"],
        false,
        "pile write failure surfaces as success=false (never a fabricated move)"
    );
    assert!(
        h.owns(&core),
        "Core still owns the item (move is not an ownership op)"
    );
    let pile = h
        .bridge
        .block_on({
            let p = h.piles.clone();
            let c = core.clone();
            async move { p.get(&c).await }
        })
        .unwrap();
    assert_ne!(
        pile.as_deref(),
        Some("trade"),
        "pile was NOT updated (no silent divergence)"
    );
    "MOVE pile-fail: success=false, Core owns, pile not updated (coherent)".into()
}

/// MARKET RESERVE failure → no debit, no grant, listing stays legal (active).
fn case_market_reserve_failure(h: &FailHarness) -> String {
    // List a genuinely-owned card: the server resolves card_id + resourceId from
    // Core inventory via the wire id (you can only list what you own).
    let (item_id, _core) = h.mint_one();
    set_balance(&h.client, 50_000);
    let list = h.dispatch(
        "POST",
        "/ut/game/fifa17/auctionhouse",
        format!(r#"{{"itemData":{{"id":{item_id}}},"buyNowPrice":1000,"startingBid":500}}"#)
            .as_bytes(),
    );
    let trade_id = bj(&list)["id"].as_i64().expect("trade id");
    let before = h.client.balance().unwrap();
    h.market.fault().arm("reserve", 1);
    let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}");
    assert_eq!(resp.status, 503, "reserve infra failure is fail-closed 503");
    assert_eq!(
        h.client.balance().unwrap(),
        before,
        "no debit when reserve fails"
    );
    assert!(
        !h.owns(&format!("market-buy:{trade_id}")),
        "no card minted when reserve fails"
    );
    assert_eq!(
        h.listing_state(&trade_id.to_string()).as_deref(),
        Some("active"),
        "listing stays active (legal) after a reserve failure"
    );
    "MARKET reserve-fail: 503, no debit, no mint, listing active".into()
}

/// MARKET Core purchase_item failure AFTER reserve → reservation rolls back to
/// active, no debit, no mint.
fn case_market_purchase_failure(h: &FailHarness) -> String {
    let (item_id, _core) = h.mint_one();
    set_balance(&h.client, 50_000);
    let list = h.dispatch(
        "POST",
        "/ut/game/fifa17/auctionhouse",
        format!(r#"{{"itemData":{{"id":{item_id}}},"buyNowPrice":1000,"startingBid":500}}"#)
            .as_bytes(),
    );
    let trade_id = bj(&list)["id"].as_i64().expect("trade id");
    let before = h.client.balance().unwrap();
    h.econ.arm("purchase_item", 1);
    let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}");
    assert_eq!(
        resp.status, 503,
        "purchase failure after reserve is fail-closed 503"
    );
    assert_eq!(
        h.client.balance().unwrap(),
        before,
        "no debit when Core purchase fails"
    );
    assert!(
        !h.owns(&format!("market-buy:{trade_id}")),
        "no card minted when Core purchase fails"
    );
    assert_eq!(
        h.listing_state(&trade_id.to_string()).as_deref(),
        Some("active"),
        "reservation rolled back to active (buyable again)"
    );
    "MARKET purchase-fail: 503, reservation rolled back to active, no debit, no mint".into()
}

/// THE critical case. MARKET COMPLETE-SALE failure AFTER a SUCCESSFUL Core
/// purchase must NOT leave the listing buyable with the buyer already debited +
/// minted. Expected (current design): the listing is stuck in `reserved` (NOT
/// active), so no further `active -> reserved` CAS can succeed → not buyable,
/// with exactly one debit + one mint. Returns ("SAFE"|"E3", detail).
fn case_market_complete_sale_failure(h: &FailHarness) -> (String, String) {
    let (item_id, _core) = h.mint_one();
    set_balance(&h.client, 50_000);
    let list = h.dispatch(
        "POST",
        "/ut/game/fifa17/auctionhouse",
        format!(r#"{{"itemData":{{"id":{item_id}}},"buyNowPrice":1000,"startingBid":500}}"#)
            .as_bytes(),
    );
    let trade_id = bj(&list)["id"].as_i64().expect("trade id");
    let mint_id = format!("market-buy:{trade_id}");
    let before = h.client.balance().unwrap();

    // Fail the complete_sale that runs AFTER Core has debited + minted.
    h.market.fault().arm("complete_sale", 1);
    let resp = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}");
    assert_eq!(
        resp.status, 200,
        "purchase committed in Core (complete_sale is post-commit)"
    );

    // Core-side commit really happened: exactly one debit + one mint.
    assert_eq!(
        h.client.balance().unwrap(),
        before - 1000,
        "exactly one debit"
    );
    let mints = h
        .client
        .all_owned()
        .unwrap()
        .iter()
        .filter(|it| it.owned_card_id == mint_id)
        .count();
    assert_eq!(mints, 1, "exactly one mint");

    let state = h.listing_state(&trade_id.to_string());
    // Second buy attempt AFTER the committed purchase.
    let after = h.client.balance().unwrap();
    let retry = h.dispatch("POST", &format!("/ut/game/fifa17/trade/{trade_id}"), b"{}");
    let buyable_again = bj(&retry)["auctionInfo"]
        .as_array()
        .map(|a| !a.is_empty())
        .unwrap_or(false);
    let debited_again = h.client.balance().unwrap() != after;
    let minted_again = h
        .client
        .all_owned()
        .unwrap()
        .iter()
        .filter(|it| it.owned_card_id == mint_id)
        .count()
        > 1;

    if buyable_again || debited_again || minted_again {
        return (
            "E3".into(),
            format!(
                "ATOMICITY DEFECT: after a committed purchase, listing state={state:?}, \
                 buyable_again={buyable_again} debited_again={debited_again} minted_again={minted_again}"
            ),
        );
    }
    assert_eq!(
        state.as_deref(),
        Some("reserved"),
        "committed-but-uncompleted listing is left in `reserved`, not `active`"
    );
    (
        "SAFE".into(),
        "complete_sale failure leaves listing=reserved (not buyable); exactly one \
         debit + one mint; retry returned empty, no further debit/mint"
            .to_string(),
    )
}

fn run_all_failures(base: &str, dir: &std::path::Path) -> (String, String, bool) {
    wait_ready(base);
    let h = build_fail_harness(base, dir);
    let mut lines = vec![
        case_buy_core_failure(&h),
        case_open_redeem_failure(&h),
        case_open_generator_failure(&h),
        case_open_pile_persist_failure(&h),
        case_open_identity_failure(&h),
        case_quicksell_core_failure(&h),
        case_move_pile_failure(&h),
        case_market_reserve_failure(&h),
        case_market_purchase_failure(&h),
    ];
    let (verdict, detail) = case_market_complete_sale_failure(&h);
    lines.push(format!("MARKET complete-sale-fail [{verdict}]: {detail}"));
    let e3 = verdict == "E3";
    (lines.join("\n"), verdict, e3)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn economy_failure_injection() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-fail-{}-{}",
        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 (summary, verdict, e3) = tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || run_all_failures(&b1, &d1))
            .join()
            .expect("failure thread")
    })
    .await
    .expect("failures");
    h1.abort();
    std::fs::remove_dir_all(&dir).ok();

    eprintln!("economy_failure summary (complete-sale verdict={verdict}, e3={e3}):\n{summary}");
    assert!(!e3, "unrecoverable partial-state (E3) detected: {summary}");
}