finance-rust

  1//! In-process pub/sub hub for the live data stream.
  2//!
  3//! One [`Hub`] lives in `AppState`. The scheduler publishes quote and
  4//! market-session events into it; each `/stream` SSE connection subscribes and
  5//! forwards them to a browser.
  6//!
  7//! The hub also carries the **interest registry** — a count, per ticker, of
  8//! how many connected clients are currently displaying it. This is what makes
  9//! intraday polling demand-driven: the scheduler asks [`Hub::viewed`] for the
 10//! tickers with at least one live viewer and polls only those. With nobody
 11//! watching, `viewed()` is empty and the intraday job does no network work at
 12//! all (see `scheduler::run_intraday`).
 13//!
 14//! For a Python reader: this is a tiny in-memory pub/sub plus a `Counter` of
 15//! who's looking at what — no external broker, just a `tokio` broadcast
 16//! channel and a `Mutex<HashMap>`.
 17
 18use std::collections::HashMap;
 19use std::sync::Mutex;
 20
 21use serde::Serialize;
 22use tokio::sync::broadcast;
 23
 24/// Broadcast channel depth. One full universe sweep (~144 quote events) sits
 25/// well under this; a subscriber that still lags is handled by skipping the
 26/// gap (see `routes::stream`), never by stalling a publisher.
 27const CHANNEL_CAPACITY: usize = 1024;
 28
 29/// One message pushed to every connected `/stream` client. Serialized as the
 30/// SSE event payload; the `kind` tag is unused on the wire (the SSE `event:`
 31/// field carries the type) but keeps the JSON self-describing.
 32#[derive(Debug, Clone, Serialize)]
 33#[serde(tag = "kind", rename_all = "snake_case")]
 34pub enum StreamEvent {
 35    Quote(QuoteUpdate),
 36    Market { session: String },
 37    /// A background-data state change — a job started or finished, a
 38    /// `fetch_log` row landed. Carries no payload: it is a nudge telling an
 39    /// open `/health` page to pull a fresh snapshot from `/api/health`.
 40    /// Published by the scheduler; see `routes::health`.
 41    Health,
 42}
 43
 44/// A live quote, shaped for the browser to patch `data-field` nodes in place.
 45#[derive(Debug, Clone, Serialize)]
 46pub struct QuoteUpdate {
 47    pub ticker: String,
 48    pub price: f64,
 49    pub prev_close: Option<f64>,
 50    pub change_abs: Option<f64>,
 51    pub change_pct: Option<f64>,
 52    pub market_state: Option<String>,
 53}
 54
 55impl QuoteUpdate {
 56    /// Build an update, deriving the day change from `price` and `prev_close`.
 57    pub fn new(
 58        ticker: String,
 59        price: f64,
 60        prev_close: Option<f64>,
 61        market_state: Option<String>,
 62    ) -> Self {
 63        let (change_abs, change_pct) = match prev_close {
 64            Some(p) if p != 0.0 => (Some(price - p), Some((price - p) / p * 100.0)),
 65            Some(p) => (Some(price - p), None),
 66            None => (None, None),
 67        };
 68        Self {
 69            ticker,
 70            price,
 71            prev_close,
 72            change_abs,
 73            change_pct,
 74            market_state,
 75        }
 76    }
 77}
 78
 79/// The pub/sub hub plus the per-ticker viewer-interest registry.
 80pub struct Hub {
 81    tx: broadcast::Sender<StreamEvent>,
 82    /// ticker -> number of connected clients currently viewing it.
 83    interest: Mutex<HashMap<String, u32>>,
 84}
 85
 86impl Default for Hub {
 87    fn default() -> Self {
 88        Self::new()
 89    }
 90}
 91
 92impl Hub {
 93    pub fn new() -> Self {
 94        let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
 95        Self {
 96            tx,
 97            interest: Mutex::new(HashMap::new()),
 98        }
 99    }
100
101    /// Subscribe a new client. The returned receiver yields every event
102    /// published from now on.
103    pub fn subscribe(&self) -> broadcast::Receiver<StreamEvent> {
104        self.tx.subscribe()
105    }
106
107    /// Publish an event to all subscribers. A send with no subscribers is not
108    /// an error — it simply goes nowhere.
109    pub fn publish(&self, event: StreamEvent) {
110        let _ = self.tx.send(event);
111    }
112
113    /// Register that a client has begun viewing `tickers`.
114    pub fn add_interest(&self, tickers: &[String]) {
115        let mut map = self.interest.lock().unwrap();
116        for t in tickers {
117            *map.entry(t.clone()).or_insert(0) += 1;
118        }
119    }
120
121    /// Drop a client's interest in `tickers` (called when its stream ends). A
122    /// ticker's entry is removed once its viewer count falls back to zero, so
123    /// `viewed()` stays a tight set.
124    pub fn remove_interest(&self, tickers: &[String]) {
125        let mut map = self.interest.lock().unwrap();
126        for t in tickers {
127            if let Some(n) = map.get_mut(t) {
128                *n = n.saturating_sub(1);
129                if *n == 0 {
130                    map.remove(t);
131                }
132            }
133        }
134    }
135
136    /// The tickers with at least one live viewer right now.
137    pub fn viewed(&self) -> Vec<String> {
138        self.interest.lock().unwrap().keys().cloned().collect()
139    }
140}