featherbit/admin/
stores.rs

1//! Admin CRUD for the `stores:` resource (named redis/valkey connections).
2//!
3//! Responses always carry the RAW stored config — `${ENV}` placeholders are
4//! never resolved here (they resolve only when a client is built). Deleting
5//! a store still referenced by any plugin config is rejected with
6//! `409 {"error":"in_use","referrers":[...]}` — a new envelope, since the
7//! reference is not discoverable via graph recompilation (a store used only
8//! by a shared plugin_config compiles fine without the node being wired).
9//! The same guard also covers `acme.storage` (`system.yaml`), since an ACME
10//! `type: store` storage config references a store outside `gateway.yaml`
11//! entirely and would otherwise go undetected.
12
13use std::sync::Arc;
14
15use axum::extract::{Path, State};
16use axum::http::StatusCode;
17use axum::response::IntoResponse;
18use axum::routing::get;
19use axum::{Json, Router};
20
21use crate::config::{GatewayConfig, NodeConfig, StoreConfig};
22use crate::state::SharedState;
23
24pub fn router() -> Router<Arc<SharedState>> {
25    Router::new()
26        .route("/api/stores", get(list_stores).post(create_store))
27        .route(
28            "/api/stores/{name}",
29            get(get_store).put(update_store).delete(delete_store),
30        )
31        .route("/api/stores/{name}/ping", axum::routing::post(ping_store))
32}
33
34async fn list_stores(State(state): State<Arc<SharedState>>) -> impl IntoResponse {
35    let gw = state.gateway.read().await;
36    Json(gw.stores.clone()).into_response()
37}
38
39async fn get_store(
40    State(state): State<Arc<SharedState>>,
41    Path(name): Path<String>,
42) -> impl IntoResponse {
43    let gw = state.gateway.read().await;
44    match gw.stores.iter().find(|s| s.name == name) {
45        Some(s) => Json(s.clone()).into_response(),
46        None => (
47            StatusCode::NOT_FOUND,
48            Json(serde_json::json!({"error": "not_found"})),
49        )
50            .into_response(),
51    }
52}
53
54async fn create_store(
55    State(state): State<Arc<SharedState>>,
56    Json(store): Json<StoreConfig>,
57) -> impl IntoResponse {
58    let candidate = {
59        let gw = state.gateway.read().await;
60        if gw.stores.iter().any(|s| s.name == store.name) {
61            return (
62                StatusCode::CONFLICT,
63                Json(serde_json::json!({"error": "store already exists"})),
64            )
65                .into_response();
66        }
67        let mut candidate = gw.clone();
68        candidate.stores.push(store);
69        candidate
70    };
71    match state.config_store.clone().commit(&state, candidate).await {
72        Ok(_) => (
73            StatusCode::CREATED,
74            Json(serde_json::json!({"status": "created"})),
75        )
76            .into_response(),
77        Err(e) => (
78            StatusCode::BAD_REQUEST,
79            Json(serde_json::json!({"error": e})),
80        )
81            .into_response(),
82    }
83}
84
85async fn update_store(
86    State(state): State<Arc<SharedState>>,
87    Path(name): Path<String>,
88    Json(mut store): Json<StoreConfig>,
89) -> impl IntoResponse {
90    store.name = name.clone();
91    let candidate = {
92        let gw = state.gateway.read().await;
93        let mut candidate = gw.clone();
94        if let Some(existing) = candidate.stores.iter_mut().find(|s| s.name == name) {
95            *existing = store;
96        } else {
97            candidate.stores.push(store);
98        }
99        candidate
100    };
101    match state.config_store.clone().commit(&state, candidate).await {
102        Ok(_) => Json(serde_json::json!({"status": "updated"})).into_response(),
103        Err(e) => (
104            StatusCode::BAD_REQUEST,
105            Json(serde_json::json!({"error": e})),
106        )
107            .into_response(),
108    }
109}
110
111async fn delete_store(
112    State(state): State<Arc<SharedState>>,
113    Path(name): Path<String>,
114) -> impl IntoResponse {
115    let candidate = {
116        let gw = state.gateway.read().await;
117        let mut referrers = store_referrers(&gw, &name);
118        if let Some(crate::config::AcmeConfig {
119            storage: crate::config::AcmeStorageConfig::Store { store, .. },
120            ..
121        }) = &state.system.acme
122        {
123            if store == &name {
124                referrers.push("acme.storage (system.yaml)".to_string());
125            }
126        }
127        if !referrers.is_empty() {
128            return (
129                StatusCode::CONFLICT,
130                Json(serde_json::json!({"error": "in_use", "referrers": referrers})),
131            )
132                .into_response();
133        }
134        let mut candidate = gw.clone();
135        let before = candidate.stores.len();
136        candidate.stores.retain(|s| s.name != name);
137        if candidate.stores.len() == before {
138            return (
139                StatusCode::NOT_FOUND,
140                Json(serde_json::json!({"error": "not_found"})),
141            )
142                .into_response();
143        }
144        candidate
145    };
146    match state.config_store.clone().commit(&state, candidate).await {
147        Ok(_) => Json(serde_json::json!({"status": "deleted"})).into_response(),
148        Err(e) => (
149            StatusCode::BAD_REQUEST,
150            Json(serde_json::json!({"error": e})),
151        )
152            .into_response(),
153    }
154}
155
156/// Connectivity check: resolves the store's config (env placeholders included)
157/// and PINGs it, bounded by the store's own connect_timeout_ms. The response
158/// never echoes resolved connection details — only latency and version.
159#[cfg(feature = "redis-store")]
160async fn ping_store(
161    State(state): State<Arc<SharedState>>,
162    Path(name): Path<String>,
163) -> impl IntoResponse {
164    let cfg = {
165        let gw = state.gateway.read().await;
166        match gw.stores.iter().find(|s| s.name == name) {
167            Some(s) => s.clone(),
168            None => {
169                return (
170                    StatusCode::NOT_FOUND,
171                    Json(serde_json::json!({"error": "not_found"})),
172                )
173                    .into_response()
174            }
175        }
176    };
177    let client = match crate::stores::redis_store::RedisStoreClient::build(&cfg) {
178        Ok(c) => c,
179        Err(e) => {
180            return (
181                StatusCode::BAD_REQUEST,
182                Json(serde_json::json!({"error": e})),
183            )
184                .into_response()
185        }
186    };
187    match tokio::time::timeout(client.connect_timeout(), client.ping()).await {
188        Err(_) => (
189            StatusCode::GATEWAY_TIMEOUT,
190            Json(serde_json::json!({
191                "error": "ping_timeout",
192                "message": format!("no reply within connect_timeout_ms ({}ms)", cfg.connect_timeout_ms),
193            })),
194        )
195            .into_response(),
196        Ok(Err(e)) => (
197            StatusCode::BAD_GATEWAY,
198            Json(serde_json::json!({"error": e})),
199        )
200            .into_response(),
201        Ok(Ok(info)) => Json(serde_json::json!({
202            "status": "ok",
203            "latency_ms": info.latency_ms,
204            "version": info.version,
205        }))
206        .into_response(),
207    }
208}
209
210#[cfg(not(feature = "redis-store"))]
211async fn ping_store(
212    State(_state): State<Arc<SharedState>>,
213    Path(_name): Path<String>,
214) -> impl IntoResponse {
215    (
216        StatusCode::NOT_IMPLEMENTED,
217        Json(serde_json::json!({
218            "error": "this binary was built without the redis-store feature"
219        })),
220    )
221        .into_response()
222}
223
224/// Everything that references store `name`: node config `store:` keys (flat,
225/// for limit-count/workflow), nested `session.store` (session plugins —
226/// wired in a later plan, scanned now so the guard never lags the feature),
227/// and `workflow`'s per-rule action params (`config.rules[].actions[][1].store`
228/// — actions are `[name, params]` pairs; see `plugins/native/workflow.rs`).
229pub(crate) fn store_referrers(gw: &GatewayConfig, name: &str) -> Vec<String> {
230    fn config_references(
231        config: &std::collections::HashMap<String, serde_json::Value>,
232        name: &str,
233    ) -> bool {
234        config.get("store").and_then(|v| v.as_str()) == Some(name)
235            || config.get("session_store").and_then(|v| v.as_str()) == Some(name)
236            || config
237                .get("session")
238                .and_then(|v| v.get("store"))
239                .and_then(|v| v.as_str())
240                == Some(name)
241            || config
242                .get("rules")
243                .and_then(|v| v.as_array())
244                .is_some_and(|rules| rules.iter().any(|rule| rule_references(rule, name)))
245    }
246
247    fn rule_references(rule: &serde_json::Value, name: &str) -> bool {
248        rule.get("actions")
249            .and_then(|v| v.as_array())
250            .is_some_and(|actions| actions.iter().any(|action| action_references(action, name)))
251    }
252
253    fn action_references(action: &serde_json::Value, name: &str) -> bool {
254        action
255            .as_array()
256            .and_then(|a| a.get(1))
257            .and_then(|params| params.as_object())
258            .and_then(|params| params.get("store"))
259            .and_then(|v| v.as_str())
260            == Some(name)
261    }
262    fn scan_nodes(nodes: &[NodeConfig], owner: &str, name: &str, out: &mut Vec<String>) {
263        for n in nodes {
264            if config_references(&n.config, name) {
265                out.push(format!("{} node '{}'", owner, n.id));
266            }
267        }
268    }
269    let mut refs = Vec::new();
270    for p in &gw.policies {
271        scan_nodes(&p.nodes, &format!("policy '{}'", p.name), name, &mut refs);
272    }
273    for s in &gw.supernodes {
274        scan_nodes(
275            &s.nodes,
276            &format!("supernode '{}'", s.name),
277            name,
278            &mut refs,
279        );
280    }
281    for pc in &gw.plugin_configs {
282        if config_references(&pc.config, name) {
283            refs.push(format!("plugin_config '{}'", pc.name));
284        }
285    }
286    refs
287}
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292    use crate::config::{GatewayConfig, SystemConfig};
293    use crate::config_store::FileConfigStore;
294    use axum::body::Body;
295    use axum::http::{Request, StatusCode};
296    use tower::ServiceExt;
297
298    fn test_state(gateway_yaml: &str) -> Arc<SharedState> {
299        let system: SystemConfig = serde_yaml::from_str("{}").unwrap();
300        let gateway: GatewayConfig = serde_yaml::from_str(gateway_yaml).unwrap();
301        Arc::new(
302            SharedState::new(
303                system,
304                gateway,
305                None,
306                Arc::new(FileConfigStore::new(std::path::PathBuf::from(
307                    "gateway.yaml",
308                ))),
309            )
310            .unwrap(),
311        )
312    }
313
314    fn app(state: Arc<SharedState>) -> Router {
315        router().with_state(state)
316    }
317
318    const VALID_STORE: &str = r#"{
319        "name": "s1",
320        "type": "redis",
321        "url": "${TEST_STORES_URL:-redis://127.0.0.1:6379}"
322    }"#;
323
324    async fn send(state: &Arc<SharedState>, req: Request<Body>) -> (StatusCode, serde_json::Value) {
325        let resp = app(state.clone()).oneshot(req).await.unwrap();
326        let status = resp.status();
327        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
328            .await
329            .unwrap();
330        let v = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
331        (status, v)
332    }
333
334    #[tokio::test]
335    async fn test_crud_roundtrip_keeps_placeholders_raw() {
336        let state = test_state("{}");
337        let (status, _) = send(
338            &state,
339            Request::post("/api/stores")
340                .header("content-type", "application/json")
341                .body(Body::from(VALID_STORE))
342                .unwrap(),
343        )
344        .await;
345        assert_eq!(status, StatusCode::CREATED);
346
347        // Duplicate create → 409.
348        let (status, _) = send(
349            &state,
350            Request::post("/api/stores")
351                .header("content-type", "application/json")
352                .body(Body::from(VALID_STORE))
353                .unwrap(),
354        )
355        .await;
356        assert_eq!(status, StatusCode::CONFLICT);
357
358        // GET serves the RAW placeholder, never a resolved value.
359        let (status, body) = send(
360            &state,
361            Request::get("/api/stores/s1").body(Body::empty()).unwrap(),
362        )
363        .await;
364        assert_eq!(status, StatusCode::OK);
365        assert_eq!(
366            body["url"].as_str().unwrap(),
367            "${TEST_STORES_URL:-redis://127.0.0.1:6379}"
368        );
369
370        // PUT upserts; invalid type is rejected by commit-time validation.
371        let (status, body) = send(
372            &state,
373            Request::put("/api/stores/s1")
374                .header("content-type", "application/json")
375                .body(Body::from(
376                    r#"{"name":"s1","type":"memcached","url":"redis://x"}"#,
377                ))
378                .unwrap(),
379        )
380        .await;
381        assert_eq!(status, StatusCode::BAD_REQUEST);
382        assert!(
383            body["error"].as_str().unwrap().contains("unknown type"),
384            "{body}"
385        );
386
387        // DELETE, then 404.
388        let (status, _) = send(
389            &state,
390            Request::delete("/api/stores/s1")
391                .body(Body::empty())
392                .unwrap(),
393        )
394        .await;
395        assert_eq!(status, StatusCode::OK);
396        let (status, _) = send(
397            &state,
398            Request::get("/api/stores/s1").body(Body::empty()).unwrap(),
399        )
400        .await;
401        assert_eq!(status, StatusCode::NOT_FOUND);
402    }
403
404    #[tokio::test]
405    async fn test_delete_referenced_store_is_409_with_referrers() {
406        let state = test_state(
407            r#"
408stores:
409  - name: s1
410    type: redis
411    url: redis://127.0.0.1:6379
412plugin_configs:
413  - name: shared-lc
414    type: limit-count
415    config: { count: 1, time_window: 60, policy: redis, store: s1 }
416"#,
417        );
418        let (status, body) = send(
419            &state,
420            Request::delete("/api/stores/s1")
421                .body(Body::empty())
422                .unwrap(),
423        )
424        .await;
425        assert_eq!(status, StatusCode::CONFLICT);
426        assert_eq!(body["error"], "in_use");
427        let refs: Vec<String> = body["referrers"]
428            .as_array()
429            .unwrap()
430            .iter()
431            .map(|v| v.as_str().unwrap().to_string())
432            .collect();
433        assert!(
434            refs.contains(&"plugin_config 'shared-lc'".to_string()),
435            "{refs:?}"
436        );
437    }
438
439    #[tokio::test]
440    async fn test_delete_referenced_by_workflow_action_is_409_with_referrers() {
441        let state = test_state(
442            r#"
443stores:
444  - name: s1
445    type: redis
446    url: redis://127.0.0.1:6379
447plugin_configs:
448  - name: shared-wf
449    type: workflow
450    config:
451      rules:
452        - case: [["uri", "==", "/x"]]
453          actions:
454            - ["limit-count", { count: 1, time_window: 1, policy: redis, store: s1 }]
455"#,
456        );
457        let (status, body) = send(
458            &state,
459            Request::delete("/api/stores/s1")
460                .body(Body::empty())
461                .unwrap(),
462        )
463        .await;
464        assert_eq!(status, StatusCode::CONFLICT);
465        assert_eq!(body["error"], "in_use");
466        let refs: Vec<String> = body["referrers"]
467            .as_array()
468            .unwrap()
469            .iter()
470            .map(|v| v.as_str().unwrap().to_string())
471            .collect();
472        assert!(
473            refs.contains(&"plugin_config 'shared-wf'".to_string()),
474            "{refs:?}"
475        );
476    }
477
478    /// The UI's SchemaForm writes the FLAT `session_store` key (the nested
479    /// `session.store` form is hand-written YAML); the delete guard must see
480    /// both, or a store referenced only by a UI-authored session plugin
481    /// config could be deleted out from under it.
482    #[tokio::test]
483    async fn test_delete_store_referenced_by_flat_session_store_is_409() {
484        let state = test_state(
485            r#"
486stores:
487  - name: s1
488    type: redis
489    url: redis://127.0.0.1:6379
490plugin_configs:
491  - name: ui-oidc
492    type: openid-connect
493    config: { session_storage: redis, session_store: s1 }
494"#,
495        );
496        let (status, body) = send(
497            &state,
498            Request::delete("/api/stores/s1")
499                .body(Body::empty())
500                .unwrap(),
501        )
502        .await;
503        assert_eq!(status, StatusCode::CONFLICT, "{body}");
504        assert_eq!(body["error"], "in_use");
505        let refs: Vec<String> = body["referrers"]
506            .as_array()
507            .unwrap()
508            .iter()
509            .map(|v| v.as_str().unwrap().to_string())
510            .collect();
511        assert!(
512            refs.contains(&"plugin_config 'ui-oidc'".to_string()),
513            "{refs:?}"
514        );
515    }
516
517    #[tokio::test]
518    #[cfg(feature = "redis-store")]
519    async fn test_delete_store_used_by_acme_storage_is_409() {
520        let system: SystemConfig = serde_yaml::from_str(
521            "acme:\n  terms_of_service_agreed: true\n  storage:\n    type: store\n    store: s1\n    encryption_key: k\n",
522        )
523        .unwrap();
524        let gateway: GatewayConfig = serde_yaml::from_str(
525            "stores:\n  - name: s1\n    type: redis\n    url: redis://127.0.0.1:6379\n",
526        )
527        .unwrap();
528        let state = Arc::new(
529            SharedState::new(
530                system,
531                gateway,
532                None,
533                Arc::new(FileConfigStore::new("gateway.yaml".into())),
534            )
535            .unwrap(),
536        );
537        let req = Request::builder()
538            .method("DELETE")
539            .uri("/api/stores/s1")
540            .body(Body::empty())
541            .unwrap();
542        let resp = app(state).oneshot(req).await.unwrap();
543        assert_eq!(resp.status(), StatusCode::CONFLICT);
544        let body: serde_json::Value = serde_json::from_slice(
545            &axum::body::to_bytes(resp.into_body(), usize::MAX)
546                .await
547                .unwrap(),
548        )
549        .unwrap();
550        assert_eq!(body["error"], "in_use");
551        assert!(body["referrers"]
552            .as_array()
553            .unwrap()
554            .iter()
555            .any(|r| r.as_str().unwrap().contains("acme.storage")));
556    }
557
558    /// Ping on an unknown store is 404; on an unreachable store it is a 502
559    /// or 504 within the configured timeout — never a hang, never a 200.
560    #[tokio::test]
561    #[cfg(feature = "redis-store")]
562    async fn test_ping_unknown_and_unreachable() {
563        // Port 1 is reserved/closed; 300ms timeout keeps the test fast.
564        let state = test_state(
565            "stores:\n  - name: dead\n    type: redis\n    url: redis://127.0.0.1:1\n    connect_timeout_ms: 300\n",
566        );
567        let (status, _) = send(
568            &state,
569            Request::post("/api/stores/nope/ping")
570                .body(Body::empty())
571                .unwrap(),
572        )
573        .await;
574        assert_eq!(status, StatusCode::NOT_FOUND);
575
576        let (status, body) = send(
577            &state,
578            Request::post("/api/stores/dead/ping")
579                .body(Body::empty())
580                .unwrap(),
581        )
582        .await;
583        assert!(
584            status == StatusCode::BAD_GATEWAY || status == StatusCode::GATEWAY_TIMEOUT,
585            "{status} {body}"
586        );
587    }
588
589    /// Live ping; skipped unless FEATHERBIT_TEST_REDIS_URL is set.
590    #[tokio::test]
591    #[cfg(feature = "redis-store")]
592    async fn test_ping_live() {
593        let Ok(url) = std::env::var("FEATHERBIT_TEST_REDIS_URL") else {
594            eprintln!("skipping test_ping_live: FEATHERBIT_TEST_REDIS_URL not set");
595            return;
596        };
597        let state = test_state(&format!(
598            "stores:\n  - name: live\n    type: redis\n    url: {url}\n"
599        ));
600        let (status, body) = send(
601            &state,
602            Request::post("/api/stores/live/ping")
603                .body(Body::empty())
604                .unwrap(),
605        )
606        .await;
607        assert_eq!(status, StatusCode::OK, "{body}");
608        assert_eq!(body["status"], "ok");
609        assert!(body["latency_ms"].is_u64(), "{body}");
610        assert!(body["version"].is_string(), "{body}");
611    }
612}
featherbit/admin/stores.rs

featherbit/admin/
stores.rs
