//! Lifecycle simulation tests — death/repair/cache/routing behavior.
//!
//! Tests that node death correctly triggers:
//! - Repair queue population for re-replication
//! - Cache invalidation of stale entries
//! - Routing table cleanup
//! - Recovery after partition heals

use simulation::distribution::properties::{
    check_repair_queue_populated, check_routing_table_bounded,
};
use simulation::distribution::sim::{
    run_simulation_with_nodes, DistributionSimConfig, NetworkFault, Partition, SimAction,
};

fn default_config() -> DistributionSimConfig {
    DistributionSimConfig::default()
}

// ────────────────────────────────────────────────────────────────────────────
// 1. Dead node's actors populate repair queue and invalidate cache
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn dead_node_triggers_repair_queue_and_cache_invalidation() {
    // Given: 5-node cluster, 2 actors/node. Node 2 is killed at round 10.
    let config = DistributionSimConfig {
        name: "death-repair-cache".into(),
        num_nodes: 5,
        num_rounds: 80,
        ticks_per_round: 3,
        actors_per_node: 2,
        kill_schedule: vec![(10, 2)],
        ..default_config()
    };

    let (trace, nodes) = run_simulation_with_nodes(config);

    // Then: at least one survivor should have a non-empty repair queue
    let result = check_repair_queue_populated(&trace, 10);
    assert!(
        result.passed,
        "repair queue should be populated after node death: {}",
        result.actual
    );

    // And: no survivor's cache should contain entries pointing to the dead node
    let dead_node_id = {
        // Find the node_id for node 2 from round snapshots before death
        // We can check from the surviving nodes
        // Node 2 is dead (None), so we check survivors' caches
        let mut stale_count = 0;
        for node in nodes.iter().filter_map(|n| n.as_ref()) {
            for (_actor, cached_on) in node.cache().entries() {
                // The dead node's entries should have been invalidated
                // We can't easily get node 2's ID here, but we can check
                // that no survivor caches an actor on a node not in their members
                let alive_ids: Vec<_> = node.members().iter().map(|m| m.node_id).collect();
                if !alive_ids.contains(&cached_on) && cached_on != node.node_id() {
                    stale_count += 1;
                }
            }
        }
        stale_count
    };

    assert_eq!(
        dead_node_id, 0,
        "no survivor should have cache entries pointing to non-member nodes"
    );
}

// ────────────────────────────────────────────────────────────────────────────
// 2. Revived node starts fresh (empty directory)
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn revived_node_has_empty_directory() {
    // Given: 5-node cluster, 2 actors/node.
    // Node 2 killed at round 10, revived at round 50.
    let config = DistributionSimConfig {
        name: "revive-fresh".into(),
        num_nodes: 5,
        num_rounds: 100,
        ticks_per_round: 3,
        actors_per_node: 2,
        kill_schedule: vec![(10, 2)],
        revive_schedule: vec![(50, 2)],
        ..default_config()
    };

    let (_trace, nodes) = run_simulation_with_nodes(config);

    // Then: the revived node should have an empty directory
    // (it's a fresh DistributedNode, not carrying over old state)
    let revived = nodes[2].as_ref().expect("node 2 should be revived");
    assert_eq!(
        revived.directory().entry_count(), 0,
        "revived node should start with empty directory"
    );

    // And: the revived node should have rejoined the cluster
    assert!(
        !revived.members().is_empty(),
        "revived node should have some cluster members"
    );
}

// ────────────────────────────────────────────────────────────────────────────
// 3. Cache invalidation tracks membership changes
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn cache_shrinks_after_node_death() {
    // Given: 5-node cluster with actors, all caches populated during setup.
    // When: node 1 is killed
    // Then: cache_size should decrease for survivors after death detection.
    let config = DistributionSimConfig {
        name: "cache-invalidation".into(),
        num_nodes: 5,
        num_rounds: 80,
        ticks_per_round: 3,
        actors_per_node: 3,
        kill_schedule: vec![(10, 1)],
        ..default_config()
    };

    let (trace, _nodes) = run_simulation_with_nodes(config);

    // Check that cache_size decreased for at least some survivors after death
    // Before death (round 9), survivors should have cache entries
    // After death detection, cache for dead node's actors should be invalidated
    let pre_death_round = 8; // 0-indexed round 9
    let post_detection_round = 39; // well after SWIM detection

    if pre_death_round < trace.snapshots_per_round.len()
        && post_detection_round < trace.snapshots_per_round.len()
    {
        let pre_cache_max: usize = trace.snapshots_per_round[pre_death_round]
            .iter()
            .filter(|(_, s)| s.is_alive)
            .map(|(_, s)| s.cache_size)
            .max()
            .unwrap_or(0);

        let post_cache_sizes: Vec<usize> = trace.snapshots_per_round[post_detection_round]
            .iter()
            .filter(|(_, s)| s.is_alive)
            .map(|(_, s)| s.cache_size)
            .collect();

        // After death, some survivors should have fewer cache entries
        // (the dead node's actors were invalidated)
        let any_decreased = post_cache_sizes.iter().any(|&s| s < pre_cache_max);
        assert!(
            any_decreased || pre_cache_max == 0,
            "cache should shrink after node death, pre_max={pre_cache_max}, post={post_cache_sizes:?}"
        );
    }
}

// ────────────────────────────────────────────────────────────────────────────
// 4. Routing table recovers after partition heals (dead reprobe)
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn routing_table_recovers_after_partition_heals() {
    // Given: 6-node cluster, partition at round 10, heal at round 40
    // With dead_reprobe_interval=10, nodes re-discover dead members
    let config = DistributionSimConfig {
        name: "rt-recovery".into(),
        num_nodes: 6,
        num_rounds: 120,
        ticks_per_round: 3,
        actors_per_node: 0,
        swim: distribution::swim::probe::SwimConfig {
            probe_interval: 1,
            probe_timeout: 3,
            indirect_probes: 1,
            suspicion_timeout: 5,
            dead_reprobe_interval: 10,
        },
        network_faults: vec![
            NetworkFault::Partition {
                round: 10,
                partition: Partition {
                    side_a: vec![0, 1, 2],
                    side_b: vec![3, 4, 5],
                    asymmetric: false,
                },
            },
            NetworkFault::Heal { round: 40 },
        ],
        ..default_config()
    };

    let (_trace, nodes) = run_simulation_with_nodes(config);

    // Then: after healing, all nodes should have recovered routing tables
    // Each node should see at least 4 of 5 other nodes in their routing table
    for (i, maybe_node) in nodes.iter().enumerate() {
        if let Some(node) = maybe_node {
            assert!(
                node.routing_table().len() >= 4,
                "node {i} should have ≥4 RT entries after partition heals, got {}",
                node.routing_table().len()
            );
        }
    }
}

// ────────────────────────────────────────────────────────────────────────────
// 5. Routing table tracks alive membership (bounded invariant)
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn routing_table_bounded_by_alive_count() {
    // Run a simulation with deaths and verify the routing table invariant
    let config = DistributionSimConfig {
        name: "rt-bounded".into(),
        num_nodes: 8,
        num_rounds: 80,
        ticks_per_round: 3,
        actors_per_node: 1,
        kill_schedule: vec![(15, 2), (25, 5)],
        ..default_config()
    };

    let (trace, _) = run_simulation_with_nodes(config);

    let result = check_routing_table_bounded(&trace);
    assert!(
        result.passed,
        "routing table should never exceed alive count: {}",
        result.actual
    );
}

// ────────────────────────────────────────────────────────────────────────────
// 6. Combined: partition + death during partition + heal + verify
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn partition_then_death_during_partition_then_heal() {
    // Given: 6 nodes, partition at r=10, node 2 (side A) killed during partition
    // at r=20, heal at r=40. Node 2 had actors and a registry name.
    // Use high suspicion_timeout so cross-partition nodes stay Suspect (not Dead),
    // while within-partition death of node 2 is detected after timeout expires.
    let config = DistributionSimConfig {
        name: "partition-death-heal".into(),
        num_nodes: 6,
        num_rounds: 150,
        ticks_per_round: 3,
        actors_per_node: 2,
        network_faults: vec![
            NetworkFault::Partition {
                round: 10,
                partition: Partition {
                    side_a: vec![0, 1, 2],
                    side_b: vec![3, 4, 5],
                    asymmetric: false,
                },
            },
            NetworkFault::Heal { round: 40 },
        ],
        action_schedule: vec![
            (5, SimAction::RegisterName { node_idx: 2, name: "doomed-svc".into() }),
            (5, SimAction::RegisterName { node_idx: 4, name: "stable-svc".into() }),
        ],
        kill_schedule: vec![(20, 2)],
        swim: distribution::swim::probe::SwimConfig {
            probe_interval: 1,
            probe_timeout: 3,
            indirect_probes: 1,
            // >90 ticks (30 rounds × 3 ticks) so cross-partition nodes stay Suspect during
            // the 30-round partition. Node 2 (truly dead) gets declared dead ~33 rounds
            // after kill, well after partition heals.
            suspicion_timeout: 100,
            dead_reprobe_interval: 10,
        },
        ..default_config()
    };

    let (_trace, nodes) = run_simulation_with_nodes(config);

    let survivors: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect();
    assert_eq!(survivors.len(), 5, "5 of 6 should survive");

    // "doomed-svc" should be tombstoned (owner node 2 died)
    for node in &survivors {
        assert!(
            node.resolve_name("doomed-svc").is_none(),
            "doomed-svc should be tombstoned after owner died during partition"
        );
    }

    // "stable-svc" should still resolve (node 4 alive throughout)
    for node in &survivors {
        assert!(
            node.resolve_name("stable-svc").is_some(),
            "stable-svc should resolve (owner survived partition)"
        );
    }

    // After partition heal + dead reprobe, routing tables should recover
    // (at least 4 entries for each surviving node)
    for node in &survivors {
        assert!(
            node.routing_table().len() >= 3,
            "surviving node should have ≥3 RT entries after partition heals, got {}",
            node.routing_table().len()
        );
    }
}

// ────────────────────────────────────────────────────────────────────────────
// 7. Registry GC: tombstones are garbage-collected after TTL
// ────────────────────────────────────────────────────────────────────────────

#[test]
fn registry_tombstones_gc_after_ttl() {
    // Given: short tombstone TTL and GC interval, register then unregister a name.
    // Then do many more register/unregister operations to advance the logical clock
    // (which is used for TTL comparison). After enough clock advancement, the
    // original tombstone should be garbage-collected.
    let config = DistributionSimConfig {
        name: "registry-gc".into(),
        num_nodes: 5,
        num_rounds: 80,
        ticks_per_round: 3,
        actors_per_node: 0,
        // Very short GC: TTL=5 logical clock ticks, GC runs every 3 ticks
        registry_tombstone_ttl: Some(5),
        registry_gc_interval: Some(3),
        action_schedule: vec![
            (5, SimAction::RegisterName { node_idx: 0, name: "ephemeral".into() }),
            (10, SimAction::UnregisterName { node_idx: 0, name: "ephemeral".into() }),
            // Additional operations to advance the logical clock past the TTL
            (15, SimAction::RegisterName { node_idx: 1, name: "churn-1".into() }),
            (16, SimAction::RegisterName { node_idx: 2, name: "churn-2".into() }),
            (17, SimAction::RegisterName { node_idx: 3, name: "churn-3".into() }),
            (18, SimAction::RegisterName { node_idx: 4, name: "churn-4".into() }),
            (19, SimAction::RegisterName { node_idx: 1, name: "churn-5".into() }),
            (20, SimAction::RegisterName { node_idx: 2, name: "churn-6".into() }),
        ],
        ..default_config()
    };

    let (trace, nodes) = run_simulation_with_nodes(config);

    // Shortly after unregister (round 12), tombstones should exist
    let mid_tombstones: usize = trace.snapshots_per_round
        .get(11) // round 12
        .map(|snaps| {
            snaps.iter()
                .filter(|(_, s)| s.is_alive)
                .map(|(_, s)| s.registry_tombstone_count)
                .sum()
        })
        .unwrap_or(0);

    assert!(
        mid_tombstones > 0,
        "tombstones should exist shortly after unregister"
    );

    // After many more register operations advance the clock, the "ephemeral" tombstone
    // should be GC'd (its age exceeds TTL=5 in logical clock terms)
    let alive_nodes: Vec<_> = nodes.iter().filter_map(|n| n.as_ref()).collect();

    // Check that "ephemeral" resolves to None on all nodes (whether GC'd or still tombstoned)
    for node in &alive_nodes {
        assert!(
            node.resolve_name("ephemeral").is_none(),
            "ephemeral should not resolve (tombstoned or GC'd)"
        );
    }

    // At least some nodes should have GC'd the tombstone (clock advanced past TTL)
    let nodes_with_ephemeral_tombstone: usize = alive_nodes
        .iter()
        .filter(|n| {
            n.registry().entries().any(|e| e.name == "ephemeral" && e.tombstone)
        })
        .count();

    assert!(
        nodes_with_ephemeral_tombstone < alive_nodes.len(),
        "at least some nodes should have GC'd the 'ephemeral' tombstone, but {} of {} still have it",
        nodes_with_ephemeral_tombstone,
        alive_nodes.len()
    );
}