hyperhive/hive-c0re/src/dashboard/build_logs.rs

//! Build-log endpoints for the dashboard.
//!
//! Header lists (all-agents + per-agent), the full row by id, a `text/plain`
//! download, and an SSE stream that delivers incremental stdout/stderr while
//! a build runs (closing once it finishes or the row is vacuum-reaped).

use std::convert::Infallible;

use axum::{
    extract::{Path as AxumPath, State},
    http::StatusCode,
    response::{
        IntoResponse, Response,
        sse::{Event, KeepAlive, Sse},
    },
};
use serde::{Deserialize, Serialize};
use tokio_stream::Stream;
use tokio_stream::wrappers::ReceiverStream;
use utoipa::IntoParams;

use super::{AppState, Ident, error_response};
use crate::build_logs::{BuildLogFull, BuildLogHeader};

#[derive(Deserialize, IntoParams)]
pub(super) struct BuildLogsAllQuery {
    /// Max rows to return. Capped at 100. Default 30.
    #[serde(default)]
    limit: Option<usize>,
}

/// Most-recent build log headers across
/// all agents, newest first.
///
/// Same JSON shape as the per-agent endpoint.
#[utoipa::path(
    get,
    path = "/api/build-logs",
    params(BuildLogsAllQuery),
    responses(
        (status = 200, description = "recent build log headers, newest first", body = Vec<BuildLogHeader>),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_logs_all(
    State(state): State<AppState>,
    axum::extract::Query(q): axum::extract::Query<BuildLogsAllQuery>,
) -> Response {
    let limit = q.limit.unwrap_or(30);
    match state.coord.build_logs.list_recent_all(limit) {
        Ok(rows) => axum::Json(rows).into_response(),
        Err(e) => error_response(&format!("build-logs all: {e:#}")),
    }
}

#[derive(Deserialize, IntoParams)]
pub(super) struct BuildLogsQuery {
    /// Maximum number of rows to return. Capped server-side at 50
    /// (see `build_logs::list_recent_for_agent`). Default 10.
    #[serde(default)]
    limit: Option<usize>,
}

/// Most-recent build log
/// headers for one agent, newest first.
///
/// Returns `Vec<BuildLogHeader>` (JSON). Backs the per-agent log chip
/// in the agent card and the side-panel header list.
#[utoipa::path(
    get,
    path = "/api/build-logs/{agent}",
    params(
        ("agent" = String, Path, description = "agent name"),
        BuildLogsQuery,
    ),
    responses(
        (status = 200, description = "recent build log headers for the agent, newest first", body = Vec<BuildLogHeader>),
        (status = 400, description = "bad agent name"),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_logs_agent(
    State(state): State<AppState>,
    AxumPath(name): AxumPath<String>,
    axum::extract::Query(q): axum::extract::Query<BuildLogsQuery>,
) -> Response {
    let name = match Ident::parse(&name) {
        Ok(n) => n,
        Err(reason) => {
            return (StatusCode::BAD_REQUEST, format!("bad agent name: {reason}")).into_response();
        }
    };
    let limit = q.limit.unwrap_or(10);
    match state
        .coord
        .build_logs
        .list_recent_for_agent(name.as_str(), limit)
    {
        Ok(rows) => axum::Json(rows).into_response(),
        Err(e) => error_response(&format!("build-logs {name}: {e:#}")),
    }
}

/// Full build log row (stdout +
/// stderr concatenated) by id.
///
/// Returns `BuildLogFull` (JSON), or HTTP 404 when the id doesn't
/// exist (vacuum-reaped, or the operator passed a stale id from a
/// refresh race).
#[utoipa::path(
    get,
    path = "/api/build-logs/id/{id}",
    params(("id" = i64, Path, description = "build log row id")),
    responses(
        (status = 200, description = "full build log row", body = BuildLogFull),
        (status = 404, description = "no such build log row"),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_log_full(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
) -> Response {
    match state.coord.build_logs.get_full(id) {
        Ok(Some(log)) => axum::Json(log).into_response(),
        Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
        Err(e) => error_response(&format!("build-log {id}: {e:#}")),
    }
}

/// The build log for a **queue node**,
/// resolved node id → log-row id → full log.
///
/// Same `BuildLogFull` JSON (`stdout` / `stderr` + header) as
/// `get_build_log_full`; HTTP 404 when the node has no linked log (the
/// client gates the request on the wire node's `build_log_id`, but a vacuum
/// race can still 404). This is the on-demand live-log-panel fetch,
/// distinct from the `build_log_id` on the wire — that id is for
/// deep-linking to the BUILD L0GS tab's full history view, not for
/// fetching the log content itself.
#[utoipa::path(
    get,
    path = "/api/build-log/{node_id}",
    params(("node_id" = u64, Path, description = "job-queue node id")),
    responses(
        (status = 200, description = "full build log row for the node's linked log", body = BuildLogFull),
        (status = 404, description = "node has no linked build log, or the log row is gone"),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_log_for_node(
    State(state): State<AppState>,
    AxumPath(node_id): AxumPath<u64>,
) -> Response {
    match state.coord.build_logs.id_for_node(node_id) {
        Some(log_id) => get_build_log_full(State(state), AxumPath(log_id)).await,
        None => (
            StatusCode::NOT_FOUND,
            format!("node #{node_id} has no build log"),
        )
            .into_response(),
    }
}

/// The node's build log as `text/plain`
/// for download (delegates to `get_build_log_raw` after resolving the node id).
#[utoipa::path(
    get,
    path = "/api/build-log/{node_id}/raw",
    params(("node_id" = u64, Path, description = "job-queue node id")),
    responses(
        (status = 200, description = "build log text for download", body = String, content_type = "text/plain"),
        (status = 404, description = "node has no linked build log, or the log row is gone"),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_log_raw_for_node(
    State(state): State<AppState>,
    AxumPath(node_id): AxumPath<u64>,
) -> Response {
    match state.coord.build_logs.id_for_node(node_id) {
        Some(log_id) => get_build_log_raw(State(state), AxumPath(log_id)).await,
        None => (
            StatusCode::NOT_FOUND,
            format!("node #{node_id} has no build log"),
        )
            .into_response(),
    }
}

/// JSON frame sent on the `/api/build-logs/id/{id}/stream` SSE channel.
/// `stdout_append` / `stderr_append` carry only the new bytes since the
/// last frame; `done = true` means the build finished and the stream
/// will close after this frame.
#[derive(Serialize)]
struct BuildLogFrame {
    stdout_append: String,
    stderr_append: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    status: Option<String>,
    done: bool,
}

/// SSE stream that delivers
/// incremental stdout/stderr as a build runs.
///
/// The client connects when it opens a running-build panel; the
/// stream closes automatically once the build finishes (or the row
/// disappears due to a vacuum). Each frame is a JSON-serialised
/// `BuildLogFrame`. The first frame always carries the full
/// accumulated log so far (cursors start at 0); subsequent frames
/// carry only new bytes. `done: true` on the final frame signals the
/// browser to close the `EventSource`.
#[utoipa::path(
    get,
    path = "/api/build-logs/id/{id}/stream",
    params(("id" = i64, Path, description = "build log row id")),
    responses(
        (status = 200, description = "server-sent event stream; each event's \
            `data` is a JSON-serialised `BuildLogFrame` \
            (stdout_append/stderr_append/status/done)",
            body = String, content_type = "text/event-stream"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_log_stream(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(32);
    let logs = state.coord.build_logs.clone();

    tokio::spawn(async move {
        let mut notify_rx = logs.subscribe_notifications();
        let mut stdout_cursor = 0usize;
        let mut stderr_cursor = 0usize;

        // ── initial snapshot ──────────────────────────────────────────
        match logs.get_progress(id, stdout_cursor, stderr_cursor) {
            Ok(Some(prog)) => {
                stdout_cursor += prog.stdout_append.len();
                stderr_cursor += prog.stderr_append.len();
                let done = prog.finished_at.is_some();
                if let Ok(json) = serde_json::to_string(&BuildLogFrame {
                    stdout_append: prog.stdout_append,
                    stderr_append: prog.stderr_append,
                    status: prog.status,
                    done,
                }) {
                    let _ = tx.send(Ok(Event::default().data(json))).await;
                }
                if done {
                    return;
                }
            }
            Ok(None) => {
                // Row missing — send a single error event and exit.
                let _ = tx
                    .send(Ok(Event::default()
                        .event("error")
                        .data(format!("build log #{id} not found"))))
                    .await;
                return;
            }
            Err(e) => {
                let _ = tx
                    .send(Ok(Event::default()
                        .event("error")
                        .data(format!("build log #{id}: {e:#}"))))
                    .await;
                return;
            }
        }

        // ── live delta loop ───────────────────────────────────────────
        loop {
            match notify_rx.recv().await {
                // Notification for a different build — ignore and wait
                // for the next one.
                Ok(notif_id) if notif_id != id => {}
                Ok(_) => {
                    match logs.get_progress(id, stdout_cursor, stderr_cursor) {
                        Ok(Some(prog)) => {
                            stdout_cursor += prog.stdout_append.len();
                            stderr_cursor += prog.stderr_append.len();
                            let done = prog.finished_at.is_some();
                            if let Ok(json) = serde_json::to_string(&BuildLogFrame {
                                stdout_append: prog.stdout_append,
                                stderr_append: prog.stderr_append,
                                status: prog.status,
                                done,
                            }) && tx.send(Ok(Event::default().data(json))).await.is_err()
                            {
                                return; // browser disconnected
                            }
                            if done {
                                return;
                            }
                        }
                        Ok(None) | Err(_) => return, // vacuum reaped row / channel closed
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
                Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
            }
        }
    });

    Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::default())
}

/// Full log as `text/plain` for
/// download.
///
/// Stdout and stderr are concatenated with a `--- stderr ---`
/// separator (same layout the JS side-panel renders). The
/// `Content-Disposition` header triggers a browser download with a
/// descriptive filename so the operator can save and share the log.
#[utoipa::path(
    get,
    path = "/api/build-logs/id/{id}/raw",
    params(("id" = i64, Path, description = "build log row id")),
    responses(
        (status = 200, description = "build log text for download", body = String, content_type = "text/plain"),
        (status = 404, description = "no such build log row"),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "build_logs"
)]
pub(super) async fn get_build_log_raw(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
) -> Response {
    match state.coord.build_logs.get_full(id) {
        Ok(Some(log)) => {
            let mut text = log.stdout;
            if !log.stderr.is_empty() {
                text.push_str("\n--- stderr ---\n");
                text.push_str(&log.stderr);
            }
            (
                StatusCode::OK,
                [
                    ("content-type", "text/plain; charset=utf-8".to_string()),
                    (
                        "content-disposition",
                        format!(
                            "attachment; filename=\"build-log-{}-{}.txt\"",
                            log.header.agent, id
                        ),
                    ),
                ],
                text,
            )
                .into_response()
        }
        Ok(None) => (StatusCode::NOT_FOUND, format!("build log #{id} not found")).into_response(),
        Err(e) => error_response(&format!("build-log {id}: {e:#}")),
    }
}