opendeviationbar-streaming 13.78.1

//! Trade dispatch: fan-out of trades to threshold processors with midnight reset.

use super::types::*;
use crate::ring_buffer::ConcurrentRingBuffer;
use opendeviationbar_core::processor::OpenDeviationBarProcessor;
use opendeviationbar_core::{OpenDeviationBar, Tick};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::sync::watch;

/// Type alias for the shared sink collection used in fan-out dispatch (Issue #318).
///
/// When `clickhouse-sink` feature is enabled, this is `Some(Arc<Mutex<Vec<Box<dyn BarSink>>>>)`.
/// When disabled, callers pass `None`. The type is always present to avoid
/// duplicating function signatures with `#[cfg]`.
#[cfg(feature = "clickhouse-sink")]
pub(crate) type ExtraSinks =
    Option<std::sync::Arc<std::sync::Mutex<Vec<Box<dyn crate::engine::traits::BarSink>>>>>;
#[cfg(not(feature = "clickhouse-sink"))]
pub(crate) type ExtraSinks = Option<()>;

/// Process a single trade through all threshold processors, handling midnight resets,
/// bar emission to ring buffer, forming bar updates, and metrics.
///
/// Extracted from the symbol_task loop to enable burst-atomic frame processing (Issue #273).
///
/// Issue #318: When `clickhouse-sink` feature is enabled, completed bars are also
/// dispatched to `extra_sinks` (e.g., ClickHouseWriterSink) AFTER the ring buffer push.
/// Sink errors never block ring buffer delivery to Python.
#[allow(clippy::too_many_arguments)]
pub(crate) fn process_trade_through_processors(
    trade: &Tick,
    processors: &mut [(u32, OpenDeviationBarProcessor)],
    last_days: &mut [i64],
    last_trade_timestamps_us: &mut [i64], // Phase 52: per-processor last trade timestamp for Week gap detection
    bar_buffer: &ConcurrentRingBuffer<CompletedBar>,
    metrics: &LiveEngineMetrics,
    symbol_arc: &Arc<str>,
    symbol: &str,
    forming_tx_map: &HashMap<u32, watch::Sender<Option<FormingBar>>>,
    committed_floors: &mut HashMap<u32, i64>,
    last_forming_update: &mut HashMap<u32, i64>, // Issue #300 (MEM-08): per-threshold 1Hz throttle
    ouroboros_mode: OuroborosMode,               // Phase 16: per-symbol mode (Day or Aion)
    extra_sinks: &ExtraSinks,                    // Issue #318: fan-out to BarSink implementations
) {
    let _ = &extra_sinks; // Suppress unused warning when clickhouse-sink disabled
    for (idx, (threshold, processor)) in processors.iter_mut().enumerate() {
        // Week gap check: if inter-trade gap exceeds max_gap_us, emit orphan and reset.
        // Runs BEFORE midnight check. Week mode also skips midnight resets (handled in maybe_reset_at_midnight).
        if let OuroborosMode::Week { max_gap_us } = ouroboros_mode
            && let Some(orphan) = maybe_reset_at_week_gap(
                processor,
                trade.timestamp,
                &mut last_trade_timestamps_us[idx],
                max_gap_us,
            )
        {
            // Rust-side dedup: skip orphan bars within already-committed range
            if let Some(&floor) = committed_floors.get(threshold)
                && orphan.last_agg_trade_id > 0
                && orphan.last_agg_trade_id <= floor
            {
                tracing::debug!(%symbol, threshold, last_tid = orphan.last_agg_trade_id, floor, "rust-dedup: skipping week-gap orphan (within committed range)");
                continue;
            }
            metrics.bars_emitted.fetch_add(1, Ordering::Relaxed);
            let completed = CompletedBar {
                symbol: symbol_arc.clone(),
                threshold_decimal_bps: *threshold,
                bar: orphan,
            };
            // Update floor after emission
            committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
            let was_added = bar_buffer.push(completed.clone());
            if !was_added {
                metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                tracing::warn!(%symbol, threshold, "ring buffer full, old bar dropped (week-gap orphan)");
            }
            // Issue #318: Fan-out to extra sinks (ClickHouseWriterSink) AFTER ring buffer push
            #[cfg(feature = "clickhouse-sink")]
            if let Some(sinks) = extra_sinks {
                crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
            }
        }

        // Midnight reset: if trade crosses day boundary, emit orphan and reset
        if let Some(orphan) = maybe_reset_at_midnight(
            processor,
            trade.timestamp,
            &mut last_days[idx],
            ouroboros_mode,
        ) {
            // Rust-side dedup: skip orphan bars within already-committed range
            if let Some(&floor) = committed_floors.get(threshold)
                && orphan.last_agg_trade_id > 0
                && orphan.last_agg_trade_id <= floor
            {
                tracing::debug!(%symbol, threshold, last_tid = orphan.last_agg_trade_id, floor, "rust-dedup: skipping orphan (within committed range)");
                continue;
            }
            metrics.bars_emitted.fetch_add(1, Ordering::Relaxed);
            let completed = CompletedBar {
                symbol: symbol_arc.clone(),
                threshold_decimal_bps: *threshold,
                bar: orphan,
            };
            // Update floor after emission
            committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
            let was_added = bar_buffer.push(completed.clone());
            if !was_added {
                metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                tracing::warn!(%symbol, threshold, "ring buffer full, old bar dropped (midnight orphan)");
            }
            // Issue #318: Fan-out to extra sinks (ClickHouseWriterSink) AFTER ring buffer push
            #[cfg(feature = "clickhouse-sink")]
            if let Some(sinks) = extra_sinks {
                crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
            }
        }

        match processor.process_single_trade(trade) {
            Ok(Some(bar)) => {
                // Rust-side dedup: skip bars within already-committed range
                if let Some(&floor) = committed_floors.get(threshold)
                    && bar.last_agg_trade_id > 0
                    && bar.last_agg_trade_id <= floor
                {
                    tracing::warn!(
                        %symbol, threshold,
                        first_tid = bar.first_agg_trade_id,
                        last_tid = bar.last_agg_trade_id,
                        floor,
                        "rust-dedup: SUPPRESSING bar (last_tid <= committed floor) — #345 telemetry"
                    );
                    metrics.bars_suppressed.fetch_add(1, Ordering::Relaxed);
                    // Still clear forming bar state
                    if let Some(tx) = forming_tx_map.get(threshold) {
                        let _ = tx.send(None);
                    }
                    continue;
                }

                // Issue #214: Clear forming bar BEFORE emitting completed bar
                if let Some(tx) = forming_tx_map.get(threshold) {
                    let _ = tx.send(None);
                }

                metrics.bars_emitted.fetch_add(1, Ordering::Relaxed);
                let completed = CompletedBar {
                    symbol: symbol_arc.clone(),
                    threshold_decimal_bps: *threshold,
                    bar,
                };
                // #345 telemetry: detect Stathera gap at emission time
                if let Some(&floor) = committed_floors.get(threshold) {
                    let expected_first = floor + 1;
                    if completed.bar.first_agg_trade_id > expected_first
                        && completed.bar.first_agg_trade_id > 0
                        && floor > 0
                    {
                        let gap = completed.bar.first_agg_trade_id - expected_first;
                        tracing::warn!(
                            %symbol, threshold,
                            expected_first,
                            actual_first = completed.bar.first_agg_trade_id,
                            prev_last = floor,
                            gap,
                            bar_last = completed.bar.last_agg_trade_id,
                            bar_trades = completed.bar.last_agg_trade_id - completed.bar.first_agg_trade_id + 1,
                            "STATHERA-GAP-AT-EMISSION: bar first_tid != prev_last+1 — trades missing from processor"
                        );
                    }
                }
                // #345 telemetry: log every @100 bar emission for trace analysis
                if *threshold == 100 {
                    tracing::info!(
                        %symbol,
                        first = completed.bar.first_agg_trade_id,
                        last = completed.bar.last_agg_trade_id,
                        trades = completed.bar.last_agg_trade_id - completed.bar.first_agg_trade_id + 1,
                        "BAR100-EMIT"
                    );
                }
                // Update floor after emission
                committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
                let was_added = bar_buffer.push(completed.clone());
                if !was_added {
                    metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                    metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                    tracing::warn!(%symbol, threshold, "ring buffer full, old bar dropped");
                }
                // Issue #318: Fan-out to extra sinks AFTER ring buffer push
                #[cfg(feature = "clickhouse-sink")]
                if let Some(sinks) = extra_sinks {
                    crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
                }
                let current_depth = bar_buffer.len() as u64;
                let _ = metrics
                    .max_queue_depth
                    .fetch_max(current_depth, Ordering::Relaxed);

                // Issue #300 (MEM-08): Reset forming bar throttle -- new bar started,
                // send forming state immediately on next trade.
                last_forming_update.insert(*threshold, 0);
            }
            Ok(None) => {
                // Trade absorbed, no bar completed yet.
                // Issue #214 + MEM-08: Update forming bar snapshot for consumers.
                // Throttled to 1 Hz -- consumers poll at 1s intervals, so 99.3% of
                // updates are immediately overwritten by watch::send().
                let last_us = last_forming_update.entry(*threshold).or_insert(0);
                if trade.timestamp - *last_us >= 1_000_000 {
                    // 1 second in microseconds
                    if let Some(tx) = forming_tx_map.get(threshold)
                        && let Some(incomplete) = processor.get_incomplete_bar()
                    {
                        let _ = tx.send(Some(FormingBar {
                            symbol: symbol_arc.clone(),
                            threshold_decimal_bps: *threshold,
                            bar: incomplete,
                            last_trade_timestamp_us: trade.timestamp,
                        }));
                        *last_us = trade.timestamp;
                    }
                }
            }
            Err(e) => {
                tracing::warn!(%symbol, threshold = *threshold, ?e, "trade processing error");
            }
        }
    }
}

/// Check if a trade crosses a UTC midnight boundary relative to the processor's last trade,
/// and if so, reset the processor at ouroboros. Returns the orphan bar if one was in progress.
///
/// This prevents cross-midnight bars from being produced. The orphan bar (if any) is emitted
/// as a completed bar with `is_orphan=true`. The trade that triggered the reset is NOT consumed --
/// it will be processed normally after this function returns.
pub(crate) fn maybe_reset_at_midnight(
    processor: &mut OpenDeviationBarProcessor,
    trade_timestamp_us: i64,
    last_day: &mut i64,
    ouroboros_mode: OuroborosMode,
) -> Option<OpenDeviationBar> {
    // Aion and Week modes: never reset at midnight -- bars span across day boundaries.
    // Week mode uses its own gap-based check (maybe_reset_at_week_gap), not midnight.
    if matches!(
        ouroboros_mode,
        OuroborosMode::Aion | OuroborosMode::Week { .. }
    ) {
        // Still track the day for metrics/logging, but never trigger reset
        let trade_day = trade_timestamp_us / DAY_US;
        *last_day = trade_day;
        return None;
    }

    // Day mode: existing behavior unchanged
    let trade_day = trade_timestamp_us / DAY_US;
    if *last_day >= 0 && trade_day != *last_day {
        *last_day = trade_day;
        // Reset processor -- any in-progress bar becomes an orphan
        processor.reset_at_ouroboros()
    } else {
        *last_day = trade_day;
        None
    }
}

/// Check if the gap between consecutive trades exceeds the week-mode threshold,
/// and if so, reset the processor at ouroboros. Returns the orphan bar if one was in progress.
///
/// Reserved for modes where intentional data voids create multi-hour gaps.
/// When gap > max_gap_us (default 4 hours), the forming bar is emitted as orphan
/// and the processor resets for the new trading week.
///
/// The first trade ever (last_trade_ts == 0) never triggers a reset -- there is no
/// previous trade to measure a gap against.
pub(crate) fn maybe_reset_at_week_gap(
    processor: &mut OpenDeviationBarProcessor,
    trade_timestamp_us: i64,
    last_trade_ts: &mut i64,
    max_gap_us: i64,
) -> Option<OpenDeviationBar> {
    if *last_trade_ts > 0 {
        let gap = trade_timestamp_us - *last_trade_ts;
        if gap > max_gap_us {
            *last_trade_ts = trade_timestamp_us;
            return processor.reset_at_ouroboros();
        }
    }
    *last_trade_ts = trade_timestamp_us;
    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::processor::OpenDeviationBarProcessor;
    use opendeviationbar_core::{FixedPoint, Tick};

    /// Default gap threshold: 4 hours in microseconds.
    const FOUR_HOURS_US: i64 = 14_400_000_000;

    fn make_tick(id: i64, price_f64: f64, timestamp_us: i64) -> Tick {
        let price_str = format!("{price_f64:.8}");
        Tick {
            ref_id: id,
            price: FixedPoint::from_str(&price_str).unwrap(),
            volume: FixedPoint::from_str("1.00000000").unwrap(),
            first_sub_id: id,
            last_sub_id: id,
            timestamp: timestamp_us,
            is_buyer_maker: false,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        }
    }

    #[test]
    fn test_week_variant_construction_and_traits() {
        // Week variant with max_gap_us can be constructed and is Copy/Clone/PartialEq
        let mode = OuroborosMode::Week {
            max_gap_us: FOUR_HOURS_US,
        };
        let mode2 = mode; // Copy
        let mode3 = mode.clone(); // Clone
        assert_eq!(mode, mode2); // PartialEq
        assert_eq!(mode, mode3);
        assert_ne!(mode, OuroborosMode::Aion);
        assert_ne!(mode, OuroborosMode::Day);

        // Different max_gap_us values are not equal
        let mode_different = OuroborosMode::Week {
            max_gap_us: 1_000_000,
        };
        assert_ne!(mode, mode_different);
    }

    #[test]
    fn test_week_gap_first_trade_no_reset() {
        // First trade ever (last_trade_ts == 0) should NOT trigger reset
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let mut last_ts: i64 = 0;

        // Feed one trade to create a forming bar
        let tick = make_tick(1, 100.0, 1_000_000_000_000);
        let _ = processor.process_single_trade(&tick);
        assert!(processor.get_incomplete_bar().is_some());

        // Call maybe_reset_at_week_gap with last_ts == 0 (first trade)
        let result = maybe_reset_at_week_gap(
            &mut processor,
            1_000_000_000_000,
            &mut last_ts,
            FOUR_HOURS_US,
        );
        assert!(result.is_none(), "first trade should not trigger reset");
        // last_ts should be updated
        assert_eq!(last_ts, 1_000_000_000_000);
    }

    #[test]
    fn test_week_gap_small_gap_no_reset() {
        // Gap < max_gap_us should NOT trigger reset
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let base_ts = 1_000_000_000_000i64;

        // Feed a trade to build a forming bar
        let tick1 = make_tick(1, 100.0, base_ts);
        let _ = processor.process_single_trade(&tick1);

        // Set last_ts to base_ts (simulating first trade already tracked)
        let mut last_ts = base_ts;

        // Next trade 1 hour later (3_600_000_000 us < 4h threshold)
        let next_ts = base_ts + 3_600_000_000;
        let result = maybe_reset_at_week_gap(&mut processor, next_ts, &mut last_ts, FOUR_HOURS_US);
        assert!(result.is_none(), "gap < threshold should not trigger reset");
        assert_eq!(last_ts, next_ts, "last_ts should be updated");
        // Forming bar should still exist
        assert!(processor.get_incomplete_bar().is_some());
    }

    #[test]
    fn test_week_gap_large_gap_triggers_reset() {
        // Gap > max_gap_us should trigger reset and emit orphan bar
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let base_ts = 1_000_000_000_000i64;

        // Feed trades to build a forming bar
        let tick1 = make_tick(1, 100.0, base_ts);
        let _ = processor.process_single_trade(&tick1);
        let tick2 = make_tick(2, 100.001, base_ts + 1_000_000);
        let _ = processor.process_single_trade(&tick2);
        assert!(processor.get_incomplete_bar().is_some());

        let mut last_ts = base_ts + 1_000_000;

        // Next trade 5 hours later (> 4h threshold)
        let weekend_ts = base_ts + 18_000_000_000; // 5 hours
        let result =
            maybe_reset_at_week_gap(&mut processor, weekend_ts, &mut last_ts, FOUR_HOURS_US);
        assert!(result.is_some(), "gap > threshold should trigger reset");
        let orphan = result.unwrap();
        // Orphan should contain the trades we fed
        assert_eq!(orphan.first_agg_trade_id, 1);
        assert_eq!(orphan.last_agg_trade_id, 2);
        // last_ts should be updated to the new trade timestamp
        assert_eq!(last_ts, weekend_ts);
        // Processor should be fresh (no forming bar)
        assert!(processor.get_incomplete_bar().is_none());
    }

    #[test]
    fn test_week_gap_updates_last_ts_on_both_paths() {
        // Verify last_ts is updated whether reset happens or not
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Path 1: No reset (gap within threshold)
        let mut last_ts = 1_000_000_000_000i64;
        let next_ts = last_ts + 1_000_000; // 1 second gap
        let tick = make_tick(1, 100.0, next_ts);
        let _ = processor.process_single_trade(&tick);

        let result = maybe_reset_at_week_gap(&mut processor, next_ts, &mut last_ts, FOUR_HOURS_US);
        assert!(result.is_none());
        assert_eq!(last_ts, next_ts, "non-reset path should update last_ts");

        // Path 2: Reset (gap exceeds threshold)
        let weekend_ts = next_ts + 20_000_000_000; // 20 seconds > threshold if we use small threshold
        let result = maybe_reset_at_week_gap(
            &mut processor,
            weekend_ts,
            &mut last_ts,
            1_000_000, // 1 second threshold for testing
        );
        assert!(
            result.is_some(),
            "should trigger reset with small threshold"
        );
        assert_eq!(last_ts, weekend_ts, "reset path should update last_ts");
    }

    #[test]
    fn test_midnight_returns_none_for_week_mode() {
        // Week mode should NOT trigger midnight resets (same as Aion)
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Feed a trade before midnight
        let before_midnight = 86_399_000_000i64; // 23:59:59 in microseconds
        let tick1 = make_tick(1, 100.0, before_midnight);
        let _ = processor.process_single_trade(&tick1);

        let mut last_day = -1i64;
        let mode = OuroborosMode::Week {
            max_gap_us: FOUR_HOURS_US,
        };

        // First call sets last_day
        let result = maybe_reset_at_midnight(&mut processor, before_midnight, &mut last_day, mode);
        assert!(result.is_none());

        // Cross midnight -- should still return None for Week mode
        let after_midnight = 86_401_000_000i64; // 00:00:01 next day
        let result = maybe_reset_at_midnight(&mut processor, after_midnight, &mut last_day, mode);
        assert!(
            result.is_none(),
            "Week mode should not trigger midnight reset"
        );
        // Forming bar should still exist
        assert!(processor.get_incomplete_bar().is_some());
    }

    #[test]
    fn test_week_checkpoint_roundtrip() {
        // WK-03: Checkpoint roundtrip after week-gap reset preserves fresh processor state
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        let base_ts = 1_000_000_000_000i64;

        // Phase 1: Feed several trades to build a forming bar
        for i in 0..5i64 {
            let tick = make_tick(i + 1, 100.0 + (i as f64) * 0.001, base_ts + i * 1_000_000);
            let _ = processor.process_single_trade(&tick);
        }
        assert!(processor.get_incomplete_bar().is_some());
        let bar_before = processor.get_incomplete_bar().unwrap();
        assert_eq!(bar_before.first_agg_trade_id, 1);
        assert_eq!(bar_before.last_agg_trade_id, 5);

        // Phase 2: Trigger week-gap reset (simulating weekend void)
        let mut last_ts = base_ts + 4_000_000;
        let weekend_ts = base_ts + 20_000_000_000; // 20 seconds later (> 4h if we used real gap)
        let orphan = maybe_reset_at_week_gap(
            &mut processor,
            weekend_ts,
            &mut last_ts,
            1_000_000, // 1-second threshold for test
        );
        assert!(orphan.is_some(), "should emit orphan at week gap");
        let orphan_bar = orphan.unwrap();
        assert_eq!(orphan_bar.first_agg_trade_id, 1);
        assert_eq!(orphan_bar.last_agg_trade_id, 5);

        // Phase 3: Create checkpoint on the now-fresh processor
        let checkpoint = processor.create_checkpoint("TESTUSDT");
        assert!(
            !checkpoint.has_incomplete_bar(),
            "processor should be fresh after reset"
        );

        // Phase 4: Restore from checkpoint
        let restored = OpenDeviationBarProcessor::from_checkpoint(checkpoint).unwrap();

        // Phase 5: Feed a new trade -- should open a new bar (fresh state)
        let mut restored = restored;
        let new_tick = make_tick(100, 200.0, weekend_ts + 1_000_000);
        let result = restored.process_single_trade(&new_tick).unwrap();
        assert!(result.is_none(), "first trade opens bar, no breach");
        let new_bar = restored.get_incomplete_bar().unwrap();
        assert_eq!(
            new_bar.first_agg_trade_id, 100,
            "new bar starts from new trade"
        );
        assert_eq!(new_bar.last_agg_trade_id, 100);
    }
}