//! Common value types shared by producers, consumers, and admin calls.
//!
//! ```no_run
//! use kafkit_client::{KafkaMessage, RecordHeader};
//!
//! let message = KafkaMessage::new("created")
//!     .with_topic("orders")
//!     .with_key("order-42")
//!     .with_header(RecordHeader::new("source", "checkout"));
//! ```
//!
use std::collections::HashMap;

use anyhow::bail;
use bytes::Bytes;
use uuid::Uuid;

#[derive(Debug, Clone, PartialEq, Eq)]
/// Record Header.
pub struct RecordHeader {
    /// Record key.
    pub key: String,
    /// Record value.
    pub value: Option<Bytes>,
}

impl RecordHeader {
    /// Creates a new value.
    pub fn new(key: impl Into<String>, value: impl Into<Bytes>) -> Self {
        Self {
            key: key.into(),
            value: Some(value.into()),
        }
    }

    /// Null.
    pub fn null(key: impl Into<String>) -> Self {
        Self {
            key: key.into(),
            value: None,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
/// Topic Partition.
pub struct TopicPartition {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
}

impl TopicPartition {
    /// Creates a new value.
    pub fn new(topic: impl Into<String>, partition: i32) -> Self {
        Self {
            topic: topic.into(),
            partition,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
/// Subscription Pattern.
pub struct SubscriptionPattern {
    pattern: String,
}

impl SubscriptionPattern {
    /// Creates a new value.
    pub fn new(pattern: impl Into<String>) -> Self {
        Self {
            pattern: pattern.into(),
        }
    }

    /// Pattern.
    pub fn pattern(&self) -> &str {
        &self.pattern
    }

    /// Converts this value into inner.
    pub fn into_inner(self) -> String {
        self.pattern
    }
}

impl From<String> for SubscriptionPattern {
    fn from(value: String) -> Self {
        Self::new(value)
    }
}

impl From<&str> for SubscriptionPattern {
    fn from(value: &str) -> Self {
        Self::new(value)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Topic Partition Offset.
pub struct TopicPartitionOffset {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Kafka offset.
    pub offset: i64,
}

impl TopicPartitionOffset {
    /// Creates a new value.
    pub fn new(topic: impl Into<String>, partition: i32, offset: i64) -> Self {
        Self {
            topic: topic.into(),
            partition,
            offset,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Topic Partition Timestamp.
pub struct TopicPartitionTimestamp {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Timestamp.
    pub timestamp: i64,
}

impl TopicPartitionTimestamp {
    /// Creates a new value.
    pub fn new(topic: impl Into<String>, partition: i32, timestamp: i64) -> Self {
        Self {
            topic: topic.into(),
            partition,
            timestamp,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Offset And Timestamp.
pub struct OffsetAndTimestamp {
    /// Kafka offset.
    pub offset: i64,
    /// Timestamp.
    pub timestamp: i64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Topic Partition Offset And Timestamp.
pub struct TopicPartitionOffsetAndTimestamp {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Kafka offset.
    pub offset: i64,
    /// Timestamp.
    pub timestamp: i64,
}

impl TopicPartitionOffsetAndTimestamp {
    /// Creates a new value.
    pub fn new(topic: impl Into<String>, partition: i32, offset: i64, timestamp: i64) -> Self {
        Self {
            topic: topic.into(),
            partition,
            offset,
            timestamp,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Topic Partition Info.
pub struct TopicPartitionInfo {
    /// Partition number.
    pub partition: i32,
    /// Leader Id.
    pub leader_id: i32,
    /// Leader Epoch.
    pub leader_epoch: i32,
    /// Replica Nodes.
    pub replica_nodes: Vec<i32>,
    /// Isr Nodes.
    pub isr_nodes: Vec<i32>,
    /// Offline Replicas.
    pub offline_replicas: Vec<i32>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Consumer Group Metadata.
pub struct ConsumerGroupMetadata {
    /// Group Id.
    pub group_id: String,
    /// Generation Id.
    pub generation_id: i32,
    /// Member Id.
    pub member_id: String,
    /// Group Instance Id.
    pub group_instance_id: Option<String>,
}

impl ConsumerGroupMetadata {
    /// Creates a new value.
    pub fn new(group_id: impl Into<String>) -> Self {
        Self {
            group_id: group_id.into(),
            generation_id: -1,
            member_id: String::new(),
            group_instance_id: None,
        }
    }

    /// Sets generation id and returns the updated value.
    pub fn with_generation_id(mut self, generation_id: i32) -> Self {
        self.generation_id = generation_id;
        self
    }

    /// Sets member id and returns the updated value.
    pub fn with_member_id(mut self, member_id: impl Into<String>) -> Self {
        self.member_id = member_id.into();
        self
    }

    /// Sets group instance id and returns the updated value.
    pub fn with_group_instance_id(mut self, group_instance_id: impl Into<String>) -> Self {
        self.group_instance_id = Some(group_instance_id.into());
        self
    }
}

#[derive(Debug, Clone)]
/// Produce Record.
pub struct ProduceRecord {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: Option<i32>,
    /// Timestamp.
    pub timestamp: Option<i64>,
    /// Headers.
    pub headers: Vec<RecordHeader>,
    /// Record key.
    pub key: Option<Bytes>,
    /// Record value.
    pub value: Option<Bytes>,
}

impl ProduceRecord {
    /// Creates a new value.
    pub fn new(
        topic: impl Into<String>,
        partition: impl Into<Option<i32>>,
        value: impl Into<Bytes>,
    ) -> Self {
        Self {
            topic: topic.into(),
            partition: partition.into(),
            timestamp: None,
            headers: Vec::new(),
            key: None,
            value: Some(value.into()),
        }
    }

    /// Without Partition.
    pub fn without_partition(topic: impl Into<String>, value: impl Into<Bytes>) -> Self {
        Self::new(topic, None::<i32>, value)
    }

    /// To Partition.
    pub fn to_partition(topic: impl Into<String>, partition: i32, value: impl Into<Bytes>) -> Self {
        Self::new(topic, partition, value)
    }

    /// Tombstone.
    pub fn tombstone(topic: impl Into<String>, partition: impl Into<Option<i32>>) -> Self {
        Self {
            topic: topic.into(),
            partition: partition.into(),
            timestamp: None,
            headers: Vec::new(),
            key: None,
            value: None,
        }
    }

    /// Sets partition and returns the updated value.
    pub fn with_partition(mut self, partition: i32) -> Self {
        self.partition = Some(partition);
        self
    }

    /// Sets key and returns the updated value.
    pub fn with_key(mut self, key: impl Into<Bytes>) -> Self {
        self.key = Some(key.into());
        self
    }

    /// Sets timestamp and returns the updated value.
    pub fn with_timestamp(mut self, timestamp: i64) -> Self {
        self.timestamp = Some(timestamp);
        self
    }

    /// Sets null value and returns the updated value.
    pub fn with_null_value(mut self) -> Self {
        self.value = None;
        self
    }

    /// Sets header and returns the updated value.
    pub fn with_header(mut self, header: RecordHeader) -> Self {
        self.headers.push(header);
        self
    }

    /// Sets headers and returns the updated value.
    pub fn with_headers<I>(mut self, headers: I) -> Self
    where
        I: IntoIterator<Item = RecordHeader>,
    {
        self.headers.extend(headers);
        self
    }
}

#[derive(Debug, Clone)]
/// Produce Ack.
pub struct ProduceAck {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Base Offset.
    pub base_offset: i64,
    /// Kafka offset.
    pub offset: i64,
    /// Log Append Time Ms.
    pub log_append_time_ms: i64,
}

#[derive(Debug, Clone)]
/// Consumer Record.
pub struct ConsumerRecord {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Kafka offset.
    pub offset: i64,
    /// Timestamp.
    pub timestamp: i64,
    /// Headers.
    pub headers: Vec<RecordHeader>,
    /// Record key.
    pub key: Option<Bytes>,
    /// Record value.
    pub value: Option<Bytes>,
}

#[derive(Debug, Clone, Default)]
/// Consumer Records.
pub struct ConsumerRecords {
    records: Vec<ConsumerRecord>,
}

impl ConsumerRecords {
    /// Creates a new value.
    pub fn new(records: Vec<ConsumerRecord>) -> Self {
        Self { records }
    }

    /// Returns whether empty.
    pub fn is_empty(&self) -> bool {
        self.records.is_empty()
    }

    /// Returns the number of records.
    pub fn len(&self) -> usize {
        self.records.len()
    }

    /// Iterates over the records without taking ownership.
    pub fn iter(&self) -> impl Iterator<Item = &ConsumerRecord> {
        self.records.iter()
    }

    /// Converts this value into inner.
    pub fn into_inner(self) -> Vec<ConsumerRecord> {
        self.records
    }

    /// Commit Offsets.
    pub fn commit_offsets(&self) -> Vec<CommitOffset> {
        let mut offsets = HashMap::<TopicPartitionKey, i64>::new();
        for record in &self.records {
            offsets.insert(
                TopicPartitionKey::new(record.topic.clone(), record.partition),
                record.offset + 1,
            );
        }

        offsets
            .into_iter()
            .map(|(key, offset)| CommitOffset {
                topic: key.topic,
                partition: key.partition,
                offset,
            })
            .collect()
    }
}

#[derive(Debug, Clone)]
/// Commit Offset.
pub struct CommitOffset {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Kafka offset.
    pub offset: i64,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
/// Topic Partition Key.
pub struct TopicPartitionKey {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
}

impl TopicPartitionKey {
    /// Creates a new value.
    pub fn new(topic: String, partition: i32) -> Self {
        Self { topic, partition }
    }
}

impl From<TopicPartitionOffset> for CommitOffset {
    fn from(value: TopicPartitionOffset) -> Self {
        Self {
            topic: value.topic,
            partition: value.partition,
            offset: value.offset,
        }
    }
}

impl From<TopicPartition> for TopicPartitionKey {
    fn from(value: TopicPartition) -> Self {
        Self::new(value.topic, value.partition)
    }
}

impl From<&TopicPartition> for TopicPartitionKey {
    fn from(value: &TopicPartition) -> Self {
        Self::new(value.topic.clone(), value.partition)
    }
}

#[derive(Debug, Clone)]
/// Assigned Partition.
pub struct AssignedPartition {
    /// Record key.
    pub key: TopicPartitionKey,
    /// Topic Id.
    pub topic_id: Uuid,
    /// Leader Id.
    pub leader_id: i32,
    /// Leader Epoch.
    pub leader_epoch: i32,
    /// Fetch Offset.
    pub fetch_offset: i64,
}

#[derive(Debug, Clone)]
/// Kafka Message.
pub struct KafkaMessage {
    pub(crate) topic: Option<String>,
    pub(crate) partition: Option<i32>,
    pub(crate) timestamp: Option<i64>,
    pub(crate) headers: Vec<RecordHeader>,
    pub(crate) key: Option<Bytes>,
    pub(crate) value: Option<Bytes>,
}

impl KafkaMessage {
    /// Creates a new value.
    pub fn new(value: impl Into<Bytes>) -> Self {
        Self {
            topic: None,
            partition: None,
            timestamp: None,
            headers: Vec::new(),
            key: None,
            value: Some(value.into()),
        }
    }

    /// Tombstone.
    pub fn tombstone() -> Self {
        Self {
            topic: None,
            partition: None,
            timestamp: None,
            headers: Vec::new(),
            key: None,
            value: None,
        }
    }

    /// Sets topic and returns the updated value.
    pub fn with_topic(mut self, topic: impl Into<String>) -> Self {
        self.topic = Some(topic.into());
        self
    }

    /// Sets partition and returns the updated value.
    pub fn with_partition(mut self, partition: i32) -> Self {
        self.partition = Some(partition);
        self
    }

    /// Sets key and returns the updated value.
    pub fn with_key(mut self, key: impl Into<Bytes>) -> Self {
        self.key = Some(key.into());
        self
    }

    /// Sets timestamp and returns the updated value.
    pub fn with_timestamp(mut self, timestamp: i64) -> Self {
        self.timestamp = Some(timestamp);
        self
    }

    /// Sets null value and returns the updated value.
    pub fn with_null_value(mut self) -> Self {
        self.value = None;
        self
    }

    /// Sets header and returns the updated value.
    pub fn with_header(mut self, header: RecordHeader) -> Self {
        self.headers.push(header);
        self
    }

    /// Sets headers and returns the updated value.
    pub fn with_headers<I>(mut self, headers: I) -> Self
    where
        I: IntoIterator<Item = RecordHeader>,
    {
        self.headers.extend(headers);
        self
    }

    /// Converts this value into record.
    pub fn into_record(
        self,
        default_topic: Option<&str>,
        default_partition: Option<i32>,
    ) -> anyhow::Result<ProduceRecord> {
        let topic = match self.topic.as_deref().or(default_topic) {
            Some(t) => {
                let t = t.trim().to_owned();
                if t.is_empty() {
                    bail!("topic must be non-empty");
                }
                t
            }
            None => bail!("message requires a topic"),
        };
        let partition = self.partition.or(default_partition);
        let mut record = match self.value {
            Some(value) => ProduceRecord::without_partition(topic, value),
            None => ProduceRecord::tombstone(topic, None::<i32>),
        };
        if let Some(p) = partition {
            if p < 0 {
                bail!("partition must be non-negative");
            }
            record = record.with_partition(p);
        }
        if let Some(key) = self.key {
            record = record.with_key(key);
        }
        if let Some(timestamp) = self.timestamp {
            record = record.with_timestamp(timestamp);
        }
        if !self.headers.is_empty() {
            record = record.with_headers(self.headers);
        }
        Ok(record)
    }
}

impl<T> From<T> for KafkaMessage
where
    T: Into<Bytes>,
{
    fn from(value: T) -> Self {
        Self::new(value)
    }
}

impl From<ProduceRecord> for KafkaMessage {
    fn from(record: ProduceRecord) -> Self {
        Self {
            topic: Some(record.topic),
            partition: record.partition,
            timestamp: record.timestamp,
            headers: record.headers,
            key: record.key,
            value: record.value,
        }
    }
}

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

    #[test]
    fn consumer_records_commit_offsets_uses_next_offset() {
        let records = ConsumerRecords::new(vec![
            ConsumerRecord {
                topic: "topic-a".to_owned(),
                partition: 1,
                offset: 5,
                timestamp: 0,
                headers: Vec::new(),
                key: None,
                value: None,
            },
            ConsumerRecord {
                topic: "topic-a".to_owned(),
                partition: 1,
                offset: 8,
                timestamp: 0,
                headers: Vec::new(),
                key: None,
                value: None,
            },
        ]);

        let offsets = records.commit_offsets();
        assert_eq!(offsets.len(), 1);
        assert_eq!(offsets[0].offset, 9);
    }

    #[test]
    fn produce_record_builders_set_timestamp_and_headers() {
        let record = ProduceRecord::new("topic-a", 0, b"value".as_slice())
            .with_timestamp(1234)
            .with_header(RecordHeader::new("trace-id", b"abc".as_slice()))
            .with_header(RecordHeader::null("nullable"));

        assert_eq!(record.timestamp, Some(1234));
        assert_eq!(record.headers.len(), 2);
        assert_eq!(record.headers[0].key, "trace-id");
        assert_eq!(record.headers[0].value.as_deref(), Some(&b"abc"[..]));
        assert_eq!(record.headers[1].key, "nullable");
        assert!(record.headers[1].value.is_none());
    }

    #[test]
    fn produce_record_tombstone_builder_sets_null_value() {
        let record = ProduceRecord::tombstone("topic-a", 3).with_key("key");

        assert_eq!(record.topic, "topic-a");
        assert_eq!(record.partition, Some(3));
        assert_eq!(record.key.as_deref(), Some(&b"key"[..]));
        assert!(record.value.is_none());
    }
}
