//! The ingest server for opt-in data collection (docs/telemetry.md). //! //! - `POST /v1/batch`: a zstd batch from the launcher, checked by `horizon_telemetry::batch`. //! - `DELETE /v1/installs/`: everything an install sent goes. //! - `GET /v1/aggregate?days=30`: the anonymous totals the website reads, cached. //! - `GET /v1/benchmarks/summary`: players' benchmark results grouped by GPU and CPU, groups of at //! least 3 installs only (`summary`), cached; the website's Benchmarks page reads it. //! - `POST /v1/live`: a game's anonymous "I'm playing" ping (`live`, in memory only). //! - `GET /v1/live`: how many play now, rounded, cached briefly. //! - `/hub/…`: the hub's votes (`hub`), bug reports and suggestions (`intake`). //! - `GET /healthz`. //! //! Abuse limits: a token bucket per IP address (kept in memory for an hour, never stored), a //! daily item cap per install, and the batch size limits. The live count has its own, larger //! bucket per address (several players can share one) and a cap on the sessions it holds. Raw //! rows expire after `retention_days`; the daily totals stay. One small process and one SQLite //! file: see `README.md` for running it. use std::collections::HashMap; use std::hash::{Hash, RandomState}; use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use axum::Router; use axum::extract::{ConnectInfo, Path, Query, Request, State}; use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post}; use horizon_telemetry::batch; pub mod classes; pub mod github; pub mod hub; pub mod intake; pub mod live; pub mod store; pub mod summary; use github::Filer; use horizon_telemetry::live::{MAX_PING_BYTES, Ping}; use live::Live; use store::Store; pub struct Config { /// Days raw rows are kept. pub retention_days: u32, /// Groups in the public totals need at least this many install-days. pub min_install_days: u64, /// Requests an IP address may make an hour (with a burst of a tenth of it). pub requests_per_hour: f64, /// Items one install may send a day. pub items_per_day: u32, /// Behind a proxy (Railway, Fly…): the client's address is the proxy's `X-Real-IP`, else the /// last `X-Forwarded-For` entry. pub trust_proxy: bool, /// The website allowed to read the totals from a browser. pub allow_origin: String, /// Live count requests (pings and reads) an IP address may make an hour, with a burst of a /// tenth of it. A game pings once a minute, so the default lets ten play behind one address. pub live_per_hour: f64, /// How long a session counts after its last ping. pub live_window: Duration, /// Sessions the live count holds at most; a new one past that is refused. pub live_max_sessions: usize, /// The hub's votes and reads of their counts an IP address may make an hour. pub hub_per_hour: f64, /// Bug reports and suggestions together an IP address may send an hour. pub hub_posts_per_hour: f64, /// Bug reports taken a day, from everyone: each one opens a GitHub issue. pub hub_reports_per_day: u32, /// Suggestions taken a day, from everyone. pub hub_suggestions_per_day: u32, /// Days reports and suggestions are kept (Kira has relayed them long before). pub hub_keep_days: u32, /// The bearer token `GET /hub/suggestions` wants (Kira's); unset, it answers 404. pub hub_token: Option, /// Opens the website's bug reports as issues; unset, `POST /hub/report` answers 503. pub github_token: Option, /// Where those issues go, `owner/name`. pub github_repo: String, } impl Default for Config { fn default() -> Self { Self { retention_days: 180, min_install_days: 5, requests_per_hour: 60.0, items_per_day: 100, trust_proxy: false, allow_origin: "https://playhorizon.dev".into(), live_per_hour: 600.0, live_window: horizon_telemetry::live::WINDOW, live_max_sessions: 200_000, hub_per_hour: 600.0, hub_posts_per_hour: 20.0, hub_reports_per_day: 30, hub_suggestions_per_day: 200, hub_keep_days: 90, hub_token: None, github_token: None, github_repo: "horizon-rust/horizon".into(), } } } impl Config { /// The defaults, with `HORIZON_TELEMETRY_RETENTION_DAYS`, `…_MIN_INSTALL_DAYS`, /// `…_REQUESTS_PER_HOUR`, `…_ITEMS_PER_DAY`, `…_TRUST_PROXY` (1), `…_ALLOW_ORIGIN`, /// `…_LIVE_PER_HOUR`, `…_LIVE_WINDOW_SECONDS`, `…_LIVE_MAX_SESSIONS`, `…_HUB_PER_HOUR`, /// `…_HUB_POSTS_PER_HOUR`, `…_HUB_REPORTS_PER_DAY`, `…_HUB_SUGGESTIONS_PER_DAY`, `…_HUB_KEEP_DAYS`, /// `…_HUB_TOKEN`, `…_GITHUB_TOKEN` and `…_GITHUB_REPO`. Empty tokens count as unset. pub fn from_env() -> Self { let var = |k: &str| std::env::var(format!("HORIZON_TELEMETRY_{k}")).ok(); let secret = |k: &str| { var(k) .map(|v| v.trim().to_owned()) .filter(|v| !v.is_empty()) }; let d = Self::default(); Self { retention_days: var("RETENTION_DAYS") .and_then(|v| v.parse().ok()) .unwrap_or(d.retention_days), min_install_days: var("MIN_INSTALL_DAYS") .and_then(|v| v.parse().ok()) .unwrap_or(d.min_install_days), requests_per_hour: var("REQUESTS_PER_HOUR") .and_then(|v| v.parse().ok()) .unwrap_or(d.requests_per_hour), items_per_day: var("ITEMS_PER_DAY") .and_then(|v| v.parse().ok()) .unwrap_or(d.items_per_day), trust_proxy: var("TRUST_PROXY").is_some_and(|v| v == "1"), allow_origin: var("ALLOW_ORIGIN").unwrap_or(d.allow_origin), live_per_hour: var("LIVE_PER_HOUR") .and_then(|v| v.parse().ok()) .unwrap_or(d.live_per_hour), live_window: var("LIVE_WINDOW_SECONDS") .and_then(|v| v.parse().ok()) .map_or(d.live_window, Duration::from_secs), live_max_sessions: var("LIVE_MAX_SESSIONS") .and_then(|v| v.parse().ok()) .unwrap_or(d.live_max_sessions), hub_per_hour: var("HUB_PER_HOUR") .and_then(|v| v.parse().ok()) .unwrap_or(d.hub_per_hour), hub_posts_per_hour: var("HUB_POSTS_PER_HOUR") .and_then(|v| v.parse().ok()) .unwrap_or(d.hub_posts_per_hour), hub_reports_per_day: var("HUB_REPORTS_PER_DAY") .and_then(|v| v.parse().ok()) .unwrap_or(d.hub_reports_per_day), hub_suggestions_per_day: var("HUB_SUGGESTIONS_PER_DAY") .and_then(|v| v.parse().ok()) .unwrap_or(d.hub_suggestions_per_day), hub_keep_days: var("HUB_KEEP_DAYS") .and_then(|v| v.parse().ok()) .unwrap_or(d.hub_keep_days), hub_token: secret("HUB_TOKEN"), github_token: secret("GITHUB_TOKEN"), github_repo: var("GITHUB_REPO").unwrap_or(d.github_repo), } } } pub struct App { store: Mutex, config: Config, /// Tokens left and when they were counted, per client address. buckets: Mutex, /// The same for the live count's requests. live_buckets: Mutex, /// Who is playing now (`live`). live: Mutex, /// `GET /v1/live`'s answer and when it was made. live_cache: Mutex>, /// `/v1/aggregate` answers by `days`, with when they were made. cache: Mutex>, /// `/v1/benchmarks/summary`'s answer and when it was made. summary_cache: Mutex>, /// Today, in days since 1970 (a test moves it). today: Box u32 + Send + Sync>, /// The key the hub's buckets hash addresses with, drawn at start and never stored. hub_key: RandomState, /// The hub's votes and vote counts, per hashed address. hub_buckets: Mutex, /// The hub's reports and suggestions, per hashed address. hub_post_buckets: Mutex, /// `GET /hub/votes`'s answer and when it was made. votes_cache: Mutex>, /// Opens bug reports as GitHub issues; None without a token. filer: Option>, } type Buckets = HashMap; const CACHE_FOR: Duration = Duration::from_secs(600); fn status(code: StatusCode, message: &str) -> Response { (code, message.to_owned()).into_response() } impl App { pub fn new(store: Store, config: Config) -> Arc { Self::with_clock(store, config, Box::new(horizon_telemetry::today)) } pub fn with_clock( store: Store, config: Config, today: Box u32 + Send + Sync>, ) -> Arc { let filer = config .github_token .clone() .map(|token| Arc::from(github::github(config.github_repo.clone(), token))); Self::with_filer(store, config, today, filer) } /// [`Self::with_clock`] with the hub's bug reports opened by `filer` (tests give their own). pub fn with_filer( store: Store, config: Config, today: Box u32 + Send + Sync>, filer: Option>, ) -> Arc { let live = Live::new(config.live_window, config.live_max_sessions); Arc::new(Self { store: Mutex::new(store), config, buckets: Mutex::default(), live_buckets: Mutex::default(), live: Mutex::new(live), live_cache: Mutex::default(), cache: Mutex::default(), summary_cache: Mutex::default(), today, hub_key: RandomState::new(), hub_buckets: Mutex::default(), hub_post_buckets: Mutex::default(), votes_cache: Mutex::default(), filer, }) } fn store(&self) -> std::sync::MutexGuard<'_, Store> { self.store.lock().unwrap_or_else(|e| e.into_inner()) } /// Takes a token for `ip`; false when it has none left. fn allow(&self, ip: IpAddr) -> bool { take(&self.buckets, self.config.requests_per_hour, ip) } /// [`Self::allow`] for the live count's requests. fn allow_live(&self, ip: IpAddr) -> bool { take(&self.live_buckets, self.config.live_per_hour, ip) } fn live(&self) -> std::sync::MutexGuard<'_, Live> { self.live.lock().unwrap_or_else(|e| e.into_inner()) } /// The address requests are counted under, for the rate limits only: kept in memory, never /// stored or logged. Behind a proxy (Railway) the TCP peer is the proxy, so it is the proxy's /// `X-Real-IP`, else the last `X-Forwarded-For` entry: the one the proxy added, which a client /// can't choose (the first one it can). An IPv6 client counts by its /64. fn client(&self, headers: &HeaderMap, peer: Option) -> IpAddr { let header = |name: &str| headers.get(name)?.to_str().ok(); let forwarded = self .config .trust_proxy .then(|| { header("x-real-ip") .and_then(|ip| ip.trim().parse().ok()) .or_else(|| { header("x-forwarded-for")? .rsplit(',') .next()? .trim() .parse() .ok() }) }) .flatten(); rate_key( forwarded .or(peer.map(|p| p.ip())) .unwrap_or(IpAddr::from([0, 0, 0, 0])), ) } /// The day an install's newest row came in, if it has any left. pub fn newest_day(&self, install: &[u8]) -> Option { self.store().newest_day(install).ok().flatten() } /// Drops the cached public answers (after a deletion or expiry). fn forget_cached(&self) { self.cache.lock().unwrap_or_else(|e| e.into_inner()).clear(); *self.summary_cache.lock().unwrap_or_else(|e| e.into_inner()) = None; } /// Daily upkeep: totals rebuilt, expired rows removed, idle rate-limit entries and live /// sessions forgotten, the hub's old reports and suggestions removed. Reports still waiting for /// their issue are [`Self::file_pending`]'s. pub fn maintain(&self) -> rusqlite::Result { let oldest = (self.today)().saturating_sub(self.config.retention_days); let removed = self.store().expire(oldest)?; self.expire_intake()?; self.forget_hub_buckets(); for buckets in [&self.buckets, &self.live_buckets] { buckets .lock() .unwrap_or_else(|e| e.into_inner()) .retain(|_, (_, at)| at.elapsed() < Duration::from_secs(3600)); } self.live().forget_expired(Instant::now()); if removed > 0 { self.forget_cached(); } Ok(removed) } } /// Takes a token for `key` from its bucket in `buckets` (`per_hour`, bursts of a tenth of it); /// false when it has none left. fn take(buckets: &Mutex>, per_hour: f64, key: K) -> bool { let rate = per_hour / 3600.0; let burst = (per_hour / 10.0).max(1.0); let mut buckets = buckets.lock().unwrap_or_else(|e| e.into_inner()); let now = Instant::now(); let (tokens, at) = buckets.entry(key).or_insert((burst, now)); *tokens = (*tokens + now.duration_since(*at).as_secs_f64() * rate).min(burst); *at = now; if *tokens < 1.0 { return false; } *tokens -= 1.0; true } /// The address a client's requests count under: IPv4 (an IPv4-mapped IPv6 address as IPv4 too), /// or an IPv6 address's /64, which one home or host holds whole, so addresses rotated inside it /// don't get fresh buckets. fn rate_key(ip: IpAddr) -> IpAddr { match ip { IpAddr::V6(v6) => match v6.to_ipv4_mapped() { Some(v4) => IpAddr::V4(v4), None => { let s = v6.segments(); IpAddr::V6(std::net::Ipv6Addr::new(s[0], s[1], s[2], s[3], 0, 0, 0, 0)) } }, v4 => v4, } } fn peer(req: &Request) -> Option { req.extensions() .get::>() .map(|c| c.0) } async fn receive(State(app): State>, req: Request) -> Response { let ip = app.client(req.headers(), peer(&req)); if !app.allow(ip) { return status(StatusCode::TOO_MANY_REQUESTS, "too many requests"); } let zstd = req .headers() .get(header::CONTENT_ENCODING) .is_some_and(|v| v.as_bytes().eq_ignore_ascii_case(b"zstd")); if !zstd { return status( StatusCode::UNSUPPORTED_MEDIA_TYPE, "send zstd (Content-Encoding: zstd)", ); } let Ok(body) = axum::body::to_bytes(req.into_body(), batch::MAX_COMPRESSED).await else { return status(StatusCode::PAYLOAD_TOO_LARGE, "a batch is at most 256 KiB"); }; let batch = match batch::decode(&body) { Ok(batch) => batch, Err(e) => return status(StatusCode::BAD_REQUEST, &e), }; let today = (app.today)(); let install = store::install_key(&batch.install_id).unwrap_or_default(); let mut store = app.store(); let sent = store.count_on(&install, today).unwrap_or(u32::MAX); if sent.saturating_add(batch.items.len() as u32) > app.config.items_per_day { return status( StatusCode::TOO_MANY_REQUESTS, "this install's daily limit is reached", ); } match store.insert(&batch, today) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(e) => { eprintln!("horizon-telemetry-server: could not store a batch: {e}"); status(StatusCode::SERVICE_UNAVAILABLE, "try again later") } } } async fn forget(State(app): State>, Path(id): Path, req: Request) -> Response { if !app.allow(app.client(req.headers(), peer(&req))) { return status(StatusCode::TOO_MANY_REQUESTS, "too many requests"); } let Some(install) = store::install_key(&id) else { return status(StatusCode::BAD_REQUEST, "not an install ID"); }; match app.store().delete_install(&install) { Ok(_) => { app.forget_cached(); StatusCode::NO_CONTENT.into_response() } Err(e) => { eprintln!("horizon-telemetry-server: could not delete an install: {e}"); status(StatusCode::SERVICE_UNAVAILABLE, "try again later") } } } #[derive(serde::Deserialize)] struct Window { days: Option, } async fn aggregate(State(app): State>, Query(window): Query) -> Response { let days = window.days.unwrap_or(30).clamp(1, 366); let cached = app .cache .lock() .unwrap_or_else(|e| e.into_inner()) .get(&days) .filter(|(at, _)| at.elapsed() < CACHE_FOR) .map(|(_, json)| json.clone()); let json = match cached { Some(json) => json, None => { let to = (app.today)(); let made = app.store() .aggregate(to + 1 - days.min(to + 1), to, app.config.min_install_days); match made { Ok(value) => { let json = value.to_string(); app.cache .lock() .unwrap_or_else(|e| e.into_inner()) .insert(days, (Instant::now(), json.clone())); json } Err(e) => { eprintln!("horizon-telemetry-server: could not build the totals: {e}"); return status(StatusCode::SERVICE_UNAVAILABLE, "try again later"); } } } }; json_response(&app, json, CACHE_FOR.as_secs()) } /// `GET /v1/benchmarks/summary`: players' benchmark groups (`summary`), made again at most every /// ten minutes. async fn benchmarks(State(app): State>) -> Response { let cached = app .summary_cache .lock() .unwrap_or_else(|e| e.into_inner()) .as_ref() .filter(|(at, _)| at.elapsed() < CACHE_FOR) .map(|(_, json)| json.clone()); let json = match cached { Some(json) => json, None => { let oldest = (app.today)().saturating_sub(app.config.retention_days); let uploads = match app.store().benchmark_uploads(oldest) { Ok(uploads) => uploads, Err(e) => { eprintln!("horizon-telemetry-server: could not read the benchmarks: {e}"); return status(StatusCode::SERVICE_UNAVAILABLE, "try again later"); } }; let json = summary::summary(&uploads, summary::MIN_INSTALLS, &utc_now()).to_string(); *app.summary_cache.lock().unwrap_or_else(|e| e.into_inner()) = Some((Instant::now(), json.clone())); json } }; json_response(&app, json, CACHE_FOR.as_secs()) } /// Now as `YYYY-MM-DDTHH:MM:SSZ`. fn utc_now() -> String { let secs = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_secs()); let (days, rest) = (secs / 86_400, secs % 86_400); // Days since 1970 to a civil date (Howard Hinnant's algorithm). let z = days as i64 + 719_468; let era = z.div_euclid(146_097); let doe = z - era * 146_097; let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); let mp = (5 * doy + 2) / 153; let day = doy - (153 * mp + 2) / 5 + 1; let month = if mp < 10 { mp + 3 } else { mp - 9 }; let year = yoe + era * 400 + i64::from(month <= 2); format!( "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", rest / 3600, rest / 60 % 60, rest % 60 ) } /// `POST /v1/live`: one "I'm playing". 204, or 400 for a bad ping, 413 for one too large, 429 /// over the address's limit, 503 when the table is full. async fn ping(State(app): State>, req: Request) -> Response { if !app.allow_live(app.client(req.headers(), peer(&req))) { return status(StatusCode::TOO_MANY_REQUESTS, "too many requests"); } let Ok(body) = axum::body::to_bytes(req.into_body(), MAX_PING_BYTES).await else { return status(StatusCode::PAYLOAD_TOO_LARGE, "a ping is at most 512 bytes"); }; let ping: Ping = match serde_json::from_slice(&body) { Ok(ping) => ping, Err(e) => return status(StatusCode::BAD_REQUEST, &e.to_string()), }; if let Err(e) = ping.check() { return status(StatusCode::BAD_REQUEST, e); } match app.live().ping(&ping, Instant::now()) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(live::Full) => status(StatusCode::SERVICE_UNAVAILABLE, "try again later"), } } /// `GET /v1/live`: the rounded live count (`horizon_telemetry::live::Count`), made again at /// most every eighth of the window. async fn playing(State(app): State>, req: Request) -> Response { if !app.allow_live(app.client(req.headers(), peer(&req))) { return status(StatusCode::TOO_MANY_REQUESTS, "too many requests"); } let fresh_for = app.config.live_window / 8; let json = { let mut cache = app.live_cache.lock().unwrap_or_else(|e| e.into_inner()); match &*cache { Some((at, json)) if at.elapsed() < fresh_for => json.clone(), _ => { let json = serde_json::to_string(&app.live().count(Instant::now())).unwrap_or_default(); *cache = Some((Instant::now(), json.clone())); json } } }; json_response(&app, json, fresh_for.as_secs()) } fn json_response(app: &App, json: String, max_age: u64) -> Response { let origin = HeaderValue::from_str(&app.config.allow_origin).unwrap_or(HeaderValue::from_static("*")); let cache = HeaderValue::from_str(&format!("public, max-age={max_age}")) .unwrap_or(HeaderValue::from_static("no-cache")); ( [ ( header::CONTENT_TYPE, HeaderValue::from_static("application/json"), ), (header::CACHE_CONTROL, cache), (header::ACCESS_CONTROL_ALLOW_ORIGIN, origin), ], json, ) .into_response() } pub fn router(app: Arc) -> Router { Router::new() .route("/v1/batch", post(receive)) .route("/v1/live", post(ping).get(playing)) .route("/v1/installs/{id}", delete(forget)) .route("/v1/aggregate", get(aggregate)) .route("/v1/benchmarks/summary", get(benchmarks)) .route("/hub/vote", post(hub::vote).options(hub::preflight)) .route("/hub/votes", get(hub::votes)) .route("/hub/report", post(intake::report).options(hub::preflight)) .route( "/hub/suggest", post(intake::suggest).options(hub::preflight), ) .route("/hub/suggestions", get(intake::suggestions)) .route("/healthz", get(|| async { "ok" })) .with_state(app) } #[cfg(test)] mod tests { use super::*; #[test] fn ipv6_rates_count_by_64() { let ip = |s: &str| s.parse::().unwrap(); // Rotated inside a /64: one key. Another /64: another key. assert_eq!(rate_key(ip("2001:db8::1")), rate_key(ip("2001:db8::99"))); assert_ne!(rate_key(ip("2001:db8::1")), rate_key(ip("2001:db8:1::1"))); // IPv4-mapped counts as the IPv4 it is. assert_eq!( rate_key(ip("::ffff:203.0.113.7")), IpAddr::from([203, 0, 113, 7]) ); // Plain IPv4 unchanged. assert_eq!(rate_key(ip("203.0.113.8")), IpAddr::from([203, 0, 113, 8])); } }