pub mod manager;
pub mod state;

pub use manager::{DefaultNamespaceManager, ManagerConfig, NamespaceQuota};

use std::future::Future;
use std::pin::Pin;

use crate::error::Result;
use crate::types::{
    ClientId, FetchResult, GroupName, MessageEnvelope, MessageId, NamespaceId, SubscribeResult,
    SubscribeStart, TopicInfo, TopicName, TopicOffset,
};

/// Result of a Nack operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NackResult {
    /// Message was successfully written to the DLQ topic.
    DeadLettered,
    /// DLQ is disabled (empty suffix) — message was dropped from inflight.
    Dropped,
}

/// Result of a batch ack operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct BatchAckResult {
    /// Number of message IDs that were found in inflight and successfully acked.
    pub acked: u32,
    /// Number of message IDs not found in inflight (already acked or expired).
    pub unknown: u32,
}

/// Central coordination layer between TCP/HTTP servers and the queue engine.
///
/// Manages subscriber/consumer state, fan-out delivery, consumer groups,
/// interceptor pipelines, and dead-letter routing.
pub trait NamespaceManager: Send + Sync {
    /// Subscribe a client to a topic, optionally as part of a consumer group.
    ///
    /// Returns a [`SubscribeResult`] containing the delivery channel, the actual
    /// starting offset, and the number of matched topics (for wildcard
    /// subscriptions; 0 for exact match).
    fn subscribe(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
        start_from: SubscribeStart,
    ) -> impl Future<Output = Result<SubscribeResult>> + Send;

    /// Remove a client's subscription from a topic.
    fn unsubscribe(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
    ) -> impl Future<Output = Result<()>> + Send;

    /// Publish a message to a topic, running publish interceptors first.
    fn publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> impl Future<Output = Result<TopicOffset>> + Send;

    /// Acknowledge delivery of a message by its [`MessageId`].
    ///
    /// The manager uses the reverse subscription index to locate the correct
    /// inflight map (individual or group) and remove the entry. This matches
    /// the wire protocol, which uses `MessageId` in `Frame::Ack`.
    fn ack(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        message_id: MessageId,
    ) -> impl Future<Output = Result<()>> + Send;

    /// Acknowledge multiple messages in one call, grouped by topic internally.
    ///
    /// Default implementation calls `ack()` in a loop. Implementations can
    /// override for efficiency (single lock per topic instead of per message).
    fn batch_ack(
        &self,
        ns: &NamespaceId,
        client: ClientId,
        acks: Vec<(MessageId, TopicName)>,
    ) -> impl Future<Output = BatchAckResult> + Send {
        async move {
            let mut acked: u32 = 0;
            let mut unknown: u32 = 0;
            for (message_id, topic) in acks {
                match self.ack(ns, &topic, client, message_id).await {
                    Ok(()) => acked += 1,
                    Err(_) => unknown += 1,
                }
            }
            BatchAckResult { acked, unknown }
        }
    }

    /// Reject a delivered message and route it to the dead-letter queue.
    ///
    /// Looks up the inflight entry by `message_id`, sets `_fafb.dlq.*` system
    /// headers with `DlqReason::ClientNack` metadata, and publishes to the
    /// DLQ topic. Returns `NackResult::Dropped` when DLQ is disabled (empty
    /// suffix). Returns `Err` when the DLQ write fails — the message stays
    /// inflight for redelivery.
    fn nack(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        message_id: MessageId,
        reason: String,
    ) -> impl Future<Output = Result<NackResult>> + Send;

    /// Register a new connection to a namespace.
    ///
    /// Checks the per-namespace connection limit and increments the connection
    /// count. Returns `Error::ResourceExhausted` if the limit is reached.
    /// Records the client-to-namespace mapping for use during disconnect.
    fn connect(
        &self,
        ns: &NamespaceId,
        client: ClientId,
    ) -> impl Future<Output = Result<()>> + Send;

    /// Clean up all subscriptions for a disconnected client.
    fn disconnect(&self, client: &ClientId) -> impl Future<Output = Result<()>> + Send;

    /// List all topics with published data in a namespace.
    fn list_topics(&self, ns: &NamespaceId) -> impl Future<Output = Result<Vec<TopicName>>> + Send;

    /// Return combined queue and subscriber statistics for a topic.
    fn topic_info(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = Result<TopicInfo>> + Send;

    /// Return the lowest offset still needed by any active subscriber or consumer
    /// group for this topic. Returns `None` if no subscribers exist.
    fn min_live_offset(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = Option<TopicOffset>> + Send;

    /// Return the topic's current head offset (the next offset that will be
    /// assigned on publish). Returns `TopicOffset::ZERO` if the topic has
    /// never been published to. This is a lightweight alternative to
    /// `topic_info()` when only the head offset is needed.
    fn head_offset(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = Result<TopicOffset>> + Send;

    /// Find the offset of the first message at or after `timestamp_millis`.
    /// Returns `None` if no message exists at or after the timestamp.
    /// Used by the TCP session to resolve `SubscribeFrom` timestamps before
    /// calling `subscribe()`.
    fn offset_at_timestamp(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        timestamp_millis: u64,
    ) -> impl Future<Output = Result<Option<TopicOffset>>> + Send;

    /// Fetch up to `max_count` messages starting at `offset`.
    ///
    /// Stateless read — does not create subscriptions, delivery tasks,
    /// or inflight tracking. The caller manages offset progression.
    ///
    /// Returns messages available in `[offset, head_offset)` up to `max_count`,
    /// respecting the frame size budget (`max_response_bytes`). If `offset` is
    /// below `base_offset` (evicted from memory), reads from storage via cold batch.
    ///
    /// When `headers_enabled` is true, header wire overhead is included in the
    /// per-message byte budget. When false (legacy clients without `CAP_HEADERS`),
    /// header overhead is excluded since headers won't appear on the wire.
    ///
    /// Returns `Error::MessageTooLarge` if the first message at the requested
    /// offset exceeds `max_response_bytes`.
    fn fetch(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
        max_count: u32,
        max_response_bytes: usize,
        headers_enabled: bool,
    ) -> impl Future<Output = Result<FetchResult>> + Send;

    /// Apply a replicated consumer group offset. Called on followers
    /// when receiving CommitOffset replication entries.
    /// Creates the consumer group if it doesn't exist locally.
    /// Sets the group's next_offset to the given offset.
    fn commit_offset(
        &self,
        _ns: &NamespaceId,
        _topic: &TopicName,
        _group: &GroupName,
        _offset: TopicOffset,
    ) -> impl Future<Output = Result<()>> + Send {
        async { Ok(()) }
    }

    /// Signal all delivery tasks to stop and await their completion.
    fn shutdown(&self) -> impl Future<Output = Result<()>> + Send;

    /// Return the last subscribe timestamp (epoch millis) for a topic.
    /// Returns `None` if no delivery state exists, `Some(0)` if state exists but no subscribe yet.
    fn topic_last_subscribe(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = Option<u64>> + Send;

    /// Returns true if any individual subscriber or consumer group member is connected.
    fn has_active_subscribers(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = bool> + Send;

    /// Returns true if any inflight messages are pending ack.
    fn has_inflight_messages(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = bool> + Send;

    /// Tear down all delivery state for a topic: abort delivery tasks, remove TopicDeliveryState.
    /// Wildcard subscription entries in NamespaceState are NOT removed.
    fn teardown_topic(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> impl Future<Output = Result<()>> + Send;
}

/// Intercepts messages before they are written to the queue engine.
///
/// Uses `Pin<Box<dyn Future>>` return type for dyn-compatibility so
/// interceptors can be stored in `Vec<Box<dyn PublishInterceptor>>`.
pub trait PublishInterceptor: Send + Sync {
    fn intercept_publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> Pin<Box<dyn Future<Output = Result<MessageEnvelope>> + Send + '_>>;
}

/// Intercepts messages before they are delivered to a subscriber.
///
/// Returning `Ok(None)` suppresses delivery to this particular client.
pub trait DeliveryInterceptor: Send + Sync {
    fn intercept_delivery(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        msg: MessageEnvelope,
    ) -> Pin<Box<dyn Future<Output = Result<Option<MessageEnvelope>>> + Send + '_>>;
}
