pushkin 0.2.0

//! Phase 4 tasks 3+6: daemon lifecycle verbs and the canonical-binary
//! guard (spec §8.4, §11). Committed first, read-only hereafter (charter
//! §4.1, N10). Pins: the first `daemon start` pins the invoking binary as
//! canonical (trust-on-first-use, the consent pattern); a NON-canonical
//! binary is refused start/stop/restart and offered a read-only daemon
//! instead; read-only daemons live on private sockets, answer checks and
//! pings (flagged `readOnly`), and refuse protocol mutations (Shutdown —
//! their lifecycle belongs to the OS session that spawned them, not to
//! the wire); `status` is a question and works for anyone; a stale socket
//! file left by an unclean exit does not wedge the next start.

use assert_cmd::Command;
use pushkin_daemon::protocol::{Request, Response, PROTOCOL_VERSION};
use pushkin_daemon::server;
use std::fs;
use std::path::Path;
use std::time::{Duration, Instant};

const MANIFEST: &str = r#"
version = 1
canonical = "json-schema-2020-12"
authoring = "zod"

[[contracts]]
name = "user"
source = "contracts/user.zod.ts"
emit = ["zod"]

[[mappings]]
glob = "app/api/**/*.ts"
contracts = ["user"]
require = "boundary-validation"

[gates]
protected_paths = ["pushkin.toml"]
"#;

const CANONICAL_FILE: &str = ".pushkin/daemon.canonical";
const SOCKET_FILE: &str = ".pushkin/daemon.sock";

fn repo() -> std::io::Result<tempfile::TempDir> {
    let dir = tempfile::tempdir()?;
    fs::write(dir.path().join("pushkin.toml"), MANIFEST)?;
    Ok(dir)
}

fn daemon_cmd(dir: &Path, args: &[&str]) -> (i32, String, String) {
    let Ok(mut command) = Command::cargo_bin("pushkin") else {
        return (-1, String::new(), "cargo_bin resolution failed".to_owned());
    };
    let mut full = vec!["daemon"];
    full.extend_from_slice(args);
    let Ok(output) = command.current_dir(dir).args(&full).output() else {
        return (-1, String::new(), "spawn failed".to_owned());
    };
    (
        output.status.code().unwrap_or(-1),
        String::from_utf8_lossy(&output.stdout).into_owned(),
        String::from_utf8_lossy(&output.stderr).into_owned(),
    )
}

fn wait_socket_gone(socket: &Path) -> bool {
    let deadline = Instant::now() + Duration::from_secs(3);
    while Instant::now() < deadline {
        if !socket.exists() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(25));
    }
    false
}

fn stop(dir: &Path) {
    let _ = daemon_cmd(dir, &["stop"]);
}

fn kill_pid(pid: u32) {
    let _ = std::process::Command::new("kill")
        .arg(pid.to_string())
        .status();
}

#[test]
fn first_start_pins_canonical_binary() {
    let dir = repo().unwrap();
    let (code, _, stderr) = daemon_cmd(dir.path(), &["start"]);
    assert_eq!(code, 0, "first start must succeed: {stderr}");

    let pinned = fs::read_to_string(dir.path().join(CANONICAL_FILE)).unwrap();
    let expected = Command::cargo_bin("pushkin").unwrap();
    let expected = Path::new(expected.get_program())
        .canonicalize()
        .unwrap()
        .to_string_lossy()
        .into_owned();
    assert_eq!(
        pinned.trim(),
        expected,
        "first start pins the invoking binary as canonical"
    );
    stop(dir.path());
}

#[test]
fn non_canonical_binary_refused_lifecycle_verbs() {
    let dir = repo().unwrap();
    fs::create_dir_all(dir.path().join(".pushkin")).unwrap();
    fs::write(dir.path().join(CANONICAL_FILE), "/usr/bin/false\n").unwrap();

    for verb in ["start", "stop", "restart"] {
        let (code, _, stderr) = daemon_cmd(dir.path(), &[verb]);
        assert_ne!(code, 0, "non-canonical `{verb}` must be refused");
        assert!(
            stderr.contains("read-only"),
            "`{verb}` refusal must offer the read-only alternative: {stderr}"
        );
    }
    assert!(
        !dir.path().join(SOCKET_FILE).exists(),
        "a refused start must not leave a daemon behind"
    );
}

#[test]
fn daemon_status_reports_socket_and_pid() {
    let dir = repo().unwrap();
    let (code, _, stderr) = daemon_cmd(dir.path(), &["start"]);
    assert_eq!(code, 0, "start failed: {stderr}");

    let (code, stdout, _) = daemon_cmd(dir.path(), &["status"]);
    assert_eq!(code, 0);
    assert!(
        stdout.contains("pid:"),
        "status must report a pid: {stdout}"
    );
    assert!(
        stdout.contains(SOCKET_FILE),
        "status must name the socket: {stdout}"
    );
    stop(dir.path());

    let (code, stdout, _) = daemon_cmd(dir.path(), &["status"]);
    assert_eq!(code, 1, "status of a stopped daemon is unhealthy");
    assert!(stdout.contains("not running"), "got: {stdout}");
}

#[test]
fn stop_removes_socket_file() {
    let dir = repo().unwrap();
    let (code, _, stderr) = daemon_cmd(dir.path(), &["start"]);
    assert_eq!(code, 0, "start failed: {stderr}");
    let socket = dir.path().join(SOCKET_FILE);
    assert!(socket.exists(), "start must leave a listening socket");

    let (code, _, stderr) = daemon_cmd(dir.path(), &["stop"]);
    assert_eq!(code, 0, "stop failed: {stderr}");
    assert!(
        wait_socket_gone(&socket),
        "stop must remove the socket file"
    );
}

#[test]
fn restart_survives_stale_socket_file() {
    let dir = repo().unwrap();
    // An unclean exit leaves a dead socket file behind.
    fs::create_dir_all(dir.path().join(".pushkin")).unwrap();
    fs::write(dir.path().join(SOCKET_FILE), "stale").unwrap();

    let (code, _, stderr) = daemon_cmd(dir.path(), &["restart"]);
    assert_eq!(
        code, 0,
        "restart must recover from a stale socket: {stderr}"
    );

    let (code, stdout, _) = daemon_cmd(dir.path(), &["status"]);
    assert_eq!(code, 0, "daemon must be live after restart: {stdout}");
    stop(dir.path());
}

#[test]
fn read_only_daemon_offered_on_private_socket() {
    let dir = repo().unwrap();
    fs::create_dir_all(dir.path().join(".pushkin")).unwrap();
    fs::write(dir.path().join(CANONICAL_FILE), "/usr/bin/false\n").unwrap();

    let (code, stdout, stderr) = daemon_cmd(dir.path(), &["start", "--read-only"]);
    assert_eq!(code, 0, "read-only start is open to any binary: {stderr}");
    let socket_line = stdout
        .lines()
        .find(|line| line.starts_with("socket: "))
        .unwrap_or_else(|| panic!("read-only start must print its socket: {stdout}"));
    let socket = Path::new(socket_line.trim_start_matches("socket: ").trim());

    assert!(socket.exists(), "private socket must be live");
    assert_ne!(
        socket,
        dir.path().join(SOCKET_FILE),
        "read-only daemons never squat the canonical socket"
    );

    let info = match server::request_at(
        socket,
        &Request::Ping {
            v: PROTOCOL_VERSION,
        },
    ) {
        Ok(Response::Pong { info }) => info,
        other => panic!("expected pong over the private socket, got {other:?}"),
    };
    assert!(info.read_only, "the daemon must flag itself read-only");
    kill_pid(info.pid);
}

#[test]
fn read_only_daemon_rejects_state_mutation() {
    let dir = repo().unwrap();
    fs::create_dir_all(dir.path().join(".pushkin")).unwrap();
    fs::write(dir.path().join(CANONICAL_FILE), "/usr/bin/false\n").unwrap();

    let (code, stdout, _) = daemon_cmd(dir.path(), &["start", "--read-only"]);
    assert_eq!(code, 0);
    let socket_line = stdout
        .lines()
        .find(|line| line.starts_with("socket: "))
        .unwrap_or_else(|| panic!("read-only start must print its socket: {stdout}"));
    let socket = Path::new(socket_line.trim_start_matches("socket: ").trim());

    // Shutdown over the wire is a control mutation; a read-only daemon's
    // lifecycle belongs to the OS session that spawned it, not the socket.
    match server::request_at(
        socket,
        &Request::Shutdown {
            v: PROTOCOL_VERSION,
        },
    ) {
        Ok(Response::Error { message }) => {
            assert!(message.contains("read-only"), "got: {message}");
        }
        other => panic!("read-only daemon must refuse shutdown, got {other:?}"),
    }
    assert!(socket.exists(), "the refused shutdown must not kill it");

    let info = match server::request_at(
        socket,
        &Request::Ping {
            v: PROTOCOL_VERSION,
        },
    ) {
        Ok(Response::Pong { info }) => info,
        other => panic!("daemon must still answer after refusing, got {other:?}"),
    };
    kill_pid(info.pid);
}