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

//! Scheduled-prompt + rebuild-queue endpoints for the dashboard.
//!
//! Operator-direct schedule CRUD (`/api/schedules` GET/POST, `{id}` PATCH,
//! `{id}/cancel` + `{id}/fire-now`) — the operator click *is* the approval,
//! so these skip the manager approval gate. Also the co-located
//! `/api/rebuild-queue/{id}/cancel` endpoint.

use axum::{
    extract::{Path as AxumPath, State},
    http::StatusCode,
    response::{IntoResponse, Response},
};

use problem_details::ProblemDetails;

use crate::scheduled_prompts::ScheduleNotFoundOrCancelled;
use crate::scheduled_prompts_worker::FireNowReport;

use super::{AppState, error_problem, error_response};

/// Snapshot of every schedule for the
/// scheduled-prompts tab.
///
/// Returns the wire shape directly so the frontend can render
/// without an extra translation layer.
// `hive_sh4re::schedule::WireSchedule` (the actual body) has no `ToSchema` — adding
// one would pull `utoipa` into the wire-types crate for a single dashboard
// endpoint. `serde_json::Value` placeholder; see the batch report.
#[utoipa::path(
    get,
    path = "/api/schedules",
    responses(
        (status = 200, description = "every schedule, wire shape", body = Vec<serde_json::Value>),
        (status = 500, description = "sqlite read failed"),
    ),
    tag = "schedules"
)]
pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
    match state.coord.scheduled_prompts.list() {
        Ok(rows) => {
            let mut wire: Vec<hive_sh4re::schedule::WireSchedule> = rows
                .into_iter()
                .map(crate::socket_server::schedule_to_wire_public)
                .collect();
            // Drop ghost targets (agents that no longer exist) so the
            // table never shows dead columns. Uses the reliable async
            // roster snapshot here on the request path.
            let live: std::collections::HashSet<String> = state
                .coord
                .containers_snapshot()
                .await
                .into_iter()
                .map(|c| c.name)
                .collect();
            crate::socket_server::filter_ghost_schedule_targets(&mut wire, &live);
            axum::Json(wire).into_response()
        }
        Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
    }
}

/// Operator-direct schedule creation
/// (mara: "user can add them manually").
///
/// Accepts the same `SchedulePromptPayload` shape as the manager
/// request flow but skips the approval gate — the operator click
/// *is* the approval. The schedule lands directly with
/// `source = Operator` and the worker picks it up at fire time.
#[utoipa::path(
    post,
    path = "/api/schedules",
    // `hive_sh4re::manager::SchedulePromptPayload` (the actual body) has no `ToSchema` —
    // same reasoning as the `Vec<serde_json::Value>` placeholder on
    // `api_schedules` above.
    request_body(content = serde_json::Value, description = "SchedulePromptPayload wire shape"),
    responses(
        (status = 200, description = "created; body carries the new row id", body = serde_json::Value),
        (status = 400, description = "no targets, empty body, or interval_seconds == 0"),
        (status = 500, description = "submit failed"),
    ),
    tag = "schedules"
)]
pub(super) async fn post_schedule_new(
    State(state): State<AppState>,
    axum::Json(payload): axum::Json<hive_sh4re::manager::SchedulePromptPayload>,
) -> Result<Response, ProblemDetails> {
    if payload.targets.is_empty() {
        return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST)
            .with_detail("schedule must have at least one target"));
    }
    if payload.body.trim().is_empty() {
        return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST)
            .with_detail("schedule body must be non-empty"));
    }
    if let Some(0) = payload.interval_seconds {
        return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST)
            .with_detail("interval_seconds must be > 0 (use None for one-shot)"));
    }
    let new = crate::scheduled_prompts::NewSchedule {
        owner: hive_sh4re::manager::OPERATOR_RECIPIENT.to_owned(),
        targets: payload.targets,
        body: payload.body,
        first_fire_at_unix: payload.first_fire_at_unix,
        interval_seconds: payload.interval_seconds,
        description: payload.description,
        source: crate::scheduled_prompts::ScheduleSource::Operator,
    };
    match state.coord.scheduled_prompts.submit(&new) {
        Ok(id) => {
            state.coord.emit_schedules_snapshot();
            Ok(axum::Json(serde_json::json!({"id": id})).into_response())
        }
        Err(e) => Err(error_problem(&format!("schedule submit: {e:#}"))),
    }
}

/// Optional JSON body for `fire-now`. Absent / empty body ⇒
/// `reset_timer = false` (back-compat: cadence stays intact).
#[derive(serde::Deserialize, Default, utoipa::ToSchema)]
pub(super) struct FireNowBody {
    #[serde(default)]
    reset_timer: bool,
}

/// Operator-initiated
/// out-of-band fire of a scheduled prompt.
///
/// Runs the per-target fan-out once immediately and reports
/// per-target outcome counts. One-shot schedules are consumed
/// (cancelled) by a manual fire — the operator's intent is "send
/// this now, the scheduled time was wrong." For recurring schedules
/// the cadence stays intact unless the body carries
/// `{"reset_timer": true}`, in which case the countdown is re-armed
/// from now (`next_fire_at = now + interval`).
#[utoipa::path(
    post,
    path = "/api/schedules/{id}/fire-now",
    params(("id" = i64, Path, description = "schedule row id")),
    request_body(content = FireNowBody, description = "optional; absent body means reset_timer = false"),
    responses(
        (status = 200, description = "fired; per-target outcome counts", body = FireNowReport),
        (status = 500, description = "schedule missing, cancelled, or fully drained"),
    ),
    tag = "schedules"
)]
pub(super) async fn post_schedule_fire_now(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
    body: Option<axum::Json<FireNowBody>>,
) -> Response {
    let reset_timer = body.is_some_and(|axum::Json(b)| b.reset_timer);
    match crate::scheduled_prompts_worker::fire_now(&state.coord, id, reset_timer).await {
        Ok(report) => {
            state.coord.emit_schedules_snapshot();
            axum::Json(report).into_response()
        }
        Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")),
    }
}

/// Drop still-queued work from the job
/// queue.
///
/// `id` is a **node** id. A DAG's root cancels the whole group (the scheduler
/// cascades to pending descendants), which is what the dashboard's cancel
/// button sends today — a DAG id *is* its root node's id. An interior node
/// cancels just that branch.
///
/// Refuses `Running` / terminal nodes: an
/// in-flight node owns the agent's nix store + nixos-container update
/// lock and can't be safely interrupted from the queue side. Always
/// returns 200; the body is `{"cancelled": true}` on a successful
/// flip to Cancelled, `{"cancelled": false}` when the DAG was
/// Running / terminal / gone. On success a fresh `RebuildQueueChanged`
/// snapshot fires so the state flip surfaces live.
#[utoipa::path(
    post,
    path = "/api/rebuild-queue/{id}/cancel",
    params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")),
    responses((status = 200, description = "whether the DAG was cancelled", body = serde_json::Value)),
    tag = "schedules"
)]
pub(super) async fn post_rebuild_queue_cancel(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<u64>,
) -> Response {
    if state.coord.job_queue.cancel(id) {
        // Any terminal side effect is the DAG's own spared tail node, which the
        // scheduler picks up on its next pass — nothing to fire from here.
        state.coord.emit_rebuild_queue_snapshot();
        axum::Json(serde_json::json!({"cancelled": true})).into_response()
    } else {
        axum::Json(serde_json::json!({"cancelled": false})).into_response()
    }
}

#[derive(serde::Deserialize, Default, utoipa::ToSchema)]
pub(super) struct CancelScheduleForm {
    /// `None` / absent / empty array → cancel whole schedule.
    #[serde(default)]
    targets: Option<Vec<String>>,
}

#[derive(serde::Deserialize, Default, utoipa::ToSchema)]
#[allow(
    clippy::option_option,
    reason = "double-Option carries three-state PATCH semantics on the wire \
              (missing key = leave alone, JSON null = clear, value = set); \
              collapsing to a single Option would lose the 'clear' state"
)]
pub(super) struct EditScheduleForm {
    #[serde(default)]
    body: Option<String>,
    /// Double-`Option` semantics on the wire: missing key = leave
    /// alone, explicit `null` = clear, value = set. serde's
    /// `deserialize_with` trick to distinguish missing from null:
    /// we wrap each editable field in its own helper. Simpler
    /// here — keep them plain `Option<Option<_>>` and document
    /// that the dashboard caller passes JSON `null` to clear.
    #[serde(default, deserialize_with = "deserialize_some")]
    description: Option<Option<String>>,
    #[serde(default, deserialize_with = "deserialize_some")]
    interval_seconds: Option<Option<u64>>,
    #[serde(default)]
    next_fire_at_unix: Option<i64>,
    /// New targets to add. Replace-on-conflict: re-adding a
    /// previously-cancelled target drops the tombstone and the
    /// target starts fresh (operator intent on re-add = "this
    /// target is active again, fresh start").
    #[serde(default)]
    targets_add: Option<Vec<String>>,
    /// Targets to cancel. Same path as `cancel_targets`:
    /// tombstones preserve audit and the parent schedule
    /// auto-cancels when no active targets remain.
    #[serde(default)]
    targets_remove: Option<Vec<String>>,
}

/// serde adaptor: turns missing-key into `None`, explicit-null
/// into `Some(None)`, value into `Some(Some(v))`. Standard trick
/// for distinguishing "field absent" from "field set to null" in
/// JSON PATCH bodies.
fn deserialize_some<'de, T, D>(deserializer: D) -> Result<Option<T>, D::Error>
where
    T: serde::Deserialize<'de>,
    D: serde::Deserializer<'de>,
{
    T::deserialize(deserializer).map(Some)
}

/// Partial update of an existing
/// schedule.
///
/// Mutable fields: `body`, `description`, `interval_seconds`,
/// `next_fire_at_unix`, plus the target set via `targets_add` /
/// `targets_remove`. Both target lists are applied in the same
/// transaction as the scalar fields with removes-before-adds;
/// re-adding a previously-removed target resets per-target history
/// (fresh start); draining all targets auto-cancels the parent
/// schedule. JSON body uses missing-key = "leave alone", explicit
/// null = "clear" for `description` + `interval_seconds`. Cancelled
/// schedules are refused — submit a new one instead. Returns the
/// updated `WireSchedule` so the caller's post-edit refresh has the
/// new state inline.
#[utoipa::path(
    patch,
    path = "/api/schedules/{id}",
    params(("id" = i64, Path, description = "schedule row id")),
    responses(
        (status = 200, description = "updated, wire shape", body = serde_json::Value),
        (status = 500, description = "update failed (cancelled, not found, ...)"),
    ),
    tag = "schedules"
)]
pub(super) async fn patch_schedule(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
    axum::Json(form): axum::Json<EditScheduleForm>,
) -> Response {
    let patch = crate::scheduled_prompts::UpdateSchedule {
        body: form.body,
        description: form.description,
        interval_seconds: form.interval_seconds,
        next_fire_at_unix: form.next_fire_at_unix,
        targets_add: form.targets_add,
        targets_remove: form.targets_remove,
    };
    if let Err(e) = state.coord.scheduled_prompts.update(id, patch) {
        return error_response(&format!("edit schedule {id}: {e:#}"));
    }
    match state.coord.scheduled_prompts.get(id) {
        Ok(Some(s)) => {
            let wire = crate::socket_server::schedule_to_wire_public(s);
            state.coord.emit_schedules_snapshot();
            axum::Json(wire).into_response()
        }
        Ok(None) => error_response(&format!("edit schedule {id}: row vanished post-update")),
        Err(e) => error_response(&format!("re-read schedule {id}: {e:#}")),
    }
}

/// Pause a schedule so the worker
/// skips it until explicitly resumed.
///
/// Idempotent; no-op on an already-paused row. Returns 404 when the
/// schedule is cancelled or not found.
#[utoipa::path(
    post,
    path = "/api/schedules/{id}/pause",
    params(("id" = i64, Path, description = "schedule row id")),
    responses(
        (status = 200, description = "paused", body = String),
        (status = 404, description = "schedule cancelled or not found"),
        (status = 500, description = "pause failed"),
    ),
    tag = "schedules"
)]
pub(super) async fn post_schedule_pause(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
) -> Response {
    match state.coord.scheduled_prompts.pause(id) {
        Ok(()) => {
            state.coord.emit_schedules_snapshot();
            (StatusCode::OK, "ok").into_response()
        }
        Err(e) => {
            if e.downcast_ref::<ScheduleNotFoundOrCancelled>().is_some() {
                (StatusCode::NOT_FOUND, format!("{e}")).into_response()
            } else {
                error_response(&format!("pause schedule {id}: {e:#}"))
            }
        }
    }
}

/// Resume a paused schedule.
///
/// Idempotent; no-op on an already-active row. Returns 404 when the
/// schedule is cancelled or not found.
#[utoipa::path(
    post,
    path = "/api/schedules/{id}/resume",
    params(("id" = i64, Path, description = "schedule row id")),
    responses(
        (status = 200, description = "resumed", body = String),
        (status = 404, description = "schedule cancelled or not found"),
        (status = 500, description = "resume failed"),
    ),
    tag = "schedules"
)]
pub(super) async fn post_schedule_resume(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
) -> Response {
    match state.coord.scheduled_prompts.resume(id) {
        Ok(()) => {
            state.coord.emit_schedules_snapshot();
            (StatusCode::OK, "ok").into_response()
        }
        Err(e) => {
            if e.downcast_ref::<ScheduleNotFoundOrCancelled>().is_some() {
                (StatusCode::NOT_FOUND, format!("{e}")).into_response()
            } else {
                error_response(&format!("resume schedule {id}: {e:#}"))
            }
        }
    }
}

/// Operator-side cancel
/// (whole schedule when no `targets` field, partial when one is
/// provided).
///
/// Operator bypasses the topology check; the manager surface
/// enforces it for agent callers.
#[utoipa::path(
    post,
    path = "/api/schedules/{id}/cancel",
    params(("id" = i64, Path, description = "schedule row id")),
    responses(
        (status = 200, description = "cancelled (whole or partial)", body = String),
        (status = 500, description = "cancel failed"),
    ),
    tag = "schedules"
)]
pub(super) async fn post_schedule_cancel(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<i64>,
    body: Option<axum::Json<CancelScheduleForm>>,
) -> Response {
    let targets = body
        .and_then(|axum::Json(b)| b.targets)
        .filter(|t| !t.is_empty());
    let result = match targets.as_deref() {
        Some(list) => state.coord.scheduled_prompts.cancel_targets(id, list),
        None => state.coord.scheduled_prompts.cancel_all(id),
    };
    match result {
        Ok(()) => {
            state.coord.emit_schedules_snapshot();
            (StatusCode::OK, "ok").into_response()
        }
        Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")),
    }
}