//! Deterministic service supervision tests: native tasks stand in for guest lifetimes.
use super::*;
use crate::{NativePlugin, runtime::Runtime};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot, watch};

struct Rig {
    host: Host,
    starts: mpsc::UnboundedReceiver<(String, oneshot::Sender<()>)>,
    _world: tempfile::TempDir,
}
impl Rig {
    async fn new() -> Self {
        let world = tempfile::tempdir().unwrap();
        let runtime = Runtime::new(world.path()).unwrap();
        let host = Host::test_world(world.path(), runtime).await.unwrap();
        let (tx, starts) = mpsc::unbounded_channel();
        let plugin: NativePlugin = Arc::new(move |ctx, method, _| {
            let tx = tx.clone();
            Box::pin(async move {
                if method != "serve" {
                    return Ok(Value::Null);
                }
                let (end, ended) = oneshot::channel();
                tx.send((ctx.pid, end)).unwrap();
                let _ = ended.await;
                Ok(Value::Null)
            })
        });
        host.register_native("demo", plugin.clone());
        host.register_native("other", plugin);
        Self {
            host,
            starts,
            _world: world,
        }
    }
    async fn start(&mut self, method: &str) -> (String, oneshot::Sender<()>) {
        let host = self.host.clone();
        let method = method.to_string();
        tokio::spawn(async move {
            host.start_service(method, json!({"kept":true})).await;
        });
        self.next().await
    }
    async fn next(&mut self) -> (String, oneshot::Sender<()>) {
        // Reuse the host integration call bound (activation.rs::call).
        tokio::time::timeout(Duration::from_secs(60), self.starts.recv())
            .await
            .unwrap()
            .unwrap()
    }
    fn ended(&self, pid: &str) -> watch::Receiver<bool> {
        self.host.0.state.lock().unwrap().processes[pid]
            .ended
            .subscribe()
    }
    async fn crash(&self, pid: &str) {
        let mut ended = self.ended(pid);
        // Abort execution without asking the supervisor for an intentional stop.
        self.host.0.state.lock().unwrap().processes[pid]
            .abort
            .as_ref()
            .unwrap()
            .abort();
        ended.wait_for(|done| *done).await.unwrap();
    }
    async fn clean(&self, pid: &str, end: oneshot::Sender<()>) {
        let mut ended = self.ended(pid);
        end.send(()).unwrap();
        ended.wait_for(|done| *done).await.unwrap();
    }
    fn assert_single(&self, pid: &str, registrations: usize) {
        let state = self.host.0.state.lock().unwrap();
        assert_eq!(state.services.len(), registrations);
        let p = &state.processes[pid];
        assert_eq!(p.info.params, json!({"kept":true}));
        assert!(p.info.detached);
        assert_eq!(p.caller, "core");
        assert_eq!(
            state
                .services
                .iter()
                .filter(|s| s.owner.as_deref() == Some(pid))
                .count(),
            1
        );
    }
    fn no_start(&mut self) {
        assert!(self.starts.try_recv().is_err());
    }
}

#[tokio::test(start_paused = true)]
async fn crash_before_and_after_readiness_observes_process_not_response() {
    let mut rig = Rig::new().await;
    let (pid, _end) = rig.start("demo.serve").await;
    rig.assert_single(&pid, 1);
    rig.crash(&pid).await; // No ready RPC response was sent.
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    rig.host.answer(&pid, Ok(json!({"ready":true})));
    tokio::task::yield_now().await;
    rig.no_start(); // Ready does not end the service lifetime.
    rig.crash(&pid).await;
    drop(end);
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    rig.clean(&pid, end).await;
    rig.no_start(); // A clean, intentional return is not a crash.
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn fifth_restarts_sixth_gives_up_and_journals_once() {
    let mut rig = Rig::new().await;
    let (mut pid, mut end) = rig.start("demo.serve").await;
    for crash in 1..=CRASH_LIMIT {
        rig.crash(&pid).await;
        drop(end);
        (pid, end) = rig.next().await;
        rig.assert_single(&pid, 1);
        assert_eq!(
            rig.host.0.state.lock().unwrap().services[0].crashes.len(),
            crash
        );
    }
    rig.crash(&pid).await;
    drop(end);
    rig.no_start();
    let logs = rig
        .host
        .call("core.log", json!({"grep":"service gave up"}))
        .await
        .unwrap();
    assert_eq!(logs.as_array().unwrap().len(), 1);
    assert!(
        logs[0]["line"]
            .as_str()
            .unwrap()
            .contains("more than 5 crashes within 10 seconds")
    );
    assert!(!rig.host.0.state.lock().unwrap().services[0].enabled);
    // A retired hook cannot bypass give-up while its supervisor has no owner.
    rig.host.register_native(
        "fallback",
        Arc::new(move |ctx, _, _| {
            Box::pin(async move {
                ctx.call(
                    "core.call",
                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
                )
                .await
            })
        }),
    );
    assert_eq!(
        rig.host
            .call("fallback.rpc.retired", json!({}))
            .await
            .unwrap(),
        Value::Null
    );
    rig.no_start();
    // Neither time passing nor an upgrade re-arms a given-up service.
    tokio::time::advance(CRASH_WINDOW).await;
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
    rig.host.begin_upgrade(upgrade);
    tokio::task::yield_now().await;
    rig.no_start();
    rig.host.stop().await;
}

#[test]
fn sliding_window_boundary_and_progress() {
    let now = Instant::now();
    let service = || Service {
        method: "demo.serve".into(),
        params: json!({}),
        request: json!("request"),
        owner: None,
        enabled: true,
        pending: false,
        version: None,
        crashes: VecDeque::new(),
    };
    let mut before = service();
    let mut exact = service();
    let mut after = service();
    for _ in 0..CRASH_LIMIT {
        assert!(before.crashed(now));
        assert!(exact.crashed(now));
        assert!(after.crashed(now));
    }
    assert!(!before.crashed(now + CRASH_WINDOW - Duration::from_nanos(1)));
    assert!(exact.crashed(now + CRASH_WINDOW));
    assert_eq!(exact.crashes.len(), 1);
    assert!(after.crashed(now + CRASH_WINDOW + Duration::from_nanos(1)));
    // Repeated progress beyond the window never accumulates old failures.
    for step in 2..=CRASH_LIMIT as u32 + 1 {
        assert!(exact.crashed(now + CRASH_WINDOW * step));
        assert_eq!(exact.crashes.len(), 1);
    }
}

#[tokio::test(start_paused = true)]
async fn services_have_independent_crash_budgets() {
    let mut rig = Rig::new().await;
    let (mut a, mut a_end) = rig.start("demo.serve").await;
    let (b, b_end) = rig.start("other.serve").await;
    for _ in 0..CRASH_LIMIT {
        rig.crash(&a).await;
        drop(a_end);
        (a, a_end) = rig.next().await;
    }
    rig.crash(&a).await;
    drop(a_end);
    rig.crash(&b).await;
    drop(b_end);
    let (b, b_end) = rig.next().await;
    rig.assert_single(&b, 2);
    rig.clean(&b, b_end).await;
    rig.no_start();
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn upgrade_crash_race_has_one_restart_and_retired_fallback_cannot_duplicate_it() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
    rig.crash(&pid).await; // Exits between snapshot and begin_upgrade.
    drop(end);
    rig.host.begin_upgrade(upgrade);
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    assert!(
        rig.host.0.state.lock().unwrap().services[0]
            .crashes
            .is_empty()
    );
    // A native retired hook exercises the same routing path as a guest fallback.
    rig.host.register_native(
        "fallback",
        Arc::new(move |ctx, _, _| {
            Box::pin(async move {
                ctx.call(
                    "core.call",
                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
                )
                .await
            })
        }),
    );
    assert_eq!(
        rig.host
            .call("fallback.rpc.retired", json!({}))
            .await
            .unwrap(),
        Value::Null
    );
    rig.no_start();
    rig.clean(&pid, end).await;
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn crash_before_upgrade_snapshot_also_has_one_restart() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    rig.crash(&pid).await;
    drop(end);
    // Restart may be pending admission or already admitted: snapshot handles both.
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
    rig.host.begin_upgrade(upgrade);
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    rig.no_start();
    rig.clean(&pid, end).await;
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn pending_admission_is_single_owned_across_upgrade() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    let host = rig.host.clone();
    let admission = host.0.admission.write().await;
    rig.crash(&pid).await;
    drop(end);
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
    rig.host.begin_upgrade(upgrade);
    assert!(rig.host.0.state.lock().unwrap().services[0].pending);
    drop(admission);
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    rig.no_start();
    rig.clean(&pid, end).await;
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn shutdown_rejects_restart_already_waiting_for_admission() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    let host = rig.host.clone();
    let admission = host.0.admission.write().await;
    rig.crash(&pid).await;
    drop(end);
    rig.host.stop().await;
    drop(admission);
    tokio::task::yield_now().await;
    rig.no_start();
    assert!(rig.host.processes().is_empty());
}

#[tokio::test(start_paused = true)]
async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disabled() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    rig.clean(&pid, end).await;
    rig.no_start();
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
    rig.host.begin_upgrade(upgrade);
    let (pid, end) = rig.next().await;
    rig.assert_single(&pid, 1);
    // An explicit stop stays disabled even if the task subsequently exits cleanly.
    let mut ended = rig.host.begin_stop(&pid).unwrap();
    end.send(()).unwrap();
    ended.wait_for(|done| *done).await.unwrap();
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "newer");
    rig.host.begin_upgrade(upgrade);
    tokio::task::yield_now().await;
    rig.no_start();
    rig.host.stop().await;
}

#[tokio::test(start_paused = true)]
async fn intentional_stop_and_shutdown_never_restart_even_during_upgrade() {
    let mut rig = Rig::new().await;
    let (pid, end) = rig.start("demo.serve").await;
    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
    // An explicit kill after supersession overrides the automatic upgrade restart.
    rig.host
        .call("core.kill", json!({"pid":pid}))
        .await
        .unwrap();
    drop(end);
    rig.host.begin_upgrade(upgrade);
    tokio::task::yield_now().await;
    rig.no_start();
    let (_pid, end) = rig.start("other.serve").await;
    rig.host.stop().await;
    drop(end);
    rig.no_start();
}