forge_core/workflow/

traits.rs

use std::future::Future;
use std::pin::Pin;
use std::str::FromStr;
use std::time::Duration;

use serde::{Serialize, de::DeserializeOwned};

use super::context::WorkflowContext;
use crate::Result;

/// Trait for workflow handlers.
pub trait ForgeWorkflow: Send + Sync + 'static {
    /// Input type for the workflow.
    type Input: DeserializeOwned + Serialize + Send + Sync;
    /// Output type for the workflow.
    type Output: Serialize + Send;

    /// Get workflow metadata.
    fn info() -> WorkflowInfo;

    /// Execute the workflow.
    fn execute(
        ctx: &WorkflowContext,
        input: Self::Input,
    ) -> Pin<Box<dyn Future<Output = Result<Self::Output>> + Send + '_>>;
}

/// Workflow metadata.
#[derive(Debug, Clone)]
pub struct WorkflowInfo {
    /// Workflow logical name (stable across versions).
    pub name: &'static str,
    /// User-facing version identifier (e.g. "2026-03", "v2", "signup-fix-1").
    pub version: &'static str,
    /// Derived signature from the persisted contract. Used as the hard runtime safety gate.
    pub signature: &'static str,
    /// Whether this is the active version (new runs start here).
    pub is_active: bool,
    /// Whether this version is deprecated (kept for draining old runs).
    pub is_deprecated: bool,
    /// Default timeout for the entire workflow.
    pub timeout: Duration,
    /// Default timeout for outbound HTTP requests made by the workflow.
    pub http_timeout: Option<Duration>,
    /// Whether the workflow is public (no auth required).
    pub is_public: bool,
    /// Required role for authorization (implies auth required).
    pub required_role: Option<&'static str>,
}

impl Default for WorkflowInfo {
    fn default() -> Self {
        Self {
            name: "",
            version: "v1",
            signature: "",
            is_active: true,
            is_deprecated: false,
            timeout: Duration::from_secs(86400), // 24 hours
            http_timeout: None,
            is_public: false,
            required_role: None,
        }
    }
}

/// Workflow execution status.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkflowStatus {
    /// Workflow is created but not started.
    Created,
    /// Workflow is actively running.
    Running,
    /// Workflow is waiting for an external event or timer.
    Waiting,
    /// Workflow completed successfully.
    Completed,
    /// Workflow failed and is running compensation.
    Compensating,
    /// Workflow compensation completed.
    Compensated,
    /// Workflow failed (compensation also failed or not available).
    Failed,
    /// Blocked: the workflow version is not present in the current binary.
    BlockedMissingVersion,
    /// Blocked: the workflow version exists but its signature does not match.
    BlockedSignatureMismatch,
    /// Blocked: no handler registered for this workflow name at all.
    BlockedMissingHandler,
    /// Explicitly retired by an operator. Terminal, preserves audit trail.
    RetiredUnresumable,
    /// Explicitly cancelled by an operator. Terminal, preserves audit trail.
    CancelledByOperator,
}

impl WorkflowStatus {
    /// Convert to string for database storage.
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Created => "created",
            Self::Running => "running",
            Self::Waiting => "waiting",
            Self::Completed => "completed",
            Self::Compensating => "compensating",
            Self::Compensated => "compensated",
            Self::Failed => "failed",
            Self::BlockedMissingVersion => "blocked_missing_version",
            Self::BlockedSignatureMismatch => "blocked_signature_mismatch",
            Self::BlockedMissingHandler => "blocked_missing_handler",
            Self::RetiredUnresumable => "retired_unresumable",
            Self::CancelledByOperator => "cancelled_by_operator",
        }
    }

    /// Check if the workflow is terminal (no longer running).
    pub fn is_terminal(&self) -> bool {
        matches!(
            self,
            Self::Completed
                | Self::Compensated
                | Self::Failed
                | Self::RetiredUnresumable
                | Self::CancelledByOperator
        )
    }

    /// Check if the workflow is blocked and cannot make progress.
    pub fn is_blocked(&self) -> bool {
        matches!(
            self,
            Self::BlockedMissingVersion
                | Self::BlockedSignatureMismatch
                | Self::BlockedMissingHandler
        )
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParseWorkflowStatusError(pub String);

impl std::fmt::Display for ParseWorkflowStatusError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "invalid workflow status: '{}'", self.0)
    }
}

impl std::error::Error for ParseWorkflowStatusError {}

impl FromStr for WorkflowStatus {
    type Err = ParseWorkflowStatusError;

    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
        match s {
            "created" => Ok(Self::Created),
            "running" => Ok(Self::Running),
            "waiting" => Ok(Self::Waiting),
            "completed" => Ok(Self::Completed),
            "compensating" => Ok(Self::Compensating),
            "compensated" => Ok(Self::Compensated),
            "failed" => Ok(Self::Failed),
            "blocked_missing_version" => Ok(Self::BlockedMissingVersion),
            "blocked_signature_mismatch" => Ok(Self::BlockedSignatureMismatch),
            "blocked_missing_handler" => Ok(Self::BlockedMissingHandler),
            "retired_unresumable" => Ok(Self::RetiredUnresumable),
            "cancelled_by_operator" => Ok(Self::CancelledByOperator),
            _ => Err(ParseWorkflowStatusError(s.to_string())),
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::indexing_slicing)]
mod tests {
    use super::*;

    #[test]
    fn test_workflow_info_default() {
        let info = WorkflowInfo::default();
        assert_eq!(info.name, "");
        assert_eq!(info.version, "v1");
        assert!(info.is_active);
        assert!(!info.is_deprecated);
    }

    #[test]
    fn test_workflow_status_conversion() {
        assert_eq!(WorkflowStatus::Running.as_str(), "running");
        assert_eq!(WorkflowStatus::Completed.as_str(), "completed");
        assert_eq!(WorkflowStatus::Compensating.as_str(), "compensating");
        assert_eq!(
            WorkflowStatus::BlockedMissingVersion.as_str(),
            "blocked_missing_version"
        );
        assert_eq!(
            WorkflowStatus::CancelledByOperator.as_str(),
            "cancelled_by_operator"
        );

        assert_eq!(
            "running".parse::<WorkflowStatus>(),
            Ok(WorkflowStatus::Running)
        );
        assert_eq!(
            "completed".parse::<WorkflowStatus>(),
            Ok(WorkflowStatus::Completed)
        );
        assert_eq!(
            "blocked_missing_version".parse::<WorkflowStatus>(),
            Ok(WorkflowStatus::BlockedMissingVersion)
        );
        assert_eq!(
            "cancelled_by_operator".parse::<WorkflowStatus>(),
            Ok(WorkflowStatus::CancelledByOperator)
        );
    }

    #[test]
    fn test_workflow_status_is_terminal() {
        assert!(!WorkflowStatus::Running.is_terminal());
        assert!(!WorkflowStatus::Waiting.is_terminal());
        assert!(WorkflowStatus::Completed.is_terminal());
        assert!(WorkflowStatus::Failed.is_terminal());
        assert!(WorkflowStatus::Compensated.is_terminal());
        assert!(WorkflowStatus::RetiredUnresumable.is_terminal());
        assert!(WorkflowStatus::CancelledByOperator.is_terminal());
    }

    #[test]
    fn test_workflow_status_is_blocked() {
        assert!(!WorkflowStatus::Running.is_blocked());
        assert!(WorkflowStatus::BlockedMissingVersion.is_blocked());
        assert!(WorkflowStatus::BlockedSignatureMismatch.is_blocked());
        assert!(WorkflowStatus::BlockedMissingHandler.is_blocked());
    }
}