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 /health` — the data-health page — and `GET /api/health`, its JSON feed.
2//!
3//! The page lays the background-data machinery open: each endpoint guard's
4//! circuit-breaker state and how much of its per-hour request budget is spent,
5//! every scheduler job's state / last success / next run, and a tail of the
6//! `fetch_log`.
7//!
8//! The page route renders an initial snapshot embedded in the HTML so it draws
9//! without a round trip. From there the page stays live off the Phase 5 SSE
10//! hub: the scheduler publishes a `health` event whenever a job changes state
11//! or logs a row, and the page answers each one by re-pulling `/api/health`.
12//! Both routes build the same [`Health`] snapshot via [`snapshot`].
13
14use axum::{extract::State, response::Response, routing::get, Json, Router};
15use serde::Serialize;
16use sqlx::SqlitePool;
17
18use crate::db::now_ms;
19use crate::render::render;
20use crate::AppState;
21
22pub fn router() -> Router<AppState> {
23 Router::new()
24 .route("/health", get(page))
25 .route("/api/health", get(api))
26}
27
28/// The whole data-health picture in one payload: every endpoint guard, every
29/// scheduler job, and a tail of the fetch log.
30#[derive(Serialize)]
31struct Health {
32 /// When this snapshot was built, UTC epoch-ms.
33 generated_at: i64,
34 /// True while any job sits in the `fetching` state — drives the page's
35 /// live "fetching now" banner.
36 fetching: bool,
37 endpoints: Vec<Endpoint>,
38 jobs: Vec<Job>,
39 log: Vec<LogRow>,
40}
41
42/// One upstream's request guard, shaped for the page.
43#[derive(Serialize)]
44struct Endpoint {
45 endpoint: String,
46 label: String,
47 /// `closed` | `open` | `half_open`.
48 state: String,
49 fail_streak: i64,
50 trip_count: i64,
51 opened_at: Option<i64>,
52 retry_at: Option<i64>,
53 /// Requests let through in the current rolling budget hour.
54 hour_count: i64,
55 /// Start of that hour, UTC epoch-ms (the budget resets an hour later).
56 hour_start: Option<i64>,
57 hourly_budget: i64,
58 /// `hour_count` as a 0..100 share of the budget, for the meter fill.
59 budget_pct: f64,
60 last_ok_at: Option<i64>,
61 last_error: Option<String>,
62 last_error_at: Option<i64>,
63}
64
65/// One scheduler job's `data_status` row, with a human label and description.
66#[derive(Serialize)]
67struct Job {
68 job: String,
69 label: String,
70 description: String,
71 /// `idle` | `fetching` | `ok` | `stale` | `error`.
72 state: String,
73 last_ok_at: Option<i64>,
74 last_error: Option<String>,
75 last_error_at: Option<i64>,
76 next_run_at: Option<i64>,
77 updated_at: i64,
78}
79
80/// One `fetch_log` row — a passthrough of the table, newest first.
81#[derive(Serialize, sqlx::FromRow)]
82struct LogRow {
83 job: String,
84 provider: String,
85 ticker: Option<String>,
86 /// `ok` | `error` | `skipped`.
87 status: String,
88 detail: Option<String>,
89 rows: Option<i64>,
90 duration_ms: Option<i64>,
91 started_at: i64,
92}
93
94/// The `endpoint_guard` columns the page needs.
95#[derive(sqlx::FromRow)]
96struct GuardRow {
97 endpoint: String,
98 state: String,
99 fail_streak: i64,
100 trip_count: i64,
101 opened_at: Option<i64>,
102 retry_at: Option<i64>,
103 hour_start: Option<i64>,
104 hour_count: i64,
105 hourly_budget: i64,
106 last_ok_at: Option<i64>,
107 last_error: Option<String>,
108 last_error_at: Option<i64>,
109}
110
111/// The `data_status` columns the page needs.
112#[derive(sqlx::FromRow)]
113struct StatusRow {
114 job: String,
115 state: String,
116 last_ok_at: Option<i64>,
117 last_error: Option<String>,
118 last_error_at: Option<i64>,
119 next_run_at: Option<i64>,
120 updated_at: i64,
121}
122
123/// Build the full health snapshot. Three small reads — `endpoint_guard` holds
124/// a handful of rows, `data_status` one per job, and the log tail is capped.
125async fn snapshot(pool: &SqlitePool) -> Health {
126 let guards: Vec<GuardRow> = sqlx::query_as(
127 "SELECT endpoint, state, fail_streak, trip_count, opened_at, retry_at, \
128 hour_start, hour_count, hourly_budget, last_ok_at, last_error, last_error_at \
129 FROM endpoint_guard ORDER BY endpoint",
130 )
131 .fetch_all(pool)
132 .await
133 .unwrap_or_default();
134
135 let statuses: Vec<StatusRow> = sqlx::query_as(
136 "SELECT job, state, last_ok_at, last_error, last_error_at, next_run_at, updated_at \
137 FROM data_status",
138 )
139 .fetch_all(pool)
140 .await
141 .unwrap_or_default();
142
143 let log: Vec<LogRow> = sqlx::query_as(
144 "SELECT job, provider, ticker, status, detail, rows, duration_ms, started_at \
145 FROM fetch_log ORDER BY started_at DESC LIMIT 50",
146 )
147 .fetch_all(pool)
148 .await
149 .unwrap_or_default();
150
151 let endpoints: Vec<Endpoint> = guards.into_iter().map(to_endpoint).collect();
152
153 let mut jobs: Vec<Job> = statuses.into_iter().map(to_job).collect();
154 jobs.sort_by_key(|j| job_rank(&j.job));
155 let fetching = jobs.iter().any(|j| j.state == "fetching");
156
157 Health {
158 generated_at: now_ms(),
159 fetching,
160 endpoints,
161 jobs,
162 log,
163 }
164}
165
166fn to_endpoint(g: GuardRow) -> Endpoint {
167 // Rounded to two places: this only drives a meter's CSS width, and a tidy
168 // figure keeps the JSON payload readable.
169 let budget_pct = if g.hourly_budget > 0 {
170 let raw = g.hour_count as f64 / g.hourly_budget as f64 * 100.0;
171 (raw.clamp(0.0, 100.0) * 100.0).round() / 100.0
172 } else {
173 0.0
174 };
175 Endpoint {
176 label: endpoint_label(&g.endpoint).to_string(),
177 endpoint: g.endpoint,
178 state: g.state,
179 fail_streak: g.fail_streak,
180 trip_count: g.trip_count,
181 opened_at: g.opened_at,
182 retry_at: g.retry_at,
183 hour_start: g.hour_start,
184 hour_count: g.hour_count,
185 hourly_budget: g.hourly_budget,
186 budget_pct,
187 last_ok_at: g.last_ok_at,
188 last_error: g.last_error,
189 last_error_at: g.last_error_at,
190 }
191}
192
193fn to_job(s: StatusRow) -> Job {
194 let (label, description) = job_meta(&s.job);
195 Job {
196 label: label.to_string(),
197 description: description.to_string(),
198 job: s.job,
199 state: s.state,
200 last_ok_at: s.last_ok_at,
201 last_error: s.last_error,
202 last_error_at: s.last_error_at,
203 next_run_at: s.next_run_at,
204 updated_at: s.updated_at,
205 }
206}
207
208/// A human label for a known upstream id.
209fn endpoint_label(endpoint: &str) -> &str {
210 match endpoint {
211 "yahoo" => "Yahoo Finance · quotes, intraday & daily history",
212 "sec" => "SEC EDGAR · fundamentals & filings",
213 other => other,
214 }
215}
216
217/// Human label and one-line description per scheduler job. Since the demand-only
218/// refocus the timed jobs are the intraday poll and the active home sweep (all
219/// the old sweeps were removed; deep data is fetched on demand when a page is
220/// viewed — see the guard usage in the Endpoints section above). An unknown job
221/// falls back to its raw id and no description.
222fn job_meta(job: &str) -> (&str, &str) {
223 match job {
224 "intraday" => (
225 "Intraday quotes",
226 "Live quotes from Yahoo for the symbols a browser is viewing, on a \
227 ~5-minute cadence. Nothing is polled when nobody is on the site.",
228 ),
229 "home" => (
230 "Home sweep",
231 "Re-quotes the dashboard's overview instruments and every watchlist \
232 symbol on a 15-minute cadence, viewer or not, so the home page \
233 always opens fresh. Off-hours only the ~24h instruments are polled.",
234 ),
235 "prune" => (
236 "Prune",
237 "Local cleanup of aged intraday bars and fetch-log rows (no network).",
238 ),
239 other => (other, ""),
240 }
241}
242
243/// Display order for the jobs list; the fallback keeps any future job at the end.
244fn job_rank(job: &str) -> u8 {
245 match job {
246 "intraday" => 0,
247 "home" => 1,
248 "prune" => 2,
249 _ => 3,
250 }
251}
252
253/// `GET /health` — the page, with the current snapshot embedded so it renders
254/// without a round trip; the page's script keeps it live from there.
255async fn page(State(state): State<AppState>) -> Response {
256 let snap = snapshot(&state.pool).await;
257 let json = serde_json::to_string(&snap).unwrap_or_else(|_| "null".to_string());
258 let extra = minijinja::context! {
259 title => "Data health",
260 health_json => embed_json(&json),
261 };
262 render(&state, "pages/health.html", "/health", extra)
263}
264
265/// `GET /api/health` — the same snapshot as JSON, polled by the page whenever
266/// the SSE hub signals a data change.
267async fn api(State(state): State<AppState>) -> Json<Health> {
268 Json(snapshot(&state.pool).await)
269}
270
271/// Escape a JSON string for safe embedding inside a `<script>` element. Only
272/// `<`, `>` and `&` matter (they could otherwise close the tag or open a
273/// comment); replaced with their `\uXXXX` forms, which a JSON parser reads back
274/// identically. Structural JSON never contains these characters, so a blanket
275/// replace touches only string contents.
276fn embed_json(json: &str) -> String {
277 json.replace('<', "\\u003c")
278 .replace('>', "\\u003e")
279 .replace('&', "\\u0026")
280}