opendeviationbar-streaming 13.78.1

//! StreamManager: managed streaming with rate-limited gap-fill (Issue #161).
//!
//! Wraps `LiveBarEngine` with:
//! - Proactive rate limiting via `AdaptiveRateLimiter` (Issue #162)
//! - Trade-ID continuity tracking seeded from ClickHouse
//! - Metrics for rate limiter utilization and gap-fill operations
//!
//! # Architecture
//!
//! ```text
//! Binance WS → LiveBarEngine → ring buffer → StreamManager.next_bar()
//!                    │                              │
//!              AdaptiveRateLimiter           trade-ID tracking
//!              (REST gap-fill pacing)        (ClickHouse seeds)
//! ```
//!
//! The StreamManager is a **thin orchestration layer** — it does NOT replace
//! LiveBarEngine's internal gap-fill. Instead, it:
//! 1. Seeds trade-ID tracking from ClickHouse `last_agg_trade_id` values
//! 2. Passes the rate limiter to the engine for gap-fill pacing
//! 3. Tracks per-symbol trade-ID continuity for diagnostic purposes

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use opendeviationbar_core::checkpoint::Checkpoint;
use opendeviationbar_providers::binance::{AdaptiveRateLimiter, HistoricalDataLoader};
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use crate::gap::{GapFillCommand, GapFillResult, GapFillSender};
use crate::live_engine::{
    CompletedBar, FormingBar, FormingBarWatches, LiveBarEngine, LiveEngineConfig, LiveEngineMetrics,
};

/// Configuration for the StreamManager.
#[derive(Debug, Clone)]
pub struct StreamManagerConfig {
    /// Underlying engine configuration.
    pub engine_config: LiveEngineConfig,
    /// Shared rate limiter for REST API calls.
    pub rate_limiter: Arc<AdaptiveRateLimiter>,
    /// Whether shadow validation is enabled (Phase 2, diagnostic only).
    pub shadow_validation_enabled: bool,
    /// Interval between shadow validation probes (seconds).
    pub shadow_interval_secs: u64,
    /// Maximum trades to backfill inline (larger gaps deferred to kintsugi).
    pub max_inline_gap_fill: i64,
    /// ClickHouse seed values: last_agg_trade_id per (symbol, threshold).
    /// Allows detecting gaps from the very first WS trade after startup.
    pub clickhouse_seeds: HashMap<(String, u32), i64>,
}

impl StreamManagerConfig {
    /// Create config with sensible defaults.
    pub fn new(engine_config: LiveEngineConfig, rate_limiter: Arc<AdaptiveRateLimiter>) -> Self {
        Self {
            engine_config,
            rate_limiter,
            shadow_validation_enabled: false,
            shadow_interval_secs: 5,
            max_inline_gap_fill: 100_000,
            clickhouse_seeds: HashMap::new(),
        }
    }

    /// Seed trade-ID tracking from ClickHouse for a (symbol, threshold) pair.
    pub fn with_seed(mut self, symbol: String, threshold: u32, last_ref_id: i64) -> Self {
        self.clickhouse_seeds
            .insert((symbol, threshold), last_ref_id);
        self
    }

    /// Enable shadow validation (Phase 2).
    pub fn with_shadow_validation(mut self, enabled: bool) -> Self {
        self.shadow_validation_enabled = enabled;
        self
    }
}

/// Metrics specific to the StreamManager layer.
#[derive(Debug, Default)]
pub struct StreamManagerMetrics {
    /// Bars where trade-ID continuity was verified.
    pub continuity_verified: AtomicU64,
    /// Bars where trade-ID gap was detected (gap > 0).
    pub continuity_gaps_detected: AtomicU64,
    /// Shadow validation probes executed.
    pub shadow_probes: AtomicU64,
    /// Shadow validation divergences detected.
    pub shadow_divergences: AtomicU64,
}

/// Managed streaming engine with rate limiting and trade-ID continuity.
///
/// Drop-in replacement for `LiveBarEngine` with additional orchestration.
pub struct StreamManager {
    engine: LiveBarEngine,
    rate_limiter: Arc<AdaptiveRateLimiter>,
    /// Per-symbol last contiguous trade ID (tracks continuity).
    /// Shared with shadow validation background task.
    trade_id_state: Arc<std::sync::Mutex<HashMap<Arc<str>, i64>>>,
    /// StreamManager-specific metrics.
    sm_metrics: Arc<StreamManagerMetrics>,
    /// Shadow validation cancellation token.
    shadow_cancel: Option<CancellationToken>,
    config: StreamManagerConfig,
    /// Issue #214: Watch receivers for forming bar snapshots.
    /// Extracted from engine before start() to avoid borrow conflicts with next_bar(&mut self).
    /// Stored here so Python can call get_forming_bars() concurrently with next_bar().
    forming_bar_watches: Option<FormingBarWatches>,
    /// Issue #257: Gap event receiver, extracted from engine for Python consumers.
    gap_event_rx: Option<tokio::sync::mpsc::Receiver<crate::GapEvent>>,
    /// Issue #257: Per-symbol gap-fill command senders for on-demand fill_gap().
    /// Cloneable and `&self`-safe — no borrow conflict with next_bar(&mut self).
    gap_fill_senders: HashMap<String, GapFillSender>,
    /// Issue #318: ClickHouse writer flush thread metrics (Plan 02).
    /// Present when `clickhouse-sink` feature is enabled AND OPENDEVIATIONBAR_CH_HOSTS was set.
    #[cfg(feature = "clickhouse-sink")]
    ch_writer_metrics: Option<Arc<crate::clickhouse_writer::flush_thread::FlushThreadMetrics>>,
}

impl StreamManager {
    /// Create a new StreamManager.
    ///
    /// Does NOT start streaming — call `start()` after creation.
    pub fn new(config: StreamManagerConfig) -> Self {
        // P0.1 (Issue #641): Per-threshold resume floors (never cross-threshold min/max).
        // Initialize gap_detector_seeds FROM clickhouse_seeds, taking per-threshold values ONLY.
        // Previously: took max() across all thresholds per symbol, which could drag stale
        // thresholds forward. Now: each threshold keeps its own resume point independent.
        let mut trade_id_state: HashMap<Arc<str>, i64> =
            HashMap::with_capacity(config.engine_config.symbol_thresholds.len());
        for ((symbol, _threshold), last_tid) in &config.clickhouse_seeds {
            let entry = trade_id_state
                .entry(Arc::from(symbol.as_str()))
                .or_insert(0);
            // Note: For gap_detector (per-symbol), we still take max() across thresholds.
            // This is correct: gap detection fires at trade-ID discontinuity (symbol-level).
            // The per-threshold resume validation happens in symbol_task via validate_seed_tid().
            *entry = (*entry).max(*last_tid);
        }

        let seeded_count = trade_id_state.len();
        let mut engine_config = config.engine_config.clone();
        engine_config.rate_limiter = Some(config.rate_limiter.clone());
        // Propagate CH seeds to LiveEngineConfig so symbol_task() can seed
        // each TradeIdGapDetector. Converts Arc<str> keys back to String.
        engine_config.gap_detector_seeds = trade_id_state
            .iter()
            .map(|(k, v)| (k.to_string(), *v))
            .collect();
        let mut engine = LiveBarEngine::new(engine_config);

        // P0.1: Pass per-threshold clickhouse_seeds to engine for per-threshold validation.
        // Previously: engine only had symbol-level seeds, couldn't validate per-threshold.
        engine.set_clickhouse_seeds(config.clickhouse_seeds.clone());

        // Issue #214: Extract forming bar watches before engine is started.
        // Stored separately so Python can read them without conflicting with
        // next_bar(&mut self) borrows.
        let forming_bar_watches = engine.take_forming_bar_watches();

        // Issue #257: Extract gap event receiver before engine is started.
        let gap_event_rx = engine.take_gap_event_receiver();
        // Issue #257: Get per-symbol gap-fill senders for on-demand fill_gap().
        let gap_fill_senders = engine.gap_fill_senders();

        // Issue #318 Plan 02: Extract CH writer metrics from engine for get_combined_metrics()
        #[cfg(feature = "clickhouse-sink")]
        let ch_writer_metrics = engine.ch_writer_metrics().cloned();

        let total_pairs: usize = config
            .engine_config
            .symbol_thresholds
            .values()
            .map(|v| v.len())
            .sum::<usize>();
        info!(
            symbols = config.engine_config.symbol_thresholds.len(),
            thresholds = total_pairs,
            seeded_symbols = seeded_count,
            rate_limiter_budget = config.rate_limiter.remaining(),
            forming_bar_watches = forming_bar_watches.as_ref().map_or(0, |w| w.len()),
            "StreamManager created"
        );

        Self {
            engine,
            rate_limiter: config.rate_limiter.clone(),
            trade_id_state: Arc::new(std::sync::Mutex::new(trade_id_state)),
            sm_metrics: Arc::new(StreamManagerMetrics::default()),
            shadow_cancel: None,
            config,
            forming_bar_watches,
            gap_event_rx,
            gap_fill_senders,
            #[cfg(feature = "clickhouse-sink")]
            ch_writer_metrics,
        }
    }

    /// Seed trade-ID continuity tracker for a (symbol, threshold) pair.
    ///
    /// Call before `start()`. The seeded value is the `last_agg_trade_id`
    /// from ClickHouse — the first WS trade triggers gap detection if its
    /// trade ID doesn't follow contiguously.
    pub fn seed_trade_id(&mut self, symbol: &str, _threshold: u32, last_ref_id: i64) {
        let mut state = self.trade_id_state.lock().unwrap();
        let entry = state.entry(Arc::from(symbol)).or_insert(0);
        if last_ref_id > *entry {
            *entry = last_ref_id;
        }
    }

    /// Start WebSocket connections only — buffer trades without processing (#286).
    ///
    /// Call this BEFORE loading checkpoints or seeding gap detectors.
    /// Trades buffer in mpsc channels (capacity 1000 per symbol).
    /// Then call `start()` to begin processing the buffered + live trades.
    pub fn start_ws(&mut self) {
        self.engine.start_ws();
    }

    /// Fill historical gap from REST before starting WS processing (#286).
    ///
    /// Creates processors, feeds REST trades, preserves state for `start()`.
    /// Call sequence: `start_ws()` → `fill_from_rest()` → `start()`
    pub async fn fill_from_rest(
        &mut self,
    ) -> Result<u64, opendeviationbar_core::processor::ProcessingError> {
        // Propagate runtime trade_id_state to engine so fill_from_rest
        // knows where to start REST pagination. seed_trade_id() updates
        // trade_id_state but not engine's gap_detector_seeds.
        {
            let state = self
                .trade_id_state
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let seeds: HashMap<String, i64> =
                state.iter().map(|(k, v)| (k.to_string(), *v)).collect();
            self.engine.set_gap_detector_seeds(seeds);
        }
        self.engine.fill_from_rest().await
    }

    /// Start the streaming engine (WebSocket connections + processing).
    ///
    /// If `start_ws()` was called first, reuses the already-buffered WS
    /// channels (zero-gap transition). Otherwise, creates WS connections inline.
    /// If shadow validation is enabled, spawns a background probe loop.
    /// Idempotent: calling start() twice is safe (shadow task guard).
    pub fn start(&mut self) -> Result<(), opendeviationbar_core::processor::ProcessingError> {
        self.engine.start()?;

        // Guard: only spawn shadow validation once (prevents duplicate tasks on double-start)
        if self.config.shadow_validation_enabled && self.shadow_cancel.is_none() {
            self.spawn_shadow_validation();
        }

        Ok(())
    }

    /// Spawn a background task that periodically fetches the latest REST trade
    /// and compares with WS-tracked trade IDs. Diagnostic only — divergences
    /// are logged and metriced, not auto-corrected.
    fn spawn_shadow_validation(&mut self) {
        let cancel = CancellationToken::new();
        self.shadow_cancel = Some(cancel.clone());

        let symbols = self.config.engine_config.symbols();
        let interval_secs = self.config.shadow_interval_secs;
        let rate_limiter = self.rate_limiter.clone();
        let sm_metrics = self.sm_metrics.clone();
        let trade_id_state = self.trade_id_state.clone();

        info!(
            symbols = symbols.len(),
            interval_secs, "Shadow validation loop started"
        );

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
            interval.tick().await; // skip first immediate tick

            loop {
                tokio::select! {
                    _ = cancel.cancelled() => {
                        info!("Shadow validation loop cancelled");
                        break;
                    }
                    _ = interval.tick() => {
                        // Adaptive skip: back off when rate limiter is under pressure
                        if rate_limiter.utilization() > 0.6 {
                            debug!("Shadow validation skipped: rate limiter utilization > 60%");
                            continue;
                        }

                        for symbol in &symbols {
                            let loader = HistoricalDataLoader::new(symbol)
                                .with_rate_limiter(rate_limiter.clone());

                            match loader.fetch_latest_aggtrade().await {
                                Ok(trade) => {
                                    sm_metrics.shadow_probes.fetch_add(1, Ordering::Relaxed);
                                    let rest_tid = trade.ref_id;

                                    let ws_tid = trade_id_state
                                        .lock()
                                        .unwrap_or_else(|e| e.into_inner())
                                        .get(symbol.as_str())
                                        .copied()
                                        .unwrap_or(0);

                                    if ws_tid > 0 && rest_tid > ws_tid + 100 {
                                        // REST is significantly ahead — WS may be lagging
                                        sm_metrics.shadow_divergences.fetch_add(1, Ordering::Relaxed);
                                        warn!(
                                            symbol = %symbol,
                                            rest_latest_tid = rest_tid,
                                            ws_latest_tid = ws_tid,
                                            drift = rest_tid - ws_tid,
                                            "Shadow validation: REST ahead of WS by {} trades",
                                            rest_tid - ws_tid
                                        );
                                    } else {
                                        debug!(
                                            symbol = %symbol,
                                            rest_tid,
                                            ws_tid,
                                            "Shadow validation: OK"
                                        );
                                    }
                                }
                                Err(e) => {
                                    debug!(
                                        symbol = %symbol,
                                        error = %e,
                                        "Shadow validation: REST fetch failed"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        });
    }

    /// Poll for the next completed bar with timeout.
    ///
    /// Same interface as `LiveBarEngine::next_bar()` but additionally:
    /// - Tracks trade-ID continuity per symbol
    /// - Updates StreamManager metrics
    pub async fn next_bar(&mut self, timeout: Duration) -> Option<CompletedBar> {
        let bar = self.engine.next_bar(timeout).await?;

        // Track trade-ID continuity
        let first_tid = bar.bar.first_agg_trade_id;
        let last_tid = bar.bar.last_agg_trade_id;

        if first_tid > 0 && last_tid > 0 {
            let mut state = self
                .trade_id_state
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            let prev_tid = state.get(&*bar.symbol).copied().unwrap_or(0);
            if prev_tid > 0 {
                let expected = prev_tid + 1;
                if first_tid == expected {
                    self.sm_metrics
                        .continuity_verified
                        .fetch_add(1, Ordering::Relaxed);
                } else if first_tid > expected {
                    let gap = first_tid - expected;
                    self.sm_metrics
                        .continuity_gaps_detected
                        .fetch_add(1, Ordering::Relaxed);
                    debug!(
                        symbol = %bar.symbol,
                        threshold = bar.threshold_decimal_bps,
                        expected_first_tid = expected,
                        actual_first_tid = first_tid,
                        gap_trades = gap,
                        "trade-ID continuity gap detected"
                    );
                }
                // Overlaps (first_tid < expected) are not flagged here —
                // ReplacingMergeTree handles dedup at ClickHouse level.
            }
            state.insert(bar.symbol.clone(), last_tid);
        }

        Some(bar)
    }

    /// Drain all completed bars from the ring buffer synchronously.
    ///
    /// Returns bars in FIFO order. No async, no polling.
    /// Used by sync_flush to reliably drain REST bars before start().
    pub fn drain_bars(&self) -> Vec<crate::live_engine::CompletedBar> {
        self.engine.drain_bars()
    }

    /// Drain all completed bars as an Arrow RecordBatch (#313, v7.2).
    ///
    /// Returns bars in FIFO order as a single RecordBatch with all bar fields
    /// plus `_symbol` (Utf8) and `_threshold_decimal_bps` (UInt32) metadata columns.
    /// Enables zero-copy Polars grouping in Python without per-bar dict allocation.
    ///
    /// Purely synchronous — no async, no block_on. Safe to call before start().
    #[cfg(feature = "arrow")]
    pub fn drain_bars_arrow(&self) -> arrow_array::RecordBatch {
        let bars = self.engine.drain_bars();
        crate::arrow_export::completed_bars_to_record_batch(&bars)
    }

    /// Graceful stop (signals all tasks to wind down).
    pub fn stop(&mut self) {
        if let Some(cancel) = self.shadow_cancel.take() {
            cancel.cancel();
        }
        self.engine.stop();
    }

    /// Collect processor checkpoints for state recovery.
    pub async fn collect_checkpoints(&mut self, timeout: Duration) -> HashMap<String, Checkpoint> {
        self.engine.collect_checkpoints(timeout).await
    }

    /// Get the shutdown token for graceful coordination.
    pub fn shutdown_token(&self) -> tokio_util::sync::CancellationToken {
        self.engine.shutdown_token()
    }

    /// Inject a processor checkpoint before start.
    pub fn set_initial_checkpoint(&mut self, symbol: &str, threshold: u32, checkpoint: Checkpoint) {
        self.engine
            .set_initial_checkpoint(symbol, threshold, checkpoint);
    }

    /// Engine-level metrics (trades, bars, reconnections, drops).
    pub fn engine_metrics(&self) -> &LiveEngineMetrics {
        self.engine.metrics()
    }

    /// StreamManager-level metrics (continuity tracking, shadow validation).
    pub fn stream_manager_metrics(&self) -> &Arc<StreamManagerMetrics> {
        &self.sm_metrics
    }

    /// Rate limiter utilization [0.0, 1.0+].
    pub fn rate_limiter_utilization(&self) -> f32 {
        self.rate_limiter.utilization()
    }

    /// Rate limiter remaining budget.
    pub fn rate_limiter_remaining(&self) -> u32 {
        self.rate_limiter.remaining()
    }

    /// Whether engine has been started.
    pub fn is_started(&self) -> bool {
        self.engine.is_started()
    }

    /// Take the forming bar watch receivers (Issue #214).
    ///
    /// Must be called BEFORE `start()`. Returns watch receivers for all
    /// (symbol, threshold) pairs. The caller (PyStreamManager) reads these
    /// at 1Hz to get the latest forming bar snapshot.
    pub fn take_forming_bar_watches(&mut self) -> Option<FormingBarWatches> {
        self.forming_bar_watches.take()
    }

    /// Take the gap event receiver (Issue #257).
    ///
    /// Returns the receiver for gap events emitted by the engine's
    /// `TradeIdGapDetector` instances. Can only be called once.
    pub fn take_gap_event_receiver(
        &mut self,
    ) -> Option<tokio::sync::mpsc::Receiver<crate::GapEvent>> {
        self.gap_event_rx.take()
    }

    /// Submit an on-demand gap-fill command (Issue #257).
    ///
    /// Uses `&self` — safe to call concurrently with `next_bar(&mut self)`.
    /// Sends a command to the symbol's task and awaits the fill result via oneshot.
    ///
    /// Returns `None` if the symbol is unknown or the command channel is closed.
    pub async fn fill_gap(
        &self,
        symbol: &str,
        from_tid: i64,
        to_tid: i64,
    ) -> Option<GapFillResult> {
        let sender = self.gap_fill_senders.get(symbol)?;
        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
        let cmd = GapFillCommand {
            symbol: Arc::from(symbol),
            from_tid,
            to_tid,
            response_tx,
        };
        sender.send(cmd).await.ok()?;
        response_rx.await.ok()
    }

    /// Get a gap-fill sender for a specific symbol (Issue #257).
    ///
    /// Returns a cloneable sender that can be used to submit gap-fill commands
    /// from any thread. Primarily used by PyO3 bindings.
    pub fn gap_fill_sender(&self, symbol: &str) -> Option<GapFillSender> {
        self.gap_fill_senders.get(symbol).cloned()
    }

    /// Read current forming bar snapshots from all watch receivers (Issue #214).
    ///
    /// Returns a Vec of the latest forming bar for each (symbol, threshold) pair
    /// where a forming bar exists (skips None entries). This is a non-blocking
    /// read via `watch::Receiver::borrow()`.
    ///
    /// NOTE: This requires &self access to the watches. If watches were taken
    /// via `take_forming_bar_watches()`, this returns an empty Vec — the caller
    /// should read from the taken watches directly.
    pub fn get_forming_bars(&self) -> Vec<FormingBar> {
        match &self.forming_bar_watches {
            Some(watches) => watches
                .values()
                .filter_map(|rx| rx.borrow().clone())
                .collect(),
            None => Vec::new(),
        }
    }

    /// Get a snapshot of combined metrics (engine + stream manager).
    pub fn get_combined_metrics(&self) -> HashMap<String, u64> {
        let engine = self.engine.metrics();
        let sm = &self.sm_metrics;
        let mut m = HashMap::new();
        m.insert(
            "trades_received".into(),
            engine.trades_received.load(Ordering::Relaxed),
        );
        m.insert(
            "bars_emitted".into(),
            engine.bars_emitted.load(Ordering::Relaxed),
        );
        m.insert(
            "reconnections".into(),
            engine.reconnections.load(Ordering::Relaxed),
        );
        m.insert(
            "dropped_bars".into(),
            engine.dropped_bars.load(Ordering::Relaxed),
        );
        m.insert("gap_fills".into(), engine.gap_fills.load(Ordering::Relaxed));
        m.insert(
            "gap_trades_recovered".into(),
            engine.gap_trades_recovered.load(Ordering::Relaxed),
        );
        m.insert(
            "bars_suppressed".into(),
            engine.bars_suppressed.load(Ordering::Relaxed),
        );
        m.insert(
            "trades_skipped_monotonicity".into(),
            engine.trades_skipped_monotonicity.load(Ordering::Relaxed),
        );
        m.insert(
            "continuity_verified".into(),
            sm.continuity_verified.load(Ordering::Relaxed),
        );
        m.insert(
            "continuity_gaps_detected".into(),
            sm.continuity_gaps_detected.load(Ordering::Relaxed),
        );
        m.insert(
            "shadow_probes".into(),
            sm.shadow_probes.load(Ordering::Relaxed),
        );
        m.insert(
            "shadow_divergences".into(),
            sm.shadow_divergences.load(Ordering::Relaxed),
        );
        m.insert(
            "rate_limiter_remaining".into(),
            self.rate_limiter.remaining() as u64,
        );

        // Issue #318 Plan 02: Include CH writer metrics when available
        #[cfg(feature = "clickhouse-sink")]
        if let Some(ref ch_metrics) = self.ch_writer_metrics {
            m.insert(
                "ch_writer_bars_flushed".into(),
                ch_metrics.bars_flushed.load(Ordering::Relaxed),
            );
            m.insert(
                "ch_writer_bars_failed".into(),
                ch_metrics.bars_failed.load(Ordering::Relaxed),
            );
            m.insert(
                "ch_writer_flush_count".into(),
                ch_metrics.flush_count.load(Ordering::Relaxed),
            );
            m.insert(
                "ch_writer_last_flush_latency_us".into(),
                ch_metrics.last_flush_latency_us.load(Ordering::Relaxed),
            );
            m.insert(
                "ch_writer_dead_letter_count".into(),
                ch_metrics.dead_letter_count.load(Ordering::Relaxed),
            );
            m.insert(
                "ch_writer_dead_letter_replayed".into(),
                ch_metrics.dead_letter_replayed.load(Ordering::Relaxed),
            );
        }

        m
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_providers::binance::shared_binance_limiter;

    fn sym_thresh(pairs: &[(&str, &[u32])]) -> HashMap<String, Vec<u32>> {
        let mut m = HashMap::new();
        for (sym, thrs) in pairs {
            m.insert((*sym).to_string(), thrs.to_vec());
        }
        m
    }

    #[test]
    fn test_config_defaults() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250, 500])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter);
        assert!(!config.shadow_validation_enabled);
        assert_eq!(config.shadow_interval_secs, 5);
        assert_eq!(config.max_inline_gap_fill, 100_000);
        assert!(config.clickhouse_seeds.is_empty());
    }

    #[test]
    fn test_config_with_seeds() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250, 500])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter)
            .with_seed("BTCUSDT".into(), 250, 1_000_000)
            .with_seed("BTCUSDT".into(), 500, 999_000);

        assert_eq!(config.clickhouse_seeds.len(), 2);
        assert_eq!(config.clickhouse_seeds[&("BTCUSDT".into(), 250)], 1_000_000);
    }

    #[test]
    fn test_stream_manager_creation_with_seeds() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250, 500])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter)
            .with_seed("BTCUSDT".into(), 250, 1_000_000)
            .with_seed("BTCUSDT".into(), 500, 999_000);

        let sm = StreamManager::new(config);
        // Should take max(1_000_000, 999_000) = 1_000_000 for BTCUSDT
        let state = sm.trade_id_state.lock().unwrap();
        assert_eq!(state.get("BTCUSDT" as &str), Some(&1_000_000));
        assert!(!sm.is_started());
    }

    #[test]
    fn test_combined_metrics_keys() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter);
        let sm = StreamManager::new(config);

        let metrics = sm.get_combined_metrics();
        assert!(metrics.contains_key("trades_received"));
        assert!(metrics.contains_key("continuity_verified"));
        assert!(metrics.contains_key("continuity_gaps_detected"));
        assert!(metrics.contains_key("rate_limiter_remaining"));
    }

    #[test]
    fn test_shadow_validation_config() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter).with_shadow_validation(true);
        assert!(config.shadow_validation_enabled);
        assert_eq!(config.shadow_interval_secs, 5);
    }

    #[cfg(feature = "clickhouse-sink")]
    #[test]
    fn test_combined_metrics_includes_ch_writer_keys_when_metrics_set() {
        // Issue #318: Verify ch_writer_* keys appear in get_combined_metrics()
        // when FlushThreadMetrics are available.
        use crate::clickhouse_writer::flush_thread::FlushThreadMetrics;
        use std::sync::atomic::Ordering;

        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter);
        let mut sm = StreamManager::new(config);

        // Manually inject FlushThreadMetrics to simulate CH writer being active
        let metrics = Arc::new(FlushThreadMetrics::default());
        metrics.bars_flushed.store(42, Ordering::Relaxed);
        metrics.bars_failed.store(3, Ordering::Relaxed);
        metrics.flush_count.store(10, Ordering::Relaxed);
        metrics.last_flush_latency_us.store(5000, Ordering::Relaxed);
        metrics.dead_letter_count.store(1, Ordering::Relaxed);
        metrics.dead_letter_replayed.store(7, Ordering::Relaxed);
        sm.ch_writer_metrics = Some(metrics);

        let combined = sm.get_combined_metrics();

        // Existing keys still present
        assert!(combined.contains_key("trades_received"));
        assert!(combined.contains_key("bars_emitted"));

        // CH writer keys present with correct values
        assert_eq!(combined.get("ch_writer_bars_flushed"), Some(&42));
        assert_eq!(combined.get("ch_writer_bars_failed"), Some(&3));
        assert_eq!(combined.get("ch_writer_flush_count"), Some(&10));
        assert_eq!(combined.get("ch_writer_last_flush_latency_us"), Some(&5000));
        assert_eq!(combined.get("ch_writer_dead_letter_count"), Some(&1));
        assert_eq!(combined.get("ch_writer_dead_letter_replayed"), Some(&7));
    }

    #[cfg(feature = "clickhouse-sink")]
    #[test]
    fn test_combined_metrics_no_ch_writer_keys_when_none() {
        // When ch_writer_metrics is explicitly None, ch_writer_* keys should NOT appear.
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter);
        let mut sm = StreamManager::new(config);

        // Force ch_writer_metrics to None (simulates no CH hosts configured)
        sm.ch_writer_metrics = None;

        let combined = sm.get_combined_metrics();

        // Existing keys still present
        assert!(combined.contains_key("trades_received"));

        // CH writer keys should NOT be present
        assert!(!combined.contains_key("ch_writer_bars_flushed"));
        assert!(!combined.contains_key("ch_writer_bars_failed"));
        assert!(!combined.contains_key("ch_writer_flush_count"));
    }

    #[test]
    fn test_rate_limiter_utilization() {
        let engine_config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let limiter = shared_binance_limiter();
        let config = StreamManagerConfig::new(engine_config, limiter);
        let sm = StreamManager::new(config);

        assert!(sm.rate_limiter_utilization() < f32::EPSILON);
        assert_eq!(sm.rate_limiter_remaining(), 4800); // 6000 * 0.8
    }
}