liminal-server 0.21.1

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use super::{Lifecycle, LifecycleError, Placer, ServiceState};
use crate::config::types::ServiceTemplateConfig;

/// Records every placement and refuses those whose child id is listed.
#[derive(Debug, Default)]
struct FakeLys {
    placed: Mutex<Vec<(String, String, String)>>,
    refuse: Option<&'static str>,
}

impl Placer for FakeLys {
    fn place(
        &self,
        operation: &str,
        child: (&str, &str),
        parent: (&str, &str),
    ) -> Result<(), String> {
        if self.refuse.is_some_and(|kind| kind == child.0) {
            return Err("WithheldFromAgents".to_owned());
        }
        assert_eq!(parent, ("liminal.network", "net"));
        self.placed
            .lock()
            .map_err(|_| "poisoned".to_owned())?
            .push((operation.to_owned(), child.0.to_owned(), child.1.to_owned()));
        Ok(())
    }
}

/// A fresh directory under the target's temporary space.
fn scratch(name: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!("liminal-lifecycle-{name}-{}", std::process::id()));
    let _gone = std::fs::remove_dir_all(&dir);
    dir
}

/// A template running `script` under `/bin/sh -c`.
fn template(script: &str) -> ServiceTemplateConfig {
    ServiceTemplateConfig {
        name: "echo".to_owned(),
        command: PathBuf::from("/bin/sh"),
        args: vec!["-c".to_owned(), script.to_owned()],
        subjects: vec!["echo".to_owned()],
        env: Vec::new(),
        stop_timeout_ms: 2_000,
    }
}

fn open(dir: &Path, script: &str, lys: Arc<FakeLys>) -> Arc<Lifecycle> {
    let lifecycle = Lifecycle::open(
        &[template(script)],
        dir.to_path_buf(),
        "net".to_owned(),
        Some(lys),
    )
    .unwrap_or_else(|error| unreachable!("the registry opens: {error}"));
    lifecycle.listening("127.0.0.1:1".to_owned());
    lifecycle
}

/// Waits up to five seconds for `done`.
fn eventually(mut done: impl FnMut() -> bool) -> bool {
    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if done() {
            return true;
        }
        // ast-grep-allow: no-sleep-in-tests: waits on a file or record another process writes; no in-process signal crosses that boundary
        std::thread::sleep(Duration::from_millis(20));
    }
    done()
}

#[test]
fn create_places_the_service_and_subjects_then_records_asleep() -> Result<(), LifecycleError> {
    let dir = scratch("create");
    let lys = Arc::new(FakeLys::default());
    let lifecycle = open(&dir, "true", Arc::clone(&lys));
    let record = lifecycle.create("echo", "one", "agent-a", "op-1", 10)?;
    assert_eq!(record.state, ServiceState::Asleep);
    assert_eq!(record.subjects, vec!["echo.one".to_owned()]);
    assert_eq!(record.created_by, "agent-a");
    let placed = lys.placed.lock().map(|p| p.clone()).unwrap_or_default();
    let kinds: Vec<(&str, &str)> = placed
        .iter()
        .map(|(_, kind, id)| (kind.as_str(), id.as_str()))
        .collect();
    assert_eq!(
        kinds,
        vec![
            ("liminal.service", record.id.as_str()),
            ("liminal.subject", "echo.one")
        ]
    );
    // The same operation from the same holder is the same service, placed once.
    assert_eq!(
        lifecycle.create("echo", "one", "agent-a", "op-1", 11)?,
        record
    );
    assert_eq!(lys.placed.lock().map(|p| p.len()).unwrap_or_default(), 2);
    Ok(())
}

#[test]
fn create_refused_by_lys_records_nothing() {
    let dir = scratch("refused");
    let lys = Arc::new(FakeLys {
        refuse: Some("liminal.subject"),
        ..FakeLys::default()
    });
    let lifecycle = open(&dir, "true", lys);
    let refused = lifecycle.create("echo", "one", "agent-a", "op-1", 10);
    assert_eq!(
        refused,
        Err(LifecycleError::Placement("WithheldFromAgents".to_owned()))
    );
    let listed = lifecycle.list(|_| true).unwrap_or_default();
    assert_eq!(listed["services"], serde_json::json!([]));
}

#[test]
fn create_refuses_an_unknown_template_a_bad_name_and_a_taken_name() -> Result<(), LifecycleError> {
    let dir = scratch("names");
    let lifecycle = open(&dir, "true", Arc::new(FakeLys::default()));
    assert_eq!(
        lifecycle.create("nope", "one", "a", "op-1", 1),
        Err(LifecycleError::UnknownTemplate("nope".to_owned()))
    );
    assert_eq!(
        lifecycle.create("echo", "o.ne", "a", "op-1", 1),
        Err(LifecycleError::NameInvalid)
    );
    lifecycle.create("echo", "one", "a", "op-1", 1)?;
    assert_eq!(
        lifecycle.create("echo", "one", "b", "op-2", 1),
        Err(LifecycleError::NameTaken("one".to_owned()))
    );
    Ok(())
}

#[test]
fn registry_survives_a_restart_with_every_transition() -> Result<(), LifecycleError> {
    let dir = scratch("restart");
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, "true", Arc::clone(&lys));
    let made = first.create("echo", "one", "agent-a", "op-1", 10)?;
    let held = first.retire(&made.id, "agent-a", 20)?;
    drop(first);
    let second = open(&dir, "true", lys);
    let read = second.record(&made.id)?;
    assert_eq!(read, held);
    assert_eq!(read.state, ServiceState::Retired);
    assert_eq!(read.transitions.len(), 2);
    Ok(())
}

#[test]
fn deploy_starts_the_template_and_goes_live_on_registration() -> Result<(), LifecycleError> {
    let dir = scratch("deploy");
    let lifecycle = open(&dir, "exec sleep 30", Arc::new(FakeLys::default()));
    let made = lifecycle.create("echo", "one", "agent-a", "op-1", 10)?;
    lifecycle.deploy(&made.id, "agent-b")?;
    // A second deploy while it runs does nothing.
    assert!(matches!(
        lifecycle.deploy(&made.id, "agent-b"),
        Err(LifecycleError::NotNow { running: true, .. })
    ));
    lifecycle.mark_live(&made.id, 30)?;
    let live = lifecycle.record(&made.id)?;
    assert_eq!(live.state, ServiceState::Live);
    assert_eq!(
        live.transitions.last().map(|t| t.holder.as_str()),
        Some("agent-b")
    );
    let held = lifecycle.hold(&made.id, "agent-b", 40)?;
    assert_eq!(held.state, ServiceState::Held);
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    lifecycle.wake(&made.id, "agent-b")?;
    lifecycle.retire(&made.id, "agent-b", 50)?;
    assert_eq!(lifecycle.record(&made.id)?.state, ServiceState::Retired);
    Ok(())
}

#[test]
fn hold_stops_within_the_timeout_even_when_sigterm_is_ignored() -> Result<(), LifecycleError> {
    let dir = scratch("stubborn");
    let lifecycle = open(
        &dir,
        "trap '' TERM; while true; do sleep 1; done",
        Arc::new(FakeLys::default()),
    );
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    lifecycle.hold(&made.id, "a", 2)?;
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    assert_eq!(lifecycle.record(&made.id)?.state, ServiceState::Held);
    Ok(())
}

#[test]
fn a_process_credential_admits_only_its_own_subjects() -> Result<(), LifecycleError> {
    let dir = scratch("token");
    let file = dir.join("token");
    let mut written = template("printf %s \"$LIMINAL_TOKEN\" > \"$0\"; exec sleep 30");
    written.args.push(file.display().to_string());
    let lifecycle = Lifecycle::open(
        &[written],
        dir,
        "net".to_owned(),
        Some(Arc::new(FakeLys::default())),
    )?;
    lifecycle.listening("127.0.0.1:1".to_owned());
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    assert!(eventually(
        || std::fs::metadata(&file).is_ok_and(|m| m.len() == 64)
    ));
    let token = std::fs::read(&file).unwrap_or_default();
    let grant = lifecycle.admit_process(&token);
    assert_eq!(
        grant.as_ref().map(|g| g.service.as_str()),
        Some(made.id.as_str())
    );
    assert!(grant.as_ref().is_some_and(|g| g.serves("echo.one")));
    assert!(grant.as_ref().is_some_and(|g| !g.serves("echo.two")));
    assert_eq!(lifecycle.admit_process(b"not-a-token"), None);
    lifecycle.hold(&made.id, "a", 2)?;
    assert_eq!(lifecycle.admit_process(&token), None, "a stop revokes it");
    Ok(())
}

#[test]
fn an_unasked_exit_is_recorded_and_not_restarted() -> Result<(), LifecycleError> {
    let dir = scratch("exit");
    let lifecycle = open(&dir, "exit 3", Arc::new(FakeLys::default()));
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    assert!(eventually(|| {
        lifecycle
            .record(&made.id)
            .is_ok_and(|record| record.transitions.len() == 2)
    }));
    let record = lifecycle.record(&made.id)?;
    assert_eq!(record.state, ServiceState::Asleep);
    let note = record
        .transitions
        .last()
        .and_then(|t| t.note.clone())
        .unwrap_or_default();
    assert!(
        note.contains("exited unasked") && note.contains('3'),
        "{note}"
    );
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    Ok(())
}

#[test]
fn restart_live_starts_what_was_live() -> Result<(), LifecycleError> {
    let dir = scratch("boot");
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, "exec sleep 30", Arc::clone(&lys));
    let made = first.create("echo", "one", "a", "op-1", 1)?;
    first.deploy(&made.id, "a")?;
    first.mark_live(&made.id, 2)?;
    first.shutdown();
    assert_eq!(
        first.record(&made.id)?.state,
        ServiceState::Live,
        "shutdown keeps state"
    );
    drop(first);
    let second = open(&dir, "exec sleep 30", lys);
    assert_eq!(second.restart_live(3), Vec::new());
    assert_eq!(second.list(|_| true)?["services"][0]["running"], true);
    second.shutdown();
    Ok(())
}

#[test]
fn an_exit_after_shutdown_began_keeps_live_and_the_next_start_restarts_it()
-> Result<(), LifecycleError> {
    let dir = scratch("disconnect");
    std::fs::create_dir_all(&dir).map_err(|error| LifecycleError::Store(error.to_string()))?;
    // the worker's connection: the process leaves when its writer closes
    let connection = dir.join("connection");
    let made_fifo = std::process::Command::new("/usr/bin/mkfifo")
        .arg(&connection)
        .status()
        .map_err(|error| LifecycleError::Store(error.to_string()))?;
    assert!(made_fifo.success(), "mkfifo {}", connection.display());
    let script = format!("read _ < '{}'; exit 1", connection.display());
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, &script, Arc::clone(&lys));
    let made = first.create("echo", "one", "a", "op-1", 1)?;
    first.deploy(&made.id, "a")?;
    first.mark_live(&made.id, 2)?;
    let before = first.record(&made.id)?.transitions.len();
    first.shutdown_begun();
    // the server drains: the connection closes and the process exits 1
    drop(
        std::fs::OpenOptions::new()
            .write(true)
            .open(&connection)
            .map_err(|error| LifecycleError::Store(error.to_string()))?,
    );
    assert!(eventually(|| {
        first
            .list(|_| true)
            .is_ok_and(|listed| listed["services"][0]["running"] == false)
    }));
    let record = first.record(&made.id)?;
    assert_eq!(record.state, ServiceState::Live, "the exit kept live");
    assert_eq!(record.transitions.len(), before, "nothing recorded");
    first.shutdown();
    drop(first);
    let second = open(&dir, "exec sleep 30", lys);
    assert_eq!(second.restart_live(3), Vec::new());
    assert_eq!(second.list(|_| true)?["services"][0]["running"], true);
    second.shutdown();
    Ok(())
}

#[test]
fn no_placer_refuses_every_create_by_name() {
    let dir = scratch("noplacer");
    let lifecycle = Lifecycle::open(&[template("true")], dir, "net".to_owned(), None)
        .unwrap_or_else(|error| unreachable!("the registry opens: {error}"));
    assert_eq!(
        lifecycle.create("echo", "one", "a", "op-1", 1),
        Err(LifecycleError::NoPlacer)
    );
}