oxirs-stream 0.4.1

//! # Message Queue Bridge Module
//!
//! This module provides comprehensive message queue integration for external systems:
//! - Protocol bridging between different message queue systems
//! - Format conversion and message transformation
//! - Routing rules and message filtering
//! - External system adapters
//! - Performance monitoring and diagnostics

use anyhow::{anyhow, Result};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, RwLock};
use tokio::time::interval;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// Message queue bridge manager
pub struct MessageBridgeManager {
    /// Registered bridges
    bridges: Arc<RwLock<HashMap<String, MessageBridge>>>,
    /// Bridge configurations
    configs: Arc<RwLock<HashMap<String, BridgeConfig>>>,
    /// Message transformers
    transformers: Arc<RwLock<HashMap<String, Box<dyn MessageTransformer + Send + Sync>>>>,
    /// Routing engine
    router: Arc<RoutingEngine>,
    /// Statistics
    stats: Arc<RwLock<BridgeStats>>,
    /// Event notifier
    event_notifier: broadcast::Sender<BridgeNotification>,
}

/// Message bridge
#[derive(Clone)]
struct MessageBridge {
    /// Bridge ID
    id: String,
    /// Bridge type
    bridge_type: BridgeType,
    /// Source configuration
    source: ExternalSystemConfig,
    /// Target configuration
    target: ExternalSystemConfig,
    /// Message transformer
    transformer: String,
    /// Routing rules
    routing_rules: Vec<RoutingRule>,
    /// Bridge status
    status: BridgeStatus,
    /// Statistics
    stats: BridgeStatistics,
    /// Created timestamp
    created_at: Instant,
    /// Last activity
    last_activity: Option<Instant>,
}

/// Bridge types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum BridgeType {
    /// Bidirectional bridge
    Bidirectional,
    /// Source to target only
    SourceToTarget,
    /// Target to source only
    TargetToSource,
    /// Fanout (one source, multiple targets)
    Fanout,
    /// Fanin (multiple sources, one target)
    Fanin,
}

/// External system configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExternalSystemConfig {
    /// System type
    pub system_type: ExternalSystemType,
    /// Connection configuration
    pub connection: ConnectionConfig,
    /// Format configuration
    pub format: FormatConfig,
    /// Security configuration
    pub security: SecurityConfig,
}

/// External system types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ExternalSystemType {
    /// Apache Kafka
    Kafka {
        brokers: Vec<String>,
        topics: Vec<String>,
        consumer_group: Option<String>,
    },
    /// RabbitMQ
    RabbitMQ {
        url: String,
        exchange: String,
        routing_key: String,
        queue: Option<String>,
    },
    /// Amazon SQS
    AmazonSQS {
        region: String,
        queue_url: String,
        credentials: AwsCredentials,
    },
    /// Azure Service Bus
    AzureServiceBus {
        connection_string: String,
        queue_name: String,
    },
    /// Google Cloud Pub/Sub
    GooglePubSub {
        project_id: String,
        topic: String,
        subscription: Option<String>,
    },
    /// Apache Pulsar
    Pulsar {
        service_url: String,
        topics: Vec<String>,
        subscription: Option<String>,
    },
    /// Redis Pub/Sub
    RedisPubSub { url: String, channels: Vec<String> },
    /// HTTP REST API
    HttpRest {
        base_url: String,
        endpoints: HashMap<String, String>,
        headers: HashMap<String, String>,
    },
    /// WebSocket
    WebSocket { url: String, protocols: Vec<String> },
    /// File system
    FileSystem {
        directory: String,
        pattern: String,
        watch_mode: bool,
    },
    /// MQTT broker
    Mqtt {
        broker_url: String,
        client_id: String,
        topic_subscriptions: Vec<String>,
        qos: u8,
        username: Option<String>,
        password: Option<String>,
    },
    /// OPC UA server
    OpcUa {
        endpoint_url: String,
        security_policy: String,
        user_identity: String,
        node_subscriptions: Vec<String>,
    },
    /// Eclipse Sparkplug B (MQTT-based Industry 4.0)
    SparkplugB {
        broker_url: String,
        group_id: String,
        edge_node_id: String,
        device_ids: Vec<String>,
    },
}

/// AWS credentials
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AwsCredentials {
    pub access_key_id: String,
    pub secret_access_key: String,
    pub session_token: Option<String>,
}

/// Connection configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionConfig {
    /// Connection timeout
    pub timeout: Duration,
    /// Keep alive interval
    pub keep_alive: Duration,
    /// Retry configuration
    pub retry: RetryConfig,
    /// SSL/TLS configuration
    pub tls: Option<TlsConfig>,
}

/// TLS configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TlsConfig {
    pub enabled: bool,
    pub verify_certificate: bool,
    pub certificate_path: Option<String>,
    pub private_key_path: Option<String>,
    pub ca_certificate_path: Option<String>,
}

/// Retry configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    pub max_attempts: u32,
    pub initial_delay: Duration,
    pub max_delay: Duration,
    pub exponential_backoff: bool,
}

/// Format configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FormatConfig {
    /// Message format
    pub format: MessageFormat,
    /// Encoding
    pub encoding: String,
    /// Compression
    pub compression: Option<CompressionType>,
    /// Schema validation
    pub schema_validation: bool,
}

/// Message formats
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum MessageFormat {
    /// JSON format
    Json,
    /// Apache Avro
    Avro { schema: String },
    /// Protocol Buffers
    Protobuf { schema: String },
    /// XML format
    Xml,
    /// Plain text
    Text,
    /// Binary format
    Binary,
    /// Custom format
    Custom { transformer: String },
}

/// Compression types
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CompressionType {
    Gzip,
    Snappy,
    Lz4,
    Zstd,
}

/// Security configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecurityConfig {
    /// Authentication method
    pub auth: AuthenticationMethod,
    /// Encryption settings
    pub encryption: EncryptionConfig,
    /// Access control
    pub access_control: AccessControlConfig,
}

/// Authentication methods
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AuthenticationMethod {
    None,
    BasicAuth {
        username: String,
        password: String,
    },
    BearerToken {
        token: String,
    },
    ApiKey {
        key: String,
        header: String,
    },
    OAuth2 {
        client_id: String,
        client_secret: String,
        token_url: String,
    },
    SaslPlain {
        username: String,
        password: String,
    },
    SaslScramSha256 {
        username: String,
        password: String,
    },
    Certificate {
        cert_path: String,
        key_path: String,
    },
}

/// Encryption configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EncryptionConfig {
    pub enabled: bool,
    pub algorithm: Option<String>,
    pub key_id: Option<String>,
}

/// Access control configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AccessControlConfig {
    pub read_permissions: Vec<String>,
    pub write_permissions: Vec<String>,
    pub admin_permissions: Vec<String>,
}

/// Routing rule
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RoutingRule {
    /// Rule name
    pub name: String,
    /// Rule condition
    pub condition: RuleCondition,
    /// Rule action
    pub action: RuleAction,
    /// Rule priority
    pub priority: u32,
    /// Rule enabled
    pub enabled: bool,
}

/// Rule condition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RuleCondition {
    /// Always match
    Always,
    /// Match event type
    EventType { types: Vec<String> },
    /// Match graph
    Graph { patterns: Vec<String> },
    /// Match subject pattern
    SubjectPattern { regex: String },
    /// Match predicate
    Predicate { predicates: Vec<String> },
    /// Custom expression
    Expression { expr: String },
    /// Composite condition
    Composite {
        operator: LogicalOperator,
        conditions: Vec<RuleCondition>,
    },
}

/// Logical operators
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum LogicalOperator {
    And,
    Or,
    Not,
}

/// Rule action
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RuleAction {
    /// Forward message
    Forward,
    /// Drop message
    Drop,
    /// Transform message
    Transform { transformer: String },
    /// Route to specific target
    Route { target: String },
    /// Duplicate message
    Duplicate { targets: Vec<String> },
}

/// Bridge status
#[derive(Debug, Clone, PartialEq)]
enum BridgeStatus {
    Active,
    #[allow(dead_code)]
    Paused,
    Stopped,
    #[allow(dead_code)]
    Failed {
        reason: String,
    },
}

/// Bridge configuration
#[derive(Debug, Clone)]
pub struct BridgeConfig {
    /// Maximum message queue size
    pub max_queue_size: usize,
    /// Batch size for processing
    pub batch_size: usize,
    /// Processing interval
    pub processing_interval: Duration,
    /// Enable monitoring
    pub enable_monitoring: bool,
    /// Enable dead letter queue
    pub enable_dlq: bool,
    /// Message TTL
    pub message_ttl: Duration,
}

impl Default for BridgeConfig {
    fn default() -> Self {
        Self {
            max_queue_size: 10000,
            batch_size: 100,
            processing_interval: Duration::from_millis(100),
            enable_monitoring: true,
            enable_dlq: true,
            message_ttl: Duration::from_secs(24 * 60 * 60),
        }
    }
}

/// Bridge statistics
#[derive(Debug, Clone, Default)]
pub struct BridgeStatistics {
    /// Messages received
    pub messages_received: u64,
    /// Messages sent
    pub messages_sent: u64,
    /// Messages dropped
    pub messages_dropped: u64,
    /// Messages failed
    pub messages_failed: u64,
    /// Transform errors
    pub transform_errors: u64,
    /// Average processing time
    pub avg_processing_time: Duration,
    /// Last activity
    pub last_activity: Option<Instant>,
}

/// Manager statistics
#[derive(Debug, Clone, Default)]
pub struct BridgeStats {
    /// Total bridges
    pub total_bridges: usize,
    /// Active bridges
    pub active_bridges: usize,
    /// Total messages processed
    pub total_messages: u64,
    /// Failed messages
    pub failed_messages: u64,
    /// Average processing time
    pub avg_processing_time: Duration,
}

/// Bridge notification events
#[derive(Debug, Clone)]
pub enum BridgeNotification {
    /// Bridge created
    BridgeCreated { id: String, bridge_type: BridgeType },
    /// Bridge started
    BridgeStarted { id: String },
    /// Bridge stopped
    BridgeStopped { id: String },
    /// Bridge failed
    BridgeFailed { id: String, reason: String },
    /// Message processed
    MessageProcessed {
        bridge_id: String,
        message_id: String,
        duration: Duration,
    },
    /// Message failed
    MessageFailed {
        bridge_id: String,
        message_id: String,
        error: String,
    },
}

/// Message transformer trait
pub trait MessageTransformer {
    /// Transform message from source format to target format
    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage>;

    /// Get transformer name
    fn name(&self) -> &str;

    /// Get supported formats
    fn supported_formats(&self) -> (MessageFormat, MessageFormat);
}

/// External message representation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExternalMessage {
    /// Message ID
    pub id: String,
    /// Message headers
    pub headers: HashMap<String, String>,
    /// Message payload
    pub payload: Vec<u8>,
    /// Message format
    pub format: MessageFormat,
    /// Timestamp
    pub timestamp: chrono::DateTime<chrono::Utc>,
    /// Source system
    pub source: String,
    /// Message metadata
    pub metadata: HashMap<String, String>,
}

/// Routing engine
struct RoutingEngine {
    /// Global routing rules
    _global_rules: Arc<RwLock<Vec<RoutingRule>>>,
    /// Bridge-specific rules cache
    _rule_cache: Arc<RwLock<HashMap<String, Vec<RoutingRule>>>>,
}

impl MessageBridgeManager {
    /// Create a new message bridge manager
    pub async fn new() -> Result<Self> {
        let (tx, _) = broadcast::channel(1000);

        Ok(Self {
            bridges: Arc::new(RwLock::new(HashMap::new())),
            configs: Arc::new(RwLock::new(HashMap::new())),
            transformers: Arc::new(RwLock::new(HashMap::new())),
            router: Arc::new(RoutingEngine::new()),
            stats: Arc::new(RwLock::new(BridgeStats::default())),
            event_notifier: tx,
        })
    }

    /// Register a message transformer
    pub async fn register_transformer(
        &self,
        transformer: Box<dyn MessageTransformer + Send + Sync>,
    ) {
        let name = transformer.name().to_string();
        self.transformers.write().await.insert(name, transformer);
        info!("Registered message transformer");
    }

    /// Create a message bridge
    pub async fn create_bridge(
        &self,
        bridge_type: BridgeType,
        source: ExternalSystemConfig,
        target: ExternalSystemConfig,
        transformer: String,
        routing_rules: Vec<RoutingRule>,
        config: BridgeConfig,
    ) -> Result<String> {
        // Validate transformer exists
        if !self.transformers.read().await.contains_key(&transformer) {
            return Err(anyhow!("Transformer not found: {}", transformer));
        }

        // Generate bridge ID
        let bridge_id = Uuid::new_v4().to_string();

        // Create bridge
        let bridge = MessageBridge {
            id: bridge_id.clone(),
            bridge_type: bridge_type.clone(),
            source,
            target,
            transformer,
            routing_rules,
            status: BridgeStatus::Stopped,
            stats: BridgeStatistics::default(),
            created_at: Instant::now(),
            last_activity: None,
        };

        // Register bridge
        self.bridges.write().await.insert(bridge_id.clone(), bridge);
        self.configs.write().await.insert(bridge_id.clone(), config);

        // Update statistics
        let mut stats = self.stats.write().await;
        stats.total_bridges += 1;
        drop(stats);

        // Notify
        let _ = self.event_notifier.send(BridgeNotification::BridgeCreated {
            id: bridge_id.clone(),
            bridge_type,
        });

        info!("Created message bridge: {}", bridge_id);
        Ok(bridge_id)
    }

    /// Whether a real transport is implemented for the given external system.
    ///
    /// Only the FileSystem transport is implemented in pure Rust today. Network
    /// transports (Kafka/RabbitMQ/Redis/HTTP/etc.) are not, so bridges using
    /// them must fail loud at start rather than reporting Active while silently
    /// transferring zero messages.
    fn transport_implemented(system_type: &ExternalSystemType) -> bool {
        matches!(system_type, ExternalSystemType::FileSystem { .. })
    }

    /// Start a bridge
    pub async fn start_bridge(&self, bridge_id: &str) -> Result<()> {
        // Validate that both endpoints have an implemented transport BEFORE
        // marking the bridge active. This upholds the fail-loud contract: a
        // bridge to an unimplemented backend must error, not silently no-op.
        {
            let bridges = self.bridges.read().await;
            let bridge = bridges
                .get(bridge_id)
                .ok_or_else(|| anyhow!("Bridge not found"))?;

            if !Self::transport_implemented(&bridge.source.system_type) {
                return Err(anyhow!(
                    "Bridge source transport {:?} is not implemented; cannot start bridge {}",
                    bridge.source.system_type,
                    bridge_id
                ));
            }
            if !Self::transport_implemented(&bridge.target.system_type) {
                return Err(anyhow!(
                    "Bridge target transport {:?} is not implemented; cannot start bridge {}",
                    bridge.target.system_type,
                    bridge_id
                ));
            }
        }

        let bridge_exists = {
            let mut bridges = self.bridges.write().await;
            if let Some(bridge) = bridges.get_mut(bridge_id) {
                bridge.status = BridgeStatus::Active;
                true
            } else {
                false
            }
        };

        if !bridge_exists {
            return Err(anyhow!("Bridge not found"));
        }

        // Start bridge processing
        self.start_bridge_processing(bridge_id).await?;

        // Update statistics
        self.stats.write().await.active_bridges += 1;

        // Notify
        let _ = self.event_notifier.send(BridgeNotification::BridgeStarted {
            id: bridge_id.to_string(),
        });

        info!("Started bridge: {}", bridge_id);
        Ok(())
    }

    /// Stop a bridge
    pub async fn stop_bridge(&self, bridge_id: &str) -> Result<()> {
        let mut bridges = self.bridges.write().await;
        let bridge = bridges
            .get_mut(bridge_id)
            .ok_or_else(|| anyhow!("Bridge not found"))?;

        bridge.status = BridgeStatus::Stopped;

        // Update statistics
        self.stats.write().await.active_bridges = bridges
            .values()
            .filter(|b| b.status == BridgeStatus::Active)
            .count();

        // Notify
        let _ = self.event_notifier.send(BridgeNotification::BridgeStopped {
            id: bridge_id.to_string(),
        });

        info!("Stopped bridge: {}", bridge_id);
        Ok(())
    }

    /// Start bridge processing
    async fn start_bridge_processing(&self, bridge_id: &str) -> Result<()> {
        // Clone all necessary data before spawning the task
        let bridge = {
            let bridges_guard = self.bridges.read().await;
            bridges_guard
                .get(bridge_id)
                .ok_or_else(|| anyhow!("Bridge not found"))?
                .clone()
        };

        let config = {
            let configs_guard = self.configs.read().await;
            configs_guard
                .get(bridge_id)
                .ok_or_else(|| anyhow!("Bridge config not found"))?
                .clone()
        };

        let bridges = self.bridges.clone();
        let transformers = self.transformers.clone();
        let router = self.router.clone();
        let stats = self.stats.clone();
        let event_notifier = self.event_notifier.clone();
        let bridge_id = bridge_id.to_string();

        tokio::spawn(async move {
            let mut interval = interval(config.processing_interval);
            let mut message_queue = VecDeque::new();

            loop {
                interval.tick().await;

                // Check if bridge is still active
                let status = {
                    let bridges_guard = bridges.read().await;
                    bridges_guard.get(&bridge_id).map(|b| b.status.clone())
                };

                if let Some(BridgeStatus::Active) = status {
                    // Process messages from source
                    match MessageBridgeManager::receive_messages(&bridge.source, &config).await {
                        Ok(messages) => {
                            for message in messages {
                                message_queue.push_back(message);

                                // Limit queue size
                                if message_queue.len() > config.max_queue_size {
                                    message_queue.pop_front();
                                    warn!("Bridge queue full, dropping oldest message");
                                }
                            }
                        }
                        Err(e) => {
                            error!("Failed to receive messages for bridge {}: {}", bridge_id, e);
                        }
                    }

                    // Process queued messages in batches
                    let batch_size = config.batch_size.min(message_queue.len());
                    if batch_size > 0 {
                        let batch: Vec<_> = message_queue.drain(..batch_size).collect();

                        for message in batch {
                            let start_time = Instant::now();

                            match MessageBridgeManager::process_message(
                                &bridge,
                                &message,
                                &transformers,
                                &router,
                            )
                            .await
                            {
                                Ok(_) => {
                                    let duration = start_time.elapsed();

                                    // Update bridge statistics
                                    MessageBridgeManager::update_bridge_stats(
                                        &bridges, &bridge_id, true, duration,
                                    )
                                    .await;
                                    stats.write().await.total_messages += 1;

                                    let _ =
                                        event_notifier.send(BridgeNotification::MessageProcessed {
                                            bridge_id: bridge_id.clone(),
                                            message_id: message.id.clone(),
                                            duration,
                                        });
                                }
                                Err(e) => {
                                    let duration = start_time.elapsed();

                                    error!(
                                        "Failed to process message {} in bridge {}: {}",
                                        message.id, bridge_id, e
                                    );

                                    // Update bridge statistics
                                    MessageBridgeManager::update_bridge_stats(
                                        &bridges, &bridge_id, false, duration,
                                    )
                                    .await;
                                    stats.write().await.failed_messages += 1;

                                    let _ =
                                        event_notifier.send(BridgeNotification::MessageFailed {
                                            bridge_id: bridge_id.clone(),
                                            message_id: message.id.clone(),
                                            error: e.to_string(),
                                        });

                                    // Send to dead letter queue if enabled
                                    if config.enable_dlq {
                                        // This would implement DLQ logic
                                        warn!("Message sent to dead letter queue: {}", message.id);
                                    }
                                }
                            }
                        }
                    }
                } else {
                    // Bridge is not active, exit loop
                    break;
                }
            }
        });

        Ok(())
    }

    /// Receive messages from external system
    async fn receive_messages(
        source: &ExternalSystemConfig,
        config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        match &source.system_type {
            ExternalSystemType::Kafka {
                brokers,
                topics,
                consumer_group,
            } => Self::receive_kafka_messages(brokers, topics, consumer_group, config).await,
            ExternalSystemType::RabbitMQ {
                url,
                exchange,
                routing_key,
                queue,
            } => Self::receive_rabbitmq_messages(url, exchange, routing_key, queue, config).await,
            ExternalSystemType::RedisPubSub { url, channels } => {
                Self::receive_redis_messages(url, channels, config).await
            }
            ExternalSystemType::HttpRest {
                base_url,
                endpoints,
                headers,
            } => Self::receive_http_messages(base_url, endpoints, headers, config).await,
            ExternalSystemType::FileSystem {
                directory,
                pattern,
                watch_mode,
            } => Self::receive_file_messages(directory, pattern, *watch_mode, config).await,
            other => Err(anyhow!(
                "Message receiving is not implemented for system type {:?}",
                other
            )),
        }
    }

    /// Receive messages from Kafka
    async fn receive_kafka_messages(
        _brokers: &[String],
        _topics: &[String],
        _consumer_group: &Option<String>,
        _config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        Err(anyhow!(
            "Kafka bridge consumer transport is not implemented"
        ))
    }

    /// Receive messages from RabbitMQ
    async fn receive_rabbitmq_messages(
        _url: &str,
        _exchange: &str,
        _routing_key: &str,
        _queue: &Option<String>,
        _config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        Err(anyhow!(
            "RabbitMQ bridge consumer transport is not implemented"
        ))
    }

    /// Receive messages from Redis
    async fn receive_redis_messages(
        _url: &str,
        _channels: &[String],
        _config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        Err(anyhow!(
            "Redis Pub/Sub bridge consumer transport is not implemented"
        ))
    }

    /// Receive messages from HTTP endpoints
    async fn receive_http_messages(
        _base_url: &str,
        _endpoints: &HashMap<String, String>,
        _headers: &HashMap<String, String>,
        _config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        Err(anyhow!(
            "HTTP REST bridge consumer transport is not implemented"
        ))
    }

    /// Receive messages from the file system.
    ///
    /// Reads files under `directory` whose file name matches `pattern` (a simple
    /// glob supporting a single leading/trailing `*`, or `*` for everything).
    /// Each matched file becomes one [`ExternalMessage`] whose payload is the
    /// file contents; the file is then deleted so it is consumed exactly once.
    async fn receive_file_messages(
        directory: &str,
        pattern: &str,
        _watch_mode: bool,
        config: &BridgeConfig,
    ) -> Result<Vec<ExternalMessage>> {
        let mut messages = Vec::new();

        let mut read_dir = match tokio::fs::read_dir(directory).await {
            Ok(read_dir) => read_dir,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(messages),
            Err(e) => {
                return Err(anyhow!(
                    "Failed to read bridge source directory {}: {}",
                    directory,
                    e
                ))
            }
        };

        while let Some(entry) = read_dir.next_entry().await? {
            if messages.len() >= config.max_queue_size {
                break;
            }
            let path = entry.path();
            if !path.is_file() {
                continue;
            }
            let file_name = match path.file_name().and_then(|n| n.to_str()) {
                Some(name) => name.to_string(),
                None => continue,
            };
            if !Self::file_name_matches(&file_name, pattern) {
                continue;
            }

            let payload = tokio::fs::read(&path).await?;
            let mut metadata = HashMap::new();
            metadata.insert("file_name".to_string(), file_name.clone());
            metadata.insert("directory".to_string(), directory.to_string());

            messages.push(ExternalMessage {
                id: Uuid::new_v4().to_string(),
                headers: HashMap::new(),
                payload,
                format: MessageFormat::Binary,
                timestamp: chrono::Utc::now(),
                source: format!("file://{directory}/{file_name}"),
                metadata,
            });

            // Consume the file so it is not re-read on the next poll.
            tokio::fs::remove_file(&path).await?;
        }

        Ok(messages)
    }

    /// Match a file name against a simple glob pattern.
    ///
    /// Supports `*` (match everything), `*.ext` (suffix), `prefix*` (prefix),
    /// and exact names. This intentionally covers the common cases without a
    /// regex dependency.
    fn file_name_matches(file_name: &str, pattern: &str) -> bool {
        if pattern == "*" || pattern.is_empty() {
            return true;
        }
        match (pattern.strip_prefix('*'), pattern.strip_suffix('*')) {
            (Some(suffix), _) if !pattern.ends_with('*') => file_name.ends_with(suffix),
            (_, Some(prefix)) if !pattern.starts_with('*') => file_name.starts_with(prefix),
            _ => file_name == pattern,
        }
    }

    /// Process a message through the bridge
    async fn process_message(
        bridge: &MessageBridge,
        message: &ExternalMessage,
        transformers: &Arc<RwLock<HashMap<String, Box<dyn MessageTransformer + Send + Sync>>>>,
        router: &Arc<RoutingEngine>,
    ) -> Result<()> {
        // Apply routing rules
        let action = router
            .evaluate_rules(&bridge.routing_rules, message)
            .await?;

        match action {
            RuleAction::Drop => {
                debug!("Message dropped by routing rule: {}", message.id);
                return Ok(());
            }
            RuleAction::Forward => {
                // Continue with normal processing
            }
            RuleAction::Transform { transformer } => {
                // Apply specific transformer
                let transformed = {
                    let transformers_guard = transformers.read().await;
                    let transformer = transformers_guard
                        .get(&transformer)
                        .ok_or_else(|| anyhow!("Transformer not found: {}", transformer))?;
                    transformer.transform(message)?
                };

                return Self::send_message(&bridge.target, &transformed).await;
            }
            _ => {
                // Handle other actions
                warn!("Routing action not implemented: {:?}", action);
            }
        }

        // Apply default transformation
        let transformed = {
            let transformers_guard = transformers.read().await;
            let transformer = transformers_guard
                .get(&bridge.transformer)
                .ok_or_else(|| anyhow!("Transformer not found: {}", bridge.transformer))?;
            transformer.transform(message)?
        };

        // Send to target
        Self::send_message(&bridge.target, &transformed).await
    }

    /// Send message to external system
    async fn send_message(target: &ExternalSystemConfig, message: &ExternalMessage) -> Result<()> {
        match &target.system_type {
            ExternalSystemType::Kafka {
                brokers, topics, ..
            } => Self::send_kafka_message(brokers, topics, message).await,
            ExternalSystemType::RabbitMQ {
                url,
                exchange,
                routing_key,
                ..
            } => Self::send_rabbitmq_message(url, exchange, routing_key, message).await,
            ExternalSystemType::RedisPubSub { url, channels } => {
                Self::send_redis_message(url, channels, message).await
            }
            ExternalSystemType::HttpRest {
                base_url,
                endpoints,
                headers,
            } => Self::send_http_message(base_url, endpoints, headers, message).await,
            ExternalSystemType::FileSystem { directory, .. } => {
                Self::send_file_message(directory, message).await
            }
            other => Err(anyhow!(
                "Message sending is not implemented for system type {:?}",
                other
            )),
        }
    }

    /// Send message to Kafka
    async fn send_kafka_message(
        _brokers: &[String],
        _topics: &[String],
        _message: &ExternalMessage,
    ) -> Result<()> {
        Err(anyhow!(
            "Kafka bridge producer transport is not implemented"
        ))
    }

    /// Send message to RabbitMQ
    async fn send_rabbitmq_message(
        _url: &str,
        _exchange: &str,
        _routing_key: &str,
        _message: &ExternalMessage,
    ) -> Result<()> {
        Err(anyhow!(
            "RabbitMQ bridge producer transport is not implemented"
        ))
    }

    /// Send message to Redis
    async fn send_redis_message(
        _url: &str,
        _channels: &[String],
        _message: &ExternalMessage,
    ) -> Result<()> {
        Err(anyhow!(
            "Redis Pub/Sub bridge producer transport is not implemented"
        ))
    }

    /// Send message via HTTP
    async fn send_http_message(
        _base_url: &str,
        _endpoints: &HashMap<String, String>,
        _headers: &HashMap<String, String>,
        _message: &ExternalMessage,
    ) -> Result<()> {
        Err(anyhow!(
            "HTTP REST bridge producer transport is not implemented"
        ))
    }

    /// Send a message to the file system by writing its payload to a new file in
    /// `directory` (named after the message id).
    async fn send_file_message(directory: &str, message: &ExternalMessage) -> Result<()> {
        tokio::fs::create_dir_all(directory).await?;
        let file_path = std::path::Path::new(directory).join(format!("{}.msg", message.id));
        tokio::fs::write(&file_path, &message.payload).await?;
        debug!("Wrote bridge message {} to {:?}", message.id, file_path);
        Ok(())
    }

    /// Update bridge statistics
    async fn update_bridge_stats(
        bridges: &Arc<RwLock<HashMap<String, MessageBridge>>>,
        bridge_id: &str,
        success: bool,
        duration: Duration,
    ) {
        let mut bridges_guard = bridges.write().await;
        if let Some(bridge) = bridges_guard.get_mut(bridge_id) {
            bridge.last_activity = Some(Instant::now());

            if success {
                bridge.stats.messages_sent += 1;
            } else {
                bridge.stats.messages_failed += 1;
            }

            // Update average processing time
            let total_messages = bridge.stats.messages_sent + bridge.stats.messages_failed;
            let avg_nanos = bridge.stats.avg_processing_time.as_nanos() as u64;
            let duration_nanos = duration.as_nanos() as u64;
            if let Some(new_avg_nanos) =
                (avg_nanos * (total_messages - 1) + duration_nanos).checked_div(total_messages)
            {
                bridge.stats.avg_processing_time = Duration::from_nanos(new_avg_nanos);
            }
        }
    }

    /// Get bridge statistics
    pub async fn get_bridge_stats(&self, bridge_id: &str) -> Result<BridgeStatistics> {
        let bridges = self.bridges.read().await;
        let bridge = bridges
            .get(bridge_id)
            .ok_or_else(|| anyhow!("Bridge not found"))?;

        Ok(bridge.stats.clone())
    }

    /// Get manager statistics
    pub async fn get_stats(&self) -> BridgeStats {
        self.stats.read().await.clone()
    }

    /// List all bridges
    pub async fn list_bridges(&self) -> Vec<BridgeInfo> {
        let bridges = self.bridges.read().await;
        bridges
            .values()
            .map(|b| BridgeInfo {
                id: b.id.clone(),
                bridge_type: b.bridge_type.clone(),
                status: format!("{:?}", b.status),
                created_at: b.created_at.elapsed(),
                last_activity: b.last_activity.map(|t| t.elapsed()),
                messages_processed: b.stats.messages_sent + b.stats.messages_failed,
                success_rate: if b.stats.messages_sent + b.stats.messages_failed > 0 {
                    b.stats.messages_sent as f64
                        / (b.stats.messages_sent + b.stats.messages_failed) as f64
                } else {
                    0.0
                },
            })
            .collect()
    }

    /// Subscribe to bridge notifications
    pub fn subscribe(&self) -> broadcast::Receiver<BridgeNotification> {
        self.event_notifier.subscribe()
    }
}

/// Bridge information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BridgeInfo {
    pub id: String,
    pub bridge_type: BridgeType,
    pub status: String,
    pub created_at: Duration,
    pub last_activity: Option<Duration>,
    pub messages_processed: u64,
    pub success_rate: f64,
}

impl RoutingEngine {
    /// Create a new routing engine
    fn new() -> Self {
        Self {
            _global_rules: Arc::new(RwLock::new(Vec::new())),
            _rule_cache: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Evaluate routing rules for a message
    async fn evaluate_rules(
        &self,
        rules: &[RoutingRule],
        message: &ExternalMessage,
    ) -> Result<RuleAction> {
        // Sort rules by priority
        let mut sorted_rules = rules.to_vec();
        sorted_rules.sort_by_key(|r| r.priority);

        // Evaluate rules in priority order
        for rule in sorted_rules.iter().filter(|r| r.enabled) {
            if self.evaluate_condition(&rule.condition, message).await? {
                return Ok(rule.action.clone());
            }
        }

        // Default action is forward
        Ok(RuleAction::Forward)
    }

    /// Evaluate a rule condition
    #[allow(clippy::only_used_in_recursion)]
    fn evaluate_condition<'a>(
        &'a self,
        condition: &'a RuleCondition,
        message: &'a ExternalMessage,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<bool>> + Send + 'a>> {
        Box::pin(async move {
            match condition {
                RuleCondition::Always => Ok(true),
                RuleCondition::EventType { types } => {
                    let unknown = "unknown".to_string();
                    let event_type = message
                        .headers
                        .get("event_type")
                        .or_else(|| message.metadata.get("event_type"))
                        .unwrap_or(&unknown);
                    Ok(types.contains(event_type))
                }
                RuleCondition::Graph { patterns } => {
                    let graph = message
                        .headers
                        .get("graph")
                        .or_else(|| message.metadata.get("graph"));
                    if let Some(g) = graph {
                        Ok(patterns.iter().any(|p| g.contains(p)))
                    } else {
                        Ok(false)
                    }
                }
                RuleCondition::SubjectPattern { regex } => {
                    let subject = message
                        .headers
                        .get("subject")
                        .or_else(|| message.metadata.get("subject"));
                    if let Some(s) = subject {
                        let re = regex::Regex::new(regex)
                            .map_err(|e| anyhow!("Invalid regex: {}", e))?;
                        Ok(re.is_match(s))
                    } else {
                        Ok(false)
                    }
                }
                RuleCondition::Predicate { predicates } => {
                    let predicate = message
                        .headers
                        .get("predicate")
                        .or_else(|| message.metadata.get("predicate"));
                    if let Some(p) = predicate {
                        Ok(predicates.contains(p))
                    } else {
                        Ok(false)
                    }
                }
                RuleCondition::Expression { expr } => {
                    // This would implement expression evaluation
                    warn!("Expression evaluation not implemented: {}", expr);
                    Ok(false)
                }
                RuleCondition::Composite {
                    operator,
                    conditions,
                } => match operator {
                    LogicalOperator::And => {
                        for cond in conditions {
                            if !self.evaluate_condition(cond, message).await? {
                                return Ok(false);
                            }
                        }
                        Ok(true)
                    }
                    LogicalOperator::Or => {
                        for cond in conditions {
                            if self.evaluate_condition(cond, message).await? {
                                return Ok(true);
                            }
                        }
                        Ok(false)
                    }
                    LogicalOperator::Not => {
                        if conditions.len() != 1 {
                            return Err(anyhow!("NOT operator requires exactly one condition"));
                        }
                        Ok(!self.evaluate_condition(&conditions[0], message).await?)
                    }
                },
            }
        })
    }
}

/// JSON message transformer
pub struct JsonTransformer;

impl MessageTransformer for JsonTransformer {
    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage> {
        // This would implement JSON transformation logic
        Ok(message.clone())
    }

    fn name(&self) -> &str {
        "json"
    }

    fn supported_formats(&self) -> (MessageFormat, MessageFormat) {
        (MessageFormat::Json, MessageFormat::Json)
    }
}

/// RDF to JSON transformer
pub struct RdfToJsonTransformer;

impl MessageTransformer for RdfToJsonTransformer {
    fn transform(&self, message: &ExternalMessage) -> Result<ExternalMessage> {
        // This would implement RDF to JSON transformation
        let mut transformed = message.clone();
        transformed.format = MessageFormat::Json;

        // Transform payload from RDF to JSON
        // For now, just pass through

        Ok(transformed)
    }

    fn name(&self) -> &str {
        "rdf-to-json"
    }

    fn supported_formats(&self) -> (MessageFormat, MessageFormat) {
        (MessageFormat::Text, MessageFormat::Json) // Assuming RDF as text
    }
}

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

    #[tokio::test]
    async fn test_bridge_creation() {
        let manager = MessageBridgeManager::new().await.unwrap();

        let source = ExternalSystemConfig {
            system_type: ExternalSystemType::Kafka {
                brokers: vec!["localhost:9092".to_string()],
                topics: vec!["source-topic".to_string()],
                consumer_group: Some("test-group".to_string()),
            },
            connection: ConnectionConfig {
                timeout: Duration::from_secs(30),
                keep_alive: Duration::from_secs(60),
                retry: RetryConfig {
                    max_attempts: 3,
                    initial_delay: Duration::from_millis(100),
                    max_delay: Duration::from_secs(10),
                    exponential_backoff: true,
                },
                tls: None,
            },
            format: FormatConfig {
                format: MessageFormat::Json,
                encoding: "utf-8".to_string(),
                compression: None,
                schema_validation: false,
            },
            security: SecurityConfig {
                auth: AuthenticationMethod::None,
                encryption: EncryptionConfig {
                    enabled: false,
                    algorithm: None,
                    key_id: None,
                },
                access_control: AccessControlConfig {
                    read_permissions: vec![],
                    write_permissions: vec![],
                    admin_permissions: vec![],
                },
            },
        };

        let target = source.clone(); // Same config for simplicity

        // Register transformer
        manager
            .register_transformer(Box::new(JsonTransformer))
            .await;

        let bridge_id = manager
            .create_bridge(
                BridgeType::SourceToTarget,
                source,
                target,
                "json".to_string(),
                vec![],
                BridgeConfig::default(),
            )
            .await
            .unwrap();

        assert!(!bridge_id.is_empty());

        let bridges = manager.list_bridges().await;
        assert_eq!(bridges.len(), 1);
        assert_eq!(bridges[0].id, bridge_id);
    }

    #[tokio::test]
    async fn test_routing_rules() {
        let engine = RoutingEngine::new();

        let rule = RoutingRule {
            name: "test-rule".to_string(),
            condition: RuleCondition::EventType {
                types: vec!["triple_added".to_string()],
            },
            action: RuleAction::Forward,
            priority: 1,
            enabled: true,
        };

        let mut message = ExternalMessage {
            id: "test".to_string(),
            headers: HashMap::new(),
            payload: vec![],
            format: MessageFormat::Json,
            timestamp: chrono::Utc::now(),
            source: "test".to_string(),
            metadata: HashMap::new(),
        };

        message
            .headers
            .insert("event_type".to_string(), "triple_added".to_string());

        let action = engine.evaluate_rules(&[rule], &message).await.unwrap();
        assert!(matches!(action, RuleAction::Forward));
    }

    #[test]
    fn regression_file_name_matches_glob() {
        assert!(MessageBridgeManager::file_name_matches("a.json", "*"));
        assert!(MessageBridgeManager::file_name_matches("a.json", "*.json"));
        assert!(!MessageBridgeManager::file_name_matches("a.txt", "*.json"));
        assert!(MessageBridgeManager::file_name_matches("data_1", "data_*"));
        assert!(!MessageBridgeManager::file_name_matches("other", "data_*"));
        assert!(MessageBridgeManager::file_name_matches("exact", "exact"));
    }

    #[tokio::test]
    async fn regression_file_transport_roundtrip() {
        let dir = std::env::temp_dir().join(format!("oxirs-bridge-{}", Uuid::new_v4()));
        tokio::fs::create_dir_all(&dir).await.unwrap();
        let dir_str = dir.to_string_lossy().to_string();

        // Write a message via the real file sender.
        let message = ExternalMessage {
            id: "msg-1".to_string(),
            headers: HashMap::new(),
            payload: b"hello-bridge".to_vec(),
            format: MessageFormat::Binary,
            timestamp: chrono::Utc::now(),
            source: "test".to_string(),
            metadata: HashMap::new(),
        };
        MessageBridgeManager::send_file_message(&dir_str, &message)
            .await
            .unwrap();

        // Receiving must actually read the file back (not return empty).
        let received = MessageBridgeManager::receive_file_messages(
            &dir_str,
            "*",
            false,
            &BridgeConfig::default(),
        )
        .await
        .unwrap();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].payload, b"hello-bridge");

        // File is consumed: a second receive returns nothing.
        let again = MessageBridgeManager::receive_file_messages(
            &dir_str,
            "*",
            false,
            &BridgeConfig::default(),
        )
        .await
        .unwrap();
        assert!(again.is_empty());

        let _ = tokio::fs::remove_dir_all(&dir).await;
    }

    #[tokio::test]
    async fn regression_start_bridge_rejects_unimplemented_transport() {
        let manager = MessageBridgeManager::new().await.unwrap();
        manager
            .register_transformer(Box::new(JsonTransformer))
            .await;

        let kafka = ExternalSystemConfig {
            system_type: ExternalSystemType::Kafka {
                brokers: vec!["localhost:9092".to_string()],
                topics: vec!["t".to_string()],
                consumer_group: None,
            },
            connection: ConnectionConfig {
                timeout: Duration::from_secs(1),
                keep_alive: Duration::from_secs(1),
                retry: RetryConfig {
                    max_attempts: 1,
                    initial_delay: Duration::from_millis(1),
                    max_delay: Duration::from_millis(1),
                    exponential_backoff: false,
                },
                tls: None,
            },
            format: FormatConfig {
                format: MessageFormat::Json,
                encoding: "utf-8".to_string(),
                compression: None,
                schema_validation: false,
            },
            security: SecurityConfig {
                auth: AuthenticationMethod::None,
                encryption: EncryptionConfig {
                    enabled: false,
                    algorithm: None,
                    key_id: None,
                },
                access_control: AccessControlConfig {
                    read_permissions: vec![],
                    write_permissions: vec![],
                    admin_permissions: vec![],
                },
            },
        };

        let bridge_id = manager
            .create_bridge(
                BridgeType::SourceToTarget,
                kafka.clone(),
                kafka,
                "json".to_string(),
                vec![],
                BridgeConfig::default(),
            )
            .await
            .unwrap();

        // Starting a bridge with an unimplemented transport must fail loud.
        assert!(manager.start_bridge(&bridge_id).await.is_err());
    }
}