opendeviationbar-streaming 13.79.0

//! TradeIdGapDetector: extracted gap-detection logic (Issue #257).
//!
//! Previously embedded in `live_engine.rs` symbol_task (~lines 722-744).
//! Extracted for independent unit testing with mock trade sequences.

use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::debug;

use super::event::GapEvent;

/// Detects trade-ID gaps in a stream of aggregate trades.
///
/// Maintains per-detector state (last known trade ID) and emits
/// `GapEvent`s through an optional channel when gaps are detected.
pub struct TradeIdGapDetector {
    /// Symbol this detector is tracking.
    symbol: Arc<str>,
    /// Last contiguous trade ID seen.
    last_known_tid: Option<i64>,
    /// Channel for emitting gap events (None if no consumer).
    gap_event_tx: Option<mpsc::Sender<GapEvent>>,
    /// Total gaps detected (for metrics).
    gaps_detected: u64,
}

impl TradeIdGapDetector {
    /// Create a new detector for the given symbol.
    pub fn new(symbol: Arc<str>, gap_event_tx: Option<mpsc::Sender<GapEvent>>) -> Self {
        Self {
            symbol,
            last_known_tid: None,
            gap_event_tx,
            gaps_detected: 0,
        }
    }

    /// Seed with a known last trade ID (e.g., from ClickHouse or checkpoint).
    pub fn seed(&mut self, last_tid: i64) {
        self.last_known_tid = Some(last_tid);
    }

    /// Check an incoming trade ID for gaps.
    ///
    /// Returns `Some((last_known_id, gap_size))` if a gap is detected,
    /// `None` if the trade is contiguous or this is the first trade.
    ///
    /// Always updates the internal state to track the incoming ID.
    pub fn check(&mut self, incoming_tid: i64) -> Option<(i64, i64)> {
        // Note: we don't update last_known_tid here -- the caller should call
        // update_last_tid() after processing the trade (including any gap-fill trades)
        let last_id = self.last_known_tid?;
        if incoming_tid <= last_id + 1 {
            return None;
        }

        let gap_size = incoming_tid - last_id - 1;
        self.gaps_detected += 1;

        // Emit gap event (non-blocking, best-effort)
        if let Some(ref tx) = self.gap_event_tx {
            let event = GapEvent {
                symbol: self.symbol.clone(),
                expected_tid: last_id + 1,
                actual_tid: incoming_tid,
                gap_size,
                detected_at_ms: chrono::Utc::now().timestamp_millis(),
                filled_inline: false, // Updated later if fill succeeds
            };
            // try_send: don't block the hot path if channel is full
            if tx.try_send(event).is_err() {
                debug!(
                    symbol = %self.symbol,
                    gap_size,
                    "gap event channel full, event dropped"
                );
            }
        }

        Some((last_id, gap_size))
    }

    /// Update the last known trade ID after processing.
    ///
    /// Called after a trade (or batch of gap-fill trades) has been fully
    /// processed through all threshold processors.
    pub fn update_last_tid(&mut self, tid: i64) {
        self.last_known_tid = Some(tid);
    }

    /// Mark the most recently emitted gap event as filled inline.
    ///
    /// Called after `puck_fill()` successfully recovers trades.
    /// Sends an updated `GapEvent` with `filled_inline = true`.
    pub fn mark_filled(&self, last_known_id: i64, gap_size: i64) {
        if let Some(ref tx) = self.gap_event_tx {
            let event = GapEvent {
                symbol: self.symbol.clone(),
                expected_tid: last_known_id + 1,
                actual_tid: last_known_id + gap_size + 1,
                gap_size,
                detected_at_ms: chrono::Utc::now().timestamp_millis(),
                filled_inline: true,
            };
            let _ = tx.try_send(event);
        }
    }

    /// Total gaps detected by this detector.
    pub fn gaps_detected(&self) -> u64 {
        self.gaps_detected
    }

    /// Current last known trade ID.
    pub fn last_known_tid(&self) -> Option<i64> {
        self.last_known_tid
    }
}

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

    #[test]
    fn test_first_trade_no_gap() {
        let (tx, _rx) = mpsc::channel(16);
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));
        // First trade should not detect a gap (no prior state)
        assert!(detector.check(100).is_none());
    }

    #[test]
    fn test_contiguous_trades_no_gap() {
        let (tx, _rx) = mpsc::channel(16);
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));
        detector.update_last_tid(100);
        // Next trade is contiguous
        assert!(detector.check(101).is_none());
    }

    #[test]
    fn test_gap_detected() {
        let (tx, mut rx) = mpsc::channel(16);
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));
        detector.update_last_tid(100);

        // Trade 105 arrives, gap of 4 trades (101, 102, 103, 104)
        let result = detector.check(105);
        assert!(result.is_some());
        let (last_id, gap_size) = result.unwrap();
        assert_eq!(last_id, 100);
        assert_eq!(gap_size, 4);
        assert_eq!(detector.gaps_detected(), 1);

        // Verify gap event was emitted
        let event = rx.try_recv().unwrap();
        assert_eq!(&*event.symbol, "BTCUSDT");
        assert_eq!(event.expected_tid, 101);
        assert_eq!(event.actual_tid, 105);
        assert_eq!(event.gap_size, 4);
        assert!(!event.filled_inline);
    }

    #[test]
    fn test_seeded_gap_detection() {
        let (tx, mut rx) = mpsc::channel(16);
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));
        detector.seed(1000);

        // First WS trade after seed with a gap
        let result = detector.check(1010);
        assert!(result.is_some());
        let (last_id, gap_size) = result.unwrap();
        assert_eq!(last_id, 1000);
        assert_eq!(gap_size, 9);

        let event = rx.try_recv().unwrap();
        assert_eq!(event.expected_tid, 1001);
    }

    #[test]
    fn test_mark_filled() {
        let (tx, mut rx) = mpsc::channel(16);
        let detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));

        detector.mark_filled(100, 5);

        let event = rx.try_recv().unwrap();
        assert!(event.filled_inline);
        assert_eq!(event.gap_size, 5);
    }

    #[test]
    fn test_no_channel_still_detects() {
        // Without a channel, gaps are still detected (for metrics)
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), None);
        detector.update_last_tid(100);

        let result = detector.check(105);
        assert!(result.is_some());
        assert_eq!(detector.gaps_detected(), 1);
    }

    #[test]
    fn test_multi_gap_counting() {
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), None);
        detector.update_last_tid(100);

        assert!(detector.check(110).is_some());
        detector.update_last_tid(110);

        assert!(detector.check(120).is_some());
        detector.update_last_tid(120);

        assert_eq!(detector.gaps_detected(), 2);
    }

    #[test]
    fn test_overlap_not_gap() {
        // Overlapping trade IDs (first_tid < expected) should NOT be flagged as gaps
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), None);
        detector.update_last_tid(100);

        // Trade 90 arrives (older/duplicate) — not a gap
        assert!(detector.check(90).is_none());
        assert_eq!(detector.gaps_detected(), 0);
    }

    #[test]
    fn test_single_trade_gap() {
        let (tx, mut rx) = mpsc::channel(16);
        let mut detector = TradeIdGapDetector::new(Arc::from("BTCUSDT"), Some(tx));
        detector.update_last_tid(100);

        // Exactly 1 trade missing (101)
        let result = detector.check(102);
        assert!(result.is_some());
        let (_, gap_size) = result.unwrap();
        assert_eq!(gap_size, 1);

        let event = rx.try_recv().unwrap();
        assert_eq!(event.gap_size, 1);
    }
}