Skip to main content

max / synckit

17.4 KB · 507 lines History Blame Raw
1 //! Device registration and the core encrypted push/pull.
2 //!
3 //! Registers a device, lists the account's devices, and moves change entries
4 //! to and from the server. Rows are encrypted before push and decrypted after
5 //! pull, so the transport only ever carries ciphertext; the `*_filtered` and
6 //! `*_rich` variants narrow the pull or return per-row sync metadata.
7
8 use bytes::Bytes;
9 use tracing::instrument;
10 use uuid::Uuid;
11
12 use crate::{
13 crypto,
14 error::Result,
15 ids::DeviceId,
16 types::{
17 ChangeEntry, Device, FilteredPullRequest, PullChangeEntry, PullFilter, PullRequest,
18 PullResponse, PulledChange, PushResponse, RegisterDeviceRequest, SyncStatus,
19 WirePushRequest,
20 },
21 };
22
23 use super::SyncKitClient;
24 use super::helpers::{Idempotency, check_response};
25
26 impl SyncKitClient {
27 // ── Devices ──
28
29 /// Register a device for sync.
30 ///
31 /// If a device with the same name already exists for this user/app, the
32 /// server upserts: it updates the existing device's platform and
33 /// `last_seen_at` rather than creating a duplicate.
34 #[instrument(skip(self))]
35 pub async fn register_device(&self, device_name: &str, platform: &str) -> Result<Device> {
36 let token = self.require_token()?;
37
38 let body = Bytes::from(serde_json::to_vec(&RegisterDeviceRequest {
39 device_name: device_name.to_string(),
40 platform: platform.to_string(),
41 })?);
42
43 self.retry_request_json(
44 Idempotency::IdempotentWrite {
45 on: "device registration (server-deduped)",
46 },
47 || {
48 let req = self
49 .http
50 .post(&self.endpoints.devices)
51 .bearer_auth(&token)
52 .header("content-type", "application/json")
53 .body(body.clone());
54 async move { check_response(req.send().await?).await }
55 },
56 )
57 .await
58 }
59
60 /// List all devices for the current user.
61 #[instrument(skip(self))]
62 pub async fn list_devices(&self) -> Result<Vec<Device>> {
63 let token = self.require_token()?;
64
65 self.retry_request_json(Idempotency::ReadOnly, || {
66 let req = self.http.get(&self.endpoints.devices).bearer_auth(&token);
67 async move { check_response(req.send().await?).await }
68 })
69 .await
70 }
71
72 // ── Push / Pull ──
73
74 /// Push changes to the server. Encrypts `data` fields automatically.
75 /// Returns the server cursor after the push.
76 ///
77 /// Retries on transient failures (network errors, 5xx, 429) with exponential backoff.
78 #[instrument(skip(self, changes))]
79 pub async fn push(&self, device_id: DeviceId, changes: Vec<ChangeEntry>) -> Result<i64> {
80 let token = self.require_token()?;
81
82 // Every change now seals an HLC envelope (Deletes included), so the master
83 // key is needed whenever there is anything to push.
84 let key_holder = if changes.is_empty() {
85 crypto::ZeroizeOnDrop([0u8; 32])
86 } else {
87 self.require_master_key()?
88 };
89 let master_key: &[u8; 32] = &key_holder;
90 let wire_changes = changes
91 .into_iter()
92 .map(|c| Self::encrypt_change_with_key(c, master_key))
93 .collect::<Result<Vec<_>>>()?;
94
95 let body = Bytes::from(serde_json::to_vec(&WirePushRequest {
96 device_id,
97 batch_id: Uuid::new_v4(),
98 changes: wire_changes,
99 })?);
100
101 let push_resp: PushResponse = self
102 .retry_request_json(Idempotency::Keyed, || {
103 let req = self
104 .http
105 .post(&self.endpoints.push)
106 .bearer_auth(&token)
107 .header("content-type", "application/json")
108 .body(body.clone());
109 async move { check_response(req.send().await?).await }
110 })
111 .await?;
112 Ok(push_resp.cursor)
113 }
114
115 /// Pull changes from the server since the given cursor.
116 /// Decrypts `data` fields automatically.
117 /// Returns (changes, new_cursor, has_more).
118 ///
119 /// Retries on transient failures (network errors, 5xx, 429) with exponential backoff.
120 ///
121 /// **Contract, persist the cursor only after applying.** The returned cursor
122 /// must be written to durable storage *after* the returned changes are
123 /// applied, in the same transaction where possible. Persisting the cursor
124 /// first and crashing before apply silently drops a batch; the SDK does not
125 /// dedup re-delivered changes, so apply must also be idempotent per
126 /// `(table, row_id, hlc)`.
127 #[instrument(skip(self))]
128 pub async fn pull(
129 &self,
130 device_id: DeviceId,
131 cursor: i64,
132 ) -> Result<(Vec<ChangeEntry>, i64, bool)> {
133 let body = Bytes::from(serde_json::to_vec(&PullRequest { device_id, cursor })?);
134 self.pull_inner(body, Self::decrypt_change_with_key).await
135 }
136
137 /// Pull changes from the server with optional table and timestamp filters.
138 /// Decrypts `data` fields automatically.
139 /// Returns (changes, new_cursor, has_more).
140 ///
141 /// Identical to [`pull`](SyncKitClient::pull) when the filter is empty/default.
142 #[instrument(skip(self, filter))]
143 pub async fn pull_filtered(
144 &self,
145 device_id: DeviceId,
146 cursor: i64,
147 filter: PullFilter,
148 ) -> Result<(Vec<ChangeEntry>, i64, bool)> {
149 let body = Bytes::from(serde_json::to_vec(&FilteredPullRequest {
150 device_id,
151 cursor,
152 tables: filter.tables,
153 since: filter.since,
154 })?);
155 self.pull_inner(body, Self::decrypt_change_with_key).await
156 }
157
158 /// Pull changes from the server, preserving `device_id` and `seq` metadata.
159 ///
160 /// Same HTTP call and decryption as [`pull`](SyncKitClient::pull), but returns
161 /// [`PulledChange`] wrappers that retain server metadata needed for conflict
162 /// detection. Returns (changes, new_cursor, has_more).
163 #[instrument(skip(self))]
164 pub async fn pull_rich(
165 &self,
166 device_id: DeviceId,
167 cursor: i64,
168 ) -> Result<(Vec<PulledChange>, i64, bool)> {
169 let body = Bytes::from(serde_json::to_vec(&PullRequest { device_id, cursor })?);
170 self.pull_inner(body, Self::decrypt_change_to_pulled).await
171 }
172
173 /// Pull changes with filters, preserving `device_id` and `seq` metadata.
174 ///
175 /// Same as [`pull_rich`](SyncKitClient::pull_rich) but with table/timestamp
176 /// filtering support. Returns (changes, new_cursor, has_more).
177 #[instrument(skip(self, filter))]
178 pub async fn pull_filtered_rich(
179 &self,
180 device_id: DeviceId,
181 cursor: i64,
182 filter: PullFilter,
183 ) -> Result<(Vec<PulledChange>, i64, bool)> {
184 let body = Bytes::from(serde_json::to_vec(&FilteredPullRequest {
185 device_id,
186 cursor,
187 tables: filter.tables,
188 since: filter.since,
189 })?);
190 self.pull_inner(body, Self::decrypt_change_to_pulled).await
191 }
192
193 /// Shared pull implementation: sends the request, extracts the master key,
194 /// and decrypts each change using the provided function.
195 ///
196 /// If a pending rotation key is cached on the client, entries are decrypted
197 /// with key selection based on each entry's `key_id` field.
198 async fn pull_inner<T, F>(&self, body: Bytes, decrypt_fn: F) -> Result<(Vec<T>, i64, bool)>
199 where
200 F: Fn(PullChangeEntry, &[u8; 32]) -> Result<T>,
201 {
202 let token = self.require_token()?;
203
204 let pull_resp: PullResponse = self
205 .retry_request_json(Idempotency::ReadOnly, || {
206 let req = self
207 .http
208 .post(&self.endpoints.pull)
209 .bearer_auth(&token)
210 .header("content-type", "application/json")
211 .body(body.clone());
212 async move { check_response(req.send().await?).await }
213 })
214 .await?;
215
216 // Extract key once for the entire batch (only needed if any entry has data)
217 let has_data = pull_resp.changes.iter().any(|c| c.data.is_some());
218 let key_holder = if has_data {
219 self.require_master_key()?
220 } else {
221 crypto::ZeroizeOnDrop([0u8; 32])
222 };
223 let master_key: &[u8; 32] = &key_holder;
224
225 // Check for pending rotation key (multi-key decryption)
226 let pending_guard = self.pending_key.read();
227 let has_pending = pending_guard.is_some() && has_data;
228
229 let changes = if has_pending {
230 let pending = pending_guard.as_ref().unwrap();
231 let primary_key_id = *self.master_key_id.read();
232 pull_resp
233 .changes
234 .into_iter()
235 .map(|c| {
236 Self::decrypt_with_rotation_keys(
237 c,
238 master_key,
239 primary_key_id,
240 &pending.key,
241 pending.key_id,
242 &decrypt_fn,
243 )
244 })
245 .collect::<Result<Vec<_>>>()?
246 } else {
247 drop(pending_guard);
248 pull_resp
249 .changes
250 .into_iter()
251 .map(|c| decrypt_fn(c, master_key))
252 .collect::<Result<Vec<_>>>()?
253 };
254
255 Ok((changes, pull_resp.cursor, pull_resp.has_more))
256 }
257
258 /// Get sync status (total changes, latest cursor).
259 #[instrument(skip(self))]
260 pub async fn status(&self) -> Result<SyncStatus> {
261 let token = self.require_token()?;
262
263 self.retry_request_json(Idempotency::ReadOnly, || {
264 let req = self.http.get(&self.endpoints.status).bearer_auth(&token);
265 async move { check_response(req.send().await?).await }
266 })
267 .await
268 }
269 }
270
271 #[cfg(test)]
272 mod tests {
273 use chrono::Utc;
274 use uuid::Uuid;
275
276 use crate::ids::{AppId, DeviceId, UserId};
277 use crate::types::*;
278
279 // ── Type serialization / deserialization ──
280
281 #[test]
282 fn change_entry_serialization_roundtrip() {
283 let entry = ChangeEntry {
284 table: "tasks".to_string(),
285 op: ChangeOp::Insert,
286 row_id: Uuid::new_v4().to_string(),
287 timestamp: Utc::now(),
288 hlc: Hlc::zero(DeviceId::nil()),
289 data: Some(serde_json::json!({"title": "Test task", "done": false})),
290 extra: serde_json::Map::default(),
291 };
292
293 let json = serde_json::to_string(&entry).unwrap();
294 let deserialized: ChangeEntry = serde_json::from_str(&json).unwrap();
295
296 assert_eq!(deserialized.table, entry.table);
297 assert_eq!(deserialized.op, entry.op);
298 assert_eq!(deserialized.row_id, entry.row_id);
299 assert_eq!(deserialized.data, entry.data);
300 }
301
302 #[test]
303 fn change_entry_with_none_data_omits_field() {
304 let entry = ChangeEntry {
305 table: "tasks".to_string(),
306 op: ChangeOp::Delete,
307 row_id: "abc-123".to_string(),
308 timestamp: Utc::now(),
309 hlc: Hlc::zero(DeviceId::nil()),
310 data: None,
311 extra: serde_json::Map::default(),
312 };
313
314 let json = serde_json::to_string(&entry).unwrap();
315 assert!(!json.contains("\"data\""));
316 }
317
318 #[test]
319 fn change_entry_deserialization_with_missing_data() {
320 let json = r#"{
321 "table": "events",
322 "op": "DELETE",
323 "row_id": "evt-1",
324 "timestamp": "2025-01-15T10:00:00Z"
325 }"#;
326
327 let entry: ChangeEntry = serde_json::from_str(json).unwrap();
328 assert_eq!(entry.table, "events");
329 assert_eq!(entry.op, ChangeOp::Delete);
330 assert!(entry.data.is_none());
331 }
332
333 #[test]
334 fn device_serialization_roundtrip() {
335 let device = Device {
336 id: DeviceId::new(Uuid::new_v4()),
337 app_id: AppId::new(Uuid::new_v4()),
338 user_id: UserId::new(Uuid::new_v4()),
339 device_name: "MacBook Pro".to_string(),
340 platform: "macos".to_string(),
341 last_seen_at: Utc::now(),
342 created_at: Utc::now(),
343 };
344
345 let json = serde_json::to_string(&device).unwrap();
346 let deserialized: Device = serde_json::from_str(&json).unwrap();
347
348 assert_eq!(deserialized.id, device.id);
349 assert_eq!(deserialized.device_name, device.device_name);
350 assert_eq!(deserialized.platform, device.platform);
351 }
352
353 #[test]
354 fn sync_status_deserialization() {
355 let json = r#"{"total_changes": 42, "latest_cursor": 100}"#;
356 let status: SyncStatus = serde_json::from_str(json).unwrap();
357 assert_eq!(status.total_changes, 42);
358 assert_eq!(status.latest_cursor, Some(100));
359 }
360
361 #[test]
362 fn sync_status_with_null_cursor() {
363 let json = r#"{"total_changes": 0, "latest_cursor": null}"#;
364 let status: SyncStatus = serde_json::from_str(json).unwrap();
365 assert_eq!(status.total_changes, 0);
366 assert_eq!(status.latest_cursor, None);
367 }
368
369 #[test]
370 fn register_device_request_serialization() {
371 let req = RegisterDeviceRequest {
372 device_name: "iPhone 15".to_string(),
373 platform: "ios".to_string(),
374 };
375
376 let json = serde_json::to_string(&req).unwrap();
377 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
378 assert_eq!(parsed["device_name"], "iPhone 15");
379 assert_eq!(parsed["platform"], "ios");
380 }
381
382 // ── Wire types ──
383
384 #[test]
385 fn wire_push_request_serialization() {
386 let device_id = DeviceId::new(Uuid::new_v4());
387 let req = WirePushRequest {
388 device_id,
389 batch_id: Uuid::new_v4(),
390 changes: vec![WireChangeEntry {
391 table: "tasks".to_string(),
392 op: ChangeOp::Insert,
393 row_id: "r1".to_string(),
394 timestamp: Utc::now(),
395 data: Some(serde_json::json!("encrypted-blob")),
396 }],
397 };
398
399 let json = serde_json::to_string(&req).unwrap();
400 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
401 assert_eq!(parsed["device_id"], device_id.to_string());
402 assert_eq!(parsed["changes"].as_array().unwrap().len(), 1);
403 }
404
405 #[test]
406 fn pull_request_serialization() {
407 let device_id = DeviceId::new(Uuid::new_v4());
408 let req = PullRequest {
409 device_id,
410 cursor: 42,
411 };
412
413 let json = serde_json::to_string(&req).unwrap();
414 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
415 assert_eq!(parsed["device_id"], device_id.to_string());
416 assert_eq!(parsed["cursor"], 42);
417 }
418
419 #[test]
420 fn pull_response_deserialization() {
421 let device_id = Uuid::new_v4();
422 let json = format!(
423 r#"{{
424 "changes": [
425 {{
426 "seq": 1,
427 "device_id": "{device_id}",
428 "table": "tasks",
429 "op": "INSERT",
430 "row_id": "r1",
431 "timestamp": "2025-06-01T12:00:00Z",
432 "data": "encrypted"
433 }}
434 ],
435 "cursor": 5,
436 "has_more": true
437 }}"#
438 );
439
440 let resp: PullResponse = serde_json::from_str(&json).unwrap();
441 assert_eq!(resp.changes.len(), 1);
442 assert_eq!(resp.cursor, 5);
443 assert!(resp.has_more);
444 assert_eq!(resp.changes[0].seq, 1);
445 assert_eq!(resp.changes[0].table, "tasks");
446 }
447
448 #[test]
449 fn pull_response_empty_changes() {
450 let json = r#"{"changes": [], "cursor": 0, "has_more": false}"#;
451 let resp: PullResponse = serde_json::from_str(json).unwrap();
452 assert!(resp.changes.is_empty());
453 assert_eq!(resp.cursor, 0);
454 assert!(!resp.has_more);
455 }
456
457 #[test]
458 fn push_response_deserialization() {
459 let json = r#"{"cursor": 99}"#;
460 let resp: PushResponse = serde_json::from_str(json).unwrap();
461 assert_eq!(resp.cursor, 99);
462 }
463
464 // ── ChangeOp display and parsing ──
465
466 #[test]
467 fn change_op_display() {
468 assert_eq!(ChangeOp::Insert.to_string(), "INSERT");
469 assert_eq!(ChangeOp::Update.to_string(), "UPDATE");
470 assert_eq!(ChangeOp::Delete.to_string(), "DELETE");
471 }
472
473 #[test]
474 fn change_op_from_str_valid() {
475 assert_eq!(ChangeOp::from_str_opt("INSERT"), Some(ChangeOp::Insert));
476 assert_eq!(ChangeOp::from_str_opt("UPDATE"), Some(ChangeOp::Update));
477 assert_eq!(ChangeOp::from_str_opt("DELETE"), Some(ChangeOp::Delete));
478 }
479
480 #[test]
481 fn change_op_from_str_invalid() {
482 assert_eq!(ChangeOp::from_str_opt("insert"), None);
483 assert_eq!(ChangeOp::from_str_opt("UPSERT"), None);
484 assert_eq!(ChangeOp::from_str_opt(""), None);
485 }
486
487 // ── Malformed response types ──
488
489 #[test]
490 fn pull_response_missing_changes_fails() {
491 let json = r#"{"cursor": 0, "has_more": false}"#;
492 assert!(serde_json::from_str::<PullResponse>(json).is_err());
493 }
494
495 #[test]
496 fn pull_response_missing_cursor_fails() {
497 let json = r#"{"changes": [], "has_more": false}"#;
498 assert!(serde_json::from_str::<PullResponse>(json).is_err());
499 }
500
501 #[test]
502 fn push_response_missing_cursor_fails() {
503 let json = r"{}";
504 assert!(serde_json::from_str::<PushResponse>(json).is_err());
505 }
506 }
507