rust-rabbit 1.2.2

use crate::{
    connection::Connection,
    error::RustRabbitError,
    message::{ErrorType, MassTransitEnvelope, MessageEnvelope},
    retry::RetryConfig,
};
use futures_lite::stream::StreamExt;
use lapin::{
    options::{BasicAckOptions, BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions},
    types::{AMQPValue, FieldTable},
    BasicProperties, Channel,
};
use serde::de::DeserializeOwned;
use serde::Serialize;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tracing::{debug, error, warn};

// Helper functions for reading/writing headers
const HEADER_RETRY_ATTEMPT: &str = "x-retry-attempt";
const HEADER_CORRELATION_ID: &str = "x-correlation-id";

/// Read retry_attempt from AMQP headers, defaulting to 0 if not present
fn read_retry_attempt(properties: &BasicProperties) -> u32 {
    if let Some(headers) = properties.headers() {
        // Iterate over headers to find the key
        for (header_key, value) in headers {
            if header_key.as_str() == HEADER_RETRY_ATTEMPT {
                if let AMQPValue::LongLongInt(attempt) = value {
                    return *attempt as u32;
                }
            }
        }
    }
    0
}

/// Read correlation_id from AMQP headers
fn read_correlation_id(properties: &BasicProperties) -> Option<String> {
    if let Some(headers) = properties.headers() {
        // Iterate over headers to find the key
        for (header_key, value) in headers {
            if header_key.as_str() == HEADER_CORRELATION_ID {
                if let AMQPValue::LongString(corr_id) = value {
                    return Some(corr_id.to_string());
                }
            }
        }
    }
    None
}

/// Create headers with retry_attempt and correlation_id
///
/// This function is kept for potential future use or testing scenarios
/// where header creation is needed outside of the main message processing flow.
#[allow(dead_code)]
fn create_headers(retry_attempt: u32, correlation_id: Option<&str>) -> FieldTable {
    let mut headers = FieldTable::default();
    headers.insert(
        HEADER_RETRY_ATTEMPT.into(),
        AMQPValue::LongLongInt(retry_attempt as i64),
    );
    if let Some(corr_id) = correlation_id {
        headers.insert(
            HEADER_CORRELATION_ID.into(),
            AMQPValue::LongString(corr_id.into()),
        );
    }
    headers
}

/// Update headers with new retry_attempt, preserving existing headers
fn update_headers_with_retry(
    existing_headers: Option<&FieldTable>,
    retry_attempt: u32,
) -> FieldTable {
    let mut headers = match existing_headers {
        Some(h) => h.clone(),
        None => FieldTable::default(),
    };
    headers.insert(
        HEADER_RETRY_ATTEMPT.into(),
        AMQPValue::LongLongInt(retry_attempt as i64),
    );
    headers
}

/// Consumer configuration builder
pub struct ConsumerBuilder {
    connection: Arc<Connection>,
    queue_name: String,
    exchange_name: Option<String>,
    routing_key: Option<String>,
    retry_config: Option<RetryConfig>,
    prefetch_count: Option<u16>,
    auto_ack: bool,
}

impl ConsumerBuilder {
    pub fn new(connection: Arc<Connection>, queue_name: impl Into<String>) -> Self {
        Self {
            connection,
            queue_name: queue_name.into(),
            exchange_name: None,
            routing_key: None,
            retry_config: None,
            prefetch_count: Some(10),
            auto_ack: true,
        }
    }

    /// Bind to an exchange with routing key
    pub fn bind_to_exchange(
        mut self,
        exchange: impl Into<String>,
        routing_key: impl Into<String>,
    ) -> Self {
        self.exchange_name = Some(exchange.into());
        self.routing_key = Some(routing_key.into());
        self
    }

    /// Set routing key (for use with bind_to_exchange)
    pub fn routing_key(mut self, routing_key: impl Into<String>) -> Self {
        self.routing_key = Some(routing_key.into());
        self
    }

    /// Configure retry behavior
    pub fn with_retry(mut self, retry_config: RetryConfig) -> Self {
        self.retry_config = Some(retry_config);
        self
    }

    /// Set TTL for dead letter queue (auto-cleanup failed messages)
    /// This is a convenience method that modifies the retry_config if it exists
    ///
    /// # Example
    /// ```rust,no_run
    /// # use rust_rabbit::{Connection, Consumer, RetryConfig};
    /// # use std::time::Duration;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let connection = Connection::new("amqp://localhost").await?;
    /// let consumer = Consumer::builder(connection, "orders")
    ///     .with_retry(RetryConfig::exponential_default())
    ///     .with_dlq_ttl(Duration::from_secs(86400))  // 1 day
    ///     .build();
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_dlq_ttl(mut self, ttl: Duration) -> Self {
        if let Some(retry_config) = self.retry_config.as_mut() {
            retry_config.dlq_ttl = Some(ttl);
        }
        self
    }

    /// Set prefetch count
    pub fn with_prefetch(mut self, count: u16) -> Self {
        self.prefetch_count = Some(count);
        self
    }

    /// Disable auto-acknowledge (manual ack required)
    pub fn manual_ack(mut self) -> Self {
        self.auto_ack = false;
        self
    }

    /// Build the consumer
    pub fn build(self) -> Consumer {
        Consumer {
            connection: self.connection,
            queue_name: self.queue_name,
            exchange_name: self.exchange_name,
            routing_key: self.routing_key,
            retry_config: self.retry_config,
            prefetch_count: self.prefetch_count.unwrap_or(10),
            auto_ack: self.auto_ack,
        }
    }
}

/// Simplified Consumer for message consumption
pub struct Consumer {
    connection: Arc<Connection>,
    queue_name: String,
    exchange_name: Option<String>,
    routing_key: Option<String>,
    retry_config: Option<RetryConfig>,
    prefetch_count: u16,
    auto_ack: bool,
}

impl Consumer {
    /// Create a new consumer builder
    pub fn builder(connection: Arc<Connection>, queue_name: impl Into<String>) -> ConsumerBuilder {
        ConsumerBuilder::new(connection, queue_name)
    }

    /// Create retry queue with TTL
    async fn create_retry_queue(
        &self,
        channel: &Channel,
        retry_attempt: u32,
        delay: std::time::Duration,
    ) -> Result<String, RustRabbitError> {
        let retry_queue_name = format!("{}.retry.{}", self.queue_name, retry_attempt);
        let delay_ms = delay.as_millis() as i64;

        // Create retry queue with TTL that routes back to original queue
        let mut args = FieldTable::default();
        args.insert("x-message-ttl".into(), AMQPValue::LongLongInt(delay_ms));
        args.insert(
            "x-dead-letter-exchange".into(),
            AMQPValue::LongString("".into()),
        ); // Default exchange
        args.insert(
            "x-dead-letter-routing-key".into(),
            AMQPValue::LongString(self.queue_name.clone().into()),
        );

        channel
            .queue_declare(
                &retry_queue_name,
                QueueDeclareOptions {
                    durable: true,
                    ..Default::default()
                },
                args,
            )
            .await?;

        debug!(
            "Created retry queue: {} with TTL: {}ms",
            retry_queue_name, delay_ms
        );
        Ok(retry_queue_name)
    }

    /// Create DLQ (Dead Letter Queue) with optional TTL
    async fn create_dlq(&self, channel: &Channel) -> Result<String, RustRabbitError> {
        let dlq_name = format!("{}.dlq", self.queue_name);

        // Build queue arguments with optional TTL
        let mut args = FieldTable::default();
        if let Some(retry_config) = &self.retry_config {
            if let Some(ttl) = &retry_config.dlq_ttl {
                let ttl_ms = ttl.as_millis() as i64;
                args.insert("x-message-ttl".into(), AMQPValue::LongLongInt(ttl_ms));
                debug!("DLQ TTL: {}ms", ttl_ms);
            }
        }

        channel
            .queue_declare(
                &dlq_name,
                QueueDeclareOptions {
                    durable: true,
                    ..Default::default()
                },
                args,
            )
            .await?;

        debug!("Created DLQ: {}", dlq_name);
        Ok(dlq_name)
    }

    /// Send message to retry queue with delay
    async fn send_to_retry_queue(
        &self,
        channel: &Channel,
        message_data: &[u8],
        retry_attempt: u32,
        delay: std::time::Duration,
    ) -> Result<(), RustRabbitError> {
        let retry_queue_name = self
            .create_retry_queue(channel, retry_attempt, delay)
            .await?;

        // Publish to retry queue
        channel
            .basic_publish(
                "", // Default exchange
                &retry_queue_name,
                BasicPublishOptions::default(),
                message_data,
                BasicProperties::default()
                    .with_content_type("application/json".into())
                    .with_delivery_mode(2), // Persistent
            )
            .await?
            .await?;

        debug!("Sent message to retry queue: {}", retry_queue_name);
        Ok(())
    }

    /// Send message to retry queue with delay and custom headers
    async fn send_to_retry_queue_with_headers(
        &self,
        channel: &Channel,
        message_data: &[u8],
        retry_attempt: u32,
        delay: std::time::Duration,
        headers: FieldTable,
    ) -> Result<(), RustRabbitError> {
        let retry_queue_name = self
            .create_retry_queue(channel, retry_attempt, delay)
            .await?;

        // Publish to retry queue with headers
        channel
            .basic_publish(
                "", // Default exchange
                &retry_queue_name,
                BasicPublishOptions::default(),
                message_data,
                BasicProperties::default()
                    .with_content_type("application/json".into())
                    .with_delivery_mode(2) // Persistent
                    .with_headers(headers),
            )
            .await?
            .await?;

        debug!(
            "Sent message to retry queue with headers: {}",
            retry_queue_name
        );
        Ok(())
    }

    /// Send message to DLQ
    async fn send_to_dlq_simple(
        &self,
        channel: &Channel,
        message_data: &[u8],
    ) -> Result<(), RustRabbitError> {
        let dlq_name = self.create_dlq(channel).await?;

        // Publish to DLQ
        channel
            .basic_publish(
                "", // Default exchange
                &dlq_name,
                BasicPublishOptions::default(),
                message_data,
                BasicProperties::default()
                    .with_content_type("application/json".into())
                    .with_delivery_mode(2), // Persistent
            )
            .await?
            .await?;

        debug!("Sent message to DLQ: {}", dlq_name);
        Ok(())
    }

    /// Create delay exchange using RabbitMQ delayed message exchange plugin
    /// Requires rabbitmq_delayed_message_exchange plugin to be installed on RabbitMQ
    async fn create_delay_exchange(&self, channel: &Channel) -> Result<String, RustRabbitError> {
        if let Some(retry_config) = &self.retry_config {
            let delay_exchange = retry_config.get_delay_exchange(&self.queue_name);

            // Declare delay exchange with x-delayed-type argument
            let mut args = FieldTable::default();
            args.insert(
                "x-delayed-type".into(),
                AMQPValue::LongString("direct".into()),
            );

            channel
                .exchange_declare(
                    &delay_exchange,
                    lapin::ExchangeKind::Custom("x-delayed-message".to_string()),
                    lapin::options::ExchangeDeclareOptions {
                        durable: true,
                        ..Default::default()
                    },
                    args,
                )
                .await?;

            debug!(
                "Created delay exchange: {} (x-delayed-message type)",
                delay_exchange
            );
            Ok(delay_exchange)
        } else {
            Err(RustRabbitError::Retry(
                "Retry config not configured".to_string(),
            ))
        }
    }

    /// Send message to delay exchange with x-delay header for retry
    /// Message will be automatically routed back to the original queue after delay
    async fn send_to_delay_exchange(
        &self,
        channel: &Channel,
        message_data: &[u8],
        delay: std::time::Duration,
    ) -> Result<(), RustRabbitError> {
        let delay_exchange = self.create_delay_exchange(channel).await?;
        let delay_ms = delay.as_millis() as i64;

        // Publish to delay exchange with x-delay header
        // The message will be re-delivered to original queue after delay
        channel
            .basic_publish(
                &delay_exchange,
                &self.queue_name, // Routing key: original queue name
                BasicPublishOptions::default(),
                message_data,
                BasicProperties::default()
                    .with_content_type("application/json".into())
                    .with_delivery_mode(2) // Persistent
                    .with_headers({
                        let mut headers = FieldTable::default();
                        headers.insert("x-delay".into(), AMQPValue::LongLongInt(delay_ms));
                        headers
                    }),
            )
            .await?
            .await?;

        debug!(
            "Sent message to delay exchange: {} with delay: {}ms",
            delay_exchange, delay_ms
        );
        Ok(())
    }

    /// Send message to delay exchange with x-delay header and custom headers for retry
    /// Message will be automatically routed back to the original queue after delay
    async fn send_to_delay_exchange_with_headers(
        &self,
        channel: &Channel,
        message_data: &[u8],
        delay: std::time::Duration,
        mut headers: FieldTable,
    ) -> Result<(), RustRabbitError> {
        let delay_exchange = self.create_delay_exchange(channel).await?;
        let delay_ms = delay.as_millis() as i64;

        // Add x-delay header to existing headers
        headers.insert("x-delay".into(), AMQPValue::LongLongInt(delay_ms));

        // Publish to delay exchange with headers
        // The message will be re-delivered to original queue after delay
        channel
            .basic_publish(
                &delay_exchange,
                &self.queue_name, // Routing key: original queue name
                BasicPublishOptions::default(),
                message_data,
                BasicProperties::default()
                    .with_content_type("application/json".into())
                    .with_delivery_mode(2) // Persistent
                    .with_headers(headers),
            )
            .await?
            .await?;

        debug!(
            "Sent message to delay exchange with headers: {} with delay: {}ms",
            delay_exchange, delay_ms
        );
        Ok(())
    }

    /// Start consuming messages with smart MassTransit detection
    /// Handler receives just the payload type T (no wrapper)
    pub async fn consume<T, H, Fut>(&self, handler: H) -> Result<(), RustRabbitError>
    where
        T: DeserializeOwned + Send + Clone + Sync + 'static + Serialize,
        H: Fn(T) -> Fut + Send + Sync + Clone + 'static,
        Fut: Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send,
    {
        let channel = self.connection.create_channel().await?;

        // Set prefetch count
        channel
            .basic_qos(
                self.prefetch_count,
                lapin::options::BasicQosOptions::default(),
            )
            .await?;

        // Setup infrastructure (queues, exchanges)
        self.setup_infrastructure(&channel).await?;

        // Start consuming
        let mut consumer = channel
            .basic_consume(
                &self.queue_name,
                "",
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await?;

        let semaphore = Arc::new(Semaphore::new(self.prefetch_count as usize));

        debug!("Started consuming from queue: {}", self.queue_name);

        // Process messages
        while let Some(delivery_result) = consumer.next().await {
            let delivery = delivery_result?;
            let permit = semaphore.clone().acquire_owned().await.unwrap();
            let handler_clone = handler.clone();
            let auto_ack = self.auto_ack;
            let channel_clone = Arc::new(channel.clone());
            let retry_config = self.retry_config.clone();
            let consumer_self = Consumer {
                connection: self.connection.clone(),
                queue_name: self.queue_name.clone(),
                exchange_name: self.exchange_name.clone(),
                routing_key: self.routing_key.clone(),
                retry_config: self.retry_config.clone(),
                prefetch_count: self.prefetch_count,
                auto_ack: self.auto_ack,
            };

            tokio::spawn(async move {
                let _permit = permit;
                let delivery_tag = delivery.delivery_tag;
                let properties = delivery.properties;

                // Smart detection: Try MassTransit format first
                let (payload, correlation_id_from_mt) =
                    match MassTransitEnvelope::from_slice(&delivery.data) {
                        Ok(mt_envelope) => {
                            // MassTransit format detected - extract payload
                            match mt_envelope.extract_message::<T>() {
                                Ok(data) => {
                                    debug!("Detected MassTransit format, extracted payload");
                                    (data, mt_envelope.correlation_id().map(|s| s.to_string()))
                                }
                                Err(e) => {
                                    error!(
                                        "Failed to extract payload from MassTransit envelope: {}",
                                        e
                                    );
                                    if auto_ack {
                                        if let Err(e) = channel_clone
                                            .basic_nack(
                                                delivery_tag,
                                                lapin::options::BasicNackOptions {
                                                    multiple: false,
                                                    requeue: false,
                                                },
                                            )
                                            .await
                                        {
                                            error!("Failed to nack malformed message: {}", e);
                                        }
                                    }
                                    return;
                                }
                            }
                        }
                        Err(_) => {
                            // Not MassTransit format - try direct deserialization
                            match serde_json::from_slice::<T>(&delivery.data) {
                                Ok(data) => {
                                    debug!("Direct format detected");
                                    (data, None)
                                }
                                Err(e) => {
                                    error!("Failed to deserialize message: {}", e);
                                    if auto_ack {
                                        if let Err(e) = channel_clone
                                            .basic_nack(
                                                delivery_tag,
                                                lapin::options::BasicNackOptions {
                                                    multiple: false,
                                                    requeue: false,
                                                },
                                            )
                                            .await
                                        {
                                            error!("Failed to nack malformed message: {}", e);
                                        }
                                    }
                                    return;
                                }
                            }
                        }
                    };

                // Read metadata from headers
                let retry_attempt = read_retry_attempt(&properties);
                let correlation_id =
                    correlation_id_from_mt.or_else(|| read_correlation_id(&properties));

                // If we got correlation_id from MassTransit but it's not in headers, we should add it
                // But for now, we'll just use it for retries

                // Process message - handler receives just T
                match handler_clone(payload.clone()).await {
                    Ok(()) => {
                        if auto_ack {
                            if let Err(e) = channel_clone
                                .basic_ack(delivery_tag, BasicAckOptions::default())
                                .await
                            {
                                error!("Failed to ack message: {}", e);
                            }
                        }
                        debug!("Message processed successfully");
                    }
                    Err(e) => {
                        error!("Handler error: {}", e);
                        if auto_ack {
                            // Check if retry is configured
                            if let Some(retry_cfg) = &retry_config {
                                if retry_attempt < retry_cfg.max_retries {
                                    // Calculate delay for next retry
                                    if let Some(delay) = retry_cfg.calculate_delay(retry_attempt) {
                                        warn!(
                                            "Scheduling retry {} with delay {:?} for message",
                                            retry_attempt + 1,
                                            delay
                                        );

                                        // Serialize payload (raw, no wrapper)
                                        let payload_bytes = match serde_json::to_vec(&payload) {
                                            Ok(bytes) => bytes,
                                            Err(e) => {
                                                error!(
                                                    "Failed to serialize payload for retry: {}",
                                                    e
                                                );
                                                if let Err(e) = channel_clone
                                                    .basic_nack(
                                                        delivery_tag,
                                                        lapin::options::BasicNackOptions {
                                                            multiple: false,
                                                            requeue: false,
                                                        },
                                                    )
                                                    .await
                                                {
                                                    error!("Failed to nack message: {}", e);
                                                }
                                                return;
                                            }
                                        };

                                        // Update headers with new retry_attempt, preserving existing headers
                                        let mut updated_headers = update_headers_with_retry(
                                            properties.headers().as_ref(),
                                            retry_attempt + 1,
                                        );

                                        // Preserve correlation_id in headers if present
                                        if let Some(corr_id) = &correlation_id {
                                            updated_headers.insert(
                                                HEADER_CORRELATION_ID.into(),
                                                AMQPValue::LongString(corr_id.clone().into()),
                                            );
                                        }

                                        // Send via appropriate strategy (TTL or DelayedExchange)
                                        let send_result = if matches!(
                                            retry_cfg.delay_strategy,
                                            crate::retry::DelayStrategy::DelayedExchange
                                        ) {
                                            consumer_self
                                                .send_to_delay_exchange_with_headers(
                                                    &channel_clone,
                                                    &payload_bytes,
                                                    delay,
                                                    updated_headers,
                                                )
                                                .await
                                        } else {
                                            consumer_self
                                                .send_to_retry_queue_with_headers(
                                                    &channel_clone,
                                                    &payload_bytes,
                                                    retry_attempt + 1,
                                                    delay,
                                                    updated_headers,
                                                )
                                                .await
                                        };

                                        if let Err(e) = send_result {
                                            error!("Failed to send retry message: {}", e);
                                            if let Err(e) = channel_clone
                                                .basic_nack(
                                                    delivery_tag,
                                                    lapin::options::BasicNackOptions {
                                                        multiple: false,
                                                        requeue: false,
                                                    },
                                                )
                                                .await
                                            {
                                                error!("Failed to nack message: {}", e);
                                            }
                                            return;
                                        }

                                        // ACK original message (it's now queued for retry)
                                        if let Err(e) = channel_clone
                                            .basic_ack(delivery_tag, BasicAckOptions::default())
                                            .await
                                        {
                                            error!("Failed to ack message after retry: {}", e);
                                        }
                                    } else {
                                        // No more retries, send to DLQ
                                        warn!("Retry exhausted, sending to DLQ");
                                        let payload_bytes = match serde_json::to_vec(&payload) {
                                            Ok(bytes) => bytes,
                                            Err(e) => {
                                                error!(
                                                    "Failed to serialize payload for DLQ: {}. Creating error payload.",
                                                    e
                                                );
                                                // Create descriptive error payload instead of using raw delivery.data
                                                serde_json::to_vec(&serde_json::json!({
                                                    "error": "Failed to serialize payload for DLQ",
                                                    "serialization_error": e.to_string(),
                                                    "retry_attempt": retry_attempt,
                                                    "original_message_size": delivery.data.len()
                                                }))
                                                .unwrap_or_else(|_| {
                                                    // Last resort: minimal error message
                                                    format!(
                                                        r#"{{"error":"Serialization failed for DLQ","attempt":{}}}"#,
                                                        retry_attempt
                                                    )
                                                    .into_bytes()
                                                })
                                            }
                                        };
                                        if let Err(e) = consumer_self
                                            .send_to_dlq_simple(&channel_clone, &payload_bytes)
                                            .await
                                        {
                                            error!("Failed to send to DLQ: {}", e);
                                        }
                                        if let Err(e) = channel_clone
                                            .basic_ack(delivery_tag, BasicAckOptions::default())
                                            .await
                                        {
                                            error!("Failed to ack message after DLQ: {}", e);
                                        }
                                    }
                                } else {
                                    // Retry exhausted, send to DLQ
                                    warn!("Max retries reached, sending to DLQ");
                                    let payload_bytes = match serde_json::to_vec(&payload) {
                                        Ok(bytes) => bytes,
                                        Err(e) => {
                                            error!(
                                                "Failed to serialize payload for DLQ: {}. Creating error payload.",
                                                e
                                            );
                                            // Create descriptive error payload instead of using raw delivery.data
                                            serde_json::to_vec(&serde_json::json!({
                                                "error": "Failed to serialize payload for DLQ",
                                                "serialization_error": e.to_string(),
                                                "retry_attempt": retry_attempt,
                                                "original_message_size": delivery.data.len()
                                            }))
                                            .unwrap_or_else(|_| {
                                                // Last resort: minimal error message
                                                format!(
                                                    r#"{{"error":"Serialization failed for DLQ","attempt":{}}}"#,
                                                    retry_attempt
                                                )
                                                .into_bytes()
                                            })
                                        }
                                    };
                                    if let Err(e) = consumer_self
                                        .send_to_dlq_simple(&channel_clone, &payload_bytes)
                                        .await
                                    {
                                        error!("Failed to send to DLQ: {}", e);
                                    }
                                    if let Err(e) = channel_clone
                                        .basic_ack(delivery_tag, BasicAckOptions::default())
                                        .await
                                    {
                                        error!("Failed to ack message after DLQ: {}", e);
                                    }
                                }
                            } else {
                                // No retry config, just nack
                                if let Err(e) = channel_clone
                                    .basic_nack(
                                        delivery_tag,
                                        lapin::options::BasicNackOptions {
                                            multiple: false,
                                            requeue: false,
                                        },
                                    )
                                    .await
                                {
                                    error!("Failed to nack message: {}", e);
                                }
                            }
                        }
                    }
                }
            });
        }

        Ok(())
    }

    /// Setup queue and exchange infrastructure
    async fn setup_infrastructure(&self, channel: &Channel) -> Result<(), RustRabbitError> {
        // Declare queue
        channel
            .queue_declare(
                &self.queue_name,
                QueueDeclareOptions {
                    durable: true,
                    ..Default::default()
                },
                FieldTable::default(),
            )
            .await?;

        // Bind to exchange if specified
        if let (Some(exchange), Some(routing_key)) = (&self.exchange_name, &self.routing_key) {
            channel
                .queue_bind(
                    &self.queue_name,
                    exchange,
                    routing_key,
                    lapin::options::QueueBindOptions::default(),
                    FieldTable::default(),
                )
                .await?;
        }

        // Setup delay exchange if using DelayedExchange strategy
        if let Some(retry_config) = &self.retry_config {
            if matches!(
                retry_config.delay_strategy,
                crate::retry::DelayStrategy::DelayedExchange
            ) {
                let delay_exchange = self.create_delay_exchange(channel).await?;

                // Bind delay exchange to original queue
                channel
                    .queue_bind(
                        &self.queue_name,
                        &delay_exchange,
                        &self.queue_name, // Routing key: original queue name
                        lapin::options::QueueBindOptions::default(),
                        FieldTable::default(),
                    )
                    .await?;

                debug!(
                    "Bound queue {} to delay exchange {}",
                    self.queue_name, delay_exchange
                );
            }
        }

        Ok(())
    }

    /// Start consuming message envelopes with full retry support
    pub async fn consume_envelopes<T, H, Fut>(&self, handler: H) -> Result<(), RustRabbitError>
    where
        T: DeserializeOwned + Send + Clone + Sync + 'static + serde::Serialize,
        H: Fn(MessageEnvelope<T>) -> Fut + Send + Sync + Clone + 'static,
        Fut: Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send,
    {
        let channel = self.connection.create_channel().await?;
        let retry_config = self.retry_config.clone();

        // Set prefetch count
        channel
            .basic_qos(
                self.prefetch_count,
                lapin::options::BasicQosOptions::default(),
            )
            .await?;

        // Setup queue and exchange
        self.setup_infrastructure(&channel).await?;

        // Create consumer
        let mut consumer = channel
            .basic_consume(
                &self.queue_name,
                "rust-rabbit-envelope-consumer",
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await?;

        let semaphore = Arc::new(Semaphore::new(self.prefetch_count as usize));

        debug!(
            "Started consuming envelopes from queue: {}",
            self.queue_name
        );

        // Process message envelopes with retry support
        while let Some(delivery_result) = consumer.next().await {
            let delivery = delivery_result?;
            let permit = semaphore.clone().acquire_owned().await.unwrap();
            let handler_clone = handler.clone();
            let auto_ack = self.auto_ack;
            let channel_clone = Arc::new(channel.clone());
            let retry_config_clone = retry_config.clone();
            let queue_name = self.queue_name.clone();
            let connection = self.connection.clone();

            tokio::spawn(async move {
                let _permit = permit;

                // Try to deserialize as MessageEnvelope
                match serde_json::from_slice::<MessageEnvelope<T>>(&delivery.data) {
                    Ok(mut envelope) => {
                        debug!(
                            "Processing envelope {} (attempt {}/{})",
                            envelope.metadata.message_id,
                            envelope.metadata.retry_attempt + 1,
                            envelope.metadata.max_retries + 1
                        );

                        // Process message
                        match handler_clone(envelope.clone()).await {
                            Ok(()) => {
                                if auto_ack {
                                    if let Err(e) = channel_clone
                                        .basic_ack(
                                            delivery.delivery_tag,
                                            BasicAckOptions::default(),
                                        )
                                        .await
                                    {
                                        error!("Failed to ack message: {}", e);
                                    }
                                }
                                debug!(
                                    "Envelope {} processed successfully",
                                    envelope.metadata.message_id
                                );
                            }
                            Err(e) => {
                                error!(
                                    "Handler error for envelope {}: {}",
                                    envelope.metadata.message_id, e
                                );

                                // Determine error type (simplified classification)
                                let error_type = classify_error(e.as_ref());

                                // Add error to envelope
                                envelope = envelope.with_error(
                                    &e.to_string(),
                                    error_type,
                                    Some(&format!("Queue: {}", queue_name)),
                                );

                                if auto_ack {
                                    // Check if we should retry
                                    if let Some(retry_cfg) = &retry_config_clone {
                                        if !envelope.is_retry_exhausted() {
                                            // Calculate delay and schedule retry
                                            if let Some(delay) = retry_cfg
                                                .calculate_delay(envelope.metadata.retry_attempt)
                                            {
                                                warn!(
                                                    "Scheduling retry {} for envelope {} with delay {:?}",
                                                    envelope.metadata.retry_attempt + 1,
                                                    envelope.metadata.message_id,
                                                    delay
                                                );

                                                // Increment retry attempt in envelope
                                                envelope.metadata.retry_attempt += 1;

                                                // Serialize updated envelope
                                                match serde_json::to_vec(&envelope) {
                                                    Ok(retry_payload) => {
                                                        // Create consumer instance for access to methods
                                                        let consumer_self = Consumer {
                                                            connection: connection.clone(),
                                                            queue_name: queue_name.clone(),
                                                            exchange_name: None,
                                                            routing_key: None,
                                                            retry_config: retry_config_clone
                                                                .clone(),
                                                            prefetch_count: 10,
                                                            auto_ack: true,
                                                        };

                                                        // Send via appropriate strategy (TTL or DelayedExchange)
                                                        let send_result = if matches!(
                                                            retry_config_clone
                                                                .as_ref()
                                                                .map(|c| c.delay_strategy),
                                                            Some(crate::retry::DelayStrategy::DelayedExchange)
                                                        ) {
                                                            consumer_self
                                                                .send_to_delay_exchange(
                                                                    &channel_clone,
                                                                    &retry_payload,
                                                                    delay,
                                                                )
                                                                .await
                                                        } else {
                                                            consumer_self
                                                                .send_to_retry_queue(
                                                                    &channel_clone,
                                                                    &retry_payload,
                                                                    envelope.metadata.retry_attempt,
                                                                    delay,
                                                                )
                                                                .await
                                                        };

                                                        if let Err(e) = send_result {
                                                            error!("Failed to send envelope for retry: {}", e);
                                                            // Fallback to simple nack
                                                            if let Err(e) = channel_clone
                                                                .basic_nack(
                                                                    delivery.delivery_tag,
                                                                    lapin::options::BasicNackOptions {
                                                                        multiple: false,
                                                                        requeue: false,
                                                                    },
                                                                )
                                                                .await
                                                            {
                                                                error!("Failed to nack message: {}", e);
                                                            }
                                                            return;
                                                        }

                                                        // ACK original message (it's now queued for retry)
                                                        if let Err(e) = channel_clone
                                                            .basic_ack(
                                                                delivery.delivery_tag,
                                                                BasicAckOptions::default(),
                                                            )
                                                            .await
                                                        {
                                                            error!("Failed to ack message after retry: {}", e);
                                                        }
                                                    }
                                                    Err(e) => {
                                                        error!("Failed to serialize envelope for retry: {}", e);
                                                        // Fallback to simple nack
                                                        if let Err(e) = channel_clone
                                                            .basic_nack(
                                                                delivery.delivery_tag,
                                                                lapin::options::BasicNackOptions {
                                                                    multiple: false,
                                                                    requeue: false,
                                                                },
                                                            )
                                                            .await
                                                        {
                                                            error!("Failed to nack message: {}", e);
                                                        }
                                                    }
                                                }
                                            } else {
                                                // No more retries, send to DLQ
                                                Self::send_to_dlq(
                                                    &envelope,
                                                    retry_cfg,
                                                    &connection,
                                                    &queue_name,
                                                )
                                                .await;

                                                // ACK original message
                                                if let Err(e) = channel_clone
                                                    .basic_ack(
                                                        delivery.delivery_tag,
                                                        BasicAckOptions::default(),
                                                    )
                                                    .await
                                                {
                                                    error!(
                                                        "Failed to ack message after DLQ: {}",
                                                        e
                                                    );
                                                }
                                            }
                                        } else {
                                            // Retry exhausted, send to DLQ
                                            warn!(
                                                "Retry exhausted for envelope {}",
                                                envelope.metadata.message_id
                                            );
                                            Self::send_to_dlq(
                                                &envelope,
                                                retry_cfg,
                                                &connection,
                                                &queue_name,
                                            )
                                            .await;

                                            // ACK original message
                                            if let Err(e) = channel_clone
                                                .basic_ack(
                                                    delivery.delivery_tag,
                                                    BasicAckOptions::default(),
                                                )
                                                .await
                                            {
                                                error!("Failed to ack message after DLQ: {}", e);
                                            }
                                        }
                                    } else {
                                        // No retry config, just nack
                                        if let Err(e) = channel_clone
                                            .basic_nack(
                                                delivery.delivery_tag,
                                                lapin::options::BasicNackOptions {
                                                    multiple: false,
                                                    requeue: false,
                                                },
                                            )
                                            .await
                                        {
                                            error!("Failed to nack message: {}", e);
                                        }
                                    }
                                }
                            }
                        }
                    }
                    Err(e) => {
                        error!("Failed to deserialize message envelope: {}", e);
                        if auto_ack {
                            // Reject malformed messages
                            if let Err(e) = channel_clone
                                .basic_nack(
                                    delivery.delivery_tag,
                                    lapin::options::BasicNackOptions {
                                        multiple: false,
                                        requeue: false,
                                    },
                                )
                                .await
                            {
                                error!("Failed to nack malformed envelope: {}", e);
                            }
                        }
                    }
                }
            });
        }

        Ok(())
    }

    /// Send failed message to Dead Letter Queue
    async fn send_to_dlq<T>(
        envelope: &MessageEnvelope<T>,
        retry_config: &RetryConfig,
        connection: &Arc<Connection>,
        queue_name: &str,
    ) where
        T: serde::Serialize,
    {
        match connection.create_channel().await {
            Ok(dlq_channel) => {
                let dlq_name = retry_config.get_dead_letter_queue(queue_name);

                // Declare DLQ
                if let Err(e) = dlq_channel
                    .queue_declare(
                        &dlq_name,
                        QueueDeclareOptions {
                            durable: true,
                            ..Default::default()
                        },
                        FieldTable::default(),
                    )
                    .await
                {
                    error!("Failed to declare DLQ {}: {}", dlq_name, e);
                    return;
                }

                // Publish to DLQ with failure summary
                let failure_summary = envelope.get_failure_summary();
                let dlq_payload = serde_json::json!({
                    "envelope": envelope,
                    "failure_summary": failure_summary,
                    "sent_to_dlq_at": chrono::Utc::now(),
                });

                if let Ok(payload_bytes) = serde_json::to_vec(&dlq_payload) {
                    if let Err(e) = dlq_channel
                        .basic_publish(
                            "",
                            &dlq_name,
                            lapin::options::BasicPublishOptions::default(),
                            &payload_bytes,
                            lapin::BasicProperties::default(),
                        )
                        .await
                    {
                        error!("Failed to publish to DLQ {}: {}", dlq_name, e);
                    } else {
                        warn!(
                            "Sent envelope {} to DLQ: {}",
                            envelope.metadata.message_id, failure_summary
                        );
                    }
                }
            }
            Err(e) => {
                error!("Failed to create DLQ channel: {}", e);
            }
        }
    }
}

/// Classify error type based on error message (simplified heuristics)
fn classify_error(error: &(dyn std::error::Error + Send + Sync)) -> ErrorType {
    let error_msg = error.to_string().to_lowercase();

    if error_msg.contains("timeout")
        || error_msg.contains("connection")
        || error_msg.contains("network")
        || error_msg.contains("temporary")
    {
        ErrorType::Transient
    } else if error_msg.contains("rate limit")
        || error_msg.contains("quota")
        || error_msg.contains("resource")
    {
        ErrorType::Resource
    } else if error_msg.contains("validation")
        || error_msg.contains("authentication")
        || error_msg.contains("authorization")
        || error_msg.contains("invalid")
        || error_msg.contains("bad request")
    {
        ErrorType::Permanent
    } else {
        ErrorType::Unknown
    }
}