call.rs

use chrono::Utc;

use std::collections::HashMap;

use crate::cache::{CacheEntry, compute_cache_key};
use crate::context::Context;
use crate::output;

use super::super::{DurableEngine, Result};
use crate::task_ext::call_parts;

/// Execute a Call task - invokes functions (user-defined, catalog, or built-in protocols)
pub async fn exec_call_task(
    engine: &DurableEngine,
    task_name: &str,
    call_task: &serverless_workflow_core::models::task::CallTaskDefinition,
    ctx: &Context,
) -> Result<serde_json::Value> {
    let (function_name, with_params) = call_parts(call_task)?;

    // Evaluate expressions in with parameters
    let current_data = ctx.state.data.read().await.clone();
    let evaluated_with_params_value = crate::expressions::evaluate_value_with_input(
        &serde_json::to_value(&with_params)?,
        &current_data,
        &ctx.metadata.initial_input,
    )?;

    // Convert back to HashMap
    let evaluated_with_params: HashMap<String, serde_json::Value> =
        serde_json::from_value(evaluated_with_params_value.clone())?;

    let params = evaluated_with_params_value.clone();
    let cache_key = compute_cache_key(task_name, &params);

    if let Some(cached) = ctx.services.cache.get(&cache_key).await? {
        output::format_cache_hit(
            task_name,
            &cache_key,
            Some(&cached.timestamp.format("%Y-%m-%d %H:%M:%S").to_string()),
        );
        return Ok(cached.output);
    }

    output::format_cache_miss(task_name, &cache_key);

    // Note: TaskStarted event is now emitted centrally in exec_task()

    // First check user-defined functions
    let function_result = if let Some(function_def) = ctx
        .metadata
        .workflow
        .use_
        .as_ref()
        .and_then(|use_| use_.functions.as_ref())
        .and_then(|funcs| funcs.get(function_name))
    {
        // User-defined function
        use serverless_workflow_core::models::task::TaskDefinition;
        let (call_type, func_params) = match function_def {
            TaskDefinition::Call(call_def) => call_parts(call_def)?,
            TaskDefinition::Do(_)
            | TaskDefinition::Fork(_)
            | TaskDefinition::Emit(_)
            | TaskDefinition::For(_)
            | TaskDefinition::Listen(_)
            | TaskDefinition::Raise(_)
            | TaskDefinition::Run(_)
            | TaskDefinition::Set(_)
            | TaskDefinition::Switch(_)
            | TaskDefinition::Try(_)
            | TaskDefinition::Wait(_) => {
                return Err(super::super::Error::Configuration {
                    message: format!("Function {function_name} is not a call task"),
                });
            }
        };
        let mut merged_params = func_params;
        merged_params.extend(evaluated_with_params.clone());

        let executor =
            engine
                .executors
                .get(call_type)
                .ok_or(super::super::Error::TaskExecution {
                    message: format!("No executor for call type: {call_type}"),
                })?;

        let final_params = serde_json::to_value(&merged_params)?;
        executor.exec(task_name, &final_params, ctx, None).await?
    } else if let Some(catalog_result) = engine
        .try_load_catalog_function(function_name, &evaluated_with_params, ctx)
        .await?
    {
        // Catalog function - execute as nested workflow
        catalog_result
    } else {
        // Built-in protocol
        let executor =
            engine
                .executors
                .get(function_name)
                .ok_or(super::super::Error::TaskExecution {
                    message: format!("No executor for call type: {function_name}"),
                })?;

        let final_params = serde_json::to_value(&evaluated_with_params)?;
        executor.exec(task_name, &final_params, ctx, None).await?
    };

    let mut result = function_result;

    // Apply output filtering if specified
    if let Some(output_config) = &call_task.common().output
        && let Some(as_value) = &output_config.as_
    {
        let task_input = ctx.state.task_input.read().await.clone();

        // Handle both string (expression) and object (field mapping) forms
        if let Some(expr_str) = as_value.as_str() {
            // String form: jq expression (not wrapped in ${})
            // Evaluate the jq expression on the result with access to $input
            result = crate::expressions::evaluate_jq_expression_with_context(
                expr_str,
                &result,
                &task_input,
            )?;
        } else if let Some(as_obj) = as_value.as_object() {
            // Object form: map fields with expressions (may be wrapped in ${})
            let mut transformed = serde_json::Map::new();
            for (key, value) in as_obj {
                if let Some(expr_str) = value.as_str() {
                    let evaluated = crate::expressions::evaluate_expression_with_input(
                        expr_str,
                        &result,
                        &task_input,
                    )?;
                    transformed.insert(key.clone(), evaluated);
                }
            }
            result = serde_json::Value::Object(transformed);
        }
    }

    let cache_entry = CacheEntry {
        key: cache_key.clone(),
        inputs: params,
        output: result.clone(),
        timestamp: Utc::now(),
    };
    ctx.services.cache.set(cache_entry).await?;

    Ok(result)
}