use std::collections::HashMap;
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use arc_swap::ArcSwap;
use clap::Parser;
use serde::{Deserialize, Serialize};

use crate::error::Error;
use crate::queue::topic::BackpressurePolicy;
use crate::types::{ApiKey, NamespaceId, Permission};

/// Policy for handling subscribers that fall behind.
///
/// - `DropForSlow`: Skip delivery to slow subscribers; increment drop counter.
/// - `DisconnectSlow`: After N consecutive drops, close the subscriber's connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SlowConsumerPolicy {
    #[serde(rename = "drop_for_slow")]
    DropForSlow,
    #[serde(rename = "disconnect_slow")]
    DisconnectSlow,
}

fn default_slow_consumer_policy() -> SlowConsumerPolicy {
    SlowConsumerPolicy::DropForSlow
}

fn default_slow_consumer_threshold() -> u64 {
    100
}

fn default_slow_consumer_disconnect_after() -> u64 {
    1000
}

fn default_subscriber_channel_buffer() -> usize {
    1024
}

/// Controls when `fsync` is called on segment files.
///
/// - `EveryWrite`: fsync after every `append()` — maximum durability, lowest throughput.
/// - `EveryBatch`: fsync during each background flush cycle (default).
/// - `OsManaged`: no explicit fsync — rely on OS page-cache writeback.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum FsyncStrategy {
    #[serde(rename = "every_write")]
    EveryWrite,
    #[serde(rename = "every_batch")]
    EveryBatch,
    #[serde(rename = "os_managed")]
    OsManaged,
}

fn default_fsync_strategy() -> FsyncStrategy {
    FsyncStrategy::EveryBatch
}

fn default_flush_interval() -> u64 {
    100
}

fn default_flush_size() -> u64 {
    4_194_304 // 4 MB
}

fn default_max_open_fds() -> usize {
    1024
}

fn default_max_segments() -> usize {
    64
}

fn default_retention_interval() -> u64 {
    60
}

fn default_index_interval() -> u32 {
    256
}

fn default_merge_threshold() -> u64 {
    16 * 1024 * 1024 // 16 MiB
}

fn default_compaction_interval() -> u64 {
    300 // 5 minutes
}

/// Parse a boolean from an environment variable value.
/// Accepts "true", "false", "1", "0" (case-insensitive).
fn parse_bool(s: &str) -> Option<bool> {
    match s.to_ascii_lowercase().as_str() {
        "true" | "1" => Some(true),
        "false" | "0" => Some(false),
        _ => None,
    }
}

/// Parse a backpressure policy from an environment variable value.
fn parse_backpressure_policy(s: &str) -> Option<BackpressurePolicy> {
    match s {
        "Reject" => Some(BackpressurePolicy::Reject),
        "DropOldest" => Some(BackpressurePolicy::DropOldest),
        "Block" => Some(BackpressurePolicy::Block),
        _ => None,
    }
}

/// Server-specific CLI arguments (used with the `serve` subcommand).
#[derive(Debug, Clone, Default, Parser)]
#[command(about = "Run the fafb-q server")]
pub struct ServeArgs {
    /// Path to config TOML file
    #[arg(short, long)]
    pub config: Option<PathBuf>,

    /// Print the resolved configuration (secrets redacted) and exit
    #[arg(long)]
    pub dump_config: bool,

    /// Override TCP bind address (e.g., "0.0.0.0:9876")
    #[arg(long)]
    pub tcp_bind: Option<String>,

    /// Override HTTP bind address (e.g., "0.0.0.0:8080")
    #[arg(long)]
    pub http_bind: Option<String>,

    /// Override storage data directory
    #[arg(long)]
    pub data_dir: Option<String>,

    /// Override log level (e.g., "debug", "info,fafb_q::storage=trace")
    #[arg(long)]
    pub log_level: Option<String>,
}

/// Discover the config file path using a 4-tier search:
/// 1. CLI `--config` flag (highest priority)
/// 2. `FAFB_CONFIG` environment variable
/// 3. `./config.toml` in the current directory
/// 4. `/etc/fafb-q/config.toml` system-wide
///
/// Returns `None` when no config file is found. Note: with no config file,
/// `load_defaults()` will fail validation because namespaces cannot be defined
/// via env vars. The `None` path exists to produce a clear, helpful error
/// message rather than a cryptic "file not found".
pub fn find_config_path(
    cli_path: Option<PathBuf>,
    env: impl Fn(&str) -> Option<String>,
) -> Option<PathBuf> {
    if let Some(path) = cli_path {
        return Some(path);
    }

    if let Some(path) = env("FAFB_CONFIG") {
        let p = PathBuf::from(path);
        if p.exists() {
            return Some(p);
        }
        eprintln!(
            "WARNING: FAFB_CONFIG points to non-existent file: {}",
            p.display()
        );
    }

    let local = PathBuf::from("config.toml");
    if local.exists() {
        return Some(local);
    }

    let system = PathBuf::from("/etc/fafb-q/config.toml");
    if system.exists() {
        return Some(system);
    }

    None
}

/// Warn if the config file is world-readable (Unix only).
/// Config files often contain API keys and should not be readable by other users.
/// Only warns about world-readable (other-read bit), not group-readable,
/// since group-read (0o644) is a common and often intentional default.
#[cfg(unix)]
pub fn check_config_permissions(path: &Path) {
    use std::os::unix::fs::PermissionsExt;
    if let Ok(meta) = std::fs::metadata(path) {
        let mode = meta.permissions().mode();
        if mode & 0o004 != 0 {
            eprintln!(
                "WARNING: config file {} is world-readable (mode {:o}) — consider chmod 640 or 600 to protect API keys",
                path.display(),
                mode
            );
        }
    }
}

#[cfg(not(unix))]
pub fn check_config_permissions(_path: &Path) {}

/// Metrics / observability configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct MetricsConfig {
    pub enabled: bool,
}

impl Default for MetricsConfig {
    fn default() -> Self {
        Self { enabled: true }
    }
}

/// Logging configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct LoggingConfig {
    /// Log level filter. Supports `RUST_LOG` syntax
    /// (e.g. `"info"`, `"info,fafb_q::storage=debug"`).
    /// Overridden by the `RUST_LOG` environment variable when set.
    /// Default: `"info"`.
    pub level: String,
    /// Output format: `"text"` or `"json"`. Default: `"text"`.
    pub format: String,
    /// Write log output to stdout. Default: `true`.
    pub stdout: bool,
    /// Write log output to rotating files. Default: `false`.
    pub file_enabled: bool,
    /// Directory for log files. Default: `"./logs"`.
    pub file_dir: String,
    /// Filename prefix for log files. Default: `"fafb-q"`.
    pub file_prefix: String,
    /// Rotation period: `"daily"` or `"hourly"`. Default: `"daily"`.
    pub rotation: String,
    /// Delete log files older than this many days. Default: 14.
    pub max_age_days: u32,
    /// Delete oldest log files when total directory size exceeds this (MB).
    /// Default: 50.
    pub max_total_size_mb: u64,
}

impl Default for LoggingConfig {
    fn default() -> Self {
        Self {
            level: "info".to_owned(),
            format: "text".to_owned(),
            stdout: true,
            file_enabled: false,
            file_dir: "./logs".to_owned(),
            file_prefix: "fafb-q".to_owned(),
            rotation: "daily".to_owned(),
            max_age_days: 14,
            max_total_size_mb: 50,
        }
    }
}

/// Deploy / graceful-shutdown configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct DeployConfig {
    /// How long to wait in drain mode before force-closing connections (seconds).
    /// During this window, clients receive GoingAway and should disconnect
    /// voluntarily. Default: 30.
    pub drain_timeout_secs: u64,
    /// How long after drain completes to wait for engine/storage shutdown (seconds).
    /// Default: 10.
    pub shutdown_timeout_secs: u64,
    /// Reason sent in GoingAway frames: "upgrade", "maintenance", or "shutdown".
    /// Default: "upgrade".
    pub going_away_reason: String,
    /// API key for admin endpoints (`/admin/*`). Must be >= 24 characters.
    /// If not set, admin endpoints return 403.
    pub admin_api_key: Option<String>,
}

impl Default for DeployConfig {
    fn default() -> Self {
        Self {
            drain_timeout_secs: 30,
            shutdown_timeout_secs: 10,
            going_away_reason: "upgrade".to_owned(),
            admin_api_key: None,
        }
    }
}

/// TLS / mTLS configuration.
///
/// Shared across TCP and HTTP listeners unless overridden per-listener.
/// When `enabled = false` (default), the server runs without encryption
/// and logs a startup warning.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TlsConfig {
    /// Master switch — when false, TLS is completely disabled.
    pub enabled: bool,
    /// Path to PEM certificate chain file.
    pub cert_file: String,
    /// Path to PEM private key file.
    pub key_file: String,
    /// Path to PEM CA bundle for mTLS client verification.
    /// Empty string means no client certificate is required.
    pub client_ca_file: String,
    /// Minimum TLS version: "1.2" or "1.3". Default: "1.2".
    pub min_version: String,
    /// Check cert files for changes this often (seconds). 0 = no auto-reload.
    pub reload_interval_secs: u64,
    /// Ciphersuite allowlist. Empty = rustls defaults.
    #[serde(default)]
    pub ciphersuite_allowlist: Vec<String>,
    /// TLS handshake timeout in seconds. Default: 10.
    pub handshake_timeout_secs: u64,
}

impl Default for TlsConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            cert_file: String::new(),
            key_file: String::new(),
            client_ca_file: String::new(),
            min_version: "1.2".to_owned(),
            reload_interval_secs: 3600,
            ciphersuite_allowlist: Vec::new(),
            handshake_timeout_secs: 10,
        }
    }
}

impl TlsConfig {
    /// Produce a resolved `TlsConfig` for a specific listener by merging
    /// per-listener overrides with the shared `[tls]` config.
    ///
    /// Non-empty per-listener values take precedence over shared values.
    /// All other fields (enabled, min_version, etc.) are inherited as-is.
    pub fn resolved_for_listener(
        &self,
        cert_override: Option<&String>,
        key_override: Option<&String>,
        ca_override: Option<&String>,
    ) -> TlsConfig {
        TlsConfig {
            cert_file: cert_override
                .filter(|s| !s.is_empty())
                .cloned()
                .unwrap_or_else(|| self.cert_file.clone()),
            key_file: key_override
                .filter(|s| !s.is_empty())
                .cloned()
                .unwrap_or_else(|| self.key_file.clone()),
            client_ca_file: ca_override
                .filter(|s| !s.is_empty())
                .cloned()
                .unwrap_or_else(|| self.client_ca_file.clone()),
            ..self.clone()
        }
    }
}

/// Hot-reload configuration (Tier 3 — cannot be hot-reloaded itself).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct HotReloadConfig {
    /// Watch the config file for changes and auto-reload. Default: false.
    pub watch_file: bool,
    /// Debounce window in seconds — changes within this window after the
    /// first detected change are coalesced into a single reload. Default: 2.
    pub debounce_secs: u64,
}

impl Default for HotReloadConfig {
    fn default() -> Self {
        Self {
            watch_file: false,
            debounce_secs: 2,
        }
    }
}

/// Cluster configuration (Tier 3 restart-only for core fields, Tier 1 for timeouts).
///
/// When `enabled = false` (default), no clustering code runs and there is zero overhead.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct ClusterConfig {
    /// Master switch — restart-only (Tier 3).
    pub enabled: bool,
    /// Unique ID for this node (must be > 0 when enabled) — Tier 3.
    pub node_id: u64,
    /// Bind address for node-to-node communication — Tier 3.
    pub bind: String,
    /// All other nodes in the cluster — Tier 3.
    pub peers: Vec<String>,

    /// Leader → follower heartbeat frequency in ms — Tier 1 (hot-reload safe).
    pub heartbeat_interval_ms: u64,
    /// Missed heartbeats before entering Leader Suspect — Tier 1.
    pub failover_timeout_ms: u64,
    /// Auto-promote timeout when no client corroboration — Tier 1.
    pub failover_no_clients_timeout_ms: u64,

    /// Stream replay buffer entry count — Tier 1.
    pub replication_buffer_max_entries: usize,
    /// Stream replay buffer byte limit — Tier 1.
    pub replication_buffer_max_bytes: usize,

    /// NTP server for time sync (empty = disabled) — Tier 1.
    pub time_server: String,
    /// NTP re-sync frequency in seconds — Tier 1.
    pub time_sync_interval_secs: u64,
    /// Skew warning threshold in ms — Tier 1.
    pub clock_skew_warn_threshold_ms: u64,

    /// Shared secret for node-to-node auth (≥32 chars when enabled) — Tier 3.
    pub cluster_secret: String,
}

impl Default for ClusterConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            node_id: 0,
            bind: "0.0.0.0:9877".to_owned(),
            peers: Vec::new(),
            heartbeat_interval_ms: 1000,
            failover_timeout_ms: 5000,
            failover_no_clients_timeout_ms: 30_000,
            replication_buffer_max_entries: 100_000,
            replication_buffer_max_bytes: 256 * 1024 * 1024, // 256 MB
            time_server: "pool.ntp.org".to_owned(),
            time_sync_interval_secs: 300,
            clock_skew_warn_threshold_ms: 500,
            cluster_secret: String::new(),
        }
    }
}

/// Top-level server configuration.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServerConfig {
    #[serde(default)]
    pub tcp: TcpConfig,
    #[serde(default)]
    pub http: HttpConfig,
    #[serde(default)]
    pub storage: StorageConfig,
    #[serde(default)]
    pub queue: QueueConfig,
    #[serde(default)]
    pub metrics: MetricsConfig,
    #[serde(default)]
    pub tls: TlsConfig,
    #[serde(default)]
    pub logging: LoggingConfig,
    #[serde(default)]
    pub deploy: DeployConfig,
    #[serde(default)]
    pub hot_reload: HotReloadConfig,
    #[serde(default)]
    pub cluster: ClusterConfig,
    #[serde(default)]
    pub namespaces: HashMap<NamespaceId, NamespaceConfig>,
}

/// TCP listener configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TcpConfig {
    pub bind_address: String,
    pub port: u16,
    /// Maximum frame size in bytes for the TCP protocol codec.
    /// Frames exceeding this size are rejected. Default: 1MB.
    pub max_message_size: u32,
    /// Seconds of inactivity before the server closes a connection.
    /// Clients must send periodic Ping frames to prevent timeout.
    /// 0 means no timeout. Default: 30.
    pub idle_timeout_secs: u64,
    /// Maximum number of concurrent TCP connections. 0 means unlimited.
    /// Default: 0.
    pub max_connections: usize,
    /// Per-subscriber delivery channel buffer size. Default: 1024.
    #[serde(default = "default_subscriber_channel_buffer")]
    pub subscriber_channel_buffer: usize,
    /// Policy for handling slow consumers. Default: drop_for_slow.
    #[serde(default = "default_slow_consumer_policy")]
    pub slow_consumer_policy: SlowConsumerPolicy,
    /// Consecutive send failures before marking a subscriber as slow. Default: 100.
    #[serde(default = "default_slow_consumer_threshold")]
    pub slow_consumer_threshold: u64,
    /// Consecutive drops before disconnecting a slow subscriber
    /// (only applies when policy is `disconnect_slow`). Default: 1000.
    #[serde(default = "default_slow_consumer_disconnect_after")]
    pub slow_consumer_disconnect_after: u64,
    /// Per-listener TLS cert override (overrides `[tls].cert_file` for TCP only).
    #[serde(default)]
    pub tls_cert_file: Option<String>,
    /// Per-listener TLS key override (overrides `[tls].key_file` for TCP only).
    #[serde(default)]
    pub tls_key_file: Option<String>,
    /// Per-listener mTLS CA override (overrides `[tls].client_ca_file` for TCP only).
    #[serde(default)]
    pub tls_client_ca_file: Option<String>,
}

impl Default for TcpConfig {
    fn default() -> Self {
        Self {
            bind_address: "127.0.0.1".to_owned(),
            port: 9876,
            max_message_size: 1_048_576,
            idle_timeout_secs: 30,
            max_connections: 0,
            subscriber_channel_buffer: default_subscriber_channel_buffer(),
            slow_consumer_policy: default_slow_consumer_policy(),
            slow_consumer_threshold: default_slow_consumer_threshold(),
            slow_consumer_disconnect_after: default_slow_consumer_disconnect_after(),
            tls_cert_file: None,
            tls_key_file: None,
            tls_client_ca_file: None,
        }
    }
}

/// HTTP server configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct HttpConfig {
    pub bind_address: String,
    pub port: u16,
    /// Maximum response body size for HTTP fetch endpoint in bytes.
    /// Larger than the TCP frame limit to serve as a fallback for oversized messages.
    /// Default: 4MB (4_194_304).
    #[serde(default = "default_max_fetch_response_bytes")]
    pub max_fetch_response_bytes: usize,
    /// Per-listener TLS cert override (overrides `[tls].cert_file` for HTTP only).
    #[serde(default)]
    pub tls_cert_file: Option<String>,
    /// Per-listener TLS key override (overrides `[tls].key_file` for HTTP only).
    #[serde(default)]
    pub tls_key_file: Option<String>,
    /// Per-listener mTLS CA override (overrides `[tls].client_ca_file` for HTTP only).
    #[serde(default)]
    pub tls_client_ca_file: Option<String>,
}

fn default_max_fetch_response_bytes() -> usize {
    4_194_304 // 4MB
}

impl Default for HttpConfig {
    fn default() -> Self {
        Self {
            bind_address: "127.0.0.1".to_owned(),
            port: 8080,
            max_fetch_response_bytes: default_max_fetch_response_bytes(),
            tls_cert_file: None,
            tls_key_file: None,
            tls_client_ca_file: None,
        }
    }
}

/// Storage layer configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct StorageConfig {
    pub data_dir: String,
    /// Fsync strategy for segment files.
    #[serde(default = "default_fsync_strategy")]
    pub fsync_strategy: FsyncStrategy,
    /// Interval in milliseconds between background flush cycles.
    #[serde(default = "default_flush_interval")]
    pub batch_flush_interval_ms: u64,
    /// Trigger an early flush when a topic accumulates this many unflushed bytes.
    /// 0 means size-based flushing is disabled (only interval-based). Default: 4MB.
    #[serde(default = "default_flush_size")]
    pub batch_flush_size_bytes: u64,
    /// Maximum open file descriptors for segment files. When the limit is
    /// reached the least-recently-used topic's file handle is evicted and
    /// closed automatically.
    #[serde(default = "default_max_open_fds")]
    pub max_open_fds: usize,
    /// Maximum number of segment files retained per topic.
    /// Oldest segments are deleted during rotation when the limit is exceeded.
    #[serde(default = "default_max_segments")]
    pub max_segments_per_topic: usize,
    /// Interval in seconds between retention enforcement sweeps. Default: 60.
    #[serde(default = "default_retention_interval")]
    pub retention_check_interval_secs: u64,
    /// Records between sparse index entries in `.idx` files. Default: 256.
    /// Lower values = faster seek but larger index files.
    #[serde(default = "default_index_interval")]
    pub index_interval: u32,
    /// Merge consecutive segments smaller than this byte count.
    /// Default: 16 MiB. Set to 0 to disable merge compaction.
    #[serde(default = "default_merge_threshold")]
    pub segment_merge_threshold_bytes: u64,
    /// Seconds between compaction sweeps. Default: 300 (5 min).
    #[serde(default = "default_compaction_interval")]
    pub compaction_interval_secs: u64,
}

impl Default for StorageConfig {
    fn default() -> Self {
        Self {
            data_dir: "./data".to_owned(),
            fsync_strategy: default_fsync_strategy(),
            batch_flush_interval_ms: default_flush_interval(),
            batch_flush_size_bytes: default_flush_size(),
            max_open_fds: default_max_open_fds(),
            max_segments_per_topic: default_max_segments(),
            retention_check_interval_secs: default_retention_interval(),
            index_interval: default_index_interval(),
            segment_merge_threshold_bytes: default_merge_threshold(),
            compaction_interval_secs: default_compaction_interval(),
        }
    }
}

/// Queue engine configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct QueueConfig {
    /// Maximum number of messages buffered in-memory per topic.
    pub max_depth: usize,
    /// Policy when a topic's buffer is full.
    pub backpressure_policy: BackpressurePolicy,
    /// Maximum redelivery attempts before routing to the dead-letter queue.
    pub max_redelivery_attempts: u32,
    /// Suffix appended to a topic name to form its dead-letter topic.
    /// Empty string disables DLQ routing.
    pub dlq_topic_suffix: String,
    /// Default maximum bytes per topic (0 = unlimited). Default: 256MB.
    pub default_topic_max_bytes: u64,
    /// Global maximum memory budget across all topics (0 = unlimited). Default: 4GB.
    pub global_max_memory_bytes: u64,
    /// Maximum number of consumer groups per topic. Default: 1000.
    pub max_consumer_groups_per_topic: usize,
    /// When true, publishes are accepted as long as disk has space.
    /// Memory becomes a cache; DropOldest is forced for all topics.
    pub disk_backed: bool,
    /// Reject publishes when disk usage exceeds this percentage (0–100).
    /// Only applies when `disk_backed = true`. Default: 90.
    pub disk_capacity_threshold_pct: f64,
    /// Maximum number of headers per message. Default: 64.
    pub max_headers_per_message: usize,
    /// Maximum size of a single header key in bytes. Default: 256.
    pub max_header_key_bytes: usize,
    /// Maximum size of a single header value in bytes. Default: 8192.
    pub max_header_value_bytes: usize,
    /// Maximum total header bytes (all keys + values) per message. Default: 32768.
    pub max_header_bytes_total: usize,
    /// Base inflight timeout in seconds before a delivered message is considered
    /// timed out and eligible for redelivery. Default: 300 (5 minutes).
    pub inflight_timeout_secs: u64,
    /// Maximum backoff cap in seconds for redelivery exponential backoff.
    /// Default: 3600 (1 hour).
    pub max_redelivery_backoff_secs: u64,
}

impl Default for QueueConfig {
    fn default() -> Self {
        Self {
            max_depth: 100_000,
            backpressure_policy: BackpressurePolicy::DropOldest,
            max_redelivery_attempts: 5,
            dlq_topic_suffix: "__dlq".to_owned(),
            default_topic_max_bytes: 268_435_456,
            global_max_memory_bytes: 4_294_967_296,
            max_consumer_groups_per_topic: 1_000,
            disk_backed: false,
            disk_capacity_threshold_pct: 90.0,
            max_headers_per_message: 64,
            max_header_key_bytes: 256,
            max_header_value_bytes: 8_192,
            max_header_bytes_total: 32_768,
            inflight_timeout_secs: 300,
            max_redelivery_backoff_secs: 3600,
        }
    }
}

/// Retention policy for segment cleanup. All fields are optional —
/// None means no limit for that dimension.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct RetentionPolicy {
    /// Delete segments whose newest message is older than this many seconds.
    pub max_age_secs: Option<u64>,
    /// Delete oldest segments when total topic size exceeds this byte count.
    pub max_bytes: Option<u64>,
    /// Keep at most this many segments per topic. Oldest deleted first.
    pub max_segments: Option<usize>,
    /// Reap entire topic after this many seconds of inactivity. None = no reaping.
    pub topic_inactive_secs: Option<u64>,
    /// Grace period before hard delete. None = 86400 (24 hours).
    pub topic_reap_grace_period_secs: Option<u64>,
    /// Per-topic overrides. First matching pattern wins. Empty = use namespace default.
    #[serde(default)]
    pub topic_inactive_overrides: Vec<TopicInactiveOverride>,
}

impl RetentionPolicy {
    /// Returns true if any retention limit is configured.
    pub fn is_active(&self) -> bool {
        self.max_age_secs.is_some()
            || self.max_bytes.is_some()
            || self.max_segments.is_some()
            || self.topic_inactive_secs.is_some()
    }
}

/// Per-topic override for inactivity GC thresholds. Evaluated in order; first match wins.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct TopicInactiveOverride {
    /// NATS-style wildcard pattern (e.g., "system.>", "tmp.*").
    pub pattern: String,
    /// Inactivity threshold in seconds. 0 = never reap matching topics.
    pub inactive_secs: u64,
}

/// Pre-compiled version of TopicInactiveOverride for use at sweep time.
#[derive(Debug, Clone)]
pub struct CompiledInactiveOverride {
    pub pattern: crate::wildcard::TopicPattern,
    pub inactive_secs: u64,
}

/// Per-namespace resource quota configuration.
/// Used at runtime by the namespace manager to enforce limits.
#[derive(Debug, Clone, Default, PartialEq, Serialize)]
pub struct NamespaceQuota {
    pub max_total_messages: Option<u64>,
    pub max_topics: Option<usize>,
    pub topic_max_depth: Option<usize>,
    pub topic_max_bytes: Option<u64>,
    pub backpressure_policy: Option<BackpressurePolicy>,
    pub max_connections: Option<usize>,
    pub max_disk_bytes: Option<u64>,
    /// Default TTL in seconds applied to messages without explicit TTL. 0 = no default.
    pub default_ttl_secs: u64,
    /// Maximum TTL in seconds a publisher can set. 0 = unlimited.
    pub max_ttl_secs: u64,
    /// Minimum TTL in seconds a publisher can set. 0 = no minimum.
    pub min_ttl_secs: u64,
}

/// Per-namespace configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NamespaceConfig {
    pub api_key: ApiKey,
    pub permissions: Vec<Permission>,
    #[serde(default = "default_max_redelivery")]
    pub max_redelivery_attempts: u32,
    #[serde(default = "default_dlq_suffix")]
    pub dlq_topic_suffix: String,
    #[serde(default)]
    pub max_total_messages: Option<u64>,
    #[serde(default)]
    pub max_topics: Option<usize>,
    #[serde(default)]
    pub topic_max_depth: Option<usize>,
    #[serde(default)]
    pub topic_max_bytes: Option<u64>,
    #[serde(default)]
    pub backpressure_policy: Option<BackpressurePolicy>,
    /// Maximum concurrent connections to this namespace. None means unlimited.
    #[serde(default)]
    pub max_connections: Option<usize>,
    /// Maximum total bytes on disk for this namespace's segments.
    /// None means unlimited (subject to global disk threshold only).
    #[serde(default)]
    pub max_disk_bytes: Option<u64>,
    /// Retention policy for this namespace's topics.
    /// Defaults to no retention (keep forever).
    #[serde(default)]
    pub retention: RetentionPolicy,
    /// Default TTL (in seconds) applied to messages without an explicit TTL.
    /// 0 = no default TTL (messages never expire unless publisher sets one).
    #[serde(default)]
    pub default_ttl_secs: u64,
    /// Maximum TTL (in seconds) a publisher can set. 0 = unlimited.
    /// Messages with TTL exceeding this are clamped to this value.
    #[serde(default)]
    pub max_ttl_secs: u64,
    /// Minimum TTL (in seconds) a publisher can set. 0 = no minimum.
    /// Messages with a non-zero TTL below this are rejected.
    #[serde(default)]
    pub min_ttl_secs: u64,
}

fn default_max_redelivery() -> u32 {
    5
}

fn default_dlq_suffix() -> String {
    "__dlq".to_owned()
}

/// Configuration validation errors.
#[derive(Debug, PartialEq, Eq)]
pub enum ConfigError {
    PortOutOfRange { field: String, port: u16 },
    PortConflict { port: u16, bind_address: String },
    NoNamespaces,
    InvalidNamespaceName(String),
    ApiKeyTooShort { namespace: String, len: usize },
    EmptyPermissions(String),
    DuplicateApiKey(String),
    MaxMessageSizeTooSmall { size: u32 },
    IdleTimeoutTooSmall { secs: u64 },
    InvalidEnvVar { var: String, reason: String },
    InvalidLogging { reason: String },
    InvalidDeployConfig { reason: String },
    InvalidTlsConfig { reason: String },
    InvalidTtlConfig { namespace: String, reason: String },
    InvalidClusterConfig { reason: String },
}

impl fmt::Display for ConfigError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::PortOutOfRange { field, port } => {
                write!(f, "{field} port {port} is out of range (1024..=65535)")
            }
            Self::PortConflict { port, bind_address } => {
                write!(
                    f,
                    "TCP and HTTP share the same bind address {bind_address} and port {port}"
                )
            }
            Self::NoNamespaces => write!(f, "at least one namespace must be configured"),
            Self::InvalidNamespaceName(name) => {
                write!(
                    f,
                    "invalid namespace name: {name:?} (must match [a-zA-Z0-9_-]+)"
                )
            }
            Self::ApiKeyTooShort { namespace, len } => {
                write!(
                    f,
                    "namespace {namespace:?} API key is too short ({len} chars, minimum 24)"
                )
            }
            Self::EmptyPermissions(namespace) => {
                write!(f, "namespace {namespace:?} has no permissions")
            }
            Self::DuplicateApiKey(key) => {
                write!(f, "duplicate API key across namespaces: {key:?}")
            }
            Self::MaxMessageSizeTooSmall { size } => {
                write!(
                    f,
                    "tcp.max_message_size is too small ({size} bytes, minimum 1024)"
                )
            }
            Self::IdleTimeoutTooSmall { secs } => {
                write!(
                    f,
                    "tcp.idle_timeout_secs is too small ({secs}s, minimum 10 when non-zero)"
                )
            }
            Self::InvalidEnvVar { var, reason } => {
                write!(f, "invalid env var {var}: {reason}")
            }
            Self::InvalidLogging { reason } => {
                write!(f, "invalid logging config: {reason}")
            }
            Self::InvalidDeployConfig { reason } => {
                write!(f, "invalid deploy config: {reason}")
            }
            Self::InvalidTlsConfig { reason } => {
                write!(f, "invalid TLS config: {reason}")
            }
            Self::InvalidTtlConfig {
                namespace, reason, ..
            } => {
                write!(f, "namespace {namespace:?} invalid TTL config: {reason}")
            }
            Self::InvalidClusterConfig { reason } => {
                write!(f, "invalid cluster config: {reason}")
            }
        }
    }
}

impl ServerConfig {
    /// Validate the configuration, collecting all errors (no short-circuiting).
    pub fn validate(&self) -> Vec<ConfigError> {
        let mut errors = Vec::new();

        // Port range checks
        if !(1024..=65535).contains(&self.tcp.port) {
            errors.push(ConfigError::PortOutOfRange {
                field: "tcp".to_owned(),
                port: self.tcp.port,
            });
        }
        if !(1024..=65535).contains(&self.http.port) {
            errors.push(ConfigError::PortOutOfRange {
                field: "http".to_owned(),
                port: self.http.port,
            });
        }

        // Max message size minimum
        if self.tcp.max_message_size < 1024 {
            errors.push(ConfigError::MaxMessageSizeTooSmall {
                size: self.tcp.max_message_size,
            });
        }

        // Idle timeout minimum (when non-zero)
        if self.tcp.idle_timeout_secs > 0 && self.tcp.idle_timeout_secs < 10 {
            errors.push(ConfigError::IdleTimeoutTooSmall {
                secs: self.tcp.idle_timeout_secs,
            });
        }

        // Port conflict between TCP and HTTP
        if self.tcp.port == self.http.port && self.tcp.bind_address == self.http.bind_address {
            errors.push(ConfigError::PortConflict {
                port: self.tcp.port,
                bind_address: self.tcp.bind_address.clone(),
            });
        }

        // At least one namespace
        if self.namespaces.is_empty() {
            errors.push(ConfigError::NoNamespaces);
        }

        // Per-namespace checks
        for (ns_id, ns_config) in &self.namespaces {
            let name = &ns_id.0;

            // Name validation: [a-zA-Z0-9_-]+
            if name.is_empty()
                || !name
                    .chars()
                    .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
            {
                errors.push(ConfigError::InvalidNamespaceName(name.clone()));
            }

            // API key minimum length
            if ns_config.api_key.0.len() < 24 {
                errors.push(ConfigError::ApiKeyTooShort {
                    namespace: name.clone(),
                    len: ns_config.api_key.0.len(),
                });
            }

            // Non-empty permissions
            if ns_config.permissions.is_empty() {
                errors.push(ConfigError::EmptyPermissions(name.clone()));
            }

            // TTL validation
            if ns_config.default_ttl_secs > 0
                && ns_config.max_ttl_secs > 0
                && ns_config.default_ttl_secs > ns_config.max_ttl_secs
            {
                errors.push(ConfigError::InvalidTtlConfig {
                    namespace: name.clone(),
                    reason: format!(
                        "default_ttl_secs ({}) exceeds max_ttl_secs ({})",
                        ns_config.default_ttl_secs, ns_config.max_ttl_secs
                    ),
                });
            }
            if ns_config.min_ttl_secs > 0
                && ns_config.max_ttl_secs > 0
                && ns_config.min_ttl_secs > ns_config.max_ttl_secs
            {
                errors.push(ConfigError::InvalidTtlConfig {
                    namespace: name.clone(),
                    reason: format!(
                        "min_ttl_secs ({}) exceeds max_ttl_secs ({})",
                        ns_config.min_ttl_secs, ns_config.max_ttl_secs
                    ),
                });
            }
            // u32 millis caps at ~49.7 days = 4_294_967 seconds
            const MAX_TTL_SECS: u64 = 4_294_967;
            if ns_config.default_ttl_secs > MAX_TTL_SECS {
                tracing::warn!(
                    "namespace {:?} default_ttl_secs ({}) exceeds u32 millis range (~49 days), will be clamped",
                    name,
                    ns_config.default_ttl_secs
                );
            }
            if ns_config.max_ttl_secs > MAX_TTL_SECS {
                tracing::warn!(
                    "namespace {:?} max_ttl_secs ({}) exceeds u32 millis range (~49 days), will be clamped",
                    name,
                    ns_config.max_ttl_secs
                );
            }
        }

        // Duplicate API keys across namespaces
        let mut seen_keys: HashMap<&str, &str> = HashMap::new();
        for (ns_id, ns_config) in &self.namespaces {
            let key = ns_config.api_key.0.as_str();
            if seen_keys.contains_key(key) {
                errors.push(ConfigError::DuplicateApiKey(key.to_owned()));
            } else {
                seen_keys.insert(key, &ns_id.0);
            }
        }

        // Deploy config validation
        if self.deploy.shutdown_timeout_secs < 1 {
            errors.push(ConfigError::InvalidDeployConfig {
                reason: "shutdown_timeout_secs must be >= 1".to_owned(),
            });
        }
        if !matches!(
            self.deploy.going_away_reason.as_str(),
            "upgrade" | "maintenance" | "shutdown"
        ) {
            errors.push(ConfigError::InvalidDeployConfig {
                reason: format!(
                    "going_away_reason must be \"upgrade\", \"maintenance\", or \"shutdown\", got {:?}",
                    self.deploy.going_away_reason
                ),
            });
        }
        if let Some(ref key) = self.deploy.admin_api_key
            && key.len() < 24
        {
            errors.push(ConfigError::InvalidDeployConfig {
                reason: "admin_api_key must be >= 24 characters when set".to_owned(),
            });
        }

        // TLS config validation
        if self.tls.enabled {
            if self.tls.cert_file.is_empty() {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: "cert_file is required when TLS is enabled".to_owned(),
                });
            } else if !Path::new(&self.tls.cert_file).exists() {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("cert_file not found: {:?}", self.tls.cert_file),
                });
            }
            if self.tls.key_file.is_empty() {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: "key_file is required when TLS is enabled".to_owned(),
                });
            } else if !Path::new(&self.tls.key_file).exists() {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("key_file not found: {:?}", self.tls.key_file),
                });
            }
            if !self.tls.client_ca_file.is_empty() && !Path::new(&self.tls.client_ca_file).exists()
            {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("client_ca_file not found: {:?}", self.tls.client_ca_file),
                });
            }
        }
        if !matches!(self.tls.min_version.as_str(), "1.2" | "1.3") {
            errors.push(ConfigError::InvalidTlsConfig {
                reason: format!(
                    "min_version must be \"1.2\" or \"1.3\", got {:?}",
                    self.tls.min_version
                ),
            });
        }

        // Per-listener TLS override validation
        fn validate_listener_tls_override(
            errors: &mut Vec<ConfigError>,
            listener: &str,
            cert: &Option<String>,
            key: &Option<String>,
            ca: &Option<String>,
        ) {
            if let Some(path) = cert
                && !path.is_empty()
                && !Path::new(path).exists()
            {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("{listener}.tls_cert_file not found: {path:?}"),
                });
            }
            if let Some(path) = key
                && !path.is_empty()
                && !Path::new(path).exists()
            {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("{listener}.tls_key_file not found: {path:?}"),
                });
            }
            if let Some(path) = ca
                && !path.is_empty()
                && !Path::new(path).exists()
            {
                errors.push(ConfigError::InvalidTlsConfig {
                    reason: format!("{listener}.tls_client_ca_file not found: {path:?}"),
                });
            }
        }
        validate_listener_tls_override(
            &mut errors,
            "tcp",
            &self.tcp.tls_cert_file,
            &self.tcp.tls_key_file,
            &self.tcp.tls_client_ca_file,
        );
        validate_listener_tls_override(
            &mut errors,
            "http",
            &self.http.tls_cert_file,
            &self.http.tls_key_file,
            &self.http.tls_client_ca_file,
        );

        // Cluster config validation
        if self.cluster.enabled {
            if self.cluster.node_id == 0 {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "cluster.node_id must be > 0".to_owned(),
                });
            }
            if self.cluster.peers.is_empty() {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "cluster.peers must contain at least one peer".to_owned(),
                });
            }
            if self.cluster.cluster_secret.len() < 32 {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "cluster.cluster_secret must be at least 32 characters".to_owned(),
                });
            }
            // This node's bind address should not appear in its own peer list.
            // Note: this is a string comparison — it won't catch equivalent addresses
            // like "0.0.0.0:9877" vs "192.168.1.1:9877". The peer handshake in
            // Phase 24.2 validates node identity at connection time as a second line
            // of defense.
            if self.cluster.peers.contains(&self.cluster.bind) {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "cluster.peers must not include this node's own bind address"
                        .to_owned(),
                });
            }
            if self.cluster.failover_timeout_ms < self.cluster.heartbeat_interval_ms * 3 {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "failover_timeout_ms must be at least 3x heartbeat_interval_ms"
                        .to_owned(),
                });
            }
            if self.cluster.failover_no_clients_timeout_ms < self.cluster.failover_timeout_ms {
                errors.push(ConfigError::InvalidClusterConfig {
                    reason: "failover_no_clients_timeout_ms must be >= failover_timeout_ms"
                        .to_owned(),
                });
            }

            #[cfg(not(feature = "cluster"))]
            errors.push(ConfigError::InvalidClusterConfig {
                reason: "cluster.enabled requires --features cluster".to_owned(),
            });
        }

        // Logging config validation
        if !matches!(self.logging.format.as_str(), "text" | "json") {
            errors.push(ConfigError::InvalidLogging {
                reason: format!(
                    "format must be \"text\" or \"json\", got {:?}",
                    self.logging.format
                ),
            });
        }
        if !matches!(self.logging.rotation.as_str(), "daily" | "hourly") {
            errors.push(ConfigError::InvalidLogging {
                reason: format!(
                    "rotation must be \"daily\" or \"hourly\", got {:?}",
                    self.logging.rotation
                ),
            });
        }
        if self.logging.file_enabled && self.logging.max_age_days == 0 {
            errors.push(ConfigError::InvalidLogging {
                reason: "max_age_days must be > 0 when file logging is enabled".to_owned(),
            });
        }

        errors
    }

    /// Apply environment variable overrides. Returns errors for invalid values.
    /// The `env` closure is injectable for testability.
    pub fn apply_env_overrides(
        &mut self,
        env: impl Fn(&str) -> Option<String>,
    ) -> Vec<ConfigError> {
        let mut errors = Vec::new();

        if let Some(val) = env("FAFB_TCP_ADDRESS") {
            self.tcp.bind_address = val;
        }
        if let Some(val) = env("FAFB_TCP_PORT") {
            match val.parse::<u16>() {
                Ok(port) => self.tcp.port = port,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_TCP_PORT".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_MAX_MESSAGE_SIZE") {
            match val.parse::<u32>() {
                Ok(size) => self.tcp.max_message_size = size,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_MAX_MESSAGE_SIZE".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_TCP_IDLE_TIMEOUT") {
            match val.parse::<u64>() {
                Ok(secs) => self.tcp.idle_timeout_secs = secs,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_TCP_IDLE_TIMEOUT".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_TCP_MAX_CONNECTIONS") {
            match val.parse::<usize>() {
                Ok(n) => self.tcp.max_connections = n,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_TCP_MAX_CONNECTIONS".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_HTTP_ADDRESS") {
            self.http.bind_address = val;
        }
        if let Some(val) = env("FAFB_HTTP_PORT") {
            match val.parse::<u16>() {
                Ok(port) => self.http.port = port,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_HTTP_PORT".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_DATA_DIR") {
            self.storage.data_dir = val;
        }
        if let Some(val) = env("FAFB_ADMIN_API_KEY") {
            self.deploy.admin_api_key = if val.is_empty() { None } else { Some(val) };
        }
        if let Some(val) = env("FAFB_TLS_CERT_FILE") {
            self.tls.cert_file = val;
        }
        if let Some(val) = env("FAFB_TLS_KEY_FILE") {
            self.tls.key_file = val;
        }
        if let Some(val) = env("FAFB_TLS_CLIENT_CA_FILE") {
            self.tls.client_ca_file = val;
        }

        // -- Queue engine env vars --
        if let Some(val) = env("FAFB_GLOBAL_MAX_MEMORY_BYTES") {
            match val.parse::<u64>() {
                Ok(v) => self.queue.global_max_memory_bytes = v,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_GLOBAL_MAX_MEMORY_BYTES".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_MAX_DEPTH") {
            match val.parse::<usize>() {
                Ok(v) => self.queue.max_depth = v,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_MAX_DEPTH".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_DISK_BACKED") {
            match parse_bool(&val) {
                Some(v) => self.queue.disk_backed = v,
                None => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_DISK_BACKED".to_owned(),
                    reason: format!("expected true/false/1/0, got '{val}'"),
                }),
            }
        }
        if let Some(val) = env("FAFB_DISK_CAPACITY_THRESHOLD_PCT") {
            match val.parse::<f64>() {
                Ok(v) => self.queue.disk_capacity_threshold_pct = v,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_DISK_CAPACITY_THRESHOLD_PCT".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }
        if let Some(val) = env("FAFB_BACKPRESSURE_POLICY") {
            match parse_backpressure_policy(&val) {
                Some(v) => self.queue.backpressure_policy = v,
                None => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_BACKPRESSURE_POLICY".to_owned(),
                    reason: format!("expected 'Reject', 'DropOldest', or 'Block', got '{val}'"),
                }),
            }
        }

        // -- Logging env vars --
        if let Some(val) = env("FAFB_LOG_LEVEL") {
            self.logging.level = val;
        }
        if let Some(val) = env("FAFB_LOG_FORMAT") {
            match val.as_str() {
                "text" | "json" => self.logging.format = val,
                _ => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_LOG_FORMAT".to_owned(),
                    reason: format!("expected 'text' or 'json', got '{val}'"),
                }),
            }
        }

        // -- Metrics env vars --
        if let Some(val) = env("FAFB_METRICS_ENABLED") {
            match parse_bool(&val) {
                Some(v) => self.metrics.enabled = v,
                None => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_METRICS_ENABLED".to_owned(),
                    reason: format!("expected true/false/1/0, got '{val}'"),
                }),
            }
        }

        // -- Storage env vars --
        if let Some(val) = env("FAFB_MAX_OPEN_FDS") {
            match val.parse::<usize>() {
                Ok(v) => self.storage.max_open_fds = v,
                Err(e) => errors.push(ConfigError::InvalidEnvVar {
                    var: "FAFB_MAX_OPEN_FDS".to_owned(),
                    reason: e.to_string(),
                }),
            }
        }

        errors
    }

    /// Load configuration from a TOML file, apply env overrides, and validate.
    pub fn load(path: &Path) -> crate::error::Result<Self> {
        let contents = std::fs::read_to_string(path)
            .map_err(|e| Error::Config(format!("failed to read config file: {e}")))?;

        let mut config: ServerConfig = toml::from_str(&contents)?;

        Self::apply_env_and_validate(&mut config)?;
        Ok(config)
    }

    /// Create a configuration from compiled defaults + env overrides, and validate.
    /// Used when no config file is found.
    pub fn load_defaults() -> crate::error::Result<Self> {
        let mut config = ServerConfig::default();

        Self::apply_env_and_validate(&mut config)?;
        Ok(config)
    }

    fn apply_env_and_validate(config: &mut Self) -> crate::error::Result<()> {
        let mut all_errors: Vec<ConfigError> = Vec::new();

        let env_errors = config.apply_env_overrides(|key| std::env::var(key).ok());
        all_errors.extend(env_errors);

        let validation_errors = config.validate();
        all_errors.extend(validation_errors);

        if !all_errors.is_empty() {
            let msg = all_errors
                .iter()
                .map(|e| e.to_string())
                .collect::<Vec<_>>()
                .join("; ");
            return Err(Error::Config(msg));
        }

        Ok(())
    }

    /// Apply CLI flag overrides. Called after env overrides so CLI wins.
    pub fn apply_cli_overrides(&mut self, args: &ServeArgs) {
        if let Some(ref bind) = args.tcp_bind {
            if let Some((addr, port_str)) = bind.rsplit_once(':') {
                self.tcp.bind_address = addr.to_owned();
                if let Ok(port) = port_str.parse::<u16>() {
                    self.tcp.port = port;
                }
            } else {
                self.tcp.bind_address = bind.clone();
            }
        }
        if let Some(ref bind) = args.http_bind {
            if let Some((addr, port_str)) = bind.rsplit_once(':') {
                self.http.bind_address = addr.to_owned();
                if let Ok(port) = port_str.parse::<u16>() {
                    self.http.port = port;
                }
            } else {
                self.http.bind_address = bind.clone();
            }
        }
        if let Some(ref dir) = args.data_dir {
            self.storage.data_dir = dir.clone();
        }
        if let Some(ref level) = args.log_level {
            self.logging.level = level.clone();
        }
    }

    /// Serialize to pretty TOML with API keys redacted.
    pub fn dump_redacted(&self) -> String {
        let mut config = self.clone();
        for ns in config.namespaces.values_mut() {
            ns.api_key = ApiKey(redact_secret(&ns.api_key.0));
        }
        if let Some(ref key) = config.deploy.admin_api_key {
            config.deploy.admin_api_key = Some(redact_secret(key));
        }
        toml::to_string_pretty(&config).unwrap_or_else(|e| format!("# serialization error: {e}"))
    }
}

/// Redact a secret, keeping only the last 4 chars visible.
fn redact_secret(s: &str) -> String {
    if s.len() <= 4 {
        "****".to_owned()
    } else {
        format!("****{}", &s[s.len() - 4..])
    }
}

// ---------------------------------------------------------------------------
// Runtime config — the subset of ServerConfig that can change at runtime
// via hot-reload. Stored in an ArcSwap for lock-free reads.
// ---------------------------------------------------------------------------

/// Auth-relevant fields extracted from NamespaceConfig.
/// Used to detect auth changes on config reload (Tier 2).
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
pub struct NamespaceAuthConfig {
    pub api_key: ApiKey,
    pub permissions: Vec<Permission>,
}

/// Configuration fields that can be changed at runtime via hot-reload.
/// Stored in an ArcSwap for lock-free reads. Fields that require a restart
/// (bind addresses, data directory, fsync strategy, disk_backed mode) are
/// intentionally excluded — see Phase 17 Tier classification.
#[derive(Debug, Clone, Serialize)]
pub struct RuntimeConfig {
    // --- Queue engine (Tier 1) ---
    pub max_depth: usize,
    pub default_topic_max_bytes: u64,
    pub backpressure_policy: BackpressurePolicy,
    pub global_max_memory_bytes: u64,
    pub max_consumer_groups_per_topic: usize,
    pub disk_capacity_threshold_pct: f64,
    pub max_redelivery_attempts: u32,
    pub dlq_topic_suffix: String,
    pub max_headers_per_message: usize,
    pub max_header_key_bytes: usize,
    pub max_header_value_bytes: usize,
    pub max_header_bytes_total: usize,
    pub inflight_timeout_secs: u64,
    pub max_redelivery_backoff_secs: u64,

    // --- Manager delivery (Tier 1) ---
    pub cold_batch_size: usize,

    // --- TCP / connection behavior (Tier 1) ---
    pub idle_timeout_secs: u64,
    pub max_connections: usize,
    pub subscriber_channel_buffer: usize,
    pub slow_consumer_policy: SlowConsumerPolicy,
    pub slow_consumer_threshold: u64,
    pub slow_consumer_disconnect_after: u64,
    pub max_message_size: u32,

    // --- HTTP (Tier 1) ---
    pub max_fetch_response_bytes: usize,

    // --- Storage behavior (Tier 1) ---
    pub max_segments_per_topic: usize,
    pub retention_check_interval_secs: u64,
    pub segment_merge_threshold_bytes: u64,
    pub compaction_interval_secs: u64,
    pub batch_flush_interval_ms: u64,
    pub batch_flush_size_bytes: u64,
    pub max_open_fds: usize,

    // --- Deploy (Tier 1) ---
    pub drain_timeout_secs: u64,
    pub shutdown_timeout_secs: u64,
    pub going_away_reason: String,

    // --- Per-namespace (Tier 1 + Tier 2) ---
    pub namespace_quotas: HashMap<NamespaceId, NamespaceQuota>,
    pub namespace_retention: HashMap<NamespaceId, RetentionPolicy>,
    /// Pre-compiled `topic_inactive_overrides` patterns. Only populated for namespaces
    /// that have at least one override. Runtime-only; not serialized.
    #[serde(skip)]
    pub compiled_inactive_overrides: HashMap<NamespaceId, Vec<CompiledInactiveOverride>>,

    // --- Auth (Tier 2 — requires auth store rebuild) ---
    pub namespace_auth: HashMap<NamespaceId, NamespaceAuthConfig>,
}

impl RuntimeConfig {
    /// Build a `RuntimeConfig` from the full `ServerConfig`, extracting only
    /// the Tier 1 and Tier 2 fields that can change at runtime.
    pub fn from_server_config(config: &ServerConfig) -> Self {
        let mut namespace_quotas = HashMap::new();
        let mut namespace_retention = HashMap::new();
        let mut namespace_auth = HashMap::new();
        let mut compiled_inactive_overrides = HashMap::new();

        for (ns_id, ns_cfg) in &config.namespaces {
            namespace_quotas.insert(
                ns_id.clone(),
                NamespaceQuota {
                    max_total_messages: ns_cfg.max_total_messages,
                    max_topics: ns_cfg.max_topics,
                    topic_max_depth: ns_cfg.topic_max_depth,
                    topic_max_bytes: ns_cfg.topic_max_bytes,
                    backpressure_policy: ns_cfg.backpressure_policy,
                    max_connections: ns_cfg.max_connections,
                    max_disk_bytes: ns_cfg.max_disk_bytes,
                    default_ttl_secs: ns_cfg.default_ttl_secs,
                    max_ttl_secs: ns_cfg.max_ttl_secs,
                    min_ttl_secs: ns_cfg.min_ttl_secs,
                },
            );
            namespace_retention.insert(ns_id.clone(), ns_cfg.retention.clone());
            namespace_auth.insert(
                ns_id.clone(),
                NamespaceAuthConfig {
                    api_key: ns_cfg.api_key.clone(),
                    permissions: ns_cfg.permissions.clone(),
                },
            );

            // Compile topic_inactive_overrides patterns for this namespace.
            let mut compiled = Vec::new();
            for ov in &ns_cfg.retention.topic_inactive_overrides {
                let pattern =
                    crate::wildcard::TopicPattern::parse(&ov.pattern).unwrap_or_else(|e| {
                        panic!(
                            "invalid topic_inactive_overrides pattern '{}' in namespace {}: {e}",
                            ov.pattern, ns_id.0
                        )
                    });
                compiled.push(CompiledInactiveOverride {
                    pattern,
                    inactive_secs: ov.inactive_secs,
                });
            }
            if !compiled.is_empty() {
                compiled_inactive_overrides.insert(ns_id.clone(), compiled);
            }
        }

        RuntimeConfig {
            // Queue engine
            max_depth: config.queue.max_depth,
            default_topic_max_bytes: config.queue.default_topic_max_bytes,
            backpressure_policy: config.queue.backpressure_policy,
            global_max_memory_bytes: config.queue.global_max_memory_bytes,
            max_consumer_groups_per_topic: config.queue.max_consumer_groups_per_topic,
            disk_capacity_threshold_pct: config.queue.disk_capacity_threshold_pct,
            max_redelivery_attempts: config.queue.max_redelivery_attempts,
            dlq_topic_suffix: config.queue.dlq_topic_suffix.clone(),
            max_headers_per_message: config.queue.max_headers_per_message,
            max_header_key_bytes: config.queue.max_header_key_bytes,
            max_header_value_bytes: config.queue.max_header_value_bytes,
            max_header_bytes_total: config.queue.max_header_bytes_total,
            inflight_timeout_secs: config.queue.inflight_timeout_secs,
            max_redelivery_backoff_secs: config.queue.max_redelivery_backoff_secs,

            // Manager delivery — cold_batch_size lives on ManagerConfig, not in TOML.
            // Use the compiled default since it's not exposed in the config file.
            cold_batch_size: 256,

            // TCP
            idle_timeout_secs: config.tcp.idle_timeout_secs,
            max_connections: config.tcp.max_connections,
            subscriber_channel_buffer: config.tcp.subscriber_channel_buffer,
            slow_consumer_policy: config.tcp.slow_consumer_policy,
            slow_consumer_threshold: config.tcp.slow_consumer_threshold,
            slow_consumer_disconnect_after: config.tcp.slow_consumer_disconnect_after,
            max_message_size: config.tcp.max_message_size,

            // HTTP
            max_fetch_response_bytes: config.http.max_fetch_response_bytes,

            // Storage
            max_segments_per_topic: config.storage.max_segments_per_topic,
            retention_check_interval_secs: config.storage.retention_check_interval_secs,
            segment_merge_threshold_bytes: config.storage.segment_merge_threshold_bytes,
            compaction_interval_secs: config.storage.compaction_interval_secs,
            batch_flush_interval_ms: config.storage.batch_flush_interval_ms,
            batch_flush_size_bytes: config.storage.batch_flush_size_bytes,
            max_open_fds: config.storage.max_open_fds,

            // Deploy
            drain_timeout_secs: config.deploy.drain_timeout_secs,
            shutdown_timeout_secs: config.deploy.shutdown_timeout_secs,
            going_away_reason: config.deploy.going_away_reason.clone(),

            // Per-namespace
            namespace_quotas,
            namespace_retention,
            compiled_inactive_overrides,
            namespace_auth,
        }
    }

    /// Serialize the runtime config to a JSON `serde_json::Value` with API keys
    /// redacted to `****<last4>`. Safe for admin introspection endpoints.
    ///
    /// Builds a redacted copy of `namespace_auth` at the struct level before
    /// serializing, so redaction is structural (not path-based JSON walking).
    pub fn to_json_redacted(&self) -> serde_json::Value {
        let redacted = RedactedRuntimeConfig::from(self);
        serde_json::to_value(&redacted).unwrap_or(serde_json::json!({}))
    }
}

/// A view of `RuntimeConfig` with API keys redacted. Used exclusively for
/// serialization to JSON in admin introspection endpoints.
#[derive(Serialize)]
struct RedactedRuntimeConfig<'a> {
    // Flat borrow all non-auth fields from RuntimeConfig
    max_depth: usize,
    default_topic_max_bytes: u64,
    backpressure_policy: &'a BackpressurePolicy,
    global_max_memory_bytes: u64,
    max_consumer_groups_per_topic: usize,
    disk_capacity_threshold_pct: f64,
    max_redelivery_attempts: u32,
    dlq_topic_suffix: &'a str,
    max_headers_per_message: usize,
    max_header_key_bytes: usize,
    max_header_value_bytes: usize,
    max_header_bytes_total: usize,
    inflight_timeout_secs: u64,
    max_redelivery_backoff_secs: u64,
    cold_batch_size: usize,
    idle_timeout_secs: u64,
    max_connections: usize,
    subscriber_channel_buffer: usize,
    slow_consumer_policy: &'a SlowConsumerPolicy,
    slow_consumer_threshold: u64,
    slow_consumer_disconnect_after: u64,
    max_message_size: u32,
    max_fetch_response_bytes: usize,
    max_segments_per_topic: usize,
    retention_check_interval_secs: u64,
    segment_merge_threshold_bytes: u64,
    compaction_interval_secs: u64,
    batch_flush_interval_ms: u64,
    batch_flush_size_bytes: u64,
    max_open_fds: usize,
    drain_timeout_secs: u64,
    shutdown_timeout_secs: u64,
    going_away_reason: &'a str,
    namespace_quotas: &'a HashMap<NamespaceId, NamespaceQuota>,
    namespace_retention: &'a HashMap<NamespaceId, RetentionPolicy>,
    namespace_auth: HashMap<&'a NamespaceId, RedactedAuthConfig>,
}

#[derive(Serialize)]
struct RedactedAuthConfig {
    api_key: String,
    permissions: Vec<Permission>,
}

impl<'a> From<&'a RuntimeConfig> for RedactedRuntimeConfig<'a> {
    fn from(cfg: &'a RuntimeConfig) -> Self {
        let namespace_auth = cfg
            .namespace_auth
            .iter()
            .map(|(ns, auth)| {
                (
                    ns,
                    RedactedAuthConfig {
                        api_key: redact_secret(&auth.api_key.0),
                        permissions: auth.permissions.clone(),
                    },
                )
            })
            .collect();

        RedactedRuntimeConfig {
            max_depth: cfg.max_depth,
            default_topic_max_bytes: cfg.default_topic_max_bytes,
            backpressure_policy: &cfg.backpressure_policy,
            global_max_memory_bytes: cfg.global_max_memory_bytes,
            max_consumer_groups_per_topic: cfg.max_consumer_groups_per_topic,
            disk_capacity_threshold_pct: cfg.disk_capacity_threshold_pct,
            max_redelivery_attempts: cfg.max_redelivery_attempts,
            dlq_topic_suffix: &cfg.dlq_topic_suffix,
            max_headers_per_message: cfg.max_headers_per_message,
            max_header_key_bytes: cfg.max_header_key_bytes,
            max_header_value_bytes: cfg.max_header_value_bytes,
            max_header_bytes_total: cfg.max_header_bytes_total,
            inflight_timeout_secs: cfg.inflight_timeout_secs,
            max_redelivery_backoff_secs: cfg.max_redelivery_backoff_secs,
            cold_batch_size: cfg.cold_batch_size,
            idle_timeout_secs: cfg.idle_timeout_secs,
            max_connections: cfg.max_connections,
            subscriber_channel_buffer: cfg.subscriber_channel_buffer,
            slow_consumer_policy: &cfg.slow_consumer_policy,
            slow_consumer_threshold: cfg.slow_consumer_threshold,
            slow_consumer_disconnect_after: cfg.slow_consumer_disconnect_after,
            max_message_size: cfg.max_message_size,
            max_fetch_response_bytes: cfg.max_fetch_response_bytes,
            max_segments_per_topic: cfg.max_segments_per_topic,
            retention_check_interval_secs: cfg.retention_check_interval_secs,
            segment_merge_threshold_bytes: cfg.segment_merge_threshold_bytes,
            compaction_interval_secs: cfg.compaction_interval_secs,
            batch_flush_interval_ms: cfg.batch_flush_interval_ms,
            batch_flush_size_bytes: cfg.batch_flush_size_bytes,
            max_open_fds: cfg.max_open_fds,
            drain_timeout_secs: cfg.drain_timeout_secs,
            shutdown_timeout_secs: cfg.shutdown_timeout_secs,
            going_away_reason: &cfg.going_away_reason,
            namespace_quotas: &cfg.namespace_quotas,
            namespace_retention: &cfg.namespace_retention,
            namespace_auth,
        }
    }
}

/// Shared, atomically-swappable config handle. Clone is cheap (Arc clone).
/// `ArcSwap::load()` is ~2ns — safe to call on every publish/subscribe/deliver.
pub type SharedConfig = Arc<ArcSwap<RuntimeConfig>>;

/// Create a new `SharedConfig` from a `ServerConfig`.
pub fn shared_config(config: &ServerConfig) -> SharedConfig {
    Arc::new(ArcSwap::from_pointee(RuntimeConfig::from_server_config(
        config,
    )))
}

// ---------------------------------------------------------------------------
// Config reload — diffing, notifications, and the reload function
// ---------------------------------------------------------------------------

/// A single field-level change detected during config reload.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ConfigChange {
    /// Dotted field path (e.g. "queue.max_depth", "namespaces.my_ns.api_key").
    pub field: String,
    /// Human-readable description of the change.
    pub description: String,
    /// Whether this change requires active notification to a subsystem
    /// (Tier 2 change — e.g. auth store rebuild, namespace creation).
    pub requires_notification: bool,
}

/// Summary of all changes detected between two `RuntimeConfig` snapshots.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ConfigDiff {
    /// Tier 1 + Tier 2 changes that were applied.
    pub applied: Vec<ConfigChange>,
    /// Tier 3 changes that were detected but NOT applied (require restart).
    pub ignored: Vec<ConfigChange>,
    /// Namespaces present in new config but not in old.
    pub namespaces_added: Vec<NamespaceId>,
    /// Namespaces present in old config but not in new.
    pub namespaces_removed: Vec<NamespaceId>,
    /// True if any namespace's API key or permissions changed (Tier 2).
    /// Computed during `diff()` — avoids fragile string matching on field names.
    pub auth_changed: bool,
}

impl ConfigDiff {
    /// Returns true if no changes were detected at all.
    pub fn is_empty(&self) -> bool {
        self.applied.is_empty()
            && self.ignored.is_empty()
            && self.namespaces_added.is_empty()
            && self.namespaces_removed.is_empty()
            && !self.auth_changed
    }

    /// Returns true if any Tier 2 change requires notification.
    pub fn requires_notification(&self) -> bool {
        self.auth_changed
            || !self.namespaces_added.is_empty()
            || !self.namespaces_removed.is_empty()
    }
}

/// Notification sent to subsystems when Tier 2 config changes are applied.
/// Subscribers (auth store, namespace manager) react to create/remove state.
#[derive(Debug, Clone)]
pub struct ReloadNotification {
    /// True if any namespace's API key or permissions changed.
    pub auth_changed: bool,
    /// Namespaces that were added in the new config.
    pub namespaces_added: Vec<NamespaceId>,
    /// Namespaces that were removed in the new config.
    pub namespaces_removed: Vec<NamespaceId>,
}

/// Errors that can occur during config reload.
#[derive(Debug)]
pub enum ReloadError {
    /// No config file path available (server started with compiled defaults).
    NoConfigFile,
    /// Failed to read the config file from disk.
    ReadError(std::io::Error),
    /// Failed to parse the TOML config file.
    ParseError(toml::de::Error),
    /// Config validation failed (one or more errors).
    ValidationError(Vec<ConfigError>),
    /// Namespace removal blocked — active connections exist.
    NamespaceRemovalBlocked {
        namespace: NamespaceId,
        active_connections: u64,
    },
    /// Reload rejected because the server is draining or shutting down.
    ServerDraining,
}

impl fmt::Display for ReloadError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ReloadError::NoConfigFile => {
                write!(f, "no config file path available (started with defaults)")
            }
            ReloadError::ReadError(e) => write!(f, "failed to read config file: {e}"),
            ReloadError::ParseError(e) => write!(f, "failed to parse config: {e}"),
            ReloadError::ValidationError(errors) => {
                write!(f, "config validation failed: ")?;
                for (i, e) in errors.iter().enumerate() {
                    if i > 0 {
                        write!(f, "; ")?;
                    }
                    write!(f, "{e}")?;
                }
                Ok(())
            }
            ReloadError::NamespaceRemovalBlocked {
                namespace,
                active_connections,
            } => write!(
                f,
                "cannot remove namespace '{namespace}': {active_connections} active connections"
            ),
            ReloadError::ServerDraining => {
                write!(f, "config reload rejected: server is draining")
            }
        }
    }
}

/// Shared state for serializing config reload operations.
///
/// Both the HTTP endpoint and the future file watcher acquire the mutex
/// before proceeding, preventing concurrent reloads from racing.
/// Callback type for checking active connections per namespace during reload.
/// Returns 0 when no connections exist or namespace is unknown.
pub type NamespaceConnectionChecker = Arc<
    dyn Fn(&NamespaceId) -> std::pin::Pin<Box<dyn std::future::Future<Output = u64> + Send>>
        + Send
        + Sync,
>;

pub struct ReloadState {
    /// Mutex to serialize concurrent reload attempts.
    lock: tokio::sync::Mutex<()>,
    /// The config file path to re-read on reload.
    pub config_path: Arc<PathBuf>,
    /// The shared runtime config (ArcSwap).
    pub shared_config: SharedConfig,
    /// The full ServerConfig — kept for Tier 3 diffing.
    pub current_server_config: tokio::sync::RwLock<ServerConfig>,
    /// Watch channel for notifying subsystems of Tier 2 changes.
    pub reload_notify: tokio::sync::watch::Sender<Option<ReloadNotification>>,
    /// CLI args preserved from startup — re-applied on every reload so CLI
    /// overrides are not regressed back to file values.
    pub cli_args: ServeArgs,
    /// Checker for namespace active connections. Set once after manager is
    /// created; read during reload to block namespace removal with active
    /// connections.
    pub ns_checker: std::sync::OnceLock<NamespaceConnectionChecker>,
    /// Server lifecycle — reload is rejected when draining/shutting down.
    pub lifecycle: Option<Arc<crate::server::state::ServerLifecycle>>,
}

impl ReloadState {
    pub fn new(
        config_path: Arc<PathBuf>,
        shared_config: SharedConfig,
        server_config: ServerConfig,
        reload_notify: tokio::sync::watch::Sender<Option<ReloadNotification>>,
        cli_args: ServeArgs,
    ) -> Self {
        Self {
            lock: tokio::sync::Mutex::new(()),
            config_path,
            shared_config,
            current_server_config: tokio::sync::RwLock::new(server_config),
            reload_notify,
            cli_args,
            ns_checker: std::sync::OnceLock::new(),
            lifecycle: None,
        }
    }

    /// Set the server lifecycle for drain-aware reload rejection.
    pub fn with_lifecycle(mut self, lifecycle: Arc<crate::server::state::ServerLifecycle>) -> Self {
        self.lifecycle = Some(lifecycle);
        self
    }

    /// Perform a config reload: read file, parse, validate, diff, apply.
    ///
    /// Acquires the internal mutex to prevent concurrent reloads. Returns the
    /// diff on success, or a `ReloadError` on failure.
    ///
    /// Precedence on reload matches startup: CLI flags > env vars > file > defaults.
    pub async fn reload(&self) -> std::result::Result<ConfigDiff, ReloadError> {
        // Reject reload when server is draining or shutting down — applying
        // config changes during drain can disrupt the orderly shutdown sequence.
        if self.lifecycle.as_ref().is_some_and(|l| l.is_draining()) {
            return Err(ReloadError::ServerDraining);
        }

        let _guard = self.lock.lock().await;

        // 1. Read + parse + validate (async file read to avoid blocking the runtime)
        let toml_str = tokio::fs::read_to_string(self.config_path.as_ref())
            .await
            .map_err(ReloadError::ReadError)?;
        let mut new_config: ServerConfig =
            toml::from_str(&toml_str).map_err(ReloadError::ParseError)?;
        new_config.apply_env_overrides(|k| std::env::var(k).ok());
        new_config.apply_cli_overrides(&self.cli_args);
        let errors = new_config.validate();
        if !errors.is_empty() {
            return Err(ReloadError::ValidationError(errors));
        }

        // 2. Detect Tier 3 changes (logged as warnings, not applied)
        let current = self.current_server_config.read().await;
        let tier3_changes = detect_restart_only_changes(&current, &new_config);
        for change in &tier3_changes {
            tracing::warn!(
                field = %change.field,
                "config change requires restart (ignored): {}",
                change.description
            );
        }

        // 3. Build new RuntimeConfig and diff against current
        let new_runtime = RuntimeConfig::from_server_config(&new_config);
        let old_runtime = self.shared_config.load();
        let mut diff = new_runtime.diff(&old_runtime);
        diff.ignored = tier3_changes;

        // 4. Check namespace removal safety — reject if active connections exist.
        if !diff.namespaces_removed.is_empty() {
            if let Some(check_fn) = self.ns_checker.get() {
                for ns in &diff.namespaces_removed {
                    let active = check_fn(ns).await;
                    if active > 0 {
                        return Err(ReloadError::NamespaceRemovalBlocked {
                            namespace: ns.clone(),
                            active_connections: active,
                        });
                    }
                }
            }
            tracing::warn!(
                namespaces = ?diff.namespaces_removed,
                "namespace removal detected — no active connections, proceeding"
            );
        }

        // 5. If nothing changed, return early
        if diff.is_empty() {
            return Ok(diff);
        }

        // 6. Apply via ArcSwap::store()
        self.shared_config.store(Arc::new(new_runtime));

        // 7. Send Tier 2 notifications if needed
        if diff.requires_notification() {
            let notification = ReloadNotification {
                auth_changed: diff.auth_changed,
                namespaces_added: diff.namespaces_added.clone(),
                namespaces_removed: diff.namespaces_removed.clone(),
            };
            // Ignore send error — no receivers means no one is listening yet
            let _ = self.reload_notify.send(Some(notification));
        }

        // 8. Update stored ServerConfig for future diffs
        drop(current); // release read lock
        *self.current_server_config.write().await = new_config;

        tracing::info!(
            applied = diff.applied.len(),
            ignored = diff.ignored.len(),
            namespaces_added = diff.namespaces_added.len(),
            namespaces_removed = diff.namespaces_removed.len(),
            "config reloaded"
        );

        Ok(diff)
    }
}

/// Spawn a file watcher that monitors the config file for changes and triggers
/// a reload with debounce. Watches the parent directory (not just the file) to
/// handle Kubernetes ConfigMap symlink swaps and editor temp-file-then-rename
/// patterns (vim, emacs).
///
/// The watcher coalesces rapid changes (e.g., editor writes) into a single
/// reload using a debounce window.
///
/// Returns a `JoinHandle` for crash-isolated monitoring with backoff respawn.
pub fn spawn_config_watcher(
    reload_state: Arc<ReloadState>,
    debounce_secs: u64,
    lifecycle: Arc<crate::server::state::ServerLifecycle>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        use notify::{RecommendedWatcher, RecursiveMode, Watcher};

        let config_path = reload_state.config_path.as_ref().clone();
        let watch_dir = match config_path.parent() {
            Some(dir) => dir.to_path_buf(),
            None => {
                tracing::error!(
                    path = %config_path.display(),
                    "cannot determine parent directory for config file watcher"
                );
                return;
            }
        };
        let config_filename = config_path.file_name().map(|f| f.to_os_string());

        let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(16);

        // Build the synchronous notify watcher with a sender that signals the async task.
        let watcher_result = {
            let config_filename = config_filename.clone();
            let watch_dir_clone = watch_dir.clone();
            RecommendedWatcher::new(
                move |event: notify::Result<notify::Event>| {
                    if let Ok(ev) = event {
                        // Filter: only react to events touching the config file itself.
                        // This covers direct edits, rename-into (editors), and symlink changes.
                        let relevant = ev.paths.is_empty()
                            || ev.paths.iter().any(|p| {
                                // Match the exact config file, the parent directory
                                // (symlink swap), or any event in the watched directory
                                // (kqueue on macOS may report the directory path for
                                // file content changes).
                                if let Some(ref expected) = config_filename {
                                    p.file_name()
                                        .map(|f| f == expected.as_os_str())
                                        .unwrap_or(false)
                                        || p == &watch_dir_clone
                                        || p.starts_with(&watch_dir_clone)
                                } else {
                                    true
                                }
                            });
                        if relevant {
                            let _ = tx.try_send(());
                        }
                    }
                },
                notify::Config::default(),
            )
        };

        let mut watcher = match watcher_result {
            Ok(w) => w,
            Err(e) => {
                tracing::error!(
                    error = %e,
                    dir = %watch_dir.display(),
                    "failed to create config file watcher — HTTP reload still works"
                );
                return;
            }
        };

        if let Err(e) = watcher.watch(&watch_dir, RecursiveMode::NonRecursive) {
            tracing::error!(
                error = %e,
                dir = %watch_dir.display(),
                "failed to start watching directory — HTTP reload still works"
            );
            return;
        }

        // Also watch the config file itself — on macOS kqueue, directory
        // watches don't always fire for content changes to existing files.
        if config_path.exists()
            && let Err(e) = watcher.watch(&config_path, RecursiveMode::NonRecursive)
        {
            tracing::warn!(
                error = %e,
                path = %config_path.display(),
                "could not watch config file directly (directory watch still active)"
            );
        }

        tracing::info!(
            path = %config_path.display(),
            dir = %watch_dir.display(),
            debounce_secs = debounce_secs,
            "config file watcher started"
        );

        let debounce = std::time::Duration::from_secs(debounce_secs.max(1));
        let mut shutdown_rx = lifecycle.subscribe();

        loop {
            tokio::select! {
                _ = shutdown_rx.changed() => {
                    tracing::info!("config file watcher stopping (shutdown)");
                    break;
                }
                Some(()) = rx.recv() => {
                    // Debounce: wait for the debounce window, draining any further
                    // events that arrive during that period.
                    tokio::time::sleep(debounce).await;
                    while rx.try_recv().is_ok() {}

                    // Mtime stability check: if the file was modified within the
                    // last 500ms, wait a bit longer — the file may still be
                    // mid-write (e.g., large config, slow disk, editor flushing).
                    let stability = std::time::Duration::from_millis(500);
                    for _ in 0..3 {
                        let recent = config_path.metadata().ok().and_then(|m| {
                            m.modified().ok().and_then(|t| t.elapsed().ok())
                        });
                        if let Some(age) = recent.filter(|a| a < &stability) {
                            tokio::time::sleep(stability - age).await;
                            while rx.try_recv().is_ok() {}
                            continue;
                        }
                        break;
                    }

                    tracing::info!("config file change detected, reloading");
                    let start = std::time::Instant::now();
                    match reload_state.reload().await {
                        Ok(diff) if diff.is_empty() => {
                            let elapsed = start.elapsed();
                            metrics::counter!("fafb_config_reloads_total",
                                "trigger" => "file_watcher", "result" => "unchanged")
                                .increment(1);
                            metrics::histogram!("fafb_config_reload_duration_seconds")
                                .record(elapsed.as_secs_f64());
                            tracing::debug!(
                                elapsed_ms = elapsed.as_millis() as u64,
                                "config file unchanged after reload"
                            );
                        }
                        Ok(diff) => {
                            let elapsed = start.elapsed();
                            let changes = diff.applied.len();
                            metrics::counter!("fafb_config_reloads_total",
                                "trigger" => "file_watcher", "result" => "success")
                                .increment(1);
                            metrics::histogram!("fafb_config_reload_duration_seconds")
                                .record(elapsed.as_secs_f64());
                            metrics::gauge!("fafb_config_last_reload_timestamp")
                                .set(crate::types::now_millis() as f64 / 1000.0);
                            metrics::gauge!("fafb_config_last_reload_changes")
                                .set(changes as f64);
                            tracing::info!(
                                changes = changes,
                                elapsed_ms = elapsed.as_millis() as u64,
                                "config reloaded via file watcher"
                            );
                        }
                        Err(e) => {
                            let elapsed = start.elapsed();
                            metrics::counter!("fafb_config_reloads_total",
                                "trigger" => "file_watcher", "result" => "error")
                                .increment(1);
                            metrics::histogram!("fafb_config_reload_duration_seconds")
                                .record(elapsed.as_secs_f64());
                            tracing::warn!(
                                error = %e,
                                elapsed_ms = elapsed.as_millis() as u64,
                                "config file watcher reload failed — keeping previous config"
                            );
                        }
                    }
                }
            }
        }

        // Keep watcher alive until task exits; dropping it deregisters watches.
        drop(watcher);
        tracing::info!("config file watcher stopped");
    })
}

impl RuntimeConfig {
    /// Compare `self` (new config) against `old` and return a diff of all changes.
    ///
    /// Uses explicit destructuring so the compiler will fail if a field is
    /// added to `RuntimeConfig` without updating this function.
    pub fn diff(&self, old: &RuntimeConfig) -> ConfigDiff {
        let mut applied = Vec::new();
        let mut namespaces_added = Vec::new();
        let mut namespaces_removed = Vec::new();
        let mut auth_changed = false;

        // Destructure to get compiler enforcement of exhaustiveness.
        let RuntimeConfig {
            max_depth,
            default_topic_max_bytes,
            backpressure_policy,
            global_max_memory_bytes,
            max_consumer_groups_per_topic,
            disk_capacity_threshold_pct,
            max_redelivery_attempts,
            dlq_topic_suffix,
            max_headers_per_message,
            max_header_key_bytes,
            max_header_value_bytes,
            max_header_bytes_total,
            inflight_timeout_secs,
            max_redelivery_backoff_secs,
            cold_batch_size,
            idle_timeout_secs,
            max_connections,
            subscriber_channel_buffer,
            slow_consumer_policy,
            slow_consumer_threshold,
            slow_consumer_disconnect_after,
            max_message_size,
            max_fetch_response_bytes,
            max_segments_per_topic,
            retention_check_interval_secs,
            segment_merge_threshold_bytes,
            compaction_interval_secs,
            batch_flush_interval_ms,
            batch_flush_size_bytes,
            max_open_fds,
            drain_timeout_secs,
            shutdown_timeout_secs,
            going_away_reason,
            namespace_quotas,
            namespace_retention,
            compiled_inactive_overrides: _, // derived from namespace_retention; no separate diff needed
            namespace_auth,
        } = self;

        // --- Queue engine (Tier 1) ---
        if *max_depth != old.max_depth {
            applied.push(ConfigChange {
                field: "queue.max_depth".into(),
                description: format!("{} -> {}", old.max_depth, max_depth),
                requires_notification: false,
            });
        }
        if *default_topic_max_bytes != old.default_topic_max_bytes {
            applied.push(ConfigChange {
                field: "queue.default_topic_max_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.default_topic_max_bytes, default_topic_max_bytes
                ),
                requires_notification: false,
            });
        }
        if *backpressure_policy != old.backpressure_policy {
            applied.push(ConfigChange {
                field: "queue.backpressure_policy".into(),
                description: format!("{:?} -> {:?}", old.backpressure_policy, backpressure_policy),
                requires_notification: false,
            });
        }
        if *global_max_memory_bytes != old.global_max_memory_bytes {
            applied.push(ConfigChange {
                field: "queue.global_max_memory_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.global_max_memory_bytes, global_max_memory_bytes
                ),
                requires_notification: false,
            });
        }
        if *max_consumer_groups_per_topic != old.max_consumer_groups_per_topic {
            applied.push(ConfigChange {
                field: "queue.max_consumer_groups_per_topic".into(),
                description: format!(
                    "{} -> {}",
                    old.max_consumer_groups_per_topic, max_consumer_groups_per_topic
                ),
                requires_notification: false,
            });
        }
        if (*disk_capacity_threshold_pct - old.disk_capacity_threshold_pct).abs() > f64::EPSILON {
            applied.push(ConfigChange {
                field: "queue.disk_capacity_threshold_pct".into(),
                description: format!(
                    "{} -> {}",
                    old.disk_capacity_threshold_pct, disk_capacity_threshold_pct
                ),
                requires_notification: false,
            });
        }
        if *max_redelivery_attempts != old.max_redelivery_attempts {
            applied.push(ConfigChange {
                field: "queue.max_redelivery_attempts".into(),
                description: format!(
                    "{} -> {}",
                    old.max_redelivery_attempts, max_redelivery_attempts
                ),
                requires_notification: false,
            });
        }
        if *dlq_topic_suffix != old.dlq_topic_suffix {
            applied.push(ConfigChange {
                field: "queue.dlq_topic_suffix".into(),
                description: format!("{:?} -> {:?}", old.dlq_topic_suffix, dlq_topic_suffix),
                requires_notification: false,
            });
        }
        if *max_headers_per_message != old.max_headers_per_message {
            applied.push(ConfigChange {
                field: "queue.max_headers_per_message".into(),
                description: format!(
                    "{} -> {}",
                    old.max_headers_per_message, max_headers_per_message
                ),
                requires_notification: false,
            });
        }
        if *max_header_key_bytes != old.max_header_key_bytes {
            applied.push(ConfigChange {
                field: "queue.max_header_key_bytes".into(),
                description: format!("{} -> {}", old.max_header_key_bytes, max_header_key_bytes),
                requires_notification: false,
            });
        }
        if *max_header_value_bytes != old.max_header_value_bytes {
            applied.push(ConfigChange {
                field: "queue.max_header_value_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.max_header_value_bytes, max_header_value_bytes
                ),
                requires_notification: false,
            });
        }
        if *max_header_bytes_total != old.max_header_bytes_total {
            applied.push(ConfigChange {
                field: "queue.max_header_bytes_total".into(),
                description: format!(
                    "{} -> {}",
                    old.max_header_bytes_total, max_header_bytes_total
                ),
                requires_notification: false,
            });
        }
        if *inflight_timeout_secs != old.inflight_timeout_secs {
            applied.push(ConfigChange {
                field: "queue.inflight_timeout_secs".into(),
                description: format!("{} -> {}", old.inflight_timeout_secs, inflight_timeout_secs),
                requires_notification: false,
            });
        }
        if *max_redelivery_backoff_secs != old.max_redelivery_backoff_secs {
            applied.push(ConfigChange {
                field: "queue.max_redelivery_backoff_secs".into(),
                description: format!(
                    "{} -> {}",
                    old.max_redelivery_backoff_secs, max_redelivery_backoff_secs
                ),
                requires_notification: false,
            });
        }

        // --- Manager delivery (Tier 1) ---
        if *cold_batch_size != old.cold_batch_size {
            applied.push(ConfigChange {
                field: "manager.cold_batch_size".into(),
                description: format!("{} -> {}", old.cold_batch_size, cold_batch_size),
                requires_notification: false,
            });
        }

        // --- TCP / connection behavior (Tier 1) ---
        if *idle_timeout_secs != old.idle_timeout_secs {
            applied.push(ConfigChange {
                field: "tcp.idle_timeout_secs".into(),
                description: format!("{} -> {}", old.idle_timeout_secs, idle_timeout_secs),
                requires_notification: false,
            });
        }
        if *max_connections != old.max_connections {
            applied.push(ConfigChange {
                field: "tcp.max_connections".into(),
                description: format!("{} -> {}", old.max_connections, max_connections),
                requires_notification: false,
            });
        }
        if *subscriber_channel_buffer != old.subscriber_channel_buffer {
            applied.push(ConfigChange {
                field: "tcp.subscriber_channel_buffer".into(),
                description: format!(
                    "{} -> {}",
                    old.subscriber_channel_buffer, subscriber_channel_buffer
                ),
                requires_notification: false,
            });
        }
        if *slow_consumer_policy != old.slow_consumer_policy {
            applied.push(ConfigChange {
                field: "tcp.slow_consumer_policy".into(),
                description: format!(
                    "{:?} -> {:?}",
                    old.slow_consumer_policy, slow_consumer_policy
                ),
                requires_notification: false,
            });
        }
        if *slow_consumer_threshold != old.slow_consumer_threshold {
            applied.push(ConfigChange {
                field: "tcp.slow_consumer_threshold".into(),
                description: format!(
                    "{} -> {}",
                    old.slow_consumer_threshold, slow_consumer_threshold
                ),
                requires_notification: false,
            });
        }
        if *slow_consumer_disconnect_after != old.slow_consumer_disconnect_after {
            applied.push(ConfigChange {
                field: "tcp.slow_consumer_disconnect_after".into(),
                description: format!(
                    "{} -> {}",
                    old.slow_consumer_disconnect_after, slow_consumer_disconnect_after
                ),
                requires_notification: false,
            });
        }
        if *max_message_size != old.max_message_size {
            applied.push(ConfigChange {
                field: "tcp.max_message_size".into(),
                description: format!("{} -> {}", old.max_message_size, max_message_size),
                requires_notification: false,
            });
        }

        // --- HTTP (Tier 1) ---
        if *max_fetch_response_bytes != old.max_fetch_response_bytes {
            applied.push(ConfigChange {
                field: "http.max_fetch_response_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.max_fetch_response_bytes, max_fetch_response_bytes
                ),
                requires_notification: false,
            });
        }

        // --- Storage behavior (Tier 1) ---
        if *max_segments_per_topic != old.max_segments_per_topic {
            applied.push(ConfigChange {
                field: "storage.max_segments_per_topic".into(),
                description: format!(
                    "{} -> {}",
                    old.max_segments_per_topic, max_segments_per_topic
                ),
                requires_notification: false,
            });
        }
        if *retention_check_interval_secs != old.retention_check_interval_secs {
            applied.push(ConfigChange {
                field: "storage.retention_check_interval_secs".into(),
                description: format!(
                    "{} -> {}",
                    old.retention_check_interval_secs, retention_check_interval_secs
                ),
                requires_notification: false,
            });
        }
        if *segment_merge_threshold_bytes != old.segment_merge_threshold_bytes {
            applied.push(ConfigChange {
                field: "storage.segment_merge_threshold_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.segment_merge_threshold_bytes, segment_merge_threshold_bytes
                ),
                requires_notification: false,
            });
        }
        if *compaction_interval_secs != old.compaction_interval_secs {
            applied.push(ConfigChange {
                field: "storage.compaction_interval_secs".into(),
                description: format!(
                    "{} -> {}",
                    old.compaction_interval_secs, compaction_interval_secs
                ),
                requires_notification: false,
            });
        }
        if *batch_flush_interval_ms != old.batch_flush_interval_ms {
            applied.push(ConfigChange {
                field: "storage.batch_flush_interval_ms".into(),
                description: format!(
                    "{} -> {}",
                    old.batch_flush_interval_ms, batch_flush_interval_ms
                ),
                requires_notification: false,
            });
        }
        if *batch_flush_size_bytes != old.batch_flush_size_bytes {
            applied.push(ConfigChange {
                field: "storage.batch_flush_size_bytes".into(),
                description: format!(
                    "{} -> {}",
                    old.batch_flush_size_bytes, batch_flush_size_bytes
                ),
                requires_notification: false,
            });
        }
        if *max_open_fds != old.max_open_fds {
            applied.push(ConfigChange {
                field: "storage.max_open_fds".into(),
                description: format!("{} -> {}", old.max_open_fds, max_open_fds),
                requires_notification: false,
            });
        }

        // --- Deploy (Tier 1) ---
        if *drain_timeout_secs != old.drain_timeout_secs {
            applied.push(ConfigChange {
                field: "deploy.drain_timeout_secs".into(),
                description: format!("{} -> {}", old.drain_timeout_secs, drain_timeout_secs),
                requires_notification: false,
            });
        }
        if *shutdown_timeout_secs != old.shutdown_timeout_secs {
            applied.push(ConfigChange {
                field: "deploy.shutdown_timeout_secs".into(),
                description: format!("{} -> {}", old.shutdown_timeout_secs, shutdown_timeout_secs),
                requires_notification: false,
            });
        }
        if *going_away_reason != old.going_away_reason {
            applied.push(ConfigChange {
                field: "deploy.going_away_reason".into(),
                description: format!("{:?} -> {:?}", old.going_away_reason, going_away_reason),
                requires_notification: false,
            });
        }

        // --- Per-namespace quotas (Tier 1) ---
        for (ns, quota) in namespace_quotas {
            match old.namespace_quotas.get(ns) {
                Some(old_quota) if old_quota != quota => {
                    applied.push(ConfigChange {
                        field: format!("namespaces.{ns}.quota"),
                        description: "quota changed".into(),
                        requires_notification: false,
                    });
                }
                None => {} // handled by namespaces_added below
                _ => {}
            }
        }

        // --- Per-namespace retention (Tier 1) ---
        for (ns, ret) in namespace_retention {
            match old.namespace_retention.get(ns) {
                Some(old_ret) if old_ret != ret => {
                    applied.push(ConfigChange {
                        field: format!("namespaces.{ns}.retention"),
                        description: "retention policy changed".into(),
                        requires_notification: false,
                    });
                }
                None => {} // handled by namespaces_added below
                _ => {}
            }
        }

        // --- Per-namespace auth (Tier 2) ---
        for (ns, auth) in namespace_auth {
            match old.namespace_auth.get(ns) {
                Some(old_auth) if old_auth != auth => {
                    auth_changed = true;
                    applied.push(ConfigChange {
                        field: format!("namespaces.{ns}.auth"),
                        description: "API key or permissions changed".into(),
                        requires_notification: true,
                    });
                }
                None => {} // handled by namespaces_added below
                _ => {}
            }
        }

        // --- Namespace additions / removals ---
        for ns in namespace_auth.keys() {
            if !old.namespace_auth.contains_key(ns) {
                namespaces_added.push(ns.clone());
            }
        }
        for ns in old.namespace_auth.keys() {
            if !namespace_auth.contains_key(ns) {
                namespaces_removed.push(ns.clone());
            }
        }

        ConfigDiff {
            applied,
            ignored: Vec::new(), // Tier 3 changes filled in by caller
            namespaces_added,
            namespaces_removed,
            auth_changed,
        }
    }
}

/// Detect Tier 3 (restart-only) field changes between two `ServerConfig` snapshots.
///
/// Returns a list of changes with descriptions. These are logged as warnings
/// and included in the reload response as "ignored" — they are NOT applied.
pub fn detect_restart_only_changes(old: &ServerConfig, new: &ServerConfig) -> Vec<ConfigChange> {
    let mut changes = Vec::new();

    if old.tcp.bind_address != new.tcp.bind_address {
        changes.push(ConfigChange {
            field: "tcp.bind_address".into(),
            description: format!(
                "{:?} -> {:?} (requires restart)",
                old.tcp.bind_address, new.tcp.bind_address
            ),
            requires_notification: false,
        });
    }
    if old.tcp.port != new.tcp.port {
        changes.push(ConfigChange {
            field: "tcp.port".into(),
            description: format!("{} -> {} (requires restart)", old.tcp.port, new.tcp.port),
            requires_notification: false,
        });
    }
    if old.http.bind_address != new.http.bind_address {
        changes.push(ConfigChange {
            field: "http.bind_address".into(),
            description: format!(
                "{:?} -> {:?} (requires restart)",
                old.http.bind_address, new.http.bind_address
            ),
            requires_notification: false,
        });
    }
    if old.http.port != new.http.port {
        changes.push(ConfigChange {
            field: "http.port".into(),
            description: format!("{} -> {} (requires restart)", old.http.port, new.http.port),
            requires_notification: false,
        });
    }
    if old.storage.data_dir != new.storage.data_dir {
        changes.push(ConfigChange {
            field: "storage.data_dir".into(),
            description: format!(
                "{:?} -> {:?} (requires restart)",
                old.storage.data_dir, new.storage.data_dir
            ),
            requires_notification: false,
        });
    }
    if old.storage.fsync_strategy != new.storage.fsync_strategy {
        changes.push(ConfigChange {
            field: "storage.fsync_strategy".into(),
            description: format!(
                "{:?} -> {:?} (requires restart)",
                old.storage.fsync_strategy, new.storage.fsync_strategy
            ),
            requires_notification: false,
        });
    }
    if old.storage.index_interval != new.storage.index_interval {
        changes.push(ConfigChange {
            field: "storage.index_interval".into(),
            description: format!(
                "{} -> {} (requires restart)",
                old.storage.index_interval, new.storage.index_interval
            ),
            requires_notification: false,
        });
    }
    if old.queue.disk_backed != new.queue.disk_backed {
        changes.push(ConfigChange {
            field: "queue.disk_backed".into(),
            description: format!(
                "{} -> {} (requires restart)",
                old.queue.disk_backed, new.queue.disk_backed
            ),
            requires_notification: false,
        });
    }
    if old.deploy.admin_api_key != new.deploy.admin_api_key {
        changes.push(ConfigChange {
            field: "deploy.admin_api_key".into(),
            description: "admin API key changed (requires restart)".into(),
            requires_notification: false,
        });
    }
    if old.hot_reload.watch_file != new.hot_reload.watch_file {
        changes.push(ConfigChange {
            field: "hot_reload.watch_file".into(),
            description: format!(
                "{} -> {} (requires restart)",
                old.hot_reload.watch_file, new.hot_reload.watch_file
            ),
            requires_notification: false,
        });
    }
    if old.hot_reload.debounce_secs != new.hot_reload.debounce_secs {
        changes.push(ConfigChange {
            field: "hot_reload.debounce_secs".into(),
            description: format!(
                "{} -> {} (requires restart)",
                old.hot_reload.debounce_secs, new.hot_reload.debounce_secs
            ),
            requires_notification: false,
        });
    }

    changes
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use tempfile::NamedTempFile;

    fn valid_toml() -> &'static str {
        r#"
[tcp]
bind_address = "127.0.0.1"
port = 9876

[http]
bind_address = "127.0.0.1"
port = 8080

[storage]
data_dir = "./data"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe", "Publish", "Query"]
"#
    }

    fn parse(toml_str: &str) -> ServerConfig {
        toml::from_str(toml_str).expect("failed to parse TOML")
    }

    #[test]
    fn defaults_are_applied() {
        let config = parse(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.tcp.bind_address, "127.0.0.1");
        assert_eq!(config.tcp.port, 9876);
        assert_eq!(config.http.bind_address, "127.0.0.1");
        assert_eq!(config.http.port, 8080);
        assert_eq!(config.storage.data_dir, "./data");
    }

    #[test]
    fn valid_full_config() {
        let config = parse(valid_toml());
        assert!(config.validate().is_empty());
        assert!(
            config
                .namespaces
                .contains_key(&NamespaceId("default".into()))
        );
    }

    #[test]
    fn namespace_id_as_toml_key() {
        let config = parse(
            r#"
[namespaces.my-ns_01]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Publish"]
"#,
        );
        assert!(
            config
                .namespaces
                .contains_key(&NamespaceId("my-ns_01".into()))
        );
    }

    #[test]
    fn reject_port_below_1024() {
        let config = parse(
            r#"
[tcp]
port = 80

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(errors.iter().any(|e| matches!(
            e,
            ConfigError::PortOutOfRange { field, port } if field == "tcp" && *port == 80
        )));
    }

    #[test]
    fn reject_port_above_65535() {
        // u16 can't hold 70000, so this is a deserialization error, not a validation error
        let result = toml::from_str::<ServerConfig>(
            r#"
[tcp]
port = 70000

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert!(result.is_err());
    }

    #[test]
    fn reject_port_conflict() {
        let config = parse(
            r#"
[tcp]
bind_address = "127.0.0.1"
port = 8080

[http]
bind_address = "127.0.0.1"
port = 8080

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(
            errors
                .iter()
                .any(|e| matches!(e, ConfigError::PortConflict { .. }))
        );
    }

    #[test]
    fn allow_same_port_different_bind() {
        let config = parse(
            r#"
[tcp]
bind_address = "127.0.0.1"
port = 8080

[http]
bind_address = "0.0.0.0"
port = 8080

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(
            !errors
                .iter()
                .any(|e| matches!(e, ConfigError::PortConflict { .. }))
        );
    }

    #[test]
    fn reject_no_namespaces() {
        let config: ServerConfig = toml::from_str("").unwrap();
        let errors = config.validate();
        assert!(
            errors
                .iter()
                .any(|e| matches!(e, ConfigError::NoNamespaces))
        );
    }

    #[test]
    fn reject_invalid_namespace_name() {
        let mut config = parse(valid_toml());
        config.namespaces.insert(
            NamespaceId("bad name!".into()),
            NamespaceConfig {
                api_key: ApiKey("abcdefghijklmnopqrstuvwx".into()),
                permissions: vec![Permission::Subscribe],
                max_redelivery_attempts: 5,
                dlq_topic_suffix: "__dlq".to_owned(),
                max_total_messages: None,
                max_topics: None,
                topic_max_depth: None,
                topic_max_bytes: None,
                backpressure_policy: None,
                max_connections: None,
                max_disk_bytes: None,
                retention: RetentionPolicy::default(),
                default_ttl_secs: 0,
                max_ttl_secs: 0,
                min_ttl_secs: 0,
            },
        );
        let errors = config.validate();
        assert!(
            errors.iter().any(
                |e| matches!(e, ConfigError::InvalidNamespaceName(name) if name == "bad name!")
            )
        );
    }

    #[test]
    fn reject_api_key_too_short() {
        let config = parse(
            r#"
[namespaces.default]
api_key = "short"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(errors.iter().any(
            |e| matches!(e, ConfigError::ApiKeyTooShort { namespace, .. } if namespace == "default")
        ));
    }

    #[test]
    fn reject_empty_permissions() {
        let config = parse(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = []
"#,
        );
        let errors = config.validate();
        assert!(
            errors
                .iter()
                .any(|e| matches!(e, ConfigError::EmptyPermissions(ns) if ns == "default"))
        );
    }

    #[test]
    fn reject_duplicate_api_key() {
        let config = parse(
            r#"
[namespaces.ns1]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]

[namespaces.ns2]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Publish"]
"#,
        );
        let errors = config.validate();
        assert!(
            errors
                .iter()
                .any(|e| matches!(e, ConfigError::DuplicateApiKey(_)))
        );
    }

    #[test]
    fn env_override_addresses() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_TCP_ADDRESS" => Some("0.0.0.0".into()),
            "FAFB_HTTP_ADDRESS" => Some("0.0.0.0".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.tcp.bind_address, "0.0.0.0");
        assert_eq!(config.http.bind_address, "0.0.0.0");
    }

    #[test]
    fn env_override_ports() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_TCP_PORT" => Some("4000".into()),
            "FAFB_HTTP_PORT" => Some("5000".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.tcp.port, 4000);
        assert_eq!(config.http.port, 5000);
    }

    #[test]
    fn env_override_data_dir() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DATA_DIR" => Some("/tmp/fafb".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.storage.data_dir, "/tmp/fafb");
    }

    #[test]
    fn max_message_size_default() {
        let config = parse(valid_toml());
        assert_eq!(config.tcp.max_message_size, 1_048_576);
    }

    #[test]
    fn max_message_size_custom() {
        let config = parse(
            r#"
[tcp]
max_message_size = 4194304

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.tcp.max_message_size, 4_194_304);
        assert!(config.validate().is_empty());
    }

    #[test]
    fn reject_max_message_size_too_small() {
        let config = parse(
            r#"
[tcp]
max_message_size = 512

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(
            errors
                .iter()
                .any(|e| matches!(e, ConfigError::MaxMessageSizeTooSmall { size } if *size == 512))
        );
    }

    #[test]
    fn env_override_max_message_size() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_MESSAGE_SIZE" => Some("2097152".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.tcp.max_message_size, 2_097_152);
    }

    #[test]
    fn env_override_invalid_port() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_TCP_PORT" => Some("not_a_number".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(
            matches!(&errors[0], ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_TCP_PORT")
        );
    }

    #[test]
    fn load_valid_config_from_file() {
        let mut file = NamedTempFile::new().unwrap();
        write!(file, "{}", valid_toml()).unwrap();
        let config = ServerConfig::load(file.path()).unwrap();
        assert_eq!(config.tcp.port, 9876);
        assert_eq!(config.namespaces.len(), 1);
    }

    #[test]
    fn storage_config_defaults() {
        let config: StorageConfig = Default::default();
        assert_eq!(config.data_dir, "./data");
        assert_eq!(config.fsync_strategy, FsyncStrategy::EveryBatch);
        assert_eq!(config.batch_flush_interval_ms, 100);
        assert_eq!(config.max_open_fds, 1024);
        assert_eq!(config.max_segments_per_topic, 64);
    }

    #[test]
    fn fsync_strategy_parse_every_write() {
        let config = parse(
            r#"
[storage]
fsync_strategy = "every_write"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.storage.fsync_strategy, FsyncStrategy::EveryWrite);
    }

    #[test]
    fn fsync_strategy_parse_every_batch() {
        let config = parse(
            r#"
[storage]
fsync_strategy = "every_batch"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.storage.fsync_strategy, FsyncStrategy::EveryBatch);
    }

    #[test]
    fn fsync_strategy_parse_os_managed() {
        let config = parse(
            r#"
[storage]
fsync_strategy = "os_managed"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.storage.fsync_strategy, FsyncStrategy::OsManaged);
    }

    #[test]
    fn fsync_strategy_defaults_to_every_batch() {
        let config = parse(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.storage.fsync_strategy, FsyncStrategy::EveryBatch);
    }

    #[test]
    fn storage_config_custom_values() {
        let config = parse(
            r#"
[storage]
data_dir = "/custom/path"
fsync_strategy = "os_managed"
batch_flush_interval_ms = 200
max_open_fds = 512
max_segments_per_topic = 32

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.storage.data_dir, "/custom/path");
        assert_eq!(config.storage.fsync_strategy, FsyncStrategy::OsManaged);
        assert_eq!(config.storage.batch_flush_interval_ms, 200);
        assert_eq!(config.storage.max_open_fds, 512);
        assert_eq!(config.storage.max_segments_per_topic, 32);
    }

    #[test]
    fn metrics_config_defaults_to_enabled() {
        let config = parse(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert!(config.metrics.enabled);
    }

    #[test]
    fn metrics_config_parse_disabled() {
        let config = parse(
            r#"
[metrics]
enabled = false

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert!(!config.metrics.enabled);
    }

    #[test]
    fn metrics_config_parse_enabled() {
        let config = parse(
            r#"
[metrics]
enabled = true

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert!(config.metrics.enabled);
    }

    // -- LoggingConfig tests --

    #[test]
    fn logging_config_defaults() {
        let config: LoggingConfig = Default::default();
        assert_eq!(config.level, "info");
        assert_eq!(config.format, "text");
        assert!(config.stdout);
        assert!(!config.file_enabled);
        assert_eq!(config.file_dir, "./logs");
        assert_eq!(config.file_prefix, "fafb-q");
        assert_eq!(config.rotation, "daily");
        assert_eq!(config.max_age_days, 14);
        assert_eq!(config.max_total_size_mb, 50);
    }

    #[test]
    fn logging_config_absent_uses_defaults() {
        // Config with no [logging] section should use defaults
        let config = parse(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.logging.level, "info");
        assert!(config.logging.stdout);
        assert!(!config.logging.file_enabled);
    }

    #[test]
    fn logging_config_custom_values() {
        let config = parse(
            r#"
[logging]
level = "debug"
format = "json"
stdout = false
file_enabled = true
file_dir = "/var/log/fafb"
file_prefix = "myapp"
rotation = "hourly"
max_age_days = 7
max_total_size_mb = 100

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.logging.level, "debug");
        assert_eq!(config.logging.format, "json");
        assert!(!config.logging.stdout);
        assert!(config.logging.file_enabled);
        assert_eq!(config.logging.file_dir, "/var/log/fafb");
        assert_eq!(config.logging.file_prefix, "myapp");
        assert_eq!(config.logging.rotation, "hourly");
        assert_eq!(config.logging.max_age_days, 7);
        assert_eq!(config.logging.max_total_size_mb, 100);
    }

    #[test]
    fn logging_config_invalid_format_rejected() {
        let config = parse(
            r#"
[logging]
format = "xml"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(errors.iter().any(
            |e| matches!(e, ConfigError::InvalidLogging { reason } if reason.contains("format"))
        ));
    }

    #[test]
    fn logging_config_invalid_rotation_rejected() {
        let config = parse(
            r#"
[logging]
rotation = "weekly"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(errors.iter().any(
            |e| matches!(e, ConfigError::InvalidLogging { reason } if reason.contains("rotation"))
        ));
    }

    #[test]
    fn logging_config_zero_max_age_rejected_when_file_enabled() {
        let config = parse(
            r#"
[logging]
file_enabled = true
max_age_days = 0

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(errors
            .iter()
            .any(|e| matches!(e, ConfigError::InvalidLogging { reason } if reason.contains("max_age_days"))));
    }

    #[test]
    fn logging_config_valid_passes() {
        let config = parse(
            r#"
[logging]
level = "debug,fafb_q::storage=trace"
format = "json"
rotation = "hourly"
file_enabled = true
max_age_days = 7

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let errors = config.validate();
        assert!(
            !errors
                .iter()
                .any(|e| matches!(e, ConfigError::InvalidLogging { .. })),
            "expected no logging errors, got: {errors:?}"
        );
    }

    #[test]
    fn logging_config_partial_override() {
        // Only override some fields; rest should be defaults
        let config = parse(
            r#"
[logging]
level = "warn"
file_enabled = true

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert_eq!(config.logging.level, "warn");
        assert!(config.logging.file_enabled);
        // Defaults for unspecified fields
        assert_eq!(config.logging.format, "text");
        assert!(config.logging.stdout);
        assert_eq!(config.logging.rotation, "daily");
    }

    #[test]
    fn admin_api_key_none_passes_validation() {
        let mut config = parse(valid_toml());
        config.deploy.admin_api_key = None;
        let errors = config.validate();
        let deploy_errors: Vec<_> = errors
            .iter()
            .filter(|e| matches!(e, ConfigError::InvalidDeployConfig { reason } if reason.contains("admin_api_key")))
            .collect();
        assert!(deploy_errors.is_empty());
    }

    #[test]
    fn admin_api_key_too_short_fails_validation() {
        let mut config = parse(valid_toml());
        config.deploy.admin_api_key = Some("short-key".into());
        let errors = config.validate();
        assert!(errors.iter().any(|e| matches!(
            e,
            ConfigError::InvalidDeployConfig { reason } if reason.contains("admin_api_key")
        )));
    }

    #[test]
    fn admin_api_key_valid_passes_validation() {
        let mut config = parse(valid_toml());
        config.deploy.admin_api_key = Some("this-key-is-at-least-24-characters-long".into());
        let errors = config.validate();
        let deploy_errors: Vec<_> = errors
            .iter()
            .filter(|e| matches!(e, ConfigError::InvalidDeployConfig { reason } if reason.contains("admin_api_key")))
            .collect();
        assert!(deploy_errors.is_empty());
    }

    #[test]
    fn env_override_admin_api_key() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_ADMIN_API_KEY" => Some("my-admin-key-for-testing-purposes".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(
            config.deploy.admin_api_key.as_deref(),
            Some("my-admin-key-for-testing-purposes")
        );
    }

    #[test]
    fn env_override_admin_api_key_empty_clears() {
        let mut config = parse(valid_toml());
        config.deploy.admin_api_key = Some("existing-key-that-is-long-enough".into());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_ADMIN_API_KEY" => Some(String::new()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(config.deploy.admin_api_key.is_none());
    }

    // -- Unknown key rejection tests --

    #[test]
    fn reject_unknown_key_in_tcp() {
        let result = toml::from_str::<ServerConfig>(
            r#"
[tcp]
max_mesage_size = 1024

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("max_mesage_size"),
            "error should mention the unknown field, got: {err}"
        );
    }

    #[test]
    fn reject_unknown_key_in_top_level() {
        let result = toml::from_str::<ServerConfig>(
            r#"
unknown_section = "value"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        assert!(result.is_err());
    }

    #[test]
    fn reject_unknown_key_in_namespace() {
        let result = toml::from_str::<ServerConfig>(
            r#"
[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
max_total_mesages = 1000
"#,
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("max_total_mesages"),
            "error should mention the unknown field, got: {err}"
        );
    }

    #[test]
    fn reject_unknown_key_in_storage() {
        let result = toml::from_str::<ServerConfig>(
            r#"
[storage]
data_directory = "./data"

[namespaces.default]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe"]
"#,
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("data_directory"),
            "error should mention the unknown field, got: {err}"
        );
    }

    // -- Config file discovery tests --

    fn no_env(_key: &str) -> Option<String> {
        None
    }

    #[test]
    fn find_config_path_cli_takes_precedence() {
        let cli = PathBuf::from("/my/custom/config.toml");
        // CLI path is returned even if the file doesn't exist — caller handles errors.
        let result = find_config_path(Some(cli.clone()), no_env);
        assert_eq!(result, Some(cli));
    }

    #[test]
    fn find_config_path_returns_none_when_no_files() {
        // With no CLI arg, no env var, and no files on disk, should return None.
        let result = find_config_path(None, no_env);
        // We can't guarantee None in all environments (./config.toml might exist),
        // so just verify the function doesn't panic.
        let _ = result;
    }

    #[test]
    fn find_config_path_env_var_takes_precedence_over_cwd() {
        let mut file = NamedTempFile::new().unwrap();
        write!(file, "{}", valid_toml()).unwrap();
        let env_path = file.path().to_path_buf();

        let result = find_config_path(None, |key| {
            if key == "FAFB_CONFIG" {
                Some(env_path.to_string_lossy().into_owned())
            } else {
                None
            }
        });
        assert_eq!(result, Some(env_path));
    }

    #[test]
    fn find_config_path_env_var_nonexistent_warns_and_falls_through() {
        // FAFB_CONFIG points to a non-existent file — should fall through.
        let result = find_config_path(None, |key| {
            if key == "FAFB_CONFIG" {
                Some("/nonexistent/path/config.toml".into())
            } else {
                None
            }
        });
        // Falls through to CWD/system search (likely None in test env)
        let _ = result;
    }

    #[test]
    fn find_config_path_cli_overrides_env_var() {
        let cli = PathBuf::from("/cli/config.toml");
        let result = find_config_path(Some(cli.clone()), |key| {
            if key == "FAFB_CONFIG" {
                Some("/env/config.toml".into())
            } else {
                None
            }
        });
        assert_eq!(result, Some(cli));
    }

    #[test]
    fn load_defaults_fails_without_namespaces() {
        // load_defaults will fail validation (no namespaces), but should not panic.
        // This is the expected behavior — namespaces cannot be defined via env vars,
        // so the defaults path produces a clear error message.
        let result = ServerConfig::load_defaults();
        assert!(
            result.is_err(),
            "defaults with no namespaces should fail validation"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("namespace"),
            "error should mention namespaces: {err}"
        );
    }

    // -- CLI override tests --

    #[test]
    fn cli_tcp_bind_overrides_config() {
        let mut config = parse(valid_toml());
        let args = ServeArgs {
            config: None,
            dump_config: false,
            tcp_bind: Some("0.0.0.0:5555".into()),
            http_bind: None,
            data_dir: None,
            log_level: None,
        };
        config.apply_cli_overrides(&args);
        assert_eq!(config.tcp.bind_address, "0.0.0.0");
        assert_eq!(config.tcp.port, 5555);
    }

    #[test]
    fn cli_http_bind_overrides_config() {
        let mut config = parse(valid_toml());
        let args = ServeArgs {
            config: None,
            dump_config: false,
            tcp_bind: None,
            http_bind: Some("0.0.0.0:3000".into()),
            data_dir: None,
            log_level: None,
        };
        config.apply_cli_overrides(&args);
        assert_eq!(config.http.bind_address, "0.0.0.0");
        assert_eq!(config.http.port, 3000);
    }

    #[test]
    fn cli_data_dir_and_log_level_override_config() {
        let mut config = parse(valid_toml());
        let args = ServeArgs {
            config: None,
            dump_config: false,
            tcp_bind: None,
            http_bind: None,
            data_dir: Some("/tmp/fafb-data".into()),
            log_level: Some("debug".into()),
        };
        config.apply_cli_overrides(&args);
        assert_eq!(config.storage.data_dir, "/tmp/fafb-data");
        assert_eq!(config.logging.level, "debug");
    }

    #[test]
    fn cli_bind_address_only_without_port() {
        let mut config = parse(valid_toml());
        let args = ServeArgs {
            config: None,
            dump_config: false,
            tcp_bind: Some("0.0.0.0".into()),
            http_bind: None,
            data_dir: None,
            log_level: None,
        };
        config.apply_cli_overrides(&args);
        assert_eq!(config.tcp.bind_address, "0.0.0.0");
        assert_eq!(config.tcp.port, 9876); // unchanged
    }

    // -- Config dump tests --

    #[test]
    fn dump_redacted_hides_api_keys() {
        let config = parse(valid_toml());
        let dump = config.dump_redacted();
        assert!(
            !dump.contains("abcdefghijklmnopqrstuvwx"),
            "dump should not contain raw API key"
        );
        assert!(dump.contains("****"), "dump should contain redacted marker");
    }

    #[test]
    fn dump_redacted_hides_admin_api_key() {
        let mut config = parse(valid_toml());
        config.deploy.admin_api_key = Some("my-super-secret-admin-key-1234".into());
        let dump = config.dump_redacted();
        assert!(
            !dump.contains("my-super-secret-admin-key-1234"),
            "dump should not contain raw admin key"
        );
        assert!(dump.contains("****1234"), "dump should show last 4 chars");
    }

    #[test]
    fn redact_secret_keeps_last_four() {
        assert_eq!(redact_secret("abcdefghijklmnop"), "****mnop");
        assert_eq!(redact_secret("1234"), "****");
        assert_eq!(redact_secret("ab"), "****");
    }

    #[test]
    fn dump_roundtrips_as_valid_toml() {
        let config = parse(valid_toml());
        let dump = config.dump_redacted();
        // Should be valid TOML (even though API keys are redacted and won't pass validation)
        let _: ServerConfig = toml::from_str(&dump).expect("dump should be valid TOML");
    }

    // -- CI config file tests --

    #[test]
    fn example_config_parses_and_validates() {
        let path =
            std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("config/config.example.toml");
        let contents = std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("failed to read {}: {e}", path.display()));
        let config: ServerConfig =
            toml::from_str(&contents).expect("config.example.toml should parse as valid TOML");
        let errors = config.validate();
        assert!(
            errors.is_empty(),
            "config.example.toml should pass validation, got: {errors:?}"
        );
    }

    #[test]
    fn sdk_test_config_parses_and_validates() {
        let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("config/config.sdk-test.toml");
        let contents = std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("failed to read {}: {e}", path.display()));
        let config: ServerConfig =
            toml::from_str(&contents).expect("config.sdk-test.toml should parse as valid TOML");
        let errors = config.validate();
        assert!(
            errors.is_empty(),
            "config.sdk-test.toml should pass validation, got: {errors:?}"
        );
    }

    // -- parse_bool tests --

    #[test]
    fn parse_bool_accepts_true_values() {
        assert_eq!(parse_bool("true"), Some(true));
        assert_eq!(parse_bool("TRUE"), Some(true));
        assert_eq!(parse_bool("True"), Some(true));
        assert_eq!(parse_bool("1"), Some(true));
    }

    #[test]
    fn parse_bool_accepts_false_values() {
        assert_eq!(parse_bool("false"), Some(false));
        assert_eq!(parse_bool("FALSE"), Some(false));
        assert_eq!(parse_bool("False"), Some(false));
        assert_eq!(parse_bool("0"), Some(false));
    }

    #[test]
    fn parse_bool_rejects_invalid() {
        assert_eq!(parse_bool("yes"), None);
        assert_eq!(parse_bool("no"), None);
        assert_eq!(parse_bool(""), None);
        assert_eq!(parse_bool("2"), None);
    }

    // -- New env var override tests --

    #[test]
    fn env_override_global_max_memory_bytes() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_GLOBAL_MAX_MEMORY_BYTES" => Some("8589934592".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.queue.global_max_memory_bytes, 8_589_934_592);
    }

    #[test]
    fn env_override_global_max_memory_bytes_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_GLOBAL_MAX_MEMORY_BYTES" => Some("not_a_number".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_GLOBAL_MAX_MEMORY_BYTES"
        ));
    }

    #[test]
    fn env_override_max_depth() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_DEPTH" => Some("500000".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.queue.max_depth, 500_000);
    }

    #[test]
    fn env_override_max_depth_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_DEPTH" => Some("abc".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(
            matches!(&errors[0], ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_MAX_DEPTH")
        );
    }

    #[test]
    fn env_override_disk_backed() {
        let mut config = parse(valid_toml());
        assert!(!config.queue.disk_backed);
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_BACKED" => Some("true".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(config.queue.disk_backed);
    }

    #[test]
    fn env_override_disk_backed_numeric() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_BACKED" => Some("1".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(config.queue.disk_backed);

        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_BACKED" => Some("0".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(!config.queue.disk_backed);
    }

    #[test]
    fn env_override_disk_backed_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_BACKED" => Some("yes".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_DISK_BACKED"
        ));
    }

    #[test]
    fn env_override_disk_capacity_threshold_pct() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_CAPACITY_THRESHOLD_PCT" => Some("85.5".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!((config.queue.disk_capacity_threshold_pct - 85.5).abs() < f64::EPSILON);
    }

    #[test]
    fn env_override_disk_capacity_threshold_pct_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_CAPACITY_THRESHOLD_PCT" => Some("high".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_DISK_CAPACITY_THRESHOLD_PCT"
        ));
    }

    #[test]
    fn env_override_backpressure_policy() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_BACKPRESSURE_POLICY" => Some("Reject".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.queue.backpressure_policy, BackpressurePolicy::Reject);
    }

    #[test]
    fn env_override_backpressure_policy_drop_oldest() {
        let mut config = parse(valid_toml());
        config.queue.backpressure_policy = BackpressurePolicy::Reject;
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_BACKPRESSURE_POLICY" => Some("DropOldest".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(
            config.queue.backpressure_policy,
            BackpressurePolicy::DropOldest
        );
    }

    #[test]
    fn env_override_backpressure_policy_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_BACKPRESSURE_POLICY" => Some("drop_oldest".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_BACKPRESSURE_POLICY"
        ));
    }

    #[test]
    fn env_override_log_level() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_LOG_LEVEL" => Some("debug,fafb_q::storage=trace".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.logging.level, "debug,fafb_q::storage=trace");
    }

    #[test]
    fn env_override_log_format() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_LOG_FORMAT" => Some("json".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.logging.format, "json");
    }

    #[test]
    fn env_override_log_format_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_LOG_FORMAT" => Some("xml".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_LOG_FORMAT"
        ));
        // Original value preserved on error
        assert_eq!(config.logging.format, "text");
    }

    #[test]
    fn env_override_metrics_enabled() {
        let mut config = parse(valid_toml());
        assert!(config.metrics.enabled);
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_METRICS_ENABLED" => Some("false".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(!config.metrics.enabled);
    }

    #[test]
    fn env_override_metrics_enabled_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_METRICS_ENABLED" => Some("nope".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_METRICS_ENABLED"
        ));
    }

    #[test]
    fn env_override_max_open_fds() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_OPEN_FDS" => Some("2048".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.storage.max_open_fds, 2048);
    }

    #[test]
    fn env_override_max_open_fds_invalid() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_OPEN_FDS" => Some("-1".into()),
            _ => None,
        });
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            &errors[0],
            ConfigError::InvalidEnvVar { var, .. } if var == "FAFB_MAX_OPEN_FDS"
        ));
    }

    // -- Multi-env-var interaction tests --

    #[test]
    fn env_override_disk_backed_with_threshold_pct() {
        let mut config = parse(valid_toml());
        assert!(!config.queue.disk_backed);
        assert!((config.queue.disk_capacity_threshold_pct - 90.0).abs() < f64::EPSILON);

        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_DISK_BACKED" => Some("true".into()),
            "FAFB_DISK_CAPACITY_THRESHOLD_PCT" => Some("75".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert!(config.queue.disk_backed);
        assert!((config.queue.disk_capacity_threshold_pct - 75.0).abs() < f64::EPSILON);
    }

    #[test]
    fn env_override_multiple_queue_settings() {
        let mut config = parse(valid_toml());
        let errors = config.apply_env_overrides(|key| match key {
            "FAFB_MAX_DEPTH" => Some("200000".into()),
            "FAFB_GLOBAL_MAX_MEMORY_BYTES" => Some("8589934592".into()),
            "FAFB_BACKPRESSURE_POLICY" => Some("Reject".into()),
            "FAFB_LOG_LEVEL" => Some("warn".into()),
            "FAFB_METRICS_ENABLED" => Some("0".into()),
            _ => None,
        });
        assert!(errors.is_empty());
        assert_eq!(config.queue.max_depth, 200_000);
        assert_eq!(config.queue.global_max_memory_bytes, 8_589_934_592);
        assert_eq!(config.queue.backpressure_policy, BackpressurePolicy::Reject);
        assert_eq!(config.logging.level, "warn");
        assert!(!config.metrics.enabled);
    }

    // -- RuntimeConfig and SharedConfig tests (Phase 17.5) --

    #[test]
    fn runtime_config_from_server_config_maps_queue_fields() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.max_depth, config.queue.max_depth);
        assert_eq!(
            rt.default_topic_max_bytes,
            config.queue.default_topic_max_bytes
        );
        assert_eq!(rt.backpressure_policy, config.queue.backpressure_policy);
        assert_eq!(
            rt.global_max_memory_bytes,
            config.queue.global_max_memory_bytes
        );
        assert_eq!(
            rt.max_consumer_groups_per_topic,
            config.queue.max_consumer_groups_per_topic
        );
        assert!(
            (rt.disk_capacity_threshold_pct - config.queue.disk_capacity_threshold_pct).abs()
                < f64::EPSILON
        );
        assert_eq!(
            rt.max_redelivery_attempts,
            config.queue.max_redelivery_attempts
        );
        assert_eq!(rt.dlq_topic_suffix, config.queue.dlq_topic_suffix);
    }

    #[test]
    fn runtime_config_from_server_config_maps_tcp_fields() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.idle_timeout_secs, config.tcp.idle_timeout_secs);
        assert_eq!(rt.max_connections, config.tcp.max_connections);
        assert_eq!(
            rt.subscriber_channel_buffer,
            config.tcp.subscriber_channel_buffer
        );
        assert_eq!(rt.slow_consumer_policy, config.tcp.slow_consumer_policy);
        assert_eq!(
            rt.slow_consumer_threshold,
            config.tcp.slow_consumer_threshold
        );
        assert_eq!(
            rt.slow_consumer_disconnect_after,
            config.tcp.slow_consumer_disconnect_after
        );
        assert_eq!(rt.max_message_size, config.tcp.max_message_size);
    }

    #[test]
    fn runtime_config_from_server_config_maps_http_fields() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);
        assert_eq!(
            rt.max_fetch_response_bytes,
            config.http.max_fetch_response_bytes
        );
    }

    #[test]
    fn runtime_config_from_server_config_maps_storage_fields() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(
            rt.max_segments_per_topic,
            config.storage.max_segments_per_topic
        );
        assert_eq!(
            rt.retention_check_interval_secs,
            config.storage.retention_check_interval_secs
        );
        assert_eq!(
            rt.segment_merge_threshold_bytes,
            config.storage.segment_merge_threshold_bytes
        );
        assert_eq!(
            rt.compaction_interval_secs,
            config.storage.compaction_interval_secs
        );
        assert_eq!(
            rt.batch_flush_interval_ms,
            config.storage.batch_flush_interval_ms
        );
        assert_eq!(
            rt.batch_flush_size_bytes,
            config.storage.batch_flush_size_bytes
        );
        assert_eq!(rt.max_open_fds, config.storage.max_open_fds);
    }

    #[test]
    fn runtime_config_from_server_config_maps_deploy_fields() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.drain_timeout_secs, config.deploy.drain_timeout_secs);
        assert_eq!(
            rt.shutdown_timeout_secs,
            config.deploy.shutdown_timeout_secs
        );
        assert_eq!(rt.going_away_reason, config.deploy.going_away_reason);
    }

    #[test]
    fn runtime_config_from_server_config_maps_namespace_quotas() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        // valid_toml() defines at least one namespace
        assert!(!rt.namespace_quotas.is_empty());
        assert_eq!(rt.namespace_quotas.len(), config.namespaces.len());

        for (ns_id, ns_cfg) in &config.namespaces {
            let quota = rt.namespace_quotas.get(ns_id).expect("quota should exist");
            assert_eq!(quota.max_total_messages, ns_cfg.max_total_messages);
            assert_eq!(quota.max_topics, ns_cfg.max_topics);
            assert_eq!(quota.topic_max_depth, ns_cfg.topic_max_depth);
            assert_eq!(quota.topic_max_bytes, ns_cfg.topic_max_bytes);
            assert_eq!(quota.backpressure_policy, ns_cfg.backpressure_policy);
            assert_eq!(quota.max_connections, ns_cfg.max_connections);
            assert_eq!(quota.max_disk_bytes, ns_cfg.max_disk_bytes);
        }
    }

    #[test]
    fn runtime_config_from_server_config_maps_namespace_retention() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.namespace_retention.len(), config.namespaces.len());
        for (ns_id, ns_cfg) in &config.namespaces {
            let ret = rt
                .namespace_retention
                .get(ns_id)
                .expect("retention should exist");
            assert_eq!(ret.max_age_secs, ns_cfg.retention.max_age_secs);
            assert_eq!(ret.max_bytes, ns_cfg.retention.max_bytes);
            assert_eq!(ret.max_segments, ns_cfg.retention.max_segments);
        }
    }

    #[test]
    fn runtime_config_from_server_config_maps_namespace_auth() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.namespace_auth.len(), config.namespaces.len());
        for (ns_id, ns_cfg) in &config.namespaces {
            let auth = rt.namespace_auth.get(ns_id).expect("auth should exist");
            assert_eq!(auth.api_key, ns_cfg.api_key);
            assert_eq!(auth.permissions, ns_cfg.permissions);
        }
    }

    #[test]
    fn runtime_config_empty_namespaces() {
        let config = ServerConfig::default();
        let rt = RuntimeConfig::from_server_config(&config);
        assert!(rt.namespace_quotas.is_empty());
        assert!(rt.namespace_retention.is_empty());
        assert!(rt.namespace_auth.is_empty());
    }

    #[test]
    fn shared_config_load_returns_consistent_snapshot() {
        let config = parse(valid_toml());
        let sc = shared_config(&config);

        let guard = sc.load();
        assert_eq!(guard.max_depth, config.queue.max_depth);
        assert_eq!(guard.idle_timeout_secs, config.tcp.idle_timeout_secs);
    }

    #[test]
    fn shared_config_store_swaps_atomically() {
        let config = parse(valid_toml());
        let sc = shared_config(&config);

        // Read original
        let original_depth = sc.load().max_depth;

        // Build a new RuntimeConfig with different max_depth
        let mut new_rt = RuntimeConfig::from_server_config(&config);
        new_rt.max_depth = original_depth + 42;
        sc.store(Arc::new(new_rt));

        // Verify new value is visible
        assert_eq!(sc.load().max_depth, original_depth + 42);
    }

    #[test]
    fn shared_config_clone_is_cheap() {
        let config = parse(valid_toml());
        let sc = shared_config(&config);
        let sc2 = Arc::clone(&sc);

        // Both point to the same ArcSwap
        assert_eq!(sc.load().max_depth, sc2.load().max_depth);

        // Mutation through one is visible through the other
        let mut new_rt = RuntimeConfig::from_server_config(&config);
        new_rt.max_depth = 999;
        sc.store(Arc::new(new_rt));
        assert_eq!(sc2.load().max_depth, 999);
    }

    #[test]
    fn runtime_config_multiple_namespaces() {
        use crate::types::Permission;

        let toml_str = r#"
            [tcp]
            bind_address = "127.0.0.1"
            port = 9876

            [namespaces.alpha]
            api_key = "aaaaaaaaaaaaaaaaaaaaaaaaaaaa"
            permissions = ["Subscribe", "Publish"]
            max_total_messages = 1000
            max_topics = 10
            max_connections = 50

            [namespaces.beta]
            api_key = "bbbbbbbbbbbbbbbbbbbbbbbbbbbb"
            permissions = ["Query"]
            max_disk_bytes = 1073741824

            [namespaces.beta.retention]
            max_age_secs = 3600
        "#;
        let config: ServerConfig = toml::from_str(toml_str).unwrap();
        let rt = RuntimeConfig::from_server_config(&config);

        assert_eq!(rt.namespace_quotas.len(), 2);
        assert_eq!(rt.namespace_auth.len(), 2);
        assert_eq!(rt.namespace_retention.len(), 2);

        // Check alpha
        let alpha_q = &rt.namespace_quotas[&NamespaceId("alpha".into())];
        assert_eq!(alpha_q.max_total_messages, Some(1000));
        assert_eq!(alpha_q.max_topics, Some(10));
        assert_eq!(alpha_q.max_connections, Some(50));

        let alpha_auth = &rt.namespace_auth[&NamespaceId("alpha".into())];
        assert_eq!(alpha_auth.permissions.len(), 2);
        assert!(alpha_auth.permissions.contains(&Permission::Subscribe));
        assert!(alpha_auth.permissions.contains(&Permission::Publish));

        // Check beta
        let beta_q = &rt.namespace_quotas[&NamespaceId("beta".into())];
        assert_eq!(beta_q.max_disk_bytes, Some(1_073_741_824));

        let beta_ret = &rt.namespace_retention[&NamespaceId("beta".into())];
        assert_eq!(beta_ret.max_age_secs, Some(3600));
    }

    /// Exhaustiveness guard for `RuntimeConfig::from_server_config()`.
    ///
    /// This test destructures every field of `RuntimeConfig`. If a new field
    /// is added to the struct but not listed here, the compiler emits:
    ///   "pattern does not mention field `new_field`"
    ///
    /// When you add a field to `RuntimeConfig`:
    /// 1. Map it in `from_server_config()`
    /// 2. Add it to the destructure below
    /// 3. Add an assertion that it was correctly mapped
    #[test]
    fn runtime_config_exhaustiveness_guard() {
        let config = parse(valid_toml());
        let rt = RuntimeConfig::from_server_config(&config);

        // Destructure ALL fields — compiler enforces completeness.
        let RuntimeConfig {
            // Queue engine
            max_depth,
            default_topic_max_bytes,
            backpressure_policy,
            global_max_memory_bytes,
            max_consumer_groups_per_topic,
            disk_capacity_threshold_pct,
            max_redelivery_attempts,
            ref dlq_topic_suffix,
            // Headers
            max_headers_per_message: _,
            max_header_key_bytes: _,
            max_header_value_bytes: _,
            max_header_bytes_total: _,
            // Redelivery backoff
            inflight_timeout_secs: _,
            max_redelivery_backoff_secs: _,
            // Manager delivery
            cold_batch_size,
            // TCP
            idle_timeout_secs,
            max_connections,
            subscriber_channel_buffer,
            slow_consumer_policy,
            slow_consumer_threshold,
            slow_consumer_disconnect_after,
            max_message_size,
            // HTTP
            max_fetch_response_bytes,
            // Storage
            max_segments_per_topic,
            retention_check_interval_secs,
            segment_merge_threshold_bytes,
            compaction_interval_secs,
            batch_flush_interval_ms,
            batch_flush_size_bytes,
            max_open_fds,
            // Deploy
            drain_timeout_secs,
            shutdown_timeout_secs,
            ref going_away_reason,
            // Per-namespace
            ref namespace_quotas,
            ref namespace_retention,
            compiled_inactive_overrides: _,
            ref namespace_auth,
        } = rt;

        // Spot-check a few fields to ensure the destructure isn't just
        // binding to defaults — the per-field tests above cover all values.
        assert_eq!(max_depth, config.queue.max_depth);
        assert_eq!(idle_timeout_secs, config.tcp.idle_timeout_secs);
        assert_eq!(
            max_fetch_response_bytes,
            config.http.max_fetch_response_bytes
        );
        assert_eq!(
            max_segments_per_topic,
            config.storage.max_segments_per_topic
        );
        assert_eq!(drain_timeout_secs, config.deploy.drain_timeout_secs);
        assert_eq!(
            default_topic_max_bytes,
            config.queue.default_topic_max_bytes
        );
        assert_eq!(backpressure_policy, config.queue.backpressure_policy);
        assert_eq!(
            global_max_memory_bytes,
            config.queue.global_max_memory_bytes
        );
        assert_eq!(
            max_consumer_groups_per_topic,
            config.queue.max_consumer_groups_per_topic
        );
        assert!(
            (disk_capacity_threshold_pct - config.queue.disk_capacity_threshold_pct).abs()
                < f64::EPSILON
        );
        assert_eq!(
            max_redelivery_attempts,
            config.queue.max_redelivery_attempts
        );
        assert_eq!(dlq_topic_suffix, &config.queue.dlq_topic_suffix);
        assert_eq!(cold_batch_size, 256); // ManagerConfig default
        assert_eq!(max_connections, config.tcp.max_connections);
        assert_eq!(
            subscriber_channel_buffer,
            config.tcp.subscriber_channel_buffer
        );
        assert_eq!(slow_consumer_policy, config.tcp.slow_consumer_policy);
        assert_eq!(slow_consumer_threshold, config.tcp.slow_consumer_threshold);
        assert_eq!(
            slow_consumer_disconnect_after,
            config.tcp.slow_consumer_disconnect_after
        );
        assert_eq!(max_message_size, config.tcp.max_message_size);
        assert_eq!(
            retention_check_interval_secs,
            config.storage.retention_check_interval_secs
        );
        assert_eq!(
            segment_merge_threshold_bytes,
            config.storage.segment_merge_threshold_bytes
        );
        assert_eq!(
            compaction_interval_secs,
            config.storage.compaction_interval_secs
        );
        assert_eq!(
            batch_flush_interval_ms,
            config.storage.batch_flush_interval_ms
        );
        assert_eq!(
            batch_flush_size_bytes,
            config.storage.batch_flush_size_bytes
        );
        assert_eq!(max_open_fds, config.storage.max_open_fds);
        assert_eq!(shutdown_timeout_secs, config.deploy.shutdown_timeout_secs);
        assert_eq!(going_away_reason, &config.deploy.going_away_reason);
        assert!(!namespace_quotas.is_empty());
        assert!(!namespace_retention.is_empty());
        assert!(!namespace_auth.is_empty());
    }

    // -----------------------------------------------------------------------
    // Phase 17.6 — Config diffing tests
    // -----------------------------------------------------------------------

    fn make_runtime_config() -> RuntimeConfig {
        let config = parse(valid_toml());
        RuntimeConfig::from_server_config(&config)
    }

    #[test]
    fn diff_no_changes() {
        let a = make_runtime_config();
        let b = make_runtime_config();
        let diff = b.diff(&a);
        assert!(diff.is_empty());
        assert!(diff.applied.is_empty());
        assert!(diff.namespaces_added.is_empty());
        assert!(diff.namespaces_removed.is_empty());
        assert!(!diff.auth_changed);
    }

    #[test]
    fn diff_detects_tier1_scalar_changes() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        b.max_depth = 999_999;
        b.idle_timeout_secs = 600;
        b.max_fetch_response_bytes = 8_000_000;
        b.drain_timeout_secs = 60;
        b.max_segments_per_topic = 200;
        let diff = b.diff(&a);
        assert_eq!(diff.applied.len(), 5);
        let fields: Vec<&str> = diff.applied.iter().map(|c| c.field.as_str()).collect();
        assert!(fields.contains(&"queue.max_depth"));
        assert!(fields.contains(&"tcp.idle_timeout_secs"));
        assert!(fields.contains(&"http.max_fetch_response_bytes"));
        assert!(fields.contains(&"deploy.drain_timeout_secs"));
        assert!(fields.contains(&"storage.max_segments_per_topic"));
        // All Tier 1 — no notifications required
        assert!(diff.applied.iter().all(|c| !c.requires_notification));
    }

    #[test]
    fn diff_detects_namespace_auth_change() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        // Change the API key for the "default" namespace
        if let Some(auth) = b.namespace_auth.get_mut(&NamespaceId("default".into())) {
            auth.api_key = ApiKey("new_key_that_is_long_enough!!".into());
        }
        let diff = b.diff(&a);
        assert_eq!(diff.applied.len(), 1);
        assert_eq!(diff.applied[0].field, "namespaces.default.auth");
        assert!(diff.applied[0].requires_notification);
        assert!(diff.auth_changed);
    }

    #[test]
    fn diff_detects_namespace_added() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        let ns = NamespaceId("new_ns".into());
        b.namespace_auth.insert(
            ns.clone(),
            NamespaceAuthConfig {
                api_key: ApiKey("abcdefghijklmnopqrstuvwx".into()),
                permissions: vec![Permission::Publish],
            },
        );
        b.namespace_quotas
            .insert(ns.clone(), NamespaceQuota::default());
        b.namespace_retention
            .insert(ns.clone(), RetentionPolicy::default());
        let diff = b.diff(&a);
        assert_eq!(diff.namespaces_added.len(), 1);
        assert_eq!(diff.namespaces_added[0], ns);
        assert!(diff.requires_notification());
    }

    #[test]
    fn diff_detects_namespace_removed() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        b.namespace_auth.remove(&NamespaceId("default".into()));
        b.namespace_quotas.remove(&NamespaceId("default".into()));
        b.namespace_retention.remove(&NamespaceId("default".into()));
        let diff = b.diff(&a);
        assert_eq!(diff.namespaces_removed.len(), 1);
        assert_eq!(diff.namespaces_removed[0], NamespaceId("default".into()));
        assert!(diff.requires_notification());
    }

    #[test]
    fn diff_detects_quota_change() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        if let Some(quota) = b.namespace_quotas.get_mut(&NamespaceId("default".into())) {
            quota.max_total_messages = Some(50_000);
        }
        let diff = b.diff(&a);
        assert_eq!(diff.applied.len(), 1);
        assert_eq!(diff.applied[0].field, "namespaces.default.quota");
        assert!(!diff.applied[0].requires_notification);
    }

    #[test]
    fn diff_detects_retention_change() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        b.namespace_retention.insert(
            NamespaceId("default".into()),
            RetentionPolicy {
                max_age_secs: Some(86400),
                ..Default::default()
            },
        );
        let diff = b.diff(&a);
        assert_eq!(diff.applied.len(), 1);
        assert_eq!(diff.applied[0].field, "namespaces.default.retention");
    }

    #[test]
    fn diff_detects_float_change() {
        let a = make_runtime_config();
        let mut b = make_runtime_config();
        b.disk_capacity_threshold_pct = 95.0;
        let diff = b.diff(&a);
        assert_eq!(diff.applied.len(), 1);
        assert_eq!(diff.applied[0].field, "queue.disk_capacity_threshold_pct");
    }

    // -----------------------------------------------------------------------
    // Phase 17.6 — Tier 3 change detection tests
    // -----------------------------------------------------------------------

    #[test]
    fn tier3_no_changes() {
        let a = parse(valid_toml());
        let b = parse(valid_toml());
        let changes = detect_restart_only_changes(&a, &b);
        assert!(changes.is_empty());
    }

    #[test]
    fn tier3_detects_bind_address_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.tcp.bind_address = "0.0.0.0".to_owned();
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "tcp.bind_address");
        assert!(changes[0].description.contains("requires restart"));
    }

    #[test]
    fn tier3_detects_port_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.tcp.port = 1234;
        b.http.port = 5678;
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 2);
        let fields: Vec<&str> = changes.iter().map(|c| c.field.as_str()).collect();
        assert!(fields.contains(&"tcp.port"));
        assert!(fields.contains(&"http.port"));
    }

    #[test]
    fn tier3_detects_data_dir_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.storage.data_dir = "/new/data".to_owned();
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "storage.data_dir");
    }

    #[test]
    fn tier3_detects_disk_backed_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.queue.disk_backed = true;
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "queue.disk_backed");
    }

    #[test]
    fn tier3_detects_admin_api_key_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.deploy.admin_api_key = Some("new_secret_admin_key_12345678".to_owned());
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "deploy.admin_api_key");
    }

    #[test]
    fn tier3_detects_fsync_strategy_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.storage.fsync_strategy = FsyncStrategy::EveryWrite;
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "storage.fsync_strategy");
    }

    #[test]
    fn tier3_detects_index_interval_change() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.storage.index_interval = 512;
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "storage.index_interval");
    }

    #[test]
    fn tier3_detects_multiple_changes() {
        let a = parse(valid_toml());
        let mut b = parse(valid_toml());
        b.tcp.bind_address = "0.0.0.0".to_owned();
        b.tcp.port = 1111;
        b.http.bind_address = "0.0.0.0".to_owned();
        b.http.port = 2222;
        b.storage.data_dir = "/tmp/data".to_owned();
        b.storage.fsync_strategy = FsyncStrategy::EveryWrite;
        b.storage.index_interval = 128;
        b.queue.disk_backed = true;
        b.deploy.admin_api_key = Some("admin_key_long_enough_12345678".to_owned());
        let changes = detect_restart_only_changes(&a, &b);
        assert_eq!(changes.len(), 9);
    }

    // -----------------------------------------------------------------------
    // Phase 17.6 — ConfigDiff helper method tests
    // -----------------------------------------------------------------------

    #[test]
    fn config_diff_is_empty() {
        let diff = ConfigDiff {
            applied: vec![],
            ignored: vec![],
            namespaces_added: vec![],
            namespaces_removed: vec![],
            auth_changed: false,
        };
        assert!(diff.is_empty());
    }

    #[test]
    fn config_diff_not_empty_with_applied() {
        let diff = ConfigDiff {
            applied: vec![ConfigChange {
                field: "test".into(),
                description: "test".into(),
                requires_notification: false,
            }],
            ignored: vec![],
            namespaces_added: vec![],
            namespaces_removed: vec![],
            auth_changed: false,
        };
        assert!(!diff.is_empty());
    }

    #[test]
    fn config_diff_not_empty_with_auth_changed() {
        let diff = ConfigDiff {
            applied: vec![],
            ignored: vec![],
            namespaces_added: vec![],
            namespaces_removed: vec![],
            auth_changed: true,
        };
        assert!(!diff.is_empty());
        assert!(diff.requires_notification());
    }

    #[test]
    fn config_diff_requires_notification_on_auth_change() {
        let diff = ConfigDiff {
            applied: vec![ConfigChange {
                field: "namespaces.test.auth".into(),
                description: "changed".into(),
                requires_notification: true,
            }],
            ignored: vec![],
            namespaces_added: vec![],
            namespaces_removed: vec![],
            auth_changed: true,
        };
        assert!(diff.requires_notification());
    }

    #[test]
    fn config_diff_requires_notification_on_namespace_add() {
        let diff = ConfigDiff {
            applied: vec![],
            ignored: vec![],
            namespaces_added: vec![NamespaceId("new".into())],
            namespaces_removed: vec![],
            auth_changed: false,
        };
        assert!(diff.requires_notification());
    }

    // -----------------------------------------------------------------------
    // Phase 17.6 — Diff exhaustiveness guard
    // -----------------------------------------------------------------------

    #[test]
    fn diff_exhaustiveness_guard() {
        // This test ensures that every field of RuntimeConfig is covered in
        // the diff() method. If a new field is added to RuntimeConfig without
        // updating diff(), this test will fail to compile.
        let config = make_runtime_config();
        let RuntimeConfig {
            max_depth: _,
            default_topic_max_bytes: _,
            backpressure_policy: _,
            global_max_memory_bytes: _,
            max_consumer_groups_per_topic: _,
            disk_capacity_threshold_pct: _,
            max_redelivery_attempts: _,
            dlq_topic_suffix: _,
            max_headers_per_message: _,
            max_header_key_bytes: _,
            max_header_value_bytes: _,
            max_header_bytes_total: _,
            inflight_timeout_secs: _,
            max_redelivery_backoff_secs: _,
            cold_batch_size: _,
            idle_timeout_secs: _,
            max_connections: _,
            subscriber_channel_buffer: _,
            slow_consumer_policy: _,
            slow_consumer_threshold: _,
            slow_consumer_disconnect_after: _,
            max_message_size: _,
            max_fetch_response_bytes: _,
            max_segments_per_topic: _,
            retention_check_interval_secs: _,
            segment_merge_threshold_bytes: _,
            compaction_interval_secs: _,
            batch_flush_interval_ms: _,
            batch_flush_size_bytes: _,
            max_open_fds: _,
            drain_timeout_secs: _,
            shutdown_timeout_secs: _,
            going_away_reason: _,
            namespace_quotas: _,
            namespace_retention: _,
            compiled_inactive_overrides: _,
            namespace_auth: _,
        } = config;
        // If this compiles, all fields are covered.
    }

    // -----------------------------------------------------------------------
    // Phase 17.6 — ReloadError Display tests
    // -----------------------------------------------------------------------

    #[test]
    fn reload_error_display() {
        let e = ReloadError::ServerDraining;
        assert!(e.to_string().contains("draining"));

        let e = ReloadError::NoConfigFile;
        assert!(e.to_string().contains("no config file"));

        let e = ReloadError::ReadError(std::io::Error::new(std::io::ErrorKind::NotFound, "gone"));
        assert!(e.to_string().contains("failed to read"));

        let e = ReloadError::ValidationError(vec![ConfigError::NoNamespaces]);
        assert!(e.to_string().contains("validation failed"));

        let e = ReloadError::NamespaceRemovalBlocked {
            namespace: NamespaceId("test".into()),
            active_connections: 5,
        };
        assert!(e.to_string().contains("test"));
        assert!(e.to_string().contains("5"));
    }

    // -----------------------------------------------------------------------
    // Phase 17.8 — HotReloadConfig, config introspection, Tier 3 detection
    // -----------------------------------------------------------------------

    #[test]
    fn hot_reload_config_defaults() {
        let cfg = HotReloadConfig::default();
        assert!(!cfg.watch_file);
        assert_eq!(cfg.debounce_secs, 2);
    }

    #[test]
    fn hot_reload_config_parses_from_toml() {
        let toml = r#"
[hot_reload]
watch_file = true
debounce_secs = 5

[namespaces.test]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe", "Publish"]
"#;
        let config: ServerConfig = toml::from_str(toml).expect("should parse");
        assert!(config.hot_reload.watch_file);
        assert_eq!(config.hot_reload.debounce_secs, 5);
    }

    #[test]
    fn hot_reload_config_unknown_field_rejected() {
        let toml = r#"
[hot_reload]
watch_file = true
debounce_secs = 2
unknown_field = 42

[namespaces.test]
api_key = "abcdefghijklmnopqrstuvwx"
permissions = ["Subscribe", "Publish"]
"#;
        let result = toml::from_str::<ServerConfig>(toml);
        assert!(result.is_err(), "unknown field should be rejected");
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("unknown_field"),
            "error should name the field: {err}"
        );
    }

    #[test]
    fn hot_reload_config_omitted_uses_defaults() {
        // A config with no [hot_reload] section should use defaults
        let config = parse(valid_toml());
        assert!(!config.hot_reload.watch_file);
        assert_eq!(config.hot_reload.debounce_secs, 2);
    }

    #[test]
    fn detect_restart_only_changes_hot_reload() {
        let old = parse(valid_toml());
        let mut new = old.clone();

        // No changes — empty list
        let changes = detect_restart_only_changes(&old, &new);
        assert!(changes.is_empty());

        // Change watch_file
        new.hot_reload.watch_file = true;
        let changes = detect_restart_only_changes(&old, &new);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "hot_reload.watch_file");

        // Change debounce_secs
        new.hot_reload.watch_file = false; // reset
        new.hot_reload.debounce_secs = 10;
        let changes = detect_restart_only_changes(&old, &new);
        assert_eq!(changes.len(), 1);
        assert_eq!(changes[0].field, "hot_reload.debounce_secs");
    }

    #[test]
    fn runtime_config_to_json_redacted() {
        let config = parse(valid_toml());
        let runtime = RuntimeConfig::from_server_config(&config);
        let json = runtime.to_json_redacted();

        // API key should be redacted
        let auth = json.get("namespace_auth").unwrap();
        let ns_auth = auth.get("default").unwrap();
        let redacted_key = ns_auth.get("api_key").unwrap().as_str().unwrap();
        assert!(
            redacted_key.starts_with("****"),
            "key should be redacted: {redacted_key}"
        );
        assert!(
            !redacted_key.contains("abcdefghijklmnopqrstuvwx"),
            "full key should not appear in redacted output"
        );
        assert!(
            redacted_key.ends_with("uvwx"),
            "last 4 chars should be visible: {redacted_key}"
        );

        // Other fields should be present
        assert!(json.get("max_depth").is_some());
        assert!(json.get("global_max_memory_bytes").is_some());
    }

    #[test]
    fn hot_reload_config_serializes() {
        let cfg = HotReloadConfig {
            watch_file: true,
            debounce_secs: 5,
        };
        let toml = toml::to_string_pretty(&cfg).expect("should serialize");
        assert!(toml.contains("watch_file = true"));
        assert!(toml.contains("debounce_secs = 5"));
    }

    #[test]
    fn server_config_with_hot_reload_roundtrips() {
        let config = parse(valid_toml());
        let dump = config.dump_redacted();
        let reparsed: ServerConfig = toml::from_str(&dump).expect("dump should roundtrip");
        assert_eq!(reparsed.hot_reload.watch_file, config.hot_reload.watch_file);
        assert_eq!(
            reparsed.hot_reload.debounce_secs,
            config.hot_reload.debounce_secs
        );
    }

    // ── Cluster config validation ──────────────────────────────────

    fn cluster_config_valid() -> ClusterConfig {
        ClusterConfig {
            enabled: true,
            node_id: 1,
            bind: "0.0.0.0:9877".to_owned(),
            peers: vec!["192.168.1.2:9877".to_owned()],
            cluster_secret: "a]b]c]d]e]f]g]h]i]j]k]l]m]n]o]p]".to_owned(),
            ..Default::default()
        }
    }

    #[test]
    fn cluster_disabled_skips_validation() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            enabled: false,
            node_id: 0,    // would fail if enabled
            peers: vec![], // would fail if enabled
            ..Default::default()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { .. }))
            .collect();
        assert!(errors.is_empty(), "disabled cluster should skip validation");
    }

    #[test]
    fn cluster_node_id_must_be_nonzero() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            node_id: 0,
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("node_id")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_peers_must_be_nonempty() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            peers: vec![],
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("peers")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_secret_minimum_length() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            cluster_secret: "too-short".to_owned(),
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("cluster_secret")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_bind_not_in_peers() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            bind: "192.168.1.1:9877".to_owned(),
            peers: vec!["192.168.1.1:9877".to_owned()], // same as bind
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("bind address")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_failover_timeout_minimum() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            heartbeat_interval_ms: 1000,
            failover_timeout_ms: 2000, // less than 3x heartbeat
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("failover_timeout")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_no_clients_timeout_minimum() {
        let mut config = parse(valid_toml());
        config.cluster = ClusterConfig {
            failover_timeout_ms: 5000,
            failover_no_clients_timeout_ms: 4000, // less than failover_timeout
            ..cluster_config_valid()
        };
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("failover_no_clients")))
            .collect();
        assert_eq!(errors.len(), 1);
    }

    #[test]
    fn cluster_valid_config_passes() {
        let mut config = parse(valid_toml());
        config.cluster = cluster_config_valid();
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { .. }))
            // When compiled without --features cluster, the "requires feature" error is expected.
            .filter(|e| !matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("requires --features")))
            .collect();
        assert!(
            errors.is_empty(),
            "valid cluster config should pass, got: {errors:?}"
        );
    }

    #[test]
    #[cfg(not(feature = "cluster"))]
    fn cluster_enabled_without_feature_errors() {
        let mut config = parse(valid_toml());
        config.cluster = cluster_config_valid();
        let errors: Vec<_> = config
            .validate()
            .into_iter()
            .filter(|e| matches!(e, ConfigError::InvalidClusterConfig { reason } if reason.contains("requires --features")))
            .collect();
        assert_eq!(errors.len(), 1, "should error when feature not compiled");
    }

    // -----------------------------------------------------------------------
    // Task 3 — TopicInactiveOverride / topic_inactive_secs config tests
    // -----------------------------------------------------------------------

    #[test]
    fn retention_policy_with_topic_inactive_secs() {
        let toml_str = r#"
            [storage]
            data_dir = "/tmp/data"

            [namespaces.alpha]
            api_key = "alpha-key-min-24-chars-long"
            permissions = ["Publish"]

            [namespaces.alpha.retention]
            topic_inactive_secs = 604800
            topic_reap_grace_period_secs = 43200

            [[namespaces.alpha.retention.topic_inactive_overrides]]
            pattern = "system.>"
            inactive_secs = 0

            [[namespaces.alpha.retention.topic_inactive_overrides]]
            pattern = "tmp.*"
            inactive_secs = 3600
        "#;
        let config: ServerConfig = toml::from_str(toml_str).unwrap();
        let alpha = &config.namespaces[&NamespaceId("alpha".into())];
        assert_eq!(alpha.retention.topic_inactive_secs, Some(604800));
        assert_eq!(alpha.retention.topic_reap_grace_period_secs, Some(43200));
        assert_eq!(alpha.retention.topic_inactive_overrides.len(), 2);
        assert_eq!(
            alpha.retention.topic_inactive_overrides[0].pattern,
            "system.>"
        );
        assert_eq!(alpha.retention.topic_inactive_overrides[0].inactive_secs, 0);
        assert_eq!(alpha.retention.topic_inactive_overrides[1].pattern, "tmp.*");
        assert_eq!(
            alpha.retention.topic_inactive_overrides[1].inactive_secs,
            3600
        );
    }

    #[test]
    fn retention_is_active_with_only_topic_inactive_secs() {
        let policy = RetentionPolicy {
            max_age_secs: None,
            max_bytes: None,
            max_segments: None,
            topic_inactive_secs: Some(86400),
            topic_reap_grace_period_secs: None,
            topic_inactive_overrides: vec![],
        };
        assert!(policy.is_active());
    }

    #[test]
    fn retention_policy_default_has_no_gc() {
        let policy = RetentionPolicy::default();
        assert_eq!(policy.topic_inactive_secs, None);
        assert_eq!(policy.topic_reap_grace_period_secs, None);
        assert!(policy.topic_inactive_overrides.is_empty());
        assert!(!policy.is_active());
    }

    #[test]
    #[should_panic(expected = "invalid topic_inactive_overrides pattern")]
    fn invalid_override_pattern_panics_at_config_load() {
        let toml_str = r#"
            [storage]
            data_dir = "/tmp/data"

            [namespaces.beta]
            api_key = "beta-key-min-24-chars-long00"
            permissions = ["Publish"]

            [[namespaces.beta.retention.topic_inactive_overrides]]
            pattern = "bad..pattern"
            inactive_secs = 600
        "#;
        let config: ServerConfig = toml::from_str(toml_str).unwrap();
        let _ = RuntimeConfig::from_server_config(&config);
    }

    #[test]
    fn runtime_config_compiles_inactive_overrides() {
        let toml_str = r#"
            [storage]
            data_dir = "/tmp/data"

            [namespaces.alpha]
            api_key = "alpha-key-min-24-chars-long"
            permissions = ["Publish"]

            [[namespaces.alpha.retention.topic_inactive_overrides]]
            pattern = "system.>"
            inactive_secs = 0

            [[namespaces.alpha.retention.topic_inactive_overrides]]
            pattern = "tmp.*"
            inactive_secs = 3600
        "#;
        let config: ServerConfig = toml::from_str(toml_str).unwrap();
        let rt = RuntimeConfig::from_server_config(&config);
        let ns_id = NamespaceId("alpha".into());
        let compiled = rt.compiled_inactive_overrides.get(&ns_id).unwrap();
        assert_eq!(compiled.len(), 2);
        assert!(compiled[0].pattern.matches("system.events"));
        assert_eq!(compiled[0].inactive_secs, 0);
        assert!(compiled[1].pattern.matches("tmp.foo"));
        assert_eq!(compiled[1].inactive_secs, 3600);
    }

    #[test]
    fn runtime_config_no_overrides_omitted_from_compiled_map() {
        let config: ServerConfig = toml::from_str(valid_toml()).unwrap();
        let rt = RuntimeConfig::from_server_config(&config);
        // valid_toml() has no topic_inactive_overrides — map should be empty
        assert!(rt.compiled_inactive_overrides.is_empty());
    }
}
