OpenFUT/openfut-utas-host/src/pile_store.rs

//! 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,
    fault: crate::market_store::StoreFault,
}

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,
            fault: crate::market_store::StoreFault::default(),
        })
    }

    /// A shared handle to this store's test-only fault switch (inert unless a
    /// test arms it). Production never calls it.
    pub fn fault(&self) -> crate::market_store::StoreFault {
        self.fault.clone()
    }

    /// 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> {
        if self.fault.tripped("set") {
            return Err(PileError::Db("injected pile set fault".into()));
        }
        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))
            }
        }
    }

    /// Every Core owned-instance id currently recorded in `pile`, oldest first.
    /// Used to render the `GET /purchased` reveal (the FIFA "purchased" pile).
    pub async fn list_by_pile(&self, pile: &str) -> Result<Vec<String>, PileError> {
        let rows = sqlx::query(
            "SELECT core_item_id FROM item_pile WHERE pile = ? ORDER BY updated_at ASC",
        )
        .bind(pile)
        .fetch_all(&self.pool)
        .await
        .map_err(db)?;
        Ok(rows
            .iter()
            .map(|r| r.get::<String, _>("core_item_id"))
            .collect())
    }
    /// Remove stale presentation metadata after Core has consumed an item.
    /// Projection paths still intersect Core ownership, so failure is safe.
    pub async fn remove(&self, core_item_id: &str) -> Result<(), PileError> {
        sqlx::query("DELETE FROM item_pile WHERE core_item_id = ?")
            .bind(core_item_id)
            .execute(&self.pool)
            .await
            .map_err(db)?;
        Ok(())
    }
}

#[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")
        );
    }

    #[tokio::test]
    async fn list_by_pile_filters_and_reflects_moves() {
        let db = TempDb::new();
        let store = PileStore::open(db.path()).await.unwrap();
        store.set("a", "purchased").await.unwrap();
        store.set("b", "purchased").await.unwrap();
        store.set("c", "club").await.unwrap();
        let mut purchased = store.list_by_pile("purchased").await.unwrap();
        purchased.sort();
        assert_eq!(purchased, vec!["a".to_string(), "b".to_string()]);
        // Moving a card out of the purchased pile drops it from the reveal set.
        store.set("a", "club").await.unwrap();
        assert_eq!(
            store.list_by_pile("purchased").await.unwrap(),
            vec!["b".to_string()]
        );
    }
}