Skip to main content

max / synckit

36.7 KB · 1137 lines History Blame Raw
1 //! Request/response types matching the MNW SyncKit server API.
2
3 use crate::ids::{AppId, DeviceId, GroupId, UserId};
4 use chrono::{DateTime, Utc};
5 use serde::{Deserialize, Serialize};
6 use std::fmt;
7 use uuid::Uuid;
8
9 // ── Change operations ──
10
11 /// The operation type for a sync change entry.
12 ///
13 /// Serializes to/from uppercase strings (`"INSERT"`, `"UPDATE"`, `"DELETE"`)
14 /// matching the server wire protocol. Invalid values are rejected at
15 /// deserialization time.
16 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
17 #[serde(rename_all = "UPPERCASE")]
18 pub enum ChangeOp {
19 Insert,
20 Update,
21 Delete,
22 }
23
24 impl fmt::Display for ChangeOp {
25 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
26 match self {
27 ChangeOp::Insert => write!(f, "INSERT"),
28 ChangeOp::Update => write!(f, "UPDATE"),
29 ChangeOp::Delete => write!(f, "DELETE"),
30 }
31 }
32 }
33
34 impl ChangeOp {
35 /// Parse from an uppercase string, returning `None` for unrecognized values.
36 ///
37 /// Case-sensitive: only `"INSERT"`, `"UPDATE"`, `"DELETE"` are accepted.
38 /// Useful when reading raw op strings from a local database.
39 pub fn from_str_opt(s: &str) -> Option<Self> {
40 match s {
41 "INSERT" => Some(ChangeOp::Insert),
42 "UPDATE" => Some(ChangeOp::Update),
43 "DELETE" => Some(ChangeOp::Delete),
44 _ => None,
45 }
46 }
47 }
48
49 // ── Hybrid logical clock ──
50
51 /// A hybrid logical clock (HLC) timestamp: a wall-clock millisecond reading kept
52 /// monotonic by a tiebreak counter, with the originating device as a final
53 /// tiebreak. HLCs are the ordering used for conflict resolution.
54 ///
55 /// Why not a bare wall-clock timestamp: two edits in the same millisecond would
56 /// be unordered. The HLC keeps causality (it never goes backwards and advances
57 /// past anything it has seen) and breaks exact ties deterministically by `node`,
58 /// so every device resolves a conflict identically.
59 ///
60 /// The HLC does NOT, on its own, stop a device with a fast/hostile clock from
61 /// winning every LWW conflict, the highest `wall_ms` still wins. That is bounded
62 /// separately, at comparison time, by the forward-drift cap in
63 /// [`resolve_lww_at`](crate::conflict::resolve_lww_at) (see `MAX_HLC_DRIFT_MS`).
64 ///
65 /// Ordering is lexicographic: `(wall_ms, counter, node)`. The `node` field makes
66 /// the value globally unique across devices, so resolution is convergent without
67 /// any side-channel, comparing two complete HLCs yields the same winner on every
68 /// device.
69 ///
70 /// The value travels inside the E2E-encrypted change payload (the SyncKit server
71 /// never sees or orders by it), so adding it required no server-side change.
72 ///
73 /// `Hlc` deliberately does **not** implement `Default`. A defaulted HLC would
74 /// carry the nil [`DeviceId`], a non-unique, always-losing clock, and minting
75 /// one by accident (`Hlc::default()`) silently produced a change that loses every
76 /// conflict. Mint a real clock with [`Hlc::tick`]/[`Hlc::observe`]; the only
77 /// sanctioned nil-node clock is the explicit legacy floor ([`hlc_legacy_floor`]),
78 /// used solely for pre-HLC entries that arrive with no embedded clock.
79 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
80 pub struct Hlc {
81 /// Wall-clock component, milliseconds since the Unix epoch.
82 pub wall_ms: i64,
83 /// Monotonic tiebreak within a single `wall_ms`, reset to 0 when `wall_ms` advances.
84 pub counter: u32,
85 /// The device that minted this HLC. Final, deterministic tiebreak, and the
86 /// component that makes the value globally unique across devices.
87 pub node: DeviceId,
88 }
89
90 impl Hlc {
91 /// The zero HLC for `node`. Used as the initial clock state and as the floor
92 /// for legacy entries that predate HLC support.
93 pub fn zero(node: DeviceId) -> Self {
94 Hlc {
95 wall_ms: 0,
96 counter: 0,
97 node,
98 }
99 }
100
101 /// Synthesize an HLC for a legacy change that carried no embedded clock,
102 /// deriving the wall component from its `client_timestamp`. This keeps
103 /// pre-HLC entries orderable against new ones during the transition.
104 pub fn from_legacy(timestamp_ms: i64, node: DeviceId) -> Self {
105 Hlc {
106 wall_ms: timestamp_ms,
107 counter: 0,
108 node,
109 }
110 }
111
112 /// Advance the clock for a **local** event happening at `now_ms` on `node`.
113 ///
114 /// Standard HLC send rule: if physical time has moved past the last reading,
115 /// adopt it and reset the counter; otherwise hold the wall component and bump
116 /// the counter so the new event still sorts strictly after the previous one.
117 pub fn tick(prev: Hlc, now_ms: i64, node: DeviceId) -> Hlc {
118 if now_ms > prev.wall_ms {
119 Hlc {
120 wall_ms: now_ms,
121 counter: 0,
122 node,
123 }
124 } else {
125 bump_counter(prev.wall_ms, prev.counter, node)
126 }
127 }
128
129 /// Advance the clock on **receiving** `remote`, at local physical time
130 /// `now_ms` on `node`.
131 ///
132 /// Standard HLC receive rule: take the max wall component across the local
133 /// clock, the incoming HLC, and physical time, then derive a counter that is
134 /// strictly greater than any counter seen at that wall component. This keeps
135 /// the local clock ahead of everything observed, so subsequent local writes
136 /// causally follow the remote change.
137 pub fn observe(prev: Hlc, remote: Hlc, now_ms: i64, node: DeviceId) -> Hlc {
138 let wall = prev.wall_ms.max(remote.wall_ms).max(now_ms);
139 let base = if wall == prev.wall_ms && wall == remote.wall_ms {
140 prev.counter.max(remote.counter)
141 } else if wall == prev.wall_ms {
142 prev.counter
143 } else if wall == remote.wall_ms {
144 remote.counter
145 } else {
146 // Physical time strictly leads both clocks: counter resets cleanly.
147 return Hlc {
148 wall_ms: wall,
149 counter: 0,
150 node,
151 };
152 };
153 bump_counter(wall, base, node)
154 }
155 }
156
157 /// The legacy HLC floor: a zero clock on the nil [`DeviceId`]. This is the only
158 /// sanctioned way to obtain a nil-node clock, used for pre-HLC entries that
159 /// deserialize without an embedded clock (they must lose to any real HLC). It is
160 /// wired as the `serde` default for [`ChangeEntry::hlc`]; everywhere else a clock
161 /// is minted via [`Hlc::tick`]/[`Hlc::observe`], so the old silent `Hlc::default()`
162 /// nil-node foot-gun no longer compiles.
163 pub(crate) fn hlc_legacy_floor() -> Hlc {
164 Hlc::zero(DeviceId::nil())
165 }
166
167 /// Increment a counter at `wall_ms`, rolling into the wall component on the
168 /// (astronomically unlikely, but attacker-reachable via a hostile peer's
169 /// `counter: u32::MAX`) overflow instead of wrapping to 0, which would send the
170 /// clock *backwards* and break the monotonicity HLC exists to guarantee.
171 fn bump_counter(wall_ms: i64, counter: u32, node: DeviceId) -> Hlc {
172 match counter.checked_add(1) {
173 Some(next) => Hlc {
174 wall_ms,
175 counter: next,
176 node,
177 },
178 None => match wall_ms.checked_add(1) {
179 Some(wall_ms) => Hlc {
180 wall_ms,
181 counter: 0,
182 node,
183 },
184 // Absolute ceiling {i64::MAX, u32::MAX}: no representable greater HLC
185 // exists. Clamp (stay put) rather than reset the counter to 0, which
186 // would move the clock backwards and break monotonicity. Unreachable
187 // in practice (year ~292 million); a hostile maximal HLC is poisoned
188 // and loses in LWW regardless.
189 None => Hlc {
190 wall_ms,
191 counter: u32::MAX,
192 node,
193 },
194 },
195 }
196 }
197
198 // ── Auth ──
199
200 #[derive(Serialize)]
201 pub(crate) struct AuthRequest<'a> {
202 pub email: &'a str,
203 pub password: &'a str,
204 pub api_key: &'a str,
205 /// Developer-defined SDK key. Identifies which billing slot uploads
206 /// from this session count against. The developer's backend picks one
207 /// key per user/workspace/org.
208 pub key: &'a str,
209 }
210
211 #[derive(Deserialize)]
212 pub(crate) struct AuthResponse {
213 pub token: String,
214 pub user_id: UserId,
215 pub app_id: AppId,
216 }
217
218 // ── Devices ──
219
220 #[derive(Serialize)]
221 pub(crate) struct RegisterDeviceRequest {
222 pub device_name: String,
223 pub platform: String,
224 }
225
226 #[derive(Debug, Clone, Deserialize, Serialize)]
227 #[non_exhaustive]
228 pub struct Device {
229 /// Server-assigned device UUID.
230 pub id: DeviceId,
231 /// The SyncKit app this device belongs to.
232 pub app_id: AppId,
233 /// The user who owns this device.
234 pub user_id: UserId,
235 /// Human-readable name (e.g., "MacBook Pro").
236 pub device_name: String,
237 /// OS identifier (e.g., "macos", "linux", "windows").
238 pub platform: String,
239 /// Last time this device synced.
240 pub last_seen_at: DateTime<Utc>,
241 /// When this device was first registered.
242 pub created_at: DateTime<Utc>,
243 }
244
245 // ── Groups ──
246
247 /// A group the authenticated user belongs to, as returned by
248 /// [`SyncKitClient::list_groups`](crate::SyncKitClient::list_groups).
249 #[derive(Debug, Clone, Deserialize)]
250 #[non_exhaustive]
251 pub struct SyncGroup {
252 /// Server-assigned group id.
253 pub id: GroupId,
254 /// The app this group belongs to.
255 pub app_id: AppId,
256 /// The admin who mints the GCK and manages membership.
257 pub admin_user_id: UserId,
258 /// Human-readable group name.
259 pub name: String,
260 /// Current GCK generation. A change here means the group key rotated and a
261 /// fresh grant must be fetched.
262 pub gck_version: i32,
263 /// When the group was created.
264 pub created_at: DateTime<Utc>,
265 }
266
267 /// A member of a group, as returned by
268 /// [`SyncKitClient::list_members`](crate::SyncKitClient::list_members). Grants are
269 /// not included, each member fetches only their own.
270 #[derive(Debug, Clone, Deserialize)]
271 #[non_exhaustive]
272 pub struct GroupMember {
273 /// The member's account id.
274 pub user_id: UserId,
275 /// The member's account email, so a consumer's admin UI can identify them
276 /// rather than showing a bare user id.
277 pub email: String,
278 /// `"admin"` or `"member"`.
279 pub role: String,
280 /// When they were added.
281 pub added_at: DateTime<Utc>,
282 }
283
284 /// The caller's sealed Group Content Key grant, as returned by
285 /// [`SyncKitClient::group_grant`](crate::SyncKitClient::group_grant). Opened with
286 /// the member's identity private key to recover the GCK.
287 #[derive(Debug, Clone, Deserialize)]
288 #[non_exhaustive]
289 pub struct GroupGrant {
290 /// The GCK sealed to this member's identity public key (base64), opaque to
291 /// the server.
292 pub sealed_gck: String,
293 /// The GCK generation this grant was sealed under.
294 pub gck_version: i32,
295 }
296
297 // ── Push / Pull ──
298
299 /// A change entry for pushing to the server.
300 /// `data` is plaintext here, the client encrypts it before sending.
301 #[derive(Debug, Clone, Serialize, Deserialize)]
302 pub struct ChangeEntry {
303 /// The source table name (opaque to server, meaningful to the app).
304 pub table: String,
305 /// Insert, Update, or Delete.
306 pub op: ChangeOp,
307 /// App-assigned row identifier (typically a UUID string).
308 pub row_id: String,
309 /// When this change was made on the originating device. Retained for the
310 /// server's incremental-pull cursor and for display; conflict ordering uses
311 /// [`Self::hlc`], not this field.
312 pub timestamp: DateTime<Utc>,
313 /// Hybrid logical clock for this change, the value conflict resolution
314 /// orders by. The app mints it from its persistent clock at the time the
315 /// local change is recorded (see [`Hlc::tick`]). It travels inside the
316 /// E2E-encrypted payload, so the server never sees it. A legacy entry that
317 /// arrives without an embedded clock defaults to the explicit
318 /// [`hlc_legacy_floor`] (nil node, always loses) rather than a silent
319 /// `Hlc::default()`.
320 #[serde(default = "hlc_legacy_floor")]
321 pub hlc: Hlc,
322 /// The row payload as JSON. `None` for Delete operations.
323 /// Serialized as absent (not null) when `None`.
324 #[serde(skip_serializing_if = "Option::is_none")]
325 pub data: Option<serde_json::Value>,
326 /// Forward-compat capture of any fields a newer client added to this
327 /// (decrypted, E2E) payload that this build does not know about. Preserved
328 /// verbatim through deserialize -> re-serialize so an older client cannot
329 /// silently *drop* a field a newer client relies on (e.g. a future tombstone
330 /// marker). Empty on current-format entries, so it serializes to nothing and
331 /// the wire format is unchanged.
332 ///
333 /// The guarantee is **round-trip preservation only**. Conflict resolution
334 /// ([`resolve_lww`](crate::conflict::resolve_lww) /
335 /// [`resolve_field_merge`](crate::conflict::resolve_field_merge)) orders and
336 /// merges on `data` alone, it does not read `extra`. A newer client that
337 /// wants a field in `extra` to influence the winner must also promote it into
338 /// `data`; parking it only in `extra` preserves it across the merge but does
339 /// not change which side wins.
340 #[serde(flatten)]
341 pub extra: serde_json::Map<String, serde_json::Value>,
342 }
343
344 /// Wire format sent to server (data is already encrypted).
345 #[derive(Serialize)]
346 pub(crate) struct WirePushRequest {
347 /// The device sending the changes.
348 pub device_id: DeviceId,
349 /// Client-generated UUID for idempotent push. If a push with the same
350 /// batch_id was already committed, the server returns the existing cursor.
351 pub batch_id: Uuid,
352 /// Encrypted change entries ready for the wire.
353 pub changes: Vec<WireChangeEntry>,
354 }
355
356 #[derive(Debug, Serialize)]
357 pub(crate) struct WireChangeEntry {
358 pub table: String,
359 pub op: ChangeOp,
360 pub row_id: String,
361 pub timestamp: DateTime<Utc>,
362 pub data: Option<serde_json::Value>,
363 }
364
365 #[derive(Deserialize)]
366 pub(crate) struct PushResponse {
367 pub cursor: i64,
368 }
369
370 #[derive(Serialize)]
371 pub(crate) struct PullRequest {
372 pub device_id: DeviceId,
373 pub cursor: i64,
374 }
375
376 #[derive(Deserialize)]
377 pub(crate) struct PullResponse {
378 pub changes: Vec<PullChangeEntry>,
379 pub cursor: i64,
380 pub has_more: bool,
381 }
382
383 /// A change entry received from the server during a pull.
384 ///
385 /// `seq` and `device_id` are present in the server response and parsed for
386 /// completeness, but not used by the SDK, consumers track cursors, not
387 /// individual sequence numbers.
388 #[derive(Deserialize, Clone)]
389 pub(crate) struct PullChangeEntry {
390 #[allow(dead_code)]
391 pub seq: i64,
392 #[allow(dead_code)]
393 pub device_id: DeviceId,
394 pub table: String,
395 pub op: ChangeOp,
396 pub row_id: String,
397 pub timestamp: DateTime<Utc>,
398 pub data: Option<serde_json::Value>,
399 /// Which encryption key was used. None means key_id 1 (pre-rotation).
400 #[serde(default)]
401 pub key_id: Option<i32>,
402 }
403
404 // ── Filtered pull ──
405
406 /// Optional filters for [`SyncKitClient::pull_filtered`].
407 ///
408 /// Both fields are optional and compose with AND. An empty/default filter
409 /// is equivalent to an unfiltered pull.
410 #[derive(Debug, Clone, Default, Serialize)]
411 pub struct PullFilter {
412 /// Only return entries for these table names.
413 /// `None` and `Some(vec![])` are treated identically (no table filter).
414 #[serde(skip_serializing_if = "PullFilter::tables_is_empty")]
415 pub tables: Option<Vec<String>>,
416 /// Only return entries with `client_timestamp >= since`.
417 #[serde(skip_serializing_if = "Option::is_none")]
418 pub since: Option<DateTime<Utc>>,
419 }
420
421 impl PullFilter {
422 #[allow(
423 clippy::ref_option,
424 reason = "serde skip_serializing_if requires a fn(&T) -> bool signature"
425 )]
426 fn tables_is_empty(tables: &Option<Vec<String>>) -> bool {
427 match tables {
428 None => true,
429 Some(v) => v.is_empty(),
430 }
431 }
432 }
433
434 #[derive(Serialize)]
435 pub(crate) struct FilteredPullRequest {
436 pub device_id: DeviceId,
437 pub cursor: i64,
438 #[serde(skip_serializing_if = "PullFilter::tables_is_empty")]
439 pub tables: Option<Vec<String>>,
440 #[serde(skip_serializing_if = "Option::is_none")]
441 pub since: Option<DateTime<Utc>>,
442 }
443
444 // ── Pulled change (rich metadata) ──
445
446 /// A change entry from pull with server metadata preserved.
447 ///
448 /// Wraps [`ChangeEntry`] with `device_id` and `seq` fields that are normally
449 /// discarded during `pull()`. Used by [`SyncKitClient::pull_rich`] for
450 /// conflict detection, the `device_id` identifies whether a change came from
451 /// another device, and `seq` provides total server ordering.
452 #[derive(Debug, Clone)]
453 #[non_exhaustive]
454 pub struct PulledChange {
455 /// The decrypted change entry.
456 pub entry: ChangeEntry,
457 /// The device that originated this change.
458 pub device_id: DeviceId,
459 /// Server sequence number (total ordering).
460 pub seq: i64,
461 }
462
463 // ── Keys ──
464
465 #[derive(Serialize)]
466 pub(crate) struct PutKeyRequest {
467 pub encrypted_key: String,
468 /// Expected key version for optimistic concurrency control.
469 pub expected_version: i32,
470 }
471
472 #[derive(Deserialize)]
473 #[allow(dead_code)]
474 pub(crate) struct GetKeyResponse {
475 pub encrypted_key: String,
476 /// Server-side key version, incremented on each password change.
477 #[serde(default)]
478 pub key_version: Option<i32>,
479 /// Current active key identifier.
480 #[serde(default)]
481 pub key_id: Option<i32>,
482 /// If a rotation is in progress, the new key envelope and its key_id.
483 #[serde(default)]
484 pub pending_key: Option<PendingKeyInfo>,
485 }
486
487 #[derive(Debug, Clone, Deserialize)]
488 #[allow(dead_code)]
489 pub(crate) struct PendingKeyInfo {
490 pub encrypted_key: String,
491 pub key_id: i32,
492 }
493
494 // ── Key Rotation ──
495
496 #[derive(Serialize)]
497 pub(crate) struct BeginRotationRequest {
498 pub device_id: DeviceId,
499 pub new_encrypted_key: String,
500 pub expected_key_version: i32,
501 }
502
503 #[derive(Deserialize)]
504 pub(crate) struct BeginRotationResponse {
505 pub rotation_id: Uuid,
506 pub target_seq: i64,
507 pub new_key_id: i32,
508 }
509
510 #[derive(Serialize)]
511 pub(crate) struct RotationEntriesRequest {
512 pub rotation_id: Uuid,
513 pub after_seq: i64,
514 }
515
516 #[derive(Deserialize)]
517 pub(crate) struct RotationEntriesResponse {
518 pub entries: Vec<RotationEntry>,
519 pub has_more: bool,
520 }
521
522 #[derive(Deserialize)]
523 pub(crate) struct RotationEntry {
524 pub seq: i64,
525 /// Source table, echoed by the server so re-encryption can recompute the
526 /// entry's AEAD associated data (`aad_for_entry(table, row_id)`).
527 pub table: String,
528 /// App-assigned row identifier, see [`Self::table`].
529 pub row_id: String,
530 pub data: Option<serde_json::Value>,
531 }
532
533 #[derive(Serialize)]
534 pub(crate) struct RotationBatchRequest {
535 pub rotation_id: Uuid,
536 pub entries: Vec<RotationBatchEntry>,
537 }
538
539 #[derive(Serialize)]
540 pub(crate) struct RotationBatchEntry {
541 pub seq: i64,
542 pub data: Option<serde_json::Value>,
543 }
544
545 #[derive(Deserialize)]
546 #[allow(dead_code)]
547 pub(crate) struct RotationBatchResponse {
548 pub updated_count: u64,
549 }
550
551 #[derive(Serialize)]
552 pub(crate) struct CompleteRotationRequest {
553 pub rotation_id: Uuid,
554 }
555
556 // ── OAuth ──
557 //
558 // The token request is sent as form-encoded params built inline in
559 // `authenticate_with_code` (the endpoint expects `application/x-www-form-urlencoded`,
560 // not JSON), so there is no request struct here, only the response.
561
562 #[derive(Deserialize)]
563 pub(crate) struct OAuthTokenResponse {
564 pub access_token: String,
565 #[allow(dead_code)]
566 pub token_type: String,
567 #[allow(dead_code)]
568 pub expires_in: i64,
569 pub user_id: UserId,
570 pub app_id: AppId,
571 }
572
573 // ── Status ──
574
575 #[derive(Debug, Deserialize)]
576 #[non_exhaustive]
577 pub struct SyncStatus {
578 /// Number of changelog entries on the server for this app/user.
579 pub total_changes: i64,
580 /// Sequence number of the most recent change. `None` if no changes exist.
581 pub latest_cursor: Option<i64>,
582 }
583
584 // ── Blobs ──
585
586 #[derive(Serialize)]
587 pub(crate) struct BlobUploadUrlRequest {
588 pub hash: String,
589 pub size_bytes: i64,
590 }
591
592 #[derive(Deserialize)]
593 #[non_exhaustive]
594 pub struct BlobUploadUrlResponse {
595 /// Presigned S3 PUT URL. Empty string when `already_exists` is true.
596 pub upload_url: String,
597 /// True if the server already has a blob with this hash (skip upload).
598 ///
599 /// Consumer contract: this is a *server-asserted* bool, and a consumer that
600 /// acts on it skips sending bytes it holds locally. A server that answers
601 /// `true` for content it does not actually have makes the upload a silent
602 /// no-op, so the blob is absent later. That is a liveness failure,
603 /// not an integrity one: nothing accepts unverified bytes as a result. The
604 /// download path re-hashes what it receives and rejects any mismatch
605 /// (`SyncKitError::IntegrityFailed`), so a lying server can withhold a blob
606 /// but cannot substitute one. Treat a skipped upload as unconfirmed rather
607 /// than as proof the server holds the content.
608 pub already_exists: bool,
609 }
610
611 #[derive(Serialize)]
612 pub(crate) struct BlobConfirmRequest {
613 pub hash: String,
614 pub size_bytes: i64,
615 }
616
617 // ── Blob multipart session ──
618 //
619 // The chunked transport for blobs above the server's one-shot PUT ceiling.
620 // `size_bytes` throughout is the *ciphertext* length (`blob_encrypted_len` of
621 // the plaintext), which both sides need to derive the same part geometry.
622
623 #[derive(Serialize)]
624 pub(crate) struct BlobMultipartStartRequest {
625 pub hash: String,
626 pub size_bytes: i64,
627 }
628
629 #[derive(Deserialize)]
630 pub(crate) struct BlobMultipartStartResponse {
631 pub upload_id: String,
632 /// Ciphertext bytes per part; every part but the last is exactly this long.
633 pub part_size: u64,
634 pub part_count: u32,
635 /// Server already holds this content address, no session was opened. Same
636 /// consumer contract as [`BlobUploadUrlResponse::already_exists`].
637 pub already_exists: bool,
638 }
639
640 #[derive(Serialize)]
641 pub(crate) struct BlobMultipartPartsRequest {
642 pub hash: String,
643 pub upload_id: String,
644 pub size_bytes: i64,
645 pub first_part: u32,
646 pub count: u32,
647 /// SHA-256 of each requested part, base64 of the raw digest, aligned with
648 /// `first_part..first_part + count`. The server binds these into the
649 /// presigned URLs as `x-amz-checksum-sha256`, so S3 rehashes each part and
650 /// refuses a corrupted one at write time.
651 pub checksums: Vec<String>,
652 }
653
654 #[derive(Deserialize)]
655 pub(crate) struct BlobMultipartPartsResponse {
656 pub parts: Vec<BlobMultipartPartUrl>,
657 }
658
659 #[derive(Deserialize)]
660 pub(crate) struct BlobMultipartPartUrl {
661 pub part_number: i32,
662 /// The `Content-Length` this URL was signed for. The PUT must send exactly
663 /// this many bytes.
664 pub content_length: u64,
665 pub url: String,
666 }
667
668 #[derive(Serialize)]
669 pub(crate) struct BlobMultipartCompleteRequest {
670 pub hash: String,
671 pub upload_id: String,
672 pub parts: Vec<BlobMultipartCompletedPart>,
673 }
674
675 #[derive(Serialize)]
676 pub(crate) struct BlobMultipartCompletedPart {
677 pub part_number: i32,
678 pub etag: String,
679 }
680
681 #[derive(Serialize)]
682 pub(crate) struct BlobMultipartAbortRequest {
683 pub hash: String,
684 pub upload_id: String,
685 }
686
687 #[derive(Serialize)]
688 pub(crate) struct BlobDownloadUrlRequest {
689 pub hash: String,
690 }
691
692 #[derive(Deserialize)]
693 pub(crate) struct BlobDownloadUrlResponse {
694 pub download_url: String,
695 }
696
697 #[cfg(test)]
698 mod tests {
699 use super::*;
700 use serde_json::json;
701
702 #[test]
703 fn change_op_serde_roundtrip() {
704 for (variant, expected_str) in [
705 (ChangeOp::Insert, "\"INSERT\""),
706 (ChangeOp::Update, "\"UPDATE\""),
707 (ChangeOp::Delete, "\"DELETE\""),
708 ] {
709 let serialized = serde_json::to_string(&variant).unwrap();
710 assert_eq!(serialized, expected_str);
711 let deserialized: ChangeOp = serde_json::from_str(&serialized).unwrap();
712 assert_eq!(deserialized, variant);
713 }
714 }
715
716 #[test]
717 fn change_op_display_matches_serde() {
718 for variant in [ChangeOp::Insert, ChangeOp::Update, ChangeOp::Delete] {
719 let display = variant.to_string();
720 let serde_str = serde_json::to_string(&variant).unwrap();
721 // serde wraps in quotes, Display does not
722 assert_eq!(format!("\"{display}\""), serde_str);
723 }
724 }
725
726 #[test]
727 fn change_op_from_str_opt_rejects_lowercase() {
728 assert_eq!(ChangeOp::from_str_opt("insert"), None);
729 assert_eq!(ChangeOp::from_str_opt("update"), None);
730 assert_eq!(ChangeOp::from_str_opt("delete"), None);
731 }
732
733 #[test]
734 fn change_op_from_str_opt_rejects_unknown() {
735 assert_eq!(ChangeOp::from_str_opt("UPSERT"), None);
736 assert_eq!(ChangeOp::from_str_opt(""), None);
737 assert_eq!(ChangeOp::from_str_opt("MERGE"), None);
738 }
739
740 #[test]
741 fn change_op_is_copy_and_eq() {
742 let op = ChangeOp::Insert;
743 let copied = op; // Copy
744 assert_eq!(op, copied);
745 }
746
747 #[test]
748 fn change_op_hash_works() {
749 use std::collections::HashSet;
750 let mut set = HashSet::new();
751 set.insert(ChangeOp::Insert);
752 set.insert(ChangeOp::Update);
753 set.insert(ChangeOp::Delete);
754 set.insert(ChangeOp::Insert); // duplicate
755 assert_eq!(set.len(), 3);
756 }
757
758 #[test]
759 fn change_entry_serialization_omits_none_data() {
760 let entry = ChangeEntry {
761 table: "t".into(),
762 op: ChangeOp::Delete,
763 row_id: "r".into(),
764 timestamp: chrono::Utc::now(),
765 hlc: Hlc::zero(DeviceId::nil()),
766 data: None,
767 extra: serde_json::Map::default(),
768 };
769 let json = serde_json::to_string(&entry).unwrap();
770 assert!(
771 !json.contains("\"data\""),
772 "None data should be omitted: {json}"
773 );
774 }
775
776 #[test]
777 fn change_entry_serialization_includes_some_data() {
778 let entry = ChangeEntry {
779 table: "t".into(),
780 op: ChangeOp::Insert,
781 row_id: "r".into(),
782 timestamp: chrono::Utc::now(),
783 hlc: Hlc::zero(DeviceId::nil()),
784 data: Some(json!({"k": "v"})),
785 extra: serde_json::Map::default(),
786 };
787 let json = serde_json::to_string(&entry).unwrap();
788 assert!(
789 json.contains("\"data\""),
790 "Some data should be present: {json}"
791 );
792 }
793
794 #[test]
795 fn change_entry_preserves_unknown_fields_across_roundtrip() {
796 let json = r#"{
797 "table": "tasks",
798 "op": "INSERT",
799 "row_id": "r1",
800 "timestamp": "2025-01-15T10:00:00Z",
801 "data": {"title": "test"},
802 "future_marker": "should survive",
803 "another_unknown": 42
804 }"#;
805 let entry: ChangeEntry = serde_json::from_str(json).unwrap();
806 assert_eq!(entry.table, "tasks");
807 assert_eq!(entry.op, ChangeOp::Insert);
808 assert_eq!(entry.data.as_ref().unwrap()["title"], "test");
809 // A newer client's fields are captured, not dropped...
810 assert_eq!(entry.extra["future_marker"], "should survive");
811 assert_eq!(entry.extra["another_unknown"], 42);
812 // ...and round-trip back out, so an older client re-serializing this
813 // entry cannot silently discard a resolution-relevant field.
814 let reserialized = serde_json::to_value(&entry).unwrap();
815 assert_eq!(reserialized["future_marker"], "should survive");
816 assert_eq!(reserialized["another_unknown"], 42);
817 }
818
819 #[test]
820 fn device_deserialization_with_iso_timestamps() {
821 let json = r#"{
822 "id": "550e8400-e29b-41d4-a716-446655440000",
823 "app_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
824 "user_id": "550e8400-e29b-41d4-a716-446655440001",
825 "device_name": "MacBook Pro",
826 "platform": "macos",
827 "last_seen_at": "2025-06-15T14:30:00.123Z",
828 "created_at": "2025-01-01T00:00:00Z"
829 }"#;
830 let device: Device = serde_json::from_str(json).unwrap();
831 assert_eq!(device.device_name, "MacBook Pro");
832 assert_eq!(device.platform, "macos");
833 assert_eq!(
834 device.id.as_uuid(),
835 Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap()
836 );
837 }
838
839 #[test]
840 fn sync_status_with_zero_total_changes() {
841 let json = r#"{"total_changes": 0, "latest_cursor": null}"#;
842 let status: SyncStatus = serde_json::from_str(json).unwrap();
843 assert_eq!(status.total_changes, 0);
844 assert!(status.latest_cursor.is_none());
845 }
846
847 #[test]
848 fn blob_upload_url_response_already_exists() {
849 let json = r#"{"upload_url": "", "already_exists": true}"#;
850 let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
851 assert!(resp.already_exists);
852 assert!(resp.upload_url.is_empty());
853 }
854
855 #[test]
856 fn change_op_debug_format() {
857 assert_eq!(format!("{:?}", ChangeOp::Insert), "Insert");
858 assert_eq!(format!("{:?}", ChangeOp::Update), "Update");
859 assert_eq!(format!("{:?}", ChangeOp::Delete), "Delete");
860 }
861
862 // ── PullFilter ──
863
864 #[test]
865 fn pull_filter_serialization_with_both_fields() {
866 let filter = PullFilter {
867 tables: Some(vec!["tasks".to_string(), "events".to_string()]),
868 since: Some("2025-06-15T12:00:00Z".parse().unwrap()),
869 };
870 let json = serde_json::to_string(&filter).unwrap();
871 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
872 assert_eq!(parsed["tables"].as_array().unwrap().len(), 2);
873 assert!(parsed["since"].is_string());
874 }
875
876 #[test]
877 fn pull_filter_serialization_with_none_fields() {
878 let filter = PullFilter::default();
879 let json = serde_json::to_string(&filter).unwrap();
880 // None fields should be omitted entirely
881 assert!(!json.contains("tables"));
882 assert!(!json.contains("since"));
883 assert_eq!(json, "{}");
884 }
885
886 #[test]
887 fn pull_filter_empty_tables_vec_omitted() {
888 let filter = PullFilter {
889 tables: Some(vec![]),
890 since: None,
891 };
892 let json = serde_json::to_string(&filter).unwrap();
893 assert!(
894 !json.contains("tables"),
895 "empty tables vec should be omitted: {json}"
896 );
897 assert_eq!(json, "{}");
898 }
899
900 #[test]
901 fn filtered_pull_request_includes_filter_fields() {
902 let req = FilteredPullRequest {
903 device_id: DeviceId::new(
904 Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(),
905 ),
906 cursor: 42,
907 tables: Some(vec!["tasks".to_string()]),
908 since: Some("2025-01-01T00:00:00Z".parse().unwrap()),
909 };
910 let json = serde_json::to_string(&req).unwrap();
911 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
912 assert_eq!(parsed["cursor"], 42);
913 assert_eq!(parsed["tables"].as_array().unwrap().len(), 1);
914 assert!(parsed["since"].is_string());
915 }
916
917 // ── Hybrid logical clock ──
918
919 #[test]
920 fn hlc_orders_by_wall_then_counter_then_node() {
921 let a = DeviceId::new(Uuid::from_u128(1));
922 let b = DeviceId::new(Uuid::from_u128(2));
923 // wall dominates
924 assert!(
925 Hlc {
926 wall_ms: 1,
927 counter: 9,
928 node: b
929 } < Hlc {
930 wall_ms: 2,
931 counter: 0,
932 node: a
933 }
934 );
935 // then counter
936 assert!(
937 Hlc {
938 wall_ms: 5,
939 counter: 0,
940 node: b
941 } < Hlc {
942 wall_ms: 5,
943 counter: 1,
944 node: a
945 }
946 );
947 // then node
948 assert!(
949 Hlc {
950 wall_ms: 5,
951 counter: 1,
952 node: a
953 } < Hlc {
954 wall_ms: 5,
955 counter: 1,
956 node: b
957 }
958 );
959 }
960
961 #[test]
962 fn hlc_tick_adopts_physical_time_and_resets_counter() {
963 let node = DeviceId::new(Uuid::from_u128(1));
964 let prev = Hlc {
965 wall_ms: 1000,
966 counter: 4,
967 node,
968 };
969 let next = Hlc::tick(prev, 2000, node);
970 assert_eq!(
971 next,
972 Hlc {
973 wall_ms: 2000,
974 counter: 0,
975 node
976 }
977 );
978 }
979
980 #[test]
981 fn hlc_tick_bumps_counter_when_clock_has_not_advanced() {
982 let node = DeviceId::new(Uuid::from_u128(1));
983 // Physical time equal to or behind the last reading: hold wall, bump counter,
984 // so the new local event still sorts strictly after the previous one.
985 let prev = Hlc {
986 wall_ms: 1000,
987 counter: 4,
988 node,
989 };
990 assert_eq!(
991 Hlc::tick(prev, 1000, node),
992 Hlc {
993 wall_ms: 1000,
994 counter: 5,
995 node
996 }
997 );
998 assert_eq!(
999 Hlc::tick(prev, 500, node),
1000 Hlc {
1001 wall_ms: 1000,
1002 counter: 5,
1003 node
1004 }
1005 );
1006 // Strictly increasing under a stuck clock.
1007 let mut h = Hlc::zero(node);
1008 let mut last = h;
1009 for _ in 0..100 {
1010 h = Hlc::tick(h, 0, node);
1011 assert!(h > last);
1012 last = h;
1013 }
1014 }
1015
1016 #[test]
1017 fn hlc_observe_stays_ahead_of_a_skewed_fast_remote() {
1018 let me = DeviceId::new(Uuid::from_u128(1));
1019 let them = DeviceId::new(Uuid::from_u128(2));
1020 // Our physical clock is at 1000; a remote with a fast clock sends 9000.
1021 let local = Hlc {
1022 wall_ms: 1000,
1023 counter: 0,
1024 node: me,
1025 };
1026 let remote = Hlc {
1027 wall_ms: 9000,
1028 counter: 3,
1029 node: them,
1030 };
1031 let merged = Hlc::observe(local, remote, 1000, me);
1032 // We adopt the remote wall and a strictly-greater counter, so our next
1033 // local write causally follows the remote change despite the skew.
1034 assert_eq!(merged.wall_ms, 9000);
1035 assert!(merged.counter > remote.counter);
1036 assert_eq!(merged.node, me);
1037 let next_local = Hlc::tick(merged, 1001, me);
1038 assert!(
1039 next_local > remote,
1040 "a later local write must outrank the skewed remote"
1041 );
1042 }
1043
1044 #[test]
1045 fn hlc_observe_advances_to_physical_time_when_it_leads() {
1046 let me = DeviceId::new(Uuid::from_u128(1));
1047 let them = DeviceId::new(Uuid::from_u128(2));
1048 let local = Hlc {
1049 wall_ms: 1000,
1050 counter: 2,
1051 node: me,
1052 };
1053 let remote = Hlc {
1054 wall_ms: 1500,
1055 counter: 0,
1056 node: them,
1057 };
1058 // Physical time (3000) leads both: adopt it, counter resets.
1059 let merged = Hlc::observe(local, remote, 3000, me);
1060 assert_eq!(
1061 merged,
1062 Hlc {
1063 wall_ms: 3000,
1064 counter: 0,
1065 node: me
1066 }
1067 );
1068 }
1069
1070 #[test]
1071 fn hlc_counter_overflow_rolls_into_wall_not_backwards() {
1072 let me = DeviceId::new(Uuid::from_u128(1));
1073 // A local tick at a saturated counter must not wrap to 0 (which would
1074 // move the clock backwards); it rolls the wall component forward instead.
1075 let prev = Hlc {
1076 wall_ms: 1000,
1077 counter: u32::MAX,
1078 node: me,
1079 };
1080 let next = Hlc::tick(prev, 1000, me);
1081 assert!(
1082 next > prev,
1083 "tick at counter::MAX must stay strictly increasing"
1084 );
1085 assert_eq!(
1086 next,
1087 Hlc {
1088 wall_ms: 1001,
1089 counter: 0,
1090 node: me
1091 }
1092 );
1093 }
1094
1095 #[test]
1096 fn hlc_observe_hostile_max_counter_stays_monotonic() {
1097 let me = DeviceId::new(Uuid::from_u128(1));
1098 let them = DeviceId::new(Uuid::from_u128(2));
1099 // A hostile/buggy peer sends counter: u32::MAX at our exact wall_ms.
1100 let local = Hlc {
1101 wall_ms: 5000,
1102 counter: 3,
1103 node: me,
1104 };
1105 let remote = Hlc {
1106 wall_ms: 5000,
1107 counter: u32::MAX,
1108 node: them,
1109 };
1110 let merged = Hlc::observe(local, remote, 5000, me);
1111 assert!(
1112 merged > local,
1113 "observe must not wrap backwards on a MAX remote counter"
1114 );
1115 assert!(merged > remote);
1116 assert_eq!(
1117 merged,
1118 Hlc {
1119 wall_ms: 5001,
1120 counter: 0,
1121 node: me
1122 }
1123 );
1124 }
1125
1126 #[test]
1127 fn hlc_serde_roundtrip() {
1128 let h = Hlc {
1129 wall_ms: 123,
1130 counter: 4,
1131 node: DeviceId::new(Uuid::from_u128(7)),
1132 };
1133 let j = serde_json::to_value(h).unwrap();
1134 assert_eq!(serde_json::from_value::<Hlc>(j).unwrap(), h);
1135 }
1136 }
1137