opendeviationbar-streaming 13.70.3

//! Per-symbol WebSocket task with independent reconnection.

use super::puck::puck_fill;
use super::trade_dispatch::{ExtraSinks, process_trade_through_processors};
use super::types::*;
use crate::gap::{GapEvent, GapFillResult, TradeIdGapDetector};
use crate::ring_buffer::ConcurrentRingBuffer;
use opendeviationbar_core::Tick;
use opendeviationbar_core::checkpoint::Checkpoint;
use opendeviationbar_core::processor::OpenDeviationBarProcessor;
use opendeviationbar_providers::binance::{AdaptiveRateLimiter, ReconnectionPolicy};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::sync::{mpsc, watch};
use tokio_util::sync::CancellationToken;

/// Maximum number of trades to drain from the channel in a single burst-atomic frame.
/// Prevents unbounded draining if the channel is continuously fed.
/// 1024 is generous -- typical same-ms bursts are 5-50 trades.
const MAX_BURST_DRAIN: usize = 1024;

/// Per-symbol WebSocket task with independent reconnection.
/// Issue #91: Each symbol gets its own backoff state (barter-rs pattern).
///
/// TODO(#279): Migrate from per-symbol WebSocket connections to combined streams.
/// Empirical testing (2026-03-16) showed 2x9 hybrid (2 combined connections,
/// 9 symbols each) delivers 4x throughput with zero errors vs 18 individual
/// connections which suffer recv timeouts on low-volume symbols.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn symbol_task(
    symbol: String,
    mut processors: Vec<(u32, OpenDeviationBarProcessor)>,
    bar_buffer: ConcurrentRingBuffer<CompletedBar>, // Issue #96 Task #9: Ring buffer replaces mpsc
    checkpoint_tx: mpsc::Sender<(String, u32, Checkpoint)>,
    _policy: ReconnectionPolicy,
    shutdown: CancellationToken,
    metrics: Arc<LiveEngineMetrics>,
    rate_limiter: Option<Arc<AdaptiveRateLimiter>>,
    forming_txs: Vec<(u32, watch::Sender<Option<FormingBar>>)>, // Issue #214
    gap_event_tx: mpsc::Sender<GapEvent>,                       // Issue #257
    mut gap_fill_rx: Option<crate::gap::GapFillReceiver>, // Issue #257: on-demand fill commands
    ch_seed_tid: Option<i64>, // ClickHouse seed for gap detector (from StreamManager)
    pre_buffered_rx: Option<mpsc::Receiver<Tick>>, // Issue #286: pre-spawned WS receiver
    ouroboros_mode: OuroborosMode, // Phase 16: per-symbol mode (Day or Aion)
    extra_sinks: ExtraSinks, // Issue #318: fan-out to BarSink implementations
    mut checkpoint_snapshot_rx: mpsc::Receiver<()>, // Proactive checkpoint snapshot requests
    junction_semaphore: Arc<tokio::sync::Semaphore>, // Serialize junction fills across symbols (#347)
) {
    // Issue #96: Create Arc<str> once per symbol task -- cheap clone for each CompletedBar
    let symbol_arc: Arc<str> = Arc::from(symbol.as_str());

    // Issue #214: Index forming bar senders by threshold for O(1) lookup
    let forming_tx_map: HashMap<u32, watch::Sender<Option<FormingBar>>> =
        forming_txs.into_iter().collect();

    // Issue #286: Reuse pre-buffered WS receiver if available (from start_ws()),
    // otherwise create a new WS connection inline (legacy path).
    let mut trade_rx = if let Some(rx) = pre_buffered_rx {
        tracing::info!(%symbol, "using pre-buffered WS receiver (zero-gap start)");
        rx
    } else {
        // Combined stream requires start_ws() to be called first.
        // Per-symbol fallback removed -- all symbols use combined WS (COMB-04).
        tracing::error!(%symbol, "combined stream requires pre-buffered receiver -- call start_ws() before start()");
        return;
    };

    // Issue #257: Create gap detector for this symbol
    let mut gap_detector = TradeIdGapDetector::new(symbol_arc.clone(), Some(gap_event_tx));
    // Seed gap detector: max(checkpoint_tid, ch_seed_tid).
    // Checkpoint seed comes from processor state (if restored from checkpoint file).
    // CH seed comes from StreamManager's trade_id_state (seeded from ClickHouse).
    // Using max ensures the detector knows the highest trade ID from ANY source.
    let checkpoint_tid = processors.first().and_then(|(_, p)| p.last_agg_trade_id());
    let effective_seed = match (checkpoint_tid, ch_seed_tid) {
        (Some(cp), Some(ch)) => Some(cp.max(ch)),
        (Some(cp), None) => Some(cp),
        (None, Some(ch)) => Some(ch),
        (None, None) => None,
    };
    if let Some(seed) = effective_seed {
        gap_detector.seed(seed);
        tracing::info!(
            %symbol,
            seed,
            checkpoint_tid = checkpoint_tid.unwrap_or(-1),
            ch_seed_tid = ch_seed_tid.unwrap_or(-1),
            "gap detector seeded"
        );
    }

    tracing::info!(%symbol, thresholds = ?processors.iter().map(|(t, _)| *t).collect::<Vec<_>>(), "symbol task started");

    // Rust-side dedup: per-threshold committed floor.
    // Bars with last_agg_trade_id <= floor are already in ClickHouse -- skip emission.
    // Floor is initialized from max(processor.last_completed_bar_tid, ch_seed_tid) per threshold.
    // v1.4: Uses last_completed_bar_tid (not last_agg_trade_id) to avoid suppressing
    // junction bars whose trade IDs overlap with the forming bar tail.
    // After each bar emission, floor is updated to the emitted bar's last_agg_trade_id.
    let mut committed_floors: HashMap<u32, i64> = HashMap::new();
    for (threshold, processor) in processors.iter() {
        let proc_tid = processor.last_completed_bar_tid().unwrap_or(-1);
        let ch_tid = ch_seed_tid.unwrap_or(-1);
        let floor = proc_tid.max(ch_tid);
        if floor > 0 {
            committed_floors.insert(*threshold, floor);
            tracing::info!(%symbol, threshold, floor, completed_tid = proc_tid, ch_tid, "rust-dedup: floor initialized (v1.4: completed bar tid)");
        }
    }

    // Track current UTC day per processor for midnight reset (#264)
    let mut last_days: Vec<i64> = processors.iter().map(|_| -1i64).collect();

    // Phase 52: Track last trade timestamp per processor for Week gap detection.
    // Initialized to 0 (not -1). maybe_reset_at_week_gap() guards on *last_trade_ts > 0
    // so the first trade is safe (never triggers reset).
    let mut last_trade_timestamps_us: Vec<i64> = processors.iter().map(|_| 0i64).collect();

    // Issue #300 (MEM-08): Per-threshold forming bar throttle timestamps (microseconds).
    // Initialized to 0 so the first forming bar update is always sent immediately.
    let mut last_forming_update: HashMap<u32, i64> = HashMap::new();

    // --- JUNCTION FILL: Bridge REST->WS gap (Issue #XXX) ---
    // Peek first WS trade to determine if a junction gap exists.
    // This runs ONCE at startup, before the main select loop.

    // #347: Timed drain — wait 50ms after first trade to let more WS trades buffer,
    // reducing the junction gap that puck_fill must bridge via REST.
    let first_ws_trade = match trade_rx.recv().await {
        Some(t) => t,
        None => {
            tracing::warn!(%symbol, "WS channel closed before first trade, exiting symbol_task");
            return;
        }
    };
    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
    let mut ws_buffer: Vec<Tick> = Vec::new();
    while let Ok(t) = trade_rx.try_recv() {
        ws_buffer.push(t);
        if ws_buffer.len() >= MAX_BURST_DRAIN {
            break;
        }
    }
    let ws_buffer_depth = ws_buffer.len();

    // JUNC-01/JUNC-02: Fill junction gap using the SAME processors (preserves forming bar)
    let last_rest_tid = processors.first().and_then(|(_, p)| p.last_agg_trade_id());
    if let Some(rest_tid) = last_rest_tid {
        let junction_gap = first_ws_trade.ref_id - rest_tid - 1;
        if junction_gap > 0 {
            // Compute forming bar trade count across all processors
            let forming_bar_trades = processors
                .first()
                .and_then(|(_, p)| p.get_incomplete_bar())
                .map(|b| b.individual_trade_count)
                .unwrap_or(0);

            tracing::info!(
                %symbol,
                junction_gap_tid_delta = junction_gap,
                forming_bar_trades,
                ws_buffer_depth,
                rest_last_tid = rest_tid,
                ws_first_tid = first_ws_trade.ref_id,
                "junction.telemetry"
            );

            // #347: Serialize junction fills across symbols to prevent rate limiter
            // exhaustion. Without this, 16+ concurrent puck_fills overwhelm the
            // 3000 weight/min budget and REST fetches fail silently.
            let _junction_permit = junction_semaphore.acquire().await
                .expect("junction semaphore closed");

            // Temporarily boost rate limiter ceiling during junction fill
            if let Some(ref rl) = rate_limiter {
                rl.set_ceiling(4800);
            }

            let recovered = puck_fill(
                &symbol,
                rest_tid,
                first_ws_trade.ref_id,
                &mut processors,
                &bar_buffer,
                &metrics,
                &shutdown,
                rate_limiter.clone(),
                &forming_tx_map,
                "junction_fill",
                &mut committed_floors,
                ouroboros_mode,
                &extra_sinks,
            )
            .await;

            // Restore rate limiter ceiling
            if let Some(ref rl) = rate_limiter {
                rl.set_ceiling(3000);
            }

            // Drop permit so next symbol can proceed
            drop(_junction_permit);

            // #347: Loud failure telemetry — junction gap unrecovered
            if recovered == 0 && junction_gap > 0 {
                tracing::error!(
                    %symbol,
                    junction_gap,
                    rest_last_tid = rest_tid,
                    ws_first_tid = first_ws_trade.ref_id,
                    "junction.UNRECOVERED: puck_fill returned 0 trades for non-zero gap \
                     — Stathera continuity broken, kintsugi must repair"
                );
                metrics.gap_fills.fetch_add(1, Ordering::Relaxed);
            }
        } else {
            let forming_bar_trades = processors
                .first()
                .and_then(|(_, p)| p.get_incomplete_bar())
                .map(|b| b.individual_trade_count)
                .unwrap_or(0);
            tracing::info!(
                %symbol,
                junction_gap_tid_delta = 0i64,
                forming_bar_trades,
                ws_buffer_depth,
                "junction.telemetry"
            );
        }
    } else {
        tracing::info!(
            %symbol,
            junction_gap_tid_delta = 0i64,
            forming_bar_trades = 0u64,
            ws_buffer_depth,
            "junction.telemetry"
        );
    }

    // JUNC-03: Re-seed gap detector past the processor's high-water mark (#295).
    // After fill_from_rest + junction fill, the processor has seen trades up to
    // current_last_tid. Gap detector must be seeded past this point so it doesn't
    // fire puck_fill on WS buffer trades that fall within the already-processed
    // REST range. Without this, WS buffer gaps trigger duplicate bar emission
    // (puck_fill bypasses committed_floors -> overlapping bars in CH).
    // Issue #345: Use max() across all thresholds (gap detector is per-symbol,
    // must be seeded past the highest TID ANY processor has seen).
    let current_last_tid = processors
        .iter()
        .filter_map(|(_, p)| p.last_agg_trade_id())
        .max()
        .unwrap_or(-1);
    let gap_detector_seed = current_last_tid.max(first_ws_trade.ref_id);
    gap_detector.seed(gap_detector_seed);

    // JUNC-04: Monotonicity guard -- drop WS trade if already processed during fill_from_rest/junction fill
    if first_ws_trade.ref_id > current_last_tid {
        process_trade_through_processors(
            &first_ws_trade,
            &mut processors,
            &mut last_days,
            &mut last_trade_timestamps_us,
            &bar_buffer,
            &metrics,
            &symbol_arc,
            &symbol,
            &forming_tx_map,
            &mut committed_floors,
            &mut last_forming_update,
            ouroboros_mode,
            &extra_sinks,
        );
        gap_detector.update_last_tid(first_ws_trade.ref_id);
    } else {
        tracing::debug!(
            %symbol,
            ws_tid = first_ws_trade.ref_id,
            current_last_tid,
            "dropping overlapping WS trade at junction"
        );
    }
    // --- END JUNCTION FILL ---

    // JUNC-05: Process buffered WS trades with monotonicity guard (#295).
    // WS buffer has been accumulating since start_ws() -- many trades overlap
    // with fill_from_rest's range. Skip trades already processed to prevent:
    // (a) processor state corruption (double-counted volume, wrong trade counts)
    // (b) gap detector firing puck_fill on already-covered TID ranges
    // (c) puck_fill emitting duplicate bars that bypass committed_floors
    let mut ws_skipped = 0u64;
    let mut ws_processed = 0u64;
    // Track the high-water mark: max TID the processor has seen (from REST + junction fill)
    let mut high_water_tid = current_last_tid;
    for trade in &ws_buffer {
        if trade.ref_id <= high_water_tid {
            ws_skipped += 1;
            continue;
        }
        // Trade is beyond REST/junction fill range -- process normally
        if let Some((last_id, gap_size)) = gap_detector.check(trade.ref_id) {
            let recovered = puck_fill(
                &symbol,
                last_id,
                trade.ref_id,
                &mut processors,
                &bar_buffer,
                &metrics,
                &shutdown,
                rate_limiter.clone(),
                &forming_tx_map,
                "junction_fill",
                &mut committed_floors,
                ouroboros_mode,
                &extra_sinks,
            )
            .await;
            if recovered > 0 {
                gap_detector.mark_filled(last_id, gap_size);
            }
        }
        process_trade_through_processors(
            trade,
            &mut processors,
            &mut last_days,
            &mut last_trade_timestamps_us,
            &bar_buffer,
            &metrics,
            &symbol_arc,
            &symbol,
            &forming_tx_map,
            &mut committed_floors,
            &mut last_forming_update,
            ouroboros_mode,
            &extra_sinks,
        );
        gap_detector.update_last_tid(trade.ref_id);
        high_water_tid = trade.ref_id;
        ws_processed += 1;
    }
    if !ws_buffer.is_empty() {
        tracing::info!(
            %symbol,
            ws_buffer_total = ws_buffer.len(),
            ws_skipped,
            ws_processed,
            high_water_tid,
            "junction: WS buffer processed (skipped trades within REST range)"
        );
    }

    // Issue #96 RUST-03: Reuse burst frame allocation across select loop iterations.
    // Previously allocated Vec::with_capacity(32) per iteration -- now cleared and reused.
    let mut frame: Vec<Tick> = Vec::with_capacity(32);

    // Process trades -> bars
    loop {
        tokio::select! {
            // Issue #257: On-demand gap-fill commands from Python fill_gap()
            Some(cmd) = async {
                match gap_fill_rx.as_mut() {
                    Some(rx) => rx.recv().await,
                    None => std::future::pending().await,
                }
            } => {
                let start_time = std::time::Instant::now();
                tracing::info!(
                    %symbol, from_tid = cmd.from_tid, to_tid = cmd.to_tid,
                    "processing on-demand gap-fill command"
                );

                let _gap_size = cmd.to_tid - cmd.from_tid;
                let recovered = puck_fill(
                    &symbol,
                    cmd.from_tid - 1, // last_known_id (fill expects exclusive start)
                    cmd.to_tid,
                    &mut processors,
                    &bar_buffer,
                    &metrics,
                    &shutdown,
                    rate_limiter.clone(),
                    &forming_tx_map,
                    "on_demand_gap_fill",
                    &mut committed_floors,
                    ouroboros_mode,
                    &extra_sinks,
                )
                .await;
                // Advance gap detector so it won't re-detect this range (#290)
                if recovered > 0 {
                    gap_detector.update_last_tid(cmd.to_tid);
                }
                let (trades_fetched, bars_produced, partial) =
                    (recovered as usize, 0usize, false);

                let duration_ms = start_time.elapsed().as_millis() as u64;
                let result = GapFillResult {
                    trades_fetched,
                    bars_produced,
                    duration_ms,
                    partial,
                };
                // Send result back -- ignore error if caller dropped
                let _ = cmd.response_tx.send(result);
            }
            trade = trade_rx.recv() => {
                match trade {
                    Some(first_trade) => {
                        // Issue #273: Burst-atomic frame processing.
                        //
                        // Collect the triggering trade plus all immediately-buffered
                        // trades from the channel into a single "frame". This ensures
                        // that when a same-millisecond burst of trades arrives (e.g.,
                        // 19 trades at ts=1773485713955), ALL of them are processed
                        // through the threshold processors atomically -- no select-loop
                        // yield or gap-fill interruption can split the burst.
                        //
                        // Without this, the select loop processes one trade per
                        // iteration. If a bar closes on trade N and trades N+1..N+k
                        // share the same millisecond, the loop could yield between
                        // them (e.g., to handle a gap-fill command), leaving N+1..N+k
                        // unprocessed until the next select iteration. In edge cases,
                        // those trades may never arrive (WS delivery gap at TCP level)
                        // and the gap detector can't help because they were "expected
                        // to be contiguous" from a timestamp perspective.
                        //
                        // By draining all buffered trades into a frame BEFORE any
                        // processing, we guarantee atomic handling of bursts.
                        frame.clear();
                        frame.push(first_trade);
                        for _ in 0..MAX_BURST_DRAIN {
                            match trade_rx.try_recv() {
                                Ok(t) => frame.push(t),
                                Err(_) => break, // Channel empty or closed
                            }
                        }

                        let frame_size = frame.len();
                        metrics.trades_received.fetch_add(frame_size as u64, Ordering::Relaxed);

                        if frame_size > 1 {
                            tracing::debug!(
                                %symbol, frame_size,
                                first_tid = frame[0].ref_id,
                                last_tid = frame[frame_size - 1].ref_id,
                                "burst-atomic frame: processing {} trades atomically",
                                frame_size
                            );
                        }

                        // Process each trade in the frame sequentially
                        for trade in &frame {
                            // Issue #257: Gap detection via TradeIdGapDetector
                            if let Some((last_id, gap_size)) = gap_detector.check(trade.ref_id) {
                                metrics.reconnections.fetch_add(1, Ordering::Relaxed);
                                metrics.gap_fills.fetch_add(1, Ordering::Relaxed);
                                let recovered = puck_fill(
                                    &symbol,
                                    last_id,
                                    trade.ref_id,
                                    &mut processors,
                                    &bar_buffer,
                                    &metrics,
                                    &shutdown,
                                    rate_limiter.clone(),
                                    &forming_tx_map,
                                    "normal_gap_fill",
                                    &mut committed_floors,
                                    ouroboros_mode,
                                    &extra_sinks,
                                )
                                .await;
                                metrics
                                    .gap_trades_recovered
                                    .fetch_add(recovered, Ordering::Relaxed);

                                // Mark gap as filled inline if trades were recovered
                                if recovered > 0 {
                                    gap_detector.mark_filled(last_id, gap_size);
                                }
                            }

                            // Issue #273: Process trade through all threshold processors
                            process_trade_through_processors(
                                trade,
                                &mut processors,
                                &mut last_days,
                                &mut last_trade_timestamps_us,
                                &bar_buffer,
                                &metrics,
                                &symbol_arc,
                                &symbol,
                                &forming_tx_map,
                                &mut committed_floors,
                                &mut last_forming_update,
                                ouroboros_mode,
                                &extra_sinks,
                            );

                            // Issue #257: Update gap detector with this trade's ID
                            gap_detector.update_last_tid(trade.ref_id);
                        }
                    }
                    None => {
                        // WebSocket channel closed (terminal error or shutdown)
                        tracing::info!(%symbol, "trade channel closed");
                        break;
                    }
                }
            }
            // Proactive checkpoint snapshot — fires during active trading
            Some(_) = checkpoint_snapshot_rx.recv() => {
                for (threshold, processor) in &processors {
                    let cp = processor.create_checkpoint(&symbol);
                    let _ = checkpoint_tx.send((symbol.clone(), *threshold, cp)).await;
                }
            }
            () = shutdown.cancelled() => {
                tracing::info!(%symbol, "shutdown requested");
                break;
            }
        }
    }

    // Extract checkpoints from all processors before task exits
    for (threshold, processor) in &processors {
        let cp = processor.create_checkpoint(&symbol);
        tracing::info!(
            %symbol, threshold,
            has_incomplete_bar = cp.has_incomplete_bar(),
            "checkpoint extracted on shutdown"
        );
        if checkpoint_tx
            .send((symbol.clone(), *threshold, cp))
            .await
            .is_err()
        {
            tracing::warn!(%symbol, threshold, "checkpoint channel closed");
        }
    }

    tracing::info!(%symbol, "symbol task ended");
}