szal 1.0.0

//! Workflow event bus — powered by majra pub/sub.
//!
//! Publishes workflow lifecycle events to topics that external systems
//! can subscribe to for monitoring, logging, and orchestration.
//!
//! ## Topic hierarchy
//!
//! ```text
//! szal/flow/{flow_name}/started
//! szal/flow/{flow_name}/completed
//! szal/flow/{flow_name}/failed
//! szal/step/{step_name}/started
//! szal/step/{step_name}/completed
//! szal/step/{step_name}/failed
//! szal/step/{step_name}/retry
//! szal/step/{step_name}/rollback
//! ```
//!
//! Subscribe with wildcards: `szal/flow/#` for all flow events,
//! `szal/step/*/failed` for all step failures.

#[cfg(feature = "majra")]
use majra::pubsub::PubSub;
use serde::{Deserialize, Serialize};

/// A workflow lifecycle event.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowEvent {
    pub event_type: EventType,
    pub flow_name: Option<String>,
    pub step_name: Option<String>,
    pub step_id: Option<String>,
    pub attempt: Option<u32>,
    pub duration_ms: Option<u64>,
    pub error: Option<String>,
    pub timestamp: chrono::DateTime<chrono::Utc>,
}

/// Event types emitted during workflow execution.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum EventType {
    FlowStarted,
    FlowCompleted,
    FlowFailed,
    FlowRolledBack,
    StepStarted,
    StepCompleted,
    StepFailed,
    StepRetry,
    StepRollback,
    StepSkipped,
    StepTimeout,
}

impl std::fmt::Display for EventType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::FlowStarted => write!(f, "flow_started"),
            Self::FlowCompleted => write!(f, "flow_completed"),
            Self::FlowFailed => write!(f, "flow_failed"),
            Self::FlowRolledBack => write!(f, "flow_rolled_back"),
            Self::StepStarted => write!(f, "step_started"),
            Self::StepCompleted => write!(f, "step_completed"),
            Self::StepFailed => write!(f, "step_failed"),
            Self::StepRetry => write!(f, "step_retry"),
            Self::StepRollback => write!(f, "step_rollback"),
            Self::StepSkipped => write!(f, "step_skipped"),
            Self::StepTimeout => write!(f, "step_timeout"),
        }
    }
}

impl WorkflowEvent {
    fn new(event_type: EventType) -> Self {
        Self {
            event_type,
            flow_name: None,
            step_name: None,
            step_id: None,
            attempt: None,
            duration_ms: None,
            error: None,
            timestamp: chrono::Utc::now(),
        }
    }

    #[must_use]
    pub fn with_flow(mut self, name: &str) -> Self {
        self.flow_name = Some(name.into());
        self
    }

    #[must_use]
    pub fn with_step(mut self, name: &str, id: &str) -> Self {
        self.step_name = Some(name.into());
        self.step_id = Some(id.into());
        self
    }

    #[must_use]
    pub fn with_duration(mut self, ms: u64) -> Self {
        self.duration_ms = Some(ms);
        self
    }

    #[must_use]
    pub fn with_attempt(mut self, attempt: u32) -> Self {
        self.attempt = Some(attempt);
        self
    }

    #[must_use]
    pub fn with_error(mut self, error: &str) -> Self {
        self.error = Some(error.into());
        self
    }

    #[must_use]
    pub fn flow_started(flow_name: &str) -> Self {
        Self::new(EventType::FlowStarted).with_flow(flow_name)
    }

    #[must_use]
    pub fn flow_completed(flow_name: &str, duration_ms: u64) -> Self {
        Self::new(EventType::FlowCompleted)
            .with_flow(flow_name)
            .with_duration(duration_ms)
    }

    #[must_use]
    pub fn flow_failed(flow_name: &str, error: &str) -> Self {
        Self::new(EventType::FlowFailed)
            .with_flow(flow_name)
            .with_error(error)
    }

    #[must_use]
    pub fn flow_rolled_back(flow_name: &str) -> Self {
        Self::new(EventType::FlowRolledBack).with_flow(flow_name)
    }

    #[must_use]
    pub fn step_started(step_name: &str, step_id: &str) -> Self {
        Self::new(EventType::StepStarted).with_step(step_name, step_id)
    }

    #[must_use]
    pub fn step_completed(step_name: &str, step_id: &str, duration_ms: u64, attempt: u32) -> Self {
        Self::new(EventType::StepCompleted)
            .with_step(step_name, step_id)
            .with_duration(duration_ms)
            .with_attempt(attempt)
    }

    #[must_use]
    pub fn step_failed(step_name: &str, step_id: &str, error: &str, attempt: u32) -> Self {
        Self::new(EventType::StepFailed)
            .with_step(step_name, step_id)
            .with_error(error)
            .with_attempt(attempt)
    }

    #[must_use]
    pub fn step_retry(step_name: &str, step_id: &str, attempt: u32) -> Self {
        Self::new(EventType::StepRetry)
            .with_step(step_name, step_id)
            .with_attempt(attempt)
    }

    #[must_use]
    pub fn step_skipped(step_name: &str, step_id: &str, reason: &str) -> Self {
        Self::new(EventType::StepSkipped)
            .with_step(step_name, step_id)
            .with_error(reason)
    }

    #[must_use]
    pub fn step_timeout(step_name: &str, step_id: &str, timeout_ms: u64) -> Self {
        Self::new(EventType::StepTimeout)
            .with_step(step_name, step_id)
            .with_duration(timeout_ms)
    }

    #[must_use]
    pub fn step_rollback(step_name: &str, step_id: &str) -> Self {
        Self::new(EventType::StepRollback).with_step(step_name, step_id)
    }

    /// Build the topic string for this event.
    #[must_use]
    pub fn topic(&self) -> String {
        match self.event_type {
            EventType::FlowStarted
            | EventType::FlowCompleted
            | EventType::FlowFailed
            | EventType::FlowRolledBack => {
                let name = self.flow_name.as_deref().unwrap_or("unknown");
                format!("szal/flow/{name}/{}", self.event_type)
            }
            _ => {
                let name = self.step_name.as_deref().unwrap_or("unknown");
                format!("szal/step/{name}/{}", self.event_type)
            }
        }
    }
}

/// Workflow event bus backed by majra pub/sub.
#[cfg(feature = "majra")]
pub struct EventBus {
    pubsub: PubSub,
}

#[cfg(feature = "majra")]
impl EventBus {
    /// Create a new event bus.
    #[must_use]
    pub fn new() -> Self {
        Self {
            pubsub: PubSub::new(),
        }
    }

    /// Publish a workflow event.
    pub fn publish(&self, event: &WorkflowEvent) {
        let topic = event.topic();
        let payload = match serde_json::to_value(event) {
            Ok(v) => v,
            Err(e) => {
                tracing::warn!(error = %e, "failed to serialize workflow event");
                return;
            }
        };
        self.pubsub.publish(&topic, payload);
    }

    /// Subscribe to workflow events matching a pattern.
    ///
    /// Examples:
    /// - `"szal/flow/#"` — all flow events
    /// - `"szal/step/*/step_failed"` — all step failures
    /// - `"szal/#"` — everything
    pub fn subscribe(
        &self,
        pattern: &str,
    ) -> tokio::sync::broadcast::Receiver<majra::pubsub::TopicMessage> {
        self.pubsub.subscribe(pattern)
    }
}

#[cfg(feature = "majra")]
impl Default for EventBus {
    fn default() -> Self {
        Self::new()
    }
}

/// Create an [`EventSink`](crate::engine::EventSink) that maps workflow events to tracing spans.
///
/// Each event becomes a tracing event with structured fields following
/// OpenTelemetry semantic conventions. Consumers wire a `tracing-opentelemetry`
/// subscriber layer to export these as OTel spans to a collector.
///
/// Fields emitted: `workflow.event_type`, `workflow.flow_name`, `workflow.step_name`,
/// `workflow.step_id`, `workflow.attempt`, `workflow.duration_ms`, `workflow.error`,
/// `workflow.status`.
#[must_use]
pub fn otel_event_sink() -> std::sync::Arc<dyn Fn(WorkflowEvent) + Send + Sync> {
    std::sync::Arc::new(|event: WorkflowEvent| {
        let event_type = event.event_type.to_string();
        let flow = event.flow_name.as_deref().unwrap_or("");
        let step = event.step_name.as_deref().unwrap_or("");
        let step_id = event.step_id.as_deref().unwrap_or("");
        let status = match event.event_type {
            EventType::FlowCompleted | EventType::StepCompleted => "ok",
            EventType::FlowFailed | EventType::StepFailed | EventType::StepTimeout => "error",
            EventType::StepSkipped => "skipped",
            EventType::FlowRolledBack | EventType::StepRollback => "rolled_back",
            _ => "unset",
        };

        match event.event_type {
            EventType::FlowStarted
            | EventType::FlowCompleted
            | EventType::FlowFailed
            | EventType::FlowRolledBack => {
                tracing::info!(
                    workflow.event_type = %event_type,
                    workflow.flow_name = %flow,
                    workflow.status = %status,
                    workflow.duration_ms = event.duration_ms.unwrap_or(0),
                    workflow.error = event.error.as_deref().unwrap_or(""),
                    "workflow.flow"
                );
            }
            _ => {
                tracing::info!(
                    workflow.event_type = %event_type,
                    workflow.flow_name = %flow,
                    workflow.step_name = %step,
                    workflow.step_id = %step_id,
                    workflow.status = %status,
                    workflow.attempt = event.attempt.unwrap_or(0),
                    workflow.duration_ms = event.duration_ms.unwrap_or(0),
                    workflow.error = event.error.as_deref().unwrap_or(""),
                    "workflow.step"
                );
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn event_topic_flow() {
        let e = WorkflowEvent::flow_started("deploy");
        assert_eq!(e.topic(), "szal/flow/deploy/flow_started");
    }

    #[test]
    fn event_topic_step() {
        let e = WorkflowEvent::step_completed("build", "abc-123", 500, 1);
        assert_eq!(e.topic(), "szal/step/build/step_completed");
    }

    #[test]
    fn event_serde_roundtrip() {
        let e = WorkflowEvent::step_failed("deploy", "id-1", "timeout", 3);
        let json = serde_json::to_string(&e).unwrap();
        let back: WorkflowEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(back.event_type, EventType::StepFailed);
        assert_eq!(back.attempt, Some(3));
    }

    #[test]
    fn event_type_display() {
        assert_eq!(EventType::FlowStarted.to_string(), "flow_started");
        assert_eq!(EventType::StepRetry.to_string(), "step_retry");
    }

    #[test]
    fn builder_with_flow() {
        let e = WorkflowEvent::flow_started("deploy");
        assert_eq!(e.flow_name, Some("deploy".into()));
    }

    #[test]
    fn builder_with_step() {
        let e = WorkflowEvent::step_started("build", "id-1");
        assert_eq!(e.step_name, Some("build".into()));
        assert_eq!(e.step_id, Some("id-1".into()));
    }

    #[test]
    fn builder_with_duration() {
        let e = WorkflowEvent::step_completed("s", "id", 500, 1);
        assert_eq!(e.duration_ms, Some(500));
    }

    #[test]
    fn builder_with_attempt() {
        let e = WorkflowEvent::step_retry("s", "id", 3);
        assert_eq!(e.attempt, Some(3));
    }

    #[test]
    fn builder_with_error() {
        let e = WorkflowEvent::flow_failed("f", "oops");
        assert_eq!(e.error, Some("oops".into()));
    }

    #[test]
    fn builder_chaining() {
        let e = WorkflowEvent::step_failed("s", "id", "e", 2);
        assert_eq!(e.event_type, EventType::StepFailed);
        assert_eq!(e.step_name, Some("s".into()));
        assert_eq!(e.step_id, Some("id".into()));
        assert_eq!(e.error, Some("e".into()));
        assert_eq!(e.attempt, Some(2));
    }

    #[test]
    fn flow_rolled_back_topic() {
        let e = WorkflowEvent::flow_rolled_back("deploy");
        assert_eq!(e.topic(), "szal/flow/deploy/flow_rolled_back");
    }

    #[cfg(feature = "majra")]
    #[tokio::test]
    async fn event_bus_publish_subscribe() {
        let bus = EventBus::new();
        let mut sub = bus.subscribe("szal/flow/#");

        bus.publish(&WorkflowEvent::flow_started("test"));

        let msg = sub.recv().await.unwrap();
        assert_eq!(msg.topic, "szal/flow/test/flow_started");
    }
}