use std::time::Duration;

use config::queue::AppQueueSettings;
use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
use rdkafka::client::DefaultClientContext;
use rdkafka::config::ClientConfig;
use rdkafka::error::{KafkaError, RDKafkaErrorCode};

use crate::error::QueueError;

pub async fn ensure_topics(cfg: &AppQueueSettings, topics: &[String]) -> Result<(), QueueError> {
    if topics.is_empty() {
        return Ok(());
    }

    let mut client_cfg = ClientConfig::new();
    client_cfg.set("bootstrap.servers", &cfg.brokers);
    client_cfg.set("client.id", &cfg.client_id);
    for (k, v) in &cfg.additional_properties {
        client_cfg.set(k, v);
    }
    let admin: AdminClient<DefaultClientContext> =
        client_cfg.create().map_err(QueueError::Kafka)?;

    let new_topics: Vec<NewTopic<'_>> = topics
        .iter()
        .map(|name| NewTopic::new(name.as_str(), 1, TopicReplication::Fixed(1)))
        .collect();

    let opts = AdminOptions::new().operation_timeout(Some(Duration::from_secs(10)));
    let results = admin
        .create_topics(new_topics.iter(), &opts)
        .await
        .map_err(QueueError::Kafka)?;

    for result in results {
        if let Err((_, code)) = result {
            if code != RDKafkaErrorCode::TopicAlreadyExists {
                return Err(QueueError::Kafka(KafkaError::AdminOp(code)));
            }
        }
    }
    Ok(())
}