From 580d80a86e912b182c16a6ee7dd0931026842d06 Mon Sep 17 00:00:00 2001 From: OpenFUT Agent Date: Thu, 13 Aug 2026 21:30:10 +0000 Subject: [PATCH] feat(fifa17): wire async economy handlers into the host via a runtime bridge (unrouted) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bridges the synchronous thread-per-connection host to the async transfer-market/pile handlers WITHOUT flipping the classifier. classify() is untouched; production still proxies every economy route to Python. The new dispatch is exercised only by the integration harness via Server::try_handle_economy — handler wiring, not authority cutover. async_bridge.rs: AsyncBridge owns ONE process-lifetime multi-threaded Tokio runtime, shared by every connection via Arc. block_on() runs a future from the sync dispatch thread; if invoked from within an ambient runtime it offloads onto its own runtime + a std channel instead of panicking ("cannot start a runtime from within a runtime"). 4 unit tests incl. the nested-runtime-safety case and concurrent multi-thread drivers. lib.rs: EconomyRoute + classify_economy (mirrors the Python route table: credits, purchasegroup, store/transaction, purchased, item DELETE/PUT, ut/delete match/item/trade, auctionhouse/transfermarket, tradePile, trade). EconomyServices (Core econ transport + durable MarketStore/PileStore + the bridge + the pack-content pool), attached via Server::with_economy (kept out of `new`/`from_config` so existing tests build a DB-less Server; production from_config attachment is the barrier step). Server::try_handle_economy dispatches: sync handlers (credits/purchasegroup/store-buy/pack-open/ quick-sell/match) inline; async handlers (market list/query/buy/cancel, move) on the bridge via owned `async move` blocks. build_content_pool derives the resolvable FIFA∩Core candidate pool from Core content. market.rs: FIX the load-bearing hazard the FakeEconomy tests missed — the async market handlers call the BLOCKING reqwest Core client, which panics (reqwest::blocking::wait::enter) when run while a Tokio runtime is entered. off_runtime() hops each Core call to a fresh OS thread with no runtime entered, so blocking is legal. handle_move_items resolver gains `+ Sync` (future must be Send for the bridge). tests/economy_integration.rs: economy_full_sequence_through_dispatch drives the WHOLE cluster through the REAL Server dispatch + bridge against a live in-process Core (seeded with fifa17 dev content: 100k coins + owned cards), on a plain OS thread (direct bridge path), over the real blocking HttpCoreClient — no fakes: Store BUY (pool draw + shape + debit 400 + mint 5), credits, quick-sell (reverse-resolve + credit), match WIN (+400), market list->query->buy->query(sold)->second-buy-fails(no double debit), cancel (cancelled not buyable), move-items, then reopen the durable market/pile stores from disk (sold + pile persist). Deterministic. start_core_seeded loads fifa17 dev content so /collection renders real definitions. Tests: host 68 lib (+4 bridge) + 2 economy_integration + 24 host_test, all green; adapter unchanged-green; clippy -D warnings + fmt clean. --- openfut-utas-host/Cargo.toml | 4 + openfut-utas-host/src/async_bridge.rs | 151 ++++++ openfut-utas-host/src/lib.rs | 279 +++++++++++ openfut-utas-host/src/market.rs | 23 +- .../tests/economy_integration.rs | 447 +++++++++++++++++- 5 files changed, 898 insertions(+), 6 deletions(-) create mode 100644 openfut-utas-host/src/async_bridge.rs diff --git a/openfut-utas-host/Cargo.toml b/openfut-utas-host/Cargo.toml index 57edc00..64d5bd4 100644 --- a/openfut-utas-host/Cargo.toml +++ b/openfut-utas-host/Cargo.toml @@ -23,6 +23,10 @@ sqlx = { version = "0.7", features = ["sqlite", "runtime-tokio"] } # 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" +# One process-lifetime runtime drives the async transfer-market / pile handlers +# (sqlx runtime-tokio). Created once in `Server::from_config`, shared via `Arc`; +# the `AsyncBridge` bridges the synchronous thread-per-connection dispatch to it. +tokio = { version = "1", features = ["rt-multi-thread"] } [dev-dependencies] parking_lot = "0.12" diff --git a/openfut-utas-host/src/async_bridge.rs b/openfut-utas-host/src/async_bridge.rs new file mode 100644 index 0000000..cdd37e4 --- /dev/null +++ b/openfut-utas-host/src/async_bridge.rs @@ -0,0 +1,151 @@ +//! Sync → async execution bridge for the economy handlers. +//! +//! The UTAS host is a **synchronous, thread-per-connection** server +//! ([`crate::Server::serve_listener`] spawns one `std::thread` per socket and +//! [`crate::Server::handle_with_ip`] runs to completion on that plain OS +//! thread). Most handlers — `/club`, squad, credits, match reward, the Store +//! writers — are fully synchronous (Core is reached over a **blocking** reqwest +//! client). Only the durable transfer-market + pile handlers are `async`, +//! because their store is `sqlx` with the `runtime-tokio` feature. +//! +//! Rather than sprinkle per-request runtimes, the host owns ONE long-lived +//! multi-threaded Tokio runtime for the whole process lifetime and drives the +//! few async handlers on it with [`AsyncBridge::block_on`]. +//! +//! ## Execution contract +//! +//! * Exactly one runtime exists for the host lifetime; it is created once +//! (`Server::from_config`) and shared by every connection thread via `Arc`. +//! * `block_on` is called from the synchronous dispatch thread, which is NOT a +//! Tokio worker, so the common path is a direct [`tokio::runtime::Runtime::block_on`]. +//! * **Nested-runtime safety:** should dispatch ever be invoked from inside a +//! Tokio context (it is not in production, but a future async caller or a test +//! might), calling `block_on`/`Handle::block_on` on the current thread would +//! panic (*"Cannot start a runtime from within a runtime"*). We detect an +//! ambient runtime with [`tokio::runtime::Handle::try_current`] and, in that +//! case, offload the future onto our own runtime and block the caller on a +//! std channel — no runtime is entered on the current thread, so it cannot +//! panic. +//! * The blocking Core client used *inside* the async handlers is safe here: it +//! runs its own private runtime on a dedicated background thread, so calling +//! it from a runtime worker parks the worker on a channel rather than entering +//! a runtime. The host client is process-lifetime (`Arc`) and never dropped in +//! an async context. + +use std::future::Future; + +use tokio::runtime::{Builder, Handle, Runtime}; + +/// A process-lifetime Tokio runtime plus a nested-safe blocking bridge. +pub struct AsyncBridge { + rt: Runtime, +} + +impl AsyncBridge { + /// Build the shared multi-threaded runtime. Multi-threaded (not + /// current-thread) so that a handler which parks a worker on the blocking + /// Core client still leaves other workers to drive `sqlx` IO to completion. + pub fn new() -> std::io::Result { + let rt = Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .thread_name("openfut-econ") + .build()?; + Ok(Self { rt }) + } + + /// A cloneable handle to the shared runtime (used by callers that want to + /// `spawn` onto the same executor rather than block). + pub fn handle(&self) -> Handle { + self.rt.handle().clone() + } + + /// Drive `fut` to completion from a synchronous caller, returning its output. + /// + /// The `Send + 'static` bound lets the nested-safety fallback move the future + /// onto our runtime; dispatch satisfies it by building `async move` blocks + /// that own `Arc` clones of the shared services (never per-request state). + pub fn block_on(&self, fut: F) -> F::Output + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + match Handle::try_current() { + // Common path: a plain OS dispatch thread, no ambient runtime. + Err(_) => self.rt.block_on(fut), + // Ambient runtime present: driving on the current thread would panic. + // Offload to our runtime and block on a std channel — the current + // thread never enters a runtime, so it is panic-free. + Ok(_) => { + let (tx, rx) = std::sync::mpsc::sync_channel(1); + self.rt.spawn(async move { + let _ = tx.send(fut.await); + }); + rx.recv().expect("AsyncBridge offloaded task panicked") + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn block_on_from_plain_thread_runs_future() { + let bridge = AsyncBridge::new().unwrap(); + let out = bridge.block_on(async { 2 + 3 }); + assert_eq!(out, 5); + } + + #[test] + fn block_on_from_plain_thread_drives_async_io() { + // A future with an actual await point (timer) completes on our runtime. + let bridge = AsyncBridge::new().unwrap(); + let out = bridge.block_on(async { + tokio::task::yield_now().await; + 41 + 1 + }); + assert_eq!(out, 42); + } + + #[test] + fn block_on_is_nested_runtime_safe() { + // Simulate the hazardous case: dispatch invoked from WITHIN a Tokio + // context. A naive `Runtime::block_on` here panics; the bridge must not. + let bridge = AsyncBridge::new().unwrap(); + let ambient = Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let out = ambient.block_on(async { + // We are now inside `ambient`. Calling the bridge must offload, not + // panic with "Cannot start a runtime from within a runtime". + bridge.block_on(async { + tokio::task::yield_now().await; + 7 * 6 + }) + }); + assert_eq!(out, 42); + } + + #[test] + fn block_on_concurrently_from_many_threads() { + // The shared runtime is driven by many dispatch threads at once, exactly + // as thread-per-connection dispatch does. + let bridge = std::sync::Arc::new(AsyncBridge::new().unwrap()); + let mut handles = Vec::new(); + for i in 0..16i64 { + let b = bridge.clone(); + handles.push(std::thread::spawn(move || { + b.block_on(async move { + tokio::task::yield_now().await; + i * 2 + }) + })); + } + let sum: i64 = handles.into_iter().map(|h| h.join().unwrap()).sum(); + assert_eq!(sum, (0..16i64).map(|i| i * 2).sum::()); + } +} diff --git a/openfut-utas-host/src/lib.rs b/openfut-utas-host/src/lib.rs index 6447c9a..f45a7fc 100644 --- a/openfut-utas-host/src/lib.rs +++ b/openfut-utas-host/src/lib.rs @@ -34,6 +34,7 @@ //! faked (see `club_response`). With today's empty mapping, `/club` returns //! `{"itemData":[]}` — the honest state until Core inventory is asset-backed. +pub mod async_bridge; pub mod config; pub mod economy_store; pub mod market; @@ -56,6 +57,7 @@ use openfut_adapter_fifa17::fut::entities::Fifa17Entities; use openfut_adapter_fifa17::fut::owned_query::{ is_special_rareflag, map_to_core, parse_club_query, MapError, }; +use openfut_adapter_fifa17::fut::pack_content::GeneratedCandidate; use openfut_adapter_fifa17::fut::squad::{parse_squad_put, save_ack, SquadWireResolver}; use openfut_adapter_fifa17::fut::squad_ext::{ build_squad_write, Fifa17SquadExtensionV1, SquadBuildError, EXT_NAMESPACE, EXT_SCHEMA_VERSION, @@ -69,6 +71,7 @@ use openfut_adapter_fifa17::fut::store_session::{ validate_capability, SessionStore, StoreMode, SENTINEL_PACK_ID, }; use openfut_identity::ExternalIdentityStore; +use rand::SeedableRng; use serde_json::{json, Value}; use config::HostConfig; @@ -162,6 +165,87 @@ fn is_numeric_squad_tail(tail: &str) -> bool { } } +/// A FIFA17 economy route, classified separately from [`classify`]. This is the +/// *target* ownership map for the economy cutover. It is deliberately NOT wired +/// into [`Server::handle_with_ip`] yet: handler wiring and authority cutover are +/// distinct steps. Until the single barrier commit flips the whole cluster, +/// production classification ([`classify`]) still sends every one of these to +/// Python; only integration tests drive them through [`Server::try_handle_economy`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EconomyRoute { + /// `GET …/user/credits` — coins + unopened-pack count. + Credits, + /// `GET …/store/purchasegroup` — catalogue + owned packs, full-generated. + PurchaseGroup, + /// `PUT …/store/transaction` — Store BUY (open-on-buy). + StoreBuy, + /// `POST …/purchased` — open a pack / redeem an owned entitlement. + PackOpen, + /// `DELETE …/item/` — single-card quick-sell. + QuickSellPath, + /// `POST /ut/delete/game//item` — bulk quick-sell. + QuickSellBody, + /// `PUT …/item` — FutMoveCard pile move. + MoveItems, + /// `POST /ut/delete/game//match` — match END (the coin-crediting call). + MatchEnd, + /// `…/auctionhouse` | `…/transfermarket` — list-for-sale / browse. + MarketList, + /// `GET …/tradePile` — the user's own active listings. + MarketQuery, + /// `…/trade/` — view / buy-now. + MarketBuy, + /// `DELETE /ut/delete/game//trade/` — cancel a listing. + MarketCancel, +} + +/// `item/` — the single-card quick-sell tail (DELETE). +fn is_item_id_tail(tail: &str) -> bool { + match tail.strip_prefix("item/") { + Some(id) => !id.is_empty() && id.bytes().all(|b| b.is_ascii_digit()), + None => false, + } +} + +/// Classify a FIFA17 economy route from method + path, mirroring the Python +/// oracle's route table (`utas_server.py` §1418-1553). Returns `None` for any +/// non-economy path. Path is already query-stripped by the caller. +pub fn classify_economy(method: &str, path: &str) -> Option { + let get = method.eq_ignore_ascii_case("GET"); + let put = method.eq_ignore_ascii_case("PUT"); + let post = method.eq_ignore_ascii_case("POST"); + let delete = method.eq_ignore_ascii_case("DELETE"); + + // The `/ut/delete/game//…` family is NOT `/ut/game/…`-prefixed. + if let Some(rest) = path.strip_prefix("/ut/delete/game/") { + if let Some((_sku, tail)) = rest.split_once('/') { + if tail == "item" && post { + return Some(EconomyRoute::QuickSellBody); + } + if tail.starts_with("trade") && delete { + return Some(EconomyRoute::MarketCancel); + } + if tail == "match" && post { + return Some(EconomyRoute::MatchEnd); + } + } + return None; + } + + match ut_tail(path) { + Some("user/credits") if get => Some(EconomyRoute::Credits), + Some(t) if get && t.starts_with("store/purchasegroup") => Some(EconomyRoute::PurchaseGroup), + Some("store/transaction") if put => Some(EconomyRoute::StoreBuy), + Some("purchased") if post => Some(EconomyRoute::PackOpen), + Some(t) if delete && is_item_id_tail(t) => Some(EconomyRoute::QuickSellPath), + Some("item") if put => Some(EconomyRoute::MoveItems), + Some(t) if (t == "auctionhouse" || t == "transfermarket") => Some(EconomyRoute::MarketList), + Some("tradePile") if get => Some(EconomyRoute::MarketQuery), + Some(t) if t.starts_with("trade") => Some(EconomyRoute::MarketBuy), + _ => None, + } +} + // ───────────────────────────── Core access boundary ───────────────────────── /// Failure reaching or reading OpenFUT Core. @@ -1594,6 +1678,61 @@ impl PassClient { // ───────────────────────────── Server ─────────────────────────────────────── +/// The durable services the FIFA17 economy handlers need, wired into [`Server`] +/// ONCE at construction (never per request). All are process-lifetime `Arc`s: +/// the Core economy transport, the two host-owned durable SQLite stores (listing +/// and pile), the shared Tokio runtime bridge, and the pack-content candidate +/// pool (the resolvable FIFA∩Core card universe). +#[derive(Clone)] +pub struct EconomyServices { + pub econ: Arc, + pub market: Arc, + pub piles: Arc, + pub bridge: Arc, + /// The resolvable FIFA∩Core card universe a pack can award (empty → the Store + /// fail-closes: it draws nothing and debits nothing). + pub pool: Arc>, +} + +/// Build the pack-content candidate pool from Core's current content, evidenced +/// by the owned inventory: every distinct owned card definition that resolves to +/// a real FIFA asset id is a candidate (`gold` = rating ≥ 75; `special` from the +/// catalog `rareflag > 1`). This is the resolvable FIFA∩Core card universe — the +/// cards a pack can award and the shared shaper can render. An empty pool (no +/// content, or Core unreachable) is fail-closed by construction: the generator +/// returns no cards, so the Store neither mints nor debits. +pub fn build_content_pool( + core: &dyn CoreAccess, + resolver: &Fifa17IdentityResolver, +) -> Vec { + let owned = match core.all_owned() { + Ok(v) => v, + Err(_) => return Vec::new(), + }; + let mut seen = std::collections::HashSet::new(); + let mut pool = Vec::new(); + for item in &owned { + if !seen.insert(item.card_id.clone()) { + continue; + } + let Some(id) = resolver.resolve(item) else { + continue; + }; + pool.push(GeneratedCandidate { + card_id: item.card_id.clone(), + rating: item.rating, + position: item.position.clone(), + nation: item.nation.clone(), + league: item.league.clone(), + club: item.club.clone(), + attributes: item.attributes, + gold: item.rating >= 75, + special: id.rareflag > 1, + }); + } + pool +} + /// The migration host. Cheap to clone (all shared state is `Arc`). #[derive(Clone)] pub struct Server { @@ -1613,6 +1752,11 @@ pub struct Server { sessions: Arc>, /// Monotonic clock origin for the session/pending TTLs. start: Instant, + /// FIFA17 economy authority services (Core transport + durable listing/pile + /// stores + runtime bridge + content pool). `None` until wired via + /// [`Server::with_economy`]; the economy dispatch is inert without it, and + /// `handle_with_ip` does not consult it until the classifier barrier. + economy: Option>, } impl Server { @@ -1632,6 +1776,7 @@ impl Server { persona_id, sessions: Arc::new(Mutex::new(SessionStore::new())), start: Instant::now(), + economy: None, } } @@ -1658,6 +1803,7 @@ impl Server { persona_id: cfg.persona_id, sessions: Arc::new(Mutex::new(SessionStore::new())), start: Instant::now(), + economy: None, }) } @@ -1671,6 +1817,139 @@ impl Server { } } + /// Attach the FIFA17 economy authority services. Kept separate from + /// construction so the (many) squad/club tests build a `Server` without a + /// database, while the economy integration path wires real durable stores + + /// the runtime bridge once. + pub fn with_economy(mut self, economy: Arc) -> Self { + self.economy = Some(economy); + self + } + + /// Dispatch a FIFA17 economy route to its Rust handler, or `None` if the path + /// is not an economy route (or no economy services are wired). This is the + /// handler-wiring entry point exercised by the integration harness; it is + /// deliberately NOT called by `handle_with_ip` yet — handler wiring and the + /// authority cutover are distinct steps, and the classifier barrier is one + /// later coherent flip. Sync handlers run inline; the async transfer-market / + /// move handlers run on the shared runtime via the bridge. + pub fn try_handle_economy( + &self, + method: &str, + target: &str, + headers: &[(String, String)], + body: &[u8], + client_ip: Option<&str>, + ) -> Option { + let svc = self.economy.as_ref()?; + let path = target.split('?').next().unwrap_or(target); + let route = classify_economy(method, path)?; + use crate::economy_store::{ + handle_pack_open, handle_quick_sell_body, handle_quick_sell_path, handle_store_buy, + CoreItemLookup, QuickSellDeps, StoreDeps, + }; + let resp = match route { + EconomyRoute::Credits => handle_credits(svc.econ.as_ref()), + EconomyRoute::PurchaseGroup => { + let sid = header(headers, "x-ut-sid").unwrap_or(""); + let mode = + self.sessions + .lock() + .unwrap() + .empty_mypacks_mode(sid, client_ip, self.now()); + handle_purchasegroup(svc.econ.as_ref(), mode) + } + EconomyRoute::StoreBuy => { + let mut rng = rand::rngs::StdRng::from_entropy(); + let deps = StoreDeps { + econ: svc.econ.as_ref(), + assets: self.resolver.as_ref(), + entities: self.entities.as_ref(), + pool: svc.pool.as_ref(), + }; + handle_store_buy(body, &deps, &mut rng) + } + EconomyRoute::PackOpen => { + let mut rng = rand::rngs::StdRng::from_entropy(); + let deps = StoreDeps { + econ: svc.econ.as_ref(), + assets: self.resolver.as_ref(), + entities: self.entities.as_ref(), + pool: svc.pool.as_ref(), + }; + handle_pack_open(body, &deps, &mut rng) + } + EconomyRoute::QuickSellPath => { + let id = ut_tail(path) + .and_then(|t| t.strip_prefix("item/")) + .and_then(|d| d.parse::().ok())?; + let lookup = CoreItemLookup { + core: self.core.as_ref(), + }; + let deps = QuickSellDeps { + econ: svc.econ.as_ref(), + reverse: self.resolver.as_ref(), + items: &lookup, + }; + handle_quick_sell_path(id, &deps) + } + EconomyRoute::QuickSellBody => { + let lookup = CoreItemLookup { + core: self.core.as_ref(), + }; + let deps = QuickSellDeps { + econ: svc.econ.as_ref(), + reverse: self.resolver.as_ref(), + items: &lookup, + }; + handle_quick_sell_body(body, &deps) + } + EconomyRoute::MatchEnd => handle_match_end(svc.econ.as_ref(), body), + EconomyRoute::MoveItems => { + let (bridge, piles, resolver) = + (svc.bridge.clone(), svc.piles.clone(), self.resolver.clone()); + let body = body.to_vec(); + bridge.block_on(async move { + crate::market::handle_move_items(&body, resolver.as_ref(), piles.as_ref()).await + }) + } + EconomyRoute::MarketList => { + let (bridge, market, econ) = + (svc.bridge.clone(), svc.market.clone(), svc.econ.clone()); + let (m, body) = (method.to_string(), body.to_vec()); + bridge.block_on(async move { + crate::market::handle_market_list(&m, &body, econ.as_ref(), market.as_ref()) + .await + }) + } + EconomyRoute::MarketQuery => { + let (bridge, market, econ) = + (svc.bridge.clone(), svc.market.clone(), svc.econ.clone()); + bridge.block_on(async move { + crate::market::handle_market_query("active", econ.as_ref(), market.as_ref()) + .await + }) + } + EconomyRoute::MarketBuy => { + let (bridge, market, econ) = + (svc.bridge.clone(), svc.market.clone(), svc.econ.clone()); + let (m, p, body) = (method.to_string(), path.to_string(), body.to_vec()); + bridge.block_on(async move { + crate::market::handle_market_buy(&m, &p, &body, econ.as_ref(), market.as_ref()) + .await + }) + } + EconomyRoute::MarketCancel => { + let (bridge, market) = (svc.bridge.clone(), svc.market.clone()); + let (p, owner) = (path.to_string(), client_ip.map(|s| s.to_string())); + bridge.block_on(async move { + crate::market::handle_market_cancel(&p, owner.as_deref(), market.as_ref()).await + }) + } + }; + Some(resp) + } + /// 4-arg entrypoint (tests + callers without a peer address). Session-bound /// routes fall back to a `None` client IP. pub fn handle( diff --git a/openfut-utas-host/src/market.rs b/openfut-utas-host/src/market.rs index c88d000..38168a9 100644 --- a/openfut-utas-host/src/market.rs +++ b/openfut-utas-host/src/market.rs @@ -102,11 +102,26 @@ fn auction_record(l: &Listing) -> Value { }) } +/// Run a BLOCKING closure — the blocking Core client — on a fresh OS thread that +/// has NO Tokio runtime entered, then block until it finishes. The market +/// handlers are `async` and driven on the shared runtime by the host bridge, but +/// `CoreEconomy` is a `reqwest::blocking` client: calling it while a runtime is +/// entered panics (`reqwest::blocking::wait::enter`). Hopping to a plain thread +/// keeps the Core call off the runtime, so blocking is legal. Cheap relative to +/// the network round-trip it guards. +fn off_runtime(f: F) -> T +where + F: FnOnce() -> T + Send, + T: Send, +{ + std::thread::scope(|s| s.spawn(f).join().expect("off-runtime Core call panicked")) +} + /// Current Core balance as the `credits` field, or 0 if Core is unreachable /// (used only to decorate an already-decided response; the transactional path /// never trusts a fabricated balance). fn credits_or_zero(econ: &dyn CoreEconomy) -> i64 { - econ.balance().unwrap_or(0) + off_runtime(|| econ.balance()).unwrap_or(0) } /// `/auctionhouse` — search (GET), list-for-sale (POST FutISStart), relist (PUT). @@ -280,7 +295,7 @@ pub async fn handle_market_buy( // 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() { + let balance = match off_runtime(|| econ.balance()) { Ok(b) => b, Err(_) => { let _ = store.rollback_reservation(&id).await; @@ -298,7 +313,7 @@ pub async fn handle_market_buy( // 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) { + match off_runtime(|| econ.purchase_item(price, &minted_item_id, &listing.card_id)) { Ok(new_balance) => { // Core has taken the coins and minted the item; finalise the listing. // If completing fails, Core is still authoritative — report success. @@ -336,7 +351,7 @@ pub async fn handle_market_buy( /// never a fabricated move. pub async fn handle_move_items( body: &[u8], - resolver: &dyn SquadWireResolver, + resolver: &(dyn SquadWireResolver + Sync), pile_store: &PileStore, ) -> WireResponse { let b = parse_body(body); diff --git a/openfut-utas-host/tests/economy_integration.rs b/openfut-utas-host/tests/economy_integration.rs index 3eec81c..dee3676 100644 --- a/openfut-utas-host/tests/economy_integration.rs +++ b/openfut-utas-host/tests/economy_integration.rs @@ -11,12 +11,21 @@ //! Safety: uses only a temp directory + `127.0.0.1:0` ephemeral ports. Never //! touches the production Core DB, production ports/containers, or `.105`. +use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog; +use openfut_adapter_fifa17::fut::entities::Fifa17Entities; use openfut_adapter_fifa17::fut::store_session::StoreMode; +use openfut_identity::JsonIdentityStore; +use openfut_utas_host::async_bridge::AsyncBridge; +use openfut_utas_host::market_store::MarketStore; +use openfut_utas_host::pile_store::PileStore; use openfut_utas_host::{ - handle_credits, handle_match_end, handle_purchasegroup, overlay_massinfo_economy, CoreEconomy, - HttpCoreClient, + build_content_pool, handle_credits, handle_match_end, handle_purchasegroup, + overlay_massinfo_economy, CoreAccess, CoreEconomy, EconomyServices, Fifa17IdentityResolver, + HttpCoreClient, PassClient, Server, }; use serde_json::{json, Value}; +use std::collections::HashMap; +use std::sync::Arc; /// Boot a Core instance against `db_url`, serving on an ephemeral port. Returns /// the serve task handle and its base URL. @@ -45,6 +54,53 @@ async fn start_core(db_url: &str) -> (tokio::task::JoinHandle<()>, String) { (handle, format!("http://{addr}")) } +/// Boot a Core with the FIFA17 dev CONTENT loaded (so `/collection` renders real +/// card definitions) and, on first boot, the dev inventory SEEDED (a fifa17 +/// profile + club with 100k coins + one owned instance per definition). On +/// restart pass `seed=false`: content is reloaded but the durable DB is left as +/// is, proving persistence rather than re-seeding. +async fn start_core_seeded(db_url: &str, seed: bool) -> (tokio::task::JoinHandle<()>, String) { + use openfut_core::config::Config; + use openfut_core::seed::{seed_fifa17_dev, FIFA17_GAME}; + use openfut_core::services::card_db::CardDb; + + let data_dir = "../openfut-core/data"; + let pool = openfut_core::db::init_pool(db_url, 5) + .await + .expect("core pool"); + openfut_core::db::run_migrations(&pool) + .await + .expect("core migrations"); + if seed { + let mut card_db = CardDb::load(data_dir).expect("card_db load"); + card_db + .load_game_dev(data_dir, FIFA17_GAME) + .expect("load fifa17 dev content"); + seed_fifa17_dev(&pool, &card_db) + .await + .expect("seed fifa17 dev inventory"); + } + let cfg = Config { + listen_addr: "127.0.0.1:0".into(), + database_url: "sqlite::memory:".into(), + data_dir: data_dir.into(), + max_connections: 5, + dev_content_games: vec![FIFA17_GAME.to_string()], + content_packs: Vec::new(), + }; + let app = openfut_core::app::build(pool, cfg) + .await + .expect("core app::build"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral"); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + (handle, format!("http://{addr}")) +} + fn wait_ready(base: &str) { let http = reqwest::blocking::Client::new(); // Generous ceiling (~30s): returns on first success, so it only ever waits @@ -194,3 +250,390 @@ async fn economy_end_to_end_and_restart_persistence() { std::fs::remove_dir_all(&dir).ok(); } + +// ─────────────── Full economy sequence through the real host dispatch ──────── +// +// Everything below drives the ACTUAL `Server::try_handle_economy` path (the same +// dispatch + async bridge the barrier will route production through), against a +// live in-process Core over the real blocking `HttpCoreClient`. No fakes. The +// sequence runs on a plain OS thread (no ambient Tokio runtime), exactly like the +// thread-per-connection server, so the bridge takes its DIRECT `block_on` path. + +/// Facts captured from the write sequence, re-checked after reopening the stores. +struct SeqResult { + sold_listing: String, + moved_core_id: String, +} + +/// Build a real `Server` with economy authority wired: real Core transport, a +/// catalog derived from the seeded content (so the pack pool + shaper resolve), +/// a persistent identity store, and the two durable SQLite stores opened via the +/// bridge. Returns the server, a direct Core client for balance assertions, and +/// the identity resolver (to reverse a wire id for the restart pile check). +fn build_econ_server( + base: &str, + dir: &std::path::Path, +) -> (Server, HttpCoreClient, Arc) { + let probe = HttpCoreClient::new(base, "fifa17"); + let owned = probe.all_owned().expect("core collection"); + assert!(!owned.is_empty(), "seed must grant a starter collection"); + // A catalog mapping every seeded definition to a real-looking asset id (in + // production this is the shipped FIFA catalog; here it is derived so the E2E + // exercises real shaping without a static fixture). + let mut entries = String::new(); + let mut seen = std::collections::HashSet::new(); + let mut asset = 20000u32; + for it in &owned { + if !seen.insert(it.card_id.clone()) { + continue; + } + if !entries.is_empty() { + entries.push(','); + } + entries.push_str(&format!("\"{}\":{{\"asset_id\":{asset}}}", it.card_id)); + asset += 1; + } + let doc = format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}"); + let catalog = Fifa17CardCatalog::from_json_str(&doc).expect("catalog"); + let store = JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).unwrap(); + let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store))); + let entities = Arc::new(Fifa17Entities::from_maps( + HashMap::new(), + HashMap::new(), + HashMap::new(), + )); + let core: Arc = Arc::new(HttpCoreClient::new(base, "fifa17")); + let bridge = Arc::new(AsyncBridge::new().unwrap()); + let market_path = dir.join("market.db").to_string_lossy().into_owned(); + let market = Arc::new( + bridge + .block_on(async move { MarketStore::open(&market_path).await }) + .expect("market store"), + ); + let pile_path = dir.join("pile.db").to_string_lossy().into_owned(); + let piles = Arc::new( + bridge + .block_on(async move { PileStore::open(&pile_path).await }) + .expect("pile store"), + ); + let econ: Arc = Arc::new(HttpCoreClient::new(base, "fifa17")); + let pool = Arc::new(build_content_pool(core.as_ref(), resolver.as_ref())); + assert!( + !pool.is_empty(), + "content pool derived from real Core content" + ); + let services = Arc::new(EconomyServices { + econ, + market, + piles, + bridge, + pool, + }); + let server = Server::new( + core, + entities, + resolver.clone(), + Arc::new(PassClient::new("http://127.0.0.1:9")), + 33068179, + ) + .with_economy(services); + (server, probe, resolver) +} + +/// Drive the whole writer+reader cluster through the real dispatch. Panics on any +/// mismatch. Runs on a plain OS thread (no ambient runtime). +fn economy_sequence(base: &str, dir: &std::path::Path) -> SeqResult { + wait_ready(base); + // Core is seeded (start_core_seeded): a fifa17 profile with 100k coins + one + // owned instance per definition. No /auth/local — the profile already exists. + let (server, client, resolver) = build_econ_server(base, dir); + let start = client.balance().unwrap(); + assert!(start >= 5000, "seeded dev balance present ({start})"); + + // 1) Store BUY (pack 1 = Bronze, price 400, 5 cards): debit + mint + reveal. + let buy = server + .try_handle_economy( + "PUT", + "/ut/game/fifa17/store/transaction", + &[], + br#"{"packId":1}"#, + None, + ) + .expect("store BUY routed"); + assert_eq!(buy.status, 200, "BUY 200"); + let bv: Value = serde_json::from_slice(&buy.body).unwrap(); + let items = bv["createPackResponse"]["itemList"] + .as_array() + .expect("itemList") + .clone(); + assert_eq!(items.len(), 5, "pack 1 awards 5 cards"); + assert_eq!(bv["createPackResponse"]["numberItems"], 5); + assert!( + items[0]["id"].as_i64().unwrap() >= 100_000_000, + "minted wire id above the FIFA floor" + ); + assert_eq!( + client.balance().unwrap(), + start - 400, + "BUY debited exactly 400" + ); + + // 2) credits reads the SAME Core authority. + let cr = server + .try_handle_economy("GET", "/ut/game/fifa17/user/credits", &[], b"", None) + .unwrap(); + let crv: Value = serde_json::from_slice(&cr.body).unwrap(); + assert_eq!( + crv["currencies"][0]["funds"], + start - 400, + "credits == Core balance" + ); + + // 3) Quick-sell one minted card: reverse-resolve wire -> Core id, credit. + let sell_wire = items[0]["id"].as_i64().unwrap(); + let before = client.balance().unwrap(); + let qs = server + .try_handle_economy( + "DELETE", + &format!("/ut/game/fifa17/item/{sell_wire}"), + &[], + b"", + None, + ) + .expect("quick-sell routed"); + assert_eq!(qs.status, 200); + let qv: Value = serde_json::from_slice(&qs.body).unwrap(); + assert_eq!(qv["items"].as_array().unwrap().len(), 1, "sold exactly one"); + let after_sell = client.balance().unwrap(); + assert!( + after_sell > before, + "quick-sell credited ({before}->{after_sell})" + ); + assert_eq!( + qv["totalCredits"].as_i64().unwrap(), + after_sell, + "totalCredits == absolute Core balance" + ); + + // 4) Match END reward through dispatch (WIN = +400). + let before_match = client.balance().unwrap(); + let mm = server + .try_handle_economy( + "POST", + "/ut/delete/game/fifa17/match", + &[], + br#"{"endReason":"WIN"}"#, + None, + ) + .expect("match routed"); + assert_eq!(mm.status, 200); + assert_eq!( + client.balance().unwrap(), + before_match + 400, + "WIN credited +400 via Core" + ); + + // 5) MARKET buy-now (async handlers via the bridge): list -> query -> buy -> + // query -> second buy fails, exactly one debit + one sale. + let list = server + .try_handle_economy( + "POST", + "/ut/game/fifa17/auctionhouse", + &[], + br#"{"itemData":{"id":777,"resourceId":777},"buyNowPrice":1000,"startingBid":500}"#, + None, + ) + .expect("market list routed"); + let lv: Value = serde_json::from_slice(&list.body).unwrap(); + let trade_id = lv["id"].as_i64().expect("trade id"); + let trade_path = format!("/ut/game/fifa17/trade/{trade_id}"); + + let q1 = server + .try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None) + .unwrap(); + let q1v: Value = serde_json::from_slice(&q1.body).unwrap(); + assert_eq!( + q1v["auctionInfo"].as_array().unwrap().len(), + 1, + "listing active before buy" + ); + + let before_buy = client.balance().unwrap(); + let buy1 = server + .try_handle_economy("POST", &trade_path, &[], b"{}", None) + .expect("market buy routed"); + assert_eq!(buy1.status, 200); + assert_eq!( + client.balance().unwrap(), + before_buy - 1000, + "buy debited exactly the buy-now price" + ); + + let q2 = server + .try_handle_economy("GET", "/ut/game/fifa17/tradePile", &[], b"", None) + .unwrap(); + let q2v: Value = serde_json::from_slice(&q2.body).unwrap(); + assert_eq!( + q2v["auctionInfo"].as_array().unwrap().len(), + 0, + "listing sold: no longer active" + ); + + let after_buy = client.balance().unwrap(); + let buy2 = server + .try_handle_economy("POST", &trade_path, &[], b"{}", None) + .unwrap(); + let buy2v: Value = serde_json::from_slice(&buy2.body).unwrap(); + assert_eq!( + buy2v["auctionInfo"].as_array().unwrap().len(), + 0, + "second buy sees a closed auction" + ); + assert_eq!( + client.balance().unwrap(), + after_buy, + "second buy does NOT debit again" + ); + + // 6) MARKET cancel: a cancelled listing cannot be bought. + let clist = server + .try_handle_economy( + "POST", + "/ut/game/fifa17/auctionhouse", + &[], + br#"{"itemData":{"id":888,"resourceId":888},"buyNowPrice":1000,"startingBid":500}"#, + None, + ) + .unwrap(); + let cancel_id = serde_json::from_slice::(&clist.body).unwrap()["id"] + .as_i64() + .unwrap(); + server + .try_handle_economy( + "DELETE", + &format!("/ut/delete/game/fifa17/trade/{cancel_id}"), + &[], + b"", + None, + ) + .expect("market cancel routed"); + let bal_before_cancel_buy = client.balance().unwrap(); + let cancel_buy = server + .try_handle_economy( + "POST", + &format!("/ut/game/fifa17/trade/{cancel_id}"), + &[], + b"{}", + None, + ) + .unwrap(); + assert_eq!( + serde_json::from_slice::(&cancel_buy.body).unwrap()["auctionInfo"] + .as_array() + .unwrap() + .len(), + 0, + "cancelled listing is not buyable" + ); + assert_eq!( + client.balance().unwrap(), + bal_before_cancel_buy, + "buying a cancelled listing does not debit" + ); + + // 7) Move a still-owned minted card to the trade pile (durable pile metadata). + let move_wire = items[1]["id"].as_i64().unwrap(); + let mv = server + .try_handle_economy( + "PUT", + "/ut/game/fifa17/item", + &[], + format!(r#"{{"itemData":[{{"id":{move_wire},"pile":"trade"}}]}}"#).as_bytes(), + None, + ) + .expect("move routed"); + let mvv: Value = serde_json::from_slice(&mv.body).unwrap(); + assert_eq!(mvv["itemData"][0]["success"], true, "move recorded"); + let moved_core_id = resolver + .owned_id_for_wire(move_wire) + .expect("moved item reverses to a Core id"); + + SeqResult { + sold_listing: trade_id.to_string(), + moved_core_id, + } +} + +/// Reopen the durable market/pile SQLite stores from the SAME files (a fresh +/// process would do exactly this) and prove the sold listing and the pile move +/// persisted. Core-side coin/inventory persistence across a full Core restart is +/// proven by `economy_end_to_end_and_restart_persistence`; here the focus is the +/// host-owned durable stores. +fn verify_store_durability(dir: &std::path::Path, seq: &SeqResult) { + let bridge = AsyncBridge::new().unwrap(); + let market_path = dir.join("market.db").to_string_lossy().into_owned(); + let market = bridge + .block_on(async move { MarketStore::open(&market_path).await }) + .unwrap(); + let sold = seq.sold_listing.clone(); + let m2 = market.clone(); + let listing = bridge + .block_on(async move { m2.get_listing(&sold).await }) + .expect("sold listing survives reopen"); + assert_eq!(listing.state, "sold", "sold stays sold after reopen"); + let pile_path = dir.join("pile.db").to_string_lossy().into_owned(); + let piles = bridge + .block_on(async move { PileStore::open(&pile_path).await }) + .unwrap(); + let moved = seq.moved_core_id.clone(); + let p2 = piles.clone(); + let pile = bridge + .block_on(async move { p2.get(&moved).await }) + .unwrap(); + assert_eq!( + pile.as_deref(), + Some("trade"), + "pile move persisted after reopen" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn economy_full_sequence_through_dispatch() { + let dir = std::env::temp_dir().join(format!( + "openfut-econ-dispatch-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + let db_url = format!("sqlite://{}/econ.db", dir.display()); + + // Core #1: run the full write sequence on a plain OS thread (direct bridge + // path). The join runs in spawn_blocking so Core's runtime keeps serving. + let (h1, base1) = start_core_seeded(&db_url, true).await; + let (b1, d1) = (base1.clone(), dir.clone()); + let seq = tokio::task::spawn_blocking(move || { + let t = std::thread::spawn(move || economy_sequence(&b1, &d1)); + t.join().expect("sequence thread") + }) + .await + .expect("write sequence"); + h1.abort(); + + // Durable host stores: reopen the market/pile files from disk (as a fresh + // process would) and prove sold/pile state persisted. No Core rebuild — the + // synthetic market mint uses the wire resourceId as a placeholder card id, + // which Core's content preflight (correctly) rejects on reboot; Core-side + // coin/inventory restart persistence is covered by the sibling test. + let d2 = dir.clone(); + tokio::task::spawn_blocking(move || { + let t = std::thread::spawn(move || verify_store_durability(&d2, &seq)); + t.join().expect("durability thread") + }) + .await + .expect("durability phase"); + + std::fs::remove_dir_all(&dir).ok(); +}