huskies/server/src/gateway/mod.rs

//! Multi-project gateway — entrypoint wiring and route tree.
//!
//! When `huskies --gateway` is used, the server starts in gateway mode.
//! Business logic lives in `service::gateway`, HTTP handlers in `http::gateway`.
//! This file contains only the `run` entrypoint and `build_gateway_route` wiring.

/// Gateway rebuild — builds the new binary and launches the detached trampoline.
pub mod rebuild;

use crate::http::gateway::*;
use crate::rebuild::ShutdownReason;
use crate::service::gateway::{self, GatewayState};
use poem::EndpointExt;
use std::path::Path;
use std::sync::Arc;

// Re-export public types that callers reference as `crate::gateway::*`.
pub use crate::service::gateway::{
    GatewayConfig, GatewayState as GatewayStateType, GatewayStatusEvent, ProjectEntry,
    broadcast_status_event, fetch_all_project_pipeline_statuses, format_aggregate_status_compact,
    spawn_gateway_broadcaster_forwarder, spawn_gateway_notification_poller,
    subscribe_status_events,
};

/// Build the complete gateway route tree.
///
/// Extracted from `run` so that tests can construct the full route tree and
/// catch duplicate-route panics before they reach production.
pub fn build_gateway_route(state_arc: Arc<GatewayState>) -> impl poem::Endpoint {
    poem::Route::new()
        .at("/bot-config", poem::get(gateway_bot_config_page_handler))
        .at("/api/gateway", poem::get(gateway_api_handler))
        .at(
            "/api/gateway/projects",
            poem::post(gateway_add_project_handler),
        )
        .at(
            "/api/gateway/projects/:name",
            poem::delete(gateway_remove_project_handler),
        )
        .at(
            "/api/gateway/bot-config",
            poem::get(gateway_bot_config_get_handler).post(gateway_bot_config_save_handler),
        )
        .at(
            "/mcp",
            poem::post(gateway_mcp_post_handler).get(gateway_mcp_get_handler),
        )
        // Agent join endpoints.
        .at("/gateway/mode", poem::get(gateway_mode_handler))
        .at(
            "/gateway/tokens",
            poem::post(gateway_generate_token_handler),
        )
        .at(
            "/gateway/events/push",
            poem::get(gateway_event_push_handler),
        )
        // Agent registration via CRDT-sync WebSocket.
        .at("/crdt-sync", poem::get(gateway_crdt_sync_handler))
        // Sled uplink: permission-forwarding WebSocket from sleds to gateway.
        .at("/api/sled-uplink", poem::get(gateway_sled_uplink_handler))
        // Agent management REST endpoints.
        .at(
            "/gateway/agents/:id/assign",
            poem::post(gateway_assign_agent_handler),
        )
        // Binary self-update: serve the gateway binary so sleds can download it.
        .at(
            "/api/huskies-binary",
            poem::get(crate::http::serve_binary_handler),
        )
        .data(state_arc)
}

/// Start the gateway HTTP server. This is the entry point when `--gateway` is used.
pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> {
    // Enforce one-active-gateway invariant: acquire an exclusive flock on the
    // pidfile before doing anything else.  A second gateway start while one is
    // running will fail here with a clear error.  The flock is held for the
    // lifetime of `_pidfile_guard`; it is released automatically when this
    // process exits, allowing the next gateway (spawned by the trampoline) to
    // acquire it.
    let _pidfile_guard =
        crate::pidfile::acquire_gateway_pidfile().map_err(std::io::Error::other)?;

    let config_dir = config_path
        .parent()
        .unwrap_or(std::path::Path::new("."))
        .to_path_buf();

    let config = gateway::io::load_config(config_path).map_err(std::io::Error::other)?;

    // Initialise the CRDT so gateway_config.active_project is persisted across restarts.
    let crdt_db = config_dir.join("gateway.db");
    if let Err(e) = crate::crdt_state::init(&crdt_db).await {
        crate::slog!(
            "[gateway] Warning: CRDT init failed ({e}); active-project selection will not persist"
        );
    }

    let state =
        GatewayState::new(config, config_dir.clone(), port).map_err(std::io::Error::other)?;
    let state_arc = Arc::new(state);

    let active = state_arc.active_project.read().await.clone();
    crate::slog!("[gateway] Starting gateway on port {port}, active project: {active}");
    crate::slog!(
        "[gateway] Registered projects: {}",
        state_arc
            .projects
            .read()
            .await
            .keys()
            .cloned()
            .collect::<Vec<_>>()
            .join(", ")
    );

    // Write `.mcp.json` so that the gateway's bot connects to this gateway's MCP endpoint.
    if let Err(e) = gateway::io::write_gateway_mcp_json(&config_dir, port) {
        crate::slog!("[gateway] Warning: could not write .mcp.json: {e}");
    }

    // Spawn the Matrix bot if `.huskies/bot.toml` exists in the config directory.
    let (bot_abort, bot_shutdown_tx) = gateway::io::spawn_gateway_bot(
        &config_dir,
        Arc::clone(&state_arc.active_project),
        Arc::clone(&state_arc.projects),
        port,
        Some(state_arc.event_tx.clone()),
        Arc::clone(&state_arc.perm_rx),
    );
    *state_arc.bot_handle.lock().await = bot_abort;
    *state_arc.bot_shutdown_tx.lock().await = Some(bot_shutdown_tx);

    let route = build_gateway_route(state_arc.clone());

    let host = std::env::var("HUSKIES_HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
    let addr = format!("{host}:{port}");
    crate::slog!("[gateway] Listening on {addr}");

    let result = poem::Server::new(poem::listener::TcpListener::bind(&addr))
        .run(route)
        .await;

    // Best-effort shutdown notification: signal the Matrix bot so it can post
    // "going offline" before the process exits.  Mirror of main.rs:346.
    {
        let guard = state_arc.bot_shutdown_tx.lock().await;
        if let Some(tx) = guard.as_ref() {
            let _ = tx.send(Some(ShutdownReason::Manual));
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(1500)).await;

    result
}

#[cfg(test)]
mod tests;