async-nodes 0.1.0

use std::{
    any::{Any, type_name},
    fmt,
    sync::Arc,
};

use indexmap::IndexMap;

use super::{
    error::BspError,
    graph::{NodeId, OutputPin, PortIndex},
    node::NodeOutcome,
};

/// A marker type used to propagate bypassed execution paths in dynamic control-flow topologies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BspSkip;

/// A thread-safe, reference-counted container wrapping arbitrary Rust types.
#[derive(Clone)]
pub struct BspValue {
    inner: Arc<dyn Any + Send + Sync>,
    type_name: &'static str,
}

impl fmt::Debug for BspValue {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BspValue")
            .field("type_name", &self.type_name)
            .finish_non_exhaustive()
    }
}

impl BspValue {
    pub fn new<T: Any + Send + Sync>(val: T) -> Self {
        Self {
            inner: Arc::new(val) as Arc<dyn Any + Send + Sync>,
            type_name: type_name::<T>(),
        }
    }

    #[must_use]
    pub const fn type_name(&self) -> &'static str {
        self.type_name
    }

    #[must_use]
    pub fn is<T: Any>(&self) -> bool {
        self.inner.as_ref().is::<T>()
    }

    pub fn downcast<T: Any + Send + Sync>(self) -> Result<Arc<T>, Self> {
        if self.is::<T>() {
            let type_name = self.type_name;
            match self.inner.downcast::<T>() {
                Ok(arc_t) => Ok(arc_t),
                Err(arc_any) => Err(Self {
                    inner: arc_any,
                    type_name,
                }),
            }
        } else {
            Err(self)
        }
    }

    #[must_use]
    pub fn downcast_ref<T: Any>(&self) -> Option<&T> {
        self.inner.as_ref().downcast_ref::<T>()
    }

    pub fn downcast_mut<T: Any>(&mut self) -> Option<&mut T> {
        Arc::get_mut(&mut self.inner)?.downcast_mut::<T>()
    }
}

// =========================================================================
// TRAITS FOR MACRO INTEGRATION & STATIC DISPATCH
// =========================================================================

/// Translates a raw `BspValue` back into a typed parameter.
pub trait BspFromValue: Sized {
    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError>;
}

/// Serializes a typed output back into a type-erased `BspValue` container.
pub trait BspToValue {
    fn to_value(self) -> BspValue;
}

/// Standardized boundary compiler that converts functional return types into `NodeOutcome`.
pub trait BspOutputPacker {
    fn output_ports() -> Vec<PortIndex>;
    fn pack(self, node_id: NodeId, node_name: &str) -> Result<NodeOutcome, BspError>;

    /// Packs typed outputs into the raw port map — the form
    /// [`crate::node::DataflowNode`] implementors return.
    ///
    /// The default routes through [`BspOutputPacker::pack`]; a `Suspend` outcome has no
    /// port map and is rejected as a validation error (dataflow nodes never suspend —
    /// suspend-capable nodes implement [`crate::node::BspNode`] directly).
    fn pack_outputs(
        self,
        node_id: NodeId,
        node_name: &str,
    ) -> Result<IndexMap<PortIndex, BspValue>, BspError>
    where
        Self: Sized,
    {
        match self.pack(node_id, node_name)? {
            NodeOutcome::Success(outputs) => Ok(outputs),
            NodeOutcome::Suspend { .. } => Err(BspError::ValidationError(format!(
                "node '{node_name}' suspended: suspend outcomes have no port map; \
                 implement BspNode directly for suspend-capable nodes"
            ))),
        }
    }

    /// Returns whether this output compiles dynamic outputs that bypass static validation checks.
    #[must_use]
    fn is_dynamic() -> bool {
        false
    }
}

// Support wrapping directly into Arc references
impl<T: Any + Send + Sync> BspFromValue for Arc<T> {
    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
        let found_type = value.type_name();
        value.downcast::<T>().map_err(|_| BspError::TypeMismatch {
            node_id,
            port_index: port,
            expected: type_name::<T>(),
            found: found_type,
        })
    }
}

impl<T: Any + Send + Sync> BspToValue for Arc<T> {
    fn to_value(self) -> BspValue {
        BspValue {
            inner: self as Arc<dyn Any + Send + Sync>,
            type_name: type_name::<T>(),
        }
    }
}

impl<T: BspFromValue> BspFromValue for Option<T> {
    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
        if value.is::<BspSkip>() {
            Ok(None)
        } else {
            T::from_value(value, node_id, port).map(Some)
        }
    }
}

impl<T: BspToValue + 'static> BspToValue for Option<T> {
    fn to_value(self) -> BspValue {
        self.map_or_else(|| BspValue::new(BspSkip), BspToValue::to_value)
    }
}

impl<T: BspToValue + 'static> BspOutputPacker for Option<T> {
    fn output_ports() -> Vec<PortIndex> {
        vec![0]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, self.to_value());
        Ok(NodeOutcome::Success(outputs))
    }
}

// Core suspension pass-through implementations
impl BspOutputPacker for NodeOutcome {
    fn output_ports() -> Vec<PortIndex> {
        vec![0]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        Ok(self)
    }
    fn is_dynamic() -> bool {
        true
    }
}

// Support returning Arc directly
impl<T: Any + Send + Sync> BspOutputPacker for Arc<T> {
    fn output_ports() -> Vec<PortIndex> {
        vec![0]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, self.to_value());
        Ok(NodeOutcome::Success(outputs))
    }
}

// Void return support
impl BspOutputPacker for () {
    fn output_ports() -> Vec<PortIndex> {
        Vec::new()
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        Ok(NodeOutcome::Success(IndexMap::new()))
    }
}

// Single generic output compiler routing standard results beautifully
impl<T: BspOutputPacker, E: std::fmt::Display> BspOutputPacker for Result<T, E> {
    fn output_ports() -> Vec<PortIndex> {
        T::output_ports()
    }
    fn pack(self, node_id: NodeId, node_name: &str) -> Result<NodeOutcome, BspError> {
        match self {
            Ok(val) => val.pack(node_id, node_name),
            Err(e) => Err(BspError::NodeExecutionFailed {
                node_id,
                node_name: node_name.to_string(),
                message: e.to_string(),
            }),
        }
    }
    fn is_dynamic() -> bool {
        T::is_dynamic()
    }
}

// =========================================================================
// MULTIPLE-PORT TUPLE OUTPUT PACKING
// =========================================================================

impl<T1, T2> BspOutputPacker for (T1, T2)
where
    T1: BspToValue + 'static,
    T2: BspToValue + 'static,
{
    fn output_ports() -> Vec<PortIndex> {
        vec![0, 1]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, self.0.to_value());
        outputs.insert(1, self.1.to_value());
        Ok(NodeOutcome::Success(outputs))
    }
}

impl<T1, T2, T3> BspOutputPacker for (T1, T2, T3)
where
    T1: BspToValue + 'static,
    T2: BspToValue + 'static,
    T3: BspToValue + 'static,
{
    fn output_ports() -> Vec<PortIndex> {
        vec![0, 1, 2]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, self.0.to_value());
        outputs.insert(1, self.1.to_value());
        outputs.insert(2, self.2.to_value());
        Ok(NodeOutcome::Success(outputs))
    }
}

// =========================================================================
// BLANKET AGGREGATION & VECTOR PACKING
// =========================================================================

impl<T: BspFromValue + Clone + Send + Sync + 'static> BspFromValue for Vec<T> {
    fn from_value(value: BspValue, node_id: NodeId, port: PortIndex) -> Result<Self, BspError> {
        if value.is::<BspSkip>() {
            return Ok(Self::new());
        }

        // Resolves recursive vector wrapping bugs by validating direct type downcasts
        if value.is::<Self>() {
            let arc_vec = value
                .downcast::<Self>()
                .map_err(|err| BspError::TypeMismatch {
                    node_id,
                    port_index: port,
                    expected: std::any::type_name::<Self>(),
                    found: err.type_name(), // Uses the returned BspValue ownership to extract the type name
                })?;
            return Ok((*arc_vec).clone());
        }

        if let Some(vec_bsp) = value.downcast_ref::<Vec<BspValue>>() {
            let mut result = Self::with_capacity(vec_bsp.len());
            for val in vec_bsp {
                if !val.is::<BspSkip>() {
                    result.push(T::from_value(val.clone(), node_id, port)?);
                }
            }
            Ok(result)
        } else {
            Ok(vec![T::from_value(value, node_id, port)?])
        }
    }
}

impl<T: BspToValue + Send + Sync + 'static> BspToValue for Vec<T> {
    fn to_value(self) -> BspValue {
        BspValue::new(self)
    }
}

impl<T: BspToValue + Send + Sync + 'static> BspOutputPacker for Vec<T> {
    fn output_ports() -> Vec<PortIndex> {
        vec![0]
    }
    fn pack(self, _node_id: NodeId, _node_name: &str) -> Result<NodeOutcome, BspError> {
        let mut outputs = IndexMap::new();
        outputs.insert(0, self.to_value());
        Ok(NodeOutcome::Success(outputs))
    }
}

// =========================================================================
// ERGONOMIC MAP DOWNCASTING EXTENSIONS
// =========================================================================

pub trait BspOutputMapExt {
    fn get_typed<T: std::any::Any + Send + Sync + Clone>(
        &self,
        pin: OutputPin,
    ) -> Result<T, BspError>;
}

impl BspOutputMapExt for IndexMap<OutputPin, BspValue> {
    fn get_typed<T: std::any::Any + Send + Sync + Clone>(
        &self,
        pin: OutputPin,
    ) -> Result<T, BspError> {
        let value = self.get(&pin).ok_or(BspError::NodeNotFound {
            node_id: pin.node_id,
        })?;
        let found_type = value.type_name();
        let arc_t = value
            .clone()
            .downcast::<T>()
            .map_err(|_| BspError::TypeMismatch {
                node_id: pin.node_id,
                port_index: pin.port,
                expected: std::any::type_name::<T>(),
                found: found_type,
            })?;
        Ok((*arc_t).clone())
    }
}

// =========================================================================
// PRIMITIVE TYPE REGISTRATION
// =========================================================================

#[macro_export]
macro_rules! register_bsp_type {
    ($t:ty) => {
        impl $crate::value::BspFromValue for $t {
            fn from_value(
                value: $crate::value::BspValue,
                node_id: $crate::graph::NodeId,
                port: $crate::graph::PortIndex,
            ) -> Result<Self, $crate::error::BspError> {
                let found_type = value.type_name();
                let arc_t =
                    value
                        .downcast::<$t>()
                        .map_err(|_| $crate::error::BspError::TypeMismatch {
                            node_id,
                            port_index: port,
                            expected: std::any::type_name::<$t>(),
                            found: found_type,
                        })?;
                Ok((*arc_t).clone())
            }
        }

        impl $crate::value::BspToValue for $t {
            fn to_value(self) -> $crate::value::BspValue {
                $crate::value::BspValue::new(self)
            }
        }

        impl $crate::value::BspOutputPacker for $t {
            fn output_ports() -> Vec<$crate::graph::PortIndex> {
                vec![0]
            }
            fn pack(
                self,
                _node_id: $crate::graph::NodeId,
                _node_name: &str,
            ) -> Result<$crate::node::NodeOutcome, $crate::error::BspError> {
                let mut outputs = ::indexmap::IndexMap::new();
                outputs.insert(0, $crate::value::BspToValue::to_value(self));
                Ok($crate::node::NodeOutcome::Success(outputs))
            }
        }
    };
}

// Standard primitives registration
register_bsp_type!(i64);
register_bsp_type!(f64);
register_bsp_type!(bool);
register_bsp_type!(String);
register_bsp_type!(serde_json::Value);
register_bsp_type!(BspSkip);