OpenFUT/openfut-utas-host/tests/economy_concurrency.rs

//! Real host↔Core economy CONCURRENCY proofs.
//!
//! Every race here drives the REAL host dispatch (`Server::try_handle_economy`)
//! from several plain `std::thread`s against ONE live in-process Core (axum,
//! ephemeral loopback port, disposable temp SQLite) seeded with the FIFA17 dev
//! inventory. No fakes: the Core economy transport is the real blocking
//! `HttpCoreClient`, the durable listing/pile stores are the real SQLite stores,
//! and the identity store is the real `JsonIdentityStore`.
//!
//! Execution contract (mirrors `economy_integration.rs`): the blocking Core
//! client MUST NOT run while a Tokio runtime is entered on the thread, so all
//! economy work runs on plain `std::thread`s joined under a single
//! `spawn_blocking`, and the async bridge always takes its direct `block_on`
//! path off-runtime. Each racer is its own plain OS thread holding a cheap
//! `Server` clone (all shared state is `Arc`), exactly like the production
//! thread-per-connection server.
//!
//! Safety: temp dir + `127.0.0.1:0` only. Never touches production/`.105`.

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_identity::{ExternalIdentityStore, 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::{
    build_content_pool, CoreAccess, CoreEconomy, EconomyServices, Fifa17IdentityResolver,
    HttpCoreClient, PassClient, Server, WireResponse,
};
use serde_json::Value;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

/// Race iterations per case (the brief asks for 50–100 fresh-state repeats).
const ITERS: usize = 50;

const GAME: &str = "fifa17";
const OWNED_KIND: &str = "owned-item";
const OWNED_FLOOR: i64 = 100_000_001;

// ───────────────────────────── Core boot (seeded) ───────────────────────────

/// Boot a Core with the FIFA17 dev CONTENT loaded and, on first boot, the dev
/// inventory SEEDED (fifa17 profile + club, 100k coins, one owned instance per
/// definition). Identical to the reference harness.
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();
    for _ in 0..1500 {
        if let Ok(r) = http.get(format!("{base}/health")).send() {
            if r.status().is_success() {
                return;
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
    panic!("core did not become ready at {base}");
}

// ───────────────────────────── Harness ──────────────────────────────────────

/// Everything a race needs: the real `Server`, a direct client for Core-state
/// assertions, the identity resolver + its concrete store (watermark checks),
/// and a valid wire `resourceId` that reverse-maps to a real Core `card_id`.
struct Harness {
    server: Server,
    client: HttpCoreClient,
    resolver: Arc<Fifa17IdentityResolver>,
    ident: Arc<JsonIdentityStore>,
    sample_resource: i64,
}

/// Build the real economy `Server` from the seeded Core content (same wiring as
/// `economy_integration::build_econ_server`, plus a retained concrete identity
/// store handle for watermark assertions).
fn build_harness(base: &str, dir: &std::path::Path) -> Harness {
    let probe = HttpCoreClient::new(base, GAME);
    let owned = probe.all_owned().expect("core collection");
    assert!(!owned.is_empty(), "seed must grant a starter collection");
    let mut entries = String::new();
    let mut seen = 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 ident = Arc::new(
        JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).expect("identity"),
    );
    let dyn_store: Arc<dyn ExternalIdentityStore> = ident.clone();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, dyn_store));
    let entities = Arc::new(Fifa17Entities::from_maps(
        HashMap::new(),
        HashMap::new(),
        HashMap::new(),
    ));
    let core: Arc<dyn CoreAccess> = Arc::new(HttpCoreClient::new(base, GAME));
    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<dyn CoreEconomy> = Arc::new(HttpCoreClient::new(base, GAME));
    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);
    Harness {
        server,
        client: HttpCoreClient::new(base, GAME),
        resolver,
        ident,
        sample_resource: 20000,
    }
}

// ───────────────────────────── Race helpers ─────────────────────────────────

type Req = (&'static str, String, Vec<u8>);

/// Dispatch each request on its own plain OS thread (no ambient runtime, so the
/// bridge takes its direct `block_on` path), returning the responses in order.
fn fire(server: &Server, reqs: Vec<Req>) -> Vec<WireResponse> {
    let handles: Vec<_> = reqs
        .into_iter()
        .map(|(m, p, b)| {
            let s = server.clone();
            std::thread::spawn(move || {
                s.try_handle_economy(m, &p, &[], &b, None)
                    .expect("economy route matched")
            })
        })
        .collect();
    handles.into_iter().map(|h| h.join().unwrap()).collect()
}

fn bj(r: &WireResponse) -> Value {
    serde_json::from_slice(&r.body).unwrap_or(Value::Null)
}

/// Drive the club balance to exactly `target` via the real Core economy API.
fn set_balance(client: &HttpCoreClient, target: i64) {
    let cur = client.balance().unwrap();
    if cur > target {
        client
            .purchase_entitlement(cur - target, "econ-test-drain")
            .expect("drain");
    } else if cur < target {
        client.grant_reward(target - cur).expect("top up");
    }
    assert_eq!(client.balance().unwrap(), target, "balance set");
}

/// The on-wire quick-sell `discardValue` tiers (host `economy_store::quick_sell_value`),
/// replicated to assert the EXACT credit — the client-visible contract.
fn qs_value(rating: u8) -> i64 {
    match rating {
        r if r >= 85 => 1500,
        r if r >= 80 => 900,
        r if r >= 75 => 600,
        r if r >= 65 => 300,
        _ => 150,
    }
}

fn owns(client: &HttpCoreClient, core_id: &str) -> bool {
    client
        .all_owned()
        .unwrap()
        .iter()
        .any(|it| it.owned_card_id == core_id)
}

fn rating_of(client: &HttpCoreClient, core_id: &str) -> u8 {
    client
        .all_owned()
        .unwrap()
        .into_iter()
        .find(|it| it.owned_card_id == core_id)
        .expect("owned")
        .rating
}

/// Mint one owned card via a real Store BUY (pack 1). Returns (wire_id, core_id).
/// Balance is set high enough that the buy always succeeds.
fn mint_one(h: &Harness) -> (i64, String) {
    set_balance(&h.client, 50_000);
    let resp = h
        .server
        .try_handle_economy(
            "PUT",
            "/ut/game/fifa17/store/transaction",
            &[],
            br#"{"packId":1}"#,
            None,
        )
        .expect("store buy routed");
    assert_eq!(resp.status, 200, "mint buy 200");
    let items = bj(&resp)["createPackResponse"]["itemList"]
        .as_array()
        .expect("itemList")
        .clone();
    let wire = items[0]["id"].as_i64().expect("wire id");
    let core = h.resolver.owned_id_for_wire(wire).expect("reverse");
    (wire, core)
}

// ───────────────────────────── Cases ────────────────────────────────────────

/// A: two Store BUYs when only ONE is affordable → exactly one 200 + one 461,
/// final balance 0, exactly one debit.
fn case_a_two_buys(h: &Harness) -> String {
    let (mut ok, mut refused) = (0u32, 0u32);
    for _ in 0..ITERS {
        set_balance(&h.client, 400); // exactly one Bronze pack (price 400)
        let rs = fire(
            &h.server,
            vec![
                (
                    "PUT",
                    "/ut/game/fifa17/store/transaction".into(),
                    b"{\"packId\":1}".to_vec(),
                ),
                (
                    "PUT",
                    "/ut/game/fifa17/store/transaction".into(),
                    b"{\"packId\":1}".to_vec(),
                ),
            ],
        );
        let statuses: Vec<u16> = rs.iter().map(|r| r.status).collect();
        let wins = statuses.iter().filter(|&&s| s == 200).count();
        let refs = statuses.iter().filter(|&&s| s == 461).count();
        assert_eq!(wins, 1, "exactly one BUY wins (statuses {statuses:?})");
        assert_eq!(
            refs, 1,
            "exactly one BUY refused 461 (statuses {statuses:?})"
        );
        // The winner minted 5 cards; the loser minted nothing.
        for r in &rs {
            if r.status == 200 {
                assert_eq!(
                    bj(r)["createPackResponse"]["itemList"]
                        .as_array()
                        .unwrap()
                        .len(),
                    5
                );
            }
        }
        assert_eq!(h.client.balance().unwrap(), 0, "exactly one 400 debit → 0");
        ok += wins as u32;
        refused += refs as u32;
    }
    format!("A two-buys: {ITERS} iters, {ok} wins / {refused} refused, always 1+1, final=0")
}

/// B: duplicate OPEN of the same seeded "70" entitlement → exactly one
/// redemption, inventory granted once (11 cards), entitlement consumed once,
/// the loser fails safe.
fn case_b_dup_open(h: &Harness) -> String {
    let mut winners = 0u32;
    for _ in 0..ITERS {
        // Seed exactly one unopened "70" entitlement (cost 0).
        h.client
            .purchase_entitlement(0, "70")
            .expect("seed pack 70");
        let owned_before = h.client.all_owned().unwrap().len();
        let rs = fire(
            &h.server,
            vec![
                (
                    "POST",
                    "/ut/game/fifa17/purchased".into(),
                    b"{\"packId\":70}".to_vec(),
                ),
                (
                    "POST",
                    "/ut/game/fifa17/purchased".into(),
                    b"{\"packId\":70}".to_vec(),
                ),
            ],
        );
        // A real open echoes packId==70; the loser is 503 or an empty reveal.
        let real_opens = rs
            .iter()
            .filter(|r| r.status == 200 && bj(r)["packId"].as_u64() == Some(70))
            .count();
        assert_eq!(real_opens, 1, "exactly one redemption");
        let owned_after = h.client.all_owned().unwrap().len();
        assert_eq!(
            owned_after - owned_before,
            11,
            "inventory granted once (11 cards)"
        );
        // Entitlement consumed exactly once → no unopened "70" remains.
        let unopened_70 = h
            .client
            .entitlements()
            .unwrap()
            .iter()
            .filter(|e| e.definition_id == "70")
            .count();
        assert_eq!(unopened_70, 0, "entitlement consumed once");
        winners += real_opens as u32;
    }
    format!("B dup-open: {ITERS} iters, {winners} single redemptions, +11 once, ent consumed once")
}

/// C: duplicate QUICK-SELL of the same wire id → exactly one sell + one credit +
/// one removal.
fn case_c_dup_quicksell(h: &Harness) -> String {
    let mut sells = 0u32;
    for _ in 0..ITERS {
        let (wire, core) = mint_one(h);
        let value = qs_value(rating_of(&h.client, &core));
        let before = h.client.balance().unwrap();
        let path = format!("/ut/game/fifa17/item/{wire}");
        let rs = fire(
            &h.server,
            vec![
                ("DELETE", path.clone(), Vec::new()),
                ("DELETE", path.clone(), Vec::new()),
            ],
        );
        // Exactly one response actually accounted the sale (items lists the wire).
        let sold = rs
            .iter()
            .filter(|r| {
                r.status == 200
                    && bj(r)["items"]
                        .as_array()
                        .map(|a| a.iter().any(|x| x["id"].as_i64() == Some(wire)))
                        .unwrap_or(false)
            })
            .count();
        assert_eq!(sold, 1, "exactly one quick-sell accounted the item");
        assert!(!owns(&h.client, &core), "item removed exactly once");
        assert_eq!(
            h.client.balance().unwrap(),
            before + value,
            "credited exactly one discardValue (no double credit)"
        );
        sells += sold as u32;
    }
    format!("C dup-quicksell: {ITERS} iters, {sells} single sells, one credit + one removal each")
}

/// D: two buyers of the SAME listing → one success, one closed/empty; sold once;
/// exactly one debit; exactly one mint.
fn case_d_two_market_buyers(h: &Harness) -> String {
    let mut wins = 0u32;
    for i in 0..ITERS {
        set_balance(&h.client, 50_000);
        let item_id = 500_000 + i as i64; // unique listing per iteration
        let list = h
            .server
            .try_handle_economy(
                "POST",
                "/ut/game/fifa17/auctionhouse",
                &[],
                format!(
                    r#"{{"itemData":{{"id":{item_id},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#,
                    h.sample_resource
                )
                .as_bytes(),
                None,
            )
            .expect("list routed");
        let trade_id = bj(&list)["id"].as_i64().expect("trade id");
        let path = format!("/ut/game/fifa17/trade/{trade_id}");
        let before = h.client.balance().unwrap();
        let rs = fire(
            &h.server,
            vec![
                ("POST", path.clone(), b"{}".to_vec()),
                ("POST", path.clone(), b"{}".to_vec()),
            ],
        );
        // Winner returns a non-empty auctionInfo; loser an empty one.
        let non_empty = rs
            .iter()
            .filter(|r| {
                bj(r)["auctionInfo"]
                    .as_array()
                    .map(|a| !a.is_empty())
                    .unwrap_or(false)
            })
            .count();
        assert_eq!(non_empty, 1, "exactly one buyer wins the listing");
        assert_eq!(
            h.client.balance().unwrap(),
            before - 1000,
            "exactly one 1000 debit"
        );
        // Exactly one mint of the deterministic won-card id.
        let mint_id = format!("market-buy:{trade_id}");
        let mints = h
            .client
            .all_owned()
            .unwrap()
            .iter()
            .filter(|it| it.owned_card_id == mint_id)
            .count();
        assert_eq!(mints, 1, "exactly one card minted for the sold listing");
        wins += non_empty as u32;
    }
    format!(
        "D two-market-buyers: {ITERS} iters, {wins} single winners, sold once, 1 debit + 1 mint"
    )
}

/// E: match REWARD + Store BUY concurrently → final balance is one legal
/// serialization (no lost update). Reward (+400) and buy (−400) commute, so the
/// final balance must equal the start exactly.
fn case_e_reward_and_buy(h: &Harness) -> String {
    let start = 8_000i64;
    for _ in 0..ITERS {
        set_balance(&h.client, start);
        let rs = fire(
            &h.server,
            vec![
                (
                    "POST",
                    "/ut/delete/game/fifa17/match".into(),
                    b"{\"endReason\":\"WIN\"}".to_vec(),
                ),
                (
                    "PUT",
                    "/ut/game/fifa17/store/transaction".into(),
                    b"{\"packId\":1}".to_vec(),
                ),
            ],
        );
        assert!(rs.iter().all(|r| r.status == 200), "both ops succeed");
        assert_eq!(
            h.client.balance().unwrap(),
            start,
            "reward(+400) and buy(-400) both applied: no lost update"
        );
    }
    format!("E reward+buy: {ITERS} iters, final==start ({start}) every time (no lost update)")
}

/// F: MOVE + QUICK-SELL of the same item → one coherent final state (item sold
/// once, one credit, and no SOLD item rendered in any pile view).
fn case_f_move_and_quicksell(h: &Harness) -> String {
    for _ in 0..ITERS {
        let (wire, core) = mint_one(h);
        let value = qs_value(rating_of(&h.client, &core));
        let before = h.client.balance().unwrap();
        let del_path = format!("/ut/game/fifa17/item/{wire}");
        let rs = fire(
            &h.server,
            vec![
                (
                    "PUT",
                    "/ut/game/fifa17/item".into(),
                    format!(r#"{{"itemData":[{{"id":{wire},"pile":"trade"}}]}}"#).into_bytes(),
                ),
                ("DELETE", del_path, Vec::new()),
            ],
        );
        assert!(rs.iter().all(|r| r.status == 200), "both ops respond 200");
        // Exactly one real disposal: Core no longer owns the item, credited once.
        assert!(
            !owns(&h.client, &core),
            "item sold (Core no longer owns it)"
        );
        assert_eq!(
            h.client.balance().unwrap(),
            before + value,
            "exactly one credit (move never credits)"
        );
        // Coherence: the sold item is not rendered in the purchased-pile reveal.
        let reveal = h
            .server
            .try_handle_economy("GET", "/ut/game/fifa17/purchased", &[], b"", None)
            .expect("reveal routed");
        let shown = bj(&reveal)["itemData"]
            .as_array()
            .map(|a| a.iter().any(|x| x["id"].as_i64() == Some(wire)))
            .unwrap_or(false);
        assert!(!shown, "no sold item left in a visible pile");
    }
    format!("F move+quicksell: {ITERS} iters, sold once + one credit + no sold item in a pile")
}

/// G: LIST + QUICK-SELL of the same owned item → only one legal transition of
/// the real item's Core ownership (the sell). The listing is a synthetic-seller
/// row that does not dispose of the owned instance, so it never becomes a second
/// credit or a double removal.
fn case_g_list_and_quicksell(h: &Harness) -> String {
    for i in 0..ITERS {
        let (wire, core) = mint_one(h);
        let value = qs_value(rating_of(&h.client, &core));
        let before = h.client.balance().unwrap();
        let list_body = format!(
            r#"{{"itemData":{{"id":{wire},"resourceId":{}}},"buyNowPrice":1000,"startingBid":500}}"#,
            h.sample_resource
        );
        let del_path = format!("/ut/game/fifa17/item/{wire}");
        let rs = fire(
            &h.server,
            vec![
                (
                    "POST",
                    "/ut/game/fifa17/auctionhouse".into(),
                    list_body.into_bytes(),
                ),
                ("DELETE", del_path, Vec::new()),
            ],
        );
        assert!(rs.iter().all(|r| r.status == 200), "both ops respond 200");
        // The one legal ownership transition: the item is sold exactly once.
        assert!(!owns(&h.client, &core), "item removed exactly once");
        assert_eq!(
            h.client.balance().unwrap(),
            before + value,
            "credited exactly once (listing does not credit)"
        );
        let _ = i;
    }
    format!(
        "G list+quicksell: {ITERS} iters, exactly one ownership transition (sell); listing is synthetic"
    )
}

/// H: concurrent MINTS (Store BUYs) → no duplicate FIFA wire ids across all
/// mints, every wire id reverse-resolves, and the identity watermark stays
/// monotonic (the next id to issue is strictly above every issued id).
fn case_h_concurrent_mints(h: &Harness) -> String {
    let mut all: HashSet<i64> = HashSet::new();
    let mut total = 0usize;
    for _ in 0..ITERS {
        set_balance(&h.client, 50_000);
        let reqs: Vec<Req> = (0..4)
            .map(|_| {
                (
                    "PUT",
                    "/ut/game/fifa17/store/transaction".to_string(),
                    b"{\"packId\":1}".to_vec(),
                )
            })
            .collect();
        let rs = fire(&h.server, reqs);
        for r in &rs {
            assert_eq!(r.status, 200, "concurrent buy 200");
            for it in bj(r)["createPackResponse"]["itemList"].as_array().unwrap() {
                let wire = it["id"].as_i64().expect("wire id");
                assert!(
                    all.insert(wire),
                    "wire id {wire} is globally unique (no duplicate)"
                );
                assert!(
                    h.resolver.owned_id_for_wire(wire).is_some(),
                    "wire id {wire} reverse-resolves to a Core instance"
                );
                total += 1;
            }
        }
    }
    // Monotonic watermark: the next id the store would issue is strictly above
    // every id it has ever issued.
    let next = h.ident.peek_next_external(GAME, OWNED_KIND, OWNED_FLOOR);
    let max_seen = all.iter().copied().max().unwrap_or(OWNED_FLOOR - 1);
    assert!(next > max_seen, "watermark {next} > max issued {max_seen}");
    assert!(
        all.iter().all(|&w| w >= OWNED_FLOOR),
        "every wire id ≥ floor"
    );
    format!(
        "H concurrent-mints: {ITERS}×4 buys, {total} mints, all unique + reversible, watermark {next} > {max_seen}"
    )
}

fn run_all_races(base: &str, dir: &std::path::Path) -> String {
    wait_ready(base);
    let h = build_harness(base, dir);
    [
        case_a_two_buys(&h),
        case_b_dup_open(&h),
        case_c_dup_quicksell(&h),
        case_d_two_market_buyers(&h),
        case_e_reward_and_buy(&h),
        case_f_move_and_quicksell(&h),
        case_g_list_and_quicksell(&h),
        case_h_concurrent_mints(&h),
    ]
    .join("\n")
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn economy_concurrency_races() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-conc-{}-{}",
        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());

    let (h1, base1) = start_core_seeded(&db_url, true).await;
    let (b1, d1) = (base1.clone(), dir.clone());
    let summary = tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || run_all_races(&b1, &d1))
            .join()
            .expect("race thread")
    })
    .await
    .expect("races");
    h1.abort();
    std::fs::remove_dir_all(&dir).ok();

    eprintln!("economy_concurrency summary:\n{summary}");
}