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 super::{AppState, error_problem, error_response};

/// `GET /api/schedules` — snapshot of every schedule for the
/// scheduled-prompts tab. Returns the wire shape directly
/// so the frontend can render without an extra translation layer.
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::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:#}")),
    }
}

/// `POST /api/schedules` — 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.
pub(super) async fn post_schedule_new(
    State(state): State<AppState>,
    axum::Json(payload): axum::Json<hive_sh4re::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::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)]
pub(super) struct FireNowBody {
    #[serde(default)]
    reset_timer: bool,
}

/// `POST /api/schedules/{id}/fire-now` — 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`).
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:#}")),
    }
}

/// `POST /api/rebuild-queue/{id}/cancel` — drop a still-fully-queued
/// DAG from the job queue. Refuses `Running` / terminal DAGs: 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.
pub(super) async fn post_rebuild_queue_cancel(
    State(state): State<AppState>,
    AxumPath(id): AxumPath<u64>,
) -> Response {
    if let Some(terminal) = state.coord.job_queue.cancel(id) {
        // Fire the DAG's inline terminal hook (power-op intent revert / approval
        // resolution) off the cancel roll-up, then surface the flip live.
        crate::job_queue::exec::run_terminal_hook(&state.coord, &terminal).await;
        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)]
pub(super) struct CancelScheduleForm {
    /// `None` / absent / empty array → cancel whole schedule.
    #[serde(default)]
    targets: Option<Vec<String>>,
}

#[derive(serde::Deserialize, Default)]
#[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)
}

/// `PATCH /api/schedules/{id}` — 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.
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:#}")),
    }
}

/// `POST /api/schedules/{id}/pause` — 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.
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:#}"))
            }
        }
    }
}

/// `POST /api/schedules/{id}/resume` — resume a paused schedule.
/// Idempotent; no-op on an already-active row. Returns 404 when the
/// schedule is cancelled or not found.
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:#}"))
            }
        }
    }
}

/// `POST /api/schedules/{id}/cancel` — 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.
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:#}")),
    }
}