datum-mq 0.11.2

use std::{
    sync::{
        Arc, Condvar, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use crate::native::{
    AutoOffsetReset, FetchTuning, GroupAssignor, NativeCommitPolicy, NativeKafkaConsumerConfig,
    NativeKafkaSource, StartOffset, TopicPartitionAssignment,
};
use crate::offset::{OffsetCommitter, OffsetPosition};
use crate::{
    CommitPolicy, ConsumerRecord, KafkaBatchOffset, KafkaConsumerSettings, KafkaMetrics,
    KafkaOffset, KafkaTimestamp, MqError, MqResult, Offset, Subscription, TopicPartition,
};
use bytes::Bytes;
use datum::{Source, SourceWithContext, StreamError};

/// Kafka source entry points.
pub struct KafkaSource;

#[derive(Debug, Clone)]
struct KafkaEnvelope {
    record: ConsumerRecord,
    offset: KafkaOffset,
}

/// Payload-only Kafka record emitted by [`KafkaSource::committable_payload_batches`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KafkaPayloadRecord {
    pub partition: i32,
    pub offset: i64,
    pub timestamp: KafkaTimestamp,
    payload_start: usize,
    payload_end: usize,
}

impl KafkaPayloadRecord {
    #[must_use]
    pub fn payload<'a>(&self, batch: &'a KafkaPayloadBatch) -> &'a [u8] {
        &batch.payloads[self.payload_start..self.payload_end]
    }
}

/// A payload batch plus one committable Kafka watermark per touched partition.
#[derive(Debug, Clone)]
pub struct KafkaPayloadBatch {
    records: Vec<KafkaPayloadRecord>,
    payloads: Bytes,
    offset: KafkaBatchOffset,
    active_partitions: Vec<TopicPartition>,
}

impl KafkaPayloadBatch {
    #[must_use]
    pub fn records(&self) -> &[KafkaPayloadRecord] {
        &self.records
    }

    #[must_use]
    pub fn len(&self) -> usize {
        self.records.len()
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.records.is_empty()
    }

    #[must_use]
    pub fn payload(&self, record: &KafkaPayloadRecord) -> &[u8] {
        record.payload(self)
    }

    #[must_use]
    pub fn offset(&self) -> &KafkaBatchOffset {
        &self.offset
    }

    /// Topic/partitions assigned to the consumer when this batch was emitted.
    ///
    /// The list includes partitions that did not contribute a record to this
    /// batch. Event-time consumers use it to min-reduce watermarks across idle
    /// or slower Kafka partitions before exposing a merged stream.
    #[must_use]
    pub fn active_partitions(&self) -> &[TopicPartition] {
        &self.active_partitions
    }

    /// Commits this batch after the downstream batch checkpoint succeeds.
    pub fn commit(&self) -> MqResult<()> {
        self.offset.commit()
    }
}

impl KafkaSource {
    /// Creates a plain source of owned Kafka records.
    ///
    /// Use [`KafkaSource::committable`] when records should advance consumer-group
    /// offsets after downstream checkpoints.
    #[must_use]
    pub fn plain(
        settings: KafkaConsumerSettings,
        subscription: Subscription,
    ) -> Source<ConsumerRecord, KafkaControl> {
        native_envelope_source(settings, subscription).map(|envelope| envelope.record)
    }

    /// Creates a source whose records carry committable Kafka offsets as context.
    ///
    /// Call [`KafkaOffset::commit`](crate::KafkaOffset::commit) only after the
    /// corresponding downstream checkpoint or side effect succeeds.
    #[must_use]
    pub fn committable(
        settings: KafkaConsumerSettings,
        subscription: Subscription,
    ) -> SourceWithContext<ConsumerRecord, KafkaOffset, KafkaControl> {
        native_envelope_source(settings, subscription)
            .as_source_with_context(|envelope| envelope.offset.clone())
            .map(|envelope| envelope.record)
    }

    /// Creates a payload-only batched source with committable partition watermarks.
    ///
    /// This is the throughput-oriented counterpart to [`KafkaSource::committable`].
    /// It emits one Datum element per Kafka poll batch, copies payload bytes into a
    /// batch buffer, and exposes a single [`KafkaBatchOffset`] for the highest
    /// offset in each touched topic/partition. Downstream code should call
    /// [`KafkaPayloadBatch::commit`] only after the whole batch checkpoint succeeds.
    #[must_use]
    pub fn committable_payload_batches(
        settings: KafkaConsumerSettings,
        subscription: Subscription,
    ) -> Source<KafkaPayloadBatch, KafkaControl> {
        native_payload_batch_source(settings, subscription)
    }
}

fn failed_controlled_source<Out: Send + 'static>(
    control: KafkaControl,
    message: String,
) -> Source<Out, KafkaControl> {
    Source::unfold_resource(
        move || -> datum::StreamResult<()> { Err(StreamError::Failed(message.clone())) },
        |_| Ok(None),
        |_| Ok(()),
    )
    .map_materialized_value(move |_| control.clone())
}

fn native_payload_batch_source(
    settings: KafkaConsumerSettings,
    subscription: Subscription,
) -> Source<KafkaPayloadBatch, KafkaControl> {
    let metrics = KafkaMetrics::default();
    let control = KafkaControl::new(metrics.clone());
    let config = match native_config_from_settings(&settings, &subscription) {
        Ok(config) => config,
        Err(error) => {
            return failed_controlled_source(control, error.to_string());
        }
    };
    NativeKafkaSource::payload_batches(config)
        .map(native_batch_to_mq)
        .map_materialized_value({
            let control = control.clone();
            move |native_control| {
                let delegate = Arc::new(NativeControlDelegate { native_control });
                delegate.refresh_metrics(&control.state.metrics);
                control.register_delegate(delegate);
                control.clone()
            }
        })
}

fn native_envelope_source(
    settings: KafkaConsumerSettings,
    subscription: Subscription,
) -> Source<KafkaEnvelope, KafkaControl> {
    let metrics = KafkaMetrics::default();
    let control = KafkaControl::new(metrics.clone());
    let config = match native_config_from_settings(&settings, &subscription) {
        Ok(config) => config,
        Err(error) => {
            return failed_controlled_source(control, error.to_string());
        }
    };
    NativeKafkaSource::payload_batches(config)
        .map_concat(native_batch_to_envelopes)
        .map_materialized_value({
            let control = control.clone();
            move |native_control| {
                let delegate = Arc::new(NativeControlDelegate { native_control });
                delegate.refresh_metrics(&control.state.metrics);
                control.register_delegate(delegate);
                control.clone()
            }
        })
}

/// Materialized control for a Kafka source.
#[derive(Debug, Clone)]
pub struct KafkaControl {
    state: Arc<SourceControlState>,
}

struct SourceControlState {
    draining: AtomicBool,
    shutdown_now: AtomicBool,
    metrics: KafkaMetrics,
    delegate: Mutex<Option<Arc<dyn KafkaControlDelegate>>>,
    wait_lock: Mutex<()>,
    wait_cv: Condvar,
}

impl std::fmt::Debug for SourceControlState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SourceControlState")
            .field("draining", &self.draining.load(Ordering::Relaxed))
            .field("shutdown_now", &self.shutdown_now.load(Ordering::Relaxed))
            .field("metrics", &self.metrics.snapshot())
            .finish_non_exhaustive()
    }
}

trait KafkaControlDelegate: Send + Sync {
    fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()>;
    fn shutdown_now(&self);
    fn refresh_metrics(&self, metrics: &KafkaMetrics);
}

impl KafkaControl {
    fn new(metrics: KafkaMetrics) -> Self {
        Self {
            state: Arc::new(SourceControlState {
                draining: AtomicBool::new(false),
                shutdown_now: AtomicBool::new(false),
                metrics,
                delegate: Mutex::new(None),
                wait_lock: Mutex::new(()),
                wait_cv: Condvar::new(),
            }),
        }
    }

    #[must_use]
    pub fn metrics(&self) -> KafkaMetrics {
        if let Some(delegate) = self.delegate() {
            delegate.refresh_metrics(&self.state.metrics);
        }
        self.state.metrics.clone()
    }

    #[must_use]
    pub fn is_draining(&self) -> bool {
        self.state.draining.load(Ordering::SeqCst)
    }

    /// Stops new emission and waits for already emitted offsets to be committed.
    ///
    /// The source iterator completes on its next pull once outstanding offsets are
    /// drained. This is the Datum equivalent of the Alpakka `DrainingControl`
    /// hook; `datum-agent` jobs can call this before stopping the materialized
    /// graph.
    pub fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()> {
        self.state.draining.store(true, Ordering::SeqCst);
        self.state.wait_cv.notify_all();
        if let Some(delegate) = self.delegate() {
            let result = delegate.drain_and_shutdown(timeout);
            delegate.refresh_metrics(&self.state.metrics);
            return result;
        }
        let deadline = Instant::now() + timeout;
        let mut guard = self
            .state
            .wait_lock
            .lock()
            .map_err(|_| MqError::Failed("Kafka source drain lock poisoned".to_owned()))?;
        loop {
            if self.metrics().snapshot().outstanding == 0 {
                return Ok(());
            }
            let now = Instant::now();
            if now >= deadline {
                return Err(MqError::DrainTimeout);
            }
            let remaining = deadline.saturating_duration_since(now);
            let (next_guard, result) = self
                .state
                .wait_cv
                .wait_timeout(guard, remaining)
                .map_err(|_| MqError::Failed("Kafka source drain wait poisoned".to_owned()))?;
            guard = next_guard;
            if result.timed_out() && self.metrics().snapshot().outstanding != 0 {
                return Err(MqError::DrainTimeout);
            }
        }
    }

    /// Stops the source without claiming uncommitted offsets.
    pub fn shutdown_now(&self) {
        self.state.shutdown_now.store(true, Ordering::SeqCst);
        self.state.wait_cv.notify_all();
        if let Some(delegate) = self.delegate() {
            delegate.shutdown_now();
            delegate.refresh_metrics(&self.state.metrics);
        }
    }

    fn register_delegate(&self, delegate: Arc<dyn KafkaControlDelegate>) {
        if let Ok(mut current) = self.state.delegate.lock() {
            *current = Some(delegate);
        }
    }

    fn delegate(&self) -> Option<Arc<dyn KafkaControlDelegate>> {
        self.state
            .delegate
            .lock()
            .ok()
            .and_then(|delegate| delegate.clone())
    }
}

struct NativeControlDelegate {
    native_control: crate::native::NativeKafkaControl,
}

impl KafkaControlDelegate for NativeControlDelegate {
    fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()> {
        self.native_control
            .drain_and_shutdown(timeout)
            .map_err(native_error)
    }

    fn shutdown_now(&self) {
        self.native_control.shutdown_now();
    }

    fn refresh_metrics(&self, metrics: &KafkaMetrics) {
        let snapshot = self.native_control.metrics();
        metrics.replace_from(crate::KafkaMetricsSnapshot {
            assigned_partitions: snapshot.assigned_partitions,
            revoked_partitions: snapshot.revoked_partitions,
            lost_partitions: snapshot.lost_partitions,
            rebalances: snapshot.rebalances,
            emitted: snapshot.emitted,
            committed_offsets: snapshot.committed_offsets,
            commit_failures: snapshot.broker_commit_failures,
            paused: false,
            outstanding: snapshot.outstanding,
            producer_in_flight: 0,
            produced: 0,
            delivery_failures: 0,
            queue_full: 0,
            high_watermark: u64_to_i64(snapshot.high_watermark),
            committed_watermark: u64_to_i64(snapshot.committed_watermark),
        });
    }
}

fn u64_to_i64(value: u64) -> i64 {
    i64::try_from(value).unwrap_or(i64::MAX)
}

fn native_batch_to_mq(batch: crate::native::KafkaPayloadBatch) -> KafkaPayloadBatch {
    let commit_batch = batch.clone();
    let parts = batch.into_parts();
    let records = parts
        .records
        .into_iter()
        .map(|record| {
            let range = record.payload_range();
            KafkaPayloadRecord {
                partition: record.partition,
                offset: record.offset,
                timestamp: native_timestamp(record.timestamp),
                payload_start: range.start,
                payload_end: range.end,
            }
        })
        .collect::<Vec<_>>();
    let positions = parts
        .watermarks
        .iter()
        .map(|watermark| {
            OffsetPosition::new(
                watermark.topic.clone(),
                watermark.partition,
                watermark.offset - 1,
            )
        })
        .collect::<Vec<_>>();
    let active_partitions = parts
        .active_partitions
        .into_iter()
        .map(|partition| TopicPartition::new(partition.topic, partition.partition))
        .collect();
    KafkaPayloadBatch {
        records,
        payloads: parts.payloads,
        offset: KafkaBatchOffset::custom(positions, move || {
            commit_batch.commit().map_err(native_error)
        }),
        active_partitions,
    }
}

fn native_batch_to_envelopes(batch: crate::native::KafkaPayloadBatch) -> Vec<KafkaEnvelope> {
    let batch = Arc::new(batch);
    batch
        .records()
        .iter()
        .map(|record| {
            let topic = record.topic(&batch).to_owned();
            let position = OffsetPosition::new(topic.clone(), record.partition, record.offset);
            let offset_committer: Arc<dyn OffsetCommitter> = Arc::new(NativeRecordCommitter {
                batch: Arc::clone(&batch),
            });
            KafkaEnvelope {
                record: ConsumerRecord {
                    topic,
                    partition: record.partition,
                    offset: record.offset,
                    timestamp: native_timestamp(record.timestamp),
                    key: None,
                    payload: Some(Bytes::copy_from_slice(batch.payload(record))),
                    headers: Vec::new(),
                },
                offset: KafkaOffset::new(position, offset_committer),
            }
        })
        .collect()
}

struct NativeRecordCommitter {
    batch: Arc<crate::native::KafkaPayloadBatch>,
}

impl OffsetCommitter for NativeRecordCommitter {
    fn stage_offset(&self, position: OffsetPosition) -> MqResult<()> {
        self.commit_position(position)
    }

    fn commit_offset(&self, position: OffsetPosition) -> MqResult<()> {
        self.commit_position(position)
    }
}

impl NativeRecordCommitter {
    fn commit_position(&self, position: OffsetPosition) -> MqResult<()> {
        self.batch
            .commit_offset(position.topic, position.partition, position.offset)
            .map_err(native_error)
    }
}

fn native_config_from_settings(
    settings: &KafkaConsumerSettings,
    subscription: &Subscription,
) -> MqResult<NativeKafkaConsumerConfig> {
    subscription.validate()?;
    validate_native_boundary(settings)?;
    let bootstrap = settings.config.get("bootstrap.servers").ok_or_else(|| {
        MqError::InvalidConfig(
            "native Kafka consumer backend requires bootstrap.servers".to_owned(),
        )
    })?;
    let mut config = match subscription {
        Subscription::Assignment(partitions) => NativeKafkaConsumerConfig::new(
            bootstrap.to_owned(),
            partitions
                .iter()
                .map(native_assignment)
                .collect::<MqResult<Vec<_>>>()?,
        ),
        Subscription::Topics(topics) => {
            NativeKafkaConsumerConfig::subscribe(bootstrap.to_owned(), topics.clone())
        }
        Subscription::Pattern(_) => {
            return Err(unsupported_native(
                "pattern subscriptions are not supported by the native backend",
            ));
        }
    };
    config.group_id = settings.group_id.clone();
    config.commit_policy = match settings.commit_policy {
        CommitPolicy::Manual => NativeCommitPolicy::Manual,
        CommitPolicy::External => NativeCommitPolicy::External,
        CommitPolicy::AutoKafka => {
            return Err(unsupported_native(
                "Kafka auto-commit is not supported by the native backend",
            ));
        }
    };
    config.commit_batch_size = settings.commit_batch_size;
    config.commit_interval = settings.commit_interval;
    config.fetch = native_fetch_tuning(settings)?;
    config.auto_offset_reset = native_auto_offset_reset(settings)?;
    config.client.security = crate::native::native_security_config(&settings.config)?;
    if let Some(client_id) = settings.config.get("client.id") {
        config.client.client_id = client_id.to_owned();
    }
    if let Some(request_timeout_ms) = parse_u64_config(settings, "request.timeout.ms")? {
        config.client.request_timeout = Duration::from_millis(request_timeout_ms);
    }
    if let Some(group_instance_id) = settings.config.get("group.instance.id") {
        config.group_instance_id = Some(group_instance_id.to_owned());
    }
    if let Some(assignor) = settings.config.get("partition.assignment.strategy") {
        config.group_assignor = native_group_assignor(assignor)?;
    }
    if let Some(session_timeout_ms) = parse_u64_config(settings, "session.timeout.ms")? {
        config.session_timeout = Duration::from_millis(session_timeout_ms);
    }
    if let Some(rebalance_timeout_ms) = parse_u64_config(settings, "max.poll.interval.ms")? {
        config.rebalance_timeout = Duration::from_millis(rebalance_timeout_ms);
    }
    if let Some(heartbeat_interval_ms) = parse_u64_config(settings, "heartbeat.interval.ms")? {
        config.heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);
    }
    Ok(config)
}

fn validate_native_boundary(settings: &KafkaConsumerSettings) -> MqResult<()> {
    for (key, value) in settings.config.properties() {
        let key_lower = key.to_ascii_lowercase();
        let value_lower = value.trim().to_ascii_lowercase();
        if matches!(key_lower.as_str(), "transactional.id") || key_lower.starts_with("transaction.")
        {
            return Err(unsupported_native(
                "transactions are not supported by the native backend",
            ));
        }
        if matches!(
            key_lower.as_str(),
            "compression.type" | "compression.codec" | "topic.compression.codec"
        ) {
            match value_lower.as_str() {
                "" | "none" | "uncompressed" | "gzip" | "snappy" | "lz4" | "zstd" => {}
                other => {
                    return Err(MqError::InvalidConfig(format!(
                        "native Kafka consumer compression setting must be none, gzip, snappy, lz4, or zstd; got {other}"
                    )));
                }
            }
        }
        if key_lower == "enable.auto.commit" && raw_bool_true(&value_lower) {
            return Err(unsupported_native(
                "Kafka auto-commit is not supported by the native backend; use CommitPolicy::Manual or External",
            ));
        }
        if key_lower == "enable.auto.offset.store" && raw_bool_true(&value_lower) {
            return Err(unsupported_native(
                "Kafka auto offset store is not supported by the native backend; use CommitPolicy::Manual or External",
            ));
        }
        if key_lower == "api.version.request" && !raw_bool_true(&value_lower) {
            return Err(unsupported_native(
                "broker/version fallback is outside the native support matrix; native requires ApiVersions negotiation against Kafka 2.8+",
            ));
        }
        if key_lower == "broker.version.fallback" {
            return Err(unsupported_native(
                "broker.version.fallback is outside the native support matrix; native requires Kafka 2.8+ ApiVersions negotiation",
            ));
        }
    }
    if settings
        .config
        .get("isolation.level")
        .is_some_and(|value| value.eq_ignore_ascii_case("read_committed"))
    {
        return Err(unsupported_native(
            "read_committed isolation and transactions are not supported by the native backend",
        ));
    }
    if let Some(group_protocol) = settings.config.get("group.protocol") {
        match group_protocol.trim().to_ascii_lowercase().as_str() {
            "" | "classic" => {}
            "consumer" => {
                return Err(unsupported_native(
                    "KIP-848 group.protocol=consumer is not supported by the native backend; use classic groups",
                ));
            }
            other => {
                return Err(unsupported_native(format!(
                    "group.protocol={other} is not supported by the native backend; use classic groups"
                )));
            }
        }
    }
    Ok(())
}

fn native_assignment(
    partition: &crate::TopicPartitionOffset,
) -> MqResult<TopicPartitionAssignment> {
    let start_offset = match partition.offset {
        Offset::Beginning => StartOffset::Beginning,
        Offset::End => StartOffset::End,
        Offset::Stored => StartOffset::Committed,
        Offset::Offset(offset) => StartOffset::Offset(offset),
        Offset::OffsetTail(_) => {
            return Err(unsupported_native(
                "tail-relative assignment offsets are not supported by the native backend",
            ));
        }
        Offset::Invalid => {
            return Err(MqError::InvalidConfig(format!(
                "native Kafka assignment for {}:{} has invalid start offset",
                partition.topic, partition.partition
            )));
        }
    };
    Ok(TopicPartitionAssignment {
        topic: partition.topic.clone(),
        partition: partition.partition,
        start_offset,
    })
}

fn native_fetch_tuning(settings: &KafkaConsumerSettings) -> MqResult<FetchTuning> {
    let mut tuning = FetchTuning {
        output_records: settings.poll_batch_size,
        output_payload_bytes: settings.poll_batch_size.saturating_mul(256).max(1),
        include_timestamps: settings.include_payload_timestamps,
        high_watermark: settings.high_watermark,
        low_watermark: settings.low_watermark,
        ..FetchTuning::default()
    };
    if let Some(value) = parse_i32_config(settings, "fetch.wait.max.ms")? {
        tuning.max_wait_ms = value;
    }
    if let Some(value) = parse_i32_config(settings, "fetch.min.bytes")? {
        tuning.min_bytes = value;
    }
    if let Some(value) = parse_i32_config(settings, "fetch.max.bytes")? {
        tuning.max_bytes = value;
    }
    if let Some(value) = parse_i32_config(settings, "max.partition.fetch.bytes")?
        .or(parse_i32_config(settings, "fetch.message.max.bytes")?)
    {
        tuning.partition_max_bytes = value;
    }
    Ok(tuning)
}

fn native_auto_offset_reset(settings: &KafkaConsumerSettings) -> MqResult<AutoOffsetReset> {
    match settings
        .config
        .get("auto.offset.reset")
        .unwrap_or("latest")
        .trim()
        .to_ascii_lowercase()
        .as_str()
    {
        "earliest" | "smallest" => Ok(AutoOffsetReset::Earliest),
        "latest" | "largest" => Ok(AutoOffsetReset::Latest),
        "error" | "none" => Ok(AutoOffsetReset::Error),
        other => Err(MqError::InvalidConfig(format!(
            "native Kafka consumer backend does not recognize auto.offset.reset={other}"
        ))),
    }
}

fn native_group_assignor(value: &str) -> MqResult<GroupAssignor> {
    if value.contains(',') {
        return Err(unsupported_native(
            "native Kafka consumer backend supports one classic assignor at a time (cooperative-sticky or range), not an assignor list",
        ));
    }
    let normalized = value.trim().to_ascii_lowercase();
    if normalized == "cooperative-sticky" || normalized.contains("cooperativestickyassignor") {
        Ok(GroupAssignor::CooperativeSticky)
    } else if normalized == "range" || normalized.contains("rangeassignor") {
        Ok(GroupAssignor::Range)
    } else {
        Err(unsupported_native(format!(
            "partition.assignment.strategy={value} is not supported by the native backend"
        )))
    }
}

fn parse_i32_config(settings: &KafkaConsumerSettings, key: &str) -> MqResult<Option<i32>> {
    settings
        .config
        .get(key)
        .map(|value| {
            value.parse::<i32>().map_err(|_| {
                MqError::InvalidConfig(format!(
                    "native Kafka consumer backend expected integer {key}, got {value}"
                ))
            })
        })
        .transpose()
}

fn parse_u64_config(settings: &KafkaConsumerSettings, key: &str) -> MqResult<Option<u64>> {
    settings
        .config
        .get(key)
        .map(|value| {
            value.parse::<u64>().map_err(|_| {
                MqError::InvalidConfig(format!(
                    "native Kafka consumer backend expected integer millisecond {key}, got {value}"
                ))
            })
        })
        .transpose()
}

fn raw_bool_true(value: &str) -> bool {
    matches!(value.trim(), "1" | "true" | "TRUE" | "True" | "yes" | "YES")
}

fn native_timestamp(timestamp: crate::native::KafkaTimestamp) -> KafkaTimestamp {
    match timestamp {
        crate::native::KafkaTimestamp::NotAvailable => KafkaTimestamp::NotAvailable,
        crate::native::KafkaTimestamp::CreateTime(value) => KafkaTimestamp::CreateTime(value),
        crate::native::KafkaTimestamp::LogAppendTime(value) => KafkaTimestamp::LogAppendTime(value),
    }
}

fn native_error(error: crate::native::KafkaClientError) -> MqError {
    match error {
        crate::native::KafkaClientError::InvalidConfig(message) => {
            MqError::InvalidConfig(format!("native Kafka consumer backend: {message}"))
        }
        crate::native::KafkaClientError::Unsupported(message) => MqError::InvalidConfig(format!(
            "native Kafka consumer backend: {message}; {}",
            crate::consumer_backend_hint()
        )),
        crate::native::KafkaClientError::AssignmentLost { topic, partition } => {
            MqError::AssignmentLost { topic, partition }
        }
        crate::native::KafkaClientError::Tls { broker, message } => {
            MqError::NativeTls { broker, message }
        }
        crate::native::KafkaClientError::Sasl {
            broker,
            mechanism,
            message,
        } => MqError::NativeSasl {
            broker,
            mechanism,
            message,
        },
        other => MqError::Failed(format!("native Kafka consumer backend failed: {other}")),
    }
}

fn unsupported_native(message: impl Into<String>) -> MqError {
    MqError::InvalidConfig(format!(
        "native Kafka consumer backend unsupported boundary: {}; {}",
        message.into(),
        crate::consumer_backend_hint()
    ))
}

#[cfg(test)]
mod native_tests {
    use super::*;
    use crate::KafkaConsumerBackend;

    fn native_settings() -> KafkaConsumerSettings {
        KafkaConsumerSettings::new("127.0.0.1:9092", "group")
            .with_consumer_backend(KafkaConsumerBackend::Native)
    }

    fn native_error_for(settings: KafkaConsumerSettings) -> String {
        native_config_from_settings(&settings, &Subscription::topics(["topic"]))
            .expect_err("native config should reject unsupported boundary")
            .to_string()
    }

    #[test]
    fn native_backend_rejects_unsupported_boundaries() {
        let cases = [
            (
                native_settings()
                    .with("security.protocol", "ssl")
                    .with("ssl.endpoint.identification.algorithm", ""),
                "verification",
            ),
            (
                native_settings()
                    .with("security.protocol", "sasl_ssl")
                    .with("sasl.mechanism", "GSSAPI")
                    .with("sasl.username", "user")
                    .with("sasl.password", "secret"),
                "GSSAPI",
            ),
            (
                native_settings().with("isolation.level", "read_committed"),
                "read_committed",
            ),
            (
                native_settings().with("transactional.id", "tx"),
                "transactions",
            ),
            (
                native_settings().with("group.protocol", "consumer"),
                "KIP-848",
            ),
            (
                native_settings().with("api.version.request", "false"),
                "support matrix",
            ),
        ];

        for (settings, expected) in cases {
            let error = native_error_for(settings);
            assert!(
                error.contains(expected),
                "expected '{error}' to name '{expected}'"
            );
            assert!(
                error.contains("only the native Kafka consumer backend"),
                "unsupported boundary must identify the native-only backend: {error}"
            );
        }
    }

    #[test]
    fn native_backend_accepts_tls_and_ratified_sasl_mechanisms() {
        for codec in ["none", "gzip", "snappy", "lz4", "zstd"] {
            assert!(
                native_config_from_settings(
                    &native_settings().with("compression.type", codec),
                    &Subscription::topics(["topic"]),
                )
                .is_ok(),
                "codec={codec}"
            );
        }
        assert!(
            native_config_from_settings(
                &native_settings().with("security.protocol", "ssl"),
                &Subscription::topics(["topic"]),
            )
            .is_ok()
        );
        for mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"] {
            let settings = native_settings()
                .with("security.protocol", "sasl_ssl")
                .with("sasl.mechanism", mechanism)
                .with("sasl.username", "user")
                .with("sasl.password", "secret");
            assert!(
                native_config_from_settings(&settings, &Subscription::topics(["topic"])).is_ok(),
                "mechanism={mechanism}"
            );
        }
    }

    #[test]
    fn native_backend_rejects_auto_commit_and_pattern_subscriptions() {
        let error = native_config_from_settings(
            &native_settings().with_commit_policy(CommitPolicy::AutoKafka),
            &Subscription::topics(["topic"]),
        )
        .expect_err("native auto commit should be rejected")
        .to_string();
        assert!(error.contains("auto-commit"));

        let error =
            native_config_from_settings(&native_settings(), &Subscription::pattern("topic-.*"))
                .expect_err("native pattern subscription should be rejected")
                .to_string();
        assert!(error.contains("pattern subscriptions"));
    }
}