//! The CLOSED vocabulary of instance mutations Core can execute for a caller
//! that is not in-process.
//!
//! [`crate::services::consume::consume_item`] takes an
//! [`ItemMutation`] — an in-process closure, which cannot cross an HTTP
//! boundary. A game host applying a consumable over HTTP therefore cannot SUPPLY
//! its effect; it can only DESCRIBE one, and Core executes the description.
//!
//! That does not move the game's formula into Core. The caller still owns every
//! number: how many match-contracts a given card grants a given target is the
//! adapter's reversed per-game table, and it arrives here as `amount`. Core owns
//! only what it can prove without knowing the game — the arithmetic, the clamp,
//! the ownership scope, the loan invariant and the transaction. This is the same
//! split quick-sell already uses (the host prices the item, Core moves it).
//!
//! The enum is closed ON PURPOSE. A generic "set field X to value Y" escape
//! hatch would hand the host arbitrary write access to Core state and make every
//! present and future invariant unenforceable; a new effect is a new variant,
//! validated here, reviewed here.

use chrono::Utc;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sqlx::SqliteConnection;

use crate::{
    error::{AppError, AppResult},
    models::card::OwnedCard,
    services::consume::{ConsumeContext, ItemMutation, MutationFuture},
};

/// One wire-describable mutation of a target instance.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum InstanceEffect {
    /// Add match-contracts to the target, saturating at `cap`.
    ///
    /// `default_when_unset` is what a target Core tracks no contract for counts
    /// as — the caller's pack-fresh starting value (FIFA 17: 7). Core has no such
    /// default of its own, which is precisely why `owned_cards.contract_matches`
    /// is nullable; see migration 0028.
    AddContractMatches {
        amount: i64,
        cap: i64,
        default_when_unset: i64,
    },
    /// Attach an attribute training effect to the target, REPLACING any the
    /// instance already carries.
    ///
    /// `attribute_index` is a slot in Core's own six-attribute card model, in
    /// `CardDefinition` declaration order (0 pace .. 5 physical), or `None` for
    /// an effect that boosts ALL SIX slots. Naming the slot rather than the
    /// game's attribute is what keeps this game-neutral: that FIFA 17's "GK
    /// speed" is slot 4 is the adapter's reversed knowledge, and it stays there.
    ///
    /// REPLACEMENT, NOT ACCUMULATION, and at most one effect per instance. Both
    /// halves are the caller's game rule, but they are enforced here because the
    /// storage shape is Core's: FIFA 17's own documentation states "you can only
    /// boost one attribute or all six" and "when you apply a new training card
    /// to a player, he loses the improved attributes of previous training cards.
    /// It does not accumulate, it replaces."
    ///
    /// `max_amount` is the caller's authored ceiling for the specific family
    /// (FIFA 17: 15 single-attribute, 10 for the rare all-six card). Core cannot
    /// know it, but it can refuse anything above the number the caller itself
    /// declares, which is what stops a host describing a "+99 pace" no card
    /// could grant.
    ApplyTraining {
        attribute_index: Option<i64>,
        amount: i64,
        max_amount: i64,
    },
}

impl ItemMutation for InstanceEffect {
    fn apply<'c>(
        &'c self,
        tx: &'c mut SqliteConnection,
        ctx: &'c ConsumeContext,
    ) -> MutationFuture<'c> {
        match self {
            InstanceEffect::AddContractMatches {
                amount,
                cap,
                default_when_unset,
            } => Box::pin(add_contract_matches(
                tx,
                ctx,
                *amount,
                *cap,
                *default_when_unset,
            )),
            InstanceEffect::ApplyTraining {
                attribute_index,
                amount,
                max_amount,
            } => Box::pin(apply_training(
                tx,
                ctx,
                *attribute_index,
                *amount,
                *max_amount,
            )),
        }
    }
}

/// The target instance this effect mutates. An effect that writes to an instance
/// has nothing to write to when the application is club-scoped, so that is a
/// refusal rather than a silent no-op that still spends the source.
fn require_target<'a>(ctx: &'a ConsumeContext, effect: &str) -> AppResult<&'a OwnedCard> {
    ctx.target
        .as_ref()
        .ok_or_else(|| AppError::BadRequest(format!("effect '{effect}' requires a target item")))
}

async fn add_contract_matches(
    tx: &mut SqliteConnection,
    ctx: &ConsumeContext,
    amount: i64,
    cap: i64,
    default_when_unset: i64,
) -> AppResult<Value> {
    let target = require_target(ctx, "add_contract_matches")?;

    if amount < 1 {
        return Err(AppError::BadRequest(format!(
            "add_contract_matches amount must be >= 1, got {amount}"
        )));
    }
    if cap < 1 {
        return Err(AppError::BadRequest(format!(
            "add_contract_matches cap must be >= 1, got {cap}"
        )));
    }
    if default_when_unset < 0 {
        return Err(AppError::BadRequest(format!(
            "add_contract_matches default_when_unset must be >= 0, got {default_when_unset}"
        )));
    }
    // A loan item is borrowed for a fixed number of matches
    // (`loan_matches_remaining`); topping up its contract would pretend to extend
    // something Core does not own. Core models loans, so the invariant is Core's.
    if target.is_loan {
        return Err(AppError::BadRequest(format!(
            "contracts cannot be applied to a loan item ('{}')",
            target.id
        )));
    }

    // Read-modify-write INSIDE the caller's transaction, deliberately re-reading
    // rather than trusting the snapshot in `ctx`: the read and the write then sit
    // in one contiguous critical section under the same write lock, so two
    // concurrent applies serialise instead of both computing from one `before`
    // and losing an update.
    let before = sqlx::query_scalar::<_, i64>(
        "SELECT COALESCE(contract_matches, ?) FROM owned_cards WHERE id = ? AND club_id = ?",
    )
    .bind(default_when_unset)
    .bind(&target.id)
    .bind(&ctx.club_id)
    .fetch_optional(&mut *tx)
    .await?
    .ok_or_else(|| AppError::NotFound(format!("target item '{}' not found", target.id)))?;

    // Saturating, because `before + amount` is caller-supplied arithmetic and an
    // overflow must not panic a request thread; the clamp makes the sum's exact
    // magnitude irrelevant anyway.
    let after = before.saturating_add(amount).min(cap);

    let updated =
        sqlx::query("UPDATE owned_cards SET contract_matches = ? WHERE id = ? AND club_id = ?")
            .bind(after)
            .bind(&target.id)
            .bind(&ctx.club_id)
            .execute(&mut *tx)
            .await?
            .rows_affected();
    if updated != 1 {
        return Err(AppError::Conflict(format!(
            "target item '{}' changed under us",
            target.id
        )));
    }

    // `granted` is what the caller's table awarded, NOT `after - before`: the cap
    // can swallow part of it, and the two numbers answer different questions
    // (what the card was worth vs. what the instance now holds).
    Ok(json!({
        "kind": "add_contract_matches",
        "granted": amount,
        "before": before,
        "after": after,
    }))
}

/// Core's own six-attribute card model is this wide. A slot outside it cannot
/// name anything Core can project.
const ATTRIBUTE_SLOTS: i64 = 6;

async fn apply_training(
    tx: &mut SqliteConnection,
    ctx: &ConsumeContext,
    attribute_index: Option<i64>,
    amount: i64,
    max_amount: i64,
) -> AppResult<Value> {
    let target = require_target(ctx, "apply_training")?;

    if let Some(slot) = attribute_index {
        if !(0..ATTRIBUTE_SLOTS).contains(&slot) {
            return Err(AppError::BadRequest(format!(
                "apply_training attribute_index must be 0..{ATTRIBUTE_SLOTS} or absent, got {slot}"
            )));
        }
    }
    if amount < 1 {
        return Err(AppError::BadRequest(format!(
            "apply_training amount must be >= 1, got {amount}"
        )));
    }
    // The caller declares its own family's authored ceiling and is then held to
    // it. Without this a host could describe an arbitrary boost through a
    // vocabulary that exists precisely to prevent that.
    if !(1..=99).contains(&max_amount) {
        return Err(AppError::BadRequest(format!(
            "apply_training max_amount must be 1..=99, got {max_amount}"
        )));
    }
    if amount > max_amount {
        return Err(AppError::BadRequest(format!(
            "apply_training amount {amount} exceeds the caller's declared maximum {max_amount}"
        )));
    }
    // Same reasoning as contracts: a loan is borrowed for a fixed run of
    // matches, so durably improving it would outlive the thing it is attached
    // to. Conservative and consistent rather than reversed -- no FIFA 17 source
    // speaks to training a loan item.
    if target.is_loan {
        return Err(AppError::BadRequest(format!(
            "training cannot be applied to a loan item ('{}')",
            target.id
        )));
    }

    // REPLACE. One instance carries at most one training effect, and a new card
    // supersedes whatever was there -- including an effect on a DIFFERENT slot,
    // because FIFA 17 allows "one attribute or all six" and never a mixture.
    // Delete-then-insert inside the caller's transaction, so the old effect can
    // never survive a failed insert and the two can never coexist.
    let previous = sqlx::query_as::<_, (Option<i64>, i64, String)>(
        "SELECT attribute_index, amount, source_card_id FROM owned_card_training \
         WHERE owned_card_id = ?",
    )
    .bind(&target.id)
    .fetch_optional(&mut *tx)
    .await?;

    sqlx::query("DELETE FROM owned_card_training WHERE owned_card_id = ?")
        .bind(&target.id)
        .execute(&mut *tx)
        .await?;

    sqlx::query(
        "INSERT INTO owned_card_training \
           (owned_card_id, attribute_index, amount, source_card_id, applied_at) \
         VALUES (?, ?, ?, ?, ?)",
    )
    .bind(&target.id)
    .bind(attribute_index)
    .bind(amount)
    .bind(&ctx.source.card_id)
    .bind(Utc::now().to_rfc3339())
    .execute(&mut *tx)
    .await?;

    Ok(json!({
        "kind": "apply_training",
        "attribute_index": attribute_index,
        "amount": amount,
        // Named `granted` as well as `amount` so every effect's recorded outcome
        // answers "what did this card award" under one key, whatever the family.
        "granted": amount,
        // `before`/`after` describe the training this instance holds, not the
        // attribute's value: Core stores effects, and the attribute total is a
        // projection over a definition Core does not consult here. `before` is
        // the magnitude of the effect this one replaced, 0 when there was none.
        "before": previous.as_ref().map(|(_, a, _)| *a).unwrap_or(0),
        "after": amount,
        // What was displaced, so the outcome records the replacement rather than
        // silently overwriting history.
        "replaced": previous.map(|(slot, amt, src)| json!({
            "attribute_index": slot,
            "amount": amt,
            "source_card_id": src,
        })),
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db;
    use crate::models::card::ContentKind;
    use crate::services::consume::{
        consume_item, ConsumeRequest, ConsumeTarget, SourceConsumption,
    };

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

    /// One club holding contract consumables and three player targets: one Core
    /// tracks no contract for, one part-way through its contract, and one on loan.
    async fn fixture() -> (tempfile::TempDir, 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");

        sqlx::query("INSERT INTO profiles (id, username, created_at, updated_at) VALUES (?,?,?,?)")
            .bind("prof")
            .bind("prof")
            .bind(TS)
            .bind(TS)
            .execute(&pool)
            .await
            .expect("profile");
        sqlx::query(
            "INSERT INTO clubs (id, profile_id, name, coins, created_at, updated_at) \
             VALUES ('club', 'prof', 'club', 0, ?, ?)",
        )
        .bind(TS)
        .bind(TS)
        .execute(&pool)
        .await
        .expect("club");

        for (id, kind, is_loan, contract) in [
            ("card-1", ContentKind::Consumable, 0, None),
            ("card-2", ContentKind::Consumable, 0, None),
            ("card-3", ContentKind::Consumable, 0, None),
            ("fresh", ContentKind::Player, 0, None),
            ("used", ContentKind::Player, 0, Some(90i64)),
            ("loaned", ContentKind::Player, 1, None),
        ] {
            sqlx::query(
                "INSERT INTO owned_cards \
                 (id, club_id, card_id, is_loan, acquired_at, content_kind, contract_matches) \
                 VALUES (?, 'club', ?, ?, ?, ?, ?)",
            )
            .bind(id)
            .bind(format!("def-{id}"))
            .bind(is_loan)
            .bind(TS)
            .bind(kind.as_str())
            .bind(contract)
            .execute(&pool)
            .await
            .expect("owned card");
        }
        (dir, pool)
    }

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

    fn grant(amount: i64) -> InstanceEffect {
        InstanceEffect::AddContractMatches {
            amount,
            cap: 99,
            default_when_unset: 7,
        }
    }

    fn train(attribute_index: i64, amount: i64) -> InstanceEffect {
        InstanceEffect::ApplyTraining {
            attribute_index: Some(attribute_index),
            amount,
            max_amount: 15,
        }
    }

    /// The rare card: every slot, ceiling 10.
    fn train_all(amount: i64) -> InstanceEffect {
        InstanceEffect::ApplyTraining {
            attribute_index: None,
            amount,
            max_amount: 10,
        }
    }

    async fn training_of(pool: &db::Pool, id: &str) -> Vec<(Option<i64>, i64, String)> {
        sqlx::query_as::<_, (Option<i64>, i64, String)>(
            "SELECT attribute_index, amount, source_card_id FROM owned_card_training \
             WHERE owned_card_id = ?",
        )
        .bind(id)
        .fetch_all(pool)
        .await
        .expect("read training")
    }

    async fn source_exists(pool: &db::Pool, id: &str) -> bool {
        sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM owned_cards WHERE id = ?")
            .bind(id)
            .fetch_one(pool)
            .await
            .expect("count source")
            == 1
    }

    /// The whole point: one card, one slot, one consumed source, recorded
    /// against the definition that granted it.
    #[tokio::test]
    async fn training_attaches_to_the_slot_and_spends_the_card() {
        let (_dir, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(4, 10),
        )
        .await
        .expect("apply");

        assert!(out.applied);
        assert!(out.source_destroyed);
        assert_eq!(
            out.effect,
            json!({
                "kind": "apply_training",
                "attribute_index": 4,
                "amount": 10,
                "granted": 10,
                "before": 0,
                "after": 10,
                "replaced": null,
            })
        );
        assert_eq!(
            training_of(&pool, "fresh").await,
            vec![(Some(4), 10, "def-card-1".to_string())]
        );
        assert!(!source_exists(&pool, "card-1").await);
    }

    /// A second card on the SAME slot REPLACES the first and does not
    /// accumulate: 10 then 15 leaves 15, never 25.
    #[tokio::test]
    async fn a_second_training_on_the_same_slot_replaces_rather_than_accumulating() {
        let (_dir, pool) = fixture().await;
        consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(4, 10),
        )
        .await
        .expect("first apply");

        let out = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-2", "card-2", "fresh"),
            &train(4, 15),
        )
        .await
        .expect("second apply replaces");

        assert_eq!(
            training_of(&pool, "fresh").await,
            vec![(Some(4), 15, "def-card-2".to_string())],
            "the newer effect must stand alone, not sum to 25"
        );
        assert_eq!(out.effect["before"], json!(10));
        assert_eq!(out.effect["after"], json!(15));
        assert_eq!(out.effect["replaced"]["amount"], json!(10));
        // Both cards were legitimately spent.
        assert!(!source_exists(&pool, "card-2").await);
    }

    /// A card on a DIFFERENT slot also replaces: FIFA 17 permits "one attribute
    /// or all six", never a mixture, so two slots must never be boosted at once.
    #[tokio::test]
    async fn a_training_on_a_different_slot_still_replaces_the_previous_one() {
        let (_dir, pool) = fixture().await;
        for (identity, source, slot, amount) in
            [("act-1", "card-1", 4, 10), ("act-2", "card-2", 1, 5)]
        {
            consume_item(
                &pool,
                "prof",
                "club",
                &apply_request(identity, source, "fresh"),
                &train(slot, amount),
            )
            .await
            .expect("apply");
        }
        assert_eq!(
            training_of(&pool, "fresh").await,
            vec![(Some(1), 5, "def-card-2".to_string())],
            "only the newest effect may remain"
        );
    }

    /// The rare card boosts every slot, and replaces a single-attribute effect
    /// exactly like any other new card.
    #[tokio::test]
    async fn the_all_six_card_stores_no_slot_and_replaces_a_single_attribute_effect() {
        let (_dir, pool) = fixture().await;
        consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(4, 15),
        )
        .await
        .expect("single-attribute apply");

        consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-2", "card-2", "fresh"),
            &train_all(10),
        )
        .await
        .expect("all-six apply");

        assert_eq!(
            training_of(&pool, "fresh").await,
            vec![(None, 10, "def-card-2".to_string())],
            "the all-six effect stores a NULL slot and stands alone"
        );
    }

    /// The all-six card authors at most +10; a caller declaring that ceiling
    /// cannot then push +15 through it.
    #[tokio::test]
    async fn the_all_six_ceiling_is_enforced_independently() {
        let (_dir, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &InstanceEffect::ApplyTraining {
                attribute_index: None,
                amount: 15,
                max_amount: 10,
            },
        )
        .await
        .expect_err("an over-ceiling all-six boost must be refused");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert!(training_of(&pool, "fresh").await.is_empty());
        assert!(source_exists(&pool, "card-1").await);
    }

    /// The caller declares its own family's ceiling and is held to it. This is
    /// what stops a host describing a boost no card could grant.
    #[tokio::test]
    async fn an_amount_above_the_callers_declared_maximum_is_refused() {
        let (_dir, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(0, 99),
        )
        .await
        .expect_err("over-max must be refused");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert!(training_of(&pool, "fresh").await.is_empty());
        assert!(source_exists(&pool, "card-1").await);
    }

    /// A slot outside Core's six-attribute model names nothing projectable.
    #[tokio::test]
    async fn a_slot_outside_the_card_model_is_refused() {
        let (_dir, pool) = fixture().await;
        for slot in [-1i64, 6, 99] {
            let err = consume_item(
                &pool,
                "prof",
                "club",
                &apply_request("act-x", "card-1", "fresh"),
                &train(slot, 5),
            )
            .await
            .expect_err("out-of-range slot must be refused");
            assert!(
                matches!(err, AppError::BadRequest(_)),
                "slot {slot}: {err:?}"
            );
        }
        assert!(source_exists(&pool, "card-1").await);
    }

    /// Same reasoning as contracts: a loan outlives neither its match budget nor
    /// the improvement, so training it is refused rather than quietly wasted.
    #[tokio::test]
    async fn training_a_loan_item_is_refused_and_the_source_survives() {
        let (_dir, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "loaned"),
            &train(0, 5),
        )
        .await
        .expect_err("loan target must be refused");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert!(training_of(&pool, "loaned").await.is_empty());
        assert!(source_exists(&pool, "card-1").await);
    }

    /// A transport retry of the SAME action must not train twice or spend two
    /// cards — the guard is the same one contracts rely on.
    #[tokio::test]
    async fn a_training_replay_neither_trains_nor_charges_twice() {
        let (_dir, pool) = fixture().await;
        let first = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(2, 15),
        )
        .await
        .expect("first");
        assert!(first.applied);

        let replay = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(2, 15),
        )
        .await
        .expect("replay");
        assert!(!replay.applied, "a replay must not report a fresh apply");
        assert_eq!(replay.effect, first.effect);
        assert_eq!(
            training_of(&pool, "fresh").await,
            vec![(Some(2), 15, "def-card-1".to_string())]
        );
    }

    /// Destroying a trained instance must not leave its training behind — the
    /// FK cascade is what guarantees a quick-sold card cannot haunt the table.
    #[tokio::test]
    async fn training_dies_with_the_instance_it_is_attached_to() {
        let (_dir, pool) = fixture().await;
        consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &train(3, 5),
        )
        .await
        .expect("apply");
        assert_eq!(training_of(&pool, "fresh").await.len(), 1);

        sqlx::query("DELETE FROM owned_cards WHERE id = 'fresh'")
            .execute(&pool)
            .await
            .expect("delete instance");
        assert!(training_of(&pool, "fresh").await.is_empty());
    }

    /// APPLY vs QUICK-SELL on the LAST copy of a source. Exactly one may win:
    /// the card is either spent on the target or sold for coins, never both.
    ///
    /// Both paths are single Core transactions over the same row, so the loser
    /// must fail rather than operate on an already-gone source. This is the race
    /// a real player creates by hammering Enter on the consumables screen while a
    /// quick-sell is in flight.
    #[tokio::test]
    async fn apply_and_quick_sell_cannot_both_spend_one_card() {
        use crate::services::economy::{self, SaleBuyer, SaleTerms};

        // Repeated because a race that only sometimes interleaves would pass by
        // luck on a single attempt.
        for round in 0..12 {
            let (_dir, pool) = fixture().await;
            let apply_pool = pool.clone();
            let sell_pool = pool.clone();

            let applied = tokio::spawn(async move {
                consume_item(
                    &apply_pool,
                    "prof",
                    "club",
                    &apply_request("act-race", "card-1", "fresh"),
                    &train(0, 5),
                )
                .await
            });
            let sold = tokio::spawn(async move {
                economy::settle_sale(
                    &sell_pool,
                    "card-1",
                    "club",
                    SaleBuyer::Outside,
                    SaleTerms { gross: 100, fee: 0 },
                )
                .await
            });

            let applied = applied.await.expect("apply task");
            let sold = sold.await.expect("sell task");

            let trained = !training_of(&pool, "fresh").await.is_empty();
            let coins = sqlx::query_scalar::<_, i64>("SELECT coins FROM clubs WHERE id='club'")
                .fetch_one(&pool)
                .await
                .expect("coins");

            match (applied.is_ok(), sold.is_ok()) {
                (true, false) => {
                    assert!(trained, "round {round}: apply won but left no training");
                    assert_eq!(coins, 0, "round {round}: apply won but coins moved");
                }
                (false, true) => {
                    assert!(!trained, "round {round}: sale won but training was written");
                    assert_eq!(coins, 100, "round {round}: sale won but paid nothing");
                }
                (a, s) => panic!("round {round}: exactly one must win, got apply={a} sale={s}"),
            }
            // Either way the card is gone exactly once.
            assert!(
                !source_exists(&pool, "card-1").await,
                "round {round}: the source survived a winner"
            );
        }
    }

    /// Two CONCURRENT applies of the same last copy, under DIFFERENT identities
    /// (so the replay guard is not what separates them) and onto different
    /// slots. One must win outright: one training written, one card spent, one
    /// audit row.
    #[tokio::test]
    async fn two_concurrent_applies_of_one_card_produce_exactly_one_effect() {
        for round in 0..12 {
            let (_dir, pool) = fixture().await;
            let a_pool = pool.clone();
            let b_pool = pool.clone();

            let a = tokio::spawn(async move {
                consume_item(
                    &a_pool,
                    "prof",
                    "club",
                    &apply_request("act-a", "card-1", "fresh"),
                    &train(0, 5),
                )
                .await
            });
            let b = tokio::spawn(async move {
                consume_item(
                    &b_pool,
                    "prof",
                    "club",
                    &apply_request("act-b", "card-1", "fresh"),
                    &train(1, 5),
                )
                .await
            });
            let (a, b) = (a.await.expect("a"), b.await.expect("b"));

            assert!(
                a.is_ok() ^ b.is_ok(),
                "round {round}: exactly one apply must win, got a={:?} b={:?}",
                a.is_ok(),
                b.is_ok()
            );
            assert_eq!(
                training_of(&pool, "fresh").await.len(),
                1,
                "round {round}: exactly one training effect must exist"
            );
            let audits = sqlx::query_scalar::<_, i64>(
                "SELECT COUNT(*) FROM consumable_applications WHERE profile_id = 'prof'",
            )
            .fetch_one(&pool)
            .await
            .expect("audit count");
            assert_eq!(audits, 1, "round {round}: exactly one audit row");
            assert!(!source_exists(&pool, "card-1").await);
        }
    }

    async fn contract_of(pool: &db::Pool, id: &str) -> Option<i64> {
        sqlx::query_scalar::<_, Option<i64>>(
            "SELECT contract_matches FROM owned_cards WHERE id = ?",
        )
        .bind(id)
        .fetch_one(pool)
        .await
        .expect("read contract")
    }

    /// A target Core tracks no contract for counts as the CALLER's pack-fresh
    /// default, not as zero — the whole reason the column is nullable.
    #[tokio::test]
    async fn an_unset_target_starts_from_the_callers_default() {
        let (_dir, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-1", "card-1", "fresh"),
            &grant(15),
        )
        .await
        .expect("apply");

        assert!(out.applied);
        assert!(out.source_destroyed);
        assert_eq!(
            out.effect,
            json!({ "kind": "add_contract_matches", "granted": 15, "before": 7, "after": 22 })
        );
        assert_eq!(contract_of(&pool, "fresh").await, Some(22));
    }

    /// A stored value is added to, never replaced by the default.
    #[tokio::test]
    async fn an_existing_value_is_added_to() {
        let (_dir, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-2", "card-1", "used"),
            &grant(3),
        )
        .await
        .expect("apply");
        assert_eq!(
            out.effect,
            json!({ "kind": "add_contract_matches", "granted": 3, "before": 90, "after": 93 })
        );
        assert_eq!(contract_of(&pool, "used").await, Some(93));
    }

    /// The cap clamps the STORED total but not the REPORTED grant: they answer
    /// different questions, and a caller reconciling its own wire response needs
    /// the amount its table awarded.
    #[tokio::test]
    async fn the_cap_clamps_the_total_but_not_the_reported_grant() {
        let (_dir, pool) = fixture().await;
        let out = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-3", "card-1", "used"),
            &grant(28),
        )
        .await
        .expect("apply");
        assert_eq!(
            out.effect,
            json!({ "kind": "add_contract_matches", "granted": 28, "before": 90, "after": 99 })
        );
        assert_eq!(contract_of(&pool, "used").await, Some(99));
    }

    /// A loan is borrowed for a fixed run of matches; its contract is not Core's
    /// to extend. The refusal must also unwind the charge.
    #[tokio::test]
    async fn a_loan_target_is_refused_and_the_source_survives() {
        let (_dir, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-loan", "card-1", "loaned"),
            &grant(15),
        )
        .await
        .expect_err("a loan cannot take a contract");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
        assert_eq!(contract_of(&pool, "loaned").await, None);
        let sources =
            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM owned_cards WHERE id = 'card-1'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(sources, 1, "a refused effect must not spend the source");
    }

    /// This effect writes to an instance, so a club-scoped application has
    /// nothing to write to — refuse rather than spend the source for nothing.
    #[tokio::test]
    async fn a_club_scoped_application_is_refused() {
        let (_dir, pool) = fixture().await;
        let err = consume_item(
            &pool,
            "prof",
            "club",
            &ConsumeRequest {
                action_identity: "act-club",
                source_owned_card_id: "card-1",
                expected_source_kind: ContentKind::Consumable,
                consumption: SourceConsumption::DestroyInstance,
                target: ConsumeTarget::Club,
            },
            &grant(15),
        )
        .await
        .expect_err("no target to contract");
        assert!(matches!(err, AppError::BadRequest(_)), "got {err:?}");
    }

    /// Core validates the caller's numbers instead of trusting them: a zero or
    /// negative grant, a zero cap and a negative default are all nonsense, and a
    /// nonsense application must not silently succeed as a no-op.
    #[tokio::test]
    async fn nonsensical_parameters_are_refused() {
        let (_dir, pool) = fixture().await;
        for (identity, effect) in [
            (
                "bad-amount-0",
                InstanceEffect::AddContractMatches {
                    amount: 0,
                    cap: 99,
                    default_when_unset: 7,
                },
            ),
            (
                "bad-amount-neg",
                InstanceEffect::AddContractMatches {
                    amount: -5,
                    cap: 99,
                    default_when_unset: 7,
                },
            ),
            (
                "bad-cap",
                InstanceEffect::AddContractMatches {
                    amount: 1,
                    cap: 0,
                    default_when_unset: 7,
                },
            ),
            (
                "bad-default",
                InstanceEffect::AddContractMatches {
                    amount: 1,
                    cap: 99,
                    default_when_unset: -1,
                },
            ),
        ] {
            let Err(err) = consume_item(
                &pool,
                "prof",
                "club",
                &apply_request(identity, "card-1", "fresh"),
                &effect,
            )
            .await
            else {
                panic!("{identity} must be refused");
            };
            assert!(
                matches!(err, AppError::BadRequest(_)),
                "{identity}: got {err:?}"
            );
        }
        assert_eq!(
            contract_of(&pool, "fresh").await,
            None,
            "a refused application leaves the target untracked"
        );
    }

    /// The replay guard covers the effect too: a repeated `action_identity`
    /// echoes the recorded outcome, adds no second grant, and spends no second
    /// card.
    #[tokio::test]
    async fn a_replay_neither_grants_nor_charges_twice() {
        let (_dir, pool) = fixture().await;
        let first = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-replay", "card-1", "fresh"),
            &grant(15),
        )
        .await
        .expect("first");
        assert!(first.applied);

        let replay = consume_item(
            &pool,
            "prof",
            "club",
            &apply_request("act-replay", "card-1", "fresh"),
            &grant(15),
        )
        .await
        .expect("replay");
        assert!(!replay.applied, "a replay must not re-apply");
        assert_eq!(
            replay.effect, first.effect,
            "the recorded outcome is echoed"
        );
        assert_eq!(contract_of(&pool, "fresh").await, Some(22), "granted once");
        assert_eq!(
            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM owned_cards WHERE id LIKE 'card-%'")
                .fetch_one(&pool)
                .await
                .unwrap(),
            2,
            "exactly one consumable was spent"
        );
    }

    /// The wire vocabulary is closed: the documented body parses, and an
    /// undescribed effect is rejected at the boundary rather than reaching a
    /// fallback.
    #[test]
    fn the_effect_vocabulary_is_closed() {
        let parsed: InstanceEffect = serde_json::from_str(
            r#"{"kind":"add_contract_matches","amount":15,"cap":99,"default_when_unset":7}"#,
        )
        .expect("the documented effect body must parse");
        assert!(matches!(
            parsed,
            InstanceEffect::AddContractMatches {
                amount: 15,
                cap: 99,
                default_when_unset: 7
            }
        ));
        assert!(
            serde_json::from_str::<InstanceEffect>(r#"{"kind":"set_rating","value":99}"#).is_err(),
            "an effect Core does not implement must be refused, never guessed"
        );
    }
}