pub mod admin;
pub mod consumer;
pub mod error;
pub mod metrics;
pub mod producer;
pub mod topic;
pub mod typed;

use std::sync::Arc;

use config::queue::AppQueueSettings;

pub use consumer::{AutoOffsetReset, RawMessage, SubscribeHandle, SubscriptionMode};
pub use error::QueueError;
pub use producer::KafkaPublisher;
pub use typed::{TypedMessage, TypedMessageMeta, TypedQueue};

#[derive(Clone)]
pub struct Queue {
    inner: Arc<QueueInner>,
}

struct QueueInner {
    publisher: KafkaPublisher,
    cfg: Arc<AppQueueSettings>,
}

impl Queue {
    pub fn publisher(&self) -> &KafkaPublisher {
        &self.inner.publisher
    }

    pub fn config(&self) -> &AppQueueSettings {
        &self.inner.cfg
    }

    pub(crate) fn config_arc(&self) -> Arc<AppQueueSettings> {
        self.inner.cfg.clone()
    }

    pub(crate) fn topic_prefix(&self) -> String {
        self.inner.cfg.topic_prefix.clone()
    }

    pub async fn publish_raw(
        &self,
        topic: &str,
        key: Option<&[u8]>,
        payload: &[u8],
    ) -> Result<(), QueueError> {
        self.inner.publisher.publish(topic, key, payload).await
    }

    pub fn subscribe_raw<F, Fut>(
        &self,
        mode: SubscriptionMode,
        topics: &[&str],
        handler: F,
    ) -> Result<SubscribeHandle, QueueError>
    where
        F: Fn(RawMessage) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<(), QueueError>> + Send + 'static,
    {
        if topics.is_empty() {
            return Err(QueueError::Topic(
                "subscribe requires at least one topic".to_string(),
            ));
        }
        let cfg = self.config_arc();
        let (consumer, group_id) = consumer::build_consumer(&cfg, &mode, topics)?;
        let handler: consumer::RawHandler = Arc::new(move |raw| Box::pin(handler(raw)));
        Ok(consumer::spawn_raw_loop(cfg, consumer, group_id, handler))
    }

    pub fn typed<T>(&self) -> TypedQueue<'_, T>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Send + Sync + 'static,
    {
        TypedQueue::new(self)
    }

    pub async fn ensure_topics(&self, topics: &[String]) -> Result<(), QueueError> {
        admin::ensure_topics(&self.inner.cfg, topics).await
    }

    pub async fn ensure_typed_topic<T>(&self) -> Result<(), QueueError>
    where
        T: 'static,
    {
        let topic = topic::topic_for_type::<T>(&self.topic_prefix());
        self.ensure_topics(&[topic]).await
    }

    pub async fn shutdown(&self) -> Result<(), QueueError> {
        Ok(())
    }
}

pub struct QueueBuilder {
    cfg: AppQueueSettings,
}

impl QueueBuilder {
    pub fn from_config(cfg: AppQueueSettings) -> Self {
        Self { cfg }
    }

    pub async fn build(self) -> Result<Option<Queue>, QueueError> {
        if !self.cfg.enabled {
            return Ok(None);
        }
        self.cfg.is_valid().map_err(QueueError::Config)?;
        let publisher = KafkaPublisher::from_config(&self.cfg)?;
        Ok(Some(Queue {
            inner: Arc::new(QueueInner {
                publisher,
                cfg: Arc::new(self.cfg),
            }),
        }))
    }
}