qualia_client_core/wellfair/
sync_relay_server.rs

1//! A minimal HTTP relay for the sync transport (T3.1) — the server counterpart to
2//! [`HttpRelayTransport`](super::sync_transport::HttpRelayTransport).
3//!
4//! A **dumb op bus**: it stores published operations (append-only, dedup by operation id) and serves
5//! pulls from a cursor. It does **no validation** — trust is the receiving node's inbox
6//! ([`validate_operation`](super::sync_protocol::validate_operation)), so a compromised relay can
7//! only ever cause rejections, never admission of bad data. Peers rendezvous through it.
8//!
9//! Endpoints:
10//! - `POST /sync/publish` — body `{"ops":[...]}`; stores new ops, returns `{"ops":[]}`.
11//! - `GET  /sync/pull?since={n}` — returns `{"ops":[...]}` for ops at index `>= n` (relay order).
12//!
13//! Native-only (`tiny_http`). Runs a background accept loop with graceful shutdown on drop.
14#![cfg(not(target_arch = "wasm32"))]
15
16use std::net::SocketAddr;
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::{Arc, Mutex};
19use std::thread::JoinHandle;
20use std::time::Duration;
21
22use tiny_http::{Header, Method, Request, Response, Server};
23
24use super::sync_protocol::SyncOperation;
25use super::sync_transport::SyncOpsBody;
26
27type OpStore = Arc<Mutex<Vec<SyncOperation>>>;
28
29/// A running relay. Stop it explicitly with [`SyncRelayServer::stop`] or let `Drop` do it.
30pub struct SyncRelayServer {
31    addr: SocketAddr,
32    shutdown: Arc<AtomicBool>,
33    handle: Option<JoinHandle<()>>,
34    store: OpStore,
35}
36
37impl SyncRelayServer {
38    /// Bind and start the relay. Use `"127.0.0.1:0"` for an OS-assigned port; read the bound
39    /// address back via [`SyncRelayServer::addr`] / [`SyncRelayServer::base_url`].
40    pub fn start(bind: &str) -> Result<Self, String> {
41        let server = Server::http(bind).map_err(|e| e.to_string())?;
42        let addr = server
43            .server_addr()
44            .to_ip()
45            .ok_or_else(|| "relay bound to a non-IP address".to_string())?;
46        let shutdown = Arc::new(AtomicBool::new(false));
47        let store: OpStore = Arc::new(Mutex::new(Vec::new()));
48        let server = Arc::new(server);
49
50        let handle = {
51            let shutdown = shutdown.clone();
52            let store = store.clone();
53            let server = server.clone();
54            std::thread::spawn(move || loop {
55                if shutdown.load(Ordering::Relaxed) {
56                    break;
57                }
58                match server.recv_timeout(Duration::from_millis(100)) {
59                    Ok(Some(req)) => handle_request(req, &store),
60                    Ok(None) => continue, // timeout — re-check the shutdown flag
61                    Err(_) => break,
62                }
63            })
64        };
65
66        Ok(Self {
67            addr,
68            shutdown,
69            handle: Some(handle),
70            store,
71        })
72    }
73
74    pub fn addr(&self) -> SocketAddr {
75        self.addr
76    }
77
78    pub fn base_url(&self) -> String {
79        format!("http://{}", self.addr)
80    }
81
82    /// Number of operations the relay currently holds.
83    pub fn op_count(&self) -> usize {
84        self.store.lock().map(|v| v.len()).unwrap_or(0)
85    }
86
87    /// Signal shutdown and join the accept thread.
88    pub fn stop(&mut self) {
89        self.shutdown.store(true, Ordering::Relaxed);
90        if let Some(h) = self.handle.take() {
91            let _ = h.join();
92        }
93    }
94}
95
96impl Drop for SyncRelayServer {
97    fn drop(&mut self) {
98        self.stop();
99    }
100}
101
102fn json_response(code: u16, body: &SyncOpsBody) -> Response<std::io::Cursor<Vec<u8>>> {
103    let json = serde_json::to_string(body).unwrap_or_else(|_| "{\"ops\":[]}".to_string());
104    let header = Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..])
105        .expect("static header is valid");
106    Response::from_string(json)
107        .with_status_code(code)
108        .with_header(header)
109}
110
111fn text_response(code: u16, msg: &str) -> Response<std::io::Cursor<Vec<u8>>> {
112    Response::from_string(msg).with_status_code(code)
113}
114
115fn parse_since(query: &str) -> u64 {
116    query
117        .split('&')
118        .find_map(|kv| kv.strip_prefix("since="))
119        .and_then(|v| v.parse().ok())
120        .unwrap_or(0)
121}
122
123fn handle_request(mut req: Request, store: &OpStore) {
124    let method = req.method().clone();
125    let url = req.url().to_string();
126    let (path, query) = url.split_once('?').unwrap_or((url.as_str(), ""));
127
128    // Read the body up front (for POST) so the mutable borrow is released before responding.
129    let mut body = String::new();
130    if method == Method::Post {
131        let _ = req.as_reader().read_to_string(&mut body);
132    }
133
134    let response = match (&method, path) {
135        (Method::Post, "/sync/publish") => match serde_json::from_str::<SyncOpsBody>(&body) {
136            Ok(parsed) => {
137                if let Ok(mut s) = store.lock() {
138                    for op in parsed.ops {
139                        if !s.iter().any(|e| e.operation_id == op.operation_id) {
140                            s.push(op);
141                        }
142                    }
143                }
144                json_response(200, &SyncOpsBody { ops: Vec::new() })
145            }
146            Err(e) => text_response(400, &format!("bad json: {e}")),
147        },
148        (Method::Get, "/sync/pull") => {
149            let since = parse_since(query);
150            let ops = store
151                .lock()
152                .map(|s| {
153                    let start = (since as usize).min(s.len());
154                    s[start..].to_vec()
155                })
156                .unwrap_or_default();
157            json_response(200, &SyncOpsBody { ops })
158        }
159        _ => text_response(404, "not found"),
160    };
161
162    let _ = req.respond(response);
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168    use crate::wellfair::sync_protocol::SyncOperation;
169    use crate::wellfair::sync_transport::{HttpRelayTransport, SyncTransport};
170
171    fn signed(id: &str, summary: &str, lamport: u64) -> SyncOperation {
172        SyncOperation::new(
173            id,
174            format!("urn:wellfair:ledger_entry:{id}"),
175            "ledger_entry",
176            "did:wf:remote",
177            "Restricted",
178            summary,
179            lamport,
180            1_700_000_000,
181        )
182        .with_signature("deadbeef")
183    }
184
185    #[test]
186    fn http_transport_round_trips_through_the_relay() {
187        let server = SyncRelayServer::start("127.0.0.1:0").expect("relay starts");
188        let base = server.base_url();
189
190        let node_a = HttpRelayTransport::new(&base);
191        let node_b = HttpRelayTransport::new(&base);
192
193        node_a
194            .publish(&[signed("x", "1", 1), signed("y", "2", 2)])
195            .expect("publish");
196        // A second node pulls what A published, over real HTTP.
197        let pulled = node_b.pull(0).expect("pull");
198        assert_eq!(pulled.len(), 2);
199        assert_eq!(pulled[0].operation_id, "x");
200        assert_eq!(server.op_count(), 2);
201
202        // Cursor works: nothing new after index 2.
203        assert!(node_b.pull(2).expect("pull cursor").is_empty());
204
205        // Dedup at the relay: re-publishing 'x' does not grow the store.
206        node_a.publish(&[signed("x", "1", 1)]).expect("republish");
207        assert_eq!(server.op_count(), 2);
208    }
209}
qualia_client_core/wellfair/sync_relay_server.rs

qualia_client_core/wellfair/
sync_relay_server.rs
