santui_core/
sync.rs

1use std::sync::Mutex;
2
3use crate::auth::AuthHandle;
4
5/// A pending key-value operation that needs to be synced to the server.
6#[derive(Debug, Clone)]
7pub enum SyncOp {
8    Set {
9        plugin: String,
10        key: String,
11        value: String,
12    },
13    Delete {
14        plugin: String,
15        key: String,
16    },
17}
18
19/// Best-effort sync client that pushes local DB writes to a remote santui-server.
20///
21/// - Writes are queued by [`enqueue`](Self::enqueue).
22/// - On each [`try_sync`](Self::try_sync) call (driven from the main loop), the
23///   queue is drained and pushed to the server via HTTP.
24/// - If the server is unreachable or auth fails, the ops stay queued and are
25///   retried on the next tick.
26#[derive(Debug)]
27pub struct SyncClient {
28    pub server_url: String,
29    jwt: Mutex<Option<String>>,
30    pending: Mutex<Vec<SyncOp>>,
31}
32
33impl SyncClient {
34    pub fn new(server_url: String) -> Self {
35        SyncClient {
36            server_url,
37            jwt: Mutex::new(None),
38            pending: Mutex::new(Vec::new()),
39        }
40    }
41
42    /// Queue an operation for the next sync cycle.
43    pub fn enqueue(&self, op: SyncOp) {
44        if let Ok(mut pending) = self.pending.lock() {
45            pending.push(op);
46        }
47    }
48
49    /// Attempt to sync all pending operations to the server.
50    ///
51    /// If no JWT is cached, tries to authenticate using the provided auth
52    /// handle's bearer token.  Failures are logged — ops stay queued for
53    /// the next call.
54    pub fn try_sync(&self, auth: &Option<std::sync::Arc<dyn AuthHandle>>) {
55        // Step 1: make sure we have a JWT.
56        if self.jwt.lock().map_or(true, |j| j.is_none()) {
57            if let Some(auth) = auth {
58                if let Some(bearer) = auth.bearer_token() {
59                    // Try both providers.
60                    for provider in &["github", "google"] {
61                        match self.exchange_token(provider, &bearer) {
62                            Ok(jwt) => {
63                                if let Ok(mut j) = self.jwt.lock() {
64                                    *j = Some(jwt);
65                                }
66                                break;
67                            }
68                            Err(e) => log::debug!("[sync] {provider} auth failed: {e}"),
69                        }
70                    }
71                }
72            }
73        }
74
75        // Step 2: drain the queue.
76        let ops: Vec<SyncOp> = self
77            .pending
78            .lock()
79            .map_or_else(|_| Vec::new(), |mut p| std::mem::take(&mut *p));
80
81        if ops.is_empty() {
82            return;
83        }
84
85        let jwt_guard = match self.jwt.lock() {
86            Ok(g) => g,
87            Err(_) => return,
88        };
89        let jwt = match jwt_guard.as_ref() {
90            Some(j) => j.clone(),
91            None => {
92                // No valid JWT — ops stay queued.
93                if let Ok(mut pending) = self.pending.lock() {
94                    pending.extend(ops);
95                }
96                return;
97            }
98        };
99        drop(jwt_guard);
100
101        for op in &ops {
102            match op {
103                SyncOp::Set { plugin, key, value } => {
104                    let body = serde_json::json!({
105                        "token": &jwt,
106                        "values": [{"key": key, "value": value}],
107                    });
108                    let url = format!("{}/api/v1/data/{}", self.server_url, plugin);
109                    let result = ureq::post(&url)
110                        .header("Content-Type", "application/json")
111                        .send_json(&body);
112                    if let Err(e) = result {
113                        log::debug!("[sync] push failed for {plugin}/{key}: {e}");
114                        if let Ok(mut pending) = self.pending.lock() {
115                            pending.push(op.clone());
116                        }
117                    }
118                }
119                SyncOp::Delete { plugin, key } => {
120                    let url = format!(
121                        "{}/api/v1/data/{}/{}?token={}",
122                        self.server_url, plugin, key, jwt
123                    );
124                    if let Err(e) = ureq::delete(&url).call() {
125                        log::debug!("[sync] delete failed for {plugin}/{key}: {e}");
126                        if let Ok(mut pending) = self.pending.lock() {
127                            pending.push(op.clone());
128                        }
129                    }
130                }
131            }
132        }
133    }
134
135    fn exchange_token(
136        &self,
137        provider: &str,
138        bearer: &str,
139    ) -> Result<String, Box<dyn std::error::Error>> {
140        let url = format!("{}/auth/login", self.server_url);
141        let body = serde_json::json!({
142            "provider": provider,
143            "token": bearer,
144        });
145        let mut resp = ureq::post(&url)
146            .header("Content-Type", "application/json")
147            .send_json(&body)?;
148        let text = resp.body_mut().read_to_string()?;
149        let parsed: serde_json::Value = serde_json::from_str(&text)?;
150        parsed["jwt"]
151            .as_str()
152            .map(|s| s.to_string())
153            .ok_or_else(|| "no jwt in response".into())
154    }
155}
santui_core/sync.rs

santui_core/
sync.rs
