af-workflow 0.4.0

//! Generic (non-business) node implementations.
//!
//! These are foundation nodes every product can share. External execution stays
//! behind the governed Agent platform; workflow nodes remain deterministic.
//!
//! Each node holds its raw config `Value` and resolves templates per-event in
//! `process` (matching the Python `resolved_config` discipline), then reads
//! fields. Required-key checks happen in the factory at compile time.

use std::collections::HashMap;

use async_trait::async_trait;
use serde_json::{Map, Value};

use crate::event::Event;
use crate::node::{StepNode, WorkflowContext};
use crate::registry::{NodeError, NodeRegistry};
use crate::result::StepResult;
use crate::template::resolve_config;

/// Register every builtin step + ingress type.
pub fn register_builtins(r: &mut NodeRegistry) {
    // Ingress sources — execution owned by the runner; registered by name so
    // the compiler recognizes them as legal chain heads.
    r.register_ingress("ingress.cron");
    r.register_ingress("ingress.fixed_rate");
    r.register_ingress("ingress.fixed_delay");
    r.register_ingress("ingress.event");
    r.register_ingress("ingress.manual");

    // Step nodes.
    r.register_step("transform.state_append", TransformStateAppend::factory);
    r.register_step("transform.state_set", TransformStateSet::factory);
    r.register_step("transform.state_read", TransformStateRead::factory);
    r.register_step("transform.set_fields", TransformSetFields::factory);
    r.register_step(
        "transform.state_publish_cross_branch",
        TransformStatePublishCrossBranch::factory,
    );
    r.register_step(
        "transform.state_read_cross_branch",
        TransformStateReadCrossBranch::factory,
    );
    r.register_step(
        "transform.state_append_cross_branch",
        TransformStateAppendCrossBranch::factory,
    );
    r.register_step("decision.consensus_voting", ConsensusVoting::factory);
    r.register_step("filter.required_fields", FilterRequiredFields::factory);
    r.register_step("sink.log", SinkLog::factory);

    register_builtin_schemas(r);
}

fn register_builtin_schemas(registry: &mut NodeRegistry) {
    use crate::registry::{FieldSpec as Field, FieldType, NodeSchema};

    let schema = |fields| NodeSchema { fields };
    registry.register_schema(
        "transform.state_append",
        schema(vec![
            Field::required("path", FieldType::String),
            Field::required("key", FieldType::String),
            Field::optional("max_len", FieldType::Number),
            Field::optional("ttl_seconds", FieldType::Number),
        ]),
    );
    registry.register_schema(
        "transform.state_set",
        schema(vec![
            Field::required("key", FieldType::String),
            Field::required("value", FieldType::Any),
            Field::optional("ttl_seconds", FieldType::Number),
        ]),
    );
    registry.register_schema(
        "transform.state_read",
        schema(vec![
            Field::required("key", FieldType::String),
            Field::required("into", FieldType::String),
            Field::optional("default", FieldType::Any),
        ]),
    );
    registry.register_schema(
        "transform.set_fields",
        schema(vec![Field::required("fields", FieldType::Object)]),
    );
    for node_type in [
        "transform.state_publish_cross_branch",
        "transform.state_append_cross_branch",
    ] {
        registry.register_schema(
            node_type,
            schema(vec![
                Field::required("path", FieldType::String),
                Field::required("key", FieldType::String),
                Field::optional("max_len", FieldType::Number),
                Field::optional("ttl_seconds", FieldType::Number),
            ]),
        );
    }
    registry.register_schema(
        "transform.state_read_cross_branch",
        schema(vec![
            Field::required("key", FieldType::String),
            Field::required("into", FieldType::String),
            Field::optional("default", FieldType::Any),
        ]),
    );
    registry.register_schema(
        "decision.consensus_voting",
        schema(vec![
            Field::required("signals_path", FieldType::String),
            Field::optional("mode", FieldType::String),
            Field::optional("quorum", FieldType::Number),
            Field::optional("threshold", FieldType::Number),
        ]),
    );
    registry.register_schema(
        "filter.required_fields",
        schema(vec![Field::required("fields", FieldType::Array)]),
    );
    registry.register_schema(
        "sink.log",
        schema(vec![
            Field::required("message", FieldType::String),
            Field::optional("level", FieldType::String),
        ]),
    );
}

// ── helpers ──────────────────────────────────────────────────────────────

fn require_keys(node_type: &str, config: &Value, keys: &[&str]) -> Result<(), NodeError> {
    let obj = config.as_object().ok_or_else(|| NodeError::InvalidConfig {
        node_type: node_type.to_string(),
        reason: "config must be an object".to_string(),
    })?;
    for k in keys {
        if !obj.contains_key(*k) {
            return Err(NodeError::InvalidConfig {
                node_type: node_type.to_string(),
                reason: format!("missing required key '{k}'"),
            });
        }
    }
    Ok(())
}

fn resolve(config: &Value, ctx: &WorkflowContext, event: &Event) -> Result<Value, StepResult> {
    let payload = Value::Object(event.payload.clone());
    resolve_config(config, &ctx.config, &ctx.instance_metadata, Some(&payload))
        .map_err(|e| StepResult::drop(e.to_string()))
}

fn cfg_str<'a>(cfg: &'a Value, key: &str) -> Option<&'a str> {
    cfg.get(key).and_then(|v| v.as_str())
}

fn cfg_usize(cfg: &Value, key: &str) -> Option<usize> {
    cfg.get(key).and_then(|v| v.as_u64()).map(|n| n as usize)
}

fn cross_branch_key(key: &str) -> String {
    format!("__cross_branch__.{key}")
}

// ── transform.state_append ─────────────────────────────────────────────────

/// Append `payload[path]` to a branch-scoped sliding-window list `key`.
struct TransformStateAppend {
    config: Value,
}

impl TransformStateAppend {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("transform.state_append", config, &["path", "key"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStateAppend {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(c) => c,
            Err(drop) => return drop,
        };
        let path = cfg_str(&cfg, "path").unwrap_or_default();
        let key = cfg_str(&cfg, "key").unwrap_or_default();
        let item = match event.payload_path(path) {
            Some(v) => v.clone(),
            None => return StepResult::drop(format!("state_append: path '{path}' missing")),
        };
        let max_len = cfg_usize(&cfg, "max_len");
        let ttl = cfg
            .get("ttl_seconds")
            .and_then(|v| v.as_u64())
            .map(std::time::Duration::from_secs);
        ctx.state.append(&ctx.scoped(key), item, max_len, ttl).await;
        StepResult::Pass(event.clone())
    }
}

// ── transform.state_set ────────────────────────────────────────────────────

/// Set a branch-scoped key to a (possibly templated) value.
struct TransformStateSet {
    config: Value,
}

impl TransformStateSet {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("transform.state_set", config, &["key", "value"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStateSet {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(c) => c,
            Err(drop) => return drop,
        };
        let key = cfg_str(&cfg, "key").unwrap_or_default().to_string();
        let value = cfg.get("value").cloned().unwrap_or(Value::Null);
        let ttl = cfg
            .get("ttl_seconds")
            .and_then(|v| v.as_u64())
            .map(std::time::Duration::from_secs);
        ctx.state.set(&ctx.scoped(&key), value, ttl).await;
        StepResult::Pass(event.clone())
    }
}

// ── transform.state_read ───────────────────────────────────────────────────

/// Read a branch-scoped key into `payload[into]`. Drops if the key is absent
/// and no `default` is configured.
struct TransformStateRead {
    config: Value,
}

impl TransformStateRead {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("transform.state_read", config, &["key", "into"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStateRead {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(c) => c,
            Err(drop) => return drop,
        };
        let key = cfg_str(&cfg, "key").unwrap_or_default();
        let into = cfg_str(&cfg, "into").unwrap_or_default().to_string();
        let value = match ctx.state.get(&ctx.scoped(key)).await {
            Some(v) => v,
            None => match cfg.get("default") {
                Some(d) => d.clone(),
                None => return StepResult::drop(format!("state_read: key '{key}' absent")),
            },
        };
        let mut patch = Map::new();
        patch.insert(into, value);
        StepResult::Pass(event.with_payload(patch))
    }
}

// ── explicit cross-branch state ────────────────────────────────────────────

struct TransformStatePublishCrossBranch {
    config: Value,
}

impl TransformStatePublishCrossBranch {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys(
            "transform.state_publish_cross_branch",
            config,
            &["path", "key"],
        )?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStatePublishCrossBranch {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(cfg) => cfg,
            Err(drop) => return drop,
        };
        let path = cfg_str(&cfg, "path").unwrap_or_default();
        let key = cfg_str(&cfg, "key").unwrap_or_default();
        let Some(value) = event.payload_path(path).cloned() else {
            return StepResult::drop(format!("cross-branch publish: path '{path}' missing"));
        };
        let ttl = cfg
            .get("ttl_seconds")
            .and_then(Value::as_u64)
            .map(std::time::Duration::from_secs);
        ctx.state.set(&cross_branch_key(key), value, ttl).await;
        StepResult::Pass(event.clone())
    }
}

struct TransformStateReadCrossBranch {
    config: Value,
}

impl TransformStateReadCrossBranch {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys(
            "transform.state_read_cross_branch",
            config,
            &["key", "into"],
        )?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStateReadCrossBranch {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(cfg) => cfg,
            Err(drop) => return drop,
        };
        let key = cfg_str(&cfg, "key").unwrap_or_default();
        let target = cfg_str(&cfg, "into").unwrap_or_default().to_string();
        let value = match ctx.state.get(&cross_branch_key(key)).await {
            Some(value) => value,
            None => match cfg.get("default") {
                Some(value) => value.clone(),
                None => return StepResult::drop(format!("cross-branch key '{key}' absent")),
            },
        };
        let mut patch = Map::new();
        patch.insert(target, value);
        StepResult::Pass(event.with_payload(patch))
    }
}

struct TransformStateAppendCrossBranch {
    config: Value,
}

impl TransformStateAppendCrossBranch {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys(
            "transform.state_append_cross_branch",
            config,
            &["path", "key"],
        )?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformStateAppendCrossBranch {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(cfg) => cfg,
            Err(drop) => return drop,
        };
        let path = cfg_str(&cfg, "path").unwrap_or_default();
        let key = cfg_str(&cfg, "key").unwrap_or_default();
        let Some(value) = event.payload_path(path).cloned() else {
            return StepResult::drop(format!("cross-branch append: path '{path}' missing"));
        };
        let ttl = cfg
            .get("ttl_seconds")
            .and_then(Value::as_u64)
            .map(std::time::Duration::from_secs);
        ctx.state
            .append(
                &cross_branch_key(key),
                value,
                cfg_usize(&cfg, "max_len"),
                ttl,
            )
            .await;
        StepResult::Pass(event.clone())
    }
}

// ── transform.set_fields ───────────────────────────────────────────────────

/// Merge a static/templated `fields` object into the event payload.
struct TransformSetFields {
    config: Value,
}

impl TransformSetFields {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("transform.set_fields", config, &["fields"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for TransformSetFields {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(c) => c,
            Err(drop) => return drop,
        };
        let fields = match cfg.get("fields").and_then(|v| v.as_object()) {
            Some(f) => f.clone(),
            None => return StepResult::drop("set_fields: 'fields' must be an object"),
        };
        StepResult::Pass(event.with_payload(fields))
    }
}

// ── generic consensus ──────────────────────────────────────────────────────

/// Vote over `{choice, weight?}` objects. Quorum counts votes; weighted mode
/// selects the highest total only when its lead exceeds `threshold`.
struct ConsensusVoting {
    config: Value,
}

impl ConsensusVoting {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("decision.consensus_voting", config, &["signals_path"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for ConsensusVoting {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(cfg) => cfg,
            Err(drop) => return drop,
        };
        let path = cfg_str(&cfg, "signals_path").unwrap_or("signals");
        let Some(signals) = event.payload_path(path).and_then(Value::as_array) else {
            return StepResult::drop("consensus: no signals");
        };
        if signals.is_empty() {
            return StepResult::drop("consensus: no signals");
        }

        let mut totals: HashMap<&str, (u64, f64)> = HashMap::new();
        for signal in signals {
            let Some(choice) = signal.get("choice").and_then(Value::as_str) else {
                continue;
            };
            let entry = totals.entry(choice).or_default();
            entry.0 += 1;
            entry.1 += signal.get("weight").and_then(Value::as_f64).unwrap_or(1.0);
        }
        if totals.is_empty() {
            return StepResult::drop("consensus: no valid choices");
        }

        let weighted = cfg_str(&cfg, "mode") == Some("weighted");
        let mut ranked = totals.into_iter().collect::<Vec<_>>();
        ranked.sort_by(|left, right| {
            let left_score = if weighted {
                left.1 .1
            } else {
                left.1 .0 as f64
            };
            let right_score = if weighted {
                right.1 .1
            } else {
                right.1 .0 as f64
            };
            right_score.total_cmp(&left_score)
        });
        let (choice, (votes, weight)) = ranked[0];
        let score = if weighted { weight } else { votes as f64 };
        let runner_up = ranked
            .get(1)
            .map(|(_, (votes, weight))| if weighted { *weight } else { *votes as f64 })
            .unwrap_or(0.0);
        let required = if weighted {
            cfg.get("threshold").and_then(Value::as_f64).unwrap_or(0.0)
        } else {
            cfg.get("quorum").and_then(Value::as_u64).unwrap_or(1) as f64
        };
        if (!weighted && score < required) || (weighted && score - runner_up <= required) {
            return StepResult::drop("consensus: threshold not met");
        }
        if score == runner_up {
            return StepResult::drop("consensus: tie");
        }

        let mut patch = Map::new();
        patch.insert(
            "consensus".into(),
            serde_json::json!({ "choice": choice, "score": score }),
        );
        StepResult::Pass(event.with_payload(patch))
    }
}

// ── filter.required_fields ─────────────────────────────────────────────────

/// Pass only if every listed payload path is present; otherwise Drop.
struct FilterRequiredFields {
    fields: Vec<String>,
}

impl FilterRequiredFields {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        let fields = config
            .get("fields")
            .and_then(|v| v.as_array())
            .ok_or_else(|| NodeError::InvalidConfig {
                node_type: "filter.required_fields".to_string(),
                reason: "'fields' must be an array of paths".to_string(),
            })?
            .iter()
            .filter_map(|v| v.as_str().map(str::to_string))
            .collect();
        Ok(Box::new(Self { fields }))
    }
}

#[async_trait]
impl StepNode for FilterRequiredFields {
    async fn process(&self, event: &Event, _ctx: &WorkflowContext) -> StepResult {
        for path in &self.fields {
            if event.payload_path(path).is_none() {
                return StepResult::drop(format!("required field '{path}' missing"));
            }
        }
        StepResult::Pass(event.clone())
    }
}

// ── sink.log ───────────────────────────────────────────────────────────────

/// Emit one structured log line and stop propagation (terminal).
struct SinkLog {
    config: Value,
}

impl SinkLog {
    fn factory(config: &Value) -> Result<Box<dyn StepNode>, NodeError> {
        require_keys("sink.log", config, &["message"])?;
        Ok(Box::new(Self {
            config: config.clone(),
        }))
    }
}

#[async_trait]
impl StepNode for SinkLog {
    async fn process(&self, event: &Event, ctx: &WorkflowContext) -> StepResult {
        let cfg = match resolve(&self.config, ctx, event) {
            Ok(c) => c,
            Err(drop) => return drop,
        };
        let message = cfg_str(&cfg, "message").unwrap_or("sink.log");
        let level = cfg_str(&cfg, "level").unwrap_or("info");
        match level {
            "debug" => tracing::debug!(target: "wf.sink", event_id = %event.id, "{message}"),
            "warning" | "warn" => {
                tracing::warn!(target: "wf.sink", event_id = %event.id, "{message}")
            }
            "error" => tracing::error!(target: "wf.sink", event_id = %event.id, "{message}"),
            _ => tracing::info!(target: "wf.sink", event_id = %event.id, "{message}"),
        }
        StepResult::drop("sink.log emitted")
    }
}