poimen-memory/crates/mem-cli/src/queue_worker_dlq.rs

/// Dead Letter Queue handler using gateway queue adapter (kmsvc).
///
/// Extracts that fail contradiction detection or entity validation
/// are sent to the DLQ topic for async reprocessing or analysis.

use serde::{Deserialize, Serialize};
use chrono::{DateTime, Utc};
use std::collections::HashMap;

/// DLQ message sent to kmsvc
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DlqMessage {
    pub id: String,
    pub original_content: String,
    pub extraction_type: String,  // "entity" | "edge"
    pub error_type: String,       // "contradiction_high" | "extraction_failed" | "validation_failed"
    pub error_details: String,
    pub retry_count: i32,
    pub max_retries: i32,
    pub created_at: DateTime<Utc>,
}

impl DlqMessage {
    pub fn new(
        original_content: String,
        extraction_type: &str,
        error_type: &str,
        error_details: String,
    ) -> Self {
        Self {
            id: uuid::Uuid::new_v4().to_string(),
            original_content,
            extraction_type: extraction_type.to_string(),
            error_type: error_type.to_string(),
            error_details,
            retry_count: 0,
            max_retries: 3,
            created_at: Utc::now(),
        }
    }
}

/// DLQ handler for gateway queue adapter
pub struct DlqHandler {
    // Uses GatewayQueueAdapter under the hood (injected at AppState level)
    // This struct just defines the message format and retry logic
}

impl DlqHandler {
    /// Build message for kmsvc DLQ topic
    pub fn format_for_queue(msg: &DlqMessage) -> serde_json::Value {
        serde_json::json!({
            "id": msg.id,
            "original_content": msg.original_content,
            "extraction_type": msg.extraction_type,
            "error_type": msg.error_type,
            "error_details": msg.error_details,
            "retry_count": msg.retry_count,
            "max_retries": msg.max_retries,
            "created_at": msg.created_at.to_rfc3339(),
        })
    }
    
    /// Build queue attributes for kmsvc
    pub fn queue_attributes(msg: &DlqMessage) -> HashMap<String, String> {
        let mut attrs = HashMap::new();
        attrs.insert("extraction_type".to_string(), msg.extraction_type.clone());
        attrs.insert("error_type".to_string(), msg.error_type.clone());
        attrs.insert("retry_count".to_string(), msg.retry_count.to_string());
        attrs.insert("max_retries".to_string(), msg.max_retries.to_string());
        attrs.insert("created_at".to_string(), msg.created_at.to_rfc3339());
        attrs
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_dlq_message_creation() {
        let msg = DlqMessage::new(
            "test content".to_string(),
            "entity",
            "extraction_failed",
            "LLM timeout".to_string(),
        );
        
        assert_eq!(msg.extraction_type, "entity");
        assert_eq!(msg.error_type, "extraction_failed");
        assert_eq!(msg.retry_count, 0);
        assert_eq!(msg.max_retries, 3);
    }
    
    #[test]
    fn test_dlq_message_format() {
        let msg = DlqMessage::new(
            "test content".to_string(),
            "edge",
            "contradiction_high",
            "confidence < 0.7".to_string(),
        );
        
        let formatted = DlqHandler::format_for_queue(&msg);
        assert_eq!(formatted["extraction_type"], "edge");
        assert_eq!(formatted["error_type"], "contradiction_high");
    }
    
    #[test]
    fn test_queue_attributes() {
        let msg = DlqMessage::new(
            "test".to_string(),
            "entity",
            "validation_failed",
            "missing name field".to_string(),
        );
        
        let attrs = DlqHandler::queue_attributes(&msg);
        assert_eq!(attrs.get("extraction_type"), Some(&"entity".to_string()));
        assert_eq!(attrs.get("retry_count"), Some(&"0".to_string()));
    }
}