nxtgauge-backend-rust/crates/wallet/src/hold.rs

//! Tracecoin holds (escrow) and bucket operations.
//!
//! A "hold" is a soft-reserve on a user's tracecoins. The hold is
//! counted against the user's `available` balance but is not yet a
//! permanent debit. The hold lifecycle is:
//!
//!   ACTIVE → SETTLED    (action completed, hold becomes a permanent debit
//!                        with a linked ledger row)
//!   ACTIVE → RELEASED   (user or system cancelled, the tracecoins are
//!                        returned to the user's `available` balance)
//!   ACTIVE → EXPIRED    (the cron's expire_due_holds() saw that
//!                        `expires_at` had passed; same as RELEASED but
//!                        the audit trail is "expired" not "cancelled")
//!
//! Holds are the right primitive when the user-facing action has a
//! long lifetime (e.g. a 24-hour lead request that may or may not be
//! accepted) and we want to keep the tracecoins earmarked without
//! double-counting them as spent.
//!
//! `reserve_for_lead_request` (in lib.rs) is implemented on top of
//! `hold_for_action` internally so the wallet stays consistent.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sqlx::{FromRow, PgPool, Postgres, Row, Transaction};
use uuid::Uuid;

// ──────────────────────────────────────────────────────────────────────────
// Hold
// ──────────────────────────────────────────────────────────────────────────

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "UPPERCASE")]
pub enum HoldStatus {
    Active,
    Settled,
    Released,
    Expired,
}

impl HoldStatus {
    pub fn as_str(self) -> &'static str {
        match self {
            HoldStatus::Active => "ACTIVE",
            HoldStatus::Settled => "SETTLED",
            HoldStatus::Released => "RELEASED",
            HoldStatus::Expired => "EXPIRED",
        }
    }
}

#[derive(Debug, Clone, Serialize, FromRow)]
pub struct Hold {
    pub id: Uuid,
    pub wallet_id: Uuid,
    pub user_id: Uuid,
    pub amount: i32,
    pub reason: String,
    pub reference_id: Option<Uuid>,
    pub status: String,
    pub expires_at: Option<DateTime<Utc>>,
    pub settled_at: Option<DateTime<Utc>>,
    pub settled_ledger_id: Option<Uuid>,
    pub released_at: Option<DateTime<Utc>>,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum HoldReason {
    LeadRequest,
    ContactUnlock,
    AiCredit,
    FeatureReservation,
    System,
}

impl HoldReason {
    pub fn as_str(self) -> &'static str {
        match self {
            HoldReason::LeadRequest => "LEAD_REQUEST",
            HoldReason::ContactUnlock => "CONTACT_UNLOCK",
            HoldReason::AiCredit => "AI_CREDIT",
            HoldReason::FeatureReservation => "FEATURE_RESERVATION",
            HoldReason::System => "SYSTEM",
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum HoldError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),
    #[error("hold not found")]
    NotFound,
    #[error("hold not in ACTIVE state (current: {0})")]
    NotActive(String),
    #[error("hold amount {amount} exceeds available {available}")]
    Insufficient { amount: i32, available: i32 },
    #[error("hold already exists for reference {0}")]
    AlreadyExists(Uuid),
    #[error("invalid hold amount: {0}")]
    InvalidAmount(i32),
    #[error("invalid input: {0}")]
    InvalidInput(String),
    #[error("wallet error: {0}")]
    Wallet(#[from] crate::WalletError),
}

pub type HoldResult<T> = Result<T, HoldError>;

// ──────────────────────────────────────────────────────────────────────────
// Hold operations
// ──────────────────────────────────────────────────────────────────────────

/// Create a hold for the given amount. If `reference_id` is supplied
/// and a hold already exists for it, the existing hold is returned
/// (idempotent).
///
/// On success, returns `(hold, available_after)`. The hold reduces the
/// user's `available` balance by `amount` but does NOT deduct from
/// `balance` until the hold is settled.
pub async fn place(
    pool: &PgPool,
    user_id: Uuid,
    amount: i32,
    reason: HoldReason,
    reference_id: Option<Uuid>,
    expires_at: Option<DateTime<Utc>>,
) -> HoldResult<(Hold, i32)> {
    if amount <= 0 {
        return Err(HoldError::InvalidAmount(amount));
    }

    let mut tx = pool.begin().await?;
    let wallet_id = crate::lock_wallet(&mut tx, user_id).await?.0;

    // Idempotency: if a hold already exists for this reference_id, return it.
    if let Some(reference_id) = reference_id {
        let existing: Option<(Uuid, Uuid, Uuid, Uuid, i32, String, Option<Uuid>, String, Option<DateTime<Utc>>, Option<DateTime<Utc>>, Option<Uuid>, Option<DateTime<Utc>>, DateTime<Utc>)> = sqlx::query_as(
            r#"
            SELECT id, wallet_id, user_id, user_id, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at
            FROM tracecoin_holds
            WHERE reference_id = $1
            "#,
        )
        .bind(reference_id)
        .fetch_optional(&mut *tx)
        .await?;

        if let Some((id, wallet_id, user_id, _, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at)) = existing {
            tx.commit().await?;
            return Ok((Hold {
                id,
                wallet_id,
                user_id,
                amount,
                reason,
                reference_id,
                status,
                expires_at,
                settled_at,
                settled_ledger_id,
                released_at,
                created_at,
            }, 0));
        }
    }

    // Read current available balance
    let balance: i32 = sqlx::query_scalar("SELECT balance FROM tracecoin_wallets WHERE id = $1")
        .bind(wallet_id)
        .fetch_one(&mut *tx)
        .await?;
    let reserved: i32 = sqlx::query_scalar("SELECT reserved FROM tracecoin_wallets WHERE id = $1")
        .bind(wallet_id)
        .fetch_one(&mut *tx)
        .await?;
    let available = balance - reserved;

    if available < amount {
        return Err(HoldError::Insufficient {
            amount,
            available,
        });
    }

    // Insert the hold
    let hold_id = Uuid::new_v4();
    sqlx::query(
        r#"
        INSERT INTO tracecoin_holds (
            id, wallet_id, user_id, amount, reason, reference_id, status, expires_at
        )
        VALUES ($1, $2, $3, $4, $5, $6, 'ACTIVE', $7)
        "#,
    )
    .bind(hold_id)
    .bind(wallet_id)
    .bind(user_id)
    .bind(amount)
    .bind(reason.as_str())
    .bind(reference_id)
    .bind(expires_at)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;

    let new_available = available - amount;
    Ok((
        Hold {
            id: hold_id,
            wallet_id,
            user_id,
            amount,
            reason: reason.as_str().to_string(),
            reference_id,
            status: HoldStatus::Active.as_str().to_string(),
            expires_at,
            settled_at: None,
            settled_ledger_id: None,
            released_at: None,
            created_at: Utc::now(),
        },
        new_available,
    ))
}

/// Settle an active hold. The tracecoins are permanently debited
/// (`balance` decreases) and a ledger row is written. Idempotent on
/// the hold id.
pub async fn settle(
    pool: &PgPool,
    hold_id: Uuid,
) -> HoldResult<()> {
    let mut tx = pool.begin().await?;

    let row: Option<(Uuid, Uuid, Uuid, i32, String)> = sqlx::query_as(
        r#"
        SELECT id, wallet_id, user_id, amount, status
        FROM tracecoin_holds
        WHERE id = $1
        FOR UPDATE
        "#,
    )
    .bind(hold_id)
    .fetch_optional(&mut *tx)
    .await?;

    let (id, wallet_id, user_id, amount, status) = match row {
        Some(r) => r,
        None => return Err(HoldError::NotFound),
    };

    if status != "ACTIVE" {
        return Err(HoldError::NotActive(status));
    }

    // Lock wallet
    let (wallet_id_locked, _balance, _reserved) = crate::lock_wallet_inner(&mut tx, wallet_id).await?;

    // Insert a DEBIT ledger row
    let new_balance: i32 = sqlx::query_scalar(
        r#"
        UPDATE tracecoin_wallets
        SET balance = balance - $1, updated_at = NOW()
        WHERE id = $2
        RETURNING balance
        "#,
    )
    .bind(amount)
    .bind(wallet_id_locked)
    .fetch_one(&mut *tx)
    .await?;

    if new_balance < 0 {
        return Err(HoldError::InvalidInput(format!(
            "balance went negative after settling hold {} (user {})",
            id, user_id
        )));
    }

    let ledger_id: Uuid = sqlx::query_scalar(
        r#"
        INSERT INTO tracecoin_ledger (
            wallet_id, type, amount, balance_after, reason,
            reference_id, actor_user_id, metadata
        )
        VALUES ($1, 'DEBIT', $2, $3, 'HOLD_SETTLED', $4, NULL, $5)
        RETURNING id
        "#,
    )
    .bind(wallet_id_locked)
    .bind(-amount)
    .bind(new_balance)
    .bind(id)
    .bind(serde_json::json!({"hold_id": id, "settled_at": Utc::now()}))
    .fetch_one(&mut *tx)
    .await?;

    sqlx::query(
        r#"
        UPDATE tracecoin_holds
        SET status = 'SETTLED',
            settled_at = NOW(),
            settled_ledger_id = $1
        WHERE id = $2
        "#,
    )
    .bind(ledger_id)
    .bind(id)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(())
}

/// Release an active hold (cancel). The tracecoins are returned to
/// the user's `available` balance. Idempotent.
pub async fn release(
    pool: &PgPool,
    hold_id: Uuid,
) -> HoldResult<()> {
    let mut tx = pool.begin().await?;

    let row: Option<(Uuid, Uuid, String)> = sqlx::query_as(
        r#"
        SELECT id, wallet_id, status
        FROM tracecoin_holds
        WHERE id = $1
        FOR UPDATE
        "#,
    )
    .bind(hold_id)
    .fetch_optional(&mut *tx)
    .await?;

    let (id, _wallet_id, status) = match row {
        Some(r) => r,
        None => return Err(HoldError::NotFound),
    };

    if status != "ACTIVE" {
        return Err(HoldError::NotActive(status));
    }

    sqlx::query(
        r#"
        UPDATE tracecoin_holds
        SET status = 'RELEASED', released_at = NOW()
        WHERE id = $1
        "#,
    )
    .bind(id)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;
    Ok(())
}

/// Auto-release every hold whose `expires_at` is in the past and is
/// still ACTIVE. Returns the number of holds released.
///
/// Designed for the cron loop. Idempotent and safe under concurrent
/// invocation: each hold is locked with FOR UPDATE.
pub async fn expire_due_holds(pool: &PgPool) -> HoldResult<u64> {
    let now = Utc::now();

    let due: Vec<(Uuid, Uuid)> = sqlx::query_as(
        r#"
        SELECT id, wallet_id
        FROM tracecoin_holds
        WHERE status = 'ACTIVE'
          AND expires_at IS NOT NULL
          AND expires_at < $1
        "#,
    )
    .bind(now)
    .fetch_all(pool)
    .await?;

    let mut count = 0u64;
    for (id, _wallet_id) in due {
        let mut tx = pool.begin().await?;
        let row: Option<(Uuid, Uuid, String)> = sqlx::query_as(
            r#"
            SELECT id, wallet_id, status
            FROM tracecoin_holds
            WHERE id = $1
            FOR UPDATE
            "#,
        )
        .bind(id)
        .fetch_optional(&mut *tx)
        .await?;

        if let Some((id, _, status)) = row {
            if status == "ACTIVE" {
                let updated = sqlx::query(
                    r#"
                    UPDATE tracecoin_holds
                    SET status = 'EXPIRED', released_at = NOW()
                    WHERE id = $1 AND status = 'ACTIVE'
                    "#,
                )
                .bind(id)
                .execute(&mut *tx)
                .await?;
                if updated.rows_affected() > 0 {
                    count += 1;
                }
            }
        }
        tx.commit().await.ok();
    }

    Ok(count)
}

/// List all active holds for a user, oldest first.
pub async fn list_active_for_user(
    pool: &PgPool,
    user_id: Uuid,
) -> HoldResult<Vec<Hold>> {
    let rows: Vec<(Uuid, Uuid, Uuid, Uuid, i32, String, Option<Uuid>, String, Option<DateTime<Utc>>, Option<DateTime<Utc>>, Option<Uuid>, Option<DateTime<Utc>>, DateTime<Utc>)> = sqlx::query_as(
        r#"
        SELECT id, wallet_id, user_id, user_id, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at
        FROM tracecoin_holds
        WHERE user_id = $1 AND status = 'ACTIVE'
        ORDER BY created_at ASC
        "#,
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    Ok(rows
        .into_iter()
        .map(|(id, wallet_id, user_id, _, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at)| Hold {
            id,
            wallet_id,
            user_id,
            amount,
            reason,
            reference_id,
            status,
            expires_at,
            settled_at,
            settled_ledger_id,
            released_at,
            created_at,
        })
        .collect())
}

/// List all holds (any status) for a user, newest first.
pub async fn list_for_user(
    pool: &PgPool,
    user_id: Uuid,
    limit: i64,
) -> HoldResult<Vec<Hold>> {
    let limit = limit.clamp(1, 200);
    let rows: Vec<(Uuid, Uuid, Uuid, Uuid, i32, String, Option<Uuid>, String, Option<DateTime<Utc>>, Option<DateTime<Utc>>, Option<Uuid>, Option<DateTime<Utc>>, DateTime<Utc>)> = sqlx::query_as(
        r#"
        SELECT id, wallet_id, user_id, user_id, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at
        FROM tracecoin_holds
        WHERE user_id = $1
        ORDER BY created_at DESC
        LIMIT $2
        "#,
    )
    .bind(user_id)
    .bind(limit)
    .fetch_all(pool)
    .await?;

    Ok(rows
        .into_iter()
        .map(|(id, wallet_id, user_id, _, amount, reason, reference_id, status, expires_at, settled_at, settled_ledger_id, released_at, created_at)| Hold {
            id,
            wallet_id,
            user_id,
            amount,
            reason,
            reference_id,
            status,
            expires_at,
            settled_at,
            settled_ledger_id,
            released_at,
            created_at,
        })
        .collect())
}

// Suppress unused-import warnings for items used by the macro.
#[allow(dead_code)]
fn _phantom_tx<'a>(_: &Transaction<'a, Postgres>) {}