opendeviationbar-streaming 13.79.0

//! OdbEngine: unified source→verify→fan-out engine loop (Issue #178, Phase 2)

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

use tokio_util::sync::CancellationToken;

use super::traits::{BarSink, BarSource, EngineClock, SinkError};
use crate::live_engine::CompletedBar;

/// Metrics tracked by the engine loop.
#[derive(Debug, Default)]
pub struct OdbEngineMetrics {
    pub bars_produced: AtomicU64,
    pub bars_dropped: AtomicU64,
    pub trade_id_gaps: AtomicU64,
    pub trade_id_verified: AtomicU64,
}

/// Unified engine: source → Stathera verification → fan-out to sinks.
///
/// Same `run()` loop works for live AND backtest — swap the source, not the engine.
/// Stathera verification happens ONCE in the engine, not per-consumer.
/// Each sink is fault-isolated — ClickHouse outage doesn't block strategy.
pub struct OdbEngine {
    source: Box<dyn BarSource>,
    sinks: Vec<Box<dyn BarSink>>,
    clock: Box<dyn EngineClock>,
    shutdown: CancellationToken,
    /// Per-symbol last contiguous trade ID for Stathera verification.
    trade_id_state: HashMap<Arc<str>, i64>,
    /// Per-symbol last bar timestamp for Argus watchdog metrics.
    last_bar_time_ms: HashMap<Arc<str>, i64>,
    metrics: Arc<OdbEngineMetrics>,
    poll_timeout: Duration,
}

impl OdbEngine {
    pub fn new(
        source: Box<dyn BarSource>,
        sinks: Vec<Box<dyn BarSink>>,
        clock: Box<dyn EngineClock>,
        shutdown: CancellationToken,
    ) -> Self {
        Self {
            source,
            sinks,
            clock,
            shutdown,
            trade_id_state: HashMap::new(),
            last_bar_time_ms: HashMap::new(),
            metrics: Arc::new(OdbEngineMetrics::default()),
            poll_timeout: Duration::from_secs(5),
        }
    }

    /// Set the poll timeout for `source.next_bar()`.
    pub fn with_poll_timeout(mut self, timeout: Duration) -> Self {
        self.poll_timeout = timeout;
        self
    }

    /// Seed trade-ID continuity tracker from ClickHouse.
    pub fn seed_trade_id(&mut self, symbol: &str, last_ref_id: i64) {
        let entry = self.trade_id_state.entry(Arc::from(symbol)).or_insert(0);
        if last_ref_id > *entry {
            *entry = last_ref_id;
        }
    }

    /// Get engine metrics.
    pub fn metrics(&self) -> &Arc<OdbEngineMetrics> {
        &self.metrics
    }

    /// Get last bar time per symbol (for Argus watchdog).
    pub fn last_bar_times(&self) -> &HashMap<Arc<str>, i64> {
        &self.last_bar_time_ms
    }

    /// Get combined metrics as a flat map (for PyO3 export).
    pub fn get_combined_metrics(&self) -> HashMap<String, u64> {
        let mut m = HashMap::new();
        m.insert(
            "bars_produced".into(),
            self.metrics.bars_produced.load(Ordering::Relaxed),
        );
        m.insert(
            "bars_dropped".into(),
            self.metrics.bars_dropped.load(Ordering::Relaxed),
        );
        m.insert(
            "trade_id_gaps".into(),
            self.metrics.trade_id_gaps.load(Ordering::Relaxed),
        );
        m.insert(
            "trade_id_verified".into(),
            self.metrics.trade_id_verified.load(Ordering::Relaxed),
        );
        m
    }

    /// Run the engine loop. Blocks until shutdown is signalled or source is exhausted.
    pub async fn run(&mut self) {
        tracing::info!(sinks = self.sinks.len(), "OdbEngine starting");

        loop {
            tokio::select! {
                bar = self.source.next_bar(self.poll_timeout) => {
                    match bar {
                        Some(bar) => {
                            self.verify_stathera(&bar);
                            self.update_watchdog(&bar);
                            self.metrics.bars_produced.fetch_add(1, Ordering::Relaxed);
                            self.fan_out(&bar);
                        }
                        None => {
                            // For live sources, None means timeout — continue polling.
                            // Check shutdown to avoid busy-looping after source exhaustion.
                            if self.shutdown.is_cancelled() {
                                break;
                            }
                        }
                    }
                }
                () = self.shutdown.cancelled() => {
                    tracing::info!("OdbEngine shutdown requested");
                    break;
                }
            }
        }

        for sink in &mut self.sinks {
            if let Err(e) = sink.flush() {
                tracing::warn!(sink = sink.name(), ?e, "sink flush error on shutdown");
            }
        }

        tracing::info!(
            bars_produced = self.metrics.bars_produced.load(Ordering::Relaxed),
            bars_dropped = self.metrics.bars_dropped.load(Ordering::Relaxed),
            "OdbEngine stopped"
        );
    }

    /// Verify Stathera trade-ID continuity for this bar.
    fn verify_stathera(&mut self, bar: &CompletedBar) {
        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 {
            return;
        }

        let prev_tid = self.trade_id_state.get(&*bar.symbol).copied().unwrap_or(0);
        if prev_tid > 0 {
            let expected = prev_tid + 1;
            if first_tid == expected {
                self.metrics
                    .trade_id_verified
                    .fetch_add(1, Ordering::Relaxed);
            } else if first_tid > expected {
                let gap = first_tid - expected;
                self.metrics.trade_id_gaps.fetch_add(1, Ordering::Relaxed);
                tracing::debug!(
                    symbol = %bar.symbol,
                    threshold = bar.threshold_decimal_bps,
                    expected_first_tid = expected,
                    actual_first_tid = first_tid,
                    gap_trades = gap,
                    "Stathera: trade-ID continuity gap"
                );
            }
            // Overlaps (first_tid < expected) not flagged — ReplacingMergeTree handles dedup.
        }
        self.trade_id_state.insert(bar.symbol.clone(), last_tid);
    }

    /// Update per-symbol last bar time for Argus watchdog.
    fn update_watchdog(&mut self, bar: &CompletedBar) {
        let bar_time_ms = self.clock.now_ms();
        self.last_bar_time_ms
            .insert(bar.symbol.clone(), bar_time_ms);
    }

    /// Fan-out bar to all sinks with fault isolation.
    fn fan_out(&mut self, bar: &CompletedBar) {
        for sink in &mut self.sinks {
            match sink.on_bar(bar) {
                Ok(()) => {}
                Err(SinkError::Recoverable(msg)) => {
                    self.metrics.bars_dropped.fetch_add(1, Ordering::Relaxed);
                    tracing::warn!(sink = sink.name(), %msg, "sink recoverable error");
                }
                Err(SinkError::Unrecoverable(msg)) => {
                    tracing::error!(sink = sink.name(), %msg, "sink unrecoverable error");
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::engine::clock::LiveClock;
    use opendeviationbar_core::{FixedPoint, OpenDeviationBar, Tick};
    use std::sync::Arc;

    /// Mock BarSource that yields bars from a Vec.
    struct VecBarSource {
        bars: std::collections::VecDeque<CompletedBar>,
    }

    impl VecBarSource {
        fn new(bars: Vec<CompletedBar>) -> Self {
            Self { bars: bars.into() }
        }
    }

    #[async_trait::async_trait]
    impl BarSource for VecBarSource {
        async fn next_bar(&mut self, timeout: Duration) -> Option<CompletedBar> {
            match self.bars.pop_front() {
                Some(bar) => Some(bar),
                None => {
                    // Mimic LiveBarEngine: sleep for timeout when no bars available.
                    // This prevents busy-looping and allows tokio::select! to pick
                    // the shutdown branch.
                    tokio::time::sleep(timeout).await;
                    None
                }
            }
        }
        fn snapshot(&self) -> Option<super::super::traits::SourceCheckpoint> {
            None
        }
    }

    /// Recording sink that captures all bars.
    struct RecordingSink {
        bars: Vec<CompletedBar>,
    }

    impl RecordingSink {
        fn new() -> Self {
            Self { bars: Vec::new() }
        }
    }

    impl BarSink for RecordingSink {
        fn on_bar(&mut self, bar: &CompletedBar) -> Result<(), SinkError> {
            self.bars.push(bar.clone());
            Ok(())
        }
        fn flush(&mut self) -> Result<(), SinkError> {
            Ok(())
        }
        fn name(&self) -> &str {
            "recording"
        }
    }

    fn make_completed_bar(symbol: &str, threshold: u32, tid: i64, ts_ms: u64) -> CompletedBar {
        let trade = Tick {
            ref_id: tid,
            price: FixedPoint::from_str("50000.0").unwrap(),
            volume: FixedPoint::from_str("1.0").unwrap(),
            first_sub_id: tid,
            last_sub_id: tid,
            timestamp: opendeviationbar_core::normalize_timestamp(ts_ms),
            is_buyer_maker: false,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        };
        CompletedBar {
            symbol: Arc::from(symbol),
            threshold_decimal_bps: threshold,
            bar: OpenDeviationBar::new(&trade),
        }
    }

    #[tokio::test]
    async fn test_engine_processes_all_bars() {
        let bars = vec![
            make_completed_bar("BTCUSDT", 250, 1, 1_700_000_000_000),
            make_completed_bar("BTCUSDT", 250, 2, 1_700_000_001_000),
            make_completed_bar("ETHUSDT", 500, 1, 1_700_000_002_000),
        ];

        let source = VecBarSource::new(bars);
        let shutdown = CancellationToken::new();
        let shutdown_clone = shutdown.clone();

        // Shutdown after source is drained (VecBarSource returns None)
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            shutdown_clone.cancel();
        });

        let recording = RecordingSink::new();
        let recording_ptr = &recording as *const RecordingSink as usize;
        let _ = recording_ptr; // unused, we'll check metrics instead

        let mut engine = OdbEngine::new(
            Box::new(source),
            vec![Box::new(recording)],
            Box::new(LiveClock),
            shutdown,
        )
        .with_poll_timeout(Duration::from_millis(50));

        engine.run().await;
        assert_eq!(engine.metrics().bars_produced.load(Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn test_stathera_gap_detection() {
        // Bar 1: tid=100, Bar 2: tid=200 (gap of 99)
        let bars = vec![
            make_completed_bar("BTCUSDT", 250, 100, 1_700_000_000_000),
            make_completed_bar("BTCUSDT", 250, 200, 1_700_000_001_000),
        ];

        let source = VecBarSource::new(bars);
        let shutdown = CancellationToken::new();
        let shutdown_clone = shutdown.clone();

        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            shutdown_clone.cancel();
        });

        let mut engine = OdbEngine::new(Box::new(source), vec![], Box::new(LiveClock), shutdown)
            .with_poll_timeout(Duration::from_millis(50));

        engine.run().await;
        assert_eq!(engine.metrics().trade_id_gaps.load(Ordering::Relaxed), 1);
        assert_eq!(
            engine.metrics().trade_id_verified.load(Ordering::Relaxed),
            0
        );
    }

    #[tokio::test]
    async fn test_stathera_continuity_verified() {
        // Contiguous: tid=100 then tid=101
        let bars = vec![
            make_completed_bar("BTCUSDT", 250, 100, 1_700_000_000_000),
            make_completed_bar("BTCUSDT", 250, 101, 1_700_000_001_000),
        ];

        let source = VecBarSource::new(bars);
        let shutdown = CancellationToken::new();
        let shutdown_clone = shutdown.clone();

        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            shutdown_clone.cancel();
        });

        let mut engine = OdbEngine::new(Box::new(source), vec![], Box::new(LiveClock), shutdown)
            .with_poll_timeout(Duration::from_millis(50));

        engine.run().await;
        assert_eq!(
            engine.metrics().trade_id_verified.load(Ordering::Relaxed),
            1
        );
        assert_eq!(engine.metrics().trade_id_gaps.load(Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn test_sink_fault_isolation() {
        /// Sink that always fails.
        struct FailingSink;
        impl BarSink for FailingSink {
            fn on_bar(&mut self, _bar: &CompletedBar) -> Result<(), SinkError> {
                Err(SinkError::Recoverable("test failure".into()))
            }
            fn flush(&mut self) -> Result<(), SinkError> {
                Ok(())
            }
            fn name(&self) -> &str {
                "failing"
            }
        }

        let bars = vec![make_completed_bar("BTCUSDT", 250, 1, 1_700_000_000_000)];
        let source = VecBarSource::new(bars);
        let shutdown = CancellationToken::new();
        let shutdown_clone = shutdown.clone();

        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            shutdown_clone.cancel();
        });

        let recording = RecordingSink::new();
        let mut engine = OdbEngine::new(
            Box::new(source),
            vec![Box::new(FailingSink), Box::new(recording)],
            Box::new(LiveClock),
            shutdown,
        )
        .with_poll_timeout(Duration::from_millis(50));

        engine.run().await;
        // Engine processed the bar despite the failing sink
        assert_eq!(engine.metrics().bars_produced.load(Ordering::Relaxed), 1);
        // The failing sink caused a drop
        assert_eq!(engine.metrics().bars_dropped.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn test_seed_trade_id() {
        let bars = vec![make_completed_bar("BTCUSDT", 250, 1001, 1_700_000_000_000)];
        let source = VecBarSource::new(bars);
        let shutdown = CancellationToken::new();
        let shutdown_clone = shutdown.clone();

        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            shutdown_clone.cancel();
        });

        let mut engine = OdbEngine::new(Box::new(source), vec![], Box::new(LiveClock), shutdown)
            .with_poll_timeout(Duration::from_millis(50));

        // Seed from ClickHouse: last known tid=1000
        engine.seed_trade_id("BTCUSDT", 1000);

        engine.run().await;
        // 1001 follows 1000+1 = contiguous
        assert_eq!(
            engine.metrics().trade_id_verified.load(Ordering::Relaxed),
            1
        );
    }
}