hypercall_db_diesel/

database_handler.rs

//! Unified database handler that owns a connection pool and implements
//! all persistence traits from `hypercall_db`.
//!
//! This is the standalone Diesel persistence layer. It has NO dependency
//! on the root `hypercall` crate and owns its own `DbPool` directly.
//! All SQL is inline or delegated to local modules (`crate::ledger_ops`,
//! `crate::settlement_ops`, `crate::order_status_materialization`).
//!
//! Trait implementations live in sibling modules:
//! - `crate::instruments` -- InstrumentReader / InstrumentWriter
//! - `crate::tiers`       -- TierReader / TierWriter
//! - `crate::mmp`         -- MmpConfigReader / MmpConfigWriter
//! - `crate::orders`      -- OrderReader / OrderWriter + fill ledger helper
//! - `crate::transaction` -- Transactional (+ DieselTransaction)
//! - `crate::replay`      -- JournalReplayReader
//! - `crate::rfq`         -- RfqReader / RfqWriter
//! - `crate::settlements` -- SettlementReader / SettlementWriter

use crate::db_auth::DbAuthConfig;
use anyhow::{Context, Result};
use diesel::backend::Backend;
use diesel::connection::SimpleConnection;
use diesel::migration::{
    Migration, MigrationMetadata, MigrationName, MigrationSource, MigrationVersion,
};
use diesel::pg::Pg;
use diesel::pg::PgConnection;
use diesel::prelude::*;
use diesel::r2d2::{self, ConnectionManager};
use diesel::{Connection, RunQueryDsl};
use diesel_migrations::{EmbeddedMigrations, MigrationHarness};
use std::fmt;
use std::sync::Arc;
use tracing::info;

/// Connection manager that resolves the database URL on every `connect()` call.
///
/// In password mode the URL is a captured `String` (zero-cost clone via `Arc`).
/// In RDS IAM mode the closure reads the latest cached token from the provider.
pub struct DynConnectionManager {
    url_source: Arc<dyn Fn() -> String + Send + Sync>,
}

impl DynConnectionManager {
    /// Construct from a [`DbAuthConfig`]. Password mode captures a static URL;
    /// IAM mode calls `current_url()` on each connection attempt.
    pub fn new(auth: &DbAuthConfig) -> Self {
        let auth = auth.clone();
        Self {
            url_source: Arc::new(move || auth.current_url()),
        }
    }
}

impl r2d2::ManageConnection for DynConnectionManager {
    type Connection = PgConnection;
    type Error = r2d2::Error;

    fn connect(&self) -> std::result::Result<PgConnection, r2d2::Error> {
        let url = (self.url_source)();
        PgConnection::establish(&url).map_err(r2d2::Error::ConnectionError)
    }

    fn is_valid(&self, conn: &mut PgConnection) -> std::result::Result<(), r2d2::Error> {
        conn.batch_execute("SELECT 1")
            .map_err(r2d2::Error::QueryError)
    }

    fn has_broken(&self, conn: &mut PgConnection) -> bool {
        use diesel::r2d2::R2D2Connection;
        std::thread::panicking() || conn.is_broken()
    }
}

impl std::fmt::Debug for DynConnectionManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DynConnectionManager").finish()
    }
}

/// Synchronous r2d2 connection pool for Postgres.
pub type DbPool = diesel::r2d2::Pool<DynConnectionManager>;

/// Legacy pool type alias for backward compatibility with code that constructs
/// pools directly with `ConnectionManager`.
pub type LegacyDbPool = diesel::r2d2::Pool<ConnectionManager<PgConnection>>;

fn wallet_from_bytes(bytes: &[u8], label: &str) -> Result<hypercall_types::WalletAddress> {
    let arr: [u8; 20] = bytes.try_into().map_err(|_| {
        anyhow::anyhow!(
            "invalid {} address length {}, expected 20",
            label,
            bytes.len()
        )
    })?;
    Ok(hypercall_types::WalletAddress::from(arr))
}

fn rsm_deposit_match_from_row(
    request_id: String,
    account: Vec<u8>,
    token: Vec<u8>,
    tx_hash: String,
    log_index: i64,
    amount_wei: String,
    label: &str,
) -> Result<hypercall_db::RsmUsdcDepositMatch> {
    let account = wallet_from_bytes(
        account.as_slice(),
        &format!("{label} account {tx_hash}:{log_index}"),
    )?;
    let token = wallet_from_bytes(
        token.as_slice(),
        &format!("{label} token {tx_hash}:{log_index}"),
    )?;
    Ok(hypercall_db::RsmUsdcDepositMatch {
        request_id,
        account,
        tx_hash,
        log_index,
        amount_wei,
        token,
    })
}

/// Read the option token deployment parameters from environment variables.
/// Used by `save_market_and_instrument_sync` to derive option token addresses.
pub(crate) fn current_option_token_deployment() -> Result<hypercall_types::OptionTokenDeployment> {
    use alloy::primitives::{Address, B256};
    use std::str::FromStr;

    let registry_str = std::env::var("OPTION_REGISTRY_ADDRESS").map_err(|_| {
        anyhow::anyhow!(
            "Option token derivation requires OPTION_REGISTRY_ADDRESS from backend contracts config"
        )
    })?;
    let init_hash_str =
        std::env::var("OPTION_TOKEN_BEACON_PROXY_INIT_CODE_HASH").map_err(|_| {
            anyhow::anyhow!(
            "Option token derivation requires OPTION_TOKEN_BEACON_PROXY_INIT_CODE_HASH from backend contracts config"
        )
        })?;

    let option_registry = Address::from_str(registry_str.trim()).map_err(|e| {
        anyhow::anyhow!(
            "Option token derivation requires valid OPTION_REGISTRY_ADDRESS: {}",
            e
        )
    })?;
    let beacon_proxy_init_code_hash = B256::from_str(init_hash_str.trim()).map_err(|e| {
        anyhow::anyhow!(
            "Option token derivation requires valid OPTION_TOKEN_BEACON_PROXY_INIT_CODE_HASH: {}",
            e
        )
    })?;

    Ok(hypercall_types::OptionTokenDeployment::new(
        option_registry,
        beacon_proxy_init_code_hash,
    ))
}

/// Embed migrations relative to hypercall-rs/db-diesel/Cargo.toml.
pub const MIGRATIONS: EmbeddedMigrations = diesel_migrations::embed_migrations!("../../migrations");

const ORDER_INFOS_ACTIVE_SYMBOL_INDEX_MIGRATION: &str =
    "2026-06-01-000001_order_infos_active_symbol_index";

/// Diesel runs each embedded migration file with one `batch_execute` call.
/// PostgreSQL rejects `CREATE/DROP INDEX CONCURRENTLY` when it appears in a
/// multi-statement batch, even outside an explicit transaction. This source
/// replaces the one affected migration with a splitter that executes each
/// statement separately while preserving Diesel's migration ledger behavior.
struct HypercallMigrations;

impl MigrationSource<Pg> for HypercallMigrations {
    fn migrations(&self) -> diesel::migration::Result<Vec<Box<dyn Migration<Pg>>>> {
        MIGRATIONS
            .migrations()?
            .into_iter()
            .map(|migration| {
                if migration.name().to_string() == ORDER_INFOS_ACTIVE_SYMBOL_INDEX_MIGRATION {
                    Ok(
                        Box::new(SplitStatementMigration::order_infos_active_symbol_index())
                            as Box<dyn Migration<Pg>>,
                    )
                } else {
                    Ok(migration)
                }
            })
            .collect()
    }
}

#[derive(Debug)]
struct SplitStatementMigration {
    name: StaticMigrationName,
    metadata: NonTransactionalMigrationMetadata,
    up_sql: &'static str,
    down_sql: &'static str,
}

impl SplitStatementMigration {
    fn order_infos_active_symbol_index() -> Self {
        Self {
            name: StaticMigrationName(ORDER_INFOS_ACTIVE_SYMBOL_INDEX_MIGRATION),
            metadata: NonTransactionalMigrationMetadata,
            up_sql: include_str!(
                "../../../migrations/2026-06-01-000001_order_infos_active_symbol_index/up.sql"
            ),
            down_sql: include_str!(
                "../../../migrations/2026-06-01-000001_order_infos_active_symbol_index/down.sql"
            ),
        }
    }

    fn execute_statements<DB: Backend>(
        &self,
        conn: &mut dyn diesel::connection::BoxableConnection<DB>,
        sql: &str,
    ) -> diesel::migration::Result<()> {
        for statement in sql.split(';').map(str::trim).filter(|s| !s.is_empty()) {
            conn.batch_execute(statement).map_err(|error| {
                let message = format!(
                    "failed to run statement in migration {}: {}",
                    self.name, error
                );
                Box::new(std::io::Error::new(std::io::ErrorKind::Other, message))
                    as Box<dyn std::error::Error + Send + Sync>
            })?;
        }
        Ok(())
    }
}

impl Migration<Pg> for SplitStatementMigration {
    fn run(
        &self,
        conn: &mut dyn diesel::connection::BoxableConnection<Pg>,
    ) -> diesel::migration::Result<()> {
        self.execute_statements(conn, self.up_sql)
    }

    fn revert(
        &self,
        conn: &mut dyn diesel::connection::BoxableConnection<Pg>,
    ) -> diesel::migration::Result<()> {
        self.execute_statements(conn, self.down_sql)
    }

    fn metadata(&self) -> &dyn MigrationMetadata {
        &self.metadata
    }

    fn name(&self) -> &dyn MigrationName {
        &self.name
    }
}

#[derive(Debug)]
struct StaticMigrationName(&'static str);

impl MigrationName for StaticMigrationName {
    fn version(&self) -> MigrationVersion<'_> {
        self.0.split('_').next().unwrap_or(self.0).into()
    }
}

impl fmt::Display for StaticMigrationName {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.0)
    }
}

#[derive(Debug)]
struct NonTransactionalMigrationMetadata;

impl MigrationMetadata for NonTransactionalMigrationMetadata {
    fn run_in_transaction(&self) -> bool {
        false
    }
}

/// Sets statement_timeout and lock_timeout on every new Postgres connection.
#[derive(Debug)]
struct PgTimeoutCustomizer {
    statement_timeout_ms: u32,
    lock_timeout_ms: u32,
}

impl r2d2::CustomizeConnection<PgConnection, r2d2::Error> for PgTimeoutCustomizer {
    fn on_acquire(&self, conn: &mut PgConnection) -> std::result::Result<(), r2d2::Error> {
        conn.batch_execute(&format!(
            "SET statement_timeout = '{}'; SET lock_timeout = '{}'",
            self.statement_timeout_ms, self.lock_timeout_ms,
        ))
        .map_err(r2d2::Error::QueryError)
    }
}

/// Redact credentials from a database URL for safe logging.
fn redact_database_url(url: &str) -> String {
    url.find('@')
        .map(|at| format!("postgres://***@{}", &url[at + 1..]))
        .unwrap_or_else(|| "postgres://***".to_string())
}

/// Build a raw r2d2 pool with the given auth config, for callers that manage
/// the pool lifecycle themselves (e.g. `integrated.rs`).
pub fn build_db_pool(
    auth: &DbAuthConfig,
    pool_max: u32,
    statement_timeout_ms: u32,
    lock_timeout_ms: u32,
) -> Result<DbPool> {
    let manager = DynConnectionManager::new(auth);
    r2d2::Pool::builder()
        .max_size(pool_max)
        .min_idle(Some(0))
        .connection_timeout(std::time::Duration::from_secs(5))
        .idle_timeout(Some(std::time::Duration::from_secs(300)))
        .connection_customizer(Box::new(PgTimeoutCustomizer {
            statement_timeout_ms,
            lock_timeout_ms,
        }))
        .build(manager)
        .with_context(|| "Failed to create connection pool")
}

/// Synchronous persistence handler. Owns an r2d2 connection pool and
/// implements all engine-path traits from `hypercall_db` (orders,
/// instruments, tiers, settlements, journal replay, archiver, RFQ).
///
/// Constructors run Diesel migrations and `ensure_enum_values` by default.
/// Use [`Self::new_readonly`] or [`Self::with_pool_no_migrations`] to skip.
pub struct DatabaseHandler {
    pool: Arc<DbPool>,
}

impl DatabaseHandler {
    /// Create a new handler: build pool + run migrations.
    pub fn new(database_url: &str) -> Result<Self> {
        Self::new_with_pool_size(database_url, 3)
    }

    /// Create a new handler with explicit pool size: build pool + run migrations.
    pub fn new_with_pool_size(database_url: &str, pool_max: u32) -> Result<Self> {
        Self::new_with_auth(DbAuthConfig::password(database_url), pool_max)
    }

    /// Create a new handler with [`DbAuthConfig`]: build pool + run migrations.
    pub fn new_with_auth(auth: DbAuthConfig, pool_max: u32) -> Result<Self> {
        let handler = Self::build_pool(&auth, pool_max, 30_000, 10_000)?;
        handler.run_migrations().with_context(|| {
            format!(
                "Failed to run migrations for database (auth={:?}): {}",
                auth,
                redact_database_url(&auth.current_url())
            )
        })?;
        Ok(handler)
    }

    /// Create a handler with custom statement/lock timeouts, no migrations.
    /// Used by db-archiver which needs longer timeouts for batch operations.
    pub fn new_with_timeouts(
        database_url: &str,
        pool_max: u32,
        statement_timeout_ms: u32,
        lock_timeout_ms: u32,
    ) -> Result<Self> {
        Self::new_with_timeouts_auth(
            DbAuthConfig::password(database_url),
            pool_max,
            statement_timeout_ms,
            lock_timeout_ms,
        )
    }

    /// Create a handler with custom timeouts and [`DbAuthConfig`], no migrations.
    pub fn new_with_timeouts_auth(
        auth: DbAuthConfig,
        pool_max: u32,
        statement_timeout_ms: u32,
        lock_timeout_ms: u32,
    ) -> Result<Self> {
        let redacted = redact_database_url(&auth.current_url());
        info!(
            "Initializing DatabaseHandler (custom timeouts: stmt={}ms, lock={}ms): {}",
            statement_timeout_ms, lock_timeout_ms, redacted
        );
        Self::build_pool(&auth, pool_max, statement_timeout_ms, lock_timeout_ms)
    }

    /// Create a handler with pool only, no migrations (for read-only replicas).
    pub fn new_readonly(database_url: &str, pool_max: u32) -> Result<Self> {
        Self::new_readonly_auth(DbAuthConfig::password(database_url), pool_max)
    }

    /// Create a readonly handler with [`DbAuthConfig`], no migrations.
    pub fn new_readonly_auth(auth: DbAuthConfig, pool_max: u32) -> Result<Self> {
        let redacted = redact_database_url(&auth.current_url());
        info!(
            "Initializing readonly DatabaseHandler (migrations skipped): {}",
            redacted
        );
        Self::build_pool(&auth, pool_max, 30_000, 10_000)
    }

    /// Wrap an existing pool + run migrations.
    pub fn with_pool(pool: Arc<DbPool>) -> Result<Self> {
        info!("Creating DatabaseHandler with existing pool");
        let handler = Self { pool };
        handler
            .run_migrations()
            .with_context(|| "Failed to run migrations")?;
        Ok(handler)
    }

    /// Wrap an existing pool, skip migrations.
    pub fn with_pool_no_migrations(pool: Arc<DbPool>) -> Self {
        info!("Creating DatabaseHandler with existing pool (migrations skipped)");
        Self { pool }
    }

    /// Expose the pool (Arc-cloned).
    pub fn pool(&self) -> Arc<DbPool> {
        self.pool.clone()
    }

    /// Clone the inner pool (not the Arc).
    pub fn get_pool(&self) -> DbPool {
        (*self.pool).clone()
    }

    /// Run pending Diesel migrations + ensure enum values.
    pub fn run_migrations(&self) -> Result<()> {
        let mut conn = self.pool.get()?;

        match conn.run_pending_migrations(HypercallMigrations) {
            Ok(migrations) => {
                info!(
                    "Database migrations completed successfully: {} migrations applied",
                    migrations.len()
                );
            }
            Err(e) => return Err(anyhow::anyhow!("Migration failed: {}", e)),
        }

        Self::ensure_enum_values(&mut conn)?;
        Self::ensure_directive_action_key_enum(&mut conn)?;
        Self::ensure_directive_outbox_wallet_address(&mut conn)?;
        Self::ensure_real_liquidation_schema(&mut conn)?;

        Ok(())
    }

    fn ensure_real_liquidation_schema(conn: &mut PgConnection) -> Result<()> {
        diesel::sql_query(
            "ALTER TABLE liquidation_states
                ADD COLUMN IF NOT EXISTS liquidation_mode TEXT,
                ADD COLUMN IF NOT EXISTS target_equity NUMERIC,
                ADD COLUMN IF NOT EXISTS escalation_deadline BIGINT,
                ADD COLUMN IF NOT EXISTS last_reprice_at BIGINT,
                ADD COLUMN IF NOT EXISTS partial_order_request_ids JSONB,
                ADD COLUMN IF NOT EXISTS partial_order_client_ids JSONB,
                ADD COLUMN IF NOT EXISTS partial_bonus_bps INTEGER,
                ADD COLUMN IF NOT EXISTS request_id TEXT,
                ADD COLUMN IF NOT EXISTS tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS chain_start_time BIGINT,
                ADD COLUMN IF NOT EXISTS margin_needed NUMERIC,
                ADD COLUMN IF NOT EXISTS stop_request_id TEXT,
                ADD COLUMN IF NOT EXISTS stop_tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS resolved_winner BYTEA,
                ADD COLUMN IF NOT EXISTS resolved_bonus NUMERIC,
                ADD COLUMN IF NOT EXISTS resolution_tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS last_observed_block BIGINT,
                ADD COLUMN IF NOT EXISTS updated_at_ms BIGINT",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure liquidation_states schema: {}", e))?;

        diesel::sql_query(
            "ALTER TABLE liquidation_history
                ADD COLUMN IF NOT EXISTS liquidation_mode TEXT,
                ADD COLUMN IF NOT EXISTS maintenance_margin NUMERIC,
                ADD COLUMN IF NOT EXISTS request_id TEXT,
                ADD COLUMN IF NOT EXISTS tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS margin_needed NUMERIC,
                ADD COLUMN IF NOT EXISTS winner_address BYTEA,
                ADD COLUMN IF NOT EXISTS bonus NUMERIC,
                ADD COLUMN IF NOT EXISTS details JSONB NOT NULL DEFAULT '{}'::jsonb",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure liquidation_history schema: {}", e))?;

        diesel::sql_query(
            "UPDATE liquidation_history
             SET maintenance_margin = equity - mm_required
             WHERE maintenance_margin IS NULL",
        )
        .execute(conn)
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to backfill liquidation_history.maintenance_margin: {}",
                e
            )
        })?;

        diesel::sql_query(
            "ALTER TABLE liquidation_history
             ALTER COLUMN maintenance_margin SET NOT NULL",
        )
        .execute(conn)
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to enforce liquidation_history.maintenance_margin not-null: {}",
                e
            )
        })?;

        diesel::sql_query(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_liquidation_history_replay_guard
             ON liquidation_history (
                 wallet_address, previous_state, new_state, timestamp,
                 COALESCE(auction_id, ''), COALESCE(request_id, ''), COALESCE(tx_hash, '')
             )",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure liquidation_history replay index: {}", e))?;

        diesel::sql_query(
            "ALTER TABLE liquidation_auctions
                ADD COLUMN IF NOT EXISTS target_equity NUMERIC,
                ADD COLUMN IF NOT EXISTS request_id TEXT,
                ADD COLUMN IF NOT EXISTS tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS chain_start_time BIGINT,
                ADD COLUMN IF NOT EXISTS margin_needed NUMERIC,
                ADD COLUMN IF NOT EXISTS stop_request_id TEXT,
                ADD COLUMN IF NOT EXISTS stop_tx_hash TEXT,
                ADD COLUMN IF NOT EXISTS bonus NUMERIC,
                ADD COLUMN IF NOT EXISTS last_observed_block BIGINT",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure liquidation_auctions schema: {}", e))?;

        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS rsm_signer_nonces (
                signer_address BYTEA PRIMARY KEY,
                next_nonce BIGINT NOT NULL,
                last_synced_nonce BIGINT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure rsm_signer_nonces table: {}", e))?;

        diesel::sql_query(
            "CREATE TABLE IF NOT EXISTS rsm_signer_requests (
                request_id TEXT PRIMARY KEY,
                signer_address BYTEA NOT NULL,
                account_address BYTEA NOT NULL,
                action BYTEA NOT NULL,
                nonce BIGINT NOT NULL,
                status TEXT NOT NULL,
                directive BYTEA,
                signature TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )",
        )
        .execute(conn)
        .map_err(|e| anyhow::anyhow!("Failed to ensure rsm_signer_requests table: {}", e))?;

        diesel::sql_query(
            "ALTER TABLE rsm_signer_requests ADD COLUMN IF NOT EXISTS signer_address BYTEA",
        )
        .execute(conn)
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to ensure rsm_signer_requests signer_address column: {}",
                e
            )
        })?;

        diesel::sql_query("DELETE FROM rsm_signer_requests WHERE signer_address IS NULL")
            .execute(conn)
            .map_err(|e| {
                anyhow::anyhow!(
                    "Failed to remove rsm_signer_requests rows missing signer_address: {}",
                    e
                )
            })?;

        diesel::sql_query(
            "ALTER TABLE rsm_signer_requests ALTER COLUMN signer_address SET NOT NULL",
        )
        .execute(conn)
        .map_err(|e| {
            anyhow::anyhow!(
                "Failed to require rsm_signer_requests signer_address: {}",
                e
            )
        })?;

        Ok(())
    }

    // ------------------------------------------------------------------
    // Internal helpers
    // ------------------------------------------------------------------

    fn build_pool(
        auth: &DbAuthConfig,
        pool_max: u32,
        statement_timeout_ms: u32,
        lock_timeout_ms: u32,
    ) -> Result<Self> {
        let redacted = redact_database_url(&auth.current_url());
        info!(
            "Initializing DatabaseHandler (auth={:?}): {}",
            auth, redacted
        );
        let manager = DynConnectionManager::new(auth);
        let pool = r2d2::Pool::builder()
            .max_size(pool_max)
            .min_idle(Some(0))
            .connection_timeout(std::time::Duration::from_secs(5))
            .idle_timeout(Some(std::time::Duration::from_secs(300)))
            .connection_customizer(Box::new(PgTimeoutCustomizer {
                statement_timeout_ms,
                lock_timeout_ms,
            }))
            .build(manager)
            .with_context(|| {
                format!(
                    "Failed to create connection pool for database: {}",
                    redacted
                )
            })?;

        Ok(Self {
            pool: Arc::new(pool),
        })
    }

    /// Ensure all required Postgres enum values exist.
    fn ensure_enum_values(conn: &mut PgConnection) -> Result<()> {
        let stmts = [
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'RfqExecute'",
            "ALTER TYPE engine_event_type ADD VALUE IF NOT EXISTS 'RfqFilled'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'ReplaceOrder' AFTER 'CancelOrder'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'PriceUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'IvUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'TierUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'HypercorePositionUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'MmpConfigUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'LiquidationBonusUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'ApproveAgent'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'RevokeAgent'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'NonceAdvance'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'HypercoreEquityUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'OptionDepositUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'OptionWithdrawalUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'CashWithdrawalUpdate'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'RecordPmVaultDeposit'",
            "ALTER TYPE engine_command_type ADD VALUE IF NOT EXISTS 'RequestPmVaultWithdrawal'",
        ];

        for stmt in &stmts {
            diesel::sql_query(*stmt)
                .execute(conn)
                .map_err(|e| anyhow::anyhow!("Failed to ensure enum value: {} - {}", stmt, e))?;
        }

        info!("Ensured all required Postgres enum values exist");
        Ok(())
    }

    /// Ensure the directive outbox action-key enum exists even if an older
    /// duplicate-version migration was recorded before this enum migration ran.
    fn ensure_directive_action_key_enum(conn: &mut PgConnection) -> Result<()> {
        diesel::sql_query(
            r#"
            DO $$
            BEGIN
                IF NOT EXISTS (
                    SELECT 1
                    FROM pg_type
                    WHERE typname = 'directive_action_key'
                ) THEN
                    CREATE TYPE directive_action_key AS ENUM (
                        'hl_limit_order',
                        'hl_cancel_by_oid',
                        'hl_cancel_by_cloid',
                        'hc_update_api_wallet',
                        'hl_send_asset',
                        'hc_transfer_option',
                        'rsm_hl_limit_order',
                        'rsm_hl_cancel_by_oid',
                        'rsm_hl_cancel_by_cloid',
                        'rsm_hl_send_asset',
                        'system_credit_token',
                        'system_credit_option',
                        'system_start_liquidation',
                        'system_stop_liquidation',
                        'system_withdraw_token'
                    );
                END IF;
            END $$;
            "#,
        )
        .execute(conn)
        .context("Failed to ensure directive_action_key enum type exists")?;

        let values = [
            "hl_limit_order",
            "hl_cancel_by_oid",
            "hl_cancel_by_cloid",
            "hc_update_api_wallet",
            "hl_send_asset",
            "hc_transfer_option",
            "rsm_hl_limit_order",
            "rsm_hl_cancel_by_oid",
            "rsm_hl_cancel_by_cloid",
            "rsm_hl_send_asset",
            "system_credit_token",
            "system_credit_option",
            "system_start_liquidation",
            "system_stop_liquidation",
            "system_withdraw_token",
        ];
        for value in values {
            diesel::sql_query(format!(
                "ALTER TYPE directive_action_key ADD VALUE IF NOT EXISTS '{}'",
                value
            ))
            .execute(conn)
            .with_context(|| format!("Failed to ensure directive_action_key enum value {value}"))?;
        }

        diesel::sql_query(
            r#"
            ALTER TABLE directive_outbox
                ALTER COLUMN action_key TYPE directive_action_key
                USING action_key::text::directive_action_key
            "#,
        )
        .execute(conn)
        .context("Failed to ensure directive_outbox.action_key uses directive_action_key")?;

        info!("Ensured directive_action_key enum and directive_outbox schema exist");
        Ok(())
    }

    fn ensure_directive_outbox_wallet_address(conn: &mut PgConnection) -> Result<()> {
        diesel::sql_query(
            r#"
            ALTER TABLE directive_outbox
                ADD COLUMN IF NOT EXISTS wallet_address BYTEA
            "#,
        )
        .execute(conn)
        .context("Failed to ensure directive_outbox.wallet_address column exists")?;

        diesel::sql_query(
            r#"
            UPDATE directive_outbox
            SET wallet_address = account_address
            WHERE wallet_address IS NULL
            "#,
        )
        .execute(conn)
        .context("Failed to backfill directive_outbox.wallet_address")?;

        diesel::sql_query(
            r#"
            ALTER TABLE directive_outbox
                ALTER COLUMN wallet_address SET NOT NULL
            "#,
        )
        .execute(conn)
        .context("Failed to ensure directive_outbox.wallet_address is NOT NULL")?;

        diesel::sql_query(
            r#"
            CREATE INDEX IF NOT EXISTS idx_directive_outbox_withdrawal_wallet_created
                ON directive_outbox (wallet_address, created_ts_ms DESC)
            "#,
        )
        .execute(conn)
        .context("Failed to ensure directive_outbox wallet history index")?;

        info!("Ensured directive_outbox.wallet_address schema exists");
        Ok(())
    }

    /// Map an `OrderUpdateStatus` enum to its DB string representation.
    pub(crate) fn order_update_status_to_db(
        status: hypercall_types::OrderUpdateStatus,
    ) -> &'static str {
        match status {
            hypercall_types::OrderUpdateStatus::Acked => "ACKED",
            hypercall_types::OrderUpdateStatus::Open => "OPEN",
            hypercall_types::OrderUpdateStatus::PartiallyFilled => "PARTIALLY_FILLED",
            hypercall_types::OrderUpdateStatus::Filled => "FILLED",
            hypercall_types::OrderUpdateStatus::Canceled => "CANCELED",
            hypercall_types::OrderUpdateStatus::Rejected => "REJECTED",
        }
    }

    /// Log and count option_token_address unique constraint violations.
    pub(crate) fn observe_diesel_option_token_violation(err: &diesel::result::Error) {
        if let diesel::result::Error::DatabaseError(
            diesel::result::DatabaseErrorKind::UniqueViolation,
            ref info,
        ) = err
        {
            if info
                .constraint_name()
                .is_some_and(|c| c.contains("option_token_address"))
            {
                tracing::warn!(
                    "Option token address unique constraint violation (likely re-listing): {}",
                    info.message()
                );
                metrics::counter!("ht_diesel_option_token_violation_total").increment(1);
            }
        }
    }

    /// Persist the engine state snapshot's last_command_id to Postgres so
    /// the db-archiver can compute a safe deletion boundary. Must be called
    /// after every successful snapshot write to disk.
    pub fn update_snapshot_boundary_sync(&self, last_command_id: i64) -> Result<()> {
        let mut conn = self.pool.get()?;
        diesel::sql_query(
            "INSERT INTO engine_snapshot_boundary (id, last_command_id, updated_at) \
             VALUES (1, $1, now()) \
             ON CONFLICT (id) DO UPDATE SET last_command_id = $1, updated_at = now()",
        )
        .bind::<diesel::sql_types::BigInt, _>(last_command_id)
        .execute(&mut conn)?;
        Ok(())
    }

    // ===== Directive deposit/withdrawal methods =====

    /// Check whether a directive outbox row exists for the given directive_id.
    pub fn directive_outbox_exists_sync(&self, directive_id: &str) -> Result<bool> {
        use diesel::sql_types::Text;

        #[derive(diesel::QueryableByName)]
        struct ExistsRow {
            #[diesel(sql_type = diesel::sql_types::Bool)]
            exists: bool,
        }

        let mut conn = self.pool.get()?;
        let row = diesel::sql_query(
            "SELECT EXISTS (
                 SELECT 1 FROM directive_outbox WHERE directive_id = $1
             ) AS exists",
        )
        .bind::<Text, _>(directive_id)
        .get_result::<ExistsRow>(&mut conn)?;
        Ok(row.exists)
    }

    /// Get the maximum observed block from rsm_deposit_credits.
    pub async fn get_max_rsm_deposit_credit_observed_block(&self) -> Result<Option<u64>> {
        use diesel::sql_types::{BigInt, Nullable};

        #[derive(diesel::QueryableByName)]
        struct MaxBlockRow {
            #[diesel(sql_type = Nullable<BigInt>)]
            max_block: Option<i64>,
        }

        let mut conn = self.pool.get()?;
        let row = diesel::sql_query(
            "SELECT COALESCE(
                MIN(observed_block) FILTER (WHERE status <> 'submitted'),
                MAX(observed_block)
             ) AS max_block
             FROM rsm_deposit_credits",
        )
        .get_result::<MaxBlockRow>(&mut conn)?;

        row.max_block
            .map(|value| {
                u64::try_from(value).map_err(|_| {
                    anyhow::anyhow!(
                        "negative rsm_deposit_credits observed_block {} persisted in database",
                        value
                    )
                })
            })
            .transpose()
    }

    /// Ensure a user_tiers row exists for the wallet that will receive deposit credit.
    pub async fn ensure_observed_deposit_account(
        &self,
        account: &hypercall_types::WalletAddress,
    ) -> Result<()> {
        use diesel::sql_types::Binary;

        let mut conn = self.pool.get()?;
        diesel::sql_query(
            "INSERT INTO user_tiers (wallet_address, tier, margin_mode, version)
             VALUES ($1, 'tier2', 'standard', 1)
             ON CONFLICT (wallet_address) DO NOTHING",
        )
        .bind::<Binary, _>(account)
        .execute(&mut conn)?;
        Ok(())
    }

    /// Claim an RSM deposit credit (idempotent upsert).
    pub async fn claim_rsm_deposit_credit(
        &self,
        input: &hypercall_db::RsmDepositCreditClaimInput,
    ) -> Result<hypercall_db::RsmDepositCreditClaimRecord> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct ClaimRow {
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = BigInt)]
            observed_block: i64,
            #[diesel(sql_type = Binary)]
            account: hypercall_types::WalletAddress,
            #[diesel(sql_type = Binary)]
            token: hypercall_types::WalletAddress,
            #[diesel(sql_type = Text)]
            amount_wei: String,
            #[diesel(sql_type = Text)]
            credit_kind: String,
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Text)]
            status: String,
        }

        let mut conn = self.pool.get()?;
        let row = diesel::sql_query(
            "WITH inserted AS (
                 INSERT INTO rsm_deposit_credits
                     (tx_hash, log_index, observed_block, account, token, amount_wei, credit_kind, request_id)
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                 ON CONFLICT (tx_hash, log_index) DO NOTHING
                 RETURNING tx_hash, log_index, observed_block, account, token, amount_wei, credit_kind, request_id, status
             )
             SELECT tx_hash, log_index, observed_block, account, token, amount_wei, credit_kind, request_id, status
             FROM inserted
             UNION ALL
             SELECT tx_hash, log_index, observed_block, account, token, amount_wei, credit_kind, request_id, status
             FROM rsm_deposit_credits
             WHERE tx_hash = $1 AND log_index = $2
             LIMIT 1",
        )
        .bind::<Text, _>(&input.tx_hash)
        .bind::<BigInt, _>(input.log_index)
        .bind::<BigInt, _>(input.observed_block)
        .bind::<Binary, _>(&input.account)
        .bind::<Binary, _>(&input.token)
        .bind::<Text, _>(&input.amount_wei)
        .bind::<Text, _>(&input.credit_kind)
        .bind::<Text, _>(&input.request_id)
        .get_result::<ClaimRow>(&mut conn)?;

        Ok(hypercall_db::RsmDepositCreditClaimRecord {
            tx_hash: row.tx_hash,
            log_index: row.log_index,
            observed_block: row.observed_block,
            account: row.account,
            token: row.token,
            amount_wei: row.amount_wei,
            credit_kind: row.credit_kind,
            request_id: row.request_id,
            status: row.status,
        })
    }

    /// Mark an RSM deposit credit as submitted.
    pub async fn mark_rsm_deposit_credit_submitted(&self, request_id: &str) -> Result<()> {
        use diesel::sql_types::Text;

        let mut conn = self.pool.get()?;
        let updated = diesel::sql_query(
            "UPDATE rsm_deposit_credits
             SET status = 'submitted', error = NULL, updated_at = now()
             WHERE request_id = $1",
        )
        .bind::<Text, _>(request_id)
        .execute(&mut conn)?;
        if updated != 1 {
            anyhow::bail!(
                "expected 1 rsm_deposit_credits row updated for request_id {}, got {}",
                request_id,
                updated
            );
        }
        Ok(())
    }

    /// Mark an RSM deposit credit as failed.
    pub async fn mark_rsm_deposit_credit_failed(
        &self,
        request_id: &str,
        error: &str,
    ) -> Result<()> {
        use diesel::sql_types::Text;

        let mut conn = self.pool.get()?;
        let updated = diesel::sql_query(
            "UPDATE rsm_deposit_credits
             SET status = 'failed', error = $2, updated_at = now()
             WHERE request_id = $1",
        )
        .bind::<Text, _>(request_id)
        .bind::<Text, _>(error)
        .execute(&mut conn)?;
        if updated != 1 {
            anyhow::bail!(
                "expected 1 rsm_deposit_credits row failed for request_id {}, got {}",
                request_id,
                updated
            );
        }
        Ok(())
    }

    /// Return the oldest pending Exchange.UsdcDeposit event for a HyperCore cash amount.
    pub async fn pending_rsm_usdc_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Binary)]
            account: Vec<u8>,
            #[diesel(sql_type = Binary)]
            token: Vec<u8>,
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = Text)]
            amount_wei: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT request_id, account, token, tx_hash, log_index, amount_wei
             FROM rsm_deposit_credits
             WHERE credit_kind = 'usdc'
               AND status = 'pending'
               AND amount_wei = $1
             ORDER BY observed_block ASC, log_index ASC
             LIMIT 2",
        )
        .bind::<Text, _>(amount_wei)
        .load::<Row>(&mut conn)?;

        if rows.len() > 1 {
            anyhow::bail!(
                "ambiguous pending Exchange.UsdcDeposit attribution for amount_wei {}",
                amount_wei
            );
        }

        rows.into_iter()
            .next()
            .map(|row| {
                let account = wallet_from_bytes(
                    row.account.as_slice(),
                    &format!("pending USDC deposit {}:{}", row.tx_hash, row.log_index),
                )?;
                let token = wallet_from_bytes(
                    row.token.as_slice(),
                    &format!(
                        "pending USDC deposit token {}:{}",
                        row.tx_hash, row.log_index
                    ),
                )?;
                Ok(hypercall_db::RsmUsdcDepositMatch {
                    request_id: row.request_id,
                    account,
                    tx_hash: row.tx_hash,
                    log_index: row.log_index,
                    amount_wei: row.amount_wei,
                    token,
                })
            })
            .transpose()
    }

    /// Return the pending Exchange.UsdcDeposit event for a CoreWriter writer-action EVM tx hash.
    pub async fn pending_rsm_usdc_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Binary)]
            account: Vec<u8>,
            #[diesel(sql_type = Binary)]
            token: Vec<u8>,
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = Text)]
            amount_wei: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT request_id, account, token, tx_hash, log_index, amount_wei
             FROM rsm_deposit_credits
             WHERE credit_kind = 'usdc'
               AND status = 'pending'
               AND tx_hash = $1
             ORDER BY observed_block ASC, log_index ASC",
        )
        .bind::<Text, _>(evm_tx_hash)
        .load::<Row>(&mut conn)?;

        if rows.is_empty() {
            return Ok(None);
        }

        let amount_matched_rows: Vec<_> = rows
            .into_iter()
            .filter(|row| row.amount_wei == amount_wei)
            .collect();

        if amount_matched_rows.is_empty() {
            anyhow::bail!(
                "CoreWriter evm_tx_hash {} matched pending Exchange.UsdcDeposit rows, but none had amount_wei {}",
                evm_tx_hash,
                amount_wei
            );
        }

        if amount_matched_rows.len() > 1 {
            anyhow::bail!(
                "ambiguous pending Exchange.UsdcDeposit attribution for evm_tx_hash {} amount_wei {}",
                evm_tx_hash,
                amount_wei
            );
        }

        let row = amount_matched_rows
            .into_iter()
            .next()
            .expect("amount_matched_rows is non-empty");
        let account = wallet_from_bytes(
            row.account.as_slice(),
            &format!("pending USDC deposit {}:{}", row.tx_hash, row.log_index),
        )?;
        let token = wallet_from_bytes(
            row.token.as_slice(),
            &format!(
                "pending USDC deposit token {}:{}",
                row.tx_hash, row.log_index
            ),
        )?;
        Ok(Some(hypercall_db::RsmUsdcDepositMatch {
            request_id: row.request_id,
            account,
            tx_hash: row.tx_hash,
            log_index: row.log_index,
            amount_wei: row.amount_wei,
            token,
        }))
    }

    /// Return a PM liquidity Exchange deposit by CoreWriter writer-action EVM tx hash.
    pub async fn pm_liquidity_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Binary)]
            account: Vec<u8>,
            #[diesel(sql_type = Binary)]
            token: Vec<u8>,
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = Text)]
            amount_wei: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT request_id, account, token, tx_hash, log_index, amount_wei
             FROM rsm_deposit_credits
             WHERE credit_kind = 'pm_liquidity'
               AND tx_hash = $1
               AND matched_hypercore_event_hash IS NULL
             ORDER BY observed_block ASC, log_index ASC",
        )
        .bind::<Text, _>(evm_tx_hash)
        .load::<Row>(&mut conn)?;

        if rows.is_empty() {
            return Ok(None);
        }

        let amount_matched_rows: Vec<_> = rows
            .into_iter()
            .filter(|row| row.amount_wei == amount_wei)
            .collect();

        if amount_matched_rows.is_empty() {
            anyhow::bail!(
                "CoreWriter evm_tx_hash {} matched PM liquidity deposit rows, but none had amount_wei {}",
                evm_tx_hash,
                amount_wei
            );
        }

        if amount_matched_rows.len() > 1 {
            anyhow::bail!(
                "ambiguous PM liquidity deposit attribution for evm_tx_hash {} amount_wei {}",
                evm_tx_hash,
                amount_wei
            );
        }

        let row = amount_matched_rows
            .into_iter()
            .next()
            .expect("amount_matched_rows is non-empty");
        rsm_deposit_match_from_row(
            row.request_id,
            row.account,
            row.token,
            row.tx_hash,
            row.log_index,
            row.amount_wei,
            "PM liquidity deposit",
        )
        .map(Some)
    }

    /// Return the oldest PM liquidity Exchange deposit for a HyperCore cash amount.
    pub async fn pm_liquidity_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Binary)]
            account: Vec<u8>,
            #[diesel(sql_type = Binary)]
            token: Vec<u8>,
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = Text)]
            amount_wei: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT request_id, account, token, tx_hash, log_index, amount_wei
             FROM rsm_deposit_credits
             WHERE credit_kind = 'pm_liquidity'
               AND amount_wei = $1
               AND matched_hypercore_event_hash IS NULL
             ORDER BY observed_block ASC, log_index ASC
             LIMIT 2",
        )
        .bind::<Text, _>(amount_wei)
        .load::<Row>(&mut conn)?;

        if rows.len() > 1 {
            anyhow::bail!(
                "ambiguous PM liquidity deposit attribution for amount_wei {}",
                amount_wei
            );
        }

        rows.into_iter()
            .next()
            .map(|row| {
                rsm_deposit_match_from_row(
                    row.request_id,
                    row.account,
                    row.token,
                    row.tx_hash,
                    row.log_index,
                    row.amount_wei,
                    "PM liquidity deposit",
                )
            })
            .transpose()
    }

    /// Mark a PM liquidity Exchange deposit as consumed by a HyperCore cash event.
    pub async fn mark_pm_liquidity_deposit_hypercore_matched(
        &self,
        request_id: &str,
        event_hash: &str,
    ) -> Result<()> {
        use diesel::sql_types::Text;

        let mut conn = self.pool.get()?;
        let updated = diesel::sql_query(
            "UPDATE rsm_deposit_credits
             SET matched_hypercore_event_hash = $2, updated_at = now()
             WHERE request_id = $1
               AND credit_kind = 'pm_liquidity'
               AND (matched_hypercore_event_hash IS NULL OR matched_hypercore_event_hash = $2)",
        )
        .bind::<Text, _>(request_id)
        .bind::<Text, _>(event_hash)
        .execute(&mut conn)?;
        if updated != 1 {
            anyhow::bail!(
                "expected 1 PM liquidity deposit matched for request_id {}, got {}",
                request_id,
                updated
            );
        }
        Ok(())
    }

    /// Return a pending Exchange.UsdcDeposit request for an already credited HyperCore event.
    pub async fn pending_rsm_usdc_deposit_for_credited_hypercore_event(
        &self,
        event_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        use diesel::sql_types::{BigInt, Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            request_id: String,
            #[diesel(sql_type = Binary)]
            account: Vec<u8>,
            #[diesel(sql_type = Binary)]
            token: Vec<u8>,
            #[diesel(sql_type = Text)]
            tx_hash: String,
            #[diesel(sql_type = BigInt)]
            log_index: i64,
            #[diesel(sql_type = Text)]
            amount_wei: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT r.request_id, r.account, r.token, r.tx_hash, r.log_index, r.amount_wei
             FROM hypercore_cash_ledger_events h
             JOIN rsm_deposit_credits r
               ON r.account = h.wallet
              AND r.credit_kind = 'usdc'
              AND r.status = 'pending'
              AND r.amount_wei = $2
             WHERE h.event_hash = $1
               AND h.delta_type = 'pool_transfer'
               AND h.ledger_event_id IS NOT NULL
             ORDER BY r.observed_block ASC, r.log_index ASC
             LIMIT 2",
        )
        .bind::<Text, _>(event_hash)
        .bind::<Text, _>(amount_wei)
        .load::<Row>(&mut conn)?;

        if rows.len() > 1 {
            anyhow::bail!(
                "ambiguous pending Exchange.UsdcDeposit replay attribution for HyperCore event {} amount_wei {}",
                event_hash,
                amount_wei
            );
        }

        rows.into_iter()
            .next()
            .map(|row| {
                let account = wallet_from_bytes(
                    row.account.as_slice(),
                    &format!(
                        "pending replay USDC deposit {}:{}",
                        row.tx_hash, row.log_index
                    ),
                )?;
                let token = wallet_from_bytes(
                    row.token.as_slice(),
                    &format!(
                        "pending replay USDC deposit token {}:{}",
                        row.tx_hash, row.log_index
                    ),
                )?;
                Ok(hypercall_db::RsmUsdcDepositMatch {
                    request_id: row.request_id,
                    account,
                    tx_hash: row.tx_hash,
                    log_index: row.log_index,
                    amount_wei: row.amount_wei,
                    token,
                })
            })
            .transpose()
    }

    /// Return the credited wallet for an already-applied HyperCore cash event.
    pub async fn credited_wallet_for_hypercore_cash_event(
        &self,
        event_hash: &str,
    ) -> Result<Option<hypercall_types::WalletAddress>> {
        use diesel::sql_types::{Binary, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Binary)]
            wallet: Vec<u8>,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT wallet
             FROM hypercore_cash_ledger_events
             WHERE event_hash = $1
               AND delta_type = 'pool_transfer'
               AND ledger_event_id IS NOT NULL
             ORDER BY id ASC
             LIMIT 2",
        )
        .bind::<Text, _>(event_hash)
        .load::<Row>(&mut conn)?;

        if rows.len() > 1 {
            anyhow::bail!(
                "duplicate credited wallet rows for hypercore cash event hash {}",
                event_hash
            );
        }

        rows.into_iter()
            .next()
            .map(|row| wallet_from_bytes(row.wallet.as_slice(), "credited hypercore cash event"))
            .transpose()
    }

    /// Return true when this HyperCore cash event is already recorded as non-crediting.
    pub async fn non_crediting_hypercore_cash_event(
        &self,
        event_hash: &str,
        amount_usdc: rust_decimal::Decimal,
    ) -> Result<bool> {
        use diesel::sql_types::{Bool, Numeric, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Bool)]
            exists: bool,
        }

        let mut conn = self.pool.get()?;
        let row = diesel::sql_query(
            "SELECT EXISTS (
                 SELECT 1
                 FROM hypercore_cash_ledger_events
                 WHERE event_hash = $1
                   AND delta_type = 'pool_transfer'
                   AND amount_usdc = $2
                   AND status = 'submitted'
                   AND ledger_event_id IS NULL
             ) AS exists",
        )
        .bind::<Text, _>(event_hash)
        .bind::<Numeric, _>(amount_usdc)
        .get_result::<Row>(&mut conn)?;
        Ok(row.exists)
    }

    /// Return recent cash deposit attribution rows for the admin panel.
    pub async fn list_recent_cash_deposit_monitoring_rows(
        &self,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<hypercall_db::DepositMonitoringRow>> {
        use diesel::sql_types::{BigInt, Binary, Nullable, Text};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Text)]
            source: String,
            #[diesel(sql_type = Text)]
            status: String,
            #[diesel(sql_type = Text)]
            correlation_status: String,
            #[diesel(sql_type = Binary)]
            wallet: Vec<u8>,
            #[diesel(sql_type = Text)]
            amount_usdc: String,
            #[diesel(sql_type = Text)]
            event_hash: String,
            #[diesel(sql_type = Nullable<Text>)]
            tx_hash: Option<String>,
            #[diesel(sql_type = Nullable<Text>)]
            request_id: Option<String>,
            #[diesel(sql_type = Nullable<BigInt>)]
            observed_block: Option<i64>,
            #[diesel(sql_type = Nullable<BigInt>)]
            log_index: Option<i64>,
            #[diesel(sql_type = Nullable<BigInt>)]
            ledger_event_id: Option<i64>,
            #[diesel(sql_type = Text)]
            created_at: String,
            #[diesel(sql_type = Text)]
            updated_at: String,
        }

        let mut conn = self.pool.get()?;
        let rows = diesel::sql_query(
            "SELECT source, status, correlation_status, wallet, amount_usdc, event_hash, tx_hash, request_id,
                    observed_block, log_index, ledger_event_id, created_at, updated_at
             FROM (
                 SELECT
                     'hypercore'::text AS source,
                     status,
                     'cash_ledger_event'::text AS correlation_status,
                     wallet,
                     amount_usdc::text AS amount_usdc,
                     event_hash,
                     NULL::text AS tx_hash,
                     NULL::text AS request_id,
                     NULL::bigint AS observed_block,
                     NULL::bigint AS log_index,
                     ledger_event_id,
                     created_at::text AS created_at,
                     updated_at::text AS updated_at
                 FROM hypercore_cash_ledger_events
                 WHERE delta_type IN ('deposit', 'pool_transfer')
                 UNION ALL
                 SELECT
                     CASE WHEN credit_kind = 'usdc' THEN 'exchange_usdc' ELSE 'rsm_option' END AS source,
                     status,
                     CASE
                         WHEN credit_kind = 'usdc' AND status = 'submitted' THEN 'writer_action_correlated'
                         WHEN credit_kind = 'usdc' AND status = 'pending' THEN 'pending_writer_action'
                         WHEN credit_kind = 'usdc' AND status = 'failed' THEN 'failed'
                         ELSE 'not_applicable'
                     END AS correlation_status,
                     account AS wallet,
                     CASE
                         WHEN credit_kind = 'usdc'
                         THEN (amount_wei::numeric / 1000000)::text
                         ELSE amount_wei
                     END AS amount_usdc,
                     tx_hash AS event_hash,
                     tx_hash,
                     request_id,
                     observed_block,
                     log_index,
                     NULL::bigint AS ledger_event_id,
                     created_at::text AS created_at,
                     updated_at::text AS updated_at
                 FROM rsm_deposit_credits
             ) deposits
             ORDER BY created_at DESC, event_hash DESC
             LIMIT $1 OFFSET $2",
        )
        .bind::<BigInt, _>(limit)
        .bind::<BigInt, _>(offset)
        .load::<Row>(&mut conn)?;

        rows.into_iter()
            .map(|row| {
                Ok(hypercall_db::DepositMonitoringRow {
                    source: row.source,
                    status: row.status,
                    correlation_status: row.correlation_status,
                    wallet: wallet_from_bytes(row.wallet.as_slice(), "deposit monitoring wallet")?,
                    amount_usdc: row.amount_usdc,
                    event_hash: row.event_hash,
                    tx_hash: row.tx_hash,
                    request_id: row.request_id,
                    observed_block: row.observed_block,
                    log_index: row.log_index,
                    ledger_event_id: row.ledger_event_id,
                    created_at: row.created_at,
                    updated_at: row.updated_at,
                })
            })
            .collect()
    }

    /// Apply a HyperCore cash deposit (idempotent).
    pub async fn apply_hypercore_cash_deposit(
        &self,
        input: &hypercall_db::HypercoreCashLedgerApply,
    ) -> Result<hypercall_db::HypercoreCashLedgerApplyResult> {
        use diesel::sql_types::{BigInt, Binary, Nullable, Numeric, Text};
        use diesel::Connection;
        use diesel::OptionalExtension;
        use rust_decimal::Decimal;

        if input.amount_usdc <= Decimal::ZERO {
            anyhow::bail!(
                "hypercore cash deposit amount must be positive for {} event_hash={}",
                input.wallet,
                input.event_hash
            );
        }

        #[derive(diesel::QueryableByName)]
        struct InsertedEventRow {
            #[diesel(sql_type = BigInt)]
            id: i64,
        }

        #[derive(diesel::QueryableByName)]
        struct LedgerEventRow {
            #[diesel(sql_type = BigInt)]
            id: i64,
        }

        #[derive(diesel::QueryableByName)]
        struct ExistingEventRow {
            #[diesel(sql_type = BigInt)]
            id: i64,
            #[diesel(sql_type = Nullable<BigInt>)]
            ledger_event_id: Option<i64>,
            #[diesel(sql_type = Nullable<Numeric>)]
            balance_after: Option<Decimal>,
        }

        let mut conn = self.pool.get()?;
        let row = conn.transaction::<hypercall_db::HypercoreCashLedgerApplyResult, diesel::result::Error, _>(|conn| {
            let inserted = diesel::sql_query(
                "INSERT INTO hypercore_cash_ledger_events
                    (wallet, event_hash, event_time_ms, delta_type, amount_usdc)
                 VALUES ($1, $2, $3, 'pool_transfer', $4)
                 ON CONFLICT (wallet, event_hash, delta_type) DO NOTHING
                 RETURNING id",
            )
            .bind::<Binary, _>(&input.wallet)
            .bind::<Text, _>(&input.event_hash)
            .bind::<BigInt, _>(input.event_time_ms)
            .bind::<Numeric, _>(input.amount_usdc)
            .get_result::<InsertedEventRow>(conn)
            .optional()?;

            let Some(inserted) = inserted else {
                let existing = diesel::sql_query(
                    "SELECT id, ledger_event_id, balance_after
                     FROM hypercore_cash_ledger_events
                     WHERE wallet = $1 AND event_hash = $2 AND delta_type = 'pool_transfer'
                     FOR UPDATE",
                )
                .bind::<Binary, _>(&input.wallet)
                .bind::<Text, _>(&input.event_hash)
                .get_result::<ExistingEventRow>(conn)?;

                if let Some(ledger_event_id) = existing.ledger_event_id {
                    return Ok(hypercall_db::HypercoreCashLedgerApplyResult {
                        applied: false,
                        balance_after: existing.balance_after,
                        ledger_event_id: Some(
                            u64::try_from(ledger_event_id).unwrap_or_else(|_| {
                                panic!(
                                    "STATE_CORRUPTION: negative ledger_event_id {} for hypercore deposit {}",
                                    ledger_event_id, input.event_hash
                                )
                            }),
                        ),
                    });
                }

                let ledger = diesel::sql_query(
                    "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type)
                     VALUES ($1, $2, $3, 'deposit')
                     RETURNING id",
                )
                .bind::<Binary, _>(&input.wallet)
                .bind::<BigInt, _>(input.event_time_ms)
                .bind::<Numeric, _>(input.amount_usdc)
                .get_result::<LedgerEventRow>(conn)?;

                diesel::sql_query(
                    "UPDATE hypercore_cash_ledger_events
                     SET ledger_event_id = $2,
                         status = 'submitted',
                         error = NULL,
                         updated_at = NOW()
                     WHERE id = $1 AND ledger_event_id IS NULL",
                )
                .bind::<BigInt, _>(existing.id)
                .bind::<BigInt, _>(ledger.id)
                .execute(conn)?;

                return Ok(hypercall_db::HypercoreCashLedgerApplyResult {
                    applied: true,
                    balance_after: None,
                    ledger_event_id: Some(u64::try_from(ledger.id).unwrap_or_else(|_| {
                        panic!(
                            "STATE_CORRUPTION: negative ledger_event_id {} for hypercore deposit {}",
                            ledger.id, input.event_hash
                        )
                    })),
                });
            };

            let ledger = diesel::sql_query(
                "INSERT INTO ledger_events (wallet, event_ts_ms, delta, event_type)
                 VALUES ($1, $2, $3, 'deposit')
                 RETURNING id",
            )
            .bind::<Binary, _>(&input.wallet)
            .bind::<BigInt, _>(input.event_time_ms)
            .bind::<Numeric, _>(input.amount_usdc)
            .get_result::<LedgerEventRow>(conn)?;

            diesel::sql_query(
                "UPDATE hypercore_cash_ledger_events
                 SET ledger_event_id = $2,
                     status = 'submitted',
                     error = NULL,
                     updated_at = NOW()
                 WHERE id = $1",
            )
            .bind::<BigInt, _>(inserted.id)
            .bind::<BigInt, _>(ledger.id)
            .execute(conn)?;

            Ok(hypercall_db::HypercoreCashLedgerApplyResult {
                applied: true,
                balance_after: None,
                ledger_event_id: Some(u64::try_from(ledger.id).unwrap_or_else(|_| {
                    panic!(
                        "STATE_CORRUPTION: negative ledger_event_id {} for hypercore deposit {}",
                        ledger.id, input.event_hash
                    )
                })),
            })
        })?;
        Ok(row)
    }

    /// Record an observed Exchange HyperCore deposit that must not credit engine cash.
    pub async fn record_hypercore_cash_deposit_non_crediting(
        &self,
        input: &hypercall_db::HypercoreCashLedgerApply,
    ) -> Result<()> {
        use diesel::sql_types::{BigInt, Binary, Numeric, Text};
        use rust_decimal::Decimal;

        if input.amount_usdc <= Decimal::ZERO {
            anyhow::bail!(
                "non-crediting hypercore cash deposit amount must be positive for {} event_hash={}",
                input.wallet,
                input.event_hash
            );
        }

        let mut conn = self.pool.get()?;
        diesel::sql_query(
            "INSERT INTO hypercore_cash_ledger_events
                (wallet, event_hash, event_time_ms, delta_type, amount_usdc, status, error)
             VALUES ($1, $2, $3, 'pool_transfer', $4, 'submitted', NULL)
             ON CONFLICT (wallet, event_hash, delta_type) DO UPDATE
             SET status = CASE
                     WHEN hypercore_cash_ledger_events.ledger_event_id IS NULL
                     THEN 'submitted'
                     ELSE hypercore_cash_ledger_events.status
                 END,
                 error = CASE
                     WHEN hypercore_cash_ledger_events.ledger_event_id IS NULL
                     THEN NULL
                     ELSE hypercore_cash_ledger_events.error
                 END,
                 updated_at = NOW()",
        )
        .bind::<Binary, _>(&input.wallet)
        .bind::<Text, _>(&input.event_hash)
        .bind::<BigInt, _>(input.event_time_ms)
        .bind::<Numeric, _>(input.amount_usdc)
        .execute(&mut conn)?;
        Ok(())
    }

    /// Record a HyperCore cash deposit that cannot yet be credited because
    /// margin mode is unknown. The row pins the replay watermark until a later
    /// replay can apply the credit with a durable ledger sequence.
    pub async fn record_hypercore_cash_deposit_pending_margin_mode(
        &self,
        input: &hypercall_db::HypercoreCashLedgerApply,
    ) -> Result<()> {
        use diesel::sql_types::{BigInt, Binary, Numeric, Text};
        use rust_decimal::Decimal;

        if input.amount_usdc <= Decimal::ZERO {
            anyhow::bail!(
                "hypercore cash deposit amount must be positive for {} event_hash={}",
                input.wallet,
                input.event_hash
            );
        }

        let mut conn = self.pool.get()?;
        diesel::sql_query(
            "INSERT INTO hypercore_cash_ledger_events
                (wallet, event_hash, event_time_ms, delta_type, amount_usdc, status, error)
             VALUES ($1, $2, $3, 'pool_transfer', $4, 'pending_margin_mode', 'missing margin mode')
             ON CONFLICT (wallet, event_hash, delta_type) DO UPDATE
             SET status = CASE
                     WHEN hypercore_cash_ledger_events.ledger_event_id IS NULL
                     THEN 'pending_margin_mode'
                     ELSE hypercore_cash_ledger_events.status
                 END,
                 error = CASE
                     WHEN hypercore_cash_ledger_events.ledger_event_id IS NULL
                     THEN 'missing margin mode'
                     ELSE hypercore_cash_ledger_events.error
                 END,
                 updated_at = NOW()",
        )
        .bind::<Binary, _>(&input.wallet)
        .bind::<Text, _>(&input.event_hash)
        .bind::<BigInt, _>(input.event_time_ms)
        .bind::<Numeric, _>(input.amount_usdc)
        .execute(&mut conn)?;
        Ok(())
    }

    /// Get the exchange-wide cash ledger watermark (max event_time_ms).
    pub fn get_exchange_cash_ledger_watermark_sync(&self) -> Result<Option<i64>> {
        use diesel::sql_types::{BigInt, Nullable};

        #[derive(diesel::QueryableByName)]
        struct Row {
            #[diesel(sql_type = Nullable<BigInt>)]
            last_event_time_ms: Option<i64>,
        }

        let mut conn = self.pool.get()?;
        // Only consider deposit/pool_transfer rows for the watermark. Local
        // withdraw reservations can have timestamps newer than the last
        // observed on-chain event and would push the watermark forward,
        // causing the HTTP catch-up to skip older unseen deposits.
        // Pending uncredited rows pin the watermark to their oldest timestamp
        // so a missing-margin-mode deposit stays replayable until credited.
        let row = diesel::sql_query(
            "WITH pending_uncredited AS (
                 SELECT MIN(event_time_ms)::bigint AS event_time_ms
                 FROM hypercore_cash_ledger_events
                 WHERE delta_type IN ('deposit', 'pool_transfer')
                   AND ledger_event_id IS NULL
                   AND status = 'pending_margin_mode'
             ),
             credited AS (
                 SELECT MAX(event_time_ms)::bigint AS event_time_ms
                 FROM hypercore_cash_ledger_events
                 WHERE delta_type IN ('deposit', 'pool_transfer')
                   AND (ledger_event_id IS NOT NULL OR status = 'submitted')
             )
             SELECT COALESCE(
                 (SELECT event_time_ms FROM pending_uncredited),
                 (SELECT event_time_ms FROM credited)
             ) AS last_event_time_ms",
        )
        .get_result::<Row>(&mut conn)?;
        Ok(row.last_event_time_ms)
    }

    /// Load the current nonce state for a signer address (sync, used by startup code).
    pub async fn get_rsm_signer_nonce(
        &self,
        signer: &hypercall_types::WalletAddress,
    ) -> Result<Option<hypercall_db::RsmSignerNonceRecord>> {
        use diesel::OptionalExtension;

        let mut conn = self.pool.get()?;
        let result = crate::schema::rsm_signer_nonces::table
            .filter(crate::schema::rsm_signer_nonces::signer_address.eq(signer))
            .first::<crate::models::RsmSignerNonceRecord>(&mut conn)
            .optional()?;
        Ok(result.map(Into::into))
    }

    /// Look up an option instrument by its on-chain token address.
    pub async fn get_option_instrument_for_credit(
        &self,
        token: &hypercall_types::WalletAddress,
    ) -> Result<Option<hypercall_db::OptionInstrumentForCredit>> {
        use diesel::sql_types::{BigInt, Binary, Numeric, Text};
        use diesel::OptionalExtension;

        #[derive(diesel::QueryableByName)]
        struct InstrumentRow {
            #[diesel(sql_type = Text)]
            id: String,
            #[diesel(sql_type = Text)]
            underlying: String,
            #[diesel(sql_type = BigInt)]
            expiry: i64,
            #[diesel(sql_type = Numeric)]
            strike: rust_decimal::Decimal,
            #[diesel(sql_type = Text)]
            option_type: String,
        }

        let mut conn = self.pool.get()?;
        let row = diesel::sql_query(
            "SELECT id, underlying, expiry::bigint AS expiry, strike, option_type
             FROM instruments
             WHERE option_token_address = $1
             LIMIT 1",
        )
        .bind::<Binary, _>(token)
        .get_result::<InstrumentRow>(&mut conn)
        .optional()?;

        Ok(row.map(|row| hypercall_db::OptionInstrumentForCredit {
            symbol: row.id,
            underlying: row.underlying,
            expiry: row.expiry,
            strike: row.strike,
            option_type: row.option_type,
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn replaces_multi_statement_concurrent_index_migration() {
        let migrations = HypercallMigrations
            .migrations()
            .expect("embedded migrations should load");
        let migration = migrations
            .iter()
            .find(|migration| {
                migration.name().to_string() == ORDER_INFOS_ACTIVE_SYMBOL_INDEX_MIGRATION
            })
            .expect("order_infos active symbol migration should be present");

        assert_eq!(migration.name().version().to_string(), "2026-06-01-000001");
        assert!(!migration.metadata().run_in_transaction());
    }

    #[test]
    fn dyn_connection_manager_calls_closure_on_connect() {
        let call_count = Arc::new(AtomicUsize::new(0));
        let cc = call_count.clone();
        let mgr = DynConnectionManager {
            url_source: Arc::new(move || {
                cc.fetch_add(1, Ordering::SeqCst);
                "postgres://invalid-will-fail/db".to_string()
            }),
        };

        // connect() will fail (no real DB), but the closure should be called
        let _ = r2d2::ManageConnection::connect(&mgr);
        let _ = r2d2::ManageConnection::connect(&mgr);
        let _ = r2d2::ManageConnection::connect(&mgr);
        assert_eq!(call_count.load(Ordering::SeqCst), 3);
    }

    #[test]
    fn dyn_connection_manager_sees_url_changes() {
        let url = Arc::new(std::sync::Mutex::new(
            "postgres://first@host/db".to_string(),
        ));
        let url_clone = url.clone();
        let mgr = DynConnectionManager {
            url_source: Arc::new(move || url_clone.lock().unwrap().clone()),
        };

        // First call sees "first"
        let _ = r2d2::ManageConnection::connect(&mgr);

        // Swap URL
        *url.lock().unwrap() = "postgres://second@host/db".to_string();

        // Second call sees "second" — verifying dynamic resolution
        let _ = r2d2::ManageConnection::connect(&mgr);
    }

    #[test]
    fn dyn_connection_manager_from_password_auth() {
        let auth = DbAuthConfig::password("postgres://u:p@host/db");
        let mgr = DynConnectionManager::new(&auth);
        // Closure resolves the static URL
        let url = (mgr.url_source)();
        assert_eq!(url, "postgres://u:p@host/db");
    }

    #[cfg(feature = "rds-iam")]
    #[test]
    fn dyn_connection_manager_from_iam_auth_sees_rotation() {
        let parsed = crate::rds_iam::ParsedDbUrl::parse("postgres://user@host/db").unwrap();
        let mock_gen = crate::rds_iam::test_support::MockTokenGenerator::new(vec![]);
        let provider =
            crate::rds_iam::RdsIamTokenProvider::with_generator(parsed, Box::new(mock_gen));
        provider.store_url("postgres://user:token_v1@host/db".into());

        let auth = DbAuthConfig::rds_iam_with_provider(provider.clone());
        let mgr = DynConnectionManager::new(&auth);

        assert_eq!((mgr.url_source)(), "postgres://user:token_v1@host/db");

        provider.store_url("postgres://user:token_v2@host/db".into());
        assert_eq!((mgr.url_source)(), "postgres://user:token_v2@host/db");
    }
}