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.
This commit is contained in:
OpenFUT Agent
2026-08-13 20:47:57 +00:00
parent 0b31abe1d1
commit 4d2b8b9be3
10 changed files with 2835 additions and 0 deletions
Generated
+3
View File
@@ -3113,6 +3113,7 @@ name = "openfut-adapter-fifa17"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"openfut-protocol-blaze", "openfut-protocol-blaze",
"rand",
"serde", "serde",
"serde_json", "serde_json",
] ]
@@ -3285,8 +3286,10 @@ dependencies = [
"openfut-http", "openfut-http",
"openfut-identity", "openfut-identity",
"parking_lot", "parking_lot",
"rand",
"reqwest", "reqwest",
"serde_json", "serde_json",
"sqlx",
"tokio", "tokio",
] ]
+4
View File
@@ -15,6 +15,10 @@ openfut-protocol-blaze = { path = "../openfut-protocol-blaze" }
# streak would be reinventing a solved problem in the riskiest possible place. # streak would be reinventing a solved problem in the riskiest possible place.
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" 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] [dev-dependencies]
# Differential fixtures are JSONL; the runtime dependency already covers it. # Differential fixtures are JSONL; the runtime dependency already covers it.
+1
View File
@@ -11,6 +11,7 @@ pub mod economy_policy;
pub mod entities; pub mod entities;
pub mod item; pub mod item;
pub mod owned_query; pub mod owned_query;
pub mod pack_content;
pub mod squad; pub mod squad;
pub mod squad_ext; pub mod squad_ext;
pub mod squad_projection; pub mod squad_projection;
@@ -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());
}
}
+9
View File
@@ -14,6 +14,15 @@ serde_json = "1"
# Plain-HTTP client for Core queries and Python passthrough. UTAS is plaintext # 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. # HTTP (worker D: no wrap_socket, no cert), so no TLS backend is linked.
reqwest = { version = "0.11", default-features = false, features = ["blocking", "json"] } 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] [dev-dependencies]
parking_lot = "0.12" parking_lot = "0.12"
File diff suppressed because it is too large Load Diff
+4
View File
@@ -35,6 +35,10 @@
//! `{"itemData":[]}` — the honest state until Core inventory is asset-backed. //! `{"itemData":[]}` — the honest state until Core inventory is asset-backed.
pub mod config; 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::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream}; use std::net::{TcpListener, TcpStream};
+757
View File
@@ -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")
);
}
}
+608
View File
@@ -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"));
}
}
+187
View File
@@ -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")
);
}
}