pushkin-daemon 0.2.0

//! Phase 4 task 2: UDS IPC + cold fallback (spec §4.3). Committed first,
//! read-only hereafter (charter §4.1, N10). Pins: a served check round
//! trip returns the SAME uniform result JSON the cold path produces; a
//! shim that cannot reach the daemon falls back to the cold path (the
//! gate never fails open because the daemon is down); concurrent shim
//! connections each get their own uncorrupted response; the socket lives
//! under `.pushkin/` and is removed on shutdown.
//!
//! Daemon auto-start on first shim connect is pinned at the CLI layer
//! (`daemon_lifecycle.rs`) where the spawning binary exists.

use pushkin_core::envelope::Decision;
use pushkin_core::manifest::Manifest;
use pushkin_core::pipeline::{check_write, WriteRequest};
use pushkin_daemon::protocol::{Request, Response, PROTOCOL_VERSION};
use pushkin_daemon::server;
use std::path::Path;

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 NONCONFORMING: &str = "export async function POST(req: Request) {\n\
  const body = await req.json();\n\
  return Response.json({ name: body.name });\n\
}\n";

const CONFORMING: &str = "import { UserCreateSchema } from \"contracts/user.zod\";\n\
export async function POST(req: Request) {\n\
  const body = UserCreateSchema.parse(await req.json());\n\
  return Response.json(body);\n\
}\n";

const HANDLER_PATH: &str = "app/api/users/route.ts";

fn manifest() -> Result<Manifest, pushkin_core::manifest::ManifestError> {
    Manifest::parse(MANIFEST)
}

fn check_request(content: &str) -> Request {
    Request::Check {
        v: PROTOCOL_VERSION,
        file_path: HANDLER_PATH.to_owned(),
        content: content.to_owned(),
    }
}

/// Serve on a background thread inside `dir` until shutdown; panics in the
/// server thread surface as test failures via the returned handle. Callers
/// parse the manifest inside `#[test]` fns (sanctioned unwrap territory).
fn spawn_daemon(
    dir: &Path,
    m: Manifest,
) -> std::thread::JoinHandle<Result<(), server::ServerError>> {
    let root = dir.to_path_buf();
    let handle = std::thread::spawn(move || server::serve(&root, m));
    // F61 — readiness is a daemon that ANSWERS, not a socket file that exists.
    //
    // The file appears at `bind()` and is only a proxy for liveness. Worse, the
    // wait it replaced fell through SILENTLY when its budget ran out, so a
    // daemon that was slow to start (or a `serve()` that returned early) showed
    // up as `expected a check response, got NotRunning` — a panic naming the
    // protocol when the real story was that nothing was listening yet. A ping
    // round trip is the actual readiness condition, and a timeout here now says
    // so in its own words.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
    let mut answered = false;
    while std::time::Instant::now() < deadline {
        if matches!(
            server::request(
                dir,
                &Request::Ping {
                    v: PROTOCOL_VERSION
                }
            ),
            Ok(Response::Pong { .. })
        ) {
            answered = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(10));
    }
    assert!(
        answered,
        "daemon never answered a ping within 30s: it failed to start, or \
         serve() returned before binding"
    );
    handle
}

fn shutdown(dir: &Path) {
    let _ = server::request(
        dir,
        &Request::Shutdown {
            v: PROTOCOL_VERSION,
        },
    );
}

#[test]
fn uds_round_trip_returns_uniform_result_json() {
    let dir = tempfile::tempdir().unwrap();
    let handle = spawn_daemon(dir.path(), manifest().unwrap());

    let warm = match server::request(dir.path(), &check_request(NONCONFORMING)) {
        Ok(Response::Check { result }) => result,
        other => panic!("expected a check response, got {other:?}"),
    };
    // The warm result IS the cold result: same rule, same decision, same
    // fix hint — serialized through the same envelope type.
    let cold = check_write(
        &manifest().unwrap(),
        &WriteRequest {
            file_path: HANDLER_PATH.to_owned(),
            content: NONCONFORMING.to_owned(),
        },
    );
    assert_eq!(warm.decision, Decision::Block);
    assert_eq!(warm.violations.len(), cold.violations.len());
    assert_eq!(warm.violations[0].rule, cold.violations[0].rule);
    assert_eq!(warm.violations[0].fix_hint, cold.violations[0].fix_hint);
    assert_eq!(warm.violations[0].line, cold.violations[0].line);

    let allow = match server::request(dir.path(), &check_request(CONFORMING)) {
        Ok(Response::Check { result }) => result,
        other => panic!("expected a check response, got {other:?}"),
    };
    assert_eq!(allow.decision, Decision::Allow);
    assert!(allow.violations.is_empty());

    shutdown(dir.path());
    let _ = handle.join();
}

#[test]
fn request_against_dead_socket_reports_not_running() {
    let dir = tempfile::tempdir().unwrap();
    // No daemon: the transport error is typed, so the shim can distinguish
    // "fall back to cold" from a protocol failure.
    let err = server::request(dir.path(), &check_request(NONCONFORMING))
        .expect_err("no daemon is listening");
    assert!(
        matches!(
            err,
            server::ServerError::NotRunning | server::ServerError::Io(_)
        ),
        "unreachable daemon must be a transport error, got {err:?}"
    );
}

#[test]
fn concurrent_shim_requests_each_get_their_own_response() {
    let dir = tempfile::tempdir().unwrap();
    let handle = spawn_daemon(dir.path(), manifest().unwrap());

    let mut workers = Vec::new();
    for i in 0..8 {
        let root = dir.path().to_path_buf();
        workers.push(std::thread::spawn(move || {
            let content = if i % 2 == 0 {
                NONCONFORMING
            } else {
                CONFORMING
            };
            let response = server::request(&root, &check_request(content));
            (i, response)
        }));
    }
    for worker in workers {
        let (i, response) = worker.join().unwrap();
        let result = match response {
            Ok(Response::Check { result }) => result,
            other => panic!("worker {i}: expected check response, got {other:?}"),
        };
        let expected = if i % 2 == 0 {
            Decision::Block
        } else {
            Decision::Allow
        };
        assert_eq!(result.decision, expected, "worker {i} got a crossed wire");
    }

    shutdown(dir.path());
    let _ = handle.join();
}

#[test]
fn ping_reports_daemon_info() {
    let dir = tempfile::tempdir().unwrap();
    let handle = spawn_daemon(dir.path(), manifest().unwrap());

    match server::request(
        dir.path(),
        &Request::Ping {
            v: PROTOCOL_VERSION,
        },
    ) {
        Ok(Response::Pong { info }) => {
            assert!(info.pid > 0);
            assert!(!info.version.is_empty());
        }
        other => panic!("expected pong, got {other:?}"),
    }

    shutdown(dir.path());
    let _ = handle.join();
}

#[test]
fn socket_lives_under_dot_pushkin_and_shutdown_removes_it() {
    let dir = tempfile::tempdir().unwrap();
    let handle = spawn_daemon(dir.path(), manifest().unwrap());

    let socket = server::socket_path(dir.path());
    assert!(
        socket.starts_with(dir.path().join(".pushkin")),
        "socket must live inside the gate-owned .pushkin/ dir"
    );
    assert!(socket.exists(), "serving daemon exposes its socket");

    shutdown(dir.path());
    let _ = handle.join();
    assert!(!socket.exists(), "shutdown must remove the socket file");
}