xbp 10.40.1

//! Loopback HTTP API for worktree-watch (local dashboard + automation).
//!
//! Bound to `127.0.0.1` only. Default port `17890` (see
//! `apps/worktree-watch-app/API_CONTRACT.md` and `API_BASE` in `app.js`).
//!
//! Started by:
//! - `xbp worktree-watch tray`
//! - `xbp worktree-watch start` / `start --detach` (inside the watcher process)

use crate::commands::worktree_watch::{
    collect_worktree_watch_api_repositories, collect_worktree_watch_api_stats,
    collect_worktree_watch_api_status, run_worktree_watch_start_detached, run_worktree_watch_stop,
    run_worktree_watch_sync, WorktreeWatchStartOptions, WorktreeWatchStopOptions,
    WorktreeWatchSyncOptions, WorktreeWatchTargetOptions,
    WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES,
};
use axum::extract::{Request, State};
use axum::http::{header, HeaderValue, Method, StatusCode};
use axum::middleware::{self, Next};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::Deserialize;
use serde_json::{json, Value};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

/// Default loopback port advertised by the static dashboard (`app.js`).
pub const DEFAULT_WORKTREE_WATCH_API_PORT: u16 = 17890;

const ENV_API_PORT: &str = "XBP_WORKTREE_WATCH_API_PORT";
const ENV_API_DISABLE: &str = "XBP_WORKTREE_WATCH_API";

static API_SPAWNED: AtomicBool = AtomicBool::new(false);

#[derive(Clone)]
struct ApiState {
    target: WorktreeWatchTargetOptions,
    sync_interval_seconds: u64,
    stats_gap_minutes: u64,
    last_error: Arc<Mutex<Option<String>>>,
    /// PID of the process hosting this API (always the watcher/tray process).
    host_pid: u32,
}

#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct StartBody {
    #[serde(default)]
    sync_interval_seconds: Option<u64>,
}

#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SyncBody {
    #[serde(default)]
    dry_run: bool,
    #[serde(default)]
    resync: bool,
}

/// Resolve the listen port from env, else the default.
pub fn resolve_worktree_watch_api_port(explicit: Option<u16>) -> Option<u16> {
    if let Some(port) = explicit {
        return Some(port);
    }
    if let Ok(value) = std::env::var(ENV_API_DISABLE) {
        let lowered = value.trim().to_ascii_lowercase();
        if matches!(
            lowered.as_str(),
            "0" | "false" | "off" | "disable" | "disabled"
        ) {
            return None;
        }
    }
    if let Ok(value) = std::env::var(ENV_API_PORT) {
        let trimmed = value.trim();
        if trimmed.is_empty() {
            return Some(DEFAULT_WORKTREE_WATCH_API_PORT);
        }
        match trimmed.parse::<u16>() {
            Ok(0) => return None,
            Ok(port) => return Some(port),
            Err(_) => {
                eprintln!(
                    "warning: invalid {ENV_API_PORT}={value:?}; using {DEFAULT_WORKTREE_WATCH_API_PORT}"
                );
            }
        }
    }
    Some(DEFAULT_WORKTREE_WATCH_API_PORT)
}

/// Spawn the loopback API in a background thread (for the blocking tray loop).
///
/// Safe to call once per process; later calls are no-ops.
///
/// Uses `std::sync::mpsc` (not `tokio::sync::oneshot::blocking_recv`) so this
/// can be called from inside an existing Tokio runtime without panicking.
pub fn spawn_worktree_watch_api_in_background(
    target: WorktreeWatchTargetOptions,
    sync_interval_seconds: u64,
    port: Option<u16>,
) -> Option<u16> {
    let Some(port) = resolve_worktree_watch_api_port(port) else {
        return None;
    };
    if API_SPAWNED.swap(true, Ordering::SeqCst) {
        return Some(port);
    }

    // std mpsc: callers include async `watch_foreground` (Tokio worker) and the
    // sync tray loop. tokio oneshot::blocking_recv panics inside a runtime.
    let (ready_tx, ready_rx) = mpsc::channel::<Result<u16, String>>();
    thread::Builder::new()
        .name("xbp-worktree-watch-api".into())
        .spawn(move || {
            let runtime = match tokio::runtime::Builder::new_multi_thread()
                .enable_all()
                .worker_threads(2)
                .thread_name("xbp-ww-api")
                .build()
            {
                Ok(rt) => rt,
                Err(error) => {
                    let _ = ready_tx.send(Err(format!("API runtime: {error}")));
                    return;
                }
            };
            runtime.block_on(async move {
                match bind_and_serve(target, sync_interval_seconds, port, ready_tx).await {
                    Ok(()) => {}
                    Err(error) => {
                        tracing::warn!("worktree-watch API exited: {error}");
                    }
                }
            });
        })
        .ok()?;

    match ready_rx.recv_timeout(Duration::from_secs(10)) {
        Ok(Ok(bound)) => Some(bound),
        Ok(Err(error)) => {
            API_SPAWNED.store(false, Ordering::SeqCst);
            eprintln!("worktree-watch API: {error}");
            None
        }
        Err(_) => {
            API_SPAWNED.store(false, Ordering::SeqCst);
            None
        }
    }
}

async fn bind_and_serve(
    target: WorktreeWatchTargetOptions,
    sync_interval_seconds: u64,
    port: u16,
    ready_tx: mpsc::Sender<Result<u16, String>>,
) -> Result<(), String> {
    let addr = SocketAddr::from(([127, 0, 0, 1], port));
    let listener = match tokio::net::TcpListener::bind(addr).await {
        Ok(listener) => listener,
        Err(error) => {
            let message = format!(
                "failed to bind http://{addr} ({error}). Set {ENV_API_PORT} or free the port. Dashboard: apps/worktree-watch-app/"
            );
            let _ = ready_tx.send(Err(message.clone()));
            return Err(message);
        }
    };
    let bound = listener
        .local_addr()
        .map_err(|error| format!("Failed to read bound address: {error}"))?;
    let _ = ready_tx.send(Ok(bound.port()));

    let state = ApiState {
        target,
        sync_interval_seconds,
        stats_gap_minutes: WORKTREE_WATCH_API_DEFAULT_STATS_GAP_MINUTES,
        last_error: Arc::new(Mutex::new(None)),
        host_pid: std::process::id(),
    };

    let app = Router::new()
        .route("/api/worktree-watch/status", get(handle_status))
        .route("/api/worktree-watch/stats", get(handle_stats))
        .route("/api/worktree-watch/repositories", get(handle_repositories))
        .route("/api/worktree-watch/start", post(handle_start))
        .route("/api/worktree-watch/stop", post(handle_stop))
        .route("/api/worktree-watch/sync", post(handle_sync))
        .route("/api/worktree-watch/health", get(handle_health))
        .route("/", get(handle_root))
        .layer(middleware::from_fn(cors_middleware))
        .with_state(state);

    axum::serve(listener, app)
        .await
        .map_err(|error| format!("worktree-watch API server error: {error}"))
}

fn set_last_error(state: &ApiState, message: Option<String>) {
    if let Ok(mut guard) = state.last_error.lock() {
        *guard = message;
    }
}

fn read_last_error(state: &ApiState) -> Option<String> {
    state.last_error.lock().ok().and_then(|guard| guard.clone())
}

async fn handle_root() -> impl IntoResponse {
    Json(json!({
        "service": "xbp-worktree-watch",
        "endpoints": [
            "GET /api/worktree-watch/status",
            "GET /api/worktree-watch/stats",
            "GET /api/worktree-watch/repositories",
            "POST /api/worktree-watch/start",
            "POST /api/worktree-watch/stop",
            "POST /api/worktree-watch/sync",
            "GET /api/worktree-watch/health",
        ],
    }))
}

async fn handle_health(State(state): State<ApiState>) -> impl IntoResponse {
    Json(json!({
        "ok": true,
        "service": "xbp-worktree-watch",
        "target": describe_target(&state.target),
    }))
}

async fn handle_status(State(state): State<ApiState>) -> Response {
    let last_error = read_last_error(&state);
    let target = state.target.clone();
    let host_pid = state.host_pid;
    // Blocking disk/process work must not stall the async API runtime.
    let result = tokio::task::spawn_blocking(move || {
        collect_worktree_watch_api_status(&target, last_error.as_deref())
            .map(|body| mark_api_host_as_running(body, &target, host_pid))
    })
    .await;
    match result {
        Ok(Ok(body)) => (StatusCode::OK, Json(body)).into_response(),
        Ok(Err(error)) => error_response(StatusCode::INTERNAL_SERVER_ERROR, "status_failed", error),
        Err(error) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "status_failed",
            format!("status task failed: {error}"),
        ),
    }
}

/// The loopback API only runs inside a live watcher/tray process, so status
/// must never report "not running" while this host is serving.
fn mark_api_host_as_running(
    mut body: Value,
    target: &WorktreeWatchTargetOptions,
    host_pid: u32,
) -> Value {
    let Some(object) = body.as_object_mut() else {
        return body;
    };
    let repo_count = object
        .get("repositoryCount")
        .and_then(Value::as_u64)
        .unwrap_or(0) as usize;
    object.insert("anyRunning".to_string(), json!(true));
    object.insert("runningWatchers".to_string(), json!(repo_count.max(1)));
    object.insert("allRunning".to_string(), json!(repo_count > 0));
    if target.parent.is_some() {
        object.insert("parentWatcherRunning".to_string(), json!(true));
        object.insert("parentWatcherPid".to_string(), json!(host_pid));
    }
    if let Some(repos) = object
        .get_mut("repositories")
        .and_then(|value| value.as_array_mut())
    {
        for repo in repos {
            if let Some(repo_obj) = repo.as_object_mut() {
                repo_obj.insert("running".to_string(), json!(true));
                if repo_obj.get("pid").and_then(Value::as_u64).is_none() {
                    repo_obj.insert("pid".to_string(), json!(host_pid));
                }
            }
        }
    }
    if let Some(line) = object.get("statsLine").and_then(Value::as_str) {
        // Keep free-form line but ensure it doesn't say zero watching when live.
        let _ = line;
    }
    object.insert(
        "statsLine".to_string(),
        json!(format!(
            "{repo_count} repo(s) · {watching} watching · {unsynced} unsynced · {files} file(s)",
            watching = repo_count.max(1),
            unsynced = object
                .get("totalUnsyncedRecords")
                .and_then(Value::as_u64)
                .unwrap_or(0),
            files = object
                .get("totalUnsyncedFiles")
                .and_then(Value::as_u64)
                .unwrap_or(0),
        )),
    );
    body
}

async fn handle_stats(State(state): State<ApiState>) -> Response {
    let target = state.target.clone();
    let gap = state.stats_gap_minutes;
    let result =
        tokio::task::spawn_blocking(move || collect_worktree_watch_api_stats(&target, gap)).await;
    match result {
        Ok(Ok(body)) => (StatusCode::OK, Json(body)).into_response(),
        Ok(Err(error)) => error_response(StatusCode::INTERNAL_SERVER_ERROR, "stats_failed", error),
        Err(error) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "stats_failed",
            format!("stats task failed: {error}"),
        ),
    }
}

async fn handle_repositories(State(state): State<ApiState>) -> Response {
    let target = state.target.clone();
    let result =
        tokio::task::spawn_blocking(move || collect_worktree_watch_api_repositories(&target)).await;
    match result {
        Ok(Ok(body)) => (StatusCode::OK, Json(body)).into_response(),
        Ok(Err(error)) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "repositories_failed",
            error,
        ),
        Err(error) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "repositories_failed",
            format!("repositories task failed: {error}"),
        ),
    }
}

async fn handle_start(
    State(state): State<ApiState>,
    body: Result<Json<StartBody>, axum::extract::rejection::JsonRejection>,
) -> Response {
    let body = body.map(|Json(v)| v).unwrap_or_default();
    let sync_interval = body
        .sync_interval_seconds
        .unwrap_or(state.sync_interval_seconds);

    // Detached-only entrypoint: avoids pulling `watch_foreground` into this
    // async type graph (which would cycle with the API server future).
    let result = run_worktree_watch_start_detached(WorktreeWatchStartOptions {
        target: state.target.clone(),
        detach: true,
        sync_interval_seconds: sync_interval,
        once: false,
    })
    .map(|_| "Started detached worktree watcher.".to_string());

    match result {
        Ok(message) => {
            set_last_error(&state, None);
            let status = collect_worktree_watch_api_status(&state.target, None).ok();
            let any_running = status
                .as_ref()
                .and_then(|v| v.get("anyRunning"))
                .and_then(Value::as_bool)
                .unwrap_or(true);
            (
                StatusCode::OK,
                Json(json!({
                    "ok": true,
                    "message": message,
                    "anyRunning": any_running,
                    "status": status,
                })),
            )
                .into_response()
        }
        Err(error) => {
            set_last_error(&state, Some(error.clone()));
            error_response(StatusCode::INTERNAL_SERVER_ERROR, "start_failed", error)
        }
    }
}

async fn handle_stop(
    State(state): State<ApiState>,
    _body: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
) -> Response {
    let target = state.target.clone();
    let result = if !target.repos.is_empty() && target.parent.is_none() {
        for repo in &target.repos {
            if let Err(error) = run_worktree_watch_stop(WorktreeWatchStopOptions {
                target: WorktreeWatchTargetOptions {
                    repo: Some(repo.clone()),
                    parent: None,
                    repos: Vec::new(),
                },
                force: false,
            }) {
                set_last_error(&state, Some(error.clone()));
                return error_response(StatusCode::INTERNAL_SERVER_ERROR, "stop_failed", error);
            }
        }
        Ok("Stop requested for configured repo(s).".to_string())
    } else {
        run_worktree_watch_stop(WorktreeWatchStopOptions {
            target,
            force: false,
        })
        .map(|_| "Stop requested.".to_string())
    };

    match result {
        Ok(message) => {
            set_last_error(&state, None);
            let status = collect_worktree_watch_api_status(&state.target, None).ok();
            let any_running = status
                .as_ref()
                .and_then(|v| v.get("anyRunning"))
                .and_then(Value::as_bool)
                .unwrap_or(false);
            (
                StatusCode::OK,
                Json(json!({
                    "ok": true,
                    "message": message,
                    "anyRunning": any_running,
                    "status": status,
                })),
            )
                .into_response()
        }
        Err(error) => {
            set_last_error(&state, Some(error.clone()));
            error_response(StatusCode::INTERNAL_SERVER_ERROR, "stop_failed", error)
        }
    }
}

async fn handle_sync(
    State(state): State<ApiState>,
    body: Result<Json<SyncBody>, axum::extract::rejection::JsonRejection>,
) -> Response {
    let body = body.map(|Json(v)| v).unwrap_or_default();
    match run_worktree_watch_sync(WorktreeWatchSyncOptions {
        target: state.target.clone(),
        dry_run: body.dry_run,
        resync: body.resync,
    })
    .await
    {
        Ok(()) => {
            set_last_error(&state, None);
            let status = collect_worktree_watch_api_status(&state.target, None).ok();
            let unsynced_files = status
                .as_ref()
                .and_then(|v| {
                    v.get("totalUnsyncedFiles")
                        .or_else(|| v.get("unsyncedFiles"))
                })
                .and_then(Value::as_u64)
                .unwrap_or(0);
            let unsynced_records = status
                .as_ref()
                .and_then(|v| {
                    v.get("totalUnsyncedRecords")
                        .or_else(|| v.get("unsyncedRecords"))
                })
                .and_then(Value::as_u64)
                .unwrap_or(0);
            let last_sync = status.as_ref().and_then(|v| {
                v.get("repositories")
                    .and_then(Value::as_array)
                    .and_then(|repos| repos.first())
                    .and_then(|repo| repo.get("lastSync"))
                    .cloned()
            });
            (
                StatusCode::OK,
                Json(json!({
                    "ok": true,
                    "message": if body.dry_run {
                        "Dry-run sync complete"
                    } else {
                        "Sync complete"
                    },
                    "unsyncedFiles": unsynced_files,
                    "unsyncedRecords": unsynced_records,
                    "lastSync": last_sync,
                    "status": status,
                })),
            )
                .into_response()
        }
        Err(error) => {
            set_last_error(&state, Some(error.clone()));
            error_response(StatusCode::INTERNAL_SERVER_ERROR, "sync_failed", error)
        }
    }
}

fn error_response(status: StatusCode, error: &str, message: String) -> Response {
    (
        status,
        Json(json!({
            "ok": false,
            "error": error,
            "message": message,
        })),
    )
        .into_response()
}

fn describe_target(target: &WorktreeWatchTargetOptions) -> String {
    if let Some(parent) = target.parent.as_ref() {
        return format!("parent:{}", parent.display());
    }
    if !target.repos.is_empty() {
        return format!("{} repo(s)", target.repos.len());
    }
    if let Some(repo) = target.repo.as_ref() {
        return repo.display().to_string();
    }
    "current repo".to_string()
}

async fn cors_middleware(req: Request, next: Next) -> Response {
    if req.method() == Method::OPTIONS {
        let mut response = StatusCode::NO_CONTENT.into_response();
        insert_cors_headers(response.headers_mut());
        return response;
    }

    let mut response = next.run(req).await;
    insert_cors_headers(response.headers_mut());
    response
}

fn insert_cors_headers(headers: &mut axum::http::HeaderMap) {
    headers.insert(
        header::ACCESS_CONTROL_ALLOW_ORIGIN,
        HeaderValue::from_static("*"),
    );
    headers.insert(
        header::ACCESS_CONTROL_ALLOW_METHODS,
        HeaderValue::from_static("GET, POST, OPTIONS"),
    );
    headers.insert(
        header::ACCESS_CONTROL_ALLOW_HEADERS,
        HeaderValue::from_static("Content-Type, Accept"),
    );
    headers.insert(
        header::ACCESS_CONTROL_MAX_AGE,
        HeaderValue::from_static("86400"),
    );
}

/// Log a one-line URL for operators (and the static dashboard).
pub fn print_api_listen_hint(port: u16) {
    println!(
        "Worktree-watch local API: http://127.0.0.1:{port}  (dashboard: apps/worktree-watch-app/)"
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{Read, Write};
    use std::net::TcpStream;
    use std::sync::Mutex;

    /// Serialize spawn tests: `API_SPAWNED` is process-global.
    static SPAWN_TEST_LOCK: Mutex<()> = Mutex::new(());

    fn empty_target() -> WorktreeWatchTargetOptions {
        WorktreeWatchTargetOptions {
            repo: None,
            parent: None,
            repos: Vec::new(),
        }
    }

    fn reset_api_spawned_for_test() {
        API_SPAWNED.store(false, Ordering::SeqCst);
    }

    /// Minimal loopback GET; avoids async HTTP clients in unit tests.
    fn http_get_loopback(port: u16, path: &str) -> String {
        let mut stream = TcpStream::connect(("127.0.0.1", port))
            .unwrap_or_else(|e| panic!("connect 127.0.0.1:{port}: {e}"));
        stream
            .set_read_timeout(Some(Duration::from_secs(3)))
            .expect("read timeout");
        stream
            .set_write_timeout(Some(Duration::from_secs(3)))
            .expect("write timeout");
        let request =
            format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nConnection: close\r\n\r\n");
        stream
            .write_all(request.as_bytes())
            .unwrap_or_else(|e| panic!("write request: {e}"));
        let mut body = String::new();
        stream
            .read_to_string(&mut body)
            .unwrap_or_else(|e| panic!("read response: {e}"));
        body
    }

    #[test]
    fn default_port_constant() {
        assert_eq!(DEFAULT_WORKTREE_WATCH_API_PORT, 17890);
    }

    #[test]
    fn resolve_port_honors_explicit_value() {
        assert_eq!(resolve_worktree_watch_api_port(Some(12345)), Some(12345));
        // Explicit 0 is allowed so callers can request an OS-assigned ephemeral port.
        assert_eq!(resolve_worktree_watch_api_port(Some(0)), Some(0));
    }

    /// Regression: `tokio::sync::oneshot::Receiver::blocking_recv` panics with
    /// "Cannot block the current thread from within a runtime" when
    /// `spawn_worktree_watch_api_in_background` is called from
    /// `watch_foreground` / `watch_parent_foreground` (async on the CLI Tokio
    /// runtime). Ready-wait must use `std::sync::mpsc` instead.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn spawn_api_from_inside_tokio_runtime_does_not_panic() {
        let _guard = SPAWN_TEST_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        reset_api_spawned_for_test();

        // Must not panic when invoked on a Tokio worker (the production path).
        let bound = spawn_worktree_watch_api_in_background(empty_target(), 60, Some(0));
        let port = bound.expect("API should bind on an ephemeral port");
        assert_ne!(port, 0, "bound port must be OS-assigned, not 0");

        let response = http_get_loopback(port, "/api/worktree-watch/health");
        assert!(
            response.contains("200") || response.contains("\"ok\""),
            "health response unexpected: {response}"
        );
        assert!(
            response.contains("xbp-worktree-watch") || response.contains("ok"),
            "health body unexpected: {response}"
        );
    }

    /// Same ready-channel path as tray (`run_worktree_watch_tray`), which is
    /// sync and must keep working after the oneshot→mpsc switch.
    #[test]
    fn spawn_api_from_sync_context_binds_and_serves_health() {
        let _guard = SPAWN_TEST_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        reset_api_spawned_for_test();

        let bound = spawn_worktree_watch_api_in_background(empty_target(), 120, Some(0));
        let port = bound.expect("API should bind from a plain OS thread");
        assert_ne!(port, 0);

        let root = http_get_loopback(port, "/");
        assert!(
            root.contains("worktree-watch") || root.contains("endpoints"),
            "root JSON unexpected: {root}"
        );
    }

    /// Second spawn in the same process is a no-op (idempotent).
    #[test]
    fn spawn_api_is_idempotent_after_first_success() {
        let _guard = SPAWN_TEST_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        reset_api_spawned_for_test();

        let first = spawn_worktree_watch_api_in_background(empty_target(), 60, Some(0))
            .expect("first spawn");
        assert_ne!(first, 0);

        // Requested port is returned when already spawned (not re-bound).
        let second = spawn_worktree_watch_api_in_background(empty_target(), 60, Some(9999));
        assert_eq!(second, Some(9999));
    }

    /// Documents the failure mode the mpsc ready-wait replaces: calling
    /// `oneshot::Receiver::blocking_recv` from a Tokio worker panics.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn tokio_oneshot_blocking_recv_panics_inside_runtime() {
        let (tx, rx) = tokio::sync::oneshot::channel::<u16>();
        std::thread::spawn(move || {
            let _ = tx.send(1);
        });

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _ = rx.blocking_recv();
        }));
        assert!(
            result.is_err(),
            "expected oneshot blocking_recv to panic inside a Tokio runtime"
        );
        let panic_payload = result.unwrap_err();
        let message = panic_payload
            .downcast_ref::<String>()
            .map(|s| s.as_str())
            .or_else(|| panic_payload.downcast_ref::<&str>().copied())
            .unwrap_or("");
        assert!(
            message.contains("Cannot block the current thread from within a runtime")
                || message.contains("block the current thread"),
            "unexpected panic message: {message:?}"
        );
    }
}