repos
/ finance-rust master

finance-rust

mirror archived upstream

Single-binary self-hosted market watcher for stocks, ETFs, indexes, and futures: live charts, key stats, fundamentals, SEC filings, and SSE streaming.

axumdockerfinancerustself-hostedsqlitestocksvite

9.0 KB · 281 lines · Rust Raw History
  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}