calimero-node-primitives 0.11.0-rc.2

//! Wire protocol types for sync stream communication.
//!
//! This module contains the message types used for all sync protocol
//! communication over network streams:
//!
//! - [`StreamMessage`]: Top-level message wrapper (Init or Message)
//! - [`InitPayload`]: Initial request types (blob share, key share, delta, snapshot, etc.)
//! - [`MessagePayload`]: Response and follow-up message types
//!
//! # Protocol Flow
//!
//! ```text
//! Initiator                              Responder
//! │                                            │
//! │ ── StreamMessage::Init { payload } ──────► │
//! │                                            │
//! │ ◄── StreamMessage::Message { payload } ── │
//! │                                            │
//! │ ... (continue as needed) ...               │
//! └────────────────────────────────────────────┘
//! ```
//!
//! # Adding New Protocols
//!
//! To add a new sync protocol's wire messages:
//! 1. Add request variant to [`InitPayload`]
//! 2. Add response variant(s) to [`MessagePayload`]
//! 3. Update re-exports in `sync.rs`

use std::borrow::Cow;

use borsh::{BorshDeserialize, BorshSerialize};
use calimero_crypto::Nonce;
use calimero_primitives::blobs::BlobId;
use calimero_primitives::context::ContextId;
use calimero_primitives::hash::Hash;
use calimero_primitives::identity::PublicKey;

use super::hash_comparison::TreeNode;
use super::levelwise::LevelNode;
use super::snapshot::SnapshotError;

/// Maximum depth allowed in TreeNodeRequest.
///
/// Prevents malicious peers from requesting expensive deep traversals.
/// Handlers should validate against this limit before processing.
pub const MAX_TREE_REQUEST_DEPTH: u8 = 16;

// =============================================================================
// Stream Message Wrapper
// =============================================================================

/// Top-level message for sync stream communication.
///
/// All sync protocol messages are wrapped in this enum, which provides:
/// - Context and identity information (in Init)
/// - Sequence tracking (in Message)
/// - Nonce for replay protection
#[derive(Debug, BorshSerialize, BorshDeserialize)]
pub enum StreamMessage<'a> {
    /// Initial message to start a sync operation.
    Init {
        /// Context being synchronized.
        context_id: ContextId,
        /// Identity of the sending party.
        party_id: PublicKey,
        /// The specific request payload.
        payload: InitPayload,
        /// Nonce for the next message.
        next_nonce: Nonce,
    },
    /// Follow-up message in an ongoing sync operation.
    Message {
        /// Sequence number for ordering.
        ///
        /// # Wire Format Change
        ///
        /// Changed from `usize` to `u64` for cross-platform portability.
        /// This is a breaking wire format change - nodes must be upgraded
        /// together to avoid deserialization failures.
        sequence_id: u64,
        /// The message payload.
        payload: MessagePayload<'a>,
        /// Nonce for the next message.
        next_nonce: Nonce,
    },
    /// Opaque error - reveals nothing about node state.
    ///
    /// Used when something goes wrong but we don't want to leak
    /// information to potentially malicious peers (e.g. unverified
    /// dialer, cross-namespace stream leak).
    OpaqueError,
    /// Typed "I am a valid peer for this context but haven't materialised
    /// it locally yet" response. Sent by the receiver when the inbound
    /// stream's `dialer_verified` check passed (the dialer IS a member of
    /// the context's group) but the receiver itself has no local entry
    /// for the context — e.g. they opted out of auto-follow, or the
    /// `JoinContext` is still in flight.
    ///
    /// The initiator MUST treat this as benign: do not increment
    /// `failure_count`, do not apply exponential backoff, just drop this
    /// peer for this round and continue. Closes the Ronit/Fran incident
    /// path where namespace-fallback peer selection picked a non-following
    /// peer and the resulting `OpaqueError` cascaded into 256s backoff
    /// against a peer that fundamentally cannot serve the context.
    ///
    /// # Wire Format Change
    ///
    /// New variant at index 3, appended to the enum tail. Borsh enum
    /// serialization is by declaration order, so old nodes will fail to
    /// deserialize this variant. Coordinated upgrade required across the
    /// network — same constraint as the `sequence_id` u64 change
    /// documented on the `Message` variant. Pre-upgrade peers fall back
    /// to the previous behaviour (no benign signal) and continue working
    /// for all other variants.
    NotMaterialized,
}

// =============================================================================
// Init Payload (Requests)
// =============================================================================

/// Initial request payloads for various sync protocols.
///
/// Each variant represents a different type of sync request that can
/// be initiated by a node.
#[derive(Clone, Debug, BorshSerialize, BorshDeserialize)]
pub enum InitPayload {
    /// Request to share a blob.
    BlobShare {
        /// ID of the blob to share.
        blob_id: BlobId,
    },

    /// Request a specific delta by ID (for DAG gap filling).
    DeltaRequest {
        /// Context for the delta.
        context_id: ContextId,
        /// ID of the specific delta to request.
        delta_id: [u8; 32],
    },

    /// Request peer's current DAG heads for catchup.
    DagHeadsRequest {
        /// Context to get DAG heads for.
        context_id: ContextId,
    },

    /// Request snapshot boundary negotiation.
    SnapshotBoundaryRequest {
        /// Context for snapshot sync.
        context_id: ContextId,
        /// Optional requested cutoff timestamp.
        requested_cutoff_timestamp: Option<u64>,
    },

    /// Request to stream snapshot pages.
    SnapshotStreamRequest {
        /// Context for snapshot sync.
        context_id: ContextId,
        /// Root hash that was negotiated in boundary request.
        boundary_root_hash: Hash,
        /// Maximum pages per response.
        page_limit: u16,
        /// Maximum bytes per response.
        byte_limit: u32,
        /// Resume cursor from previous page (for pagination).
        resume_cursor: Option<Vec<u8>>,
    },

    /// Request tree node(s) for HashComparison sync (CIP §4).
    ///
    /// Used by the HashComparison protocol to request subtrees from a peer
    /// for Merkle tree comparison.
    TreeNodeRequest {
        /// Context being synchronized.
        context_id: ContextId,
        /// ID of the node to request (root hash or entity ID).
        node_id: [u8; 32],
        /// Maximum depth to traverse from this node.
        /// None means only the requested node, Some(1) includes immediate children.
        max_depth: Option<u8>,
    },

    /// Request nodes at a specific level for LevelWise sync (CIP Appendix B).
    ///
    /// Used by the LevelWise protocol for breadth-first tree synchronization,
    /// optimized for wide, shallow trees (depth ≤ 2).
    LevelWiseRequest {
        /// Context being synchronized.
        context_id: ContextId,
        /// Level to request (0 = root's children, 1 = grandchildren, etc.).
        level: u32,
        /// Parent IDs to fetch children for.
        /// - `None` = fetch all nodes at this level
        /// - `Some(ids)` = fetch only children of specified parents
        parent_ids: Option<Vec<[u8; 32]>>,
    },

    /// Push local-only entities to the peer for bidirectional HashComparison sync.
    ///
    /// When the initiator detects subtrees that exist locally but not on the peer
    /// (`RemoteMissing` or `Different` with `local_only_children`), it collects
    /// the leaf data and pushes it to the responder via this message.
    ///
    /// The responder applies CRDT merge (Invariant I5) for each entity.
    EntityPush {
        /// Context being synchronized.
        context_id: ContextId,
        /// Leaf entities to push to the peer.
        entities: Vec<super::hash_comparison::TreeLeafData>,
    },

    /// Push entity *deletions* (tombstones) to the peer during HashComparison.
    ///
    /// The tree comparison is add-wins, so a deletion has to be propagated
    /// explicitly or a peer that still holds the entry never converges (the
    /// clear split-brain). The responder applies each via the authenticated
    /// `Action::DeleteRef` path (delete-wins by HLC).
    EntityDeletePush {
        /// Context being synchronized.
        context_id: ContextId,
        /// Tombstones to apply on the peer.
        deletions: Vec<super::hash_comparison::EntityDeletion>,
    },

    /// Request encrypted payloads for namespace governance skeletons.
    /// Used during lazy backfill when a member joins a new group and
    /// needs to decrypt previously-stored opaque skeletons.
    NamespaceBackfillRequest {
        namespace_id: [u8; 32],
        /// Delta IDs for which we have skeletons but need full payloads.
        delta_ids: Vec<[u8; 32]>,
    },

    /// Direct request to join a namespace. The joiner sends their signed
    /// invitation and public key; the responder validates and returns the
    /// group key + context list in one shot.
    NamespaceJoinRequest {
        namespace_id: [u8; 32],
        /// Borsh-serialized SignedGroupOpenInvitation
        invitation_bytes: Vec<u8>,
        /// The joiner's public key for ECDH key wrapping
        joiner_public_key: PublicKey,
    },

    /// Direct request to materialise inherited membership in an Open
    /// subgroup (issue #2357). Lets the inherited self-join path skip
    /// the gossip-only `MemberJoinedOpen` → `KeyDelivery` round-trip
    /// that times out in small clusters where the gossipsub mesh stays
    /// empty (#2293).
    ///
    /// Responder validates that `joiner_public_key` has
    /// `MembershipPath::Inherited` to `subgroup_id` (proof of
    /// Open-chain authorisation), then wraps the subgroup key via
    /// ECDH and replies with `OpenSubgroupJoinResponse`. Mirrors
    /// `NamespaceJoinRequest`'s direct-stream determinism for the
    /// inherited case.
    OpenSubgroupJoinRequest {
        namespace_id: [u8; 32],
        subgroup_id: [u8; 32],
        /// The joiner's public key for ECDH key wrapping.
        joiner_public_key: PublicKey,
    },

    /// Reconcile per-`Shared`-entity rotation logs at the end of a sync
    /// session (core#2716/#2703).
    ///
    /// A `SharedStorage` writer-set rotation is hash-neutral (the writers
    /// live in `Metadata`, which is `#[borsh(skip)]` out of the Merkle
    /// hash), so HashComparison — which reconciles entity trees by hash and
    /// prunes hash-equal subtrees — never carries it, and a node that
    /// catches up to a peer's branch via HC never learns the rotation. Its
    /// `resolve_local` writer set then stays stale and nodes disagree on who
    /// is a writer even though the data converged.
    ///
    /// This message ships the sender's per-`Shared`-entity rotation logs so
    /// the receiver can UNION them into its own (dedup by `delta_id`). The
    /// receiver replies with [`MessagePayload::RotationLogSyncResponse`]
    /// carrying its own logs, so one round-trip reconciles both directions.
    ///
    /// New variant appended to the enum tail (borsh tags by declaration
    /// order); an older peer that can't decode it simply errors the stream,
    /// which the initiator treats as "nothing to reconcile" — graceful,
    /// best-effort degradation.
    RotationLogSyncRequest {
        /// Context being synchronized.
        context_id: ContextId,
        /// `(entity_id, borsh(RotationLog))` for each `Shared` anchor the
        /// sender holds locally.
        logs: Vec<([u8; 32], Vec<u8>)>,
    },

    /// Pull-based recovery request for a group key the requester is
    /// already an admitted member of but does not yet hold locally.
    ///
    /// This is the durable replacement for the old on-DAG `KeyDelivery`
    /// governance op: instead of an existing member *pushing* the key
    /// once (and only once) when it first applies the join, the joiner
    /// — which is online and syncing — *pulls* the key from its sync
    /// peer every sync round until it has it. The responder validates
    /// that `requester_public_key` is a current member of `group_id`,
    /// then ECDH-wraps the group key and replies with
    /// [`GroupKeyResponse`](MessagePayload::GroupKeyResponse). A
    /// responder that doesn't hold the key replies with an empty
    /// envelope and the joiner tries another peer next round.
    ///
    /// **Borsh ordering**: appended at the tail of `InitPayload` (after
    /// `RotationLogSyncRequest`) so all existing variant discriminants are
    /// unchanged.
    GroupKeyRequest {
        namespace_id: [u8; 32],
        group_id: [u8; 32],
        /// The requester's namespace identity public key, used both for
        /// the membership check and as the ECDH wrap recipient.
        requester_public_key: PublicKey,
    },
}

// =============================================================================
// Message Payload (Responses)
// =============================================================================

/// Response and follow-up message payloads.
///
/// Each variant represents a different type of response or continuation
/// message in a sync protocol exchange.
#[derive(Debug, BorshSerialize, BorshDeserialize)]
pub enum MessagePayload<'a> {
    /// Blob data chunk.
    BlobShare {
        /// Chunk of blob data.
        chunk: Cow<'a, [u8]>,
    },

    /// Response to DeltaRequest containing the requested delta.
    DeltaResponse {
        /// The serialized delta data.
        delta: Cow<'a, [u8]>,
        /// Signing identity of the node that authored this delta.
        /// Required — the responder only serves deltas that have a
        /// recorded author (rows missing the field return
        /// `DeltaNotFound`, forcing the initiator to fall back to
        /// snapshot sync where the per-entity signature path applies).
        /// Initiator runs `membership_status_at` against this author
        /// unconditionally; there is no legacy-accept escape hatch.
        author_id: calimero_primitives::identity::PublicKey,
        /// Serialized `calimero_context_config::types::GovernancePosition`
        /// (borsh bytes) at the delta's sign time. Pairs with
        /// `author_id` for the apply-time `membership_status_at` check.
        /// `None` only for non-group contexts where the author has no
        /// governance cut to cite — initiator skips the membership
        /// check in that case (there's nothing to check against).
        governance_position_blob: Option<Cow<'a, [u8]>>,
        /// Ed25519 signature by `author_id`'s identity key over the
        /// canonical [`super::delta_auth::DeltaSignaturePayload`].
        /// Closes the anti-impersonation gap on the delta envelope:
        /// without this, a current group-key holder could relabel a
        /// foreign delta as their own (or vice versa) since
        /// `membership_status_at` would pass for both members.
        ///
        /// `Option` because legacy rows (deltas authored before the
        /// envelope-signature feature landed) have no signature on
        /// file — the responder forwards `None` for those rather than
        /// withholding the delta entirely. Freshly-authored deltas
        /// always carry `Some(_)` (`internal_execute` signs against
        /// the same `governance_position` it persists on the row).
        /// Initiators MUST verify any present signature; `None` is
        /// tolerated only for that legacy-row case and will tighten
        /// to required once those rows have aged out of every peer's
        /// storage.
        delta_signature: Option<[u8; 64]>,
    },

    /// Delta not found response.
    DeltaNotFound,

    /// Response to DagHeadsRequest containing peer's current heads and root hash.
    DagHeadsResponse {
        /// Current DAG head hashes.
        dag_heads: Vec<[u8; 32]>,
        /// Current root hash.
        root_hash: Hash,
    },

    /// Response to SnapshotBoundaryRequest.
    SnapshotBoundaryResponse {
        /// Authoritative boundary timestamp (nanoseconds since epoch).
        boundary_timestamp: u64,
        /// Root hash for the boundary state.
        boundary_root_hash: Hash,
        /// Peer's DAG heads at the boundary.
        dag_heads: Vec<[u8; 32]>,
    },

    /// A page of snapshot data.
    SnapshotPage {
        /// Compressed payload data.
        payload: Cow<'a, [u8]>,
        /// Uncompressed length for validation.
        uncompressed_len: u32,
        /// Cursor for resuming (None if complete).
        cursor: Option<Vec<u8>>,
        /// Total page count.
        page_count: u64,
        /// Pages sent so far.
        sent_count: u64,
        /// Grand total of shippable `Entity` records across the whole snapshot
        /// — every entity with both an `Index` and an `Entry`, counted from the
        /// sender's full boundary scan and stable across bursts. Excludes
        /// orphans that are never shipped, so it's the exact denominator the
        /// receiver's cumulative applied count reaches (percent → 100, ETA →
        /// 0). `0` means "unknown" — an empty snapshot, or a peer too old to
        /// advertise it; the receiver then reports raw progress only.
        total_records: u64,
    },

    /// Snapshot sync error.
    SnapshotError {
        /// The error that occurred.
        error: SnapshotError,
    },

    /// Response to TreeNodeRequest for HashComparison sync (CIP §4).
    ///
    /// Contains tree nodes from the requested subtree for Merkle comparison.
    TreeNodeResponse {
        /// Tree nodes in the requested subtree.
        ///
        /// For a request with max_depth=0: contains just the requested node.
        /// For max_depth=1: contains the node and its immediate children.
        nodes: Vec<TreeNode>,
        /// True if the requested node was not found.
        not_found: bool,
    },

    /// Response to LevelWiseRequest for LevelWise sync (CIP Appendix B).
    ///
    /// Contains all nodes at the requested level for breadth-first comparison.
    LevelWiseResponse {
        /// Level these nodes are at.
        level: u32,
        /// Nodes at this level.
        ///
        /// Each node includes:
        /// - `id` and `hash` for comparison
        /// - `parent_id` for tree structure
        /// - `leaf_data` if this is a leaf (includes full entity data for CRDT merge)
        nodes: Vec<LevelNode>,
        /// Whether there are more levels below this one.
        has_more_levels: bool,
        /// Authenticated tombstones for children of the queried parents that
        /// this responder has deleted. Lets a holder that still has the entity
        /// (the initiator) learn of the deletion and apply it delete-wins —
        /// the LevelWise analogue of `TreeNode::deleted_children`. Without it a
        /// cleared responder returns no nodes at level 0 and the initiator's
        /// loop exits before ever seeing the deletion (holder-initiates clear).
        deleted_children: Vec<super::hash_comparison::EntityDeletion>,
    },

    /// Acknowledgment of received EntityPush for bidirectional HashComparison sync.
    ///
    /// Sent by the responder after applying CRDT merge for pushed entities.
    EntityPushAck {
        /// Number of entities successfully applied via CRDT merge.
        applied_count: u32,
    },

    /// Acknowledgment of a received `EntityDeletePush`.
    ///
    /// Sent by the responder after applying delete-wins for pushed tombstones.
    EntityDeletePushAck {
        /// Number of deletions successfully applied (delete-wins) by the responder.
        applied_count: u32,
    },

    /// Response containing namespace governance delta payloads for backfill.
    NamespaceBackfillResponse {
        /// Pairs of (delta_id, borsh(SignedNamespaceOp)).
        /// Only includes deltas the responder has full payloads for.
        deltas: Vec<([u8; 32], Vec<u8>)>,
    },

    /// Response to NamespaceJoinRequest with everything the joiner needs.
    NamespaceJoinResponse {
        /// ECDH-wrapped group key envelope (borsh-serialized KeyEnvelope).
        /// Empty if the responder doesn't hold the group key.
        key_envelope_bytes: Vec<u8>,
        /// Context IDs registered under this namespace/group.
        context_ids: Vec<ContextId>,
        /// The application ID used by contexts in this group.
        application_id: [u8; 32],
        /// All namespace governance ops (borsh-serialized SignedNamespaceOp)
        /// so the joiner can replay the full governance history.
        governance_ops: Vec<Vec<u8>>,
        /// Namespace's `default_capabilities` value at the moment the
        /// invitation is fulfilled. Issue #2256: traveling this with the
        /// bundle eliminates the joiner-side hard-coded fallback (the
        /// `create_group` constant) so admin overrides via
        /// `DefaultCapabilitiesSet` are respected even before the
        /// governance op finishes propagating to the joiner.
        default_capabilities: u32,
    },

    /// The responder rejected the join request.
    NamespaceJoinRejected { reason: String },

    /// Response to `OpenSubgroupJoinRequest` (issue #2357). Carries the
    /// ECDH-wrapped subgroup-key envelope; empty if the responder
    /// doesn't hold the key (joiner should try another peer).
    OpenSubgroupJoinResponse {
        /// ECDH-wrapped subgroup-key envelope (borsh-serialized
        /// `KeyEnvelope`).
        key_envelope_bytes: Vec<u8>,
    },

    /// The responder rejected the open-subgroup-join request — e.g.,
    /// joiner has no `MembershipPath::Inherited` to the subgroup, or
    /// the subgroup doesn't belong to the named namespace.
    OpenSubgroupJoinRejected { reason: String },

    /// Response to [`InitPayload::RotationLogSyncRequest`] carrying the
    /// responder's own per-`Shared`-entity rotation logs so the initiator
    /// can union them. See that variant for the full rationale.
    RotationLogSyncResponse {
        /// `(entity_id, borsh(RotationLog))` for each `Shared` anchor the
        /// responder holds locally.
        logs: Vec<([u8; 32], Vec<u8>)>,
    },

    /// Response to [`GroupKeyRequest`](InitPayload::GroupKeyRequest).
    /// Carries the ECDH-wrapped group-key envelope; empty if the
    /// responder doesn't hold the key or the requester isn't a member
    /// (the joiner should try another peer on its next sync round).
    ///
    /// `responder_identity` is the responder's namespace identity — the
    /// trust anchor a keyless bootstrap joiner uses to seed the
    /// namespace admin when it receives the root-group key without an
    /// invitation (replaces the old KeyDelivery-signer trust anchor).
    ///
    /// **Borsh ordering**: appended at the tail of `MessagePayload` (after
    /// `RotationLogSyncResponse`) so all existing variant discriminants are
    /// unchanged.
    GroupKeyResponse {
        /// ECDH-wrapped group-key envelope (borsh-serialized
        /// `KeyEnvelope`). Empty ⇒ no key delivered.
        key_envelope_bytes: Vec<u8>,
        /// Responder's namespace identity public key (the wrap sender).
        responder_identity: PublicKey,
    },
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_init_payload_tree_node_request() {
        let request = InitPayload::TreeNodeRequest {
            context_id: ContextId::from([1u8; 32]),
            node_id: [2u8; 32],
            max_depth: Some(1),
        };

        let encoded = borsh::to_vec(&request).expect("serialize");
        let decoded: InitPayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            InitPayload::TreeNodeRequest {
                context_id,
                node_id,
                max_depth,
            } => {
                assert_eq!(*context_id.as_ref(), [1u8; 32]);
                assert_eq!(node_id, [2u8; 32]);
                assert_eq!(max_depth, Some(1));
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_tree_node_response() {
        use crate::sync::hash_comparison::{LeafMetadata, TreeLeafData, TreeNode};

        let leaf_data = TreeLeafData::new(
            [10u8; 32],
            vec![1, 2, 3],
            LeafMetadata::new(
                crate::sync::hash_comparison::CrdtType::lww_register("test"),
                100,
                [0u8; 32],
            ),
        );
        let node = TreeNode::leaf([1u8; 32], [2u8; 32], leaf_data);

        let response = MessagePayload::TreeNodeResponse {
            nodes: vec![node],
            not_found: false,
        };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::TreeNodeResponse { nodes, not_found } => {
                assert_eq!(nodes.len(), 1);
                assert!(!not_found);
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_tree_node_response_not_found() {
        let response = MessagePayload::TreeNodeResponse {
            nodes: vec![],
            not_found: true,
        };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::TreeNodeResponse { nodes, not_found } => {
                assert!(nodes.is_empty());
                assert!(not_found);
            }
            _ => panic!("wrong variant"),
        }
    }

    // =========================================================================
    // LevelWise Wire Protocol Tests
    // =========================================================================

    #[test]
    fn test_init_payload_levelwise_request_full_level() {
        let request = InitPayload::LevelWiseRequest {
            context_id: ContextId::from([1u8; 32]),
            level: 0,
            parent_ids: None,
        };

        let encoded = borsh::to_vec(&request).expect("serialize");
        let decoded: InitPayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            InitPayload::LevelWiseRequest {
                context_id,
                level,
                parent_ids,
            } => {
                assert_eq!(*context_id.as_ref(), [1u8; 32]);
                assert_eq!(level, 0);
                assert!(parent_ids.is_none());
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_init_payload_levelwise_request_with_parents() {
        let parents = vec![[10u8; 32], [20u8; 32], [30u8; 32]];
        let request = InitPayload::LevelWiseRequest {
            context_id: ContextId::from([2u8; 32]),
            level: 1,
            parent_ids: Some(parents.clone()),
        };

        let encoded = borsh::to_vec(&request).expect("serialize");
        let decoded: InitPayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            InitPayload::LevelWiseRequest {
                context_id,
                level,
                parent_ids,
            } => {
                assert_eq!(*context_id.as_ref(), [2u8; 32]);
                assert_eq!(level, 1);
                assert_eq!(parent_ids, Some(parents));
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_levelwise_response_internal_nodes() {
        use crate::sync::levelwise::LevelNode;

        let nodes = vec![
            LevelNode::internal([1u8; 32], [10u8; 32], None),
            LevelNode::internal([2u8; 32], [20u8; 32], None),
        ];

        let response = MessagePayload::LevelWiseResponse {
            level: 0,
            nodes: nodes.clone(),
            has_more_levels: true,
            deleted_children: vec![],
        };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::LevelWiseResponse {
                level,
                nodes: decoded_nodes,
                has_more_levels,
                deleted_children,
            } => {
                assert_eq!(level, 0);
                assert_eq!(decoded_nodes.len(), 2);
                assert!(has_more_levels);
                assert!(decoded_nodes[0].is_internal());
                assert!(decoded_nodes[1].is_internal());
                assert!(deleted_children.is_empty());
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_levelwise_response_with_leaves() {
        use crate::sync::hash_comparison::{CrdtType, LeafMetadata, TreeLeafData};
        use crate::sync::levelwise::LevelNode;

        let metadata = LeafMetadata::new(CrdtType::lww_register("test"), 100, [0u8; 32]);
        let leaf_data = TreeLeafData::new([5u8; 32], vec![1, 2, 3, 4], metadata);

        let nodes = vec![
            LevelNode::internal([1u8; 32], [10u8; 32], None),
            LevelNode::leaf([2u8; 32], [20u8; 32], Some([1u8; 32]), leaf_data),
        ];

        let response = MessagePayload::LevelWiseResponse {
            level: 1,
            nodes,
            has_more_levels: false,
            deleted_children: vec![],
        };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::LevelWiseResponse {
                level,
                nodes: decoded_nodes,
                has_more_levels,
                deleted_children: _,
            } => {
                assert_eq!(level, 1);
                assert_eq!(decoded_nodes.len(), 2);
                assert!(!has_more_levels);
                assert!(decoded_nodes[0].is_internal());
                assert!(decoded_nodes[1].is_leaf());
                assert_eq!(decoded_nodes[1].parent_id, Some([1u8; 32]));
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_levelwise_response_empty() {
        let response = MessagePayload::LevelWiseResponse {
            level: 2,
            nodes: vec![],
            has_more_levels: false,
            deleted_children: vec![],
        };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::LevelWiseResponse {
                level,
                nodes,
                has_more_levels,
                deleted_children,
            } => {
                assert_eq!(level, 2);
                assert!(nodes.is_empty());
                assert!(!has_more_levels);
                assert!(deleted_children.is_empty());
            }
            _ => panic!("wrong variant"),
        }
    }

    // =========================================================================
    // EntityPush / EntityPushAck Wire Protocol Tests
    // =========================================================================

    #[test]
    fn test_init_payload_entity_push_roundtrip() {
        use crate::sync::hash_comparison::{CrdtType, LeafMetadata, TreeLeafData};

        let metadata = LeafMetadata::new(CrdtType::lww_register("test"), 100, [0u8; 32]);
        let leaf = TreeLeafData::new([10u8; 32], vec![1, 2, 3, 4], metadata);

        let request = InitPayload::EntityPush {
            context_id: ContextId::from([5u8; 32]),
            entities: vec![leaf],
        };

        let encoded = borsh::to_vec(&request).expect("serialize");
        let decoded: InitPayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            InitPayload::EntityPush {
                context_id,
                entities,
            } => {
                assert_eq!(*context_id.as_ref(), [5u8; 32]);
                assert_eq!(entities.len(), 1);
                assert_eq!(entities[0].key, [10u8; 32]);
                assert_eq!(entities[0].value, vec![1, 2, 3, 4]);
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_init_payload_entity_push_empty() {
        let request = InitPayload::EntityPush {
            context_id: ContextId::from([1u8; 32]),
            entities: vec![],
        };

        let encoded = borsh::to_vec(&request).expect("serialize");
        let decoded: InitPayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            InitPayload::EntityPush { entities, .. } => {
                assert!(entities.is_empty());
            }
            _ => panic!("wrong variant"),
        }
    }

    #[test]
    fn test_message_payload_entity_push_ack_roundtrip() {
        let response = MessagePayload::EntityPushAck { applied_count: 42 };

        let encoded = borsh::to_vec(&response).expect("serialize");
        let decoded: MessagePayload = borsh::from_slice(&encoded).expect("deserialize");

        match decoded {
            MessagePayload::EntityPushAck { applied_count } => {
                assert_eq!(applied_count, 42);
            }
            _ => panic!("wrong variant"),
        }
    }
}