//! Request/response types matching the MNW SyncKit server API. use crate::ids::{AppId, DeviceId, GroupId, UserId}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use std::fmt; use uuid::Uuid; // ── Change operations ── /// The operation type for a sync change entry. /// /// Serializes to/from uppercase strings (`"INSERT"`, `"UPDATE"`, `"DELETE"`) /// matching the server wire protocol. Invalid values are rejected at /// deserialization time. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "UPPERCASE")] pub enum ChangeOp { /// A row was inserted. Insert, /// A row was updated. Update, /// A row was deleted. Delete, } impl fmt::Display for ChangeOp { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { ChangeOp::Insert => write!(f, "INSERT"), ChangeOp::Update => write!(f, "UPDATE"), ChangeOp::Delete => write!(f, "DELETE"), } } } impl ChangeOp { /// Parse from an uppercase string, returning `None` for unrecognized values. /// /// Case-sensitive: only `"INSERT"`, `"UPDATE"`, `"DELETE"` are accepted. /// Useful when reading raw op strings from a local database. pub fn from_str_opt(s: &str) -> Option { match s { "INSERT" => Some(ChangeOp::Insert), "UPDATE" => Some(ChangeOp::Update), "DELETE" => Some(ChangeOp::Delete), _ => None, } } } // ── Hybrid logical clock ── /// A hybrid logical clock (HLC) timestamp: a wall-clock millisecond reading kept /// monotonic by a tiebreak counter, with the originating device as a final /// tiebreak. HLCs are the ordering used for conflict resolution. /// /// Why not a bare wall-clock timestamp: two edits in the same millisecond would /// be unordered. The HLC keeps causality (it never goes backwards and advances /// past anything it has seen) and breaks exact ties deterministically by `node`, /// so every device resolves a conflict identically. /// /// The HLC does NOT, on its own, stop a device with a fast/hostile clock from /// winning every LWW conflict, the highest `wall_ms` still wins. That is bounded /// separately, at comparison time, by the forward-drift cap in /// [`resolve_lww_at`](crate::conflict::resolve_lww_at) (see `MAX_HLC_DRIFT_MS`). /// /// Ordering is lexicographic: `(wall_ms, counter, node)`. The `node` field makes /// the value globally unique across devices, so resolution is convergent without /// any side-channel, comparing two complete HLCs yields the same winner on every /// device. /// /// The value travels inside the E2E-encrypted change payload (the SyncKit server /// never sees or orders by it), so adding it required no server-side change. /// /// `Hlc` deliberately does **not** implement `Default`. A defaulted HLC would /// carry the nil [`DeviceId`], a non-unique, always-losing clock, and minting /// one by accident (`Hlc::default()`) silently produced a change that loses every /// conflict. Mint a real clock with [`Hlc::tick`]/[`Hlc::observe`]; the only /// sanctioned nil-node clock is the explicit legacy floor ([`hlc_legacy_floor`]), /// used solely for pre-HLC entries that arrive with no embedded clock. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] pub struct Hlc { /// Wall-clock component, milliseconds since the Unix epoch. pub wall_ms: i64, /// Monotonic tiebreak within a single `wall_ms`, reset to 0 when `wall_ms` advances. pub counter: u32, /// The device that minted this HLC. Final, deterministic tiebreak, and the /// component that makes the value globally unique across devices. pub node: DeviceId, } impl Hlc { /// The zero HLC for `node`. Used as the initial clock state and as the floor /// for legacy entries that predate HLC support. pub fn zero(node: DeviceId) -> Self { Hlc { wall_ms: 0, counter: 0, node, } } /// Synthesize an HLC for a legacy change that carried no embedded clock, /// deriving the wall component from its `client_timestamp`. This keeps /// pre-HLC entries orderable against new ones during the transition. pub fn from_legacy(timestamp_ms: i64, node: DeviceId) -> Self { Hlc { wall_ms: timestamp_ms, counter: 0, node, } } /// Advance the clock for a **local** event happening at `now_ms` on `node`. /// /// Standard HLC send rule: if physical time has moved past the last reading, /// adopt it and reset the counter; otherwise hold the wall component and bump /// the counter so the new event still sorts strictly after the previous one. pub fn tick(prev: Hlc, now_ms: i64, node: DeviceId) -> Hlc { if now_ms > prev.wall_ms { Hlc { wall_ms: now_ms, counter: 0, node, } } else { bump_counter(prev.wall_ms, prev.counter, node) } } /// Advance the clock on **receiving** `remote`, at local physical time /// `now_ms` on `node`. /// /// Standard HLC receive rule: take the max wall component across the local /// clock, the incoming HLC, and physical time, then derive a counter that is /// strictly greater than any counter seen at that wall component. This keeps /// the local clock ahead of everything observed, so subsequent local writes /// causally follow the remote change. pub fn observe(prev: Hlc, remote: Hlc, now_ms: i64, node: DeviceId) -> Hlc { let wall = prev.wall_ms.max(remote.wall_ms).max(now_ms); let base = if wall == prev.wall_ms && wall == remote.wall_ms { prev.counter.max(remote.counter) } else if wall == prev.wall_ms { prev.counter } else if wall == remote.wall_ms { remote.counter } else { // Physical time strictly leads both clocks: counter resets cleanly. return Hlc { wall_ms: wall, counter: 0, node, }; }; bump_counter(wall, base, node) } } /// The legacy HLC floor: a zero clock on the nil [`DeviceId`]. This is the only /// sanctioned way to obtain a nil-node clock, used for pre-HLC entries that /// deserialize without an embedded clock (they must lose to any real HLC). It is /// wired as the `serde` default for [`ChangeEntry::hlc`]; everywhere else a clock /// is minted via [`Hlc::tick`]/[`Hlc::observe`], so the old silent `Hlc::default()` /// nil-node foot-gun no longer compiles. pub(crate) fn hlc_legacy_floor() -> Hlc { Hlc::zero(DeviceId::nil()) } /// Increment a counter at `wall_ms`, rolling into the wall component on the /// (astronomically unlikely, but attacker-reachable via a hostile peer's /// `counter: u32::MAX`) overflow instead of wrapping to 0, which would send the /// clock *backwards* and break the monotonicity HLC exists to guarantee. fn bump_counter(wall_ms: i64, counter: u32, node: DeviceId) -> Hlc { match counter.checked_add(1) { Some(next) => Hlc { wall_ms, counter: next, node, }, None => match wall_ms.checked_add(1) { Some(wall_ms) => Hlc { wall_ms, counter: 0, node, }, // Absolute ceiling {i64::MAX, u32::MAX}: no representable greater HLC // exists. Clamp (stay put) rather than reset the counter to 0, which // would move the clock backwards and break monotonicity. Unreachable // in practice (year ~292 million); a hostile maximal HLC is poisoned // and loses in LWW regardless. None => Hlc { wall_ms, counter: u32::MAX, node, }, }, } } // ── Auth ── #[derive(Serialize)] pub(crate) struct AuthRequest<'a> { pub email: &'a str, pub password: &'a str, pub api_key: &'a str, /// Developer-defined SDK key. Identifies which billing slot uploads /// from this session count against. The developer's backend picks one /// key per user/workspace/org. pub key: &'a str, } #[derive(Deserialize)] pub(crate) struct AuthResponse { pub token: String, pub user_id: UserId, pub app_id: AppId, } // ── Devices ── #[derive(Serialize)] pub(crate) struct RegisterDeviceRequest { pub device_name: String, pub platform: String, } /// A registered device belonging to a user, as returned by the server. #[derive(Debug, Clone, Deserialize, Serialize)] #[non_exhaustive] pub struct Device { /// Server-assigned device UUID. pub id: DeviceId, /// The SyncKit app this device belongs to. pub app_id: AppId, /// The user who owns this device. pub user_id: UserId, /// Human-readable name (e.g., "MacBook Pro"). pub device_name: String, /// OS identifier (e.g., "macos", "linux", "windows"). pub platform: String, /// Last time this device synced. pub last_seen_at: DateTime, /// When this device was first registered. pub created_at: DateTime, } // ── Groups ── /// A group the authenticated user belongs to, as returned by /// [`SyncKitClient::list_groups`](crate::SyncKitClient::list_groups). #[derive(Debug, Clone, Deserialize)] #[non_exhaustive] pub struct SyncGroup { /// Server-assigned group id. pub id: GroupId, /// The app this group belongs to. pub app_id: AppId, /// The admin who mints the GCK and manages membership. pub admin_user_id: UserId, /// Human-readable group name. pub name: String, /// Current GCK generation. A change here means the group key rotated and a /// fresh grant must be fetched. pub gck_version: i32, /// When the group was created. pub created_at: DateTime, } /// A member of a group, as returned by /// [`SyncKitClient::list_members`](crate::SyncKitClient::list_members). Grants are /// not included, each member fetches only their own. #[derive(Debug, Clone, Deserialize)] #[non_exhaustive] pub struct GroupMember { /// The member's account id. pub user_id: UserId, /// The member's account email, so a consumer's admin UI can identify them /// rather than showing a bare user id. pub email: String, /// `"admin"` or `"member"`. pub role: String, /// When they were added. pub added_at: DateTime, } /// The caller's sealed Group Content Key grant, as returned by /// [`SyncKitClient::group_grant`](crate::SyncKitClient::group_grant). Opened with /// the member's identity private key to recover the GCK. #[derive(Debug, Clone, Deserialize)] #[non_exhaustive] pub struct GroupGrant { /// The GCK sealed to this member's identity public key (base64), opaque to /// the server. pub sealed_gck: String, /// The GCK generation this grant was sealed under. pub gck_version: i32, } // ── Push / Pull ── /// A change entry for pushing to the server. /// `data` is plaintext here, the client encrypts it before sending. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChangeEntry { /// The source table name (opaque to server, meaningful to the app). pub table: String, /// Insert, Update, or Delete. pub op: ChangeOp, /// App-assigned row identifier (typically a UUID string). pub row_id: String, /// When this change was made on the originating device. Retained for the /// server's incremental-pull cursor and for display; conflict ordering uses /// [`Self::hlc`], not this field. pub timestamp: DateTime, /// Hybrid logical clock for this change, the value conflict resolution /// orders by. The app mints it from its persistent clock at the time the /// local change is recorded (see [`Hlc::tick`]). It travels inside the /// E2E-encrypted payload, so the server never sees it. A legacy entry that /// arrives without an embedded clock defaults to the explicit /// [`hlc_legacy_floor`] (nil node, always loses) rather than a silent /// `Hlc::default()`. #[serde(default = "hlc_legacy_floor")] pub hlc: Hlc, /// The row payload as JSON. `None` for Delete operations. /// Serialized as absent (not null) when `None`. #[serde(skip_serializing_if = "Option::is_none")] pub data: Option, /// Forward-compat capture of any fields a newer client added to this /// (decrypted, E2E) payload that this build does not know about. Preserved /// verbatim through deserialize -> re-serialize so an older client cannot /// silently *drop* a field a newer client relies on (e.g. a future tombstone /// marker). Empty on current-format entries, so it serializes to nothing and /// the wire format is unchanged. /// /// The guarantee is **round-trip preservation only**. Conflict resolution /// ([`resolve_lww`](crate::conflict::resolve_lww) / /// [`resolve_field_merge`](crate::conflict::resolve_field_merge)) orders and /// merges on `data` alone, it does not read `extra`. A newer client that /// wants a field in `extra` to influence the winner must also promote it into /// `data`; parking it only in `extra` preserves it across the merge but does /// not change which side wins. #[serde(flatten)] pub extra: serde_json::Map, } /// Wire format sent to server (data is already encrypted). #[derive(Serialize)] pub(crate) struct WirePushRequest { /// The device sending the changes. pub device_id: DeviceId, /// Client-generated UUID for idempotent push. If a push with the same /// batch_id was already committed, the server returns the existing cursor. pub batch_id: Uuid, /// Encrypted change entries ready for the wire. pub changes: Vec, } #[derive(Debug, Serialize)] pub(crate) struct WireChangeEntry { pub table: String, pub op: ChangeOp, pub row_id: String, pub timestamp: DateTime, pub data: Option, } #[derive(Deserialize)] pub(crate) struct PushResponse { pub cursor: i64, } #[derive(Serialize)] pub(crate) struct PullRequest { pub device_id: DeviceId, pub cursor: i64, } #[derive(Deserialize)] pub(crate) struct PullResponse { pub changes: Vec, pub cursor: i64, pub has_more: bool, } /// A change entry received from the server during a pull. /// /// `seq` and `device_id` are present in the server response and parsed for /// completeness, but not used by the SDK, consumers track cursors, not /// individual sequence numbers. #[derive(Deserialize, Clone)] pub(crate) struct PullChangeEntry { #[allow(dead_code)] pub seq: i64, #[allow(dead_code)] pub device_id: DeviceId, pub table: String, pub op: ChangeOp, pub row_id: String, pub timestamp: DateTime, pub data: Option, /// Which encryption key was used. None means key_id 1 (pre-rotation). #[serde(default)] pub key_id: Option, } // ── Filtered pull ── /// Optional filters for [`SyncKitClient::pull_filtered`]. /// /// Both fields are optional and compose with AND. An empty/default filter /// is equivalent to an unfiltered pull. #[derive(Debug, Clone, Default, Serialize)] pub struct PullFilter { /// Only return entries for these table names. /// `None` and `Some(vec![])` are treated identically (no table filter). #[serde(skip_serializing_if = "PullFilter::tables_is_empty")] pub tables: Option>, /// Only return entries with `client_timestamp >= since`. #[serde(skip_serializing_if = "Option::is_none")] pub since: Option>, } impl PullFilter { #[allow( clippy::ref_option, reason = "serde skip_serializing_if requires a fn(&T) -> bool signature" )] fn tables_is_empty(tables: &Option>) -> bool { match tables { None => true, Some(v) => v.is_empty(), } } } #[derive(Serialize)] pub(crate) struct FilteredPullRequest { pub device_id: DeviceId, pub cursor: i64, #[serde(skip_serializing_if = "PullFilter::tables_is_empty")] pub tables: Option>, #[serde(skip_serializing_if = "Option::is_none")] pub since: Option>, } // ── Pulled change (rich metadata) ── /// A change entry from pull with server metadata preserved. /// /// Wraps [`ChangeEntry`] with `device_id` and `seq` fields that are normally /// discarded during `pull()`. Used by [`SyncKitClient::pull_rich`] for /// conflict detection, the `device_id` identifies whether a change came from /// another device, and `seq` provides total server ordering. #[derive(Debug, Clone)] #[non_exhaustive] pub struct PulledChange { /// The decrypted change entry. pub entry: ChangeEntry, /// The device that originated this change. pub device_id: DeviceId, /// Server sequence number (total ordering). pub seq: i64, } // ── Keys ── #[derive(Serialize)] pub(crate) struct PutKeyRequest { pub encrypted_key: String, /// Expected key version for optimistic concurrency control. pub expected_version: i32, } #[derive(Deserialize)] #[allow(dead_code)] pub(crate) struct GetKeyResponse { pub encrypted_key: String, /// Server-side key version, incremented on each password change. #[serde(default)] pub key_version: Option, /// Current active key identifier. #[serde(default)] pub key_id: Option, /// If a rotation is in progress, the new key envelope and its key_id. #[serde(default)] pub pending_key: Option, } #[derive(Debug, Clone, Deserialize)] #[allow(dead_code)] pub(crate) struct PendingKeyInfo { pub encrypted_key: String, pub key_id: i32, } // ── Key Rotation ── #[derive(Serialize)] pub(crate) struct BeginRotationRequest { pub device_id: DeviceId, pub new_encrypted_key: String, pub expected_key_version: i32, } #[derive(Deserialize)] pub(crate) struct BeginRotationResponse { pub rotation_id: Uuid, pub target_seq: i64, pub new_key_id: i32, } #[derive(Serialize)] pub(crate) struct RotationEntriesRequest { pub rotation_id: Uuid, pub after_seq: i64, } #[derive(Deserialize)] pub(crate) struct RotationEntriesResponse { pub entries: Vec, pub has_more: bool, } #[derive(Deserialize)] pub(crate) struct RotationEntry { pub seq: i64, /// Source table, echoed by the server so re-encryption can recompute the /// entry's AEAD associated data (`aad_for_entry(table, row_id)`). pub table: String, /// App-assigned row identifier, see [`Self::table`]. pub row_id: String, pub data: Option, } #[derive(Serialize)] pub(crate) struct RotationBatchRequest { pub rotation_id: Uuid, pub entries: Vec, } #[derive(Serialize)] pub(crate) struct RotationBatchEntry { pub seq: i64, pub data: Option, } #[derive(Deserialize)] #[allow(dead_code)] pub(crate) struct RotationBatchResponse { pub updated_count: u64, } #[derive(Serialize)] pub(crate) struct CompleteRotationRequest { pub rotation_id: Uuid, } // ── OAuth ── // // The token request is sent as form-encoded params built inline in // `authenticate_with_code` (the endpoint expects `application/x-www-form-urlencoded`, // not JSON), so there is no request struct here, only the response. #[derive(Deserialize)] pub(crate) struct OAuthTokenResponse { pub access_token: String, #[allow(dead_code)] pub token_type: String, #[allow(dead_code)] pub expires_in: i64, pub user_id: UserId, pub app_id: AppId, } // ── Status ── /// Server-side sync status for the authenticated app/user. #[derive(Debug, Deserialize)] #[non_exhaustive] pub struct SyncStatus { /// Number of changelog entries on the server for this app/user. pub total_changes: i64, /// Sequence number of the most recent change. `None` if no changes exist. pub latest_cursor: Option, } // ── Blobs ── #[derive(Serialize)] pub(crate) struct BlobUploadUrlRequest { pub hash: String, pub size_bytes: i64, } /// Server response to a blob upload-URL request. #[derive(Deserialize)] #[non_exhaustive] pub struct BlobUploadUrlResponse { /// Presigned S3 PUT URL. Empty string when `already_exists` is true. pub upload_url: String, /// True if the server already has a blob with this hash (skip upload). /// /// Consumer contract: this is a *server-asserted* bool, and a consumer that /// acts on it skips sending bytes it holds locally. A server that answers /// `true` for content it does not actually have makes the upload a silent /// no-op, so the blob is absent later. That is a liveness failure, /// not an integrity one: nothing accepts unverified bytes as a result. The /// download path re-hashes what it receives and rejects any mismatch /// (`SyncKitError::IntegrityFailed`), so a lying server can withhold a blob /// but cannot substitute one. Treat a skipped upload as unconfirmed rather /// than as proof the server holds the content. pub already_exists: bool, } #[derive(Serialize)] pub(crate) struct BlobConfirmRequest { pub hash: String, pub size_bytes: i64, } // ── Blob multipart session ── // // The chunked transport for blobs above the server's one-shot PUT ceiling. // `size_bytes` throughout is the *ciphertext* length (`blob_encrypted_len` of // the plaintext), which both sides need to derive the same part geometry. #[derive(Serialize)] pub(crate) struct BlobMultipartStartRequest { pub hash: String, pub size_bytes: i64, } #[derive(Deserialize)] pub(crate) struct BlobMultipartStartResponse { pub upload_id: String, /// Ciphertext bytes per part; every part but the last is exactly this long. pub part_size: u64, pub part_count: u32, /// Server already holds this content address, no session was opened. Same /// consumer contract as [`BlobUploadUrlResponse::already_exists`]. pub already_exists: bool, } #[derive(Serialize)] pub(crate) struct BlobMultipartPartsRequest { pub hash: String, pub upload_id: String, pub size_bytes: i64, pub first_part: u32, pub count: u32, /// SHA-256 of each requested part, base64 of the raw digest, aligned with /// `first_part..first_part + count`. The server binds these into the /// presigned URLs as `x-amz-checksum-sha256`, so S3 rehashes each part and /// refuses a corrupted one at write time. pub checksums: Vec, } #[derive(Deserialize)] pub(crate) struct BlobMultipartPartsResponse { pub parts: Vec, } #[derive(Deserialize)] pub(crate) struct BlobMultipartPartUrl { pub part_number: i32, /// The `Content-Length` this URL was signed for. The PUT must send exactly /// this many bytes. pub content_length: u64, pub url: String, } #[derive(Serialize)] pub(crate) struct BlobMultipartCompleteRequest { pub hash: String, pub upload_id: String, pub parts: Vec, } #[derive(Serialize)] pub(crate) struct BlobMultipartCompletedPart { pub part_number: i32, pub etag: String, } #[derive(Serialize)] pub(crate) struct BlobMultipartAbortRequest { pub hash: String, pub upload_id: String, } #[derive(Serialize)] pub(crate) struct BlobDownloadUrlRequest { pub hash: String, } #[derive(Deserialize)] pub(crate) struct BlobDownloadUrlResponse { pub download_url: String, } #[cfg(test)] mod tests { use super::*; use serde_json::json; #[test] fn change_op_serde_roundtrip() { for (variant, expected_str) in [ (ChangeOp::Insert, "\"INSERT\""), (ChangeOp::Update, "\"UPDATE\""), (ChangeOp::Delete, "\"DELETE\""), ] { let serialized = serde_json::to_string(&variant).unwrap(); assert_eq!(serialized, expected_str); let deserialized: ChangeOp = serde_json::from_str(&serialized).unwrap(); assert_eq!(deserialized, variant); } } #[test] fn change_op_display_matches_serde() { for variant in [ChangeOp::Insert, ChangeOp::Update, ChangeOp::Delete] { let display = variant.to_string(); let serde_str = serde_json::to_string(&variant).unwrap(); // serde wraps in quotes, Display does not assert_eq!(format!("\"{display}\""), serde_str); } } #[test] fn change_op_from_str_opt_rejects_lowercase() { assert_eq!(ChangeOp::from_str_opt("insert"), None); assert_eq!(ChangeOp::from_str_opt("update"), None); assert_eq!(ChangeOp::from_str_opt("delete"), None); } #[test] fn change_op_from_str_opt_rejects_unknown() { assert_eq!(ChangeOp::from_str_opt("UPSERT"), None); assert_eq!(ChangeOp::from_str_opt(""), None); assert_eq!(ChangeOp::from_str_opt("MERGE"), None); } #[test] fn change_op_is_copy_and_eq() { let op = ChangeOp::Insert; let copied = op; // Copy assert_eq!(op, copied); } #[test] fn change_op_hash_works() { use std::collections::HashSet; let mut set = HashSet::new(); set.insert(ChangeOp::Insert); set.insert(ChangeOp::Update); set.insert(ChangeOp::Delete); set.insert(ChangeOp::Insert); // duplicate assert_eq!(set.len(), 3); } #[test] fn change_entry_serialization_omits_none_data() { let entry = ChangeEntry { table: "t".into(), op: ChangeOp::Delete, row_id: "r".into(), timestamp: chrono::Utc::now(), hlc: Hlc::zero(DeviceId::nil()), data: None, extra: serde_json::Map::default(), }; let json = serde_json::to_string(&entry).unwrap(); assert!( !json.contains("\"data\""), "None data should be omitted: {json}" ); } #[test] fn change_entry_serialization_includes_some_data() { let entry = ChangeEntry { table: "t".into(), op: ChangeOp::Insert, row_id: "r".into(), timestamp: chrono::Utc::now(), hlc: Hlc::zero(DeviceId::nil()), data: Some(json!({"k": "v"})), extra: serde_json::Map::default(), }; let json = serde_json::to_string(&entry).unwrap(); assert!( json.contains("\"data\""), "Some data should be present: {json}" ); } #[test] fn change_entry_preserves_unknown_fields_across_roundtrip() { let json = r#"{ "table": "tasks", "op": "INSERT", "row_id": "r1", "timestamp": "2025-01-15T10:00:00Z", "data": {"title": "test"}, "future_marker": "should survive", "another_unknown": 42 }"#; let entry: ChangeEntry = serde_json::from_str(json).unwrap(); assert_eq!(entry.table, "tasks"); assert_eq!(entry.op, ChangeOp::Insert); assert_eq!(entry.data.as_ref().unwrap()["title"], "test"); // A newer client's fields are captured, not dropped... assert_eq!(entry.extra["future_marker"], "should survive"); assert_eq!(entry.extra["another_unknown"], 42); // ...and round-trip back out, so an older client re-serializing this // entry cannot silently discard a resolution-relevant field. let reserialized = serde_json::to_value(&entry).unwrap(); assert_eq!(reserialized["future_marker"], "should survive"); assert_eq!(reserialized["another_unknown"], 42); } #[test] fn device_deserialization_with_iso_timestamps() { let json = r#"{ "id": "550e8400-e29b-41d4-a716-446655440000", "app_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8", "user_id": "550e8400-e29b-41d4-a716-446655440001", "device_name": "MacBook Pro", "platform": "macos", "last_seen_at": "2025-06-15T14:30:00.123Z", "created_at": "2025-01-01T00:00:00Z" }"#; let device: Device = serde_json::from_str(json).unwrap(); assert_eq!(device.device_name, "MacBook Pro"); assert_eq!(device.platform, "macos"); assert_eq!( device.id.as_uuid(), Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap() ); } #[test] fn sync_status_with_zero_total_changes() { let json = r#"{"total_changes": 0, "latest_cursor": null}"#; let status: SyncStatus = serde_json::from_str(json).unwrap(); assert_eq!(status.total_changes, 0); assert!(status.latest_cursor.is_none()); } #[test] fn blob_upload_url_response_already_exists() { let json = r#"{"upload_url": "", "already_exists": true}"#; let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap(); assert!(resp.already_exists); assert!(resp.upload_url.is_empty()); } #[test] fn change_op_debug_format() { assert_eq!(format!("{:?}", ChangeOp::Insert), "Insert"); assert_eq!(format!("{:?}", ChangeOp::Update), "Update"); assert_eq!(format!("{:?}", ChangeOp::Delete), "Delete"); } // ── PullFilter ── #[test] fn pull_filter_serialization_with_both_fields() { let filter = PullFilter { tables: Some(vec!["tasks".to_string(), "events".to_string()]), since: Some("2025-06-15T12:00:00Z".parse().unwrap()), }; let json = serde_json::to_string(&filter).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["tables"].as_array().unwrap().len(), 2); assert!(parsed["since"].is_string()); } #[test] fn pull_filter_serialization_with_none_fields() { let filter = PullFilter::default(); let json = serde_json::to_string(&filter).unwrap(); // None fields should be omitted entirely assert!(!json.contains("tables")); assert!(!json.contains("since")); assert_eq!(json, "{}"); } #[test] fn pull_filter_empty_tables_vec_omitted() { let filter = PullFilter { tables: Some(vec![]), since: None, }; let json = serde_json::to_string(&filter).unwrap(); assert!( !json.contains("tables"), "empty tables vec should be omitted: {json}" ); assert_eq!(json, "{}"); } #[test] fn filtered_pull_request_includes_filter_fields() { let req = FilteredPullRequest { device_id: DeviceId::new( Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(), ), cursor: 42, tables: Some(vec!["tasks".to_string()]), since: Some("2025-01-01T00:00:00Z".parse().unwrap()), }; let json = serde_json::to_string(&req).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["cursor"], 42); assert_eq!(parsed["tables"].as_array().unwrap().len(), 1); assert!(parsed["since"].is_string()); } // ── Hybrid logical clock ── #[test] fn hlc_orders_by_wall_then_counter_then_node() { let a = DeviceId::new(Uuid::from_u128(1)); let b = DeviceId::new(Uuid::from_u128(2)); // wall dominates assert!( Hlc { wall_ms: 1, counter: 9, node: b } < Hlc { wall_ms: 2, counter: 0, node: a } ); // then counter assert!( Hlc { wall_ms: 5, counter: 0, node: b } < Hlc { wall_ms: 5, counter: 1, node: a } ); // then node assert!( Hlc { wall_ms: 5, counter: 1, node: a } < Hlc { wall_ms: 5, counter: 1, node: b } ); } #[test] fn hlc_tick_adopts_physical_time_and_resets_counter() { let node = DeviceId::new(Uuid::from_u128(1)); let prev = Hlc { wall_ms: 1000, counter: 4, node, }; let next = Hlc::tick(prev, 2000, node); assert_eq!( next, Hlc { wall_ms: 2000, counter: 0, node } ); } #[test] fn hlc_tick_bumps_counter_when_clock_has_not_advanced() { let node = DeviceId::new(Uuid::from_u128(1)); // Physical time equal to or behind the last reading: hold wall, bump counter, // so the new local event still sorts strictly after the previous one. let prev = Hlc { wall_ms: 1000, counter: 4, node, }; assert_eq!( Hlc::tick(prev, 1000, node), Hlc { wall_ms: 1000, counter: 5, node } ); assert_eq!( Hlc::tick(prev, 500, node), Hlc { wall_ms: 1000, counter: 5, node } ); // Strictly increasing under a stuck clock. let mut h = Hlc::zero(node); let mut last = h; for _ in 0..100 { h = Hlc::tick(h, 0, node); assert!(h > last); last = h; } } #[test] fn hlc_observe_stays_ahead_of_a_skewed_fast_remote() { let me = DeviceId::new(Uuid::from_u128(1)); let them = DeviceId::new(Uuid::from_u128(2)); // Our physical clock is at 1000; a remote with a fast clock sends 9000. let local = Hlc { wall_ms: 1000, counter: 0, node: me, }; let remote = Hlc { wall_ms: 9000, counter: 3, node: them, }; let merged = Hlc::observe(local, remote, 1000, me); // We adopt the remote wall and a strictly-greater counter, so our next // local write causally follows the remote change despite the skew. assert_eq!(merged.wall_ms, 9000); assert!(merged.counter > remote.counter); assert_eq!(merged.node, me); let next_local = Hlc::tick(merged, 1001, me); assert!( next_local > remote, "a later local write must outrank the skewed remote" ); } #[test] fn hlc_observe_advances_to_physical_time_when_it_leads() { let me = DeviceId::new(Uuid::from_u128(1)); let them = DeviceId::new(Uuid::from_u128(2)); let local = Hlc { wall_ms: 1000, counter: 2, node: me, }; let remote = Hlc { wall_ms: 1500, counter: 0, node: them, }; // Physical time (3000) leads both: adopt it, counter resets. let merged = Hlc::observe(local, remote, 3000, me); assert_eq!( merged, Hlc { wall_ms: 3000, counter: 0, node: me } ); } #[test] fn hlc_counter_overflow_rolls_into_wall_not_backwards() { let me = DeviceId::new(Uuid::from_u128(1)); // A local tick at a saturated counter must not wrap to 0 (which would // move the clock backwards); it rolls the wall component forward instead. let prev = Hlc { wall_ms: 1000, counter: u32::MAX, node: me, }; let next = Hlc::tick(prev, 1000, me); assert!( next > prev, "tick at counter::MAX must stay strictly increasing" ); assert_eq!( next, Hlc { wall_ms: 1001, counter: 0, node: me } ); } #[test] fn hlc_observe_hostile_max_counter_stays_monotonic() { let me = DeviceId::new(Uuid::from_u128(1)); let them = DeviceId::new(Uuid::from_u128(2)); // A hostile/buggy peer sends counter: u32::MAX at our exact wall_ms. let local = Hlc { wall_ms: 5000, counter: 3, node: me, }; let remote = Hlc { wall_ms: 5000, counter: u32::MAX, node: them, }; let merged = Hlc::observe(local, remote, 5000, me); assert!( merged > local, "observe must not wrap backwards on a MAX remote counter" ); assert!(merged > remote); assert_eq!( merged, Hlc { wall_ms: 5001, counter: 0, node: me } ); } #[test] fn hlc_serde_roundtrip() { let h = Hlc { wall_ms: 123, counter: 4, node: DeviceId::new(Uuid::from_u128(7)), }; let j = serde_json::to_value(h).unwrap(); assert_eq!(serde_json::from_value::(j).unwrap(), h); } }