lashlang/

trigger.rs

use std::collections::{BTreeMap, BTreeSet};

use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::artifact::{ModuleArtifact, ModuleRef, ProcessRef, RequiredSurfaceRef};
use crate::ast::{AstString, Expr, TypeExpr, TypeField, format_type_expr};
use crate::linker::{NamedDataType, ResourceCatalog};
use crate::runtime::{
    LASH_HOST_VALUE_KEY, LASH_HOST_VALUE_TYPE_KEY, LASH_MODULE_REF_KEY, LASH_PROCESS_NAME_KEY,
    LASH_PROCESS_REF_KEY, LASH_PROCESS_VALUE_KEY, LASH_REQUIRED_SURFACE_REF_KEY,
};

const TRIGGERS_RESOURCE_TYPE: &str = "Triggers";
const TRIGGERS_ALIAS: &str = "triggers";
const TRIGGER_REGISTRATION_TYPE: &str = "TriggerRegistration";
pub const LASH_TRIGGER_EVENT_KEY: &str = "$lash.trigger.event";

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TriggerHostOperation {
    Register,
    List,
    Cancel,
}

impl TriggerHostOperation {
    pub const fn host_operation(self) -> &'static str {
        match self {
            Self::Register => "triggers.register",
            Self::List => "triggers.list",
            Self::Cancel => "triggers.cancel",
        }
    }

    pub const fn receiver_method(self) -> &'static str {
        match self {
            Self::Register => "register",
            Self::List => "list",
            Self::Cancel => "cancel",
        }
    }

    pub fn from_host_operation(operation: &str) -> Option<Self> {
        [Self::Register, Self::List, Self::Cancel]
            .into_iter()
            .find(|candidate| candidate.host_operation() == operation)
    }

    pub fn from_receiver_method(operation: &str) -> Option<Self> {
        [Self::Register, Self::List, Self::Cancel]
            .into_iter()
            .find(|candidate| candidate.receiver_method() == operation)
    }

    pub fn input_ty(self) -> TypeExpr {
        match self {
            Self::Register => TypeExpr::Object(vec![
                required_field("source", TypeExpr::Dict),
                required_field(
                    "target",
                    TypeExpr::Process {
                        input: Box::new(TypeExpr::Any),
                        output: Box::new(TypeExpr::Any),
                        input_count: 1,
                    },
                ),
                required_field("inputs", TypeExpr::Dict),
                optional_field("name", TypeExpr::Str),
            ]),
            Self::List => TypeExpr::Object(vec![
                optional_field(
                    "target",
                    TypeExpr::Process {
                        input: Box::new(TypeExpr::Any),
                        output: Box::new(TypeExpr::Any),
                        input_count: 1,
                    },
                ),
                optional_field("name", TypeExpr::Str),
                optional_field("source_type", TypeExpr::Str),
                optional_field("enabled", TypeExpr::Bool),
            ]),
            Self::Cancel => TypeExpr::Object(vec![required_field(
                "handle",
                TypeExpr::TriggerHandle(Box::new(TypeExpr::Any)),
            )]),
        }
    }

    pub fn output_ty(self) -> TypeExpr {
        match self {
            Self::Register => TypeExpr::TriggerHandle(Box::new(TypeExpr::Any)),
            Self::List => TypeExpr::List(Box::new(TypeExpr::Ref(TRIGGER_REGISTRATION_TYPE.into()))),
            Self::Cancel => TypeExpr::Bool,
        }
    }
}

pub fn is_trigger_resource_type(resource_type: &str) -> bool {
    resource_type == TRIGGERS_RESOURCE_TYPE
}

pub fn add_trigger_resource_operations(catalog: &mut ResourceCatalog) {
    for operation in [
        TriggerHostOperation::Register,
        TriggerHostOperation::List,
        TriggerHostOperation::Cancel,
    ] {
        catalog.add_module_operation(
            [TRIGGERS_ALIAS],
            TRIGGERS_RESOURCE_TYPE,
            operation.receiver_method(),
            operation.host_operation(),
            operation.input_ty(),
            operation.output_ty(),
        );
    }
}

fn required_field(name: &'static str, ty: TypeExpr) -> TypeField {
    TypeField {
        name: name.into(),
        ty,
        optional: false,
    }
}

fn optional_field(name: &'static str, ty: TypeExpr) -> TypeField {
    TypeField {
        name: name.into(),
        ty,
        optional: true,
    }
}

pub struct TriggerRegistrationCall<'expr> {
    pub source: &'expr Expr,
    pub target: &'expr Expr,
    pub inputs: &'expr Expr,
    pub name: Option<&'expr Expr>,
}

pub struct TriggerListCall<'expr> {
    pub entries: &'expr [(AstString, Expr)],
}

pub struct TriggerCancelCall<'expr> {
    pub handle: &'expr Expr,
}

pub fn register_call_args(
    args: &[Expr],
) -> Result<TriggerRegistrationCall<'_>, TriggerCallShapeError> {
    let entries = record_entries(args).ok_or(TriggerCallShapeError::Registration)?;
    Ok(TriggerRegistrationCall {
        source: required_entry(entries, "source").ok_or(TriggerCallShapeError::Registration)?,
        target: required_entry(entries, "target").ok_or(TriggerCallShapeError::Registration)?,
        inputs: required_entry(entries, "inputs").ok_or(TriggerCallShapeError::Registration)?,
        name: required_entry(entries, "name"),
    })
}

pub fn list_call_args(args: &[Expr]) -> Result<TriggerListCall<'_>, TriggerCallShapeError> {
    let entries = record_entries(args).ok_or(TriggerCallShapeError::List)?;
    for (name, _) in entries {
        match name.as_str() {
            "target" | "name" | "source_type" | "enabled" => {}
            _ => return Err(TriggerCallShapeError::List),
        }
    }
    Ok(TriggerListCall { entries })
}

pub fn cancel_call_args(args: &[Expr]) -> Result<TriggerCancelCall<'_>, TriggerCallShapeError> {
    let entries = record_entries(args).ok_or(TriggerCallShapeError::Cancel)?;
    Ok(TriggerCancelCall {
        handle: required_entry(entries, "handle").ok_or(TriggerCallShapeError::Cancel)?,
    })
}

fn record_entries(args: &[Expr]) -> Option<&[(AstString, Expr)]> {
    let [Expr::Record(entries)] = args else {
        return None;
    };
    Some(entries)
}

fn required_entry<'expr>(entries: &'expr [(AstString, Expr)], name: &str) -> Option<&'expr Expr> {
    entries
        .iter()
        .find_map(|(entry_name, expr)| (entry_name.as_str() == name).then_some(expr))
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TriggerCallShapeError {
    Registration,
    List,
    Cancel,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct TriggerRegistrationRequest {
    pub source: HostValue,
    pub target: TriggerTargetIdentity,
    pub inputs: TriggerInputTemplate,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
}

impl TriggerRegistrationRequest {
    pub fn decode(request: &serde_json::Value) -> Result<Self, TriggerRequestDecodeError> {
        let operation = TriggerHostOperation::Register;
        Ok(Self {
            source: HostValue::decode(required_json_field(request, "source", operation)?)
                .map_err(TriggerRequestDecodeError::from)?,
            target: TriggerTargetIdentity::decode(
                required_json_field(request, "target", operation)?,
                "trigger target",
            )?,
            inputs: TriggerInputTemplate::decode(required_json_field(
                request, "inputs", operation,
            )?)?,
            name: request
                .get("name")
                .and_then(serde_json::Value::as_str)
                .map(ToOwned::to_owned),
        })
    }
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct TriggerInputTemplate {
    entries: BTreeMap<String, TriggerInputBinding>,
}

impl TriggerInputTemplate {
    pub fn new(entries: BTreeMap<String, TriggerInputBinding>) -> Self {
        Self { entries }
    }

    pub fn decode(value: &serde_json::Value) -> Result<Self, TriggerRequestDecodeError> {
        let map = value
            .as_object()
            .ok_or_else(|| TriggerRequestDecodeError::InvalidField {
                operation: TriggerHostOperation::Register.host_operation(),
                field: "inputs",
                message: "expected an object mapping process params to values".to_string(),
            })?;
        let mut entries = BTreeMap::new();
        for (name, value) in map {
            let binding = if is_trigger_event_placeholder_value(value) {
                TriggerInputBinding::Event
            } else {
                TriggerInputBinding::Fixed {
                    value: value.clone(),
                }
            };
            entries.insert(name.clone(), binding);
        }
        Ok(Self { entries })
    }

    pub fn entries(&self) -> impl Iterator<Item = (&str, &TriggerInputBinding)> {
        self.entries
            .iter()
            .map(|(name, binding)| (name.as_str(), binding))
    }

    pub fn get(&self, name: &str) -> Option<&TriggerInputBinding> {
        self.entries.get(name)
    }

    pub fn contains_event(&self) -> bool {
        self.entries
            .values()
            .any(|binding| matches!(binding, TriggerInputBinding::Event))
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TriggerInputBinding {
    Event,
    Fixed { value: serde_json::Value },
}

impl TriggerInputBinding {
    pub fn as_fixed(&self) -> Option<&serde_json::Value> {
        match self {
            Self::Fixed { value } => Some(value),
            Self::Event => None,
        }
    }
}

pub fn trigger_event_placeholder_expr() -> Expr {
    Expr::Record(vec![(LASH_TRIGGER_EVENT_KEY.into(), Expr::Bool(true))])
}

fn is_trigger_event_placeholder_value(value: &serde_json::Value) -> bool {
    let Some(map) = value.as_object() else {
        return false;
    };
    map.len() == 1
        && map
            .get(LASH_TRIGGER_EVENT_KEY)
            .and_then(serde_json::Value::as_bool)
            == Some(true)
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct TriggerListRequest {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target: Option<TriggerTargetIdentity>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_type: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub enabled: Option<bool>,
}

impl TriggerListRequest {
    pub fn decode(request: &serde_json::Value) -> Result<Self, TriggerRequestDecodeError> {
        let map = request
            .as_object()
            .ok_or_else(|| TriggerRequestDecodeError::InvalidField {
                operation: TriggerHostOperation::List.host_operation(),
                field: "filters",
                message: "expected a record of trigger filters".to_string(),
            })?;
        for key in map.keys() {
            match key.as_str() {
                "target" | "name" | "source_type" | "enabled" => {}
                _ => {
                    return Err(TriggerRequestDecodeError::InvalidField {
                        operation: TriggerHostOperation::List.host_operation(),
                        field: "filters",
                        message: format!("unknown filter `{key}`"),
                    });
                }
            }
        }
        Ok(Self {
            target: request
                .get("target")
                .map(|value| TriggerTargetIdentity::decode(value, "triggers.list target"))
                .transpose()?,
            name: optional_string_filter(request, "name", TriggerHostOperation::List)?,
            source_type: optional_string_filter(
                request,
                "source_type",
                TriggerHostOperation::List,
            )?,
            enabled: optional_bool_filter(request, "enabled", TriggerHostOperation::List)?,
        })
    }
}

fn optional_string_filter(
    request: &serde_json::Value,
    field: &'static str,
    operation: TriggerHostOperation,
) -> Result<Option<String>, TriggerRequestDecodeError> {
    request
        .get(field)
        .map(|value| {
            value.as_str().map(ToOwned::to_owned).ok_or_else(|| {
                TriggerRequestDecodeError::InvalidField {
                    operation: operation.host_operation(),
                    field,
                    message: "expected a string".to_string(),
                }
            })
        })
        .transpose()
}

fn optional_bool_filter(
    request: &serde_json::Value,
    field: &'static str,
    operation: TriggerHostOperation,
) -> Result<Option<bool>, TriggerRequestDecodeError> {
    request
        .get(field)
        .map(|value| {
            value
                .as_bool()
                .ok_or_else(|| TriggerRequestDecodeError::InvalidField {
                    operation: operation.host_operation(),
                    field,
                    message: "expected a boolean".to_string(),
                })
        })
        .transpose()
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TriggerCancelRequest {
    pub handle: String,
}

impl TriggerCancelRequest {
    pub fn decode(request: &serde_json::Value) -> Result<Self, TriggerRequestDecodeError> {
        let value = required_json_field(request, "handle", TriggerHostOperation::Cancel)?;
        let handle = value
            .as_str()
            .map(ToOwned::to_owned)
            .or_else(|| {
                value
                    .get("id")
                    .and_then(serde_json::Value::as_str)
                    .map(ToOwned::to_owned)
            })
            .ok_or_else(|| TriggerRequestDecodeError::InvalidField {
                operation: TriggerHostOperation::Cancel.host_operation(),
                field: "handle",
                message: "expected trigger handle string or object with `id`".to_string(),
            })?;
        Ok(Self { handle })
    }
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct HostValue {
    pub source_type: String,
    pub value: serde_json::Value,
}

impl HostValue {
    pub fn new(source_type: impl Into<String>, value: serde_json::Value) -> Self {
        Self {
            source_type: source_type.into(),
            value,
        }
    }

    pub fn decode(source: &serde_json::Value) -> Result<Self, HostValueError> {
        let source_type = source
            .get(LASH_HOST_VALUE_TYPE_KEY)
            .and_then(serde_json::Value::as_str)
            .map(ToOwned::to_owned)
            .ok_or(HostValueError::InvalidHostValue)?;
        let value = source
            .get(LASH_HOST_VALUE_KEY)
            .cloned()
            .ok_or(HostValueError::InvalidHostValue)?;
        Ok(Self { source_type, value })
    }

    pub fn encode(
        source_type: impl Into<String>,
        value: impl Serialize,
    ) -> Result<serde_json::Value, HostValueError> {
        let source_type = source_type.into();
        let value =
            serde_json::to_value(value).map_err(|err| HostValueError::MalformedPayload {
                source_type: source_type.clone(),
                message: err.to_string(),
            })?;
        Ok(Self::new(source_type, value).to_json())
    }

    pub fn decode_as<T: serde::de::DeserializeOwned>(
        &self,
        resources: &ResourceCatalog,
    ) -> Result<T, HostValueError> {
        resources.decode_host_value_as(&self.source_type, self.value.clone())
    }

    pub fn to_json(&self) -> serde_json::Value {
        serde_json::json!({
            LASH_HOST_VALUE_TYPE_KEY: self.source_type,
            LASH_HOST_VALUE_KEY: self.value,
        })
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Error)]
pub enum HostValueError {
    #[error("host value must be a host value constructor result")]
    InvalidHostValue,
    #[error("host value `{source_type}` is not declared in the resource catalog")]
    UnknownSourceType { source_type: String },
    #[error("host value `{source_type}` payload is invalid: {message}")]
    MalformedPayload {
        source_type: String,
        message: String,
    },
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TriggerTargetIdentity {
    pub module_ref: ModuleRef,
    pub required_surface_ref: RequiredSurfaceRef,
    pub process_ref: ProcessRef,
    pub process_name: String,
}

impl TriggerTargetIdentity {
    pub fn decode(
        value: &serde_json::Value,
        label: &'static str,
    ) -> Result<Self, TriggerRequestDecodeError> {
        if value
            .get(LASH_PROCESS_VALUE_KEY)
            .and_then(serde_json::Value::as_bool)
            != Some(true)
        {
            return Err(TriggerRequestDecodeError::InvalidTarget {
                label,
                message: "must be a process value".to_string(),
            });
        }
        Ok(Self {
            module_ref: decode_json_field(value, LASH_MODULE_REF_KEY, label)?,
            required_surface_ref: decode_json_field(value, LASH_REQUIRED_SURFACE_REF_KEY, label)?,
            process_ref: decode_json_field(value, LASH_PROCESS_REF_KEY, label)?,
            process_name: value
                .get(LASH_PROCESS_NAME_KEY)
                .and_then(serde_json::Value::as_str)
                .ok_or_else(|| TriggerRequestDecodeError::InvalidTarget {
                    label,
                    message: format!("missing {LASH_PROCESS_NAME_KEY}"),
                })?
                .to_string(),
        })
    }

    pub fn matches(
        &self,
        module_ref: &ModuleRef,
        required_surface_ref: &RequiredSurfaceRef,
        process_ref: &ProcessRef,
        process_name: &str,
    ) -> bool {
        self.module_ref == *module_ref
            && self.required_surface_ref == *required_surface_ref
            && self.process_ref == *process_ref
            && self.process_name == process_name
    }
}

pub fn event_type_for_source(
    resources: &ResourceCatalog,
    source_type: &str,
) -> Result<NamedDataType, TriggerRequestDecodeError> {
    resources
        .resolve_trigger_source(source_type)
        .map(|binding| binding.event_type().clone())
        .ok_or_else(|| TriggerRequestDecodeError::UnknownSourceType {
            source_type: source_type.to_string(),
        })
}

#[derive(Clone, Debug, PartialEq, Eq, Error)]
pub enum TriggerRequestDecodeError {
    #[error("{operation} requires `{field}`")]
    MissingField {
        operation: &'static str,
        field: &'static str,
    },
    #[error("{operation} field `{field}` is invalid: {message}")]
    InvalidField {
        operation: &'static str,
        field: &'static str,
        message: String,
    },
    #[error("trigger source must be a host value constructor result")]
    InvalidSource,
    #[error("{label} {message}")]
    InvalidTarget {
        label: &'static str,
        message: String,
    },
    #[error("host value `{source_type}` is not registered as a trigger source")]
    UnknownSourceType { source_type: String },
}

impl From<HostValueError> for TriggerRequestDecodeError {
    fn from(err: HostValueError) -> Self {
        match err {
            HostValueError::InvalidHostValue => Self::InvalidSource,
            HostValueError::UnknownSourceType { source_type } => {
                Self::UnknownSourceType { source_type }
            }
            HostValueError::MalformedPayload { message, .. } => Self::InvalidField {
                operation: TriggerHostOperation::Register.host_operation(),
                field: "source",
                message,
            },
        }
    }
}

fn required_json_field<'json>(
    request: &'json serde_json::Value,
    field: &'static str,
    operation: TriggerHostOperation,
) -> Result<&'json serde_json::Value, TriggerRequestDecodeError> {
    request
        .get(field)
        .ok_or_else(|| TriggerRequestDecodeError::MissingField {
            operation: operation.host_operation(),
            field,
        })
}

fn decode_json_field<T: serde::de::DeserializeOwned>(
    value: &serde_json::Value,
    field: &'static str,
    label: &'static str,
) -> Result<T, TriggerRequestDecodeError> {
    serde_json::from_value(value.get(field).cloned().ok_or_else(|| {
        TriggerRequestDecodeError::InvalidTarget {
            label,
            message: format!("missing {field}"),
        }
    })?)
    .map_err(|err| TriggerRequestDecodeError::InvalidTarget {
        label,
        message: format!("invalid {field}: {err}"),
    })
}

#[derive(Clone, Debug, PartialEq)]
pub struct TriggerTargetValidation {
    pub inputs: TriggerInputTemplate,
    pub event_ty: TypeExpr,
}

pub fn validate_trigger_target(
    target: &TriggerTargetIdentity,
    event_ty: &NamedDataType,
    inputs: &TriggerInputTemplate,
    artifact: &ModuleArtifact,
) -> Result<TriggerTargetValidation, TriggerTargetValidationError> {
    if artifact.required_surface_ref != target.required_surface_ref {
        return Err(TriggerTargetValidationError::RequiredSurfaceMismatch {
            process_name: target.process_name.clone(),
            target_surface: target.required_surface_ref.to_string(),
            artifact_surface: artifact.required_surface_ref.to_string(),
        });
    }
    let Some(exported_process_name) = artifact.process_name_for_ref(&target.process_ref) else {
        return Err(TriggerTargetValidationError::ProcessRefMismatch {
            module_ref: target.module_ref.to_string(),
            process_name: target.process_name.clone(),
            process_ref: format!("{:?}", target.process_ref),
        });
    };
    if exported_process_name != target.process_name {
        return Err(TriggerTargetValidationError::ProcessRefMismatch {
            module_ref: target.module_ref.to_string(),
            process_name: target.process_name.clone(),
            process_ref: format!("{:?}", target.process_ref),
        });
    }
    let process = artifact
        .canonical_ir
        .process(exported_process_name)
        .ok_or_else(|| TriggerTargetValidationError::MissingProcess {
            module_ref: target.module_ref.to_string(),
            process_name: target.process_name.clone(),
        })?;
    for (input_name, _) in inputs.entries() {
        if !process
            .params
            .iter()
            .any(|param| param.name.as_str() == input_name)
        {
            return Err(TriggerTargetValidationError::UnknownInput {
                process_name: target.process_name.clone(),
                input: input_name.to_string(),
            });
        }
    }
    if !inputs.contains_event() {
        return Err(TriggerTargetValidationError::MissingEventInput {
            process_name: target.process_name.clone(),
        });
    }
    let aliases = type_aliases(artifact);
    let event_ty = resolve_type_refs(
        &event_ty.to_ref_ty(),
        &aliases,
        &artifact.required_surface.resources,
    );
    for param in &process.params {
        let Some(input) = inputs.get(param.name.as_str()) else {
            return Err(TriggerTargetValidationError::MissingInput {
                process_name: target.process_name.clone(),
                input: param.name.to_string(),
            });
        };
        let input_ty = resolve_type_refs(&param.ty, &aliases, &artifact.required_surface.resources);
        match input {
            TriggerInputBinding::Event => {
                if !is_resolved_type_assignable(&event_ty, &input_ty) {
                    return Err(TriggerTargetValidationError::EventMismatch {
                        event: format_type_expr(&event_ty),
                        process_name: target.process_name.clone(),
                        input_name: param.name.to_string(),
                        input: format_type_expr(&input_ty),
                    });
                }
            }
            TriggerInputBinding::Fixed { value } => {
                validate_fixed_input_value(
                    value,
                    &input_ty,
                    &artifact.required_surface.resources,
                    target.process_name.as_str(),
                    param.name.as_str(),
                )?;
            }
        }
    }
    Ok(TriggerTargetValidation {
        inputs: inputs.clone(),
        event_ty,
    })
}

fn validate_fixed_input_value(
    value: &serde_json::Value,
    input_ty: &TypeExpr,
    resources: &ResourceCatalog,
    process_name: &str,
    input_name: &str,
) -> Result<(), TriggerTargetValidationError> {
    let TypeExpr::Ref(resource_type) = input_ty else {
        return Ok(());
    };
    if !resources.has_resource_type(resource_type.as_str()) {
        return Ok(());
    }
    match crate::runtime::from_json(value.clone()) {
        crate::Value::Resource(handle) if handle.resource_type == *resource_type => Ok(()),
        crate::Value::Resource(handle) => Err(TriggerTargetValidationError::FixedInputMismatch {
            process_name: process_name.to_string(),
            input: input_name.to_string(),
            expected: resource_type.to_string(),
            actual: handle.resource_type,
        }),
        _ => Err(TriggerTargetValidationError::FixedInputMismatch {
            process_name: process_name.to_string(),
            input: input_name.to_string(),
            expected: resource_type.to_string(),
            actual: "value".to_string(),
        }),
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Error)]
pub enum TriggerTargetValidationError {
    #[error(
        "trigger target `{process_name}` required surface mismatch: target has {target_surface}, artifact has {artifact_surface}"
    )]
    RequiredSurfaceMismatch {
        process_name: String,
        target_surface: String,
        artifact_surface: String,
    },
    #[error(
        "trigger target artifact `{module_ref}` does not export process `{process_name}` as requested ref {process_ref}"
    )]
    ProcessRefMismatch {
        module_ref: String,
        process_name: String,
        process_ref: String,
    },
    #[error("trigger target artifact `{module_ref}` is missing process `{process_name}`")]
    MissingProcess {
        module_ref: String,
        process_name: String,
    },
    #[error("trigger target `{process_name}` input `{input}` is not mapped")]
    MissingInput { process_name: String, input: String },
    #[error("trigger target `{process_name}` has no input `{input}`")]
    UnknownInput { process_name: String, input: String },
    #[error("trigger target `{process_name}` inputs must map at least one param to trigger.event")]
    MissingEventInput { process_name: String },
    #[error(
        "trigger source emits {event}, but target `{process_name}` input `{input_name}` expects {input}"
    )]
    EventMismatch {
        event: String,
        process_name: String,
        input_name: String,
        input: String,
    },
    #[error(
        "trigger target `{process_name}` input `{input}` has incompatible fixed authority type: expected {expected}, got {actual}"
    )]
    FixedInputMismatch {
        process_name: String,
        input: String,
        expected: String,
        actual: String,
    },
}

fn resolve_type_refs(
    ty: &TypeExpr,
    aliases: &BTreeMap<String, TypeExpr>,
    resources: &ResourceCatalog,
) -> TypeExpr {
    resolve_type_refs_inner(ty, aliases, Some(resources), &mut BTreeSet::new())
}

fn resolve_type_refs_inner(
    ty: &TypeExpr,
    aliases: &BTreeMap<String, TypeExpr>,
    resources: Option<&ResourceCatalog>,
    seen: &mut BTreeSet<String>,
) -> TypeExpr {
    match ty {
        TypeExpr::Ref(name) if seen.insert(name.to_string()) => {
            let resolved = if let Some(ty) = aliases.get(name.as_str()) {
                resolve_type_refs_inner(ty, aliases, resources, seen)
            } else if let Some(data_type) =
                resources.and_then(|resources| resources.resolve_named_data_type(name.as_str()))
            {
                data_type.ty().clone()
            } else {
                ty.clone()
            };
            seen.remove(name.as_str());
            resolved
        }
        TypeExpr::List(item) => TypeExpr::List(Box::new(resolve_type_refs_inner(
            item, aliases, resources, seen,
        ))),
        TypeExpr::Object(fields) => TypeExpr::Object(
            fields
                .iter()
                .map(|field| TypeField {
                    name: field.name.clone(),
                    ty: resolve_type_refs_inner(&field.ty, aliases, resources, seen),
                    optional: field.optional,
                })
                .collect(),
        ),
        TypeExpr::Union(items) => TypeExpr::Union(
            items
                .iter()
                .map(|item| resolve_type_refs_inner(item, aliases, resources, seen))
                .collect(),
        ),
        TypeExpr::Process {
            input,
            output,
            input_count,
        } => TypeExpr::Process {
            input: Box::new(resolve_type_refs_inner(input, aliases, resources, seen)),
            output: Box::new(resolve_type_refs_inner(output, aliases, resources, seen)),
            input_count: *input_count,
        },
        TypeExpr::TriggerHandle(event) => TypeExpr::TriggerHandle(Box::new(
            resolve_type_refs_inner(event, aliases, resources, seen),
        )),
        _ => ty.clone(),
    }
}

pub fn is_resolved_type_assignable(source: &TypeExpr, target: &TypeExpr) -> bool {
    if matches!(target, TypeExpr::Any) {
        return true;
    }
    if source == target {
        return true;
    }

    match (source, target) {
        (TypeExpr::Any, _) => false,
        (TypeExpr::Union(sources), _) => sources
            .iter()
            .all(|source| is_resolved_type_assignable(source, target)),
        (_, TypeExpr::Union(targets)) => targets
            .iter()
            .any(|target| is_resolved_type_assignable(source, target)),
        (TypeExpr::Int, TypeExpr::Float) => true,
        (TypeExpr::Enum(_), TypeExpr::Str) => true,
        (TypeExpr::Enum(sources), TypeExpr::Enum(targets)) => {
            sources.iter().all(|source| targets.contains(source))
        }
        (TypeExpr::List(source), TypeExpr::List(target)) => {
            is_resolved_type_assignable(source, target)
        }
        (TypeExpr::Object(_), TypeExpr::Dict) => true,
        (TypeExpr::Object(source), TypeExpr::Object(target)) => {
            object_type_assignable(source, target)
        }
        (TypeExpr::Ref(source), TypeExpr::Ref(target)) => source == target,
        (
            TypeExpr::Process {
                input: source_input,
                output: source_output,
                input_count: source_count,
            },
            TypeExpr::Process {
                input: target_input,
                output: target_output,
                input_count: target_count,
            },
        ) => {
            source_count == target_count
                && is_resolved_type_assignable(source_input, target_input)
                && is_resolved_type_assignable(source_output, target_output)
        }
        (TypeExpr::TriggerHandle(source), TypeExpr::TriggerHandle(target)) => {
            is_resolved_type_assignable(source, target)
        }
        _ => false,
    }
}

fn object_type_assignable(source: &[TypeField], target: &[TypeField]) -> bool {
    target.iter().all(|target_field| {
        let Some(source_field) = source
            .iter()
            .find(|source_field| source_field.name == target_field.name)
        else {
            return target_field.optional;
        };
        if !target_field.optional && source_field.optional {
            return false;
        }
        is_resolved_type_assignable(&source_field.ty, &target_field.ty)
    })
}

fn type_aliases(artifact: &ModuleArtifact) -> BTreeMap<String, TypeExpr> {
    artifact
        .canonical_ir
        .declarations
        .iter()
        .filter_map(|declaration| match declaration {
            crate::Declaration::Type(decl) => Some((decl.name.to_string(), decl.ty.clone())),
            _ => None,
        })
        .collect()
}

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

    #[derive(Debug, Deserialize, PartialEq)]
    struct ScheduleSource {
        expr: String,
        #[serde(default)]
        tz: Option<String>,
    }

    fn resources() -> ResourceCatalog {
        let mut resources = ResourceCatalog::new();
        resources
            .add_trigger_source_constructor(
                ["cron", "Schedule"],
                TypeExpr::Object(vec![
                    TypeField {
                        name: "expr".into(),
                        ty: TypeExpr::Str,
                        optional: false,
                    },
                    TypeField {
                        name: "tz".into(),
                        ty: TypeExpr::Str,
                        optional: true,
                    },
                ]),
                NamedDataType::object(
                    "cron.Tick",
                    vec![TypeField {
                        name: "fired_at".into(),
                        ty: TypeExpr::Str,
                        optional: false,
                    }],
                )
                .expect("valid cron tick type"),
            )
            .expect("valid cron schedule source");
        resources
    }

    #[test]
    fn host_value_encode_decode_and_typed_decode_round_trip() {
        let value = serde_json::json!({
            "expr": "*/10 * * * * *",
            "tz": "UTC",
        });
        let encoded = HostValue::encode("cron.Schedule", value).expect("host value encode");
        let decoded = HostValue::decode(&encoded).expect("host value decode");
        let payload: ScheduleSource = decoded
            .decode_as(&resources())
            .expect("typed host value payload");

        assert_eq!(
            payload,
            ScheduleSource {
                expr: "*/10 * * * * *".to_string(),
                tz: Some("UTC".to_string()),
            }
        );
    }

    #[test]
    fn host_value_typed_decode_rejects_unknown_source_type() {
        let decoded = HostValue::new("missing.Source", serde_json::json!({ "expr": "*" }));
        let err = decoded
            .decode_as::<ScheduleSource>(&resources())
            .expect_err("unknown source type should fail");

        assert!(
            matches!(err, HostValueError::UnknownSourceType { source_type } if source_type == "missing.Source")
        );
    }

    #[test]
    fn host_value_typed_decode_reports_malformed_payload() {
        let decoded = HostValue::new("cron.Schedule", serde_json::json!({ "expr": 1 }));
        let err = decoded
            .decode_as::<ScheduleSource>(&resources())
            .expect_err("malformed source payload should fail");

        assert!(
            matches!(err, HostValueError::MalformedPayload { source_type, .. } if source_type == "cron.Schedule")
        );
    }
}