silicon_dm_client/
lib.rs

1//! Stateless Silicon DM client. Callers own credentials, retry policy and durable cursors.
2//! No IAM application secrets, local files or backend-internal operations are required.
3pub mod models;
4pub mod relay;
5#[cfg(feature = "runtime")]
6pub mod runtime;
7pub use models::*;
8pub use silicon_dm_protocol::Envelope;
9
10use reqwest::{Client as HttpClient, Method, RequestBuilder};
11use serde::{Serialize, de::DeserializeOwned};
12use serde_json::{Value, json};
13use std::time::Duration;
14use tokio::net::TcpStream;
15use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite::client::IntoClientRequest};
16use url::Url;
17use uuid::Uuid;
18
19pub type Result<T> = std::result::Result<T, Error>;
20pub type Socket = WebSocketStream<MaybeTlsStream<TcpStream>>;
21
22/// HTTP failures preserve the response body, including current-draft conflict data.
23#[derive(Debug, thiserror::Error)]
24pub enum Error {
25    #[error("invalid client configuration: {0}")]
26    Configuration(String),
27    #[error("DM transport failed: {0}")]
28    Transport(#[from] reqwest::Error),
29    #[error("DM returned HTTP {status}: {code}: {message}")]
30    Api {
31        status: u16,
32        code: String,
33        message: String,
34        body: Box<Value>,
35        request_id: Option<String>,
36        retry_after: Option<String>,
37    },
38    #[error("invalid DM response: {0}")]
39    Decode(#[from] serde_json::Error),
40    #[error("WebSocket connection failed: {0}")]
41    WebSocket(#[from] tokio_tungstenite::tungstenite::Error),
42}
43impl Error {
44    /// Retrying a mutation requires reusing its original idempotency key.
45    pub fn retryable(&self) -> bool {
46        matches!(self, Self::Transport(_) | Self::WebSocket(_))
47            || matches!(
48                self,
49                Self::Api {
50                    status: 408 | 429 | 500..=599,
51                    ..
52                }
53            )
54    }
55    pub fn unauthorized(&self) -> bool {
56        matches!(self, Self::Api { status: 401, .. })
57    }
58}
59
60/// Immutable connection configuration. Debug intentionally omits bearer/test secrets.
61#[derive(Clone)]
62pub struct Client {
63    http: HttpClient,
64    base: Url,
65    token: Option<String>,
66    organization: Option<String>,
67    test_key: Option<String>,
68    websocket_limit: usize,
69    testing_generation: Option<i64>,
70}
71impl Client {
72    /// Accepts an origin or an `/api/v1` base. HTTP is limited to loopback hosts.
73    pub fn new(base: impl AsRef<str>) -> Result<Self> {
74        let mut base =
75            Url::parse(base.as_ref()).map_err(|e| Error::Configuration(e.to_string()))?;
76        validate_endpoint(&base)?;
77        if base.query().is_some() || base.fragment().is_some() {
78            return Err(Error::Configuration(
79                "base URL must not have a query or fragment".into(),
80            ));
81        }
82        let path = base.path().trim_end_matches('/');
83        let path = if path.is_empty() {
84            "/api/v1/".to_owned()
85        } else {
86            format!("{path}/")
87        };
88        base.set_path(&path);
89        let http = HttpClient::builder()
90            .redirect(reqwest::redirect::Policy::none())
91            .timeout(Duration::from_secs(45))
92            .user_agent(concat!("silicon-dm-client/", env!("CARGO_PKG_VERSION")))
93            .build()?;
94        Ok(Self {
95            http,
96            base,
97            token: None,
98            organization: None,
99            test_key: None,
100            websocket_limit: 128 * 1024 * 1024,
101            testing_generation: None,
102        })
103    }
104    pub fn with_auth(mut self, token: impl Into<String>, organization: impl Into<String>) -> Self {
105        self.token = Some(token.into());
106        self.organization = Some(organization.into());
107        self
108    }
109    /// Mandatory IAM sandbox selection remains the backend's responsibility.
110    pub fn with_test_key(mut self, key: impl Into<String>) -> Result<Self> {
111        let key = key.into();
112        if key.len() != 32 || !key.bytes().all(|b| b.is_ascii_alphanumeric()) {
113            return Err(Error::Configuration(
114                "testing environment key must be 32 alphanumeric characters".into(),
115            ));
116        }
117        self.test_key = Some(key);
118        Ok(self)
119    }
120    pub fn without_test(mut self) -> Self {
121        self.test_key = None;
122        self.testing_generation = None;
123        self
124    }
125    /// Binds requests to a previously observed sandbox generation. A cleaned or
126    /// rotated environment rejects stale requests rather than applying them anew.
127    pub fn with_testing_generation(mut self, generation: i64) -> Result<Self> {
128        if generation < 1 {
129            return Err(Error::Configuration(
130                "testing generation must be positive".into(),
131            ));
132        }
133        self.testing_generation = Some(generation);
134        Ok(self)
135    }
136    /// Sets a bounded encoded WebSocket message/frame limit, up to 3 GiB.
137    /// The default 128 MiB accommodates DM's default API body ceiling.
138    pub fn with_websocket_limit(mut self, max_bytes: usize) -> Result<Self> {
139        if max_bytes == 0 || max_bytes as u64 > 3 * 1024 * 1024 * 1024 {
140            return Err(Error::Configuration(
141                "WebSocket limit must be between 1 byte and 3 GiB".into(),
142            ));
143        }
144        self.websocket_limit = max_bytes;
145        Ok(self)
146    }
147    fn endpoint(&self, path: &str) -> Result<Url> {
148        self.base
149            .join(path)
150            .map_err(|e| Error::Configuration(e.to_string()))
151    }
152    fn request(&self, method: Method, path: &str) -> Result<RequestBuilder> {
153        let mut request = self.http.request(method, self.endpoint(path)?);
154        if let Some(token) = &self.token {
155            request = request.bearer_auth(token);
156        }
157        if let Some(org) = &self.organization {
158            request = request.header("X-Org-ID", org);
159        }
160        if let Some(key) = &self.test_key {
161            request = request.header("X-Testing-Environment-Key", key);
162        }
163        if let Some(generation) = self.testing_generation {
164            request = request.header("X-Testing-Environment-Generation", generation);
165        }
166        Ok(request)
167    }
168    async fn send(&self, builder: RequestBuilder) -> Result<reqwest::Response> {
169        let mut request = builder.build()?;
170        if let Some(body) = request.body().and_then(reqwest::Body::as_bytes) {
171            let kind =
172                silicon_dm_protocol::http_type(request.method().as_str(), request.url().path());
173            // Preserve already-serialized JSON bytes, avoiding a second parsed message allocation.
174            let mut encoded = format!("{{\"type\":\"{kind}\",\"data\":").into_bytes();
175            encoded.extend_from_slice(body);
176            encoded.push(b'}');
177            *request.body_mut() = Some(encoded.into());
178        }
179        Ok(self.http.execute(request).await?)
180    }
181    async fn json<T: DeserializeOwned>(&self, request: RequestBuilder) -> Result<T> {
182        let response = checked(self.send(request).await?).await?;
183        Ok(serde_json::from_slice::<Envelope<T>>(&response.bytes().await?)?.data)
184    }
185    async fn empty(&self, request: RequestBuilder) -> Result<()> {
186        checked(self.send(request).await?).await?;
187        Ok(())
188    }
189    pub async fn login(&self, slt: &str, key: &str) -> Result<Tokens> {
190        self.json(
191            self.request(Method::POST, "auth/login")?
192                .header("Idempotency-Key", key)
193                .json(&json!({"slt": slt})),
194        )
195        .await
196    }
197    pub async fn refresh(&self, refresh_token: &str, key: &str) -> Result<Tokens> {
198        self.json(
199            self.request(Method::POST, "auth/refresh")?
200                .header("Idempotency-Key", key)
201                .json(&json!({"refresh_token": refresh_token})),
202        )
203        .await
204    }
205    pub async fn logout(&self, token: &str, key: &str) -> Result<()> {
206        self.empty(
207            self.request(Method::POST, "auth/logout")?
208                .header("Idempotency-Key", key)
209                .json(&json!({"token":token})),
210        )
211        .await
212    }
213    /// Public IAM application information; does not require login.
214    pub async fn iam(&self) -> Result<IamInfo> {
215        self.json(self.request(Method::GET, "iam")?).await
216    }
217    pub async fn me(&self) -> Result<Identity> {
218        self.json(self.request(Method::GET, "auth/me")?).await
219    }
220    pub async fn conversations(&self, page: &PageRequest) -> Result<Page<Conversation>> {
221        self.json(self.request(Method::GET, "conversations")?.query(page))
222            .await
223    }
224    pub async fn create_conversation(
225        &self,
226        participants: &[String],
227        key: &str,
228    ) -> Result<Conversation> {
229        self.json(
230            self.request(Method::POST, "conversations")?
231                .header("Idempotency-Key", key)
232                .json(&json!({"participant_ids":participants})),
233        )
234        .await
235    }
236    pub async fn messages(
237        &self,
238        conversation: Uuid,
239        page: &PageRequest,
240        include_bundled: bool,
241    ) -> Result<Page<Message>> {
242        self.json(
243            self.request(
244                Method::GET,
245                &format!("conversations/{conversation}/messages"),
246            )?
247            .query(page)
248            .query(&[("include_bundled_members", include_bundled)]),
249        )
250        .await
251    }
252    pub async fn send_message(
253        &self,
254        conversation: Uuid,
255        message: &MessageCreate,
256        key: &str,
257    ) -> Result<Message> {
258        self.json(
259            self.request(
260                Method::POST,
261                &format!("conversations/{conversation}/messages"),
262            )?
263            .header("Idempotency-Key", key)
264            .json(message),
265        )
266        .await
267    }
268    pub async fn message(&self, conversation: Uuid, message: Uuid) -> Result<Message> {
269        self.json(self.request(
270            Method::GET,
271            &format!("conversations/{conversation}/messages/{message}"),
272        )?)
273        .await
274    }
275    pub async fn edit_message(
276        &self,
277        conversation: Uuid,
278        message: Uuid,
279        content: &MessageCreate,
280        version: i64,
281        key: &str,
282    ) -> Result<Message> {
283        self.json(
284            self.request(
285                Method::PATCH,
286                &format!("conversations/{conversation}/messages/{message}"),
287            )?
288            .header("If-Match", version)
289            .header("Idempotency-Key", key)
290            .json(content),
291        )
292        .await
293    }
294    pub async fn delete_message(
295        &self,
296        conversation: Uuid,
297        message: Uuid,
298        version: i64,
299        key: &str,
300    ) -> Result<Message> {
301        self.json(
302            self.request(
303                Method::DELETE,
304                &format!("conversations/{conversation}/messages/{message}"),
305            )?
306            .header("If-Match", version)
307            .header("Idempotency-Key", key),
308        )
309        .await
310    }
311    pub async fn record_receipt(
312        &self,
313        conversation: Uuid,
314        message: Uuid,
315        status: ReceiptStatus,
316        device: &str,
317    ) -> Result<Message> {
318        self.json(
319            self.request(
320                Method::POST,
321                &format!("conversations/{conversation}/messages/{message}/receipts"),
322            )?
323            .json(&json!({"status":status,"device_id":device})),
324        )
325        .await
326    }
327    pub async fn draft(&self, conversation: Uuid) -> Result<Draft> {
328        self.json(self.request(Method::GET, &format!("conversations/{conversation}/draft"))?)
329            .await
330    }
331    pub async fn put_draft(
332        &self,
333        conversation: Uuid,
334        draft: &DraftInput,
335        version: i64,
336    ) -> Result<Draft> {
337        self.json(
338            self.request(Method::PUT, &format!("conversations/{conversation}/draft"))?
339                .header("If-Match", version)
340                .json(draft),
341        )
342        .await
343    }
344    pub async fn delete_draft(&self, conversation: Uuid) -> Result<()> {
345        self.empty(self.request(
346            Method::DELETE,
347            &format!("conversations/{conversation}/draft"),
348        )?)
349        .await
350    }
351    pub async fn create_bundle(
352        &self,
353        conversation: Uuid,
354        bundle: &BundleCreate,
355        key: &str,
356    ) -> Result<Bundle> {
357        self.json(
358            self.request(
359                Method::POST,
360                &format!("conversations/{conversation}/bundles"),
361            )?
362            .header("Idempotency-Key", key)
363            .json(bundle),
364        )
365        .await
366    }
367    pub async fn bundle(&self, conversation: Uuid, bundle: Uuid) -> Result<Bundle> {
368        self.json(self.request(
369            Method::GET,
370            &format!("conversations/{conversation}/bundles/{bundle}"),
371        )?)
372        .await
373    }
374    pub async fn presence(&self, actor: &str) -> Result<Presence> {
375        let encoded: String = url::form_urlencoded::byte_serialize(actor.as_bytes()).collect();
376        self.json(self.request(Method::GET, &format!("presence/{encoded}"))?)
377            .await
378    }
379    pub async fn gifs(&self, kind: GifList<'_>) -> Result<Page<Gif>> {
380        let req = match kind {
381            GifList::Trending => self.request(Method::GET, "gifs/trending")?,
382            GifList::Recent => self.request(Method::GET, "gifs/recent")?,
383            GifList::Search(q) => self.request(Method::GET, "gifs/search")?.query(&[("q", q)]),
384        };
385        self.json(req).await
386    }
387    pub async fn create_test_environment(
388        &self,
389        input: &TestEnvironmentCreate,
390        key: &str,
391    ) -> Result<TestEnvironment> {
392        self.json(
393            self.request(Method::POST, "testing-environments")?
394                .header("Idempotency-Key", key)
395                .json(input),
396        )
397        .await
398    }
399    pub async fn test_environments(&self, include_deleted: bool) -> Result<Page<TestEnvironment>> {
400        self.json(
401            self.request(Method::GET, "testing-environments")?
402                .query(&[("include_deleted", include_deleted)]),
403        )
404        .await
405    }
406    pub async fn test_environment(&self, id: Uuid) -> Result<TestEnvironment> {
407        self.json(self.request(Method::GET, &format!("testing-environments/{id}"))?)
408            .await
409    }
410    pub async fn update_test_environment(
411        &self,
412        id: Uuid,
413        input: &TestEnvironmentUpdate,
414        key: &str,
415    ) -> Result<TestEnvironment> {
416        self.json(
417            self.request(Method::PATCH, &format!("testing-environments/{id}"))?
418                .header("Idempotency-Key", key)
419                .json(input),
420        )
421        .await
422    }
423    pub async fn test_environment_key(&self, id: Uuid) -> Result<TestEnvironmentKey> {
424        self.json(self.request(Method::GET, &format!("testing-environments/{id}/key"))?)
425            .await
426    }
427    pub async fn rotate_test_environment_key(
428        &self,
429        id: Uuid,
430        key: &str,
431    ) -> Result<TestEnvironmentKey> {
432        self.json(
433            self.request(
434                Method::POST,
435                &format!("testing-environments/{id}/rotate-key"),
436            )?
437            .header("Idempotency-Key", key),
438        )
439        .await
440    }
441    pub async fn restore_test_environment(&self, id: Uuid, key: &str) -> Result<TestEnvironment> {
442        self.json(
443            self.request(Method::POST, &format!("testing-environments/{id}/restore"))?
444                .header("Idempotency-Key", key),
445        )
446        .await
447    }
448    pub async fn clean_test_environment(&self, id: Uuid, key: &str) -> Result<()> {
449        self.empty(
450            self.request(Method::POST, &format!("testing-environments/{id}/clean"))?
451                .header("Idempotency-Key", key),
452        )
453        .await
454    }
455    pub async fn delete_test_environment(&self, id: Uuid, key: &str) -> Result<()> {
456        self.empty(
457            self.request(Method::DELETE, &format!("testing-environments/{id}"))?
458                .header("Idempotency-Key", key),
459        )
460        .await
461    }
462    /// Connects without reading/writing a cursor or automatically acknowledging messages.
463    pub async fn connect(&self, actors: &[String], device_id: &str) -> Result<Socket> {
464        self.connect_with_generation(actors, device_id, None).await
465    }
466    /// Supply the last observed sandbox generation; changed generations reset local cursors.
467    pub async fn connect_with_generation(
468        &self,
469        actors: &[String],
470        device_id: &str,
471        testing_generation: Option<i64>,
472    ) -> Result<Socket> {
473        let mut url = self.endpoint("ws")?;
474        url.set_scheme(if self.base.scheme() == "https" {
475            "wss"
476        } else {
477            "ws"
478        })
479        .map_err(|()| Error::Configuration("invalid WebSocket scheme".into()))?;
480        {
481            let mut query = url.query_pairs_mut();
482            query
483                .append_pair(
484                    "org_id",
485                    self.organization.as_deref().ok_or_else(|| {
486                        Error::Configuration("organization is required for WebSocket".into())
487                    })?,
488                )
489                .append_pair("device_id", device_id);
490            for actor in actors {
491                query.append_pair("actors", actor);
492            }
493        }
494        if let Some(generation) = testing_generation {
495            url.query_pairs_mut()
496                .append_pair("testing_generation", &generation.to_string());
497        }
498        let mut req = url.as_str().into_client_request()?;
499        let token = self
500            .token
501            .as_ref()
502            .ok_or_else(|| Error::Configuration("bearer token is required for WebSocket".into()))?;
503        req.headers_mut().insert(
504            "Authorization",
505            format!("Bearer {token}")
506                .parse()
507                .map_err(|_| Error::Configuration("invalid bearer header".into()))?,
508        );
509        if let Some(key) = &self.test_key {
510            req.headers_mut().insert(
511                "X-Testing-Environment-Key",
512                key.parse()
513                    .map_err(|_| Error::Configuration("invalid test header".into()))?,
514            );
515        }
516        let configuration = tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
517            .max_message_size(Some(self.websocket_limit))
518            .max_frame_size(Some(self.websocket_limit))
519            .max_write_buffer_size(self.websocket_limit.saturating_add(128 * 1024 + 1));
520        let (socket, _) =
521            tokio_tungstenite::connect_async_with_config(req, Some(configuration), true).await?;
522        Ok(socket)
523    }
524}
525pub enum GifList<'a> {
526    Trending,
527    Search(&'a str),
528    Recent,
529}
530
531pub fn validate_endpoint(url: &Url) -> Result<()> {
532    let loopback = matches!(
533        url.host_str(),
534        Some("localhost" | "dm.localhost" | "127.0.0.1" | "[::1]" | "::1")
535    );
536    if url.scheme() != "https" && !(url.scheme() == "http" && loopback) {
537        return Err(Error::Configuration(
538            "use HTTPS, or HTTP on loopback for local development".into(),
539        ));
540    }
541    if !url.username().is_empty() || url.password().is_some() {
542        return Err(Error::Configuration(
543            "URL credentials are not supported".into(),
544        ));
545    }
546    Ok(())
547}
548async fn checked(response: reqwest::Response) -> Result<reqwest::Response> {
549    if response.status().is_success() {
550        return Ok(response);
551    }
552    let status = response.status().as_u16();
553    let request_id = response
554        .headers()
555        .get("x-request-id")
556        .and_then(|v| v.to_str().ok())
557        .map(str::to_owned);
558    let retry_after = response
559        .headers()
560        .get("retry-after")
561        .and_then(|v| v.to_str().ok())
562        .map(str::to_owned);
563    let body = response
564        .json::<Envelope<Value>>()
565        .await
566        .map(|e| e.data)
567        .unwrap_or(Value::Null);
568    let code = body
569        .pointer("/error/code")
570        .and_then(Value::as_str)
571        .unwrap_or("http_error")
572        .to_owned();
573    let message = body
574        .pointer("/error/message")
575        .and_then(Value::as_str)
576        .unwrap_or("request failed; inspect response body")
577        .to_owned();
578    Err(Error::Api {
579        status,
580        code,
581        message,
582        body: Box::new(body),
583        request_id,
584        retry_after,
585    })
586}
587/// Available package release information. A linked Rust library cannot replace itself:
588/// applications must update their dependency and rebuild to load a newer library.
589#[derive(Clone, Debug, Serialize)]
590pub struct UpdateInfo {
591    pub package: String,
592    pub current_version: String,
593    pub latest_version: String,
594    pub rebuild_command: String,
595}
596pub async fn check_update() -> Result<UpdateInfo> {
597    let body = checked(
598        HttpClient::builder()
599            .timeout(Duration::from_secs(8))
600            .user_agent(concat!("silicon-dm-client/", env!("CARGO_PKG_VERSION")))
601            .build()?
602            .get("https://crates.io/api/v1/crates/silicon-dm-client")
603            .send()
604            .await?,
605    )
606    .await?
607    .json::<Value>()
608    .await?;
609    let latest = body
610        .pointer("/crate/max_stable_version")
611        .and_then(Value::as_str)
612        .ok_or_else(|| Error::Configuration("registry did not return a stable version".into()))?;
613    Ok(UpdateInfo {
614        package: "silicon-dm-client".into(),
615        current_version: env!("CARGO_PKG_VERSION").into(),
616        latest_version: latest.into(),
617        rebuild_command: "cargo update -p silicon-dm-client && cargo build --release".into(),
618    })
619}
silicon_dm_client/lib.rs

silicon_dm_client/
lib.rs
