//! FIFA 17 transfer-market + item-move handlers, Core-backed and durable.
//!
//! These implement the market and FutMoveCard routes against two authorities:
//! Core owns coins + item ownership (via [`CoreEconomy`]); the host owns the
//! durable *listing* lifecycle ([`MarketStore`]) and *pile* location
//! ([`PileStore`]). They are fail-closed like the rest of the economy cluster:
//! a Core failure yields a controlled response and NEVER a Python fallback that
//! would reintroduce a second writer.
//!
//! ## Synthetic-seller model
//!
//! A buy-now debits the buyer and MINTS the won card into their club
//! (`CoreEconomy::purchase_item`) — there is no real counterparty and no seller
//! credit, matching the single-player oracle. The buy is race-safe: the listing
//! is reserved with an atomic compare-and-swap BEFORE any coin movement, so two
//! concurrent buyers resolve to exactly one debit and one sold listing; the
//! loser sees a closed (empty) auction. On any Core failure the reservation is
//! rolled back so the listing becomes buyable again (no coins lost, no phantom
//! sale).
//!
//! ## Piles are not ownership
//!
//! `handle_move_items` records only the pile keyed by the Core owned-instance id
//! (reverse-resolved from the FIFA wire id). Core stays the sole ownership
//! authority — the move never mints, transfers, or duplicates an inventory row.

use serde_json::{json, Value};

use openfut_adapter_fifa17::fut::squad::SquadWireResolver;

use crate::market_store::{Listing, MarketError, MarketStore};
use crate::pile_store::PileStore;
use crate::{CoreEconomy, CoreError, WireResponse};

/// FIFA trade-id numbering base (mirrors the oracle's `_TRADE_ID_BASE`).
const TRADE_ID_BASE: i64 = 900_000_000;

fn json_body(status: u16, body: &Value) -> WireResponse {
    let bytes = serde_json::to_vec(body).unwrap_or_else(|_| b"{}".to_vec());
    WireResponse {
        status,
        headers: vec![("Content-Type".to_string(), "application/json".to_string())],
        body: bytes,
    }
}

fn ok_json(body: &Value) -> WireResponse {
    json_body(200, body)
}

fn parse_body(body: &[u8]) -> Value {
    serde_json::from_slice(body).unwrap_or_else(|_| json!({}))
}

/// Extract the numeric trade id that follows `/trade/` in a path, as a string
/// (the listing id space is numeric-string).
fn trade_id_from_path(path: &str) -> Option<String> {
    let tail = path.split("/trade/").nth(1)?;
    let digits: String = tail.chars().take_while(|c| c.is_ascii_digit()).collect();
    if digits.is_empty() {
        None
    } else {
        Some(digits)
    }
}

/// Shape one listing into the FIFA auction record (0x18013e410 fields), sourced
/// from durable listing state rather than a hardcoded sample pool.
fn auction_record(l: &Listing) -> Value {
    let trade_id: i64 = l.listing_id.parse().unwrap_or(0);
    // resourceId is the FIFA wire identity the client listed (never the Core
    // card id). 0 means "no art", a valid int — never a fabricated FIFA asset.
    let resource = l.wire_resource_id.or(l.wire_item_id).unwrap_or(0);
    let item_id = l.wire_item_id.unwrap_or(trade_id);
    let (trade_state, item_state, bid_state, current_bid) = match l.state.as_str() {
        "active" => ("active", "forSale", "none", 0),
        _ => ("closed", "free", "highest", l.buy_now_price),
    };
    json!({
        "tradeId": trade_id,
        "itemData": {
            "id": item_id,
            "resourceId": resource,
            "itemState": item_state,
            "untradeable": false,
        },
        "tradeState": trade_state,
        "buyNowPrice": l.buy_now_price,
        "startingBid": l.start_price,
        "currentBid": current_bid,
        "bidState": bid_state,
        "expires": 3600,
        "sellerName": l.owner.clone().unwrap_or_else(|| "EASFC".to_string()),
        "sellerEstablished": 1,
        "watched": false,
        "coinsProcessed": 0,
    })
}

/// Run a BLOCKING closure — the blocking Core client — on a fresh OS thread that
/// has NO Tokio runtime entered, then block until it finishes. The market
/// handlers are `async` and driven on the shared runtime by the host bridge, but
/// `CoreEconomy` is a `reqwest::blocking` client: calling it while a runtime is
/// entered panics (`reqwest::blocking::wait::enter`). Hopping to a plain thread
/// keeps the Core call off the runtime, so blocking is legal. Cheap relative to
/// the network round-trip it guards.
fn off_runtime<T, F>(f: F) -> T
where
    F: FnOnce() -> T + Send,
    T: Send,
{
    std::thread::scope(|s| s.spawn(f).join().expect("off-runtime Core call panicked"))
}

/// Current Core balance as the `credits` field, or 0 if Core is unreachable
/// (used only to decorate an already-decided response; the transactional path
/// never trusts a fabricated balance).
fn credits_or_zero(econ: &dyn CoreEconomy) -> i64 {
    off_runtime(|| econ.balance()).unwrap_or(0)
}

/// Maps a FIFA wire `resourceId` to the authoritative Core `card_id` a synthetic
/// buy mints. Backed by the FIFA17 catalog reverse index; unknown → `None`
/// (fail closed, never fabricated). Core stays unaware of FIFA resource ids.
pub trait MarketCardResolver: Send + Sync {
    fn card_id_for_resource(&self, resource_id: i64) -> Option<String>;
}

/// `/auctionhouse` — search (GET), list-for-sale (POST FutISStart), relist (PUT).
///
/// * GET returns the durable active auctions plus the FutGetAuctionCount ints,
///   in the oracle's `_market_body` shape.
/// * POST lists a club item: resolve its wire `resourceId` to the authoritative
///   Core `card_id`, persist both, and return `{"id": tradeId}`. An unmappable
///   resource fails closed (persists nothing).
/// * PUT (relist-all) is an ack `{}`.
pub async fn handle_market_list(
    method: &str,
    body: &[u8],
    econ: &dyn CoreEconomy,
    store: &MarketStore,
    mapper: &dyn MarketCardResolver,
) -> WireResponse {
    match method {
        "POST" => {
            let b = parse_body(body);
            let item_data = b.get("itemData");
            let wire_item_id = item_data
                .and_then(|d| d.get("id"))
                .and_then(Value::as_i64)
                .or_else(|| b.get("itemId").and_then(Value::as_i64));
            let start = b.get("startingBid").and_then(Value::as_i64).unwrap_or(150);
            let buy_now = b.get("buyNowPrice").and_then(Value::as_i64).unwrap_or(0);
            let Some(item_id) = wire_item_id else {
                // No item to list: mirror the oracle's fresh-id ack, persist nothing.
                return ok_json(&json!({ "id": TRADE_ID_BASE }));
            };
            // Resolve the FIFA wire resourceId to the authoritative Core card id
            // the synthetic buy will MINT. Fail closed on an unmappable resource:
            // persist nothing (a non-existent listing cannot be bought), so a bad
            // resource never becomes a mint Core's content preflight would reject.
            let Some(resource_id) = item_data
                .and_then(|d| d.get("resourceId"))
                .and_then(Value::as_i64)
            else {
                return ok_json(&json!({ "id": TRADE_ID_BASE }));
            };
            let Some(core_card_id) = mapper.card_id_for_resource(resource_id) else {
                return ok_json(&json!({ "id": TRADE_ID_BASE }));
            };
            // Trade-id space is offset from the wire item id, so each owned item
            // maps to a unique, stable auction id (no modular wraparound).
            let trade_id = TRADE_ID_BASE + item_id;
            let listing_id = trade_id.to_string();
            let seller = b.get("sellerName").and_then(Value::as_str);
            match store
                .create_listing(
                    &listing_id,
                    &core_card_id,
                    None,
                    Some(item_id),
                    Some(resource_id),
                    start,
                    buy_now,
                    seller,
                )
                .await
            {
                Ok(_) | Err(MarketError::Conflict) => ok_json(&json!({ "id": trade_id })),
                Err(_) => json_body(503, &json!({ "error": "market_store" })),
            }
        }
        "PUT" => ok_json(&json!({})),
        _ => {
            // GET: browse the durable active auctions.
            let listings = match store.query_listings("active").await {
                Ok(l) => l,
                Err(_) => return json_body(503, &json!({ "error": "market_store" })),
            };
            let auctions: Vec<Value> = listings.iter().map(auction_record).collect();
            ok_json(&json!({
                "auctionInfo": auctions,
                "credits": credits_or_zero(econ),
                "total": auctions.len(),
                "duplicateItemIdList": [],
                "count": 0,
                "maxAuctionsAllowed": 100,
                "offered": 0,
                "selling": auctions.len(),
                "sold": 0,
            }))
        }
    }
}

/// Query listings in a given `state` (e.g. the user's own sale pile is the
/// `active` set). Returns the oracle's tradePile shape.
pub async fn handle_market_query(
    state: &str,
    econ: &dyn CoreEconomy,
    store: &MarketStore,
) -> WireResponse {
    let listings = match store.query_listings(state).await {
        Ok(l) => l,
        Err(_) => return json_body(503, &json!({ "error": "market_store" })),
    };
    let auctions: Vec<Value> = listings.iter().map(auction_record).collect();
    ok_json(&json!({
        "auctionInfo": auctions,
        "credits": credits_or_zero(econ),
        "total": auctions.len(),
    }))
}

/// `DELETE /ut/delete/game/<sku>/trade/<id>` — remove a listing from the sale
/// pile. Cancels the `active` listing once; the oracle always acks `{}`, so a
/// missing/already-closed listing is not surfaced as an error to the client
/// (the sale pile simply no longer shows it).
pub async fn handle_market_cancel(
    path: &str,
    owner: Option<&str>,
    store: &MarketStore,
) -> WireResponse {
    if let Some(id) = trade_id_from_path(path) {
        // Idempotent from the client's view: NotFound / already-closed still acks.
        let _ = store.cancel_listing(&id, owner).await;
    }
    ok_json(&json!({}))
}

/// `/trade/<id>` — view (GET) or buy-now / bid (POST/PUT). Buy-now is the
/// synthetic-seller path: reserve (CAS) → Core `purchase_item` mint+debit →
/// complete the sale; any Core failure rolls the reservation back.
pub async fn handle_market_buy(
    method: &str,
    path: &str,
    body: &[u8],
    econ: &dyn CoreEconomy,
    store: &MarketStore,
) -> WireResponse {
    let Some(id) = trade_id_from_path(path) else {
        return ok_json(&json!({ "auctionInfo": [], "credits": credits_or_zero(econ) }));
    };

    if method != "POST" && method != "PUT" {
        // GET: view one auction.
        let rec = match store.get_listing(&id).await {
            Ok(l) => vec![auction_record(&l)],
            Err(_) => vec![],
        };
        return ok_json(&json!({ "auctionInfo": rec, "credits": credits_or_zero(econ) }));
    }

    let listing = match store.get_listing(&id).await {
        Ok(l) => l,
        // Unknown/closed auction: empty body (client treats as gone), never an error.
        Err(_) => return ok_json(&json!({ "auctionInfo": [], "credits": credits_or_zero(econ) })),
    };

    let b = parse_body(body);
    let bid = b
        .get("bid")
        .and_then(Value::as_i64)
        .unwrap_or(listing.buy_now_price);

    // A simple bid below buy-now: we are the sole bidder — echo the raised bid,
    // no coin movement, no reservation.
    if bid < listing.buy_now_price {
        let mut rec = auction_record(&listing);
        rec["currentBid"] = json!(bid);
        rec["bidState"] = json!("highest");
        return ok_json(&json!({ "auctionInfo": [rec], "credits": credits_or_zero(econ) }));
    }

    // BUY NOW. Reserve first (atomic CAS): only one concurrent buyer wins.
    match store.reserve_listing(&id).await {
        Ok(true) => {}
        // Lost the race / already reserved / sold / cancelled: closed auction.
        Ok(false) | Err(MarketError::NotFound) => {
            return ok_json(&json!({ "auctionInfo": [], "credits": credits_or_zero(econ) }))
        }
        Err(_) => return json_body(503, &json!({ "error": "market_store" })),
    }

    let price = listing.buy_now_price;

    // Pre-check funds so an affordability failure is a clean 461, distinct from a
    // Core transport failure (503). Core's purchase is still the atomic authority.
    let balance = match off_runtime(|| econ.balance()) {
        Ok(b) => b,
        Err(_) => {
            let _ = store.rollback_reservation(&id).await;
            return json_body(503, &json!({ "error": "core_unreachable" }));
        }
    };
    if balance < price {
        let _ = store.rollback_reservation(&id).await;
        return json_body(
            461,
            &json!({ "reason": "insufficient_coins", "credits": balance }),
        );
    }

    // Mint the won card into the buyer's club. A listing sells at most once (the
    // CAS guarantees it), so a deterministic minted id is safe and idempotent.
    let minted_item_id = format!("market-buy:{id}");
    match off_runtime(|| econ.purchase_item(price, &minted_item_id, &listing.card_id)) {
        Ok(new_balance) => {
            // Core has taken the coins and minted the item; finalise the listing.
            // If completing fails, Core is still authoritative — report success.
            if let Err(e) = store.complete_sale(&id).await {
                eprintln!("utas-host WARN market complete_sale({id}) after mint failed: {e}");
            }
            let mut rec = auction_record(&listing);
            rec["tradeState"] = json!("closed");
            rec["bidState"] = json!("highest");
            rec["currentBid"] = json!(price);
            rec["itemData"]["itemState"] = json!("free");
            ok_json(&json!({ "auctionInfo": [rec], "credits": new_balance }))
        }
        // Insufficient funds surfaced by Core (concurrent debit) -> 461.
        Err(CoreError::Status(400)) => {
            let _ = store.rollback_reservation(&id).await;
            json_body(
                461,
                &json!({ "reason": "insufficient_coins", "credits": balance }),
            )
        }
        // Any other Core failure: fail closed, restore the listing.
        Err(_) => {
            let _ = store.rollback_reservation(&id).await;
            json_body(503, &json!({ "error": "core_unreachable" }))
        }
    }
}

/// `PUT /ut/game/<sku>/item` — FutMoveCard. Move each requested owned item to its
/// target pile via [`PileStore`], reverse-resolving the FIFA wire id to the Core
/// owned-instance id. Core remains the ownership authority — this records ONLY
/// the pile, never an ownership row. Returns the per-item verdict ack
/// (`{"itemData":[{id,pile,success}]}`); an unresolved id is `success:false`,
/// never a fabricated move.
pub async fn handle_move_items(
    body: &[u8],
    resolver: &(dyn SquadWireResolver + Sync),
    pile_store: &PileStore,
) -> WireResponse {
    let b = parse_body(body);
    let Some(items) = b.get("itemData").and_then(Value::as_array) else {
        return ok_json(&json!({ "itemData": [] }));
    };

    let mut verdicts = Vec::with_capacity(items.len());
    for item in items {
        let Some(wire) = item.get("id").and_then(Value::as_i64) else {
            continue;
        };
        let pile = item
            .get("pile")
            .and_then(Value::as_str)
            .unwrap_or("club")
            .to_string();
        let success = match resolver.owned_id_for_wire(wire) {
            Some(core_id) => pile_store.set(&core_id, &pile).await.is_ok(),
            None => false,
        };
        verdicts.push(json!({ "id": wire, "pile": pile, "success": success }));
    }
    ok_json(&json!({ "itemData": verdicts }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{EconomyEntitlement, EconomyGrantItem, EconomyPurchase};
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicI64, AtomicU64, AtomicUsize, Ordering};
    use std::sync::Arc;

    // ---- temp DB helpers ---------------------------------------------------

    struct TempDb(String);
    impl TempDb {
        fn new(tag: &str) -> Self {
            static N: AtomicU64 = AtomicU64::new(0);
            let n = N.fetch_add(1, Ordering::SeqCst);
            let path = std::env::temp_dir()
                .join(format!("ofut-market-h-{tag}-{}-{n}.db", std::process::id()));
            TempDb(path.to_string_lossy().into_owned())
        }
        fn path(&self) -> &str {
            &self.0
        }
    }
    impl Drop for TempDb {
        fn drop(&mut self) {
            for suffix in ["", "-wal", "-shm"] {
                let _ = std::fs::remove_file(format!("{}{suffix}", self.0));
            }
        }
    }

    // ---- CoreEconomy double that debits coins and counts purchase calls ----

    struct CountingEconomy {
        balance: AtomicI64,
        purchase_calls: AtomicUsize,
        fail: bool,
    }
    impl CountingEconomy {
        fn with_balance(balance: i64) -> Self {
            CountingEconomy {
                balance: AtomicI64::new(balance),
                purchase_calls: AtomicUsize::new(0),
                fail: false,
            }
        }
        fn failing() -> Self {
            CountingEconomy {
                balance: AtomicI64::new(0),
                purchase_calls: AtomicUsize::new(0),
                fail: true,
            }
        }
    }
    impl CoreEconomy for CountingEconomy {
        fn balance(&self) -> Result<i64, CoreError> {
            if self.fail {
                Err(CoreError::Status(500))
            } else {
                Ok(self.balance.load(Ordering::SeqCst))
            }
        }
        fn entitlements(&self) -> Result<Vec<EconomyEntitlement>, CoreError> {
            Ok(vec![])
        }
        fn purchase_entitlement(
            &self,
            _cost: i64,
            _definition_id: &str,
        ) -> Result<EconomyPurchase, CoreError> {
            Err(CoreError::Status(500))
        }
        fn redeem_entitlement(
            &self,
            _entitlement_id: &str,
            _items: &[EconomyGrantItem],
        ) -> Result<String, CoreError> {
            Err(CoreError::Status(500))
        }
        fn sell_item(&self, _item_id: &str, _price: i64) -> Result<i64, CoreError> {
            Err(CoreError::Status(500))
        }
        fn grant_reward(&self, _amount: i64) -> Result<i64, CoreError> {
            Err(CoreError::Status(500))
        }
        fn purchase_item(
            &self,
            cost: i64,
            _item_id: &str,
            _card_id: &str,
        ) -> Result<i64, CoreError> {
            self.purchase_calls.fetch_add(1, Ordering::SeqCst);
            if self.fail {
                return Err(CoreError::Status(500));
            }
            // Atomic debit: reject (and do NOT debit) if it would go negative,
            // mirroring Core's BadRequest(400) on insufficient funds.
            let mut cur = self.balance.load(Ordering::SeqCst);
            loop {
                if cur < cost {
                    return Err(CoreError::Status(400));
                }
                match self.balance.compare_exchange(
                    cur,
                    cur - cost,
                    Ordering::SeqCst,
                    Ordering::SeqCst,
                ) {
                    Ok(_) => return Ok(cur - cost),
                    Err(actual) => cur = actual,
                }
            }
        }
        fn purchase_items(
            &self,
            _cost: i64,
            _items: &[EconomyGrantItem],
        ) -> Result<i64, CoreError> {
            Err(CoreError::Status(500))
        }
    }

    // ---- SquadWireResolver double -----------------------------------------

    struct MapResolver(HashMap<i64, String>);
    impl MapResolver {
        fn new(pairs: &[(i64, &str)]) -> Self {
            MapResolver(pairs.iter().map(|(w, c)| (*w, c.to_string())).collect())
        }
    }
    impl SquadWireResolver for MapResolver {
        fn owned_id_for_wire(&self, wire: i64) -> Option<String> {
            self.0.get(&wire).cloned()
        }
    }

    /// Permissive test resolver: maps any wire resourceId to its own string, so
    /// the existing list tests keep their prior card-id semantics. A dedicated
    /// test covers the unknown-resource fail-closed path.
    struct AllowAllResolver;
    impl MarketCardResolver for AllowAllResolver {
        fn card_id_for_resource(&self, resource_id: i64) -> Option<String> {
            Some(resource_id.to_string())
        }
    }

    /// Test resolver that maps nothing (every resourceId is unknown).
    struct DenyAllResolver;
    impl MarketCardResolver for DenyAllResolver {
        fn card_id_for_resource(&self, _resource_id: i64) -> Option<String> {
            None
        }
    }

    async fn store_at(tag: &str) -> (MarketStore, TempDb) {
        let db = TempDb::new(tag);
        let store = MarketStore::open(db.path()).await.unwrap();
        (store, db)
    }

    async fn seed_listing(store: &MarketStore, id: &str, buy_now: i64) {
        store
            .create_listing(id, "169193", None, None, Some(169193), 400, buy_now, None)
            .await
            .unwrap();
    }

    fn parse(resp: &WireResponse) -> Value {
        serde_json::from_slice(&resp.body).unwrap()
    }

    // ---- listing / auctionhouse -------------------------------------------

    #[tokio::test]
    async fn list_post_persists_and_returns_trade_id() {
        let (store, _d) = store_at("post").await;
        let econ = CountingEconomy::with_balance(10_000);
        let body = json!({ "itemData": { "id": 100004617, "resourceId": 169193 },
                           "startingBid": 300, "buyNowPrice": 2500 });
        let resp = handle_market_list(
            "POST",
            body.to_string().as_bytes(),
            &econ,
            &store,
            &AllowAllResolver,
        )
        .await;
        assert_eq!(resp.status, 200);
        let trade_id = parse(&resp)["id"].as_i64().unwrap();
        assert_eq!(trade_id, TRADE_ID_BASE + 100004617);
        // Persisted + browsable.
        let listed = store.get_listing(&trade_id.to_string()).await.unwrap();
        assert_eq!(listed.buy_now_price, 2500);
        let browse = handle_market_list("GET", b"", &econ, &store, &AllowAllResolver).await;
        let b = parse(&browse);
        assert_eq!(b["auctionInfo"].as_array().unwrap().len(), 1);
        assert_eq!(b["credits"], 10_000);
        assert_eq!(b["maxAuctionsAllowed"], 100);
    }

    #[tokio::test]
    async fn list_put_is_ack() {
        let (store, _d) = store_at("put").await;
        let econ = CountingEconomy::with_balance(0);
        let resp = handle_market_list("PUT", b"", &econ, &store, &AllowAllResolver).await;
        assert_eq!(resp.status, 200);
        assert_eq!(parse(&resp), json!({}));
    }

    #[tokio::test]
    async fn list_unknown_resource_fails_closed_no_listing() {
        // An unmappable wire resourceId must NOT create a listing (a synthetic buy
        // would otherwise mint a card id Core cannot resolve). Acks neutrally.
        let (store, _d) = store_at("deny").await;
        let econ = CountingEconomy::with_balance(10_000);
        let body = json!({ "itemData": { "id": 100004617, "resourceId": 424242 },
                           "startingBid": 300, "buyNowPrice": 2500 });
        let resp = handle_market_list(
            "POST",
            body.to_string().as_bytes(),
            &econ,
            &store,
            &DenyAllResolver,
        )
        .await;
        assert_eq!(resp.status, 200);
        assert_eq!(parse(&resp)["id"].as_i64().unwrap(), TRADE_ID_BASE);
        // Nothing persisted at the would-be trade id: not buyable.
        assert!(matches!(
            store
                .get_listing(&(TRADE_ID_BASE + 100004617).to_string())
                .await,
            Err(MarketError::NotFound)
        ));
    }

    #[tokio::test]
    async fn list_persists_core_card_and_wire_resource_across_reopen() {
        // The listing carries BOTH the authoritative Core card_id (for the mint)
        // and the FIFA wire resourceId (for the auction record), durably.
        let db = TempDb::new("reopen-map");
        let trade_id;
        {
            let store = MarketStore::open(db.path()).await.unwrap();
            let econ = CountingEconomy::with_balance(10_000);
            // A resolver that maps resourceId 20801 -> Core card_id "card_pl_042".
            struct FixedResolver;
            impl MarketCardResolver for FixedResolver {
                fn card_id_for_resource(&self, resource_id: i64) -> Option<String> {
                    (resource_id == 20801).then(|| "card_pl_042".to_string())
                }
            }
            let body = json!({ "itemData": { "id": 100004900, "resourceId": 20801 },
                               "startingBid": 300, "buyNowPrice": 2500 });
            let resp = handle_market_list(
                "POST",
                body.to_string().as_bytes(),
                &econ,
                &store,
                &FixedResolver,
            )
            .await;
            trade_id = parse(&resp)["id"].as_i64().unwrap();
        }
        // Reopen from the same file: both identities survive.
        let reopened = MarketStore::open(db.path()).await.unwrap();
        let l = reopened.get_listing(&trade_id.to_string()).await.unwrap();
        assert_eq!(l.card_id, "card_pl_042", "Core card_id minted on buy");
        assert_eq!(
            l.wire_resource_id,
            Some(20801),
            "wire resourceId for the record"
        );
    }

    #[tokio::test]
    async fn query_returns_active_pile() {
        let (store, _d) = store_at("query").await;
        seed_listing(&store, "900000005", 2500).await;
        let econ = CountingEconomy::with_balance(50);
        let resp = handle_market_query("active", &econ, &store).await;
        let b = parse(&resp);
        assert_eq!(b["auctionInfo"].as_array().unwrap().len(), 1);
        assert_eq!(b["auctionInfo"][0]["tradeId"], 900000005i64);
        assert_eq!(b["credits"], 50);
    }

    #[tokio::test]
    async fn buy_now_debits_mints_and_closes() {
        let (store, _d) = store_at("buy").await;
        seed_listing(&store, "900000010", 2500).await;
        let econ = CountingEconomy::with_balance(10_000);
        let resp = handle_market_buy(
            "POST",
            "/ut/game/fifa17/trade/900000010",
            b"{}",
            &econ,
            &store,
        )
        .await;
        assert_eq!(resp.status, 200);
        let b = parse(&resp);
        assert_eq!(b["auctionInfo"][0]["tradeState"], "closed");
        assert_eq!(b["credits"], 7500);
        assert_eq!(econ.purchase_calls.load(Ordering::SeqCst), 1);
        assert_eq!(store.get_listing("900000010").await.unwrap().state, "sold");
    }

    #[tokio::test]
    async fn buy_now_insufficient_is_461_and_no_debit() {
        let (store, _d) = store_at("poor").await;
        seed_listing(&store, "900000011", 2500).await;
        let econ = CountingEconomy::with_balance(100);
        let resp = handle_market_buy(
            "POST",
            "/ut/game/fifa17/trade/900000011",
            b"{}",
            &econ,
            &store,
        )
        .await;
        assert_eq!(resp.status, 461);
        assert_eq!(parse(&resp)["reason"], "insufficient_coins");
        // Reservation rolled back -> still buyable, no debit happened.
        assert_eq!(
            store.get_listing("900000011").await.unwrap().state,
            "active"
        );
        assert_eq!(econ.balance().unwrap(), 100);
    }

    #[tokio::test]
    async fn buy_core_failure_rolls_back_and_503() {
        let (store, _d) = store_at("coredown").await;
        seed_listing(&store, "900000012", 2500).await;
        let econ = CountingEconomy::failing();
        let resp = handle_market_buy(
            "POST",
            "/ut/game/fifa17/trade/900000012",
            b"{}",
            &econ,
            &store,
        )
        .await;
        assert_eq!(resp.status, 503);
        assert_eq!(
            store.get_listing("900000012").await.unwrap().state,
            "active"
        );
    }

    #[tokio::test]
    async fn buy_unknown_auction_is_empty_ok() {
        let (store, _d) = store_at("gone").await;
        let econ = CountingEconomy::with_balance(10_000);
        let resp = handle_market_buy(
            "POST",
            "/ut/game/fifa17/trade/900099999",
            b"{}",
            &econ,
            &store,
        )
        .await;
        assert_eq!(resp.status, 200);
        assert!(parse(&resp)["auctionInfo"].as_array().unwrap().is_empty());
        assert_eq!(econ.purchase_calls.load(Ordering::SeqCst), 0);
    }

    /// Two buyers hit the SAME listing concurrently: exactly one sale, exactly
    /// one debit; the loser sees a closed (empty) auction.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn two_buyers_exactly_one_sale_one_debit() {
        let (store, _d) = store_at("race").await;
        seed_listing(&store, "900000020", 2500).await;
        let store = Arc::new(store);
        let econ = Arc::new(CountingEconomy::with_balance(10_000));

        let mk = || {
            let s = store.clone();
            let e = econ.clone();
            tokio::spawn(async move {
                let r =
                    handle_market_buy("POST", "/ut/game/fifa17/trade/900000020", b"{}", &*e, &s)
                        .await;
                let body: Value = serde_json::from_slice(&r.body).unwrap();
                (r.status, body["auctionInfo"].as_array().unwrap().len())
            })
        };
        let (a, b) = (mk(), mk());
        let (ra, rb) = (a.await.unwrap(), b.await.unwrap());

        // Exactly one buy produced a (closed) auction record; the other is empty.
        let winners = [ra, rb].iter().filter(|(_, n)| *n == 1).count();
        assert_eq!(winners, 1, "exactly one buyer wins: {ra:?} {rb:?}");
        assert_eq!(
            econ.purchase_calls.load(Ordering::SeqCst),
            1,
            "exactly one Core debit"
        );
        assert_eq!(econ.balance().unwrap(), 7500, "debited exactly once");
        assert_eq!(store.get_listing("900000020").await.unwrap().state, "sold");
    }

    // ---- cancel ------------------------------------------------------------

    #[tokio::test]
    async fn cancel_acks_and_cancels() {
        let (store, _d) = store_at("cancel").await;
        seed_listing(&store, "900000030", 2500).await;
        let resp =
            handle_market_cancel("/ut/delete/game/fifa17/trade/900000030", None, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(parse(&resp), json!({}));
        assert_eq!(
            store.get_listing("900000030").await.unwrap().state,
            "cancelled"
        );
        // A cancel of a nonexistent trade still acks (client-idempotent).
        let resp2 =
            handle_market_cancel("/ut/delete/game/fifa17/trade/900099999", None, &store).await;
        assert_eq!(resp2.status, 200);
    }

    // ---- move items --------------------------------------------------------

    #[tokio::test]
    async fn move_between_club_and_tradepile() {
        let db = TempDb::new("move");
        let piles = PileStore::open(db.path()).await.unwrap();
        let resolver = MapResolver::new(&[(100004617, "core-uuid-7")]);

        let to_trade = json!({ "itemData": [{ "id": 100004617, "pile": "trade" }] });
        let resp = handle_move_items(to_trade.to_string().as_bytes(), &resolver, &piles).await;
        assert_eq!(resp.status, 200);
        let b = parse(&resp);
        assert_eq!(b["itemData"][0]["success"], true);
        assert_eq!(b["itemData"][0]["pile"], "trade");
        assert_eq!(b["itemData"][0]["id"], 100004617i64);
        assert_eq!(
            piles.get("core-uuid-7").await.unwrap().as_deref(),
            Some("trade")
        );

        // Move back to the club.
        let to_club = json!({ "itemData": [{ "id": 100004617, "pile": "club" }] });
        let resp2 = handle_move_items(to_club.to_string().as_bytes(), &resolver, &piles).await;
        assert_eq!(parse(&resp2)["itemData"][0]["success"], true);
        assert_eq!(
            piles.get("core-uuid-7").await.unwrap().as_deref(),
            Some("club")
        );
    }

    #[tokio::test]
    async fn move_unknown_wire_is_success_false() {
        let db = TempDb::new("moveunk");
        let piles = PileStore::open(db.path()).await.unwrap();
        let resolver = MapResolver::new(&[]);
        let body = json!({ "itemData": [{ "id": 42, "pile": "club" }] });
        let resp = handle_move_items(body.to_string().as_bytes(), &resolver, &piles).await;
        let b = parse(&resp);
        assert_eq!(b["itemData"][0]["success"], false);
        assert_eq!(piles.get("core-uuid-7").await.unwrap(), None);
    }

    /// Pile state persists across a store reopen (durable, not in-memory).
    #[tokio::test]
    async fn move_persists_across_reopen() {
        let db = TempDb::new("movepersist");
        let path = db.path();
        let resolver = MapResolver::new(&[(7, "core-7")]);
        {
            let piles = PileStore::open(path).await.unwrap();
            let body = json!({ "itemData": [{ "id": 7, "pile": "purchased" }] });
            handle_move_items(body.to_string().as_bytes(), &resolver, &piles).await;
        }
        let reopened = PileStore::open(path).await.unwrap();
        assert_eq!(
            reopened.get("core-7").await.unwrap().as_deref(),
            Some("purchased")
        );
    }
}