rifts 0.3.10

//! In-memory topic store (Rift spec section 9).
//!
//! This module provides the [`TopicStore`] — a process-wide, concurrent map
//! from topic name to [`TopicEntry`] — and the [`TopicEntry`] type that holds
//! all per-topic state: the profile, replay log, latest snapshot, and
//! subscriber/publisher counts.
//!
//! # Concurrency
//!
//! Topics are stored in a [`DashMap`] for lock-free concurrent access from
//! many connections. Each [`TopicEntry`] internally uses [`RwLock`] guards
//! for the mutable profile and log, and atomics for the subscriber/publisher
//! counts.
//!
//! # Replay and retention
//!
//! When a topic's retention policy allows it, appended messages are kept in a
//! bounded replay log. The [`TopicEntry::append`] method enforces the
//! retention policy on every insert, evicting stale entries so the log never
//! exceeds the configured bounds. The [`TopicEntry::range`] method supports
//! offset-based range queries for replay.
//!
//! # Snapshots
//!
//! If `snapshot_enabled` is set on the topic profile, every appended message
//! also replaces the latest snapshot. New subscribers can use
//! [`TopicEntry::snapshot`] to retrieve the current state without replaying
//! the full log.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use bytes::Bytes;
use dashmap::DashMap;
use parking_lot::RwLock;

use crate::error::{Result, RiftError, TopicReject};
use crate::now_ms;
use crate::topic::profile::TopicProfile;
use crate::topic::retention::RetentionPolicy;

/// Validate a topic name against the rules defined in spec section 9.1.
///
/// A valid topic name must:
///
/// - Be non-empty.
/// - Be at most 256 characters long.
/// - Not start with the reserved `$` prefix (reserved for system topics).
/// - Not contain control characters.
/// - Be valid UTF-8.
///
/// Returns `Ok(())` on success or a [`TopicReject::InvalidName`] error
/// describing which rule was violated.
pub fn validate_name(name: &str) -> Result<()> {
    if name.is_empty() {
        return Err(RiftError::Topic(TopicReject::InvalidName(
            "empty topic".into(),
        )));
    }
    if name.len() > 256 {
        return Err(RiftError::Topic(TopicReject::InvalidName(format!(
            "name too long: {} > 256",
            name.len()
        ))));
    }
    if name.starts_with('$') {
        return Err(RiftError::Topic(TopicReject::InvalidName(format!(
            "name starts with reserved '$' prefix: {}",
            name
        ))));
    }
    if name.chars().any(|c| c.is_control()) {
        return Err(RiftError::Topic(TopicReject::InvalidName(
            "name contains control characters".into(),
        )));
    }
    // `name` is a `&str` so it is already guaranteed to be valid UTF-8;
    // no extra check is needed here. The `is_control()` check above
    // rejects ASCII control characters; non-ASCII characters are
    // accepted as topic names.
    Ok(())
}

/// A subscriber registration identifier.
///
/// Each subscriber to a topic is assigned a unique `SubscriberId` that
/// allows the broker to remove the subscriber cheaply (by id) without
/// scanning a list. The inner `u64` is generated by a process-wide
/// atomic counter.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SubscriberId(pub u64);

/// A single entry in a topic's replay log.
///
/// Each log entry corresponds to one published message and stores the
/// broker-assigned offset, publisher metadata, the serialized payload,
/// and the sender's timestamp. Log entries are kept in offset order and
/// are evicted according to the topic's [`RetentionPolicy`].
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct LogEntry {
    /// Monotonically increasing offset assigned by the broker.
    pub offset: i64,

    /// The publisher's session id, if known. Used for per-publisher
    /// ordering and audit trails.
    pub publisher_session: Option<String>,

    /// Unique message identifier assigned by the publisher.
    pub message_id: String,

    /// Message class discriminator (e.g. `"event"`, `"state"`).
    pub class: String,

    /// Optional event name within the topic.
    pub event: Option<String>,

    /// The serialized message payload.
    pub payload: Bytes,

    /// Sender timestamp in milliseconds since the Unix epoch.
    /// If zero at append time the broker fills in `now_ms()`.
    pub timestamp: i64,

    /// Server-side append timestamp in milliseconds since the Unix
    /// epoch. Set by the broker when the entry is persisted; used by
    /// retention policies so that a future or skewed publisher
    /// timestamp cannot prevent an entry from being evicted.
    /// `None` for entries persisted before this field was introduced.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub appended_at: Option<i64>,
}

/// Internal state of a single topic.
///
/// A `TopicEntry` is created the first time a topic name is referenced
/// (via [`TopicStore::get_or_create`]) and lives as long as the topic
/// exists in the store. It holds the topic's profile, replay log, latest
/// snapshot, and atomic subscriber/publisher counts.
#[derive(Debug)]
pub struct TopicEntry {
    /// The topic's human-readable name.
    pub name: String,

    /// The topic's configurable profile (retention, ordering, limits, etc.).
    /// Wrapped in an `RwLock` so it can be updated at runtime.
    pub profile: RwLock<TopicProfile>,

    /// Whether the topic has been administratively closed.
    /// Closed topics reject new publishes and subscriptions.
    pub closed: parking_lot::Mutex<bool>,

    /// The replay log, sorted by offset. Bounded by the retention policy
    /// in `profile.retention`.
    pub log: RwLock<Vec<LogEntry>>,

    /// Current number of active subscribers on this topic.
    pub subscriber_count: AtomicU64,

    /// Current number of active publishers on this topic.
    pub publisher_count: AtomicU64,

    /// The most recently appended log entry, used as the topic snapshot.
    /// Only populated when `profile.snapshot_enabled` is `true`.
    pub latest_snapshot: RwLock<Option<LogEntry>>,
}

impl TopicEntry {
    /// Create a new topic entry with the given name and profile.
    ///
    /// The entry starts with an empty log, zero subscribers, zero
    /// publishers, and no snapshot.
    fn new(name: String, profile: TopicProfile) -> Self {
        Self {
            name,
            profile: RwLock::new(profile),
            closed: parking_lot::Mutex::new(false),
            log: RwLock::new(Vec::new()),
            subscriber_count: AtomicU64::new(0),
            publisher_count: AtomicU64::new(0),
            latest_snapshot: RwLock::new(None),
        }
    }

    /// Return the highest offset stored in the replay log, or `0` if
    /// the log is empty.
    ///
    /// The authoritative offset sequence is managed by the broker's
    /// `OffsetStore`; this method derives the head from the actual log
    /// entries and is suitable for range-query planning and resume
    /// evaluation.
    pub fn head_offset(&self) -> i64 {
        self.log.read().last().map(|e| e.offset).unwrap_or(0)
    }

    /// Check whether the topic can accept another subscriber.
    ///
    /// Returns `true` if the current subscriber count is below the
    /// `max_subscribers` limit in the topic profile.
    pub fn can_subscribe(&self) -> bool {
        let limit = self.profile.read().max_subscribers;
        self.subscriber_count.load(Ordering::Relaxed) < limit as u64
    }

    /// Check whether the topic can accept another publisher.
    ///
    /// Returns `true` if the current publisher count is below the
    /// `max_publishers` limit in the topic profile.
    pub fn can_publish(&self) -> bool {
        let limit = self.profile.read().max_publishers;
        self.publisher_count.load(Ordering::Relaxed) < limit as u64
    }

    /// Atomically increment the subscriber count by one.
    pub fn inc_subscriber(&self) {
        self.subscriber_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Atomically decrement the subscriber count by one (saturating at 0).
    pub fn dec_subscriber(&self) {
        self.subscriber_count.fetch_sub(1, Ordering::Relaxed);
    }

    /// Atomically increment the publisher count by one.
    pub fn inc_publisher(&self) {
        self.publisher_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Atomically decrement the publisher count by one (saturating at 0).
    pub fn dec_publisher(&self) {
        self.publisher_count.fetch_sub(1, Ordering::Relaxed);
    }

    /// Atomically reserve a subscriber slot: returns `true` and
    /// increments the count if the topic is below its
    /// `max_subscribers` limit, otherwise returns `false`
    /// without changing the count. This replaces the previous
    /// check-then-act pattern (`can_subscribe()` + `inc_subscriber()`)
    /// which had a TOCTOU race.
    pub fn try_inc_subscriber(&self) -> bool {
        let limit = self.profile.read().max_subscribers as u64;
        let mut cur = self.subscriber_count.load(Ordering::Relaxed);
        loop {
            if cur >= limit {
                return false;
            }
            match self.subscriber_count.compare_exchange_weak(
                cur,
                cur + 1,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => return true,
                Err(observed) => cur = observed,
            }
        }
    }

    /// Atomically reserve a publisher slot, mirroring
    /// [`try_inc_subscriber`](Self::try_inc_subscriber).
    pub fn try_inc_publisher(&self) -> bool {
        let limit = self.profile.read().max_publishers as u64;
        let mut cur = self.publisher_count.load(Ordering::Relaxed);
        loop {
            if cur >= limit {
                return false;
            }
            match self.publisher_count.compare_exchange_weak(
                cur,
                cur + 1,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => return true,
                Err(observed) => cur = observed,
            }
        }
    }

    /// Append a new log entry to this topic, enforcing the retention policy.
    ///
    /// If the entry's `timestamp` is `0` it is filled in with `now_ms()`.
    /// After insertion the retention policy is applied:
    ///
    /// - `None` — the log is cleared entirely.
    /// - `Count(n)` — oldest entries beyond `n` are drained.
    /// - `Size(max)` — oldest entries are drained until total payload
    ///   size is within `max` bytes.
    /// - `Ttl(dur)` — entries older than `dur` are removed.
    /// - `Latest` — all entries except the one just appended are removed.
    /// - `Durable` — no in-memory eviction (the external store handles it).
    /// - `None` — clears the entire log including the entry just
    ///   appended. This is the documented "fire-and-forget"
    ///   behavior: nothing is retained for replay.
    ///
    /// If `snapshot_enabled` is set on the profile the entry also replaces
    /// the latest snapshot.
    pub fn append(&self, mut entry: LogEntry) {
        let now = now_ms();
        if entry.timestamp == 0 {
            entry.timestamp = now;
        }
        // Stamp the server-side append time. TTL retention in external
        // stores (e.g. sled) uses this rather than the (possibly skewed
        // or malicious) publisher timestamp.
        entry.appended_at = Some(now);
        let profile = self.profile.read().clone();
        // Take the snapshot branch first so we can move `entry`
        // rather than cloning it.
        let mut log = self.log.write();
        log.push(entry.clone());
        match profile.retention {
            RetentionPolicy::None => log.clear(),
            RetentionPolicy::Count(n) => {
                if log.len() > n {
                    let drop = log.len() - n;
                    log.drain(0..drop);
                }
            }
            RetentionPolicy::Size(max_bytes) => {
                let mut total: usize = log.iter().map(|e| e.payload.len()).sum();
                let mut idx = 0;
                while total > max_bytes && idx < log.len() {
                    total -= log[idx].payload.len();
                    idx += 1;
                }
                if idx > 0 {
                    log.drain(0..idx);
                }
            }
            RetentionPolicy::Ttl(ttl) => {
                let ttl_ms = ttl.as_millis() as i64;
                log.retain(|e| {
                    // Prefer the broker's append time when present.
                    let ts = e.appended_at.unwrap_or(e.timestamp);
                    now - ts <= ttl_ms
                });
            }
            RetentionPolicy::Latest => {
                log.retain(|e| e.offset == entry.offset);
            }
            RetentionPolicy::Durable => {
                // External store handles persistence; keep a bounded
                // in-memory window for fast range queries. Evict
                // oldest entries when the log grows beyond the cap
                // to prevent unbounded memory growth.
                const DURABLE_MEMORY_CAP: usize = 10_000;
                if log.len() > DURABLE_MEMORY_CAP {
                    let drop = log.len() - DURABLE_MEMORY_CAP;
                    log.drain(0..drop);
                }
            }
        }
        if profile.snapshot_enabled {
            *self.latest_snapshot.write() = Some(entry);
        }
    }

    /// Return log entries whose offset is in the half-open range
    /// `[from, to)`.
    ///
    /// This is used by the replay path to fetch the range of messages
    /// a late-joining subscriber needs to catch up to the live stream.
    /// The returned vector is sorted by offset in ascending order.
    pub fn range(&self, from: i64, to: i64) -> Vec<LogEntry> {
        self.log
            .read()
            .iter()
            .filter(|e| e.offset >= from && e.offset < to)
            .cloned()
            .collect()
    }

    /// Return the latest snapshot for this topic, if any.
    ///
    /// Returns `None` if no messages have been appended yet or if
    /// `snapshot_enabled` is `false` on the topic profile.
    pub fn snapshot(&self) -> Option<LogEntry> {
        self.latest_snapshot.read().clone()
    }
}

/// Process-wide topic store — a concurrent map from topic name to entry.
///
/// Internally the [`DashMap`] is wrapped in an [`Arc`] so that cloning a
/// `TopicStore` (which the broker and the router both do) shares the same
/// underlying map. This design allows multiple subsystems to hold their own
/// `TopicStore` handle without duplicating data.
#[derive(Clone, Debug, Default)]
pub struct TopicStore {
    inner: Arc<DashMap<String, Arc<TopicEntry>>>,
}

impl TopicStore {
    /// Create an empty topic store with no pre-existing topics.
    pub fn new() -> Self {
        Self::default()
    }

    /// Look up a topic by name, or auto-create it with the given default
    /// profile if it does not yet exist.
    ///
    /// The topic name is validated before lookup; invalid names produce a
    /// [`TopicReject::InvalidName`] error. If the topic already exists
    /// the `default_profile` is ignored and the existing entry is returned.
    pub fn get_or_create(
        &self,
        name: &str,
        default_profile: TopicProfile,
    ) -> Result<Arc<TopicEntry>> {
        validate_name(name)?;
        Ok(self
            .inner
            .entry(name.to_string())
            .or_insert_with(|| Arc::new(TopicEntry::new(name.to_string(), default_profile)))
            .value()
            .clone())
    }

    /// Look up a topic by name.
    ///
    /// Returns `None` if the topic does not exist. Does **not** validate
    /// the name — use [`validate_name`] separately if needed.
    pub fn get(&self, name: &str) -> Option<Arc<TopicEntry>> {
        self.inner.get(name).map(|e| e.clone())
    }

    /// Returns `true` if a topic with the given name exists in the store.
    pub fn exists(&self, name: &str) -> bool {
        self.inner.contains_key(name)
    }

    /// Remove a topic from the store.
    ///
    /// Returns the removed [`TopicEntry`] if it existed, or `None`
    /// otherwise. After removal the topic can be re-created via
    /// [`get_or_create`](Self::get_or_create).
    pub fn remove(&self, name: &str) -> Option<Arc<TopicEntry>> {
        self.inner.remove(name).map(|(_, e)| e)
    }

    /// Return a snapshot of all topic names currently in the store.
    pub fn names(&self) -> Vec<String> {
        self.inner.iter().map(|kv| kv.key().clone()).collect()
    }

    /// Return per-topic statistics as a `BTreeMap` from topic name to
    /// `(subscriber_count, publisher_count, head_offset)`.
    ///
    /// The map is sorted by topic name. This is intended for admin/diagnostic
    /// endpoints and metrics exporters.
    pub fn stats(&self) -> BTreeMap<String, (u64, u64, i64)> {
        self.inner
            .iter()
            .map(|kv| {
                let e = kv.value();
                (
                    kv.key().clone(),
                    (
                        e.subscriber_count.load(Ordering::Relaxed),
                        e.publisher_count.load(Ordering::Relaxed),
                        e.head_offset(),
                    ),
                )
            })
            .collect()
    }
}

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

    fn sample_entry(offset: i64, payload: &[u8]) -> LogEntry {
        LogEntry {
            offset,
            publisher_session: None,
            message_id: format!("m-{offset}"),
            class: "event".into(),
            event: Some("e".into()),
            payload: Bytes::copy_from_slice(payload),
            timestamp: 0,
            appended_at: None,
        }
    }

    #[test]
    fn name_validation() {
        assert!(super::validate_name("room/1").is_ok());
        assert!(super::validate_name("user/abc").is_ok());
        assert!(super::validate_name("").is_err());
        assert!(super::validate_name("$system").is_err());
        assert!(super::validate_name(&"x".repeat(257)).is_err());
    }

    #[test]
    fn get_or_create_idempotent() {
        let store = TopicStore::new();
        let a = store
            .get_or_create("room/1", TopicProfile::default())
            .unwrap();
        let b = store
            .get_or_create("room/1", TopicProfile::default())
            .unwrap();
        assert!(Arc::ptr_eq(&a, &b));
    }

    #[test]
    fn append_count_retention() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    retention: RetentionPolicy::Count(2),
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        for i in 1..=5 {
            entry.append(sample_entry(i, b"x"));
        }
        let log = entry.log.read();
        assert_eq!(log.len(), 2);
        assert_eq!(log[0].offset, 4);
        assert_eq!(log[1].offset, 5);
    }

    #[test]
    fn append_latest_retention() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    retention: RetentionPolicy::Latest,
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        entry.append(sample_entry(1, b"a"));
        entry.append(sample_entry(2, b"b"));
        let log = entry.log.read();
        assert_eq!(log.len(), 1);
        assert_eq!(log[0].offset, 2);
    }

    #[test]
    fn range_query() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    retention: RetentionPolicy::Count(100),
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        for i in 1..=5 {
            entry.append(sample_entry(i, b"x"));
        }
        // `range` is half-open: [from, to).
        let got = entry.range(2, 4);
        assert_eq!(got.iter().map(|e| e.offset).collect::<Vec<_>>(), vec![2, 3]);
        let got_inclusive_end = entry.range(2, 5);
        assert_eq!(
            got_inclusive_end
                .iter()
                .map(|e| e.offset)
                .collect::<Vec<_>>(),
            vec![2, 3, 4]
        );
    }

    #[test]
    fn snapshot_keeps_latest() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    retention: RetentionPolicy::Count(10),
                    snapshot_enabled: true,
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        entry.append(sample_entry(1, b"a"));
        entry.append(sample_entry(2, b"b"));
        let s = entry.snapshot().unwrap();
        assert_eq!(s.offset, 2);
    }

    #[test]
    fn head_offset_reflects_log() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    retention: RetentionPolicy::Count(100),
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        assert_eq!(entry.head_offset(), 0);
        entry.append(sample_entry(1, b"a"));
        assert_eq!(entry.head_offset(), 1);
        entry.append(sample_entry(5, b"b"));
        assert_eq!(entry.head_offset(), 5);
    }

    #[test]
    fn subscriber_limit_check() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    max_subscribers: 2,
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        assert!(entry.can_subscribe());
        entry.inc_subscriber();
        entry.inc_subscriber();
        assert!(!entry.can_subscribe());
    }

    #[test]
    fn publisher_limit_check() {
        let store = TopicStore::new();
        let entry = store
            .get_or_create(
                "t1",
                TopicProfile {
                    max_publishers: 1,
                    ..TopicProfile::default()
                },
            )
            .unwrap();
        assert!(entry.can_publish());
        entry.inc_publisher();
        assert!(!entry.can_publish());
    }
}