mecha10-core 0.8.0

//! Dead Letter Queue (DLQ) for Failed Message Processing
//!
//! Provides a mechanism to capture and handle messages that fail processing repeatedly.
//! This prevents infinite retry loops and data loss while enabling debugging and recovery.
//!
//! # Use Cases
//!
//! - Messages that fail validation
//! - Messages that cause processing errors
//! - Messages that exceed retry limits
//! - Poison messages that would block a queue
//!
//! # Example
//!
//! ```rust,no_run
//! use mecha10_core::prelude::*;
//! use mecha10_core::topics::sensor;
//! use mecha10_core::messages::Image;
//! use mecha10_core::dead_letter::{DeadLetterQueue, FailedMessage};
//!
//! # async fn process_image(image: &Image) -> Result<Image> { Ok(image.clone()) }
//! # async fn example(ctx: &Context) -> Result<()> {
//! let dlq = DeadLetterQueue::new("vision_processing");
//!
//! let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
//!
//! while let Some(image) = images.recv().await {
//!     match process_image(&image).await {
//!         Ok(result) => {
//!             ctx.publish_to(sensor::CAMERA_RGB, &result).await?;
//!         }
//!         Err(e) => {
//!             // Send to dead letter queue
//!             dlq.send(
//!                 &image,
//!                 "vision_processing",
//!                 &e.to_string(),
//!                 None
//!             ).await?;
//!         }
//!     }
//! }
//! # Ok(())
//! # }
//! ```

use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use crate::prelude::now_micros;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FailedMessage<T> {
    pub message: T,

    pub failed_at: u64,

    pub source: String,

    pub error: String,

    pub retry_count: u32,

    pub metadata: HashMap<String, String>,

    pub stack_trace: Option<String>,
}

impl<T: Message + Serialize> Message for FailedMessage<T> where FailedMessage<T>: Send + Sync + 'static {}

#[derive(Debug, Clone)]
pub struct DeadLetterConfig {
    pub max_size: usize,

    pub persist: bool,

    pub persist_path: Option<String>,

    pub publish_to_topic: bool,

    pub topic_prefix: String,
}

impl Default for DeadLetterConfig {
    fn default() -> Self {
        Self {
            max_size: 1000,
            persist: false,
            persist_path: None,
            publish_to_topic: true,
            topic_prefix: "/dlq".to_string(),
        }
    }
}

#[derive(Clone)]
pub struct DeadLetterQueue {
    name: String,
    config: DeadLetterConfig,
    messages: Arc<RwLock<Vec<serde_json::Value>>>,
    stats: Arc<RwLock<DLQStats>>,
}

#[derive(Debug, Clone, Default)]
struct DLQStats {
    total_failed: u64,
    total_recovered: u64,
    total_dropped: u64,
}

impl DeadLetterQueue {
    /// Create a new dead letter queue
    ///
    /// # Arguments
    ///
    /// * `name` - Name for this DLQ (e.g., "vision_processor")
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10_core::dead_letter::DeadLetterQueue;
    ///
    /// let dlq = DeadLetterQueue::new("my_processor");
    /// ```
    pub fn new(name: &str) -> Self {
        Self::with_config(name, DeadLetterConfig::default())
    }

    /// Create a new dead letter queue with custom configuration
    pub fn with_config(name: &str, config: DeadLetterConfig) -> Self {
        info!(
            "Creating dead letter queue '{}' (max_size: {}, persist: {})",
            name, config.max_size, config.persist
        );

        Self {
            name: name.to_string(),
            config,
            messages: Arc::new(RwLock::new(Vec::new())),
            stats: Arc::new(RwLock::new(DLQStats::default())),
        }
    }

    /// Send a failed message to the dead letter queue
    ///
    /// # Arguments
    ///
    /// * `message` - The message that failed processing
    /// * `source` - Source topic/queue
    /// * `error` - Error message
    /// * `retry_count` - Number of retries attempted
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::dead_letter::DeadLetterQueue;
    /// # async fn example(dlq: &DeadLetterQueue, image: &Image) -> Result<()> {
    /// dlq.send(
    ///     image,
    ///     "sensor/camera/rgb",
    ///     "Invalid image format",
    ///     Some(3)
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn send<T: Message + Serialize + Clone>(
        &self,
        message: &T,
        source: &str,
        error: &str,
        retry_count: Option<u32>,
    ) -> Result<()> {
        self.send_with_metadata(message, source, error, retry_count, HashMap::new())
            .await
    }

    /// Send a failed message with additional metadata
    ///
    /// # Arguments
    ///
    /// * `message` - The message that failed
    /// * `source` - Source topic/queue
    /// * `error` - Error message
    /// * `retry_count` - Number of retries
    /// * `metadata` - Additional debugging context
    pub async fn send_with_metadata<T: Message + Serialize + Clone>(
        &self,
        message: &T,
        source: &str,
        error: &str,
        retry_count: Option<u32>,
        metadata: HashMap<String, String>,
    ) -> Result<()> {
        let failed_msg = FailedMessage {
            message: message.clone(),
            failed_at: now_micros(),
            source: source.to_string(),
            error: error.to_string(),
            retry_count: retry_count.unwrap_or(0),
            metadata,
            stack_trace: None,
        };

        {
            let mut stats = self.stats.write().await;
            stats.total_failed += 1;
        }

        let json = serde_json::to_value(&failed_msg)
            .map_err(|e| Mecha10Error::SerializationFailed(format!("Failed to serialize DLQ message: {}", e)))?;

        {
            let mut messages = self.messages.write().await;

            if messages.len() >= self.config.max_size {
                warn!(
                    "DLQ '{}' at capacity ({}), dropping oldest message",
                    self.name, self.config.max_size
                );
                messages.remove(0);

                let mut stats = self.stats.write().await;
                stats.total_dropped += 1;
            }

            messages.push(json.clone());
        }

        error!(
            "Message sent to DLQ '{}' from '{}': {} (retry_count: {})",
            self.name,
            source,
            error,
            retry_count.unwrap_or(0)
        );

        if self.config.persist {
            self.persist_message(&json).await?;
        }

        if self.config.publish_to_topic {
            self.publish_to_dlq_topic(failed_msg.clone(), source).await?;
        }

        Ok(())
    }

    /// Get all failed messages
    ///
    /// # Returns
    ///
    /// Vector of failed messages as JSON values
    pub async fn get_all(&self) -> Vec<serde_json::Value> {
        let messages = self.messages.read().await;
        messages.clone()
    }

    /// Get failed messages from a specific source
    pub async fn get_by_source(&self, source: &str) -> Vec<serde_json::Value> {
        let messages = self.messages.read().await;
        messages
            .iter()
            .filter(|msg| {
                msg.get("source")
                    .and_then(|s| s.as_str())
                    .map(|s| s == source)
                    .unwrap_or(false)
            })
            .cloned()
            .collect()
    }

    /// Get the number of failed messages
    pub async fn len(&self) -> usize {
        let messages = self.messages.read().await;
        messages.len()
    }

    /// Check if the DLQ is empty
    pub async fn is_empty(&self) -> bool {
        self.len().await == 0
    }

    /// Clear all failed messages
    ///
    /// Use with caution - messages will be lost unless persisted elsewhere.
    pub async fn clear(&self) {
        let mut messages = self.messages.write().await;
        let count = messages.len();
        messages.clear();
        info!("Cleared {} messages from DLQ '{}'", count, self.name);
    }

    /// Remove and return a failed message for retry
    ///
    /// # Arguments
    ///
    /// * `index` - Index of the message to remove
    ///
    /// # Returns
    ///
    /// The removed message, or None if index is out of bounds
    pub async fn remove(&self, index: usize) -> Option<serde_json::Value> {
        let mut messages = self.messages.write().await;

        if index < messages.len() {
            let msg = messages.remove(index);

            let mut stats = self.stats.write().await;
            stats.total_recovered += 1;

            debug!("Removed message {} from DLQ '{}'", index, self.name);
            Some(msg)
        } else {
            None
        }
    }

    /// Get statistics about the DLQ
    ///
    /// # Returns
    ///
    /// (total_failed, total_recovered, total_dropped, current_count)
    pub async fn stats(&self) -> (u64, u64, u64, usize) {
        let stats = self.stats.read().await;
        let current = self.len().await;
        (stats.total_failed, stats.total_recovered, stats.total_dropped, current)
    }

    /// Persist a message to disk (if configured)
    async fn persist_message(&self, message: &serde_json::Value) -> Result<()> {
        if let Some(path) = &self.config.persist_path {
            let json_str = serde_json::to_string(message)
                .map_err(|e| Mecha10Error::SerializationFailed(format!("Failed to serialize: {}", e)))?;

            tokio::fs::write(path, json_str).await?;

            debug!("Persisted DLQ message to {}", path);
        }

        Ok(())
    }

    /// Publish failed message to DLQ topic
    async fn publish_to_dlq_topic<T: Message + Serialize + Clone>(
        &self,
        _failed_msg: FailedMessage<T>,
        source: &str,
    ) -> Result<()> {
        debug!(
            "Would publish to DLQ topic: {}/{}/{}",
            self.config.topic_prefix, self.name, source
        );
        Ok(())
    }
}

crate::debug_struct_fields!(DeadLetterQueue {
    "name" => name,
    "config" => config,
});

pub trait DeadLetterExt<T> {
    /// Process messages with automatic DLQ handling
    fn process_with_dlq<F, Fut, R>(
        self,
        dlq: &DeadLetterQueue,
        source: &str,
        processor: F,
    ) -> impl std::future::Future<Output = Result<()>>
    where
        F: Fn(T) -> Fut,
        Fut: std::future::Future<Output = Result<R>>;
}