use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use tokio::sync::{Mutex, RwLock, mpsc, watch};
use tokio::task::JoinHandle;
use tracing;

pub use crate::config::NamespaceQuota;
use crate::config::{SharedConfig, SlowConsumerPolicy};
use crate::dlq::{DlqReason, build_dlq_headers};
use crate::error::{Error, Result};
use crate::namespace::state::{
    ConsumerGroupState, InflightEntry, NamespaceState, SubscriberDeliveryState, TopicDeliveryState,
    WildcardGroupSubInfo, WildcardSubInfo,
};
use crate::namespace::{DeliveryInterceptor, NackResult, NamespaceManager, PublishInterceptor};
use crate::queue::{PublishResult, QueueEngine};
use crate::types::{
    ClientId, FetchResult, GroupName, MessageEnvelope, MessageId, NamespaceId, PublishedMessage,
    SubscribeResult, SubscribeStart, TopicInfo, TopicName, TopicOffset, now_millis,
};
use crate::wildcard::{SubscriptionId, TopicPattern, contains_wildcard};

/// Snapshot of wildcard group subscription info for auto-attach on new topic creation.
type WildcardGroupInfo = (
    SubscriptionId,
    TopicName,
    GroupName,
    HashMap<ClientId, mpsc::Sender<PublishedMessage>>,
);

/// Snapshot of inflight entry state when DLQ routing is needed.
/// Captured under lock so the lock can be dropped before the async DLQ write.
struct DlqCheckResult {
    delivery_count: u32,
    first_delivered_at: u64,
    dlq_failures: u32,
}

/// Bundles all context needed to route a message to the dead-letter queue.
/// Built from a `DlqCheckResult` snapshot + caller-determined reason.
struct DlqContext {
    original_offset: TopicOffset,
    first_delivered_at: u64,
    delivery_count: u32,
    dlq_failures: u32,
    reason: DlqReason,
}

impl DlqContext {
    /// Build a routing context from a check snapshot and the caller's reason.
    fn from_check(check: &DlqCheckResult, offset: TopicOffset, reason: DlqReason) -> Self {
        Self {
            original_offset: offset,
            first_delivered_at: check.first_delivered_at,
            delivery_count: check.delivery_count,
            dlq_failures: check.dlq_failures,
            reason,
        }
    }
}

/// Result of a successful DLQ write. Carries the DLQ timestamp so the
/// time-in-flight metric uses the exact value embedded in the envelope.
struct DlqWriteResult {
    dlq_timestamp: u64,
}

/// Result of attempting to deliver a single message to a subscriber channel.
enum ColdDeliverResult {
    /// Message sent successfully, offset advanced.
    Sent,
    /// Channel full — caller should push message back and sleep.
    ChannelFull,
    /// Channel closed — subscriber is gone, caller should break.
    ChannelClosed,
}

/// Send a cold-path message to an individual subscriber and update state.
///
/// Shared by the cold-batch drain path and the first-cold-batch-message path.
#[allow(clippy::too_many_arguments)]
async fn cold_deliver_individual(
    msg: &PublishedMessage,
    offset: TopicOffset,
    tx: &mpsc::Sender<PublishedMessage>,
    topic_state: &Mutex<TopicDeliveryState>,
    client: ClientId,
    ns: &NamespaceId,
    topic: &TopicName,
    base_timeout: Duration,
    max_backoff: Duration,
) -> ColdDeliverResult {
    // Skip expired messages — advance offset without delivering
    let ttl_now = now_millis();
    if msg.envelope.is_expired(ttl_now) {
        metrics::counter!(
            "fafb_expired_skipped_total",
            "namespace" => ns.0.clone(),
            "topic" => topic.0.clone(),
        )
        .increment(1);
        let mut state = topic_state.lock().await;
        if let Some(sub) = state.individual.get_mut(&client) {
            sub.next_offset = offset.next();
            sub.inflight.remove(&offset);
        }
        return ColdDeliverResult::Sent;
    }
    // Record remaining TTL at delivery time (cold path)
    if let Some(remaining) = msg.envelope.remaining_ttl(ttl_now) {
        metrics::histogram!("fafb_message_remaining_ttl_seconds").record(remaining as f64 / 1000.0);
    }
    match tx.try_send(msg.clone()) {
        Ok(()) => {
            metrics::counter!("fafb_messages_delivered_total",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                "mode" => "individual"
            )
            .increment(1);
            let mut state = topic_state.lock().await;
            if let Some(sub) = state.individual.get_mut(&client) {
                sub.next_offset = offset.next();
                let now = now_millis();
                let (delivery_count, first_delivered_at) = sub
                    .inflight
                    .get(&offset)
                    .map(|e| (e.delivery_count + 1, e.first_delivered_at))
                    .unwrap_or((1, now));
                let delay =
                    InflightEntry::redelivery_delay(delivery_count, base_timeout, max_backoff);
                sub.inflight.insert(
                    offset,
                    InflightEntry {
                        offset,
                        message_id: msg.envelope.id,
                        delivered_to: client,
                        delivered_at: Instant::now(),
                        delivery_count,
                        first_delivered_at,
                        dlq_failures: 0,
                        next_redeliver_at: Instant::now() + delay,
                        msg_timestamp: msg.envelope.timestamp,
                        msg_ttl_ms: msg.envelope.ttl_ms,
                    },
                );
                sub.last_activity = Instant::now();
                sub.consecutive_send_failures = 0;
            }
            ColdDeliverResult::Sent
        }
        Err(mpsc::error::TrySendError::Full(_)) => ColdDeliverResult::ChannelFull,
        Err(mpsc::error::TrySendError::Closed(_)) => ColdDeliverResult::ChannelClosed,
    }
}

/// Send a cold-path message to a consumer group member and update state.
#[allow(clippy::too_many_arguments)]
async fn cold_deliver_group(
    msg: &PublishedMessage,
    offset: TopicOffset,
    member_id: ClientId,
    member_tx: &mpsc::Sender<PublishedMessage>,
    topic_state: &Mutex<TopicDeliveryState>,
    group_name: &GroupName,
    ns: &NamespaceId,
    topic: &TopicName,
    base_timeout: Duration,
    max_backoff: Duration,
) -> ColdDeliverResult {
    // Skip expired messages — advance offset without delivering
    let ttl_now = now_millis();
    if msg.envelope.is_expired(ttl_now) {
        metrics::counter!(
            "fafb_expired_skipped_total",
            "namespace" => ns.0.clone(),
            "topic" => topic.0.clone(),
        )
        .increment(1);
        let mut state = topic_state.lock().await;
        if let Some(group) = state.groups.get_mut(group_name) {
            group.next_offset = offset.next();
            group.inflight.remove(&offset);
        }
        return ColdDeliverResult::Sent;
    }
    // Record remaining TTL at delivery time (cold path)
    if let Some(remaining) = msg.envelope.remaining_ttl(ttl_now) {
        metrics::histogram!("fafb_message_remaining_ttl_seconds").record(remaining as f64 / 1000.0);
    }
    match member_tx.try_send(msg.clone()) {
        Ok(()) => {
            metrics::counter!("fafb_messages_delivered_total",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                "mode" => "group"
            )
            .increment(1);
            let mut state = topic_state.lock().await;
            if let Some(group) = state.groups.get_mut(group_name) {
                let now = now_millis();
                let (delivery_count, first_delivered_at) = group
                    .inflight
                    .get(&offset)
                    .map(|e| (e.delivery_count + 1, e.first_delivered_at))
                    .unwrap_or((1, now));
                let delay =
                    InflightEntry::redelivery_delay(delivery_count, base_timeout, max_backoff);
                group.inflight.insert(
                    offset,
                    InflightEntry {
                        offset,
                        message_id: msg.envelope.id,
                        delivered_to: member_id,
                        delivered_at: Instant::now(),
                        delivery_count,
                        first_delivered_at,
                        dlq_failures: 0,
                        next_redeliver_at: Instant::now() + delay,
                        msg_timestamp: msg.envelope.timestamp,
                        msg_ttl_ms: msg.envelope.ttl_ms,
                    },
                );
                group.next_offset = offset.next();
                group.last_activity = Instant::now();
                group.consecutive_send_failures = 0;
            }
            ColdDeliverResult::Sent
        }
        Err(mpsc::error::TrySendError::Full(_)) => ColdDeliverResult::ChannelFull,
        Err(mpsc::error::TrySendError::Closed(_)) => ColdDeliverResult::ChannelClosed,
    }
}

/// Grace period before empty consumer groups are cleaned up.
const EMPTY_GROUP_GRACE_PERIOD: Duration = Duration::from_secs(60);

/// A single subscription entry in the client reverse index.
#[derive(Debug, Clone)]
struct ClientSubEntry {
    namespace: NamespaceId,
    /// The topic name or wildcard pattern used to subscribe.
    topic_or_pattern: TopicName,
    group: Option<GroupName>,
    /// Wildcard state — `Some` for wildcard subscriptions, `None` for exact match.
    wildcard: Option<WildcardSubState>,
}

impl ClientSubEntry {
    fn is_wildcard(&self) -> bool {
        self.wildcard.is_some()
    }
}

/// Tracks a wildcard subscription's expanded state.
#[derive(Debug, Clone)]
struct WildcardSubState {
    pattern: TopicPattern,
    sub_id: SubscriptionId,
    /// Concrete topics this wildcard currently matches.
    matched_topics: Vec<TopicName>,
}

/// Reverse index: client -> list of subscription entries.
type ClientSubscriptionMap = Arc<RwLock<HashMap<ClientId, Vec<ClientSubEntry>>>>;

/// Shared context passed to delivery tasks.
struct DeliveryTaskContext<Q: QueueEngine> {
    engine: Arc<Q>,
    ns: NamespaceId,
    topic: TopicName,
    topic_state: Arc<Mutex<TopicDeliveryState>>,
    delivery_interceptors: Arc<Vec<Box<dyn DeliveryInterceptor>>>,
    /// Pre-computed flag to skip interceptor clone on the hot path.
    has_interceptors: bool,
    shared_config: SharedConfig,
    inflight_timeout_ms: Arc<AtomicU64>,
    max_redelivery_backoff_ms: Arc<AtomicU64>,
    shutdown_rx: watch::Receiver<bool>,
}

/// Reverse index: client -> namespace they connected to.
type ClientNamespaceMap = Arc<RwLock<HashMap<ClientId, NamespaceId>>>;

/// Configuration for the namespace manager.
///
/// Most manager settings are now read from `SharedConfig` at runtime for
/// hot-reload support. Only `delivery_channel_capacity` remains here because
/// mpsc channels are fixed-size at creation — changing it only affects new
/// subscriptions.
#[derive(Debug, Clone)]
pub struct ManagerConfig {
    /// Buffer size for subscriber delivery channels. New subscriptions use
    /// this value; existing channels keep their original capacity.
    pub delivery_channel_capacity: usize,
    /// Maximum number of concrete topics a single wildcard subscription can
    /// match. Subscribe is rejected if the pattern matches more topics.
    pub max_topics_per_wildcard: usize,
}

impl Default for ManagerConfig {
    fn default() -> Self {
        Self {
            delivery_channel_capacity: 256,
            max_topics_per_wildcard: 1000,
        }
    }
}

/// Concrete namespace manager implementation, generic over the queue engine.
///
/// ## Lock ordering
///
/// When multiple locks must be held, they must be acquired in this order to
/// prevent deadlocks:
///
/// 1. `ns_state.subscription_trie` — read in publish (new topic), write in subscribe/unsubscribe
/// 2. `ns_state.wildcard_subs` — read in publish (new topic), write in subscribe/unsubscribe
///    2b. `ns_state.wildcard_group_subs` — read in publish (new topic), write in subscribe/unsubscribe
/// 3. `self.client_subscriptions` — read/write in subscribe, unsubscribe, publish (auto-attach), disconnect
/// 4. `ns_state.topics` — read in unsubscribe/disconnect, write via `get_or_create_topic`
/// 5. `topic_state` (individual `Mutex<TopicDeliveryState>`) — locked per-topic for subscriber mutations
///
/// The publish path (hot) acquires: trie(R) → wildcard_subs(R) → wildcard_group_subs(R) → client_subscriptions(R/W) → topics(W) → topic_state(W)
/// The subscribe path acquires: topics(W) → topic_state(W) → trie(W) → wildcard_subs(W)/wildcard_group_subs(W) → client_subscriptions(W)
///
/// These orderings are safe because publish holds only read locks on trie/wildcard_subs/wildcard_group_subs
/// while subscribe holds write locks — RwLock permits concurrent readers. No path holds
/// a write lock on an earlier resource while acquiring a later one in the opposite order.
pub struct DefaultNamespaceManager<Q: QueueEngine> {
    engine: Arc<Q>,
    namespaces: Arc<RwLock<HashMap<NamespaceId, Arc<NamespaceState>>>>,
    shared_config: SharedConfig,
    publish_interceptors: Vec<Box<dyn PublishInterceptor>>,
    delivery_interceptors: Arc<Vec<Box<dyn DeliveryInterceptor>>>,
    client_subscriptions: ClientSubscriptionMap,
    /// Reverse index: client -> namespace they connected to.
    client_namespaces: ClientNamespaceMap,
    /// Initial delivery channel capacity — used for new subscriptions.
    /// Cannot change for existing channels (mpsc is fixed-size at creation).
    delivery_channel_capacity: usize,
    max_topics_per_wildcard: usize,
    inflight_timeout_ms: Arc<AtomicU64>,
    max_redelivery_backoff_ms: Arc<AtomicU64>,
    shutdown_tx: watch::Sender<bool>,
    /// Monotonic counter for generating unique SubscriptionIds.
    next_sub_id: AtomicU64,
}

impl<Q: QueueEngine + 'static> DefaultNamespaceManager<Q> {
    /// Create a new namespace manager. Pre-populates namespace state for each
    /// configured namespace ID.
    pub fn new(
        engine: Arc<Q>,
        namespace_ids: impl IntoIterator<Item = NamespaceId>,
        shared_config: SharedConfig,
        config: ManagerConfig,
    ) -> Self {
        let mut namespaces = HashMap::new();
        for ns_id in namespace_ids {
            namespaces.insert(ns_id, Arc::new(NamespaceState::new()));
        }
        let (shutdown_tx, _) = watch::channel(false);
        let cfg = shared_config.load();
        let initial_inflight_ms = cfg.inflight_timeout_secs.saturating_mul(1000);
        let initial_backoff_ms = cfg.max_redelivery_backoff_secs.saturating_mul(1000);
        drop(cfg);
        Self {
            engine,
            namespaces: Arc::new(RwLock::new(namespaces)),
            shared_config,
            publish_interceptors: Vec::new(),
            delivery_interceptors: Arc::new(Vec::new()),
            client_subscriptions: Arc::new(RwLock::new(HashMap::new())),
            client_namespaces: Arc::new(RwLock::new(HashMap::new())),
            delivery_channel_capacity: config.delivery_channel_capacity,
            max_topics_per_wildcard: config.max_topics_per_wildcard,
            inflight_timeout_ms: Arc::new(AtomicU64::new(initial_inflight_ms)),
            max_redelivery_backoff_ms: Arc::new(AtomicU64::new(initial_backoff_ms)),
            shutdown_tx,
            next_sub_id: AtomicU64::new(1),
        }
    }

    /// Accelerate the inflight timeout during drain so unacked messages
    /// are resolved before shutdown. Also resets existing inflight entries'
    /// deadlines so long-backoff entries don't outlive the drain window.
    pub async fn set_inflight_timeout(&self, timeout: Duration) {
        self.inflight_timeout_ms
            .store(timeout.as_millis() as u64, Ordering::Release);
        tracing::info!(
            inflight_timeout_ms = timeout.as_millis() as u64,
            "inflight timeout updated"
        );
        // Walk all topics and cap existing deadlines to now + timeout
        let namespaces = self.namespaces.read().await;
        for ns_state in namespaces.values() {
            let topics = ns_state.topics.read().await;
            for topic_state in topics.values() {
                let mut state = topic_state.lock().await;
                state.reset_inflight_deadlines(timeout);
            }
        }
    }

    /// Add publish and delivery interceptors. Builder-style.
    pub fn with_interceptors(
        mut self,
        publish: Vec<Box<dyn PublishInterceptor>>,
        delivery: Vec<Box<dyn DeliveryInterceptor>>,
    ) -> Self {
        self.publish_interceptors = publish;
        self.delivery_interceptors = Arc::new(delivery);
        self
    }

    /// Register a namespace — creates state if it doesn't exist (idempotent).
    /// Used during config hot-reload when new namespaces are added.
    pub async fn register_namespace(&self, ns: &NamespaceId) {
        let mut namespaces = self.namespaces.write().await;
        if namespaces.contains_key(ns) {
            tracing::debug!(namespace = %ns, "namespace already registered (idempotent)");
            return;
        }
        namespaces.insert(ns.clone(), Arc::new(NamespaceState::new()));
        tracing::info!(namespace = %ns, "namespace registered via hot-reload");
    }

    /// Return the number of active connections for a namespace.
    /// Used to check if it's safe to remove a namespace during hot-reload.
    pub async fn active_connections_for_namespace(&self, ns: &NamespaceId) -> u64 {
        let namespaces = self.namespaces.read().await;
        namespaces
            .get(ns)
            .map(|state| {
                state
                    .connection_count
                    .load(std::sync::atomic::Ordering::Relaxed) as u64
            })
            .unwrap_or(0)
    }

    /// Spawn a background task that watches for reload notifications and
    /// registers newly added namespaces.
    pub fn spawn_reload_watcher(
        self: &Arc<Self>,
        mut reload_rx: tokio::sync::watch::Receiver<Option<crate::config::ReloadNotification>>,
    ) -> JoinHandle<()> {
        let mgr = Arc::clone(self);
        tokio::spawn(async move {
            while reload_rx.changed().await.is_ok() {
                let notification = reload_rx.borrow().clone();
                if let Some(ref notif) = notification {
                    for ns in &notif.namespaces_added {
                        mgr.register_namespace(ns).await;
                    }
                }
            }
        })
    }

    /// Look up the namespace state, returning `NamespaceNotFound` if not configured.
    async fn get_namespace(&self, ns: &NamespaceId) -> Result<Arc<NamespaceState>> {
        let namespaces = self.namespaces.read().await;
        namespaces
            .get(ns)
            .cloned()
            .ok_or_else(|| Error::NamespaceNotFound(ns.clone()))
    }

    /// Get or create the topic delivery state within a namespace.
    async fn get_or_create_topic(
        ns_state: &NamespaceState,
        topic: &TopicName,
    ) -> Arc<Mutex<TopicDeliveryState>> {
        // Fast path: read lock
        {
            let topics = ns_state.topics.read().await;
            if let Some(t) = topics.get(topic) {
                return Arc::clone(t);
            }
        }
        // Slow path: write lock, double-check
        let mut topics = ns_state.topics.write().await;
        if let Some(t) = topics.get(topic) {
            return Arc::clone(t);
        }
        let t = Arc::new(Mutex::new(TopicDeliveryState::new()));
        topics.insert(topic.clone(), Arc::clone(&t));
        t
    }

    /// Build a DeliveryTaskContext from the manager's current config.
    fn make_delivery_context(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        topic_state: &Arc<Mutex<TopicDeliveryState>>,
    ) -> DeliveryTaskContext<Q> {
        DeliveryTaskContext {
            engine: Arc::clone(&self.engine),
            ns: ns.clone(),
            topic: topic.clone(),
            topic_state: Arc::clone(topic_state),
            has_interceptors: !self.delivery_interceptors.is_empty(),
            delivery_interceptors: Arc::clone(&self.delivery_interceptors),
            shared_config: Arc::clone(&self.shared_config),
            inflight_timeout_ms: Arc::clone(&self.inflight_timeout_ms),
            max_redelivery_backoff_ms: Arc::clone(&self.max_redelivery_backoff_ms),
            shutdown_rx: self.shutdown_tx.subscribe(),
        }
    }

    /// Spawn a delivery task for an individual subscriber.
    /// Returns the JoinHandle so the caller can store it.
    fn spawn_individual_delivery_task(
        ctx: DeliveryTaskContext<Q>,
        client: ClientId,
    ) -> JoinHandle<()> {
        /// How often the individual delivery task sweeps inflight entries
        /// for timed-out messages that need delivery_count escalation or DLQ routing.
        const INFLIGHT_SWEEP_INTERVAL: Duration = Duration::from_secs(1);

        let DeliveryTaskContext {
            engine,
            ns,
            topic,
            topic_state,
            delivery_interceptors,
            has_interceptors,
            shared_config,
            inflight_timeout_ms,
            max_redelivery_backoff_ms,
            mut shutdown_rx,
        } = ctx;
        let engine_for_dlq = Arc::clone(&engine);
        tokio::spawn(async move {
            let mut cold_batch: Option<VecDeque<PublishedMessage>> = None;
            let mut last_sweep = Instant::now();
            // Cache config values; refreshed when a message is consumed.
            // Initial load provides values for the first loop iteration (including
            // the inflight sweep which runs before any message is consumed).
            let cfg = shared_config.load();
            #[allow(unused_assignments)]
            let mut max_redelivery = cfg.max_redelivery_attempts;
            #[allow(unused_assignments)]
            let mut dlq_suffix = cfg.dlq_topic_suffix.clone();
            #[allow(unused_assignments)]
            let mut slow_consumer_policy = cfg.slow_consumer_policy;
            #[allow(unused_assignments)]
            let mut slow_consumer_threshold = cfg.slow_consumer_threshold;
            #[allow(unused_assignments)]
            let mut slow_consumer_disconnect_after = cfg.slow_consumer_disconnect_after;
            #[allow(unused_assignments)]
            let mut cold_batch_size = cfg.cold_batch_size;
            drop(cfg);
            // Backoff timeouts: cached per message consume, driven by SharedConfig.
            // The sweep reads atomics directly for drain-acceleration responsiveness.
            let mut base_timeout = Duration::from_secs(shared_config.load().inflight_timeout_secs);
            let mut max_backoff =
                Duration::from_secs(shared_config.load().max_redelivery_backoff_secs);
            macro_rules! refresh_config {
                () => {{
                    let cfg = shared_config.load();
                    max_redelivery = cfg.max_redelivery_attempts;
                    dlq_suffix = cfg.dlq_topic_suffix.clone();
                    slow_consumer_policy = cfg.slow_consumer_policy;
                    slow_consumer_threshold = cfg.slow_consumer_threshold;
                    slow_consumer_disconnect_after = cfg.slow_consumer_disconnect_after;
                    cold_batch_size = cfg.cold_batch_size;
                    base_timeout = Duration::from_secs(cfg.inflight_timeout_secs);
                    max_backoff = Duration::from_secs(cfg.max_redelivery_backoff_secs);
                }};
            }

            loop {
                // --- Inflight timeout sweep ---
                // Scan for unacked messages past the inflight timeout.
                // Entries below max_redelivery get their delivery_count incremented
                // and timer reset. Entries exceeding the threshold are routed to DLQ.
                if last_sweep.elapsed() >= INFLIGHT_SWEEP_INTERVAL {
                    last_sweep = Instant::now();
                    // Sweep uses atomics so drain-accelerated timeouts take effect immediately.
                    let sweep_base =
                        Duration::from_millis(inflight_timeout_ms.load(Ordering::Acquire));
                    let sweep_max =
                        Duration::from_millis(max_redelivery_backoff_ms.load(Ordering::Acquire));
                    let now = Instant::now();

                    // Phase 1: Under lock, escalate below-threshold entries and
                    // snapshot above-threshold candidates for async DLQ routing.
                    // Expired messages (TTL elapsed) are routed to DLQ immediately
                    // without incrementing delivery_count.
                    let dlq_candidates = {
                        let mut state = topic_state.lock().await;
                        let Some(sub) = state.individual.get_mut(&client) else {
                            break;
                        };
                        let sweep_now_ms = now_millis();
                        let mut candidates: Vec<(TopicOffset, DlqCheckResult, bool)> = Vec::new();
                        for (&offset, entry) in sub.inflight.iter_mut() {
                            if now < entry.next_redeliver_at {
                                continue;
                            }
                            // Check TTL expiry first — expired messages go directly
                            // to DLQ without wasting a redelivery cycle.
                            if entry.is_expired(sweep_now_ms) {
                                candidates.push((
                                    offset,
                                    DlqCheckResult {
                                        delivery_count: entry.delivery_count,
                                        first_delivered_at: entry.first_delivered_at,
                                        dlq_failures: entry.dlq_failures,
                                    },
                                    true, // is_expired
                                ));
                                continue;
                            }
                            let new_count = entry.delivery_count + 1;
                            if new_count > max_redelivery {
                                // Needs DLQ routing — snapshot state, don't modify entry
                                // so a failed DLQ write retries immediately next sweep.
                                candidates.push((
                                    offset,
                                    DlqCheckResult {
                                        delivery_count: new_count,
                                        first_delivered_at: entry.first_delivered_at,
                                        dlq_failures: entry.dlq_failures,
                                    },
                                    false, // not expired, max attempts exceeded
                                ));
                            } else {
                                // Below threshold — escalate count, set next backoff
                                entry.delivery_count = new_count;
                                entry.delivered_at = Instant::now();
                                let delay = InflightEntry::redelivery_delay(
                                    new_count, sweep_base, sweep_max,
                                );
                                entry.next_redeliver_at = Instant::now() + delay;
                                metrics::counter!(
                                    "fafb_redelivery_total",
                                    "namespace" => ns.0.clone(),
                                    "topic" => topic.0.clone(),
                                )
                                .increment(1);
                                metrics::histogram!("fafb_redelivery_delay_seconds",)
                                    .record(delay.as_secs_f64());
                            }
                        }
                        candidates
                    };

                    // Phase 2: Route DLQ candidates (async, no lock held).
                    for (offset, check, expired) in dlq_candidates {
                        // Re-read the message from the engine for the DLQ envelope.
                        let msg = match engine_for_dlq.consume_from(&ns, &topic, offset).await {
                            Ok(Some(msg)) => msg,
                            Ok(None) | Err(_) => {
                                // Message no longer available — remove stale entry.
                                let mut state = topic_state.lock().await;
                                if let Some(sub) = state.individual.get_mut(&client) {
                                    tracing::warn!(
                                        namespace = %ns, topic = %topic,
                                        offset = %offset,
                                        "message unavailable for DLQ routing, \
                                         removing stale inflight entry"
                                    );
                                    sub.inflight.remove(&offset);
                                }
                                continue;
                            }
                        };

                        let reason = if expired {
                            DlqReason::MessageExpired
                        } else {
                            DlqReason::MaxAttemptsExceeded
                        };
                        let ctx = DlqContext::from_check(&check, offset, reason);
                        let result = Self::route_to_dlq(
                            &engine_for_dlq,
                            &ns,
                            &topic,
                            &dlq_suffix,
                            (*msg.envelope).clone(),
                            &ctx,
                        )
                        .await;
                        let mut state = topic_state.lock().await;
                        if let Some(sub) = state.individual.get_mut(&client) {
                            Self::handle_dlq_result(
                                result,
                                &ctx,
                                &ns,
                                &topic,
                                offset,
                                &mut sub.inflight,
                                Some("dlq"),
                            );
                        }
                    }
                }

                // Drain buffered cold-path messages first
                if let Some(ref mut batch) = cold_batch {
                    if let Some(msg) = batch.pop_front() {
                        let offset = msg.offset;

                        let tx = {
                            let state = topic_state.lock().await;
                            let Some(sub) = state.individual.get(&client) else {
                                break;
                            };
                            sub.tx.clone()
                        };

                        match cold_deliver_individual(
                            &msg,
                            offset,
                            &tx,
                            &topic_state,
                            client,
                            &ns,
                            &topic,
                            base_timeout,
                            max_backoff,
                        )
                        .await
                        {
                            ColdDeliverResult::Sent => {}
                            ColdDeliverResult::ChannelFull => {
                                batch.push_front(msg);
                                tokio::time::sleep(Duration::from_millis(10)).await;
                            }
                            ColdDeliverResult::ChannelClosed => break,
                        }
                        continue;
                    } else {
                        cold_batch = None;
                        // Fall through to normal path
                    }
                }

                // Read current offset and register notify interest (brief lock).
                // FIX 4: Register notified() BEFORE consuming to avoid race.
                let (offset, notified_future, tx, is_slow, has_inflight) = {
                    let state = topic_state.lock().await;
                    let Some(sub) = state.individual.get(&client) else {
                        break; // subscriber removed
                    };
                    (
                        sub.next_offset,
                        state.notify.clone().notified_owned(),
                        sub.tx.clone(),
                        sub.is_slow,
                        !sub.inflight.is_empty(),
                    )
                };

                // Slow consumer: check policy before attempting delivery
                if is_slow {
                    match slow_consumer_policy {
                        SlowConsumerPolicy::DropForSlow => {
                            // Check if we should skip delivery (is_slow already set)
                            // We still need to consume and advance the offset
                        }
                        SlowConsumerPolicy::DisconnectSlow => {
                            let should_disconnect = {
                                let state = topic_state.lock().await;
                                if let Some(sub) = state.individual.get(&client) {
                                    sub.total_drops >= slow_consumer_disconnect_after
                                } else {
                                    break;
                                }
                            };
                            if should_disconnect {
                                tracing::warn!(
                                    namespace = %ns, topic = %topic, client = %client,
                                    "disconnecting slow consumer (drop threshold exceeded)"
                                );
                                // Drop the sender to signal the connection to close
                                let mut state = topic_state.lock().await;
                                state.individual.remove(&client);
                                state.individual_handles.remove(&client);
                                break;
                            }
                        }
                    }
                }

                // Attempt to consume (no lock held)
                let msg = match engine.consume_from(&ns, &topic, offset).await {
                    Ok(Some(msg)) => {
                        refresh_config!();
                        msg
                    }
                    Ok(None) => {
                        // Distinguish "caught up" (at head) from "evicted behind" (below base).
                        let base = engine
                            .base_offset(&ns, &topic)
                            .await
                            .unwrap_or(TopicOffset::ZERO);
                        if offset < base {
                            // Try cold batch read before skipping
                            match engine
                                .read_cold_batch(&ns, &topic, offset, cold_batch_size)
                                .await
                            {
                                Ok(msgs) if !msgs.is_empty() => {
                                    let mut batch = VecDeque::from(msgs);
                                    if let Some(first_msg) = batch.pop_front() {
                                        let first_offset = first_msg.offset;
                                        match cold_deliver_individual(
                                            &first_msg,
                                            first_offset,
                                            &tx,
                                            &topic_state,
                                            client,
                                            &ns,
                                            &topic,
                                            base_timeout,
                                            max_backoff,
                                        )
                                        .await
                                        {
                                            ColdDeliverResult::Sent => {}
                                            ColdDeliverResult::ChannelFull => {
                                                batch.push_front(first_msg);
                                                tokio::time::sleep(Duration::from_millis(10)).await;
                                            }
                                            ColdDeliverResult::ChannelClosed => break,
                                        }
                                    }
                                    cold_batch = Some(batch);
                                    continue;
                                }
                                _ => {
                                    // No data on disk — skip forward (existing behavior)
                                    let mut state = topic_state.lock().await;
                                    if let Some(sub) = state.individual.get_mut(&client) {
                                        let gap = base.0 - sub.next_offset.0;
                                        tracing::warn!(
                                            namespace = %ns, topic = %topic, client = %client,
                                            gap = gap, new_offset = %base,
                                            "consumer skipping deleted offsets"
                                        );
                                        metrics::counter!("fafb_messages_skipped_total",
                                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                        )
                                        .increment(gap);
                                        sub.next_offset = base;
                                    }
                                    continue;
                                }
                            }
                        }
                        // Caught up — wait for notification, shutdown, or sweep timer.
                        // The notified future was created before consume, so we
                        // won't miss a publish that happened in between.
                        // When inflight entries exist, wake periodically to run
                        // the timeout sweep (escalate delivery_count / route to DLQ).
                        if has_inflight {
                            tokio::select! {
                                _ = notified_future => continue,
                                _ = shutdown_rx.changed() => break,
                                _ = tokio::time::sleep(INFLIGHT_SWEEP_INTERVAL) => continue,
                            }
                        } else {
                            tokio::select! {
                                _ = notified_future => continue,
                                _ = shutdown_rx.changed() => break,
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            namespace = %ns, topic = %topic, client = %client,
                            error = %e, "delivery task consume error"
                        );
                        tokio::time::sleep(Duration::from_millis(100)).await;
                        continue;
                    }
                };

                // Run delivery interceptors (skip clone when empty)
                let envelope = if !has_interceptors {
                    Arc::clone(&msg.envelope)
                } else {
                    let intercepted = Self::run_delivery_interceptors(
                        &delivery_interceptors,
                        &ns,
                        &topic,
                        client,
                        (*msg.envelope).clone(),
                    )
                    .await;
                    match intercepted {
                        Ok(Some(e)) => Arc::new(e),
                        Ok(None) | Err(_) => {
                            // Suppressed or error — advance offset past this message
                            let mut state = topic_state.lock().await;
                            if let Some(sub) = state.individual.get_mut(&client) {
                                sub.next_offset = offset.next();
                            }
                            continue;
                        }
                    }
                };

                // If subscriber is slow and policy is DropForSlow, skip delivery
                if is_slow && slow_consumer_policy == SlowConsumerPolicy::DropForSlow {
                    let mut state = topic_state.lock().await;
                    if let Some(sub) = state.individual.get_mut(&client) {
                        sub.next_offset = offset.next();
                        sub.total_drops += 1;
                    }
                    continue;
                }

                // Skip expired messages — advance offset without delivering
                let ttl_now = now_millis();
                if msg.envelope.is_expired(ttl_now) {
                    metrics::counter!(
                        "fafb_expired_skipped_total",
                        "namespace" => ns.0.clone(),
                        "topic" => topic.0.clone(),
                    )
                    .increment(1);
                    let mut state = topic_state.lock().await;
                    if let Some(sub) = state.individual.get_mut(&client) {
                        sub.next_offset = offset.next();
                        // Remove from inflight if it was a redelivery attempt
                        sub.inflight.remove(&offset);
                    }
                    continue;
                }

                // Record remaining TTL at delivery time
                if let Some(remaining) = msg.envelope.remaining_ttl(ttl_now) {
                    metrics::histogram!("fafb_message_remaining_ttl_seconds")
                        .record(remaining as f64 / 1000.0);
                }

                // Check redelivery count for DLQ routing (individual subscriber)
                {
                    let dlq_check = {
                        let state = topic_state.lock().await;
                        state.individual.get(&client).and_then(|sub| {
                            Self::check_dlq_needed(&sub.inflight, offset, max_redelivery)
                        })
                    };
                    if let Some(dlq_check) = dlq_check {
                        // Route to DLQ — lock was dropped before async write
                        let ctx = DlqContext::from_check(
                            &dlq_check,
                            offset,
                            DlqReason::MaxAttemptsExceeded,
                        );
                        let result = Self::route_to_dlq(
                            &engine_for_dlq,
                            &ns,
                            &topic,
                            &dlq_suffix,
                            (*msg.envelope).clone(),
                            &ctx,
                        )
                        .await;
                        let mut state = topic_state.lock().await;
                        if let Some(sub) = state.individual.get_mut(&client)
                            && Self::handle_dlq_result(
                                result,
                                &ctx,
                                &ns,
                                &topic,
                                offset,
                                &mut sub.inflight,
                                Some("dlq"),
                            )
                        {
                            sub.next_offset = offset.next();
                        }
                        continue;
                    }
                }

                let delivered_msg = PublishedMessage {
                    envelope,
                    offset: msg.offset,
                };

                // Try to send to subscriber
                match tx.try_send(delivered_msg) {
                    Ok(()) => {
                        tracing::trace!(
                            namespace = %ns, topic = %topic,
                            client = %client, offset = %msg.offset,
                            "message delivered (individual)"
                        );
                        metrics::counter!("fafb_messages_delivered_total", "namespace" => ns.0.clone(), "topic" => topic.0.clone(), "mode" => "individual").increment(1);
                        {
                            let now = now_millis();
                            let latency_ms = now.saturating_sub(msg.envelope.timestamp);
                            metrics::histogram!(
                                "fafb_delivery_latency_seconds",
                                "namespace" => ns.0.clone(),
                                "topic" => topic.0.clone()
                            )
                            .record(latency_ms as f64 / 1000.0);
                        }
                        // Update state: advance offset, add inflight
                        let mut state = topic_state.lock().await;
                        if let Some(sub) = state.individual.get_mut(&client) {
                            sub.next_offset = offset.next();
                            let now = now_millis();
                            let (delivery_count, first_delivered_at) = sub
                                .inflight
                                .get(&offset)
                                .map(|e| (e.delivery_count + 1, e.first_delivered_at))
                                .unwrap_or((1, now));
                            let delay = InflightEntry::redelivery_delay(
                                delivery_count,
                                base_timeout,
                                max_backoff,
                            );
                            sub.inflight.insert(
                                offset,
                                InflightEntry {
                                    offset,
                                    message_id: msg.envelope.id,
                                    delivered_to: client,
                                    delivered_at: Instant::now(),
                                    delivery_count,
                                    first_delivered_at,
                                    dlq_failures: 0,
                                    next_redeliver_at: Instant::now() + delay,
                                    msg_timestamp: msg.envelope.timestamp,
                                    msg_ttl_ms: msg.envelope.ttl_ms,
                                },
                            );
                            let inflight_count = sub.inflight.len();
                            metrics::gauge!("fafb_inflight_messages",
                                "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                            )
                            .set(inflight_count as f64);
                            // Warn when inflight crosses 1000-message thresholds
                            if inflight_count >= 1000 && inflight_count % 1000 == 0 {
                                tracing::warn!(
                                    namespace = %ns, topic = %topic, client = %client,
                                    inflight = inflight_count,
                                    "subscriber has high inflight count — messages not being ACKed"
                                );
                            }
                            sub.last_activity = Instant::now();
                            // Reset failure counter on successful send
                            sub.consecutive_send_failures = 0;
                            if sub.is_slow {
                                sub.is_slow = false;
                                tracing::info!(
                                    namespace = %ns, topic = %topic, client = %client,
                                    "subscriber recovered from slow state"
                                );
                            }
                        }
                    }
                    Err(mpsc::error::TrySendError::Full(_)) => {
                        metrics::counter!("fafb_messages_dropped_total", "namespace" => ns.0.clone(), "topic" => topic.0.clone(), "reason" => "channel_full").increment(1);
                        // Channel full — track failure and apply slow consumer detection
                        {
                            let mut state = topic_state.lock().await;
                            if let Some(sub) = state.individual.get_mut(&client) {
                                sub.consecutive_send_failures += 1;
                                if !sub.is_slow
                                    && sub.consecutive_send_failures > slow_consumer_threshold
                                {
                                    sub.is_slow = true;
                                    metrics::counter!("fafb_slow_consumers_total", "namespace" => ns.0.clone()).increment(1);
                                    tracing::warn!(
                                        namespace = %ns, topic = %topic, client = %client,
                                        consecutive_failures = sub.consecutive_send_failures,
                                        "subscriber marked as slow"
                                    );
                                }
                            }
                        }
                        tokio::time::sleep(Duration::from_millis(10)).await;
                    }
                    Err(mpsc::error::TrySendError::Closed(_)) => {
                        // Subscriber gone
                        break;
                    }
                }
            }
        })
    }

    /// Spawn a delivery task for a consumer group.
    /// Returns the JoinHandle so the caller can store it.
    fn spawn_group_delivery_task(
        ctx: DeliveryTaskContext<Q>,
        group_name: GroupName,
    ) -> JoinHandle<()> {
        let DeliveryTaskContext {
            engine,
            ns,
            topic,
            topic_state,
            delivery_interceptors,
            has_interceptors,
            shared_config,
            inflight_timeout_ms,
            max_redelivery_backoff_ms,
            mut shutdown_rx,
        } = ctx;
        let engine_for_dlq = Arc::clone(&engine);
        tokio::spawn(async move {
            let mut cold_batch: Option<VecDeque<PublishedMessage>> = None;
            let mut last_sweep = Instant::now();

            // Cache config values; refreshed when a message is consumed.
            // Initial load provides values for the first loop iteration (including
            // the inflight sweep which runs before any message is consumed).
            let cfg = shared_config.load();
            #[allow(unused_assignments)]
            let mut max_redelivery = cfg.max_redelivery_attempts;
            #[allow(unused_assignments)]
            let mut dlq_suffix = cfg.dlq_topic_suffix.clone();
            #[allow(unused_assignments)]
            let mut slow_consumer_policy = cfg.slow_consumer_policy;
            #[allow(unused_assignments)]
            let mut slow_consumer_threshold = cfg.slow_consumer_threshold;
            #[allow(unused_assignments)]
            let mut slow_consumer_disconnect_after = cfg.slow_consumer_disconnect_after;
            #[allow(unused_assignments)]
            let mut cold_batch_size = cfg.cold_batch_size;
            drop(cfg);
            // Backoff timeouts: cached per message consume, driven by SharedConfig.
            // The sweep reads atomics directly for drain-acceleration responsiveness.
            let mut base_timeout = Duration::from_secs(shared_config.load().inflight_timeout_secs);
            let mut max_backoff =
                Duration::from_secs(shared_config.load().max_redelivery_backoff_secs);
            macro_rules! refresh_config {
                () => {{
                    let cfg = shared_config.load();
                    max_redelivery = cfg.max_redelivery_attempts;
                    dlq_suffix = cfg.dlq_topic_suffix.clone();
                    slow_consumer_policy = cfg.slow_consumer_policy;
                    slow_consumer_threshold = cfg.slow_consumer_threshold;
                    slow_consumer_disconnect_after = cfg.slow_consumer_disconnect_after;
                    cold_batch_size = cfg.cold_batch_size;
                    base_timeout = Duration::from_secs(cfg.inflight_timeout_secs);
                    max_backoff = Duration::from_secs(cfg.max_redelivery_backoff_secs);
                }};
            }

            /// How often the group delivery task sweeps inflight entries
            /// for timed-out messages that need delivery_count escalation or DLQ routing.
            const INFLIGHT_SWEEP_INTERVAL: Duration = Duration::from_secs(1);

            loop {
                // --- Inflight timeout sweep (group) ---
                // Mirror of the individual delivery task sweep: scan for
                // unacked messages past their next_redeliver_at deadline.
                if last_sweep.elapsed() >= INFLIGHT_SWEEP_INTERVAL {
                    last_sweep = Instant::now();
                    // Sweep uses atomics so drain-accelerated timeouts take effect immediately.
                    let sweep_base =
                        Duration::from_millis(inflight_timeout_ms.load(Ordering::Acquire));
                    let sweep_max =
                        Duration::from_millis(max_redelivery_backoff_ms.load(Ordering::Acquire));
                    let now = Instant::now();

                    let dlq_candidates = {
                        let mut state = topic_state.lock().await;
                        let Some(group) = state.groups.get_mut(&group_name) else {
                            break;
                        };
                        let sweep_now_ms = now_millis();
                        let mut candidates: Vec<(TopicOffset, DlqCheckResult, bool)> = Vec::new();
                        for (&offset, entry) in group.inflight.iter_mut() {
                            if now < entry.next_redeliver_at {
                                continue;
                            }
                            // Check TTL expiry first — expired messages go directly
                            // to DLQ without wasting a redelivery cycle.
                            if entry.is_expired(sweep_now_ms) {
                                candidates.push((
                                    offset,
                                    DlqCheckResult {
                                        delivery_count: entry.delivery_count,
                                        first_delivered_at: entry.first_delivered_at,
                                        dlq_failures: entry.dlq_failures,
                                    },
                                    true, // is_expired
                                ));
                                continue;
                            }
                            let new_count = entry.delivery_count + 1;
                            if new_count > max_redelivery {
                                candidates.push((
                                    offset,
                                    DlqCheckResult {
                                        delivery_count: new_count,
                                        first_delivered_at: entry.first_delivered_at,
                                        dlq_failures: entry.dlq_failures,
                                    },
                                    false, // not expired, max attempts exceeded
                                ));
                            } else {
                                entry.delivery_count = new_count;
                                entry.delivered_at = Instant::now();
                                let delay = InflightEntry::redelivery_delay(
                                    new_count, sweep_base, sweep_max,
                                );
                                entry.next_redeliver_at = Instant::now() + delay;
                                metrics::counter!(
                                    "fafb_redelivery_total",
                                    "namespace" => ns.0.clone(),
                                    "topic" => topic.0.clone(),
                                )
                                .increment(1);
                                metrics::histogram!("fafb_redelivery_delay_seconds",)
                                    .record(delay.as_secs_f64());
                            }
                        }
                        candidates
                    };

                    // Phase 2: Route DLQ candidates (async, no lock held).
                    for (offset, check, expired) in dlq_candidates {
                        let msg = match engine_for_dlq.consume_from(&ns, &topic, offset).await {
                            Ok(Some(msg)) => msg,
                            Ok(None) | Err(_) => {
                                let mut state = topic_state.lock().await;
                                if let Some(group) = state.groups.get_mut(&group_name) {
                                    tracing::warn!(
                                        namespace = %ns, topic = %topic,
                                        group = %group_name, offset = %offset,
                                        "message unavailable for DLQ routing, \
                                         removing stale inflight entry"
                                    );
                                    group.inflight.remove(&offset);
                                }
                                continue;
                            }
                        };

                        let reason = if expired {
                            DlqReason::MessageExpired
                        } else {
                            DlqReason::MaxAttemptsExceeded
                        };
                        let ctx = DlqContext::from_check(&check, offset, reason);
                        let result = Self::route_to_dlq(
                            &engine_for_dlq,
                            &ns,
                            &topic,
                            &dlq_suffix,
                            (*msg.envelope).clone(),
                            &ctx,
                        )
                        .await;
                        let mut state = topic_state.lock().await;
                        if let Some(group) = state.groups.get_mut(&group_name) {
                            Self::handle_dlq_result(
                                result,
                                &ctx,
                                &ns,
                                &topic,
                                offset,
                                &mut group.inflight,
                                Some("dlq"),
                            );
                        }
                    }
                }

                // Drain buffered cold-path messages first
                if let Some(ref mut batch) = cold_batch {
                    if let Some(msg) = batch.pop_front() {
                        let offset = msg.offset;

                        // Pick a member to deliver to (brief lock)
                        let member_info = {
                            let state = topic_state.lock().await;
                            let Some(group) = state.groups.get(&group_name) else {
                                break;
                            };
                            group.pick_member().map(|(id, tx)| (*id, tx.clone()))
                        };

                        let Some((member_id, member_tx)) = member_info else {
                            batch.push_front(msg);
                            tokio::time::sleep(Duration::from_millis(50)).await;
                            continue;
                        };

                        match cold_deliver_group(
                            &msg,
                            offset,
                            member_id,
                            &member_tx,
                            &topic_state,
                            &group_name,
                            &ns,
                            &topic,
                            base_timeout,
                            max_backoff,
                        )
                        .await
                        {
                            ColdDeliverResult::Sent => {}
                            ColdDeliverResult::ChannelFull => {
                                batch.push_front(msg);
                                tokio::time::sleep(Duration::from_millis(10)).await;
                            }
                            ColdDeliverResult::ChannelClosed => {
                                // This member is gone — put message back and try another
                                batch.push_front(msg);
                            }
                        }
                        continue;
                    } else {
                        cold_batch = None;
                        // Fall through to normal path
                    }
                }

                // Read current offset, register notify, and pick member (brief lock).
                // FIX 4: Register notified() BEFORE consuming to avoid race.
                let (offset, notified_future, member_info, is_slow, has_inflight) = {
                    let state = topic_state.lock().await;
                    let Some(group) = state.groups.get(&group_name) else {
                        break; // group removed
                    };
                    if group.members.is_empty() {
                        // No members — wait for notification (but still wake for sweep)
                        let has_inflight = !group.inflight.is_empty();
                        let notified = state.notify.clone().notified_owned();
                        drop(state);
                        if has_inflight {
                            tokio::select! {
                                _ = notified => continue,
                                _ = shutdown_rx.changed() => break,
                                _ = tokio::time::sleep(INFLIGHT_SWEEP_INTERVAL) => continue,
                            }
                        } else {
                            tokio::select! {
                                _ = notified => continue,
                                _ = shutdown_rx.changed() => break,
                            }
                        }
                    }
                    let member = group.pick_member().map(|(id, tx)| (*id, tx.clone()));
                    (
                        group.next_offset,
                        state.notify.clone().notified_owned(),
                        member,
                        group.is_slow,
                        !group.inflight.is_empty(),
                    )
                };

                // Slow consumer: check policy before attempting delivery
                if is_slow {
                    match slow_consumer_policy {
                        SlowConsumerPolicy::DropForSlow => {
                            // We still need to consume and advance the offset
                        }
                        SlowConsumerPolicy::DisconnectSlow => {
                            let should_disconnect = {
                                let state = topic_state.lock().await;
                                if let Some(group) = state.groups.get(&group_name) {
                                    group.total_drops >= slow_consumer_disconnect_after
                                } else {
                                    break;
                                }
                            };
                            if should_disconnect {
                                tracing::warn!(
                                    namespace = %ns, topic = %topic, group = %group_name,
                                    "disconnecting slow consumer group (drop threshold exceeded)"
                                );
                                let mut state = topic_state.lock().await;
                                state.groups.remove(&group_name);
                                state.group_handles.remove(&group_name);
                                metrics::gauge!("fafb_consumer_lag",
                                    "namespace" => ns.0.clone(),
                                    "topic" => topic.0.clone(),
                                    "group" => group_name.0.clone()
                                )
                                .set(0.0);
                                break;
                            }
                        }
                    }
                }

                let Some((member_id, member_tx)) = member_info else {
                    // No members available
                    tokio::time::sleep(Duration::from_millis(50)).await;
                    continue;
                };

                // Attempt to consume (no lock held)
                let msg = match engine.consume_from(&ns, &topic, offset).await {
                    Ok(Some(msg)) => {
                        refresh_config!();
                        msg
                    }
                    Ok(None) => {
                        // Distinguish "caught up" from "evicted behind"
                        let base = engine
                            .base_offset(&ns, &topic)
                            .await
                            .unwrap_or(TopicOffset::ZERO);
                        if offset < base {
                            // Try cold batch read before skipping
                            match engine
                                .read_cold_batch(&ns, &topic, offset, cold_batch_size)
                                .await
                            {
                                Ok(msgs) if !msgs.is_empty() => {
                                    let mut batch = VecDeque::from(msgs);
                                    if let Some(first_msg) = batch.pop_front() {
                                        let first_offset = first_msg.offset;
                                        match cold_deliver_group(
                                            &first_msg,
                                            first_offset,
                                            member_id,
                                            &member_tx,
                                            &topic_state,
                                            &group_name,
                                            &ns,
                                            &topic,
                                            base_timeout,
                                            max_backoff,
                                        )
                                        .await
                                        {
                                            ColdDeliverResult::Sent => {}
                                            ColdDeliverResult::ChannelFull => {
                                                batch.push_front(first_msg);
                                                tokio::time::sleep(Duration::from_millis(10)).await;
                                            }
                                            ColdDeliverResult::ChannelClosed => {
                                                batch.push_front(first_msg);
                                            }
                                        }
                                    }
                                    cold_batch = Some(batch);
                                    continue;
                                }
                                _ => {
                                    // No data on disk — skip forward (existing behavior)
                                    let mut state = topic_state.lock().await;
                                    if let Some(group) = state.groups.get_mut(&group_name) {
                                        let gap = base.0 - group.next_offset.0;
                                        tracing::warn!(
                                            namespace = %ns, topic = %topic, group = %group_name,
                                            gap = gap, new_offset = %base,
                                            "consumer group skipping deleted offsets"
                                        );
                                        metrics::counter!("fafb_messages_skipped_total",
                                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                        )
                                        .increment(gap);
                                        group.next_offset = base;
                                    }
                                    continue;
                                }
                            }
                        }
                        if has_inflight {
                            tokio::select! {
                                _ = notified_future => continue,
                                _ = shutdown_rx.changed() => break,
                                _ = tokio::time::sleep(INFLIGHT_SWEEP_INTERVAL) => continue,
                            }
                        } else {
                            tokio::select! {
                                _ = notified_future => continue,
                                _ = shutdown_rx.changed() => break,
                            }
                        }
                    }
                    Err(e) => {
                        tracing::warn!(
                            namespace = %ns, topic = %topic, group = %group_name,
                            error = %e, "group delivery task consume error"
                        );
                        tokio::time::sleep(Duration::from_millis(100)).await;
                        continue;
                    }
                };

                // Run delivery interceptors (skip clone when empty)
                let envelope = if !has_interceptors {
                    Arc::clone(&msg.envelope)
                } else {
                    let intercepted = Self::run_delivery_interceptors(
                        &delivery_interceptors,
                        &ns,
                        &topic,
                        member_id,
                        (*msg.envelope).clone(),
                    )
                    .await;
                    match intercepted {
                        Ok(Some(e)) => Arc::new(e),
                        Ok(None) | Err(_) => {
                            let mut state = topic_state.lock().await;
                            if let Some(group) = state.groups.get_mut(&group_name) {
                                group.next_offset = offset.next();
                            }
                            continue;
                        }
                    }
                };

                // If group is slow and policy is DropForSlow, skip delivery
                if is_slow && slow_consumer_policy == SlowConsumerPolicy::DropForSlow {
                    let mut state = topic_state.lock().await;
                    if let Some(group) = state.groups.get_mut(&group_name) {
                        group.next_offset = offset.next();
                        group.total_drops += 1;
                    }
                    continue;
                }

                // Skip expired messages — advance offset without delivering
                let ttl_now = now_millis();
                if msg.envelope.is_expired(ttl_now) {
                    metrics::counter!(
                        "fafb_expired_skipped_total",
                        "namespace" => ns.0.clone(),
                        "topic" => topic.0.clone(),
                    )
                    .increment(1);
                    let mut state = topic_state.lock().await;
                    if let Some(group) = state.groups.get_mut(&group_name) {
                        group.next_offset = offset.next();
                        // Remove from inflight if it was a redelivery attempt
                        group.inflight.remove(&offset);
                    }
                    continue;
                }

                // Record remaining TTL at delivery time
                if let Some(remaining) = msg.envelope.remaining_ttl(ttl_now) {
                    metrics::histogram!("fafb_message_remaining_ttl_seconds")
                        .record(remaining as f64 / 1000.0);
                }

                // Check redelivery count for DLQ routing
                {
                    let dlq_check = {
                        let state = topic_state.lock().await;
                        state.groups.get(&group_name).and_then(|group| {
                            Self::check_dlq_needed(&group.inflight, offset, max_redelivery)
                        })
                    };
                    if let Some(dlq_check) = dlq_check {
                        // Route to DLQ — lock was dropped before async write
                        let ctx = DlqContext::from_check(
                            &dlq_check,
                            offset,
                            DlqReason::MaxAttemptsExceeded,
                        );
                        let result = Self::route_to_dlq(
                            &engine_for_dlq,
                            &ns,
                            &topic,
                            &dlq_suffix,
                            (*msg.envelope).clone(),
                            &ctx,
                        )
                        .await;
                        let mut state = topic_state.lock().await;
                        if let Some(group) = state.groups.get_mut(&group_name)
                            && Self::handle_dlq_result(
                                result,
                                &ctx,
                                &ns,
                                &topic,
                                offset,
                                &mut group.inflight,
                                Some("dlq"),
                            )
                        {
                            group.next_offset = offset.next();
                        }
                        continue;
                    }
                }

                let delivered_msg = PublishedMessage {
                    envelope,
                    offset: msg.offset,
                };

                // Try to send to selected member
                match member_tx.try_send(delivered_msg) {
                    Ok(()) => {
                        tracing::trace!(
                            namespace = %ns, topic = %topic,
                            offset = %msg.offset,
                            "message delivered (group)"
                        );
                        metrics::counter!("fafb_messages_delivered_total", "namespace" => ns.0.clone(), "topic" => topic.0.clone(), "mode" => "group").increment(1);
                        {
                            let now = now_millis();
                            let latency_ms = now.saturating_sub(msg.envelope.timestamp);
                            metrics::histogram!(
                                "fafb_delivery_latency_seconds",
                                "namespace" => ns.0.clone(),
                                "topic" => topic.0.clone()
                            )
                            .record(latency_ms as f64 / 1000.0);
                        }
                        let mut state = topic_state.lock().await;
                        if let Some(group) = state.groups.get_mut(&group_name) {
                            let now = now_millis();
                            let (delivery_count, first_delivered_at) = group
                                .inflight
                                .get(&offset)
                                .map(|e| (e.delivery_count + 1, e.first_delivered_at))
                                .unwrap_or((1, now));
                            let delay = InflightEntry::redelivery_delay(
                                delivery_count,
                                base_timeout,
                                max_backoff,
                            );
                            group.inflight.insert(
                                offset,
                                InflightEntry {
                                    offset,
                                    message_id: msg.envelope.id,
                                    delivered_to: member_id,
                                    delivered_at: Instant::now(),
                                    delivery_count,
                                    first_delivered_at,
                                    dlq_failures: 0,
                                    next_redeliver_at: Instant::now() + delay,
                                    msg_timestamp: msg.envelope.timestamp,
                                    msg_ttl_ms: msg.envelope.ttl_ms,
                                },
                            );
                            metrics::gauge!("fafb_inflight_messages",
                                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                                "group" => group_name.0.clone()
                            )
                            .set(group.inflight.len() as f64);
                            group.next_offset = offset.next();
                            group.last_activity = Instant::now();
                            // Reset failure counter on successful send
                            group.consecutive_send_failures = 0;
                            if group.is_slow {
                                group.is_slow = false;
                                tracing::info!(
                                    namespace = %ns, topic = %topic, group = %group_name,
                                    "consumer group recovered from slow state"
                                );
                            }
                        }
                    }
                    Err(mpsc::error::TrySendError::Full(_)) => {
                        metrics::counter!("fafb_messages_dropped_total", "namespace" => ns.0.clone(), "topic" => topic.0.clone(), "reason" => "channel_full").increment(1);
                        // Track failure and apply slow consumer detection
                        {
                            let mut state = topic_state.lock().await;
                            if let Some(group) = state.groups.get_mut(&group_name) {
                                group.consecutive_send_failures += 1;
                                if !group.is_slow
                                    && group.consecutive_send_failures > slow_consumer_threshold
                                {
                                    group.is_slow = true;
                                    metrics::counter!("fafb_slow_consumers_total", "namespace" => ns.0.clone()).increment(1);
                                    tracing::warn!(
                                        namespace = %ns, topic = %topic, group = %group_name,
                                        consecutive_failures = group.consecutive_send_failures,
                                        "consumer group marked as slow"
                                    );
                                }
                            }
                        }
                        tokio::time::sleep(Duration::from_millis(10)).await;
                    }
                    Err(mpsc::error::TrySendError::Closed(_)) => {
                        // This member is gone — loop will pick another
                        continue;
                    }
                }
            }
        })
    }

    /// Route a message to the dead-letter queue topic.
    ///
    /// Sets `_fafb.dlq.*` system headers on the message with provenance
    /// metadata and publishes the original payload (unwrapped) to the DLQ
    /// topic.
    ///
    /// Returns a [`DlqWriteResult`] with the offset and the DLQ timestamp
    /// (so the caller can record time-in-flight metrics with the exact same
    /// value). When DLQ is disabled (empty suffix), returns a sentinel result
    /// after logging a warning. On failure, returns `Err` so the caller can
    /// leave the message in inflight for retry.
    async fn route_to_dlq(
        engine: &Q,
        ns: &NamespaceId,
        topic: &TopicName,
        dlq_suffix: &str,
        msg: MessageEnvelope,
        ctx: &DlqContext,
    ) -> crate::error::Result<DlqWriteResult> {
        if dlq_suffix.is_empty() {
            tracing::warn!(
                namespace = %ns, topic = %topic, message_id = %msg.id,
                "message exceeded max redelivery attempts, dropping (DLQ disabled)"
            );
            return Ok(DlqWriteResult { dlq_timestamp: 0 });
        }
        let dlq_topic = TopicName(format!("{}{}", topic.0, dlq_suffix));
        let dlq_ts = now_millis();
        let headers = build_dlq_headers(
            topic,
            ctx.original_offset,
            ctx.delivery_count,
            &ctx.reason,
            ctx.first_delivered_at,
            dlq_ts,
            &msg.headers,
        );
        let dlq_msg = MessageEnvelope {
            topic: dlq_topic.clone(),
            headers,
            timestamp: dlq_ts,
            ttl_ms: 0, // DLQ messages never expire — TTL cleared on routing
            ..msg
        };
        let result = engine.publish(ns, &dlq_topic, dlq_msg).await?;
        tracing::info!(
            namespace = %ns, topic = %topic, dlq_topic = %dlq_topic,
            offset = %result.offset, delivery_count = ctx.delivery_count,
            "message routed to DLQ"
        );
        Ok(DlqWriteResult {
            dlq_timestamp: dlq_ts,
        })
    }

    /// Threshold of consecutive DLQ write failures before escalating to ERROR level.
    const DLQ_FAILURE_ESCALATION_THRESHOLD: u32 = 3;

    /// Check if a message's inflight entry has exceeded max redelivery attempts.
    /// Returns `Some(DlqCheckResult)` with a snapshot of the entry's state if
    /// the message should be routed to DLQ, `None` if delivery should proceed.
    ///
    /// Call this under lock, then drop the lock before performing the async
    /// DLQ write. Pass the result to `handle_dlq_result` after the write.
    /// Record delivery attempts histogram. Called once per message resolution
    /// (ack, nack, or auto-DLQ). `outcome` is one of "ack", "nack", "dlq".
    fn record_delivery_attempts(
        ns: &NamespaceId,
        topic: &TopicName,
        outcome: &'static str,
        delivery_count: u32,
    ) {
        metrics::histogram!(
            "fafb_delivery_attempts",
            "namespace" => ns.0.clone(),
            "topic" => topic.0.clone(),
            "outcome" => outcome
        )
        .record(delivery_count as f64);
    }

    fn check_dlq_needed(
        inflight: &HashMap<TopicOffset, InflightEntry>,
        offset: TopicOffset,
        max_redelivery: u32,
    ) -> Option<DlqCheckResult> {
        let entry = inflight.get(&offset)?;
        if entry.delivery_count <= max_redelivery {
            return None;
        }
        Some(DlqCheckResult {
            delivery_count: entry.delivery_count,
            first_delivered_at: entry.first_delivered_at,
            dlq_failures: entry.dlq_failures,
        })
    }

    /// Handle the result of a DLQ routing attempt, updating inflight state
    /// and emitting metrics. Returns `true` if the message was removed from
    /// inflight (successfully routed or DLQ disabled).
    ///
    /// `delivery_attempts_outcome`: when `Some`, records `fafb_delivery_attempts`
    /// histogram with the given outcome label. Callers pass `Some("dlq")` for
    /// auto-DLQ (max attempts) and `None` for nack-initiated DLQ (nack method
    /// records its own histogram). This makes the recording contract explicit
    /// at each call site rather than implicitly matching on `DlqReason`.
    fn handle_dlq_result(
        result: crate::error::Result<DlqWriteResult>,
        ctx: &DlqContext,
        ns: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
        inflight: &mut HashMap<TopicOffset, InflightEntry>,
        delivery_attempts_outcome: Option<&'static str>,
    ) -> bool {
        match result {
            Ok(write_result) => {
                inflight.remove(&offset);
                metrics::counter!(
                    "fafb_dlq_messages_total",
                    "namespace" => ns.0.clone(),
                    "source_topic" => topic.0.clone(),
                    "reason" => ctx.reason.reason_label()
                )
                .increment(1);
                // Record time-in-flight using the same timestamp from the envelope
                if write_result.dlq_timestamp > 0 && ctx.first_delivered_at > 0 {
                    let time_in_flight_secs = write_result
                        .dlq_timestamp
                        .saturating_sub(ctx.first_delivered_at)
                        as f64
                        / 1000.0;
                    metrics::histogram!(
                        "fafb_dlq_time_in_flight_seconds",
                        "namespace" => ns.0.clone(),
                        "source_topic" => topic.0.clone()
                    )
                    .record(time_in_flight_secs);
                }
                if let Some(outcome) = delivery_attempts_outcome {
                    Self::record_delivery_attempts(ns, topic, outcome, ctx.delivery_count);
                }
                true
            }
            Err(e) => {
                let failures = ctx.dlq_failures + 1;
                // Escalate to error after repeated failures
                if failures >= Self::DLQ_FAILURE_ESCALATION_THRESHOLD {
                    tracing::error!(
                        namespace = %ns, topic = %topic, offset = %offset,
                        error = %e, consecutive_failures = failures,
                        "DLQ write repeatedly failing, message stuck in inflight"
                    );
                } else {
                    tracing::warn!(
                        namespace = %ns, topic = %topic, offset = %offset,
                        error = %e, consecutive_failures = failures,
                        "DLQ write failed, message stays inflight for retry"
                    );
                }
                // Update the failure count on the inflight entry
                if let Some(entry) = inflight.get_mut(&offset) {
                    entry.dlq_failures = failures;
                }
                metrics::counter!(
                    "fafb_dlq_write_failures_total",
                    "namespace" => ns.0.clone(),
                    "source_topic" => topic.0.clone()
                )
                .increment(1);
                false
            }
        }
    }

    /// Run delivery interceptors in sequence. Returns `Ok(None)` if any
    /// interceptor suppresses delivery, `Err` on interceptor error.
    async fn run_delivery_interceptors(
        interceptors: &[Box<dyn DeliveryInterceptor>],
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        mut envelope: MessageEnvelope,
    ) -> Result<Option<MessageEnvelope>> {
        for interceptor in interceptors {
            match interceptor
                .intercept_delivery(ns, topic, client, envelope)
                .await
            {
                Ok(Some(e)) => envelope = e,
                Ok(None) => return Ok(None),
                Err(e) => {
                    tracing::warn!(
                        namespace = %ns, topic = %topic, client = %client,
                        error = %e, "delivery interceptor error"
                    );
                    return Err(e);
                }
            }
        }
        Ok(Some(envelope))
    }

    /// Clean up expired empty consumer groups on a topic (lazy cleanup).
    async fn cleanup_expired_groups(
        ns: &NamespaceId,
        topic: &TopicName,
        topic_state: &Mutex<TopicDeliveryState>,
    ) {
        let mut state = topic_state.lock().await;
        let expired: Vec<GroupName> = state
            .groups
            .iter()
            .filter(|(_, g)| {
                g.members.is_empty()
                    && g.emptied_at
                        .is_some_and(|t| t.elapsed() > EMPTY_GROUP_GRACE_PERIOD)
            })
            .map(|(name, _)| name.clone())
            .collect();
        for name in &expired {
            state.groups.remove(name);
            // Also remove the dead group handle
            if let Some(handle) = state.group_handles.remove(name) {
                handle.abort();
            }
            // Zero the consumer lag gauge so stale label combos don't persist.
            metrics::gauge!("fafb_consumer_lag",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                "group" => name.0.clone()
            )
            .set(0.0);
        }
    }

    /// Use the client_subscriptions reverse index to find what group (if any)
    /// a client belongs to for a specific (ns, topic) pair.
    async fn find_client_group(
        &self,
        client: ClientId,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> Option<GroupName> {
        let subs = self.client_subscriptions.read().await;
        subs.get(&client).and_then(|entries| {
            entries.iter().find_map(|e| {
                if e.namespace != *ns {
                    return None;
                }
                // Exact-match entry: topic_or_pattern matches directly.
                if e.topic_or_pattern == *topic {
                    return e.group.clone();
                }
                // Wildcard group entry: check if the concrete topic is in
                // the matched_topics list.
                if let Some(ref wstate) = e.wildcard
                    && e.group.is_some()
                    && wstate.matched_topics.iter().any(|t| t == topic)
                {
                    return e.group.clone();
                }
                None
            })
        })
    }

    /// Wildcard subscribe: subscribe a client to all existing topics matching
    /// a wildcard pattern. Messages from all matched topics are fan-in
    /// multiplexed onto a single delivery channel.
    async fn subscribe_wildcard(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
        start_from: SubscribeStart,
    ) -> Result<SubscribeResult> {
        // Reject duplicate wildcard subscription (same client + pattern + group).
        {
            let subs = self.client_subscriptions.read().await;
            if let Some(entries) = subs.get(&client)
                && entries.iter().any(|e| {
                    e.namespace == *ns
                        && e.topic_or_pattern == *topic
                        && e.group == group
                        && e.is_wildcard()
                })
            {
                return Err(Error::DuplicateSubscription {
                    client,
                    topic: topic.clone(),
                });
            }
        }

        // SubscribeAt (explicit offset) makes no sense for wildcards.
        if matches!(start_from, SubscribeStart::Offset(_)) {
            return Err(Error::InvalidOperation(
                "SubscribeAt with explicit offset is not supported for wildcard patterns; \
                 use Subscribe (head) or SubscribeFrom (timestamp)"
                    .into(),
            ));
        }

        if let Some(group_name) = group {
            return self
                .subscribe_wildcard_group(ns, topic, client, group_name, start_from)
                .await;
        }

        let pattern = TopicPattern::parse(&topic.0)?;
        let ns_state = self.get_namespace(ns).await?;

        // Collect all existing topics in this namespace from the engine.
        let all_topics = self.engine.list_topics(ns).await;

        // Filter to those matching the wildcard pattern.
        let matched: Vec<TopicName> = all_topics
            .into_iter()
            .filter(|t| pattern.matches(&t.0))
            .collect();

        if matched.len() > self.max_topics_per_wildcard {
            return Err(Error::ResourceExhausted(format!(
                "wildcard pattern '{}' matches {} topics, exceeding limit of {}",
                topic,
                matched.len(),
                self.max_topics_per_wildcard
            )));
        }

        // Allocate a subscription ID for the trie.
        let sub_id = SubscriptionId(self.next_sub_id.fetch_add(1, Ordering::Relaxed));

        // One shared (tx, rx) pair — all per-topic forwarders fan into this.
        let (tx, rx) = mpsc::channel(self.delivery_channel_capacity);

        // Subscribe to each matched topic individually.
        for matched_topic in &matched {
            let topic_state = Self::get_or_create_topic(&ns_state, matched_topic).await;

            // Resolve per-topic start offset.
            let head = self
                .engine
                .head_offset(ns, matched_topic)
                .await
                .unwrap_or(TopicOffset::ZERO);
            let start_offset = match start_from {
                SubscribeStart::Head => head,
                SubscribeStart::Timestamp(ts) => {
                    match self.engine.offset_at_timestamp(ns, matched_topic, ts).await {
                        Ok(Some(offset)) => offset.min(head),
                        _ => head,
                    }
                }
                SubscribeStart::Offset(_) => unreachable!(),
            };

            let mut state = topic_state.lock().await;
            // Skip if already subscribed (e.g. overlapping wildcard patterns).
            if state.individual.contains_key(&client) {
                continue;
            }
            state.individual.insert(
                client,
                SubscriberDeliveryState::new(start_offset, tx.clone()),
            );
            drop(state);

            // Spawn delivery task for this topic.
            let ctx = self.make_delivery_context(ns, matched_topic, &topic_state);
            let handle = Self::spawn_individual_delivery_task(ctx, client);
            {
                let mut state = topic_state.lock().await;
                state.individual_handles.insert(client, handle);
            }
        }

        // Register pattern in the namespace trie and wildcard reverse index.
        {
            let mut trie = ns_state.subscription_trie.write().await;
            trie.insert(&pattern, sub_id);
        }
        {
            let mut wsubs = ns_state.wildcard_subs.write().await;
            wsubs.insert(
                sub_id,
                WildcardSubInfo {
                    client,
                    pattern_topic: topic.clone(),
                    tx: tx.clone(),
                },
            );
        }

        // Register in reverse index with wildcard state embedded.
        {
            let mut subs = self.client_subscriptions.write().await;
            subs.entry(client).or_default().push(ClientSubEntry {
                namespace: ns.clone(),
                topic_or_pattern: topic.clone(),
                group: None,
                wildcard: Some(WildcardSubState {
                    pattern,
                    sub_id,
                    matched_topics: matched.clone(),
                }),
            });
        }

        let matched_count = matched.len();
        tracing::info!(
            namespace = %ns, pattern = %topic, client = %client,
            matched_topics = matched_count,
            "wildcard subscription created"
        );
        metrics::gauge!(
            "fafb_wildcard_subscriptions_active",
            "namespace" => ns.0.clone()
        )
        .increment(1.0);
        metrics::gauge!(
            "fafb_wildcard_match_topics",
            "namespace" => ns.0.clone(),
            "pattern" => topic.0.clone()
        )
        .set(matched_count as f64);

        // Wildcard has no single offset — return 0.
        Ok(SubscribeResult {
            rx,
            start_offset: TopicOffset::ZERO,
            matched_count: matched_count.min(u16::MAX as usize) as u16,
        })
    }

    /// Wildcard subscribe with consumer group: subscribe a client to a consumer
    /// group across all topics matching a wildcard pattern.
    ///
    /// If a wildcard group subscription for the same `(pattern, group)` already
    /// exists, the client joins it (added to all per-topic groups). Otherwise,
    /// a new wildcard group subscription is created with a trie entry.
    async fn subscribe_wildcard_group(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group_name: GroupName,
        start_from: SubscribeStart,
    ) -> Result<SubscribeResult> {
        let pattern = TopicPattern::parse(&topic.0)?;
        let ns_state = self.get_namespace(ns).await?;

        let cfg = self.shared_config.load();
        let max_consumer_groups_per_topic = cfg.max_consumer_groups_per_topic;
        drop(cfg);

        // Check if there's an existing wildcard group subscription for (pattern, group).
        let existing_sub_id = {
            let wgsubs = ns_state.wildcard_group_subs.read().await;
            wgsubs
                .iter()
                .find(|(_, info)| info.pattern_topic == *topic && info.group == group_name)
                .map(|(sub_id, _)| *sub_id)
        };

        let (tx, rx) = mpsc::channel(self.delivery_channel_capacity);

        if let Some(sub_id) = existing_sub_id {
            // ── Join existing wildcard group subscription ──────────
            // Add this client as a member to all per-topic consumer groups.
            let matched_topics = {
                let wgsubs = ns_state.wildcard_group_subs.read().await;
                wgsubs
                    .get(&sub_id)
                    .map(|info| info.matched_topics.clone())
                    .unwrap_or_default()
            };

            for matched_topic in &matched_topics {
                let topic_state = Self::get_or_create_topic(&ns_state, matched_topic).await;

                let mut state = topic_state.lock().await;
                if let Some(group) = state.groups.get_mut(&group_name) {
                    if let std::collections::hash_map::Entry::Vacant(e) =
                        group.members.entry(client)
                    {
                        e.insert(tx.clone());
                        group.emptied_at = None;
                        metrics::gauge!("fafb_consumer_group_members",
                            "namespace" => ns.0.clone(),
                            "topic" => matched_topic.0.clone(),
                            "group" => group_name.0.clone()
                        )
                        .set(group.members.len() as f64);
                    }
                    // Check if group delivery task needs respawn.
                    let need_spawn = state
                        .group_handles
                        .get(&group_name)
                        .is_none_or(|h| h.is_finished());
                    drop(state);
                    if need_spawn {
                        let ctx = self.make_delivery_context(ns, matched_topic, &topic_state);
                        let handle = Self::spawn_group_delivery_task(ctx, group_name.clone());
                        let mut state = topic_state.lock().await;
                        state.group_handles.insert(group_name.clone(), handle);
                    }
                }
            }

            // Add to wildcard_group_subs members map.
            {
                let mut wgsubs = ns_state.wildcard_group_subs.write().await;
                if let Some(info) = wgsubs.get_mut(&sub_id) {
                    info.members.insert(client, tx.clone());
                }
            }

            // Register in client reverse index.
            {
                let mut subs = self.client_subscriptions.write().await;
                subs.entry(client).or_default().push(ClientSubEntry {
                    namespace: ns.clone(),
                    topic_or_pattern: topic.clone(),
                    group: Some(group_name.clone()),
                    wildcard: Some(WildcardSubState {
                        pattern,
                        sub_id,
                        matched_topics: matched_topics.clone(),
                    }),
                });
            }

            let matched_count = matched_topics.len();
            // Emit wildcard group member count metric.
            let member_count = {
                let wgsubs = ns_state.wildcard_group_subs.read().await;
                wgsubs
                    .get(&sub_id)
                    .map(|info| info.members.len())
                    .unwrap_or(0)
            };
            tracing::info!(
                namespace = %ns, pattern = %topic, client = %client,
                group = %group_name, matched_topics = matched_count,
                members = member_count,
                "client joined wildcard consumer group"
            );
            metrics::gauge!(
                "fafb_wildcard_group_members",
                "namespace" => ns.0.clone(),
                "pattern" => topic.0.clone(),
                "group" => group_name.0.clone()
            )
            .set(member_count as f64);

            return Ok(SubscribeResult {
                rx,
                start_offset: TopicOffset::ZERO,
                matched_count: matched_count.min(u16::MAX as usize) as u16,
            });
        }

        // ── Create new wildcard group subscription ──────────────
        let all_topics = self.engine.list_topics(ns).await;
        let matched: Vec<TopicName> = all_topics
            .into_iter()
            .filter(|t| pattern.matches(&t.0))
            .collect();

        if matched.len() > self.max_topics_per_wildcard {
            return Err(Error::ResourceExhausted(format!(
                "wildcard pattern '{}' matches {} topics, exceeding limit of {}",
                topic,
                matched.len(),
                self.max_topics_per_wildcard
            )));
        }

        let sub_id = SubscriptionId(self.next_sub_id.fetch_add(1, Ordering::Relaxed));

        // For each matched topic, create/join the consumer group and add this client.
        for matched_topic in &matched {
            let topic_state = Self::get_or_create_topic(&ns_state, matched_topic).await;

            // Resolve per-topic start offset.
            let head = self
                .engine
                .head_offset(ns, matched_topic)
                .await
                .unwrap_or(TopicOffset::ZERO);
            let start_offset = match start_from {
                SubscribeStart::Head => head,
                SubscribeStart::Timestamp(ts) => {
                    match self.engine.offset_at_timestamp(ns, matched_topic, ts).await {
                        Ok(Some(offset)) => offset.min(head),
                        _ => head,
                    }
                }
                SubscribeStart::Offset(_) => unreachable!(),
            };

            let mut state = topic_state.lock().await;
            let need_spawn = if let Some(group) = state.groups.get_mut(&group_name) {
                // Existing group for this topic — add member.
                if let std::collections::hash_map::Entry::Vacant(e) = group.members.entry(client) {
                    e.insert(tx.clone());
                    group.emptied_at = None;
                }
                metrics::gauge!("fafb_consumer_group_members",
                    "namespace" => ns.0.clone(),
                    "topic" => matched_topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(group.members.len() as f64);
                state
                    .group_handles
                    .get(&group_name)
                    .is_none_or(|h| h.is_finished())
            } else {
                // New group for this topic — enforce limit.
                if state.groups.len() >= max_consumer_groups_per_topic {
                    tracing::warn!(
                        namespace = %ns, topic = %matched_topic,
                        group = %group_name,
                        "wildcard group subscribe: consumer group limit exceeded for topic"
                    );
                    continue;
                }
                let mut new_group = ConsumerGroupState::new(group_name.clone(), start_offset);
                new_group.members.insert(client, tx.clone());
                metrics::gauge!("fafb_consumer_group_members",
                    "namespace" => ns.0.clone(),
                    "topic" => matched_topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(1.0);
                state.groups.insert(group_name.clone(), new_group);
                true
            };
            drop(state);

            if need_spawn {
                let ctx = self.make_delivery_context(ns, matched_topic, &topic_state);
                let handle = Self::spawn_group_delivery_task(ctx, group_name.clone());
                let mut state = topic_state.lock().await;
                state.group_handles.insert(group_name.clone(), handle);
            }
        }

        // Register pattern in trie.
        {
            let mut trie = ns_state.subscription_trie.write().await;
            trie.insert(&pattern, sub_id);
        }

        // Register in wildcard_group_subs.
        {
            let mut wgsubs = ns_state.wildcard_group_subs.write().await;
            let mut members = HashMap::new();
            members.insert(client, tx.clone());
            wgsubs.insert(
                sub_id,
                WildcardGroupSubInfo {
                    pattern_topic: topic.clone(),
                    group: group_name.clone(),
                    members,
                    matched_topics: matched.clone(),
                },
            );
        }

        // Register in client reverse index.
        {
            let mut subs = self.client_subscriptions.write().await;
            subs.entry(client).or_default().push(ClientSubEntry {
                namespace: ns.clone(),
                topic_or_pattern: topic.clone(),
                group: Some(group_name.clone()),
                wildcard: Some(WildcardSubState {
                    pattern,
                    sub_id,
                    matched_topics: matched.clone(),
                }),
            });
        }

        let matched_count = matched.len();
        tracing::info!(
            namespace = %ns, pattern = %topic, client = %client,
            group = %group_name, matched_topics = matched_count,
            "wildcard consumer group subscription created"
        );
        metrics::gauge!(
            "fafb_wildcard_subscriptions_active",
            "namespace" => ns.0.clone()
        )
        .increment(1.0);
        metrics::gauge!(
            "fafb_wildcard_match_topics",
            "namespace" => ns.0.clone(),
            "pattern" => topic.0.clone()
        )
        .set(matched_count as f64);
        metrics::gauge!(
            "fafb_wildcard_group_members",
            "namespace" => ns.0.clone(),
            "pattern" => topic.0.clone(),
            "group" => group_name.0.clone()
        )
        .set(1.0);

        Ok(SubscribeResult {
            rx,
            start_offset: TopicOffset::ZERO,
            matched_count: matched_count.min(u16::MAX as usize) as u16,
        })
    }

    /// Attach a wildcard subscriber to a newly created topic.
    ///
    /// Creates a delivery forwarder for the new topic, fanning into the
    /// wildcard subscription's shared delivery channel. Called from the
    /// publish path when `topic_created == true` and the trie matches.
    ///
    /// Accepts only the fields it needs rather than a full `WildcardSubInfo`
    /// to avoid re-parsing the pattern or cloning structs under lock.
    #[allow(clippy::too_many_arguments)]
    async fn attach_wildcard_topic(
        &self,
        ns: &NamespaceId,
        ns_state: &Arc<NamespaceState>,
        sub_id: SubscriptionId,
        client: ClientId,
        tx: &mpsc::Sender<PublishedMessage>,
        pattern_topic: &TopicName,
        new_topic: &TopicName,
    ) {
        let topic_state = Self::get_or_create_topic(ns_state, new_topic).await;

        // Build context before locking topic_state — spawn is synchronous so
        // we can insert the subscriber and handle in a single lock scope.
        let ctx = self.make_delivery_context(ns, new_topic, &topic_state);

        {
            let mut state = topic_state.lock().await;
            // Skip if already subscribed (e.g. race or overlapping wildcard).
            if state.individual.contains_key(&client) {
                return;
            }
            // Start from offset 0 so the subscriber gets the very first
            // message that created this topic.
            state.individual.insert(
                client,
                SubscriberDeliveryState::new(TopicOffset::ZERO, tx.clone()),
            );
            let handle = Self::spawn_individual_delivery_task(ctx, client);
            state.individual_handles.insert(client, handle);
        }

        // Update matched_topics in client_subscriptions.
        {
            let mut subs = self.client_subscriptions.write().await;
            if let Some(entries) = subs.get_mut(&client) {
                for entry in entries.iter_mut() {
                    if let Some(ref mut wstate) = entry.wildcard
                        && wstate.sub_id == sub_id
                    {
                        wstate.matched_topics.push(new_topic.clone());
                        break;
                    }
                }
            }
        }

        tracing::info!(
            namespace = %ns, pattern = %pattern_topic,
            new_topic = %new_topic, client = %client,
            "wildcard auto-subscribed to new topic"
        );
    }

    /// Attach a wildcard consumer group to a newly created topic.
    ///
    /// Creates a consumer group for the new topic, adds all current members,
    /// and spawns a group delivery task. Called from the publish path when
    /// `topic_created == true` and the trie matches a group wildcard sub.
    #[allow(clippy::too_many_arguments)]
    async fn attach_wildcard_group_topic(
        &self,
        ns: &NamespaceId,
        ns_state: &Arc<NamespaceState>,
        sub_id: SubscriptionId,
        group_name: &GroupName,
        pattern_topic: &TopicName,
        new_topic: &TopicName,
        members: &HashMap<ClientId, mpsc::Sender<PublishedMessage>>,
    ) {
        if members.is_empty() {
            return;
        }

        let topic_state = Self::get_or_create_topic(ns_state, new_topic).await;

        let cfg = self.shared_config.load();
        let max_consumer_groups_per_topic = cfg.max_consumer_groups_per_topic;
        drop(cfg);

        let need_spawn = {
            let mut state = topic_state.lock().await;
            if state.groups.contains_key(group_name) {
                // Group already exists (e.g. race). Add missing members.
                let group = state.groups.get_mut(group_name).unwrap();
                for (client, tx) in members {
                    if !group.members.contains_key(client) {
                        group.members.insert(*client, tx.clone());
                    }
                }
                group.emptied_at = None;
                metrics::gauge!("fafb_consumer_group_members",
                    "namespace" => ns.0.clone(),
                    "topic" => new_topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(group.members.len() as f64);
                state
                    .group_handles
                    .get(group_name)
                    .is_none_or(|h| h.is_finished())
            } else {
                if state.groups.len() >= max_consumer_groups_per_topic {
                    tracing::warn!(
                        namespace = %ns, topic = %new_topic, group = %group_name,
                        "wildcard group auto-subscribe: consumer group limit exceeded"
                    );
                    return;
                }
                let mut new_group = ConsumerGroupState::new(group_name.clone(), TopicOffset::ZERO);
                for (client, tx) in members {
                    new_group.members.insert(*client, tx.clone());
                }
                metrics::gauge!("fafb_consumer_group_members",
                    "namespace" => ns.0.clone(),
                    "topic" => new_topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(new_group.members.len() as f64);
                state.groups.insert(group_name.clone(), new_group);
                true
            }
        };

        if need_spawn {
            let ctx = self.make_delivery_context(ns, new_topic, &topic_state);
            let handle = Self::spawn_group_delivery_task(ctx, group_name.clone());
            let mut state = topic_state.lock().await;
            state.group_handles.insert(group_name.clone(), handle);
        }

        // Update matched_topics in client_subscriptions for all members.
        {
            let mut subs = self.client_subscriptions.write().await;
            for client in members.keys() {
                if let Some(entries) = subs.get_mut(client) {
                    for entry in entries.iter_mut() {
                        if let Some(ref mut wstate) = entry.wildcard
                            && wstate.sub_id == sub_id
                        {
                            wstate.matched_topics.push(new_topic.clone());
                            break;
                        }
                    }
                }
            }
        }

        // Update wildcard_group_subs matched_topics.
        {
            let mut wgsubs = ns_state.wildcard_group_subs.write().await;
            if let Some(info) = wgsubs.get_mut(&sub_id) {
                info.matched_topics.push(new_topic.clone());
            }
        }

        tracing::info!(
            namespace = %ns, pattern = %pattern_topic,
            new_topic = %new_topic, group = %group_name,
            members = members.len(),
            "wildcard group auto-subscribed to new topic"
        );
    }

    /// Wildcard unsubscribe: remove a wildcard subscription and all its
    /// per-topic delivery forwarders.
    async fn unsubscribe_wildcard(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
    ) -> Result<()> {
        // Extract the wildcard state from client_subscriptions in one lock.
        let (wstate, is_group) = {
            let mut subs = self.client_subscriptions.write().await;
            let Some(entries) = subs.get_mut(&client) else {
                return Err(Error::NotSubscribed {
                    client,
                    topic: topic.clone(),
                });
            };
            let idx = entries.iter().position(|e| {
                e.namespace == *ns
                    && e.topic_or_pattern == *topic
                    && e.group == group
                    && e.is_wildcard()
            });
            let Some(idx) = idx else {
                return Err(Error::NotSubscribed {
                    client,
                    topic: topic.clone(),
                });
            };
            let entry = entries.remove(idx);
            if entries.is_empty() {
                subs.remove(&client);
            }
            let is_group = entry.group.is_some();
            (entry.wildcard.unwrap(), is_group)
        };

        let ns_state = self.get_namespace(ns).await?;

        if is_group {
            // ── Wildcard group unsubscribe ──────────────────────────
            let group_name = group.as_ref().unwrap();

            // Check if this is the last member in the wildcard group sub.
            let is_last_member = {
                let mut wgsubs = ns_state.wildcard_group_subs.write().await;
                if let Some(info) = wgsubs.get_mut(&wstate.sub_id) {
                    info.members.remove(&client);
                    info.members.is_empty()
                } else {
                    true
                }
            };

            // Remove client from per-topic consumer groups.
            for matched_topic in &wstate.matched_topics {
                let topics = ns_state.topics.read().await;
                if let Some(topic_state) = topics.get(matched_topic) {
                    let topic_state = Arc::clone(topic_state);
                    drop(topics);
                    let mut state = topic_state.lock().await;
                    if let Some(cg) = state.groups.get_mut(group_name) {
                        cg.members.remove(&client);
                        metrics::gauge!("fafb_consumer_group_members",
                            "namespace" => ns.0.clone(),
                            "topic" => matched_topic.0.clone(),
                            "group" => group_name.0.clone()
                        )
                        .set(cg.members.len() as f64);

                        // Release inflight entries from this client; reset the group
                        // cursor if any freed offset precedes it for redelivery.
                        let to_reassign: Vec<TopicOffset> = cg
                            .inflight
                            .iter()
                            .filter(|(_, entry)| entry.delivered_to == client)
                            .map(|(offset, _)| *offset)
                            .collect();
                        let min_freed = to_reassign.iter().copied().min();
                        for offset in to_reassign {
                            cg.inflight.remove(&offset);
                        }
                        if let Some(min) = min_freed
                            && min < cg.next_offset
                        {
                            cg.next_offset = min;
                        }

                        if cg.members.is_empty() {
                            cg.emptied_at = Some(Instant::now());
                        }
                    }
                    // Notify delivery task to re-deliver any freed inflight messages.
                    state.notify.notify_one();
                }
            }

            if is_last_member {
                // Last member — remove trie entry and group sub info.
                {
                    let mut trie = ns_state.subscription_trie.write().await;
                    trie.remove(&wstate.pattern, wstate.sub_id);
                }
                {
                    let mut wgsubs = ns_state.wildcard_group_subs.write().await;
                    wgsubs.remove(&wstate.sub_id);
                }
                metrics::gauge!(
                    "fafb_wildcard_subscriptions_active",
                    "namespace" => ns.0.clone()
                )
                .decrement(1.0);
                metrics::gauge!(
                    "fafb_wildcard_match_topics",
                    "namespace" => ns.0.clone(),
                    "pattern" => topic.0.clone()
                )
                .set(0.0);
            }

            // Update wildcard group member count metric.
            if is_last_member {
                metrics::gauge!(
                    "fafb_wildcard_group_members",
                    "namespace" => ns.0.clone(),
                    "pattern" => topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(0.0);
            } else {
                let member_count = {
                    let wgsubs = ns_state.wildcard_group_subs.read().await;
                    wgsubs
                        .get(&wstate.sub_id)
                        .map(|info| info.members.len())
                        .unwrap_or(0)
                };
                metrics::gauge!(
                    "fafb_wildcard_group_members",
                    "namespace" => ns.0.clone(),
                    "pattern" => topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(member_count as f64);
            }

            tracing::info!(
                namespace = %ns, pattern = %topic, client = %client,
                group = %group_name, is_last_member,
                "client left wildcard consumer group"
            );
        } else {
            // ── Individual wildcard unsubscribe ────────────────────
            // Remove from trie and wildcard reverse index.
            {
                let mut trie = ns_state.subscription_trie.write().await;
                trie.remove(&wstate.pattern, wstate.sub_id);
            }
            {
                let mut wsubs = ns_state.wildcard_subs.write().await;
                wsubs.remove(&wstate.sub_id);
            }

            // Remove subscriber from each matched topic.
            for matched_topic in &wstate.matched_topics {
                let topics = ns_state.topics.read().await;
                if let Some(topic_state) = topics.get(matched_topic) {
                    let topic_state = Arc::clone(topic_state);
                    drop(topics);
                    let mut state = topic_state.lock().await;
                    state.individual.remove(&client);
                    state.individual_handles.remove(&client);
                }
            }

            tracing::info!(
                namespace = %ns, pattern = %topic, client = %client,
                "wildcard subscription removed"
            );
            metrics::gauge!(
                "fafb_wildcard_subscriptions_active",
                "namespace" => ns.0.clone()
            )
            .decrement(1.0);
            metrics::gauge!(
                "fafb_wildcard_match_topics",
                "namespace" => ns.0.clone(),
                "pattern" => topic.0.clone()
            )
            .set(0.0);
        }

        Ok(())
    }

    /// Validate a publish: check topic name, quota limits, and run interceptors.
    /// Does NOT write to the queue engine. Returns the (possibly modified) envelope.
    ///
    /// Used by `ClusteredManager` to validate before replication.
    pub async fn validate_publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> Result<MessageEnvelope> {
        topic.validate_for_publish()?;

        let ns_state = self.get_namespace(ns).await?;

        let cfg = self.shared_config.load();
        let quotas = cfg.namespace_quotas.clone();
        let dlq_topic_suffix = cfg.dlq_topic_suffix.clone();
        drop(cfg);

        if let Some(quota) = quotas.get(ns) {
            if let Some(max_messages) = quota.max_total_messages {
                let current = ns_state
                    .message_count
                    .load(std::sync::atomic::Ordering::Relaxed);
                let threshold = max_messages * 4 / 5;
                if current >= threshold && current < max_messages {
                    if !ns_state
                        .messages_warning_fired
                        .swap(true, std::sync::atomic::Ordering::Relaxed)
                    {
                        tracing::warn!(
                            namespace = %ns, current, max_messages,
                            "namespace message count at 80% of quota"
                        );
                    }
                } else if current < threshold {
                    ns_state
                        .messages_warning_fired
                        .store(false, std::sync::atomic::Ordering::Relaxed);
                }
                if current >= max_messages {
                    tracing::warn!(
                        namespace = %ns, current, max_messages,
                        "publish rejected: namespace message quota exceeded"
                    );
                    metrics::counter!(
                        "fafb_quota_rejected_total",
                        "namespace" => ns.0.clone(), "reason" => "max_messages"
                    )
                    .increment(1);
                    return Err(Error::ResourceExhausted(format!(
                        "namespace message quota exceeded: {current} >= {max_messages}"
                    )));
                }
            }

            if let Some(max_bytes) = quota.max_disk_bytes {
                let current = self.engine.namespace_disk_bytes(ns).await;
                let threshold = max_bytes * 4 / 5;
                if current >= threshold && current < max_bytes {
                    if !ns_state
                        .disk_bytes_warning_fired
                        .swap(true, std::sync::atomic::Ordering::Relaxed)
                    {
                        tracing::warn!(
                            namespace = %ns,
                            current_bytes = current,
                            max_bytes,
                            "namespace disk usage at 80% of quota"
                        );
                    }
                } else if current < threshold {
                    ns_state
                        .disk_bytes_warning_fired
                        .store(false, std::sync::atomic::Ordering::Relaxed);
                }
                if current >= max_bytes {
                    tracing::warn!(
                        namespace = %ns,
                        current_bytes = current,
                        max_bytes,
                        "publish rejected: namespace disk quota exceeded"
                    );
                    metrics::counter!(
                        "fafb_quota_rejected_total",
                        "namespace" => ns.0.clone(), "reason" => "max_disk_bytes"
                    )
                    .increment(1);
                    return Err(Error::ResourceExhausted(format!(
                        "namespace disk quota exceeded: {current} bytes >= {max_bytes} byte limit"
                    )));
                }
            }

            if let Some(max_topics) = quota.max_topics {
                let existing_topics = self.engine.list_topics(ns).await;
                if !existing_topics.contains(topic) {
                    let current = ns_state
                        .topic_count
                        .load(std::sync::atomic::Ordering::Relaxed);
                    if current >= max_topics {
                        tracing::warn!(
                            namespace = %ns, current, max_topics,
                            "publish rejected: namespace topic quota exceeded"
                        );
                        metrics::counter!(
                            "fafb_quota_rejected_total",
                            "namespace" => ns.0.clone(), "reason" => "max_topics"
                        )
                        .increment(1);
                        return Err(Error::ResourceExhausted(format!(
                            "namespace topic quota exceeded: {current} >= {max_topics}"
                        )));
                    }
                }
            }
        }

        let mut msg = msg;
        if !topic.0.ends_with(&dlq_topic_suffix) {
            for interceptor in &self.publish_interceptors {
                msg = interceptor.intercept_publish(ns, topic, msg).await?;
            }
        }

        Ok(msg)
    }

    /// Write a validated message to the queue engine and update all post-publish state:
    /// namespace counters, wildcard auto-subscribe, delivery task notification.
    ///
    /// Returns the full `PublishResult` (needed by `ClusteredManager` for `topic_created`).
    pub async fn apply_publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> Result<PublishResult> {
        let ns_state = self.get_namespace(ns).await?;

        let publish_start = Instant::now();
        let result = self.engine.publish(ns, topic, msg).await?;
        metrics::histogram!("fafb_publish_latency_seconds", "namespace" => ns.0.clone())
            .record(publish_start.elapsed().as_secs_f64());
        let offset = result.offset;

        metrics::counter!("fafb_messages_published_total", "namespace" => ns.0.clone(), "topic" => topic.0.clone()).increment(1);

        if result.evicted_messages > 0 {
            ns_state
                .message_count
                .fetch_update(
                    std::sync::atomic::Ordering::Relaxed,
                    std::sync::atomic::Ordering::Relaxed,
                    |v| Some(v.saturating_sub(result.evicted_messages as u64) + 1),
                )
                .ok();
        } else {
            ns_state
                .message_count
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
        if result.topic_created {
            ns_state
                .topic_count
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);

            self.auto_subscribe_wildcards(ns, &ns_state, topic).await;
        }

        self.notify_delivery(ns, &ns_state, topic, offset).await;

        Ok(result)
    }

    /// Auto-subscribe wildcard subscribers to a newly created topic.
    async fn auto_subscribe_wildcards(
        &self,
        ns: &NamespaceId,
        ns_state: &Arc<NamespaceState>,
        topic: &TopicName,
    ) {
        let match_start = Instant::now();
        let matching_sub_ids = {
            let trie = ns_state.subscription_trie.read().await;
            if trie.is_empty() {
                Vec::new()
            } else {
                trie.matching(&topic.0)
            }
        };
        if matching_sub_ids.is_empty() {
            return;
        }

        metrics::histogram!("fafb_wildcard_match_duration_seconds")
            .record(match_start.elapsed().as_secs_f64());

        let individual_infos: Vec<(
            SubscriptionId,
            ClientId,
            TopicName,
            mpsc::Sender<PublishedMessage>,
        )> = {
            let wsubs = ns_state.wildcard_subs.read().await;
            matching_sub_ids
                .iter()
                .filter_map(|sub_id| {
                    wsubs.get(sub_id).map(|info| {
                        (
                            *sub_id,
                            info.client,
                            info.pattern_topic.clone(),
                            info.tx.clone(),
                        )
                    })
                })
                .collect()
        };

        let group_infos: Vec<WildcardGroupInfo> = {
            let wgsubs = ns_state.wildcard_group_subs.read().await;
            matching_sub_ids
                .iter()
                .filter_map(|sub_id| {
                    wgsubs.get(sub_id).map(|info| {
                        (
                            *sub_id,
                            info.pattern_topic.clone(),
                            info.group.clone(),
                            info.members.clone(),
                        )
                    })
                })
                .collect()
        };

        for (sub_id, client, pattern_topic, tx) in &individual_infos {
            let current_count = {
                let subs = self.client_subscriptions.read().await;
                subs.get(client)
                    .and_then(|entries| {
                        entries.iter().find_map(|e| {
                            e.wildcard
                                .as_ref()
                                .filter(|w| w.sub_id == *sub_id)
                                .map(|w| w.matched_topics.len())
                        })
                    })
                    .unwrap_or(0)
            };
            if current_count >= self.max_topics_per_wildcard {
                tracing::warn!(
                    namespace = %ns, pattern = %pattern_topic,
                    matched = current_count, limit = self.max_topics_per_wildcard,
                    "wildcard auto-subscribe skipped: max_topics_per_wildcard exceeded"
                );
                continue;
            }

            self.attach_wildcard_topic(ns, ns_state, *sub_id, *client, tx, pattern_topic, topic)
                .await;

            metrics::counter!(
                "fafb_wildcard_auto_subscribe_total",
                "namespace" => ns.0.clone()
            )
            .increment(1);

            metrics::gauge!(
                "fafb_wildcard_match_topics",
                "namespace" => ns.0.clone(),
                "pattern" => pattern_topic.0.clone()
            )
            .set((current_count + 1) as f64);
        }

        for (sub_id, pattern_topic, group_name, members) in &group_infos {
            let current_count = {
                let wgsubs = ns_state.wildcard_group_subs.read().await;
                wgsubs
                    .get(sub_id)
                    .map(|info| info.matched_topics.len())
                    .unwrap_or(0)
            };
            if current_count >= self.max_topics_per_wildcard {
                tracing::warn!(
                    namespace = %ns, pattern = %pattern_topic,
                    group = %group_name,
                    matched = current_count, limit = self.max_topics_per_wildcard,
                    "wildcard group auto-subscribe skipped: max_topics_per_wildcard exceeded"
                );
                continue;
            }

            self.attach_wildcard_group_topic(
                ns,
                ns_state,
                *sub_id,
                group_name,
                pattern_topic,
                topic,
                members,
            )
            .await;

            metrics::counter!(
                "fafb_wildcard_auto_subscribe_total",
                "namespace" => ns.0.clone()
            )
            .increment(1);

            metrics::gauge!(
                "fafb_wildcard_match_topics",
                "namespace" => ns.0.clone(),
                "pattern" => pattern_topic.0.clone()
            )
            .set((current_count + 1) as f64);
        }
    }

    /// Wake delivery tasks and update lag metrics after a publish.
    async fn notify_delivery(
        &self,
        ns: &NamespaceId,
        ns_state: &Arc<NamespaceState>,
        topic: &TopicName,
        offset: TopicOffset,
    ) {
        let topic_state = Self::get_or_create_topic(ns_state, topic).await;

        Self::cleanup_expired_groups(ns, topic, &topic_state).await;

        let new_head = offset.next();
        let group_lags: Vec<(GroupName, f64)> = {
            let mut state = topic_state.lock().await;
            state.cached_head_offset = new_head;
            state.notify.notify_waiters();
            state
                .groups
                .iter()
                .map(|(gn, group)| {
                    let lag = new_head.0.saturating_sub(group.next_offset.0)
                        + group.inflight.len() as u64;
                    (gn.clone(), lag as f64)
                })
                .collect()
        };
        for (gn, lag) in group_lags {
            metrics::gauge!("fafb_consumer_lag",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                "group" => gn.0.clone()
            )
            .set(lag);
        }
    }

    /// 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.
    pub async fn apply_commit_offset(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        group: &GroupName,
        offset: TopicOffset,
    ) -> Result<()> {
        let ns_state = self.get_namespace(ns).await?;
        let topic_state = Self::get_or_create_topic(&ns_state, topic).await;
        let mut state = topic_state.lock().await;

        if let Some(group_state) = state.groups.get_mut(group) {
            group_state.next_offset = offset;
        } else {
            // Create the group with the replicated offset.
            let mut group_state = ConsumerGroupState::new(group.clone(), offset);
            group_state.emptied_at = Some(Instant::now());
            state.groups.insert(group.clone(), group_state);
        }

        Ok(())
    }
}

impl<Q: QueueEngine + 'static> NamespaceManager for DefaultNamespaceManager<Q> {
    async fn connect(&self, ns: &NamespaceId, client: ClientId) -> Result<()> {
        let ns_state = self.get_namespace(ns).await?;

        // Load current quotas from shared config (hot-reload aware).
        let quotas = self.shared_config.load().namespace_quotas.clone();

        // Atomically check-and-increment connection count to avoid TOCTOU race.
        let new_count = if let Some(quota) = quotas.get(ns)
            && let Some(max_conn) = quota.max_connections
        {
            match ns_state.connection_count.fetch_update(
                Ordering::Relaxed,
                Ordering::Relaxed,
                |current| {
                    if current >= max_conn {
                        None // reject — CAS fails
                    } else {
                        Some(current + 1)
                    }
                },
            ) {
                Ok(prev) => prev + 1,
                Err(_) => {
                    metrics::counter!(
                        "fafb_connections_rejected_total",
                        "reason" => "namespace_limit"
                    )
                    .increment(1);
                    tracing::warn!(
                        namespace = %ns, client = %client,
                        "connection rejected: namespace connection limit reached"
                    );
                    return Err(Error::ResourceExhausted(
                        "namespace connection limit reached".to_owned(),
                    ));
                }
            }
        } else {
            ns_state.connection_count.fetch_add(1, Ordering::Relaxed) + 1
        };

        // Record client -> namespace mapping for disconnect
        {
            let mut cn = self.client_namespaces.write().await;
            cn.insert(client, ns.clone());
        }

        tracing::debug!(
            namespace = %ns, client = %client,
            active_connections = new_count,
            "client connected to namespace"
        );
        metrics::counter!("fafb_connections_total", "namespace" => ns.0.clone()).increment(1);
        metrics::gauge!("fafb_connections_active", "namespace" => ns.0.clone())
            .set(new_count as f64);

        // 80% connection threshold warning (only when a limit is configured)
        if let Some(quota) = quotas.get(ns)
            && let Some(max_conn) = quota.max_connections
        {
            let threshold = max_conn * 4 / 5;
            if new_count >= threshold && new_count < max_conn {
                if !ns_state
                    .connections_warning_fired
                    .swap(true, Ordering::Relaxed)
                {
                    tracing::warn!(
                        namespace = %ns,
                        current = new_count,
                        max = max_conn,
                        "namespace connections at 80% of limit"
                    );
                }
            } else if new_count < threshold {
                ns_state
                    .connections_warning_fired
                    .store(false, Ordering::Relaxed);
            }
        }

        Ok(())
    }

    async fn subscribe(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
        start_from: SubscribeStart,
    ) -> Result<SubscribeResult> {
        // ── Wildcard subscribe path ──────────────────────────────
        if contains_wildcard(&topic.0) {
            return self
                .subscribe_wildcard(ns, topic, client, group, start_from)
                .await;
        }

        // ── Exact-match subscribe path (unchanged) ──────────────
        let ns_state = self.get_namespace(ns).await?;
        let topic_state = Self::get_or_create_topic(&ns_state, topic).await;

        // Lazy cleanup of expired groups
        Self::cleanup_expired_groups(ns, topic, &topic_state).await;

        let head = self.engine.head_offset(ns, topic).await?;

        // Resolve starting offset based on start_from
        let start_offset = match start_from {
            SubscribeStart::Head => head,
            SubscribeStart::Offset(requested) => {
                let clamped = requested.min(head);
                let base = self.engine.base_offset(ns, topic).await?;
                if clamped < base {
                    tracing::warn!(
                        namespace = %ns, topic = %topic, requested = %requested,
                        available = %base, "requested offset not available, clamping to oldest"
                    );
                    base
                } else {
                    clamped
                }
            }
            SubscribeStart::Timestamp(ts) => {
                match self.engine.offset_at_timestamp(ns, topic, ts).await {
                    Ok(Some(offset)) => offset.min(head),
                    _ => head,
                }
            }
        };

        let cfg = self.shared_config.load();
        let max_consumer_groups_per_topic = cfg.max_consumer_groups_per_topic;
        drop(cfg);

        let (tx, rx) = mpsc::channel(self.delivery_channel_capacity);

        match group {
            None => {
                // Individual subscription
                let mut state = topic_state.lock().await;
                if state.individual.contains_key(&client) {
                    return Err(Error::DuplicateSubscription {
                        client,
                        topic: topic.clone(),
                    });
                }
                state
                    .individual
                    .insert(client, SubscriberDeliveryState::new(start_offset, tx));
                drop(state);

                // Register in reverse index
                {
                    let mut subs = self.client_subscriptions.write().await;
                    subs.entry(client).or_default().push(ClientSubEntry {
                        namespace: ns.clone(),
                        topic_or_pattern: topic.clone(),
                        group: None,
                        wildcard: None,
                    });
                }

                // Spawn delivery task and store handle (FIX 1)
                let ctx = self.make_delivery_context(ns, topic, &topic_state);
                let handle = Self::spawn_individual_delivery_task(ctx, client);
                {
                    let mut state = topic_state.lock().await;
                    state.individual_handles.insert(client, handle);
                    state.last_subscribe_millis = crate::types::now_millis();
                }
            }
            Some(ref group_name) => {
                // Consumer group subscription
                let mut state = topic_state.lock().await;
                let (need_spawn, actual_offset) = if let Some(group) =
                    state.groups.get_mut(group_name)
                {
                    // Existing group — add member at group's current cursor
                    if group.members.contains_key(&client) {
                        return Err(Error::DuplicateSubscription {
                            client,
                            topic: topic.clone(),
                        });
                    }
                    group.members.insert(client, tx);
                    group.emptied_at = None;
                    tracing::info!(
                        namespace = %ns, topic = %topic,
                        group = %group_name, client = %client,
                        members = group.members.len(),
                        "client joined consumer group"
                    );
                    metrics::gauge!("fafb_consumer_group_members",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "group" => group_name.0.clone()
                    )
                    .set(group.members.len() as f64);

                    let group_offset = group.next_offset;
                    if start_offset != group_offset {
                        tracing::info!(
                            namespace = %ns, topic = %topic, group = %group_name,
                            requested = %start_offset, actual = %group_offset,
                            "client joined existing group; requested offset ignored, using group cursor"
                        );
                    }

                    // FIX 2: Check if the existing group task is still alive.
                    // It may have exited when the group was temporarily empty.
                    let need_spawn = state
                        .group_handles
                        .get(group_name)
                        .is_none_or(|h| h.is_finished());
                    (need_spawn, group_offset)
                } else {
                    // New group — enforce consumer group limit per topic
                    if state.groups.len() >= max_consumer_groups_per_topic {
                        tracing::warn!(
                            namespace = %ns, topic = %topic,
                            current = state.groups.len(),
                            max = max_consumer_groups_per_topic,
                            "subscribe rejected: consumer group limit exceeded"
                        );
                        metrics::counter!(
                            "fafb_quota_rejected_total",
                            "namespace" => ns.0.clone(), "reason" => "max_consumer_groups"
                        )
                        .increment(1);
                        return Err(Error::ResourceExhausted(
                            "consumer group limit exceeded".to_owned(),
                        ));
                    }
                    let mut new_group = ConsumerGroupState::new(group_name.clone(), start_offset);
                    new_group.members.insert(client, tx);
                    tracing::info!(
                        namespace = %ns, topic = %topic,
                        group = %group_name, client = %client,
                        start_offset = %start_offset,
                        "consumer group created"
                    );
                    metrics::gauge!("fafb_consumer_group_members",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "group" => group_name.0.clone()
                    )
                    .set(new_group.members.len() as f64);
                    state.groups.insert(group_name.clone(), new_group);
                    (true, start_offset) // new group uses requested offset
                };
                drop(state);

                // Register in reverse index
                {
                    let mut subs = self.client_subscriptions.write().await;
                    subs.entry(client).or_default().push(ClientSubEntry {
                        namespace: ns.clone(),
                        topic_or_pattern: topic.clone(),
                        group: Some(group_name.clone()),
                        wildcard: None,
                    });
                }

                if need_spawn {
                    let ctx = self.make_delivery_context(ns, topic, &topic_state);
                    let handle = Self::spawn_group_delivery_task(ctx, group_name.clone());
                    let mut state = topic_state.lock().await;
                    state.group_handles.insert(group_name.clone(), handle);
                    state.last_subscribe_millis = crate::types::now_millis();
                } else {
                    // Existing task — wake it so it can deliver any pending messages
                    // to the newly-joined member (e.g. inflight cursor was just reset).
                    let mut state = topic_state.lock().await;
                    state.notify.notify_one();
                    state.last_subscribe_millis = crate::types::now_millis();
                }

                return Ok(SubscribeResult {
                    rx,
                    start_offset: actual_offset,
                    matched_count: 0,
                });
            }
        }

        Ok(SubscribeResult {
            rx,
            start_offset,
            matched_count: 0,
        })
    }

    async fn unsubscribe(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        group: Option<GroupName>,
    ) -> Result<()> {
        // ── Wildcard unsubscribe path ────────────────────────────
        if contains_wildcard(&topic.0) {
            return self.unsubscribe_wildcard(ns, topic, client, group).await;
        }

        // Check if the exact topic is part of a wildcard subscription —
        // if so, reject (must unsubscribe via the pattern).
        {
            let subs = self.client_subscriptions.read().await;
            if let Some(entries) = subs.get(&client) {
                let is_part_of_wildcard = entries.iter().any(|e| {
                    e.namespace == *ns
                        && e.is_wildcard()
                        && e.wildcard
                            .as_ref()
                            .is_some_and(|w| w.matched_topics.iter().any(|t| t == topic))
                });
                if is_part_of_wildcard {
                    return Err(Error::InvalidOperation(
                        "cannot unsubscribe from a specific topic within a wildcard subscription; \
                         unsubscribe using the wildcard pattern instead"
                            .into(),
                    ));
                }
            }
        }

        // ── Exact-match unsubscribe path ─────────────────────────
        let ns_state = self.get_namespace(ns).await?;
        let topics = ns_state.topics.read().await;
        let Some(topic_state) = topics.get(topic) else {
            return Err(Error::NotSubscribed {
                client,
                topic: topic.clone(),
            });
        };
        let topic_state = Arc::clone(topic_state);
        drop(topics);

        match group {
            None => {
                let mut state = topic_state.lock().await;
                if state.individual.remove(&client).is_none() {
                    return Err(Error::NotSubscribed {
                        client,
                        topic: topic.clone(),
                    });
                }
                // Task will detect closed channel and exit. Drop handle.
                state.individual_handles.remove(&client);
            }
            Some(ref group_name) => {
                let mut state = topic_state.lock().await;
                let notify = state.notify.clone();
                let Some(group) = state.groups.get_mut(group_name) else {
                    return Err(Error::GroupNotFound {
                        group: group_name.clone(),
                        topic: topic.clone(),
                    });
                };
                if group.members.remove(&client).is_none() {
                    return Err(Error::NotSubscribed {
                        client,
                        topic: topic.clone(),
                    });
                }
                metrics::gauge!("fafb_consumer_group_members",
                    "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                    "group" => group_name.0.clone()
                )
                .set(group.members.len() as f64);

                // Release inflight entries from this client and reset the group
                // cursor if any freed offset precedes it, enabling redelivery to
                // remaining group members.
                let to_reassign: Vec<TopicOffset> = group
                    .inflight
                    .iter()
                    .filter(|(_, entry)| entry.delivered_to == client)
                    .map(|(offset, _)| *offset)
                    .collect();
                let min_freed = to_reassign.iter().copied().min();
                for offset in to_reassign {
                    group.inflight.remove(&offset);
                }
                if let Some(min) = min_freed
                    && min < group.next_offset
                {
                    tracing::info!(
                        namespace = %ns, topic = %topic,
                        group = %group_name, reset_to = %min,
                        "resetting group cursor to redeliver inflight messages \
                         after client unsubscribe"
                    );
                    group.next_offset = min;
                }

                if group.members.is_empty() {
                    group.emptied_at = Some(Instant::now());
                    tracing::info!(
                        namespace = %ns, topic = %topic,
                        group = %group_name,
                        "consumer group is now empty"
                    );
                    metrics::gauge!("fafb_consumer_lag",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "group" => group_name.0.clone()
                    )
                    .set(0.0);
                }
                drop(state);
                // Notify the delivery task so it can promptly re-deliver freed messages.
                notify.notify_one();
            }
        }

        // Update reverse index
        {
            let mut subs = self.client_subscriptions.write().await;
            if let Some(entries) = subs.get_mut(&client) {
                entries.retain(|e| {
                    !(e.namespace == *ns && e.topic_or_pattern == *topic && e.group == group)
                });
                if entries.is_empty() {
                    subs.remove(&client);
                }
            }
        }

        Ok(())
    }

    async fn publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> Result<TopicOffset> {
        let msg = self.validate_publish(ns, topic, msg).await?;
        let result = self.apply_publish(ns, topic, msg).await?;
        Ok(result.offset)
    }

    // FIX 3 + FIX 8: Ack by MessageId, using reverse index for O(1) group lookup
    async fn ack(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        message_id: MessageId,
    ) -> Result<()> {
        let ns_state = self.get_namespace(ns).await?;
        let topics = ns_state.topics.read().await;
        let Some(topic_state) = topics.get(topic) else {
            return Err(Error::NotSubscribed {
                client,
                topic: topic.clone(),
            });
        };
        let topic_state = Arc::clone(topic_state);
        drop(topics);

        // FIX 8: Use reverse index to determine if this client is an
        // individual subscriber or a group member, avoiding O(groups) scan.
        let group_name = self.find_client_group(client, ns, topic).await;

        // Capture lag info under lock: (head_offset, next_offset, inflight_len, group_name).
        // Uses cached_head_offset set by the publish path — no engine lock needed.
        let mut lag_info: Option<(TopicOffset, TopicOffset, usize, GroupName)> = None;

        let min_offset = {
            let mut state = topic_state.lock().await;

            // FIX 5: Evict expired inflight entries before computing minimum.
            state.evict_expired_inflight();

            // Read cached head offset before mutable borrows into individual/group.
            let head_offset = state.cached_head_offset;

            match group_name {
                None => {
                    // Individual subscriber
                    let Some(sub) = state.individual.get_mut(&client) else {
                        return Err(Error::NotSubscribed {
                            client,
                            topic: topic.clone(),
                        });
                    };
                    // FIX 3: Find the inflight entry by MessageId
                    let found = sub
                        .inflight
                        .iter()
                        .find(|(_, entry)| entry.message_id == message_id)
                        .map(|(offset, entry)| (*offset, entry.delivery_count, entry.delivered_at));
                    if let Some((offset, delivery_count, delivered_at)) = found {
                        sub.inflight.remove(&offset);
                        metrics::counter!("fafb_messages_acked_total",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .increment(1);
                        Self::record_delivery_attempts(ns, topic, "ack", delivery_count);
                        metrics::histogram!("fafb_ack_latency_seconds",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .record(delivered_at.elapsed().as_secs_f64());
                    } else {
                        metrics::counter!("fafb_duplicate_ack_total",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .increment(1);
                    }
                    metrics::gauge!("fafb_inflight_messages",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                    )
                    .set(sub.inflight.len() as f64);
                    sub.last_activity = Instant::now();
                }
                Some(ref gn) => {
                    // Group member — go directly to the right group
                    let Some(group) = state.groups.get_mut(gn) else {
                        return Err(Error::GroupNotFound {
                            group: gn.clone(),
                            topic: topic.clone(),
                        });
                    };
                    let found = group
                        .inflight
                        .iter()
                        .find(|(_, entry)| entry.message_id == message_id)
                        .map(|(offset, entry)| (*offset, entry.delivery_count, entry.delivered_at));
                    if let Some((offset, delivery_count, delivered_at)) = found {
                        group.inflight.remove(&offset);
                        metrics::counter!("fafb_messages_acked_total",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .increment(1);
                        Self::record_delivery_attempts(ns, topic, "ack", delivery_count);
                        metrics::histogram!("fafb_ack_latency_seconds",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .record(delivered_at.elapsed().as_secs_f64());
                    } else {
                        metrics::counter!("fafb_duplicate_ack_total",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .increment(1);
                    }
                    metrics::gauge!("fafb_inflight_messages",
                        "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                        "group" => gn.0.clone()
                    )
                    .set(group.inflight.len() as f64);
                    lag_info = Some((
                        head_offset,
                        group.next_offset,
                        group.inflight.len(),
                        gn.clone(),
                    ));
                    group.last_activity = Instant::now();
                }
            }

            state.minimum_live_offset()
        };

        // Update consumer lag gauge using cached head_offset (no engine lock).
        if let Some((head, next_offset, inflight_len, gn)) = lag_info {
            let lag = head.0.saturating_sub(next_offset.0) + inflight_len as u64;
            metrics::gauge!("fafb_consumer_lag",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                "group" => gn.0.clone()
            )
            .set(lag as f64);
        }

        // Trigger compaction if minimum advanced
        if let Some(min) = min_offset
            && let Ok(result) = self.engine.compact(ns, topic, min).await
            && result.messages_freed > 0
        {
            ns_state
                .message_count
                .fetch_update(
                    std::sync::atomic::Ordering::Relaxed,
                    std::sync::atomic::Ordering::Relaxed,
                    |v| Some(v.saturating_sub(result.messages_freed as u64)),
                )
                .ok();
        }

        Ok(())
    }

    /// Optimized batch ack: groups by topic, acquires each topic lock once,
    /// removes all matching entries, evicts once, computes minimum once.
    async fn batch_ack(
        &self,
        ns: &NamespaceId,
        client: ClientId,
        acks: Vec<(MessageId, TopicName)>,
    ) -> crate::namespace::BatchAckResult {
        use std::collections::HashMap as StdHashMap;

        if acks.is_empty() {
            return crate::namespace::BatchAckResult {
                acked: 0,
                unknown: 0,
            };
        }

        // Group message_ids by topic for single-lock-per-topic processing.
        let mut by_topic: StdHashMap<TopicName, Vec<MessageId>> = StdHashMap::new();
        for (mid, topic) in acks {
            by_topic.entry(topic).or_default().push(mid);
        }

        let ns_state = match self.get_namespace(ns).await {
            Ok(s) => s,
            Err(_) => {
                let total = by_topic.values().map(|v| v.len() as u32).sum();
                return crate::namespace::BatchAckResult {
                    acked: 0,
                    unknown: total,
                };
            }
        };

        // Resolve group name once for this client+namespace (shared across all topics).
        // We use the first topic to check — group membership is per-namespace, not per-topic,
        // so any topic will give us the group name. For wildcard groups the lookup checks
        // matched_topics, so we pass each topic individually below if needed.
        let mut total_acked: u32 = 0;
        let mut total_unknown: u32 = 0;

        for (topic, message_ids) in &by_topic {
            let topics = ns_state.topics.read().await;
            let Some(topic_state) = topics.get(topic) else {
                total_unknown += message_ids.len() as u32;
                continue;
            };
            let topic_state = Arc::clone(topic_state);
            drop(topics);

            let group_name = self.find_client_group(client, ns, topic).await;

            // Capture lag info under lock: (head, next_offset, inflight_len, group_name).
            let mut topic_lag_info: Option<(TopicOffset, TopicOffset, usize, GroupName)> = None;

            let min_offset = {
                let mut state = topic_state.lock().await;
                state.evict_expired_inflight();

                let head_offset = state.cached_head_offset;

                match group_name {
                    None => {
                        let Some(sub) = state.individual.get_mut(&client) else {
                            total_unknown += message_ids.len() as u32;
                            continue;
                        };
                        for mid in message_ids {
                            let found = sub
                                .inflight
                                .iter()
                                .find(|(_, entry)| entry.message_id == *mid)
                                .map(|(offset, entry)| {
                                    (*offset, entry.delivery_count, entry.delivered_at)
                                });
                            if let Some((offset, delivery_count, delivered_at)) = found {
                                sub.inflight.remove(&offset);
                                metrics::counter!("fafb_messages_acked_total",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .increment(1);
                                Self::record_delivery_attempts(ns, topic, "ack", delivery_count);
                                metrics::histogram!("fafb_ack_latency_seconds",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .record(delivered_at.elapsed().as_secs_f64());
                                total_acked += 1;
                            } else {
                                metrics::counter!("fafb_duplicate_ack_total",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .increment(1);
                                total_unknown += 1;
                            }
                        }
                        metrics::gauge!("fafb_inflight_messages",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                        )
                        .set(sub.inflight.len() as f64);
                        sub.last_activity = Instant::now();
                    }
                    Some(ref gn) => {
                        let Some(group) = state.groups.get_mut(gn) else {
                            total_unknown += message_ids.len() as u32;
                            continue;
                        };
                        for mid in message_ids {
                            let found = group
                                .inflight
                                .iter()
                                .find(|(_, entry)| entry.message_id == *mid)
                                .map(|(offset, entry)| {
                                    (*offset, entry.delivery_count, entry.delivered_at)
                                });
                            if let Some((offset, delivery_count, delivered_at)) = found {
                                group.inflight.remove(&offset);
                                metrics::counter!("fafb_messages_acked_total",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .increment(1);
                                Self::record_delivery_attempts(ns, topic, "ack", delivery_count);
                                metrics::histogram!("fafb_ack_latency_seconds",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .record(delivered_at.elapsed().as_secs_f64());
                                total_acked += 1;
                            } else {
                                metrics::counter!("fafb_duplicate_ack_total",
                                    "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                )
                                .increment(1);
                                total_unknown += 1;
                            }
                        }
                        metrics::gauge!("fafb_inflight_messages",
                            "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                            "group" => gn.0.clone()
                        )
                        .set(group.inflight.len() as f64);
                        topic_lag_info = Some((
                            head_offset,
                            group.next_offset,
                            group.inflight.len(),
                            gn.clone(),
                        ));
                        group.last_activity = Instant::now();
                    }
                }

                state.minimum_live_offset()
            };

            // Update consumer lag gauge using cached head_offset (no engine lock).
            if let Some((head, next_offset, inflight_len, gn)) = topic_lag_info {
                let lag = head.0.saturating_sub(next_offset.0) + inflight_len as u64;
                metrics::gauge!("fafb_consumer_lag",
                    "namespace" => ns.0.clone(), "topic" => topic.0.clone(),
                    "group" => gn.0.clone()
                )
                .set(lag as f64);
            }

            // Trigger compaction if minimum advanced.
            if let Some(min) = min_offset
                && let Ok(result) = self.engine.compact(ns, topic, min).await
                && result.messages_freed > 0
            {
                ns_state
                    .message_count
                    .fetch_update(
                        std::sync::atomic::Ordering::Relaxed,
                        std::sync::atomic::Ordering::Relaxed,
                        |v| Some(v.saturating_sub(result.messages_freed as u64)),
                    )
                    .ok();
            }
        }

        crate::namespace::BatchAckResult {
            acked: total_acked,
            unknown: total_unknown,
        }
    }

    async fn nack(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        client: ClientId,
        message_id: MessageId,
        reason: String,
    ) -> Result<NackResult> {
        let cfg = self.shared_config.load();
        let dlq_suffix = cfg.dlq_topic_suffix.clone();

        let ns_state = self.get_namespace(ns).await?;
        let topics = ns_state.topics.read().await;
        let Some(topic_state) = topics.get(topic) else {
            return Err(Error::NotSubscribed {
                client,
                topic: topic.clone(),
            });
        };
        let topic_state = Arc::clone(topic_state);
        drop(topics);

        let group_name = self.find_client_group(client, ns, topic).await;

        // Phase 1: Find the inflight entry and snapshot its state (under lock).
        // Both DLQ-enabled and DLQ-disabled paths require the message to exist
        // in inflight — error consistently if not found.
        let (offset, dlq_ctx) = {
            let state = topic_state.lock().await;
            let inflight = match &group_name {
                None => {
                    let Some(sub) = state.individual.get(&client) else {
                        return Err(Error::NotSubscribed {
                            client,
                            topic: topic.clone(),
                        });
                    };
                    &sub.inflight
                }
                Some(gn) => {
                    let Some(group) = state.groups.get(gn) else {
                        return Err(Error::GroupNotFound {
                            group: gn.clone(),
                            topic: topic.clone(),
                        });
                    };
                    &group.inflight
                }
            };

            let Some((&offset, entry)) = inflight
                .iter()
                .find(|(_, entry)| entry.message_id == message_id)
            else {
                // message_id not in inflight — already acked/expired
                return Err(Error::Codec(format!(
                    "nack: message_id {message_id} not in inflight"
                )));
            };

            let ctx = DlqContext {
                original_offset: offset,
                first_delivered_at: entry.first_delivered_at,
                delivery_count: entry.delivery_count,
                dlq_failures: entry.dlq_failures,
                reason: DlqReason::ClientNack { reason },
            };
            (offset, ctx)
        };

        // DLQ disabled — drop the message from inflight.
        if dlq_suffix.is_empty() {
            let mut state = topic_state.lock().await;
            let inflight = match &group_name {
                None => state.individual.get_mut(&client).map(|s| &mut s.inflight),
                Some(gn) => state.groups.get_mut(gn).map(|g| &mut g.inflight),
            };
            if let Some(inflight) = inflight {
                inflight.remove(&offset);
            }
            tracing::warn!(
                namespace = %ns, topic = %topic, message_id = %message_id,
                "nack: message dropped (DLQ disabled)"
            );
            metrics::counter!(
                "fafb_nack_total",
                "namespace" => ns.0.clone(),
                "topic" => topic.0.clone()
            )
            .increment(1);
            Self::record_delivery_attempts(ns, topic, "nack", dlq_ctx.delivery_count);
            return Ok(NackResult::Dropped);
        }

        // Phase 2: Re-read message from engine (no lock held).
        let msg = self.engine.consume_from(ns, topic, offset).await?;
        let Some(msg) = msg else {
            // Message evicted — remove stale inflight entry and treat as dropped.
            let mut state = topic_state.lock().await;
            match &group_name {
                None => {
                    if let Some(sub) = state.individual.get_mut(&client) {
                        sub.inflight.remove(&offset);
                    }
                }
                Some(gn) => {
                    if let Some(group) = state.groups.get_mut(gn) {
                        group.inflight.remove(&offset);
                    }
                }
            }
            tracing::warn!(
                namespace = %ns, topic = %topic, offset = %offset,
                "nack: message unavailable, removing stale inflight entry"
            );
            Self::record_delivery_attempts(ns, topic, "nack", dlq_ctx.delivery_count);
            return Ok(NackResult::Dropped);
        };

        // Phase 3: Route to DLQ.
        let result = Self::route_to_dlq(
            &self.engine,
            ns,
            topic,
            &dlq_suffix,
            (*msg.envelope).clone(),
            &dlq_ctx,
        )
        .await;

        // Capture original error before handle_dlq_result consumes it for logging.
        let original_err_msg = match &result {
            Err(e) => Some(e.to_string()),
            Ok(_) => None,
        };

        // Phase 4: Update inflight state (re-acquire lock).
        let mut state = topic_state.lock().await;
        let inflight = match &group_name {
            None => state.individual.get_mut(&client).map(|s| &mut s.inflight),
            Some(gn) => state.groups.get_mut(gn).map(|g| &mut g.inflight),
        };
        let Some(inflight) = inflight else {
            // Subscriber/group disappeared while we were routing — DLQ write
            // already happened, nothing to clean up in inflight.
            return match result {
                Ok(_) => Ok(NackResult::DeadLettered),
                Err(e) => Err(e),
            };
        };

        // If the inflight entry was removed by a concurrent ack/sweep between
        // Phase 1 and Phase 4, the DLQ write already completed — just report
        // the outcome without calling handle_dlq_result on a missing entry.
        // NOTE: We do NOT record fafb_delivery_attempts here because the
        // concurrent ack or DLQ sweep that removed the entry already recorded it.
        if !inflight.contains_key(&offset) {
            return match result {
                Ok(_) => {
                    metrics::counter!(
                        "fafb_nack_total",
                        "namespace" => ns.0.clone(),
                        "topic" => topic.0.clone()
                    )
                    .increment(1);
                    Ok(NackResult::DeadLettered)
                }
                Err(e) => Err(e),
            };
        }

        let removed = Self::handle_dlq_result(result, &dlq_ctx, ns, topic, offset, inflight, None);
        if removed {
            metrics::counter!(
                "fafb_nack_total",
                "namespace" => ns.0.clone(),
                "topic" => topic.0.clone()
            )
            .increment(1);
            Self::record_delivery_attempts(ns, topic, "nack", dlq_ctx.delivery_count);
            Ok(NackResult::DeadLettered)
        } else {
            // handle_dlq_result already logged + updated dlq_failures on the entry.
            // Propagate the original engine error rather than a generic message.
            Err(Error::Codec(
                original_err_msg.unwrap_or_else(|| "DLQ write failed".into()),
            ))
        }
    }

    async fn disconnect(&self, client: &ClientId) -> Result<()> {
        // Decrement per-namespace connection count
        let ns_for_client = {
            let mut cn = self.client_namespaces.write().await;
            cn.remove(client)
        };
        if let Some(ref ns_id) = ns_for_client
            && let Ok(ns_state) = self.get_namespace(ns_id).await
        {
            // Saturating subtraction to avoid underflow
            let prev = ns_state
                .connection_count
                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
                    Some(v.saturating_sub(1))
                })
                .unwrap_or(0);
            let new_count = prev.saturating_sub(1);
            metrics::gauge!("fafb_connections_active", "namespace" => ns_id.0.clone())
                .set(new_count as f64);
        }

        // Look up all subscriptions for this client
        let entries = {
            let mut subs = self.client_subscriptions.write().await;
            subs.remove(client).unwrap_or_default()
        };

        for entry in entries {
            let ns_id = &entry.namespace;

            // ── Wildcard subscription cleanup ────────────────────
            if let Some(wstate) = &entry.wildcard {
                if let Ok(ns_state) = self.get_namespace(ns_id).await {
                    if let Some(group_name) = entry.group.as_ref() {
                        // ── Wildcard group member disconnect ────────

                        // Remove from wildcard_group_subs members.
                        let is_last_member = {
                            let mut wgsubs = ns_state.wildcard_group_subs.write().await;
                            if let Some(info) = wgsubs.get_mut(&wstate.sub_id) {
                                info.members.remove(client);
                                info.members.is_empty()
                            } else {
                                true
                            }
                        };

                        // Remove from per-topic consumer groups.
                        for matched_topic in &wstate.matched_topics {
                            let topics = ns_state.topics.read().await;
                            if let Some(topic_state) = topics.get(matched_topic) {
                                let topic_state = Arc::clone(topic_state);
                                drop(topics);
                                let mut state = topic_state.lock().await;
                                if let Some(cg) = state.groups.get_mut(group_name) {
                                    cg.members.remove(client);
                                    metrics::gauge!("fafb_consumer_group_members",
                                        "namespace" => ns_id.0.clone(),
                                        "topic" => matched_topic.0.clone(),
                                        "group" => group_name.0.clone()
                                    )
                                    .set(cg.members.len() as f64);

                                    let to_reassign: Vec<TopicOffset> = cg
                                        .inflight
                                        .iter()
                                        .filter(|(_, ie)| ie.delivered_to == *client)
                                        .map(|(offset, _)| *offset)
                                        .collect();
                                    let min_freed = to_reassign.iter().copied().min();
                                    for offset in to_reassign {
                                        cg.inflight.remove(&offset);
                                    }
                                    if let Some(min) = min_freed
                                        && min < cg.next_offset
                                    {
                                        cg.next_offset = min;
                                    }

                                    if cg.members.is_empty() {
                                        cg.emptied_at = Some(Instant::now());
                                        metrics::gauge!("fafb_consumer_lag",
                                            "namespace" => ns_id.0.clone(),
                                            "topic" => matched_topic.0.clone(),
                                            "group" => group_name.0.clone()
                                        )
                                        .set(0.0);
                                    }
                                }
                                // Notify the delivery task so it can promptly re-deliver
                                // any inflight messages freed by this disconnect.
                                state.notify.notify_one();
                            }
                        }

                        if is_last_member {
                            let mut trie = ns_state.subscription_trie.write().await;
                            trie.remove(&wstate.pattern, wstate.sub_id);
                            drop(trie);

                            let mut wgsubs = ns_state.wildcard_group_subs.write().await;
                            wgsubs.remove(&wstate.sub_id);
                            drop(wgsubs);

                            metrics::gauge!(
                                "fafb_wildcard_subscriptions_active",
                                "namespace" => ns_id.0.clone()
                            )
                            .decrement(1.0);
                            metrics::gauge!(
                                "fafb_wildcard_match_topics",
                                "namespace" => ns_id.0.clone(),
                                "pattern" => entry.topic_or_pattern.0.clone()
                            )
                            .set(0.0);
                            metrics::gauge!(
                                "fafb_wildcard_group_members",
                                "namespace" => ns_id.0.clone(),
                                "pattern" => entry.topic_or_pattern.0.clone(),
                                "group" => group_name.0.clone()
                            )
                            .set(0.0);
                        } else {
                            let member_count = {
                                let wgsubs = ns_state.wildcard_group_subs.read().await;
                                wgsubs
                                    .get(&wstate.sub_id)
                                    .map(|info| info.members.len())
                                    .unwrap_or(0)
                            };
                            metrics::gauge!(
                                "fafb_wildcard_group_members",
                                "namespace" => ns_id.0.clone(),
                                "pattern" => entry.topic_or_pattern.0.clone(),
                                "group" => group_name.0.clone()
                            )
                            .set(member_count as f64);
                        }
                    } else {
                        // ── Individual wildcard disconnect ──────────
                        let mut trie = ns_state.subscription_trie.write().await;
                        trie.remove(&wstate.pattern, wstate.sub_id);
                        drop(trie);

                        {
                            let mut wsubs = ns_state.wildcard_subs.write().await;
                            wsubs.remove(&wstate.sub_id);
                        }

                        for matched_topic in &wstate.matched_topics {
                            let topics = ns_state.topics.read().await;
                            if let Some(topic_state) = topics.get(matched_topic) {
                                let topic_state = Arc::clone(topic_state);
                                drop(topics);
                                let mut state = topic_state.lock().await;
                                state.individual.remove(client);
                                state.individual_handles.remove(client);
                            }
                        }

                        metrics::gauge!(
                            "fafb_wildcard_subscriptions_active",
                            "namespace" => ns_id.0.clone()
                        )
                        .decrement(1.0);
                        metrics::gauge!(
                            "fafb_wildcard_match_topics",
                            "namespace" => ns_id.0.clone(),
                            "pattern" => entry.topic_or_pattern.0.clone()
                        )
                        .set(0.0);
                    }
                }
                continue;
            }

            // ── Exact-match subscription cleanup ─────────────────
            let topic_name = &entry.topic_or_pattern;
            let group = &entry.group;

            let Ok(ns_state) = self.get_namespace(ns_id).await else {
                continue;
            };
            let topics = ns_state.topics.read().await;
            let Some(topic_state) = topics.get(topic_name) else {
                continue;
            };
            let topic_state = Arc::clone(topic_state);
            drop(topics);

            let mut state = topic_state.lock().await;
            match group {
                None => {
                    state.individual.remove(client);
                    state.individual_handles.remove(client);
                }
                Some(group_name) => {
                    if let Some(group) = state.groups.get_mut(group_name) {
                        group.members.remove(client);
                        metrics::gauge!("fafb_consumer_group_members",
                            "namespace" => ns_id.0.clone(), "topic" => topic_name.0.clone(),
                            "group" => group_name.0.clone()
                        )
                        .set(group.members.len() as f64);

                        // Release inflight entries from this client and reset the
                        // group cursor if any freed offset precedes it, so the
                        // delivery task will re-deliver them to another member.
                        let to_reassign: Vec<TopicOffset> = group
                            .inflight
                            .iter()
                            .filter(|(_, entry)| entry.delivered_to == *client)
                            .map(|(offset, _)| *offset)
                            .collect();
                        let min_freed = to_reassign.iter().copied().min();
                        for offset in to_reassign {
                            group.inflight.remove(&offset);
                        }
                        if let Some(min) = min_freed
                            && min < group.next_offset
                        {
                            tracing::info!(
                                namespace = %ns_id, topic = %topic_name,
                                group = %group_name, reset_to = %min,
                                "resetting group cursor to redeliver inflight messages \
                                 after client disconnect"
                            );
                            group.next_offset = min;
                        }

                        if group.members.is_empty() {
                            group.emptied_at = Some(Instant::now());
                            tracing::info!(
                                namespace = %ns_id, topic = %topic_name,
                                group = %group_name,
                                "consumer group is now empty (client disconnected)"
                            );
                            // Zero lag gauge — group will be reaped by cleanup_expired_groups.
                            metrics::gauge!("fafb_consumer_lag",
                                "namespace" => ns_id.0.clone(),
                                "topic" => topic_name.0.clone(),
                                "group" => group_name.0.clone()
                            )
                            .set(0.0);
                        }
                    }
                }
            }
            // Notify the group delivery task so it can promptly re-deliver any
            // inflight messages whose cursor was just reset.
            state.notify.notify_one();
        }

        Ok(())
    }

    async fn list_topics(&self, ns: &NamespaceId) -> Result<Vec<TopicName>> {
        let ns_state = self.get_namespace(ns).await?;

        // Merge topics from the queue engine (have published data) and the
        // namespace state (have subscribers, may not have data yet).
        let mut seen = std::collections::HashSet::new();
        for t in self.engine.list_topics(ns).await {
            seen.insert(t);
        }
        let ns_topics = ns_state.topics.read().await;
        for t in ns_topics.keys() {
            seen.insert(t.clone());
        }

        let mut result: Vec<TopicName> = seen.into_iter().collect();
        result.sort_by(|a, b| a.0.cmp(&b.0));
        Ok(result)
    }

    async fn topic_info(&self, ns: &NamespaceId, topic: &TopicName) -> Result<TopicInfo> {
        let ns_state = self.get_namespace(ns).await?;
        let stats = self
            .engine
            .topic_stats(ns, topic)
            .await
            .ok_or_else(|| Error::TopicNotFound(topic.clone()))?;

        let topics = ns_state.topics.read().await;
        let (subscriber_count, consumer_group_count) = if let Some(topic_state) = topics.get(topic)
        {
            let state = topic_state.lock().await;
            let individual = state.individual.len();
            let groups = state.groups.len();
            let group_members: usize = state.groups.values().map(|g| g.members.len()).sum();
            (individual + group_members, groups)
        } else {
            (0, 0)
        };

        Ok(TopicInfo {
            depth: stats.depth,
            bytes_used: stats.bytes_used,
            base_offset: stats.base_offset,
            head_offset: stats.head_offset,
            subscriber_count,
            consumer_group_count,
            oldest_timestamp_millis: stats.oldest_timestamp,
            newest_timestamp_millis: stats.newest_timestamp,
        })
    }

    async fn head_offset(&self, ns: &NamespaceId, topic: &TopicName) -> Result<TopicOffset> {
        self.engine.head_offset(ns, topic).await
    }

    async fn fetch(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
        max_count: u32,
        max_response_bytes: usize,
        headers_enabled: bool,
    ) -> Result<FetchResult> {
        metrics::counter!("fafb_fetch_requests_total",
            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
        )
        .increment(1);

        // 1. Validate namespace exists
        let ns_state = self.get_namespace(ns).await?;

        // 2. Get head offset — if offset >= head, return empty result
        let head = self.engine.head_offset(ns, topic).await?;
        if offset >= head {
            return Ok(FetchResult {
                messages: Vec::new(),
                next_offset: head,
                head_offset: head,
                clamped_from: None,
            });
        }

        // 3. Get base offset — if offset < base, record clamping
        let base = self.engine.base_offset(ns, topic).await?;
        let mut clamped_from = None;
        let effective_offset = if offset < base {
            clamped_from = Some(offset);
            base
        } else {
            offset
        };

        // 4. Read messages via consume_from (handles both hot and cold paths
        //    internally — cold reads delegate to storage when offset < base).
        let max = max_count as usize;
        let mut messages = Vec::new();
        let mut total_bytes: usize = 0;
        let mut current_offset = effective_offset;
        // Scan budget: cap total messages scanned (including expired) to avoid
        // unbounded I/O when a large range of messages has expired.
        let scan_budget = max.saturating_mul(10).max(100);
        let mut scanned: usize = 0;

        while messages.len() < max && scanned < scan_budget {
            match self.engine.consume_from(ns, topic, current_offset).await? {
                Some(msg) => {
                    scanned += 1;
                    // Skip expired messages — advance offset past them
                    let ttl_now = now_millis();
                    if msg.envelope.is_expired(ttl_now) {
                        metrics::counter!(
                            "fafb_expired_skipped_fetch_total",
                            "namespace" => ns.0.clone(),
                            "topic" => topic.0.clone(),
                        )
                        .increment(1);
                        current_offset = msg.offset.next();
                        continue;
                    }
                    // Record remaining TTL at fetch time
                    if let Some(remaining) = msg.envelope.remaining_ttl(ttl_now) {
                        metrics::histogram!("fafb_message_remaining_ttl_seconds")
                            .record(remaining as f64 / 1000.0);
                    }
                    // Wire overhead: base + header section (only when client supports headers)
                    let header_wire_bytes = if headers_enabled && !msg.envelope.headers.is_empty() {
                        crate::codec::frame::FETCH_MESSAGE_HEADER_OVERHEAD
                            + msg
                                .envelope
                                .headers
                                .iter()
                                .map(|h| {
                                    crate::codec::frame::PER_HEADER_OVERHEAD
                                        + h.key.len()
                                        + h.value.len()
                                })
                                .sum::<usize>()
                    } else {
                        0
                    };
                    let msg_bytes = crate::codec::frame::FETCH_MESSAGE_OVERHEAD
                        + msg.envelope.payload.len()
                        + header_wire_bytes;
                    if total_bytes + msg_bytes > max_response_bytes {
                        if messages.is_empty() {
                            metrics::counter!(
                                "fafb_fetch_message_too_large_total",
                                "namespace" => ns.0.clone(),
                                "topic" => topic.0.clone()
                            )
                            .increment(1);
                            return Err(Error::MessageTooLarge {
                                offset: current_offset,
                                message_bytes: msg.envelope.payload.len(),
                                budget_bytes: max_response_bytes,
                            });
                        }
                        break;
                    }
                    total_bytes += msg_bytes;
                    current_offset = msg.offset.next();
                    messages.push(msg);
                }
                None => break,
            }
        }

        // 5. Compute next_offset
        let next_offset = if let Some(last) = messages.last() {
            last.offset.next()
        } else {
            head
        };

        let count = messages.len();
        metrics::histogram!("fafb_fetch_messages_returned").record(count as f64);

        tracing::debug!(
            namespace = %ns,
            topic = %topic,
            requested_offset = %offset,
            effective_offset = %effective_offset,
            returned = count,
            next_offset = %next_offset,
            head_offset = %head,
            clamped = clamped_from.is_some(),
            "fetch completed"
        );

        // Stamp subscribe activity if delivery state already exists.
        // Fetch is stateless — it does NOT create TopicDeliveryState.
        {
            let topics = ns_state.topics.read().await;
            if let Some(topic_lock) = topics.get(topic) {
                topic_lock.lock().await.last_subscribe_millis = crate::types::now_millis();
            }
        }

        Ok(FetchResult {
            messages,
            next_offset,
            head_offset: head,
            clamped_from,
        })
    }

    async fn min_live_offset(&self, ns: &NamespaceId, topic: &TopicName) -> Option<TopicOffset> {
        let ns_state = self.get_namespace(ns).await.ok()?;
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(topic)?;
        let state = topic_state.lock().await;

        let mut min: Option<TopicOffset> = None;

        // Check individual subscribers
        for sub in state.individual.values() {
            min = Some(match min {
                Some(current) => current.min(sub.next_offset),
                None => sub.next_offset,
            });
        }

        // Check consumer groups
        for group in state.groups.values() {
            min = Some(match min {
                Some(current) => current.min(group.next_offset),
                None => group.next_offset,
            });
        }

        min
    }

    async fn offset_at_timestamp(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        timestamp_millis: u64,
    ) -> Result<Option<TopicOffset>> {
        self.engine
            .offset_at_timestamp(ns, topic, timestamp_millis)
            .await
    }

    async fn commit_offset(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        group: &GroupName,
        offset: TopicOffset,
    ) -> Result<()> {
        self.apply_commit_offset(ns, topic, group, offset).await
    }

    async fn shutdown(&self) -> Result<()> {
        // Signal all delivery tasks to stop
        let _ = self.shutdown_tx.send(true);

        // Collect and abort all handles across all namespaces/topics
        let namespaces = self.namespaces.read().await;
        for ns_state in namespaces.values() {
            let topics = ns_state.topics.read().await;
            for topic_state in topics.values() {
                let mut state = topic_state.lock().await;
                state.abort_all_handles();
            }
        }

        Ok(())
    }

    async fn topic_last_subscribe(&self, ns: &NamespaceId, topic: &TopicName) -> Option<u64> {
        let ns_state = self.get_namespace(ns).await.ok()?;
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(topic)?;
        let state = topic_state.lock().await;
        Some(state.last_subscribe_millis)
    }

    async fn has_active_subscribers(&self, ns: &NamespaceId, topic: &TopicName) -> bool {
        let Ok(ns_state) = self.get_namespace(ns).await else {
            return false;
        };
        let topics = ns_state.topics.read().await;
        let Some(topic_lock) = topics.get(topic) else {
            return false;
        };
        let state = topic_lock.lock().await;
        !state.individual.is_empty() || state.groups.values().any(|g| !g.members.is_empty())
    }

    async fn has_inflight_messages(&self, ns: &NamespaceId, topic: &TopicName) -> bool {
        let Ok(ns_state) = self.get_namespace(ns).await else {
            return false;
        };
        let topics = ns_state.topics.read().await;
        let Some(topic_lock) = topics.get(topic) else {
            return false;
        };
        let state = topic_lock.lock().await;
        state.individual.values().any(|s| !s.inflight.is_empty())
            || state.groups.values().any(|g| !g.inflight.is_empty())
    }

    async fn teardown_topic(&self, ns: &NamespaceId, topic: &TopicName) -> Result<()> {
        let ns_state = self.get_namespace(ns).await?;

        let mut topics = ns_state.topics.write().await;
        if let Some(topic_lock) = topics.remove(topic) {
            let state = topic_lock.lock().await;
            // Abort all delivery task handles
            for handle in state.individual_handles.values() {
                handle.abort();
            }
            for handle in state.group_handles.values() {
                handle.abort();
            }
            drop(state);
            // Defensive: prevent underflow if topic_count drifted
            let prev = ns_state
                .topic_count
                .load(std::sync::atomic::Ordering::Relaxed);
            if prev > 0 {
                ns_state
                    .topic_count
                    .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::queue::InMemoryQueueEngine;
    use crate::queue::engine::EngineConfig;
    use crate::queue::topic::BackpressurePolicy;
    use crate::storage::mock::MockStorage;
    use crate::types::now_millis;
    use std::pin::Pin;

    fn test_shared_config() -> crate::config::SharedConfig {
        crate::config::shared_config(&crate::config::ServerConfig::default())
    }

    /// Build a SharedConfig with a custom quota for the "ns" namespace.
    fn shared_config_with_quota(quota: NamespaceQuota) -> crate::config::SharedConfig {
        let mut rt = crate::config::RuntimeConfig::from_server_config(
            &crate::config::ServerConfig::default(),
        );
        rt.namespace_quotas.insert(NamespaceId("ns".into()), quota);
        std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
    }

    /// Build a SharedConfig with custom RuntimeConfig field overrides.
    fn shared_config_with(
        f: impl FnOnce(&mut crate::config::RuntimeConfig),
    ) -> crate::config::SharedConfig {
        let mut rt = crate::config::RuntimeConfig::from_server_config(
            &crate::config::ServerConfig::default(),
        );
        f(&mut rt);
        std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
    }

    fn make_engine() -> Arc<InMemoryQueueEngine<MockStorage>> {
        Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10_000,
                backpressure: BackpressurePolicy::Reject,
                ..Default::default()
            },
        ))
    }

    fn make_manager(
        engine: Arc<InMemoryQueueEngine<MockStorage>>,
    ) -> DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>> {
        DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            test_shared_config(),
            ManagerConfig::default(),
        )
    }

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

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

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

    // -- Construction tests --

    #[test]
    fn construction_initializes_namespaces() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![
                NamespaceId("a".into()),
                NamespaceId("b".into()),
                NamespaceId("c".into()),
            ],
            test_shared_config(),
            ManagerConfig::default(),
        );
        // Config values now come from SharedConfig
        let cfg = mgr.shared_config.load();
        assert_eq!(cfg.max_redelivery_attempts, 5);
        assert_eq!(cfg.dlq_topic_suffix, "__dlq");
    }

    #[test]
    fn manager_is_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>>>();
    }

    // -- Publish tests --

    #[tokio::test]
    async fn publish_to_configured_namespace() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        let offset = mgr
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "ns", "test-topic", b"hello"),
            )
            .await
            .unwrap();
        assert_eq!(offset, TopicOffset(0));

        let offset2 = mgr
            .publish(
                &ns(),
                &topic(),
                make_envelope(2, "ns", "test-topic", b"world"),
            )
            .await
            .unwrap();
        assert_eq!(offset2, TopicOffset(1));
    }

    #[tokio::test]
    async fn publish_to_unconfigured_namespace() {
        let engine = make_engine();
        let mgr = make_manager(engine);
        let result = mgr
            .publish(
                &NamespaceId("nonexistent".into()),
                &topic(),
                make_envelope(1, "nonexistent", "test-topic", b"hello"),
            )
            .await;
        assert!(matches!(result, Err(Error::NamespaceNotFound(_))));
    }

    struct PrefixInterceptor {
        prefix: Vec<u8>,
    }

    impl PublishInterceptor for PrefixInterceptor {
        fn intercept_publish(
            &self,
            _ns: &NamespaceId,
            _topic: &TopicName,
            mut msg: MessageEnvelope,
        ) -> Pin<Box<dyn std::future::Future<Output = Result<MessageEnvelope>> + Send + '_>>
        {
            Box::pin(async move {
                let mut new_payload = self.prefix.clone();
                new_payload.extend_from_slice(&msg.payload);
                msg.payload = new_payload;
                Ok(msg)
            })
        }
    }

    struct RejectInterceptor;

    impl PublishInterceptor for RejectInterceptor {
        fn intercept_publish(
            &self,
            _ns: &NamespaceId,
            _topic: &TopicName,
            _msg: MessageEnvelope,
        ) -> Pin<Box<dyn std::future::Future<Output = Result<MessageEnvelope>> + Send + '_>>
        {
            Box::pin(async { Err(Error::PermissionDenied) })
        }
    }

    #[tokio::test]
    async fn publish_interceptor_modifies_message() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        )
        .with_interceptors(
            vec![Box::new(PrefixInterceptor {
                prefix: b"PREFIX:".to_vec(),
            })],
            vec![],
        );

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"data"),
        )
        .await
        .unwrap();

        let msg = engine
            .consume_from(&ns(), &topic(), TopicOffset(0))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"PREFIX:data");
    }

    #[tokio::test]
    async fn publish_interceptor_rejects_message() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        )
        .with_interceptors(vec![Box::new(RejectInterceptor)], vec![]);

        let result = mgr
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "ns", "test-topic", b"data"),
            )
            .await;
        assert!(matches!(result, Err(Error::PermissionDenied)));

        let head = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(head, TopicOffset::ZERO);
    }

    // -- Subscribe/delivery tests --

    #[tokio::test]
    async fn individual_subscribe_receives_messages() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"hello"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"hello");
        assert_eq!(msg.offset, TopicOffset(0));
    }

    #[tokio::test]
    async fn multiple_individual_subscribers_fan_out() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let sub1 = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx1 = sub1.rx;
        let sub2 = mgr
            .subscribe(&ns(), &topic(), ClientId(2), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx2 = sub2.rx;
        let sub3 = mgr
            .subscribe(&ns(), &topic(), ClientId(3), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx3 = sub3.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"broadcast"),
        )
        .await
        .unwrap();

        for rx in [&mut rx1, &mut rx2, &mut rx3] {
            let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(msg.envelope.payload, b"broadcast");
        }
    }

    #[tokio::test]
    async fn consumer_group_delivers_to_one_member() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let group = GroupName("workers".into());
        let sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx1 = sub1.rx;
        let sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx2 = sub2.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"task"),
        )
        .await
        .unwrap();

        let r1 = tokio::time::timeout(Duration::from_millis(500), rx1.recv()).await;
        let r2 = tokio::time::timeout(Duration::from_millis(500), rx2.recv()).await;

        let received_count = [&r1, &r2]
            .iter()
            .filter(|r| r.as_ref().ok().and_then(|o| o.as_ref()).is_some())
            .count();
        assert_eq!(
            received_count, 1,
            "exactly one group member should receive the message"
        );
    }

    #[tokio::test]
    async fn consumer_group_round_robin() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let group = GroupName("workers".into());
        let sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx1 = sub1.rx;
        let sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx2 = sub2.rx;

        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let timeout = Duration::from_secs(3);
        let mut count1 = 0;
        let mut count2 = 0;

        for _ in 0..10 {
            tokio::select! {
                Some(_) = rx1.recv() => count1 += 1,
                Some(_) = rx2.recv() => count2 += 1,
                _ = tokio::time::sleep(timeout) => break,
            }
        }

        assert!(
            count1 + count2 == 10,
            "all 10 messages should be delivered, got {} + {}",
            count1,
            count2
        );
        assert!(count1 > 0, "member 1 should receive some messages");
        assert!(count2 > 0, "member 2 should receive some messages");
    }

    #[tokio::test]
    async fn mixed_individual_and_group_delivery() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let group = GroupName("workers".into());

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut ind_rx1 = sub.rx;
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(2), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut ind_rx2 = sub.rx;
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(3),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut grp_rx1 = sub.rx;
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(4),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut grp_rx2 = sub.rx;

        for i in 0..5u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let timeout = Duration::from_secs(3);

        for rx in [&mut ind_rx1, &mut ind_rx2] {
            for _ in 0..5 {
                let msg = tokio::time::timeout(timeout, rx.recv())
                    .await
                    .expect("individual subscriber should receive message")
                    .expect("channel should not close");
                assert!(msg.envelope.payload.starts_with(b"msg-"));
            }
        }

        let mut group_total = 0;
        for _ in 0..5 {
            tokio::select! {
                Some(_) = grp_rx1.recv() => group_total += 1,
                Some(_) = grp_rx2.recv() => group_total += 1,
                _ = tokio::time::sleep(timeout) => break,
            }
        }
        assert_eq!(
            group_total, 5,
            "group should collectively receive all 5 messages"
        );
    }

    // -- Ack tests (FIX 3: ack by MessageId) --

    #[tokio::test]
    async fn ack_by_message_id_advances_cursor_and_compacts() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        for i in 0..5u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        // Receive and ack first 3 by MessageId
        for _ in 0..3 {
            let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
                .await
                .unwrap()
                .unwrap();
            mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
                .await
                .unwrap();
        }

        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    #[tokio::test]
    async fn dlq_routing_on_max_redelivery() {
        let engine = make_engine();
        let _mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        );

        let envelope = make_envelope(1, "ns", "test-topic", b"dlq-me");
        let ctx = DlqContext {
            original_offset: TopicOffset(42),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "__dlq",
            envelope,
            &ctx,
        )
        .await
        .unwrap();

        let dlq_topic = TopicName("test-topic__dlq".into());
        let msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(msg.is_some(), "message should be routed to DLQ topic");
        let dlq_msg = msg.unwrap();
        // Payload is the original (not wrapped in an envelope)
        assert_eq!(&dlq_msg.envelope.payload, b"dlq-me");
        // Provenance metadata is in system headers
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.original_offset, 42);
        assert_eq!(meta.delivery_count, 6);
        assert_eq!(meta.first_delivery_ts, 1_700_000_000_000);
        assert_eq!(meta.reason, "max_attempts");
        assert!(meta.nack_reason.is_none());
    }

    #[tokio::test]
    async fn dlq_disabled_drops_message() {
        let engine = make_engine();

        let ctx = DlqContext {
            original_offset: TopicOffset(0),
            first_delivered_at: 0,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "",
            make_envelope(1, "ns", "test-topic", b"drop-me"),
            &ctx,
        )
        .await
        .unwrap();

        let head = engine
            .head_offset(&ns(), &TopicName("test-topic".into()))
            .await
            .unwrap();
        assert_eq!(head, TopicOffset::ZERO);
    }

    #[tokio::test]
    async fn dlq_write_failure_returns_error() {
        // Engine with 1-byte memory budget + Reject policy will fail all publishes
        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                global_max_memory_bytes: 1,
                backpressure: BackpressurePolicy::Reject,
                ..Default::default()
            },
        ));

        let ctx = DlqContext {
            original_offset: TopicOffset(0),
            first_delivered_at: 0,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "__dlq",
            make_envelope(1, "ns", "test-topic", b"fail-dlq"),
            &ctx,
        )
        .await;

        assert!(
            result.is_err(),
            "DLQ write should fail when engine rejects publish"
        );

        // Verify nothing was written to the DLQ topic
        let dlq_topic = TopicName("test-topic__dlq".into());
        let msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(
            msg.is_none(),
            "no message should be in DLQ topic after failure"
        );
    }

    #[tokio::test]
    async fn dlq_write_success_returns_offset() {
        let engine = make_engine();

        let ctx = DlqContext {
            original_offset: TopicOffset(7),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 3,
            dlq_failures: 0,
            reason: DlqReason::ClientNack {
                reason: "bad data".into(),
            },
        };
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "__dlq",
            make_envelope(1, "ns", "test-topic", b"dlq-ok"),
            &ctx,
        )
        .await;

        assert!(result.is_ok(), "DLQ write should succeed");
        let write_result = result.unwrap();
        assert!(
            write_result.dlq_timestamp > 0,
            "should have a real DLQ timestamp"
        );

        // Verify message was written with system headers
        let dlq_topic = TopicName("test-topic__dlq".into());
        let msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(msg.is_some());
        let dlq_msg = msg.unwrap();
        assert_eq!(&dlq_msg.envelope.payload, b"dlq-ok");
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.original_offset, 7);
        assert_eq!(meta.delivery_count, 3);
        assert_eq!(meta.reason, "client_nack");
        assert_eq!(meta.nack_reason.as_deref(), Some("bad data"));
    }

    #[tokio::test]
    async fn dlq_disabled_returns_sentinel_offset() {
        let engine = make_engine();

        let ctx = DlqContext {
            original_offset: TopicOffset(0),
            first_delivered_at: 0,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "",
            make_envelope(1, "ns", "test-topic", b"drop-me"),
            &ctx,
        )
        .await;

        assert!(result.is_ok());
        let write_result = result.unwrap();
        assert_eq!(
            write_result.dlq_timestamp, 0,
            "disabled DLQ has zero timestamp"
        );
    }

    #[test]
    fn check_dlq_needed_below_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 3,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );

        // max_redelivery=5 → delivery_count 3 is below threshold
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::check_dlq_needed(
            &inflight,
            TopicOffset(5),
            5,
        );
        assert!(result.is_none(), "should not route to DLQ when below max");
    }

    #[test]
    fn check_dlq_needed_at_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 5, // exactly at max_redelivery
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );

        // delivery_count == max_redelivery is NOT exceeded (uses >)
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::check_dlq_needed(
            &inflight,
            TopicOffset(5),
            5,
        );
        assert!(result.is_none(), "should not route to DLQ at exactly max");
    }

    #[test]
    fn check_dlq_needed_above_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 2,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );

        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::check_dlq_needed(
            &inflight,
            TopicOffset(5),
            5,
        );
        assert!(result.is_some(), "should route to DLQ when above max");
        let check = result.unwrap();
        assert_eq!(check.delivery_count, 6);
        assert_eq!(check.first_delivered_at, 1_700_000_000_000);
        assert_eq!(check.dlq_failures, 2);
    }

    #[test]
    fn handle_dlq_result_success_removes_from_inflight() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        let ctx = DlqContext {
            original_offset: TopicOffset(5),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };

        let removed =
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                Ok(DlqWriteResult {
                    dlq_timestamp: 1_700_000_060_000,
                }),
                &ctx,
                &ns(),
                &topic(),
                TopicOffset(5),
                &mut inflight,
                Some("dlq"),
            );
        assert!(removed, "should return true on success");
        assert!(
            inflight.is_empty(),
            "message should be removed from inflight"
        );
    }

    #[test]
    fn handle_dlq_result_failure_keeps_in_inflight_and_increments_counter() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 1,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        let ctx = DlqContext {
            original_offset: TopicOffset(5),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 1,
            reason: DlqReason::MaxAttemptsExceeded,
        };

        let removed =
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                Err(Error::ResourceExhausted("test".into())),
                &ctx,
                &ns(),
                &topic(),
                TopicOffset(5),
                &mut inflight,
                Some("dlq"),
            );
        assert!(!removed, "should return false on failure");
        assert_eq!(inflight.len(), 1, "message should stay in inflight");
        assert_eq!(
            inflight[&TopicOffset(5)].dlq_failures,
            2,
            "dlq_failures should be incremented"
        );
    }

    #[test]
    fn handle_dlq_result_failure_escalation_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                // Already at threshold - 1, next failure should escalate to ERROR
                dlq_failures: 2,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        let ctx = DlqContext {
            original_offset: TopicOffset(5),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 2,
            reason: DlqReason::MaxAttemptsExceeded,
        };

        let removed =
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                Err(Error::ResourceExhausted("dlq full".into())),
                &ctx,
                &ns(),
                &topic(),
                TopicOffset(5),
                &mut inflight,
                Some("dlq"),
            );
        assert!(!removed);
        assert_eq!(
            inflight[&TopicOffset(5)].dlq_failures,
            3,
            "dlq_failures should reach escalation threshold"
        );
    }

    // -- Disconnect tests --

    #[tokio::test]
    async fn disconnect_cleans_up_individual() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let _sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        mgr.disconnect(&ClientId(1)).await.unwrap();

        let subs = mgr.client_subscriptions.read().await;
        assert!(!subs.contains_key(&ClientId(1)));
    }

    #[tokio::test]
    async fn disconnect_group_member_keeps_group() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let group = GroupName("workers".into());
        let _sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let _sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        mgr.disconnect(&ClientId(1)).await.unwrap();

        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        let grp = state.groups.get(&group).unwrap();
        assert_eq!(grp.members.len(), 1);
        assert!(grp.members.contains_key(&ClientId(2)));
        assert!(grp.emptied_at.is_none());
    }

    #[tokio::test]
    async fn empty_group_retains_state_during_grace_period() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let group = GroupName("workers".into());
        let _sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        mgr.disconnect(&ClientId(1)).await.unwrap();

        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        let grp = state.groups.get(&group);
        assert!(
            grp.is_some(),
            "group should still exist during grace period"
        );
        assert!(grp.unwrap().emptied_at.is_some());
    }

    // -- FIX 1: Shutdown test --

    #[tokio::test]
    async fn shutdown_stops_delivery_tasks() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let _sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        mgr.shutdown().await.unwrap();

        // After shutdown, task handles should be cleared
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        assert!(state.individual_handles.is_empty());
    }

    // -- FIX 2: Group task respawn on rejoin test --

    #[tokio::test]
    async fn group_task_respawns_after_all_members_leave_and_rejoin() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("workers".into());

        // Subscribe, then disconnect (task exits when group becomes empty)
        let _sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        mgr.disconnect(&ClientId(1)).await.unwrap();

        // Give the delivery task time to notice and exit
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Rejoin — should spawn a new delivery task
        let sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx2 = sub2.rx;

        // Publish and verify delivery works
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"after-rejoin"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx2.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"after-rejoin");
    }

    // -- FIX 8: Ack uses reverse index (tested implicitly by ack_by_message_id test) --

    #[tokio::test]
    async fn ack_group_member_uses_reverse_index() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("workers".into());

        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(42, "ns", "test-topic", b"group-msg"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Ack by MessageId — should use reverse index to find the group directly
        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();
    }

    // -- Per-namespace quota tests --

    #[tokio::test]
    async fn namespace_message_count_quota_enforced() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_total_messages: Some(3),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        // Publish 3 messages — should succeed
        for i in 0..3u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // 4th message should fail
        let err = mgr
            .publish(
                &ns(),
                &topic(),
                make_envelope(3, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("namespace message quota exceeded"));
    }

    #[tokio::test]
    async fn namespace_topic_count_quota_enforced() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_topics: Some(2),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        // Publish to 2 different topics — should succeed
        mgr.publish(
            &ns(),
            &TopicName("topic-1".into()),
            make_envelope(0, "ns", "topic-1", b"data"),
        )
        .await
        .unwrap();
        mgr.publish(
            &ns(),
            &TopicName("topic-2".into()),
            make_envelope(1, "ns", "topic-2", b"data"),
        )
        .await
        .unwrap();

        // Publishing to existing topic should succeed (not a new topic)
        mgr.publish(
            &ns(),
            &TopicName("topic-1".into()),
            make_envelope(2, "ns", "topic-1", b"more-data"),
        )
        .await
        .unwrap();

        // Publishing to a 3rd topic should fail
        let err = mgr
            .publish(
                &ns(),
                &TopicName("topic-3".into()),
                make_envelope(3, "ns", "topic-3", b"data"),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("namespace topic quota exceeded"));
    }

    #[tokio::test]
    async fn namespace_no_quota_means_unlimited() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            test_shared_config(),
            ManagerConfig::default(),
        );

        // Should be able to publish many messages to many topics
        for i in 0..20u64 {
            mgr.publish(
                &ns(),
                &TopicName(format!("topic-{i}")),
                make_envelope(i, "ns", &format!("topic-{i}"), b"data"),
            )
            .await
            .unwrap();
        }
    }

    // -- Per-namespace connection limit tests --

    #[tokio::test]
    async fn namespace_connection_limit_enforced() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_connections: Some(2),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        // First two connections should succeed
        mgr.connect(&ns(), ClientId(1)).await.unwrap();
        mgr.connect(&ns(), ClientId(2)).await.unwrap();

        // Third connection should fail
        let err = mgr.connect(&ns(), ClientId(3)).await.unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(
            err.to_string()
                .contains("namespace connection limit reached")
        );
    }

    #[tokio::test]
    async fn namespace_connection_limit_decrements_on_disconnect() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_connections: Some(2),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        mgr.connect(&ns(), ClientId(1)).await.unwrap();
        mgr.connect(&ns(), ClientId(2)).await.unwrap();

        // At limit
        assert!(mgr.connect(&ns(), ClientId(3)).await.is_err());

        // Disconnect one client
        mgr.disconnect(&ClientId(1)).await.unwrap();

        // Now should be able to connect again
        mgr.connect(&ns(), ClientId(3)).await.unwrap();
    }

    #[tokio::test]
    async fn namespace_connection_no_limit_means_unlimited() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            test_shared_config(),
            ManagerConfig::default(),
        );

        // Should be able to connect many clients
        for i in 0..100 {
            mgr.connect(&ns(), ClientId(i)).await.unwrap();
        }
    }

    // -- Consumer group limit per topic tests --

    #[tokio::test]
    async fn consumer_group_limit_enforced() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            shared_config_with(|rt| rt.max_consumer_groups_per_topic = 3),
            ManagerConfig::default(),
        );

        // Create 3 groups — should all succeed
        for i in 0..3u64 {
            let _sub = mgr
                .subscribe(
                    &ns(),
                    &topic(),
                    ClientId(i),
                    Some(GroupName(format!("group-{i}"))),
                    SubscribeStart::Head,
                )
                .await
                .unwrap();
        }

        // 4th group should fail
        let err = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(100),
                Some(GroupName("group-extra".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("consumer group limit exceeded"));
    }

    #[tokio::test]
    async fn consumer_group_limit_allows_existing_group_members() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            shared_config_with(|rt| rt.max_consumer_groups_per_topic = 1),
            ManagerConfig::default(),
        );

        // First group member succeeds
        let _sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(GroupName("group-1".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        // Second member of same group should also succeed (not a new group)
        let _sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(GroupName("group-1".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        // But a new group should fail
        let err = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(3),
                Some(GroupName("group-2".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
    }

    // -- Slow consumer tracking tests --

    #[tokio::test]
    async fn slow_consumer_consecutive_failures_tracked() {
        let engine = make_engine();
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            shared_config_with(|rt| {
                rt.slow_consumer_policy = SlowConsumerPolicy::DropForSlow;
                rt.slow_consumer_threshold = 5;
                rt.slow_consumer_disconnect_after = 1000;
            }),
            ManagerConfig {
                delivery_channel_capacity: 1,
                ..ManagerConfig::default()
            },
        ));

        // Subscribe but never read from the channel (capacity=1)
        let _sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        // Publish enough messages to fill the channel and trigger failures
        for i in 0..20u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Give the delivery task time to process
        tokio::time::sleep(Duration::from_millis(500)).await;

        // Check subscriber state
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        if let Some(sub) = state.individual.get(&ClientId(1)) {
            // With capacity=1, after many publishes, there should be failures
            // The subscriber should eventually be marked as slow
            assert!(
                sub.consecutive_send_failures > 0 || sub.is_slow || sub.total_drops > 0,
                "subscriber should have experienced send failures, is_slow={}, failures={}, drops={}",
                sub.is_slow,
                sub.consecutive_send_failures,
                sub.total_drops,
            );
        }
    }

    #[tokio::test]
    async fn slow_consumer_resets_on_successful_send() {
        let engine = make_engine();
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            shared_config_with(|rt| {
                rt.slow_consumer_policy = SlowConsumerPolicy::DropForSlow;
                rt.slow_consumer_threshold = 1000;
                rt.slow_consumer_disconnect_after = 10000;
            }),
            ManagerConfig::default(),
        ));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Publish and consume (successful delivery)
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"hello"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"hello");

        // After successful delivery, consecutive failures should be 0
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        let sub = state.individual.get(&ClientId(1)).unwrap();
        assert_eq!(sub.consecutive_send_failures, 0);
        assert!(!sub.is_slow);
    }

    // -- Delivery channel capacity tests --

    #[tokio::test]
    async fn configurable_delivery_channel_capacity() {
        let engine = make_engine();
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            test_shared_config(),
            ManagerConfig {
                delivery_channel_capacity: 512,
                ..ManagerConfig::default()
            },
        ));
        assert_eq!(mgr.delivery_channel_capacity, 512);
    }

    // -- Group slow consumer tracking tests --

    #[tokio::test]
    async fn group_slow_consumer_tracked() {
        let engine = make_engine();
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![NamespaceId("ns".into())],
            shared_config_with(|rt| {
                rt.slow_consumer_policy = SlowConsumerPolicy::DropForSlow;
                rt.slow_consumer_threshold = 3;
                rt.slow_consumer_disconnect_after = 1000;
            }),
            ManagerConfig {
                delivery_channel_capacity: 1,
                ..ManagerConfig::default()
            },
        ));

        let group = GroupName("slow-group".into());

        // Subscribe but never read from the channel (capacity=1)
        let _sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        // Publish enough messages to fill the channel and trigger failures
        for i in 0..20u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Give the delivery task time to process
        tokio::time::sleep(Duration::from_millis(500)).await;

        // Check group state
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        if let Some(grp) = state.groups.get(&group) {
            assert!(
                grp.consecutive_send_failures > 0 || grp.is_slow || grp.total_drops > 0,
                "consumer group should have experienced send failures, \
                 is_slow={}, failures={}, drops={}",
                grp.is_slow,
                grp.consecutive_send_failures,
                grp.total_drops,
            );
        }
    }

    // -- min_live_offset tests --

    #[tokio::test]
    async fn min_live_offset_no_subscribers() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        // Publish some messages so the topic exists
        let env = make_envelope(0, "ns", "t", b"data");
        mgr.publish(&ns(), &TopicName("t".into()), env)
            .await
            .unwrap();

        let result = mgr.min_live_offset(&ns(), &TopicName("t".into())).await;
        assert_eq!(result, None, "no subscribers means no live offset");
    }

    #[tokio::test]
    async fn min_live_offset_single_subscriber() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let t = TopicName("t".into());
        let client = ClientId(100);
        mgr.connect(&ns(), client).await.unwrap();

        // Publish a few messages first so offset advances
        for i in 0..5u64 {
            let env = make_envelope(i, "ns", "t", b"data");
            mgr.publish(&ns(), &t, env).await.unwrap();
        }

        // Subscribe — new subscriber starts at head_offset (5)
        let _sub = mgr
            .subscribe(&ns(), &t, client, None, SubscribeStart::Head)
            .await
            .unwrap();

        let result = mgr.min_live_offset(&ns(), &t).await;
        assert_eq!(
            result,
            Some(TopicOffset(5)),
            "subscriber should be at head_offset"
        );
    }

    #[tokio::test]
    async fn min_live_offset_multiple_subscribers() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let t = TopicName("t".into());
        let c1 = ClientId(100);
        let c2 = ClientId(101);
        mgr.connect(&ns(), c1).await.unwrap();
        mgr.connect(&ns(), c2).await.unwrap();

        // Publish messages
        for i in 0..10u64 {
            let env = make_envelope(i, "ns", "t", b"data");
            mgr.publish(&ns(), &t, env).await.unwrap();
        }

        // Both subscribers join at head_offset (10)
        let _sub1 = mgr
            .subscribe(&ns(), &t, c1, None, SubscribeStart::Head)
            .await
            .unwrap();
        let _sub2 = mgr
            .subscribe(&ns(), &t, c2, None, SubscribeStart::Head)
            .await
            .unwrap();

        // min_live_offset should be at most 10 (the head when they subscribed).
        // Delivery tasks may advance some offsets, but the minimum across both
        // cannot exceed the head at subscribe time.
        let result = mgr.min_live_offset(&ns(), &t).await;
        assert!(
            result.is_some(),
            "should have a live offset with two subscribers"
        );
        assert!(
            result.unwrap() <= TopicOffset(10),
            "minimum should be at most 10, got {:?}",
            result
        );
    }

    #[tokio::test]
    async fn min_live_offset_with_consumer_group() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let t = TopicName("t".into());
        let c1 = ClientId(100);
        let c2 = ClientId(101);
        mgr.connect(&ns(), c1).await.unwrap();
        mgr.connect(&ns(), c2).await.unwrap();

        // Publish messages
        for i in 0..5u64 {
            let env = make_envelope(i, "ns", "t", b"data");
            mgr.publish(&ns(), &t, env).await.unwrap();
        }

        // Individual subscriber at offset 5
        let _sub1 = mgr
            .subscribe(&ns(), &t, c1, None, SubscribeStart::Head)
            .await
            .unwrap();

        // Consumer group member at offset 5
        let _sub2 = mgr
            .subscribe(
                &ns(),
                &t,
                c2,
                Some(GroupName("grp".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap();

        // Both start at head (5), so min is 5
        let result = mgr.min_live_offset(&ns(), &t).await;
        assert_eq!(result, Some(TopicOffset(5)));
    }

    #[tokio::test]
    async fn min_live_offset_after_unsubscribe() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let t = TopicName("t".into());
        let client = ClientId(100);
        mgr.connect(&ns(), client).await.unwrap();

        let _sub = mgr
            .subscribe(&ns(), &t, client, None, SubscribeStart::Head)
            .await
            .unwrap();

        // Unsubscribe
        mgr.unsubscribe(&ns(), &t, client, None).await.unwrap();

        let result = mgr.min_live_offset(&ns(), &t).await;
        assert_eq!(result, None, "no subscribers after unsubscribe");
    }

    #[tokio::test]
    async fn delivery_skips_evicted_offsets() {
        // Test the skip-forward fix: when a subscriber's offset falls behind
        // base_offset (due to DropOldest eviction), the delivery task must
        // skip forward instead of spinning on Ok(None).
        //
        // Strategy: publish messages, subscribe (starts at head), then
        // rewind the subscriber's offset behind base_offset and trigger
        // the delivery task to verify it skips forward.
        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        ));
        let mgr = DefaultNamespaceManager::new(
            engine.clone(),
            vec![NamespaceId("ns".into())],
            test_shared_config(),
            ManagerConfig {
                delivery_channel_capacity: 64,
                ..ManagerConfig::default()
            },
        );

        let t = TopicName("t".into());

        // Publish 100 messages. DropOldest with max_depth=10 evicts 0..89.
        for i in 0..100u64 {
            mgr.publish(
                &ns(),
                &t,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let base = engine.base_offset(&ns(), &t).await.unwrap();
        assert_eq!(base, TopicOffset(90));

        // Subscribe — starts at head_offset (100), delivery task is caught up
        let client = ClientId(42);
        mgr.connect(&ns(), client).await.unwrap();
        let sub = mgr
            .subscribe(&ns(), &t, client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Rewind subscriber's next_offset to 5 (behind base_offset 90).
        // This simulates what happens when a subscriber was at offset 5
        // and DropOldest evicted past it.
        {
            let ns_map = mgr.namespaces.read().await;
            let ns_state = ns_map.get(&ns()).unwrap();
            let topics = ns_state.topics.read().await;
            let topic_state = topics.get(&t).unwrap();
            let mut state = topic_state.lock().await;
            if let Some(sub) = state.individual.get_mut(&client) {
                sub.next_offset = TopicOffset(5);
            }
            // Notify the delivery task to wake it
            state.notify.notify_waiters();
        }

        // With disk fallback, the delivery task reads evicted messages from
        // storage instead of skipping. The subscriber should receive offset 5
        // (the first message after its rewound position).
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timed out — subscriber stuck")
            .expect("channel closed");

        assert_eq!(
            msg.offset,
            TopicOffset(5),
            "first delivered offset should be 5 (read from disk, not skipped)"
        );

        mgr.shutdown().await.unwrap();
    }

    // -- Cold batch delivery tests --

    #[tokio::test]
    async fn cold_batch_individual_catches_up() {
        // A slow individual subscriber whose offset falls behind the in-memory
        // buffer must receive all messages via cold batch reads from disk,
        // starting from its rewound offset, not skipping forward.
        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        ));
        let mgr = DefaultNamespaceManager::new(
            engine.clone(),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        );

        let t = TopicName("t".into());

        // Publish 100 messages BEFORE subscribing. DropOldest with max_depth=10
        // keeps 90..99 in memory. MockStorage retains all 100 on disk.
        for i in 0..100u64 {
            mgr.publish(
                &ns(),
                &t,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let base = engine.base_offset(&ns(), &t).await.unwrap();
        assert_eq!(base, TopicOffset(90));

        // Subscribe after publishing — starts at head_offset (100), no messages pending.
        let client = ClientId(42);
        mgr.connect(&ns(), client).await.unwrap();
        let sub = mgr
            .subscribe(&ns(), &t, client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Rewind subscriber's next_offset to 0 (way behind base_offset 90).
        {
            let ns_map = mgr.namespaces.read().await;
            let ns_state = ns_map.get(&ns()).unwrap();
            let topics = ns_state.topics.read().await;
            let topic_state = topics.get(&t).unwrap();
            let mut state = topic_state.lock().await;
            if let Some(sub) = state.individual.get_mut(&client) {
                sub.next_offset = TopicOffset(0);
            }
            state.notify.notify_waiters();
        }

        // Collect messages. The subscriber should receive messages starting
        // from offset 0 (read from disk via cold batch), continuing through
        // offset 99.
        let mut received = Vec::new();
        for _ in 0..100 {
            match tokio::time::timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(msg)) => received.push(msg.offset),
                _ => break,
            }
        }

        assert!(
            received.len() >= 20,
            "cold batch should deliver many messages, got {}",
            received.len()
        );
        assert_eq!(
            received[0],
            TopicOffset(0),
            "first message should be from offset 0 (cold batch from disk)"
        );

        // Verify messages are in order
        for window in received.windows(2) {
            assert!(
                window[1] > window[0],
                "offsets should be strictly increasing: {:?} -> {:?}",
                window[0],
                window[1]
            );
        }

        mgr.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn cold_batch_group_catches_up() {
        // Same as individual test but using a consumer group. When a group's
        // offset falls behind the in-memory buffer, cold batch reads should
        // deliver messages from disk starting at the rewound offset.
        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        ));
        let mgr = DefaultNamespaceManager::new(
            engine.clone(),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        );

        let t = TopicName("t".into());
        let group = GroupName("workers".into());

        // Publish 100 messages BEFORE subscribing. DropOldest with max_depth=10
        // keeps 90..99 in memory. MockStorage retains all 100 on disk.
        for i in 0..100u64 {
            mgr.publish(
                &ns(),
                &t,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let base = engine.base_offset(&ns(), &t).await.unwrap();
        assert_eq!(base, TopicOffset(90));

        // Subscribe after publishing — group starts at head_offset (100).
        let client = ClientId(42);
        mgr.connect(&ns(), client).await.unwrap();
        let sub = mgr
            .subscribe(&ns(), &t, client, Some(group.clone()), SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Rewind group's next_offset to 0 (behind base_offset 90).
        {
            let ns_map = mgr.namespaces.read().await;
            let ns_state = ns_map.get(&ns()).unwrap();
            let topics = ns_state.topics.read().await;
            let topic_state = topics.get(&t).unwrap();
            let mut state = topic_state.lock().await;
            if let Some(grp) = state.groups.get_mut(&group) {
                grp.next_offset = TopicOffset(0);
            }
            state.notify.notify_waiters();
        }

        // Collect messages from the group channel.
        let mut received = Vec::new();
        for _ in 0..100 {
            match tokio::time::timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(msg)) => received.push(msg.offset),
                _ => break,
            }
        }

        assert!(
            received.len() >= 20,
            "cold batch should deliver many messages to group, got {}",
            received.len()
        );
        assert_eq!(
            received[0],
            TopicOffset(0),
            "first group message should be from offset 0 (cold batch from disk)"
        );

        // Verify messages are in order
        for window in received.windows(2) {
            assert!(
                window[1] > window[0],
                "group offsets should be strictly increasing: {:?} -> {:?}",
                window[0],
                window[1]
            );
        }

        mgr.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn cold_batch_transitions_to_hot_path() {
        // When a subscriber is behind the in-memory buffer, cold batch reads
        // fetch from disk. Once the subscriber catches up to base_offset, it
        // should seamlessly transition to hot path delivery from the in-memory
        // buffer, including new messages published after the transition.
        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 50,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        ));
        let mgr = DefaultNamespaceManager::new(
            engine.clone(),
            vec![ns()],
            test_shared_config(),
            ManagerConfig::default(),
        );

        let t = TopicName("t".into());

        // Publish 100 messages BEFORE subscribing. DropOldest with max_depth=50
        // keeps 50..99 in memory. MockStorage retains all 100 on disk.
        for i in 0..100u64 {
            mgr.publish(
                &ns(),
                &t,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        let base = engine.base_offset(&ns(), &t).await.unwrap();
        assert_eq!(base, TopicOffset(50));

        // Subscribe after publishing — starts at head_offset (100).
        let client = ClientId(42);
        mgr.connect(&ns(), client).await.unwrap();
        let sub = mgr
            .subscribe(&ns(), &t, client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Rewind subscriber to offset 40 (behind base but close — 10 messages
        // on disk, then 50 in memory).
        {
            let ns_map = mgr.namespaces.read().await;
            let ns_state = ns_map.get(&ns()).unwrap();
            let topics = ns_state.topics.read().await;
            let topic_state = topics.get(&t).unwrap();
            let mut state = topic_state.lock().await;
            if let Some(sub) = state.individual.get_mut(&client) {
                sub.next_offset = TopicOffset(40);
            }
            state.notify.notify_waiters();
        }

        // Receive messages covering both cold batch (40..49 from disk)
        // and hot path (50..99 from memory).
        let mut received = Vec::new();
        for _ in 0..100 {
            match tokio::time::timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(msg)) => received.push(msg.offset),
                _ => break,
            }
        }

        // Should have at least 55 messages: 10 from disk (40..49) + 50 from memory (50..99)
        assert!(
            received.len() >= 55,
            "should receive cold + hot messages, got {}",
            received.len()
        );
        assert_eq!(
            received[0],
            TopicOffset(40),
            "first message should be from offset 40 (cold batch from disk)"
        );

        // Verify the transition zone: offsets should be contiguous across cold->hot boundary
        for window in received.windows(2) {
            assert!(
                window[1] > window[0],
                "offsets should be strictly increasing across cold->hot: {:?} -> {:?}",
                window[0],
                window[1]
            );
        }

        // Verify we crossed the cold->hot boundary: should have offsets both
        // below and at/above base_offset 50
        let has_cold = received.iter().any(|o| *o < TopicOffset(50));
        let has_hot = received.iter().any(|o| *o >= TopicOffset(50));
        assert!(has_cold, "should include cold batch offsets (< 50)");
        assert!(has_hot, "should include hot path offsets (>= 50)");

        // Publish 5 more messages (100..104) and verify hot path continues.
        for i in 100..105u64 {
            mgr.publish(
                &ns(),
                &t,
                make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
            )
            .await
            .unwrap();
        }

        // Receive the new messages
        let mut new_received = Vec::new();
        for _ in 0..5 {
            match tokio::time::timeout(Duration::from_secs(3), rx.recv()).await {
                Ok(Some(msg)) => new_received.push(msg.offset),
                _ => break,
            }
        }

        assert!(
            !new_received.is_empty(),
            "hot path should deliver new messages after cold batch catchup"
        );
        // The new offsets should be >= 100 (proving the hot path resumed)
        assert!(
            new_received.iter().all(|o| o.0 >= 100),
            "new messages should have offset >= 100, got {:?}",
            new_received
        );

        mgr.shutdown().await.unwrap();
    }

    // -- Per-namespace disk quota tests --

    #[tokio::test]
    async fn publish_rejected_when_disk_quota_exceeded() {
        // Compute the actual encoded record size so the quota threshold is
        // accurate regardless of bincode overhead changes.
        let sample = PublishedMessage {
            envelope: Arc::new(make_envelope(0, "ns", "test-topic", b"data")),
            offset: TopicOffset(0),
        };
        let record_size = crate::storage::fs::encode_record(&sample).unwrap().len() as u64;

        // Allow exactly 2 records — the 3rd should be rejected
        let quota = record_size * 2;

        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_disk_bytes: Some(quota),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        for i in 0..2u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // 3rd message should fail — 2 records already at the limit
        let err = mgr
            .publish(
                &ns(),
                &topic(),
                make_envelope(2, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("namespace disk quota exceeded"));
    }

    #[tokio::test]
    async fn publish_succeeds_when_under_disk_quota() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_disk_bytes: Some(1_000_000),
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        // Publish small messages — all should succeed
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"small"),
            )
            .await
            .unwrap();
        }
    }

    #[tokio::test]
    async fn disk_quota_none_means_unlimited() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            engine,
            vec![NamespaceId("ns".into())],
            shared_config_with_quota(NamespaceQuota {
                max_disk_bytes: None,
                ..Default::default()
            }),
            ManagerConfig::default(),
        );

        // Publish many messages — none should be rejected
        for i in 0..100u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", &[0xAB; 100]),
            )
            .await
            .unwrap();
        }
    }

    // -- Seek/rewind tests --

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

    #[tokio::test]
    async fn subscribe_head_starts_at_head() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish 5 messages
        for i in 0..5u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Subscribe at head — should start at offset 5 (next to be published)
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        assert_eq!(sub.start_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn subscribe_at_offset_starts_from_offset() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish 10 messages
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Subscribe at offset 5
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                None,
                SubscribeStart::Offset(TopicOffset(5)),
            )
            .await
            .unwrap();
        let mut rx = sub.rx;
        assert_eq!(sub.start_offset, TopicOffset(5));

        // First delivered message should be at offset 5
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn subscribe_at_offset_clamps_to_head() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish 5 messages — head is at offset 5
        for i in 0..5u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Subscribe at offset 100 (past head) — should clamp to 5
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                None,
                SubscribeStart::Offset(TopicOffset(100)),
            )
            .await
            .unwrap();
        assert_eq!(sub.start_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn offset_at_timestamp_resolves_via_manager() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish messages with known timestamps
        let base_ts = 1_700_000_000_000u64;
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope_with_ts(i, "ns", "test-topic", b"data", base_ts + i * 1000),
            )
            .await
            .unwrap();
        }

        // Resolve timestamp to offset via manager (delegates to engine/storage)
        let offset = mgr
            .offset_at_timestamp(&ns(), &topic(), base_ts + 5000)
            .await
            .unwrap();
        assert_eq!(offset, Some(TopicOffset(5)));

        // Use the resolved offset to subscribe
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                None,
                SubscribeStart::Offset(offset.unwrap()),
            )
            .await
            .unwrap();
        assert_eq!(sub.start_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn offset_at_timestamp_no_match_returns_none() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish messages with known timestamps
        let base_ts = 1_700_000_000_000u64;
        for i in 0..5u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope_with_ts(i, "ns", "test-topic", b"data", base_ts + i * 1000),
            )
            .await
            .unwrap();
        }

        // Timestamp far in the future — no match
        let offset = mgr
            .offset_at_timestamp(&ns(), &topic(), base_ts + 999_999_999)
            .await
            .unwrap();
        assert_eq!(offset, None);

        // When None, the session would use SubscribeStart::Head
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        assert_eq!(sub.start_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn subscribe_at_with_consumer_group() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("workers".into());

        // Publish 10 messages
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Subscribe with group at offset 5
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Offset(TopicOffset(5)),
            )
            .await
            .unwrap();
        let mut rx = sub.rx;
        assert_eq!(sub.start_offset, TopicOffset(5));

        // First delivered message should be at offset 5
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn subscribe_group_join_returns_group_offset() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("workers".into());

        // Publish 10 messages
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // First client creates group at offset 5
        let sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Offset(TopicOffset(5)),
            )
            .await
            .unwrap();
        assert_eq!(sub1.start_offset, TopicOffset(5));

        // Second client joins the existing group requesting offset 0
        // Should get the group's current offset (5), not 0
        let sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Offset(TopicOffset(0)),
            )
            .await
            .unwrap();
        assert_eq!(sub2.start_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn subscribe_group_join_at_head_returns_group_offset() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("workers".into());

        // Publish 10 messages
        for i in 0..10u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // First client creates group at offset 3
        let sub1 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Offset(TopicOffset(3)),
            )
            .await
            .unwrap();
        assert_eq!(sub1.start_offset, TopicOffset(3));

        // Second client joins with Head (which would be 10)
        // Should still get the group's current offset (3)
        let sub2 = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(2),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        assert_eq!(sub2.start_offset, TopicOffset(3));
    }

    // -- Fetch tests --

    #[tokio::test]
    async fn fetch_returns_messages_in_range() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        // Publish 10 messages
        for i in 0..10 {
            let env = make_envelope(i, "ns", "test-topic", &[i as u8; 100]);
            mgr.publish(&ns, &topic, env).await.unwrap();
        }

        // Fetch offset 3, max_count 5
        let result = mgr
            .fetch(&ns, &topic, TopicOffset(3), 5, 1_000_000, false)
            .await
            .unwrap();
        assert_eq!(result.messages.len(), 5);
        assert_eq!(result.messages[0].offset, TopicOffset(3));
        assert_eq!(result.messages[4].offset, TopicOffset(7));
        assert_eq!(result.next_offset, TopicOffset(8));
        assert_eq!(result.head_offset, TopicOffset(10));
        assert!(result.clamped_from.is_none());
    }

    #[tokio::test]
    async fn fetch_empty_topic() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let result = mgr
            .fetch(&ns, &topic, TopicOffset(0), 10, 1_000_000, false)
            .await
            .unwrap();
        assert!(result.messages.is_empty());
        assert_eq!(result.next_offset, TopicOffset(0));
        assert_eq!(result.head_offset, TopicOffset(0));
    }

    #[tokio::test]
    async fn fetch_at_head() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        for i in 0..5 {
            let env = make_envelope(i, "ns", "test-topic", b"data");
            mgr.publish(&ns, &topic, env).await.unwrap();
        }

        // Fetch at head offset (5) — should return empty
        let result = mgr
            .fetch(&ns, &topic, TopicOffset(5), 10, 1_000_000, false)
            .await
            .unwrap();
        assert!(result.messages.is_empty());
        assert_eq!(result.next_offset, TopicOffset(5));
        assert_eq!(result.head_offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn fetch_past_head() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        for i in 0..3 {
            let env = make_envelope(i, "ns", "test-topic", b"data");
            mgr.publish(&ns, &topic, env).await.unwrap();
        }

        let result = mgr
            .fetch(&ns, &topic, TopicOffset(100), 10, 1_000_000, false)
            .await
            .unwrap();
        assert!(result.messages.is_empty());
        assert_eq!(result.next_offset, TopicOffset(3));
        assert_eq!(result.head_offset, TopicOffset(3));
    }

    #[tokio::test]
    async fn fetch_respects_max_response_bytes() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        // Publish 10 messages with 500-byte payloads
        for i in 0..10 {
            let env = make_envelope(i, "ns", "test-topic", &[0u8; 500]);
            mgr.publish(&ns, &topic, env).await.unwrap();
        }

        // Budget for ~2 messages: 2 * (20 overhead + 500 payload) = 1040 bytes
        let result = mgr
            .fetch(&ns, &topic, TopicOffset(0), 100, 1100, false)
            .await
            .unwrap();
        // Should return 2 messages (first message always included, second fits)
        assert!(result.messages.len() <= 3);
        assert!(result.messages.len() >= 2);
        assert_eq!(result.head_offset, TopicOffset(10));
    }

    #[tokio::test]
    async fn fetch_head_offset_in_result() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        for i in 0..10 {
            let env = make_envelope(i, "ns", "test-topic", b"msg");
            mgr.publish(&ns, &topic, env).await.unwrap();
        }

        // Fetch first 5
        let result = mgr
            .fetch(&ns, &topic, TopicOffset(0), 5, 1_000_000, false)
            .await
            .unwrap();
        assert_eq!(result.messages.len(), 5);
        assert_eq!(result.next_offset, TopicOffset(5));
        assert_eq!(result.head_offset, TopicOffset(10));
    }

    #[tokio::test]
    async fn fetch_message_too_large() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        // Publish a large message (1000 bytes payload)
        let env = make_envelope(0, "ns", "test-topic", &[0u8; 1000]);
        mgr.publish(&ns, &topic, env).await.unwrap();

        // Budget smaller than message + overhead
        let err = mgr
            .fetch(&ns, &topic, TopicOffset(0), 10, 500, false)
            .await
            .unwrap_err();
        assert!(err.is_message_too_large());
    }

    #[tokio::test]
    async fn fetch_message_too_large_only_on_first() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        // Publish a small message then a large one
        let env1 = make_envelope(0, "ns", "test-topic", &[0u8; 50]);
        mgr.publish(&ns, &topic, env1).await.unwrap();
        let env2 = make_envelope(1, "ns", "test-topic", &[0u8; 1000]);
        mgr.publish(&ns, &topic, env2).await.unwrap();

        // Budget fits the small one but not both
        let result = mgr
            .fetch(&ns, &topic, TopicOffset(0), 10, 200, false)
            .await
            .unwrap();
        // Should return just the small message (truncation, not error)
        assert_eq!(result.messages.len(), 1);
        assert_eq!(result.messages[0].offset, TopicOffset(0));
        assert_eq!(result.next_offset, TopicOffset(1));
    }

    #[tokio::test]
    async fn fetch_does_not_create_subscription_state() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        // Fetch
        mgr.fetch(&ns, &topic, TopicOffset(0), 10, 1_000_000, false)
            .await
            .unwrap();

        // Verify no subscription state was created
        let ns_state = mgr.get_namespace(&ns).await.unwrap();
        let topics = ns_state.topics.read().await;
        // Topic delivery state may exist from publish, but should have no subscribers
        if let Some(topic_state) = topics.get(&topic) {
            let state = topic_state.lock().await;
            assert!(state.individual.is_empty());
            assert!(state.groups.is_empty());
        }
    }

    #[tokio::test]
    async fn fetch_nonexistent_namespace() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let bad_ns = NamespaceId("nonexistent".into());
        let topic = topic();

        let err = mgr
            .fetch(&bad_ns, &topic, TopicOffset(0), 10, 1_000_000, false)
            .await
            .unwrap_err();
        assert!(matches!(err, Error::NamespaceNotFound(_)));
    }

    // -- Individual subscriber DLQ routing tests --

    #[test]
    fn check_dlq_needed_individual_inflight_below_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(3),
            InflightEntry {
                offset: TopicOffset(3),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 3,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        // max_redelivery=5, delivery_count=3 → below threshold
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::check_dlq_needed(
            &inflight,
            TopicOffset(3),
            5,
        );
        assert!(
            result.is_none(),
            "delivery_count below max should not trigger DLQ"
        );
    }

    #[test]
    fn check_dlq_needed_individual_inflight_above_threshold() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(3),
            InflightEntry {
                offset: TopicOffset(3),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        // max_redelivery=5, delivery_count=6 → exceeded
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::check_dlq_needed(
            &inflight,
            TopicOffset(3),
            5,
        );
        assert!(
            result.is_some(),
            "delivery_count exceeding max should trigger DLQ"
        );
        let check = result.unwrap();
        assert_eq!(check.delivery_count, 6);
        assert_eq!(check.first_delivered_at, 1_700_000_000_000);
    }

    #[test]
    fn handle_dlq_result_individual_success_removes_from_inflight() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(7),
            InflightEntry {
                offset: TopicOffset(7),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        let ctx = DlqContext {
            original_offset: TopicOffset(7),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };

        let removed =
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                Ok(DlqWriteResult {
                    dlq_timestamp: 1_700_000_060_000,
                }),
                &ctx,
                &ns(),
                &topic(),
                TopicOffset(7),
                &mut inflight,
                Some("dlq"),
            );
        assert!(removed, "successful DLQ write should remove from inflight");
        assert!(inflight.is_empty());
    }

    #[test]
    fn handle_dlq_result_individual_failure_keeps_in_inflight() {
        let mut inflight = HashMap::new();
        inflight.insert(
            TopicOffset(7),
            InflightEntry {
                offset: TopicOffset(7),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 6,
                first_delivered_at: 1_700_000_000_000,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );
        let ctx = DlqContext {
            original_offset: TopicOffset(7),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };

        let removed =
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                Err(Error::ResourceExhausted("dlq full".into())),
                &ctx,
                &ns(),
                &topic(),
                TopicOffset(7),
                &mut inflight,
                Some("dlq"),
            );
        assert!(!removed, "failed DLQ write should keep message in inflight");
        assert!(inflight.contains_key(&TopicOffset(7)));
        assert_eq!(inflight[&TopicOffset(7)].dlq_failures, 1);
    }

    #[tokio::test]
    async fn individual_subscriber_dlq_routing_on_max_redelivery() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Subscribe first so we start at head (offset 0)
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Publish a message after subscribing
        let envelope = make_envelope(1, "ns", "test-topic", b"individual-dlq-me");
        mgr.publish(&ns(), &topic(), envelope.clone())
            .await
            .unwrap();

        // Wait for the message to be delivered
        let delivered = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timeout waiting for delivery")
            .expect("channel closed");
        assert_eq!(delivered.envelope.payload, b"individual-dlq-me");

        // Manually set the inflight entry to have exceeded delivery count.
        // This simulates the scenario where a message has been redelivered
        // many times (e.g., via SubscribeAt to a previous offset).
        {
            let ns_state = mgr.get_namespace(&ns()).await.unwrap();
            let topics = ns_state.topics.read().await;
            let topic_state = topics.get(&topic()).unwrap();
            let mut state = topic_state.lock().await;
            let sub = state.individual.get_mut(&ClientId(1)).unwrap();
            if let Some(entry) = sub.inflight.get_mut(&TopicOffset(0)) {
                entry.delivery_count = 6; // exceeds default max_redelivery_attempts=5
            }
        }

        // Now route to DLQ using the shared helper directly
        let ctx = DlqContext {
            original_offset: TopicOffset(0),
            first_delivered_at: now_millis(),
            delivery_count: 6,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "__dlq",
            envelope,
            &ctx,
        )
        .await;

        assert!(result.is_ok(), "DLQ routing should succeed");

        // Verify message appears in DLQ topic with original payload + system headers
        let dlq_topic = TopicName("test-topic__dlq".into());
        let dlq_msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap()
            .expect("message should be in DLQ topic");

        assert_eq!(&dlq_msg.envelope.payload, b"individual-dlq-me");
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.original_offset, 0);
        assert_eq!(meta.delivery_count, 6);
        assert_eq!(meta.reason, "max_attempts");
    }

    #[tokio::test]
    async fn dlq_preserves_original_application_headers() {
        // Publish a message with application headers, route to DLQ, and verify
        // both the original app headers AND _fafb.dlq.* system headers survive.
        let engine = make_engine();
        let app_headers = vec![
            crate::types::Header {
                key: b"content-type".to_vec(),
                value: b"application/json".to_vec(),
            },
            crate::types::Header {
                key: b"correlation-id".to_vec(),
                value: b"req-abc-123".to_vec(),
            },
        ];
        let envelope = MessageEnvelope::new(
            MessageId(1),
            ns(),
            topic(),
            b"payload-with-headers".to_vec(),
            now_millis(),
        )
        .with_headers(app_headers.clone());

        let ctx = DlqContext {
            original_offset: TopicOffset(10),
            first_delivered_at: 1_700_000_000_000,
            delivery_count: 5,
            dlq_failures: 0,
            reason: DlqReason::MaxAttemptsExceeded,
        };
        let result = DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::route_to_dlq(
            &engine,
            &ns(),
            &topic(),
            "__dlq",
            envelope,
            &ctx,
        )
        .await;
        assert!(result.is_ok());

        let dlq_topic = TopicName("test-topic__dlq".into());
        let dlq_msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap()
            .expect("message should be in DLQ topic");

        // Original payload is preserved (not wrapped)
        assert_eq!(&dlq_msg.envelope.payload, b"payload-with-headers");

        // Original app headers come first
        assert!(dlq_msg.envelope.headers.len() >= 2 + 6); // 2 app + 6 system
        assert_eq!(dlq_msg.envelope.headers[0].key, b"content-type");
        assert_eq!(dlq_msg.envelope.headers[0].value, b"application/json");
        assert_eq!(dlq_msg.envelope.headers[1].key, b"correlation-id");
        assert_eq!(dlq_msg.envelope.headers[1].value, b"req-abc-123");

        // System headers are parseable
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.original_offset, 10);
        assert_eq!(meta.delivery_count, 5);
        assert_eq!(meta.reason, "max_attempts");
    }

    #[test]
    fn individual_inflight_delivery_count_increments_on_reinsert() {
        // Verify that individual subscriber inflight entries correctly
        // increment delivery_count when re-inserted at the same offset.
        let mut inflight = HashMap::new();

        // First delivery
        let now = now_millis();
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count: 1,
                first_delivered_at: now,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );

        // Simulate redelivery: check existing entry and increment
        let (delivery_count, first_delivered_at) = inflight
            .get(&TopicOffset(5))
            .map(|e| (e.delivery_count + 1, e.first_delivered_at))
            .unwrap_or((1, now_millis()));
        inflight.insert(
            TopicOffset(5),
            InflightEntry {
                offset: TopicOffset(5),
                message_id: MessageId(1),
                delivered_to: ClientId(1),
                delivered_at: Instant::now(),
                delivery_count,
                first_delivered_at,
                dlq_failures: 0,
                next_redeliver_at: Instant::now() + Duration::from_secs(300),
                msg_timestamp: 0,
                msg_ttl_ms: 0,
            },
        );

        assert_eq!(inflight[&TopicOffset(5)].delivery_count, 2);
        assert_eq!(
            inflight[&TopicOffset(5)].first_delivered_at,
            now,
            "first_delivered_at should be preserved across redeliveries"
        );
    }

    #[tokio::test]
    async fn individual_inflight_sweep_escalates_and_routes_to_dlq() {
        // End-to-end test: deliver a message to an individual subscriber,
        // don't ack it, and verify the sweep escalates delivery_count through
        // timeout cycles until it exceeds max_redelivery, then routes to DLQ.
        let engine = make_engine();
        let shared_config = shared_config_with(|rt| {
            rt.max_redelivery_attempts = 2; // low threshold for quick test
            rt.inflight_timeout_secs = 1; // short timeout so sweep triggers quickly
        });
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            shared_config,
            ManagerConfig::default(),
        ));

        // Subscribe first, then publish
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        let envelope = make_envelope(1, "ns", "test-topic", b"sweep-dlq-me");
        mgr.publish(&ns(), &topic(), envelope).await.unwrap();

        // Receive but don't ack — message enters inflight with delivery_count=1
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timeout waiting for delivery")
            .expect("channel closed");
        assert_eq!(msg.envelope.payload, b"sweep-dlq-me");

        // Wait for sweep cycles to escalate delivery_count past max_redelivery=2:
        //   Sweep 1 (after ~1s): entry timed out (1s timeout), count 1 → 2, reset timer
        //   Sweep 2 (after ~3s): entry timed out again (2s backoff), count 2 → 3 > max(2), route to DLQ
        tokio::time::sleep(Duration::from_millis(5000)).await;

        // Verify message appeared in DLQ topic with original payload + system headers
        let dlq_topic = TopicName("test-topic__dlq".into());
        let dlq_msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(
            dlq_msg.is_some(),
            "message should be routed to DLQ after sweep"
        );
        let dlq_msg = dlq_msg.unwrap();
        assert_eq!(&dlq_msg.envelope.payload, b"sweep-dlq-me");
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.original_offset, 0);
        assert_eq!(meta.delivery_count, 3); // 1 initial + 2 escalations
        assert_eq!(meta.reason, "max_attempts");

        // Verify the inflight entry was removed
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        let sub = state.individual.get(&ClientId(1)).unwrap();
        assert!(
            sub.inflight.is_empty(),
            "inflight should be empty after DLQ routing"
        );
    }

    #[tokio::test]
    async fn individual_inflight_sweep_escalates_count_below_threshold() {
        // Verify that the sweep increments delivery_count for entries below
        // the max_redelivery threshold without routing to DLQ.
        let engine = make_engine();
        let shared_config = shared_config_with(|rt| {
            rt.max_redelivery_attempts = 10; // high threshold — won't trigger DLQ
            rt.inflight_timeout_secs = 1; // short timeout so sweep triggers quickly
        });
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            shared_config,
            ManagerConfig::default(),
        ));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"escalate-me"),
        )
        .await
        .unwrap();

        // Receive but don't ack
        let _msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("timeout")
            .expect("closed");

        // Wait for the base timeout + a couple of sweep cycles
        tokio::time::sleep(Duration::from_millis(3000)).await;

        // Verify delivery_count was escalated but message is still in inflight (not DLQ'd)
        let ns_state = mgr.get_namespace(&ns()).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic()).unwrap();
        let state = topic_state.lock().await;
        let sub = state.individual.get(&ClientId(1)).unwrap();
        let entry = sub.inflight.get(&TopicOffset(0));
        assert!(entry.is_some(), "entry should still be in inflight");
        let entry = entry.unwrap();
        assert!(
            entry.delivery_count > 1,
            "delivery_count should have been escalated by sweep, got {}",
            entry.delivery_count
        );

        // Verify nothing in DLQ
        let dlq_topic = TopicName("test-topic__dlq".into());
        let dlq_msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(
            dlq_msg.is_none(),
            "no DLQ message should exist below threshold"
        );
    }

    #[tokio::test]
    async fn nack_routes_to_dlq_with_client_nack_reason() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let client = ClientId(1);
        mgr.connect(&ns(), client).await.unwrap();

        // Subscribe first (at head), then publish so message gets delivered
        let sub = mgr
            .subscribe(&ns(), &topic(), client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        let envelope = make_envelope(100, "ns", "test-topic", b"nack-me");
        mgr.publish(&ns(), &topic(), envelope).await.unwrap();

        // Wait for delivery
        let delivered = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();
        let message_id = delivered.envelope.id;

        // Nack the message with a reason
        let result = mgr
            .nack(
                &ns(),
                &topic(),
                client,
                message_id,
                "invalid payload format".into(),
            )
            .await
            .unwrap();
        assert_eq!(result, NackResult::DeadLettered);

        // Verify DLQ topic has the message with ClientNack reason in headers
        let dlq_topic = TopicName("test-topic__dlq".into());
        let dlq_msg = engine
            .consume_from(&ns(), &dlq_topic, TopicOffset(0))
            .await
            .unwrap();
        assert!(dlq_msg.is_some(), "message should appear in DLQ topic");

        let dlq_msg = dlq_msg.unwrap();
        assert_eq!(&dlq_msg.envelope.payload, b"nack-me");
        let meta = crate::dlq::parse_dlq_headers(&dlq_msg.envelope.headers)
            .expect("DLQ headers should be present");
        assert_eq!(meta.source_topic, topic().0);
        assert_eq!(meta.reason, "client_nack");
        assert_eq!(meta.nack_reason.as_deref(), Some("invalid payload format"));
        assert_eq!(meta.delivery_count, 1);
    }

    #[tokio::test]
    async fn nack_with_dlq_disabled_returns_dropped() {
        let engine = make_engine();
        let mgr = DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            shared_config_with(|rt| {
                rt.dlq_topic_suffix = String::new();
            }),
            ManagerConfig::default(),
        );

        let client = ClientId(1);
        mgr.connect(&ns(), client).await.unwrap();

        // Subscribe first, then publish
        let sub = mgr
            .subscribe(&ns(), &topic(), client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        let envelope = make_envelope(200, "ns", "test-topic", b"drop-me");
        mgr.publish(&ns(), &topic(), envelope).await.unwrap();

        let delivered = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        let result = mgr
            .nack(&ns(), &topic(), client, delivered.envelope.id, "bad".into())
            .await
            .unwrap();
        assert_eq!(result, NackResult::Dropped);
    }

    #[tokio::test]
    async fn nack_unknown_message_id_returns_error() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));

        let client = ClientId(1);
        mgr.connect(&ns(), client).await.unwrap();

        // Subscribe first, then publish
        let sub = mgr
            .subscribe(&ns(), &topic(), client, None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        let envelope = make_envelope(300, "ns", "test-topic", b"data");
        mgr.publish(&ns(), &topic(), envelope).await.unwrap();

        // Consume the message so we have inflight state
        let _delivered = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Nack with a message_id that doesn't exist in inflight
        let result = mgr
            .nack(&ns(), &topic(), client, MessageId(99999), "bad".into())
            .await;
        assert!(result.is_err(), "nack with unknown message_id should fail");
    }

    // -- Delivery attempts histogram tests (Phase 18.6) --

    #[tokio::test]
    async fn ack_records_delivery_attempts_first_delivery() {
        // Verify delivery_count=1 flows through the ack path for individual subscriber.
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"msg-1"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Ack on first delivery — delivery_count=1 recorded in histogram
        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn ack_records_delivery_attempts_group_member() {
        // Verify delivery_count flows through the ack path for group subscribers.
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));
        let group = GroupName("g1".into());

        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(group.clone()),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"msg-1"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Ack on first delivery — delivery_count=1 recorded in histogram
        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();
    }

    /// Helper: find all histogram values for a given metric name in a snapshot.
    fn histogram_values(
        snapshot: metrics_util::debugging::Snapshot,
        metric_name: &str,
    ) -> Vec<f64> {
        use metrics_util::debugging::DebugValue;
        let mut values = Vec::new();
        for (key, _, _, value) in snapshot.into_vec() {
            if key.key().name() == metric_name {
                if let DebugValue::Histogram(vs) = value {
                    values.extend(vs.iter().map(|v| v.into_inner()));
                }
            }
        }
        values
    }

    #[test]
    fn handle_dlq_result_records_delivery_attempts_for_max_attempts() {
        // Verify that Some("dlq") records the delivery_attempts histogram.
        let recorder = metrics_util::debugging::DebuggingRecorder::new();
        let snapshotter = recorder.snapshotter();

        metrics::with_local_recorder(&recorder, || {
            let mut inflight = HashMap::new();
            inflight.insert(
                TopicOffset(5),
                InflightEntry {
                    offset: TopicOffset(5),
                    message_id: MessageId(1),
                    delivered_to: ClientId(1),
                    delivered_at: Instant::now(),
                    delivery_count: 6,
                    first_delivered_at: 1_700_000_000_000,
                    dlq_failures: 0,
                    next_redeliver_at: Instant::now() + Duration::from_secs(300),
                    msg_timestamp: 0,
                    msg_ttl_ms: 0,
                },
            );
            let ctx = DlqContext {
                original_offset: TopicOffset(5),
                first_delivered_at: 1_700_000_000_000,
                delivery_count: 6,
                dlq_failures: 0,
                reason: DlqReason::MaxAttemptsExceeded,
            };

            let removed =
                DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                    Ok(DlqWriteResult {
                        dlq_timestamp: 1_700_000_060_000,
                    }),
                    &ctx,
                    &ns(),
                    &topic(),
                    TopicOffset(5),
                    &mut inflight,
                    Some("dlq"),
                );
            assert!(removed);
            assert!(inflight.is_empty());
        });

        let snap = snapshotter.snapshot();
        let values = histogram_values(snap, "fafb_delivery_attempts");
        assert_eq!(values, vec![6.0], "should record delivery_count=6");
    }

    #[test]
    fn handle_dlq_result_skips_delivery_attempts_for_client_nack() {
        // Verify that passing None skips delivery_attempts recording.
        let recorder = metrics_util::debugging::DebuggingRecorder::new();
        let snapshotter = recorder.snapshotter();

        metrics::with_local_recorder(&recorder, || {
            let mut inflight = HashMap::new();
            inflight.insert(
                TopicOffset(5),
                InflightEntry {
                    offset: TopicOffset(5),
                    message_id: MessageId(1),
                    delivered_to: ClientId(1),
                    delivered_at: Instant::now(),
                    delivery_count: 2,
                    first_delivered_at: 1_700_000_000_000,
                    dlq_failures: 0,
                    next_redeliver_at: Instant::now() + Duration::from_secs(300),
                    msg_timestamp: 0,
                    msg_ttl_ms: 0,
                },
            );
            let ctx = DlqContext {
                original_offset: TopicOffset(5),
                first_delivered_at: 1_700_000_000_000,
                delivery_count: 2,
                dlq_failures: 0,
                reason: DlqReason::ClientNack {
                    reason: "bad data".into(),
                },
            };

            let removed =
                DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::handle_dlq_result(
                    Ok(DlqWriteResult {
                        dlq_timestamp: 1_700_000_060_000,
                    }),
                    &ctx,
                    &ns(),
                    &topic(),
                    TopicOffset(5),
                    &mut inflight,
                    None,
                );
            assert!(removed);
            assert!(inflight.is_empty());
        });

        let snap = snapshotter.snapshot();
        let values = histogram_values(snap, "fafb_delivery_attempts");
        assert!(
            values.is_empty(),
            "should NOT record delivery_attempts when outcome is None"
        );
    }

    #[test]
    fn record_delivery_attempts_helper_records_correct_value() {
        // Verify the helper records the correct delivery_count value.
        let recorder = metrics_util::debugging::DebuggingRecorder::new();
        let snapshotter = recorder.snapshotter();

        metrics::with_local_recorder(&recorder, || {
            DefaultNamespaceManager::<InMemoryQueueEngine<MockStorage>>::record_delivery_attempts(
                &ns(),
                &topic(),
                "ack",
                3,
            );
        });

        let snap = snapshotter.snapshot();
        let values = histogram_values(snap, "fafb_delivery_attempts");
        assert_eq!(values, vec![3.0], "should record delivery_count=3");
    }

    #[tokio::test]
    async fn nack_records_delivery_attempts() {
        // Verify delivery_count flows through the nack path.
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"nack-me"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Nack after first delivery — delivery_count=1 recorded in histogram
        let result = mgr
            .nack(
                &ns(),
                &topic(),
                ClientId(1),
                msg.envelope.id,
                "bad data".into(),
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn nack_with_dlq_disabled_records_delivery_attempts() {
        // Verify delivery_count is recorded even when DLQ is disabled (dropped).
        let engine = make_engine();
        let mgr = Arc::new(DefaultNamespaceManager::new(
            Arc::clone(&engine),
            vec![ns()],
            shared_config_with(|rt| {
                rt.dlq_topic_suffix = String::new(); // DLQ disabled
            }),
            ManagerConfig::default(),
        ));

        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();
        let mut rx = sub.rx;

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "ns", "test-topic", b"drop-me"),
        )
        .await
        .unwrap();

        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        let result = mgr
            .nack(
                &ns(),
                &topic(),
                ClientId(1),
                msg.envelope.id,
                "unwanted".into(),
            )
            .await;
        assert!(matches!(result, Ok(NackResult::Dropped)));
    }

    // -- Phase 21.4: Delivery Observability Metrics --

    #[tokio::test]
    async fn ack_latency_histogram_records_positive_duration() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        mgr.connect(&ns(), ClientId(1)).await.unwrap();

        // Publish a message
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(0, "ns", "test-topic", b"data"),
        )
        .await
        .unwrap();

        // Subscribe as a consumer group so we get inflight tracking
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(GroupName("g1".into())),
                SubscribeStart::Offset(TopicOffset::ZERO),
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Receive the message
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        // Small delay so latency is measurably > 0
        tokio::time::sleep(Duration::from_millis(5)).await;

        // Ack the message — this should record ack latency
        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();

        // If we get here without panic, the ack latency histogram was recorded.
        // We can't directly read the histogram value without a metrics recorder,
        // but the code path exercised the `delivered_at.elapsed()` recording.
    }

    #[tokio::test]
    async fn duplicate_ack_increments_counter() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        mgr.connect(&ns(), ClientId(1)).await.unwrap();

        // Publish a message
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(0, "ns", "test-topic", b"data"),
        )
        .await
        .unwrap();

        // Subscribe as individual
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                None,
                SubscribeStart::Offset(TopicOffset::ZERO),
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Receive and ack the message
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .unwrap()
            .unwrap();

        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();

        // Ack the same message_id again — this is a duplicate ack.
        // Should not error, but should increment fafb_duplicate_ack_total.
        mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn consumer_lag_updates_on_publish_and_ack() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        mgr.connect(&ns(), ClientId(1)).await.unwrap();

        // Subscribe to a consumer group at head (offset 0)
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(GroupName("g1".into())),
                SubscribeStart::Head,
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Publish 3 messages — each publish updates consumer lag gauge
        for i in 0..3u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Head is now 3, group.next_offset starts at 0 but advances as
        // messages are delivered. Receive all 3 messages.
        let mut message_ids = Vec::new();
        for _ in 0..3 {
            let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
                .await
                .unwrap()
                .unwrap();
            message_ids.push(msg.envelope.id);
        }

        // Ack all 3 — each ack updates consumer lag gauge for the group
        for mid in message_ids {
            mgr.ack(&ns(), &topic(), ClientId(1), mid).await.unwrap();
        }

        // Verify head_offset is 3 (next offset to assign)
        let head = mgr.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(head, TopicOffset(3));
    }

    #[tokio::test]
    async fn batch_ack_records_latency_and_duplicate_metrics() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        mgr.connect(&ns(), ClientId(1)).await.unwrap();

        // Publish 3 messages
        for i in 0..3u64 {
            mgr.publish(
                &ns(),
                &topic(),
                make_envelope(i, "ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        }

        // Subscribe as group
        let sub = mgr
            .subscribe(
                &ns(),
                &topic(),
                ClientId(1),
                Some(GroupName("g1".into())),
                SubscribeStart::Offset(TopicOffset::ZERO),
            )
            .await
            .unwrap();
        let mut rx = sub.rx;

        // Receive all 3
        let mut received = Vec::new();
        for _ in 0..3 {
            let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
                .await
                .unwrap()
                .unwrap();
            received.push(msg.envelope.id);
        }

        // Ack the first message individually
        mgr.ack(&ns(), &topic(), ClientId(1), received[0])
            .await
            .unwrap();

        // Batch ack: second + third (valid) + first again (duplicate)
        let acks = vec![
            (received[1], topic()),
            (received[2], topic()),
            (received[0], topic()), // already acked — duplicate
        ];
        let result = mgr.batch_ack(&ns(), ClientId(1), acks).await;

        // 2 acked (second and third), 1 unknown (first was already acked)
        assert_eq!(result.acked, 2);
        assert_eq!(result.unknown, 1);
    }

    // ── commit_offset tests ──────────────────────────────────────

    #[tokio::test]
    async fn commit_offset_creates_group_and_sets_offset() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Publish a message first so the topic exists.
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "test-ns", "test-topic", b"data"),
        )
        .await
        .unwrap();

        // commit_offset creates the group with offset 5.
        mgr.commit_offset(
            &ns(),
            &topic(),
            &GroupName("workers".into()),
            TopicOffset(5),
        )
        .await
        .unwrap();

        // A second commit_offset updates the existing group.
        mgr.commit_offset(
            &ns(),
            &topic(),
            &GroupName("workers".into()),
            TopicOffset(10),
        )
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn commit_offset_updates_existing_group() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(1, "test-ns", "test-topic", b"data"),
        )
        .await
        .unwrap();

        // Create group at offset 0.
        mgr.commit_offset(&ns(), &topic(), &GroupName("g1".into()), TopicOffset(0))
            .await
            .unwrap();

        // Update to offset 42.
        mgr.commit_offset(&ns(), &topic(), &GroupName("g1".into()), TopicOffset(42))
            .await
            .unwrap();

        // The update should not error.
    }

    // ── validate_publish / apply_publish split tests ─────────────

    #[tokio::test]
    async fn validate_then_apply_matches_publish() {
        let engine = make_engine();
        let mgr = Arc::new(make_manager(Arc::clone(&engine)));

        // Use the split path: validate then apply.
        let msg = make_envelope(1, "test-ns", "test-topic", b"split-test");
        let validated = mgr.validate_publish(&ns(), &topic(), msg).await.unwrap();
        let result = mgr.apply_publish(&ns(), &topic(), validated).await.unwrap();
        assert_eq!(result.offset, TopicOffset(0));
        assert!(result.topic_created);

        // Message is consumable.
        let fetched = engine
            .consume_from(&ns(), &topic(), TopicOffset(0))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(fetched.envelope.payload, b"split-test");
    }

    // ── last_subscribe_millis tests ──────────────────────────────

    #[tokio::test]
    async fn subscribe_updates_last_subscribe_millis() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        let ns = ns();
        let topic = topic();

        // Publish so topic exists
        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        let ns_state = mgr.get_namespace(&ns).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic).unwrap().lock().await;
        let before = topic_state.last_subscribe_millis;
        drop(topic_state);
        drop(topics);

        std::thread::sleep(std::time::Duration::from_millis(5));

        mgr.subscribe(&ns, &topic, ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic).unwrap().lock().await;
        assert!(topic_state.last_subscribe_millis > before);
    }

    #[tokio::test]
    async fn fetch_updates_last_subscribe_millis() {
        let engine = make_engine();
        let mgr = make_manager(Arc::clone(&engine));
        let ns = ns();
        let topic = topic();

        // Publish so topic + delivery state exist
        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        // Subscribe so delivery state is created
        mgr.subscribe(&ns, &topic, ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        let ns_state = mgr.get_namespace(&ns).await.unwrap();
        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic).unwrap().lock().await;
        let before = topic_state.last_subscribe_millis;
        drop(topic_state);
        drop(topics);

        std::thread::sleep(std::time::Duration::from_millis(5));

        // Fetch updates last_subscribe_millis when delivery state exists
        mgr.fetch(&ns, &topic, TopicOffset::ZERO, 10, 1024 * 1024, false)
            .await
            .unwrap();

        let topics = ns_state.topics.read().await;
        let topic_state = topics.get(&topic).unwrap().lock().await;
        assert!(topic_state.last_subscribe_millis > before);
    }

    // -- topic_last_subscribe / has_active_subscribers / has_inflight_messages / teardown_topic tests --

    #[tokio::test]
    async fn topic_last_subscribe_returns_zero_without_subscribe() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        assert_eq!(mgr.topic_last_subscribe(&ns, &topic).await, Some(0));
    }

    #[tokio::test]
    async fn has_active_subscribers_false_when_none() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        assert!(!mgr.has_active_subscribers(&ns, &topic).await);
    }

    #[tokio::test]
    async fn has_active_subscribers_true_after_subscribe() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        mgr.connect(&ns, ClientId(1)).await.unwrap();
        mgr.subscribe(&ns, &topic, ClientId(1), None, SubscribeStart::Head)
            .await
            .unwrap();

        assert!(mgr.has_active_subscribers(&ns, &topic).await);
    }

    #[tokio::test]
    async fn teardown_topic_removes_delivery_state() {
        let engine = make_engine();
        let mgr = make_manager(engine.clone());
        let ns = ns();
        let topic = topic();

        let env = make_envelope(0, "ns", "test-topic", b"data");
        mgr.publish(&ns, &topic, env).await.unwrap();

        mgr.teardown_topic(&ns, &topic).await.unwrap();

        let ns_state = mgr.get_namespace(&ns).await.unwrap();
        let topics = ns_state.topics.read().await;
        assert!(!topics.contains_key(&topic));
    }
}
