mur_core/workflow/

compose.rs

//! Workflow composition — merge sub-workflows into a single composed workflow.

use crate::types::{Step, Workflow};
use anyhow::{Context, Result};
use chrono::Utc;
use std::collections::HashMap;

/// A composed workflow that references child workflow IDs.
#[derive(Debug, Clone)]
pub struct ComposedWorkflow {
    /// The resulting merged workflow.
    pub workflow: Workflow,
    /// IDs of child workflows that were composed.
    pub child_ids: Vec<String>,
}

/// Raw YAML structure supporting `includes` directive.
#[derive(Debug, serde::Deserialize)]
pub struct ComposableWorkflowYaml {
    pub id: String,
    pub name: String,
    #[serde(default)]
    pub description: String,
    #[serde(default)]
    pub variables: HashMap<String, String>,
    #[serde(default)]
    pub steps: Vec<super::parser::StepYaml>,
    /// List of sub-workflow IDs to include.
    #[serde(default)]
    pub includes: Vec<String>,
}

/// Compose multiple workflows into a single workflow by merging steps and variables.
///
/// Steps are concatenated in order; variables from later workflows override earlier ones
/// unless there's a naming conflict, in which case the child workflow's variable is prefixed
/// with its workflow ID.
pub fn compose_workflows(
    base_id: &str,
    base_name: &str,
    workflows: &[Workflow],
) -> Result<ComposedWorkflow> {
    anyhow::ensure!(!workflows.is_empty(), "At least one workflow is required");

    let mut merged_steps: Vec<Step> = Vec::new();
    let mut merged_variables: HashMap<String, String> = HashMap::new();
    let mut child_ids: Vec<String> = Vec::new();
    let mut seen_step_names: HashMap<String, usize> = HashMap::new();

    for wf in workflows {
        child_ids.push(wf.id.clone());

        // Merge variables, prefixing on conflict
        for (key, value) in &wf.variables {
            if merged_variables.contains_key(key) && merged_variables[key] != *value {
                // Naming conflict: prefix with workflow ID
                let prefixed_key = format!("{}.{}", wf.id, key);
                merged_variables.insert(prefixed_key, value.clone());
            } else {
                merged_variables.insert(key.clone(), value.clone());
            }
        }

        // Merge steps, deduplicating names
        for step in &wf.steps {
            let name = if let Some(count) = seen_step_names.get(&step.name) {
                format!("{}.{}", step.name, count + 1)
            } else {
                step.name.clone()
            };
            *seen_step_names.entry(step.name.clone()).or_insert(0) += 1;

            merged_steps.push(Step {
                name,
                step_type: step.step_type.clone(),
                action: step.action.clone(),
                on_failure: step.on_failure.clone(),
                breakpoint: step.breakpoint,
                breakpoint_message: step.breakpoint_message.clone(),
            });
        }
    }

    let now = Utc::now();
    Ok(ComposedWorkflow {
        workflow: Workflow {
            id: base_id.to_string(),
            name: base_name.to_string(),
            description: format!(
                "Composed workflow from: {}",
                child_ids.join(", ")
            ),
            steps: merged_steps,
            variables: merged_variables,
            schedule: None,
            created_at: now,
            updated_at: now,
        },
        child_ids,
    })
}

/// Resolve `includes` directives in a composable workflow YAML string.
///
/// Loads referenced sub-workflows from the workflows directory and merges them.
pub fn resolve_includes(yaml: &str) -> Result<Workflow> {
    let raw: ComposableWorkflowYaml =
        serde_yaml::from_str(yaml).context("Parsing composable workflow YAML")?;

    if raw.includes.is_empty() {
        // No includes — parse as a regular workflow
        return super::parser::parse_workflow_str(yaml);
    }

    // Load all available workflows to resolve includes
    let all_workflows = super::parser::load_all_workflows()?;
    let workflow_map: HashMap<&str, &Workflow> = all_workflows
        .iter()
        .map(|w| (w.id.as_str(), w))
        .collect();

    // Collect included workflows in order
    let mut child_workflows: Vec<Workflow> = Vec::new();
    for include_id in &raw.includes {
        let child = workflow_map
            .get(include_id.as_str())
            .with_context(|| format!("Included workflow '{}' not found", include_id))?;
        child_workflows.push((*child).clone());
    }

    // Build the base workflow from inline steps (if any)
    let now = Utc::now();
    let base = Workflow {
        id: raw.id.clone(),
        name: raw.name.clone(),
        description: raw.description,
        steps: raw
            .steps
            .into_iter()
            .map(super::parser::step_yaml_to_step)
            .collect(),
        variables: raw.variables,
        schedule: None,
        created_at: now,
        updated_at: now,
    };

    // Insert base as the first workflow, then append children
    let mut all = vec![base];
    all.extend(child_workflows);

    let composed = compose_workflows(&raw.id, &raw.name, &all)?;
    Ok(composed.workflow)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{FailureAction, StepType};

    fn make_workflow(id: &str, steps: Vec<(&str, &str)>, vars: Vec<(&str, &str)>) -> Workflow {
        let now = Utc::now();
        Workflow {
            id: id.into(),
            name: format!("Workflow {}", id),
            description: String::new(),
            steps: steps
                .into_iter()
                .map(|(name, action)| Step {
                    name: name.into(),
                    step_type: StepType::Execute,
                    action: action.into(),
                    on_failure: FailureAction::Abort,
                    breakpoint: false,
                    breakpoint_message: None,
                })
                .collect(),
            variables: vars
                .into_iter()
                .map(|(k, v)| (k.into(), v.into()))
                .collect(),
            schedule: None,
            created_at: now,
            updated_at: now,
        }
    }

    #[test]
    fn test_compose_merges_steps() {
        let wf1 = make_workflow("a", vec![("step1", "echo a")], vec![]);
        let wf2 = make_workflow("b", vec![("step2", "echo b")], vec![]);

        let composed = compose_workflows("composed", "Composed", &[wf1, wf2]).unwrap();
        assert_eq!(composed.workflow.steps.len(), 2);
        assert_eq!(composed.workflow.steps[0].name, "step1");
        assert_eq!(composed.workflow.steps[1].name, "step2");
        assert_eq!(composed.child_ids, vec!["a", "b"]);
    }

    #[test]
    fn test_compose_deduplicates_step_names() {
        let wf1 = make_workflow("a", vec![("deploy", "echo a")], vec![]);
        let wf2 = make_workflow("b", vec![("deploy", "echo b")], vec![]);

        let composed = compose_workflows("composed", "Composed", &[wf1, wf2]).unwrap();
        assert_eq!(composed.workflow.steps.len(), 2);
        assert_eq!(composed.workflow.steps[0].name, "deploy");
        assert_eq!(composed.workflow.steps[1].name, "deploy.2");
    }

    #[test]
    fn test_compose_merges_variables_no_conflict() {
        let wf1 = make_workflow("a", vec![], vec![("host", "localhost")]);
        let wf2 = make_workflow("b", vec![], vec![("port", "8080")]);

        let composed = compose_workflows("composed", "Composed", &[wf1, wf2]).unwrap();
        assert_eq!(composed.workflow.variables.get("host").unwrap(), "localhost");
        assert_eq!(composed.workflow.variables.get("port").unwrap(), "8080");
    }

    #[test]
    fn test_compose_handles_variable_conflicts() {
        let wf1 = make_workflow("a", vec![], vec![("host", "localhost")]);
        let wf2 = make_workflow("b", vec![], vec![("host", "production.example.com")]);

        let composed = compose_workflows("composed", "Composed", &[wf1, wf2]).unwrap();
        assert_eq!(composed.workflow.variables.get("host").unwrap(), "localhost");
        assert_eq!(
            composed.workflow.variables.get("b.host").unwrap(),
            "production.example.com"
        );
    }

    #[test]
    fn test_compose_empty_fails() {
        let result = compose_workflows("empty", "Empty", &[]);
        assert!(result.is_err());
    }
}