//! Named registry tests. Run under the scheduler: registration requires a
//! live runtime and live actors.

use smarm::{channel, name_of, register, run, spawn, unregister, whereis, RegisterError};

#[test]
fn register_whereis_roundtrip() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        register("worker", h.pid()).unwrap();
        assert_eq!(whereis("worker"), Some(h.pid()));
        assert_eq!(name_of(h.pid()).as_deref(), Some("worker"));
        tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn register_is_idempotent_for_same_binding() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        register("svc", h.pid()).unwrap();
        // Same name, same pid: a no-op Ok, not NameTaken.
        register("svc", h.pid()).unwrap();
        tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn duplicate_name_on_live_holder_is_rejected() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let (tx2, rx2) = channel::<()>();
        let a = spawn(move || {
            rx.recv().unwrap();
        });
        let b = spawn(move || {
            rx2.recv().unwrap();
        });
        register("svc", a.pid()).unwrap();
        assert_eq!(
            register("svc", b.pid()),
            Err(RegisterError::NameTaken { holder: a.pid() })
        );
        tx.send(()).unwrap();
        tx2.send(()).unwrap();
        a.join().unwrap();
        b.join().unwrap();
    });
}

#[test]
fn one_name_per_pid() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        register("first", h.pid()).unwrap();
        assert_eq!(
            register("second", h.pid()),
            Err(RegisterError::PidAlreadyRegistered { name: "first".into() })
        );
        tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn registering_a_dead_pid_is_noproc() {
    run(|| {
        let h = spawn(|| {});
        let pid = h.pid();
        h.join().unwrap();
        assert_eq!(register("ghost", pid), Err(RegisterError::NoProc));
    });
}

#[test]
fn binding_evaporates_on_death_and_name_is_reusable() {
    run(|| {
        let a = spawn(|| {});
        let pid_a = a.pid();
        register("svc", pid_a).unwrap();
        a.join().unwrap();

        // Dead holder: both lookup directions report unbound.
        assert_eq!(whereis("svc"), None);
        assert_eq!(name_of(pid_a), None);

        // And the name is free for a successor.
        let (tx, rx) = channel::<()>();
        let b = spawn(move || {
            rx.recv().unwrap();
        });
        register("svc", b.pid()).unwrap();
        assert_eq!(whereis("svc"), Some(b.pid()));
        tx.send(()).unwrap();
        b.join().unwrap();
    });
}

#[test]
fn register_over_a_dead_holder_succeeds_without_lookup_in_between() {
    // The eviction path inside register() itself (not via whereis pruning).
    run(|| {
        let a = spawn(|| {});
        let pid_a = a.pid();
        register("svc", pid_a).unwrap();
        a.join().unwrap();

        let (tx, rx) = channel::<()>();
        let b = spawn(move || {
            rx.recv().unwrap();
        });
        register("svc", b.pid()).unwrap();
        assert_eq!(whereis("svc"), Some(b.pid()));
        assert_eq!(name_of(pid_a), None);
        tx.send(()).unwrap();
        b.join().unwrap();
    });
}

#[test]
fn unregister_frees_both_directions() {
    run(|| {
        let (tx, rx) = channel::<()>();
        let h = spawn(move || {
            rx.recv().unwrap();
        });
        register("svc", h.pid()).unwrap();
        assert_eq!(unregister("svc"), Some(h.pid()));
        assert_eq!(whereis("svc"), None);
        assert_eq!(name_of(h.pid()), None);
        assert_eq!(unregister("svc"), None);

        // The pid may take a new name afterwards.
        register("svc2", h.pid()).unwrap();
        assert_eq!(name_of(h.pid()).as_deref(), Some("svc2"));
        tx.send(()).unwrap();
        h.join().unwrap();
    });
}

#[test]
fn whereis_from_another_actor_and_usable_with_runtime_apis() {
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;

    let stopped = Arc::new(AtomicBool::new(false));
    let stopped2 = stopped.clone();
    run(move || {
        let (tx, rx) = channel::<()>();
        let svc = spawn(move || {
            // Parks forever; only a request_stop ends it.
            let _ = rx.recv();
        });
        register("stoppable", svc.pid()).unwrap();

        let stopped3 = stopped2.clone();
        let client = spawn(move || {
            let pid = whereis("stoppable").expect("name must resolve cross-actor");
            // The registry's pids plug into the rest of the runtime API.
            smarm::request_stop(pid);
            stopped3.store(true, Ordering::Relaxed);
        });
        client.join().unwrap();
        svc.join().unwrap(); // a stopped actor joins Ok
        drop(tx);
    });
    assert!(stopped.load(Ordering::Relaxed));
}

#[test]
fn registry_under_concurrent_churn_multi_thread() {
    // Many actors racing to claim the same names while holders die: the
    // bimap invariant (debug-asserted internally) and Erlang semantics must
    // hold under real parallelism.
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Arc;

    let wins = Arc::new(AtomicU32::new(0));
    let wins2 = wins.clone();
    smarm::init(smarm::Config::exact(4)).run(move || {
        let mut handles = Vec::new();
        for round in 0..8 {
            let name = format!("contested-{}", round % 2);
            for _ in 0..8 {
                let name = name.clone();
                let wins = wins2.clone();
                handles.push(spawn(move || {
                    let me = smarm::self_pid();
                    match register(&name, me) {
                        Ok(()) => {
                            wins.fetch_add(1, Ordering::Relaxed);
                            smarm::yield_now();
                            // May have been pruned-by-contact never; we are
                            // alive, so our binding must still resolve to us.
                            assert_eq!(whereis(&name), Some(me));
                            assert_eq!(unregister(&name), Some(me));
                        }
                        Err(RegisterError::NameTaken { .. }) => {}
                        Err(e) => panic!("unexpected register error: {e}"),
                    }
                }));
            }
        }
        for h in handles {
            h.join().unwrap();
        }
    });
    // At least one registration per name must have succeeded.
    assert!(wins.load(Ordering::Relaxed) >= 2);
}