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

silicon_dm_client/
lib.rs
