aimdb_core/

router.rs

//! Generic message router for efficient connector dispatch
//!
//! Provides O(M) routing complexity instead of O(N×M) filtered streams.
//! Routes incoming messages directly to type-specific producers based on topic/key matching.
//!
//! This router is protocol-agnostic and can be used by any connector:
//! - MQTT: Routes topics to producers
//! - Kafka: Routes topics/partitions to producers
//! - HTTP: Routes paths to producers
//! - DDS: Routes topics to producers
//! - Shared Memory: Routes segment names to producers

#[cfg(not(feature = "std"))]
extern crate alloc;

#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, string::String, sync::Arc, vec::Vec};

#[cfg(feature = "std")]
use std::sync::Arc;

use crate::connector::{DeserializerKind, ProducerTrait};

/// A single routing entry
///
/// Maps one (resource_id, type) pair to a producer and deserializer.
/// Multiple routes can exist for the same resource_id (different types).
///
/// # Resource ID Examples
///
/// - MQTT: "sensors/temperature" (topic)
/// - Kafka: "events:0" (topic:partition)
/// - HTTP: "/api/v1/sensors" (path)
/// - DDS: "TelemetryData" (topic name)
/// - Shmem: "temperature_buffer" (segment name)
pub struct Route {
    /// Resource identifier to match (reference-counted for proper memory management)
    ///
    /// Examples: MQTT topic, Kafka topic, HTTP path, DDS topic, shmem segment
    ///
    /// Uses Arc<str> instead of &'static str to avoid memory leaks from Box::leak().
    /// This adds ~8 bytes overhead per route (Arc control block) but enables proper cleanup.
    pub resource_id: Arc<str>,

    /// Type-erased producer for this route
    pub producer: Box<dyn ProducerTrait>,

    /// Deserializer for converting bytes → typed value (raw or context-aware)
    pub deserializer: DeserializerKind,
}

/// Generic message router for connector dispatch
///
/// Routes incoming messages to appropriate producers based on resource_id.
/// Uses linear search which is efficient for <100 routes.
///
/// # Performance
///
/// - O(M) complexity where M = number of routes
/// - May check multiple routes if same resource_id maps to multiple types
/// - Typical routing time: <1μs for <50 routes
///
/// # Protocol Support
///
/// This router is protocol-agnostic. Each connector uses it with their own resource_id format:
/// - **MQTT**: `topic` (e.g., "sensors/temperature")
/// - **Kafka**: `topic` or `topic:partition` (e.g., "events" or "events:0")
/// - **HTTP**: `path` (e.g., "/api/v1/sensors")
/// - **DDS**: `topic_name` (e.g., "TelemetryData")
/// - **Shmem**: `segment_name` (e.g., "temperature_buffer")
pub struct Router {
    /// List of all registered routes
    routes: Vec<Route>,
}

impl Router {
    /// Create a new router with the given routes
    pub fn new(routes: Vec<Route>) -> Self {
        Self { routes }
    }

    /// Route a message to appropriate producer(s)
    ///
    /// # Arguments
    /// * `resource_id` - Resource identifier (topic, path, segment name, etc.)
    /// * `payload` - Raw message payload bytes
    /// * `ctx` - Optional type-erased runtime context for context-aware deserializers
    ///
    /// # Returns
    /// * `Ok(())` - Always returns Ok, even if no routes matched or processing failed.
    ///   Failures are logged (via tracing/defmt) but do not propagate as errors.
    ///
    /// # Behavior
    /// - Checks all routes that match the resource_id (may be multiple)
    /// - For `DeserializerKind::Raw`, calls the deserializer with payload only
    /// - For `DeserializerKind::Context`, calls with context + payload (skips if no context)
    /// - Logs warnings on deserialization failures but continues
    /// - Logs debug message if no routes found for resource_id
    pub async fn route(
        &self,
        resource_id: &str,
        payload: &[u8],
        ctx: Option<&Arc<dyn core::any::Any + Send + Sync>>,
    ) -> Result<(), String> {
        let mut routed = false;
        let mut matched = false;

        // Linear search through all routes
        // Note: Multiple routes may match the same resource_id (different types)
        for route in &self.routes {
            if route.resource_id.as_ref() == resource_id {
                matched = true;
                // Deserialize the payload based on deserializer kind
                let result = match &route.deserializer {
                    DeserializerKind::Raw(deser) => (deser)(payload),
                    DeserializerKind::Context(deser) => match ctx {
                        Some(ctx) => (deser)(ctx.clone(), payload),
                        None => {
                            #[cfg(feature = "tracing")]
                            tracing::warn!(
                                "Context deserializer on '{}' but no context provided, skipping",
                                resource_id
                            );

                            #[cfg(feature = "defmt")]
                            defmt::warn!(
                                "Context deserializer on '{}' but no context provided",
                                resource_id
                            );

                            continue;
                        }
                    },
                };

                match result {
                    Ok(value_any) => {
                        // Produce into the buffer
                        match route.producer.produce_any(value_any).await {
                            Ok(()) => {
                                routed = true;

                                #[cfg(feature = "tracing")]
                                tracing::debug!("Routed message on '{}' to producer", resource_id);
                            }
                            Err(_e) => {
                                #[cfg(feature = "tracing")]
                                tracing::error!(
                                    "Failed to produce message on '{}': {}",
                                    resource_id,
                                    _e
                                );

                                #[cfg(feature = "defmt")]
                                defmt::error!(
                                    "Failed to produce message on '{}': {}",
                                    resource_id,
                                    _e.as_str()
                                );
                            }
                        }
                    }
                    Err(_e) => {
                        #[cfg(feature = "tracing")]
                        tracing::warn!(
                            "Failed to deserialize message on '{}': {}",
                            resource_id,
                            _e
                        );

                        #[cfg(feature = "defmt")]
                        defmt::warn!(
                            "Failed to deserialize message on '{}': {}",
                            resource_id,
                            _e.as_str()
                        );
                    }
                }
            }
        }

        if !routed {
            if matched {
                #[cfg(feature = "tracing")]
                tracing::debug!("Route matched for '{}' but message was not produced (missing context or errors)", resource_id);

                #[cfg(feature = "defmt")]
                defmt::debug!("Route matched for '{}' but not produced", resource_id);
            } else {
                #[cfg(feature = "tracing")]
                tracing::debug!("No route found for resource: '{}'", resource_id);

                #[cfg(feature = "defmt")]
                defmt::debug!("No route found for resource: '{}'", resource_id);
            }
        }

        Ok(())
    }

    /// Get list of all resource IDs registered in this router
    ///
    /// Useful for subscribing at the protocol level (e.g., MQTT SUBSCRIBE).
    /// Returns unique resource IDs (deduplicated even if multiple routes per resource).
    pub fn resource_ids(&self) -> Vec<Arc<str>> {
        let mut ids: Vec<Arc<str>> = self.routes.iter().map(|r| r.resource_id.clone()).collect();

        // Deduplicate by converting to strings for comparison
        ids.sort_unstable_by(|a, b| a.as_ref().cmp(b.as_ref()));
        ids.dedup_by(|a, b| a.as_ref() == b.as_ref());

        ids
    }

    /// Get the number of routes in this router
    pub fn route_count(&self) -> usize {
        self.routes.len()
    }
}

/// Builder for constructing routers
///
/// Provides a fluent API for adding routes before creating the router.
///
/// # Example
///
/// ```rust,ignore
/// use aimdb_core::router::RouterBuilder;
///
/// let router = RouterBuilder::new()
///     .add_route(
///         "sensors/temperature",
///         producer_temp.clone(),
///         Arc::new(|bytes| {
///             serde_json::from_slice::<Temperature>(bytes)
///                 .map(|t| Box::new(t) as Box<dyn Any + Send>)
///                 .map_err(|e| e.to_string())
///         })
///     )
///     .add_route(
///         "sensors/humidity",
///         producer_humidity.clone(),
///         Arc::new(|bytes| {
///             serde_json::from_slice::<Humidity>(bytes)
///                 .map(|h| Box::new(h) as Box<dyn Any + Send>)
///                 .map_err(|e| e.to_string())
///         })
///     )
///     .build();
/// ```
pub struct RouterBuilder {
    routes: Vec<Route>,
}

impl RouterBuilder {
    /// Create a new router builder
    pub fn new() -> Self {
        Self { routes: Vec::new() }
    }

    /// Create a router builder from a collection of routes
    ///
    /// This is a convenience method for automatic router construction from
    /// `AimDb::collect_inbound_routes()`. The resource_ids are converted to
    /// Arc<str> for proper memory management.
    ///
    /// # Arguments
    /// * `routes` - Vector of (resource_id, producer, deserializer) tuples
    ///
    /// # Example
    /// ```rust,ignore
    /// let routes = db.collect_inbound_routes("mqtt");
    /// let router = RouterBuilder::from_routes(routes).build();
    /// connector.set_router(router).await?;
    /// ```
    pub fn from_routes(routes: Vec<(String, Box<dyn ProducerTrait>, DeserializerKind)>) -> Self {
        let mut builder = Self::new();
        for (resource_id, producer, deserializer) in routes {
            // Convert String to Arc<str> - no leaking needed!
            let resource_id_arc: Arc<str> = Arc::from(resource_id.as_str());
            builder = builder.add_route(resource_id_arc, producer, deserializer);
        }
        builder
    }

    /// Add a route to the router
    ///
    /// # Arguments
    /// * `resource_id` - Resource identifier to match (as Arc<str>)
    /// * `producer` - Producer that implements ProducerTrait
    /// * `deserializer` - Deserializer variant (raw or context-aware)
    ///
    /// # Resource ID Memory Management
    /// The resource_id is stored as Arc<str> for proper reference counting and cleanup.
    /// You can create an Arc<str> from:
    /// - String literal: `Arc::from("sensors/temperature")`
    /// - Owned String: `Arc::from(string.as_str())`
    pub fn add_route(
        mut self,
        resource_id: Arc<str>,
        producer: Box<dyn ProducerTrait>,
        deserializer: DeserializerKind,
    ) -> Self {
        self.routes.push(Route {
            resource_id,
            producer,
            deserializer,
        });
        self
    }

    /// Build the router
    ///
    /// Consumes the builder and returns a configured Router.
    pub fn build(self) -> Router {
        Router::new(self.routes)
    }

    /// Get the number of routes that will be created
    pub fn route_count(&self) -> usize {
        self.routes.len()
    }
}

impl Default for RouterBuilder {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(all(test, feature = "std"))]
mod tests {
    use super::*;
    use crate::connector::ProducerTrait;
    use core::future::Future;
    use core::pin::Pin;
    use std::any::Any;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    // Mock producer for testing
    struct MockProducer {
        call_count: Arc<AtomicUsize>,
    }

    impl ProducerTrait for MockProducer {
        fn produce_any<'a>(
            &'a self,
            _value: Box<dyn Any + Send>,
        ) -> Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>> {
            let call_count = self.call_count.clone();
            Box::pin(async move {
                call_count.fetch_add(1, Ordering::SeqCst);
                Ok(())
            })
        }
    }

    #[tokio::test]
    async fn test_single_route() {
        let call_count = Arc::new(AtomicUsize::new(0));

        let routes = vec![Route {
            resource_id: Arc::from("test/resource"),
            producer: Box::new(MockProducer {
                call_count: call_count.clone(),
            }),
            deserializer: DeserializerKind::Raw(Arc::new(|_bytes| Ok(Box::new(42i32)))),
        }];

        let router = Router::new(routes);

        router.route("test/resource", b"dummy", None).await.unwrap();

        assert_eq!(call_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_multiple_routes_same_resource() {
        let call_count1 = Arc::new(AtomicUsize::new(0));
        let call_count2 = Arc::new(AtomicUsize::new(0));

        let routes = vec![
            Route {
                resource_id: Arc::from("shared/resource"),
                producer: Box::new(MockProducer {
                    call_count: call_count1.clone(),
                }),
                deserializer: DeserializerKind::Raw(Arc::new(|_bytes| Ok(Box::new(42i32)))),
            },
            Route {
                resource_id: Arc::from("shared/resource"),
                producer: Box::new(MockProducer {
                    call_count: call_count2.clone(),
                }),
                deserializer: DeserializerKind::Raw(Arc::new(|_bytes| {
                    Ok(Box::new("test".to_string()))
                })),
            },
        ];

        let router = Router::new(routes);

        router
            .route("shared/resource", b"dummy", None)
            .await
            .unwrap();

        // Both producers should be called
        assert_eq!(call_count1.load(Ordering::SeqCst), 1);
        assert_eq!(call_count2.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_unknown_resource() {
        let routes = vec![Route {
            resource_id: Arc::from("test/resource"),
            producer: Box::new(MockProducer {
                call_count: Arc::new(AtomicUsize::new(0)),
            }),
            deserializer: DeserializerKind::Raw(Arc::new(|_bytes| Ok(Box::new(42i32)))),
        }];

        let router = Router::new(routes);

        // Should not panic on unknown resource
        router
            .route("unknown/resource", b"dummy", None)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_resource_ids_deduplication() {
        let routes = vec![
            Route {
                resource_id: Arc::from("resource1"),
                producer: Box::new(MockProducer {
                    call_count: Arc::new(AtomicUsize::new(0)),
                }),
                deserializer: DeserializerKind::Raw(Arc::new(|_bytes| Ok(Box::new(42i32)))),
            },
            Route {
                resource_id: Arc::from("resource1"), // Duplicate
                producer: Box::new(MockProducer {
                    call_count: Arc::new(AtomicUsize::new(0)),
                }),
                deserializer: DeserializerKind::Raw(Arc::new(|_bytes| {
                    Ok(Box::new("test".to_string()))
                })),
            },
            Route {
                resource_id: Arc::from("resource2"),
                producer: Box::new(MockProducer {
                    call_count: Arc::new(AtomicUsize::new(0)),
                }),
                deserializer: DeserializerKind::Raw(Arc::new(|_bytes| Ok(Box::new(99i32)))),
            },
        ];

        let router = Router::new(routes);
        let ids = router.resource_ids();

        assert_eq!(ids.len(), 2);
        assert!(ids.iter().any(|id| id.as_ref() == "resource1"));
        assert!(ids.iter().any(|id| id.as_ref() == "resource2"));
    }

    #[tokio::test]
    async fn test_context_deserializer_with_context() {
        let call_count = Arc::new(AtomicUsize::new(0));
        let call_count_clone = call_count.clone();

        let routes = vec![Route {
            resource_id: Arc::from("ctx/resource"),
            producer: Box::new(MockProducer {
                call_count: call_count.clone(),
            }),
            deserializer: DeserializerKind::Context(Arc::new(move |_ctx, _bytes| {
                Ok(Box::new(42i32) as Box<dyn Any + Send>)
            })),
        }];

        let router = Router::new(routes);

        // Provide a dummy context (just an i32 wrapped in Arc)
        let ctx: Arc<dyn Any + Send + Sync> = Arc::new(0i32);
        router
            .route("ctx/resource", b"dummy", Some(&ctx))
            .await
            .unwrap();

        assert_eq!(call_count_clone.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_context_deserializer_without_context_skips() {
        let call_count = Arc::new(AtomicUsize::new(0));

        let routes = vec![Route {
            resource_id: Arc::from("ctx/resource"),
            producer: Box::new(MockProducer {
                call_count: call_count.clone(),
            }),
            deserializer: DeserializerKind::Context(Arc::new(|_ctx, _bytes| {
                Ok(Box::new(42i32) as Box<dyn Any + Send>)
            })),
        }];

        let router = Router::new(routes);

        // No context provided — context deserializer should be skipped
        router.route("ctx/resource", b"dummy", None).await.unwrap();

        assert_eq!(call_count.load(Ordering::SeqCst), 0);
    }
}