// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::plugin::PluginEvent;
use crate::{
    StoreError, Target,
    arn::TargetID,
    error::TargetError,
    runtime::tls::{
        ReloadableTargetTls, TargetTlsInputSet, TlsReloadAdapter, config::ReloadApplyMode, fingerprint::TargetTlsGeneration,
        validate::validate_tls_material,
    },
    store::{Key, Store},
    target::{
        ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
        TargetTlsState, TargetType, build_queued_payload, build_target_tls_fingerprint, is_connectivity_error,
        open_target_queue_store, persist_queued_payload_to_store,
    },
};
use async_trait::async_trait;
use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError, KafkaCode};
use rustfs_kafka_async::{AsyncProducer, AsyncProducerConfig, Record, RequiredAcks, SaslConfig, SecurityConfig};
use rustfs_tls_runtime::{load_cert_bundle_der_bytes, load_private_key};
use std::sync::atomic::{AtomicBool, Ordering};
use std::{fmt, future::Future, marker::PhantomData, sync::Arc, time::Duration};
use tokio::sync::Mutex;
use tracing::{debug, error, info, instrument, warn};

pub(crate) const KAFKA_SASL_PLAIN: &str = "PLAIN";
pub(crate) const KAFKA_SASL_SCRAM_SHA_256: &str = "SCRAM-SHA-256";
pub(crate) const KAFKA_SASL_SCRAM_SHA_512: &str = "SCRAM-SHA-512";
const KAFKA_DELIVERY_TIMEOUT: Duration = Duration::from_secs(30);

struct KafkaDeliveryAttempt<'a> {
    armed: bool,
    poisoned: &'a AtomicBool,
}

impl KafkaDeliveryAttempt<'_> {
    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for KafkaDeliveryAttempt<'_> {
    fn drop(&mut self) {
        if self.armed {
            self.poisoned.store(true, Ordering::Release);
        }
    }
}

fn kafka_delivery_timeout() -> TargetError {
    TargetError::Timeout(format!("Kafka delivery timed out after {KAFKA_DELIVERY_TIMEOUT:?}"))
}

async fn with_serialized_kafka_delivery<P, T, Select, SelectFuture, Deliver, DeliveryFuture, Invalidate, InvalidateFuture>(
    delivery_lock: &Mutex<()>,
    delivery_poisoned: &AtomicBool,
    select_producer: Select,
    deliver: Deliver,
    invalidate: Invalidate,
) -> Result<T, TargetError>
where
    P: Send,
    T: Send,
    Select: FnOnce() -> SelectFuture + Send,
    SelectFuture: Future<Output = Result<P, TargetError>> + Send,
    Deliver: FnOnce(P) -> DeliveryFuture + Send,
    DeliveryFuture: Future<Output = Result<T, TargetError>> + Send,
    Invalidate: Fn() -> InvalidateFuture + Send,
    InvalidateFuture: Future<Output = ()> + Send,
{
    let deadline = tokio::time::Instant::now() + KAFKA_DELIVERY_TIMEOUT;
    let _delivery_guard = tokio::time::timeout_at(deadline, delivery_lock.lock())
        .await
        .map_err(|_| kafka_delivery_timeout())?;
    let mut attempt = KafkaDeliveryAttempt {
        armed: true,
        poisoned: delivery_poisoned,
    };

    if delivery_poisoned.load(Ordering::Acquire) {
        tokio::time::timeout_at(deadline, invalidate())
            .await
            .map_err(|_| kafka_delivery_timeout())?;
        delivery_poisoned.store(false, Ordering::Release);
    }

    let result = tokio::time::timeout_at(deadline, async { deliver(select_producer().await?).await })
        .await
        .map_err(|_| kafka_delivery_timeout())?;
    if result.as_ref().is_err_and(is_connectivity_error) {
        tokio::time::timeout_at(deadline, invalidate())
            .await
            .map_err(|_| kafka_delivery_timeout())?;
        delivery_poisoned.store(false, Ordering::Release);
    }
    attempt.disarm();
    result
}

/// Arguments for configuring a Kafka target
#[derive(Clone)]
pub struct KafkaArgs {
    /// Whether the target is enabled
    pub enable: bool,
    /// Comma-separated list of broker addresses (e.g. "localhost:9092,broker2:9092")
    pub brokers: Vec<String>,
    /// The topic to publish events to
    pub topic: String,
    /// Required acks: 0 = none, 1 = leader, -1 = all
    pub acks: i16,
    /// Whether to enable TLS for Kafka transport
    pub tls_enable: bool,
    /// Optional path to CA cert used for broker verification
    pub tls_ca: String,
    /// Optional path to client certificate for mTLS
    pub tls_client_cert: String,
    /// Optional path to client private key for mTLS
    pub tls_client_key: String,
    /// Whether to enable SASL authentication over the TLS transport
    pub sasl_enable: bool,
    /// SASL mechanism (PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512)
    pub sasl_mechanism: String,
    /// SASL username
    pub sasl_username: String,
    /// SASL password
    pub sasl_password: String,
    /// The directory to store events in case of failure
    pub queue_dir: String,
    /// The maximum number of events to store
    pub queue_limit: u64,
    /// The target type (audit or notify)
    pub target_type: TargetType,
}

impl fmt::Debug for KafkaArgs {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("KafkaArgs")
            .field("enable", &self.enable)
            .field("brokers", &self.brokers)
            .field("topic", &self.topic)
            .field("acks", &self.acks)
            .field("tls_enable", &self.tls_enable)
            .field("tls_ca", &self.tls_ca)
            .field("tls_client_cert", &self.tls_client_cert)
            .field(
                "tls_client_key",
                if self.tls_client_key.is_empty() {
                    &""
                } else {
                    &"***REDACTED***"
                },
            )
            .field("sasl_enable", &self.sasl_enable)
            .field("sasl_mechanism", &self.sasl_mechanism)
            .field("sasl_username", &self.sasl_username)
            .field(
                "sasl_password",
                if self.sasl_password.is_empty() {
                    &""
                } else {
                    &"***REDACTED***"
                },
            )
            .field("queue_dir", &self.queue_dir)
            .field("queue_limit", &self.queue_limit)
            .field("target_type", &self.target_type)
            .finish()
    }
}

fn normalize_kafka_sasl_mechanism(mechanism: &str) -> Result<&'static str, TargetError> {
    let mechanism = mechanism.trim();
    if mechanism.is_empty() || mechanism.eq_ignore_ascii_case(KAFKA_SASL_PLAIN) {
        return Ok(KAFKA_SASL_PLAIN);
    }
    if mechanism.eq_ignore_ascii_case(KAFKA_SASL_SCRAM_SHA_256) {
        return Ok(KAFKA_SASL_SCRAM_SHA_256);
    }
    if mechanism.eq_ignore_ascii_case(KAFKA_SASL_SCRAM_SHA_512) {
        return Ok(KAFKA_SASL_SCRAM_SHA_512);
    }
    Err(TargetError::Configuration(
        "kafka sasl_mechanism must be one of: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512".to_string(),
    ))
}

impl KafkaArgs {
    /// Validates the KafkaArgs configuration
    pub fn validate(&self) -> Result<(), TargetError> {
        if !self.enable {
            return Ok(());
        }

        if self.brokers.is_empty() {
            return Err(TargetError::Configuration("kafka brokers cannot be empty".to_string()));
        }

        if self.topic.is_empty() {
            return Err(TargetError::Configuration("kafka topic cannot be empty".to_string()));
        }

        if !matches!(self.acks, -1..=1) {
            return Err(TargetError::Configuration("kafka acks must be one of: 0, 1, -1".to_string()));
        }

        if self.tls_client_cert.is_empty() != self.tls_client_key.is_empty() {
            return Err(TargetError::Configuration(
                "kafka tls_client_cert and tls_client_key must be specified together".to_string(),
            ));
        }

        if self.sasl_enable {
            if !self.tls_enable {
                return Err(TargetError::Configuration(
                    "kafka sasl_enable requires tls_enable for SASL_SSL".to_string(),
                ));
            }
            normalize_kafka_sasl_mechanism(&self.sasl_mechanism)?;
            if self.sasl_username.is_empty() || self.sasl_password.is_empty() {
                return Err(TargetError::Configuration(
                    "kafka sasl_username and sasl_password must be specified when sasl_enable is true".to_string(),
                ));
            }
        } else if !self.sasl_mechanism.is_empty() || !self.sasl_username.is_empty() || !self.sasl_password.is_empty() {
            return Err(TargetError::Configuration(
                "kafka sasl_enable must be true when SASL fields are specified".to_string(),
            ));
        }

        if !self.queue_dir.is_empty() {
            let path = std::path::Path::new(&self.queue_dir);
            if !path.is_absolute() {
                return Err(TargetError::Configuration("kafka queueDir path should be absolute".to_string()));
            }
        }

        Ok(())
    }

    pub(crate) fn security_config(&self, validate_tls_files: bool) -> Result<Option<SecurityConfig>, TargetError> {
        if !self.tls_enable && !self.sasl_enable {
            return Ok(None);
        }

        let mut security = SecurityConfig::new();
        if !self.tls_ca.is_empty() {
            if validate_tls_files {
                let certs = load_cert_bundle_der_bytes(&self.tls_ca)
                    .map_err(|e| TargetError::Configuration(format!("Failed to parse Kafka tls_ca: {e}")))?;
                if certs.is_empty() {
                    return Err(TargetError::Configuration(
                        "Kafka tls_ca did not contain any parsable certificates".to_string(),
                    ));
                }
            }
            security = security.with_ca_cert(self.tls_ca.clone());
        }
        if !self.tls_client_cert.is_empty() && !self.tls_client_key.is_empty() {
            if validate_tls_files {
                let certs = load_cert_bundle_der_bytes(&self.tls_client_cert)
                    .map_err(|e| TargetError::Configuration(format!("Failed to parse Kafka tls_client_cert: {e}")))?;
                if certs.is_empty() {
                    return Err(TargetError::Configuration(
                        "Kafka tls_client_cert did not contain any parsable certificates".to_string(),
                    ));
                }
                let _ = load_private_key(&self.tls_client_key)
                    .map_err(|e| TargetError::Configuration(format!("Failed to parse Kafka tls_client_key: {e}")))?;
            }
            security = security.with_client_cert(self.tls_client_cert.clone(), self.tls_client_key.clone());
        }
        if self.sasl_enable {
            security = security.with_sasl(SaslConfig::new(
                normalize_kafka_sasl_mechanism(&self.sasl_mechanism)?.to_string(),
                self.sasl_username.clone(),
                self.sasl_password.clone(),
            ));
        }

        Ok(Some(security))
    }
}

/// A target that sends events to an Apache Kafka topic
pub struct KafkaTarget<E>
where
    E: PluginEvent,
{
    id: TargetID,
    args: KafkaArgs,
    store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
    producer: Arc<Mutex<Option<Arc<AsyncProducer>>>>,
    delivery_lock: Arc<Mutex<()>>,
    delivery_poisoned: Arc<AtomicBool>,
    tls_state: Arc<Mutex<TargetTlsState>>,
    /// Adapter that bridges this target to the TLS reload coordinator.
    /// When `Some`, the target uses coordinator-managed material; when `None`,
    /// it falls back to inline fingerprint-based change detection.
    tls_adapter: Option<TlsReloadAdapter<Arc<AsyncProducer>>>,
    delivery_counters: Arc<TargetDeliveryCounters>,
    _phantom: PhantomData<E>,
}

impl<E> KafkaTarget<E>
where
    E: PluginEvent,
{
    fn map_kafka_error(err: KafkaError, context: &str) -> TargetError {
        // Prefer the client's own retriable classification so transient broker
        // states (leader election / NotLeaderForPartition, coordinator load,
        // network blips, RequestTimedOut) are retried via store replay instead of
        // being dropped as permanent failures (backlog#973).
        if err.is_retriable() {
            return match &err {
                KafkaError::Connection(ConnectionError::Timeout(_)) | KafkaError::Kafka(KafkaCode::RequestTimedOut) => {
                    TargetError::Timeout(format!("{context}: {err}"))
                }
                _ => TargetError::NotConnected,
            };
        }

        // Non-retriable errors: configuration problems are permanent config
        // errors; everything else (e.g. UnknownTopicOrPartition, authorization
        // failures, oversize messages) is a permanent request-level failure.
        match &err {
            KafkaError::Config(_) => TargetError::Configuration(format!("{context}: {err}")),
            _ => TargetError::Request(format!("{context}: {err}")),
        }
    }

    /// Creates a new KafkaTarget
    #[instrument(skip(args), fields(target_id = %id))]
    pub fn new(id: String, args: KafkaArgs) -> Result<Self, TargetError> {
        args.validate()?;

        let target_id = TargetID::new(id, ChannelTargetType::Kafka.as_str().to_string());

        let queue_store = open_target_queue_store(
            &args.queue_dir,
            args.queue_limit,
            args.target_type,
            ChannelTargetType::Kafka.as_str(),
            &target_id,
            "Failed to open store for Kafka target",
        )?;

        info!(target_id = %target_id.id, "Kafka target created");
        Ok(KafkaTarget {
            id: target_id,
            args,
            store: queue_store,
            producer: Arc::new(Mutex::new(None)),
            delivery_lock: Arc::new(Mutex::new(())),
            delivery_poisoned: Arc::new(AtomicBool::new(false)),
            tls_state: Arc::new(Mutex::new(TargetTlsState::default())),
            tls_adapter: None,
            delivery_counters: Arc::new(TargetDeliveryCounters::default()),
            _phantom: PhantomData,
        })
    }

    /// Builds a Kafka producer from the current args
    async fn build_producer(&self) -> Result<AsyncProducer, TargetError> {
        let acks = match self.args.acks {
            0 => RequiredAcks::None,
            1 => RequiredAcks::One,
            _ => RequiredAcks::All,
        };

        let mut config = AsyncProducerConfig::new()
            .with_ack_timeout(KAFKA_DELIVERY_TIMEOUT)
            .with_required_acks(acks);

        if let Some(security) = self.args.security_config(true)? {
            config = config.with_security(security);
        }

        AsyncProducer::from_hosts_with_config(self.args.brokers.clone(), config)
            .await
            .map_err(|e| Self::map_kafka_error(e, "Failed to create Kafka producer"))
    }

    async fn get_or_build_producer(&self) -> Result<Arc<AsyncProducer>, TargetError> {
        // Adapter-managed path: use the material directly from the TLS reload adapter.
        if let Some(adapter) = &self.tls_adapter {
            let producer: Arc<AsyncProducer> = (*adapter.current_material()).clone();

            // Ensure the producer is also stored locally so that close() can drain it.
            {
                let mut guard = self.producer.lock().await;
                *guard = Some(Arc::clone(&producer));
            }
            return Ok(producer);
        }

        // Inline fingerprint fallback path (no coordinator).
        let next_fingerprint =
            build_target_tls_fingerprint(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key).await?;
        let tls_changed = {
            let tls_state_guard = self.tls_state.lock().await;
            tls_state_guard.needs_update(&next_fingerprint)
        };
        if tls_changed {
            let mut cached = self.producer.lock().await;
            *cached = None;
            self.tls_state.lock().await.refresh(next_fingerprint);
        }

        {
            let cached = self.producer.lock().await;
            if let Some(producer) = cached.as_ref() {
                return Ok(Arc::clone(producer));
            }
        }

        // Build the producer without holding the cache lock so a slow connect
        // does not block other senders (which only need to read the cache).
        // Re-check the cache after building in case another task raced us
        // (backlog#983).
        let producer = Arc::new(self.build_producer().await?);
        let mut cached = self.producer.lock().await;
        if let Some(existing) = cached.as_ref() {
            return Ok(Arc::clone(existing));
        }
        *cached = Some(Arc::clone(&producer));
        Ok(producer)
    }

    async fn invalidate_cached_producer(&self) {
        let mut cached = self.producer.lock().await;
        *cached = None;
        self.tls_state.lock().await.reset();
    }

    /// Serializes the event and builds a QueuedPayload
    fn build_queued_payload(&self, event: &EntityTarget<E>) -> Result<QueuedPayload, TargetError> {
        build_queued_payload(event)
    }

    /// Sends the raw body to Kafka
    #[instrument(skip(self, body, meta), fields(target_id = %self.id))]
    async fn send_body(&self, body: Vec<u8>, meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
        debug!(
            target = %self.id,
            bucket = %meta.bucket_name,
            object = %meta.object_name,
            event = %meta.event_name,
            payload_len = body.len(),
            "Sending Kafka payload"
        );

        // rustfs-kafka-async does not validate response correlation IDs. Keep
        // producer selection, send, and timeout invalidation serialized so a
        // waiter cannot reuse a connection with an unread timed-out response.
        with_serialized_kafka_delivery(
            &self.delivery_lock,
            &self.delivery_poisoned,
            || self.get_or_build_producer(),
            |producer| async move {
                // Use "<bucket>/<object>" as the message key so all events for the same
                // object hash to the same partition and preserve per-object ordering
                // across multiple partitions (backlog#983).
                let partition_key = format!("{}/{}", meta.bucket_name, meta.object_name);
                producer
                    .send(&Record::from_key_value(&self.args.topic, partition_key, body.as_slice()))
                    .await
                    .map_err(|err| Self::map_kafka_error(err, "Failed to send message to Kafka"))
            },
            || self.invalidate_cached_producer(),
        )
        .await?;

        debug!(target_id = %self.id, topic = %self.args.topic, "Event published to Kafka topic");
        self.delivery_counters.record_success();
        Ok(())
    }

    /// Clones this target into a boxed trait object
    pub fn clone_box(&self) -> Box<dyn Target<E> + Send + Sync> {
        Box::new(KafkaTarget::<E> {
            id: self.id.clone(),
            args: self.args.clone(),
            store: self.store.as_ref().map(|s| s.boxed_clone()),
            producer: Arc::clone(&self.producer),
            delivery_lock: Arc::clone(&self.delivery_lock),
            delivery_poisoned: Arc::clone(&self.delivery_poisoned),
            tls_state: Arc::clone(&self.tls_state),
            tls_adapter: self.tls_adapter.clone(),
            delivery_counters: Arc::clone(&self.delivery_counters),
            _phantom: PhantomData,
        })
    }
}

#[async_trait]
impl<E> Target<E> for KafkaTarget<E>
where
    E: PluginEvent,
{
    fn id(&self) -> TargetID {
        self.id.clone()
    }

    async fn is_active(&self) -> Result<bool, TargetError> {
        let _ = self.get_or_build_producer().await?;
        Ok(true)
    }

    async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
        let queued = match self.build_queued_payload(&event) {
            Ok(queued) => queued,
            Err(err) => {
                self.delivery_counters.record_final_failure();
                return Err(err);
            }
        };

        if let Some(store) = &self.store {
            if let Err(e) = persist_queued_payload_to_store(store.as_ref(), &queued) {
                self.delivery_counters.record_final_failure();
                return Err(e);
            }
            debug!("Event saved to store for Kafka target: {}", self.id);
            Ok(())
        } else {
            if let Err(err) = self.send_body(queued.body, &queued.meta).await {
                self.delivery_counters.record_final_failure();
                return Err(err);
            }
            Ok(())
        }
    }

    async fn send_raw_from_store(&self, key: Key, body: Vec<u8>, meta: QueuedPayloadMeta) -> Result<(), TargetError> {
        debug!("Sending queued payload from store for Kafka target: {}, key: {}", self.id, key);

        if let Err(e) = self.send_body(body, &meta).await {
            if matches!(e, TargetError::NotConnected) {
                warn!(target_id = %self.id, "Kafka not reachable, event remains in store.");
                return Err(TargetError::NotConnected);
            }
            error!(target_id = %self.id, error = %e, "Failed to send event from store.");
            return Err(e);
        }

        debug!("Event sent from store for Kafka target: {}", self.id);
        Ok(())
    }

    async fn close(&self) -> Result<(), TargetError> {
        {
            let mut guard = self.producer.lock().await;
            *guard = None;
        }

        self.tls_state.lock().await.reset();

        info!("Kafka target closed: {}", self.id);
        Ok(())
    }

    fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
        self.store.as_deref()
    }

    fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
        self.clone_box()
    }

    fn is_enabled(&self) -> bool {
        self.args.enable
    }

    fn delivery_snapshot(&self) -> TargetDeliverySnapshot {
        self.delivery_counters.snapshot(
            self.store.as_deref().map_or(0, |store| store.len() as u64),
            // Kafka targets record no terminal failures and keep no failed store.
            0,
        )
    }

    fn record_final_failure(&self) {
        self.delivery_counters.record_final_failure();
    }
}

/// Coordinated TLS hot-reload implementation for Kafka targets.
///
/// The coordinator calls these methods on a background poll loop to detect
/// TLS file changes and rebuild the producer without restarting.
#[async_trait]
impl<E> ReloadableTargetTls for KafkaTarget<E>
where
    E: PluginEvent,
{
    type Material = Arc<AsyncProducer>;

    fn tls_input_set(&self) -> TargetTlsInputSet {
        TargetTlsInputSet {
            ca_path: self.args.tls_ca.clone(),
            client_cert_path: self.args.tls_client_cert.clone(),
            client_key_path: self.args.tls_client_key.clone(),
            target_label: format!("kafka:{}", self.id.id),
        }
    }

    async fn build_tls_material(&self) -> Result<Self::Material, TargetError> {
        let producer = self.build_producer().await?;
        Ok(Arc::new(producer))
    }

    async fn apply_tls_material(
        &self,
        _generation: TargetTlsGeneration,
        material: Arc<Self::Material>,
        _mode: ReloadApplyMode,
    ) -> Result<(), TargetError> {
        let mut guard = self.producer.lock().await;
        *guard = Some((*material).clone());
        Ok(())
    }

    async fn validate_tls_files(&self) -> Result<(), TargetError> {
        validate_tls_material(&self.args.tls_ca, &self.args.tls_client_cert, &self.args.tls_client_key)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicUsize;
    use tokio::sync::Notify;

    fn base_args() -> KafkaArgs {
        KafkaArgs {
            enable: true,
            brokers: vec!["localhost:9092".to_string()],
            topic: "rustfs-events".to_string(),
            acks: 1,
            tls_enable: false,
            tls_ca: String::new(),
            tls_client_cert: String::new(),
            tls_client_key: String::new(),
            sasl_enable: false,
            sasl_mechanism: String::new(),
            sasl_username: String::new(),
            sasl_password: String::new(),
            queue_dir: String::new(),
            queue_limit: 0,
            target_type: TargetType::NotifyEvent,
        }
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_invalidates_before_the_next_delivery_selects_a_producer() {
        let delivery_lock = Arc::new(Mutex::new(()));
        let delivery_poisoned = Arc::new(AtomicBool::new(false));
        let generation = Arc::new(AtomicUsize::new(1));
        let first_entered = Arc::new(Notify::new());

        let first = {
            let delivery_lock = Arc::clone(&delivery_lock);
            let delivery_poisoned = Arc::clone(&delivery_poisoned);
            let generation = Arc::clone(&generation);
            let first_entered = Arc::clone(&first_entered);
            tokio::spawn(async move {
                with_serialized_kafka_delivery(
                    &delivery_lock,
                    &delivery_poisoned,
                    {
                        let generation = Arc::clone(&generation);
                        move || async move { Ok(generation.load(Ordering::SeqCst)) }
                    },
                    move |selected| async move {
                        assert_eq!(selected, 1);
                        first_entered.notify_one();
                        std::future::pending::<Result<usize, TargetError>>().await
                    },
                    move || {
                        let generation = Arc::clone(&generation);
                        async move { generation.store(2, Ordering::SeqCst) }
                    },
                )
                .await
            })
        };

        first_entered.notified().await;
        tokio::time::advance(Duration::from_secs(1)).await;
        let second = {
            let delivery_lock = Arc::clone(&delivery_lock);
            let delivery_poisoned = Arc::clone(&delivery_poisoned);
            let generation = Arc::clone(&generation);
            tokio::spawn(async move {
                with_serialized_kafka_delivery(
                    &delivery_lock,
                    &delivery_poisoned,
                    {
                        let generation = Arc::clone(&generation);
                        move || async move { Ok(generation.load(Ordering::SeqCst)) }
                    },
                    |selected| async move { Ok(selected) },
                    move || {
                        let generation = Arc::clone(&generation);
                        async move { generation.store(2, Ordering::SeqCst) }
                    },
                )
                .await
            })
        };

        assert!(matches!(
            first.await.expect("first delivery task should not panic"),
            Err(TargetError::Timeout(_))
        ));
        assert_eq!(
            second
                .await
                .expect("second delivery task should not panic")
                .expect("second delivery should succeed"),
            2,
            "the waiter must select a fresh producer generation after timeout invalidation"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn delivery_deadline_includes_waiting_for_the_serialization_lock() {
        let delivery_lock = Arc::new(Mutex::new(()));
        let delivery_poisoned = AtomicBool::new(false);
        let selected = Arc::new(AtomicBool::new(false));
        let _held = delivery_lock.lock().await;

        let error = with_serialized_kafka_delivery(
            &delivery_lock,
            &delivery_poisoned,
            {
                let selected = Arc::clone(&selected);
                move || async move {
                    selected.store(true, Ordering::SeqCst);
                    Ok(())
                }
            },
            |()| async { Ok(()) },
            || async {},
        )
        .await
        .expect_err("lock admission must share the absolute delivery deadline");

        assert!(matches!(error, TargetError::Timeout(_)));
        assert!(!selected.load(Ordering::SeqCst), "a timed-out waiter must not select a producer");
        assert!(!delivery_poisoned.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn cancelled_delivery_poisons_the_connection_before_the_next_selection() {
        let delivery_lock = Arc::new(Mutex::new(()));
        let delivery_poisoned = Arc::new(AtomicBool::new(false));
        let generation = Arc::new(AtomicUsize::new(1));
        let first_entered = Arc::new(Notify::new());
        let first = {
            let delivery_lock = Arc::clone(&delivery_lock);
            let delivery_poisoned = Arc::clone(&delivery_poisoned);
            let first_entered = Arc::clone(&first_entered);
            tokio::spawn(async move {
                with_serialized_kafka_delivery(
                    &delivery_lock,
                    &delivery_poisoned,
                    || async { Ok(1usize) },
                    move |_| async move {
                        first_entered.notify_one();
                        std::future::pending::<Result<(), TargetError>>().await
                    },
                    || async {},
                )
                .await
            })
        };
        first_entered.notified().await;
        first.abort();
        assert!(first.await.expect_err("first delivery should be cancelled").is_cancelled());
        assert!(delivery_poisoned.load(Ordering::Acquire));

        let selected = with_serialized_kafka_delivery(
            &delivery_lock,
            &delivery_poisoned,
            {
                let generation = Arc::clone(&generation);
                move || async move { Ok(generation.load(Ordering::SeqCst)) }
            },
            |selected| async move { Ok(selected) },
            {
                let generation = Arc::clone(&generation);
                move || {
                    let generation = Arc::clone(&generation);
                    async move { generation.store(2, Ordering::SeqCst) }
                }
            },
        )
        .await
        .expect("the next delivery should recover from cancellation poisoning");

        assert_eq!(selected, 2);
        assert!(!delivery_poisoned.load(Ordering::Acquire));
    }

    #[test]
    fn test_validate_empty_brokers() {
        let args = KafkaArgs {
            brokers: vec![],
            ..base_args()
        };
        assert!(args.validate().is_err());
    }

    #[test]
    fn test_validate_empty_topic() {
        let args = KafkaArgs {
            topic: String::new(),
            ..base_args()
        };
        assert!(args.validate().is_err());
    }

    #[test]
    fn test_validate_relative_queue_dir() {
        let args = KafkaArgs {
            queue_dir: "relative/path".to_string(),
            ..base_args()
        };
        assert!(args.validate().is_err());
    }

    #[test]
    fn test_validate_valid_args() {
        assert!(base_args().validate().is_ok());
    }

    #[test]
    fn test_validate_disabled_target_skips_validation() {
        let args = KafkaArgs {
            enable: false,
            brokers: vec![],
            topic: String::new(),
            ..base_args()
        };
        assert!(args.validate().is_ok());
    }

    #[test]
    fn test_validate_tls_client_cert_and_key_must_be_paired() {
        let args = KafkaArgs {
            tls_client_cert: "/tmp/client.crt".to_string(),
            tls_client_key: String::new(),
            ..base_args()
        };
        assert!(args.validate().is_err());
    }

    #[test]
    fn test_validate_sasl_requires_tls() {
        let args = KafkaArgs {
            sasl_enable: true,
            sasl_mechanism: KAFKA_SASL_SCRAM_SHA_512.to_string(),
            sasl_username: "user".to_string(),
            sasl_password: "secret".to_string(),
            ..base_args()
        };
        let err = args.validate().expect_err("SASL without TLS should fail");
        assert!(err.to_string().contains("requires tls_enable"));
    }

    #[test]
    fn test_validate_sasl_requires_username_and_password() {
        let args = KafkaArgs {
            tls_enable: true,
            sasl_enable: true,
            sasl_mechanism: KAFKA_SASL_PLAIN.to_string(),
            sasl_username: "user".to_string(),
            sasl_password: String::new(),
            ..base_args()
        };
        let err = args.validate().expect_err("SASL credentials should be paired");
        assert!(err.to_string().contains("sasl_username and sasl_password"));
    }

    #[test]
    fn test_validate_sasl_rejects_unsupported_mechanism() {
        let args = KafkaArgs {
            tls_enable: true,
            sasl_enable: true,
            sasl_mechanism: "OAUTHBEARER".to_string(),
            sasl_username: "user".to_string(),
            sasl_password: "secret".to_string(),
            ..base_args()
        };
        let err = args.validate().expect_err("unsupported SASL mechanism should fail");
        assert!(err.to_string().contains("sasl_mechanism must be one of"));
    }

    #[test]
    fn test_security_config_includes_sasl() {
        let args = KafkaArgs {
            tls_enable: true,
            sasl_enable: true,
            sasl_mechanism: "scram-sha-512".to_string(),
            sasl_username: "user".to_string(),
            sasl_password: "secret".to_string(),
            ..base_args()
        };

        let security = args
            .security_config(false)
            .expect("valid security config")
            .expect("security should be configured");
        let sasl = security.sasl_config().expect("SASL should be configured");

        assert_eq!(sasl.mechanism(), KAFKA_SASL_SCRAM_SHA_512);
        assert_eq!(sasl.username(), "user");
        assert_eq!(sasl.password(), "secret");
    }

    #[test]
    fn map_kafka_error_treats_transient_broker_states_as_retriable() {
        // Leader election / metadata staleness must be retried, not dropped
        // as a permanent failure (backlog#973).
        assert!(matches!(
            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::NotLeaderForPartition), "send"),
            TargetError::NotConnected
        ));
        assert!(matches!(
            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::LeaderNotAvailable), "send"),
            TargetError::NotConnected
        ));
        // RequestTimedOut is retriable and surfaced as a timeout.
        assert!(matches!(
            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::RequestTimedOut), "send"),
            TargetError::Timeout(_)
        ));
    }

    #[test]
    fn map_kafka_error_treats_permanent_broker_states_as_request_error() {
        // A missing topic/partition is a permanent condition; retrying would
        // storm the broker.
        assert!(matches!(
            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::UnknownTopicOrPartition), "send"),
            TargetError::Request(_)
        ));
        assert!(matches!(
            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Config("bad".to_string()), "send"),
            TargetError::Configuration(_)
        ));
    }

    #[test]
    fn test_debug_redacts_sasl_password_and_tls_key() {
        let rendered = format!(
            "{:?}",
            KafkaArgs {
                tls_client_key: "/tmp/client.key".to_string(),
                sasl_enable: true,
                sasl_password: "super-secret".to_string(),
                ..base_args()
            }
        );

        assert!(!rendered.contains("super-secret"));
        assert!(!rendered.contains("/tmp/client.key"));
        assert!(rendered.contains("***REDACTED***"));
    }
}