nodedb-lite 0.1.0

//! Handshake: build outgoing `HandshakeMsg`, process incoming `HandshakeAckMsg`.

use nodedb_types::sync::wire::{HandshakeAckMsg, HandshakeMsg};
use nodedb_types::wire_version::WIRE_FORMAT_VERSION;

use super::config::SyncState;
use super::state::SyncClient;

impl SyncClient {
    /// Build a handshake message from current state and the instance identity.
    ///
    /// The identity is a parameter rather than client state because it changes
    /// while the client is alive: a reported fork replaces it wholesale, and a
    /// peer-id collision rotates part of it. A handshake built from a cached
    /// copy would re-present the identity Origin just rejected.
    pub async fn build_handshake(&self, identity: &crate::identity::LiteIdentity) -> HandshakeMsg {
        let clock = self.clock.lock().await;
        let shapes = self.shapes.lock().await;

        let wire_clock = clock.to_wire();
        let mut vector_clock = std::collections::HashMap::new();
        // Origin expects: { collection: { doc_id: lamport_ts } }
        // We send a simplified version: { "_global": { peer_hex: counter } }
        vector_clock.insert("_global".to_string(), wire_clock);

        HandshakeMsg {
            jwt_token: self.config.jwt_token.clone(),
            vector_clock,
            subscribed_shapes: shapes.active_shape_ids(),
            client_version: self.config.client_version.clone(),
            lite_id: identity.lite_id.clone(),
            epoch: identity.epoch,
            wire_version: WIRE_FORMAT_VERSION,
        }
    }

    /// Process a handshake acknowledgment from Origin.
    pub async fn handle_handshake_ack(&self, ack: &HandshakeAckMsg) -> bool {
        if !ack.success {
            tracing::warn!(
                error = ack.error.as_deref().unwrap_or("unknown"),
                "handshake rejected by Origin"
            );
            return false;
        }

        *self.session_id.lock().await = Some(ack.session_id.clone());
        // New session — every collection must be re-announced before its
        // first delta so Origin materializes it, even if it was already
        // announced in a prior session.
        self.announced_collections.lock().await.clear();

        let mut clock = self.clock.lock().await;
        for (peer_hex, &counter) in &ack.server_clock {
            if let Ok(peer_id) = u64::from_str_radix(peer_hex, 16) {
                clock.advance(peer_id, counter);
            }
        }
        drop(clock);

        self.set_producer_id(ack.producer_id).await;
        self.set_accepted_epoch(ack.accepted_epoch).await;

        *self.state.lock().await = SyncState::Connected;
        tracing::info!(
            session = %ack.session_id,
            producer_id = ack.producer_id,
            accepted_epoch = ack.accepted_epoch,
            "sync handshake accepted"
        );
        true
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sync::client::SyncConfig;

    fn make_config() -> SyncConfig {
        SyncConfig::new("wss://localhost:9090/sync", "test.jwt.token")
    }

    #[tokio::test]
    async fn build_handshake() {
        let client = SyncClient::new(make_config());

        {
            let mut shapes = client.shapes().lock().await;
            shapes.subscribe(nodedb_types::sync::shape::ShapeDefinition {
                shape_id: "s1".into(),
                tenant_id: 1,
                shape_type: nodedb_types::sync::shape::ShapeType::Document {
                    collection: "orders".into(),
                    predicate: Vec::new(),
                },
                description: "test".into(),
                field_filter: vec![],
            });
        }

        let identity = crate::identity::LiteIdentity {
            lite_id: "lite-1".into(),
            epoch: 3,
            peer_id: 42,
        };
        let hs = client.build_handshake(&identity).await;
        assert_eq!(hs.jwt_token, "test.jwt.token");
        assert_eq!(hs.subscribed_shapes, vec!["s1"]);
    }

    #[tokio::test]
    async fn handle_handshake_ack_success() {
        let client = SyncClient::new(make_config());
        let ack = HandshakeAckMsg {
            success: true,
            session_id: "sess-123".into(),
            server_clock: std::collections::HashMap::new(),
            error: None,
            fork_detected: false,
            server_wire_version: 1,
            producer_id: 0,
            accepted_epoch: 0,
            // All zeros = signing unavailable for this session.
            delta_signing_key: [0u8; 32],
        };

        assert!(client.handle_handshake_ack(&ack).await);
        assert_eq!(client.state().await, SyncState::Connected);
    }

    #[tokio::test]
    async fn handle_handshake_ack_failure() {
        let client = SyncClient::new(make_config());
        let ack = HandshakeAckMsg {
            success: false,
            session_id: String::new(),
            server_clock: std::collections::HashMap::new(),
            error: Some("invalid token".into()),
            fork_detected: false,
            server_wire_version: 1,
            producer_id: 0,
            accepted_epoch: 0,
            // All zeros = signing unavailable for this session.
            delta_signing_key: [0u8; 32],
        };

        assert!(!client.handle_handshake_ack(&ack).await);
        assert_eq!(client.state().await, SyncState::Disconnected);
    }
}