use std::fmt;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use std::time::SystemTime;

use serde::{Deserialize, Serialize};

/// Global clock offset in milliseconds, set by the cluster time-sync module.
/// In non-cluster mode this stays 0 — `now_millis()` behaves identically.
static CLOCK_OFFSET_MS: AtomicI64 = AtomicI64::new(0);

/// Set the global clock offset (called by the cluster NTP sync module).
pub fn set_clock_offset(offset_ms: i64) {
    CLOCK_OFFSET_MS.store(offset_ms, Ordering::Relaxed);
}

/// Returns the current time as milliseconds since Unix epoch,
/// adjusted by the cluster clock offset.
pub fn now_millis() -> u64 {
    let raw = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as i64;
    (raw + CLOCK_OFFSET_MS.load(Ordering::Relaxed)) as u64
}

/// Format milliseconds since Unix epoch as an RFC3339 UTC string.
///
/// Output format: `YYYY-MM-DDTHH:MM:SSZ` (always UTC, second precision).
/// Uses Howard Hinnant's `civil_from_days` algorithm — zero external deps.
pub fn millis_to_rfc3339(millis: u64) -> String {
    let total_secs = millis / 1000;
    let secs_in_day = total_secs % 86400;
    let mut days = total_secs / 86400;

    let hour = secs_in_day / 3600;
    let minute = (secs_in_day % 3600) / 60;
    let second = secs_in_day % 60;

    // Convert days since epoch to year/month/day.
    days += 719_468;
    let era = days / 146_097;
    let doe = days - era * 146_097;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };

    format!("{y:04}-{m:02}-{d:02}T{hour:02}:{minute:02}:{second:02}Z")
}

use crate::error::Error;

/// Unique identifier for a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct NamespaceId(pub String);

impl fmt::Display for NamespaceId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for NamespaceId {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.is_empty() {
            return Err(Error::Config("namespace id cannot be empty".into()));
        }
        Ok(Self(s.to_owned()))
    }
}

/// Name of a queue topic within a namespace.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct TopicName(pub String);

impl fmt::Display for TopicName {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for TopicName {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.is_empty() {
            return Err(Error::Config("topic name cannot be empty".into()));
        }
        Ok(Self(s.to_owned()))
    }
}

impl TopicName {
    /// Validate that this topic name is acceptable for publishing.
    ///
    /// Publish-side rules:
    /// - Must not be empty
    /// - Must not exceed 255 bytes
    /// - Must not contain wildcard characters (`*` or `>`)
    /// - Must not contain empty segments (double dots `..`, leading `.`, trailing `.`)
    pub fn validate_for_publish(&self) -> Result<(), Error> {
        let s = &self.0;
        if s.is_empty() {
            return Err(Error::InvalidTopic("topic name cannot be empty".into()));
        }
        if s.len() > 255 {
            return Err(Error::InvalidTopic(
                "topic name exceeds 255 byte limit".into(),
            ));
        }
        if s.contains('*') || s.contains('>') {
            return Err(Error::InvalidTopic(
                "topic name cannot contain wildcard characters '*' or '>'".into(),
            ));
        }
        // Check for empty segments: leading dot, trailing dot, or double dot.
        if s.starts_with('.') || s.ends_with('.') || s.contains("..") {
            return Err(Error::InvalidTopic(
                "topic name contains empty segment (double dot)".into(),
            ));
        }
        Ok(())
    }
}

/// API key used for authentication.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ApiKey(pub String);

impl fmt::Display for ApiKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for ApiKey {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.is_empty() {
            return Err(Error::Config("api key cannot be empty".into()));
        }
        Ok(Self(s.to_owned()))
    }
}

/// Unique identifier for a message, used for ack tracking.
///
/// MessageId values must be ≥ 1. The value 0 is reserved as a sentinel byte
/// in the on-disk segment format (v2) to distinguish record versions.
/// Use [`MessageId::new()`] in production code to enforce this invariant.
/// The raw tuple constructor `MessageId(val)` is kept public for serde
/// deserialization and wire-protocol decoding (where panicking on bad input
/// is not appropriate).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct MessageId(pub u64);

impl MessageId {
    /// Create a new MessageId, panicking if the value is 0.
    ///
    /// Use this constructor in server-side code that generates message IDs
    /// (publish paths). For IDs decoded from the wire or deserialized from
    /// storage, use the tuple constructor directly.
    pub fn new(id: u64) -> Self {
        assert!(
            id != 0,
            "MessageId must be non-zero (0x00 is reserved as a storage format sentinel)"
        );
        Self(id)
    }
}

impl fmt::Display for MessageId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Unique identifier for a connected client.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct ClientId(pub u64);

impl fmt::Display for ClientId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Per-topic sequential offset within the indexed log.
///
/// Distinct from [`MessageId`]: a `MessageId` identifies a message globally
/// for ack tracking, while a `TopicOffset` is the position of a message
/// within a specific topic's log.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct TopicOffset(pub u64);

impl TopicOffset {
    /// The zero offset — first position in a topic log, or "no messages consumed yet".
    pub const ZERO: Self = Self(0);

    /// Returns the next sequential offset.
    pub fn next(self) -> Self {
        Self(self.0 + 1)
    }
}

impl fmt::Display for TopicOffset {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for TopicOffset {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let n: u64 = s
            .parse()
            .map_err(|e| Error::Config(format!("invalid topic offset: {e}")))?;
        Ok(Self(n))
    }
}

impl From<u64> for TopicOffset {
    fn from(n: u64) -> Self {
        Self(n)
    }
}

impl From<TopicOffset> for u64 {
    fn from(offset: TopicOffset) -> Self {
        offset.0
    }
}

/// Name of a consumer group within a topic.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct GroupName(pub String);

impl fmt::Display for GroupName {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl FromStr for GroupName {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.is_empty() {
            return Err(Error::Config("group name cannot be empty".into()));
        }
        Ok(Self(s.to_owned()))
    }
}

/// Combined queue and subscriber statistics for a topic.
///
/// Used by the HTTP API's info endpoint to provide a snapshot of topic state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TopicInfo {
    pub depth: usize,
    pub bytes_used: usize,
    pub base_offset: TopicOffset,
    pub head_offset: TopicOffset,
    pub subscriber_count: usize,
    pub consumer_group_count: usize,
    /// Timestamp of the oldest buffered message (millis since epoch), or None if empty.
    pub oldest_timestamp_millis: Option<u64>,
    /// Timestamp of the newest buffered message (millis since epoch), or None if empty.
    pub newest_timestamp_millis: Option<u64>,
}

/// Specifies where a new subscription should begin consuming.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubscribeStart {
    /// Start from the current head offset (live tail).
    Head,
    /// Start from a specific offset.
    Offset(TopicOffset),
    /// Start from the first message at or after a timestamp (millis since epoch).
    /// Used for wildcard subscriptions where per-topic timestamp resolution
    /// happens in the namespace manager rather than the session layer.
    Timestamp(u64),
}

/// Permissions that can be granted to an API key for a namespace.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum Permission {
    Subscribe,
    Publish,
    Query,
    Fetch,
}

/// Result of a `NamespaceManager::subscribe()` call.
///
/// Bundles the delivery channel, actual starting offset, and wildcard metadata
/// so the session layer doesn't need to re-derive information the manager
/// already computed.
#[derive(Debug)]
pub struct SubscribeResult {
    /// Channel for delivered messages.
    pub rx: tokio::sync::mpsc::Receiver<PublishedMessage>,
    /// Actual starting offset (may differ from requested due to clamping).
    pub start_offset: TopicOffset,
    /// Number of concrete topics matched (wildcard subscriptions only; 0 for exact).
    pub matched_count: u16,
}

/// Result of a stateless Fetch operation.
///
/// Returned by `NamespaceManager::fetch()`. The session layer converts this
/// into a `FetchResponse` frame, optionally preceded by an `OffsetNotAvailable`
/// warning when `clamped_from` is `Some`.
#[derive(Debug)]
pub struct FetchResult {
    pub messages: Vec<PublishedMessage>,
    pub next_offset: TopicOffset,
    pub head_offset: TopicOffset,
    /// Set when the requested offset was below the oldest available and was
    /// clamped forward. The session layer uses this to send an
    /// `OffsetNotAvailable` warning before the `FetchResponse`.
    pub clamped_from: Option<TopicOffset>,
}

/// A single message header — key-value pair of byte strings.
///
/// Both key and value are opaque byte strings. The server does not interpret
/// header contents. Keys starting with `_fafb.` are reserved for server-set
/// system headers.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Header {
    pub key: Vec<u8>,
    pub value: Vec<u8>,
}

/// Ordered list of headers on a message.
///
/// Preserves insertion order. Duplicate keys are allowed
/// (same semantics as HTTP headers).
pub type Headers = Vec<Header>;

/// Envelope wrapping a message as it flows through the system.
///
/// Represents a message before it has been assigned a position in a topic log.
/// After publishing, use [`PublishedMessage`] which pairs an envelope with its
/// assigned [`TopicOffset`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEnvelope {
    pub id: MessageId,
    pub namespace: NamespaceId,
    pub topic: TopicName,
    pub payload: Vec<u8>,
    /// Publish timestamp — milliseconds since Unix epoch.
    pub timestamp: u64,
    /// Optional key-value headers attached by the publisher.
    /// Empty vec means no headers (zero overhead in legacy path).
    pub headers: Headers,
    /// Per-message time-to-live in milliseconds. 0 = no expiry (never expires).
    /// Expired messages are skipped during delivery and fetch but remain on disk
    /// until segment retention cleans them up.
    pub ttl_ms: u32,
}

impl MessageEnvelope {
    /// Create a new envelope with default headers (empty) and TTL (0 = no expiry).
    pub fn new(
        id: MessageId,
        namespace: NamespaceId,
        topic: TopicName,
        payload: Vec<u8>,
        timestamp: u64,
    ) -> Self {
        Self {
            id,
            namespace,
            topic,
            payload,
            timestamp,
            headers: vec![],
            ttl_ms: 0,
        }
    }

    /// Set headers on this envelope (builder-style).
    pub fn with_headers(mut self, headers: Headers) -> Self {
        self.headers = headers;
        self
    }

    /// Set TTL in milliseconds on this envelope (builder-style).
    pub fn with_ttl(mut self, ttl_ms: u32) -> Self {
        self.ttl_ms = ttl_ms;
        self
    }

    /// Total bytes consumed by headers (sum of all key and value lengths).
    ///
    /// Used for memory accounting in the in-memory buffer and fetch budget
    /// calculations. Does not include per-header wire overhead — that is
    /// accounted for separately in the codec layer.
    pub fn header_bytes(&self) -> usize {
        self.headers
            .iter()
            .map(|h| h.key.len() + h.value.len())
            .sum()
    }

    /// In-memory size of this message: payload + header data bytes.
    ///
    /// Used for memory budget tracking in the queue engine and topic buffer.
    pub fn mem_size(&self) -> usize {
        self.payload.len() + self.header_bytes()
    }

    /// Check if this message has expired relative to the given timestamp.
    ///
    /// Returns `true` when `ttl_ms > 0` and the message's expiry time
    /// (`timestamp + ttl_ms`) has passed. Messages with `ttl_ms == 0`
    /// never expire.
    pub fn is_expired(&self, now_ms: u64) -> bool {
        self.ttl_ms > 0 && self.timestamp + (self.ttl_ms as u64) <= now_ms
    }

    /// Remaining TTL in milliseconds, or `None` if no TTL set or already expired.
    pub fn remaining_ttl(&self, now_ms: u64) -> Option<u64> {
        if self.ttl_ms == 0 {
            return None;
        }
        let expiry = self.timestamp + (self.ttl_ms as u64);
        if expiry <= now_ms {
            None
        } else {
            Some(expiry - now_ms)
        }
    }
}

/// A message that has been written to a topic log and assigned an offset.
///
/// This type makes it impossible to reference a published message without its
/// offset — no `Option` unwrapping needed in post-publish code paths.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PublishedMessage {
    pub envelope: Arc<MessageEnvelope>,
    pub offset: TopicOffset,
}

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

    #[test]
    fn namespace_id_display_roundtrip() {
        let id = NamespaceId("my-namespace".into());
        let s = id.to_string();
        let parsed: NamespaceId = s.parse().unwrap();
        assert_eq!(id, parsed);
    }

    #[test]
    fn namespace_id_empty_rejected() {
        assert!("".parse::<NamespaceId>().is_err());
    }

    #[test]
    fn topic_name_display_roundtrip() {
        let name = TopicName("events.user.created".into());
        let s = name.to_string();
        let parsed: TopicName = s.parse().unwrap();
        assert_eq!(name, parsed);
    }

    #[test]
    fn topic_name_empty_rejected() {
        assert!("".parse::<TopicName>().is_err());
    }

    #[test]
    fn api_key_display_roundtrip() {
        let key = ApiKey("sk-abc123".into());
        let s = key.to_string();
        let parsed: ApiKey = s.parse().unwrap();
        assert_eq!(key, parsed);
    }

    #[test]
    fn api_key_empty_rejected() {
        assert!("".parse::<ApiKey>().is_err());
    }

    #[test]
    fn message_id_display() {
        let id = MessageId(42);
        assert_eq!(id.to_string(), "42");
    }

    #[test]
    fn message_id_new_valid() {
        let id = MessageId::new(1);
        assert_eq!(id.0, 1);
        let id = MessageId::new(u64::MAX);
        assert_eq!(id.0, u64::MAX);
    }

    #[test]
    #[should_panic(expected = "MessageId must be non-zero")]
    fn message_id_new_rejects_zero() {
        MessageId::new(0);
    }

    #[test]
    fn client_id_display() {
        let id = ClientId(7);
        assert_eq!(id.to_string(), "7");
    }

    #[test]
    fn group_name_display_roundtrip() {
        let name = GroupName("my-workers".into());
        let s = name.to_string();
        let parsed: GroupName = s.parse().unwrap();
        assert_eq!(name, parsed);
    }

    #[test]
    fn group_name_empty_rejected() {
        assert!("".parse::<GroupName>().is_err());
    }

    #[test]
    fn message_envelope_fields() {
        let envelope = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![1, 2, 3],
            now_millis(),
        );
        assert_eq!(envelope.id, MessageId(1));
        assert_eq!(envelope.payload, vec![1, 2, 3]);
    }

    #[test]
    fn topic_offset_display() {
        let offset = TopicOffset(100);
        assert_eq!(offset.to_string(), "100");
    }

    #[test]
    fn topic_offset_ordering() {
        let a = TopicOffset(1);
        let b = TopicOffset(2);
        let c = TopicOffset(2);
        assert!(a < b);
        assert!(b > a);
        assert_eq!(b, c);
        assert!(a <= b);
        assert!(b >= a);
    }

    #[test]
    fn topic_offset_is_copy() {
        let a = TopicOffset(42);
        let b = a; // Copy
        assert_eq!(a, b); // `a` still usable after move — proves Copy
    }

    #[test]
    fn topic_offset_from_str() {
        let offset: TopicOffset = "42".parse().unwrap();
        assert_eq!(offset, TopicOffset(42));
        assert!("".parse::<TopicOffset>().is_err());
        assert!("not_a_number".parse::<TopicOffset>().is_err());
    }

    #[test]
    fn topic_offset_from_u64() {
        let offset = TopicOffset::from(10u64);
        assert_eq!(offset, TopicOffset(10));
        let n: u64 = offset.into();
        assert_eq!(n, 10);
    }

    #[test]
    fn topic_offset_zero_and_next() {
        assert_eq!(TopicOffset::ZERO, TopicOffset(0));
        assert_eq!(TopicOffset::ZERO.next(), TopicOffset(1));
        assert_eq!(TopicOffset(99).next(), TopicOffset(100));
    }

    #[test]
    fn published_message_has_offset() {
        let published = PublishedMessage {
            envelope: Arc::new(MessageEnvelope::new(
                MessageId(5),
                NamespaceId("ns".into()),
                TopicName("t".into()),
                vec![10, 20],
                now_millis(),
            )),
            offset: TopicOffset(99),
        };
        assert_eq!(published.offset, TopicOffset(99));
        assert_eq!(published.envelope.id, MessageId(5));
    }

    #[test]
    fn millis_to_rfc3339_epoch() {
        assert_eq!(millis_to_rfc3339(0), "1970-01-01T00:00:00Z");
    }

    #[test]
    fn millis_to_rfc3339_known_dates() {
        // 2024-01-01T00:00:00Z
        assert_eq!(millis_to_rfc3339(1_704_067_200_000), "2024-01-01T00:00:00Z");
        // 2026-03-21T23:50:45Z
        assert_eq!(millis_to_rfc3339(1_774_137_045_000), "2026-03-21T23:50:45Z");
    }

    #[test]
    fn millis_to_rfc3339_subsecond_truncated() {
        // 999ms after epoch — seconds should be 0
        assert_eq!(millis_to_rfc3339(999), "1970-01-01T00:00:00Z");
    }

    // ── validate_for_publish tests ─────────────────────────────────

    #[test]
    fn validate_publish_accepts_simple_topic() {
        TopicName("orders.us".into())
            .validate_for_publish()
            .unwrap();
    }

    #[test]
    fn validate_publish_accepts_no_dots() {
        TopicName("my-topic".into()).validate_for_publish().unwrap();
    }

    #[test]
    fn validate_publish_accepts_deep_hierarchy() {
        TopicName("a.b.c.d.e".into())
            .validate_for_publish()
            .unwrap();
    }

    #[test]
    fn validate_publish_rejects_star() {
        assert!(TopicName("orders.*".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_gt() {
        assert!(TopicName("events.>".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_bare_star() {
        assert!(TopicName("*".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_bare_gt() {
        assert!(TopicName(">".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_embedded_star() {
        assert!(
            TopicName("orders.*.us".into())
                .validate_for_publish()
                .is_err()
        );
    }

    #[test]
    fn validate_publish_rejects_double_dot() {
        assert!(
            TopicName("orders..foo".into())
                .validate_for_publish()
                .is_err()
        );
    }

    #[test]
    fn validate_publish_rejects_leading_dot() {
        assert!(TopicName(".orders".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_trailing_dot() {
        assert!(TopicName("orders.".into()).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_rejects_too_long() {
        let long = "a".repeat(256);
        assert!(TopicName(long).validate_for_publish().is_err());
    }

    #[test]
    fn validate_publish_accepts_255_bytes() {
        let exact = "a".repeat(255);
        TopicName(exact).validate_for_publish().unwrap();
    }

    #[test]
    fn validate_publish_rejects_empty() {
        assert!(TopicName(String::new()).validate_for_publish().is_err());
    }

    // ── TTL / expiry tests ──────────────────────────────────────────

    #[test]
    fn is_expired_no_ttl_never_expires() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        );
        assert!(!env.is_expired(0));
        assert!(!env.is_expired(1000));
        assert!(!env.is_expired(u64::MAX));
    }

    #[test]
    fn is_expired_before_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        // 1ms before expiry
        assert!(!env.is_expired(5999));
    }

    #[test]
    fn is_expired_exactly_at_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        // Exactly at expiry boundary (timestamp + ttl_ms == now)
        assert!(env.is_expired(6000));
    }

    #[test]
    fn is_expired_after_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        assert!(env.is_expired(6001));
    }

    #[test]
    fn remaining_ttl_no_ttl() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        );
        assert_eq!(env.remaining_ttl(500), None);
    }

    #[test]
    fn remaining_ttl_before_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        assert_eq!(env.remaining_ttl(3000), Some(3000));
    }

    #[test]
    fn remaining_ttl_at_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        assert_eq!(env.remaining_ttl(6000), None);
    }

    #[test]
    fn remaining_ttl_after_expiry() {
        let env = MessageEnvelope::new(
            MessageId(1),
            NamespaceId("ns".into()),
            TopicName("t".into()),
            vec![],
            1000,
        )
        .with_ttl(5000);
        assert_eq!(env.remaining_ttl(9000), None);
    }

    #[test]
    fn set_clock_offset_shifts_now_millis() {
        // Reset to zero first to isolate this test.
        set_clock_offset(0);
        let before = now_millis();

        // Apply a +1000ms offset.
        set_clock_offset(1000);
        let after = now_millis();

        // `after` should be ~1000ms ahead of `before` (minus the few microseconds
        // of real time elapsed between the two calls).
        assert!(
            after >= before + 900,
            "expected ~1000ms shift, got diff={}",
            after.saturating_sub(before)
        );

        // Negative offset: shift back by 2000ms.
        set_clock_offset(-2000);
        let shifted_back = now_millis();
        // shifted_back should be ~3000ms behind `after` (2000 shift + 1000 undone).
        assert!(
            shifted_back < after,
            "negative offset should reduce now_millis"
        );

        // Reset for other tests.
        set_clock_offset(0);
    }
}
