OpenFUT/openfut-utas-host/src/market.rs

//! 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::entities::ReverseEntityResolver;
use openfut_adapter_fifa17::fut::item::{shape_item, ItemIdentityResolver};
use openfut_adapter_fifa17::fut::item_state;
use openfut_adapter_fifa17::fut::non_economy;
use openfut_adapter_fifa17::fut::squad::SquadWireResolver;

use crate::economy_store::OwnedItemLookup;

use crate::market_store::{now_secs, Listing, MarketError, MarketStore};
use crate::pile_store::PileStore;
use crate::sold_experiment::{CountMode, SoldExperiment};
use crate::{CoreEconomy, CoreError, ResponseTransport, 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,
        transport: ResponseTransport::Normal,
    }
}

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.
///
/// `itemData` MUST be a full card object — the same proven-safe shape that renders
/// club/squad cards (the Python oracle's tradePile reuses its club card verbatim).
/// A 4-field stub gives the client's card view-model nothing to draw, so the
/// Transfer List shows a row with no visible card. The full card comes from the
/// snapshot persisted at listing time; a row written before snapshots existed
/// degrades to the stub (honest, not fabricated).
///
/// `state` overrides the card's `itemState`. FIFA 17's vocabulary is the
/// 12-row `{const char*, int}` table at `0x180229cc0`, and `forSale` (5) is its
/// value for an item offered for sale. The Python oracle stamps `listFS` on the
/// seller's own pile instead — a token that does NOT EXIST in FIFA 17 (zero
/// occurrences in `CardsDLL_Win64_retail.dll`, zero in 4.26 GiB of live process
/// memory) and therefore decodes to `-1` through `FUN_180166660`, i.e. the client
/// is handed an unrecognised `CARD_OFFERSTATE`. Where the binary contradicts the
/// oracle, the binary wins.
fn auction_record_as(l: &Listing, state: &str) -> Value {
    auction_record_tuned(l, state, None, 0)
}

/// [`auction_record_as`] with the two fields the staging sold experiment varies.
///
/// `sold_bid_state` overrides `bidState` for a terminal (non-active) listing, and
/// `coins_processed` sets the atom the client publishes to Flash as
/// `COINS_AWARDED`. Both default to today's production values via
/// [`auction_record_as`], so nothing changes unless the experiment is on.
///
/// Everything else is byte-identical between variants BY CONSTRUCTION: there is
/// one record builder, and the A/B changes only what is passed in here. That is
/// what makes the client's reaction attributable to the token.
fn auction_record_tuned(
    l: &Listing,
    state: &str,
    sold_bid_state: Option<&str>,
    coins_processed: i64,
) -> 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);
    // `expires` is SECONDS REMAINING (a 64-bit int), never an epoch, and the
    // client renders a LIVE COUNTDOWN from it and expects it to reach 0. A frozen
    // constant is therefore wrong on the wire even though it renders: the auction
    // never appears to age. Derived from the stored creation time plus the
    // client-supplied listing duration.
    let expires = l.expires_in_secs(now_secs());
    // An unsold auction whose clock has run out reads `expired`/`none` with
    // `expires: 0` — that is FIFA 17's relistable state. Both vocabularies are
    // closed sets read out of the client: `tradeState` decodes through a table
    // walk (`active=1 inactive=2 expired=3 closed=4`, anything else -1) and
    // `bidState` through a strcmp ladder (`none=0 outbid=1 highest=2 buyNow=3`,
    // anything else silently `none`). NEVER invent a state string — an
    // unrecognised one is swallowed as `none` and produces a plausible-looking
    // but wrong UI. There is no `won`, `lost` or `sold`.
    //
    // This is a pure PROJECTION: no row is mutated, so nothing here can race the
    // economy or need a background sweeper.
    let (trade_state, bid_state, current_bid) = match l.state.as_str() {
        "active" if expires == 0 => ("expired", "none", 0),
        "active" => ("active", "none", 0),
        // Terminal. `closed` is the only FIFA 17 token for "this auction is over
        // and something happened"; there is no `sold`. The experiment varies which
        // bidState rides along, because that is the one thing the movie can see
        // (published verbatim as YOURBID) and the native flags cannot distinguish.
        _ => (
            "closed",
            sold_bid_state.unwrap_or("highest"),
            l.buy_now_price,
        ),
    };
    let item_data = l
        .item_json
        .as_deref()
        .and_then(|s| serde_json::from_str::<Value>(s).ok())
        .filter(Value::is_object)
        .map(|mut card| {
            // Keep the wire identity and presentation state authoritative here.
            card["id"] = json!(item_id);
            card["itemState"] = json!(state);
            card["untradeable"] = json!(false);
            card
        })
        .unwrap_or_else(|| {
            json!({
                "id": item_id,
                "resourceId": resource,
                "itemState": state,
                "untradeable": false,
            })
        });
    // EXACTLY the twelve fields FIFA 17's auctionInfo deserializer (0x18013e410)
    // reads. Everything else falls through to its value-SKIP at 0x180135ff0, so an
    // extra key is not "harmless richness" — it is dead weight that misleads the
    // next reader about what the client consumes.
    //
    // In particular `tradeOwner` / `sellerId` / `offers` are NOT read by FIFA 17.
    // They were added here on the strength of contemporaneous FIFA 17 libraries
    // and are refuted by the PE's own atom table (see
    // docs/FIFA17_TRANSFER_MARKET_WIRE.md): the client cannot be told "this
    // auction is yours" through the record at all, so ownership is NOT the gate on
    // the Transfer List Actions panel.
    json!({
        "tradeId": trade_id,
        "itemData": item_data,
        "tradeState": trade_state,
        "buyNowPrice": l.buy_now_price,
        "startingBid": l.start_price,
        "currentBid": current_bid,
        "bidState": bid_state,
        "expires": expires,
        // Bounded copy, max 30 chars. The oracle stamps the player's persona here
        // and `fut_account.py` annotates that property as "UTAS sellerName", so
        // EA's house name would make the player's own listing look foreign.
        "sellerName": l
            .owner
            .clone()
            .unwrap_or_else(|| non_economy::PERSONA_DISPLAY_NAME.to_string()),
        "sellerEstablished": 1,
        "watched": false,
        // Published to Flash as COINS_AWARDED (record +0xbf, atom 0x2f4, u8).
        // Production emits 0; the experiment's third pass varies it to learn
        // whether the client treats it as informational or as a gate.
        "coinsProcessed": coins_processed,
    })
}

/// Auction record for a market/search context (`itemState: forSale`), and for the
/// closed/sold echoes the buy path returns.
fn auction_record(l: &Listing) -> Value {
    let state = if l.state == "active" {
        item_state::FOR_SALE
    } else {
        item_state::FREE
    };
    auction_record_as(l, state)
}

/// 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)
}

/// A FutISStart POST resolved to a persistable listing — all owned/`Send` data,
/// so it can cross the async persist boundary. Built by [`resolve_market_list`]
/// on the caller's thread from Core inventory (the request body carries only the
/// wire item id, never the resourceId or Core card_id).
pub struct ResolvedListing {
    pub item_id: i64,
    pub core_id: String,
    pub card_id: String,
    pub resource_id: Option<i64>,
    pub start: i64,
    pub buy_now: i64,
    pub seller: Option<String>,
    /// The full shaped FIFA card (`itemData`) snapshot for the auction record.
    /// `None` only when the item has no resolvable FIFA identity (never faked).
    pub item_json: Option<String>,
    /// Listing duration in seconds from the client's body. `None` when the client
    /// omits it, which falls back to the store's default.
    pub duration: Option<i64>,
}

/// Resolve a `/auctionhouse` POST (FutISStart) body to a [`ResolvedListing`],
/// SERVER-SIDE: the client's body carries only the wire item id, so the owned
/// card's Core `card_id` (minted on a synthetic buy) and FIFA `resourceId` (the
/// auction record) come from Core inventory. Returns `None` when the body names
/// no item OR the id does not reverse-resolve to an owned card — either way the
/// caller acks a fresh trade id and persists nothing (Core stays the ownership
/// authority; a phantom listing would mint a card the buy preflight rejects).
/// Pure identity + a single Core inventory read; run it OFF the async runtime.
pub fn resolve_market_list<E: ReverseEntityResolver>(
    body: &[u8],
    reverse: &dyn SquadWireResolver,
    resolver: &dyn ItemIdentityResolver,
    items: &dyn OwnedItemLookup,
    ent: &E,
) -> Option<ResolvedListing> {
    let b = parse_body(body);
    let item_data = b.get("itemData");
    let 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 core_id = reverse.owned_id_for_wire(item_id)?;
    let owned = items.owned_item(&core_id)?;
    // Shape the FULL card ONCE, here, with the same shaper `/club` and the squad
    // projection use — so the auction record renders identically to the club card.
    // A listing is a snapshot; storing it avoids re-resolving on every tradePile
    // poll and keeps the (non-Send) resolvers off the async persist path.
    let identity = resolver.resolve(&owned);
    let resource_id = identity.map(|id| id.resource_id as i64);
    let item_json = identity
        .map(|id| shape_item(&owned, id, ent))
        .and_then(|card| serde_json::to_string(&card).ok());
    Some(ResolvedListing {
        item_id,
        core_id,
        card_id: owned.card_id,
        resource_id,
        start: b.get("startingBid").and_then(Value::as_i64).unwrap_or(150),
        buy_now: b.get("buyNowPrice").and_then(Value::as_i64).unwrap_or(0),
        seller: b
            .get("sellerName")
            .and_then(Value::as_str)
            .map(str::to_string),
        item_json,
        // FIFA 17's ISStart body carries the listing duration in seconds. We
        // previously dropped it and reported a frozen `expires`, so the client's
        // countdown never moved and an auction could never run out.
        duration: b.get("duration").and_then(Value::as_i64).filter(|d| *d > 0),
    })
}

/// `/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 persists the pre-resolved listing (see [`resolve_market_list`]) and
///   returns `{"id": tradeId}`; an unresolved item acks a fresh id, persisting
///   nothing.
/// * PUT (relist-all) is an ack `{}`.
pub async fn handle_market_list(
    method: &str,
    resolved: Option<ResolvedListing>,
    econ: &dyn CoreEconomy,
    store: &MarketStore,
) -> WireResponse {
    match method {
        "POST" => {
            let Some(r) = resolved else {
                // No item, or the id did not resolve to an owned card: fresh-id ack.
                eprintln!(
                    "utas-host owner=RUST route=market-list POST listed=false reason=unresolved"
                );
                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 + r.item_id;
            let listing_id = trade_id.to_string();
            match store
                .create_listing(
                    &listing_id,
                    &r.card_id,
                    Some(&r.core_id),
                    Some(r.item_id),
                    r.resource_id,
                    r.start,
                    r.buy_now,
                    r.seller.as_deref(),
                    r.item_json.as_deref(),
                    r.duration,
                )
                .await
            {
                Ok(_) => {
                    eprintln!(
                        "utas-host owner=RUST route=market-list POST item_id={} listed=true trade_id={trade_id}",
                        r.item_id
                    );
                    ok_json(&json!({ "id": trade_id }))
                }
                // A row for this item already exists, so this POST is a RELIST.
                // FIFA 17 relists by re-sending ISStart, so the PK conflict is the
                // normal relist path — NOT an error, and NOT a success to swallow.
                // Acking it without resetting the clock is why relisting an expired
                // card appeared to do nothing: the client got its id back while the
                // stale row stayed expired.
                Err(MarketError::Conflict) => {
                    match store
                        .relist_listing(
                            &listing_id,
                            r.start,
                            r.buy_now,
                            r.duration,
                            r.item_json.as_deref(),
                        )
                        .await
                    {
                        Ok(_) => {
                            eprintln!(
                                "utas-host owner=RUST route=market-list POST item_id={} relisted=true trade_id={trade_id}",
                                r.item_id
                            );
                            ok_json(&json!({ "id": trade_id }))
                        }
                        // Sold or in-flight: never revive it. Ack so the screen does
                        // not wedge, but say so plainly in the log.
                        Err(e) => {
                            eprintln!(
                                "utas-host owner=RUST route=market-list POST item_id={} relisted=false reason={e:?} trade_id={trade_id}",
                                r.item_id
                            );
                            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.
///
/// Each card carries `itemState: "forSale"` (5) — FIFA 17's own token for an item
/// offered for sale. The oracle's `listFS` is not a FIFA 17 value at all and
/// decoded to `-1`; see [`auction_record_as`].
///
/// ONLY real auctions appear here. An item sitting in the trade pile with no
/// auction is deliberately absent: `plan-2026-08-06-transfer-market.md:731-733`
/// says of the `tradeState` vocabulary "`inactive` decodes but no client path
/// treats it specially; do not emit it", and RE of the FUT front-end confirmed
/// why — the trade-pile movie admits only rows it classifies as auctions
/// (`getCardsInAuction`/`isInActiveAuction`) into the action path, so an
/// `inactive` row renders and can never be acted on. In the shipped game those
/// rows were a client-side transient built by the movie itself from a
/// `TO_TRADEPILE` Flash message, never server-delivered. Nothing is stranded:
/// `/club` excludes only items with an ACTIVE listing, so an unlisted pile member
/// stays visible in the club.
pub async fn handle_market_query(
    state: &str,
    econ: &dyn CoreEconomy,
    store: &MarketStore,
    exp: SoldExperiment,
) -> WireResponse {
    let listings = match store.query_listings(state).await {
        Ok(l) => l,
        Err(_) => return json_body(503, &json!({ "error": "market_store" })),
    };
    let mut auctions: Vec<Value> = listings
        .iter()
        .map(|l| auction_record_as(l, item_state::FOR_SALE))
        .collect();
    // STAGING ONLY. FIFA 17's bulk `DELETE …/trade/sold` verb only makes sense if
    // sold rows persist in the seller's pile until acknowledged, so the experiment
    // projects uncleared sold listings alongside the active ones. Off in
    // production, where this stays exactly the Fix A invariant: active auctions
    // only.
    if exp.enabled() {
        if let Ok(sold) = store.uncleared_sold().await {
            for l in &sold {
                auctions.push(auction_record_tuned(
                    l,
                    item_state::FOR_SALE,
                    exp.bid_state,
                    exp.coins_processed,
                ));
            }
            if !sold.is_empty() {
                eprintln!(
                    "utas-host owner=RUST route=market-query SOLD-EXPERIMENT \
                     sold_rows={} bidState={:?} coinsProcessed={}",
                    sold.len(),
                    exp.bid_state,
                    exp.coins_processed
                );
            }
        }
    }
    // GetTradePile shares one deserializer (0x18013e7f0) with ISSearch and
    // ISWatchList, over exactly four members: `auctionInfo` (array), `credits`
    // (int), `duplicateItemIdList` (array of objects) and `total` (int). We were
    // omitting `duplicateItemIdList`; `[]` is the safe, recommended value.
    ok_json(&json!({
        "auctionInfo": auctions,
        "credits": credits_or_zero(econ),
        "total": auctions.len(),
        "duplicateItemIdList": [],
    }))
}

/// `GET …/tradePile/counts` — FutGetAuctionCount (the auction TALLY), a DISTINCT
/// deserializer from `/tradePile`. It reads exactly five SCALAR INTS — `count`,
/// `maxAuctionsAllowed`, `offered`, `selling`, `sold` — and skips anything else,
/// so answering it with the `auctionInfo` listing body leaves every count at its
/// constructor default (0): the hub tile shows a listing while the Transfer List
/// screen shows none. All five being ints means there is no container-type
/// freeze risk. They are the only inputs to IS_MAX_AUCTIONS, so
/// `maxAuctionsAllowed = 100` with `selling < 100` keeps the listing cap open.
/// A store read failure degrades to zeros (cosmetic tally, never fail-closed).
///
/// `sold` is NOT cosmetic: RE proved atom `sold` (0x2c9) reaches the hub tile as
/// Flash `TEXT3` under the localised caption `FUT_TF_SOLD`, so the seller really
/// does see a SOLD bucket. Production still reports 0 because we have never had a
/// sold row; the experiment reports the real count so the client can be observed.
pub async fn handle_market_counts(store: &MarketStore, exp: SoldExperiment) -> WireResponse {
    let selling = store
        .query_listings("active")
        .await
        .map(|l| l.len() as i64)
        .unwrap_or(0);
    let sold = if exp.enabled() {
        store
            .uncleared_sold()
            .await
            .map(|l| l.len() as i64)
            .unwrap_or(0)
    } else {
        0
    };
    // FIFA 17's exact meaning for `count` is unknown — live auctions, or whole
    // Transfer List membership. It is a controlled variable, never a guess.
    let count = match exp.count_mode {
        CountMode::Active => selling,
        CountMode::ActivePlusSold => selling + sold,
    };
    ok_json(&json!({
        "count": count,
        "maxAuctionsAllowed": 100,
        "offered": 0,
        "selling": selling,
        "sold": sold,
    }))
}

/// `DELETE /ut/delete/game/<sku>/trade/sold` — the bulk clear-sold verb.
///
/// PE-proven: the request builder `0x1801647c0` emits the literal `/sold` when the
/// tradeId field is zero and `/%lld` otherwise, and the client's request-name table
/// calls it `RemoveAllSoldFromTradePile`. The response body parses nothing, so `{}`
/// is the whole contract.
///
/// PRESENTATION ONLY. Settlement already happened when the sale completed; this
/// records the seller's acknowledgement. It must never move coins or ownership,
/// or a client retry would pay twice.
pub async fn handle_market_clear_sold(store: &MarketStore) -> WireResponse {
    match store.clear_sold().await {
        Ok(n) => eprintln!("utas-host owner=RUST route=market-clear-sold cleared={n}"),
        Err(e) => eprintln!("utas-host WARN market clear_sold failed: {e}"),
    }
    ok_json(&json!({}))
}

/// `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!({}))
}

/// `GET …/trade/status` — live auction-state refresh for the rows a screen is
/// showing. The Transfer List polls this CONTINUOUSLY while it is open.
///
/// This route previously fell through to the buy/view arm, where
/// `trade_id_from_path("trade/status")` cannot parse an id, so every poll was
/// answered with an EMPTY `auctionInfo` — the client kept asking for the state of
/// its own listings and was told, repeatedly, that there was none. Observed
/// directly in the live logs (`route=economy … path=…/trade/status` on repeat),
/// so unlike the `tradeOwner` change this is a CONFIRMED defect, not a candidate.
///
/// `tradeIds` is a comma-separated filter; unknown ids are simply absent from the
/// reply rather than erroring. With no filter we answer with the player's own
/// active pile, which is the only auction set this single-account market has.
pub async fn handle_market_status(
    query: Option<&str>,
    econ: &dyn CoreEconomy,
    store: &MarketStore,
    exp: SoldExperiment,
) -> WireResponse {
    let ids = trade_ids_from_query(query);
    let listings = if ids.is_empty() {
        let mut all = match store.query_listings("active").await {
            Ok(l) => l,
            Err(_) => return json_body(503, &json!({ "error": "market_store" })),
        };
        // Unfiltered poll: the experiment's sold rows are part of what the screen
        // is showing, so they must answer here too or the row would render from
        // /tradePile and then contradict its own status poll.
        if exp.enabled() {
            if let Ok(sold) = store.uncleared_sold().await {
                all.extend(sold);
            }
        }
        all
    } else {
        let mut found = Vec::with_capacity(ids.len());
        for id in &ids {
            if let Ok(l) = store.get_listing(id).await {
                found.push(l);
            }
        }
        found
    };
    let auctions: Vec<Value> = listings
        .iter()
        .map(|l| auction_record_tuned(l, item_state::FOR_SALE, exp.bid_state, exp.coins_processed))
        .collect();
    eprintln!(
        "utas-host owner=RUST route=market-status requested={} returned={} query={}",
        ids.len(),
        auctions.len(),
        query.unwrap_or("")
    );
    ok_json(&json!({
        "auctionInfo": auctions,
        "credits": credits_or_zero(econ),
    }))
}

/// Parse `tradeIds=1,2,3` (also accepts repeated `tradeIds=`) out of a raw query
/// string. Non-numeric entries are skipped rather than failing the whole poll.
fn trade_ids_from_query(query: Option<&str>) -> Vec<String> {
    let Some(q) = query else {
        return Vec::new();
    };
    q.split('&')
        .filter_map(|kv| kv.split_once('='))
        .filter(|(k, _)| k.eq_ignore_ascii_case("tradeIds") || k.eq_ignore_ascii_case("tradeId"))
        .flat_map(|(_, v)| v.split(','))
        .filter(|s| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()))
        .map(str::to_string)
        .collect()
}

/// `/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!(item_state::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.
///
/// A move to the `club` pile also ENDS any live auction on that item. This is the
/// return-to-club transition an expired transfer-list item takes, and the auction
/// that put it in the pile has to end with it: otherwise the pile reads `club`
/// while the listing row stays `active`, so the card is filtered out of `/club`
/// AND still rendered in the Transfer List, i.e. the move appears to do nothing.
/// A `reserved` (mid-sale) or `sold` row is never touched.
pub async fn handle_move_items(
    body: &[u8],
    resolver: &(dyn SquadWireResolver + Sync),
    pile_store: &PileStore,
    market: &MarketStore,
) -> 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) => {
                let moved = pile_store.set(&core_id, &pile).await.is_ok();
                if moved && pile == "club" {
                    match market.cancel_active_for_core_item(&core_id).await {
                        Ok(n) if n > 0 => eprintln!(
                            "utas-host owner=RUST route=move-items wire={wire} pile=club auction_cancelled={n}"
                        ),
                        Ok(_) => {}
                        // The pile move already succeeded and Core still owns the
                        // card; report the move honestly and log the stale auction.
                        Err(e) => eprintln!(
                            "utas-host WARN move-items wire={wire} pile=club auction cancel failed: {e:?}"
                        ),
                    }
                }
                moved
            }
            None => false,
        };
        verdicts.push(json!({ "id": wire, "pile": pile, "success": success }));
    }
    ok_json(&json!({ "itemData": verdicts }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        CoreMatchCompletion, CoreMatchReceipt, EconomyEntitlement, EconomyGrantItem,
        EconomyPurchase, EconomySale, EconomySaleReceipt,
    };
    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 ----

    /// A single coin pot stands in for the whole modelled economy: a buy-now
    /// debits it, and a club-to-club settlement debits the buyer then credits
    /// the seller out of the same pot, so the pot falls by exactly the fee —
    /// the coins the market destroys.
    struct CountingEconomy {
        balance: AtomicI64,
        purchase_calls: AtomicUsize,
        settle_calls: AtomicUsize,
        fail: bool,
    }
    impl CountingEconomy {
        fn with_balance(balance: i64) -> Self {
            CountingEconomy {
                balance: AtomicI64::new(balance),
                purchase_calls: AtomicUsize::new(0),
                settle_calls: AtomicUsize::new(0),
                fail: false,
            }
        }
        fn failing() -> Self {
            CountingEconomy {
                balance: AtomicI64::new(0),
                purchase_calls: AtomicUsize::new(0),
                settle_calls: AtomicUsize::new(0),
                fail: true,
            }
        }
        /// Atomic debit: reject (and do NOT debit) if it would go negative,
        /// mirroring Core's BadRequest(400) on insufficient funds.
        fn debit(&self, cost: i64) -> Result<i64, CoreError> {
            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,
                }
            }
        }
    }
    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));
            }
            self.debit(cost)
        }
        fn purchase_items(
            &self,
            _cost: i64,
            _items: &[EconomyGrantItem],
        ) -> Result<i64, CoreError> {
            Err(CoreError::Status(500))
        }
        fn settle_sale(&self, sale: &EconomySale<'_>) -> Result<EconomySaleReceipt, CoreError> {
            self.settle_calls.fetch_add(1, Ordering::SeqCst);
            if self.fail {
                return Err(CoreError::Status(500));
            }
            // A club buyer pays out of the pot first (and can be too poor);
            // an outside buyer is not modelled, so nobody is debited.
            let buyer_balance = match sale.buyer_club_id {
                Some(_) => Some(self.debit(sale.gross)?),
                None => None,
            };
            let proceeds = sale.gross - sale.fee;
            let seller_balance = self.balance.fetch_add(proceeds, Ordering::SeqCst) + proceeds;
            Ok(EconomySaleReceipt {
                item_id: sale.item_id.to_string(),
                card_id: format!("card-of:{}", sale.item_id),
                seller_club_id: sale.seller_club_id.unwrap_or("active-club").to_string(),
                buyer_club_id: sale.buyer_club_id.map(str::to_string),
                gross: sale.gross,
                fee: sale.fee,
                proceeds,
                seller_balance,
                buyer_balance,
                squad_slots_freed: 0,
            })
        }
        fn complete_match(
            &self,
            _m: &CoreMatchCompletion<'_>,
        ) -> Result<CoreMatchReceipt, CoreError> {
            // Match completion is not exercised through the market double.
            Err(CoreError::Status(501))
        }
    }

    // ---- 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()
        }
    }

    // ---- ItemIdentityResolver + OwnedItemLookup doubles -------------------
    use openfut_adapter_fifa17::fut::item::{CoreOwnedItem, Fifa17Identity};

    /// Owned-inventory double: core id -> CoreOwnedItem.
    struct FakeItems(HashMap<String, CoreOwnedItem>);
    impl OwnedItemLookup for FakeItems {
        fn owned_item(&self, core_id: &str) -> Option<CoreOwnedItem> {
            self.0.get(core_id).cloned()
        }
    }

    /// Identity double: resolves every owned item to a fixed FIFA resourceId.
    struct FixedIdentity(u32);
    impl ItemIdentityResolver for FixedIdentity {
        fn resolve(&self, _item: &CoreOwnedItem) -> Option<Fifa17Identity> {
            Some(Fifa17Identity {
                item_id: 0,
                asset_id: self.0,
                resource_id: self.0,
                rareflag: 1,
            })
        }
    }

    /// A CoreOwnedItem with the given core id + card id (other fields dummy).
    fn owned(core_id: &str, card_id: &str) -> CoreOwnedItem {
        CoreOwnedItem {
            owned_card_id: core_id.to_string(),
            card_id: card_id.to_string(),
            rating: 84,
            position: "ST".to_string(),
            nation: String::new(),
            league: String::new(),
            club: String::new(),
            attributes: [80, 80, 80, 80, 80, 80],
        }
    }

    /// Entity double: no reverse entity mappings, so shaped cards carry ids 0
    /// (a valid int — the shaper never fabricates an entity id).
    struct NoEntities;
    impl ReverseEntityResolver for NoEntities {
        fn league_id(&self, _name: &str) -> Option<u32> {
            None
        }
        fn nation_id(&self, _name: &str) -> Option<u32> {
            None
        }
        fn team_id(&self, _name: &str) -> Option<u32> {
            None
        }
    }

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

    #[tokio::test]
    async fn the_trade_pile_advertises_only_real_auctions() {
        // Regression guard for a wrong answer we shipped: an item sitting in the
        // trade pile with no auction was advertised as a synthetic
        // `tradeState:"inactive"` record. The corpus already forbade it
        // (plan-2026-08-06-transfer-market.md:731-733, "`inactive` decodes but no
        // client path treats it specially; do not emit it"), and RE of the FUT
        // front-end explained why: the trade-pile movie admits only rows it
        // classifies as auctions into the action path, so such a row renders and
        // can never be acted on. Only auctions belong in this body.
        let (store, _d) = store_at("pileonlyauctions").await;
        let econ = CountingEconomy::with_balance(10_000);

        // Nothing listed: an empty pile, whatever the item store holds.
        let body = parse(&handle_market_query("active", &econ, &store, SoldExperiment::OFF).await);
        assert_eq!(body["auctionInfo"].as_array().unwrap().len(), 0);
        assert_eq!(body["total"], 0);

        store
            .create_listing(
                &(TRADE_ID_BASE + 100_000_059).to_string(),
                "169193",
                Some("core-listed"),
                Some(100_000_059),
                Some(169193),
                150,
                2500,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        let body = parse(&handle_market_query("active", &econ, &store, SoldExperiment::OFF).await);
        let recs = body["auctionInfo"].as_array().unwrap();
        assert_eq!(recs.len(), 1, "the real auction, and nothing synthetic");
        assert_eq!(recs[0]["tradeState"], "active");
        assert!(
            recs.iter().all(|r| r["tradeState"] != "inactive"),
            "`inactive` must never appear on the wire: {recs:#?}"
        );
    }

    async fn seed_listing(store: &MarketStore, id: &str, buy_now: i64) {
        store
            .create_listing(
                id,
                "169193",
                None,
                None,
                Some(169193),
                400,
                buy_now,
                None,
                None,
                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);
        // The client body carries only the wire item id; the server resolves the
        // owned card's card_id + resourceId from Core inventory.
        let body = json!({ "itemData": { "id": 100004617 },
                           "startingBid": 300, "buyNowPrice": 2500 });
        let reverse = MapResolver::new(&[(100004617, "core-1")]);
        let items = FakeItems(
            [("core-1".to_string(), owned("core-1", "169193"))]
                .into_iter()
                .collect(),
        );
        let ident = FixedIdentity(169193);
        let resolved = resolve_market_list(
            body.to_string().as_bytes(),
            &reverse,
            &ident,
            &items,
            &NoEntities,
        );
        let resp = handle_market_list("POST", resolved, &econ, &store).await;
        assert_eq!(resp.status, 200);
        let trade_id = parse(&resp)["id"].as_i64().unwrap();
        assert_eq!(trade_id, TRADE_ID_BASE + 100004617);
        // Persisted with the resolved Core card_id + wire resourceId, browsable.
        let listed = store.get_listing(&trade_id.to_string()).await.unwrap();
        assert_eq!(listed.buy_now_price, 2500);
        assert_eq!(listed.card_id, "169193");
        assert_eq!(listed.wire_resource_id, Some(169193));
        // The listing carries the FULL shaped card snapshot, not a 4-field stub:
        // a stub leaves the Transfer List with an unrenderable row (the live bug).
        let snap: Value = serde_json::from_str(listed.item_json.as_deref().unwrap()).unwrap();
        assert_eq!(snap["rating"], 84);
        assert_eq!(snap["preferredPosition"], "ST");
        assert_eq!(snap["attributeList"].as_array().unwrap().len(), 6);
        // tradePile embeds that full card and stamps the seller-pile state. This
        // asserted `listFS` until FIFA17.exe disproved it: that token appears
        // nowhere in CardsDLL or in 4.26 GiB of live process memory and decoded to
        // -1, so the client was handed an unrecognised CARD_OFFERSTATE. `forSale`
        // (5) is the value in FIFA 17's own itemState table.
        let pile = handle_market_query("active", &econ, &store, SoldExperiment::OFF).await;
        let rec = parse(&pile)["auctionInfo"][0].clone();
        assert_eq!(rec["itemData"]["itemState"], "forSale");
        assert!(
            item_state::is_recovered(rec["itemData"]["itemState"].as_str().unwrap()),
            "every emitted itemState must be in FIFA 17's own 12-row table"
        );
        assert_eq!(rec["itemData"]["rating"], 84);
        assert_eq!(rec["itemData"]["id"], 100004617i64);
        assert_eq!(rec["itemData"]["resourceId"], 169193);
        let browse = handle_market_list("GET", None, &econ, &store).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", None, &econ, &store).await;
        assert_eq!(resp.status, 200);
        assert_eq!(parse(&resp), json!({}));
    }

    #[tokio::test]
    async fn list_unresolved_item_fails_closed_no_listing() {
        // A wire id that does not resolve to an owned card must NOT create a listing
        // (a synthetic buy would otherwise mint a card Core cannot resolve). Acks.
        let (store, _d) = store_at("deny").await;
        let econ = CountingEconomy::with_balance(10_000);
        let body = json!({ "itemData": { "id": 100004617 },
                           "startingBid": 300, "buyNowPrice": 2500 });
        let reverse = MapResolver::new(&[]); // maps nothing -> unresolved
        let items = FakeItems(HashMap::new());
        let ident = FixedIdentity(0);
        let resolved = resolve_market_list(
            body.to_string().as_bytes(),
            &reverse,
            &ident,
            &items,
            &NoEntities,
        );
        assert!(
            resolved.is_none(),
            "unresolved item must not build a listing"
        );
        let resp = handle_market_list("POST", resolved, &econ, &store).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);
            let body = json!({ "itemData": { "id": 100004900 },
                               "startingBid": 300, "buyNowPrice": 2500 });
            let reverse = MapResolver::new(&[(100004900, "core-9")]);
            let items = FakeItems(
                [("core-9".to_string(), owned("core-9", "card_pl_042"))]
                    .into_iter()
                    .collect(),
            );
            let ident = FixedIdentity(20801);
            let resolved = resolve_market_list(
                body.to_string().as_bytes(),
                &reverse,
                &ident,
                &items,
                &NoEntities,
            );
            let resp = handle_market_list("POST", resolved, &econ, &store).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, SoldExperiment::OFF).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 counts_is_the_five_int_tally_not_the_listing_body() {
        // FutGetAuctionCount is a DISTINCT deserializer from /tradePile: five
        // scalar ints, no auctionInfo. Answering it with the listing body leaves
        // every count at its constructor default, so the Transfer List screen
        // shows no active sale even while the hub tile reports one (live bug).
        let (store, _d) = store_at("counts").await;
        let b = parse(&handle_market_counts(&store, SoldExperiment::OFF).await);
        assert_eq!(b["count"], 0);
        assert_eq!(b["selling"], 0);

        seed_listing(&store, "900000007", 2500).await;
        let resp = handle_market_counts(&store, SoldExperiment::OFF).await;
        assert_eq!(resp.status, 200);
        let b = parse(&resp);
        assert_eq!(b["count"], 1, "tally counts the active listing");
        assert_eq!(b["selling"], 1);
        assert_eq!(b["sold"], 0);
        assert_eq!(b["offered"], 0);
        assert_eq!(
            b["maxAuctionsAllowed"], 100,
            "cap stays open for IS_MAX_AUCTIONS"
        );
        assert!(
            b.get("auctionInfo").is_none(),
            "the tally must NOT carry the listing body"
        );
        for k in ["count", "maxAuctionsAllowed", "offered", "selling", "sold"] {
            assert!(b[k].is_i64(), "{k} must be a scalar int");
        }
    }

    #[tokio::test]
    async fn auction_record_carries_exactly_the_twelve_fields_fifa17_reads() {
        // FIFA 17's auctionInfo deserializer (0x18013e410) reads TWELVE atoms and
        // value-SKIPs everything else. Emitting extras is not harmless richness: it
        // misleads the next reader about what the client consumes, and it is how
        // `tradeOwner`/`sellerId`/`offers` got added on library hearsay and then had
        // to be removed. Pin the set.
        let (store, _d) = store_at("atoms").await;
        let econ = CountingEconomy::with_balance(10_000);
        seed_listing(&store, "900000030", 2500).await;

        let body = parse(&handle_market_query("active", &econ, &store, SoldExperiment::OFF).await);
        let rec = body["auctionInfo"][0].clone();
        let mut got: Vec<&str> = rec
            .as_object()
            .unwrap()
            .keys()
            .map(String::as_str)
            .collect();
        got.sort_unstable();
        assert_eq!(
            got,
            [
                "bidState",
                "buyNowPrice",
                "coinsProcessed",
                "currentBid",
                "expires",
                "itemData",
                "sellerEstablished",
                "sellerName",
                "startingBid",
                "tradeId",
                "tradeState",
                "watched",
            ],
            "auctionInfo must be exactly FIFA 17's twelve atoms"
        );
        // "listed by user" is active / none / currentBid 0 / expires > 0. Both
        // vocabularies are closed sets; an unrecognised bidState is swallowed as
        // `none` and renders a plausible but wrong UI.
        assert_eq!(rec["tradeState"], "active");
        assert_eq!(rec["bidState"], "none");
        assert_eq!(rec["currentBid"], 0);
        assert!(rec["expires"].as_i64().unwrap() > 0);
        assert_eq!(rec["sellerName"], non_economy::PERSONA_DISPLAY_NAME);

        // GetTradePile shares the IS-list body: four members, including the
        // `duplicateItemIdList` we used to omit.
        let mut env: Vec<&str> = body
            .as_object()
            .unwrap()
            .keys()
            .map(String::as_str)
            .collect();
        env.sort_unstable();
        assert_eq!(
            env,
            ["auctionInfo", "credits", "duplicateItemIdList", "total"],
            "GetTradePile envelope is the shared IS-list body"
        );
    }

    #[tokio::test]
    async fn expires_counts_down_and_an_unsold_auction_reads_expired() {
        // `expires` is SECONDS REMAINING and the client renders a live countdown
        // that it expects to reach 0. A frozen constant means the auction never
        // ages and can never run out.
        use crate::market_store::DEFAULT_DURATION_SECS;
        let (store, _d) = store_at("clock").await;
        seed_listing(&store, "900000040", 2500).await;
        let l = store.get_listing("900000040").await.unwrap();

        let created = l.created_at.parse::<i64>().unwrap() / 1000;
        assert_eq!(
            l.expires_in_secs(created),
            DEFAULT_DURATION_SECS,
            "a fresh listing has its whole duration left"
        );
        assert_eq!(
            l.expires_in_secs(created + 600),
            DEFAULT_DURATION_SECS - 600,
            "the clock actually advances"
        );
        assert_eq!(
            l.expires_in_secs(created + DEFAULT_DURATION_SECS + 5),
            0,
            "expiry clamps at 0, never negative"
        );

        // A closed listing has no time left regardless of when it was created.
        let mut sold = l.clone();
        sold.state = "sold".into();
        assert_eq!(sold.expires_in_secs(created), 0);
    }

    #[tokio::test]
    async fn trade_status_answers_the_poll_instead_of_an_empty_set() {
        // The Transfer List polls `…/trade/status` continuously to refresh live
        // auction state. This tail has no numeric id, so it used to fall into the
        // buy/view arm and every poll was answered with an EMPTY auctionInfo —
        // observed live, and the screen never learned its own auctions' state.
        let (store, _d) = store_at("status").await;
        let econ = CountingEconomy::with_balance(10_000);
        seed_listing(&store, "900000031", 2500).await;

        // No filter: answer with the player's own active pile.
        let all = parse(&handle_market_status(None, &econ, &store, SoldExperiment::OFF).await);
        assert_eq!(
            all["auctionInfo"].as_array().unwrap().len(),
            1,
            "an unfiltered poll must not come back empty while a listing is active"
        );
        assert_eq!(all["auctionInfo"][0]["tradeId"], 900_000_031i64);
        // ISViewTrade's body is the auction list plus credits — no `total` and no
        // `duplicateItemIdList`, unlike the shared IS-list body.
        assert_eq!(all["auctionInfo"][0]["tradeState"], "active");
        assert!(all["credits"].is_i64());

        // Explicit tradeIds filter returns exactly the requested auction.
        let one = parse(
            &handle_market_status(
                Some("tradeIds=900000031"),
                &econ,
                &store,
                SoldExperiment::OFF,
            )
            .await,
        );
        assert_eq!(one["auctionInfo"].as_array().unwrap().len(), 1);
        assert_eq!(one["auctionInfo"][0]["tradeId"], 900_000_031i64);

        // An unknown id is absent, not an error: the poll must never fail closed.
        let miss = parse(
            &handle_market_status(
                Some("tradeIds=900000099"),
                &econ,
                &store,
                SoldExperiment::OFF,
            )
            .await,
        );
        assert_eq!(miss["auctionInfo"].as_array().unwrap().len(), 0);

        // Garbage is skipped rather than poisoning the whole poll.
        assert_eq!(
            trade_ids_from_query(Some("tradeIds=900000031,abc,,900000032&x=1")),
            vec!["900000031".to_string(), "900000032".to_string()]
        );
    }

    #[tokio::test]
    async fn trade_status_answers_only_about_real_auctions() {
        // The companion to `the_trade_pile_advertises_only_real_auctions`: since the
        // pile no longer advertises synthetic ids, ISVIEWTRADE has none to resolve.
        // An id it knows nothing about is simply absent from the reply rather than
        // being answered with an invented record.
        let (store, _d) = store_at("statusonlyauctions").await;
        let econ = CountingEconomy::with_balance(10_000);
        let trade_id = TRADE_ID_BASE + 100_000_122;

        let unknown = parse(
            &handle_market_status(
                Some(&format!("tradeIds={trade_id}")),
                &econ,
                &store,
                SoldExperiment::OFF,
            )
            .await,
        );
        assert_eq!(
            unknown["auctionInfo"].as_array().unwrap().len(),
            0,
            "an id with no auction resolves to nothing, not to a synthetic row"
        );

        store
            .create_listing(
                &trade_id.to_string(),
                "169193",
                Some("core-pile"),
                Some(100_000_122),
                Some(169193),
                150,
                2500,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        let one = parse(
            &handle_market_status(
                Some(&format!("tradeIds={trade_id}")),
                &econ,
                &store,
                SoldExperiment::OFF,
            )
            .await,
        );
        let recs = one["auctionInfo"].as_array().unwrap();
        assert_eq!(recs.len(), 1, "now there is a real auction to answer with");
        assert_eq!(recs[0]["tradeId"], trade_id);
        assert_eq!(recs[0]["tradeState"], "active");
    }

    #[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 returning_an_expired_listing_to_the_club_ends_its_auction() {
        // The return-to-club transition an EXPIRED transfer-list item takes. The
        // auction that put the card in the pile has to end with the move: otherwise
        // the pile reads `club` while the listing row stays `active`, so the card is
        // filtered out of /club (exclusion keys on active listings) AND still
        // rendered in the Transfer List — the move appears to do nothing at all.
        let db = TempDb::new("moveclub");
        let piles = PileStore::open(db.path()).await.unwrap();
        let market = MarketStore::open(db.path()).await.unwrap();
        let resolver = MapResolver::new(&[(100004617, "core-uuid-7")]);
        market
            .create_listing(
                "900000050",
                "169193",
                Some("core-uuid-7"),
                Some(100004617),
                Some(169193),
                400,
                2500,
                None,
                None,
                None,
            )
            .await
            .unwrap();

        let body = json!({ "itemData": [{ "id": 100004617, "pile": "club" }] });
        let resp = handle_move_items(body.to_string().as_bytes(), &resolver, &piles, &market).await;
        assert_eq!(parse(&resp)["itemData"][0]["success"], true);
        assert_eq!(
            market.get_listing("900000050").await.unwrap().state,
            "cancelled",
            "the auction ends with the return to club"
        );
        assert!(
            market.query_listings("active").await.unwrap().is_empty(),
            "no active listing remains, so /club stops hiding the card"
        );
    }

    #[tokio::test]
    async fn a_pile_move_never_disturbs_a_sale_in_flight() {
        // A reserved row is mid-sale and a sold row is already gone. Cancelling
        // either on a pile move would let one card be both sold and returned.
        let db = TempDb::new("moveinflight");
        let piles = PileStore::open(db.path()).await.unwrap();
        let market = MarketStore::open(db.path()).await.unwrap();
        let resolver = MapResolver::new(&[(1, "core-a"), (2, "core-b")]);
        for (id, core) in [("900000060", "core-a"), ("900000061", "core-b")] {
            market
                .create_listing(
                    id,
                    "169193",
                    Some(core),
                    None,
                    None,
                    400,
                    2500,
                    None,
                    None,
                    None,
                )
                .await
                .unwrap();
        }
        assert!(market.reserve_listing("900000060").await.unwrap());
        market.reserve_listing("900000061").await.unwrap();
        market.complete_sale("900000061").await.unwrap();

        let body =
            json!({ "itemData": [{ "id": 1, "pile": "club" }, { "id": 2, "pile": "club" }] });
        handle_move_items(body.to_string().as_bytes(), &resolver, &piles, &market).await;
        assert_eq!(
            market.get_listing("900000060").await.unwrap().state,
            "reserved",
            "an in-flight sale is untouched"
        );
        assert_eq!(
            market.get_listing("900000061").await.unwrap().state,
            "sold",
            "a completed sale is untouched"
        );
    }

    #[tokio::test]
    async fn move_between_club_and_tradepile() {
        let db = TempDb::new("move");
        let piles = PileStore::open(db.path()).await.unwrap();
        let market = MarketStore::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, &market).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, &market).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 market = MarketStore::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, &market).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 market = MarketStore::open(path).await.unwrap();
            let body = json!({ "itemData": [{ "id": 7, "pile": "purchased" }] });
            handle_move_items(body.to_string().as_bytes(), &resolver, &piles, &market).await;
        }
        let reopened = PileStore::open(path).await.unwrap();
        assert_eq!(
            reopened.get("core-7").await.unwrap().as_deref(),
            Some("purchased")
        );
    }

    // ── staging sold-row experiment ──────────────────────────────────────────
    //
    // The client-facing question these support: for a `closed` row, CardsDLL's
    // native flags cannot distinguish bidState `highest` from `buyNow`
    // (IS_GLOW = bidState != none, INBOX = bidState in {highest,buyNow}), but the
    // movie receives bidState verbatim as YOURBID. So the A/B is only meaningful if
    // EVERY other field is identical between variants. These tests pin that.

    async fn seed_sold(store: &MarketStore, id: &str, buy_now: i64) {
        seed_listing(store, id, buy_now).await;
        assert!(store.mark_sold(id).await.unwrap(), "listing became sold");
    }

    /// Production default: a sold listing is INVISIBLE to the seller's pile and the
    /// counts stay exactly as they ship today. Guards the Fix A invariant against
    /// the experiment leaking into production.
    #[tokio::test]
    async fn experiment_off_hides_sold_rows_entirely() {
        let (store, _d) = store_at("soldoff").await;
        let econ = CountingEconomy::with_balance(10_000);
        seed_sold(&store, "900000200", 150).await;

        let pile = parse(&handle_market_query("active", &econ, &store, SoldExperiment::OFF).await);
        assert_eq!(
            pile["auctionInfo"].as_array().unwrap().len(),
            0,
            "no sold row"
        );
        assert_eq!(pile["total"], 0);

        let counts = parse(&handle_market_counts(&store, SoldExperiment::OFF).await);
        assert_eq!(counts["sold"], 0, "production reports sold: 0");
        assert_eq!(counts["selling"], 0);
        assert_eq!(counts["count"], 0);

        let status = parse(&handle_market_status(None, &econ, &store, SoldExperiment::OFF).await);
        assert_eq!(status["auctionInfo"].as_array().unwrap().len(), 0);
    }

    /// With the experiment on, the sold row appears and carries the token under
    /// test, `tradeState: closed`, and `currentBid` = the sale price.
    #[tokio::test]
    async fn experiment_projects_the_sold_row_with_the_token_under_test() {
        for token in ["highest", "buyNow"] {
            let (store, _d) = store_at(&format!("soldon{token}")).await;
            let econ = CountingEconomy::with_balance(10_000);
            seed_sold(&store, "900000201", 150).await;
            let exp = SoldExperiment::from_values(Some(token), None, None);

            let pile = parse(&handle_market_query("active", &econ, &store, exp).await);
            let recs = pile["auctionInfo"].as_array().unwrap();
            assert_eq!(recs.len(), 1, "{token}: the sold row is shown");
            assert_eq!(recs[0]["tradeState"], "closed", "{token}");
            assert_eq!(recs[0]["bidState"], token, "{token}");
            assert_eq!(recs[0]["currentBid"], 150, "{token}: sale price");
            assert_eq!(
                recs[0]["expires"], 0,
                "{token}: a sold auction has no clock"
            );
            assert_eq!(pile["total"], 1, "{token}");
        }
    }

    /// THE experimental control: between the two variants, EXACTLY ONE field may
    /// differ. If anything else moves, the client's reaction is not attributable to
    /// the token and the whole A/B is void.
    #[tokio::test]
    async fn the_two_variants_differ_in_bidstate_and_nothing_else() {
        let mut rows = Vec::new();
        for token in ["highest", "buyNow"] {
            let (store, _d) = store_at(&format!("soldab{token}")).await;
            let econ = CountingEconomy::with_balance(10_000);
            seed_sold(&store, "900000202", 150).await;
            let exp = SoldExperiment::from_values(Some(token), None, None);
            let pile = parse(&handle_market_query("active", &econ, &store, exp).await);
            rows.push(pile["auctionInfo"][0].clone());
        }
        let (a, b) = (&rows[0], &rows[1]);
        let keys: Vec<&String> = a.as_object().unwrap().keys().collect();
        let differing: Vec<&&String> = keys
            .iter()
            .filter(|k| a[k.as_str()] != b[k.as_str()])
            .collect();
        assert_eq!(
            differing.len(),
            1,
            "exactly one field may differ between variants, saw {differing:?}"
        );
        assert_eq!(differing[0].as_str(), "bidState");
        // And the twelve-atom shape is preserved in both.
        assert_eq!(a.as_object().unwrap().len(), 12, "still twelve atoms");
        assert_eq!(b.as_object().unwrap().len(), 12);
    }

    /// `coinsProcessed` (Flash `COINS_AWARDED`) is varied INDEPENDENTLY of the
    /// bidState A/B, so the third pass cannot be confounded with the first.
    #[tokio::test]
    async fn coins_processed_varies_alone() {
        let mut rows = Vec::new();
        for cp in [None, Some("1")] {
            let (store, _d) = store_at(&format!("soldcp{}", cp.unwrap_or("0"))).await;
            let econ = CountingEconomy::with_balance(10_000);
            seed_sold(&store, "900000203", 150).await;
            let exp = SoldExperiment::from_values(Some("highest"), cp, None);
            let pile = parse(&handle_market_query("active", &econ, &store, exp).await);
            rows.push(pile["auctionInfo"][0].clone());
        }
        assert_eq!(rows[0]["coinsProcessed"], 0);
        assert_eq!(rows[1]["coinsProcessed"], 1);
        let keys: Vec<&String> = rows[0].as_object().unwrap().keys().collect();
        let differing: Vec<&&String> = keys
            .iter()
            .filter(|k| rows[0][k.as_str()] != rows[1][k.as_str()])
            .collect();
        assert_eq!(
            differing.len(),
            1,
            "only coinsProcessed may move: {differing:?}"
        );
        assert_eq!(differing[0].as_str(), "coinsProcessed");
    }

    /// Counts with a sold row present, under both `count` modes. `count`'s FIFA 17
    /// meaning is unknown, so it is a controlled variable — never a guess.
    #[tokio::test]
    async fn counts_report_sold_and_count_mode_is_controlled() {
        let (store, _d) = store_at("soldcounts").await;
        seed_sold(&store, "900000204", 150).await;
        seed_listing(&store, "900000205", 500).await; // one still active

        let active_mode = SoldExperiment::from_values(Some("highest"), None, Some("active"));
        let c = parse(&handle_market_counts(&store, active_mode).await);
        assert_eq!(c["selling"], 1, "one live auction");
        assert_eq!(c["sold"], 1, "one uncleared sale");
        assert_eq!(c["count"], 1, "active mode: count == selling");
        assert_eq!(c["maxAuctionsAllowed"], 100);
        assert_eq!(c["offered"], 0);

        let both = SoldExperiment::from_values(Some("highest"), None, Some("active_plus_sold"));
        let c2 = parse(&handle_market_counts(&store, both).await);
        assert_eq!(c2["selling"], 1);
        assert_eq!(c2["sold"], 1);
        assert_eq!(c2["count"], 2, "membership mode: count == selling + sold");
    }

    /// The bulk clear verb clears sold rows and nothing else, and it is
    /// PRESENTATION ONLY: it must not touch the economy or resurrect ownership.
    #[tokio::test]
    async fn clear_sold_removes_only_sold_rows_and_moves_no_coins() {
        let (store, _d) = store_at("soldclear").await;
        let econ = CountingEconomy::with_balance(7_777);
        seed_sold(&store, "900000206", 150).await;
        seed_listing(&store, "900000207", 500).await;
        let exp = SoldExperiment::from_values(Some("highest"), None, None);

        assert_eq!(store.uncleared_sold().await.unwrap().len(), 1);
        let resp = handle_market_clear_sold(&store).await;
        assert_eq!(parse(&resp), json!({}), "the client parses nothing");

        assert_eq!(store.uncleared_sold().await.unwrap().len(), 0, "cleared");
        // The active auction is untouched, and the sold LISTING still exists as
        // history — clearing is an acknowledgement, not a deletion of the sale.
        assert_eq!(store.query_listings("active").await.unwrap().len(), 1);
        assert_eq!(store.get_listing("900000206").await.unwrap().state, "sold");
        let pile = parse(&handle_market_query("active", &econ, &store, exp).await);
        assert_eq!(
            pile["auctionInfo"].as_array().unwrap().len(),
            1,
            "only the active one"
        );
        let c = parse(&handle_market_counts(&store, exp).await);
        assert_eq!(c["sold"], 0, "the sold bucket empties on clear");
        assert_eq!(
            econ.purchase_calls.load(Ordering::SeqCst),
            0,
            "no economy call"
        );
        assert_eq!(econ.balance().unwrap(), 7_777, "clearing moves no coins");
    }

    /// Clearing twice must be a no-op, because the client may retry.
    #[tokio::test]
    async fn clearing_sold_twice_is_idempotent() {
        let (store, _d) = store_at("soldclear2").await;
        seed_sold(&store, "900000208", 150).await;
        assert_eq!(
            store.clear_sold().await.unwrap(),
            1,
            "first clear does work"
        );
        assert_eq!(
            store.clear_sold().await.unwrap(),
            0,
            "second clears nothing"
        );
        assert_eq!(store.get_listing("900000208").await.unwrap().state, "sold");
    }

    /// A sold row must also answer its own status poll, or the Transfer List would
    /// render a row from /tradePile and then be told it does not exist.
    #[tokio::test]
    async fn sold_row_answers_its_status_poll() {
        let (store, _d) = store_at("soldstatus").await;
        let econ = CountingEconomy::with_balance(10_000);
        seed_sold(&store, "900000209", 150).await;
        let exp = SoldExperiment::from_values(Some("buyNow"), Some("1"), None);

        let one =
            parse(&handle_market_status(Some("tradeIds=900000209"), &econ, &store, exp).await);
        let recs = one["auctionInfo"].as_array().unwrap();
        assert_eq!(recs.len(), 1);
        assert_eq!(recs[0]["tradeState"], "closed");
        assert_eq!(recs[0]["bidState"], "buyNow");
        assert_eq!(recs[0]["coinsProcessed"], 1);

        let all = parse(&handle_market_status(None, &econ, &store, exp).await);
        assert_eq!(
            all["auctionInfo"].as_array().unwrap().len(),
            1,
            "unfiltered too"
        );
    }

    /// `mark_sold` is the sale transition and must happen at most once, so a
    /// duplicated counterparty settlement cannot double-sell.
    #[tokio::test]
    async fn mark_sold_is_once_only() {
        let (store, _d) = store_at("soldonce").await;
        seed_listing(&store, "900000210", 150).await;
        assert!(store.mark_sold("900000210").await.unwrap(), "first wins");
        assert!(
            !store.mark_sold("900000210").await.unwrap(),
            "second is refused"
        );
        assert!(
            !store.mark_sold("nosuchlisting").await.unwrap(),
            "unknown id"
        );
        assert_eq!(
            store.uncleared_sold().await.unwrap().len(),
            1,
            "one sold row"
        );
    }
}