mostro 0.17.4

//! Database helpers for the `bonds` table.
//!
//! Phase 0 exposes the CRUD surface later phases will need. Nothing in
//! this module hits LND or the Nostr client — it's purely storage.

use mostro_core::error::{MostroError::MostroInternalErr, ServiceError};
use sqlx::{Pool, Sqlite};
use sqlx_crud::Crud;
use uuid::Uuid;

use super::model::Bond;
use super::types::{BondRole, BondState};

/// Insert a new bond row. Returns the persisted `Bond`.
pub async fn create_bond(
    pool: &Pool<Sqlite>,
    bond: Bond,
) -> Result<Bond, mostro_core::error::MostroError> {
    bond.create(pool)
        .await
        .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// Look up the parent bond row for a given order + role. Returns `None`
/// if no bond exists for that pair (the normal case when the feature is
/// off or doesn't apply to the role).
///
/// Phase 6 introduces child slash rows that share the parent's `order_id`
/// and `role`; those rows carry a non-NULL `parent_bond_id`. The
/// `parent_bond_id IS NULL` predicate keeps this lookup pinned to the
/// parent bond so state transitions always target the right row.
pub async fn find_bond_by_order_and_role(
    pool: &Pool<Sqlite>,
    order_id: Uuid,
    role: BondRole,
) -> Result<Option<Bond>, mostro_core::error::MostroError> {
    let role_str = role.to_string();
    sqlx::query_as::<_, Bond>(
        "SELECT * FROM bonds \
         WHERE order_id = ? AND role = ? AND parent_bond_id IS NULL \
         LIMIT 1",
    )
    .bind(order_id)
    .bind(role_str)
    .fetch_optional(pool)
    .await
    .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// List every bond currently in the given state. Used by the Phase 3
/// payout scheduler.
pub async fn find_bonds_by_state(
    pool: &Pool<Sqlite>,
    state: BondState,
) -> Result<Vec<Bond>, mostro_core::error::MostroError> {
    let state_str = state.to_string();
    sqlx::query_as::<_, Bond>("SELECT * FROM bonds WHERE state = ? ORDER BY created_at ASC")
        .bind(state_str)
        .fetch_all(pool)
        .await
        .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// Look up a bond row by its Lightning payment hash. The hash uniquely
/// identifies the bond hold invoice, so this is what the LND subscriber
/// uses to correlate incoming invoice events back to a `Bond`.
pub async fn find_bond_by_hash(
    pool: &Pool<Sqlite>,
    hash: &str,
) -> Result<Option<Bond>, mostro_core::error::MostroError> {
    sqlx::query_as::<_, Bond>("SELECT * FROM bonds WHERE hash = ? LIMIT 1")
        .bind(hash)
        .fetch_optional(pool)
        .await
        .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// List every bond row that still has an outstanding LND HTLC, i.e. is in
/// `Requested` or `Locked`. Used on daemon startup to resubscribe to
/// in-flight bond hold invoices, and as the Phase 1 workhorse for the
/// "always release" exits — we filter further on `order_id` in
/// [`find_active_bonds_for_order`].
pub async fn find_active_bonds(
    pool: &Pool<Sqlite>,
) -> Result<Vec<Bond>, mostro_core::error::MostroError> {
    let requested = BondState::Requested.to_string();
    let locked = BondState::Locked.to_string();
    sqlx::query_as::<_, Bond>("SELECT * FROM bonds WHERE state IN (?, ?) ORDER BY created_at ASC")
        .bind(requested)
        .bind(locked)
        .fetch_all(pool)
        .await
        .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// List the still-outstanding bonds attached to a single order. Phase 1
/// uses this to release every bond on any order exit path (cancel,
/// release, admin actions, scheduler timeouts).
pub async fn find_active_bonds_for_order(
    pool: &Pool<Sqlite>,
    order_id: Uuid,
) -> Result<Vec<Bond>, mostro_core::error::MostroError> {
    let requested = BondState::Requested.to_string();
    let locked = BondState::Locked.to_string();
    sqlx::query_as::<_, Bond>(
        "SELECT * FROM bonds \
         WHERE order_id = ? AND state IN (?, ?) \
         ORDER BY created_at ASC",
    )
    .bind(order_id)
    .bind(requested)
    .bind(locked)
    .fetch_all(pool)
    .await
    .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// Look up the active (`Requested` or `Locked`) bond row for a given
/// `(order_id, taker_pubkey)` pair. Used by the take handlers'
/// idempotent-retry check (a taker re-emitting `take-buy` / `take-sell`
/// while their bond is still `Requested` must get back the same
/// `payment_request`, not a fresh row) and by `cancel_order_by_taker`
/// to scope the cancel to the sender's own bond under concurrent
/// taker bonds.
///
/// Filters on `parent_bond_id IS NULL` to ignore Phase 6 child slash
/// rows, mirroring `find_bond_by_order_and_role`.
pub async fn find_active_bond_by_taker(
    pool: &Pool<Sqlite>,
    order_id: Uuid,
    taker_pubkey: &str,
) -> Result<Option<Bond>, mostro_core::error::MostroError> {
    let requested = BondState::Requested.to_string();
    let locked = BondState::Locked.to_string();
    sqlx::query_as::<_, Bond>(
        "SELECT * FROM bonds \
         WHERE order_id = ? AND pubkey = ? AND state IN (?, ?) \
           AND parent_bond_id IS NULL \
         ORDER BY created_at ASC \
         LIMIT 1",
    )
    .bind(order_id)
    .bind(taker_pubkey)
    .bind(requested)
    .bind(locked)
    .fetch_optional(pool)
    .await
    .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

/// Update a bond row by primary key. Returns the persisted `Bond`.
pub async fn update_bond(
    pool: &Pool<Sqlite>,
    bond: Bond,
) -> Result<Bond, mostro_core::error::MostroError> {
    bond.update(pool)
        .await
        .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::app::bond::model::Bond;
    use crate::app::bond::types::BondRole;
    use sqlx::sqlite::SqlitePoolOptions;

    async fn setup_pool() -> Pool<Sqlite> {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect(":memory:")
            .await
            .expect("open in-memory sqlite");
        // Minimal orders table: bonds has an FK on it.
        sqlx::query(include_str!(
            "../../../migrations/20221222153301_orders.sql"
        ))
        .execute(&pool)
        .await
        .expect("orders migration");
        sqlx::query(include_str!(
            "../../../migrations/20260423120000_anti_abuse_bond.sql"
        ))
        .execute(&pool)
        .await
        .expect("bonds migration");
        sqlx::query(include_str!(
            "../../../migrations/20260518120000_bond_payout_payment_hash.sql"
        ))
        .execute(&pool)
        .await
        .expect("bond_payout_payment_hash migration");
        // SQLite doesn't enforce FKs unless asked. Turn them on so the FK to
        // `orders` is a real constraint in tests (mirrors production).
        sqlx::query("PRAGMA foreign_keys = ON")
            .execute(&pool)
            .await
            .expect("enable fk");
        pool
    }

    async fn insert_parent_order(pool: &Pool<Sqlite>, id: Uuid) {
        sqlx::query(
            r#"INSERT INTO orders (
                id, kind, event_id, status, premium, payment_method,
                amount, fiat_code, fiat_amount, created_at, expires_at
            ) VALUES (?, 'buy', ?, 'pending', 0, 'ln', 1000, 'USD', 10, 0, 0)"#,
        )
        .bind(id)
        .bind(id.simple().to_string())
        .execute(pool)
        .await
        .expect("insert parent order");
    }

    fn dummy_bond(order_id: Uuid, role: BondRole) -> Bond {
        Bond::new_requested(order_id, "a".repeat(64), role, 1_500)
    }

    #[tokio::test]
    async fn insert_and_fetch_by_order_and_role() {
        let pool = setup_pool().await;
        let order_id = Uuid::new_v4();
        insert_parent_order(&pool, order_id).await;
        let created = create_bond(&pool, dummy_bond(order_id, BondRole::Taker))
            .await
            .expect("insert");
        let fetched = find_bond_by_order_and_role(&pool, order_id, BondRole::Taker)
            .await
            .expect("query")
            .expect("row present");
        assert_eq!(fetched.id, created.id);
        assert_eq!(fetched.role, "taker");
    }

    #[tokio::test]
    async fn fetch_by_order_and_role_ignores_child_rows() {
        // Phase 6 will store child slash rows that share the parent bond's
        // `order_id` and `role`; the lookup must still return the parent.
        let pool = setup_pool().await;
        let order_id = Uuid::new_v4();
        let child_order_id = Uuid::new_v4();
        insert_parent_order(&pool, order_id).await;
        insert_parent_order(&pool, child_order_id).await;

        let parent = create_bond(&pool, dummy_bond(order_id, BondRole::Maker))
            .await
            .expect("insert parent");

        let mut child = dummy_bond(order_id, BondRole::Maker);
        child.parent_bond_id = Some(parent.id);
        child.child_order_id = Some(child_order_id);
        create_bond(&pool, child).await.expect("insert child");

        let fetched = find_bond_by_order_and_role(&pool, order_id, BondRole::Maker)
            .await
            .expect("query")
            .expect("row present");
        assert_eq!(fetched.id, parent.id);
        assert!(fetched.parent_bond_id.is_none());
    }

    #[tokio::test]
    async fn fetch_missing_returns_none() {
        let pool = setup_pool().await;
        let res = find_bond_by_order_and_role(&pool, Uuid::new_v4(), BondRole::Taker)
            .await
            .expect("query");
        assert!(res.is_none());
    }

    #[tokio::test]
    async fn find_by_hash_returns_match() {
        let pool = setup_pool().await;
        let order_id = Uuid::new_v4();
        insert_parent_order(&pool, order_id).await;

        let mut bond = dummy_bond(order_id, BondRole::Taker);
        bond.hash = Some("c".repeat(64));
        let created = create_bond(&pool, bond).await.expect("insert");

        let found = find_bond_by_hash(&pool, &"c".repeat(64))
            .await
            .expect("query")
            .expect("row present");
        assert_eq!(found.id, created.id);

        let missing = find_bond_by_hash(&pool, &"f".repeat(64))
            .await
            .expect("query");
        assert!(missing.is_none());
    }

    #[tokio::test]
    async fn active_bonds_filter_terminal_states() {
        let pool = setup_pool().await;
        let order_a = Uuid::new_v4();
        let order_b = Uuid::new_v4();
        insert_parent_order(&pool, order_a).await;
        insert_parent_order(&pool, order_b).await;
        let bond_a = create_bond(&pool, dummy_bond(order_a, BondRole::Taker))
            .await
            .unwrap();
        let bond_b = create_bond(&pool, dummy_bond(order_b, BondRole::Taker))
            .await
            .unwrap();

        // Bond B → Released (terminal): must drop out of active set.
        let mut released = bond_b.clone();
        released.state = BondState::Released.to_string();
        update_bond(&pool, released).await.unwrap();

        let active = find_active_bonds(&pool).await.unwrap();
        assert_eq!(active.len(), 1);
        assert_eq!(active[0].id, bond_a.id);

        let active_a = find_active_bonds_for_order(&pool, order_a).await.unwrap();
        assert_eq!(active_a.len(), 1);
        let active_b = find_active_bonds_for_order(&pool, order_b).await.unwrap();
        assert!(active_b.is_empty());
    }

    #[tokio::test]
    async fn find_by_state_filters() {
        let pool = setup_pool().await;
        let order_a = Uuid::new_v4();
        let order_b = Uuid::new_v4();
        insert_parent_order(&pool, order_a).await;
        insert_parent_order(&pool, order_b).await;
        let bond_a = create_bond(&pool, dummy_bond(order_a, BondRole::Taker))
            .await
            .unwrap();
        let _bond_b = create_bond(&pool, dummy_bond(order_b, BondRole::Maker))
            .await
            .unwrap();

        // Flip one to Locked.
        let mut locked = bond_a.clone();
        locked.state = BondState::Locked.to_string();
        locked.locked_at = Some(42);
        update_bond(&pool, locked).await.unwrap();

        let requested = find_bonds_by_state(&pool, BondState::Requested)
            .await
            .unwrap();
        assert_eq!(requested.len(), 1);
        assert_eq!(requested[0].order_id, order_b);

        let locked = find_bonds_by_state(&pool, BondState::Locked).await.unwrap();
        assert_eq!(locked.len(), 1);
        assert_eq!(locked[0].order_id, order_a);
    }

    #[tokio::test]
    async fn taker_context_columns_roundtrip() {
        // Concurrent-bonds rework adds taker_* columns that stash the
        // deferred take context until the bond locks. Make sure the
        // additive migration is applied and the columns round-trip
        // through insert / fetch.
        let pool = setup_pool().await;
        let order_id = Uuid::new_v4();
        insert_parent_order(&pool, order_id).await;

        let mut bond = dummy_bond(order_id, BondRole::Taker);
        bond.taker_identity = Some("d".repeat(64));
        bond.taker_trade_index = Some(42);
        bond.taker_invoice = Some("lnbc1pTAKER".to_string());
        bond.taker_fiat_amount = Some(123);
        bond.taker_amount = Some(45_678);
        bond.taker_fee = Some(89);
        bond.taker_dev_fee = Some(7);
        let created = create_bond(&pool, bond).await.unwrap();

        let fetched = find_bond_by_order_and_role(&pool, order_id, BondRole::Taker)
            .await
            .unwrap()
            .expect("bond present");
        assert_eq!(fetched.id, created.id);
        assert_eq!(
            fetched.taker_identity.as_deref(),
            Some("d".repeat(64).as_str())
        );
        assert_eq!(fetched.taker_trade_index, Some(42));
        assert_eq!(fetched.taker_invoice.as_deref(), Some("lnbc1pTAKER"));
        assert_eq!(fetched.taker_fiat_amount, Some(123));
        assert_eq!(fetched.taker_amount, Some(45_678));
        assert_eq!(fetched.taker_fee, Some(89));
        assert_eq!(fetched.taker_dev_fee, Some(7));
    }

    #[tokio::test]
    async fn find_active_bond_by_taker_scopes_to_pubkey() {
        // Two concurrent prospective takers on the same order each get
        // their own `Requested` bond. The lookup must return exactly
        // the bond belonging to the queried pubkey.
        let pool = setup_pool().await;
        let order_id = Uuid::new_v4();
        insert_parent_order(&pool, order_id).await;

        let mut bond_a = dummy_bond(order_id, BondRole::Taker);
        bond_a.pubkey = "a".repeat(64);
        let created_a = create_bond(&pool, bond_a).await.unwrap();

        let mut bond_b = dummy_bond(order_id, BondRole::Taker);
        bond_b.pubkey = "b".repeat(64);
        let created_b = create_bond(&pool, bond_b).await.unwrap();

        let found_a = find_active_bond_by_taker(&pool, order_id, &"a".repeat(64))
            .await
            .unwrap()
            .expect("bond A present");
        assert_eq!(found_a.id, created_a.id);

        let found_b = find_active_bond_by_taker(&pool, order_id, &"b".repeat(64))
            .await
            .unwrap()
            .expect("bond B present");
        assert_eq!(found_b.id, created_b.id);

        // Unrelated pubkey returns None.
        let missing = find_active_bond_by_taker(&pool, order_id, &"c".repeat(64))
            .await
            .unwrap();
        assert!(missing.is_none());

        // Released (terminal) bonds drop out of the lookup.
        let mut released = created_a.clone();
        released.state = BondState::Released.to_string();
        update_bond(&pool, released).await.unwrap();
        let after_release = find_active_bond_by_taker(&pool, order_id, &"a".repeat(64))
            .await
            .unwrap();
        assert!(after_release.is_none());
    }
}