af-workflow 0.4.0

//! Validated multi-branch workflow host.

use std::sync::Arc;

use crate::executor::CompiledBranch;
use crate::{
    validate, CompileError, Event, NodeRegistry, RunOutcome, Spec, SpecError, State, Violation,
    WorkflowContext,
};

/// Conventional id of the root branch.
pub const ROOT_BRANCH: &str = "__root__";

/// Why a spec could not be hosted.
#[derive(Debug, thiserror::Error)]
pub enum HostError {
    /// Workflow spec failed to parse.
    #[error("workflow spec failed to parse: {0}")]
    Parse(#[from] SpecError),
    /// Workflow spec '`spec_id`' has no branches.
    #[error("workflow spec '{spec_id}' has no branches")]
    NoBranch {
        /// Spec id.
        spec_id: String,
    },
    /// Workflow spec '`spec_id`' violates safety rules: `violations`.
    #[error("workflow spec '{spec_id}' violates safety rules: {violations:?}")]
    Safety {
        /// Spec that failed.
        spec_id: String,
        /// Safety rules violated.
        violations: Vec<Violation>,
    },
    /// Workflow spec '`spec_id`' failed to compile: `source`.
    #[error("workflow spec '{spec_id}' failed to compile: {source}")]
    Compile {
        /// Spec that failed.
        spec_id: String,
        /// Compile error.
        #[source]
        source: CompileError,
    },
    /// A step performs an external effect; durable spec execution runs pure/read steps only.
    #[error("workflow spec '{spec_id}' step '{node_id}' ({node_type}) performs an external effect; durable spec execution runs pure/read steps only, register a WorkflowActionProvider for it")]
    UnsupportedDurableStep {
        /// Spec that failed.
        spec_id: String,
        /// Node that cannot run durably.
        node_id: String,
        /// Its node type.
        node_type: String,
    },
    /// Workflow spec '`spec_id`' has an invalid schedule: `reason`.
    #[error("workflow spec '{spec_id}' has an invalid schedule: {reason}")]
    Schedule {
        /// Spec id.
        spec_id: String,
        /// Why the schedule is invalid.
        reason: String,
    },
}

struct BranchEntry {
    id: String,
    branch: CompiledBranch,
}

/// A parsed, validated and compiled workflow. Assembly is the only constructor,
/// so a long-running host cannot accidentally skip static validation.
pub struct WorkflowHost {
    spec_id: String,
    branches: Vec<BranchEntry>,
}

impl WorkflowHost {
    /// Parse, validate and compile a spec from JSON.
    pub fn assemble(spec_json: &str, registry: &NodeRegistry) -> Result<Self, HostError> {
        Self::from_spec(&Spec::from_json(spec_json)?, registry)
    }

    /// Validate and compile every branch of `spec`.
    pub fn from_spec(spec: &Spec, registry: &NodeRegistry) -> Result<Self, HostError> {
        if spec.branches.is_empty() {
            return Err(HostError::NoBranch {
                spec_id: spec.spec_id.clone(),
            });
        }
        if let Err(violations) = validate(spec, registry) {
            return Err(HostError::Safety {
                spec_id: spec.spec_id.clone(),
                violations,
            });
        }

        let branches = spec
            .branches
            .iter()
            .map(|branch| {
                Ok(BranchEntry {
                    id: branch.branch_id.clone(),
                    branch: CompiledBranch::compile(branch, registry).map_err(|source| {
                        HostError::Compile {
                            spec_id: spec.spec_id.clone(),
                            source,
                        }
                    })?,
                })
            })
            .collect::<Result<Vec<_>, HostError>>()?;
        Ok(Self {
            spec_id: spec.spec_id.clone(),
            branches,
        })
    }

    /// Spec id.
    pub fn spec_id(&self) -> &str {
        &self.spec_id
    }

    /// Id of the first (root) branch.
    pub fn branch_id(&self) -> &str {
        &self.branches[0].id
    }

    /// Every branch id.
    pub fn branch_ids(&self) -> impl Iterator<Item = &str> {
        self.branches.iter().map(|branch| branch.id.as_str())
    }

    /// Number of branches.
    pub fn branch_count(&self) -> usize {
        self.branches.len()
    }

    /// Context for the root branch over `state`.
    pub fn context(&self, state: Arc<dyn State>) -> WorkflowContext {
        self.context_for(self.branch_id(), state)
    }

    /// Context for `branch_id` over `state`.
    pub fn context_for(&self, branch_id: &str, state: Arc<dyn State>) -> WorkflowContext {
        WorkflowContext::new(branch_id, state)
    }

    /// Drive one event through the context's branch.
    pub async fn run_event(&self, ctx: &WorkflowContext, event: Event) -> Option<RunOutcome> {
        self.run_event_on(&ctx.branch_id, ctx, event).await
    }

    /// Drive one event through `branch_id`; `None` when the branch does not exist.
    pub async fn run_event_on(
        &self,
        branch_id: &str,
        ctx: &WorkflowContext,
        event: Event,
    ) -> Option<RunOutcome> {
        let branch = self.branches.iter().find(|branch| branch.id == branch_id)?;
        Some(branch.branch.run_event(ctx, event).await)
    }
}