krafka 0.9.1

//! Producer record types.

use std::sync::Arc;

use bytes::Bytes;

use crate::error::{KrafkaError, ProtocolErrorKind, Result};
use crate::protocol::{MAX_RECORD_HEADERS, RecordBatchBuilder, validate_topic_name};
use crate::{PartitionId, Timestamp};

/// A record to be sent to Kafka.
#[non_exhaustive]
#[must_use]
#[derive(Debug, Clone)]
pub struct ProducerRecord {
    /// Target topic.
    pub topic: String,
    /// Target partition (optional, will be computed if not set).
    pub partition: Option<PartitionId>,
    /// Record key (optional, zero-copy via `Bytes`).
    pub key: Option<Bytes>,
    /// Record value (zero-copy via `Bytes`).
    pub value: Bytes,
    /// Record timestamp (optional, will use current time if not set).
    pub timestamp: Option<Timestamp>,
    /// Record headers.
    pub headers: Vec<(String, Bytes)>,
    /// Optional record name forwarded to the schema encoder's subject-name strategy.
    ///
    /// Required when using
    /// [`crate::schema_registry::SubjectNameStrategy::RecordName`] or
    /// [`crate::schema_registry::SubjectNameStrategy::TopicRecordName`]. Pass
    /// `None` (or omit) when using the default
    /// [`crate::schema_registry::SubjectNameStrategy::TopicName`] strategy.
    pub record_name: Option<String>,
}

impl ProducerRecord {
    /// Create a new producer record.
    pub fn new(topic: impl Into<String>, value: impl Into<Bytes>) -> Self {
        Self {
            topic: topic.into(),
            partition: None,
            key: None,
            value: value.into(),
            timestamp: None,
            headers: Vec::new(),
            record_name: None,
        }
    }

    /// Set the partition.
    pub fn with_partition(mut self, partition: PartitionId) -> Self {
        self.partition = Some(partition);
        self
    }

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

    /// Clear the key (set it to `None`).
    pub fn without_key(mut self) -> Self {
        self.key = None;
        self
    }

    /// Set the timestamp.
    pub fn with_timestamp(mut self, timestamp: Timestamp) -> Self {
        self.timestamp = Some(timestamp);
        self
    }

    /// Add a header.
    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<Bytes>) -> Self {
        self.headers.push((key.into(), value.into()));
        self
    }

    /// Set the record name for schema subject-name resolution.
    ///
    /// Only needed when using
    /// [`crate::schema_registry::SubjectNameStrategy::RecordName`] or
    /// [`crate::schema_registry::SubjectNameStrategy::TopicRecordName`].
    /// Ignored by the default
    /// `TopicName` strategy.
    pub fn with_record_name(mut self, name: impl Into<String>) -> Self {
        self.record_name = Some(name.into());
        self
    }

    /// Get the key as a string (if valid UTF-8).
    #[inline]
    pub fn key_str(&self) -> Option<&str> {
        self.key.as_ref().and_then(|k| std::str::from_utf8(k).ok())
    }

    /// Get the value as a string (if valid UTF-8).
    #[inline]
    pub fn value_str(&self) -> Option<&str> {
        std::str::from_utf8(&self.value).ok()
    }

    /// Get the estimated size in bytes.
    ///
    /// Returns a conservative upper-bound on the wire-encoded size of this
    /// record within a RecordBatch v2 frame.  The estimate is used for both
    /// batch size-gating and memory backpressure; an undercount can cause
    /// batches to exceed `max_request_size` and trigger broker-side
    /// `MESSAGE_TOO_LARGE` errors.
    ///
    /// # Wire layout (RecordBatch v2 per-record)
    ///
    /// ```text
    /// signed_varint(body_size)      — record length prefix (exact)
    /// i8 attributes                 — 1 byte (fixed)
    /// signed_varlong(ts_delta)      — ≤ 5 bytes (conservative for typical batch windows)
    /// signed_varint(off_delta)      — ≤ 2 bytes (covers batches up to ~16 k records)
    /// signed_varint(key_len) + key  — exact varint + bytes
    /// signed_varint(val_len) + val  — exact varint + bytes
    /// signed_varint(hdr_count)      — exact varint
    ///   per header: varint(k_len) + k + varint(v_len) + v
    /// ```
    ///
    /// An additional per-record batch-overhead allowance is added to amortise
    /// the RecordBatch fixed header (61 bytes) and per-topic produce-request
    /// framing across records.
    #[inline]
    pub fn estimated_size(&self) -> usize {
        use crate::util::varint;

        // Unknowns at this point; conservative fixed estimates:
        //   timestamp_delta ≤ 5 bytes (covers ~67 s at ms resolution — typical batch window)
        //   offset_delta    ≤ 2 bytes (covers batches up to 16383 records)
        const TIMESTAMP_DELTA_BYTES: usize = 5;
        const OFFSET_DELTA_BYTES: usize = 2;

        let key_bytes = self.key.as_ref().map_or(0, |k| k.len());
        let val_bytes = self.value.len();

        let key_varint = match &self.key {
            Some(k) => varint::signed_varint_size(k.len() as i32),
            None => varint::signed_varint_size(-1), // null sentinel
        };
        let val_varint = varint::signed_varint_size(val_bytes as i32);
        let hdr_count_varint = varint::signed_varint_size(self.headers.len() as i32);

        let headers_wire: usize = self
            .headers
            .iter()
            .map(|(k, v)| {
                varint::signed_varint_size(k.len() as i32)
                    + k.len()
                    + varint::signed_varint_size(v.len() as i32)
                    + v.len()
            })
            .sum();

        let body_size = 1 // attributes byte
            + TIMESTAMP_DELTA_BYTES
            + OFFSET_DELTA_BYTES
            + key_varint
            + key_bytes
            + val_varint
            + val_bytes
            + hdr_count_varint
            + headers_wire;

        // Record framing: body_size is itself encoded as a signed varint prefix.
        let framing = varint::signed_varint_size(body_size as i32);

        // Amortised batch-level overhead: RecordBatch fixed header (61 bytes),
        // produce-request topic/partition framing (~20 bytes), topic String heap.
        let batch_overhead = self.topic.len() + 64;

        framing + body_size + batch_overhead
    }

    /// Validate that this record's fields do not exceed Kafka wire-format limits.
    ///
    /// Checks:
    /// - Key length fits in `i32` (Kafka bytes encoding limit of 2 GiB)
    /// - Value length fits in `i32`
    /// - Each header key fits in `i32` (record batch v2 uses varint/i32 length prefix)
    /// - Each header value fits in `i32`
    /// - Topic name fits in `i16` (Kafka string encoding limit of 32 KiB)
    pub fn validate(&self) -> Result<()> {
        // Topic name must be non-empty and fit the KafkaString (i16) length prefix.
        // Shared with admin-path ingress so the error message is stable across
        // the client.
        validate_topic_name(&self.topic)?;

        // Key is encoded as KafkaBytes (i32 length prefix)
        if let Some(ref key) = self.key
            && key.len() > i32::MAX as usize
        {
            return Err(KrafkaError::protocol_kind(
                ProtocolErrorKind::InvalidLength,
                format!(
                    "record key length {} exceeds protocol limit of {}",
                    key.len(),
                    i32::MAX
                ),
            ));
        }

        // Value is encoded as KafkaBytes (i32 length prefix)
        if self.value.len() > i32::MAX as usize {
            return Err(KrafkaError::protocol_kind(
                ProtocolErrorKind::InvalidLength,
                format!(
                    "record value length {} exceeds protocol limit of {}",
                    self.value.len(),
                    i32::MAX
                ),
            ));
        }

        // Header keys and values are encoded with varint i32 length prefixes
        // in the record batch v2 format. Limit header count to prevent
        // excessively large batches from bypassing max_request_size.
        if self.headers.len() > MAX_RECORD_HEADERS {
            return Err(KrafkaError::protocol_kind(
                ProtocolErrorKind::InvalidLength,
                format!(
                    "record has {} headers, exceeding limit of {MAX_RECORD_HEADERS}",
                    self.headers.len()
                ),
            ));
        }
        for (i, (key, value)) in self.headers.iter().enumerate() {
            if key.len() > i32::MAX as usize {
                return Err(KrafkaError::protocol_kind(
                    ProtocolErrorKind::InvalidLength,
                    format!(
                        "header[{}] key length {} exceeds protocol limit of {}",
                        i,
                        key.len(),
                        i32::MAX
                    ),
                ));
            }
            if value.len() > i32::MAX as usize {
                return Err(KrafkaError::protocol_kind(
                    ProtocolErrorKind::InvalidLength,
                    format!(
                        "header[{}] value length {} exceeds protocol limit of {}",
                        i,
                        value.len(),
                        i32::MAX
                    ),
                ));
            }
        }

        Ok(())
    }

    /// Split the public record into an interned topic handle and routed payload.
    pub(crate) fn into_routed_parts(self) -> RoutedRecordParts {
        let Self {
            topic,
            partition,
            key,
            value,
            timestamp,
            headers,
            record_name: _,
        } = self;

        RoutedRecordParts {
            topic: Arc::<str>::from(topic),
            partition,
            record: RoutedRecord {
                key,
                value,
                timestamp,
                headers,
            },
        }
    }
}

/// Interned topic handle reused across the producer routing path.
pub(crate) type TopicHandle = Arc<str>;

/// Internal payload retained after partition routing strips the topic.
#[derive(Debug, Clone)]
pub(crate) struct RoutedRecord {
    pub key: Option<Bytes>,
    pub value: Bytes,
    pub timestamp: Option<Timestamp>,
    pub headers: Vec<(String, Bytes)>,
}

impl RoutedRecord {
    #[inline]
    pub(crate) fn key_bytes(&self) -> Option<&[u8]> {
        self.key.as_deref()
    }

    #[inline]
    pub(crate) fn payload_size_bytes(&self) -> u64 {
        self.value.len() as u64 + self.key.as_ref().map(|key| key.len() as u64).unwrap_or(0)
    }

    pub(crate) fn append_to_batch_builder(
        &self,
        batch_builder: RecordBatchBuilder,
    ) -> RecordBatchBuilder {
        if self.headers.is_empty() {
            batch_builder.add_record(self.key.clone(), Some(self.value.clone()))
        } else {
            batch_builder.add_record_with_headers(
                self.key.clone(),
                Some(self.value.clone()),
                self.headers.clone(),
            )
        }
    }
}

/// Internal representation of a routed record after topic extraction.
pub(crate) struct RoutedRecordParts {
    pub topic: TopicHandle,
    pub partition: Option<PartitionId>,
    pub record: RoutedRecord,
}

/// Metadata returned after successfully sending a record.
///
/// When an idempotent producer detects a `DuplicateSequenceNumber` response,
/// it means the broker already committed the batch from a previous attempt.
/// The record is returned as `Ok(RecordMetadata)` with `offset = -1` and
/// `timestamp = -1` to signal deduplication.  Use [`is_deduplicated()`](Self::is_deduplicated)
/// to distinguish this from a normal commit, and [`is_success()`](Self::is_success) to check
/// whether a valid log offset is available.
#[non_exhaustive]
#[must_use = "contains the result of a send operation"]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecordMetadata {
    /// Topic the record was sent to.
    pub topic: String,
    /// Partition the record was sent to.
    pub partition: PartitionId,
    /// Log offset of the committed record, or `-1` when the broker deduplicated
    /// the batch (idempotent `DuplicateSequenceNumber`).  Check
    /// [`is_deduplicated()`](Self::is_deduplicated) before relying on this value.
    pub offset: i64,
    /// Broker-assigned timestamp of the record, or `-1` when deduplicated.
    pub timestamp: Timestamp,
}

impl RecordMetadata {
    /// Returns `true` if the record was committed with a known log offset.
    ///
    /// Returns `false` for deduplicated records (`offset == -1`).  Use
    /// [`is_deduplicated()`](Self::is_deduplicated) to tell those apart.
    #[inline]
    pub fn is_success(&self) -> bool {
        self.offset >= 0
    }

    /// Returns `true` when the broker deduplicated this record.
    ///
    /// An idempotent producer receives `DuplicateSequenceNumber` when the
    /// broker has already committed the batch from an earlier attempt.  The
    /// data **is** in Kafka, but the original log offset is not available;
    /// both `offset` and `timestamp` are set to `-1`.
    #[inline]
    pub fn is_deduplicated(&self) -> bool {
        self.offset == -1
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;

    #[test]
    fn test_producer_record_new() {
        let record = ProducerRecord::new("test-topic", b"hello".to_vec());
        assert_eq!(record.topic, "test-topic");
        assert_eq!(record.value.as_ref(), b"hello");
        assert!(record.key.is_none());
        assert!(record.partition.is_none());
    }

    #[test]
    fn test_producer_record_with_key() {
        let record =
            ProducerRecord::new("test-topic", b"hello".to_vec()).with_key(b"my-key".to_vec());

        assert_eq!(record.key, Some(Bytes::from_static(b"my-key")));
        assert_eq!(record.key_str(), Some("my-key"));
    }

    #[test]
    fn test_producer_record_with_partition() {
        let record = ProducerRecord::new("test-topic", b"hello".to_vec()).with_partition(5);

        assert_eq!(record.partition, Some(5));
    }

    #[test]
    fn test_producer_record_with_headers() {
        let record = ProducerRecord::new("test-topic", b"hello".to_vec())
            .with_header("h1", b"v1".to_vec())
            .with_header("h2", b"v2".to_vec());

        assert_eq!(record.headers.len(), 2);
        assert_eq!(record.headers[0].0, "h1");
        assert_eq!(record.headers[1].0, "h2");
    }

    #[test]
    fn test_producer_record_estimated_size() {
        let record =
            ProducerRecord::new("test-topic", b"hello world".to_vec()).with_key(b"key".to_vec());

        let size = record.estimated_size();
        // Must include at least key + value bytes, the varint framing overhead,
        // and the batch-level header overhead.
        assert!(size > 3 + 11 + 8, "estimated_size={size} too small");

        // Must not be unreasonably large (< 512 for this tiny record).
        assert!(size < 512, "estimated_size={size} unexpectedly large");

        // A record with no key should still estimate correctly.
        let no_key = ProducerRecord::new("test-topic", b"hello world".to_vec());
        let no_key_size = no_key.estimated_size();
        // No key → slightly smaller than with key (only null sentinel varint, no key bytes).
        assert!(no_key_size < size, "no-key estimate should be smaller");

        // A record with headers should be larger than one without.
        let with_headers = ProducerRecord::new("test-topic", b"hello world".to_vec())
            .with_header("h1", b"v1".to_vec())
            .with_header("h2", b"v2".to_vec());
        assert!(
            with_headers.estimated_size() > no_key_size,
            "headers should increase estimate"
        );
    }

    #[test]
    fn test_producer_record_into_routed_parts() {
        let record = ProducerRecord::new("test-topic", b"hello".to_vec())
            .with_partition(2)
            .with_key(b"key".to_vec())
            .with_timestamp(1234)
            .with_header("h1", b"v1".to_vec());

        let routed = record.into_routed_parts();

        assert_eq!(routed.topic.as_ref(), "test-topic");
        assert_eq!(routed.partition, Some(2));
        assert_eq!(routed.record.key, Some(Bytes::from_static(b"key")));
        assert_eq!(routed.record.value, Bytes::from_static(b"hello"));
        assert_eq!(routed.record.timestamp, Some(1234));
        assert_eq!(routed.record.headers.len(), 1);
        assert_eq!(routed.record.headers[0].0, "h1");
    }

    #[test]
    fn test_record_metadata() {
        let metadata = RecordMetadata {
            topic: "test".to_string(),
            partition: 0,
            offset: 42,
            timestamp: 1234567890000,
        };

        assert!(metadata.is_success());
        assert_eq!(metadata.offset, 42);
    }

    #[test]
    fn test_validate_valid_record() {
        let record = ProducerRecord::new("topic", b"value".to_vec())
            .with_key(b"key".to_vec())
            .with_header("h1", b"v1".to_vec());
        assert!(record.validate().is_ok());
    }

    #[test]
    fn test_validate_rejects_oversized_topic() {
        let record = ProducerRecord::new("x".repeat(i16::MAX as usize + 1), b"v".to_vec());
        let err = record.validate().unwrap_err().to_string();
        assert!(err.contains("topic name length"), "unexpected: {err}");
    }

    #[test]
    fn test_validate_accepts_header_key_within_i32_limit() {
        // Header keys use varint i32 length prefix in record batch v2,
        // so i16::MAX + 1 must be accepted (previously rejected).
        let record = ProducerRecord::new("topic", b"v".to_vec())
            .with_header("x".repeat(i16::MAX as usize + 1), b"v".to_vec());
        assert!(record.validate().is_ok());
    }

    #[test]
    fn test_validate_accepts_max_valid_sizes() {
        // Topic name max is 249 bytes (Kafka protocol limit).
        let record = ProducerRecord::new("a".repeat(249), b"v".to_vec());
        assert!(record.validate().is_ok());
    }

    #[test]
    fn test_without_key_clears_key() {
        let record = ProducerRecord::new("topic", b"value".to_vec())
            .with_key("my-key")
            .without_key();
        assert!(record.key.is_none());
    }

    #[test]
    fn test_validate_rejects_empty_topic() {
        let record = ProducerRecord::new("", b"value".to_vec());
        let err = record.validate().unwrap_err().to_string();
        assert!(err.contains("empty"), "unexpected: {err}");
    }

    #[test]
    fn test_record_metadata_equality() {
        let a = RecordMetadata {
            topic: "t".to_string(),
            partition: 0,
            offset: 1,
            timestamp: 100,
        };
        let b = RecordMetadata {
            topic: "t".to_string(),
            partition: 0,
            offset: 1,
            timestamp: 100,
        };
        assert_eq!(a, b);
    }
}