calimero-node 0.10.1-rc.42

use super::*;

use calimero_node_primitives::sync::{
    build_handshake_from_raw, estimate_entity_count, estimate_max_depth, SyncHandshake,
};
use calimero_primitives::hash::Hash;

use super::SyncManager;

/// Build a handshake using the estimation fallback path (no store available).
///
/// This mirrors the fallback in `SyncManager::build_local_handshake` when
/// `query_tree_stats` returns `None`.
fn build_estimated_handshake(root_hash: [u8; 32], dag_heads: Vec<[u8; 32]>) -> SyncHandshake {
    let entity_count = estimate_entity_count(root_hash, dag_heads.len());
    let max_depth = estimate_max_depth(entity_count);
    build_handshake_from_raw(root_hash, entity_count, max_depth, dag_heads)
}

// =========================================================================
// Tests for handshake estimation fallback
// =========================================================================

/// Fresh node (zero root_hash) should have has_state=false and entity_count=0
#[test]
fn test_build_local_handshake_fresh_node() {
    let handshake = build_estimated_handshake([0; 32], vec![]);

    assert!(
        !handshake.has_state,
        "Fresh node should have has_state=false"
    );
    assert_eq!(
        handshake.entity_count, 0,
        "Fresh node should have entity_count=0"
    );
    assert_eq!(handshake.max_depth, 0, "Fresh node should have max_depth=0");
    assert_eq!(handshake.root_hash, [0; 32]);
}

/// Initialized node should have has_state=true and entity_count >= 1
#[test]
fn test_build_local_handshake_initialized_node() {
    let handshake = build_estimated_handshake([42; 32], vec![[1; 32], [2; 32]]);

    assert!(
        handshake.has_state,
        "Initialized node should have has_state=true"
    );
    assert_eq!(
        handshake.entity_count, 2,
        "Entity count should match dag_heads length in fallback"
    );
    assert!(
        handshake.max_depth >= 1,
        "Initialized node should have max_depth >= 1"
    );
    assert_eq!(handshake.root_hash, [42; 32]);
    assert_eq!(handshake.dag_heads.len(), 2);
}

/// Initialized node with empty dag_heads should still have entity_count >= 1
#[test]
fn test_build_local_handshake_initialized_no_heads() {
    let handshake = build_estimated_handshake([42; 32], vec![]);

    assert!(handshake.has_state);
    assert_eq!(
        handshake.entity_count, 1,
        "Initialized node with no heads should have entity_count=1 (minimum)"
    );
}

// =========================================================================
// Tests for build_remote_handshake()
// =========================================================================

/// Test building remote handshake from peer state
#[test]
fn test_build_remote_handshake_with_state() {
    let peer_root_hash = Hash::from([99; 32]);
    let peer_dag_heads: Vec<[u8; 32]> = vec![[10; 32], [20; 32], [30; 32]];

    let handshake = SyncManager::build_remote_handshake(peer_root_hash, &peer_dag_heads);

    assert!(handshake.has_state);
    assert_eq!(handshake.root_hash, [99; 32]);
    assert_eq!(handshake.entity_count, 3);
    assert_eq!(handshake.dag_heads.len(), 3);
}

/// Test building remote handshake from fresh peer
#[test]
fn test_build_remote_handshake_fresh_peer() {
    let peer_root_hash = Hash::from([0; 32]);
    let peer_dag_heads: Vec<[u8; 32]> = vec![];

    let handshake = SyncManager::build_remote_handshake(peer_root_hash, &peer_dag_heads);

    assert!(!handshake.has_state);
    assert_eq!(handshake.root_hash, [0; 32]);
    assert_eq!(handshake.entity_count, 0);
    assert_eq!(handshake.max_depth, 0);
}

// =========================================================================
// Tests for protocol selection integration
// =========================================================================

/// Test that select_protocol is called correctly with built handshakes
#[test]
fn test_protocol_selection_fresh_to_initialized() {
    use calimero_node_primitives::sync::{select_protocol, SyncProtocol};

    // Fresh local node
    let local_hs = SyncHandshake::new([0; 32], 0, 0, vec![]);

    // Initialized remote node
    let remote_hs = SyncHandshake::new([42; 32], 100, 4, vec![[1; 32]]);

    let selection = select_protocol(&local_hs, &remote_hs);

    assert!(
        matches!(selection.protocol, SyncProtocol::Snapshot { .. }),
        "Fresh node syncing from initialized should use Snapshot, got {:?}",
        selection.protocol
    );
    assert!(
        selection.reason.contains("fresh node"),
        "Reason should mention fresh node"
    );
}

/// Test that same root hash results in None protocol
#[test]
fn test_protocol_selection_already_synced() {
    use calimero_node_primitives::sync::{select_protocol, SyncProtocol};

    let local_hs = SyncHandshake::new([42; 32], 50, 3, vec![[1; 32]]);
    let remote_hs = SyncHandshake::new([42; 32], 100, 4, vec![[2; 32]]);

    let selection = select_protocol(&local_hs, &remote_hs);

    assert!(
        matches!(selection.protocol, SyncProtocol::None),
        "Same root hash should result in None, got {:?}",
        selection.protocol
    );
}

/// Test max_depth calculation for various entity counts
#[test]
fn test_max_depth_calculation() {
    // Test the log16 approximation: log16(n) ≈ log2(n) / 4
    let test_cases: Vec<(u64, u32)> = vec![
        (0, 0),   // No entities
        (1, 1),   // Single entity -> depth 1
        (16, 1),  // 16 entities -> log2(16)/4 = 4/4 = 1
        (256, 2), // 256 entities -> log2(256)/4 = 8/4 = 2
    ];

    for (entity_count, expected_min_depth) in test_cases {
        let max_depth = if entity_count == 0 {
            0
        } else {
            let log2_approx = 64u32.saturating_sub(entity_count.leading_zeros());
            (log2_approx / 4).max(1).min(32)
        };

        assert!(
            max_depth >= expected_min_depth,
            "entity_count={} should have max_depth >= {}, got {}",
            entity_count,
            expected_min_depth,
            max_depth
        );
    }
}

// =========================================================================
// Tests for the #2319 dispatch-attempt backoff helper
// =========================================================================

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

    fn ctx(byte: u8) -> ContextId {
        ContextId::from([byte; 32])
    }

    #[test]
    fn no_entry_means_not_recently_attempted() {
        let map: HashMap<ContextId, time::Instant> = HashMap::new();
        assert!(!dispatch_recently_attempted(
            &map,
            &ctx(1),
            time::Duration::from_secs(5)
        ));
    }

    #[test]
    fn fresh_attempt_within_interval_is_recent() {
        let mut map = HashMap::new();
        let _ = map.insert(ctx(2), time::Instant::now());
        assert!(dispatch_recently_attempted(
            &map,
            &ctx(2),
            time::Duration::from_secs(5)
        ));
    }

    #[test]
    fn old_attempt_beyond_interval_is_not_recent() {
        let mut map = HashMap::new();
        let _ = map.insert(ctx(3), time::Instant::now() - time::Duration::from_secs(10));
        assert!(!dispatch_recently_attempted(
            &map,
            &ctx(3),
            time::Duration::from_secs(5)
        ));
    }

    #[test]
    fn other_contexts_are_unaffected() {
        let mut map = HashMap::new();
        let _ = map.insert(ctx(4), time::Instant::now());
        assert!(!dispatch_recently_attempted(
            &map,
            &ctx(5),
            time::Duration::from_secs(5)
        ));
    }
}

// =========================================================================
// Tests for the #2319 wedged-session watchdog helper
// =========================================================================

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

    fn ctx(byte: u8) -> ContextId {
        ContextId::from([byte; 32])
    }

    /// `SyncState` with `last_sync == None` (sync dispatched, no result yet).
    fn in_progress_state() -> SyncState {
        let mut s = SyncState::new();
        s.start();
        s
    }

    /// `SyncState` with `last_sync == Some(_)` (a result has cleared it).
    fn settled_state() -> SyncState {
        let mut s = SyncState::new();
        s.on_failure("prior failure".to_owned());
        s
    }

    const GRACE: time::Duration = time::Duration::from_secs(60);

    #[test]
    fn fresh_dispatch_in_progress_is_not_wedged() {
        let mut dispatched = HashMap::new();
        let _ = dispatched.insert(ctx(1), time::Instant::now());
        let mut state = HashMap::new();
        let _ = state.insert(ctx(1), in_progress_state());
        assert!(!session_dispatch_wedged(
            &dispatched,
            &state,
            &ctx(1),
            GRACE
        ));
    }

    #[test]
    fn stale_dispatch_still_in_progress_is_wedged() {
        let mut dispatched = HashMap::new();
        let _ = dispatched.insert(
            ctx(2),
            time::Instant::now() - time::Duration::from_secs(120),
        );
        let mut state = HashMap::new();
        let _ = state.insert(ctx(2), in_progress_state());
        assert!(session_dispatch_wedged(&dispatched, &state, &ctx(2), GRACE));
    }

    #[test]
    fn stale_dispatch_but_settled_is_not_wedged() {
        let mut dispatched = HashMap::new();
        let _ = dispatched.insert(
            ctx(3),
            time::Instant::now() - time::Duration::from_secs(120),
        );
        let mut state = HashMap::new();
        let _ = state.insert(ctx(3), settled_state());
        assert!(!session_dispatch_wedged(
            &dispatched,
            &state,
            &ctx(3),
            GRACE
        ));
    }

    #[test]
    fn no_dispatch_record_is_not_wedged() {
        let dispatched: HashMap<ContextId, time::Instant> = HashMap::new();
        let mut state = HashMap::new();
        let _ = state.insert(ctx(4), in_progress_state());
        assert!(!session_dispatch_wedged(
            &dispatched,
            &state,
            &ctx(4),
            GRACE
        ));
    }

    #[test]
    fn other_contexts_are_unaffected() {
        let mut dispatched = HashMap::new();
        let _ = dispatched.insert(
            ctx(5),
            time::Instant::now() - time::Duration::from_secs(120),
        );
        let mut state = HashMap::new();
        let _ = state.insert(ctx(5), in_progress_state());
        assert!(!session_dispatch_wedged(
            &dispatched,
            &state,
            &ctx(6),
            GRACE
        ));
    }
}

// =========================================================================
// `reconcile_cooldown` / `record_reconcile_*` — backoff for the
// reconcile-after-divergence path
// =========================================================================
//
// Contract:
// - `reconcile_cooldown(n)` doubles from a 30 s base, caps at 30 min.
// - A failure record bumps the counter and refreshes the timestamp.
// - A success record clears the entry entirely (no inherited cooldown).
// - `reconcile_remaining_cooldown` returns `None` outside the window.

use std::time::Duration;

use calimero_primitives::context::ContextId;
use dashmap::DashMap;

use super::{
    reconcile_cooldown, reconcile_remaining_cooldown, record_reconcile_failure,
    record_reconcile_success,
};
use crate::state::ReconcileAttempt;

fn dummy_context(n: u8) -> ContextId {
    ContextId::from([n; 32])
}

#[test]
fn reconcile_cooldown_schedule_doubles_then_caps() {
    assert_eq!(reconcile_cooldown(1), Duration::from_secs(30));
    assert_eq!(reconcile_cooldown(2), Duration::from_secs(60));
    assert_eq!(reconcile_cooldown(3), Duration::from_secs(120));
    assert_eq!(reconcile_cooldown(4), Duration::from_secs(240));
    assert_eq!(reconcile_cooldown(5), Duration::from_secs(480));
    assert_eq!(reconcile_cooldown(6), Duration::from_secs(960));
    assert_eq!(reconcile_cooldown(7), Duration::from_secs(30 * 60));
    // Cap holds for arbitrarily large counters.
    assert_eq!(reconcile_cooldown(50), Duration::from_secs(30 * 60));
    assert_eq!(reconcile_cooldown(u32::MAX), Duration::from_secs(30 * 60));
}

#[test]
fn reconcile_cooldown_zero_failures_treated_as_one() {
    // The function is only meant to be called when at least one
    // failure has been recorded; we still want a defined value at 0
    // rather than a panic or underflow.
    assert_eq!(reconcile_cooldown(0), Duration::from_secs(30));
}

#[test]
fn record_reconcile_failure_increments_counter_and_stamps_time() {
    let attempts: DashMap<ContextId, ReconcileAttempt> = DashMap::new();
    let ctx = dummy_context(1);

    assert_eq!(record_reconcile_failure(&attempts, ctx), 1);
    assert_eq!(record_reconcile_failure(&attempts, ctx), 2);
    assert_eq!(record_reconcile_failure(&attempts, ctx), 3);

    let entry = attempts.get(&ctx).expect("entry was inserted");
    assert_eq!(entry.consecutive_failures, 3);
    // Stamp should be very recent (within the last few seconds).
    assert!(entry.last_attempt_at.elapsed() < Duration::from_secs(5));
}

#[test]
fn record_reconcile_success_clears_entry() {
    let attempts: DashMap<ContextId, ReconcileAttempt> = DashMap::new();
    let ctx = dummy_context(1);

    let _ = record_reconcile_failure(&attempts, ctx);
    let _ = record_reconcile_failure(&attempts, ctx);
    assert!(attempts.contains_key(&ctx));

    record_reconcile_success(&attempts, &ctx);
    assert!(
        !attempts.contains_key(&ctx),
        "success should clear all backoff state for the context"
    );
}

#[test]
fn reconcile_remaining_cooldown_none_when_no_entry() {
    let attempts: DashMap<ContextId, ReconcileAttempt> = DashMap::new();
    let ctx = dummy_context(1);
    assert!(reconcile_remaining_cooldown(&attempts, &ctx).is_none());
}

#[test]
fn reconcile_remaining_cooldown_some_after_recent_failure() {
    let attempts: DashMap<ContextId, ReconcileAttempt> = DashMap::new();
    let ctx = dummy_context(1);
    let _ = record_reconcile_failure(&attempts, ctx);

    let (remaining, failures) =
        reconcile_remaining_cooldown(&attempts, &ctx).expect("within cooldown");
    assert_eq!(failures, 1);
    // The first cooldown is 30 s; the test runs in <1 s.
    assert!(remaining > Duration::from_secs(25));
    assert!(remaining <= Duration::from_secs(30));
}

#[test]
fn reconcile_remaining_cooldown_none_after_cooldown_lapsed() {
    let attempts: DashMap<ContextId, ReconcileAttempt> = DashMap::new();
    let ctx = dummy_context(1);
    // Synthesize an entry whose timestamp is far enough in the past
    // that even the maximum cooldown has lapsed.
    let _replaced = attempts.insert(
        ctx,
        ReconcileAttempt {
            last_attempt_at: std::time::Instant::now() - Duration::from_secs(60 * 60),
            consecutive_failures: 7,
        },
    );
    assert!(reconcile_remaining_cooldown(&attempts, &ctx).is_none());
}