fn logs_payload(state: &App) -> Value { json!({ "entries": state.logs.recent_limit(crate::logging::LIVE_WINDOW), "capacity": crate::logging::HISTORY_CAPACITY, "window": crate::logging::LIVE_WINDOW, "dropped": state.logs.dropped.load(Ordering::Relaxed) }) } #[derive(Default, Deserialize)] #[serde(default, deny_unknown_fields)] struct LogFilter { vhost: Option, q: Option, status: Option, cache: Option, limit: Option, } async fn logs( State(state): State>, Query(filter): Query, ) -> Result, ApiError> { if filter.q.as_ref().is_some_and(|q| q.len() > 256) || filter .vhost .as_ref() .is_some_and(|v| !crate::config::valid_id(v)) || filter .cache .as_ref() .is_some_and(|v| !matches!(v.as_str(), "HIT" | "MISS" | "BYPASS" | "-")) { return Err(ApiError( StatusCode::BAD_REQUEST, "invalid log filter".into(), )); } let q = filter.q.unwrap_or_default().to_ascii_lowercase(); let limit = filter .limit .unwrap_or(crate::logging::LIVE_WINDOW) .clamp(1, crate::logging::HISTORY_CAPACITY); let entries: Vec<_> = state .logs .recent() .into_iter() .filter(|e| { filter .vhost .as_ref() .is_none_or(|v| e.vhost.as_ref() == Some(v)) && filter.status.is_none_or(|v| e.status == v) && filter.cache.as_ref().is_none_or(|v| e.cache == v.as_str()) && (q.is_empty() || format!( "{} {} {} {}", e.method, e.host, e.path, e.backend.as_deref().unwrap_or("") ) .to_ascii_lowercase() .contains(&q)) }) .take(limit) .collect(); Ok(Json(json!({ "entries": entries, "capacity": crate::logging::HISTORY_CAPACITY, "window": crate::logging::LIVE_WINDOW, "dropped": state.logs.dropped.load(Ordering::Relaxed) }))) } async fn live_logs_stream( State(state): State>, headers: HeaderMap, ) -> Result { let secret = auth::cookie_secret(&headers) .ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "auth.required".into()))?; let permit = state .viewer_slots .clone() .try_acquire_owned() .map_err(|_| { ApiError( StatusCode::TOO_MANY_REQUESTS, "events.too_many_viewers".into(), ) })?; let receiver = state.logs.subscribe(); let next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10); let stream = futures_util::stream::unfold( (state, secret, permit, receiver, next_auth_check), |(state, secret, permit, mut receiver, mut next_auth_check)| async move { if tokio::time::Instant::now() >= next_auth_check { let session_seconds = state .config .load() .envelope .config .administration .session_seconds(); match state .store .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds) .await { Ok(Some(_)) => { next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10); } _ => return None, } } let event = tokio::select! { received = receiver.recv() => match received { Ok(entry) => Event::default() .event("log") .data(serde_json::to_string(&entry).unwrap_or_else(|_| "{}".into())), Err(tokio::sync::broadcast::error::RecvError::Lagged(dropped)) => { Event::default().event("gap").data(json!({"dropped": dropped}).to_string()) } Err(tokio::sync::broadcast::error::RecvError::Closed) => return None, }, _ = tokio::time::sleep_until(next_auth_check) => { let session_seconds = state .config .load() .envelope .config .administration .session_seconds(); match state .store .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds) .await { Ok(Some(_)) => { next_auth_check = tokio::time::Instant::now() + Duration::from_secs(10); Event::default().event("heartbeat").data("{}") } _ => return None, } } }; Some(( Ok::(event), (state, secret, permit, receiver, next_auth_check), )) }, ); Ok(Sse::new(stream) .keep_alive(KeepAlive::new().interval(Duration::from_secs(5))) .into_response()) } async fn audit(State(state): State>) -> Result, ApiError> { Ok(Json( json!({"entries":state.store.recent_audit().await.map_err(auth::internal)?,"capacity":crate::db::AUDIT_CAPACITY}), )) } async fn metrics(State(state): State>) -> Response { crate::prometheus::response(&state) } async fn metrics_json(State(state): State>) -> Response { crate::prometheus::json_response(&state) } async fn metrics_summary(State(state): State>) -> Response { crate::prometheus::summary_response(&state) } #[derive(Default, Deserialize)] #[serde(default, deny_unknown_fields)] struct EventsQuery { interval: Option, } async fn events( State(state): State>, headers: HeaderMap, Query(query): Query, ) -> Result { let interval = query.interval.unwrap_or(3); if ![1, 2, 3, 5, 10].contains(&interval) { return Err(ApiError( StatusCode::UNPROCESSABLE_ENTITY, "invalid SSE interval".into(), )); } let secret = auth::cookie_secret(&headers) .ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "auth.required".into()))?; let permit = state .viewer_slots .clone() .try_acquire_owned() .map_err(|_| { ApiError( StatusCode::TOO_MANY_REQUESTS, "events.too_many_viewers".into(), ) })?; let stream = futures_util::stream::unfold( (state, secret, permit, false, interval), |(state, secret, permit, delayed, interval)| async move { if delayed { tokio::time::sleep(Duration::from_secs(interval)).await; } let session_seconds = state .config .load() .envelope .config .administration .session_seconds(); match state .store .session_at_with_browser_lifetime(&secret, auth::now(), session_seconds) .await { Ok(Some(_)) => {} _ => return None, } let payload = json!({"status":live_status_payload(&state), "logs":logs_payload(&state)}) .to_string(); let event = Event::default().event("snapshot").data(payload); Some(( Ok::(event), (state, secret, permit, true, interval), )) }, ); Ok(Sse::new(stream) .keep_alive(KeepAlive::new().interval(Duration::from_secs(10))) .into_response()) } #[derive(Default, Deserialize)] #[serde(default, deny_unknown_fields)] struct TrafficQuery { vhost: Option, } async fn traffic_status( State(state): State>, Query(query): Query, ) -> Result, ApiError> { let snapshot = state.config.load_full(); if query.vhost.as_ref().is_some_and(|id| { id != "*" && !snapshot .envelope .config .vhosts .iter() .any(|vhost| &vhost.id == id) }) { return Err(ApiError( StatusCode::NOT_FOUND, "unknown vhost traffic scope".into(), )); } let scope = query.vhost.as_deref().unwrap_or("*"); let vhosts = snapshot .envelope .config .vhosts .iter() .map(|vhost| json!({"id": vhost.id, "hosts": vhost.frontend.hosts})) .collect::>(); Ok(Json(json!({ "scope": scope, "since": "process_start", "settings": { "capture_user_agent": snapshot.envelope.config.logging.capture_user_agent, "capture_country": snapshot.envelope.config.logging.capture_country }, "geoip": state.geoip.load().view(), "analytics": state.traffic.view(Some(scope)), "vhosts": vhosts }))) } #[derive(Default, Deserialize)] #[serde(default, deny_unknown_fields)] struct CacheQuery { vhost: Option, offset: Option, limit: Option, } async fn cache_status( State(state): State>, Query(query): Query, ) -> Result, ApiError> { let snapshot = state.config.load_full(); if query .vhost .as_ref() .is_some_and(|id| !snapshot.envelope.config.vhosts.iter().any(|v| &v.id == id)) { return Err(ApiError( StatusCode::NOT_FOUND, "unknown vhost cache scope".into(), )); } let scopes: Vec<_> = snapshot.envelope.config.vhosts.iter().map(|vhost| { json!({"id":vhost.id, "hosts":vhost.frontend.hosts, "metrics":state.cache.scope_view(&vhost.id), "settings":vhost.cache.resolve(&snapshot.envelope.config.cache).ok()}) }).collect(); let dashboard = state.cache.dashboard( query.vhost.as_deref(), query.offset.unwrap_or(0).min(16384), query.limit.unwrap_or(50), ); Ok(Json( json!({"settings":snapshot.envelope.config.cache, "metrics":state.cache.view(), "vhosts":scopes, "dashboard":dashboard, "bans":state.cache.bans()}), )) }