pub mod config;
pub mod consumer;
pub mod error;
pub mod producer;
pub mod topic;
pub mod typed;

use std::sync::Arc;

pub use config::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 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),
            }),
        }))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn disabled_returns_none() {
        let cfg = AppQueueSettings {
            enabled: false,
            ..Default::default()
        };
        let queue = QueueBuilder::from_config(cfg).build().await.unwrap();
        assert!(queue.is_none());
    }

    #[test]
    fn invalid_config_rejected() {
        let cfg = AppQueueSettings {
            enabled: true,
            brokers: "".into(),
            ..Default::default()
        };
        assert!(cfg.is_valid().is_err());
    }

    #[test]
    fn invalid_offset_reset_rejected() {
        let cfg = AppQueueSettings {
            enabled: true,
            auto_offset_reset: "middle".into(),
            ..Default::default()
        };
        assert!(cfg.is_valid().is_err());
    }

    #[test]
    fn invalid_compression_rejected() {
        let cfg = AppQueueSettings {
            enabled: true,
            producer_compression: "rar".into(),
            ..Default::default()
        };
        assert!(cfg.is_valid().is_err());
    }

    #[test]
    fn invalid_acks_rejected() {
        let cfg = AppQueueSettings {
            enabled: true,
            producer_acks: "2".into(),
            ..Default::default()
        };
        assert!(cfg.is_valid().is_err());
    }

    #[test]
    fn subscribe_requires_topic() {
        let cfg = AppQueueSettings::default();
        let publisher = KafkaPublisher::from_config(&cfg).unwrap();
        let queue = Queue {
            inner: Arc::new(QueueInner {
                publisher,
                cfg: Arc::new(cfg),
            }),
        };
        let err = queue
            .subscribe_raw::<_, std::future::Ready<Result<(), QueueError>>>(
                SubscriptionMode::Group {
                    group_id: "g".into(),
                    auto_offset_reset: AutoOffsetReset::Earliest,
                },
                &[],
                |_| std::future::ready(Ok(())),
            )
            .unwrap_err();
        assert!(matches!(err, QueueError::Topic(_)));
    }
}