use rdkafka::{
    ClientConfig, consumer::StreamConsumer, error::KafkaResult, producer::FutureProducer,
};
use serde::Deserialize;
use strum::IntoStaticStr;
use svix_bridge_types::{ReceiverOutput, SenderInput, SenderOutputOpts, TransformationConfig};

use crate::{KafkaProducer, Result, input::KafkaConsumer};

#[derive(Clone, Deserialize)]
#[serde(tag = "type")]
pub enum KafkaInputOpts {
    // Single-variant enum so we can require the "type": "kafka" field in deserialization
    #[serde(rename = "kafka")]
    Inner {
        /// Comma-separated list of addresses.
        ///
        /// Example: `localhost:9094`
        #[serde(rename = "kafka_bootstrap_brokers")]
        bootstrap_brokers: String,

        /// The consumer group ID, used to track the stream offset between restarts
        /// (due to host maintenance, upgrades, crashes, etc.).
        #[serde(rename = "kafka_group_id")]
        group_id: String,

        /// The topic to listen to.
        #[serde(rename = "kafka_topic")]
        topic: String,

        /// The input passed to JSON transformations.
        #[serde(rename = "kafka_transformation_input", default)]
        transformation_input: KafkaTransformationInput,

        /// The value for 'auto.offset.reset' in the kafka config.
        #[serde(rename = "kafka_auto_offset_reset", default)]
        auto_offset_reset: KafkaAutoOffsetReset,

        /// A namespace for idempotency keys from independent Kafka sources.
        #[serde(rename = "kafka_idempotency_namespace")]
        idempotency_namespace: Option<String>,

        /// The value for 'security.protocol' in the kafka config.
        #[serde(flatten)]
        security_protocol: KafkaSecurityProtocol,

        /// The 'debug' config value for rdkafka - enables more verbose logging
        /// for the selected 'contexts'
        #[serde(rename = "kafka_debug_contexts")]
        debug_contexts: Option<String>,
    },
}

#[derive(Clone, Copy, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum KafkaTransformationInput {
    #[default]
    Payload,
    Envelope,
}

impl KafkaInputOpts {
    pub(crate) fn create_consumer(self) -> KafkaResult<StreamConsumer> {
        let Self::Inner {
            bootstrap_brokers,
            group_id,
            auto_offset_reset,
            security_protocol,
            debug_contexts,
            ..
        } = self;

        let mut config = ClientConfig::new();
        config
            .set("group.id", group_id)
            .set("bootstrap.servers", bootstrap_brokers)
            .set("auto.offset.reset", <&'static str>::from(auto_offset_reset))
            // messages are committed manually after webhook delivery was successful.
            .set("enable.auto.commit", "false");

        security_protocol.apply(&mut config);
        if let Some(debug_contexts) = debug_contexts
            && !debug_contexts.is_empty()
        {
            config.set("debug", debug_contexts);
        }

        config.create()
    }
}

#[derive(Clone, Copy, Debug, Default, Deserialize, IntoStaticStr, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum KafkaAutoOffsetReset {
    #[default]
    Latest,
    Earliest,
}

#[derive(Clone, Deserialize)]
#[serde(tag = "type")]
pub enum KafkaOutputOpts {
    // Single-variant enum so we can require the "type": "kafka" field in deserialization
    #[serde(rename = "kafka")]
    Inner {
        /// Comma-separated list of addresses.
        ///
        /// Example: `localhost:9094`
        #[serde(rename = "kafka_bootstrap_brokers")]
        bootstrap_brokers: String,

        /// The topic to listen to.
        #[serde(rename = "kafka_topic")]
        topic: String,

        /// The value for 'security.protocol' in the kafka config.
        #[serde(flatten)]
        security_protocol: KafkaSecurityProtocol,

        /// The 'debug' config value for rdkafka - enables more verbose logging
        /// for the selected 'contexts'
        #[serde(rename = "kafka_debug_contexts")]
        debug_contexts: Option<String>,
    },
}

impl KafkaOutputOpts {
    pub(crate) fn create_producer(self) -> KafkaResult<FutureProducer> {
        let Self::Inner {
            bootstrap_brokers,
            security_protocol,
            debug_contexts,
            ..
        } = self;

        let mut config = ClientConfig::new();
        config.set("bootstrap.servers", bootstrap_brokers);

        security_protocol.apply(&mut config);
        if let Some(debug_contexts) = debug_contexts
            && !debug_contexts.is_empty()
        {
            config.set("debug", debug_contexts);
        }

        config.create()
    }
}

#[derive(Clone, Debug, Deserialize)]
pub enum KafkaSaslMechanism {
    #[serde(rename = "PLAIN")]
    Plain,
    #[serde(rename = "SCRAM-SHA-256")]
    ScramSha256,
    #[serde(rename = "SCRAM-SHA-512")]
    ScramSha512,
}

impl KafkaSaslMechanism {
    fn as_str(&self) -> &'static str {
        match self {
            KafkaSaslMechanism::Plain => "PLAIN",
            KafkaSaslMechanism::ScramSha256 => "SCRAM-SHA-256",
            KafkaSaslMechanism::ScramSha512 => "SCRAM-SHA-512",
        }
    }
}

#[derive(Clone, Debug, Deserialize)]
#[serde(tag = "kafka_security_protocol", rename_all = "snake_case")]
pub enum KafkaSecurityProtocol {
    Plaintext,
    Ssl,
    SaslSsl {
        #[serde(rename = "kafka_sasl_username")]
        sasl_username: String,
        #[serde(rename = "kafka_sasl_password")]
        sasl_password: String,
        /// The SASL mechanism to use. Defaults to `SCRAM-SHA-512`.
        #[serde(rename = "kafka_sasl_mechanism", default = "default_sasl_mechanism")]
        sasl_mechanism: KafkaSaslMechanism,
    },
}

fn default_sasl_mechanism() -> KafkaSaslMechanism {
    KafkaSaslMechanism::ScramSha512
}

impl KafkaSecurityProtocol {
    fn apply(self, config: &mut ClientConfig) {
        match self {
            KafkaSecurityProtocol::Plaintext => {
                config.set("security.protocol", "plaintext");
            }
            KafkaSecurityProtocol::Ssl => {
                config.set("security.protocol", "ssl");
            }
            KafkaSecurityProtocol::SaslSsl {
                sasl_username,
                sasl_password,
                sasl_mechanism,
            } => {
                config
                    .set("security.protocol", "sasl_ssl")
                    .set("sasl.mechanisms", sasl_mechanism.as_str())
                    .set("sasl.username", sasl_username)
                    .set("sasl.password", sasl_password);
            }
        }
    }
}

pub fn into_sender_input(
    name: String,
    opts: KafkaInputOpts,
    transformation: Option<TransformationConfig>,
    output: SenderOutputOpts,
) -> Result<Box<dyn SenderInput>> {
    Ok(Box::new(KafkaConsumer::new(
        name,
        opts,
        transformation,
        output,
    )?))
}

pub fn into_receiver_output(
    name: String,
    opts: KafkaOutputOpts,
) -> Result<Box<dyn ReceiverOutput>> {
    Ok(Box::new(KafkaProducer::new(name, opts)?))
}
