hypercall/

rsm_deposit_credit_observer.rs

use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use alloy::{
    primitives::{Address, B256, U256},
    providers::{DynProvider, Provider, ProviderBuilder},
    rpc::types::{BlockNumberOrTag, Filter, Log},
    sol,
    sol_types::SolEvent,
};
use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use hypercall_engine::command::{EngineCommand, RecordPmVaultDepositCommand};
use hypercall_types::WalletAddress;
use metrics::{counter, gauge};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use sha3::{Digest, Keccak256};
use tokio::sync::{broadcast, mpsc, oneshot};
use tracing::{info, warn};
use uuid::Uuid;

use crate::rsm::unified_engine::OptionDepositRequest;
use crate::shared::order_types::get_timestamp_millis;
use crate::shared::service::{Service, ServiceOwner};
use hypercall_db::{
    OptionInstrumentForCredit, RsmDepositCreditClaimInput, RsmDepositCreditClaimRecord,
};
use hypercall_db_diesel::DatabaseHandler;

const CREDIT_KIND_OPTION: &str = "option";
const CREDIT_KIND_USDC: &str = "usdc";
const CREDIT_KIND_PM_LIQUIDITY: &str = "pm_liquidity";
const RPC_TIMEOUT: Duration = Duration::from_secs(30);

sol! {
    #[sol(rpc)]
    contract Exchange {
        event Deposit(address indexed account, address indexed from, address indexed token, uint256 amount);
        event UsdcDeposit(address indexed account, address indexed from, address indexed token, uint256 amount, uint32 dstDex);
        event PmLiquidityDeposit(address indexed lp, address indexed from, address indexed token, string underlying, uint256 amount, uint32 dstDex);
        function managers(address account) external view returns (address);
    }
}

#[derive(Clone, Debug)]
pub struct RsmDepositCreditObserverConfig {
    pub poll_interval: Duration,
    pub startup_lookback_blocks: u64,
    pub max_blocks_per_poll: u64,
    pub max_blocks_per_get_logs: u64,
    pub confirmation_blocks: u64,
    pub pm_liquidity_settlement_grace_ms: i64,
}

impl RsmDepositCreditObserverConfig {
    pub fn from_runtime_config(
        config: &crate::backend_config::OnchainDepositsRuntimeConfig,
    ) -> Self {
        Self {
            poll_interval: Duration::from_millis(config.poll_interval_ms),
            startup_lookback_blocks: config.startup_lookback_blocks,
            max_blocks_per_poll: config.max_blocks_per_poll,
            max_blocks_per_get_logs: config.max_blocks_per_get_logs,
            confirmation_blocks: config.confirmation_blocks,
            pm_liquidity_settlement_grace_ms: config.pm_liquidity_settlement_grace_ms,
        }
    }
}

#[async_trait]
pub trait DepositCreditStore: Send + Sync + 'static {
    async fn max_observed_block(&self) -> Result<Option<u64>>;
    async fn register_observed_account(&self, account: &WalletAddress) -> Result<()>;
    async fn claim_deposit_credit(
        &self,
        input: &RsmDepositCreditClaimInput,
    ) -> Result<RsmDepositCreditClaimRecord>;
    async fn mark_submitted(&self, request_id: &str) -> Result<()>;
    async fn mark_failed(&self, request_id: &str, error: &str) -> Result<()>;
    async fn option_instrument_for_credit(
        &self,
        token: &WalletAddress,
    ) -> Result<Option<OptionInstrumentForCredit>>;
    async fn max_active_option_expiry_ms(&self, underlying: &str) -> Result<Option<i64>>;
}

#[async_trait]
impl DepositCreditStore for DatabaseHandler {
    async fn max_observed_block(&self) -> Result<Option<u64>> {
        self.get_max_rsm_deposit_credit_observed_block().await
    }

    async fn register_observed_account(&self, account: &WalletAddress) -> Result<()> {
        self.ensure_observed_deposit_account(account).await
    }

    async fn claim_deposit_credit(
        &self,
        input: &RsmDepositCreditClaimInput,
    ) -> Result<RsmDepositCreditClaimRecord> {
        self.claim_rsm_deposit_credit(input).await
    }

    async fn mark_submitted(&self, request_id: &str) -> Result<()> {
        self.mark_rsm_deposit_credit_submitted(request_id).await
    }

    async fn mark_failed(&self, request_id: &str, error: &str) -> Result<()> {
        self.mark_rsm_deposit_credit_failed(request_id, error).await
    }

    async fn option_instrument_for_credit(
        &self,
        token: &WalletAddress,
    ) -> Result<Option<OptionInstrumentForCredit>> {
        self.get_option_instrument_for_credit(token).await
    }

    async fn max_active_option_expiry_ms(&self, underlying: &str) -> Result<Option<i64>> {
        use hypercall_db::InstrumentReader;
        use hypercall_types::api_models::InstrumentStatus;

        let instruments =
            self.get_instruments_by_status_sync(InstrumentStatus::Active.as_db_str())?;
        instruments
            .into_iter()
            .filter(|instrument| instrument.underlying == underlying)
            .map(|instrument| {
                let expiry = u64::try_from(instrument.expiry).with_context(|| {
                    format!(
                        "Invalid negative expiry {} for active instrument {}",
                        instrument.expiry, instrument.id
                    )
                })?;
                let expiry_ts = hypercall_types::expiry_date_to_timestamp_checked(
                    &instrument.underlying,
                    expiry,
                )
                .map_err(|error| {
                    anyhow!(
                        "Invalid expiry {} for active instrument {}: {}",
                        instrument.expiry,
                        instrument.id,
                        error
                    )
                })?;
                let expiry_ms = i64::try_from(expiry_ts)
                    .context("active option expiry timestamp exceeds i64")?
                    .checked_mul(1_000)
                    .ok_or_else(|| anyhow!("active option expiry timestamp ms overflow"))?;
                Ok(expiry_ms)
            })
            .try_fold(None, |max: Option<i64>, expiry_ms: Result<i64>| {
                let expiry_ms = expiry_ms?;
                Ok(Some(
                    max.map_or(expiry_ms, |current| current.max(expiry_ms)),
                ))
            })
    }
}

#[async_trait]
pub trait DepositManagerResolver: Send + Sync + 'static {
    async fn manager_for_account(&self, account: WalletAddress) -> Result<WalletAddress>;
}

#[async_trait]
pub trait DepositCreditApplier: Send + Sync + 'static {
    async fn apply_option_deposit(&self, request: OptionDepositRequest) -> Result<()>;
    async fn apply_pm_liquidity_deposit(&self, command: RecordPmVaultDepositCommand) -> Result<()>;
}

#[async_trait]
impl DepositCreditApplier for mpsc::Sender<OptionDepositRequest> {
    async fn apply_option_deposit(&self, mut request: OptionDepositRequest) -> Result<()> {
        let (tx, rx) = oneshot::channel();
        request.applied_tx = Some(tx);
        self.send(request)
            .await
            .map_err(|_| anyhow!("option deposit receiver closed"))?;
        rx.await
            .map_err(|_| anyhow!("option deposit apply acknowledgement dropped"))?
            .map_err(|err| anyhow!(err))
    }

    async fn apply_pm_liquidity_deposit(
        &self,
        _command: RecordPmVaultDepositCommand,
    ) -> Result<()> {
        anyhow::bail!("PM liquidity deposits require a PM settlement command sender")
    }
}

fn record_deposit_observer_block_gauges(latest_block: u64, finalized_block: u64, next_block: u64) {
    gauge!("ht_rsm_deposit_credit_observer_current_block").set(latest_block as f64);
    gauge!("ht_rsm_deposit_credit_observer_finalized_block").set(finalized_block as f64);
    gauge!("ht_rsm_deposit_credit_observer_next_block").set(next_block as f64);
    gauge!("ht_rsm_deposit_credit_observer_finalized_lag_blocks")
        .set(finalized_block.saturating_sub(next_block) as f64);
    gauge!("ht_rsm_deposit_credit_observer_unfinalized_blocks")
        .set(latest_block.saturating_sub(finalized_block) as f64);
}

pub struct RsmDepositCreditObserver {
    store: Arc<dyn DepositCreditStore>,
    applier: Arc<dyn DepositCreditApplier>,
    provider: DynProvider,
    exchange_address: Address,
    usdc_address: Address,
    config: RsmDepositCreditObserverConfig,
}

impl RsmDepositCreditObserver {
    pub fn new(
        db: Arc<DatabaseHandler>,
        option_deposit_sender: mpsc::Sender<OptionDepositRequest>,
        pm_settlement_sender: mpsc::Sender<hypercall_runtime_api::PmSettlementAdminRequest>,
        rpc_url: &str,
        exchange_contract_address: &str,
        usdc_contract_address: &str,
        config: RsmDepositCreditObserverConfig,
    ) -> Result<Self> {
        let exchange_address = Address::from_str(exchange_contract_address)
            .context("invalid contracts.exchange_contract_address")?;
        let usdc_address =
            Address::from_str(usdc_contract_address).context("invalid contracts.usdc_address")?;
        let provider = ProviderBuilder::new()
            .connect_http(
                rpc_url
                    .parse()
                    .context("invalid transaction_submitter.rpc_url")?,
            )
            .erased();

        Ok(Self {
            store: db,
            applier: Arc::new(DepositCreditChannelApplier {
                option_deposit_sender,
                pm_settlement_sender,
            }),
            provider,
            exchange_address,
            usdc_address,
            config,
        })
    }

    pub async fn run_with_shutdown(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<()> {
        let mut interval = tokio::time::interval(self.config.poll_interval);
        let mut next_block = self.initial_next_block().await?;

        loop {
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("RSM deposit credit observer received shutdown signal");
                    break;
                }
                _ = interval.tick() => {
                    if let Err(error) = self.poll_chain(&mut next_block).await {
                        counter!("ht_rsm_deposit_credit_errors_total", "stage" => "poll")
                            .increment(1);
                        warn!("RSM deposit credit observer poll failed: {}", error);
                    }
                }
            }
        }

        Ok(())
    }

    async fn initial_next_block(&self) -> Result<u64> {
        let latest_block =
            rpc_timeout("get_block_number", self.provider.get_block_number()).await?;
        let finalized_block = latest_block.saturating_sub(self.config.confirmation_blocks);
        let next_block = initial_next_block(
            self.store.max_observed_block().await?,
            finalized_block,
            self.config.startup_lookback_blocks,
        )?;

        info!(
            next_block,
            latest_block,
            finalized_block,
            confirmation_blocks = self.config.confirmation_blocks,
            "RSM deposit credit observer initialized"
        );
        record_deposit_observer_block_gauges(latest_block, finalized_block, next_block);
        Ok(next_block)
    }

    async fn poll_chain(&self, next_block: &mut u64) -> Result<()> {
        let latest_block =
            rpc_timeout("get_block_number", self.provider.get_block_number()).await?;
        let finalized_block = latest_block.saturating_sub(self.config.confirmation_blocks);
        record_deposit_observer_block_gauges(latest_block, finalized_block, *next_block);
        if *next_block > finalized_block {
            gauge!("ht_rsm_deposit_credit_observer_lag_blocks").set(0.0);
            return Ok(());
        }

        let Some((from_block, to_block)) = next_poll_range(
            *next_block,
            finalized_block,
            self.config.max_blocks_per_poll,
        ) else {
            gauge!("ht_rsm_deposit_credit_observer_lag_blocks").set(0.0);
            return Ok(());
        };

        let lag_blocks = latest_block - from_block;
        gauge!("ht_rsm_deposit_credit_observer_lag_blocks").set(lag_blocks as f64);
        if to_block < latest_block {
            warn!(
                next_block = from_block,
                to_block,
                latest_block,
                finalized_block,
                lag_blocks,
                max_blocks_per_poll = self.config.max_blocks_per_poll,
                "RSM deposit credit observer is catching up"
            );
        }

        self.process_logs(from_block, to_block).await?;
        gauge!("ht_rsm_deposit_credit_observer_last_success_block").set(to_block as f64);
        *next_block = to_block
            .checked_add(1)
            .ok_or_else(|| anyhow!("deposit observer next block overflow"))?;
        record_deposit_observer_block_gauges(latest_block, finalized_block, *next_block);
        Ok(())
    }

    async fn process_logs(&self, from_block: u64, to_block: u64) -> Result<()> {
        for (chunk_from, chunk_to) in
            block_chunks_inclusive(from_block, to_block, self.config.max_blocks_per_get_logs)
        {
            let mut logs = self
                .fetch_event_logs(chunk_from, chunk_to, Exchange::Deposit::SIGNATURE_HASH)
                .await?;
            logs.extend(
                self.fetch_event_logs(chunk_from, chunk_to, Exchange::UsdcDeposit::SIGNATURE_HASH)
                    .await?,
            );
            logs.extend(
                self.fetch_event_logs(
                    chunk_from,
                    chunk_to,
                    Exchange::PmLiquidityDeposit::SIGNATURE_HASH,
                )
                .await?,
            );
            counter!("ht_rsm_deposit_credit_get_logs_total").increment(1);
            logs.sort_by_key(log_sort_key);

            for log in logs {
                self.handle_log(log).await?;
            }
        }
        Ok(())
    }

    async fn fetch_event_logs(
        &self,
        from_block: u64,
        to_block: u64,
        signature: B256,
    ) -> Result<Vec<Log>> {
        rpc_timeout(
            "get_logs",
            self.provider.get_logs(
                &Filter::new()
                    .address(self.exchange_address)
                    .from_block(BlockNumberOrTag::Number(from_block))
                    .to_block(BlockNumberOrTag::Number(to_block))
                    .event_signature(signature),
            ),
        )
        .await
        .with_context(|| {
            format!(
                "failed to fetch Exchange deposit logs signature={} from {} to {}",
                signature, from_block, to_block
            )
        })
    }

    async fn handle_log(&self, log: Log) -> Result<()> {
        let signature = log
            .topic0()
            .copied()
            .ok_or_else(|| anyhow!("Exchange deposit log missing topic0"))?;
        if signature == Exchange::UsdcDeposit::SIGNATURE_HASH {
            return self.handle_usdc_deposit_log(log).await;
        }
        if signature == Exchange::PmLiquidityDeposit::SIGNATURE_HASH {
            return self.handle_pm_liquidity_deposit_log(log).await;
        }

        let decoded = log
            .log_decode::<Exchange::Deposit>()
            .inspect_err(|_| {
                counter!("ht_rsm_deposit_credit_errors_total", "stage" => "decode").increment(1);
            })
            .context("failed to decode Exchange.Deposit log")?;
        let event = decoded.data();
        let tx_hash = log
            .transaction_hash
            .ok_or_else(|| anyhow!("Exchange.Deposit missing transaction hash"))?
            .to_string();
        let log_index = i64::try_from(
            log.log_index
                .ok_or_else(|| anyhow!("Exchange.Deposit missing log index"))?,
        )?;
        let observed_block = i64::try_from(
            log.block_number
                .ok_or_else(|| anyhow!("Exchange.Deposit missing block number"))?,
        )?;
        let account = WalletAddress::from(event.account);
        let token = WalletAddress::from(event.token);

        handle_observed_deposit(
            self.store.as_ref(),
            self.applier.as_ref(),
            self,
            ObservedDeposit {
                tx_hash,
                log_index,
                observed_block,
                account,
                token,
                amount: event.amount,
            },
        )
        .await
    }

    async fn handle_usdc_deposit_log(&self, log: Log) -> Result<()> {
        let decoded = log
            .log_decode::<Exchange::UsdcDeposit>()
            .inspect_err(|_| {
                counter!("ht_rsm_deposit_credit_errors_total", "stage" => "decode").increment(1);
            })
            .context("failed to decode Exchange.UsdcDeposit log")?;
        let event = decoded.data();
        if event.dstDex != 0 {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "unsupported_dst_dex")
                .increment(1);
            tracing::warn!(
                dst_dex = event.dstDex,
                tx_hash = ?log.transaction_hash,
                "Exchange.UsdcDeposit has unsupported dstDex {} for tx {:?}",
                event.dstDex,
                log.transaction_hash
            );
            return Ok(());
        }

        let tx_hash = log
            .transaction_hash
            .ok_or_else(|| anyhow!("Exchange.UsdcDeposit missing transaction hash"))?
            .to_string();
        let log_index = i64::try_from(
            log.log_index
                .ok_or_else(|| anyhow!("Exchange.UsdcDeposit missing log index"))?,
        )?;
        let observed_block = i64::try_from(
            log.block_number
                .ok_or_else(|| anyhow!("Exchange.UsdcDeposit missing block number"))?,
        )?;
        let claim_input = RsmDepositCreditClaimInput {
            tx_hash,
            log_index,
            observed_block,
            account: WalletAddress::from(event.account),
            token: WalletAddress::from(event.token),
            amount_wei: event.amount.to_string(),
            credit_kind: CREDIT_KIND_USDC.to_string(),
            request_id: Uuid::now_v7().to_string(),
        };
        let claim = self
            .store
            .claim_deposit_credit(&claim_input)
            .await
            .inspect_err(|_| {
                counter!("ht_rsm_deposit_credit_errors_total", "stage" => "claim").increment(1);
            })?;
        validate_claim(&claim_input, &claim);
        info!(
            account = %claim.account,
            from = %WalletAddress::from(event.from),
            token = %claim.token,
            amount_wei = %claim.amount_wei,
            request_id = %claim.request_id,
            tx_hash = %claim.tx_hash,
            log_index = claim.log_index,
            "Observed Exchange.UsdcDeposit pending HyperCore correlation"
        );
        Ok(())
    }

    async fn handle_pm_liquidity_deposit_log(&self, log: Log) -> Result<()> {
        let decoded = log
            .log_decode::<Exchange::PmLiquidityDeposit>()
            .inspect_err(|_| {
                counter!("ht_rsm_deposit_credit_errors_total", "stage" => "decode").increment(1);
            })
            .context("failed to decode Exchange.PmLiquidityDeposit log")?;
        let event = decoded.data();
        if event.dstDex != 0 {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "unsupported_dst_dex")
                .increment(1);
            warn!(
                dst_dex = event.dstDex,
                tx_hash = ?log.transaction_hash,
                "Exchange.PmLiquidityDeposit has unsupported dstDex"
            );
            return Ok(());
        }
        if event.token != self.usdc_address {
            anyhow::bail!(
                "Exchange.PmLiquidityDeposit token {} did not match configured USDC {}",
                event.token,
                self.usdc_address
            );
        }

        let tx_hash = log
            .transaction_hash
            .ok_or_else(|| anyhow!("Exchange.PmLiquidityDeposit missing transaction hash"))?
            .to_string();
        let log_index_u64 = log
            .log_index
            .ok_or_else(|| anyhow!("Exchange.PmLiquidityDeposit missing log index"))?;
        let log_index = i64::try_from(log_index_u64)?;
        let observed_block = i64::try_from(
            log.block_number
                .ok_or_else(|| anyhow!("Exchange.PmLiquidityDeposit missing block number"))?,
        )?;
        let chain_id = rpc_timeout("get_chain_id", self.provider.get_chain_id()).await?;
        handle_observed_pm_liquidity_deposit(
            self.store.as_ref(),
            self.applier.as_ref(),
            ObservedPmLiquidityDeposit {
                tx_hash,
                log_index,
                observed_block,
                lp: WalletAddress::from(event.lp),
                from: WalletAddress::from(event.from),
                token: WalletAddress::from(event.token),
                underlying: event.underlying.clone(),
                amount: event.amount,
                chain_id,
                exchange_address: self.exchange_address.to_string(),
                log_index_u64,
                settlement_grace_ms: self.config.pm_liquidity_settlement_grace_ms,
            },
        )
        .await
    }
}

async fn handle_observed_pm_liquidity_deposit(
    store: &dyn DepositCreditStore,
    applier: &dyn DepositCreditApplier,
    deposit: ObservedPmLiquidityDeposit,
) -> Result<()> {
    let request_id = Uuid::now_v7();
    let claim_input = RsmDepositCreditClaimInput {
        tx_hash: deposit.tx_hash.clone(),
        log_index: deposit.log_index,
        observed_block: deposit.observed_block,
        account: deposit.lp,
        token: deposit.token,
        amount_wei: deposit.amount.to_string(),
        credit_kind: CREDIT_KIND_PM_LIQUIDITY.to_string(),
        request_id: request_id.to_string(),
    };
    let claim = store
        .claim_deposit_credit(&claim_input)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "claim").increment(1);
        })?;
    validate_claim(&claim_input, &claim);
    if claim.status == "submitted" {
        return Ok(());
    }

    let max_listed_expiry_ts_ms = match store
        .max_active_option_expiry_ms(&deposit.underlying)
        .await?
    {
        Some(expiry) => expiry,
        None => {
            let error = format!(
                "PM liquidity deposit has no active listed options for underlying {}",
                deposit.underlying
            );
            store.mark_failed(&claim.request_id, &error).await?;
            warn!(
                request_id = %claim.request_id,
                tx_hash = %claim.tx_hash,
                log_index = claim.log_index,
                underlying = %deposit.underlying,
                "Quarantined invalid PM liquidity deposit"
            );
            return Ok(());
        }
    };
    let amount_usdc = usdc_token_amount_to_decimal(deposit.amount)?;
    let claim_request_id = Uuid::parse_str(&claim.request_id)
        .context("rsm_deposit_credits PM liquidity request_id must be a UUID")?;
    let input_digest = pm_liquidity_input_digest(
        deposit.chain_id,
        &deposit.exchange_address,
        &deposit.tx_hash,
        deposit.log_index_u64,
        &claim.account.to_string(),
        &deposit.underlying,
        &claim.amount_wei,
        max_listed_expiry_ts_ms,
        deposit.settlement_grace_ms,
    );
    applier
        .apply_pm_liquidity_deposit(RecordPmVaultDepositCommand {
            request_id: claim_request_id,
            input_digest,
            depositor: claim.account,
            underlying: deposit.underlying.clone(),
            amount_usdc,
            chain_id: deposit.chain_id,
            source_contract_address: WalletAddress::from_str(&deposit.exchange_address)
                .context("PM liquidity source contract address must be a valid EVM address")?,
            tx_hash: claim.tx_hash.clone(),
            log_index: u32::try_from(deposit.log_index_u64)
                .context("PM liquidity deposit log index exceeds u32")?,
            max_listed_expiry_ts_ms,
            settlement_grace_ms: deposit.settlement_grace_ms,
            timestamp_ms: get_timestamp_millis(),
        })
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "pm_engine_apply")
                .increment(1);
        })?;
    store
        .mark_submitted(&claim.request_id)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "mark_submitted")
                .increment(1);
        })?;
    counter!("ht_pm_liquidity_deposits_submitted_total").increment(1);
    info!(
        lp = %claim.account,
        from = %deposit.from,
        underlying = %deposit.underlying,
        amount_usdc = %amount_usdc,
        request_id = %claim.request_id,
        tx_hash = %claim.tx_hash,
        log_index = claim.log_index,
        max_listed_expiry_ts_ms,
        "Applied PM liquidity deposit for Exchange.PmLiquidityDeposit"
    );
    Ok(())
}

#[derive(Debug, Clone)]
struct ObservedPmLiquidityDeposit {
    tx_hash: String,
    log_index: i64,
    observed_block: i64,
    lp: WalletAddress,
    from: WalletAddress,
    token: WalletAddress,
    underlying: String,
    amount: U256,
    chain_id: u64,
    exchange_address: String,
    log_index_u64: u64,
    settlement_grace_ms: i64,
}

struct DepositCreditChannelApplier {
    option_deposit_sender: mpsc::Sender<OptionDepositRequest>,
    pm_settlement_sender: mpsc::Sender<hypercall_runtime_api::PmSettlementAdminRequest>,
}

#[async_trait]
impl DepositCreditApplier for DepositCreditChannelApplier {
    async fn apply_option_deposit(&self, request: OptionDepositRequest) -> Result<()> {
        self.option_deposit_sender
            .apply_option_deposit(request)
            .await
    }

    async fn apply_pm_liquidity_deposit(&self, command: RecordPmVaultDepositCommand) -> Result<()> {
        let (tx, rx) = oneshot::channel();
        self.pm_settlement_sender
            .send(hypercall_runtime_api::PmSettlementAdminRequest {
                command: EngineCommand::RecordPmVaultDeposit(command),
                applied_tx: tx,
            })
            .await
            .map_err(|_| anyhow!("PM settlement command receiver closed"))?;
        rx.await
            .map_err(|_| anyhow!("PM settlement command acknowledgement dropped"))?
            .map_err(|err| anyhow!(err))
    }
}

#[async_trait]
impl DepositManagerResolver for RsmDepositCreditObserver {
    async fn manager_for_account(&self, account: WalletAddress) -> Result<WalletAddress> {
        let exchange = Exchange::new(self.exchange_address, &self.provider);
        let manager = rpc_timeout("Exchange.managers", async {
            exchange.managers(account.inner()).call().await
        })
        .await
        .with_context(|| format!("failed to resolve Exchange manager for {account}"))?;
        if manager == Address::ZERO {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "manager_lookup")
                .increment(1);
            anyhow::bail!("Exchange.managers({account}) returned zero address");
        }
        Ok(WalletAddress::from(manager))
    }
}

async fn handle_observed_deposit(
    store: &dyn DepositCreditStore,
    applier: &dyn DepositCreditApplier,
    manager_resolver: &dyn DepositManagerResolver,
    deposit: ObservedDeposit,
) -> Result<()> {
    let manager = manager_resolver
        .manager_for_account(deposit.account)
        .await?;
    store
        .register_observed_account(&manager)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "register_account")
                .increment(1);
        })?;

    let amount_wei = deposit.amount.to_string();
    let claim_input = RsmDepositCreditClaimInput {
        tx_hash: deposit.tx_hash,
        log_index: deposit.log_index,
        observed_block: deposit.observed_block,
        account: deposit.account,
        token: deposit.token,
        amount_wei,
        credit_kind: CREDIT_KIND_OPTION.to_string(),
        request_id: Uuid::now_v7().to_string(),
    };
    let claim = store
        .claim_deposit_credit(&claim_input)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "claim").increment(1);
        })?;
    validate_claim(&claim_input, &claim);
    if claim.status == "submitted" {
        return Ok(());
    }

    let instrument = match store
        .option_instrument_for_credit(&deposit.token)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "instrument_lookup")
                .increment(1);
        })? {
        Some(instrument) => instrument,
        None => {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "unsupported_token")
                .increment(1);
            anyhow::bail!(
                "deposit token has no option instrument mapping: account={}, token={}, tx_hash={}, log_index={}",
                claim_input.account,
                claim_input.token,
                claim_input.tx_hash,
                claim_input.log_index
            );
        }
    };
    let quantity = option_token_amount_to_quantity(deposit.amount)?;
    applier
        .apply_option_deposit(OptionDepositRequest {
            request_id: claim.request_id.clone(),
            wallet: manager,
            symbol: instrument.symbol.clone(),
            quantity,
            timestamp_ms: get_timestamp_millis(),
            applied_tx: None,
        })
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "engine_apply").increment(1);
        })?;
    store
        .mark_submitted(&claim.request_id)
        .await
        .inspect_err(|_| {
            counter!("ht_rsm_deposit_credit_errors_total", "stage" => "mark_submitted")
                .increment(1);
        })?;
    counter!("ht_rsm_deposit_credit_submitted_total").increment(1);
    info!(
        account = %deposit.account,
        manager = %manager,
        token = %deposit.token,
        symbol = %instrument.symbol,
        quantity = %quantity,
        request_id = %claim.request_id,
        tx_hash = %claim.tx_hash,
        log_index = claim.log_index,
        "Applied option deposit credit for Exchange.Deposit"
    );
    Ok(())
}

#[derive(Debug, Clone)]
struct ObservedDeposit {
    tx_hash: String,
    log_index: i64,
    observed_block: i64,
    account: WalletAddress,
    token: WalletAddress,
    amount: U256,
}

#[async_trait::async_trait]
impl Service for RsmDepositCreditObserver {
    fn name(&self) -> &'static str {
        "RsmDepositCreditObserver"
    }

    fn owner(&self) -> ServiceOwner {
        ServiceOwner::Shared
    }

    async fn run(self: Arc<Self>, shutdown: crate::shared::shutdown::ShutdownRx) -> Result<()> {
        self.run_with_shutdown(shutdown).await
    }
}

fn option_token_amount_to_quantity(amount: U256) -> Result<Decimal> {
    let raw = Decimal::from_str(&amount.to_string())
        .with_context(|| format!("failed to parse option token amount {}", amount))?;
    let quantity = raw / dec!(1000000);
    if quantity <= Decimal::ZERO {
        anyhow::bail!(
            "option deposit amount {} converts to non-positive quantity",
            amount
        );
    }
    Ok(quantity)
}

fn usdc_token_amount_to_decimal(amount: U256) -> Result<Decimal> {
    let raw = Decimal::from_str(&amount.to_string())
        .with_context(|| format!("failed to parse USDC token amount {}", amount))?;
    let amount_usdc = raw / dec!(1000000);
    if amount_usdc <= Decimal::ZERO {
        anyhow::bail!(
            "PM liquidity deposit amount {} converts to non-positive USDC",
            amount
        );
    }
    Ok(amount_usdc)
}

fn pm_liquidity_input_digest(
    chain_id: u64,
    exchange_address: &str,
    tx_hash: &str,
    log_index: u64,
    lp: &str,
    underlying: &str,
    amount_wei: &str,
    max_listed_expiry_ts_ms: i64,
    settlement_grace_ms: i64,
) -> String {
    let mut hasher = Keccak256::new();
    for part in [
        "pm-liquidity-v1",
        &chain_id.to_string(),
        exchange_address,
        tx_hash,
        &log_index.to_string(),
        lp,
        underlying,
        amount_wei,
        &max_listed_expiry_ts_ms.to_string(),
        &settlement_grace_ms.to_string(),
    ] {
        hasher.update((part.len() as u64).to_le_bytes());
        hasher.update(part.as_bytes());
    }
    format!("0x{}", hex::encode(hasher.finalize()))
}

fn validate_claim(input: &RsmDepositCreditClaimInput, claim: &RsmDepositCreditClaimRecord) {
    if input.tx_hash != claim.tx_hash
        || input.log_index != claim.log_index
        || input.account != claim.account
        || input.token != claim.token
        || input.amount_wei != claim.amount_wei
        || input.credit_kind != claim.credit_kind
    {
        panic!(
            "STATE_CORRUPTION: rsm_deposit_credits claim mismatch for tx_hash={} log_index={}",
            input.tx_hash, input.log_index
        );
    }
}

fn log_sort_key(log: &Log) -> (u64, u64, u64) {
    (
        log.block_number.unwrap_or_default(),
        log.transaction_index.unwrap_or_default(),
        log.log_index.unwrap_or_default(),
    )
}

fn initial_next_block(
    max_observed_block: Option<u64>,
    latest_block: u64,
    startup_lookback_blocks: u64,
) -> Result<u64> {
    if startup_lookback_blocks == 0 {
        return Err(anyhow!("startup_lookback_blocks must be greater than zero"));
    }

    if let Some(max_observed_block) = max_observed_block {
        // Resume from the last observed block, not the next block. Blocks can
        // contain multiple deposit logs, and the claim table's tx_hash/log_index
        // uniqueness makes replaying the boundary block idempotent.
        return Ok(max_observed_block);
    }

    // First boot must not scan from genesis. The bounded lookback catches recent
    // deposits that landed before this process started without turning startup
    // into an unbounded historical backfill.
    Ok(latest_block.saturating_sub(startup_lookback_blocks.saturating_sub(1)))
}

async fn rpc_timeout<T>(
    operation: &'static str,
    future: impl std::future::Future<Output = Result<T, impl std::fmt::Display>>,
) -> Result<T> {
    tokio::time::timeout(RPC_TIMEOUT, future)
        .await
        .map_err(|_| anyhow!("{} timed out after {:?}", operation, RPC_TIMEOUT))?
        .map_err(|error| anyhow!("{} failed: {}", operation, error))
}

fn next_poll_range(
    next_block: u64,
    latest_block: u64,
    max_blocks_per_poll: u64,
) -> Option<(u64, u64)> {
    if next_block > latest_block {
        return None;
    }

    let to_block = next_block
        .saturating_add(max_blocks_per_poll.saturating_sub(1))
        .min(latest_block);
    Some((next_block, to_block))
}

fn block_chunks_inclusive(
    from_block: u64,
    to_block: u64,
    max_blocks_per_get_logs: u64,
) -> Vec<(u64, u64)> {
    assert!(
        max_blocks_per_get_logs > 0,
        "max_blocks_per_get_logs must be greater than zero"
    );
    if from_block > to_block {
        return Vec::new();
    }

    let mut chunks = Vec::new();
    let mut chunk_from = from_block;
    while chunk_from <= to_block {
        // RPC providers cap eth_getLogs spans. Keep the provider request range
        // separate from the per-poll catchup range so a tick makes bounded
        // progress while still respecting tighter upstream limits.
        let chunk_to = chunk_from
            .saturating_add(max_blocks_per_get_logs.saturating_sub(1))
            .min(to_block);
        chunks.push((chunk_from, chunk_to));
        if chunk_to == u64::MAX {
            break;
        }
        chunk_from = chunk_to + 1;
    }
    chunks
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    use rust_decimal_macros::dec;

    struct MockStore {
        claim_status: String,
        instrument: Option<OptionInstrumentForCredit>,
        max_active_option_expiry_ms: Option<i64>,
        registered_accounts: Mutex<Vec<WalletAddress>>,
        claims: Mutex<Vec<RsmDepositCreditClaimInput>>,
        submitted: Mutex<Vec<String>>,
        failed: Mutex<Vec<(String, String)>>,
    }

    impl MockStore {
        fn new(instrument: Option<OptionInstrumentForCredit>) -> Self {
            Self {
                claim_status: "pending".to_string(),
                instrument,
                max_active_option_expiry_ms: Some(1_800_000_000_000),
                registered_accounts: Mutex::new(Vec::new()),
                claims: Mutex::new(Vec::new()),
                submitted: Mutex::new(Vec::new()),
                failed: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait::async_trait]
    impl DepositCreditStore for MockStore {
        async fn max_observed_block(&self) -> Result<Option<u64>> {
            Ok(None)
        }

        async fn register_observed_account(&self, account: &WalletAddress) -> Result<()> {
            self.registered_accounts.lock().unwrap().push(*account);
            Ok(())
        }

        async fn claim_deposit_credit(
            &self,
            input: &RsmDepositCreditClaimInput,
        ) -> Result<RsmDepositCreditClaimRecord> {
            self.claims.lock().unwrap().push(input.clone());
            Ok(RsmDepositCreditClaimRecord {
                tx_hash: input.tx_hash.clone(),
                log_index: input.log_index,
                observed_block: input.observed_block,
                account: input.account,
                token: input.token,
                amount_wei: input.amount_wei.clone(),
                credit_kind: input.credit_kind.clone(),
                request_id: input.request_id.clone(),
                status: self.claim_status.clone(),
            })
        }

        async fn mark_submitted(&self, request_id: &str) -> Result<()> {
            self.submitted.lock().unwrap().push(request_id.to_string());
            Ok(())
        }

        async fn mark_failed(&self, request_id: &str, error: &str) -> Result<()> {
            self.failed
                .lock()
                .unwrap()
                .push((request_id.to_string(), error.to_string()));
            Ok(())
        }

        async fn option_instrument_for_credit(
            &self,
            _token: &WalletAddress,
        ) -> Result<Option<OptionInstrumentForCredit>> {
            Ok(self.instrument.clone())
        }

        async fn max_active_option_expiry_ms(&self, _underlying: &str) -> Result<Option<i64>> {
            Ok(self.max_active_option_expiry_ms)
        }
    }

    #[derive(Default)]
    struct MockApplier {
        applied: Mutex<Vec<OptionDepositRequest>>,
        pm_applied: Mutex<Vec<RecordPmVaultDepositCommand>>,
    }

    struct MockManagerResolver {
        manager: WalletAddress,
    }

    #[async_trait::async_trait]
    impl DepositManagerResolver for MockManagerResolver {
        async fn manager_for_account(&self, _account: WalletAddress) -> Result<WalletAddress> {
            Ok(self.manager)
        }
    }

    #[async_trait::async_trait]
    impl DepositCreditApplier for MockApplier {
        async fn apply_option_deposit(&self, request: OptionDepositRequest) -> Result<()> {
            self.applied.lock().unwrap().push(request);
            Ok(())
        }

        async fn apply_pm_liquidity_deposit(
            &self,
            command: RecordPmVaultDepositCommand,
        ) -> Result<()> {
            self.pm_applied.lock().unwrap().push(command);
            Ok(())
        }
    }

    fn btc_call_instrument() -> OptionInstrumentForCredit {
        OptionInstrumentForCredit {
            symbol: "BTC-20260130-100000-C".to_string(),
            underlying: "BTC".to_string(),
            expiry: 20260130,
            strike: dec!(100000),
            option_type: "call".to_string(),
        }
    }

    #[test]
    fn option_token_amount_to_quantity_uses_token_decimals() {
        assert_eq!(
            option_token_amount_to_quantity(U256::from(1_500_000)).unwrap(),
            dec!(1.5)
        );
    }

    #[test]
    #[should_panic(expected = "STATE_CORRUPTION")]
    fn validate_claim_panics_on_reused_log_with_different_payload() {
        let account = WalletAddress::from(Address::repeat_byte(1));
        let token = WalletAddress::from(Address::repeat_byte(2));
        let input = RsmDepositCreditClaimInput {
            tx_hash: "0xabc".to_string(),
            log_index: 1,
            observed_block: 10,
            account,
            token,
            amount_wei: "1".to_string(),
            credit_kind: "option".to_string(),
            request_id: "018f0000-0000-7000-8000-000000000001".to_string(),
        };
        let claim = RsmDepositCreditClaimRecord {
            tx_hash: input.tx_hash.clone(),
            log_index: input.log_index,
            observed_block: input.observed_block,
            account,
            token,
            amount_wei: "2".to_string(),
            credit_kind: input.credit_kind.clone(),
            request_id: input.request_id.clone(),
            status: "pending".to_string(),
        };

        validate_claim(&input, &claim);
    }

    #[test]
    fn initial_next_block_replays_watermark_block_after_restart() {
        let next = initial_next_block(Some(42), 100, 32).expect("watermark should resume");

        assert_eq!(next, 42);
    }

    #[test]
    fn initial_next_block_uses_bounded_startup_lookback() {
        let next = initial_next_block(None, 100, 32).expect("lookback should compute");

        assert_eq!(next, 69);
    }

    #[test]
    fn next_poll_range_caps_catchup_per_tick() {
        assert_eq!(next_poll_range(10, 10_000, 500), Some((10, 509)));
        assert_eq!(next_poll_range(10_001, 10_000, 500), None);
    }

    #[test]
    fn block_chunks_inclusive_splits_get_logs_ranges() {
        let chunks = block_chunks_inclusive(10, 25, 7);

        assert_eq!(chunks, vec![(10, 16), (17, 23), (24, 25)]);
    }

    #[tokio::test]
    async fn handle_observed_deposit_claims_applies_and_marks_submitted() {
        let account = WalletAddress::from(Address::repeat_byte(1));
        let manager = WalletAddress::from(Address::repeat_byte(3));
        let token = WalletAddress::from(Address::repeat_byte(2));
        let store = MockStore::new(Some(btc_call_instrument()));
        let applier = MockApplier::default();
        let manager_resolver = MockManagerResolver { manager };

        handle_observed_deposit(
            &store,
            &applier,
            &manager_resolver,
            ObservedDeposit {
                tx_hash: "0xabc".to_string(),
                log_index: 7,
                observed_block: 100,
                account,
                token,
                amount: U256::from(1_000_000),
            },
        )
        .await
        .expect("deposit should be published");

        assert_eq!(
            store.registered_accounts.lock().unwrap().as_slice(),
            &[manager]
        );

        let claims = store.claims.lock().unwrap();
        assert_eq!(claims.len(), 1);
        assert_eq!(claims[0].tx_hash, "0xabc");
        assert_eq!(claims[0].log_index, 7);
        assert_eq!(claims[0].observed_block, 100);
        assert_eq!(claims[0].account, account);
        assert_eq!(claims[0].token, token);
        assert_eq!(claims[0].amount_wei, "1000000");
        assert_eq!(claims[0].credit_kind, CREDIT_KIND_OPTION);

        let applied = applier.applied.lock().unwrap();
        assert_eq!(applied.len(), 1);
        assert_eq!(applied[0].request_id, claims[0].request_id);
        assert_eq!(applied[0].wallet, manager);
        assert_eq!(applied[0].symbol, "BTC-20260130-100000-C");
        assert_eq!(applied[0].quantity, dec!(1));

        assert_eq!(
            store.submitted.lock().unwrap().as_slice(),
            &[claims[0].request_id.clone()]
        );
    }

    #[tokio::test]
    async fn handle_observed_pm_liquidity_deposit_claims_applies_and_marks_submitted() {
        let lp = WalletAddress::from(Address::repeat_byte(11));
        let payer = WalletAddress::from(Address::repeat_byte(12));
        let token = WalletAddress::from(Address::repeat_byte(13));
        let store = MockStore {
            claim_status: "pending".to_string(),
            instrument: None,
            max_active_option_expiry_ms: Some(1_800_000_000_000),
            registered_accounts: Mutex::new(Vec::new()),
            claims: Mutex::new(Vec::new()),
            submitted: Mutex::new(Vec::new()),
            failed: Mutex::new(Vec::new()),
        };
        let applier = MockApplier::default();

        handle_observed_pm_liquidity_deposit(
            &store,
            &applier,
            ObservedPmLiquidityDeposit {
                tx_hash: "0xpm".to_string(),
                log_index: 9,
                observed_block: 100,
                lp,
                from: payer,
                token,
                underlying: "BTC".to_string(),
                amount: U256::from(50_000_000_000_u64),
                chain_id: 999,
                exchange_address: "0x1111111111111111111111111111111111111111".to_string(),
                log_index_u64: 9,
                settlement_grace_ms: 86_400_000,
            },
        )
        .await
        .expect("PM liquidity deposit should be journaled");

        let claims = store.claims.lock().unwrap();
        assert_eq!(claims.len(), 1);
        assert_eq!(claims[0].tx_hash, "0xpm");
        assert_eq!(claims[0].log_index, 9);
        assert_eq!(claims[0].account, lp);
        assert_eq!(claims[0].token, token);
        assert_eq!(claims[0].amount_wei, "50000000000");
        assert_eq!(claims[0].credit_kind, CREDIT_KIND_PM_LIQUIDITY);

        let applied = applier.pm_applied.lock().unwrap();
        assert_eq!(applied.len(), 1);
        assert_eq!(applied[0].request_id.to_string(), claims[0].request_id);
        assert_eq!(applied[0].depositor, lp);
        assert_eq!(applied[0].underlying, "BTC");
        assert_eq!(applied[0].amount_usdc, dec!(50000));
        assert_eq!(applied[0].chain_id, 999);
        assert_eq!(
            applied[0].source_contract_address,
            WalletAddress::from_str("0x1111111111111111111111111111111111111111").unwrap()
        );
        assert_eq!(applied[0].tx_hash, "0xpm");
        assert_eq!(applied[0].log_index, 9);
        assert_eq!(applied[0].max_listed_expiry_ts_ms, 1_800_000_000_000);
        assert_eq!(applied[0].settlement_grace_ms, 86_400_000);
        assert!(applied[0].input_digest.starts_with("0x"));

        assert_eq!(
            store.submitted.lock().unwrap().as_slice(),
            &[claims[0].request_id.clone()]
        );
    }

    #[tokio::test]
    async fn handle_observed_pm_liquidity_deposit_quarantines_unknown_underlying() {
        let lp = WalletAddress::from(Address::repeat_byte(11));
        let payer = WalletAddress::from(Address::repeat_byte(12));
        let token = WalletAddress::from(Address::repeat_byte(13));
        let store = MockStore {
            claim_status: "pending".to_string(),
            instrument: None,
            max_active_option_expiry_ms: None,
            registered_accounts: Mutex::new(Vec::new()),
            claims: Mutex::new(Vec::new()),
            submitted: Mutex::new(Vec::new()),
            failed: Mutex::new(Vec::new()),
        };
        let applier = MockApplier::default();

        handle_observed_pm_liquidity_deposit(
            &store,
            &applier,
            ObservedPmLiquidityDeposit {
                tx_hash: "0xpm".to_string(),
                log_index: 9,
                observed_block: 100,
                lp,
                from: payer,
                token,
                underlying: "NOT_LISTED".to_string(),
                amount: U256::from(50_000_000_000_u64),
                chain_id: 999,
                exchange_address: "0x1111111111111111111111111111111111111111".to_string(),
                log_index_u64: 9,
                settlement_grace_ms: 86_400_000,
            },
        )
        .await
        .expect("invalid PM liquidity deposit should be quarantined, not retried forever");

        let claims = store.claims.lock().unwrap();
        assert_eq!(claims.len(), 1);
        assert!(applier.pm_applied.lock().unwrap().is_empty());
        assert!(store.submitted.lock().unwrap().is_empty());
        let failed = store.failed.lock().unwrap();
        assert_eq!(failed.len(), 1);
        assert_eq!(failed[0].0, claims[0].request_id);
        assert!(
            failed[0].1.contains("no active listed options"),
            "unexpected failure reason: {}",
            failed[0].1
        );
    }

    #[tokio::test]
    async fn handle_observed_deposit_claims_then_errors_on_unsupported_token() {
        let account = WalletAddress::from(Address::repeat_byte(1));
        let manager = WalletAddress::from(Address::repeat_byte(3));
        let store = MockStore::new(None);
        let applier = MockApplier::default();
        let manager_resolver = MockManagerResolver { manager };

        let result = handle_observed_deposit(
            &store,
            &applier,
            &manager_resolver,
            ObservedDeposit {
                tx_hash: "0xabc".to_string(),
                log_index: 7,
                observed_block: 100,
                account,
                token: WalletAddress::from(Address::repeat_byte(2)),
                amount: U256::from(1_000_000),
            },
        )
        .await;

        let error = result.expect_err("unsupported token should fail the observer path");
        assert!(
            error.to_string().contains("no option instrument mapping"),
            "unexpected error: {error:#}"
        );
        assert_eq!(
            store.registered_accounts.lock().unwrap().as_slice(),
            &[manager]
        );
        let claims = store.claims.lock().unwrap();
        assert_eq!(claims.len(), 1);
        assert_eq!(claims[0].tx_hash, "0xabc");
        assert_eq!(claims[0].log_index, 7);
        assert!(applier.applied.lock().unwrap().is_empty());
        assert!(store.submitted.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn handle_observed_deposit_skips_already_submitted_claim() {
        let store = MockStore {
            claim_status: "submitted".to_string(),
            instrument: None,
            max_active_option_expiry_ms: Some(1_800_000_000_000),
            registered_accounts: Mutex::new(Vec::new()),
            claims: Mutex::new(Vec::new()),
            submitted: Mutex::new(Vec::new()),
            failed: Mutex::new(Vec::new()),
        };
        let applier = MockApplier::default();
        let manager_resolver = MockManagerResolver {
            manager: WalletAddress::from(Address::repeat_byte(3)),
        };

        handle_observed_deposit(
            &store,
            &applier,
            &manager_resolver,
            ObservedDeposit {
                tx_hash: "0xabc".to_string(),
                log_index: 7,
                observed_block: 100,
                account: WalletAddress::from(Address::repeat_byte(1)),
                token: WalletAddress::from(Address::repeat_byte(2)),
                amount: U256::from(1_000_000),
            },
        )
        .await
        .expect("submitted claim should be idempotent");

        assert!(applier.applied.lock().unwrap().is_empty());
        assert!(store.submitted.lock().unwrap().is_empty());
    }
}