agentd/store/
http.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! The **HTTP store**: the four store operations as plain HTTP
3//! requests built from `store.http.{get,put,list,delete}` templates —
4//! `GET/PUT/POST/DELETE {url}` with an optional JSON `body`, extraction from
5//! `body`/`status`/`headers`. HTTPS (loopback `http://` for dev), headers may
6//! carry `{{secret:…}}` references (resolved at dial time, never logged), the
7//! idempotency key rides as `Idempotency-Key`; `conflict_status` (default
8//! `409`) maps to `Conflict`, `404` on a read to absent.
9
10use super::mapping::{self, Vars};
11use super::{KeySeq, PutOutcome, Store, StoreError};
12use crate::config::v2::{HttpOp, StoreHttp};
13use crate::net::http::{self, Url};
14use serde_json::{Value, json};
15use std::time::Duration;
16
17pub struct HttpStore {
18    cfg: StoreHttp,
19    timeout: Duration,
20}
21
22impl HttpStore {
23    pub fn new(cfg: StoreHttp, timeout: Duration) -> Result<HttpStore, StoreError> {
24        Url::parse(&cfg.base_url)
25            .map_err(|e| StoreError::Mapping(format!("store.http.base_url: {e}")))?;
26        if cfg.get.is_none() || cfg.put.is_none() {
27            return Err(StoreError::Mapping(
28                "store.http needs `get` and `put` operations".into(),
29            ));
30        }
31        Ok(HttpStore { cfg, timeout })
32    }
33
34    fn vars(&self, key: &str, seq: Option<u64>, envelope: Option<&Value>) -> Vars {
35        let mut parts = key.splitn(4, '/');
36        let prefix = parts.next().unwrap_or("");
37        let instance = parts.next().unwrap_or("");
38        let kind = parts.next().unwrap_or("");
39        let id = parts.next().unwrap_or("");
40        let mut v = mapping::store_vars(key, seq, prefix, instance, envelope, kind, id);
41        v.insert(
42            "base_url".into(),
43            Value::String(self.cfg.base_url.trim_end_matches('/').to_string()),
44        );
45        v
46    }
47
48    /// Perform one operation; returns `(status, extraction ctx)`.
49    fn request(
50        &self,
51        op: &HttpOp,
52        vars: &Vars,
53        idempotency: Option<String>,
54    ) -> Result<(u16, Value), StoreError> {
55        let url_text = mapping::render_text(&op.url, vars)
56            .map_err(|e| StoreError::Mapping(format!("store.http url: {e}")))?;
57        let url = Url::parse(&url_text)
58            .map_err(|e| StoreError::Mapping(format!("store.http url {url_text:?}: {e}")))?;
59        let method = op.method.as_deref().unwrap_or("GET").to_ascii_uppercase();
60        let body: Vec<u8> = match &op.body {
61            Some(t) => {
62                let v = mapping::render_json(t, vars)
63                    .map_err(|e| StoreError::Mapping(format!("store.http body: {e}")))?;
64                serde_json::to_vec(&v).unwrap_or_default()
65            }
66            None => Vec::new(),
67        };
68        // Resolve header templates (secret refs → values) at dial time.
69        let env = |k: &str| std::env::var(k).ok();
70        let mut headers: Vec<(String, String)> = Vec::new();
71        for (k, v) in &self.cfg.headers {
72            let val = crate::sec::secret::resolve(v, &env)
73                .map_err(|e| StoreError::Mapping(format!("store.http header {k}: {e}")))?;
74            headers.push((k.clone(), val));
75        }
76        if !body.is_empty() {
77            headers.push(("Content-Type".into(), "application/json".into()));
78        }
79        headers.push(("Accept".into(), "application/json".into()));
80        if let Some(idem) = idempotency {
81            headers.push(("Idempotency-Key".into(), idem));
82        }
83        let hdrs: Vec<(&str, &str)> = headers
84            .iter()
85            .map(|(k, v)| (k.as_str(), v.as_str()))
86            .collect();
87        let mut stream = connect(&url, self.timeout)?;
88        let resp = http::send(
89            stream.as_mut(),
90            &url.host_header(),
91            &method,
92            &url.path,
93            &hdrs,
94            &body,
95        )
96        .map_err(|e| StoreError::Io(format!("{method} {url_text}: {e}")))?;
97        let body_json: Value = serde_json::from_slice(&resp.body)
98            .unwrap_or_else(|_| Value::String(resp.body_str().into_owned()));
99        let hmap: serde_json::Map<String, Value> = resp
100            .headers
101            .iter()
102            .map(|(k, v)| (k.clone(), Value::String(v.clone())))
103            .collect();
104        Ok((
105            resp.status,
106            json!({ "body": body_json, "status": resp.status, "headers": hmap }),
107        ))
108    }
109}
110
111fn connect(url: &Url, timeout: Duration) -> Result<Box<dyn http::Stream>, StoreError> {
112    let tcp = http::connect_tcp(&url.host, url.port, timeout)
113        .map_err(|e| StoreError::Io(format!("connect {}: {e}", url.host)))?;
114    if url.is_tls() {
115        #[cfg(feature = "tls")]
116        {
117            let s = crate::net::tls::connect(tcp, &url.host, None)
118                .map_err(|e| StoreError::Io(format!("tls {}: {e}", url.host)))?;
119            return Ok(Box::new(s));
120        }
121        #[cfg(not(feature = "tls"))]
122        {
123            return Err(StoreError::Io("https store requires --features tls".into()));
124        }
125    }
126    Ok(Box::new(tcp))
127}
128
129impl Store for HttpStore {
130    fn put(&self, key: &str, seq: u64, envelope: &Value) -> Result<PutOutcome, StoreError> {
131        let op = self.cfg.put.as_ref().expect("validated");
132        let vars = self.vars(key, Some(seq), Some(envelope));
133        let (status, ctx) = self.request(op, &vars, Some(format!("{key}#{seq}")))?;
134        let conflict = op.conflict_status.unwrap_or(409);
135        if status == conflict {
136            let latest = ctx["body"]
137                .get("latest")
138                .or_else(|| ctx["body"].get("seq"))
139                .and_then(Value::as_u64);
140            return Ok(PutOutcome::Conflict { latest_seq: latest });
141        }
142        if (200..300).contains(&status) {
143            return Ok(PutOutcome::Ok);
144        }
145        Err(StoreError::Io(format!(
146            "put {key}: HTTP {status}: {}",
147            ctx["body"]
148        )))
149    }
150
151    fn get(&self, key: &str, seq: Option<u64>) -> Result<Option<Value>, StoreError> {
152        let op = self.cfg.get.as_ref().expect("validated");
153        let vars = self.vars(key, seq, None);
154        let (status, ctx) = self.request(op, &vars, None)?;
155        if status == 404 {
156            return Ok(None);
157        }
158        if !(200..300).contains(&status) {
159            return Err(StoreError::Io(format!("get {key}: HTTP {status}")));
160        }
161        let v = match &op.value {
162            Some(x) => mapping::extract(x, &ctx).map_err(|e| StoreError::Mapping(e.0))?,
163            None => Some(ctx["body"].clone()),
164        };
165        Ok(v.filter(|v| !v.is_null()))
166    }
167
168    fn list(&self, prefix: &str) -> Result<Vec<KeySeq>, StoreError> {
169        let Some(op) = &self.cfg.list else {
170            return Err(StoreError::Unsupported("list"));
171        };
172        let mut vars = self.vars(prefix, None, None);
173        vars.insert("prefix".into(), Value::String(prefix.to_string()));
174        let (status, ctx) = self.request(op, &vars, None)?;
175        if !(200..300).contains(&status) {
176            return Err(StoreError::Io(format!("list {prefix}: HTTP {status}")));
177        }
178        let keys = match &op.keys {
179            Some(x) => mapping::extract(x, &ctx).map_err(|e| StoreError::Mapping(e.0))?,
180            None => Some(ctx["body"]["keys"].clone()),
181        };
182        let mut out = Vec::new();
183        if let Some(Value::Array(items)) = keys {
184            for it in items {
185                match it {
186                    Value::String(k) => out.push(KeySeq { key: k, seq: None }),
187                    Value::Object(o) => {
188                        if let Some(k) = o.get("key").and_then(Value::as_str) {
189                            out.push(KeySeq {
190                                key: k.to_string(),
191                                seq: o.get("seq").and_then(Value::as_u64),
192                            });
193                        }
194                    }
195                    _ => {}
196                }
197            }
198        }
199        Ok(out)
200    }
201
202    fn delete(&self, key: &str) -> Result<(), StoreError> {
203        let Some(op) = &self.cfg.delete else {
204            return Err(StoreError::Unsupported("delete"));
205        };
206        let vars = self.vars(key, None, None);
207        let (status, _) = self.request(op, &vars, None)?;
208        if (200..300).contains(&status) || status == 404 {
209            return Ok(());
210        }
211        Err(StoreError::Io(format!("delete {key}: HTTP {status}")))
212    }
213
214    fn kind(&self) -> &'static str {
215        "http"
216    }
217}
218
219#[cfg(test)]
220mod tests {
221    use super::*;
222    use std::collections::BTreeMap;
223    use std::io::{BufRead, BufReader, Read, Write};
224    use std::net::TcpListener;
225    use std::sync::{Arc, Mutex};
226
227    /// A tiny KV server: `PUT /kv/<key>?seq=N` (409 on stale seq, echoing
228    /// `{"latest": n}`), `GET /kv/<key>` (404 when absent), `GET /kv?prefix=…`,
229    /// `DELETE /kv/<key>`; records the Idempotency-Key + auth headers it saw.
230    /// (method, path, headers) per request.
231    type Seen = Arc<Mutex<Vec<(String, String, Vec<(String, String)>)>>>;
232
233    fn spawn_kv() -> (String, Seen) {
234        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
235        let addr = listener.local_addr().unwrap();
236        let seen: Seen = Arc::new(Mutex::new(Vec::new()));
237        let seen_t = seen.clone();
238        let data: Arc<Mutex<BTreeMap<String, (u64, Value)>>> =
239            Arc::new(Mutex::new(BTreeMap::new()));
240        std::thread::spawn(move || {
241            for conn in listener.incoming() {
242                let Ok(mut s) = conn else { continue };
243                let mut r = BufReader::new(s.try_clone().unwrap());
244                let mut line = String::new();
245                if r.read_line(&mut line).unwrap_or(0) == 0 {
246                    continue;
247                }
248                let mut parts = line.split_whitespace();
249                let method = parts.next().unwrap_or("").to_string();
250                let target = parts.next().unwrap_or("").to_string();
251                let mut headers = Vec::new();
252                let mut clen = 0usize;
253                loop {
254                    let mut h = String::new();
255                    if r.read_line(&mut h).unwrap_or(0) == 0 {
256                        break;
257                    }
258                    let t = h.trim_end();
259                    if t.is_empty() {
260                        break;
261                    }
262                    if let Some((k, v)) = t.split_once(':') {
263                        let k = k.trim().to_ascii_lowercase();
264                        if k == "content-length" {
265                            clen = v.trim().parse().unwrap_or(0);
266                        }
267                        headers.push((k, v.trim().to_string()));
268                    }
269                }
270                let mut body = vec![0u8; clen];
271                let _ = r.read_exact(&mut body);
272                seen_t
273                    .lock()
274                    .unwrap()
275                    .push((method.clone(), target.clone(), headers));
276                let (path, query) = target
277                    .split_once('?')
278                    .map(|(p, q)| (p.to_string(), q.to_string()))
279                    .unwrap_or((target.clone(), String::new()));
280                let mut d = data.lock().unwrap();
281                let (status, resp): (u16, Value) = match (method.as_str(), path.as_str()) {
282                    ("PUT", p) if p.starts_with("/kv/") => {
283                        let key = p.trim_start_matches("/kv/").to_string();
284                        let seq: u64 = query.trim_start_matches("seq=").parse().unwrap_or(0);
285                        if d.get(&key).is_some_and(|(cur, _)| *cur >= seq) {
286                            (409, json!({"latest": d[&key].0}))
287                        } else {
288                            let v: Value = serde_json::from_slice(&body).unwrap_or(Value::Null);
289                            d.insert(key, (seq, v));
290                            (200, json!({"ok": true}))
291                        }
292                    }
293                    ("GET", "/kv") => {
294                        let prefix = query.trim_start_matches("prefix=").to_string();
295                        let keys: Vec<Value> = d
296                            .iter()
297                            .filter(|(k, _)| k.starts_with(&prefix))
298                            .map(|(k, (s, _))| json!({"key": k, "seq": s}))
299                            .collect();
300                        (200, json!({"keys": keys}))
301                    }
302                    ("GET", p) if p.starts_with("/kv/") => {
303                        match d.get(p.trim_start_matches("/kv/")) {
304                            Some((_, v)) => (200, v.clone()),
305                            None => (404, json!({"error": "not found"})),
306                        }
307                    }
308                    ("DELETE", p) if p.starts_with("/kv/") => {
309                        d.remove(p.trim_start_matches("/kv/"));
310                        (204, Value::Null)
311                    }
312                    _ => (400, json!({"error": "bad"})),
313                };
314                let text = if resp.is_null() {
315                    String::new()
316                } else {
317                    resp.to_string()
318                };
319                let head = format!(
320                    "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
321                    text.len()
322                );
323                let _ = s.write_all(head.as_bytes());
324                let _ = s.write_all(text.as_bytes());
325            }
326        });
327        (format!("http://{addr}"), seen)
328    }
329
330    #[test]
331    fn http_store_round_trips_with_conflicts_and_headers() {
332        // SAFETY: single-threaded test; unique var name.
333        unsafe { std::env::set_var("HTTP_STORE_TEST_TOKEN", "t0k") };
334        let (base, seen) = spawn_kv();
335        let cfg = StoreHttp {
336            base_url: base.clone(),
337            headers: [(
338                "authorization".to_string(),
339                "Bearer {{secret:HTTP_STORE_TEST_TOKEN}}".to_string(),
340            )]
341            .into_iter()
342            .collect(),
343            get: Some(HttpOp {
344                method: Some("GET".into()),
345                url: "{base_url}/kv/{key}".into(),
346                body: None,
347                value: Some("body".into()),
348                keys: None,
349                conflict_status: None,
350            }),
351            put: Some(HttpOp {
352                method: Some("PUT".into()),
353                url: "{base_url}/kv/{key}?seq={seq}".into(),
354                body: Some("{envelope}".into()),
355                value: None,
356                keys: None,
357                conflict_status: Some(409),
358            }),
359            list: Some(HttpOp {
360                method: Some("GET".into()),
361                url: "{base_url}/kv?prefix={prefix}".into(),
362                body: None,
363                value: None,
364                keys: Some("body.keys".into()),
365                conflict_status: None,
366            }),
367            delete: Some(HttpOp {
368                method: Some("DELETE".into()),
369                url: "{base_url}/kv/{key}".into(),
370                body: None,
371                value: None,
372                keys: None,
373                conflict_status: None,
374            }),
375        };
376        let s = HttpStore::new(cfg, Duration::from_secs(5)).unwrap();
377        let env = json!({"v": 2, "kind": "run", "id": "1", "seq": 1, "state": {"x": 1}});
378        assert_eq!(s.put("agentd/i/run/1", 1, &env).unwrap(), PutOutcome::Ok);
379        assert_eq!(
380            s.put("agentd/i/run/1", 1, &env).unwrap(),
381            PutOutcome::Conflict {
382                latest_seq: Some(1)
383            }
384        );
385        assert_eq!(s.get("agentd/i/run/1", None).unwrap(), Some(env.clone()));
386        assert_eq!(s.get("agentd/i/run/2", None).unwrap(), None);
387        let l = s.list("agentd/i/").unwrap();
388        assert_eq!(l.len(), 1);
389        assert_eq!(l[0].seq, Some(1));
390        s.delete("agentd/i/run/1").unwrap();
391        assert_eq!(s.get("agentd/i/run/1", None).unwrap(), None);
392        // The auth header (secret resolved) and the idempotency key were sent.
393        let seen = seen.lock().unwrap();
394        let (m, t, h) = &seen[0];
395        assert_eq!(m, "PUT");
396        assert!(t.contains("seq=1"));
397        assert!(
398            h.iter()
399                .any(|(k, v)| k == "authorization" && v == "Bearer t0k"),
400            "{h:?}"
401        );
402        assert!(
403            h.iter()
404                .any(|(k, v)| k == "idempotency-key" && v == "agentd/i/run/1#1"),
405            "{h:?}"
406        );
407        unsafe { std::env::remove_var("HTTP_STORE_TEST_TOKEN") };
408    }
409}
agentd/store/http.rs

agentd/store/
http.rs
