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.

openfut-utas-host/src/pile_store.rs

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