//! 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::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use parking_lot::Mutex;

use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
use sqlx::{ConnectOptions, Connection, Row, SqlitePool};

/// Test-only durable-store fault injector, shared (cheap `Arc` clone) between a
/// store and the failure-injection tests. It is **inert in production**: nothing
/// arms it, so each guarded op reads one relaxed atomic and behaves exactly as
/// before. The economy failure tests use it to force a durable-store write to
/// fail at a chosen point (market complete-sale / reserve, pile write) — the
/// only way to exercise those recovery paths, since the stores are concrete
/// types wired straight into the handlers (no trait seam to substitute).
#[derive(Clone, Default)]
pub struct StoreFault {
    inner: Arc<StoreFaultInner>,
}

#[derive(Default)]
struct StoreFaultInner {
    any: AtomicBool,
    armed: Mutex<HashMap<&'static str, u32>>,
}

impl StoreFault {
    /// Arm `op` to fail its next `times` invocations, then heal automatically.
    pub fn arm(&self, op: &'static str, times: u32) {
        self.inner.armed.lock().insert(op, times);
        self.inner.any.store(true, Ordering::SeqCst);
    }

    /// Consume one armed unit for `op`, returning whether it should fail now.
    /// Fast path (unarmed): a single relaxed atomic load, no lock taken.
    pub fn tripped(&self, op: &'static str) -> bool {
        if !self.inner.any.load(Ordering::Relaxed) {
            return false;
        }
        let mut armed = self.inner.armed.lock();
        let fire = match armed.get_mut(op) {
            Some(n) if *n > 0 => {
                *n -= 1;
                true
            }
            _ => false,
        };
        if armed.values().all(|&n| n == 0) {
            self.inner.any.store(false, Ordering::SeqCst);
        }
        fire
    }
}

/// 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,
    /// Authoritative Core content/card id — what a synthetic buy MINTS. Distinct
    /// from the FIFA wire `resourceId` (see `wire_resource_id`).
    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>,
    /// The FIFA wire `resourceId` (versioned) the client listed, echoed back in
    /// the auction record. Never used to mint — the mint uses `card_id`.
    pub wire_resource_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,
    /// The FIFA card object (`itemData`) as shaped at listing time, serialized.
    /// A listing is a SNAPSHOT: the auction record must carry the full card the
    /// client can render (rating/position/attributes/rareflag/assetId), not a
    /// stub — a stub leaves the Transfer List with an unrenderable row. `None`
    /// only for rows written before this column existed (renders as a stub).
    pub item_json: Option<String>,
    /// Listing duration in SECONDS, as sent by the client in the `ISStart` body
    /// (`duration`). With `created_at` this is the whole auction clock: FIFA 17
    /// renders a live countdown from `expires` and expects it to reach 0, so a
    /// listing has to know when it ends. `None` for rows written before this
    /// column existed, which fall back to the default duration.
    pub duration_secs: Option<i64>,
}

/// FIFA 17 auction durations, in seconds: 3600, 10800, 21600, 43200, 86400,
/// 259200. One hour is the shortest, and the fallback when a client body omits it
/// or a pre-column row is read.
pub const DEFAULT_DURATION_SECS: i64 = 3600;

/// Seconds since the unix epoch.
pub fn now_secs() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

impl Listing {
    /// SECONDS REMAINING on this auction at `now` (unix seconds), clamped at 0 —
    /// the wire semantics of `expires`, which is never an absolute epoch.
    ///
    /// A closed/sold/cancelled listing reads 0: there is no time left on an
    /// auction that has already ended.
    pub fn expires_in_secs(&self, now: i64) -> i64 {
        if self.state != "active" {
            return 0;
        }
        let created_secs = self
            .created_at
            .parse::<i64>()
            .map(|ms| ms / 1000)
            .unwrap_or(now);
        let duration = self.duration_secs.unwrap_or(DEFAULT_DURATION_SECS);
        (created_secs + duration - now).max(0)
    }
}

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,
    wire_resource_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,
    item_json     TEXT,
    duration_secs INTEGER,
    -- Seller acknowledgement of a SOLD row, separate from the sale itself. Declared
    -- here so a fresh store never needs the ALTER path below; the additive
    -- migration exists only for stores created before this column.
    cleared_at    TEXT
)";

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"),
        wire_resource_id: row.get("wire_resource_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"),
        item_json: row.get("item_json"),
        duration_secs: row.get("duration_secs"),
    }
}

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

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)?;
        // Additive migration: `item_json` was added after the first stores shipped,
        // and `CREATE TABLE IF NOT EXISTS` will not add a column to an existing
        // file. Add it when absent so an existing market DB keeps working (old
        // rows read back `None` and render the stub card).
        let existing: Vec<String> = sqlx::query("PRAGMA table_info(listings)")
            .fetch_all(&pool)
            .await
            .map_err(db)?
            .iter()
            .map(|r| r.get::<String, _>("name"))
            .collect();
        for (col, decl) in [
            ("item_json", "TEXT"),
            ("duration_secs", "INTEGER"),
            // A SOLD listing is not the end of the seller's involvement: FIFA 17 has
            // a bulk `DELETE …/trade/sold` verb (builder 0x1801647c0, request name
            // RemoveAllSoldFromTradePile), which only makes sense if sold rows
            // PERSIST in the seller's pile until cleared. `cleared_at` records that
            // acknowledgement separately from the sale itself, so clearing a row can
            // never be mistaken for re-settling it.
            ("cleared_at", "TEXT"),
        ] {
            if !existing.iter().any(|c| c == col) {
                sqlx::query(&format!("ALTER TABLE listings ADD COLUMN {col} {decl}"))
                    .execute(&pool)
                    .await
                    .map_err(db)?;
            }
        }
        Ok(MarketStore {
            pool,
            fault: 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) -> StoreFault {
        self.fault.clone()
    }

    /// 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>,
        wire_resource_id: Option<i64>,
        start_price: i64,
        buy_now_price: i64,
        owner: Option<&str>,
        // The shaped FIFA card snapshot (`itemData`) for the auction record.
        item_json: Option<&str>,
        // Listing duration in seconds from the client's `ISStart` body; `None`
        // falls back to [`DEFAULT_DURATION_SECS`].
        duration_secs: Option<i64>,
    ) -> 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, \
             wire_resource_id, start_price, buy_now_price, owner, state, created_at, item_json, \
             duration_secs) \
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)",
        )
        .bind(listing_id)
        .bind(card_id)
        .bind(core_item_id)
        .bind(wire_item_id)
        .bind(wire_resource_id)
        .bind(start_price)
        .bind(buy_now_price)
        .bind(owner)
        .bind(&created_at)
        .bind(item_json)
        .bind(duration_secs)
        .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,
                    wire_resource_id,
                    start_price,
                    buy_now_price,
                    owner: owner.map(str::to_string),
                    state: "active".to_string(),
                    created_at,
                    item_json: item_json.map(str::to_string),
                    duration_secs,
                })
            }
            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> {
        if self.fault.tripped("reserve") {
            return Err(MarketError::Db("injected reserve fault".into()));
        }
        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.fault.tripped("complete_sale") {
            return Err(MarketError::Db("injected complete_sale fault".into()));
        }
        if self.cas(listing_id, "reserved", "sold").await? {
            Ok(())
        } else {
            Err(MarketError::Conflict)
        }
    }

    /// Mark a live listing SOLD in one step (`active | reserved -> sold`), for a
    /// sale driven by a counterparty rather than by this client's own buy-now.
    /// Returns whether this call was the one that sold it, so a replay is visible
    /// to the caller instead of silently settling twice.
    pub async fn mark_sold(&self, listing_id: &str) -> Result<bool, MarketError> {
        let affected = sqlx::query(
            "UPDATE listings SET state = 'sold' \
             WHERE listing_id = ? AND state IN ('active', 'reserved')",
        )
        .bind(listing_id)
        .execute(&self.pool)
        .await
        .map_err(db)?
        .rows_affected();
        Ok(affected == 1)
    }

    /// Sold listings the seller has NOT yet cleared, newest first.
    ///
    /// Separate from [`Self::query_listings`] because "sold" and "still shown to
    /// the seller" are different facts: a sold row stays in the pile until the
    /// client acknowledges it via the bulk clear verb.
    pub async fn uncleared_sold(&self) -> Result<Vec<Listing>, MarketError> {
        let rows = sqlx::query(
            "SELECT * FROM listings WHERE state = 'sold' AND cleared_at IS NULL \
             ORDER BY created_at DESC",
        )
        .fetch_all(&self.pool)
        .await
        .map_err(db)?;
        Ok(rows.iter().map(row_to_listing).collect())
    }

    /// Acknowledge every uncleared sold listing (the bulk `DELETE …/trade/sold`).
    /// Returns how many rows were cleared.
    ///
    /// This is PRESENTATION ONLY. It records that the seller has seen the sale; it
    /// moves no coins and no ownership, because settlement already happened when
    /// the sale completed. Clearing must never be able to pay anyone twice.
    pub async fn clear_sold(&self) -> Result<u64, MarketError> {
        Ok(sqlx::query(
            "UPDATE listings SET cleared_at = ? \
             WHERE state = 'sold' AND cleared_at IS NULL",
        )
        .bind(now_millis())
        .execute(&self.pool)
        .await
        .map_err(db)?
        .rows_affected())
    }

    /// Acknowledge ONE sold listing by id (the per-id `DELETE …/trade/{id}` form,
    /// if the client turns out to use it for sold rows). Same presentation-only
    /// contract as [`Self::clear_sold`].
    pub async fn clear_sold_one(&self, listing_id: &str) -> Result<u64, MarketError> {
        Ok(sqlx::query(
            "UPDATE listings SET cleared_at = ? \
             WHERE listing_id = ? AND state = 'sold' AND cleared_at IS NULL",
        )
        .bind(now_millis())
        .bind(listing_id)
        .execute(&self.pool)
        .await
        .map_err(db)?
        .rows_affected())
    }

    /// 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
    }

    /// RE-LIST an existing auction row for the same item: reset the clock to now
    /// and take the new prices/duration.
    ///
    /// FIFA 17 relists by sending a fresh `ISStart` POST for an item that already
    /// has a listing row, so the primary-key conflict is EXPECTED and means
    /// "relist", not "error". Treating that conflict as success is how a relist
    /// silently did nothing: the client was acked while the stale, already-expired
    /// row kept its old `created_at` and stayed expired.
    ///
    /// Only an `active` (including aged-out) or `cancelled` row may be relisted. A
    /// `sold` or `reserved` row is NEVER resurrected — the card is gone or in
    /// flight, and re-opening that auction would sell a card twice.
    pub async fn relist_listing(
        &self,
        listing_id: &str,
        start_price: i64,
        buy_now_price: i64,
        duration_secs: Option<i64>,
        item_json: Option<&str>,
    ) -> Result<Listing, MarketError> {
        let created_at = now_millis();
        let affected = sqlx::query(
            "UPDATE listings SET state = 'active', created_at = ?, start_price = ?, \
             buy_now_price = ?, duration_secs = ?, item_json = COALESCE(?, item_json) \
             WHERE listing_id = ? AND state IN ('active', 'cancelled')",
        )
        .bind(&created_at)
        .bind(start_price)
        .bind(buy_now_price)
        .bind(duration_secs)
        .bind(item_json)
        .bind(listing_id)
        .execute(&self.pool)
        .await
        .map_err(db)?
        .rows_affected();
        if affected == 0 {
            // Either no such row, or it is sold/reserved and must not be revived.
            return Err(match self.get_listing(listing_id).await {
                Ok(l) if l.state == "sold" => MarketError::Sold,
                Ok(_) => MarketError::Conflict,
                Err(e) => e,
            });
        }
        self.get_listing(listing_id).await
    }

    /// Cancel any `active` listing held by a Core owned item, returning how many
    /// rows were cancelled (0 when the item has no live auction).
    ///
    /// This is the RETURN-TO-CLUB transition: the client sends a pile move for an
    /// expired transfer-list item, and the auction that put it there has to end with
    /// it. Without this the pile says `club` while the listing row stays `active`,
    /// so the card is still filtered out of `/club` AND still rendered in the
    /// Transfer List — the item appears not to move at all.
    ///
    /// Deliberately scoped to `active`: a `reserved` row is mid-sale and a `sold`
    /// row is already gone, and cancelling either would let a card be both sold and
    /// returned.
    pub async fn cancel_active_for_core_item(
        &self,
        core_item_id: &str,
    ) -> Result<u64, MarketError> {
        Ok(sqlx::query(
            "UPDATE listings SET state = 'cancelled' \
             WHERE core_item_id = ? AND state = 'active'",
        )
        .bind(core_item_id)
        .execute(&self.pool)
        .await
        .map_err(db)?
        .rows_affected())
    }

    pub async fn has_active_for_core_item(&self, core_item_id: &str) -> Result<bool, MarketError> {
        sqlx::query_scalar(
            "SELECT EXISTS( \
                SELECT 1 FROM listings WHERE core_item_id = ? AND state = 'active' \
             )",
        )
        .bind(core_item_id)
        .fetch_one(&self.pool)
        .await
        .map_err(db)
    }
}

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

    #[tokio::test]
    async fn relist_resets_the_clock_and_takes_the_new_prices() {
        // FIFA 17 relists by re-sending ISStart for an item that already has a row,
        // so the PK conflict is the relist path. Before this existed the conflict was
        // acked as success and the stale expired row kept its old created_at, so the
        // card never came back to the market.
        let (store, _d) = temp_store().await;
        let first = seed(&store, "900000001").await;
        // Age it out by rewriting created_at to well past its default duration.
        let stale = (now_secs() - DEFAULT_DURATION_SECS - 600) * 1000;
        sqlx::query("UPDATE listings SET created_at = ? WHERE listing_id = ?")
            .bind(stale.to_string())
            .bind("900000001")
            .execute(&store.pool)
            .await
            .unwrap();
        let expired = store.get_listing("900000001").await.unwrap();
        assert_eq!(
            expired.expires_in_secs(now_secs()),
            0,
            "precondition: the listing has run out"
        );

        let relisted = store
            .relist_listing("900000001", 250, 5000, Some(10_800), None)
            .await
            .unwrap();
        assert_eq!(relisted.state, "active");
        assert_eq!(relisted.start_price, 250, "new start price applied");
        assert_eq!(relisted.buy_now_price, 5000, "new buy-now applied");
        assert_eq!(relisted.duration_secs, Some(10_800));
        assert!(
            relisted.expires_in_secs(now_secs()) > 0,
            "the clock actually restarted"
        );
        assert_ne!(
            relisted.created_at, expired.created_at,
            "created_at moved forward"
        );
        // The snapshot is preserved when the relist does not supply a new one.
        assert_eq!(relisted.item_json, first.item_json);
    }

    #[tokio::test]
    async fn relist_never_revives_a_sold_or_reserved_auction() {
        // Re-opening a sold auction would sell the same card twice.
        let (store, _d) = temp_store().await;
        seed(&store, "900000001").await;
        assert!(store.reserve_listing("900000001").await.unwrap());
        assert!(
            matches!(
                store.relist_listing("900000001", 1, 2, None, None).await,
                Err(MarketError::Conflict)
            ),
            "a reserved (in-flight) auction is not relistable"
        );
        store.complete_sale("900000001").await.unwrap();
        assert!(
            matches!(
                store.relist_listing("900000001", 1, 2, None, None).await,
                Err(MarketError::Sold)
            ),
            "a sold auction is never resurrected"
        );
        assert_eq!(store.get_listing("900000001").await.unwrap().state, "sold");

        // A cancelled listing IS relistable (the card came back to the pile).
        seed(&store, "900000002").await;
        store.cancel_listing("900000002", None).await.unwrap();
        let back = store
            .relist_listing("900000002", 300, 900, None, None)
            .await
            .unwrap();
        assert_eq!(back.state, "active");
        assert_eq!(back.start_price, 300);
    }

    #[tokio::test]
    async fn relist_of_a_missing_row_is_not_found() {
        let (store, _d) = temp_store().await;
        assert!(matches!(
            store.relist_listing("900000999", 1, 2, None, None).await,
            Err(MarketError::NotFound)
        ));
    }

    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,
                None,
                900,
                2500,
                None,
                None,
                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,
                    None,
                    1,
                    2,
                    None,
                    None,
                    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,
                None,
                900,
                2500,
                Some("alice"),
                None,
                None,
            )
            .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),
                    Some(169193),
                    900,
                    2500,
                    Some("alice"),
                    Some(r#"{"rating":84}"#),
                    None,
                )
                .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"));
        assert_eq!(
            got.item_json.as_deref(),
            Some(r#"{"rating":84}"#),
            "card snapshot survives reopen"
        );
    }
}