//! End-to-end scheduler tests: spawning, joining, panic delivery,
//! yield_now, self_pid.

use smarm::{channel, run, self_pid, spawn, spawn_under, yield_now, Signal};
use std::cell::Cell;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;

// ---------------------------------------------------------------------------
// Single root actor runs to completion
// ---------------------------------------------------------------------------

#[test]
fn root_actor_runs() {
    let captured = Arc::new(AtomicI64::new(0));
    let c = captured.clone();
    run(move || { c.store(99, Ordering::SeqCst); });
    assert_eq!(captured.load(Ordering::SeqCst), 99);
}

// ---------------------------------------------------------------------------
// Spawn child, join it
// ---------------------------------------------------------------------------

#[test]
fn spawn_and_join_returns_exit() {
    let captured = Arc::new(AtomicI64::new(0));
    let c = captured.clone();
    run(move || {
        let h = spawn(move || { c.store(7, Ordering::SeqCst); });
        let res = h.join();
        assert!(res.is_ok(), "join returned {:?}", res);
    });
    assert_eq!(captured.load(Ordering::SeqCst), 7);
}

// ---------------------------------------------------------------------------
// yield_now lets a sibling run
// ---------------------------------------------------------------------------

#[test]
fn yield_now_interleaves_actors() {
    let log: Arc<std::sync::Mutex<Vec<u8>>> = Arc::new(std::sync::Mutex::new(Vec::new()));
    let l1 = log.clone();
    let l2 = log.clone();
    run(move || {
        let h1 = spawn(move || {
            l1.lock().unwrap().push(1);
            yield_now();
            l1.lock().unwrap().push(3);
        });
        let h2 = spawn(move || {
            l2.lock().unwrap().push(2);
            yield_now();
            l2.lock().unwrap().push(4);
        });
        h1.join().unwrap();
        h2.join().unwrap();
    });
    // Both actors get their first step before either second step. Exact order
    // is FIFO: 1, 2, then 3, 4.
    assert_eq!(*log.lock().unwrap(), vec![1, 2, 3, 4]);
}

// ---------------------------------------------------------------------------
// self_pid returns this actor's pid inside the actor
// ---------------------------------------------------------------------------

#[test]
fn self_pid_is_stable_within_an_actor() {
    let pid_cell: Arc<std::sync::Mutex<Option<smarm::Pid>>> =
        Arc::new(std::sync::Mutex::new(None));
    let p2 = pid_cell.clone();
    run(move || {
        let h = spawn(move || {
            let me = self_pid();
            yield_now();
            assert_eq!(me, self_pid(), "self_pid changed across yield");
            *p2.lock().unwrap() = Some(me);
        });
        h.join().unwrap();
    });
    assert!(pid_cell.lock().unwrap().is_some());
}

// ---------------------------------------------------------------------------
// Panic is captured; join returns Err; supervisor receives Signal::Panic
// ---------------------------------------------------------------------------

#[test]
fn panicking_child_returns_join_error() {
    let saw_err = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let s = saw_err.clone();
    run(move || {
        let h = spawn(|| panic!("kaboom"));
        if h.join().is_err() {
            s.store(true, Ordering::SeqCst);
        }
    });

    assert!(saw_err.load(Ordering::SeqCst));
}

#[test]
fn supervisor_receives_panic_signal() {
    let saw_panic_signal = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let s = saw_panic_signal.clone();

    run(move || {
        // Build a supervisor actor with its own mailbox.
        let (sig_tx, sig_rx) = channel::<Signal>();
        let sup_handle = spawn(move || {
            // Wait for exactly one signal.
            let sig = sig_rx.recv().unwrap();
            if let Signal::Panic(_, _) = sig {
                s.store(true, Ordering::SeqCst);
            }
        });
        // Tell the runtime: when I spawn the next child, route signals here.
        let sup_pid = sup_handle.pid();
        smarm::scheduler::register_supervisor_channel(sup_pid, sig_tx);

        let child = spawn_under(sup_pid, || panic!("oops"));
        let _ = child.join();
        sup_handle.join().unwrap();
    });

    assert!(saw_panic_signal.load(Ordering::SeqCst));
}

// ---------------------------------------------------------------------------
// Multiple children, all complete, parent gets back control
// ---------------------------------------------------------------------------

#[test]
fn many_children_all_complete() {
    let counter = Arc::new(AtomicI64::new(0));
    let c = counter.clone();
    run(move || {
        let mut handles = Vec::new();
        for _ in 0..10 {
            let cc = c.clone();
            handles.push(spawn(move || {
                cc.fetch_add(1, Ordering::SeqCst);
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
    });
    assert_eq!(counter.load(Ordering::SeqCst), 10);
}

// ---------------------------------------------------------------------------
// Repeated yield_now inside an actor with no other actors completes
// ---------------------------------------------------------------------------

#[test]
fn yield_alone_terminates() {
    thread_local! {
        static N: Cell<i32> = const { Cell::new(0) };
    }
    N.with(|c| c.set(0));
    run(|| {
        for _ in 0..5 {
            N.with(|c| c.set(c.get() + 1));
            yield_now();
        }
    });
    assert_eq!(N.with(|c| c.get()), 5);
}