featherbit/admin/
consumers.rs

1//! Admin API endpoints for consumer CRUD. Mutations rewrite the in-memory
2//! gateway config and trigger `SharedState::reload`, which rebuilds and
3//! atomically swaps the consumer store (no graph recompile is strictly
4//! needed, but reload keeps the semantics uniform with routes/policies).
5
6use std::sync::Arc;
7
8use axum::extract::{Path, State};
9use axum::http::StatusCode;
10use axum::response::IntoResponse;
11use axum::routing::get;
12use axum::{Json, Router};
13
14use crate::consumers::ConsumerConfig;
15use crate::state::SharedState;
16
17/// Builds the router for the `/api/consumers` endpoints.
18pub fn router() -> Router<Arc<SharedState>> {
19    Router::new()
20        .route("/api/consumers", get(list_consumers).post(create_consumer))
21        .route(
22            "/api/consumers/{name}",
23            get(get_consumer)
24                .put(upsert_consumer)
25                .delete(delete_consumer),
26        )
27}
28
29/// `GET /api/consumers` — returns all declared consumers as a JSON array.
30///
31/// Note: responses include credentials verbatim — the admin API is the
32/// management plane and is Basic-Auth protected.
33async fn list_consumers(State(state): State<Arc<SharedState>>) -> impl IntoResponse {
34    let gw = state.gateway.read().await;
35    Json(&gw.consumers).into_response()
36}
37
38/// `GET /api/consumers/{name}` — returns the named consumer as JSON.
39///
40/// Errors: `404 Not Found` if no consumer with that name exists.
41async fn get_consumer(
42    State(state): State<Arc<SharedState>>,
43    Path(name): Path<String>,
44) -> impl IntoResponse {
45    let gw = state.gateway.read().await;
46    match gw.consumers.iter().find(|c| c.name == name) {
47        Some(consumer) => Json(consumer).into_response(),
48        None => (
49            StatusCode::NOT_FOUND,
50            Json(serde_json::json!({"error": "not_found"})),
51        )
52            .into_response(),
53    }
54}
55
56/// `POST /api/consumers` — creates a new consumer from the JSON body, then
57/// rebuilds the consumer store. Returns `201 Created` on success.
58///
59/// Errors: `409 Conflict` if the name is taken; `400 Bad Request` if the
60/// store rebuild rejects the config (duplicate credentials, malformed
61/// credential objects) — the previous store stays active.
62async fn create_consumer(
63    State(state): State<Arc<SharedState>>,
64    Json(consumer): Json<ConsumerConfig>,
65) -> impl IntoResponse {
66    let candidate = {
67        let gw = state.gateway.read().await;
68        if gw.consumers.iter().any(|c| c.name == consumer.name) {
69            return (
70                StatusCode::CONFLICT,
71                Json(serde_json::json!({"error": "consumer already exists"})),
72            )
73                .into_response();
74        }
75        let mut candidate = gw.clone();
76        candidate.consumers.push(consumer);
77        candidate
78    };
79
80    match state.config_store.clone().commit(&state, candidate).await {
81        Ok(_) => (
82            StatusCode::CREATED,
83            Json(serde_json::json!({"status": "created"})),
84        )
85            .into_response(),
86        Err(e) => (
87            StatusCode::BAD_REQUEST,
88            Json(serde_json::json!({"error": e})),
89        )
90            .into_response(),
91    }
92}
93
94/// `PUT /api/consumers/{name}` — creates or replaces the named consumer (the
95/// path name overrides any name in the body), matching the policies
96/// endpoint's upsert semantics. Returns `{"status": "updated"}`.
97///
98/// Errors: `400 Bad Request` if the store rebuild rejects the config.
99async fn upsert_consumer(
100    State(state): State<Arc<SharedState>>,
101    Path(name): Path<String>,
102    Json(mut consumer): Json<ConsumerConfig>,
103) -> impl IntoResponse {
104    consumer.name = name.clone();
105    let candidate = {
106        let gw = state.gateway.read().await;
107        let mut candidate = gw.clone();
108        if let Some(existing) = candidate.consumers.iter_mut().find(|c| c.name == name) {
109            *existing = consumer;
110        } else {
111            candidate.consumers.push(consumer);
112        }
113        candidate
114    };
115
116    match state.config_store.clone().commit(&state, candidate).await {
117        Ok(_) => Json(serde_json::json!({"status": "updated"})).into_response(),
118        Err(e) => (
119            StatusCode::BAD_REQUEST,
120            Json(serde_json::json!({"error": e})),
121        )
122            .into_response(),
123    }
124}
125
126/// `DELETE /api/consumers/{name}` — removes the named consumer and rebuilds
127/// the store. Returns `{"status": "deleted"}`.
128///
129/// Errors: `404 Not Found` if the consumer does not exist.
130async fn delete_consumer(
131    State(state): State<Arc<SharedState>>,
132    Path(name): Path<String>,
133) -> impl IntoResponse {
134    let candidate = {
135        let gw = state.gateway.read().await;
136        let mut candidate = gw.clone();
137        let before = candidate.consumers.len();
138        candidate.consumers.retain(|c| c.name != name);
139        if candidate.consumers.len() == before {
140            return (
141                StatusCode::NOT_FOUND,
142                Json(serde_json::json!({"error": "not_found"})),
143            )
144                .into_response();
145        }
146        candidate
147    };
148
149    match state.config_store.clone().commit(&state, candidate).await {
150        Ok(_) => Json(serde_json::json!({"status": "deleted"})).into_response(),
151        Err(e) => (
152            StatusCode::BAD_REQUEST,
153            Json(serde_json::json!({"error": e})),
154        )
155            .into_response(),
156    }
157}
featherbit/admin/consumers.rs

featherbit/admin/
consumers.rs
