use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering::Relaxed};
use std::time::Duration;

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

use crate::config::SharedConfig;
use crate::error::{Error, Result};
use crate::queue::topic::{BackpressurePolicy, Topic, TopicStats};
use crate::queue::{QueueEngine, RemoveTopicResult};
use crate::storage::StorageBackend;
use crate::types::{MessageEnvelope, NamespaceId, PublishedMessage, TopicName, TopicOffset};

type TopicMap = HashMap<(NamespaceId, TopicName), Arc<Mutex<Topic>>>;

/// Configuration for the in-memory queue engine.
#[derive(Debug, Clone)]
pub struct EngineConfig {
    pub max_depth: usize,
    pub backpressure: BackpressurePolicy,
    pub topic_max_bytes: u64,
    pub global_max_memory_bytes: u64,
    pub flush_interval_ms: u64,
    /// Trigger an early flush when a topic accumulates this many unflushed bytes.
    /// 0 means size-based flushing is disabled (interval-only). Default: 4MB.
    pub batch_flush_size_bytes: u64,
    /// When true, publishes are accepted as long as disk has space.
    /// The global memory budget becomes advisory (DropOldest keeps memory
    /// bounded) and the disk capacity threshold replaces it as the
    /// acceptance gate. Default: false.
    pub disk_backed: bool,
    /// Reject publishes when disk usage exceeds this percentage (0–100).
    /// Only applies when `disk_backed = true`. Default: 90.0.
    pub disk_capacity_threshold_pct: f64,
    /// Path to the data directory for disk capacity checks.
    /// Only used when `disk_backed = true`.
    pub data_dir: Option<std::path::PathBuf>,
}

impl Default for EngineConfig {
    fn default() -> Self {
        Self {
            max_depth: 100_000,
            backpressure: BackpressurePolicy::DropOldest,
            topic_max_bytes: 268_435_456,
            global_max_memory_bytes: 4_294_967_296,
            flush_interval_ms: 100,
            batch_flush_size_bytes: 4_194_304,
            disk_backed: false,
            disk_capacity_threshold_pct: 90.0,
            data_dir: None,
        }
    }
}

/// Concrete in-memory queue engine backed by a [`StorageBackend`] for durability.
pub struct InMemoryQueueEngine<S: StorageBackend> {
    topics: Arc<RwLock<TopicMap>>,
    storage: Arc<S>,
    shared_config: SharedConfig,
    total_bytes: AtomicU64,
    /// Set when the 80% memory warning has fired; cleared when usage drops below.
    memory_warning_fired: AtomicBool,
    flush_interval: Duration,
    shutdown_tx: watch::Sender<bool>,
    flush_handle: Mutex<Option<JoinHandle<()>>>,
    /// When true, disk capacity gates publish acceptance instead of memory budget.
    disk_backed: bool,
    /// Cached disk usage percentage (0–100), updated by the flush task.
    disk_used_pct: Arc<AtomicU32>,
    /// Set when the 80% disk warning has fired.
    disk_warning_fired: Arc<AtomicBool>,
    /// Fired once when cold-path reads begin; resets after 60s of no cold reads.
    cold_read_warning_fired: AtomicBool,
    /// Data directory path for disk capacity checks.
    data_dir: Option<PathBuf>,
}

impl<S: StorageBackend + 'static> InMemoryQueueEngine<S> {
    /// Create a new engine. Call [`init`](Self::init) before use to replay
    /// persisted state and start the background flush task.
    pub fn new(storage: Arc<S>, config: EngineConfig) -> Self {
        Self::with_shared_config(storage, config, None)
    }

    /// Create a new engine with a SharedConfig for hot-reload support.
    /// When `shared_config` is `Some`, publish-path limit checks read from it
    /// instead of owned fields, allowing runtime reconfiguration.
    pub fn with_shared_config(
        storage: Arc<S>,
        config: EngineConfig,
        shared_config: Option<SharedConfig>,
    ) -> Self {
        let (shutdown_tx, _) = watch::channel(false);

        // Create a SharedConfig from EngineConfig if none provided.
        // This ensures tests that pass EngineConfig values get matching behavior.
        let shared_config = shared_config.unwrap_or_else(|| {
            let mut rt = crate::config::RuntimeConfig::from_server_config(
                &crate::config::ServerConfig::default(),
            );
            rt.max_depth = config.max_depth;
            rt.default_topic_max_bytes = config.topic_max_bytes;
            rt.backpressure_policy = config.backpressure;
            rt.global_max_memory_bytes = config.global_max_memory_bytes;
            rt.disk_capacity_threshold_pct = config.disk_capacity_threshold_pct;
            rt.batch_flush_size_bytes = config.batch_flush_size_bytes;
            rt.batch_flush_interval_ms = config.flush_interval_ms;
            Arc::new(arc_swap::ArcSwap::from_pointee(rt))
        });

        Self {
            topics: Arc::new(RwLock::new(HashMap::new())),
            storage,
            shared_config,
            total_bytes: AtomicU64::new(0),
            memory_warning_fired: AtomicBool::new(false),
            flush_interval: Duration::from_millis(config.flush_interval_ms),
            shutdown_tx,
            flush_handle: Mutex::new(None),
            disk_backed: config.disk_backed,
            disk_used_pct: Arc::new(AtomicU32::new(0)),
            disk_warning_fired: Arc::new(AtomicBool::new(false)),
            cold_read_warning_fired: AtomicBool::new(false),
            data_dir: config.data_dir,
        }
    }

    /// Replay persisted topics from storage and start the background flush task.
    ///
    /// Topics are replayed in parallel (up to 8 concurrent I/O tasks) for
    /// faster startup on deployments with many topics.
    pub async fn init(&self) -> Result<()> {
        let persisted = self.storage.list_topics().await?;
        let total_topics = persisted.len();
        let init_start = std::time::Instant::now();

        // Replay topics in parallel, bounded by a semaphore.
        let semaphore = Arc::new(tokio::sync::Semaphore::new(8));
        let mut join_set = tokio::task::JoinSet::new();

        let cfg = self.shared_config.load();
        let max_depth = cfg.max_depth;
        let max_bytes = cfg.default_topic_max_bytes;
        let backpressure = if self.disk_backed {
            BackpressurePolicy::DropOldest
        } else {
            cfg.backpressure_policy
        };
        drop(cfg);

        for (ns, topic_name) in persisted {
            let storage = Arc::clone(&self.storage);
            let sem = Arc::clone(&semaphore);
            join_set.spawn(async move {
                let _permit = sem.acquire().await.expect("semaphore closed");
                let replay_start = std::time::Instant::now();
                let messages = storage.replay(&ns, &topic_name).await?;
                let replay_elapsed = replay_start.elapsed();
                metrics::histogram!("fafb_storage_replay_duration_seconds", "namespace" => ns.0.clone())
                    .record(replay_elapsed.as_secs_f64());
                let msg_count = messages.len();
                let bytes: usize = messages.iter().map(|m| m.envelope.payload.len()).sum();
                let t = Topic::from_replay(messages, max_depth, max_bytes, backpressure);
                tracing::info!(
                    namespace = %ns, topic = %topic_name,
                    messages = msg_count, elapsed_ms = replay_elapsed.as_millis() as u64,
                    "replayed {ns}/{topic_name}: {msg_count} messages"
                );
                Ok::<_, crate::error::Error>((ns, topic_name, t, bytes as u64))
            });
        }

        let mut topics = self.topics.write().await;
        let mut total_bytes = 0u64;
        let mut total_messages = 0u64;
        let mut failed_topics = 0u64;

        while let Some(result) = join_set.join_next().await {
            match result {
                Ok(Ok((ns, topic_name, t, bytes))) => {
                    total_messages += t.depth() as u64;
                    total_bytes += bytes;
                    topics.insert((ns, topic_name), Arc::new(Mutex::new(t)));
                }
                Ok(Err(e)) => {
                    // Replay failed for one topic — log and continue with the rest.
                    tracing::error!(error = %e, "failed to replay topic, skipping");
                    failed_topics += 1;
                }
                Err(e) => {
                    tracing::error!(error = %e, "replay task panicked, skipping");
                    failed_topics += 1;
                }
            }
        }
        drop(topics);

        let init_elapsed = init_start.elapsed();
        tracing::info!(
            total_topics = total_topics,
            total_messages = total_messages,
            failed_topics = failed_topics,
            elapsed_ms = init_elapsed.as_millis() as u64,
            "replay complete: {total_topics} topics, {total_messages} messages, \
             {failed_topics} failed in {init_elapsed:?}",
        );
        if init_elapsed.as_secs() >= 5 {
            tracing::warn!(
                elapsed_secs = init_elapsed.as_secs(),
                total_topics,
                total_messages,
                "startup replay took longer than expected — \
                 consider reducing data volume or increasing I/O capacity"
            );
        }

        if failed_topics > 0 {
            tracing::warn!(
                failed_topics = failed_topics,
                "some topics failed to replay — they will appear empty until re-published"
            );
        }

        self.total_bytes.store(total_bytes, Relaxed);

        let handle = self.spawn_flush_task(self.flush_interval);
        *self.flush_handle.lock().await = Some(handle);

        Ok(())
    }

    /// Spawn a background task that periodically flushes all topics to storage.
    ///
    /// On each cycle the task also checks each topic's accumulated dirty bytes.
    /// If a topic exceeds `batch_flush_size_bytes` the flush happens on that
    /// cycle regardless of the interval, and the counter is reset.
    fn spawn_flush_task(&self, interval: Duration) -> JoinHandle<()> {
        let storage = Arc::clone(&self.storage);
        let topics = Arc::clone(&self.topics);
        let mut shutdown_rx = self.shutdown_tx.subscribe();
        let disk_backed = self.disk_backed;
        let data_dir = self.data_dir.clone();
        let disk_used_pct = Arc::clone(&self.disk_used_pct);
        let disk_warning_fired = Arc::clone(&self.disk_warning_fired);

        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = shutdown_rx.changed() => break,
                    _ = tokio::time::sleep(interval) => {
                        let cycle_start = std::time::Instant::now();
                        let entries: Vec<_> = {
                            let map = topics.read().await;
                            map.iter().map(|(k, v)| (k.clone(), Arc::clone(v))).collect()
                        };
                        let topic_count = entries.len();
                        for ((ns, topic_name), topic_lock) in &entries {
                            // Reset dirty bytes under the topic lock, regardless
                            // of whether we flush (the interval already elapsed).
                            {
                                let mut t = topic_lock.lock().await;
                                t.take_dirty_bytes();
                            }
                            let flush_start = std::time::Instant::now();
                            if let Err(e) = storage.flush(ns, topic_name).await {
                                tracing::warn!(
                                    namespace = %ns,
                                    topic = %topic_name,
                                    error = %e,
                                    "background flush failed"
                                );
                            }
                            metrics::histogram!("fafb_storage_flush_latency_seconds")
                                .record(flush_start.elapsed().as_secs_f64());
                        }

                        // Warn once if the entire flush cycle took >2x the interval
                        let cycle_elapsed = cycle_start.elapsed();
                        if cycle_elapsed > interval * 2 {
                            tracing::warn!(
                                cycle_ms = cycle_elapsed.as_millis() as u64,
                                interval_ms = interval.as_millis() as u64,
                                topics_flushed = topic_count,
                                "slow flush cycle: took >2x the configured interval"
                            );
                        }

                        // Update disk usage cache when in disk-backed mode
                        if disk_backed
                            && let Some(ref dir) = data_dir
                                && let Ok(usage) = crate::storage::disk::disk_usage(dir) {
                                    let pct = usage.used_pct();
                                    disk_used_pct.store(pct, Relaxed);
                                    metrics::gauge!("fafb_disk_used_pct").set(pct as f64);
                                    metrics::gauge!("fafb_disk_available_bytes")
                                        .set(usage.available_bytes as f64);
                                    if pct >= 80 && !disk_warning_fired.swap(true, Relaxed) {
                                        tracing::warn!(used_pct = pct, "disk usage at 80%");
                                    } else if pct < 80 {
                                        disk_warning_fired.store(false, Relaxed);
                                    }
                                }
                    }
                }
            }
        })
    }

    /// Check if a topic has exceeded the batch flush size threshold and
    /// trigger an early flush if so. Called from the publish path.
    async fn maybe_flush_by_size(
        &self,
        ns: &NamespaceId,
        topic_name: &TopicName,
        topic_lock: &Arc<Mutex<Topic>>,
    ) {
        let batch_flush_size_bytes = self.shared_config.load().batch_flush_size_bytes;
        if batch_flush_size_bytes == 0 {
            return;
        }

        let dirty = {
            let t = topic_lock.lock().await;
            t.dirty_bytes()
        };

        if dirty >= batch_flush_size_bytes {
            let flush_start = std::time::Instant::now();
            if let Err(e) = self.storage.flush(ns, topic_name).await {
                tracing::warn!(
                    namespace = %ns, topic = %topic_name,
                    error = %e, "early size-triggered flush failed"
                );
            } else {
                metrics::histogram!("fafb_storage_flush_latency_seconds")
                    .record(flush_start.elapsed().as_secs_f64());
            }
            // Reset counter after flush
            let mut t = topic_lock.lock().await;
            t.take_dirty_bytes();
        }
    }

    /// Look up or create the topic entry for the given namespace/topic pair.
    /// Returns `(topic_lock, created)` where `created` is true if the topic
    /// was newly created by this call.
    async fn get_or_create_topic(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
    ) -> (Arc<Mutex<Topic>>, bool) {
        // Fast path: read lock
        {
            let map = self.topics.read().await;
            if let Some(t) = map.get(&(ns.clone(), topic.clone())) {
                return (Arc::clone(t), false);
            }
        }
        // Slow path: write lock, double-check. Read config for new topic defaults.
        let cfg = self.shared_config.load();
        let max_depth = cfg.max_depth;
        let max_bytes = cfg.default_topic_max_bytes;
        let backpressure = if self.disk_backed {
            BackpressurePolicy::DropOldest
        } else {
            cfg.backpressure_policy
        };
        drop(cfg);

        let mut map = self.topics.write().await;
        let key = (ns.clone(), topic.clone());
        if let Some(t) = map.get(&key) {
            return (Arc::clone(t), false);
        }
        let t = Arc::new(Mutex::new(Topic::new(max_depth, max_bytes, backpressure)));
        map.insert(key, Arc::clone(&t));
        tracing::debug!(namespace = %ns, topic = %topic, "topic created");
        (t, true)
    }

    /// Total payload bytes across all topics.
    pub fn total_bytes(&self) -> u64 {
        self.total_bytes.load(Relaxed)
    }
}

impl<S: StorageBackend + 'static> QueueEngine for InMemoryQueueEngine<S> {
    async fn publish(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
    ) -> Result<crate::queue::PublishResult> {
        let msg_mem_size = msg.mem_size() as u64;
        // Load config once for this publish call (hot-reload aware).
        let cfg = self.shared_config.load();
        let global_max_memory_bytes = cfg.global_max_memory_bytes;
        // Clamp to valid range — guards against invalid values from hot-reload.
        let disk_capacity_threshold_pct = cfg.disk_capacity_threshold_pct.clamp(1.0, 99.9);
        drop(cfg);

        if self.disk_backed {
            // Disk-backed mode: reject only when disk is full
            let pct = self.disk_used_pct.load(Relaxed) as f64;
            if pct >= disk_capacity_threshold_pct {
                tracing::warn!(
                    disk_used_pct = pct,
                    threshold_pct = disk_capacity_threshold_pct,
                    "publish rejected: disk capacity threshold exceeded"
                );
                metrics::counter!("fafb_publish_rejected_total", "reason" => "disk_full")
                    .increment(1);
                return Err(Error::ResourceExhausted(
                    "disk capacity threshold exceeded".to_string(),
                ));
            }
            // Memory budget is advisory — DropOldest keeps in-memory usage bounded
        } else if global_max_memory_bytes > 0 {
            let current = self.total_bytes.load(Relaxed);
            if current + msg_mem_size > global_max_memory_bytes {
                tracing::warn!(
                    used = current + msg_mem_size,
                    limit = global_max_memory_bytes,
                    "publish rejected: global memory budget exceeded"
                );
                metrics::counter!("fafb_publish_rejected_total", "reason" => "memory_full")
                    .increment(1);
                return Err(Error::ResourceExhausted(
                    "global memory budget exceeded".to_string(),
                ));
            }
            let threshold = global_max_memory_bytes * 4 / 5;
            if current + msg_mem_size > threshold {
                // Fire warning once; suppress until usage drops back below threshold
                if !self.memory_warning_fired.swap(true, Relaxed) {
                    tracing::warn!(
                        used = current + msg_mem_size,
                        limit = global_max_memory_bytes,
                        "global memory budget at 80%"
                    );
                }
            } else {
                // Reset the flag so the warning fires again if we cross back over
                self.memory_warning_fired.store(false, Relaxed);
            }
        }

        let (topic_lock, topic_created) = self.get_or_create_topic(ns, topic).await;

        // Wrap in Arc once — shared between the in-memory buffer and storage.
        let arc_envelope = Arc::new(msg);

        // In-memory publish first, then persist to storage page cache.
        //
        // Trade-off: if storage.append fails after in-memory publish succeeds,
        // consumers may read a message that won't survive restart. The reverse
        // ordering (persist first) would give at-most-once but risks offset
        // gaps on storage failure. Page-cache writes virtually never fail, so
        // we accept the current ordering for simplicity.
        let (offset, evicted_bytes, evicted_messages, depth, bytes_used) = {
            let mut t = topic_lock.lock().await;
            let (offset, evicted_bytes, evicted_messages) =
                t.publish(Arc::clone(&arc_envelope), ns, topic)?;
            (
                offset,
                evicted_bytes,
                evicted_messages,
                t.depth(),
                t.bytes_used(),
            )
        };

        let published = PublishedMessage {
            envelope: arc_envelope,
            offset,
        };
        self.storage.append(ns, topic, &published).await?;

        // Check if this topic needs an early flush based on accumulated dirty bytes.
        self.maybe_flush_by_size(ns, topic, &topic_lock).await;

        // Net byte change: added msg_mem_size, possibly evicted some bytes.
        let net_add = msg_mem_size.saturating_sub(evicted_bytes as u64);
        self.total_bytes.fetch_add(net_add, Relaxed);

        if evicted_messages > 0 {
            metrics::counter!("fafb_messages_evicted_total",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone()
            )
            .increment(evicted_messages as u64);
        }
        metrics::gauge!("fafb_global_memory_bytes").set(self.total_bytes.load(Relaxed) as f64);
        metrics::gauge!("fafb_queue_depth", "namespace" => ns.0.clone(), "topic" => topic.0.clone()).set(depth as f64);
        metrics::gauge!("fafb_queue_memory_bytes", "namespace" => ns.0.clone(), "topic" => topic.0.clone()).set(bytes_used as f64);

        Ok(crate::queue::PublishResult {
            offset,
            evicted_messages,
            topic_created,
        })
    }

    async fn consume_from(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        offset: TopicOffset,
    ) -> Result<Option<PublishedMessage>> {
        // Read from memory under topic lock, capturing base/head offsets.
        let (result, base_offset, head_offset) = {
            let map = self.topics.read().await;
            let Some(topic_lock) = map.get(&(ns.clone(), topic.clone())) else {
                return Ok(None);
            };
            let mut t = topic_lock.lock().await;
            let result = t.consume_from(offset).map(|env| PublishedMessage {
                envelope: env,
                offset,
            });
            (result, t.base_offset(), t.head_offset())
        }; // Both locks released here — disk I/O won't block publishes

        if result.is_some() {
            return Ok(result);
        }

        // Cold path: offset was evicted from memory but may exist on disk
        if offset < base_offset && offset < head_offset {
            metrics::counter!("fafb_cold_reads_total",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone()
            )
            .increment(1);

            let cold_start = std::time::Instant::now();
            let disk_result = self.storage.read_at(ns, topic, offset).await?;
            metrics::histogram!("fafb_cold_read_latency_seconds",
                "namespace" => ns.0.clone(), "topic" => topic.0.clone()
            )
            .record(cold_start.elapsed().as_secs_f64());

            return Ok(disk_result);
        }

        Ok(None)
    }

    async fn base_offset(&self, ns: &NamespaceId, topic: &TopicName) -> Result<TopicOffset> {
        let map = self.topics.read().await;
        let Some(topic_lock) = map.get(&(ns.clone(), topic.clone())) else {
            return Ok(TopicOffset::ZERO);
        };
        let t = topic_lock.lock().await;
        Ok(t.base_offset())
    }

    async fn head_offset(&self, ns: &NamespaceId, topic: &TopicName) -> Result<TopicOffset> {
        let map = self.topics.read().await;
        let Some(topic_lock) = map.get(&(ns.clone(), topic.clone())) else {
            return Ok(TopicOffset::ZERO);
        };
        let t = topic_lock.lock().await;
        Ok(t.head_offset())
    }

    async fn compact(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        min_offset: TopicOffset,
    ) -> Result<crate::queue::topic::CompactResult> {
        use crate::queue::topic::CompactResult;
        let map = self.topics.read().await;
        let Some(topic_lock) = map.get(&(ns.clone(), topic.clone())) else {
            return Ok(CompactResult {
                bytes_freed: 0,
                messages_freed: 0,
            });
        };
        let t_arc = Arc::clone(topic_lock);
        drop(map);

        let result = {
            let mut t = t_arc.lock().await;
            t.compact(min_offset)
        };
        self.total_bytes
            .fetch_sub(result.bytes_freed as u64, Relaxed);
        if result.messages_freed > 0 {
            tracing::debug!(
                namespace = %ns,
                topic = %topic,
                messages_freed = result.messages_freed,
                bytes_freed = result.bytes_freed,
                "in-memory compaction"
            );
        }
        Ok(result)
    }

    async fn read_cold_batch(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        start_offset: TopicOffset,
        max_count: usize,
    ) -> Result<Vec<PublishedMessage>> {
        if !self.cold_read_warning_fired.swap(true, Relaxed) {
            tracing::warn!(
                namespace = %ns, topic = %topic, offset = %start_offset,
                "consumer reading from disk (behind in-memory window)"
            );
        }
        metrics::counter!("fafb_cold_batch_reads_total",
            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
        )
        .increment(1);

        let cold_start = std::time::Instant::now();
        let result = self
            .storage
            .read_cold_batch(ns, topic, start_offset, max_count)
            .await;
        metrics::histogram!("fafb_cold_batch_read_latency_seconds",
            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
        )
        .record(cold_start.elapsed().as_secs_f64());

        result
    }

    async fn list_topics(&self, ns: &NamespaceId) -> Vec<TopicName> {
        let map = self.topics.read().await;
        map.keys()
            .filter(|(key_ns, _)| key_ns == ns)
            .map(|(_, topic)| topic.clone())
            .collect()
    }

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

    async fn namespace_disk_bytes(&self, ns: &NamespaceId) -> u64 {
        self.storage.namespace_disk_bytes(ns).await
    }

    async fn topic_stats(&self, ns: &NamespaceId, topic: &TopicName) -> Option<TopicStats> {
        let map = self.topics.read().await;
        let t = map.get(&(ns.clone(), topic.clone()))?;
        Some(t.lock().await.stats())
    }

    async fn force_flush(&self) -> Result<()> {
        // Reset dirty-byte counters so the periodic flush task doesn't double-flush.
        let entries: Vec<_> = {
            let map = self.topics.read().await;
            map.values().map(Arc::clone).collect()
        };
        for topic_lock in &entries {
            topic_lock.lock().await.take_dirty_bytes();
        }
        self.storage.flush_all().await
    }

    #[cfg(feature = "cluster")]
    async fn publish_with_offset(
        &self,
        ns: &NamespaceId,
        topic: &TopicName,
        msg: MessageEnvelope,
        offset: TopicOffset,
    ) -> Result<crate::queue::PublishResult> {
        let msg_mem_size = msg.mem_size() as u64;
        let (topic_lock, topic_created) = self.get_or_create_topic(ns, topic).await;
        let arc_envelope = Arc::new(msg);

        // Write at the given offset — append forward-only.
        let (evicted_bytes, evicted_messages, _depth, _bytes_used) = {
            let mut t = topic_lock.lock().await;
            if offset < t.head_offset() {
                // Duplicate/stale replication entry — skip silently (idempotent).
                return Ok(crate::queue::PublishResult {
                    offset,
                    evicted_messages: 0,
                    topic_created: false,
                });
            }
            // Atomically set next_offset and append in one step.
            let (_assigned, evicted_bytes, evicted_messages) =
                t.publish_at_offset(Arc::clone(&arc_envelope), ns, topic, offset)?;
            (evicted_bytes, evicted_messages, t.depth(), t.bytes_used())
        };

        let published = PublishedMessage {
            envelope: arc_envelope,
            offset,
        };
        self.storage.append(ns, topic, &published).await?;
        self.maybe_flush_by_size(ns, topic, &topic_lock).await;

        let net_add = msg_mem_size.saturating_sub(evicted_bytes as u64);
        self.total_bytes.fetch_add(net_add, Relaxed);

        Ok(crate::queue::PublishResult {
            offset,
            evicted_messages,
            topic_created,
        })
    }

    async fn topic_last_activity(&self, ns: &NamespaceId, topic: &TopicName) -> Option<u64> {
        let map = self.topics.read().await;
        let topic_lock = map.get(&(ns.clone(), topic.clone()))?;
        let t = topic_lock.lock().await;
        Some(t.last_activity_millis())
    }

    async fn remove_topic(&self, ns: &NamespaceId, topic: &TopicName) -> Result<RemoveTopicResult> {
        let mut map = self.topics.write().await;
        let key = (ns.clone(), topic.clone());
        if let Some(topic_lock) = map.remove(&key) {
            let t = topic_lock.lock().await;
            let bytes_freed = t.bytes_used() as u64;
            let messages_freed = t.depth();
            drop(t);
            self.total_bytes.fetch_sub(bytes_freed, Relaxed);
            tracing::debug!(namespace = %ns, topic = %topic, bytes_freed, messages_freed, "topic removed from engine");
            Ok(RemoveTopicResult {
                bytes_freed,
                messages_freed,
            })
        } else {
            Ok(RemoveTopicResult {
                bytes_freed: 0,
                messages_freed: 0,
            })
        }
    }

    async fn shutdown(&self) -> Result<()> {
        let _ = self.shutdown_tx.send(true);
        if let Some(handle) = self.flush_handle.lock().await.take() {
            let _ = handle.await;
        }
        self.storage.shutdown().await
    }
}

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

    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 make_engine() -> InMemoryQueueEngine<MockStorage> {
        InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10_000,
                backpressure: BackpressurePolicy::Reject,
                topic_max_bytes: 0,
                global_max_memory_bytes: 0,
                flush_interval_ms: 100,
                ..Default::default()
            },
        )
    }

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

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

    #[tokio::test]
    async fn publish_auto_creates_topic() {
        let engine = make_engine();
        let result = engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(0, "test-ns", "test-topic", b"hello"),
            )
            .await
            .unwrap();
        assert_eq!(result.offset, TopicOffset(0));
    }

    #[tokio::test]
    async fn publish_sequential_offsets() {
        let engine = make_engine();
        for i in 0..10u64 {
            let result = engine
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "test-ns", "test-topic", b"data"),
                )
                .await
                .unwrap();
            assert_eq!(result.offset, TopicOffset(i));
        }
    }

    #[tokio::test]
    async fn consume_from_existing() {
        let engine = make_engine();
        for i in 0..3u64 {
            engine
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "test-ns", "test-topic", format!("msg-{i}").as_bytes()),
                )
                .await
                .unwrap();
        }
        for i in 0..3u64 {
            let msg = engine
                .consume_from(&ns(), &topic(), TopicOffset(i))
                .await
                .unwrap()
                .unwrap();
            assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
            assert_eq!(msg.offset, TopicOffset(i));
        }
    }

    #[tokio::test]
    async fn consume_from_nonexistent_topic() {
        let engine = make_engine();
        let result = engine
            .consume_from(&ns(), &topic(), TopicOffset(0))
            .await
            .unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn head_offset_nonexistent_returns_zero() {
        let engine = make_engine();
        let offset = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(offset, TopicOffset::ZERO);
    }

    #[tokio::test]
    async fn head_offset_after_publish() {
        let engine = make_engine();
        for i in 0..5u64 {
            engine
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "test-ns", "test-topic", b"data"),
                )
                .await
                .unwrap();
        }
        let offset = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(offset, TopicOffset(5));
    }

    #[tokio::test]
    async fn compact_nonexistent_is_noop() {
        let engine = make_engine();
        engine
            .compact(&ns(), &topic(), TopicOffset(5))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn compact_removes_old_messages() {
        let engine = make_engine();
        for i in 0..5u64 {
            engine
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "test-ns", "test-topic", format!("msg-{i}").as_bytes()),
                )
                .await
                .unwrap();
        }
        engine
            .compact(&ns(), &topic(), TopicOffset(2))
            .await
            .unwrap();
        // Compacted offsets are still readable from disk (cold path).
        assert!(
            engine
                .consume_from(&ns(), &topic(), TopicOffset(0))
                .await
                .unwrap()
                .is_some()
        );
        let msg = engine
            .consume_from(&ns(), &topic(), TopicOffset(2))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"msg-2");
    }

    #[tokio::test]
    async fn topic_stats_returns_correct_data() {
        let engine = make_engine();
        for i in 0..3u64 {
            engine
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "test-ns", "test-topic", &[0u8; 10]),
                )
                .await
                .unwrap();
        }
        let stats = engine.topic_stats(&ns(), &topic()).await.unwrap();
        assert_eq!(stats.depth, 3);
        assert_eq!(stats.bytes_used, 30);
        assert_eq!(stats.base_offset, TopicOffset(0));
        assert_eq!(stats.head_offset, TopicOffset(3));
    }

    #[tokio::test]
    async fn total_bytes_tracking() {
        let engine = make_engine();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(0, "test-ns", "test-topic", &[0u8; 20]),
            )
            .await
            .unwrap();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", &[0u8; 30]),
            )
            .await
            .unwrap();
        assert_eq!(engine.total_bytes(), 50);
        engine
            .compact(&ns(), &topic(), TopicOffset(1))
            .await
            .unwrap();
        assert_eq!(engine.total_bytes(), 30);
    }

    #[tokio::test]
    async fn shutdown_completes_without_error() {
        let engine = make_engine();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(0, "test-ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        engine.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn concurrent_auto_creation_no_duplicates() {
        let engine = Arc::new(make_engine());
        let n = 50;
        let mut handles = Vec::new();
        for i in 0..n {
            let e = Arc::clone(&engine);
            handles.push(tokio::spawn(async move {
                e.publish(
                    &NamespaceId("ns".into()),
                    &TopicName("t".into()),
                    make_envelope(i, "ns", "t", b"x"),
                )
                .await
                .unwrap()
            }));
        }
        let mut offsets: Vec<u64> = Vec::new();
        for h in handles {
            offsets.push(h.await.unwrap().offset.0);
        }
        offsets.sort();
        let expected: Vec<u64> = (0..n).collect();
        assert_eq!(offsets, expected);
    }

    #[tokio::test]
    async fn replay_restores_state() {
        let storage = Arc::new(MockStorage::new());

        // Pre-populate storage
        {
            let engine = InMemoryQueueEngine::new(
                Arc::clone(&storage),
                EngineConfig {
                    max_depth: 10_000,
                    backpressure: BackpressurePolicy::Reject,
                    ..Default::default()
                },
            );
            for i in 0..5u64 {
                engine
                    .publish(
                        &ns(),
                        &topic(),
                        make_envelope(i, "test-ns", "test-topic", format!("msg-{i}").as_bytes()),
                    )
                    .await
                    .unwrap();
            }
        }

        // New engine, init from storage
        let engine = InMemoryQueueEngine::new(
            Arc::clone(&storage),
            EngineConfig {
                max_depth: 10_000,
                backpressure: BackpressurePolicy::Reject,
                ..Default::default()
            },
        );
        engine.init().await.unwrap();

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

        for i in 0..5u64 {
            let msg = engine
                .consume_from(&ns(), &topic(), TopicOffset(i))
                .await
                .unwrap()
                .unwrap();
            assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
        }

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

    #[tokio::test]
    async fn drop_oldest_evicts_and_tracks_total_bytes() {
        let engine = InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 2,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        );
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Publish 3 messages with 10-byte payloads into a depth-2 topic.
        for i in 0..3u64 {
            engine
                .publish(&ns, &topic, make_envelope(i, "ns", "t", &[0u8; 10]))
                .await
                .unwrap();
        }

        // Only 2 messages should remain (offsets 1, 2).
        let stats = engine.topic_stats(&ns, &topic).await.unwrap();
        assert_eq!(stats.depth, 2);
        assert_eq!(stats.base_offset, TopicOffset(1));
        assert_eq!(stats.head_offset, TopicOffset(3));

        // total_bytes should reflect only the 2 retained messages.
        assert_eq!(engine.total_bytes(), 20);

        // Evicted message is still readable from disk (cold path).
        assert!(
            engine
                .consume_from(&ns, &topic, TopicOffset(0))
                .await
                .unwrap()
                .is_some()
        );
        // Retained messages are present.
        assert!(
            engine
                .consume_from(&ns, &topic, TopicOffset(1))
                .await
                .unwrap()
                .is_some()
        );
    }

    #[tokio::test]
    async fn global_memory_budget_rejects_when_exceeded() {
        // Global budget of 50 bytes
        let engine = InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10_000,
                backpressure: BackpressurePolicy::Reject,
                global_max_memory_bytes: 50,
                ..Default::default()
            },
        );
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Publish 2 messages of 20 bytes each (total: 40)
        engine
            .publish(&ns, &topic, make_envelope(0, "ns", "t", &[0u8; 20]))
            .await
            .unwrap();
        engine
            .publish(&ns, &topic, make_envelope(1, "ns", "t", &[0u8; 20]))
            .await
            .unwrap();
        assert_eq!(engine.total_bytes(), 40);

        // Third message (20 bytes) would exceed budget: 40 + 20 > 50
        let err = engine
            .publish(&ns, &topic, make_envelope(2, "ns", "t", &[0u8; 20]))
            .await
            .unwrap_err();
        assert!(matches!(err, crate::error::Error::ResourceExhausted(_)));
        assert!(err.to_string().contains("global memory budget exceeded"));
    }

    #[tokio::test]
    async fn global_memory_budget_zero_means_unlimited() {
        let engine = InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 10_000,
                backpressure: BackpressurePolicy::Reject,
                global_max_memory_bytes: 0,
                ..Default::default()
            },
        );
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        for i in 0..100u64 {
            engine
                .publish(&ns, &topic, make_envelope(i, "ns", "t", &[0u8; 1000]))
                .await
                .unwrap();
        }
        assert_eq!(engine.total_bytes(), 100_000);
    }

    // Note: disk_capacity_threshold_pct validation/clamping tests were removed
    // because the engine now reads this value from SharedConfig (RuntimeConfig).
    // The validation lives in config.rs::validate().

    #[tokio::test]
    async fn read_cold_batch_returns_sequential_messages() {
        let engine = InMemoryQueueEngine::new(
            Arc::new(MockStorage::new()),
            EngineConfig {
                max_depth: 100_000,
                backpressure: BackpressurePolicy::DropOldest,
                ..Default::default()
            },
        );
        let ns = NamespaceId("ns".into());
        let topic = TopicName("t".into());

        // Publish 100 messages
        for i in 0..100u64 {
            engine
                .publish(
                    &ns,
                    &topic,
                    make_envelope(i, "ns", "t", format!("msg-{i}").as_bytes()),
                )
                .await
                .unwrap();
        }

        // Compact first 50 from memory
        engine.compact(&ns, &topic, TopicOffset(50)).await.unwrap();

        // read_cold_batch starting at offset 0, requesting 10
        let msgs = engine
            .read_cold_batch(&ns, &topic, TopicOffset(0), 10)
            .await
            .unwrap();
        assert_eq!(msgs.len(), 10, "should return 10 messages");

        // Verify they are sequential starting from offset 0
        for (i, msg) in msgs.iter().enumerate() {
            assert_eq!(msg.offset, TopicOffset(i as u64));
            assert_eq!(msg.envelope.payload, format!("msg-{i}").as_bytes());
        }
    }

    #[cfg(feature = "cluster")]
    #[tokio::test]
    async fn publish_with_offset_writes_at_given_offset() {
        let engine = make_engine();
        let result = engine
            .publish_with_offset(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", b"hello"),
                TopicOffset(0),
            )
            .await
            .unwrap();
        assert_eq!(result.offset, TopicOffset(0));
        assert!(result.topic_created);

        // Message is consumable at the given offset.
        let msg = engine
            .consume_from(&ns(), &topic(), TopicOffset(0))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"hello");
        assert_eq!(msg.offset, TopicOffset(0));

        // Head offset advanced.
        let head = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(head, TopicOffset(1));
    }

    #[cfg(feature = "cluster")]
    #[tokio::test]
    async fn publish_with_offset_duplicate_is_idempotent() {
        let engine = make_engine();
        engine
            .publish_with_offset(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", b"first"),
                TopicOffset(0),
            )
            .await
            .unwrap();

        // Publish at same offset again — should be idempotent skip.
        let result = engine
            .publish_with_offset(
                &ns(),
                &topic(),
                make_envelope(2, "test-ns", "test-topic", b"duplicate"),
                TopicOffset(0),
            )
            .await
            .unwrap();
        assert_eq!(result.offset, TopicOffset(0));
        assert!(!result.topic_created);

        // Original message still there.
        let msg = engine
            .consume_from(&ns(), &topic(), TopicOffset(0))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(msg.envelope.payload, b"first");

        // Head didn't advance past 1.
        let head = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(head, TopicOffset(1));
    }

    #[cfg(feature = "cluster")]
    #[tokio::test]
    async fn publish_with_offset_sequential() {
        let engine = make_engine();
        for i in 0..5u64 {
            let result = engine
                .publish_with_offset(
                    &ns(),
                    &topic(),
                    make_envelope(
                        i + 1,
                        "test-ns",
                        "test-topic",
                        format!("msg-{i}").as_bytes(),
                    ),
                    TopicOffset(i),
                )
                .await
                .unwrap();
            assert_eq!(result.offset, TopicOffset(i));
        }
        let head = engine.head_offset(&ns(), &topic()).await.unwrap();
        assert_eq!(head, TopicOffset(5));
    }

    #[tokio::test]
    async fn topic_last_activity_returns_none_for_missing_topic() {
        let engine = make_engine();
        assert!(engine.topic_last_activity(&ns(), &topic()).await.is_none());
    }

    #[tokio::test]
    async fn topic_last_activity_returns_timestamp_after_publish() {
        let engine = make_engine();
        let before = crate::types::now_millis();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        let activity = engine.topic_last_activity(&ns(), &topic()).await.unwrap();
        assert!(activity >= before);
    }

    #[tokio::test]
    async fn remove_topic_removes_from_map() {
        let engine = make_engine();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", b"data"),
            )
            .await
            .unwrap();
        assert!(engine.topic_last_activity(&ns(), &topic()).await.is_some());
        engine.remove_topic(&ns(), &topic()).await.unwrap();
        assert!(engine.topic_last_activity(&ns(), &topic()).await.is_none());
    }

    #[tokio::test]
    async fn remove_topic_returns_freed_stats() {
        let engine = make_engine();
        engine
            .publish(
                &ns(),
                &topic(),
                make_envelope(1, "test-ns", "test-topic", b"hello"),
            )
            .await
            .unwrap();
        let bytes_before = engine.total_bytes();
        assert!(bytes_before > 0);
        let RemoveTopicResult {
            bytes_freed,
            messages_freed,
        } = engine.remove_topic(&ns(), &topic()).await.unwrap();
        assert!(bytes_freed > 0);
        assert_eq!(messages_freed, 1);
        assert_eq!(engine.total_bytes(), 0);
    }

    #[tokio::test]
    async fn remove_topic_nonexistent_returns_zero() {
        let engine = make_engine();
        let RemoveTopicResult {
            bytes_freed: bytes,
            messages_freed: msgs,
        } = engine.remove_topic(&ns(), &topic()).await.unwrap();
        assert_eq!(bytes, 0);
        assert_eq!(msgs, 0);
    }
}
