webhooksmith_axum/
admin.rs

1//! Admin HTTP endpoints for webhooksmith operational visibility.
2//!
3//! Mount with [`admin`] to expose queue stats, endpoint listing, DLQ inspection,
4//! and bulk DLQ retry — all backed directly by your [`WebhookEngine`].
5//!
6//! ```rust,no_run
7//! use std::sync::Arc;
8//! use axum::Router;
9//! use webhooksmith::WebhookEngine;
10//! use webhooksmith_axum::admin;
11//!
12//! # async fn example(engine: Arc<WebhookEngine>) {
13//! let app: Router = Router::new()
14//!     .nest("/admin", admin(engine));
15//! # }
16//! ```
17//!
18//! # Endpoints
19//!
20//! | Method | Path | Description |
21//! |--------|------|-------------|
22//! | `GET` | `/stats` | Queue stats (pending, delivering, failed, dead, delivered) |
23//! | `GET` | `/endpoints` | Registered endpoints, paginated (`?limit=50&offset=0`) |
24//! | `GET` | `/dlq/{endpoint_id}` | Dead events for an endpoint (paginated) |
25//! | `POST` | `/dlq/{endpoint_id}/retry-all` | Re-queue all dead events + reset circuit |
26//! | `GET` | `/metrics` | Prometheus text exposition (scrape endpoint) |
27//!
28//! Query params for `/dlq/{id}`: `?limit=50&offset=0` (both optional, defaults shown).
29//!
30//! ## Prometheus metrics
31//!
32//! `GET /admin/metrics` returns standard Prometheus text format. Point your
33//! Prometheus scraper at it — no extra configuration needed.
34//!
35//! Metrics exposed:
36//! - `webhooksmith_events{status="pending|delivering|failed|dead|delivered"}` — current event counts by status
37//! - `webhooksmith_endpoints{state="enabled|disabled|circuit_open"}` — current endpoint counts by state
38
39use axum::{
40    Json, Router,
41    extract::{Path, Query, State},
42    http::StatusCode,
43    response::{IntoResponse, Response},
44    routing::{get, post},
45};
46use serde::{Deserialize, Serialize};
47use std::sync::Arc;
48use uuid::Uuid;
49use webhooksmith::WebhookEngine;
50
51// ── State ─────────────────────────────────────────────────────────────────────
52
53#[derive(Clone)]
54struct AdminState {
55    engine: Arc<WebhookEngine>,
56}
57
58// ── Request types ─────────────────────────────────────────────────────────────
59
60#[derive(Deserialize)]
61struct Pagination {
62    #[serde(default = "default_limit")]
63    limit: i64,
64    #[serde(default)]
65    offset: i64,
66}
67
68fn default_limit() -> i64 { 50 }
69
70// ── Response types ────────────────────────────────────────────────────────────
71
72#[derive(Serialize)]
73struct ErrorBody {
74    error: String,
75}
76
77fn err_json(msg: impl Into<String>) -> (StatusCode, Json<ErrorBody>) {
78    (StatusCode::INTERNAL_SERVER_ERROR, Json(ErrorBody { error: msg.into() }))
79}
80
81fn not_found(msg: impl Into<String>) -> (StatusCode, Json<ErrorBody>) {
82    (StatusCode::NOT_FOUND, Json(ErrorBody { error: msg.into() }))
83}
84
85// ── Handlers ──────────────────────────────────────────────────────────────────
86
87async fn get_stats(State(s): State<AdminState>) -> impl IntoResponse {
88    match s.engine.queue_stats().await {
89        Ok(stats) => Json(stats).into_response(),
90        Err(e) => err_json(e.to_string()).into_response(),
91    }
92}
93
94async fn get_endpoints(
95    State(s): State<AdminState>,
96    Query(page): Query<Pagination>,
97) -> impl IntoResponse {
98    match s.engine.list_endpoints_paged(page.limit.clamp(1, 200), page.offset.max(0)).await {
99        Ok(eps) => Json(eps).into_response(),
100        Err(e) => err_json(e.to_string()).into_response(),
101    }
102}
103
104async fn get_dlq(
105    State(s): State<AdminState>,
106    Path(endpoint_id): Path<Uuid>,
107    Query(page): Query<Pagination>,
108) -> impl IntoResponse {
109    // First check the endpoint exists — dead_events_paged returns Ok([]) for unknown ids.
110    match s.engine.endpoint(endpoint_id).await {
111        Ok(None) => return not_found(format!("endpoint {endpoint_id} not found")).into_response(),
112        Err(e)   => return err_json(e.to_string()).into_response(),
113        Ok(Some(_)) => {}
114    }
115    match s.engine.dead_events_paged(endpoint_id, page.limit.clamp(1, 200), page.offset.max(0)).await {
116        Ok(events) => Json(events).into_response(),
117        Err(e)     => err_json(e.to_string()).into_response(),
118    }
119}
120
121#[derive(Serialize)]
122struct RetryAllResponse {
123    retried: u64,
124}
125
126async fn retry_all_dlq(
127    State(s): State<AdminState>,
128    Path(endpoint_id): Path<Uuid>,
129) -> impl IntoResponse {
130    // Validate the endpoint exists before calling retry_all_dead.
131    // retry_all_dead returns Ok(0) for unknown ids with no error signal.
132    match s.engine.endpoint(endpoint_id).await {
133        Ok(None) => return not_found(format!("endpoint {endpoint_id} not found")).into_response(),
134        Err(e)   => return err_json(e.to_string()).into_response(),
135        Ok(Some(_)) => {}
136    }
137    match s.engine.retry_all_dead(endpoint_id).await {
138        Ok(n) => Json(RetryAllResponse { retried: n }).into_response(),
139        Err(e) => err_json(e.to_string()).into_response(),
140    }
141}
142
143// ── Prometheus metrics ────────────────────────────────────────────────────────
144
145/// Render a single Prometheus gauge line.
146fn gauge(out: &mut String, name: &str, labels: &str, value: i64) {
147    out.push_str(&format!("{name}{{{labels}}} {value}\n"));
148}
149
150async fn get_metrics(State(s): State<AdminState>) -> Response {
151    // Fetch queue stats and endpoint list concurrently.
152    let (stats_res, endpoints_res) = tokio::join!(
153        s.engine.queue_stats(),
154        s.engine.list_endpoints(),
155    );
156
157    let (stats, endpoints) = match (stats_res, endpoints_res) {
158        (Ok(st), Ok(ep)) => (st, ep),
159        (Err(e), _) | (_, Err(e)) => {
160            return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response();
161        }
162    };
163
164    let mut body = String::with_capacity(512);
165
166    // ── Event counts ──────────────────────────────────────────────────────────
167    body.push_str("# HELP webhooksmith_events Current number of webhook events by status.\n");
168    body.push_str("# TYPE webhooksmith_events gauge\n");
169    gauge(&mut body, "webhooksmith_events", r#"status="pending""#,    stats.pending);
170    gauge(&mut body, "webhooksmith_events", r#"status="delivering""#, stats.delivering);
171    gauge(&mut body, "webhooksmith_events", r#"status="failed""#,     stats.failed);
172    gauge(&mut body, "webhooksmith_events", r#"status="dead""#,       stats.dead);
173    gauge(&mut body, "webhooksmith_events", r#"status="delivered""#,  stats.delivered);
174
175    // ── Endpoint counts ───────────────────────────────────────────────────────
176    let enabled  = endpoints.iter().filter(|e| e.enabled && e.circuit_open_until.is_none()).count() as i64;
177    let disabled = endpoints.iter().filter(|e| !e.enabled).count() as i64;
178    let circuit_open = endpoints.iter().filter(|e| {
179        e.circuit_open_until.map(|t| t > chrono::Utc::now()).unwrap_or(false)
180    }).count() as i64;
181
182    body.push_str("\n# HELP webhooksmith_endpoints Current number of registered endpoints by state.\n");
183    body.push_str("# TYPE webhooksmith_endpoints gauge\n");
184    gauge(&mut body, "webhooksmith_endpoints", r#"state="enabled""#,      enabled);
185    gauge(&mut body, "webhooksmith_endpoints", r#"state="disabled""#,     disabled);
186    gauge(&mut body, "webhooksmith_endpoints", r#"state="circuit_open""#, circuit_open);
187
188    (
189        StatusCode::OK,
190        [(axum::http::header::CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8")],
191        body,
192    )
193        .into_response()
194}
195
196// ── Public entry point ────────────────────────────────────────────────────────
197
198/// Build an axum [`Router`] exposing admin endpoints backed by `engine`.
199///
200/// Mount it with `router.nest("/admin", admin(engine))` or at the root.
201///
202/// All endpoints return JSON except `/metrics` which returns Prometheus text.
203/// The circuit-breaker state (`consecutive_failures`, `circuit_open_until`)
204/// is visible on each endpoint via `GET /endpoints` and `GET /metrics`.
205pub fn admin(engine: Arc<WebhookEngine>) -> Router {
206    let state = AdminState { engine };
207    Router::new()
208        .route("/stats", get(get_stats))
209        .route("/endpoints", get(get_endpoints))
210        .route("/dlq/:endpoint_id", get(get_dlq))
211        .route("/dlq/:endpoint_id/retry-all", post(retry_all_dlq))
212        .route("/metrics", get(get_metrics))
213        .with_state(state)
214}
webhooksmith_axum/admin.rs

webhooksmith_axum/
admin.rs
