opendeviationbar-client 13.78.1

// Issue #170: SSE client for consuming completed open deviation bars from the sidecar.

use crate::transport::ReqwestTransport;
use eventsource_client::{Client, ClientBuilder, SSE};
use futures::{Stream, StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Configuration for connecting to the ODB sidecar SSE endpoint.
///
/// Empty `symbols` or `thresholds` means "subscribe to all" (no server-side filter).
#[derive(Debug, Clone)]
pub struct OdbSseConfig {
    pub host: String,
    pub port: u16,
    pub symbols: Vec<String>,
    pub thresholds: Vec<u32>,
}

/// A completed or forming bar received via SSE, with typed fields.
///
/// Captures a subset of the full sidecar payload. Always-present OHLCV and
/// microstructure fields are typed; additional fields (inter-bar, intra-bar)
/// are captured in `extra` via `#[serde(flatten)]`.
///
/// Issue #214: Added `is_forming` field to distinguish forming bars from completed bars.
/// For completed bars (`event: bar`), `is_forming` is `None` or `Some(false)`.
/// For forming bars (`event: forming`), `is_forming` is `Some(true)`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OdbBar {
    pub symbol: String,
    pub threshold: u32,
    pub open_time_us: i64,
    pub close_time_us: i64,
    pub open: f64,
    pub high: f64,
    pub low: f64,
    pub close: f64,
    pub volume: Option<f64>,
    pub vwap: Option<f64>,
    pub buy_volume: Option<f64>,
    pub sell_volume: Option<f64>,
    pub individual_trade_count: Option<u32>,
    pub agg_record_count: Option<u32>,
    pub first_ref_id: Option<i64>,
    pub last_ref_id: Option<i64>,
    // Microstructure features (always present when microstructure enabled)
    pub duration_us: Option<i64>,
    pub ofi: Option<f64>,
    pub vwap_close_deviation: Option<f64>,
    pub price_impact: Option<f64>,
    pub kyle_lambda_proxy: Option<f64>,
    pub trade_intensity: Option<f64>,
    pub volume_per_trade: Option<f64>,
    pub aggression_ratio: Option<f64>,
    pub aggregation_density: Option<f64>,
    pub turnover_imbalance: Option<f64>,
    // Orphan marker (ouroboros boundary bars)
    pub is_orphan: Option<bool>,
    /// Issue #214: True for forming (incomplete) bars, None/false for completed bars.
    /// Forming bars are pushed via `event: forming` SSE events at 1Hz.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub is_forming: Option<bool>,
    /// Additional fields from sidecar (inter-bar, intra-bar features, future fields).
    /// Preserves forward compatibility without modeling every field.
    #[serde(flatten)]
    pub extra: HashMap<String, serde_json::Value>,
}

/// Events emitted by the SSE client stream.
///
/// Issue #214: Added `FormingBar` for incomplete bar snapshots pushed at 1Hz.
/// Issue #215: Added `Checkpoint` for processor state pushed periodically.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum OdbSseEvent {
    Connected,
    Bar(Box<OdbBar>),
    /// Issue #214: Forming (incomplete) bar snapshot, pushed at 1Hz.
    /// The contained `OdbBar` has `is_forming = Some(true)`.
    FormingBar(Box<OdbBar>),
    /// Issue #215: Processor checkpoint data, pushed periodically and on shutdown.
    Checkpoint {
        checkpoints: serde_json::Value,
    },
    Heartbeat,
    /// A bar event was received but could not be deserialized.
    DeserializationError {
        error: String,
        raw_data: String,
    },
    Disconnected(String),
}

/// Client for consuming open deviation bars via SSE from the ODB sidecar.
///
/// Wraps `eventsource-client` with a reqwest transport. Handles reconnection
/// (exponential backoff + jitter), heartbeat detection, and bar deserialization.
#[derive(Clone)]
pub struct OdbSseClient {
    config: OdbSseConfig,
}

impl OdbSseClient {
    pub fn new(config: OdbSseConfig) -> Self {
        Self { config }
    }

    /// Build the SSE endpoint URL from config. Exposed for testing.
    #[doc(hidden)]
    pub fn build_url(&self) -> String {
        let mut url = format!(
            "http://{}:{}/events/bars",
            self.config.host, self.config.port
        );
        let mut params = Vec::new();
        if !self.config.symbols.is_empty() {
            params.push(format!("symbols={}", self.config.symbols.join(",")));
        }
        if !self.config.thresholds.is_empty() {
            let ts: Vec<String> = self
                .config
                .thresholds
                .iter()
                .map(ToString::to_string)
                .collect();
            params.push(format!("thresholds={}", ts.join(",")));
        }
        if !params.is_empty() {
            url.push('?');
            url.push_str(&params.join("&"));
        }
        url
    }

    /// Connect and return a stream of bar events.
    ///
    /// Handles SSE parsing, reconnection (exponential backoff + jitter),
    /// heartbeat detection, and server `retry:` directives — all via
    /// `eventsource-client`.
    pub fn connect(&self) -> impl Stream<Item = OdbSseEvent> + Send + '_ {
        let url = self.build_url();
        futures::stream::once(async move {
            let init_result = ReqwestTransport::new()
                .map_err(|e| format!("failed to build HTTP client: {e}"))
                .and_then(|transport| {
                    ClientBuilder::for_url(&url)
                        .map_err(|e| format!("invalid URL: {e}"))
                        .map(|builder| builder.build_with_transport(transport))
                });

            let client = match init_result {
                Ok(c) => c,
                Err(msg) => {
                    return futures::stream::once(async move { OdbSseEvent::Disconnected(msg) })
                        .left_stream();
                }
            };
            let stream = client.stream();

            stream
                .map_ok(|sse| match sse {
                    SSE::Connected(_) => {
                        tracing::info!(target: "odb_client", "SSE connected");
                        Some(OdbSseEvent::Connected)
                    }
                    SSE::Comment(comment) => {
                        if comment.trim() == "heartbeat" {
                            Some(OdbSseEvent::Heartbeat)
                        } else {
                            tracing::debug!(
                                target: "odb_client",
                                comment = %comment,
                                "ignoring unknown SSE comment"
                            );
                            None
                        }
                    }
                    SSE::Event(evt) => {
                        match evt.event_type.as_str() {
                            "bar" => match serde_json::from_str::<OdbBar>(&evt.data) {
                                Ok(bar) => {
                                    tracing::info!(
                                        target: "odb_client::bar",
                                        symbol = %bar.symbol,
                                        threshold = bar.threshold,
                                        close_time_us = bar.close_time_us,
                                        open = bar.open,
                                        high = bar.high,
                                        low = bar.low,
                                        close = bar.close,
                                        buy_volume = bar.buy_volume,
                                        sell_volume = bar.sell_volume,
                                        source = "sse",
                                        "bar_received"
                                    );
                                    Some(OdbSseEvent::Bar(Box::new(bar)))
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        target: "odb_client",
                                        error = %e,
                                        data = %evt.data,
                                        "failed to deserialize bar event"
                                    );
                                    Some(OdbSseEvent::DeserializationError {
                                        error: e.to_string(),
                                        raw_data: evt.data,
                                    })
                                }
                            },
                            // Issue #214: Forming bar events pushed at 1Hz
                            "forming" => match serde_json::from_str::<OdbBar>(&evt.data) {
                                Ok(bar) => {
                                    tracing::debug!(
                                        target: "odb_client::forming",
                                        symbol = %bar.symbol,
                                        threshold = bar.threshold,
                                        close = bar.close,
                                        "forming_bar_received"
                                    );
                                    Some(OdbSseEvent::FormingBar(Box::new(bar)))
                                }
                                Err(e) => {
                                    tracing::debug!(
                                        target: "odb_client",
                                        error = %e,
                                        "failed to deserialize forming bar"
                                    );
                                    Some(OdbSseEvent::DeserializationError {
                                        error: e.to_string(),
                                        raw_data: evt.data,
                                    })
                                }
                            },
                            // Issue #215: Checkpoint events pushed periodically
                            "checkpoint" => {
                                match serde_json::from_str::<serde_json::Value>(&evt.data) {
                                    Ok(data) => {
                                        let checkpoints =
                                            data.get("checkpoints").cloned().unwrap_or(data);
                                        tracing::debug!(
                                            target: "odb_client::checkpoint",
                                            "checkpoint_received"
                                        );
                                        Some(OdbSseEvent::Checkpoint { checkpoints })
                                    }
                                    Err(e) => {
                                        tracing::debug!(
                                            target: "odb_client",
                                            error = %e,
                                            "failed to deserialize checkpoint"
                                        );
                                        Some(OdbSseEvent::DeserializationError {
                                            error: e.to_string(),
                                            raw_data: evt.data,
                                        })
                                    }
                                }
                            }
                            _ => {
                                tracing::debug!(
                                    target: "odb_client",
                                    event_type = %evt.event_type,
                                    "ignoring unknown SSE event type"
                                );
                                None
                            }
                        }
                    }
                })
                .filter_map(|result| async move {
                    match result {
                        Ok(Some(event)) => Some(event),
                        Ok(None) => None, // filtered events produce no output
                        Err(e) => Some(OdbSseEvent::Disconnected(format!("{e}"))),
                    }
                })
                .chain(futures::stream::once(async {
                    OdbSseEvent::Disconnected("stream ended".into())
                }))
                .right_stream()
        })
        .flatten()
    }
}