economy(fifa17): land Store + Market writer handlers + pack generator (unrouted)

Implements the FIFA17 economy WRITER cluster on top of the landed Core
economy authority + host CoreEconomy client + identity/item-shaper infra.
Handlers are pub, unit-tested, and NOT yet routed: classify() and
ROUTE_AUTHORITY are untouched — the classifier barrier is a later single
coherent flip. No stubs; real Core-backed behavior; fail-closed on CoreError.

Pack generator (adapter fut/pack_content.rs):
  generate_pack_contents(&PackDef, &mut impl Rng, &[GeneratedCandidate])
  -> Vec<GeneratedCard>. Pure, seeded (deterministic), gold-tier split +
  special_chance gate as documented OPENFUT PLACEHOLDER policy (Python
  open_pack/_pack_body parity note inline). Fail-closed empty on empty pool.

Store/item writers (host economy_store.rs), matching oracle wire shapes:
  - handle_store_buy   PUT /store/transaction -> purchase_items (debit+mint N)
    -> createPackResponse; cancel/unknown/owned_only -> 200 {}; insufficient
    -> 461 {reason,credits}; CoreError -> 503.
  - handle_pack_open   POST /purchased -> owned_only consumes the unopened
    entitlement (redeem_entitlement, consume-once); normal packs debit+mint.
  - handle_quick_sell{_path,_body}  DELETE .../item/<id> + POST /ut/delete/.../item
    -> reverse-resolve wire->Core id (SquadWireResolver) -> sell_item ->
    {items:[{id}],totalCredits}; not-owned skipped.
  Production OwnedItemLookup = CoreItemLookup over CoreAccess.

Market (host market_store.rs / pile_store.rs / market.rs), synthetic-seller:
  - MarketStore over sqlx SQLite (WAL-once + busy_timeout=5s + BEGIN IMMEDIATE
    for writes, mirroring openfut-core::db). listings(active/reserved/sold/
    cancelled), owner-checked cancel, CAS reserve/complete_sale/rollback.
    Typed errors NotFound/Sold/Cancelled/WrongOwner/Conflict.
  - PileStore: durable pile/location metadata keyed by Core item id.
  - handle_market_{list,query,cancel,buy} + handle_move_items. Buy-now =
    reserve (CAS) -> balance precheck (461) -> Core purchase_item (mint+debit)
    -> complete_sale; any Core failure rolls the reservation back active.
    Two concurrent buyers -> exactly one sale + one debit.

Deps (additive): rand 0.8 (adapter+host), sqlx 0.7 sqlite/runtime-tokio (host).
Tests: adapter +7 (pack_content), host +43 (economy_store 20, market/store 23
incl two_reservers_exactly_one_wins, two_buyers_exactly_one_sale_one_debit,
state_survives_reopen, move_persists_across_reopen). All green; clippy
-D warnings clean; rustfmt clean.

Cargo.lock

@@ -3113,6 +3113,7 @@ name = "openfut-adapter-fifa17"
 version = "0.1.0"
 dependencies = [
  "openfut-protocol-blaze",
+ "rand",
  "serde",
  "serde_json",
 ]
@@ -3285,8 +3286,10 @@ dependencies = [
  "openfut-http",
  "openfut-identity",
  "parking_lot",
+ "rand",
  "reqwest",
  "serde_json",
+ "sqlx",
  "tokio",
 ]
 

openfut-adapter-fifa17/Cargo.toml

@@ -15,6 +15,10 @@ openfut-protocol-blaze = { path = "../openfut-protocol-blaze" }
 # streak would be reinventing a solved problem in the riskiest possible place.
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+# Seeded RNG for the Store pack-content generator (`fut::pack_content`). The
+# generator is pure over an injected `rand::Rng`, so packs are deterministic
+# under a seeded `StdRng` in tests and reproducible in production.
+rand = "0.8"
 
 [dev-dependencies]
 # Differential fixtures are JSONL; the runtime dependency already covers it.

openfut-adapter-fifa17/src/fut/mod.rs

@@ -11,6 +11,7 @@ pub mod economy_policy;
 pub mod entities;
 pub mod item;
 pub mod owned_query;
+pub mod pack_content;
 pub mod squad;
 pub mod squad_ext;
 pub mod squad_projection;

openfut-adapter-fifa17/src/fut/pack_content.rs

@@ -0,0 +1,254 @@
+//! FIFA 17 Store pack-content generator (pure, seeded).
+//!
+//! Draws the cards a Store pack awards. It is a **pure function** of
+//! `(pack definition, RNG, candidate pool)` — no IO, no Core, no catalogue
+//! lookup — so it is deterministic under a seeded [`rand::Rng`] and trivially
+//! unit-tested. The host owns the impure parts: it builds the candidate pool
+//! (only card ids that resolve in BOTH the FIFA catalogue and Core content),
+//! mints the drawn cards into Core, and shapes them onto the wire.
+//!
+//! ## Parity note — Python `open_pack` / `_pack_body`
+//! (`fifa17-recon/tools/fut_store.py:689`, `utas_server.py:3474`)
+//! 1. `open_pack(price, count, gold, tiers, special_chance)` deducts coins then
+//!    draws `count` items (mostly players); the reveal body wraps them verbatim.
+//! 2. Non-tiered draws split the pool at rating 75 by `gold` (`p[1] >= 75 == gold`)
+//!    and fall back to the whole pool when that tier is empty (`... or PACK_POOL`).
+//! 3. Each drawn player becomes a special with probability `special_chance`
+//!    (`random.random() < special_chance`).
+//! 4. `FUT_PACK_MIX` swaps ~`count // 4` players for consumables/staff extras;
+//!    we deliberately OMIT that mix (Core candidates are player defs — players-only).
+//! 5. Prices/counts/odds are the OpenFUT **PLACEHOLDER** economy (the audit found
+//!    them invented); only the wire *shape* is EA-observed/oracle-verified.
+//! 6. This port reproduces the count + gold-tier split + `special_chance` gate as
+//!    that same PLACEHOLDER policy, drawing with replacement from the pool.
+
+use rand::Rng;
+
+use crate::fut::store_catalog::PackDef;
+
+/// A candidate the host has already verified resolves in BOTH the FIFA catalogue
+/// and Core content. Carries the full Core definition the shaper needs plus the
+/// two draw-policy annotations (`gold` tier, `special` version) the host derives
+/// from the catalogue (keeping this generator pure — it never reads a catalogue).
+#[derive(Debug, Clone)]
+pub struct GeneratedCandidate {
+    /// Core card-definition id (resolves in the FIFA catalogue and Core content).
+    pub card_id: String,
+    pub rating: u8,
+    pub position: String,
+    pub nation: String,
+    pub league: String,
+    pub club: String,
+    /// [pace, shooting, passing, dribbling, defending, physical].
+    pub attributes: [u8; 6],
+    /// Gold tier (host derives this as `rating >= 75`, the oracle's split point).
+    pub gold: bool,
+    /// Special version available (host derives this from the catalogue rareflag
+    /// `> 1`); gated by [`PackDef::special_chance`].
+    pub special: bool,
+}
+
+impl GeneratedCandidate {
+    fn to_card(&self) -> GeneratedCard {
+        GeneratedCard {
+            card_id: self.card_id.clone(),
+            rating: self.rating,
+            position: self.position.clone(),
+            nation: self.nation.clone(),
+            league: self.league.clone(),
+            club: self.club.clone(),
+            attributes: self.attributes,
+        }
+    }
+}
+
+/// One card a pack awarded. Carries `card_id` (the Core definition the host mints
+/// and shapes) plus the definition fields the shared item shaper needs. It is
+/// NOT an owned instance yet — the host mints the Core instance id and allocates
+/// the numeric wire id.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GeneratedCard {
+    pub card_id: String,
+    pub rating: u8,
+    pub position: String,
+    pub nation: String,
+    pub league: String,
+    pub club: String,
+    pub attributes: [u8; 6],
+}
+
+/// Draw `pack.count` cards from `pool` with the injected RNG. Pure and
+/// deterministic under a seeded RNG. Returns an empty `Vec` (fail-closed) when
+/// the pool is empty or the pack awards no cards.
+///
+/// Policy (PLACEHOLDER — see the module parity note): draw with replacement from
+/// the pack's tier (`gold`), biasing each draw toward a special card with
+/// probability `special_chance`. An empty tier or partition falls back to the
+/// next-wider set so a draw is always possible when the pool is non-empty.
+pub fn generate_pack_contents(
+    pack: &PackDef,
+    rng: &mut impl Rng,
+    pool: &[GeneratedCandidate],
+) -> Vec<GeneratedCard> {
+    if pool.is_empty() || pack.count == 0 {
+        return Vec::new();
+    }
+    // Tier split: a gold pack draws gold-tier candidates, a non-gold pack draws
+    // non-gold; an empty tier falls back to the whole pool (oracle `... or POOL`).
+    let tier: Vec<&GeneratedCandidate> = pool.iter().filter(|c| c.gold == pack.gold).collect();
+    let tier: Vec<&GeneratedCandidate> = if tier.is_empty() {
+        pool.iter().collect()
+    } else {
+        tier
+    };
+    // Partition the tier by special so `special_chance` can bias a draw; either
+    // partition falls back to the whole tier when empty.
+    let special: Vec<&GeneratedCandidate> = tier.iter().copied().filter(|c| c.special).collect();
+    let normal: Vec<&GeneratedCandidate> = tier.iter().copied().filter(|c| !c.special).collect();
+    let chance = pack.special_chance.clamp(0.0, 1.0);
+
+    let mut out = Vec::with_capacity(pack.count as usize);
+    for _ in 0..pack.count {
+        let want_special = chance > 0.0 && rng.gen_bool(chance);
+        let sub: &[&GeneratedCandidate] = if want_special && !special.is_empty() {
+            &special
+        } else if !want_special && !normal.is_empty() {
+            &normal
+        } else {
+            &tier
+        };
+        let pick = sub[rng.gen_range(0..sub.len())];
+        out.push(pick.to_card());
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rand::rngs::StdRng;
+    use rand::SeedableRng;
+
+    fn cand(card: &str, rating: u8, gold: bool, special: bool) -> GeneratedCandidate {
+        GeneratedCandidate {
+            card_id: card.into(),
+            rating,
+            position: "ST".into(),
+            nation: "Brazil".into(),
+            league: "Premier League".into(),
+            club: "Arsenal".into(),
+            attributes: [rating; 6],
+            gold,
+            special,
+        }
+    }
+
+    /// A mixed pool: gold specials, gold normals, and a bronze tier.
+    fn pool() -> Vec<GeneratedCandidate> {
+        vec![
+            cand("g-sp-1", 90, true, true),
+            cand("g-sp-2", 88, true, true),
+            cand("g-1", 84, true, false),
+            cand("g-2", 82, true, false),
+            cand("g-3", 79, true, false),
+            cand("b-1", 64, false, false),
+            cand("b-2", 62, false, false),
+        ]
+    }
+
+    fn pack(id: u64, count: u64, gold: bool, special_chance: f64) -> PackDef {
+        PackDef {
+            id,
+            name: "Test Pack",
+            price: 1000,
+            count,
+            gold,
+            special_chance,
+            owned_only: false,
+        }
+    }
+
+    #[test]
+    fn same_seed_same_output() {
+        let pool = pool();
+        let p = pack(5, 7, true, 0.3);
+        let mut a = StdRng::seed_from_u64(42);
+        let mut b = StdRng::seed_from_u64(42);
+        assert_eq!(
+            generate_pack_contents(&p, &mut a, &pool),
+            generate_pack_contents(&p, &mut b, &pool)
+        );
+    }
+
+    #[test]
+    fn different_seeds_can_diverge() {
+        let pool = pool();
+        let p = pack(5, 7, true, 0.3);
+        let a = generate_pack_contents(&p, &mut StdRng::seed_from_u64(1), &pool);
+        let b = generate_pack_contents(&p, &mut StdRng::seed_from_u64(999), &pool);
+        // Not a hard guarantee, but with this pool/count the two seeds differ.
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn count_is_exact_and_all_cards_from_pool() {
+        let pool = pool();
+        let ids: std::collections::HashSet<&str> =
+            pool.iter().map(|c| c.card_id.as_str()).collect();
+        for &n in &[1u64, 5, 7, 11] {
+            let p = pack(6, n, true, 0.08);
+            let cards = generate_pack_contents(&p, &mut StdRng::seed_from_u64(n), &pool);
+            assert_eq!(cards.len() as u64, n);
+            for c in &cards {
+                assert!(
+                    ids.contains(c.card_id.as_str()),
+                    "drew unknown card {}",
+                    c.card_id
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn gold_pack_draws_only_gold_tier() {
+        let pool = pool();
+        let p = pack(5, 20, true, 0.03);
+        let cards = generate_pack_contents(&p, &mut StdRng::seed_from_u64(7), &pool);
+        assert!(
+            cards.iter().all(|c| c.rating >= 75),
+            "gold pack drew a bronze card"
+        );
+    }
+
+    #[test]
+    fn bronze_pack_draws_only_bronze_tier() {
+        let pool = pool();
+        let p = pack(1, 20, false, 0.005);
+        let cards = generate_pack_contents(&p, &mut StdRng::seed_from_u64(7), &pool);
+        assert!(
+            cards.iter().all(|c| c.rating < 75),
+            "bronze pack drew a gold card"
+        );
+    }
+
+    #[test]
+    fn special_chance_one_draws_only_specials() {
+        let pool = pool();
+        let special_ids: std::collections::HashSet<&str> = pool
+            .iter()
+            .filter(|c| c.special)
+            .map(|c| c.card_id.as_str())
+            .collect();
+        let p = pack(7, 11, true, 1.0);
+        let cards = generate_pack_contents(&p, &mut StdRng::seed_from_u64(3), &pool);
+        assert!(cards
+            .iter()
+            .all(|c| special_ids.contains(c.card_id.as_str())));
+    }
+
+    #[test]
+    fn empty_pool_fails_closed() {
+        let p = pack(5, 7, true, 0.03);
+        assert!(generate_pack_contents(&p, &mut StdRng::seed_from_u64(1), &[]).is_empty());
+    }
+}

openfut-utas-host/Cargo.toml

@@ -14,6 +14,15 @@ serde_json = "1"
 # Plain-HTTP client for Core queries and Python passthrough. UTAS is plaintext
 # HTTP (worker D: no wrap_socket, no cert), so no TLS backend is linked.
 reqwest = { version = "0.11", default-features = false, features = ["blocking", "json"] }
+# Durable FIFA-specific market listing + item-pile state. These are host-owned
+# FIFA policy stores (NOT generic Core inventory), backed by their own SQLite
+# file, opened exactly like openfut-core/src/db.rs::init_pool (WAL-once +
+# foreign_keys + busy_timeout, BEGIN IMMEDIATE for writes).
+sqlx = { version = "0.7", features = ["sqlite", "runtime-tokio"] }
+# Seeded RNG for the Store pack-content generator injected into the economy
+# writer handlers (`economy_store`). Production seeds it from entropy (so minted
+# Core instance ids never collide); tests seed a fixed value for reproducibility.
+rand = "0.8"
 
 [dev-dependencies]
 parking_lot = "0.12"

openfut-utas-host/src/lib.rs

@@ -35,6 +35,10 @@
 //! `{"itemData":[]}` — the honest state until Core inventory is asset-backed.
 
 pub mod config;
+pub mod economy_store;
+pub mod market;
+pub mod market_store;
+pub mod pile_store;
 
 use std::io::{BufRead, BufReader, Write};
 use std::net::{TcpListener, TcpStream};

openfut-utas-host/src/market.rs

@@ -0,0 +1,757 @@
+//! 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 card definition when numeric; fall back to the wire
+    // item id. Never a fabricated FIFA asset — 0 means "no art", a valid int.
+    let resource = l
+        .card_id
+        .parse::<i64>()
+        .ok()
+        .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,
+    })
+}
+
+/// 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 {
+    econ.balance().unwrap_or(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 lists an owned club item and returns `{"id": tradeId}`; the listing is
+///   persisted so a later buy/cancel is durable.
+/// * PUT (relist-all) is an ack `{}`.
+pub async fn handle_market_list(
+    method: &str,
+    body: &[u8],
+    econ: &dyn CoreEconomy,
+    store: &MarketStore,
+) -> 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 }));
+            };
+            // Card definition, if the client sent the full item; else the wire id
+            // string (a user listing does not drive the synthetic-seller mint).
+            let card_id = item_data
+                .and_then(|d| d.get("resourceId"))
+                .and_then(Value::as_i64)
+                .map(|r| r.to_string())
+                .unwrap_or_else(|| item_id.to_string());
+            // 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,
+                    &card_id,
+                    None,
+                    Some(item_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 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 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,
+    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()
+        }
+    }
+
+    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, 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).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).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).await;
+        assert_eq!(resp.status, 200);
+        assert_eq!(parse(&resp), json!({}));
+    }
+
+    #[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")
+        );
+    }
+}

openfut-utas-host/src/market_store.rs

@@ -0,0 +1,608 @@
+//! Durable FIFA 17 transfer-market **listing** store.
+//!
+//! This is host-owned FIFA policy state, NOT generic Core inventory. Core stays
+//! the sole authority for coins and item ownership; the market layer owns only
+//! the durable *listing* lifecycle (who is selling what, at what price, and in
+//! which state). It is backed by its own SQLite file so it survives restart.
+//!
+//! ## Concurrency model (load-bearing)
+//!
+//! The pool is opened exactly like [`openfut_core::db::init_pool`]: WAL is
+//! established once on the file before the pool opens, every pooled connection
+//! carries `foreign_keys=ON` and a 5s `busy_timeout`, and **every write runs
+//! inside a `BEGIN IMMEDIATE` transaction**. Immediate transactions take the
+//! write lock up front, so the reserve compare-and-swap is genuinely atomic
+//! across connections — the exact class of bug (deferred transactions racing a
+//! read-then-write) that was just fixed in Core. Two buyers reserving the same
+//! active listing therefore resolve to exactly one winner.
+//!
+//! ## State machine
+//!
+//! ```text
+//!   active ──reserve──▶ reserved ──complete_sale──▶ sold
+//!     │                    │
+//!     │                    └──rollback_reservation──▶ active
+//!     └──cancel──▶ cancelled
+//! ```
+//!
+//! `reserved` is a real state (a listing being paid for), so it is part of the
+//! `CHECK` constraint even though it is a transient intermediate — omitting it
+//! would make [`MarketStore::reserve_listing`] fail the constraint.
+
+use std::time::Duration;
+
+use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
+use sqlx::{ConnectOptions, Connection, Row, SqlitePool};
+
+/// Typed failure of a listing operation. `Db` wraps an infrastructure error
+/// (transport/encoding); everything else is a modelled lifecycle outcome.
+#[derive(Debug)]
+pub enum MarketError {
+    /// No listing with that id exists.
+    NotFound,
+    /// The listing has already been sold.
+    Sold,
+    /// The listing has already been cancelled.
+    Cancelled,
+    /// The caller is not the owner of the listing.
+    WrongOwner,
+    /// The listing was not in the state the transition required (e.g. a
+    /// reserved listing asked to cancel, or a duplicate id on insert).
+    Conflict,
+    /// SQLite / transport failure.
+    Db(String),
+}
+
+impl std::fmt::Display for MarketError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MarketError::NotFound => write!(f, "listing not found"),
+            MarketError::Sold => write!(f, "listing already sold"),
+            MarketError::Cancelled => write!(f, "listing already cancelled"),
+            MarketError::WrongOwner => write!(f, "listing owned by another seller"),
+            MarketError::Conflict => write!(f, "listing state conflict"),
+            MarketError::Db(e) => write!(f, "market store db error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for MarketError {}
+
+fn db(e: sqlx::Error) -> MarketError {
+    MarketError::Db(e.to_string())
+}
+
+/// One transfer-market listing row.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Listing {
+    pub listing_id: String,
+    pub card_id: String,
+    /// Set for a seller-listed owned item; `None` for a synthetic-seller
+    /// listing (the buy path mints a fresh Core item instead of transferring).
+    pub core_item_id: Option<String>,
+    /// The FIFA wire item id of a seller-listed owned item, if any.
+    pub wire_item_id: Option<i64>,
+    pub start_price: i64,
+    pub buy_now_price: i64,
+    /// Opaque seller identity; `None` for synthetic listings.
+    pub owner: Option<String>,
+    /// `active` | `reserved` | `sold` | `cancelled`.
+    pub state: String,
+    /// Creation time, unix-epoch milliseconds as a string (sortable).
+    pub created_at: String,
+}
+
+const CREATE_LISTINGS: &str = "CREATE TABLE IF NOT EXISTS listings (
+    listing_id    TEXT PRIMARY KEY,
+    card_id       TEXT NOT NULL,
+    core_item_id  TEXT,
+    wire_item_id  INTEGER,
+    start_price   INTEGER NOT NULL,
+    buy_now_price INTEGER NOT NULL,
+    owner         TEXT,
+    state         TEXT NOT NULL CHECK (state IN ('active','reserved','sold','cancelled')),
+    created_at    TEXT NOT NULL
+)";
+
+fn now_millis() -> String {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis())
+        .unwrap_or(0)
+        .to_string()
+}
+
+fn row_to_listing(row: &sqlx::sqlite::SqliteRow) -> Listing {
+    Listing {
+        listing_id: row.get("listing_id"),
+        card_id: row.get("card_id"),
+        core_item_id: row.get("core_item_id"),
+        wire_item_id: row.get("wire_item_id"),
+        start_price: row.get("start_price"),
+        buy_now_price: row.get("buy_now_price"),
+        owner: row.get("owner"),
+        state: row.get("state"),
+        created_at: row.get("created_at"),
+    }
+}
+
+/// Durable listing store over an sqlx SQLite pool. Cheap to clone (the pool is
+/// an `Arc` internally), so the same store can be shared across tasks.
+#[derive(Clone)]
+pub struct MarketStore {
+    pool: SqlitePool,
+}
+
+impl MarketStore {
+    /// Open (creating if missing) the market DB at `path`, mirroring Core's
+    /// `init_pool`: establish WAL once on the file, then open a multi-connection
+    /// pool where every connection carries foreign_keys + a busy_timeout.
+    pub async fn open(path: &str) -> Result<Self, MarketError> {
+        let opts = SqliteConnectOptions::new()
+            .filename(path)
+            .create_if_missing(true)
+            .journal_mode(SqliteJournalMode::Wal)
+            .foreign_keys(true)
+            .busy_timeout(Duration::from_secs(5));
+        // Establish WAL on the file via ONE connection BEFORE the pool opens, so
+        // pooled connections only ever re-assert an already-WAL file (see Core).
+        {
+            let mut conn = opts.clone().connect().await.map_err(db)?;
+            sqlx::query("PRAGMA journal_mode=WAL")
+                .execute(&mut conn)
+                .await
+                .map_err(db)?;
+            conn.close().await.map_err(db)?;
+        }
+        let pool = SqlitePoolOptions::new()
+            .max_connections(8)
+            .connect_with(opts)
+            .await
+            .map_err(db)?;
+        sqlx::query(CREATE_LISTINGS)
+            .execute(&pool)
+            .await
+            .map_err(db)?;
+        Ok(MarketStore { pool })
+    }
+
+    /// Insert a new `active` listing. `listing_id` is the numeric-string trade id
+    /// the client keys the auction on (the caller allocates it). Duplicate id ->
+    /// [`MarketError::Conflict`].
+    #[allow(clippy::too_many_arguments)]
+    pub async fn create_listing(
+        &self,
+        listing_id: &str,
+        card_id: &str,
+        core_item_id: Option<&str>,
+        wire_item_id: Option<i64>,
+        start_price: i64,
+        buy_now_price: i64,
+        owner: Option<&str>,
+    ) -> Result<Listing, MarketError> {
+        let created_at = now_millis();
+        let mut conn = self.pool.acquire().await.map_err(db)?;
+        sqlx::query("BEGIN IMMEDIATE")
+            .execute(&mut *conn)
+            .await
+            .map_err(db)?;
+        let res = sqlx::query(
+            "INSERT INTO listings (listing_id, card_id, core_item_id, wire_item_id, \
+             start_price, buy_now_price, owner, state, created_at) \
+             VALUES (?, ?, ?, ?, ?, ?, ?, 'active', ?)",
+        )
+        .bind(listing_id)
+        .bind(card_id)
+        .bind(core_item_id)
+        .bind(wire_item_id)
+        .bind(start_price)
+        .bind(buy_now_price)
+        .bind(owner)
+        .bind(&created_at)
+        .execute(&mut *conn)
+        .await;
+        match res {
+            Ok(_) => {
+                sqlx::query("COMMIT")
+                    .execute(&mut *conn)
+                    .await
+                    .map_err(db)?;
+                Ok(Listing {
+                    listing_id: listing_id.to_string(),
+                    card_id: card_id.to_string(),
+                    core_item_id: core_item_id.map(str::to_string),
+                    wire_item_id,
+                    start_price,
+                    buy_now_price,
+                    owner: owner.map(str::to_string),
+                    state: "active".to_string(),
+                    created_at,
+                })
+            }
+            Err(e) => {
+                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
+                // A PK clash is a caller-level conflict, not an infra failure.
+                if matches!(&e, sqlx::Error::Database(dbe) if dbe.is_unique_violation()) {
+                    Err(MarketError::Conflict)
+                } else {
+                    Err(db(e))
+                }
+            }
+        }
+    }
+
+    /// Fetch one listing, or [`MarketError::NotFound`].
+    pub async fn get_listing(&self, listing_id: &str) -> Result<Listing, MarketError> {
+        let row = sqlx::query("SELECT * FROM listings WHERE listing_id = ?")
+            .bind(listing_id)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(db)?;
+        row.as_ref()
+            .map(row_to_listing)
+            .ok_or(MarketError::NotFound)
+    }
+
+    /// All listings in `state`, oldest first.
+    pub async fn query_listings(&self, state: &str) -> Result<Vec<Listing>, MarketError> {
+        let rows = sqlx::query("SELECT * FROM listings WHERE state = ? ORDER BY created_at ASC")
+            .bind(state)
+            .fetch_all(&self.pool)
+            .await
+            .map_err(db)?;
+        Ok(rows.iter().map(row_to_listing).collect())
+    }
+
+    /// Atomic compare-and-swap of a single listing's state inside a
+    /// `BEGIN IMMEDIATE` transaction. `Ok(true)` = the row was in `from` and is
+    /// now `to`; `Ok(false)` = the row exists but was not in `from` (lost race /
+    /// wrong state); `Err(NotFound)` = no such row.
+    async fn cas(&self, listing_id: &str, from: &str, to: &str) -> Result<bool, MarketError> {
+        let mut conn = self.pool.acquire().await.map_err(db)?;
+        sqlx::query("BEGIN IMMEDIATE")
+            .execute(&mut *conn)
+            .await
+            .map_err(db)?;
+        let outcome: Result<bool, MarketError> = async {
+            let current: Option<String> =
+                sqlx::query("SELECT state FROM listings WHERE listing_id = ?")
+                    .bind(listing_id)
+                    .fetch_optional(&mut *conn)
+                    .await
+                    .map_err(db)?
+                    .map(|r| r.get::<String, _>("state"));
+            match current {
+                None => Err(MarketError::NotFound),
+                Some(s) if s == from => {
+                    sqlx::query("UPDATE listings SET state = ? WHERE listing_id = ? AND state = ?")
+                        .bind(to)
+                        .bind(listing_id)
+                        .bind(from)
+                        .execute(&mut *conn)
+                        .await
+                        .map_err(db)?;
+                    Ok(true)
+                }
+                Some(_) => Ok(false),
+            }
+        }
+        .await;
+        match &outcome {
+            Ok(_) => {
+                sqlx::query("COMMIT")
+                    .execute(&mut *conn)
+                    .await
+                    .map_err(db)?;
+            }
+            Err(_) => {
+                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
+            }
+        }
+        outcome
+    }
+
+    /// Reserve an `active` listing (`active -> reserved`). Returns whether this
+    /// caller won the reservation. Exactly one of two concurrent callers wins.
+    pub async fn reserve_listing(&self, listing_id: &str) -> Result<bool, MarketError> {
+        self.cas(listing_id, "active", "reserved").await
+    }
+
+    /// Finalise a won reservation (`reserved -> sold`). A listing not in
+    /// `reserved` is a [`MarketError::Conflict`].
+    pub async fn complete_sale(&self, listing_id: &str) -> Result<(), MarketError> {
+        if self.cas(listing_id, "reserved", "sold").await? {
+            Ok(())
+        } else {
+            Err(MarketError::Conflict)
+        }
+    }
+
+    /// Undo a reservation on a downstream failure (`reserved -> active`), so the
+    /// listing becomes buyable again. Not in `reserved` -> [`MarketError::Conflict`].
+    pub async fn rollback_reservation(&self, listing_id: &str) -> Result<(), MarketError> {
+        if self.cas(listing_id, "reserved", "active").await? {
+            Ok(())
+        } else {
+            Err(MarketError::Conflict)
+        }
+    }
+
+    /// Cancel an `active` listing once (`active -> cancelled`). If `owner` is
+    /// supplied it must match the listing's owner. Returns typed errors for
+    /// every non-active state so a double cancel is observable.
+    pub async fn cancel_listing(
+        &self,
+        listing_id: &str,
+        owner: Option<&str>,
+    ) -> Result<(), MarketError> {
+        let mut conn = self.pool.acquire().await.map_err(db)?;
+        sqlx::query("BEGIN IMMEDIATE")
+            .execute(&mut *conn)
+            .await
+            .map_err(db)?;
+        let outcome: Result<(), MarketError> = async {
+            let row = sqlx::query("SELECT state, owner FROM listings WHERE listing_id = ?")
+                .bind(listing_id)
+                .fetch_optional(&mut *conn)
+                .await
+                .map_err(db)?;
+            let row = row.ok_or(MarketError::NotFound)?;
+            let state: String = row.get("state");
+            let stored_owner: Option<String> = row.get("owner");
+            if let Some(want) = owner {
+                if stored_owner.as_deref() != Some(want) {
+                    return Err(MarketError::WrongOwner);
+                }
+            }
+            match state.as_str() {
+                "active" => {
+                    sqlx::query(
+                        "UPDATE listings SET state = 'cancelled' \
+                         WHERE listing_id = ? AND state = 'active'",
+                    )
+                    .bind(listing_id)
+                    .execute(&mut *conn)
+                    .await
+                    .map_err(db)?;
+                    Ok(())
+                }
+                "sold" => Err(MarketError::Sold),
+                "cancelled" => Err(MarketError::Cancelled),
+                _ => Err(MarketError::Conflict),
+            }
+        }
+        .await;
+        match &outcome {
+            Ok(_) => {
+                sqlx::query("COMMIT")
+                    .execute(&mut *conn)
+                    .await
+                    .map_err(db)?;
+            }
+            Err(_) => {
+                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
+            }
+        }
+        outcome
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::sync::Arc;
+
+    /// A unique temp DB path that deletes its file (and WAL/SHM sidecars) on
+    /// drop. Holding the guard keeps the file alive across store reopens.
+    struct TempDb(String);
+    impl TempDb {
+        fn new() -> 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-{}-{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));
+            }
+        }
+    }
+
+    async fn temp_store() -> (MarketStore, TempDb) {
+        let db = TempDb::new();
+        let store = MarketStore::open(db.path()).await.unwrap();
+        (store, db)
+    }
+
+    async fn seed(store: &MarketStore, id: &str) -> Listing {
+        store
+            .create_listing(id, "card_pl_001", None, None, 900, 2500, None)
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn create_get_query_roundtrip() {
+        let (store, _d) = temp_store().await;
+        let created = seed(&store, "900000001").await;
+        assert_eq!(created.state, "active");
+        assert_eq!(created.buy_now_price, 2500);
+
+        let got = store.get_listing("900000001").await.unwrap();
+        assert_eq!(got, created);
+
+        assert!(matches!(
+            store.get_listing("nope").await,
+            Err(MarketError::NotFound)
+        ));
+
+        let active = store.query_listings("active").await.unwrap();
+        assert_eq!(active.len(), 1);
+        assert!(store.query_listings("sold").await.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn duplicate_id_is_conflict() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "900000001").await;
+        assert!(matches!(
+            store
+                .create_listing("900000001", "card_pl_002", None, None, 1, 2, None)
+                .await,
+            Err(MarketError::Conflict)
+        ));
+    }
+
+    #[tokio::test]
+    async fn reserve_complete_lifecycle() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "900000001").await;
+        assert!(store.reserve_listing("900000001").await.unwrap());
+        // Second reserve of a now-reserved listing loses.
+        assert!(!store.reserve_listing("900000001").await.unwrap());
+        store.complete_sale("900000001").await.unwrap();
+        assert_eq!(store.get_listing("900000001").await.unwrap().state, "sold");
+        // Completing again (not reserved) is a conflict.
+        assert!(matches!(
+            store.complete_sale("900000001").await,
+            Err(MarketError::Conflict)
+        ));
+    }
+
+    #[tokio::test]
+    async fn rollback_restores_active() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "900000001").await;
+        assert!(store.reserve_listing("900000001").await.unwrap());
+        store.rollback_reservation("900000001").await.unwrap();
+        assert_eq!(
+            store.get_listing("900000001").await.unwrap().state,
+            "active"
+        );
+        // Buyable again after rollback.
+        assert!(store.reserve_listing("900000001").await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn cancel_once_then_errors() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "900000001").await;
+        store.cancel_listing("900000001", None).await.unwrap();
+        assert_eq!(
+            store.get_listing("900000001").await.unwrap().state,
+            "cancelled"
+        );
+        assert!(matches!(
+            store.cancel_listing("900000001", None).await,
+            Err(MarketError::Cancelled)
+        ));
+    }
+
+    #[tokio::test]
+    async fn cancel_checks_owner() {
+        let (store, _d) = temp_store().await;
+        store
+            .create_listing(
+                "900000001",
+                "card_pl_001",
+                None,
+                None,
+                900,
+                2500,
+                Some("alice"),
+            )
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.cancel_listing("900000001", Some("mallory")).await,
+            Err(MarketError::WrongOwner)
+        ));
+        store
+            .cancel_listing("900000001", Some("alice"))
+            .await
+            .unwrap();
+        assert_eq!(
+            store.get_listing("900000001").await.unwrap().state,
+            "cancelled"
+        );
+    }
+
+    #[tokio::test]
+    async fn cannot_reserve_sold_or_cancelled() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "sold_one").await;
+        store.reserve_listing("sold_one").await.unwrap();
+        store.complete_sale("sold_one").await.unwrap();
+        assert!(!store.reserve_listing("sold_one").await.unwrap());
+
+        seed(&store, "cancel_one").await;
+        store.cancel_listing("cancel_one", None).await.unwrap();
+        assert!(!store.reserve_listing("cancel_one").await.unwrap());
+    }
+
+    /// Two tasks reserve the SAME active listing concurrently -> exactly one wins.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn two_reservers_exactly_one_wins() {
+        let (store, _d) = temp_store().await;
+        seed(&store, "900000001").await;
+        let store = Arc::new(store);
+
+        let a = {
+            let s = store.clone();
+            tokio::spawn(async move { s.reserve_listing("900000001").await.unwrap() })
+        };
+        let b = {
+            let s = store.clone();
+            tokio::spawn(async move { s.reserve_listing("900000001").await.unwrap() })
+        };
+        let (ra, rb) = (a.await.unwrap(), b.await.unwrap());
+        assert_ne!(ra, rb, "exactly one reserver must win");
+        assert!(ra || rb, "one reserver must win");
+        assert_eq!(
+            store.get_listing("900000001").await.unwrap().state,
+            "reserved"
+        );
+    }
+
+    /// State survives closing and reopening the store file (durable, not in-memory).
+    #[tokio::test]
+    async fn state_survives_reopen() {
+        let db = TempDb::new();
+        let path = db.path();
+        {
+            let store = MarketStore::open(path).await.unwrap();
+            store
+                .create_listing(
+                    "900000001",
+                    "card_pl_001",
+                    Some("core-7"),
+                    Some(100004617),
+                    900,
+                    2500,
+                    Some("alice"),
+                )
+                .await
+                .unwrap();
+            store
+                .cancel_listing("900000001", Some("alice"))
+                .await
+                .unwrap();
+            // pool dropped at end of scope
+        }
+        let reopened = MarketStore::open(path).await.unwrap();
+        let got = reopened.get_listing("900000001").await.unwrap();
+        assert_eq!(got.state, "cancelled");
+        assert_eq!(got.core_item_id.as_deref(), Some("core-7"));
+        assert_eq!(got.wire_item_id, Some(100004617));
+        assert_eq!(got.owner.as_deref(), Some("alice"));
+    }
+}

openfut-utas-host/src/pile_store.rs

@@ -0,0 +1,187 @@
+//! Durable FIFA 17 **item pile / location** metadata.
+//!
+//! FIFA moves an owned card between piles (`club`, `purchased`, `trade`, …) via
+//! `PUT /ut/game/<sku>/item` (FutMoveCard). The pile is a FIFA-side display /
+//! routing concept, NOT ownership: Core remains the sole owner of the item. This
+//! store therefore keeps ONLY the pile keyed by the Core owned-instance id — it
+//! never records ownership, never mints, never duplicates an inventory row.
+//!
+//! It shares the same SQLite-file + connection discipline as
+//! [`crate::market_store`] (WAL established once, foreign_keys + busy_timeout on
+//! every connection, `BEGIN IMMEDIATE` for the upsert), so pile edits are
+//! durable and race-safe.
+
+use std::time::Duration;
+
+use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
+use sqlx::{ConnectOptions, Connection, Row, SqlitePool};
+
+/// Failure of a pile operation.
+#[derive(Debug)]
+pub enum PileError {
+    Db(String),
+}
+
+impl std::fmt::Display for PileError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PileError::Db(e) => write!(f, "pile store db error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for PileError {}
+
+fn db(e: sqlx::Error) -> PileError {
+    PileError::Db(e.to_string())
+}
+
+const CREATE_ITEM_PILE: &str = "CREATE TABLE IF NOT EXISTS item_pile (
+    core_item_id TEXT PRIMARY KEY,
+    pile         TEXT NOT NULL,
+    updated_at   TEXT NOT NULL
+)";
+
+fn now_millis() -> String {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_millis())
+        .unwrap_or(0)
+        .to_string()
+}
+
+/// Durable pile-location store. Cheap to clone (the pool is `Arc` internally).
+#[derive(Clone)]
+pub struct PileStore {
+    pool: SqlitePool,
+}
+
+impl PileStore {
+    /// Open (creating if missing) the pile DB at `path`, mirroring Core's
+    /// `init_pool` (WAL once, foreign_keys + busy_timeout per connection).
+    pub async fn open(path: &str) -> Result<Self, PileError> {
+        let opts = SqliteConnectOptions::new()
+            .filename(path)
+            .create_if_missing(true)
+            .journal_mode(SqliteJournalMode::Wal)
+            .foreign_keys(true)
+            .busy_timeout(Duration::from_secs(5));
+        {
+            let mut conn = opts.clone().connect().await.map_err(db)?;
+            sqlx::query("PRAGMA journal_mode=WAL")
+                .execute(&mut conn)
+                .await
+                .map_err(db)?;
+            conn.close().await.map_err(db)?;
+        }
+        let pool = SqlitePoolOptions::new()
+            .max_connections(8)
+            .connect_with(opts)
+            .await
+            .map_err(db)?;
+        sqlx::query(CREATE_ITEM_PILE)
+            .execute(&pool)
+            .await
+            .map_err(db)?;
+        Ok(PileStore { pool })
+    }
+
+    /// The current pile of a Core-owned item, or `None` if none is recorded.
+    pub async fn get(&self, core_item_id: &str) -> Result<Option<String>, PileError> {
+        let row = sqlx::query("SELECT pile FROM item_pile WHERE core_item_id = ?")
+            .bind(core_item_id)
+            .fetch_optional(&self.pool)
+            .await
+            .map_err(db)?;
+        Ok(row.map(|r| r.get::<String, _>("pile")))
+    }
+
+    /// Set (upsert) the pile of a Core-owned item. Durable and race-safe
+    /// (`BEGIN IMMEDIATE` + upsert).
+    pub async fn set(&self, core_item_id: &str, pile: &str) -> Result<(), PileError> {
+        let updated_at = now_millis();
+        let mut conn = self.pool.acquire().await.map_err(db)?;
+        sqlx::query("BEGIN IMMEDIATE")
+            .execute(&mut *conn)
+            .await
+            .map_err(db)?;
+        let res = sqlx::query(
+            "INSERT INTO item_pile (core_item_id, pile, updated_at) VALUES (?, ?, ?) \
+             ON CONFLICT(core_item_id) DO UPDATE SET pile = excluded.pile, \
+             updated_at = excluded.updated_at",
+        )
+        .bind(core_item_id)
+        .bind(pile)
+        .bind(&updated_at)
+        .execute(&mut *conn)
+        .await;
+        match res {
+            Ok(_) => {
+                sqlx::query("COMMIT")
+                    .execute(&mut *conn)
+                    .await
+                    .map_err(db)?;
+                Ok(())
+            }
+            Err(e) => {
+                let _ = sqlx::query("ROLLBACK").execute(&mut *conn).await;
+                Err(db(e))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicU64, Ordering};
+
+    struct TempDb(String);
+    impl TempDb {
+        fn new() -> Self {
+            static N: AtomicU64 = AtomicU64::new(0);
+            let n = N.fetch_add(1, Ordering::SeqCst);
+            let path =
+                std::env::temp_dir().join(format!("ofut-pile-{}-{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));
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn get_set_upsert() {
+        let db = TempDb::new();
+        let store = PileStore::open(db.path()).await.unwrap();
+        assert_eq!(store.get("core-1").await.unwrap(), None);
+        store.set("core-1", "club").await.unwrap();
+        assert_eq!(store.get("core-1").await.unwrap().as_deref(), Some("club"));
+        // Upsert overwrites, does not duplicate.
+        store.set("core-1", "trade").await.unwrap();
+        assert_eq!(store.get("core-1").await.unwrap().as_deref(), Some("trade"));
+    }
+
+    #[tokio::test]
+    async fn pile_survives_reopen() {
+        let db = TempDb::new();
+        let path = db.path();
+        {
+            let store = PileStore::open(path).await.unwrap();
+            store.set("core-7", "purchased").await.unwrap();
+        }
+        let reopened = PileStore::open(path).await.unwrap();
+        assert_eq!(
+            reopened.get("core-7").await.unwrap().as_deref(),
+            Some("purchased")
+        );
+    }
+}