//! Tier 1 metrics: AMQP introspection via passive `queue_declare`.
//!
//! Queries RabbitMQ for queue-level statistics (message count, consumer count)
//! using a passive `queue_declare` — a lightweight, read-only AMQP operation
//! that does not create or modify the queue.
//!
//! The collector caches an AMQP channel and automatically recreates it
//! when the channel becomes disconnected (e.g. after a RabbitMQ restart).

use std::sync::Arc;

use lapin::options::QueueDeclareOptions;
use lapin::types::FieldTable;
use tokio::sync::RwLock;

use crate::channel_factory::ChannelFactory;

/// Metrics obtained from AMQP passive `queue_declare`.
#[derive(Debug, Clone, Copy)]
pub struct Tier1Metrics {
    /// Number of ready messages in the queue.
    pub message_count: u32,
    /// Number of consumers attached to the queue.
    pub consumer_count: u32,
}

/// Collects Tier 1 metrics via an AMQP channel.
///
/// Uses a [`ChannelFactory`] to obtain channels on demand. The channel is
/// cached and reused across collection cycles. If the cached channel becomes
/// disconnected, a new one is created automatically on the next collection.
pub struct Tier1Collector {
    factory: Arc<dyn ChannelFactory>,
    channel: RwLock<Option<lapin::Channel>>,
}

impl Tier1Collector {
    /// Create a new Tier 1 collector using the given channel factory.
    ///
    /// The factory provides channels on demand, enabling automatic recovery
    /// after connection drops.
    pub fn new(factory: Arc<dyn ChannelFactory>) -> Self {
        Self {
            factory,
            channel: RwLock::new(None),
        }
    }

    /// Get a healthy cached channel, creating a new one if needed.
    async fn get_channel(&self) -> Result<lapin::Channel, lapin::Error> {
        // Fast path: cached channel is still connected
        {
            let guard = self.channel.read().await;
            if let Some(ch) = guard.as_ref() {
                if ch.status().connected() {
                    return Ok(ch.clone());
                }
            }
        }

        // Slow path: create new channel
        let mut guard = self.channel.write().await;
        // Double-check after acquiring write lock
        if let Some(ch) = guard.as_ref() {
            if ch.status().connected() {
                return Ok(ch.clone());
            }
        }

        tracing::debug!("Creating new AMQP channel for Tier 1 metrics");
        let new_ch = self.factory.create_channel().await?;
        *guard = Some(new_ch.clone());
        Ok(new_ch)
    }

    /// Collect queue statistics via passive `queue_declare`.
    ///
    /// Returns an error if the queue does not exist (the broker closes the
    /// channel on a passive declare for a missing queue, which lapin
    /// surfaces as an error). The next call will automatically create a
    /// fresh channel.
    pub async fn collect(&self, queue_name: &str) -> Result<Tier1Metrics, lapin::Error> {
        let channel = self.get_channel().await?;

        let queue = channel
            .queue_declare(
                queue_name,
                QueueDeclareOptions {
                    passive: true,
                    ..QueueDeclareOptions::default()
                },
                FieldTable::default(),
            )
            .await?;

        Ok(Tier1Metrics {
            message_count: queue.message_count(),
            consumer_count: queue.consumer_count(),
        })
    }
}
