//! RFC 016 Chunk 1 — the read primitive. These exercise exactly what the RFC
//! promised: tests that assert an actor's state, parentage, names, mailbox
//! depth, and lifecycle counts directly off `snapshot()` / `actor_info()`
//! instead of sleeping-and-hoping.

use smarm::{
    actor_info, channel, monitor, register, run, self_pid, send, snapshot, spawn, ActorState, Name,
    Pid, SNAPSHOT_FORMAT_VERSION,
};

const SVC: Name<u64> = Name::new("svc");

/// Bounded poll on the introspection result itself (not a wall-clock sleep):
/// yield until `pred` holds for the given pid, panicking if it never does.
fn spin_until(pid: Pid, mut pred: impl FnMut(&smarm::ActorInfo) -> bool) -> smarm::ActorInfo {
    for _ in 0..100_000 {
        if let Some(info) = actor_info(pid) {
            if pred(&info) {
                return info;
            }
        }
        smarm::yield_now();
    }
    panic!("actor {pid:?} never reached the expected state");
}

#[test]
fn snapshot_lists_actors_with_parent_edge() {
    run(|| {
        let (ready_tx, ready_rx) = channel::<()>();
        let (gate_tx, gate_rx) = channel::<()>();
        let (_cmd_tx, cmd_rx) = channel::<u64>();

        let h = spawn(move || {
            register(Name::<u64>::new("w"), _cmd_tx).unwrap();
            ready_tx.send(()).unwrap();
            gate_rx.recv().unwrap(); // park here until released
            drop(cmd_rx);
        });
        ready_rx.recv().unwrap();

        let me = self_pid();
        let snap = snapshot();
        assert_eq!(snap.format_version, SNAPSHOT_FORMAT_VERSION);

        let worker = snap
            .actors
            .iter()
            .find(|a| a.pid == h.pid())
            .expect("worker present in snapshot");
        assert_eq!(worker.names, vec!["w"]);
        // `spawn` records the spawning actor as the parent (D9).
        assert_eq!(worker.supervisor, me);
        assert!(!worker.trap_exit);
        assert_eq!((worker.monitors, worker.links, worker.joiners), (0, 0, 0));

        // The root itself is on-CPU (it's running this code) and rooted under
        // the forest sentinel.
        let root = snap.actors.iter().find(|a| a.pid == me).expect("root present");
        assert_eq!(root.state, ActorState::Running);
        assert_eq!(root.supervisor, smarm::Pid::new(u32::MAX, u32::MAX));

        gate_tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn parked_state_is_observable() {
    run(|| {
        let (ready_tx, ready_rx) = channel::<()>();
        let (gate_tx, gate_rx) = channel::<()>();
        let h = spawn(move || {
            ready_tx.send(()).unwrap();
            gate_rx.recv().unwrap();
        });
        ready_rx.recv().unwrap();
        // The worker has nothing to do but block on the empty gate channel, so
        // it must reach Parked.
        let info = spin_until(h.pid(), |a| a.state == ActorState::Parked);
        assert_eq!(info.state, ActorState::Parked);
        gate_tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn mailbox_depth_counts_queued_messages() {
    run(|| {
        let (ready_tx, ready_rx) = channel::<()>();
        let (gate_tx, gate_rx) = channel::<()>();
        let (cmd_tx, _cmd_rx) = channel::<u64>();

        let h = spawn(move || {
            // Publish the command inbox, then block on an unrelated gate so the
            // queued commands are never drained while we observe them.
            register(SVC, cmd_tx).unwrap();
            ready_tx.send(()).unwrap();
            gate_rx.recv().unwrap();
            drop(_cmd_rx);
        });
        ready_rx.recv().unwrap();

        for i in 0..3 {
            send(SVC, i).unwrap();
        }

        // Depth is a property of the queue, set synchronously by `send`, so it
        // reads 3 regardless of the worker's scheduling state.
        let via_snapshot = snapshot()
            .actors
            .into_iter()
            .find(|a| a.pid == h.pid())
            .expect("worker present");
        assert_eq!(via_snapshot.mailbox_depth, 3);
        assert_eq!(actor_info(h.pid()).unwrap().mailbox_depth, 3);

        gate_tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn monitor_count_is_visible() {
    run(|| {
        let (ready_tx, ready_rx) = channel::<()>();
        let (gate_tx, gate_rx) = channel::<()>();
        let h = spawn(move || {
            ready_tx.send(()).unwrap();
            gate_rx.recv().unwrap();
        });
        ready_rx.recv().unwrap();

        let _m = monitor(h.pid());
        let info = actor_info(h.pid()).expect("worker present");
        assert_eq!(info.monitors, 1);

        gate_tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn stale_and_forged_pids_return_none() {
    run(|| {
        let (ready_tx, ready_rx) = channel::<()>();
        let (gate_tx, gate_rx) = channel::<()>();
        let h = spawn(move || {
            ready_tx.send(()).unwrap();
            gate_rx.recv().unwrap();
        });
        ready_rx.recv().unwrap();
        let live = h.pid();

        // Same slot index, wrong generation → stale, no such incarnation.
        let stale = Pid::new(live.index(), live.generation().wrapping_add(7));
        assert!(actor_info(stale).is_none());

        // Out-of-range index → not in the slab at all.
        let forged = Pid::new(u32::MAX - 1, 0);
        assert!(actor_info(forged).is_none());

        gate_tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn done_actor_is_a_tombstone() {
    run(|| {
        // Hold the join handle so the slot is NOT reclaimed when the actor
        // exits: outstanding_handles stays > 0, leaving a Done tombstone to
        // observe.
        let h = spawn(|| {});
        let pid = h.pid();
        let info = spin_until(pid, |a| a.state == ActorState::Done);
        assert_eq!(info.state, ActorState::Done);
        // The Actor record is gone at finalize, so a tombstone reports root-less
        // with empty lifecycle counts.
        assert_eq!(info.supervisor, Pid::new(u32::MAX, u32::MAX));
        assert_eq!((info.monitors, info.links, info.joiners), (0, 0, 0));
        h.join().unwrap();
    });
}