OpenFUT-Core/src/services/consume.rs

//! Atomic "apply one consumable to a target" primitive.
//!
//! ONE Core transaction that does, in this order and nothing else:
//!
//!   1. answer a REPLAY — if `(profile_id, action_identity)` is already recorded,
//!      echo that outcome and touch nothing. This precedes every validation on
//!      purpose: a completed application has already DESTROYED its source, so
//!      checking the source first would answer "not found" to a retried request
//!      that in fact succeeded;
//!   2. validate the SOURCE — it exists, belongs to the club, and is the
//!      `ContentKind` the caller expected;
//!   3. validate the TARGET — nothing at all (`ConsumeTarget::Club`) or an owned
//!      instance that exists, belongs to the club, and is the expected kind;
//!   4. write the replay guard — `UNIQUE(profile_id, action_identity)` on
//!      `consumable_applications`, so a duplicate is refused BEFORE anything is
//!      mutated or consumed (same discipline as `match_completions`). Step 1 is
//!      a courtesy; THIS is the guarantee, and it holds against a writer on any
//!      other connection or process;
//!   5. apply the caller's mutation to the target;
//!   6. consume the source EXACTLY ONCE — destroy the instance, or decrement its
//!      stack and destroy it at zero;
//!   7. commit.
//!
//! Anything failing at any step rolls the whole thing back: the source is never
//! spent without the effect landing, and the effect never lands without the
//! source being spent.
//!
//! Core deliberately supplies **no per-category formula**. What a fitness card,
//! a contract, a chemistry style or a position modifier actually DOES to its
//! target is the calling game adapter's reversed behaviour, passed in as
//! [`ItemMutation`] — either in-process, or described over the wire through the
//! closed, Core-validated vocabulary in [`crate::services::instance_effect`]. An
//! unreversed behaviour must not be invented here, and the honest stopping point
//! for one is ownership + projection, i.e. not calling this function at all.

use std::future::Future;
use std::pin::Pin;

use chrono::Utc;
use serde::Serialize;
use serde_json::Value;
use sqlx::SqliteConnection;
use uuid::Uuid;

use crate::{
    db::Pool,
    error::{AppError, AppResult},
    models::card::{ContentKind, OwnedCard, OWNED_CARD_SELECT},
    services::economy,
};

/// What the transaction does to the source instance once the effect is applied.
///
/// Both variants run inside the one transaction and under the one replay guard.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SourceConsumption {
    /// Destroy the instance: exactly one `owned_cards` row is DELETEd.
    DestroyInstance,
    /// Decrement a stack by `amount`, destroying the row when it reaches zero.
    ///
    /// Only valid for a source that actually carries a stack size
    /// (`owned_cards.quantity IS NOT NULL`); a bare instance has no count to
    /// decrement and is refused rather than silently destroyed.
    DecrementStack { amount: i64 },
}

/// What the consumable is being applied to.
///
/// Kind validation is explicit at the call site: the caller states which
/// `ContentKind` the target must be, because only the caller knows that (say) a
/// chemistry style goes on a player and a manager-league modifier goes on a
/// manager.
#[derive(Debug, Clone, Copy)]
pub enum ConsumeTarget<'a> {
    /// Another owned instance of the same club.
    OwnedCard {
        owned_card_id: &'a str,
        expected_kind: ContentKind,
    },
    /// Club-scoped state rather than an owned instance (the mutation writes
    /// whatever club row it owns; Core validates only the source).
    Club,
}

/// One application request.
#[derive(Debug, Clone, Copy)]
pub struct ConsumeRequest<'a> {
    /// Opaque, stable per-application token supplied by the caller. Core never
    /// parses it; it only enforces `UNIQUE(profile_id, action_identity)`.
    pub action_identity: &'a str,
    pub source_owned_card_id: &'a str,
    /// The kind the source MUST be. A mismatch is refused.
    pub expected_source_kind: ContentKind,
    pub consumption: SourceConsumption,
    pub target: ConsumeTarget<'a>,
}

/// Validated context handed to the caller's mutation. Both rows are as they were
/// read inside the transaction, before any mutation or consumption.
#[derive(Debug, Clone)]
pub struct ConsumeContext {
    pub profile_id: String,
    pub club_id: String,
    pub source: OwnedCard,
    /// `None` for [`ConsumeTarget::Club`].
    pub target: Option<OwnedCard>,
}

/// A future returned by an [`ItemMutation`], borrowing the transaction.
pub type MutationFuture<'c> = Pin<Box<dyn Future<Output = AppResult<Value>> + Send + 'c>>;

/// The caller's effect on the target, applied inside Core's transaction.
///
/// It receives the transaction connection, so every write it makes is committed
/// or rolled back together with the source consumption. The `Value` it returns is
/// stored verbatim as the application's recorded outcome and echoed on replay —
/// Core never interprets it.
pub trait ItemMutation: Send + Sync {
    fn apply<'c>(
        &'c self,
        tx: &'c mut SqliteConnection,
        ctx: &'c ConsumeContext,
    ) -> MutationFuture<'c>;
}

impl<F> ItemMutation for F
where
    F: for<'c> Fn(&'c mut SqliteConnection, &'c ConsumeContext) -> MutationFuture<'c> + Send + Sync,
{
    fn apply<'c>(
        &'c self,
        tx: &'c mut SqliteConnection,
        ctx: &'c ConsumeContext,
    ) -> MutationFuture<'c> {
        self(tx, ctx)
    }
}

/// The outcome of an application (fresh or replayed).
#[derive(Debug, Clone, Serialize)]
pub struct ConsumeOutcome {
    /// `true` when THIS call applied the effect; `false` when it was a replay of
    /// an already-recorded application, which mutated nothing.
    pub applied: bool,
    pub action_identity: String,
    pub source_owned_card_id: String,
    /// `true` when the source instance was destroyed, `false` when a stack was
    /// decremented and survived.
    pub source_destroyed: bool,
    /// Remaining stack size after a decrement; `None` when the instance was
    /// destroyed or carried no stack.
    pub source_quantity_after: Option<i64>,
    pub target_owned_card_id: Option<String>,
    /// The caller mutation's own recorded summary, verbatim.
    pub effect: Value,
}

async fn fetch_owned(
    conn: &mut SqliteConnection,
    owned_card_id: &str,
    club_id: &str,
) -> AppResult<OwnedCard> {
    sqlx::query_as::<_, OwnedCard>(&format!("{OWNED_CARD_SELECT} WHERE id = ? AND club_id = ?"))
        .bind(owned_card_id)
        .bind(club_id)
        .fetch_optional(&mut *conn)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("owned card '{owned_card_id}' not found")))
}

fn require_kind(card: &OwnedCard, expected: ContentKind, role: &str) -> AppResult<()> {
    if card.content_kind != expected {
        return Err(AppError::BadRequest(format!(
            "{role} '{}' is content kind '{}', expected '{expected}'",
            card.id, card.content_kind
        )));
    }
    Ok(())
}

/// Whether this call did the work or found the identity already recorded.
///
/// The replay branch cannot read the recorded outcome while the transaction is
/// still open on this connection, so it is reported out of the transaction and
/// answered once the connection is free again.
enum Applied {
    Fresh(ConsumeOutcome),
    Replay,
}

/// Apply one consumable to one target, exactly once. See the module docs.
pub async fn consume_item<M: ItemMutation>(
    pool: &Pool,
    profile_id: &str,
    club_id: &str,
    req: &ConsumeRequest<'_>,
    mutation: &M,
) -> AppResult<ConsumeOutcome> {
    if req.action_identity.trim().is_empty() {
        return Err(AppError::BadRequest(
            "action_identity must not be empty".into(),
        ));
    }
    if let SourceConsumption::DecrementStack { amount } = req.consumption {
        if amount < 1 {
            return Err(AppError::BadRequest(
                "stack decrement amount must be >= 1".into(),
            ));
        }
    }
    if let ConsumeTarget::OwnedCard { owned_card_id, .. } = req.target {
        if owned_card_id == req.source_owned_card_id {
            return Err(AppError::BadRequest(
                "a consumable cannot be applied to itself".into(),
            ));
        }
    }

    // ONE connection with an explicit BEGIN IMMEDIATE, never a DEFERRED
    // `pool.begin()`: this transaction performs three reads before its first
    // write, and a deferred transaction only upgrades to a write at that first
    // write — where SQLite answers SQLITE_BUSY *immediately*, bypassing
    // `busy_timeout` (the full reasoning is on `economy::finish`). Two clients
    // applying consumables at once is ordinary traffic, so take the write lock
    // up front and let a rival wait instead of fail.
    let mut conn = pool.acquire().await?;
    sqlx::query("BEGIN IMMEDIATE").execute(&mut *conn).await?;

    let result = async {
        // 1. a replay is answered before anything is validated: a completed
        //    application has already destroyed (or drawn down) its source, so
        //    validating first would answer NotFound to a retry of a request that
        //    actually succeeded. The write lock is already held, so this read
        //    cannot race the guard INSERT below — which stays as the real
        //    guarantee for a writer on any other connection.
        if recorded(&mut conn, profile_id, req.action_identity).await? {
            return Ok(Applied::Replay);
        }

        // 2. source: owned by this club, and the kind the caller expected.
        let source = fetch_owned(&mut conn, req.source_owned_card_id, club_id).await?;
        require_kind(&source, req.expected_source_kind, "source item")?;

        // A source that is fielded in a squad cannot be consumed: `squad_players`
        // holds a FK onto `owned_cards(id)`, so the DELETE below would fail anyway.
        // Refuse explicitly instead of surfacing SQLITE_CONSTRAINT, and never
        // silently evict a lineup as a side effect of spending an item.
        let fielded = sqlx::query_scalar::<_, i64>(
            "SELECT COUNT(*) FROM squad_players WHERE owned_card_id = ?",
        )
        .bind(req.source_owned_card_id)
        .fetch_one(&mut *conn)
        .await?;
        if fielded > 0 {
            return Err(AppError::Conflict(format!(
                "source item '{}' is fielded in a squad and cannot be consumed",
                req.source_owned_card_id
            )));
        }

        // 3. target.
        let target = match req.target {
            ConsumeTarget::OwnedCard {
                owned_card_id,
                expected_kind,
            } => {
                let card = fetch_owned(&mut conn, owned_card_id, club_id).await?;
                require_kind(&card, expected_kind, "target item")?;
                Some(card)
            }
            ConsumeTarget::Club => None,
        };

        // 4. replay guard — the durable one. It precedes the mutation and the
        //    consumption, so a duplicate that got past step 1 on another
        //    connection still cannot apply a second effect or spend a second
        //    charge. The recorded `effect` is filled in below, once the mutation
        //    has produced it.
        let application_id = Uuid::new_v4().to_string();
        let now = Utc::now().to_rfc3339();
        let guard = sqlx::query(
            "INSERT INTO consumable_applications \
             (id, profile_id, action_identity, source_owned_card_id, source_card_id, \
              source_content_kind, source_consumed, source_quantity_after, \
              target_owned_card_id, effect, applied_at) \
             VALUES (?, ?, ?, ?, ?, ?, 0, NULL, ?, '', ?)",
        )
        .bind(&application_id)
        .bind(profile_id)
        .bind(req.action_identity)
        .bind(&source.id)
        .bind(&source.card_id)
        .bind(source.content_kind.as_str())
        .bind(target.as_ref().map(|t| t.id.as_str()))
        .bind(&now)
        .execute(&mut *conn)
        .await;
        match guard {
            Ok(_) => {}
            // A collision writes nothing, so this transaction has nothing to undo
            // and simply ends; the recorded outcome is read back afterwards.
            Err(sqlx::Error::Database(e)) if e.is_unique_violation() => return Ok(Applied::Replay),
            Err(e) => return Err(e.into()),
        }

        // 5. the caller's effect on the target, inside this transaction.
        let ctx = ConsumeContext {
            profile_id: profile_id.to_string(),
            club_id: club_id.to_string(),
            source,
            target,
        };
        let effect = mutation.apply(&mut conn, &ctx).await?;

        // 6. consume the source exactly once. Both paths assert rows_affected == 1,
        //    so a concurrent spend of the same instance (which lost the SQLite write
        //    lock and now sees the row gone / already decremented) fails instead of
        //    granting a second effect.
        let (source_destroyed, source_quantity_after) = match req.consumption {
            SourceConsumption::DestroyInstance => {
                let deleted = sqlx::query("DELETE FROM owned_cards WHERE id = ? AND club_id = ?")
                    .bind(&ctx.source.id)
                    .bind(club_id)
                    .execute(&mut *conn)
                    .await?
                    .rows_affected();
                if deleted != 1 {
                    return Err(AppError::Conflict(format!(
                        "source item '{}' was already consumed",
                        ctx.source.id
                    )));
                }
                (true, None)
            }
            SourceConsumption::DecrementStack { amount } => {
                let Some(have) = ctx.source.quantity else {
                    return Err(AppError::BadRequest(format!(
                        "source item '{}' carries no stack size; it can only be destroyed",
                        ctx.source.id
                    )));
                };
                if have < amount {
                    return Err(AppError::Conflict(format!(
                        "source item '{}' holds {have}, cannot consume {amount}",
                        ctx.source.id
                    )));
                }
                let remaining = have - amount;
                if remaining == 0 {
                    let deleted = sqlx::query(
                        "DELETE FROM owned_cards WHERE id = ? AND club_id = ? AND quantity = ?",
                    )
                    .bind(&ctx.source.id)
                    .bind(club_id)
                    .bind(have)
                    .execute(&mut *conn)
                    .await?
                    .rows_affected();
                    if deleted != 1 {
                        return Err(AppError::Conflict(format!(
                            "source item '{}' changed under us",
                            ctx.source.id
                        )));
                    }
                    (true, None)
                } else {
                    let updated = sqlx::query(
                        "UPDATE owned_cards SET quantity = ? WHERE id = ? AND club_id = ? \
                         AND quantity = ?",
                    )
                    .bind(remaining)
                    .bind(&ctx.source.id)
                    .bind(club_id)
                    .bind(have)
                    .execute(&mut *conn)
                    .await?
                    .rows_affected();
                    if updated != 1 {
                        return Err(AppError::Conflict(format!(
                            "source item '{}' changed under us",
                            ctx.source.id
                        )));
                    }
                    (false, Some(remaining))
                }
            }
        };

        let effect_text = serde_json::to_string(&effect)?;
        sqlx::query(
            "UPDATE consumable_applications \
             SET effect = ?, source_consumed = ?, source_quantity_after = ? WHERE id = ?",
        )
        .bind(&effect_text)
        .bind(i64::from(source_destroyed))
        .bind(source_quantity_after)
        .bind(&application_id)
        .execute(&mut *conn)
        .await?;

        Ok(Applied::Fresh(ConsumeOutcome {
            applied: true,
            action_identity: req.action_identity.to_string(),
            source_owned_card_id: ctx.source.id.clone(),
            source_destroyed,
            source_quantity_after,
            target_owned_card_id: ctx.target.as_ref().map(|t| t.id.clone()),
            effect,
        }))
    }
    .await;

    match economy::finish(&mut conn, result).await? {
        Applied::Fresh(outcome) => Ok(outcome),
        // Read the recorded outcome on the SAME connection: acquiring a second one
        // while still holding this one deadlocks a pool saturated with racing
        // appliers, which is exactly the case a replay shows up in.
        Applied::Replay => already_applied(&mut conn, profile_id, req.action_identity).await,
    }
}

/// Has this identity already been applied? Read inside the transaction, under
/// the write lock, so the answer cannot go stale before the guard INSERT.
async fn recorded(
    conn: &mut SqliteConnection,
    profile_id: &str,
    action_identity: &str,
) -> AppResult<bool> {
    let hits = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM consumable_applications \
         WHERE profile_id = ? AND action_identity = ?",
    )
    .bind(profile_id)
    .bind(action_identity)
    .fetch_one(&mut *conn)
    .await?;
    Ok(hits > 0)
}

/// Echo the recorded outcome of an application that already happened. Mutates
/// nothing and reports `applied = false`.
async fn already_applied(
    conn: &mut SqliteConnection,
    profile_id: &str,
    action_identity: &str,
) -> AppResult<ConsumeOutcome> {
    let row = sqlx::query_as::<_, (String, i64, Option<i64>, Option<String>, String)>(
        "SELECT source_owned_card_id, source_consumed, source_quantity_after, \
         target_owned_card_id, effect FROM consumable_applications \
         WHERE profile_id = ? AND action_identity = ?",
    )
    .bind(profile_id)
    .bind(action_identity)
    .fetch_optional(&mut *conn)
    .await?
    .ok_or_else(|| {
        AppError::Internal(anyhow::anyhow!(
            "consumable_applications row missing after unique violation"
        ))
    })?;
    let (source_owned_card_id, source_consumed, source_quantity_after, target, effect_text) = row;
    Ok(ConsumeOutcome {
        applied: false,
        action_identity: action_identity.to_string(),
        source_owned_card_id,
        source_destroyed: source_consumed != 0,
        source_quantity_after,
        target_owned_card_id: target,
        // Written by this module as JSON, so a parse failure is corruption, not
        // an expected case — surface it instead of quietly returning null.
        effect: serde_json::from_str(&effect_text)?,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db;
    use serde_json::json;

    const TS: &str = "2026-01-01T00:00:00Z";

    /// A file-backed pool (so a "restart" can reopen the same DB) with one club
    /// holding: a stacked consumable, bare consumables, players (one fielded), a
    /// kit, and a consumable owned by ANOTHER club.
    async fn fixture() -> (tempfile::TempDir, String, db::Pool) {
        let dir = tempfile::tempdir().expect("tempdir");
        let url = format!("sqlite://{}", dir.path().join("core.db").display());
        let pool = db::init_pool(&url, 5).await.expect("init pool");
        db::run_migrations(&pool).await.expect("migrations");

        for (profile, club) in [("prof-a", "club-a"), ("prof-b", "club-b")] {
            sqlx::query(
                "INSERT INTO profiles (id, username, created_at, updated_at) VALUES (?, ?, ?, ?)",
            )
            .bind(profile)
            .bind(profile)
            .bind(TS)
            .bind(TS)
            .execute(&pool)
            .await
            .expect("profile");
            sqlx::query(
                "INSERT INTO clubs (id, profile_id, name, coins, created_at, updated_at) \
                 VALUES (?, ?, ?, 0, ?, ?)",
            )
            .bind(club)
            .bind(profile)
            .bind(club)
            .bind(TS)
            .bind(TS)
            .execute(&pool)
            .await
            .expect("club");
        }
        for (id, club, kind, quantity) in [
            ("stack", "club-a", ContentKind::Consumable, Some(15i64)),
            ("single", "club-a", ContentKind::Consumable, None),
            ("single2", "club-a", ContentKind::Consumable, None),
            ("player", "club-a", ContentKind::Player, None),
            ("fielded", "club-a", ContentKind::Player, None),
            ("kit", "club-a", ContentKind::Kit, None),
            ("foreign", "club-b", ContentKind::Consumable, None),
        ] {
            sqlx::query(
                "INSERT INTO owned_cards \
                 (id, club_id, card_id, is_loan, acquired_at, content_kind, quantity) \
                 VALUES (?, ?, ?, 0, ?, ?, ?)",
            )
            .bind(id)
            .bind(club)
            .bind(format!("def-{id}"))
            .bind(TS)
            .bind(kind.as_str())
            .bind(quantity)
            .execute(&pool)
            .await
            .expect("owned card");
        }
        sqlx::query(
            "INSERT INTO squads (id, club_id, name, formation, created_at, updated_at) \
             VALUES ('sq-a', 'club-a', 'S', '4-4-2', ?, ?)",
        )
        .bind(TS)
        .bind(TS)
        .execute(&pool)
        .await
        .expect("squad");
        sqlx::query(
            "INSERT INTO squad_players (id, squad_id, owned_card_id, position_index) \
             VALUES ('sp-1', 'sq-a', 'fielded', 0)",
        )
        .execute(&pool)
        .await
        .expect("squad player");
        (dir, url, pool)
    }

    /// Bumps the target's training bonus — a stand-in for a caller-owned effect.
    /// Core supplies no formula; this one lives entirely in the test.
    struct BumpTraining;

    impl ItemMutation for BumpTraining {
        fn apply<'c>(
            &'c self,
            tx: &'c mut SqliteConnection,
            ctx: &'c ConsumeContext,
        ) -> MutationFuture<'c> {
            Box::pin(async move {
                let target = ctx.target.as_ref().expect("target required");
                sqlx::query(
                    "UPDATE owned_cards SET training_bonus = training_bonus + 1 WHERE id = ?",
                )
                .bind(&target.id)
                .execute(&mut *tx)
                .await?;
                Ok(json!({ "training_bonus_delta": 1 }))
            })
        }
    }

    /// A mutation that always fails, to prove the whole transaction unwinds.
    struct Failing;

    impl ItemMutation for Failing {
        fn apply<'c>(
            &'c self,
            _tx: &'c mut SqliteConnection,
            _ctx: &'c ConsumeContext,
        ) -> MutationFuture<'c> {
            Box::pin(async move { Err(AppError::BadRequest("effect refused".into())) })
        }
    }

    /// Coerces a closure into the higher-ranked shape the blanket [`ItemMutation`]
    /// impl requires — proving a caller can pass an inline effect, not just a
    /// named type.
    fn mutation<F>(f: F) -> F
    where
        F: for<'c> Fn(&'c mut SqliteConnection, &'c ConsumeContext) -> MutationFuture<'c>
            + Send
            + Sync,
    {
        f
    }

    fn req<'a>(
        identity: &'a str,
        source: &'a str,
        consumption: SourceConsumption,
        target: &'a str,
    ) -> ConsumeRequest<'a> {
        ConsumeRequest {
            action_identity: identity,
            source_owned_card_id: source,
            expected_source_kind: ContentKind::Consumable,
            consumption,
            target: ConsumeTarget::OwnedCard {
                owned_card_id: target,
                expected_kind: ContentKind::Player,
            },
        }
    }

    async fn count(pool: &db::Pool, sql: &str) -> i64 {
        sqlx::query_scalar::<_, i64>(sql)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    async fn training(pool: &db::Pool, id: &str) -> i64 {
        sqlx::query_scalar::<_, i64>("SELECT training_bonus FROM owned_cards WHERE id = ?")
            .bind(id)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    async fn quantity(pool: &db::Pool, id: &str) -> Option<i64> {
        sqlx::query_scalar::<_, Option<i64>>("SELECT quantity FROM owned_cards WHERE id = ?")
            .bind(id)
            .fetch_optional(pool)
            .await
            .unwrap()
            .flatten()
    }

    #[tokio::test]
    async fn applies_effect_and_destroys_the_instance() {
        let (_dir, _url, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "act-1",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect("apply");

        assert!(out.applied);
        assert!(out.source_destroyed);
        assert_eq!(out.source_quantity_after, None);
        assert_eq!(out.effect, json!({ "training_bonus_delta": 1 }));
        assert_eq!(training(&pool, "player").await, 1, "effect landed");
        assert_eq!(
            count(
                &pool,
                "SELECT COUNT(*) FROM owned_cards WHERE id = 'single'"
            )
            .await,
            0,
            "a consumed card must no longer be owned"
        );
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            1
        );
    }

    /// A retry of a request that ALREADY succeeded must replay, not 404. With
    /// `DestroyInstance` the source no longer exists by then, so answering the
    /// replay has to precede source validation — otherwise a client that lost the
    /// response to a successful apply is told its item was never there.
    #[tokio::test]
    async fn a_replay_survives_the_source_it_destroyed() {
        let (_dir, _url, pool) = fixture().await;
        let spend = || {
            req(
                "act-gone",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            )
        };
        let first = consume_item(&pool, "prof-a", "club-a", &spend(), &BumpTraining)
            .await
            .expect("first");
        assert!(first.applied);

        let replay = consume_item(&pool, "prof-a", "club-a", &spend(), &BumpTraining)
            .await
            .expect("a retry must replay, not fail on the destroyed source");
        assert!(!replay.applied);
        assert!(replay.source_destroyed);
        assert_eq!(replay.effect, first.effect);
        assert_eq!(training(&pool, "player").await, 1, "effect applied once");
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            1
        );
    }

    #[tokio::test]
    async fn inline_closure_effect_is_accepted() {
        let (_dir, _url, pool) = fixture().await;
        let effect = mutation(|tx: &mut SqliteConnection, ctx: &ConsumeContext| {
            let target = ctx.target.as_ref().expect("target").id.clone();
            Box::pin(async move {
                sqlx::query("UPDATE owned_cards SET chemistry_style = 'anchor' WHERE id = ?")
                    .bind(&target)
                    .execute(&mut *tx)
                    .await?;
                Ok(json!({ "chemistry_style": "anchor" }))
            }) as MutationFuture<'_>
        });
        let out = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "act-closure",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            ),
            &effect,
        )
        .await
        .expect("apply");
        assert!(out.applied);
        let style = sqlx::query_scalar::<_, String>(
            "SELECT chemistry_style FROM owned_cards WHERE id = 'player'",
        )
        .fetch_one(&pool)
        .await
        .unwrap();
        assert_eq!(style, "anchor");
    }

    /// The core replay guarantee: the same identity twice = ONE mutation and ONE
    /// charge, with the recorded outcome echoed back as `applied = false`.
    #[tokio::test]
    async fn replay_of_one_identity_mutates_once() {
        let (dir, url, pool) = fixture().await;
        let spend = |identity| {
            req(
                identity,
                "stack",
                SourceConsumption::DecrementStack { amount: 5 },
                "player",
            )
        };
        let first = consume_item(&pool, "prof-a", "club-a", &spend("act-1"), &BumpTraining)
            .await
            .expect("first");
        assert!(first.applied);
        assert_eq!(first.source_quantity_after, Some(10));

        let replay = consume_item(&pool, "prof-a", "club-a", &spend("act-1"), &BumpTraining)
            .await
            .expect("replay");
        assert!(!replay.applied, "a replay must not re-apply");
        assert_eq!(replay.source_quantity_after, Some(10));
        assert_eq!(replay.effect, json!({ "training_bonus_delta": 1 }));
        assert_eq!(training(&pool, "player").await, 1, "effect applied once");
        assert_eq!(quantity(&pool, "stack").await, Some(10), "charged once");
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            1
        );

        // RESTART: the guard is durable, not in-memory.
        pool.close().await;
        let reopened = db::init_pool(&url, 5).await.expect("reopen");
        db::run_migrations(&reopened).await.expect("migrations");
        let after_restart = consume_item(
            &reopened,
            "prof-a",
            "club-a",
            &spend("act-1"),
            &BumpTraining,
        )
        .await
        .expect("restart replay");
        assert!(!after_restart.applied);
        assert_eq!(quantity(&reopened, "stack").await, Some(10));
        assert_eq!(training(&reopened, "player").await, 1);
        drop(dir);
    }

    #[tokio::test]
    async fn concurrent_duplicates_apply_exactly_once() {
        let (_dir, url, pool) = fixture().await;
        drop(pool);
        let pool = db::init_pool(&url, 8).await.expect("pool");

        let mut handles = Vec::new();
        for _ in 0..6 {
            let p = pool.clone();
            handles.push(tokio::spawn(async move {
                consume_item(
                    &p,
                    "prof-a",
                    "club-a",
                    &req(
                        "race",
                        "stack",
                        SourceConsumption::DecrementStack { amount: 3 },
                        "player",
                    ),
                    &BumpTraining,
                )
                .await
            }));
        }
        let mut applied = 0;
        for h in handles {
            if let Ok(Ok(out)) = h.await {
                if out.applied {
                    applied += 1;
                }
            }
        }
        assert_eq!(applied, 1, "exactly one racer applies the effect");
        assert_eq!(quantity(&pool, "stack").await, Some(12), "charged once");
        assert_eq!(training(&pool, "player").await, 1);
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            1
        );
    }

    #[tokio::test]
    async fn stack_is_destroyed_when_it_reaches_zero() {
        let (_dir, _url, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "act-all",
                "stack",
                SourceConsumption::DecrementStack { amount: 15 },
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect("apply");
        assert!(out.source_destroyed);
        assert_eq!(out.source_quantity_after, None);
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM owned_cards WHERE id = 'stack'").await,
            0
        );
    }

    #[tokio::test]
    async fn overdrawing_a_stack_is_refused_whole() {
        let (_dir, _url, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "act-over",
                "stack",
                SourceConsumption::DecrementStack { amount: 16 },
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect_err("cannot spend more than is held");
        assert!(matches!(err, AppError::Conflict(_)), "got {err:?}");
        assert_eq!(quantity(&pool, "stack").await, Some(15));
        assert_eq!(training(&pool, "player").await, 0, "effect rolled back");
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            0
        );
    }

    #[tokio::test]
    async fn decrementing_a_non_stack_is_refused() {
        let (_dir, _url, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "act-nostack",
                "single",
                SourceConsumption::DecrementStack { amount: 1 },
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect_err("a bare instance has no count to decrement");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert_eq!(
            count(
                &pool,
                "SELECT COUNT(*) FROM owned_cards WHERE id = 'single'"
            )
            .await,
            1,
            "and it must NOT be silently destroyed instead"
        );
    }

    #[tokio::test]
    async fn a_failing_effect_rolls_back_the_charge() {
        let (_dir, _url, pool) = fixture().await;
        let spend = || {
            req(
                "act-fail",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            )
        };
        let err = consume_item(&pool, "prof-a", "club-a", &spend(), &Failing)
            .await
            .expect_err("effect refused");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert_eq!(
            count(
                &pool,
                "SELECT COUNT(*) FROM owned_cards WHERE id = 'single'"
            )
            .await,
            1,
            "the source must survive an unapplied effect"
        );
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            0,
            "and the guard must not block a legitimate retry"
        );

        // The retry with the SAME identity now succeeds, because nothing landed.
        let out = consume_item(&pool, "prof-a", "club-a", &spend(), &BumpTraining)
            .await
            .expect("retry");
        assert!(out.applied);
    }

    #[tokio::test]
    async fn validates_ownership_and_kinds_before_anything_moves() {
        let (_dir, _url, pool) = fixture().await;

        // Source owned by another club.
        assert!(consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "v1",
                "foreign",
                SourceConsumption::DestroyInstance,
                "player"
            ),
            &BumpTraining,
        )
        .await
        .is_err());

        // Source is not the kind the caller expected (a kit is not a consumable).
        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req("v2", "kit", SourceConsumption::DestroyInstance, "player"),
            &BumpTraining,
        )
        .await
        .expect_err("kind mismatch");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");

        // Target is not the kind the caller expected (a kit is not a player).
        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req("v3", "single", SourceConsumption::DestroyInstance, "kit"),
            &BumpTraining,
        )
        .await
        .expect_err("target kind mismatch");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");

        // Target owned by another club.
        assert!(consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "v4",
                "single",
                SourceConsumption::DestroyInstance,
                "foreign"
            ),
            &BumpTraining,
        )
        .await
        .is_err());

        // Applying an item to itself.
        assert!(consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req("v5", "single", SourceConsumption::DestroyInstance, "single"),
            &BumpTraining,
        )
        .await
        .is_err());

        // An empty identity has no replay identity at all.
        assert!(consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req("  ", "single", SourceConsumption::DestroyInstance, "player"),
            &BumpTraining,
        )
        .await
        .is_err());

        assert_eq!(count(&pool, "SELECT COUNT(*) FROM owned_cards").await, 7);
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            0
        );
        assert_eq!(training(&pool, "player").await, 0);
    }

    /// A source fielded in a squad is refused explicitly — never destroyed, and
    /// never silently evicted from the lineup as a side effect.
    #[tokio::test]
    async fn a_fielded_source_cannot_be_consumed() {
        let (_dir, _url, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &ConsumeRequest {
                action_identity: "act-fielded",
                source_owned_card_id: "fielded",
                expected_source_kind: ContentKind::Player,
                consumption: SourceConsumption::DestroyInstance,
                target: ConsumeTarget::Club,
            },
            &mutation(|_tx: &mut SqliteConnection, _ctx: &ConsumeContext| {
                Box::pin(async move { Ok(Value::Null) }) as MutationFuture<'_>
            }),
        )
        .await
        .expect_err("a fielded item cannot be consumed");
        assert!(matches!(err, AppError::Conflict(_)), "got {err:?}");
        assert_eq!(count(&pool, "SELECT COUNT(*) FROM squad_players").await, 1);
        assert_eq!(
            count(
                &pool,
                "SELECT COUNT(*) FROM owned_cards WHERE id = 'fielded'"
            )
            .await,
            1
        );
    }

    /// Lifecycle invariant: once consumed, the instance is gone — a second spend
    /// under a DIFFERENT identity cannot resurrect it.
    #[tokio::test]
    async fn a_consumed_instance_cannot_be_spent_again() {
        let (_dir, _url, pool) = fixture().await;
        consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "first",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect("first spend");

        let err = consume_item(
            &pool,
            "prof-a",
            "club-a",
            &req(
                "second",
                "single",
                SourceConsumption::DestroyInstance,
                "player",
            ),
            &BumpTraining,
        )
        .await
        .expect_err("a consumed instance is no longer owned");
        assert!(matches!(err, AppError::NotFound(_)), "got {err:?}");
        assert_eq!(training(&pool, "player").await, 1, "effect applied once");
    }

    /// Two DISTINCT instances of the same definition are two separate charges —
    /// the real profile owns exactly that shape (two copies of one resourceId),
    /// so consuming one must leave the other spendable.
    #[tokio::test]
    async fn two_instances_of_one_definition_are_two_charges() {
        let (_dir, _url, pool) = fixture().await;
        for (identity, source) in [("i1", "single"), ("i2", "single2")] {
            let out = consume_item(
                &pool,
                "prof-a",
                "club-a",
                &req(
                    identity,
                    source,
                    SourceConsumption::DestroyInstance,
                    "player",
                ),
                &BumpTraining,
            )
            .await
            .expect("spend");
            assert!(out.applied);
        }
        assert_eq!(training(&pool, "player").await, 2);
        assert_eq!(
            count(&pool, "SELECT COUNT(*) FROM consumable_applications").await,
            2
        );
    }
}