use std::collections::VecDeque;
use std::sync::Arc;

use crate::error::{Error, Result};
use crate::types::{MessageEnvelope, NamespaceId, PublishedMessage, TopicName, TopicOffset};

/// Policy applied when a topic's buffer reaches `max_depth`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum BackpressurePolicy {
    /// Reject the publish with `Error::QueueFull`.
    #[default]
    Reject,
    /// Evict the oldest message to make room.
    DropOldest,
    /// Block until space is available (falls back to `Reject` in v1).
    Block,
}

/// Result of a compaction operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompactResult {
    pub bytes_freed: usize,
    pub messages_freed: usize,
}

/// Observability snapshot of a topic's state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TopicStats {
    pub depth: usize,
    pub bytes_used: usize,
    pub base_offset: TopicOffset,
    pub head_offset: TopicOffset,
    /// Timestamp of the oldest message in the buffer (millis since epoch), or None if empty.
    pub oldest_timestamp: Option<u64>,
    /// Timestamp of the newest message in the buffer (millis since epoch), or None if empty.
    pub newest_timestamp: Option<u64>,
}

/// In-memory indexed log for a single topic.
///
/// All methods are synchronous — async wrapping happens at the engine level.
pub struct Topic {
    buffer: VecDeque<Arc<MessageEnvelope>>,
    next_offset: TopicOffset,
    base_offset: TopicOffset,
    max_depth: usize,
    max_bytes: u64,
    bytes_used: usize,
    backpressure: BackpressurePolicy,
    /// Unflushed payload bytes since the last storage flush. Used by the engine
    /// to trigger early flushes when `batch_flush_size_bytes` is exceeded.
    dirty_bytes: u64,
    /// Fire-once flag for 80% depth warning; resets when depth drops below threshold.
    depth_warning_fired: bool,
    /// Fire-once flag for 80% byte-limit warning; resets when usage drops below threshold.
    bytes_warning_fired: bool,
    /// Cumulative evictions since last warning. Warns every 10,000 evictions.
    evictions_since_warn: usize,
    /// Epoch millis of last publish on this topic. Used by topic inactivity GC.
    last_activity_millis: u64,
}

impl Topic {
    /// Create a new empty topic with the given capacity, byte limit, and backpressure policy.
    pub fn new(max_depth: usize, max_bytes: u64, backpressure: BackpressurePolicy) -> Self {
        let prealloc = max_depth.min(1024);
        Self {
            buffer: VecDeque::with_capacity(prealloc),
            next_offset: TopicOffset::ZERO,
            base_offset: TopicOffset::ZERO,
            max_depth,
            max_bytes,
            bytes_used: 0,
            backpressure,
            dirty_bytes: 0,
            depth_warning_fired: false,
            bytes_warning_fired: false,
            evictions_since_warn: 0,
            last_activity_millis: crate::types::now_millis(),
        }
    }

    /// Reconstruct a topic from replayed persisted messages.
    ///
    /// If the replayed log exceeds `max_depth`, only the most recent
    /// `max_depth` messages are retained. This prevents the buffer from
    /// exceeding capacity after a restart when messages were appended
    /// between compaction cycles.
    pub fn from_replay(
        messages: Vec<PublishedMessage>,
        max_depth: usize,
        max_bytes: u64,
        backpressure: BackpressurePolicy,
    ) -> Self {
        if messages.is_empty() {
            return Self::new(max_depth, max_bytes, backpressure);
        }

        let next_offset = TopicOffset(messages.last().unwrap().offset.0 + 1);

        // Only keep the tail if the replayed log exceeds max_depth.
        let skip = messages.len().saturating_sub(max_depth);
        let retained = &messages[skip..];

        let base_offset = retained[0].offset;
        let mut bytes_used: usize = 0;
        let buffer: VecDeque<Arc<MessageEnvelope>> = retained
            .iter()
            .map(|m| {
                bytes_used += m.envelope.mem_size();
                Arc::clone(&m.envelope)
            })
            .collect();

        Self {
            buffer,
            next_offset,
            base_offset,
            max_depth,
            max_bytes,
            bytes_used,
            backpressure,
            dirty_bytes: 0,
            depth_warning_fired: false,
            bytes_warning_fired: false,
            evictions_since_warn: 0,
            last_activity_millis: retained.last().unwrap().envelope.timestamp,
        }
    }

    /// Publish a message to this topic.
    ///
    /// Returns `(assigned_offset, evicted_bytes, evicted_messages)`.
    /// `evicted_bytes`/`evicted_messages` are non-zero only when `DropOldest`
    /// backpressure evicts messages to make room.
    ///
    /// `ns` and `topic` are only used for error reporting in `QueueFull`.
    pub fn publish(
        &mut self,
        msg: Arc<MessageEnvelope>,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> Result<(TopicOffset, usize, usize)> {
        let mut evicted_bytes = 0;
        let mut evicted_messages = 0;

        // --- 80% depth warning (checked BEFORE eviction, includes the about-to-be-added msg) ---
        let effective_depth = self.buffer.len() + 1; // current + incoming message
        let depth_threshold = self.max_depth * 4 / 5;
        if depth_threshold > 0 && effective_depth >= depth_threshold {
            if !self.depth_warning_fired {
                self.depth_warning_fired = true;
                tracing::warn!(
                    namespace = %ns, topic = %topic,
                    depth = effective_depth, max_depth = self.max_depth,
                    "queue depth at 80%"
                );
            }
        } else {
            self.depth_warning_fired = false;
        }

        // --- 80% byte-limit warning (checked BEFORE eviction, includes incoming message) ---
        let msg_mem_size = msg.mem_size();
        if self.max_bytes > 0 {
            let bytes_after = self.bytes_used + msg_mem_size;
            let bytes_threshold = self.max_bytes as usize * 4 / 5;
            if bytes_after >= bytes_threshold {
                if !self.bytes_warning_fired {
                    self.bytes_warning_fired = true;
                    tracing::warn!(
                        namespace = %ns, topic = %topic,
                        bytes_used = bytes_after,
                        max_bytes = self.max_bytes,
                        "topic byte usage at 80%"
                    );
                }
            } else {
                self.bytes_warning_fired = false;
            }
        }

        if self.buffer.len() >= self.max_depth {
            match self.backpressure {
                BackpressurePolicy::Reject | BackpressurePolicy::Block => {
                    tracing::warn!(
                        namespace = %ns, topic = %topic,
                        depth = self.buffer.len(), max_depth = self.max_depth,
                        policy = ?self.backpressure,
                        "backpressure engaged: queue depth limit reached"
                    );
                    metrics::counter!("fafb_publish_rejected_total",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "reason" => "queue_full"
                    )
                    .increment(1);
                    return Err(Error::QueueFull {
                        namespace: ns.clone(),
                        topic: topic.clone(),
                    });
                }
                BackpressurePolicy::DropOldest => {
                    if let Some(evicted) = self.buffer.pop_front() {
                        evicted_bytes = evicted.mem_size();
                        self.bytes_used -= evicted_bytes;
                        self.base_offset = TopicOffset(self.base_offset.0 + 1);
                        evicted_messages += 1;
                    }
                }
            }
        }

        if self.max_bytes > 0 && self.bytes_used + msg_mem_size > self.max_bytes as usize {
            match self.backpressure {
                BackpressurePolicy::Reject | BackpressurePolicy::Block => {
                    tracing::warn!(
                        namespace = %ns, topic = %topic,
                        bytes_used = self.bytes_used, msg_size = msg_mem_size,
                        max_bytes = self.max_bytes, policy = ?self.backpressure,
                        "backpressure engaged: topic byte limit reached"
                    );
                    metrics::counter!("fafb_publish_rejected_total",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "reason" => "topic_bytes_exceeded"
                    )
                    .increment(1);
                    return Err(Error::ResourceExhausted(format!(
                        "topic byte limit exceeded: {} + {} > {}",
                        self.bytes_used, msg_mem_size, self.max_bytes
                    )));
                }
                BackpressurePolicy::DropOldest => {
                    while self.bytes_used + msg_mem_size > self.max_bytes as usize
                        && !self.buffer.is_empty()
                    {
                        if let Some(evicted) = self.buffer.pop_front() {
                            let evicted_size = evicted.mem_size();
                            evicted_bytes += evicted_size;
                            self.bytes_used -= evicted_size;
                            self.base_offset = TopicOffset(self.base_offset.0 + 1);
                            evicted_messages += 1;
                        }
                    }
                }
            }
        }

        // --- DropOldest eviction warning: log every 10,000 cumulative evictions ---
        if evicted_messages > 0 {
            self.evictions_since_warn += evicted_messages;
            if self.evictions_since_warn >= 10_000 {
                tracing::warn!(
                    namespace = %ns, topic = %topic,
                    evictions_since_last_warn = self.evictions_since_warn,
                    "DropOldest eviction: {0} messages evicted since last warning",
                    self.evictions_since_warn,
                );
                self.evictions_since_warn = 0;
            }
        }

        let offset = self.next_offset;
        self.bytes_used += msg_mem_size;
        self.dirty_bytes += msg_mem_size as u64;
        self.buffer.push_back(msg);
        self.next_offset = offset.next();
        self.last_activity_millis = crate::types::now_millis();
        Ok((offset, evicted_bytes, evicted_messages))
    }

    /// Read the message at the given offset. Stamps `last_activity_millis` on
    /// a successful read so that fetch-only topics (which never create
    /// `TopicDeliveryState`) are still kept alive by the inactivity GC.
    ///
    /// Returns `None` if the offset has been compacted away or is past the head.
    pub fn consume_from(&mut self, offset: TopicOffset) -> Option<Arc<MessageEnvelope>> {
        if offset < self.base_offset || offset >= self.next_offset {
            return None;
        }
        let index = (offset.0 - self.base_offset.0) as usize;
        let result = self.buffer.get(index).cloned();
        if result.is_some() {
            self.last_activity_millis = crate::types::now_millis();
        }
        result
    }

    /// Remove all messages with offsets below `min_offset`.
    ///
    /// Returns the bytes and messages freed.
    pub fn compact(&mut self, min_offset: TopicOffset) -> CompactResult {
        if min_offset <= self.base_offset {
            return CompactResult {
                bytes_freed: 0,
                messages_freed: 0,
            };
        }

        let to_drain = ((min_offset.0 - self.base_offset.0) as usize).min(self.buffer.len());
        let mut bytes_freed = 0;
        for _ in 0..to_drain {
            if let Some(evicted) = self.buffer.pop_front() {
                bytes_freed += evicted.mem_size();
            }
        }
        self.bytes_used -= bytes_freed;
        self.base_offset = min_offset;
        CompactResult {
            bytes_freed,
            messages_freed: to_drain,
        }
    }

    /// Number of messages currently in the buffer.
    pub fn depth(&self) -> usize {
        self.buffer.len()
    }

    /// Total payload bytes currently buffered.
    pub fn bytes_used(&self) -> usize {
        self.bytes_used
    }

    /// Last activity timestamp (epoch millis). Updated on publish or successful consume.
    pub fn last_activity_millis(&self) -> u64 {
        self.last_activity_millis
    }

    /// Unflushed bytes accumulated since the last flush. Returns the current
    /// value and resets it to zero.
    pub fn take_dirty_bytes(&mut self) -> u64 {
        std::mem::take(&mut self.dirty_bytes)
    }

    /// Current unflushed bytes (read-only, does not reset).
    pub fn dirty_bytes(&self) -> u64 {
        self.dirty_bytes
    }

    /// The next offset to be assigned (one past the last written message).
    pub fn head_offset(&self) -> TopicOffset {
        self.next_offset
    }

    /// Publish a message at a specific pre-assigned offset (cluster replication).
    ///
    /// Atomically sets `next_offset` to `offset`, appends the message, and
    /// advances `next_offset` to `offset + 1`. Handles backpressure and
    /// eviction identically to [`publish`].
    ///
    /// The caller must ensure `offset >= head_offset()`. Duplicate offsets
    /// (below head) should be filtered before calling this method.
    #[cfg(feature = "cluster")]
    pub fn publish_at_offset(
        &mut self,
        msg: Arc<MessageEnvelope>,
        ns: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
    ) -> Result<(TopicOffset, usize, usize)> {
        self.next_offset = offset;
        self.publish(msg, ns, topic)
    }

    /// The lowest retained offset after compaction.
    pub fn base_offset(&self) -> TopicOffset {
        self.base_offset
    }

    /// Observability snapshot.
    pub fn stats(&self) -> TopicStats {
        TopicStats {
            depth: self.buffer.len(),
            bytes_used: self.bytes_used,
            base_offset: self.base_offset,
            head_offset: self.next_offset,
            oldest_timestamp: self.buffer.front().map(|e| e.timestamp),
            newest_timestamp: self.buffer.back().map(|e| e.timestamp),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::{MessageId, PublishedMessage, now_millis};

    fn make_envelope(id: u64, payload: &[u8]) -> MessageEnvelope {
        MessageEnvelope::new(
            MessageId(id + 1),
            NamespaceId("test-ns".into()),
            TopicName("test-topic".into()),
            payload.to_vec(),
            now_millis(),
        )
    }

    fn ns() -> NamespaceId {
        NamespaceId("test-ns".into())
    }

    fn topic_name() -> TopicName {
        TopicName("test-topic".into())
    }

    #[test]
    fn publish_assigns_sequential_offsets() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..5u64 {
            let (offset, evicted, _) = t
                .publish(Arc::new(make_envelope(i, b"data")), &ns(), &topic_name())
                .unwrap();
            assert_eq!(offset, TopicOffset(i));
            assert_eq!(evicted, 0);
        }
    }

    #[test]
    fn consume_from_returns_correct_message() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..3u64 {
            t.publish(
                Arc::new(make_envelope(i, format!("msg-{i}").as_bytes())),
                &ns(),
                &topic_name(),
            )
            .unwrap();
        }
        for i in 0..3u64 {
            let env = t.consume_from(TopicOffset(i)).unwrap();
            assert_eq!(env.payload, format!("msg-{i}").as_bytes());
        }
    }

    #[test]
    fn consume_from_past_head_returns_none() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        t.publish(Arc::new(make_envelope(0, b"a")), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, b"b")), &ns(), &topic_name())
            .unwrap();
        assert!(t.consume_from(TopicOffset(2)).is_none());
    }

    #[test]
    fn consume_from_before_base_returns_none() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..5u64 {
            t.publish(Arc::new(make_envelope(i, b"x")), &ns(), &topic_name())
                .unwrap();
        }
        t.compact(TopicOffset(3));
        assert!(t.consume_from(TopicOffset(1)).is_none());
    }

    #[test]
    fn consume_from_after_compaction_valid_offset() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..5u64 {
            t.publish(
                Arc::new(make_envelope(i, format!("msg-{i}").as_bytes())),
                &ns(),
                &topic_name(),
            )
            .unwrap();
        }
        t.compact(TopicOffset(2));
        let env = t.consume_from(TopicOffset(3)).unwrap();
        assert_eq!(env.payload, b"msg-3");
    }

    #[test]
    fn compact_updates_base_offset() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..5u64 {
            t.publish(Arc::new(make_envelope(i, b"x")), &ns(), &topic_name())
                .unwrap();
        }
        t.compact(TopicOffset(3));
        assert_eq!(t.base_offset(), TopicOffset(3));
    }

    #[test]
    fn compact_updates_bytes_used() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        // 3 messages with 10-byte payloads
        for i in 0..3u64 {
            t.publish(Arc::new(make_envelope(i, &[0u8; 10])), &ns(), &topic_name())
                .unwrap();
        }
        assert_eq!(t.bytes_used(), 30);
        let result = t.compact(TopicOffset(2));
        assert_eq!(result.bytes_freed, 20);
        assert_eq!(result.messages_freed, 2);
        assert_eq!(t.bytes_used(), 10);
    }

    #[test]
    fn compact_noop_below_base() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..3u64 {
            t.publish(Arc::new(make_envelope(i, b"x")), &ns(), &topic_name())
                .unwrap();
        }
        t.compact(TopicOffset(2));
        let result = t.compact(TopicOffset(1));
        assert_eq!(result.bytes_freed, 0);
        assert_eq!(result.messages_freed, 0);
        assert_eq!(t.base_offset(), TopicOffset(2));
    }

    #[test]
    fn backpressure_reject_returns_queue_full() {
        let mut t = Topic::new(2, 0, BackpressurePolicy::Reject);
        t.publish(Arc::new(make_envelope(0, b"a")), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, b"b")), &ns(), &topic_name())
            .unwrap();
        let err = t
            .publish(Arc::new(make_envelope(2, b"c")), &ns(), &topic_name())
            .unwrap_err();
        assert!(matches!(err, Error::QueueFull { .. }));
    }

    #[test]
    fn backpressure_drop_oldest_evicts() {
        let mut t = Topic::new(2, 0, BackpressurePolicy::DropOldest);
        t.publish(Arc::new(make_envelope(0, b"a")), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, b"b")), &ns(), &topic_name())
            .unwrap();
        let (offset, evicted, _) = t
            .publish(Arc::new(make_envelope(2, b"c")), &ns(), &topic_name())
            .unwrap();
        assert_eq!(offset, TopicOffset(2));
        assert_eq!(evicted, 1); // "a" was 1 byte
        assert_eq!(t.depth(), 2);
        assert_eq!(t.base_offset(), TopicOffset(1));
        // Oldest message (offset 0) is gone
        assert!(t.consume_from(TopicOffset(0)).is_none());
        // Newer messages remain
        assert_eq!(t.consume_from(TopicOffset(1)).unwrap().payload, b"b");
        assert_eq!(t.consume_from(TopicOffset(2)).unwrap().payload, b"c");
    }

    #[test]
    fn block_falls_back_to_reject() {
        let mut t = Topic::new(2, 0, BackpressurePolicy::Block);
        t.publish(Arc::new(make_envelope(0, b"a")), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, b"b")), &ns(), &topic_name())
            .unwrap();
        let err = t
            .publish(Arc::new(make_envelope(2, b"c")), &ns(), &topic_name())
            .unwrap_err();
        assert!(matches!(err, Error::QueueFull { .. }));
    }

    #[test]
    fn bytes_used_tracks_payloads() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        t.publish(Arc::new(make_envelope(0, &[0u8; 5])), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, &[0u8; 15])), &ns(), &topic_name())
            .unwrap();
        assert_eq!(t.bytes_used(), 20);
    }

    #[test]
    fn depth_after_publish_and_compact() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        for i in 0..10u64 {
            t.publish(Arc::new(make_envelope(i, b"x")), &ns(), &topic_name())
                .unwrap();
        }
        assert_eq!(t.depth(), 10);
        t.compact(TopicOffset(4));
        assert_eq!(t.depth(), 6);
    }

    #[test]
    fn from_replay_restores_state() {
        let messages: Vec<PublishedMessage> = (0..5u64)
            .map(|i| PublishedMessage {
                envelope: Arc::new(make_envelope(i, format!("msg-{i}").as_bytes())),
                offset: TopicOffset(i),
            })
            .collect();

        let mut t = Topic::from_replay(messages, 100, 0, BackpressurePolicy::Reject);
        assert_eq!(t.depth(), 5);
        assert_eq!(t.base_offset(), TopicOffset(0));
        assert_eq!(t.head_offset(), TopicOffset(5));
        assert_eq!(
            t.bytes_used(),
            (0..5).map(|i| format!("msg-{i}").len()).sum::<usize>()
        );
        for i in 0..5u64 {
            let env = t.consume_from(TopicOffset(i)).unwrap();
            assert_eq!(env.payload, format!("msg-{i}").as_bytes());
        }
    }

    #[test]
    fn from_replay_empty() {
        let t = Topic::from_replay(vec![], 100, 0, BackpressurePolicy::Reject);
        assert_eq!(t.depth(), 0);
        assert_eq!(t.base_offset(), TopicOffset::ZERO);
        assert_eq!(t.head_offset(), TopicOffset::ZERO);
        assert_eq!(t.bytes_used(), 0);
    }

    #[test]
    fn from_replay_truncates_to_max_depth() {
        // Replay 10 messages into a topic with max_depth=3.
        // Only the last 3 should be retained.
        let messages: Vec<PublishedMessage> = (0..10u64)
            .map(|i| PublishedMessage {
                envelope: Arc::new(make_envelope(i, format!("msg-{i}").as_bytes())),
                offset: TopicOffset(i),
            })
            .collect();

        let mut t = Topic::from_replay(messages, 3, 0, BackpressurePolicy::Reject);
        assert_eq!(t.depth(), 3);
        assert_eq!(t.base_offset(), TopicOffset(7));
        assert_eq!(t.head_offset(), TopicOffset(10));
        // Older messages are gone
        assert!(t.consume_from(TopicOffset(6)).is_none());
        // Retained messages are correct
        for i in 7..10u64 {
            let env = t.consume_from(TopicOffset(i)).unwrap();
            assert_eq!(env.payload, format!("msg-{i}").as_bytes());
        }
    }

    #[test]
    fn drop_oldest_returns_evicted_bytes() {
        let mut t = Topic::new(2, 0, BackpressurePolicy::DropOldest);
        t.publish(Arc::new(make_envelope(0, &[0u8; 10])), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, &[0u8; 20])), &ns(), &topic_name())
            .unwrap();
        // Third publish evicts the 10-byte message
        let (_, evicted, evicted_count) = t
            .publish(Arc::new(make_envelope(2, &[0u8; 30])), &ns(), &topic_name())
            .unwrap();
        assert_eq!(evicted, 10);
        assert_eq!(evicted_count, 1);
        assert_eq!(t.bytes_used(), 50); // 20 + 30
    }

    #[test]
    fn byte_limit_reject_policy() {
        // max_bytes = 50, each message is 20 bytes
        let mut t = Topic::new(100, 50, BackpressurePolicy::Reject);
        t.publish(Arc::new(make_envelope(0, &[0u8; 20])), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, &[0u8; 20])), &ns(), &topic_name())
            .unwrap();
        // Third message (20 bytes) would exceed limit: 40 + 20 > 50
        let err = t
            .publish(Arc::new(make_envelope(2, &[0u8; 20])), &ns(), &topic_name())
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("topic byte limit exceeded"));
    }

    #[test]
    fn byte_limit_drop_oldest_evicts_multiple() {
        // max_bytes = 50, publish two 20-byte messages, then one 40-byte message
        let mut t = Topic::new(100, 50, BackpressurePolicy::DropOldest);
        t.publish(Arc::new(make_envelope(0, &[0u8; 20])), &ns(), &topic_name())
            .unwrap();
        t.publish(Arc::new(make_envelope(1, &[0u8; 20])), &ns(), &topic_name())
            .unwrap();
        assert_eq!(t.bytes_used(), 40);

        // Publishing a 40-byte message: 40 + 40 = 80 > 50
        // Must evict both existing messages to fit
        let (offset, evicted, _) = t
            .publish(Arc::new(make_envelope(2, &[0u8; 40])), &ns(), &topic_name())
            .unwrap();
        assert_eq!(offset, TopicOffset(2));
        assert_eq!(evicted, 40); // evicted 20 + 20
        assert_eq!(t.bytes_used(), 40);
        assert_eq!(t.depth(), 1);
        assert_eq!(t.base_offset(), TopicOffset(2));
    }

    #[test]
    fn byte_limit_zero_means_unlimited() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        // Should be able to publish large messages with max_bytes = 0
        for i in 0..10u64 {
            t.publish(
                Arc::new(make_envelope(i, &[0u8; 1000])),
                &ns(),
                &topic_name(),
            )
            .unwrap();
        }
        assert_eq!(t.bytes_used(), 10_000);
    }

    #[test]
    fn byte_limit_and_depth_limit_interact() {
        // Depth limit of 5, byte limit of 100 (20 bytes each = 100 bytes for 5 msgs)
        let mut t = Topic::new(5, 100, BackpressurePolicy::Reject);
        for i in 0..5u64 {
            t.publish(Arc::new(make_envelope(i, &[0u8; 20])), &ns(), &topic_name())
                .unwrap();
        }
        // Buffer is full (depth = 5, bytes = 100)
        let err = t
            .publish(Arc::new(make_envelope(5, &[0u8; 20])), &ns(), &topic_name())
            .unwrap_err();
        // Depth limit is checked first
        assert!(matches!(err, Error::QueueFull { .. }));
    }

    #[test]
    fn new_topic_has_recent_last_activity() {
        let t = Topic::new(100, 0, BackpressurePolicy::Reject);
        let now = crate::types::now_millis();
        assert!(t.last_activity_millis() <= now);
        assert!(t.last_activity_millis() >= now - 1000);
    }

    #[test]
    fn publish_updates_last_activity() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        let before = t.last_activity_millis();
        std::thread::sleep(std::time::Duration::from_millis(5));
        t.publish(Arc::new(make_envelope(0, b"data")), &ns(), &topic_name())
            .unwrap();
        assert!(t.last_activity_millis() >= before);
    }

    #[test]
    fn consume_from_updates_last_activity() {
        let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
        t.publish(Arc::new(make_envelope(0, b"data")), &ns(), &topic_name())
            .unwrap();
        let before = t.last_activity_millis();
        std::thread::sleep(std::time::Duration::from_millis(5));
        let _ = t.consume_from(TopicOffset(0));
        assert!(t.last_activity_millis() > before);
    }

    #[test]
    fn from_replay_sets_last_activity_from_newest_message() {
        use crate::types::{MessageId, PublishedMessage};
        let messages: Vec<PublishedMessage> = (0..3u64)
            .map(|i| PublishedMessage {
                offset: TopicOffset(i),
                envelope: Arc::new(MessageEnvelope {
                    id: MessageId::new(i + 1),
                    namespace: ns(),
                    topic: topic_name(),
                    timestamp: 1000 + i * 100,
                    payload: b"data".to_vec(),
                    headers: vec![],
                    ttl_ms: 0,
                }),
            })
            .collect();
        let t = Topic::from_replay(messages, 100, 0, BackpressurePolicy::Reject);
        assert_eq!(t.last_activity_millis(), 1200);
    }

    #[test]
    fn from_replay_empty_uses_now() {
        let t = Topic::from_replay(vec![], 100, 0, BackpressurePolicy::Reject);
        let now = crate::types::now_millis();
        assert!(t.last_activity_millis() <= now);
        assert!(t.last_activity_millis() >= now - 1000);
    }

    #[test]
    fn from_replay_truncated_uses_newest_retained() {
        use crate::types::{MessageId, PublishedMessage};
        let messages: Vec<PublishedMessage> = (0..5u64)
            .map(|i| PublishedMessage {
                offset: TopicOffset(i),
                envelope: Arc::new(MessageEnvelope {
                    id: MessageId::new(i + 1),
                    namespace: ns(),
                    topic: topic_name(),
                    timestamp: 1000 + i * 100,
                    payload: b"data".to_vec(),
                    headers: vec![],
                    ttl_ms: 0,
                }),
            })
            .collect();
        let t = Topic::from_replay(messages, 2, 0, BackpressurePolicy::Reject);
        assert_eq!(t.last_activity_millis(), 1400);
    }
}
