mod common;
use common::*;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

// ── Load-Aware Placement Tests ─────────────────────────────────────────────

/// Given a multi-threaded runtime where one worker has many more actors,
/// when new actors are spawned after a few ticks (so stats propagate),
/// then they should be placed on the lighter worker.
#[test]
fn load_aware_placement_prefers_lighter_worker() {
    // 2 threads: intentionally imbalance by spawning many actors first
    let rt = std_runtime(RuntimeConfig {
        num_threads: 2,
        ..Default::default()
    });

    // Phase 1: Spawn 20 actors. With round-robin, they split ~10/10.
    let mut addrs = Vec::new();
    for _ in 0..20 {
        addrs.push(rt.spawn(CounterActor { count: 0 }).unwrap());
    }

    // Run so stats propagate, then bombard worker 0's actors with messages
    // to create mailbox depth imbalance.
    let handle = rt.run().unwrap();
    std::thread::sleep(std::time::Duration::from_millis(10));

    // Send 500 messages to the first 10 actors (likely on worker 0).
    for addr in &addrs[..10] {
        for _ in 0..50 {
            let _ = handle.runtime.send_to(*addr, Increment {
                reply_to: *addr, // self-reply to keep mailbox depth up
            });
        }
    }

    std::thread::sleep(std::time::Duration::from_millis(20));

    // Phase 2: Spawn 10 more actors. With load-aware placement,
    // they should bias toward the lighter worker.
    let mut late_addrs = Vec::new();
    for _ in 0..10 {
        late_addrs.push(handle.runtime.spawn(CounterActor { count: 0 }).unwrap());
    }

    std::thread::sleep(std::time::Duration::from_millis(20));

    let stats = handle.runtime.stats();
    handle.shutdown();
    handle.join();

    // Verify the system is operational — both workers should have actors
    let total_actors: usize = stats.workers.iter().map(|w| w.num_actors).sum();
    assert!(total_actors >= 20, "expected at least 20 actors, got {}", total_actors);

    // The lighter worker should have gotten more of the late actors.
    assert!(
        stats.workers.iter().all(|w| w.num_actors > 0),
        "both workers should have actors, got {:?}",
        stats.workers.iter().map(|w| w.num_actors).collect::<Vec<_>>()
    );
}

/// Given a single-threaded runtime (1 worker),
/// when many actors are spawned,
/// then all go to worker 0 regardless of load (no panic, no error).
#[test]
fn load_aware_placement_single_worker_degrades_gracefully() {
    let rt = std_runtime(RuntimeConfig::default());

    for _ in 0..50 {
        rt.spawn(CounterActor { count: 0 }).unwrap();
    }

    // Tick several times to let stats update
    for _ in 0..10 {
        rt.tick();
    }

    let stats = rt.stats();
    assert_eq!(stats.workers.len(), 1);
    assert_eq!(stats.workers[0].num_actors, 50);
}

/// Given a fresh runtime with no prior ticks,
/// when actors are spawned in a burst,
/// then they distribute evenly (round-robin fallback when stats are all zero).
#[test]
fn load_aware_placement_falls_back_to_round_robin_on_fresh_runtime() {
    let rt = std_runtime(RuntimeConfig {
        num_threads: 4,
        ..Default::default()
    });

    // Spawn 100 actors before any ticks (all stats are zero)
    for _ in 0..100 {
        rt.spawn(CounterActor { count: 0 }).unwrap();
    }

    let handle = rt.run().unwrap();
    std::thread::sleep(std::time::Duration::from_millis(20));

    let stats = handle.runtime.stats();
    handle.shutdown();
    handle.join();

    // With 4 workers and 100 actors, each should have ~25 (+-5).
    for w in &stats.workers {
        assert!(
            w.num_actors >= 20 && w.num_actors <= 30,
            "worker {} has {} actors, expected ~25 (round-robin)",
            w.id, w.num_actors
        );
    }
}

// ── Mailbox Backpressure Tests ─────────────────────────────────────────────

/// Given a runtime with bounded mailboxes (capacity=10, DropNewest),
/// when 50 messages are sent to an actor before any ticks,
/// then only the first 10 are delivered and the rest are dropped.
#[test]
fn bounded_mailbox_drop_newest_caps_at_capacity() {
    let rt = std_runtime(RuntimeConfig {
        default_mailbox_capacity: 10,
        mailbox_overflow: MailboxOverflow::DropNewest,
        ..Default::default()
    });

    let inbox = rt.new_inbox::<Count>().unwrap();
    let addr = rt.spawn(CounterActor { count: 0 }).unwrap();

    // Send 50 messages — only first 10 should be queued
    for _ in 0..50 {
        let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() });
    }

    // Tick enough times to process all queued messages
    for _ in 0..20 {
        rt.tick();
    }

    // Count replies — should be exactly 10 (the mailbox capacity)
    let mut replies = 0;
    while inbox.try_recv().is_some() {
        replies += 1;
    }
    assert_eq!(replies, 10, "should deliver exactly mailbox_capacity messages");

    // Stats should show drops
    let stats = rt.stats();
    let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum();
    assert_eq!(total_drops, 40, "40 messages should have been dropped");
}

/// Given a runtime with bounded mailboxes (capacity=5, DropOldest),
/// when 10 messages are sent before any tick,
/// then only the 5 most recent messages are delivered.
#[test]
fn bounded_mailbox_drop_oldest_keeps_newest() {
    let rt = std_runtime(RuntimeConfig {
        default_mailbox_capacity: 5,
        mailbox_overflow: MailboxOverflow::DropOldest,
        ..Default::default()
    });

    let inbox = rt.new_inbox::<Done>().unwrap();
    let addr = rt.spawn(DoubleActor).unwrap();

    // Send messages with values 0..10. DoubleActor replies Done(value * 2).
    for i in 0..10 {
        let _ = rt.send_to(addr, Forward {
            value: i,
            reply_to: *inbox.addr(),
        });
    }

    for _ in 0..10 {
        rt.tick();
    }

    // Collect all replies
    let mut replies = Vec::new();
    while let Some(Done(v)) = inbox.try_recv() {
        replies.push(v);
    }

    assert_eq!(replies.len(), 5, "should deliver exactly 5 messages");
    // The 5 most recent: values 5,6,7,8,9 -> doubled: 10,12,14,16,18
    assert_eq!(replies, vec![10, 12, 14, 16, 18], "should keep the newest messages");
}

/// Given a runtime with unbounded mailboxes (capacity=0, the default),
/// when many messages are sent,
/// then all are delivered (backward compatibility).
#[test]
fn unbounded_mailbox_delivers_all_messages() {
    let rt = std_runtime(RuntimeConfig::default());

    let inbox = rt.new_inbox::<Count>().unwrap();
    let addr = rt.spawn(CounterActor { count: 0 }).unwrap();

    for _ in 0..200 {
        let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() });
    }

    for _ in 0..50 {
        rt.tick();
    }

    let mut replies = 0;
    while inbox.try_recv().is_some() {
        replies += 1;
    }
    assert_eq!(replies, 200, "all 200 messages should be delivered with unbounded mailbox");

    let stats = rt.stats();
    let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum();
    assert_eq!(total_drops, 0, "no drops with unbounded mailbox");
}

/// Given bounded mailboxes with budget, when an actor processes messages
/// and frees mailbox space, then new messages should be accepted on subsequent ticks.
#[test]
fn bounded_mailbox_refills_after_processing() {
    let rt = std_runtime(RuntimeConfig {
        default_mailbox_capacity: 5,
        actor_message_budget: 5,
        mailbox_overflow: MailboxOverflow::DropNewest,
        ..Default::default()
    });

    let inbox = rt.new_inbox::<Count>().unwrap();
    let addr = rt.spawn(CounterActor { count: 0 }).unwrap();

    // Send first batch of 5 — fills mailbox exactly
    for _ in 0..5 {
        let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() });
    }

    // Tick to process all 5 (budget=5, capacity=5)
    rt.tick();

    // Send second batch of 5 — mailbox is empty, so all 5 should be accepted
    for _ in 0..5 {
        let _ = rt.send_to(addr, Increment { reply_to: *inbox.addr() });
    }

    rt.tick();

    let mut replies = 0;
    while inbox.try_recv().is_some() {
        replies += 1;
    }
    assert_eq!(replies, 10, "all 10 messages across 2 batches should be processed");

    let stats = rt.stats();
    let total_drops: u64 = stats.workers.iter().map(|w| w.messages_dropped).sum();
    assert_eq!(total_drops, 0, "no drops when mailbox drains between batches");
}

// ── Dead Actor Cleanup Tests ───────────────────────────────────────────────

/// Given an actor that panics and is poisoned,
/// when ticks continue,
/// then the actor is removed from stats and sends to its address fail.
#[test]
fn dead_actor_cleaned_up_from_stats_and_address_map() {
    let rt = std_runtime(RuntimeConfig::default());

    let good = rt.spawn(PingPongActor).unwrap();
    let bad = rt.spawn(PanicActor).unwrap();

    // Trigger panic
    let _ = rt.send_to(bad, PanicMsg);
    for _ in 0..5 { rt.tick(); }

    let stats = rt.stats();
    // Good actor still present, bad actor cleaned up
    assert_eq!(stats.workers[0].num_actors, 1, "only the healthy actor should remain");
    assert!(
        stats.actors.iter().any(|(a, _)| *a == good),
        "good actor should be in address map"
    );
    assert!(
        !stats.actors.iter().any(|(a, _)| *a == bad),
        "poisoned actor should be removed from address map"
    );

    // Sends to cleaned-up actor fail
    let result = rt.send_to(bad, PanicMsg);
    assert!(result.is_err(), "send to cleaned-up actor should fail");
}

/// Given many actors that all panic,
/// when ticks proceed,
/// then all are cleaned up and stats reflect zero actors.
#[test]
fn bulk_dead_actor_cleanup() {
    let rt = std_runtime(RuntimeConfig::default());

    let mut addrs = Vec::new();
    for _ in 0..20 {
        addrs.push(rt.spawn(PanicActor).unwrap());
    }

    // Trigger all panics
    for &addr in &addrs {
        let _ = rt.send_to(addr, PanicMsg);
    }
    for _ in 0..10 { rt.tick(); }

    let stats = rt.stats();
    assert_eq!(stats.workers[0].num_actors, 0, "all poisoned actors should be cleaned up");
    assert_eq!(
        stats.actors.len(), 0,
        "address map should be empty after all actors poisoned"
    );
}

// ── Actor Recovery Helpers ──────────────────────────────────────────────────

/// Handles Forward messages, replies Done(value * 2), panics on the panic_at-th message.
struct RestartTestActor {
    count: usize,
    panic_at: usize,
}

impl ActorInterface for RestartTestActor {
    type Incoming = Forward;
    type Response = Done;
    fn handle(&mut self, ctx: &Ctx, msg: Forward) {
        self.count += 1;
        if self.count >= self.panic_at {
            panic!("intentional panic at message {}", self.count);
        }
        let _ = ctx.send(msg.reply_to, Done(msg.value * 2));
    }
}

// ── Actor Recovery Tests ───────────────────────────────────────────────────

/// Given an actor that panics,
/// when it panics,
/// then it is poisoned and future messages are discarded.
#[test]
fn non_restartable_actor_still_poisons_on_panic() {
    let rt = std_runtime(RuntimeConfig::default());
    let inbox = rt.new_inbox::<Done>().unwrap();

    // Normal spawn — not restartable
    let addr = rt.spawn(RestartTestActor { count: 0, panic_at: 1 }).unwrap();

    let _ = rt.send_to(addr, Forward { value: 42, reply_to: *inbox.addr() });
    for _ in 0..5 { rt.tick(); }

    let stats = rt.stats();
    let total_panics: u64 = stats.workers.iter().map(|w| w.panics).sum();
    let total_restarts: u64 = stats.workers.iter().map(|w| w.restarts).sum();
    assert_eq!(total_panics, 1, "should panic");
    assert_eq!(total_restarts, 0, "should not restart (not restartable)");
}