ferra_agent/
upstream.rs

1use anyhow::{anyhow, Context, Result};
2use reqwest::Url;
3use serde::Deserialize;
4use serde_json::Value;
5
6/// HTTP client to ferra-server. No global request timeout — the watch
7/// connection is long-lived; per-request timeouts come from the
8/// `tokio::time::timeout` wrappers in the watch loop.
9#[derive(Clone)]
10pub struct UpstreamClient {
11    http: reqwest::Client,
12    base: String,
13}
14
15#[derive(Debug, Deserialize)]
16pub struct Snapshot {
17    #[allow(dead_code)]
18    pub prefix: String,
19    pub latest_event_id: i64,
20    pub items: Vec<SnapshotItem>,
21}
22
23#[derive(Debug, Deserialize)]
24pub struct SnapshotItem {
25    pub key: String,
26    pub value: Value,
27    pub event_id: i64,
28}
29
30#[derive(Debug, Deserialize)]
31pub struct GetResponse {
32    #[allow(dead_code)]
33    pub key: String,
34    pub value: Value,
35    pub event_id: i64,
36}
37
38impl UpstreamClient {
39    pub fn new(base_url: String) -> Self {
40        let http = reqwest::Client::builder()
41            .build()
42            .expect("build reqwest client");
43        Self {
44            http,
45            base: base_url.trim_end_matches('/').to_string(),
46        }
47    }
48
49    /// `GET /v1/kv?prefix=...`
50    pub async fn snapshot(&self, prefix: &str) -> Result<Snapshot> {
51        let url = Url::parse_with_params(&format!("{}/v1/kv", self.base), &[("prefix", prefix)])
52            .context("build snapshot url")?;
53        let resp = self.http.get(url).send().await.context("snapshot request")?;
54        let status = resp.status();
55        if !status.is_success() {
56            let body = resp.text().await.unwrap_or_default();
57            return Err(anyhow!("snapshot failed: HTTP {status}: {body}"));
58        }
59        Ok(resp.json().await.context("decode snapshot response")?)
60    }
61
62    /// `GET /v1/kv/{key}` — returns `None` on 404 (key was deleted between
63    /// the watch event and our follow-up GET).
64    pub async fn get_key(&self, key: &str) -> Result<Option<GetResponse>> {
65        let url = format!("{}/v1/kv/{}", self.base, encode_key_path(key));
66        let resp = self.http.get(&url).send().await.context("get_key request")?;
67        let status = resp.status();
68        if status == reqwest::StatusCode::NOT_FOUND {
69            return Ok(None);
70        }
71        if !status.is_success() {
72            let body = resp.text().await.unwrap_or_default();
73            return Err(anyhow!("get_key failed: HTTP {status}: {body}"));
74        }
75        Ok(Some(resp.json().await.context("decode get response")?))
76    }
77
78    /// `GET /v1/watch?prefix=...&since=...` — returns the open response;
79    /// caller streams the body.
80    pub async fn open_watch(&self, prefix: &str, since: i64) -> Result<reqwest::Response> {
81        let url = Url::parse_with_params(
82            &format!("{}/v1/watch", self.base),
83            &[("prefix", prefix), ("since", &since.to_string())],
84        )
85        .context("build watch url")?;
86        let resp = self
87            .http
88            .get(url)
89            .header("Accept", "text/event-stream")
90            .send()
91            .await
92            .context("watch request")?;
93        let status = resp.status();
94        if !status.is_success() {
95            let body = resp.text().await.unwrap_or_default();
96            return Err(anyhow!("watch open failed: HTTP {status}: {body}"));
97        }
98        Ok(resp)
99    }
100}
101
102/// Percent-encode a key for use as URL path segments. Preserves `/` so a key
103/// like `services/payment/timeout_ms` becomes part of the path, not a single
104/// segment.
105fn encode_key_path(key: &str) -> String {
106    use std::fmt::Write;
107    let mut s = String::with_capacity(key.len());
108    for b in key.bytes() {
109        let safe = matches!(b,
110            b'0'..=b'9' | b'A'..=b'Z' | b'a'..=b'z'
111            | b'-' | b'.' | b'_' | b'~' | b'/'
112        );
113        if safe {
114            s.push(b as char);
115        } else {
116            let _ = write!(s, "%{:02X}", b);
117        }
118    }
119    s
120}
121
122#[cfg(test)]
123mod tests {
124    use super::encode_key_path;
125
126    #[test]
127    fn preserves_slashes() {
128        assert_eq!(encode_key_path("a/b/c"), "a/b/c");
129    }
130
131    #[test]
132    fn percent_encodes_unsafe() {
133        assert_eq!(encode_key_path("a b"), "a%20b");
134        assert_eq!(encode_key_path("?#&"), "%3F%23%26");
135    }
136}
ferra_agent/upstream.rs

ferra_agent/
upstream.rs
