opendeviationbar-streaming 13.70.3

//! Puck: inline gap fill on trade-ID discontinuity.

use super::trade_dispatch::{ExtraSinks, maybe_reset_at_midnight, maybe_reset_at_week_gap};
use super::types::*;
use crate::ring_buffer::ConcurrentRingBuffer;
use opendeviationbar_core::processor::OpenDeviationBarProcessor;
use opendeviationbar_providers::binance::AdaptiveRateLimiter;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;

/// Puck: inline gap fill on trade-ID discontinuity.
///
/// When a gap is detected between the last processed trade and an incoming
/// WebSocket trade, fetches the missing trades using Binance REST API's
/// parallel `fromId` fetch and processes them through all threshold processors.
/// Sub-3s recovery for typical reconnection gaps.
///
/// Returns the number of trades recovered (0 on error or skip -- non-fatal).
#[allow(clippy::too_many_arguments)]
pub(crate) async fn puck_fill(
    symbol: &str,
    last_known_id: i64,
    first_ws_id: i64,
    processors: &mut [(u32, OpenDeviationBarProcessor)],
    bar_buffer: &ConcurrentRingBuffer<CompletedBar>,
    metrics: &LiveEngineMetrics,
    shutdown: &CancellationToken,
    rate_limiter: Option<Arc<AdaptiveRateLimiter>>,
    forming_tx_map: &HashMap<u32, watch::Sender<Option<FormingBar>>>, // Issue #214
    event_type: &str, // "junction_fill", "normal_gap_fill", or "on_demand_gap_fill"
    committed_floors: &mut HashMap<u32, i64>, // #295: dedup bars already in CH
    ouroboros_mode: OuroborosMode, // Phase 16: per-symbol mode (Day or Aion)
    extra_sinks: &ExtraSinks, // Issue #318: fan-out to BarSink implementations
) -> u64 {
    let _ = &extra_sinks; // Suppress unused warning when clickhouse-sink disabled
    let gap_size = first_ws_id - last_known_id - 1;

    tracing::info!(
        %symbol, gap_size, last_known_id, first_ws_id, event_type,
        "puck: filling gap via parallel REST API"
    );

    // Issue #257: Use parallel fetch for sub-3s gap recovery.
    // to_id is inclusive — use first_ws_id - 1 because first_ws_id
    // is the WS trade that already arrived (don't duplicate it).
    // #347: Retry with exponential backoff (1s, 2s, 4s) on failure.
    let from_id = last_known_id + 1;
    let max_retries: u32 = 3;
    let mut trades = Vec::new();
    let mut last_err = None;

    for attempt in 0..=max_retries {
        if attempt > 0 {
            let backoff_ms = 1000u64 << (attempt - 1); // 1s, 2s, 4s
            tracing::warn!(
                %symbol, attempt, backoff_ms, event_type, gap_size,
                "puck: retrying REST gap-fill after failure"
            );
            tokio::time::sleep(tokio::time::Duration::from_millis(backoff_ms)).await;
            if shutdown.is_cancelled() {
                tracing::info!(%symbol, "puck: retry aborted by shutdown");
                return 0;
            }
        }
        match opendeviationbar_providers::binance::parallel_fetch::fetch_aggtrades_parallel(
            symbol,
            from_id,
            first_ws_id - 1,
            rate_limiter.clone(),
            None, // Use default concurrency (5)
            None, // Use default REST base URL
            None, // No extra REST headers
        )
        .await
        {
            Ok(t) => {
                trades = t;
                last_err = None;
                break;
            }
            Err(e) => {
                last_err = Some(e);
            }
        }
    }

    if let Some(e) = last_err {
        tracing::error!(
            %symbol, ?e, gap_size, event_type, attempts = max_retries + 1,
            "puck: REST gap-fill FAILED after all retries — junction gap UNRECOVERED"
        );
        return 0;
    }

    // Process fetched trades sequentially through all threshold processors
    // (CRITICAL: processor is order-sensitive, trades must be in agg_trade_id order)
    let mut trades_recovered = 0u64;
    let symbol_arc: Arc<str> = Arc::from(symbol);

    // Track current UTC day per processor for midnight reset
    let mut last_days: Vec<i64> = processors.iter().map(|_| -1i64).collect();
    // Phase 52: Track last trade timestamp per processor for Week gap detection
    let mut last_trade_timestamps_us: Vec<i64> = processors.iter().map(|_| 0i64).collect();

    for trade in &trades {
        // Check shutdown periodically
        if shutdown.is_cancelled() {
            tracing::info!(%symbol, trades_recovered, "gap fill interrupted by shutdown");
            break;
        }

        // Fan out to all threshold processors
        for (idx, (threshold, processor)) in processors.iter_mut().enumerate() {
            // Week gap check: emit orphan if inter-trade gap exceeds max_gap_us
            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,
                )
            {
                    if let Some(&floor) = committed_floors.get(threshold)
                        && orphan.last_agg_trade_id > 0
                        && orphan.last_agg_trade_id <= floor
                    {
                        tracing::debug!(%symbol, threshold = *threshold, last_tid = orphan.last_agg_trade_id, floor, "puck_fill: 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,
                    };
                    committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
                    if !bar_buffer.push(completed.clone()) {
                        metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                        metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                    }
                #[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)
            {
                // #295: Dedup orphan bars against committed_floors
                if let Some(&floor) = committed_floors.get(threshold)
                    && orphan.last_agg_trade_id > 0
                    && orphan.last_agg_trade_id <= floor
                {
                    tracing::debug!(%symbol, threshold = *threshold, last_tid = orphan.last_agg_trade_id, floor, "puck_fill: 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,
                };
                committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
                if !bar_buffer.push(completed.clone()) {
                    metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                    metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                }
                // 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);
                }
            }

            match processor.process_single_trade(trade) {
                Ok(Some(bar)) => {
                    // #295: Dedup bars against committed_floors
                    if let Some(&floor) = committed_floors.get(threshold)
                        && bar.last_agg_trade_id > 0
                        && bar.last_agg_trade_id <= floor
                    {
                        tracing::debug!(%symbol, threshold = *threshold, last_tid = bar.last_agg_trade_id, floor, "puck_fill: skipping bar (within committed range)");
                        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,
                    };
                    committed_floors.insert(*threshold, completed.bar.last_agg_trade_id);
                    if !bar_buffer.push(completed.clone()) {
                        metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                        metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
                    }
                    // 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);
                    }
                }
                Ok(None) => {
                    // Issue #214: Update forming bar snapshot during gap fill
                    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,
                        }));
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        %symbol, threshold = *threshold, ?e,
                        "gap-fill trade processing error"
                    );
                }
            }
        }
        trades_recovered += 1;
    }

    tracing::info!(
        %symbol, trades_recovered, gap_size,
        "reconnection gap fill complete"
    );
    trades_recovered
}