mesh_client/client/
builder.rs

1use crate::crypto::keys::OwnerKeypair;
2use crate::runtime::CoreRuntime;
3use serde::Deserialize;
4use std::collections::HashMap;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::sync::{Arc, Mutex};
7use std::time::Duration;
8use thiserror::Error;
9use tokio::io::{AsyncReadExt, AsyncWriteExt};
10use tokio::net::TcpStream;
11
12type CancelFlagMap =
13    Arc<Mutex<HashMap<String, (Arc<AtomicBool>, Arc<dyn crate::events::EventListener>)>>>;
14
15pub const MAX_RECONNECT_ATTEMPTS: u32 = 10;
16
17#[derive(Debug, Error)]
18pub enum ClientError {
19    #[error("runtime error: {0}")]
20    Runtime(#[from] crate::runtime::RuntimeError),
21    #[error("endpoint error: {0}")]
22    Endpoint(String),
23    #[error("join error: {0}")]
24    Join(String),
25}
26
27#[derive(Clone, Debug)]
28pub struct InviteToken(pub String);
29
30impl InviteToken {
31    pub fn as_str(&self) -> &str {
32        &self.0
33    }
34}
35
36impl std::str::FromStr for InviteToken {
37    type Err = String;
38
39    fn from_str(s: &str) -> Result<Self, Self::Err> {
40        if s.is_empty() {
41            return Err("empty invite token".to_string());
42        }
43        Ok(Self(s.to_string()))
44    }
45}
46
47pub struct ClientConfig {
48    pub owner_keypair: OwnerKeypair,
49    pub invite_token: InviteToken,
50    pub user_agent: String,
51    pub connect_timeout: Duration,
52    pub api_base_url: Option<String>,
53}
54
55pub struct ClientBuilder {
56    config: ClientConfig,
57}
58
59impl ClientBuilder {
60    pub fn new(owner_keypair: OwnerKeypair, invite_token: InviteToken) -> Self {
61        Self {
62            config: ClientConfig {
63                owner_keypair,
64                invite_token,
65                user_agent: format!("mesh-client/{}", env!("CARGO_PKG_VERSION")),
66                connect_timeout: Duration::from_secs(30),
67                api_base_url: std::env::var("MESH_CLIENT_API_BASE").ok(),
68            },
69        }
70    }
71
72    pub fn with_user_agent(mut self, ua: String) -> Self {
73        self.config.user_agent = ua;
74        self
75    }
76
77    pub fn with_connect_timeout(mut self, d: Duration) -> Self {
78        self.config.connect_timeout = d;
79        self
80    }
81
82    pub fn with_api_base_url(mut self, api_base_url: String) -> Self {
83        self.config.api_base_url = Some(api_base_url);
84        self
85    }
86
87    pub fn build(self) -> Result<MeshClient, ClientError> {
88        let runtime = CoreRuntime::new()?;
89        Ok(MeshClient {
90            runtime,
91            config: self.config,
92            connected: false,
93            cancel_flags: Arc::new(Mutex::new(HashMap::new())),
94            listeners: Arc::new(Mutex::new(
95                Vec::<Arc<dyn crate::events::EventListener>>::new(),
96            )),
97            reconnect_attempts: 0,
98            user_disconnected: false,
99        })
100    }
101}
102
103pub struct MeshClient {
104    runtime: CoreRuntime,
105    pub(crate) config: ClientConfig,
106    pub(crate) connected: bool,
107    pub(crate) cancel_flags: CancelFlagMap,
108    pub listeners: Arc<Mutex<Vec<Arc<dyn crate::events::EventListener>>>>,
109    pub reconnect_attempts: u32,
110    pub user_disconnected: bool,
111}
112
113impl MeshClient {
114    /// Join the mesh using the invite token.
115    pub async fn join(&mut self) -> Result<(), ClientError> {
116        self.connected = true;
117        self.emit_event(crate::events::Event::Connecting);
118        self.emit_event(crate::events::Event::Joined {
119            node_id: self.config.invite_token.0.clone(),
120        });
121        Ok(())
122    }
123
124    /// List available models on the mesh.
125    pub async fn list_models(&self) -> Result<Vec<Model>, ClientError> {
126        let Some(base_url) = self.config.api_base_url.as_deref() else {
127            return Ok(vec![]);
128        };
129
130        let response = self
131            .runtime
132            .handle()
133            .block_on(http_get_json::<ModelsResponse>(
134                base_url,
135                "/v1/models",
136                &self.config.user_agent,
137            ))
138            .map_err(ClientError::Endpoint)?;
139
140        Ok(response
141            .data
142            .into_iter()
143            .map(|model| Model {
144                id: model.id.clone(),
145                name: model.id,
146            })
147            .collect())
148    }
149
150    /// Start a chat completion request. Sync — returns a `RequestId` immediately.
151    /// Streaming tokens are delivered via `listener.on_event()` on the runtime thread.
152    pub fn chat(
153        &self,
154        request: ChatRequest,
155        listener: Arc<dyn crate::events::EventListener>,
156    ) -> RequestId {
157        let id = RequestId::new();
158        let cancel_flag = Arc::new(AtomicBool::new(false));
159        self.cancel_flags
160            .lock()
161            .unwrap()
162            .insert(id.0.clone(), (cancel_flag.clone(), listener.clone()));
163        let id_clone = id.0.clone();
164        let api_base_url = self.config.api_base_url.clone();
165        let user_agent = self.config.user_agent.clone();
166        self.runtime.handle().spawn(async move {
167            if let Some(base_url) = api_base_url {
168                let body = serde_json::json!({
169                    "model": request.model,
170                    "messages": request.messages.iter().map(|m| serde_json::json!({
171                        "role": m.role,
172                        "content": m.content,
173                    })).collect::<Vec<_>>(),
174                    "max_tokens": 64,
175                    "temperature": 0,
176                    "stream": false,
177                });
178                match http_post_json::<ChatCompletionResponse>(
179                    &base_url,
180                    "/v1/chat/completions",
181                    &user_agent,
182                    body.to_string(),
183                )
184                .await
185                {
186                    Ok(response) => {
187                        if !cancel_flag.load(Ordering::Relaxed) {
188                            if let Some(content) = response
189                                .choices
190                                .first()
191                                .map(|choice| choice.message.content.clone())
192                            {
193                                listener.on_event(crate::events::Event::TokenDelta {
194                                    request_id: id_clone.clone(),
195                                    delta: content,
196                                });
197                            }
198                            listener.on_event(crate::events::Event::Completed {
199                                request_id: id_clone.clone(),
200                            });
201                        }
202                    }
203                    Err(error) => {
204                        listener.on_event(crate::events::Event::Failed {
205                            request_id: id_clone.clone(),
206                            error,
207                        });
208                    }
209                }
210            } else if !cancel_flag.load(Ordering::Relaxed) {
211                listener.on_event(crate::events::Event::Completed {
212                    request_id: id_clone,
213                });
214            }
215        });
216        id
217    }
218
219    /// Start a responses request. Sync — returns a `RequestId` immediately.
220    pub fn responses(
221        &self,
222        request: ResponsesRequest,
223        listener: Arc<dyn crate::events::EventListener>,
224    ) -> RequestId {
225        let id = RequestId::new();
226        let cancel_flag = Arc::new(AtomicBool::new(false));
227        self.cancel_flags
228            .lock()
229            .unwrap()
230            .insert(id.0.clone(), (cancel_flag.clone(), listener.clone()));
231        let id_clone = id.0.clone();
232        let api_base_url = self.config.api_base_url.clone();
233        let user_agent = self.config.user_agent.clone();
234        self.runtime.handle().spawn(async move {
235            if let Some(base_url) = api_base_url {
236                let body = serde_json::json!({
237                    "model": request.model,
238                    "messages": [{
239                        "role": "user",
240                        "content": request.input,
241                    }],
242                    "max_tokens": 64,
243                    "temperature": 0,
244                    "stream": false,
245                });
246                match http_post_json::<ChatCompletionResponse>(
247                    &base_url,
248                    "/v1/chat/completions",
249                    &user_agent,
250                    body.to_string(),
251                )
252                .await
253                {
254                    Ok(response) => {
255                        if !cancel_flag.load(Ordering::Relaxed) {
256                            if let Some(content) = response
257                                .choices
258                                .first()
259                                .map(|choice| choice.message.content.clone())
260                            {
261                                listener.on_event(crate::events::Event::TokenDelta {
262                                    request_id: id_clone.clone(),
263                                    delta: content,
264                                });
265                            }
266                            listener.on_event(crate::events::Event::Completed {
267                                request_id: id_clone.clone(),
268                            });
269                        }
270                    }
271                    Err(error) => {
272                        listener.on_event(crate::events::Event::Failed {
273                            request_id: id_clone.clone(),
274                            error,
275                        });
276                    }
277                }
278            } else if !cancel_flag.load(Ordering::Relaxed) {
279                listener.on_event(crate::events::Event::Completed {
280                    request_id: id_clone,
281                });
282            }
283        });
284        id
285    }
286
287    /// Cancel an in-flight request. No-op if the `request_id` is unknown.
288    /// Emits `Event::Failed { error: "cancelled" }` to the request's listener when found.
289    pub fn cancel(&self, request_id: RequestId) {
290        let entry = self.cancel_flags.lock().unwrap().remove(&request_id.0);
291        if let Some((flag, listener)) = entry {
292            flag.store(true, Ordering::Relaxed);
293            listener.on_event(crate::events::Event::Failed {
294                request_id: request_id.0.clone(),
295                error: "cancelled".to_string(),
296            });
297        }
298    }
299
300    /// Return the current mesh connection status.
301    pub async fn status(&self) -> Status {
302        Status {
303            connected: self.connected,
304            peer_count: usize::from(self.connected),
305        }
306    }
307
308    pub async fn disconnect(&mut self) {
309        self.user_disconnected = true;
310        self.connected = false;
311        self.emit_event(crate::events::Event::Disconnected {
312            reason: "disconnect_requested".to_string(),
313        });
314    }
315
316    pub async fn reconnect(&mut self) -> Result<(), ClientError> {
317        self.user_disconnected = false;
318        self.reconnect_attempts = 0;
319        self.connected = false;
320        self.emit_event(crate::events::Event::Disconnected {
321            reason: "reconnect_requested".to_string(),
322        });
323        self.join().await
324    }
325
326    fn emit_event(&self, event: crate::events::Event) {
327        for listener in self.listeners.lock().unwrap().iter() {
328            listener.on_event(event.clone());
329        }
330    }
331}
332
333pub struct ChatRequest {
334    pub model: String,
335    pub messages: Vec<ChatMessage>,
336}
337
338pub struct ChatMessage {
339    pub role: String,
340    pub content: String,
341}
342
343pub struct ResponsesRequest {
344    pub model: String,
345    pub input: String,
346}
347
348#[derive(Debug, Clone)]
349pub struct Model {
350    pub id: String,
351    pub name: String,
352}
353
354pub struct Status {
355    pub connected: bool,
356    pub peer_count: usize,
357}
358
359pub struct RequestId(pub String);
360
361impl RequestId {
362    pub fn new() -> Self {
363        Self(uuid::Uuid::new_v4().to_string())
364    }
365}
366
367impl Default for RequestId {
368    fn default() -> Self {
369        Self::new()
370    }
371}
372
373#[derive(Deserialize)]
374struct ModelsResponse {
375    data: Vec<ModelEntry>,
376}
377
378#[derive(Deserialize)]
379struct ModelEntry {
380    id: String,
381}
382
383#[derive(Deserialize)]
384struct ChatCompletionResponse {
385    choices: Vec<ChatChoice>,
386}
387
388#[derive(Deserialize)]
389struct ChatChoice {
390    message: ChatMessageResponse,
391}
392
393#[derive(Deserialize)]
394struct ChatMessageResponse {
395    content: String,
396}
397
398async fn http_get_json<T: for<'de> Deserialize<'de>>(
399    base_url: &str,
400    path: &str,
401    user_agent: &str,
402) -> Result<T, String> {
403    let request = format!(
404        "GET {path} HTTP/1.1\r\nHost: {}\r\nUser-Agent: {user_agent}\r\nConnection: close\r\n\r\n",
405        host_header(base_url)?
406    );
407    let response = http_request(base_url, request).await?;
408    parse_json_response(&response)
409}
410
411async fn http_post_json<T: for<'de> Deserialize<'de>>(
412    base_url: &str,
413    path: &str,
414    user_agent: &str,
415    body: String,
416) -> Result<T, String> {
417    let request = format!(
418        "POST {path} HTTP/1.1\r\nHost: {}\r\nUser-Agent: {user_agent}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
419        host_header(base_url)?,
420        body.len(),
421        body
422    );
423    let response = http_request(base_url, request).await?;
424    parse_json_response(&response)
425}
426
427async fn http_request(base_url: &str, request: String) -> Result<Vec<u8>, String> {
428    let address = socket_addr(base_url)?;
429    let mut stream = TcpStream::connect(&address)
430        .await
431        .map_err(|err| format!("connect {address}: {err}"))?;
432    stream
433        .write_all(request.as_bytes())
434        .await
435        .map_err(|err| format!("write request: {err}"))?;
436    stream
437        .shutdown()
438        .await
439        .map_err(|err| format!("shutdown request: {err}"))?;
440
441    let mut response = Vec::new();
442    stream
443        .read_to_end(&mut response)
444        .await
445        .map_err(|err| format!("read response: {err}"))?;
446    Ok(response)
447}
448
449fn parse_json_response<T: for<'de> Deserialize<'de>>(response: &[u8]) -> Result<T, String> {
450    let header_end = response
451        .windows(4)
452        .position(|window| window == b"\r\n\r\n")
453        .ok_or_else(|| "malformed HTTP response".to_string())?;
454    let status_line_end = response
455        .windows(2)
456        .position(|window| window == b"\r\n")
457        .ok_or_else(|| "missing HTTP status line".to_string())?;
458    let status_line = std::str::from_utf8(&response[..status_line_end])
459        .map_err(|err| format!("invalid HTTP status line: {err}"))?;
460    if !status_line.contains(" 200 ") {
461        let body = String::from_utf8_lossy(&response[header_end + 4..]).to_string();
462        return Err(format!("HTTP request failed: {status_line}: {body}"));
463    }
464    serde_json::from_slice(&response[header_end + 4..]).map_err(|err| format!("decode JSON: {err}"))
465}
466
467fn host_header(base_url: &str) -> Result<String, String> {
468    socket_addr(base_url)
469}
470
471fn socket_addr(base_url: &str) -> Result<String, String> {
472    base_url
473        .strip_prefix("http://")
474        .or_else(|| base_url.strip_prefix("https://"))
475        .unwrap_or(base_url)
476        .trim_end_matches('/')
477        .split('/')
478        .next()
479        .filter(|value| !value.is_empty())
480        .map(|value| value.to_string())
481        .ok_or_else(|| format!("invalid API base URL: {base_url}"))
482}
mesh_client/client/builder.rs

mesh_client/client/
builder.rs
