oxigdal-workflow 0.1.7

//! Temporal.io integration.

use crate::dag::{ResourceRequirements, RetryPolicy, TaskEdge, TaskNode, WorkflowDag};
use crate::engine::WorkflowDefinition;
use crate::error::{Result, WorkflowError};
use regex::Regex;
use std::collections::HashMap;

/// Header comment prefix identifying Go source generated by [`TemporalIntegration::export_workflow`].
///
/// [`TemporalIntegration::import_workflow`] only supports round-tripping this crate's own
/// generated format; this marker (together with the `workflow_id` metadata line and the
/// `go.temporal.io/sdk/workflow` import) is what it checks for.
const GENERATED_MARKER: &str = "// Code generated by oxigdal-workflow TemporalIntegration.";

/// Temporal.io integration.
pub struct TemporalIntegration;

impl TemporalIntegration {
    /// Export workflow to Temporal workflow format (Go).
    pub fn export_workflow(workflow: &WorkflowDefinition) -> Result<String> {
        let mut go_code = String::new();

        let tasks = workflow.dag.tasks();
        let has_command = tasks.iter().any(|task| Self::task_command(task).is_some());

        // Metadata header. This is not idiomatic hand-written Go, but it lets
        // `import_workflow` losslessly recover the original workflow id/name for any input
        // (including ids that aren't valid Go identifiers), rather than trying to reverse
        // `to_camel_case`, which is not invertible in general.
        go_code.push_str(GENERATED_MARKER);
        go_code.push_str(" DO NOT EDIT.\n");
        go_code.push_str(&format!("// workflow_id: {}\n", workflow.id));
        go_code.push_str(&format!("// workflow_name: {}\n", workflow.name));

        // Add package and imports
        go_code.push_str("package workflows\n\n");
        go_code.push_str("import (\n");
        go_code.push_str("    \"time\"\n");
        if has_command {
            go_code.push_str("    \"os/exec\"\n");
        }
        go_code.push_str("    \"go.temporal.io/sdk/workflow\"\n");
        go_code.push_str(")\n\n");

        // Define activity interfaces
        for (idx, task) in tasks.iter().enumerate() {
            go_code.push_str(&format!(
                "func Task{}Activity(ctx workflow.Context) error {{\n",
                idx
            ));
            go_code.push_str("    logger := workflow.GetLogger(ctx)\n");
            go_code.push_str(&format!(
                "    logger.Info(\"Executing task\", \"task_id\", \"{}\")\n",
                Self::escape_go_string(&task.id)
            ));

            if let Some(command) = Self::task_command(task) {
                go_code.push_str(&format!(
                    "    cmd := exec.Command(\"sh\", \"-c\", {})\n",
                    Self::go_string_literal(command)
                ));
                go_code.push_str("    output, err := cmd.CombinedOutput()\n");
                go_code.push_str("    if err != nil {\n");
                go_code.push_str(
                    "        logger.Error(\"Task command failed\", \"error\", err, \"output\", string(output))\n",
                );
                go_code.push_str("        return err\n");
                go_code.push_str("    }\n");
                go_code.push_str(
                    "    logger.Info(\"Task command completed\", \"output\", string(output))\n",
                );
            }

            go_code.push_str("    return nil\n");
            go_code.push_str("}\n\n");
        }

        // Define workflow
        go_code.push_str(&format!(
            "func {}Workflow(ctx workflow.Context) error {{\n",
            Self::to_camel_case(&workflow.id)
        ));

        for (idx, _task) in tasks.iter().enumerate() {
            go_code.push_str("    ao := workflow.ActivityOptions{\n");
            go_code.push_str("        StartToCloseTimeout: 1 * time.Minute,\n");
            go_code.push_str("    }\n");
            go_code.push_str(&format!(
                "    ctx{} := workflow.WithActivityOptions(ctx, ao)\n",
                idx
            ));
            go_code.push_str(&format!(
                "    err{} := workflow.ExecuteActivity(ctx{}, Task{}Activity).Get(ctx{}, nil)\n",
                idx, idx, idx, idx
            ));
            go_code.push_str(&format!("    if err{} != nil {{\n", idx));
            go_code.push_str(&format!("        return err{}\n", idx));
            go_code.push_str("    }\n\n");
        }

        go_code.push_str("    return nil\n");
        go_code.push_str("}\n");

        Ok(go_code)
    }

    /// Import workflow from Temporal workflow.
    ///
    /// Only Go source previously produced by [`Self::export_workflow`] is supported: this
    /// searches for the `oxigdal-workflow` generated-code marker, the `workflow_id` metadata
    /// comment, and `Task{N}Activity` function definitions, then rebuilds a
    /// [`WorkflowDefinition`] whose DAG is a simple sequential chain (`task-0 -> task-1 -> ...`)
    /// over as many tasks as there were activities in the source. Arbitrary hand-written
    /// Temporal workflow code is not a supported input: this function does not implement a Go
    /// parser and returns a [`WorkflowError::Integration`] describing the supported subset
    /// instead of attempting (and likely failing) to interpret unfamiliar code.
    pub fn import_workflow(workflow_code: &str) -> Result<WorkflowDefinition> {
        if !workflow_code.contains(GENERATED_MARKER)
            || !workflow_code.contains("go.temporal.io/sdk/workflow")
        {
            return Err(WorkflowError::integration(
                "temporal",
                "Import only supports Go source produced by \
                 TemporalIntegration::export_workflow (it must contain the \
                 'oxigdal-workflow TemporalIntegration' generated-code marker and import \
                 go.temporal.io/sdk/workflow); arbitrary hand-written Temporal workflows are \
                 not a supported input format",
            ));
        }

        let id_regex = Regex::new(r"(?m)^// workflow_id:\s*(.+)$").map_err(|e| {
            WorkflowError::integration("temporal", format!("Internal regex error: {}", e))
        })?;
        let name_regex = Regex::new(r"(?m)^// workflow_name:\s*(.+)$").map_err(|e| {
            WorkflowError::integration("temporal", format!("Internal regex error: {}", e))
        })?;
        let activity_regex =
            Regex::new(r"func Task(\d+)Activity\(ctx workflow\.Context\) error \{").map_err(
                |e| WorkflowError::integration("temporal", format!("Internal regex error: {}", e)),
            )?;

        let id = id_regex
            .captures(workflow_code)
            .and_then(|c| c.get(1))
            .map(|m| m.as_str().trim().to_string())
            .ok_or_else(|| {
                WorkflowError::integration(
                    "temporal",
                    "Missing '// workflow_id: <id>' metadata comment; this is not a workflow \
                     exported by TemporalIntegration::export_workflow",
                )
            })?;

        let name = name_regex
            .captures(workflow_code)
            .and_then(|c| c.get(1))
            .map(|m| m.as_str().trim().to_string())
            .unwrap_or_else(|| id.clone());

        let mut task_indices: Vec<usize> = activity_regex
            .captures_iter(workflow_code)
            .filter_map(|c| c.get(1).and_then(|m| m.as_str().parse::<usize>().ok()))
            .collect();
        task_indices.sort_unstable();
        task_indices.dedup();

        let mut dag = WorkflowDag::new();
        for (position, _original_idx) in task_indices.iter().enumerate() {
            dag.add_task(Self::sequential_task(position))?;
        }
        for position in 1..task_indices.len() {
            dag.add_dependency(
                &Self::sequential_task_id(position - 1),
                &Self::sequential_task_id(position),
                TaskEdge::default(),
            )?;
        }

        Ok(WorkflowDefinition {
            id,
            name,
            description: None,
            version: "1.0.0".to_string(),
            dag,
        })
    }

    /// Build the `position`-th task of a reconstructed sequential DAG.
    fn sequential_task(position: usize) -> TaskNode {
        TaskNode {
            id: Self::sequential_task_id(position),
            name: format!("Task {}", position),
            description: None,
            config: serde_json::json!({}),
            retry: RetryPolicy::default(),
            timeout_secs: None,
            resources: ResourceRequirements::default(),
            metadata: HashMap::new(),
        }
    }

    /// Deterministic task id used when reconstructing a sequential DAG on import.
    fn sequential_task_id(position: usize) -> String {
        format!("task-{}", position)
    }

    /// Extracts the optional shell command associated with a task, if its `config` carries one.
    ///
    /// A task opts into command execution by setting `config = {"command": "<shell command>"}`
    /// (or including a `"command"` string field alongside other config keys).
    fn task_command(task: &TaskNode) -> Option<&str> {
        task.config.get("command").and_then(|v| v.as_str())
    }

    /// Escapes a string for embedding inside a Go double-quoted string literal.
    fn escape_go_string(s: &str) -> String {
        s.replace('\\', "\\\\").replace('"', "\\\"")
    }

    /// Renders `s` as a Go double-quoted string literal (including the surrounding quotes).
    fn go_string_literal(s: &str) -> String {
        format!("\"{}\"", Self::escape_go_string(s))
    }

    /// Convert to CamelCase for Go naming.
    fn to_camel_case(s: &str) -> String {
        s.split(['-', '_'])
            .filter(|s| !s.is_empty())
            .enumerate()
            .map(|(i, s)| {
                if i == 0 {
                    s.chars()
                        .enumerate()
                        .map(|(j, c)| if j == 0 { c.to_ascii_uppercase() } else { c })
                        .collect()
                } else {
                    let mut chars = s.chars();
                    match chars.next() {
                        None => String::new(),
                        Some(first) => first.to_uppercase().chain(chars).collect(),
                    }
                }
            })
            .collect()
    }

    /// Start a Temporal workflow via API.
    #[cfg(feature = "integrations")]
    pub async fn start_workflow(
        base_url: &str,
        namespace: &str,
        workflow_id: &str,
        workflow_type: &str,
    ) -> Result<String> {
        use reqwest::Client;

        let url = format!(
            "{}/api/v1/namespaces/{}/workflows/{}",
            base_url, namespace, workflow_id
        );
        let client = Client::new();

        let response = client
            .post(&url)
            .header("Content-Type", "application/json")
            .json(&serde_json::json!({
                "workflowId": workflow_id,
                "workflowType": {
                    "name": workflow_type
                },
                "input": {}
            }))
            .send()
            .await
            .map_err(|e| {
                WorkflowError::integration("temporal", format!("Request failed: {}", e))
            })?;

        let status = response.status();

        let body = response.text().await.map_err(|e| {
            WorkflowError::integration("temporal", format!("Failed to read response: {}", e))
        })?;

        if !status.is_success() {
            return Err(WorkflowError::integration(
                "temporal",
                format!("HTTP {}: {}", status, body),
            ));
        }

        Ok(body)
    }
}

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

    fn task_with_command(id: &str, command: Option<&str>) -> TaskNode {
        TaskNode {
            id: id.to_string(),
            name: id.to_string(),
            description: None,
            config: match command {
                Some(cmd) => serde_json::json!({ "command": cmd }),
                None => serde_json::json!({}),
            },
            retry: RetryPolicy::default(),
            timeout_secs: Some(60),
            resources: ResourceRequirements::default(),
            metadata: HashMap::new(),
        }
    }

    #[test]
    fn test_export_to_temporal() {
        let workflow = WorkflowDefinition {
            id: "test-workflow".to_string(),
            name: "Test Workflow".to_string(),
            description: None,
            version: "1.0.0".to_string(),
            dag: WorkflowDag::new(),
        };

        let result = TemporalIntegration::export_workflow(&workflow);
        assert!(result.is_ok());

        let go_code = result.expect("Failed to export");
        assert!(go_code.contains("package workflows"));
        assert!(go_code.contains("go.temporal.io/sdk/workflow"));
    }

    #[test]
    fn test_export_activity_body_logs_task_id_and_runs_command() {
        let mut dag = WorkflowDag::new();
        dag.add_task(task_with_command(
            "download",
            Some("curl -O https://example.com/scene.tif"),
        ))
        .expect("add_task");
        dag.add_task(task_with_command("cloud-mask", None))
            .expect("add_task");

        let workflow = WorkflowDefinition {
            id: "satellite-pipeline".to_string(),
            name: "Satellite Pipeline".to_string(),
            description: None,
            version: "1.0.0".to_string(),
            dag,
        };

        let go_code =
            TemporalIntegration::export_workflow(&workflow).expect("export_workflow failed");

        assert!(!go_code.contains("TODO: Implement activity logic"));
        assert!(go_code.contains("logger.Info(\"Executing task\", \"task_id\", \"download\")"));
        assert!(go_code.contains("logger.Info(\"Executing task\", \"task_id\", \"cloud-mask\")"));
        assert!(
            go_code.contains(
                "exec.Command(\"sh\", \"-c\", \"curl -O https://example.com/scene.tif\")"
            )
        );
        assert!(go_code.contains("\"os/exec\""));
    }

    #[test]
    fn test_import_rejects_foreign_go_code() {
        let foreign = "package main\n\nfunc main() {}\n";
        let result = TemporalIntegration::import_workflow(foreign);
        assert!(result.is_err());
        let message = result.expect_err("expected error").to_string();
        assert!(
            message.contains("not a supported input format")
                || message.contains("TemporalIntegration::export_workflow")
        );
    }

    #[test]
    fn test_export_import_round_trip() {
        let mut dag = WorkflowDag::new();
        dag.add_task(task_with_command("download", Some("echo download")))
            .expect("add_task");
        dag.add_task(task_with_command("process", None))
            .expect("add_task");
        dag.add_task(task_with_command("upload", None))
            .expect("add_task");

        let original = WorkflowDefinition {
            id: "sentinel-2-pipeline".to_string(),
            name: "Sentinel-2 Pipeline".to_string(),
            description: Some("ignored on round trip".to_string()),
            version: "2.3.1".to_string(),
            dag,
        };

        let exported =
            TemporalIntegration::export_workflow(&original).expect("export_workflow failed");
        let imported =
            TemporalIntegration::import_workflow(&exported).expect("import_workflow failed");

        assert_eq!(imported.id, original.id);
        assert_eq!(imported.dag.task_count(), original.dag.task_count());
    }

    #[test]
    fn test_to_camel_case() {
        assert_eq!(
            TemporalIntegration::to_camel_case("test-workflow-id"),
            "TestWorkflowId"
        );
        assert_eq!(
            TemporalIntegration::to_camel_case("my_workflow"),
            "MyWorkflow"
        );
    }
}