hypercall/

hypercore_cash_ledger_observer.rs

use std::sync::Arc;
use std::time::{Duration, Instant};

use alloy::primitives::FixedBytes;
use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use futures::{SinkExt, StreamExt};
use hypercall_types::WalletAddress;
use hypercore_rs::cash_ledger::{HyperliquidLedgerRequest, LedgerUpdate, WsLedgerEnvelope};
use metrics::{counter, gauge};
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use tokio::sync::{broadcast, mpsc, oneshot};
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, error, info, warn};

use crate::read_cache::tier::TierCache;
use crate::rsm::unified_engine::DepositRequest;
use crate::shared::order_types::get_timestamp_millis;
use crate::shared::service::{Service, ServiceOwner};
use hypercall_db::HypercoreCashLedgerApply;
use hypercall_db_diesel::DatabaseHandler;

const DEFAULT_HL_INFO_URL: &str = "https://api.hyperliquid.xyz/info";
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const STARTUP_LOOKBACK_MS: i64 = 24 * 60 * 60 * 1000;
const REPLAY_OVERLAP_MS: i64 = 5 * 60 * 1000;
const WS_READ_TIMEOUT: Duration = Duration::from_secs(60);
const WS_BASE_BACKOFF: Duration = Duration::from_secs(1);
const WS_MAX_BACKOFF: Duration = Duration::from_secs(30);

#[derive(Clone, Debug)]
pub struct HypercoreCashLedgerObserverConfig {
    pub poll_interval: Duration,
    pub info_url: String,
    pub exchange_address: WalletAddress,
    pub core_deposit_wallet_address: WalletAddress,
    pub ws_url: Option<String>,
}

impl HypercoreCashLedgerObserverConfig {
    pub fn from_runtime_config(
        config: &crate::backend_config::OnchainDepositsRuntimeConfig,
        info_url: impl Into<String>,
        exchange_address: WalletAddress,
        core_deposit_wallet_address: WalletAddress,
    ) -> Self {
        Self {
            poll_interval: Duration::from_millis(config.poll_interval_ms),
            info_url: info_url.into(),
            exchange_address,
            core_deposit_wallet_address,
            ws_url: config.ws_url.clone(),
        }
    }
}

impl Default for HypercoreCashLedgerObserverConfig {
    fn default() -> Self {
        Self {
            poll_interval: Duration::from_millis(
                crate::backend_config::OnchainDepositsRuntimeConfig::default().poll_interval_ms,
            ),
            info_url: DEFAULT_HL_INFO_URL.to_string(),
            exchange_address: WalletAddress::default(),
            core_deposit_wallet_address: WalletAddress::default(),
            ws_url: None,
        }
    }
}

fn record_cash_ledger_watermark_gauges(
    now_ms: i64,
    last_event_time_ms: Option<i64>,
    replay_start_time_ms: i64,
) {
    gauge!("ht_hypercore_cash_ledger_observer_replay_start_age_seconds")
        .set(ms_age_seconds(now_ms, replay_start_time_ms));
    match last_event_time_ms {
        Some(last_event_time_ms) => {
            gauge!("ht_hypercore_cash_ledger_observer_watermark_present").set(1.0);
            gauge!("ht_hypercore_cash_ledger_observer_watermark_age_seconds")
                .set(ms_age_seconds(now_ms, last_event_time_ms));
        }
        None => {
            gauge!("ht_hypercore_cash_ledger_observer_watermark_present").set(0.0);
        }
    }
}

fn ms_age_seconds(now_ms: i64, observed_ms: i64) -> f64 {
    now_ms.saturating_sub(observed_ms) as f64 / 1_000.0
}

#[async_trait]
pub trait HypercoreCashLedgerStore: Send + Sync + 'static {
    fn exchange_watermark(&self) -> Result<Option<i64>>;
    async fn pending_rsm_usdc_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>>;
    async fn pending_rsm_usdc_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>>;
    async fn pending_rsm_usdc_deposit_for_credited_hypercore_event(
        &self,
        event_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>>;
    async fn pm_liquidity_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>>;
    async fn pm_liquidity_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>>;
    async fn mark_pm_liquidity_deposit_hypercore_matched(
        &self,
        request_id: &str,
        event_hash: &str,
    ) -> Result<()>;
    async fn credited_wallet_for_hypercore_cash_event(
        &self,
        event_hash: &str,
    ) -> Result<Option<WalletAddress>>;
    async fn non_crediting_hypercore_cash_event(
        &self,
        event_hash: &str,
        amount_usdc: Decimal,
    ) -> Result<bool>;
    async fn mark_rsm_deposit_credit_submitted(&self, request_id: &str) -> Result<()>;
    async fn record_non_crediting_deposit(&self, input: &HypercoreCashLedgerApply) -> Result<()>;
    async fn apply_deposit(
        &self,
        input: &HypercoreCashLedgerApply,
    ) -> Result<hypercall_db::HypercoreCashLedgerApplyResult>;
}

#[async_trait]
impl HypercoreCashLedgerStore for DatabaseHandler {
    fn exchange_watermark(&self) -> Result<Option<i64>> {
        self.get_exchange_cash_ledger_watermark_sync()
    }

    async fn pending_rsm_usdc_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        self.pending_rsm_usdc_deposit_for_amount(amount_wei).await
    }

    async fn pending_rsm_usdc_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        self.pending_rsm_usdc_deposit_for_evm_tx_hash(evm_tx_hash, amount_wei)
            .await
    }

    async fn pending_rsm_usdc_deposit_for_credited_hypercore_event(
        &self,
        event_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        self.pending_rsm_usdc_deposit_for_credited_hypercore_event(event_hash, amount_wei)
            .await
    }

    async fn pm_liquidity_deposit_for_evm_tx_hash(
        &self,
        evm_tx_hash: &str,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        self.pm_liquidity_deposit_for_evm_tx_hash(evm_tx_hash, amount_wei)
            .await
    }

    async fn pm_liquidity_deposit_for_amount(
        &self,
        amount_wei: &str,
    ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
        self.pm_liquidity_deposit_for_amount(amount_wei).await
    }

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

    async fn credited_wallet_for_hypercore_cash_event(
        &self,
        event_hash: &str,
    ) -> Result<Option<WalletAddress>> {
        self.credited_wallet_for_hypercore_cash_event(event_hash)
            .await
    }

    async fn non_crediting_hypercore_cash_event(
        &self,
        event_hash: &str,
        amount_usdc: Decimal,
    ) -> Result<bool> {
        self.non_crediting_hypercore_cash_event(event_hash, amount_usdc)
            .await
    }

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

    async fn record_non_crediting_deposit(&self, input: &HypercoreCashLedgerApply) -> Result<()> {
        self.record_hypercore_cash_deposit_non_crediting(input)
            .await
    }

    async fn apply_deposit(
        &self,
        input: &HypercoreCashLedgerApply,
    ) -> Result<hypercall_db::HypercoreCashLedgerApplyResult> {
        self.apply_hypercore_cash_deposit(input).await
    }
}

#[async_trait]
pub trait HypercoreCashDepositApplier: Send + Sync + 'static {
    async fn apply_deposit(&self, request: DepositRequest) -> Result<()>;
}

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

pub struct HypercoreCashLedgerObserver {
    store: Arc<dyn HypercoreCashLedgerStore>,
    applier: Arc<dyn HypercoreCashDepositApplier>,
    client: reqwest::Client,
    config: HypercoreCashLedgerObserverConfig,
    tier_cache: Arc<TierCache>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CashLedgerCreditDecision {
    Apply,
    SkipPortfolioMargin,
}

impl HypercoreCashLedgerObserver {
    pub fn new(
        db: Arc<DatabaseHandler>,
        deposit_sender: mpsc::Sender<DepositRequest>,
        config: HypercoreCashLedgerObserverConfig,
        tier_cache: Arc<TierCache>,
    ) -> Result<Self> {
        let client = reqwest::Client::builder()
            .timeout(REQUEST_TIMEOUT)
            .build()
            .context("failed to build HyperCore cash ledger HTTP client")?;
        Ok(Self {
            store: db,
            applier: Arc::new(deposit_sender),
            client,
            config,
            tier_cache,
        })
    }

    /// Decide whether the cash ledger credit can be applied.
    ///
    /// PM wallets get equity from hypercore_account_equity (Hydromancer), not
    /// from the balance ledger, so deposit credits must be skipped. Wallets
    /// without an explicit margin-mode row use Standard margin by default.
    async fn cash_ledger_credit_decision(
        &self,
        wallet: &WalletAddress,
    ) -> Result<CashLedgerCreditDecision> {
        match self.tier_cache.get_margin_mode(wallet).await {
            Ok(mode) if mode.is_portfolio() => Ok(CashLedgerCreditDecision::SkipPortfolioMargin),
            Ok(_) => Ok(CashLedgerCreditDecision::Apply),
            Err(error) => Err(error).with_context(|| {
                format!(
                    "failed to determine margin mode for {}; refusing cash ledger credit",
                    wallet
                )
            }),
        }
    }

    pub async fn run_with_shutdown(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<()> {
        match &self.config.ws_url {
            Some(ws_url) => {
                // Startup catchup: poll HTTP to recover any events missed while offline.
                // Retry up to 3 times before entering WS mode so historical gaps are not
                // silently accepted.
                info!("HyperCore cash ledger observer: running startup catchup via HTTP poll");
                let mut catchup_ok = false;
                for attempt in 1..=3u32 {
                    let poll_result = tokio::select! {
                        result = self.poll_once() => result,
                        _ = shutdown_rx.recv() => {
                            info!("HyperCore cash ledger observer: shutdown signal received during startup catchup");
                            return Ok(());
                        }
                    };

                    match poll_result {
                        Ok(()) => {
                            catchup_ok = true;
                            break;
                        }
                        Err(error) => {
                            warn!(
                                attempt,
                                "HyperCore cash ledger observer: startup catchup failed: {}", error
                            );
                            if attempt < 3 {
                                tokio::select! {
                                    _ = tokio::time::sleep(Duration::from_secs(u64::from(attempt) * 2)) => {}
                                    _ = shutdown_rx.recv() => {
                                        info!("HyperCore cash ledger observer: shutdown signal received during startup catchup backoff");
                                        return Ok(());
                                    }
                                }
                            }
                        }
                    }
                }
                if !catchup_ok {
                    return Err(anyhow!(
                        "HyperCore cash ledger observer startup catchup exhausted retries; refusing to switch to WebSocket mode with a known historical gap"
                    ));
                }

                // Switch to WebSocket live mode.
                info!(
                    "HyperCore cash ledger observer: switching to WebSocket mode at {}",
                    ws_url
                );
                self.run_websocket(ws_url.clone(), &mut shutdown_rx).await
            }
            None => {
                // Fallback: HTTP polling mode
                info!("HyperCore cash ledger observer: running in HTTP polling mode (no ws_url configured)");
                self.run_polling(shutdown_rx).await
            }
        }
    }

    /// HTTP polling loop (fallback when no ws_url is configured).
    async fn run_polling(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<()> {
        let mut interval = tokio::time::interval(self.config.poll_interval);

        loop {
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("HyperCore cash ledger observer received shutdown signal");
                    break;
                }
                _ = interval.tick() => {
                    if let Err(error) = self.poll_once().await {
                        counter!("ht_hypercore_cash_ledger_errors_total", "stage" => "poll").increment(1);
                        warn!("HyperCore cash ledger observer poll failed: {}", error);
                    }
                }
            }
        }

        Ok(())
    }

    /// WebSocket live mode with auto-reconnect and exponential backoff.
    async fn run_websocket(
        &self,
        ws_url: String,
        shutdown_rx: &mut broadcast::Receiver<()>,
    ) -> Result<()> {
        let mut consecutive_failures: u32 = 0;

        loop {
            let before = Instant::now();
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("HyperCore cash ledger observer: shutdown signal received (WS mode)");
                    return Ok(());
                }
                result = self.run_ws_connection(&ws_url) => {
                    if let Err(ref err) = result {
                        counter!("ht_hypercore_cash_ledger_errors_total", "stage" => "ws_connection").increment(1);
                        warn!("HyperCore cash ledger observer WS connection error: {}", err);
                    }

                    // If the connection stayed up for >60s it was healthy; reset backoff
                    if before.elapsed() > Duration::from_secs(60) {
                        consecutive_failures = 0;
                    }
                    consecutive_failures += 1;
                    let backoff = std::cmp::min(
                        WS_BASE_BACKOFF * 2u32.saturating_pow(consecutive_failures - 1),
                        WS_MAX_BACKOFF,
                    );
                    warn!(
                        "HyperCore cash ledger observer: WS disconnected (failures: {}), reconnecting in {:?}",
                        consecutive_failures, backoff
                    );

                    tokio::select! {
                        _ = shutdown_rx.recv() => {
                            info!("HyperCore cash ledger observer: shutdown during reconnect backoff");
                            return Ok(());
                        }
                        _ = tokio::time::sleep(backoff) => {}
                    }

                    // Catch-up via HTTP poll before reconnecting WS to recover
                    // any deposits missed during the connection gap.
                    if let Err(error) = self.poll_once().await {
                        warn!(
                            "HyperCore cash ledger observer: reconnect catch-up poll failed: {}",
                            error
                        );
                    }
                }
            }
        }
    }

    /// Single WebSocket connection lifecycle: connect, subscribe, process events.
    async fn run_ws_connection(&self, ws_url: &str) -> Result<()> {
        info!(
            "HyperCore cash ledger observer: connecting to WS at {}",
            ws_url
        );

        let (ws_stream, _) = tokio_tungstenite::connect_async(ws_url)
            .await
            .context("failed to connect to Hydromancer WebSocket")?;

        info!("HyperCore cash ledger observer: WS connected");
        gauge!("ht_hypercore_cash_ledger_ws_connected").set(1.0);

        let (mut write, mut read) = ws_stream.split();

        // Subscribe to allUserNonFundingLedgerEvents
        let sub_msg = serde_json::json!({
            "method": "subscribe",
            "subscription": {
                "type": "allUserNonFundingLedgerEvents"
            }
        });
        let sub_text = serde_json::to_string(&sub_msg)
            .context("failed to serialize WS subscription message")?;
        write
            .send(Message::Text(sub_text))
            .await
            .context("failed to send WS subscription message")?;
        info!("HyperCore cash ledger observer: subscribed to allUserNonFundingLedgerEvents");

        let mut catchup_interval = tokio::time::interval(self.config.poll_interval);
        catchup_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        catchup_interval.tick().await;
        let read_timeout = tokio::time::sleep(WS_READ_TIMEOUT);
        tokio::pin!(read_timeout);

        // Read and process messages with timeout to detect hung connections
        loop {
            tokio::select! {
                _ = &mut read_timeout => {
                    warn!(
                        "HyperCore cash ledger observer: WS read timeout ({}s), reconnecting",
                        WS_READ_TIMEOUT.as_secs()
                    );
                    break;
                },
                maybe_msg = read.next() => {
                    read_timeout.as_mut().reset(tokio::time::Instant::now() + WS_READ_TIMEOUT);
                    match maybe_msg {
                        Some(Ok(Message::Text(text))) => {
                            if let Err(error) = self.handle_ws_message(&text).await {
                                counter!("ht_hypercore_cash_ledger_errors_total", "stage" => "ws_message").increment(1);
                                warn!(
                                    "HyperCore cash ledger observer: error handling WS message: {}",
                                    error
                                );
                            }
                        }
                        Some(Ok(Message::Ping(data))) => {
                            if let Err(e) = write.send(Message::Pong(data)).await {
                                error!("HyperCore cash ledger observer: failed to send pong: {}", e);
                                break;
                            }
                        }
                        Some(Ok(Message::Close(_))) => {
                            info!("HyperCore cash ledger observer: server sent close frame");
                            break;
                        }
                        Some(Err(e)) => {
                            error!("HyperCore cash ledger observer: WS read error: {}", e);
                            break;
                        }
                        None => {
                            info!("HyperCore cash ledger observer: WS stream ended");
                            break;
                        }
                        _ => {}
                    }
                },
                _ = catchup_interval.tick() => {
                    if let Err(error) = self.poll_once().await {
                        counter!("ht_hypercore_cash_ledger_errors_total", "stage" => "ws_catchup_poll").increment(1);
                        warn!(
                            "HyperCore cash ledger observer WS catch-up poll failed: {}",
                            error
                        );
                    }
                }
            }
        }

        gauge!("ht_hypercore_cash_ledger_ws_connected").set(0.0);
        Ok(())
    }

    /// Process a single WebSocket message from the allUserNonFundingLedgerEvents stream.
    async fn handle_ws_message(&self, text: &str) -> Result<()> {
        // Try to parse as a ledger event envelope
        let envelope: WsLedgerEnvelope = match serde_json::from_str(text) {
            Ok(e) => e,
            Err(_) => {
                // Subscription confirmations and other non-data messages
                debug!(
                    "HyperCore cash ledger observer: non-data WS message: {}",
                    &text[..text.len().min(200)]
                );
                return Ok(());
            }
        };

        // Only process allUserNonFundingLedgerEvents channel messages
        if !matches!(
            envelope.channel.as_deref(),
            Some(c) if c.contains("NonFundingLedger")
        ) {
            return Ok(());
        }

        let events = match envelope.data {
            Some(events) => events,
            None => return Ok(()),
        };

        for event in events {
            self.handle_ledger_update(event, "websocket").await?;
        }

        Ok(())
    }

    /// Process a normalized HyperCore cash ledger update from websocket or HTTP polling.
    async fn handle_ledger_update(&self, update: LedgerUpdate, source: &'static str) -> Result<()> {
        let Some(transfer) = update.transfer()? else {
            return Ok(());
        };
        // hypercore-rs exposes generic transfer facts. Hypercall-specific
        // deposit semantics start here: only USDC transfers into the configured
        // exchange account can become engine cash deposits.
        if transfer.destination != self.config.exchange_address.0 {
            return Ok(());
        }
        if transfer.token != "USDC" {
            return Ok(());
        };
        let sender = WalletAddress::from(transfer.sender);
        let amount_usdc = transfer.amount;
        if amount_usdc <= Decimal::ZERO {
            anyhow::bail!(
                "pool transfer {} has non-positive amount {}",
                transfer.event_hash,
                amount_usdc
            );
        }

        let credit_resolution = self
            .resolve_usdc_deposit_credit(
                sender,
                amount_usdc,
                &transfer.event_hash.to_string(),
                transfer.writer_evm_tx_hash.as_deref(),
            )
            .await?;
        let (credit_wallet, request_id, non_crediting_reason) = match credit_resolution {
            CashCreditResolution::DirectSender => (sender, None, None),
            CashCreditResolution::CreditAccount { wallet, request_id } => {
                (wallet, request_id, None)
            }
            CashCreditResolution::SkipPmLiquidity => (sender, None, Some("pm_liquidity")),
        };
        let event_time_ms = i64::try_from(transfer.time)
            .with_context(|| format!("ledger event time {} exceeds i64 range", transfer.time))?;
        let event_hash = transfer.event_hash.to_string();
        let apply = HypercoreCashLedgerApply {
            wallet: credit_wallet,
            event_hash,
            event_time_ms,
            amount_usdc,
        };
        if let Some(reason) = non_crediting_reason {
            self.store.record_non_crediting_deposit(&apply).await?;
            counter!(
                "ht_hypercore_cash_pool_transfers_skipped_total",
                "reason" => reason
            )
            .increment(1);
            return Ok(());
        }

        // Portfolio-margin wallets get cash/equity from the Hydromancer equity
        // path. Crediting this projection as cash would double-count the same
        // HyperCore state in the engine.
        match self.cash_ledger_credit_decision(&credit_wallet).await? {
            CashLedgerCreditDecision::Apply => {}
            CashLedgerCreditDecision::SkipPortfolioMargin => {
                counter!("ht_hypercore_cash_pool_transfers_skipped_total", "reason" => "portfolio_margin")
                    .increment(1);
                warn!(
                    wallet = %credit_wallet,
                    amount_usdc = %amount_usdc,
                    event_hash = %apply.event_hash,
                    source = source,
                    "Skipping balance_ledger credit for portfolio margin wallet (equity via Hydromancer)"
                );
                return Ok(());
            }
        }
        // The DB projection write is idempotent by HyperCore event hash and
        // returns the durable ledger_event_id used as the engine command
        // sequence. Engine apply must use that sequence so replay and retries
        // cannot double-apply the same transfer.
        let result = self.store.apply_deposit(&apply).await?;
        self.apply_engine_deposit(
            credit_wallet,
            amount_usdc,
            &result,
            transfer.time,
            transfer.event_hash,
        )
        .await?;
        if let Some(request_id) = request_id {
            self.store
                .mark_rsm_deposit_credit_submitted(&request_id)
                .await?;
        }
        counter!("ht_hypercore_cash_pool_transfers_applied_total").increment(1);
        info!(
            wallet = %credit_wallet,
            hypercore_sender = %sender,
            amount_usdc = %amount_usdc,
            event_hash = %apply.event_hash,
            balance_after = ?result.balance_after,
            exchange_address = %self.config.exchange_address,
            source = source,
            "Applied Account->Exchange pool transfer to engine cash ledger"
        );
        Ok(())
    }

    async fn poll_once(&self) -> Result<()> {
        let now_ms =
            i64::try_from(get_timestamp_millis()).context("current timestamp exceeds i64 range")?;
        let last_event_time_ms = self.store.exchange_watermark()?;
        let start_time = last_event_time_ms
            .map(|last| last.saturating_sub(REPLAY_OVERLAP_MS))
            .unwrap_or_else(|| now_ms.saturating_sub(STARTUP_LOOKBACK_MS));
        record_cash_ledger_watermark_gauges(now_ms, last_event_time_ms, start_time);

        let request = HyperliquidLedgerRequest::user_non_funding_ledger_updates(
            self.config.exchange_address.to_string(),
            start_time,
        );
        let response = self
            .client
            .post(&self.config.info_url)
            .json(&request)
            .send()
            .await
            .with_context(|| {
                format!(
                    "failed to request ledger updates for exchange {}",
                    self.config.exchange_address
                )
            })?;
        let status = response.status();
        if !status.is_success() {
            let body = response
                .text()
                .await
                .unwrap_or_else(|_| "unknown error".to_string());
            anyhow::bail!(
                "ledger update request for exchange {} failed: {status}: {body}",
                self.config.exchange_address
            );
        }

        let updates: Vec<LedgerUpdate> = response.json().await.with_context(|| {
            format!(
                "failed to parse ledger updates for exchange {}",
                self.config.exchange_address
            )
        })?;
        for update in updates {
            self.handle_ledger_update(update, "http_poll").await?;
        }
        Ok(())
    }
    async fn apply_engine_deposit(
        &self,
        wallet: WalletAddress,
        amount_usdc: Decimal,
        result: &hypercall_db::HypercoreCashLedgerApplyResult,
        timestamp_ms: u64,
        source_event_hash: FixedBytes<32>,
    ) -> Result<()> {
        let Some(sequence) = result.ledger_event_id else {
            anyhow::bail!(
                "cash ledger deposit for {} source_event_hash={} has no ledger_event_id; refusing engine apply",
                wallet,
                source_event_hash
            );
        };
        self.applier
            .apply_deposit(DepositRequest {
                wallet,
                amount: amount_usdc,
                timestamp_ms,
                sequence: Some(sequence),
                source_event_hash,
                journal_request_id: hypercore_cash_deposit_request_id(sequence),
                outbox_appends: Vec::new(),
                applied_tx: None,
            })
            .await
    }

    async fn resolve_usdc_deposit_credit(
        &self,
        hypercore_sender: WalletAddress,
        amount_usdc: Decimal,
        event_hash: &str,
        writer_evm_tx_hash: Option<&str>,
    ) -> Result<CashCreditResolution> {
        if hypercore_sender != self.config.core_deposit_wallet_address
            && hypercore_sender != self.config.exchange_address
        {
            return Ok(CashCreditResolution::DirectSender);
        }

        let amount_wei = usdc_decimal_to_wei_string(amount_usdc)?;
        if let Some(wallet) = self
            .store
            .credited_wallet_for_hypercore_cash_event(event_hash)
            .await?
        {
            let request_id = self
                .store
                .pending_rsm_usdc_deposit_for_credited_hypercore_event(event_hash, &amount_wei)
                .await?
                .map(|matched| matched.request_id);
            return Ok(CashCreditResolution::CreditAccount { wallet, request_id });
        }
        if self
            .store
            .non_crediting_hypercore_cash_event(event_hash, amount_usdc)
            .await?
        {
            return Ok(CashCreditResolution::SkipPmLiquidity);
        }

        let matched = if let Some(writer_evm_tx_hash) = writer_evm_tx_hash.and_then(non_empty_str) {
            match self
                .store
                .pending_rsm_usdc_deposit_for_evm_tx_hash(writer_evm_tx_hash, &amount_wei)
                .await?
            {
                Some(matched) => matched,
                None => {
                    if let Some(pm_match) = self
                        .store
                        .pm_liquidity_deposit_for_evm_tx_hash(writer_evm_tx_hash, &amount_wei)
                        .await?
                    {
                        self.store
                            .mark_pm_liquidity_deposit_hypercore_matched(
                                &pm_match.request_id,
                                event_hash,
                            )
                            .await?;
                        info!(
                            lp = %pm_match.account,
                            evm_tx_hash = %pm_match.tx_hash,
                            evm_log_index = pm_match.log_index,
                            hypercore_event_hash = %event_hash,
                            amount_wei = %amount_wei,
                            "Matched Exchange.PmLiquidityDeposit to HyperCore cash ledger deposit; skipping user cash credit"
                        );
                        return Ok(CashCreditResolution::SkipPmLiquidity);
                    }
                    anyhow::bail!(
                        "CoreWriter evm_tx_hash {} for HyperCore USDC deposit {} amount={} has no pending Exchange.UsdcDeposit attribution",
                        writer_evm_tx_hash,
                        event_hash,
                        amount_usdc
                    );
                }
            }
        } else {
            match self
                .store
                .pending_rsm_usdc_deposit_for_amount(&amount_wei)
                .await?
            {
                Some(matched) => matched,
                None => {
                    if let Some(pm_match) = self
                        .store
                        .pm_liquidity_deposit_for_amount(&amount_wei)
                        .await?
                    {
                        self.store
                            .mark_pm_liquidity_deposit_hypercore_matched(
                                &pm_match.request_id,
                                event_hash,
                            )
                            .await?;
                        info!(
                            lp = %pm_match.account,
                            evm_tx_hash = %pm_match.tx_hash,
                            evm_log_index = pm_match.log_index,
                            hypercore_event_hash = %event_hash,
                            amount_wei = %amount_wei,
                            "Matched Exchange.PmLiquidityDeposit to HyperCore cash ledger deposit by amount; skipping user cash credit"
                        );
                        return Ok(CashCreditResolution::SkipPmLiquidity);
                    }
                    anyhow::bail!(
                        "Exchange/CoreDepositWallet-originated HyperCore USDC deposit {} amount={} has no pending Exchange.UsdcDeposit attribution",
                        event_hash,
                        amount_usdc
                    );
                }
            }
        };

        info!(
            credited_account = %matched.account,
            evm_tx_hash = %matched.tx_hash,
            evm_log_index = matched.log_index,
            hypercore_event_hash = %event_hash,
            writer_evm_tx_hash = writer_evm_tx_hash.unwrap_or(""),
            amount_wei = %amount_wei,
            "Matched Exchange.UsdcDeposit to HyperCore cash ledger deposit"
        );
        Ok(CashCreditResolution::CreditAccount {
            wallet: matched.account,
            request_id: Some(matched.request_id),
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum CashCreditResolution {
    DirectSender,
    CreditAccount {
        wallet: WalletAddress,
        request_id: Option<String>,
    },
    SkipPmLiquidity,
}

fn hypercore_cash_deposit_request_id(ledger_event_id: u64) -> String {
    const HYPERCORE_CASH_DEPOSIT_UUID_PREFIX: u128 = 0x4859_4343_4153_4800_0000_0000_0000_0000;
    uuid::Uuid::from_u128(HYPERCORE_CASH_DEPOSIT_UUID_PREFIX | u128::from(ledger_event_id))
        .to_string()
}

fn usdc_decimal_to_wei_string(amount_usdc: Decimal) -> Result<String> {
    if amount_usdc <= Decimal::ZERO {
        anyhow::bail!("USDC deposit amount must be positive, got {}", amount_usdc);
    }
    let scaled = amount_usdc * dec!(1000000);
    if scaled.fract() != Decimal::ZERO {
        anyhow::bail!(
            "USDC deposit amount {} has more than 6 decimal places",
            amount_usdc
        );
    }
    let wei = scaled
        .to_u128()
        .ok_or_else(|| anyhow!("USDC deposit amount {} exceeds u128 range", amount_usdc))?;
    Ok(wei.to_string())
}

fn non_empty_str(value: &str) -> Option<&str> {
    let trimmed = value.trim();
    if trimmed.is_empty() {
        None
    } else {
        Some(trimmed)
    }
}

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

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

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

#[cfg(test)]
mod tests {
    use super::*;
    use hypercore_rs::cash_ledger::LedgerDelta;
    use std::collections::{HashMap, HashSet, VecDeque};
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::sync::Mutex;

    // ── Config ────────────────────────────────────────────────────────

    fn exchange_addr() -> WalletAddress {
        WalletAddress::from(alloy::primitives::address!(
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
        ))
    }

    fn core_deposit_wallet_addr() -> WalletAddress {
        WalletAddress::from(alloy::primitives::address!(
            "6b9e773128f453f5c2c60935ee2de2cbc5390a24"
        ))
    }

    fn sender_addr() -> WalletAddress {
        WalletAddress::from(alloy::primitives::address!(
            "301cd221cf81ef94cd276042aacfbcd0e3795589"
        ))
    }

    fn test_event_hash(byte: u8) -> FixedBytes<32> {
        FixedBytes::from([byte; 32])
    }

    fn test_event_hash_string(byte: u8) -> String {
        test_event_hash(byte).to_string()
    }

    struct MockTierDb {
        mode: Option<hypercall_types::MarginMode>,
    }

    impl hypercall_db::TierReader for MockTierDb {
        fn get_margin_mode_sync(
            &self,
            _wallet: &WalletAddress,
        ) -> Result<hypercall_types::MarginMode> {
            Ok(self.mode.unwrap_or(hypercall_types::MarginMode::Standard))
        }

        fn get_existing_margin_mode_sync(
            &self,
            _wallet: &WalletAddress,
        ) -> Result<Option<hypercall_types::MarginMode>> {
            Ok(self.mode)
        }

        fn get_tier_defaults_sync(
            &self,
            _tier_name: &str,
        ) -> Result<Option<hypercall_db::TierDefaultsRecord>> {
            Ok(None)
        }

        fn get_user_tier_sync(
            &self,
            _wallet: &WalletAddress,
        ) -> Result<Option<hypercall_db::UserTierRecord>> {
            Ok(None)
        }

        fn get_all_user_tiers_sync(&self) -> Result<Vec<hypercall_db::UserTierRecord>> {
            Ok(Vec::new())
        }
    }

    impl hypercall_db::TierWriter for MockTierDb {
        fn save_user_tier_sync(&self, _update: &hypercall_db::UserTierUpdate) -> Result<()> {
            Ok(())
        }

        fn set_margin_mode_sync(
            &self,
            _wallet: &WalletAddress,
            _margin_mode: hypercall_types::MarginMode,
        ) -> Result<i64> {
            Ok(1)
        }

        fn insert_margin_mode_if_missing_sync(
            &self,
            _wallet: &WalletAddress,
            _margin_mode: hypercall_types::MarginMode,
        ) -> Result<Option<i64>> {
            Ok(Some(1))
        }

        fn delete_user_tier_sync(&self, _wallet: &WalletAddress) -> Result<()> {
            Ok(())
        }
    }

    struct CountingStore {
        apply_calls: AtomicUsize,
        apply_inputs: Mutex<Vec<HypercoreCashLedgerApply>>,
        apply_results: Mutex<VecDeque<hypercall_db::HypercoreCashLedgerApplyResult>>,
        pending_usdc_matches: Mutex<VecDeque<hypercall_db::RsmUsdcDepositMatch>>,
        pm_liquidity_matches: Mutex<VecDeque<hypercall_db::RsmUsdcDepositMatch>>,
        matched_pm_liquidity_requests: Mutex<Vec<(String, String)>>,
        credited_event_wallets: Mutex<HashMap<String, WalletAddress>>,
        non_crediting_events: Mutex<HashSet<(String, Decimal)>>,
        submitted_usdc_requests: Mutex<Vec<String>>,
    }

    #[async_trait::async_trait]
    impl HypercoreCashLedgerStore for CountingStore {
        fn exchange_watermark(&self) -> Result<Option<i64>> {
            Ok(None)
        }

        async fn pending_rsm_usdc_deposit_for_amount(
            &self,
            _amount_wei: &str,
        ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
            Ok(self
                .pending_usdc_matches
                .lock()
                .expect("pending_usdc_matches mutex poisoned")
                .pop_front())
        }

        async fn pending_rsm_usdc_deposit_for_evm_tx_hash(
            &self,
            evm_tx_hash: &str,
            amount_wei: &str,
        ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
            let mut matches = self
                .pending_usdc_matches
                .lock()
                .expect("pending_usdc_matches mutex poisoned");
            let Some(index) = matches
                .iter()
                .position(|row| row.tx_hash == evm_tx_hash && row.amount_wei == amount_wei)
            else {
                return Ok(None);
            };
            Ok(matches.remove(index))
        }

        async fn pending_rsm_usdc_deposit_for_credited_hypercore_event(
            &self,
            event_hash: &str,
            amount_wei: &str,
        ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
            let Some(wallet) = self
                .credited_event_wallets
                .lock()
                .expect("credited_event_wallets mutex poisoned")
                .get(event_hash)
                .copied()
            else {
                return Ok(None);
            };
            let mut matches = self
                .pending_usdc_matches
                .lock()
                .expect("pending_usdc_matches mutex poisoned");
            let Some(index) = matches
                .iter()
                .position(|row| row.account == wallet && row.amount_wei == amount_wei)
            else {
                return Ok(None);
            };
            Ok(matches.remove(index))
        }

        async fn pm_liquidity_deposit_for_evm_tx_hash(
            &self,
            evm_tx_hash: &str,
            amount_wei: &str,
        ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
            Ok(self
                .pm_liquidity_matches
                .lock()
                .expect("pm_liquidity_matches mutex poisoned")
                .iter()
                .find(|row| row.tx_hash == evm_tx_hash && row.amount_wei == amount_wei)
                .cloned())
        }

        async fn pm_liquidity_deposit_for_amount(
            &self,
            amount_wei: &str,
        ) -> Result<Option<hypercall_db::RsmUsdcDepositMatch>> {
            Ok(self
                .pm_liquidity_matches
                .lock()
                .expect("pm_liquidity_matches mutex poisoned")
                .iter()
                .find(|row| row.amount_wei == amount_wei)
                .cloned())
        }

        async fn mark_pm_liquidity_deposit_hypercore_matched(
            &self,
            request_id: &str,
            event_hash: &str,
        ) -> Result<()> {
            self.matched_pm_liquidity_requests
                .lock()
                .expect("matched_pm_liquidity_requests mutex poisoned")
                .push((request_id.to_string(), event_hash.to_string()));
            let mut matches = self
                .pm_liquidity_matches
                .lock()
                .expect("pm_liquidity_matches mutex poisoned");
            if let Some(index) = matches.iter().position(|row| row.request_id == request_id) {
                matches.remove(index);
            }
            Ok(())
        }

        async fn credited_wallet_for_hypercore_cash_event(
            &self,
            event_hash: &str,
        ) -> Result<Option<WalletAddress>> {
            Ok(self
                .credited_event_wallets
                .lock()
                .expect("credited_event_wallets mutex poisoned")
                .get(event_hash)
                .copied())
        }

        async fn non_crediting_hypercore_cash_event(
            &self,
            event_hash: &str,
            amount_usdc: Decimal,
        ) -> Result<bool> {
            Ok(self
                .non_crediting_events
                .lock()
                .expect("non_crediting_events mutex poisoned")
                .contains(&(event_hash.to_string(), amount_usdc)))
        }

        async fn mark_rsm_deposit_credit_submitted(&self, request_id: &str) -> Result<()> {
            self.submitted_usdc_requests
                .lock()
                .expect("submitted_usdc_requests mutex poisoned")
                .push(request_id.to_string());
            Ok(())
        }

        async fn record_non_crediting_deposit(
            &self,
            input: &HypercoreCashLedgerApply,
        ) -> Result<()> {
            self.non_crediting_events
                .lock()
                .expect("non_crediting_events mutex poisoned")
                .insert((input.event_hash.clone(), input.amount_usdc));
            self.apply_inputs
                .lock()
                .expect("apply_inputs mutex poisoned")
                .push(input.clone());
            Ok(())
        }

        async fn apply_deposit(
            &self,
            input: &HypercoreCashLedgerApply,
        ) -> Result<hypercall_db::HypercoreCashLedgerApplyResult> {
            self.apply_calls.fetch_add(1, Ordering::SeqCst);
            self.apply_inputs
                .lock()
                .expect("apply_inputs mutex poisoned")
                .push(input.clone());
            if let Some(result) = self
                .apply_results
                .lock()
                .expect("apply_results mutex poisoned")
                .pop_front()
            {
                return Ok(result);
            }
            Ok(hypercall_db::HypercoreCashLedgerApplyResult {
                applied: true,
                balance_after: Some(Decimal::ONE),
                ledger_event_id: Some(1),
            })
        }
    }

    struct CountingApplier {
        calls: AtomicUsize,
        fail_next: AtomicBool,
        journal_request_ids: Mutex<Vec<String>>,
    }

    #[async_trait::async_trait]
    impl HypercoreCashDepositApplier for CountingApplier {
        async fn apply_deposit(&self, request: DepositRequest) -> Result<()> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.journal_request_ids
                .lock()
                .expect("journal_request_ids mutex poisoned")
                .push(request.journal_request_id);
            if self.fail_next.swap(false, Ordering::SeqCst) {
                anyhow::bail!("injected engine apply failure");
            }
            Ok(())
        }
    }

    fn test_observer_with_margin_mode(
        mode: Option<hypercall_types::MarginMode>,
    ) -> (
        HypercoreCashLedgerObserver,
        Arc<CountingStore>,
        Arc<CountingApplier>,
    ) {
        let store = Arc::new(CountingStore {
            apply_calls: AtomicUsize::new(0),
            apply_inputs: Mutex::new(Vec::new()),
            apply_results: Mutex::new(VecDeque::new()),
            pending_usdc_matches: Mutex::new(VecDeque::new()),
            pm_liquidity_matches: Mutex::new(VecDeque::new()),
            matched_pm_liquidity_requests: Mutex::new(Vec::new()),
            credited_event_wallets: Mutex::new(HashMap::new()),
            non_crediting_events: Mutex::new(HashSet::new()),
            submitted_usdc_requests: Mutex::new(Vec::new()),
        });
        let applier = Arc::new(CountingApplier {
            calls: AtomicUsize::new(0),
            fail_next: AtomicBool::new(false),
            journal_request_ids: Mutex::new(Vec::new()),
        });
        let tier_cache = Arc::new(
            TierCache::new(Arc::new(MockTierDb { mode })).expect("tier cache should initialize"),
        );
        let observer = HypercoreCashLedgerObserver {
            store: store.clone(),
            applier: applier.clone(),
            client: reqwest::Client::new(),
            config: HypercoreCashLedgerObserverConfig {
                exchange_address: exchange_addr(),
                core_deposit_wallet_address: core_deposit_wallet_addr(),
                ..Default::default()
            },
            tier_cache,
        };
        (observer, store, applier)
    }

    #[tokio::test]
    async fn ws_missing_margin_mode_defaults_to_standard_and_credits() {
        let (observer, store, applier) = test_observer_with_margin_mode(None);
        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(1),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(sender_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("missing margin mode should default to Standard");

        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 1);
        assert_eq!(applier.calls.load(Ordering::SeqCst), 1);
        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 1);
        assert_eq!(
            apply_inputs[0].wallet,
            sender_addr(),
            "direct user HyperCore deposits must credit the sender"
        );
        assert!(
            applier
                .journal_request_ids
                .lock()
                .expect("journal_request_ids mutex poisoned")[0]
                .parse::<uuid::Uuid>()
                .ok()
                .is_some(),
            "standard-margin HyperCore cash deposits must carry a replayable engine journal request id"
        );
    }

    #[tokio::test]
    async fn ws_deposit_replays_engine_apply_after_db_applied_retry() {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        store
            .apply_results
            .lock()
            .expect("apply_results mutex poisoned")
            .extend([
                hypercall_db::HypercoreCashLedgerApplyResult {
                    applied: true,
                    balance_after: Some(Decimal::ONE),
                    ledger_event_id: Some(41),
                },
                hypercall_db::HypercoreCashLedgerApplyResult {
                    applied: false,
                    balance_after: Some(Decimal::ONE),
                    ledger_event_id: Some(41),
                },
            ]);
        applier.fail_next.store(true, Ordering::SeqCst);

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(2),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(sender_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        let first = observer
            .handle_ledger_update(event.clone(), "websocket")
            .await;
        assert!(first
            .expect_err("first apply should surface injected engine failure")
            .to_string()
            .contains("injected engine apply failure"));

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("retry should replay engine apply using durable ledger_event_id");

        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 2);
        assert_eq!(applier.calls.load(Ordering::SeqCst), 2);
        let journal_request_ids = applier
            .journal_request_ids
            .lock()
            .expect("journal_request_ids mutex poisoned");
        assert_eq!(
            journal_request_ids.len(),
            2,
            "both first apply and retry should hand the engine a restart-source request id"
        );
        assert_eq!(
            journal_request_ids[0], journal_request_ids[1],
            "retries for the same durable ledger event must reuse the same engine journal request id"
        );
        assert!(
            journal_request_ids[0]
                .parse::<uuid::Uuid>()
                .ok()
                .is_some(),
            "standard-margin HyperCore cash deposits must carry a replayable engine journal request id"
        );
    }

    #[tokio::test]
    async fn core_deposit_wallet_originated_deposit_credits_matched_usdc_account_not_exchange() {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        let credited_account = WalletAddress::from(alloy::primitives::address!(
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
        ));
        store
            .pending_usdc_matches
            .lock()
            .expect("pending_usdc_matches mutex poisoned")
            .push_back(hypercall_db::RsmUsdcDepositMatch {
                request_id: "usdc-request-1".to_string(),
                account: credited_account,
                tx_hash: "0xevm".to_string(),
                log_index: 3,
                amount_wei: "1000000".to_string(),
                token: WalletAddress::from(alloy::primitives::address!(
                    "cccccccccccccccccccccccccccccccccccccccc"
                )),
            });

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(3),
            evm_tx_hash: Some("0xevm".to_string()),
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(core_deposit_wallet_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("matched Exchange-originated deposit should credit attributed account");

        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 1);
        assert_eq!(apply_inputs[0].wallet, credited_account);
        assert_ne!(
            apply_inputs[0].wallet,
            exchange_addr(),
            "Exchange-originated deposits must not credit Exchange itself"
        );
        assert_ne!(
            apply_inputs[0].wallet,
            core_deposit_wallet_addr(),
            "CoreDepositWallet-originated deposits must not credit CoreDepositWallet itself"
        );
        assert_eq!(apply_inputs[0].event_hash, test_event_hash_string(3));
        assert_eq!(applier.calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            store
                .submitted_usdc_requests
                .lock()
                .expect("submitted_usdc_requests mutex poisoned")
                .as_slice(),
            &["usdc-request-1".to_string()]
        );
    }

    #[tokio::test]
    async fn core_deposit_wallet_originated_pm_liquidity_with_evm_tx_is_consumed_and_not_credited()
    {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        store
            .pm_liquidity_matches
            .lock()
            .expect("pm_liquidity_matches mutex poisoned")
            .push_back(hypercall_db::RsmUsdcDepositMatch {
                request_id: "pm-liquidity-request-1".to_string(),
                account: sender_addr(),
                tx_hash: "0xpm-liquidity".to_string(),
                log_index: 9,
                amount_wei: "1000000".to_string(),
                token: WalletAddress::from(alloy::primitives::address!(
                    "cccccccccccccccccccccccccccccccccccccccc"
                )),
            });

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(7),
            evm_tx_hash: Some("0xpm-liquidity".to_string()),
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(core_deposit_wallet_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("PM liquidity deposit should be recorded as non-crediting");

        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 0);
        assert_eq!(applier.calls.load(Ordering::SeqCst), 0);
        assert_eq!(
            store
                .matched_pm_liquidity_requests
                .lock()
                .expect("matched_pm_liquidity_requests mutex poisoned")
                .as_slice(),
            &[(
                "pm-liquidity-request-1".to_string(),
                test_event_hash_string(7)
            )]
        );
        assert!(store
            .pm_liquidity_matches
            .lock()
            .expect("pm_liquidity_matches mutex poisoned")
            .is_empty());
        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 1);
        assert_eq!(apply_inputs[0].event_hash, test_event_hash_string(7));
    }

    #[tokio::test]
    async fn core_deposit_wallet_originated_pm_liquidity_amount_match_is_consumed_and_not_credited()
    {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        store
            .pm_liquidity_matches
            .lock()
            .expect("pm_liquidity_matches mutex poisoned")
            .push_back(hypercall_db::RsmUsdcDepositMatch {
                request_id: "pm-liquidity-request-2".to_string(),
                account: sender_addr(),
                tx_hash: "0xpm-liquidity-amount".to_string(),
                log_index: 10,
                amount_wei: "1000000".to_string(),
                token: WalletAddress::from(alloy::primitives::address!(
                    "cccccccccccccccccccccccccccccccccccccccc"
                )),
            });

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(8),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(core_deposit_wallet_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("amount-matched PM liquidity deposit should be recorded as non-crediting");

        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 0);
        assert_eq!(applier.calls.load(Ordering::SeqCst), 0);
        assert_eq!(
            store
                .matched_pm_liquidity_requests
                .lock()
                .expect("matched_pm_liquidity_requests mutex poisoned")
                .as_slice(),
            &[(
                "pm-liquidity-request-2".to_string(),
                test_event_hash_string(8)
            )]
        );
        assert!(store
            .pm_liquidity_matches
            .lock()
            .expect("pm_liquidity_matches mutex poisoned")
            .is_empty());
        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 1);
        assert_eq!(apply_inputs[0].event_hash, test_event_hash_string(8));
    }

    #[tokio::test]
    async fn replayed_pm_liquidity_non_crediting_event_does_not_rematch_or_credit() {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        store
            .pm_liquidity_matches
            .lock()
            .expect("pm_liquidity_matches mutex poisoned")
            .push_back(hypercall_db::RsmUsdcDepositMatch {
                request_id: "pm-liquidity-request-replay".to_string(),
                account: sender_addr(),
                tx_hash: "0xpm-liquidity-replay".to_string(),
                log_index: 11,
                amount_wei: "1000000".to_string(),
                token: WalletAddress::from(alloy::primitives::address!(
                    "cccccccccccccccccccccccccccccccccccccccc"
                )),
            });

        let event = || LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(9),
            evm_tx_hash: Some("0xpm-liquidity-replay".to_string()),
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(core_deposit_wallet_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event(), "websocket")
            .await
            .expect("first PM liquidity deposit should be recorded as non-crediting");
        observer
            .handle_ledger_update(event(), "replay")
            .await
            .expect("replayed PM liquidity deposit should stay non-crediting");

        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 0);
        assert_eq!(applier.calls.load(Ordering::SeqCst), 0);
        assert_eq!(
            store
                .matched_pm_liquidity_requests
                .lock()
                .expect("matched_pm_liquidity_requests mutex poisoned")
                .as_slice(),
            &[(
                "pm-liquidity-request-replay".to_string(),
                test_event_hash_string(9)
            )],
            "replay must not try to consume the PM liquidity request twice"
        );
        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 2);
        assert!(apply_inputs
            .iter()
            .all(|input| input.event_hash == test_event_hash_string(9)));
    }

    #[tokio::test]
    async fn core_deposit_wallet_originated_deposit_without_evm_event_fails_before_credit() {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(4),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(core_deposit_wallet_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        let error = observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect_err("unmatched CoreDepositWallet-originated deposit must not be credited");
        assert!(
            error
                .to_string()
                .contains("no pending Exchange.UsdcDeposit"),
            "unexpected error: {error:#}"
        );
        assert_eq!(store.apply_calls.load(Ordering::SeqCst), 0);
        assert!(
            store
                .apply_inputs
                .lock()
                .expect("apply_inputs mutex poisoned")
                .is_empty(),
            "unattributed CoreDepositWallet-originated deposits must not write a cash ledger row"
        );
        assert_eq!(applier.calls.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn exchange_originated_deposit_replay_uses_existing_credited_wallet() {
        let (observer, store, applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        let credited_account = WalletAddress::from(alloy::primitives::address!(
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
        ));
        store
            .credited_event_wallets
            .lock()
            .expect("credited_event_wallets mutex poisoned")
            .insert(test_event_hash_string(5), credited_account);

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(5),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(exchange_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("replayed matched event should reuse credited wallet");

        let apply_inputs = store
            .apply_inputs
            .lock()
            .expect("apply_inputs mutex poisoned");
        assert_eq!(apply_inputs.len(), 1);
        assert_eq!(apply_inputs[0].wallet, credited_account);
        assert!(
            store
                .submitted_usdc_requests
                .lock()
                .expect("submitted_usdc_requests mutex poisoned")
                .is_empty(),
            "replay path must not require a pending EVM request"
        );
        assert_eq!(applier.calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn exchange_originated_deposit_replay_marks_recovered_pending_request_submitted() {
        let (observer, store, _applier) =
            test_observer_with_margin_mode(Some(hypercall_types::MarginMode::Standard));
        let credited_account = WalletAddress::from(alloy::primitives::address!(
            "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
        ));
        store
            .credited_event_wallets
            .lock()
            .expect("credited_event_wallets mutex poisoned")
            .insert(test_event_hash_string(6), credited_account);
        store
            .pending_usdc_matches
            .lock()
            .expect("pending_usdc_matches mutex poisoned")
            .push_back(hypercall_db::RsmUsdcDepositMatch {
                request_id: "usdc-request-recovered".to_string(),
                account: credited_account,
                tx_hash: "0xevm".to_string(),
                log_index: 7,
                amount_wei: "1000000".to_string(),
                token: WalletAddress::from(alloy::primitives::address!(
                    "cccccccccccccccccccccccccccccccccccccccc"
                )),
            });

        let event = LedgerUpdate {
            time: 1778789705303_u64,
            hash: test_event_hash(6),
            evm_tx_hash: None,
            delta: LedgerDelta {
                delta_type: "send".to_string(),
                usdc: None,
                amount: Some("1.0".to_string()),
                token: Some("USDC".to_string()),
                user: Some(exchange_addr().to_string()),
                destination: Some(exchange_addr().to_string()),
                evm_tx_hash: None,
            },
        };

        observer
            .handle_ledger_update(event, "websocket")
            .await
            .expect("replayed credited event should recover and submit the pending request id");

        assert_eq!(
            store
                .submitted_usdc_requests
                .lock()
                .expect("submitted_usdc_requests mutex poisoned")
                .as_slice(),
            &["usdc-request-recovered".to_string()]
        );
    }

    #[test]
    fn usdc_decimal_to_wei_string_requires_exact_six_decimals() {
        assert_eq!(
            usdc_decimal_to_wei_string(dec!(1.234567)).unwrap(),
            "1234567"
        );
        assert!(usdc_decimal_to_wei_string(dec!(1.0000001)).is_err());
    }

    #[test]
    fn config_from_runtime_includes_ws_url() {
        let runtime = crate::backend_config::OnchainDepositsRuntimeConfig {
            ws_url: Some("wss://hydromancer.example.com/ws".to_string()),
            ..Default::default()
        };
        let config = HypercoreCashLedgerObserverConfig::from_runtime_config(
            &runtime,
            "https://api.example.com/info",
            exchange_addr(),
            core_deposit_wallet_addr(),
        );
        assert_eq!(
            config.ws_url.as_deref(),
            Some("wss://hydromancer.example.com/ws")
        );
    }

    #[test]
    fn config_from_runtime_without_ws_url() {
        let runtime = crate::backend_config::OnchainDepositsRuntimeConfig::default();
        let config = HypercoreCashLedgerObserverConfig::from_runtime_config(
            &runtime,
            "https://api.example.com/info",
            exchange_addr(),
            core_deposit_wallet_addr(),
        );
        assert!(config.ws_url.is_none());
    }
}