Single-binary self-hosted market watcher for stocks, ETFs, indexes, and futures: live charts, key stats, fundamentals, SEC filings, and SSE streaming.
axumdockerfinancerustself-hostedsqlitestocksvite
1//! `GET /stream` — the Server-Sent Events endpoint.
2//!
3//! A browser opens one `EventSource` here, passing `?symbols=` with the
4//! tickers the page is showing. The connection:
5//! - registers viewer interest in those tickers with the [`Hub`], so the
6//! scheduler's intraday job knows to poll them (and unregisters on drop);
7//! - emits an initial `market` event and a `quote` snapshot so the page is
8//! immediately consistent with stored state;
9//! - then forwards live `quote` events for the registered tickers, and every
10//! `market` and `health` event, as the hub publishes them.
11
12use std::collections::HashSet;
13use std::convert::Infallible;
14use std::sync::Arc;
15
16use axum::{
17 extract::{Query, State},
18 response::sse::{Event, KeepAlive, Sse},
19 routing::get,
20 Router,
21};
22use futures_util::Stream;
23use serde::Deserialize;
24use tokio::sync::broadcast::error::RecvError;
25
26use crate::market;
27use crate::stream::{Hub, QuoteUpdate, StreamEvent};
28use crate::AppState;
29
30pub fn router() -> Router<AppState> {
31 Router::new().route("/stream", get(stream))
32}
33
34#[derive(Deserialize)]
35struct StreamQuery {
36 /// Comma-separated tickers the client is displaying. Unknown tickers are
37 /// dropped (see `validate_tickers`) so a client cannot steer the
38 /// demand-driven poller at arbitrary upstream symbols.
39 symbols: Option<String>,
40}
41
42/// Releases a connection's interest back to the hub when its SSE stream is
43/// dropped — the client navigated away, closed the tab, or lost the network.
44struct InterestGuard {
45 hub: Arc<Hub>,
46 tickers: Vec<String>,
47}
48
49impl Drop for InterestGuard {
50 fn drop(&mut self) {
51 self.hub.remove_interest(&self.tickers);
52 }
53}
54
55async fn stream(
56 Query(q): Query<StreamQuery>,
57 State(state): State<AppState>,
58) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
59 let requested: Vec<String> = q
60 .symbols
61 .unwrap_or_default()
62 .split(',')
63 .map(|s| s.trim().to_uppercase())
64 .filter(|s| !s.is_empty())
65 .collect();
66 let tickers = validate_tickers(&state, &requested).await;
67
68 state.hub.add_interest(&tickers);
69 let guard = InterestGuard {
70 hub: state.hub.clone(),
71 tickers: tickers.clone(),
72 };
73 let want: HashSet<String> = tickers.iter().cloned().collect();
74
75 // Subscribe before snapshotting so no event published in between is lost.
76 let rx = state.hub.subscribe();
77 let snapshot = quote_snapshot(&state, &tickers).await;
78 let session = market::session_at(chrono::Utc::now());
79
80 let body = async_stream::stream! {
81 // Holding the guard inside the stream ties interest to the stream's
82 // lifetime: when the client disconnects, the stream drops, the guard
83 // drops, and the interest is released.
84 let _guard = guard;
85
86 yield Ok::<_, Infallible>(sse_market(session.as_str()));
87 for qu in &snapshot {
88 yield Ok(sse_quote(qu));
89 }
90
91 let mut rx = rx;
92 loop {
93 match rx.recv().await {
94 Ok(StreamEvent::Quote(qu)) => {
95 if want.contains(&qu.ticker) {
96 yield Ok(sse_quote(&qu));
97 }
98 }
99 Ok(StreamEvent::Market { session }) => {
100 yield Ok(sse_market(&session));
101 }
102 Ok(StreamEvent::Health) => {
103 yield Ok(sse_health());
104 }
105 // A slow client fell behind the channel: skip the dropped
106 // span and carry on rather than tearing the connection down.
107 Err(RecvError::Lagged(_)) => continue,
108 Err(RecvError::Closed) => break,
109 }
110 }
111 };
112
113 Sse::new(body).keep_alive(KeepAlive::default())
114}
115
116/// Keep only the requested tickers that are real seeded symbols, sorted and
117/// deduped. Filtering here is what bounds the demand-driven poller to the
118/// known universe.
119async fn validate_tickers(state: &AppState, requested: &[String]) -> Vec<String> {
120 if requested.is_empty() {
121 return Vec::new();
122 }
123 let known: HashSet<String> = sqlx::query_scalar("SELECT ticker FROM symbols")
124 .fetch_all(&state.pool)
125 .await
126 .unwrap_or_default()
127 .into_iter()
128 .collect();
129 let mut out: Vec<String> = requested
130 .iter()
131 .filter(|t| known.contains(*t))
132 .cloned()
133 .collect();
134 out.sort();
135 out.dedup();
136 out
137}
138
139/// The latest stored quote for each requested ticker, so a freshly connected
140/// page can reconcile immediately without waiting for the next poll. The
141/// `quotes` table holds one row per symbol (~144 max), so a full scan is cheap.
142async fn quote_snapshot(state: &AppState, tickers: &[String]) -> Vec<QuoteUpdate> {
143 if tickers.is_empty() {
144 return Vec::new();
145 }
146 let want: HashSet<&str> = tickers.iter().map(String::as_str).collect();
147 let rows: Vec<(String, f64, Option<f64>, Option<String>)> =
148 sqlx::query_as("SELECT ticker, price, prev_close, market_state FROM quotes")
149 .fetch_all(&state.pool)
150 .await
151 .unwrap_or_default();
152 rows.into_iter()
153 .filter(|(t, ..)| want.contains(t.as_str()))
154 .map(|(t, price, prev, state)| QuoteUpdate::new(t, price, prev, state))
155 .collect()
156}
157
158fn sse_quote(qu: &QuoteUpdate) -> Event {
159 Event::default()
160 .event("quote")
161 .data(serde_json::to_string(qu).unwrap_or_default())
162}
163
164fn sse_market(session: &str) -> Event {
165 Event::default()
166 .event("market")
167 .data(format!("{{\"session\":\"{session}\"}}"))
168}
169
170/// A content-free nudge: an open `/health` page answers it by pulling a fresh
171/// snapshot from `/api/health`. See `routes::health`.
172fn sse_health() -> Event {
173 Event::default().event("health").data("{}")
174}