hypercall_db_diesel/

settlement_ops.rs

//! Settlement persistence operations.
//!
//! Free functions that take `&mut PgConnection` so they can be called
//! from any context (DatabaseHandler trait impl, transaction closures, tests).
//! Previously lived as private methods on DieselEventHandler.

use anyhow::{anyhow, Context, Result};
use diesel::prelude::*;
use diesel::sql_types::{BigInt, Binary, Nullable, Numeric, Text};
use hypercall_types::{MarginMode, WalletAddress};
use rust_decimal::Decimal;

use crate::models::{NewPositionExpiration, NewSettlementPayout, SettlementPayout};
use crate::schema;

/// Parse a symbol string and extract its expiry timestamp.
pub fn extract_expiry_ts(symbol: &str) -> Result<i64> {
    hypercall_settlement::SettlementInstrument::from_symbol(symbol)
        .map(|instrument| instrument.expiry_ts)
        .map_err(|e| anyhow!("Failed to parse symbol '{}': {}", symbol, e))
}

/// Validate settlement economics (entry price, cost basis, net PnL) and return as a tuple.
pub fn validate_settlement_economics_tuple(
    settlement_entry_price: Option<Decimal>,
    cost_basis: Option<Decimal>,
    net_pnl: Option<Decimal>,
    context: &str,
) -> Result<(Option<Decimal>, Option<Decimal>, Option<Decimal>)> {
    hypercall_settlement::validate_settlement_economics(
        settlement_entry_price,
        cost_basis,
        net_pnl,
        context,
    )
    .map(|economics| {
        economics
            .map(|economics| {
                (
                    Some(economics.settlement_entry_price),
                    Some(economics.cost_basis),
                    Some(economics.net_pnl),
                )
            })
            .unwrap_or((None, None, None))
    })
    .map_err(|e| anyhow!("{}", e))
}

/// Compute the ledger cash delta for a settlement based on margin mode.
/// Standard mode uses settlement_value; portfolio mode uses net_pnl.
pub fn settlement_ledger_delta_for_margin_mode(
    margin_mode: MarginMode,
    settlement_value: Decimal,
    net_pnl: Option<Decimal>,
    context: &str,
) -> Result<Decimal> {
    hypercall_settlement::settlement_cash_delta_for_margin_mode(
        margin_mode,
        settlement_value,
        net_pnl,
        context,
    )
    .map_err(|e| anyhow!("{}", e))
}

/// Idempotent settlement claim: inserts expiration + payout + ledger event
/// in one transaction. Panics on invariant violations (data corruption).
/// Returns whether this was a new claim or a duplicate replay.
pub fn claim_settlement_in_tx(
    conn: &mut PgConnection,
    expiration: &NewPositionExpiration,
    payout: &NewSettlementPayout,
    ledger_delta: Decimal,
    event_ts_ms: i64,
) -> Result<hypercall_db::SettlementResult> {
    let expiration_rows = diesel::insert_into(schema::position_expirations::table)
        .values(expiration)
        .on_conflict((
            schema::position_expirations::wallet,
            schema::position_expirations::symbol,
            schema::position_expirations::expiry_ts,
        ))
        .do_nothing()
        .execute(conn)?;

    let expiration_inserted = expiration_rows > 0;
    let existing_payout = load_settlement_payout_in_tx(conn, payout)?;

    if expiration_inserted {
        if let Some(existing) = existing_payout {
            panic!(
                "Partial settlement payout for {}/{} expiry {}: payout id={} existed before expiration claim",
                payout.wallet, payout.symbol, payout.expiry_ts, existing.id
            );
        }
        insert_settlement_payout_in_tx(conn, payout)?;
        let ledger_event_inserted = insert_settlement_ledger_event_in_tx(
            conn,
            &expiration.wallet,
            &expiration.symbol,
            ledger_delta,
            event_ts_ms,
        )?;
        if ledger_delta != Decimal::ZERO && !ledger_event_inserted {
            panic!(
                "Partial settlement ledger for {}/{}: nonzero settlement ledger event already existed before expiration claim",
                expiration.wallet, expiration.symbol
            );
        }
    } else {
        lock_and_validate_existing_settlement_claim_in_tx(conn, expiration)?;
        let Some(existing) = existing_payout else {
            panic!(
                "Partial settlement claim for {}/{} expiry {}: expiration row exists without payout",
                payout.wallet, payout.symbol, payout.expiry_ts
            );
        };
        validate_existing_settlement_payout_facts(&existing, payout);
        if existing.ledger_applied {
            validate_existing_settlement_ledger_in_tx(
                conn,
                &expiration.wallet,
                &expiration.symbol,
                ledger_delta,
            )?;
        } else {
            tracing::warn!(
                wallet = %expiration.wallet,
                symbol = %expiration.symbol,
                expiry_ts = expiration.expiry_ts,
                payout_id = existing.id,
                "Repairing legacy settlement payout with ledger_applied=false"
            );
            let ledger_event_inserted = insert_settlement_ledger_event_in_tx(
                conn,
                &expiration.wallet,
                &expiration.symbol,
                ledger_delta,
                event_ts_ms,
            )?;
            if ledger_delta != Decimal::ZERO && !ledger_event_inserted {
                panic!(
                    "Partial legacy settlement payout for {}/{} expiry {}: existing id={} has ledger_applied=false but settlement ledger event already exists",
                    payout.wallet, payout.symbol, payout.expiry_ts, existing.id
                );
            }
            mark_settlement_payout_applied_in_tx(conn, &existing)?;
        }
    }

    Ok(hypercall_db::SettlementResult {
        newly_persisted: expiration_inserted,
    })
}

fn insert_settlement_ledger_event_in_tx(
    conn: &mut PgConnection,
    wallet: &WalletAddress,
    symbol: &str,
    ledger_delta: Decimal,
    event_ts_ms: i64,
) -> Result<bool> {
    if ledger_delta == Decimal::ZERO {
        return Ok(false);
    }

    let rows_inserted = diesel::sql_query(
        "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type, reference_symbol)
          VALUES ($1, $2, $3, $4, $5)
          ON CONFLICT (wallet, reference_symbol, event_type)
          WHERE reference_symbol IS NOT NULL AND event_type = 'settlement_realized_pnl'
          DO NOTHING",
    )
    .bind::<Binary, _>(wallet)
    .bind::<BigInt, _>(event_ts_ms)
    .bind::<Numeric, _>(ledger_delta)
    .bind::<Text, _>("settlement_realized_pnl")
    .bind::<Nullable<Text>, _>(Some(symbol))
    .execute(conn)?;

    Ok(rows_inserted > 0)
}

fn lock_and_validate_existing_settlement_claim_in_tx(
    conn: &mut PgConnection,
    expiration: &NewPositionExpiration,
) -> Result<()> {
    #[derive(QueryableByName)]
    struct LockedSettlementClaim {
        #[diesel(sql_type = BigInt)]
        id: i64,
        #[diesel(sql_type = Numeric)]
        settlement_price: Decimal,
        #[diesel(sql_type = Numeric)]
        settlement_value: Decimal,
    }

    let claim = diesel::sql_query(
        "SELECT id, settlement_price, settlement_value
         FROM position_expirations
         WHERE wallet = $1 AND symbol = $2 AND expiry_ts = $3
         FOR UPDATE",
    )
    .bind::<Binary, _>(&expiration.wallet)
    .bind::<Text, _>(&expiration.symbol)
    .bind::<BigInt, _>(expiration.expiry_ts)
    .get_result::<LockedSettlementClaim>(conn)
    .with_context(|| {
        format!(
            "Failed to lock existing settlement claim for {}/{} expiry {}",
            expiration.wallet, expiration.symbol, expiration.expiry_ts
        )
    })?;

    if claim.settlement_price != expiration.settlement_price
        || claim.settlement_value != expiration.settlement_value
    {
        panic!(
            "Settlement claim mismatch for {}/{} expiry {}: existing id={} price={} value={}, replay price={} value={}",
            expiration.wallet, expiration.symbol, expiration.expiry_ts,
            claim.id, claim.settlement_price, claim.settlement_value,
            expiration.settlement_price, expiration.settlement_value
        );
    }

    Ok(())
}

fn load_settlement_payout_in_tx(
    conn: &mut PgConnection,
    payout: &NewSettlementPayout,
) -> Result<Option<SettlementPayout>> {
    use schema::settlement_payouts::dsl as sp;

    sp::settlement_payouts
        .filter(sp::wallet.eq(&payout.wallet))
        .filter(sp::symbol.eq(&payout.symbol))
        .filter(sp::expiry_ts.eq(payout.expiry_ts))
        .select(SettlementPayout::as_select())
        .first(conn)
        .optional()
        .map_err(Into::into)
}

fn validate_existing_settlement_payout_facts(
    existing: &SettlementPayout,
    payout: &NewSettlementPayout,
) {
    if existing.position_size != payout.position_size
        || existing.settlement_price != payout.settlement_price
        || existing.payout_amount != payout.payout_amount
        || existing.settlement_entry_price != payout.settlement_entry_price
        || existing.cost_basis != payout.cost_basis
        || existing.net_pnl != payout.net_pnl
    {
        panic!(
            "Settlement payout mismatch for {}/{} expiry {}: existing id={} size={} price={} payout={} entry={:?} cost_basis={:?} net_pnl={:?}, replay size={} price={} payout={} entry={:?} cost_basis={:?} net_pnl={:?}",
            payout.wallet, payout.symbol, payout.expiry_ts,
            existing.id, existing.position_size, existing.settlement_price,
            existing.payout_amount, existing.settlement_entry_price, existing.cost_basis, existing.net_pnl,
            payout.position_size, payout.settlement_price, payout.payout_amount,
            payout.settlement_entry_price, payout.cost_basis, payout.net_pnl
        );
    }
}

fn mark_settlement_payout_applied_in_tx(
    conn: &mut PgConnection,
    existing: &SettlementPayout,
) -> Result<()> {
    use schema::settlement_payouts::dsl as sp;

    let rows = diesel::update(
        sp::settlement_payouts
            .filter(sp::id.eq(existing.id))
            .filter(sp::ledger_applied.eq(false)),
    )
    .set(sp::ledger_applied.eq(true))
    .execute(conn)?;

    if rows != 1 {
        panic!(
            "Settlement payout repair race for {}/{} expiry {}: expected to mark existing id={} applied, updated {} rows",
            existing.wallet, existing.symbol, existing.expiry_ts, existing.id, rows
        );
    }

    Ok(())
}

fn insert_settlement_payout_in_tx(
    conn: &mut PgConnection,
    payout: &NewSettlementPayout,
) -> Result<()> {
    diesel::insert_into(schema::settlement_payouts::table)
        .values(payout)
        .execute(conn)?;
    Ok(())
}

fn validate_existing_settlement_ledger_in_tx(
    conn: &mut PgConnection,
    wallet: &WalletAddress,
    symbol: &str,
    ledger_delta: Decimal,
) -> Result<()> {
    #[derive(QueryableByName)]
    struct SettlementLedgerEvent {
        #[diesel(sql_type = Numeric)]
        delta: Decimal,
    }

    let event = diesel::sql_query(
        "SELECT delta
         FROM ledger_events
         WHERE wallet = $1
           AND reference_symbol = $2
           AND event_type = 'settlement_realized_pnl'
         FOR UPDATE",
    )
    .bind::<Binary, _>(wallet)
    .bind::<Nullable<Text>, _>(Some(symbol))
    .load::<SettlementLedgerEvent>(conn)?;
    let deltas = event
        .iter()
        .map(|existing| existing.delta)
        .collect::<Vec<_>>();

    validate_settlement_ledger_deltas(wallet, symbol, ledger_delta, &deltas)
}

fn validate_settlement_ledger_deltas(
    wallet: &WalletAddress,
    symbol: &str,
    ledger_delta: Decimal,
    deltas: &[Decimal],
) -> Result<()> {
    match (ledger_delta == Decimal::ZERO, deltas) {
        (true, []) => Ok(()),
        (true, [existing_delta]) => {
            panic!(
                "Settlement ledger mismatch for {}/{}: expected no zero-delta ledger event, found delta={}",
                wallet, symbol, existing_delta
            );
        }
        (false, [existing_delta]) if *existing_delta == ledger_delta => Ok(()),
        (false, [existing_delta]) => {
            panic!(
                "Settlement ledger mismatch for {}/{}: existing delta={}, replay delta={}",
                wallet, symbol, existing_delta, ledger_delta
            );
        }
        (false, []) => {
            panic!(
                "Partial settlement ledger for {}/{}: nonzero settlement has no ledger event",
                wallet, symbol
            );
        }
        (_, _) => {
            panic!(
                "Settlement ledger invariant violation for {}/{}: multiple settlement ledger events found",
                wallet, symbol
            );
        }
    }
}

fn validate_existing_settlement_claim_readonly(
    conn: &mut PgConnection,
    expiration: &NewPositionExpiration,
) -> Result<()> {
    #[derive(QueryableByName)]
    struct ExistingSettlementClaim {
        #[diesel(sql_type = BigInt)]
        id: i64,
        #[diesel(sql_type = Numeric)]
        settlement_price: Decimal,
        #[diesel(sql_type = Numeric)]
        settlement_value: Decimal,
    }

    let claim = diesel::sql_query(
        "SELECT id, settlement_price, settlement_value
         FROM position_expirations
         WHERE wallet = $1 AND symbol = $2 AND expiry_ts = $3",
    )
    .bind::<Binary, _>(&expiration.wallet)
    .bind::<Text, _>(&expiration.symbol)
    .bind::<BigInt, _>(expiration.expiry_ts)
    .get_result::<ExistingSettlementClaim>(conn)
    .with_context(|| {
        format!(
            "Failed to load existing settlement claim for {}/{} expiry {}",
            expiration.wallet, expiration.symbol, expiration.expiry_ts
        )
    })?;

    if claim.settlement_price != expiration.settlement_price
        || claim.settlement_value != expiration.settlement_value
    {
        panic!(
            "Settlement claim mismatch for {}/{} expiry {}: existing id={} price={} value={}, replay price={} value={}",
            expiration.wallet,
            expiration.symbol,
            expiration.expiry_ts,
            claim.id,
            claim.settlement_price,
            claim.settlement_value,
            expiration.settlement_price,
            expiration.settlement_value
        );
    }

    Ok(())
}

fn validate_existing_settlement_ledger_readonly(
    conn: &mut PgConnection,
    wallet: &WalletAddress,
    symbol: &str,
    ledger_delta: Decimal,
) -> Result<()> {
    #[derive(QueryableByName)]
    struct SettlementLedgerEvent {
        #[diesel(sql_type = Numeric)]
        delta: Decimal,
    }

    let event = diesel::sql_query(
        "SELECT delta
         FROM ledger_events
         WHERE wallet = $1
           AND reference_symbol = $2
           AND event_type = 'settlement_realized_pnl'",
    )
    .bind::<Binary, _>(wallet)
    .bind::<Nullable<Text>, _>(Some(symbol))
    .load::<SettlementLedgerEvent>(conn)?;
    let deltas = event
        .iter()
        .map(|existing| existing.delta)
        .collect::<Vec<_>>();

    validate_settlement_ledger_deltas(wallet, symbol, ledger_delta, &deltas)
}

/// Read-only validation of an already-applied settlement (standby replay path).
/// Panics on any inconsistency between the replay and the persisted state.
pub fn observe_applied_settlement_in_tx(
    conn: &mut PgConnection,
    expiration: &NewPositionExpiration,
    payout: &NewSettlementPayout,
    ledger_delta: Decimal,
) -> Result<hypercall_db::SettlementResult> {
    validate_existing_settlement_claim_readonly(conn, expiration)?;
    let Some(existing) = load_settlement_payout_in_tx(conn, payout)? else {
        panic!(
            "Partial settlement claim for {}/{} expiry {}: expiration row exists without payout",
            payout.wallet, payout.symbol, payout.expiry_ts
        );
    };
    validate_existing_settlement_payout_facts(&existing, payout);
    if !existing.ledger_applied {
        panic!(
            "Partial settlement payout for {}/{} expiry {}: payout id={} exists but ledger_applied=false during standby replay",
            payout.wallet, payout.symbol, payout.expiry_ts, existing.id
        );
    }
    validate_existing_settlement_ledger_readonly(
        conn,
        &expiration.wallet,
        &expiration.symbol,
        ledger_delta,
    )?;
    Ok(hypercall_db::SettlementResult {
        newly_persisted: false,
    })
}

/// Total fill count and notional volume across all fills.
pub fn query_total_fill_volume(conn: &mut PgConnection) -> Result<(i64, Decimal)> {
    #[derive(QueryableByName)]
    struct Row {
        #[diesel(sql_type = BigInt)]
        fill_count: i64,
        #[diesel(sql_type = Numeric)]
        total_notional: Decimal,
    }

    let result = diesel::sql_query(
        "SELECT COUNT(*)::bigint AS fill_count, \
                COALESCE(SUM(ABS(size) / 1000000.0 * price), 0)::numeric AS total_notional \
         FROM fills",
    )
    .get_result::<Row>(conn)?;
    Ok((result.fill_count, result.total_notional))
}