clawdb_client/
client.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use reqwest::{Client as HttpClient, header};
5use serde_json::{json, Value};
6use tracing::info;
7
8use crate::builder::ClawDBBuilder;
9use crate::error::{SdkError, SdkResult};
10use crate::models::{
11    BranchInfo, DiffResult, HealthResponse, MemoryRecord, MergeResult, ReflectJob,
12    RememberOptions, SearchHit, SearchOptions, SessionInfo, SyncActionResult,
13    SyncResult, SyncStatusResult, TxInfo,
14};
15
16// ─── Internal shared state ────────────────────────────────────────────────────
17
18struct ClawDBInner {
19    endpoint: String,
20    api_key: String,
21    agent_id: String,
22    timeout_ms: u64,
23    http: HttpClient,
24}
25
26// ─── ClawDB client ─────────────────────────────────────────────────────────
27
28/// The primary ClawDB client.
29#[derive(Clone)]
30pub struct ClawDB {
31    inner: Arc<ClawDBInner>,
32}
33
34impl ClawDB {
35    /// Create a builder to configure and construct a ClawDB client.
36    pub fn builder() -> ClawDBBuilder {
37        ClawDBBuilder::new()
38    }
39
40    /// Create a client from environment variables.
41    pub async fn from_env() -> SdkResult<Self> {
42        ClawDBBuilder::from_env().build().await
43    }
44
45    /// Automatically provision a usable endpoint following the SDK fallback order.
46    pub async fn auto_provision() -> SdkResult<Self> {
47        if let Ok(endpoint) = std::env::var("CLAWDB_URL") {
48            return ClawDBBuilder::from_env().endpoint(endpoint).build().await;
49        }
50        if let Ok(api_key) = std::env::var("CLAWDB_API_KEY") {
51            return ClawDBBuilder::from_env()
52                .api_key(api_key)
53                .endpoint("https://cloud.clawdb.dev")
54                .build()
55                .await;
56        }
57        if tokio::net::TcpStream::connect("127.0.0.1:50050").await.is_ok() {
58            return ClawDBBuilder::from_env()
59                .endpoint("http://localhost:50050")
60                .build()
61                .await;
62        }
63        Err(SdkError::Config("could not auto-provision clawdb-server".into()))
64    }
65
66    /// Create a client from an API key and endpoint.
67    pub async fn from_api_key(api_key: impl Into<String>, endpoint: impl Into<String>) -> SdkResult<Self> {
68        ClawDBBuilder::new().api_key(api_key).endpoint(endpoint).build().await
69    }
70
71    pub(crate) async fn new_internal(
72        endpoint: String,
73        api_key: String,
74        agent_id: String,
75        _workspace: String,
76        _role: String,
77        timeout_ms: u64,
78    ) -> SdkResult<Self> {
79        let mut default_headers = header::HeaderMap::new();
80        if !api_key.is_empty() {
81            let val = header::HeaderValue::from_str(&format!("Bearer {}", api_key))
82                .map_err(|_| SdkError::Config("Invalid API key characters".into()))?;
83            default_headers.insert(header::AUTHORIZATION, val);
84        }
85        let http = HttpClient::builder()
86            .default_headers(default_headers)
87            .timeout(Duration::from_millis(timeout_ms))
88            .build()
89            .map_err(|e| SdkError::Config(e.to_string()))?;
90
91        Ok(Self {
92            inner: Arc::new(ClawDBInner { endpoint, api_key, agent_id, timeout_ms, http }),
93        })
94    }
95
96    // ─── HTTP helpers ────────────────────────────────────────────────────
97
98    async fn get(&self, path: &str) -> SdkResult<Value> {
99        let url = format!("{}{}", self.inner.endpoint, path);
100        let resp = self.inner.http.get(&url).send().await.map_err(SdkError::Reqwest)?;
101        self.parse_response(resp).await
102    }
103
104    async fn post(&self, path: &str, body: Value) -> SdkResult<Value> {
105        let url = format!("{}{}", self.inner.endpoint, path);
106        let resp = self.inner.http.post(&url).json(&body).send().await.map_err(SdkError::Reqwest)?;
107        self.parse_response(resp).await
108    }
109
110    async fn patch(&self, path: &str, body: Value) -> SdkResult<Value> {
111        let url = format!("{}{}", self.inner.endpoint, path);
112        let resp = self.inner.http.patch(&url).json(&body).send().await.map_err(SdkError::Reqwest)?;
113        self.parse_response(resp).await
114    }
115
116    async fn delete(&self, path: &str) -> SdkResult<Value> {
117        let url = format!("{}{}", self.inner.endpoint, path);
118        let resp = self.inner.http.delete(&url).send().await.map_err(SdkError::Reqwest)?;
119        self.parse_response(resp).await
120    }
121
122    async fn parse_response(&self, resp: reqwest::Response) -> SdkResult<Value> {
123        let status = resp.status().as_u16();
124        let text = resp.text().await.map_err(SdkError::Reqwest)?;
125        if status >= 400 {
126            return Err(SdkError::from_http(status, &text));
127        }
128        if text.is_empty() {
129            return Ok(Value::Null);
130        }
131        serde_json::from_str(&text).map_err(SdkError::Serialization)
132    }
133
134    fn str_field(v: &Value, field: &str) -> String {
135        v[field].as_str().unwrap_or_default().to_string()
136    }
137
138    fn extract_array<T: serde::de::DeserializeOwned>(v: &Value, field: &str) -> Vec<T> {
139        let arr = if v[field].is_array() { &v[field] } else { v };
140        serde_json::from_value(arr.clone()).unwrap_or_default()
141    }
142
143    // ─── Health ──────────────────────────────────────────────────────────
144
145    /// Check server health.
146    pub async fn health(&self) -> SdkResult<HealthResponse> {
147        let v = self.get("/v1/health").await?;
148        Ok(HealthResponse {
149            status: Self::str_field(&v, "status"),
150            version: v["version"].as_str().map(str::to_string),
151        })
152    }
153
154    /// Ping the server (returns Ok if reachable).
155    pub async fn ping(&self) -> SdkResult<()> {
156        self.health().await.map(|_| ())
157    }
158
159    // ─── Sessions ────────────────────────────────────────────────────────
160
161    /// Create a new session token.
162    pub async fn create_session(&self, role: &str, scopes: &[&str], ttl_secs: u64) -> SdkResult<SessionInfo> {
163        let v = self.post("/v1/sessions", json!({
164            "role": role,
165            "scopes": scopes,
166            "ttl_secs": ttl_secs,
167        })).await?;
168        Ok(SessionInfo {
169            session_id: Self::str_field(&v, "session_id"),
170            role: Self::str_field(&v, "role"),
171            scopes: v["scopes"].as_array()
172                .map(|a| a.iter().filter_map(|s| s.as_str().map(str::to_string)).collect())
173                .unwrap_or_default(),
174            expires_at: None,
175        })
176    }
177
178    /// Validate the current session.
179    pub async fn validate_session(&self) -> SdkResult<SessionInfo> {
180        let v = self.get("/v1/sessions/me").await?;
181        Ok(SessionInfo {
182            session_id: Self::str_field(&v, "session_id"),
183            role: Self::str_field(&v, "role"),
184            scopes: v["scopes"].as_array()
185                .map(|a| a.iter().filter_map(|s| s.as_str().map(str::to_string)).collect())
186                .unwrap_or_default(),
187            expires_at: None,
188        })
189    }
190
191    /// Revoke a session by ID.
192    pub async fn revoke_session(&self, session_id: &str) -> SdkResult<bool> {
193        let v = self.delete(&format!("/v1/sessions/{}", session_id)).await?;
194        Ok(v["revoked"].as_bool().unwrap_or(true))
195    }
196
197    /// Get the count of active sessions.
198    pub async fn active_session_count(&self) -> SdkResult<u64> {
199        let v = self.get("/v1/sessions/active/count").await?;
200        Ok(v["count"].as_u64().unwrap_or(0))
201    }
202
203    // ─── Memory ──────────────────────────────────────────────────────────
204
205    /// Store a plain memory and return its ID.
206    pub async fn remember(&self, content: impl Into<String>) -> SdkResult<String> {
207        let content = content.into();
208        if content.trim().is_empty() {
209            return Err(SdkError::Validation { field: "content".into(), constraint: "must be non-empty".into() });
210        }
211        let v = self.post("/v1/memories", json!({ "content": content })).await?;
212        Ok(Self::str_field(&v, "id"))
213    }
214
215    /// Store a typed memory and return its ID.
216    pub async fn remember_typed(&self, content: impl Into<String>, opts: RememberOptions) -> SdkResult<String> {
217        let content = content.into();
218        let mut body = json!({ "content": content });
219        if let Some(mt) = &opts.memory_type {
220            body["type"] = json!(mt);
221        }
222        if let Some(tags) = &opts.tags {
223            body["tags"] = json!(tags);
224        }
225        if let Some(ttl) = opts.ttl_days {
226            body["ttl_days"] = json!(ttl);
227        }
228        let v = self.post("/v1/memories", body).await?;
229        Ok(Self::str_field(&v, "id"))
230    }
231
232    /// Update an existing memory's content.
233    pub async fn update_memory(&self, memory_id: &str, content: impl Into<String>) -> SdkResult<bool> {
234        let v = self.patch(&format!("/v1/memories/{}", memory_id), json!({ "content": content.into() })).await?;
235        Ok(v["updated"].as_bool().unwrap_or(true))
236    }
237
238    /// Semantic search over memories.
239    pub async fn search(&self, query: impl Into<String>, opts: SearchOptions) -> SdkResult<Vec<SearchHit>> {
240        let top_k = opts.top_k.unwrap_or(5);
241        let v = self.post("/v1/memories/search", json!({
242            "query": query.into(),
243            "top_k": top_k,
244            "semantic": opts.semantic.unwrap_or(true),
245        })).await?;
246        let hits: Vec<SearchHit> = Self::extract_array(&v, "hits");
247        Ok(hits)
248    }
249
250    /// Recall specific memories by ID.
251    pub async fn recall(&self, ids: &[&str]) -> SdkResult<Vec<MemoryRecord>> {
252        let v = self.post("/v1/memories/recall", json!({ "ids": ids })).await?;
253        let records: Vec<MemoryRecord> = Self::extract_array(&v, "memories");
254        Ok(records)
255    }
256
257    /// List memories with optional type filter.
258    pub async fn list_memories(&self, memory_type: Option<&str>, limit: Option<u32>) -> SdkResult<Vec<MemoryRecord>> {
259        let mut path = "/v1/memories?".to_string();
260        if let Some(mt) = memory_type { path.push_str(&format!("type={}&", mt)); }
261        if let Some(l) = limit { path.push_str(&format!("limit={}", l)); }
262        let v = self.get(&path).await?;
263        let records: Vec<MemoryRecord> = Self::extract_array(&v, "memories");
264        Ok(records)
265    }
266
267    /// Delete a memory by ID.
268    pub async fn delete_memory(&self, memory_id: &str) -> SdkResult<bool> {
269        let v = self.delete(&format!("/v1/memories/{}", memory_id)).await?;
270        Ok(v["deleted"].as_bool().unwrap_or(true))
271    }
272
273    // ─── Branches ────────────────────────────────────────────────────────
274
275    /// Fork a new branch.
276    pub async fn branch(&self, name: &str, from: Option<&str>) -> SdkResult<BranchInfo> {
277        let mut body = json!({ "name": name });
278        if let Some(f) = from { body["from_branch_id"] = json!(f); }
279        let v = self.post("/v1/branches", body).await?;
280        Ok(BranchInfo {
281            branch_id: Self::str_field(&v, "branch_id"),
282            name: Self::str_field(&v, "name"),
283            branch_json: v["branch_json"].as_str().map(str::to_string),
284        })
285    }
286
287    /// List all branches.
288    pub async fn list_branches(&self) -> SdkResult<Vec<BranchInfo>> {
289        let v = self.get("/v1/branches").await?;
290        let branches: Vec<BranchInfo> = Self::extract_array(&v, "branches");
291        Ok(branches)
292    }
293
294    /// Get a branch by ID.
295    pub async fn get_branch(&self, branch_id: &str) -> SdkResult<BranchInfo> {
296        let v = self.get(&format!("/v1/branches/{}", branch_id)).await?;
297        Ok(BranchInfo {
298            branch_id: Self::str_field(&v, "branch_id"),
299            name: Self::str_field(&v, "name"),
300            branch_json: v["branch_json"].as_str().map(str::to_string),
301        })
302    }
303
304    /// Get a branch by name.
305    pub async fn get_branch_by_name(&self, name: &str) -> SdkResult<BranchInfo> {
306        let v = self.get(&format!("/v1/branches/by-name/{}", name)).await?;
307        Ok(BranchInfo {
308            branch_id: Self::str_field(&v, "branch_id"),
309            name: Self::str_field(&v, "name"),
310            branch_json: v["branch_json"].as_str().map(str::to_string),
311        })
312    }
313
314    /// Get the trunk (main) branch.
315    pub async fn get_trunk_branch(&self) -> SdkResult<BranchInfo> {
316        let v = self.get("/v1/branches/trunk").await?;
317        Ok(BranchInfo {
318            branch_id: Self::str_field(&v, "branch_id"),
319            name: Self::str_field(&v, "name"),
320            branch_json: v["branch_json"].as_str().map(str::to_string),
321        })
322    }
323
324    /// Diff a branch against a target.
325    pub async fn diff(&self, source_branch_id: &str, target_branch_id: &str) -> SdkResult<DiffResult> {
326        let v = self.get(&format!("/v1/branches/{}/diff?target={}", source_branch_id, target_branch_id)).await?;
327        Ok(DiffResult {
328            added: v["added"].as_u64().unwrap_or(0) as u32,
329            removed: v["removed"].as_u64().unwrap_or(0) as u32,
330            modified: v["modified"].as_u64().unwrap_or(0) as u32,
331            unchanged: v["unchanged"].as_u64().unwrap_or(0) as u32,
332            divergence_score: v["divergence_score"].as_f64().unwrap_or(0.0),
333            diff_json: v["diff_json"].as_str().map(str::to_string),
334        })
335    }
336
337    /// Merge source branch into target.
338    pub async fn merge(&self, source_branch_id: &str, target_branch_id: &str, strategy: &str) -> SdkResult<MergeResult> {
339        let v = self.post(&format!("/v1/branches/{}/merge", source_branch_id), json!({
340            "target_branch_id": target_branch_id,
341            "strategy": strategy,
342        })).await?;
343        Ok(MergeResult {
344            success: v["success"].as_bool().unwrap_or(true),
345            applied: v["applied"].as_u64().unwrap_or(0) as u32,
346            skipped: v["skipped"].as_u64().unwrap_or(0) as u32,
347            conflicts: v["conflicts"].as_u64().unwrap_or(0) as u32,
348            duration_ms: v["duration_ms"].as_u64().unwrap_or(0),
349        })
350    }
351
352    /// Discard (delete) a branch.
353    pub async fn discard_branch(&self, branch_id: &str) -> SdkResult<bool> {
354        let v = self.delete(&format!("/v1/branches/{}", branch_id)).await?;
355        Ok(v["discarded"].as_bool().unwrap_or(true))
356    }
357
358    /// Archive a branch.
359    pub async fn archive_branch(&self, branch_id: &str) -> SdkResult<bool> {
360        let v = self.post(&format!("/v1/branches/{}/archive", branch_id), json!({})).await?;
361        Ok(v["archived"].as_bool().unwrap_or(true))
362    }
363
364    // ─── Sync ────────────────────────────────────────────────────────────
365
366    /// Full bidirectional sync.
367    pub async fn sync(&self) -> SdkResult<SyncResult> {
368        let v = self.post("/v1/sync", json!({})).await?;
369        Ok(SyncResult {
370            pushed: v["pushed"].as_u64().unwrap_or(0) as u32,
371            pulled: v["pulled"].as_u64().unwrap_or(0) as u32,
372            conflicts: v["conflicts"].as_u64().unwrap_or(0) as u32,
373            duration_ms: v["duration_ms"].as_u64().unwrap_or(0),
374        })
375    }
376
377    /// Push local memories to remote.
378    pub async fn push_sync(&self) -> SdkResult<SyncActionResult> {
379        let v = self.post("/v1/sync/push", json!({})).await?;
380        Ok(SyncActionResult { summary_json: Some(v.to_string()) })
381    }
382
383    /// Pull remote memories to local.
384    pub async fn pull_sync(&self) -> SdkResult<SyncActionResult> {
385        let v = self.post("/v1/sync/pull", json!({})).await?;
386        Ok(SyncActionResult { summary_json: Some(v.to_string()) })
387    }
388
389    /// Reconcile divergent sync state.
390    pub async fn reconcile_sync(&self) -> SdkResult<SyncActionResult> {
391        let v = self.post("/v1/sync/reconcile", json!({})).await?;
392        Ok(SyncActionResult { summary_json: Some(v.to_string()) })
393    }
394
395    /// Get current sync status.
396    pub async fn sync_status(&self) -> SdkResult<SyncStatusResult> {
397        let v = self.get("/v1/sync/status").await?;
398        Ok(SyncStatusResult { status_json: Some(v.to_string()) })
399    }
400
401    // ─── Reflect ─────────────────────────────────────────────────────────
402
403    /// Trigger a new reflection job.
404    pub async fn reflect(&self) -> SdkResult<ReflectJob> {
405        let v = self.post("/v1/reflect", json!({ "agent_id": self.inner.agent_id })).await?;
406        Ok(ReflectJob {
407            job_id: Self::str_field(&v, "job_id"),
408            status: Self::str_field(&v, "status"),
409            message: v["message"].as_str().map(str::to_string),
410            skipped: v["skipped"].as_bool().unwrap_or(false),
411        })
412    }
413
414    /// List reflection jobs.
415    pub async fn reflect_list_jobs(&self, agent_id: &str) -> SdkResult<Vec<ReflectJob>> {
416        let v = self.get(&format!("/v1/reflect/jobs?agent_id={}", agent_id)).await?;
417        let jobs: Vec<ReflectJob> = Self::extract_array(&v, "jobs");
418        Ok(jobs)
419    }
420
421    /// Get a specific reflection job.
422    pub async fn reflect_get_job(&self, job_id: &str) -> SdkResult<ReflectJob> {
423        let v = self.get(&format!("/v1/reflect/jobs/{}", job_id)).await?;
424        Ok(ReflectJob {
425            job_id: Self::str_field(&v, "job_id"),
426            status: Self::str_field(&v, "status"),
427            message: v["message"].as_str().map(str::to_string),
428            skipped: v["skipped"].as_bool().unwrap_or(false),
429        })
430    }
431
432    /// Get extracted facts for an agent.
433    pub async fn reflect_get_facts(&self, agent_id: &str) -> SdkResult<Value> {
434        self.get(&format!("/v1/reflect/facts/{}", agent_id)).await
435    }
436
437    /// Get preferences for an agent.
438    pub async fn reflect_get_preferences(&self, agent_id: &str) -> SdkResult<Value> {
439        self.get(&format!("/v1/reflect/preferences/{}", agent_id)).await
440    }
441
442    /// Get contradictions for an agent.
443    pub async fn reflect_get_contradictions(&self, agent_id: &str) -> SdkResult<Value> {
444        self.get(&format!("/v1/reflect/contradictions/{}", agent_id)).await
445    }
446
447    /// Resolve a specific contradiction.
448    pub async fn reflect_resolve_contradiction(
449        &self,
450        agent_id: &str,
451        contradiction_id: &str,
452        strategy: &str,
453        merged_value_json: Option<&str>,
454    ) -> SdkResult<Value> {
455        self.post(
456            &format!("/v1/reflect/contradictions/{}/{}/resolve", agent_id, contradiction_id),
457            json!({ "strategy": strategy, "merged_value_json": merged_value_json }),
458        ).await
459    }
460
461    // ─── Transactions ─────────────────────────────────────────────────────
462
463    /// Begin a new transaction.
464    pub async fn begin_tx(&self) -> SdkResult<TxInfo> {
465        let v = self.post("/v1/tx", json!({})).await?;
466        Ok(TxInfo { tx_id: Self::str_field(&v, "tx_id") })
467    }
468
469    /// Add a plain memory to a transaction.
470    pub async fn tx_remember(&self, tx_id: &str, content: impl Into<String>) -> SdkResult<String> {
471        let v = self.post(&format!("/v1/tx/{}/memories", tx_id), json!({ "content": content.into() })).await?;
472        Ok(Self::str_field(&v, "id"))
473    }
474
475    /// Add a typed memory to a transaction.
476    pub async fn tx_remember_typed(&self, tx_id: &str, content: impl Into<String>, opts: RememberOptions) -> SdkResult<String> {
477        let mut body = json!({ "content": content.into() });
478        if let Some(mt) = &opts.memory_type { body["type"] = json!(mt); }
479        if let Some(tags) = &opts.tags { body["tags"] = json!(tags); }
480        let v = self.post(&format!("/v1/tx/{}/memories/typed", tx_id), body).await?;
481        Ok(Self::str_field(&v, "id"))
482    }
483
484    /// Commit a transaction.
485    pub async fn commit_tx(&self, tx_id: &str) -> SdkResult<bool> {
486        let v = self.post(&format!("/v1/tx/{}/commit", tx_id), json!({})).await?;
487        Ok(v["committed"].as_bool().unwrap_or(true))
488    }
489
490    /// Roll back a transaction.
491    pub async fn rollback_tx(&self, tx_id: &str) -> SdkResult<bool> {
492        let v = self.post(&format!("/v1/tx/{}/rollback", tx_id), json!({})).await?;
493        Ok(v["rolled_back"].as_bool().unwrap_or(true))
494    }
495
496    /// Close the client (no-op for HTTP).
497    pub fn close(&self) {
498        info!("clawdb.close");
499    }
500
501    /// Return the configured endpoint.
502    pub fn endpoint(&self) -> &str {
503        &self.inner.endpoint
504    }
505
506    /// Return the configured agent_id.
507    pub fn agent_id(&self) -> &str {
508        &self.inner.agent_id
509    }
510}
511
512// ─── ClawDBClient alias (legacy compat) ───────────────────────────────────
513
514/// Alias for [`ClawDB`] for backwards compatibility.
515#[derive(Clone)]
516pub struct ClawDBClient {
517    pub(crate) inner: ClawDB,
518}
519
520impl ClawDBClient {
521    pub async fn auto_provision() -> SdkResult<Self> {
522        Ok(Self { inner: ClawDB::auto_provision().await? })
523    }
524
525    pub fn builder() -> crate::builder::ClawDBBuilder {
526        ClawDB::builder()
527    }
528}
529
530impl std::ops::Deref for ClawDBClient {
531    type Target = ClawDB;
532    fn deref(&self) -> &Self::Target {
533        &self.inner
534    }
535}
536
537impl std::fmt::Debug for ClawDB {
538    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
539        f.debug_struct("ClawDB")
540            .field("endpoint", &self.inner.endpoint)
541            .field("agent_id", &self.inner.agent_id)
542            .finish()
543    }
544}
545
546impl std::fmt::Debug for ClawDBClient {
547    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
548        f.debug_struct("ClawDBClient")
549            .field("inner", &self.inner)
550            .finish()
551    }
552}
clawdb_client/client.rs

clawdb_client/
client.rs
