3use std::process::Stdio;
4use std::time::Duration;

6use bytes::Bytes;
7use http_body_util::{BodyExt, Full};
8use hyper::{Method, Request, StatusCode};
9use hyper_util::rt::TokioIo;
10use serde_json::Value;
11use tokio::net::UnixStream;
12
13use crate::paths;

16pub async fn call(method: Method, path: &str, body: Option<&Value>) -> Result<(StatusCode, Value), String> {
17    let stream = UnixStream::connect(paths::socket()).await.map_err(|e| format!("no daemon on {}: {e}", paths::socket().display()))?;
18    let (mut sender, connection) =
19        hyper::client::conn::http1::handshake(TokioIo::new(stream)).await.map_err(|e| e.to_string())?;
20    tokio::spawn(connection);
21    let body = body.map(|value| Bytes::from(value.to_string())).unwrap_or_default();
22    let request = Request::builder()
23        .method(method)
24        .uri(path)
25        .header(hyper::header::HOST, "jevhooks")
26        .header(hyper::header::CONTENT_TYPE, "application/json")
27        .body(Full::new(body))
28        .map_err(|e| e.to_string())?;
29    let response = sender.send_request(request).await.map_err(|e| e.to_string())?;
30    let status = response.status();
31    let bytes = response.into_body().collect().await.map_err(|e| e.to_string())?.to_bytes();
32    let value = serde_json::from_slice(&bytes).map_err(|e| format!("the daemon's reply was not JSON: {e}"))?;
33    Ok((status, value))
34}

37fn on_path(name: &str) -> bool {
38    std::env::var_os("PATH").is_some_and(|path| std::env::split_paths(&path).any(|dir| dir.join(name).is_file()))
39}

48fn spawn() -> Result<(), String> {
49    use std::os::unix::process::CommandExt;
50    std::fs::create_dir_all(paths::state_dir()).map_err(|e| e.to_string())?;
51    let log = std::fs::OpenOptions::new()
52        .create(true)
53        .append(true)
54        .open(paths::daemon_log())
55        .map_err(|e| format!("opening {}: {e}", paths::daemon_log().display()))?;
56    let exe = std::env::current_exe().map_err(|e| e.to_string())?;
57    let has_key = std::env::var_os(jev_http::KEY).is_some_and(|key| !key.is_empty());
58    let mut command = if !has_key && on_path("op-env-run") {
59        let mut command = std::process::Command::new("op-env-run");
60        command.arg("--").arg(&exe);
61        command
62    } else {
63        std::process::Command::new(&exe)
64    };
65    command
66        .arg("serve")
67        .stdin(Stdio::null())
68        .stdout(Stdio::null())
69        .stderr(log)
70        // Its own process group, so the session's exit does not signal it.
71        .process_group(0)
72        .spawn()
73        .map(drop)
74        .map_err(|e| format!("starting the daemon: {e}"))
75}

79pub async fn ensure() -> Result<Value, String> {
80    let mine = paths::identity();
81    if let Ok((_, status)) = call(Method::GET, "/status", None).await {
82        if status.get("identity").and_then(Value::as_str) == Some(mine.as_str()) {
83            return Ok(status);
84        }
85        // A daemon from another build: ask it to stop and wait for the socket to close.
86        call(Method::POST, "/shutdown", None).await.ok();
87        for _ in 0..50 {
88            if UnixStream::connect(paths::socket()).await.is_err() {
89                break;
90            }
91            tokio::time::sleep(Duration::from_millis(100)).await;
92        }
93    }
94    spawn()?;
95    // `op-env-run` fetching the key has been measured at 3 to 15 seconds.
96    for _ in 0..300 {
97        tokio::time::sleep(Duration::from_millis(100)).await;
98        if let Ok((_, status)) = call(Method::GET, "/status", None).await {
99            return Ok(status);
100        }
101    }
102    Err(format!("the daemon did not start within 30 s; see {}", paths::daemon_log().display()))
103}