tachyon_types/

commands.rs

use myko_macros::myko_command;
use myko::command::{CommandContext, CommandError, CommandHandler};
use serde_json::Value;

use crate::{
    derive_resources, GetAllInventorys, GetPinsByIds, GetPlaybooksByIds, GetRunsByIds, OutputLine,
    OutputLineId, OutputStream, Pin, PinId, Playbook, PlaybookId, Run, RunId, RunStatus,
};

/// Run an Ansible playbook. Creates a new `Run` entity with status `Queued`;
/// the admission step in tachyon-server promotes it to `Running` once it holds
/// a lease on every resource it targets, and the run watcher then executes it.
#[myko_command]
pub struct RunPlaybook {
    pub playbook_id: PlaybookId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub extra_vars: Option<Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<String>,
}

impl CommandHandler for RunPlaybook {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        // Confirm the playbook exists.
        ctx.exec_query_first(GetPlaybooksByIds {
            ids: vec![self.playbook_id.clone()],
        })?
        .ok_or_else(|| CommandError {
            tx: ctx.tx().to_string(),
            command_id: ctx.command_id.to_string(),
            message: format!("Playbook {} not found", self.playbook_id.0),
        })?;

        // Derive the resource set from the limit against the current inventory,
        // so the queue can serialize on shared clusters. No inventory (e.g. it
        // failed to discover) means no derivable resources → the run leases
        // nothing and is admitted immediately.
        let resources = match ctx.exec_query(GetAllInventorys {})?.into_iter().next() {
            Some(inv) => derive_resources(self.limit.as_deref(), &inv),
            None => Vec::new(),
        };

        let run_id = RunId::from(uuid::Uuid::new_v4().to_string());
        let now = chrono::Utc::now().to_rfc3339();
        let run = Run {
            playbook_id: self.playbook_id,
            status: RunStatus::Queued,
            started_at: now,
            finished_at: None,
            exit_code: None,
            extra_vars: self.extra_vars,
            limit: self.limit,
            resources,
            host_stats: Default::default(),
            client_id: None, // server-stamps the launching connection on emit
            id: run_id,
        };
        ctx.emit_set(&run)?;
        Ok(())
    }
}

/// Cancel a playbook execution. Works on a `Queued` run (drops it from the
/// queue before it ever starts) or a `Running` one (signals the child process).
/// No-op if the run has already finished.
#[myko_command]
pub struct CancelRun {
    pub run_id: RunId,
}

impl CommandHandler for CancelRun {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let current = ctx
            .exec_query_first(GetRunsByIds {
                ids: vec![self.run_id.clone()],
            })?
            .ok_or_else(|| CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!("Run {} not found", self.run_id.0),
            })?;

        // Only an in-flight run (queued or running) can be cancelled; a
        // finished run stays as it is. A queued run carries no PID, so the
        // watcher's cancel handler simply finds nothing to signal.
        if !matches!(current.status, RunStatus::Queued | RunStatus::Running) {
            return Ok(());
        }

        let mut updated = (*current).clone();
        updated.status = RunStatus::Cancelled;
        ctx.emit_set(&updated)?;
        Ok(())
    }
}

/// Pin a playbook to the sidebar's "Pinned" section. Idempotent: re-pinning
/// an already-pinned playbook is a no-op. Refuses to overwrite a locked
/// (config-defined) pin.
#[myko_command]
pub struct PinPlaybook {
    pub playbook_id: PlaybookId,
}

impl CommandHandler for PinPlaybook {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        ctx.exec_query_first(GetPlaybooksByIds {
            ids: vec![self.playbook_id.clone()],
        })?
        .ok_or_else(|| CommandError {
            tx: ctx.tx().to_string(),
            command_id: ctx.command_id.to_string(),
            message: format!("Playbook {} not found", self.playbook_id.0),
        })?;

        let pin_id = PinId::from(self.playbook_id.0.clone());
        if let Some(existing) = ctx.exec_query_first(GetPinsByIds {
            ids: vec![pin_id.clone()],
        })? {
            if existing.locked {
                return Ok(());
            }
        }

        let pin = Pin {
            playbook_id: self.playbook_id,
            locked: false,
            id: pin_id,
        };
        ctx.emit_set(&pin)?;
        Ok(())
    }
}

/// Append one line of playbook output to a run. Issued by the out-of-process
/// runner as it streams `ansible-playbook` output back to the cell (the
/// in-process runner writes `OutputLine`s directly via `ctx`). `seq` is the
/// runner's monotonic per-run counter across both stdout+stderr, so the UI can
/// order lines and jump to a parsed failure's source line.
#[myko_command]
pub struct ReportRunOutput {
    pub run_id: RunId,
    pub stream: OutputStream,
    pub line: String,
    pub seq: u64,
}

impl CommandHandler for ReportRunOutput {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let entity = OutputLine {
            id: OutputLineId::from(format!("{}:{}", self.run_id.0, self.seq)),
            run_id: self.run_id,
            stream: self.stream,
            line: self.line,
            seq: self.seq,
        };
        ctx.emit_set(&entity)?;
        Ok(())
    }
}

/// Mark a run finished, reported by the runner when `ansible-playbook` exits.
/// The cell decides the terminal status: a run already `Cancelled` (a cancel
/// landed mid-run) stays cancelled; otherwise exit code 0 -> `Completed`,
/// anything else -> `Failed`. `exit_code` is `None` when the process was killed
/// by a signal or never spawned.
#[myko_command]
pub struct ReportRunFinished {
    pub run_id: RunId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_code: Option<i32>,
}

impl CommandHandler for ReportRunFinished {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        let current = ctx
            .exec_query_first(GetRunsByIds {
                ids: vec![self.run_id.clone()],
            })?
            .ok_or_else(|| CommandError {
                tx: ctx.tx().to_string(),
                command_id: ctx.command_id.to_string(),
                message: format!("Run {} not found", self.run_id.0),
            })?;

        let mut updated = (*current).clone();
        // A concurrent cancel wins; otherwise derive terminal status from exit.
        if updated.status != RunStatus::Cancelled {
            updated.status = if self.exit_code == Some(0) {
                RunStatus::Completed
            } else {
                RunStatus::Failed
            };
        }
        updated.exit_code = self.exit_code;
        updated.finished_at = Some(chrono::Utc::now().to_rfc3339());
        ctx.emit_set(&updated)?;
        Ok(())
    }
}

/// Declare the playbooks a runner has available. The runner scans its
/// `playbook_dir` and reports each as a `Playbook` entity (upsert by id).
/// Discovery lives where the playbook files are - the runner - not the cell,
/// so `RunPlaybook`'s existence check has something to resolve against.
#[myko_command]
pub struct ReportPlaybooks {
    pub playbooks: Vec<Playbook>,
}

impl CommandHandler for ReportPlaybooks {
    fn execute(self, ctx: CommandContext) -> Result<(), CommandError> {
        for pb in self.playbooks {
            ctx.emit_set(&pb)?;
        }
        Ok(())
    }
}