use std::path::PathBuf;
use std::process;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Duration;

use clap::{CommandFactory, Parser, Subcommand};
use tracing_subscriber::Layer as _;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

use fafb_q::auth::ApiKeyStore;
use fafb_q::cli;
use fafb_q::config::{
    LoggingConfig, ReloadState, ServeArgs, ServerConfig, check_config_permissions,
    find_config_path, shared_config,
};
use fafb_q::namespace::{DefaultNamespaceManager, ManagerConfig, NamespaceManager};
use fafb_q::queue::{EngineConfig, InMemoryQueueEngine, QueueEngine};
use fafb_q::retention::RetentionTask;
use fafb_q::server::http::{AppState, DrainRequest, DrainResponse, GracePendingHandle, HttpServer};
use fafb_q::server::state::ServerLifecycle;
use fafb_q::server::tcp::TcpServer;
use fafb_q::storage::{FsStorage, StorageBackend};
use fafb_q::tls::{LabeledTlsProvider, TlsProvider, TlsProviders};

// ---------------------------------------------------------------------------
// CLI argument definitions
// ---------------------------------------------------------------------------

/// fafb-q — centralized queue server and CLI tool.
///
/// When no subcommand is given, `serve` is implied. This preserves backward
/// compatibility: `fafb-q --config foo.toml` still works.
#[derive(Parser)]
#[command(
    name = "fafb-q",
    version,
    about = "Centralized queue server and CLI tool"
)]
#[command(args_conflicts_with_subcommands = true)]
struct Cli {
    #[command(subcommand)]
    command: Option<Command>,

    /// Global options shared by all non-serve commands. These are declared
    /// with `global = true` so they work on any subcommand position.
    #[command(flatten)]
    global: cli::GlobalOpts,

    /// Serve args are flattened so `fafb-q --config foo.toml` works without
    /// the explicit `serve` subcommand (backward compatibility).
    #[command(flatten)]
    serve_args: ServeArgs,
}

#[derive(Subcommand)]
enum Command {
    /// Run the server
    Serve(ServeArgs),
    /// Publish messages
    Pub {
        /// Topic in ns/topic format
        topic: String,
        /// Message payload (reads from stdin if omitted)
        payload: Option<String>,
        /// Read payload from file
        #[arg(long)]
        file: Option<String>,
        /// One message per line from stdin
        #[arg(long)]
        lines: bool,
        /// Message TTL in seconds
        #[arg(long)]
        ttl: Option<u64>,
        /// Headers as "Key: Value" (repeatable)
        #[arg(short = 'H', long = "header")]
        headers: Vec<String>,
        /// Don't wait for PublishAck
        #[arg(long)]
        no_ack: bool,
    },
    /// Subscribe to topics and stream messages
    Sub {
        /// Topic in ns/topic format (supports wildcards)
        topic: String,
        /// Subscribe from a specific offset
        #[arg(long)]
        from: Option<u64>,
        /// Subscribe from a timestamp (ISO8601 or unix millis)
        #[arg(long)]
        since: Option<String>,
        /// Consumer group name
        #[arg(short = 'g', long)]
        group: Option<String>,
        /// Receive N messages then exit
        #[arg(long)]
        count: Option<u64>,
        /// Don't send Ack (receive only)
        #[arg(long)]
        no_ack: bool,
        /// Output only payload bytes
        #[arg(long)]
        payload_only: bool,
    },
    /// Fetch messages (pull-based)
    Fetch {
        /// Topic in ns/topic format
        topic: String,
        /// Start offset (negative = relative from head: -1 = last message)
        #[arg(long, allow_hyphen_values = true, conflicts_with = "tail")]
        from: Option<i64>,
        /// Start timestamp (ISO8601 or unix millis)
        #[arg(long, conflicts_with = "tail")]
        since: Option<String>,
        /// Fetch last N messages (shorthand for --from -N --count N)
        #[arg(long, conflicts_with_all = ["from", "since"])]
        tail: Option<u32>,
        /// Max messages per batch
        #[arg(long, default_value = "100")]
        count: u32,
        /// Fetch all messages (paginate)
        #[arg(long)]
        all: bool,
        /// Output only payload bytes
        #[arg(long)]
        payload_only: bool,
    },
    /// Queue and namespace information
    Info {
        /// Namespace or ns/topic
        target: Option<String>,
    },
    /// Trigger graceful drain
    Drain {
        /// Admin API key (env: FAFB_ADMIN_API_KEY)
        #[arg(long, env = "FAFB_ADMIN_API_KEY")]
        admin_key: Option<String>,
        /// Custom drain timeout in seconds
        #[arg(long)]
        timeout: Option<u64>,
        /// Check drain status instead of triggering
        #[arg(long)]
        status: bool,
        /// Drain reason
        #[arg(long)]
        reason: Option<String>,
        /// Wait for drain to complete (poll until done)
        #[arg(long)]
        wait: bool,
    },
    /// Health check
    Health {
        /// Use liveness probe instead of readiness
        #[arg(long)]
        live: bool,
    },
    /// Configuration utilities
    Config {
        #[command(subcommand)]
        action: Option<cli::ConfigAction>,
    },
    /// Benchmarking
    #[command(subcommand_required = true, arg_required_else_help = true)]
    Bench {
        #[command(subcommand)]
        action: cli::BenchAction,
    },
    /// Generate shell completions
    Completions {
        /// Shell to generate completions for
        #[arg(value_enum)]
        shell: clap_complete::Shell,
    },
}

/// Initialize the tracing subscriber from config.
///
/// Returns the non-blocking writer guard (must be held for the program
/// lifetime) and a reload handle for dynamic log level changes.
fn init_tracing(
    config: &LoggingConfig,
) -> (
    Option<tracing_appender::non_blocking::WorkerGuard>,
    fafb_q::server::http::LogFilterReloadHandle,
) {
    use tracing_subscriber::EnvFilter;

    let env_filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(&config.level));
    let (filter_layer, reload_handle) = tracing_subscriber::reload::Layer::new(env_filter);

    let is_json = config.format.eq_ignore_ascii_case("json");

    // Stdout layer
    let stdout_layer = if config.stdout {
        if is_json {
            Some(
                tracing_subscriber::fmt::layer()
                    .json()
                    .with_writer(std::io::stdout)
                    .boxed(),
            )
        } else {
            Some(
                tracing_subscriber::fmt::layer()
                    .with_writer(std::io::stdout)
                    .boxed(),
            )
        }
    } else {
        None
    };

    // File layer + guard
    let (file_layer, guard) = if config.file_enabled {
        let rotation = match config.rotation.as_str() {
            "hourly" => tracing_appender::rolling::Rotation::HOURLY,
            _ => tracing_appender::rolling::Rotation::DAILY,
        };
        if let Err(e) = std::fs::create_dir_all(&config.file_dir) {
            eprintln!(
                "ERROR: failed to create log directory {:?}: {e}",
                config.file_dir
            );
            std::process::exit(1);
        }
        let file_appender = match tracing_appender::rolling::RollingFileAppender::builder()
            .rotation(rotation)
            .filename_prefix(&config.file_prefix)
            .filename_suffix("log")
            .build(&config.file_dir)
        {
            Ok(appender) => appender,
            Err(e) => {
                eprintln!(
                    "ERROR: failed to create log file appender in {:?}: {e}",
                    config.file_dir
                );
                std::process::exit(1);
            }
        };
        let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
        let layer = if is_json {
            tracing_subscriber::fmt::layer()
                .json()
                .with_writer(non_blocking)
                .with_ansi(false)
                .boxed()
        } else {
            tracing_subscriber::fmt::layer()
                .with_writer(non_blocking)
                .with_ansi(false)
                .boxed()
        };
        (Some(layer), Some(guard))
    } else {
        (None, None)
    };

    tracing_subscriber::registry()
        .with(filter_layer)
        .with(stdout_layer)
        .with(file_layer)
        .init();

    (guard, reload_handle)
}

/// Exponential backoff for task respawn: 1s, 2s, 4s, 8s, ... capped at 60s.
/// Await an optional `JoinHandle`, returning `Some(result)` when the handle
/// exists and completes, or pending forever when `None`. Used in `select!` arms
/// for optional background tasks like the config file watcher.
async fn option_join(
    handle: &mut Option<tokio::task::JoinHandle<()>>,
) -> Option<Result<(), tokio::task::JoinError>> {
    match handle {
        Some(h) => Some(h.await),
        None => std::future::pending().await,
    }
}

fn backoff_delay(consecutive_failures: u32) -> Duration {
    let secs = 1u64
        .checked_shl(consecutive_failures.saturating_sub(1))
        .unwrap_or(60)
        .min(60);
    Duration::from_secs(secs)
}

/// Delete log files that exceed the age or total size limits.
fn cleanup_log_files(
    dir: &str,
    prefix: &str,
    max_age_days: u32,
    max_total_bytes: u64,
) -> std::io::Result<()> {
    use std::fs;

    let max_age = Duration::from_secs(u64::from(max_age_days) * 86400);
    let now = std::time::SystemTime::now();

    let mut entries: Vec<(PathBuf, u64, std::time::SystemTime)> = Vec::new();

    for entry in fs::read_dir(dir)? {
        let entry = entry?;
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        let name = match path.file_name().and_then(|n| n.to_str()) {
            Some(n) => n,
            None => continue,
        };
        if !name.starts_with(prefix) || !name.ends_with(".log") {
            continue;
        }
        let meta = entry.metadata()?;
        let modified = meta.modified().unwrap_or(now);
        entries.push((path, meta.len(), modified));
    }

    // Delete files older than max_age_days
    entries.retain(|&(ref path, _, modified)| {
        if let Ok(age) = now.duration_since(modified)
            && age > max_age
        {
            if let Err(e) = fs::remove_file(path) {
                tracing::warn!(path = %path.display(), error = %e, "failed to delete old log file");
            } else {
                tracing::debug!(path = %path.display(), "deleted old log file");
            }
            return false;
        }
        true
    });

    // Enforce total size limit — delete oldest first
    let total: u64 = entries.iter().map(|(_, sz, _)| sz).sum();
    if total > max_total_bytes {
        entries.sort_by_key(|&(_, _, modified)| modified);
        let mut remaining = total;
        for (path, sz, _) in &entries {
            if remaining <= max_total_bytes {
                break;
            }
            if let Err(e) = fs::remove_file(path) {
                tracing::warn!(path = %path.display(), error = %e, "failed to delete log file for size limit");
            } else {
                tracing::debug!(path = %path.display(), "deleted log file (size limit)");
                remaining -= sz;
            }
        }
    }

    Ok(())
}

/// Reload TLS certificates on all providers, logging results and updating metrics.
fn reload_tls_providers(providers: &[LabeledTlsProvider]) {
    for lp in providers {
        match lp.provider.reload() {
            Ok(true) => {
                metrics::counter!("fafb_tls_reload_total",
                    "status" => "success", "listener" => lp.label
                )
                .increment(1);
                metrics::gauge!("fafb_tls_cert_expiry_seconds",
                    "listener" => lp.label
                )
                .set(lp.provider.cert_ttl_secs() as f64);
                tracing::info!(
                    listener = lp.label,
                    cert_ttl_secs = lp.provider.cert_ttl_secs(),
                    "TLS certificates reloaded"
                );
            }
            Ok(false) => {
                metrics::counter!("fafb_tls_reload_total",
                    "status" => "unchanged", "listener" => lp.label
                )
                .increment(1);
            }
            Err(e) => {
                metrics::counter!("fafb_tls_reload_total",
                    "status" => "error", "listener" => lp.label
                )
                .increment(1);
                tracing::error!(
                    listener = lp.label,
                    error = %e,
                    "TLS certificate reload failed — old cert continues serving"
                );
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let cli = Cli::parse();

    // Apply color mode before any output.
    cli::apply_color_mode(cli.global.color);

    match cli.command {
        // No subcommand given: use the flattened serve args (backward compat).
        // `fafb-q --config foo.toml` works because ServeArgs is flattened.
        None => run_server(cli.serve_args).await,

        Some(Command::Serve(args)) => run_server(args).await,

        Some(Command::Completions { shell }) => {
            let mut cmd = Cli::command();
            clap_complete::generate(shell, &mut cmd, "fafb-q", &mut std::io::stdout());
        }

        Some(cmd) => {
            let config = cli::resolve_cli_config(&cli.global);

            let result = match cmd {
                Command::Serve(_) | Command::Completions { .. } => unreachable!(),
                Command::Pub {
                    topic,
                    payload,
                    file,
                    lines,
                    ttl,
                    headers,
                    no_ack,
                } => {
                    cli::run_publish(
                        config,
                        cli::PubArgs {
                            topic,
                            payload,
                            file,
                            lines,
                            ttl_secs: ttl,
                            headers,
                            no_ack,
                        },
                    )
                    .await
                }
                Command::Sub {
                    topic,
                    from,
                    since,
                    group,
                    count,
                    no_ack,
                    payload_only,
                } => {
                    cli::run_subscribe(
                        config,
                        cli::SubArgs {
                            topic,
                            from,
                            since,
                            group,
                            count,
                            no_ack,
                            payload_only,
                        },
                    )
                    .await
                }
                Command::Fetch {
                    topic,
                    from,
                    since,
                    tail,
                    count,
                    all,
                    payload_only,
                } => {
                    cli::run_fetch(
                        config,
                        cli::FetchArgs {
                            topic,
                            from,
                            since,
                            tail,
                            count,
                            all,
                            payload_only,
                        },
                    )
                    .await
                }
                Command::Info { target } => cli::run_info(config, cli::InfoArgs { target }).await,
                Command::Drain {
                    admin_key,
                    timeout,
                    status,
                    reason,
                    wait,
                } => {
                    cli::run_drain(
                        config,
                        cli::DrainArgs {
                            admin_key,
                            timeout,
                            status,
                            reason,
                            wait,
                        },
                    )
                    .await
                }
                Command::Health { live } => cli::run_health(config, cli::HealthArgs { live }).await,
                Command::Config { action } => {
                    cli::run_config_cmd(config, action.unwrap_or(cli::ConfigAction::Example)).await
                }
                Command::Bench { action } => cli::run_bench(config, action).await,
            };

            if let Err(e) = result {
                use colored::Colorize;
                eprintln!("{} {e}", "error:".red().bold());
                process::exit(e.exit_code());
            }
        }
    }
}

/// Run the fafb-q server. This is the original main() body, extracted so
/// it can be called from the `serve` subcommand.
async fn run_server(args: ServeArgs) {
    // Load config before tracing so we can configure the subscriber from it.
    // Config errors go to stderr since tracing isn't ready yet.
    let config_path = find_config_path(args.config.clone(), |k| std::env::var(k).ok());
    let mut config = match &config_path {
        Some(path) => {
            check_config_permissions(path);
            match ServerConfig::load(path) {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("ERROR: failed to load config: {e}");
                    process::exit(1);
                }
            }
        }
        None => {
            eprintln!(
                "INFO: No config file found (searched: $FAFB_CONFIG, ./config.toml, /etc/fafb-q/config.toml). Using compiled defaults."
            );
            match ServerConfig::load_defaults() {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("ERROR: failed to validate default config: {e}");
                    process::exit(1);
                }
            }
        }
    };

    // Apply CLI flag overrides (highest priority: CLI > env > file > defaults).
    config.apply_cli_overrides(&args);

    // Build the shared runtime config for hot-reload (Phase 17.5).
    // Subsystems don't read from it yet — they still use owned copies.
    let shared_cfg = shared_config(&config);
    let config_path_arc = config_path.clone().map(Arc::new);

    // Reload notification channel — created early so the receiver is available
    // before ReloadState is constructed (which happens after lifecycle creation).
    let (reload_notify_tx, reload_notify_rx) =
        tokio::sync::watch::channel::<Option<fafb_q::config::ReloadNotification>>(None);

    if args.dump_config {
        print!("{}", config.dump_redacted());
        process::exit(0);
    }

    // Initialize tracing subscriber from config. Hold the guard so file
    // writes are flushed on shutdown.
    let (_log_guard, log_reload_handle) = init_tracing(&config.logging);

    // Log what was configured so operators can verify at a glance.
    if config.logging.file_enabled {
        tracing::info!(
            level = %config.logging.level,
            format = %config.logging.format,
            stdout = config.logging.stdout,
            file_dir = %config.logging.file_dir,
            rotation = %config.logging.rotation,
            max_age_days = config.logging.max_age_days,
            max_total_size_mb = config.logging.max_total_size_mb,
            "logging initialized (stdout + file)"
        );
    } else {
        tracing::info!(
            level = %config.logging.level,
            format = %config.logging.format,
            "logging initialized (stdout only)"
        );
    }

    // Global panic hook — ensures panics in any tokio task are logged
    // through tracing (and thus captured in log files) before unwinding.
    let default_panic_hook = std::panic::take_hook();
    std::panic::set_hook(Box::new(move |info| {
        let location = info.location().map(|l| l.to_string()).unwrap_or_default();
        let payload = if let Some(s) = info.payload().downcast_ref::<&str>() {
            (*s).to_string()
        } else if let Some(s) = info.payload().downcast_ref::<String>() {
            s.clone()
        } else {
            "unknown panic payload".to_string()
        };
        tracing::error!(
            panic.message = %payload,
            panic.location = %location,
            "task panicked"
        );
        default_panic_hook(info);
    }));

    // -- Metrics recorder --

    let prom_handle = if config.metrics.enabled {
        let handle = metrics_exporter_prometheus::PrometheusBuilder::new()
            .set_buckets(&[0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0])
            .expect("valid bucket boundaries")
            .install_recorder()
            .expect("failed to install Prometheus recorder");
        tracing::info!("Prometheus metrics recorder installed");
        metrics::gauge!("fafb_server_state").set(0.0);
        Some(handle)
    } else {
        tracing::info!("metrics disabled by configuration");
        None
    };

    tracing::info!(
        tcp = %format!("{}:{}", config.tcp.bind_address, config.tcp.port),
        http = %format!("{}:{}", config.http.bind_address, config.http.port),
        data_dir = %config.storage.data_dir,
        namespaces = config.namespaces.len(),
        "starting rust-fafb-q"
    );

    // -- Storage + Queue Engine --
    //
    // Create storage and engine objects (fast), but defer init() (slow — topic
    // replay) until after the HTTP server starts so health probes work during
    // startup.

    let storage = Arc::new(FsStorage::from_config(&config.storage));

    let engine = Arc::new(InMemoryQueueEngine::with_shared_config(
        Arc::clone(&storage),
        EngineConfig {
            max_depth: config.queue.max_depth,
            backpressure: config.queue.backpressure_policy,
            topic_max_bytes: config.queue.default_topic_max_bytes,
            global_max_memory_bytes: config.queue.global_max_memory_bytes,
            flush_interval_ms: config.storage.batch_flush_interval_ms,
            batch_flush_size_bytes: config.storage.batch_flush_size_bytes,
            disk_backed: config.queue.disk_backed,
            disk_capacity_threshold_pct: config.queue.disk_capacity_threshold_pct,
            data_dir: Some(std::path::PathBuf::from(&config.storage.data_dir)),
        },
        Some(Arc::clone(&shared_cfg)),
    ));

    // -- Auth + Namespace Manager --

    let api_key_store = Arc::new(ApiKeyStore::from_config(&config));
    let auth: Arc<dyn fafb_q::auth::Authenticator> = Arc::clone(&api_key_store) as _;

    let namespace_ids: Vec<_> = config.namespaces.keys().cloned().collect();
    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        namespace_ids,
        Arc::clone(&shared_cfg),
        ManagerConfig {
            delivery_channel_capacity: config.tcp.subscriber_channel_buffer,
            ..ManagerConfig::default()
        },
    ));

    // Spawn auth reload watcher — rebuilds API key store when config changes.
    let _auth_reload_handle =
        api_key_store.spawn_reload_watcher(Arc::clone(&shared_cfg), reload_notify_rx);

    // -- Lifecycle + Servers --

    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    let message_id_counter = Arc::new(AtomicU64::new(1));

    // Build reload state for POST /admin/reload (Phase 17.6).
    // Only available when a config file was loaded (not compiled defaults).
    // Created after lifecycle so drain-awareness is wired from the start.
    let reload_state = config_path_arc.as_ref().map(|path| {
        Arc::new(
            ReloadState::new(
                Arc::clone(path),
                Arc::clone(&shared_cfg),
                config.clone(),
                reload_notify_tx,
                args.clone(),
            )
            .with_lifecycle(Arc::clone(&lifecycle)),
        )
    });

    // Wire namespace connection checker for reload safety checks.
    if let Some(ref rs) = reload_state {
        let mgr = Arc::clone(&manager);
        let checker: fafb_q::config::NamespaceConnectionChecker = Arc::new(move |ns| {
            let mgr = Arc::clone(&mgr);
            let ns = ns.clone();
            Box::pin(async move { mgr.active_connections_for_namespace(&ns).await })
        });
        let _ = rs.ns_checker.set(checker);
    }

    // Channel for HTTP-triggered drain requests.
    let (drain_tx, mut drain_rx) = tokio::sync::mpsc::channel::<DrainRequest>(1);

    // Build TLS providers for both listeners. When the resolved configs are
    // identical (the common case — no per-listener overrides), share a single
    // provider so cert hot-reload only needs to happen once.
    let http_tls_config = config.tls.resolved_for_listener(
        config.http.tls_cert_file.as_ref(),
        config.http.tls_key_file.as_ref(),
        config.http.tls_client_ca_file.as_ref(),
    );
    let tcp_tls_config = config.tls.resolved_for_listener(
        config.tcp.tls_cert_file.as_ref(),
        config.tcp.tls_key_file.as_ref(),
        config.tcp.tls_client_ca_file.as_ref(),
    );
    let shared_tls = http_tls_config == tcp_tls_config;
    let http_tls_provider = match TlsProvider::from_config(&http_tls_config) {
        Ok(Some(provider)) => Some(Arc::new(provider)),
        Ok(None) => None,
        Err(e) => {
            tracing::error!("HTTP TLS configuration error: {e}");
            process::exit(1);
        }
    };
    let tcp_tls_provider = if shared_tls {
        if http_tls_provider.is_some() {
            tracing::info!("TCP and HTTP listeners share TLS configuration");
        }
        http_tls_provider.clone()
    } else {
        match TlsProvider::from_config(&tcp_tls_config) {
            Ok(Some(provider)) => Some(Arc::new(provider)),
            Ok(None) => None,
            Err(e) => {
                tracing::error!("TCP TLS configuration error: {e}");
                process::exit(1);
            }
        }
    };

    // Collect labeled TLS providers for hot-reload (admin endpoint, SIGHUP, background timer).
    // When shared, this is a single "shared" provider; when separate per-listener, two.
    let tls_providers: Vec<LabeledTlsProvider> = if shared_tls {
        http_tls_provider
            .iter()
            .map(|p| LabeledTlsProvider {
                label: "shared",
                provider: Arc::clone(p),
            })
            .collect()
    } else {
        let mut v = Vec::new();
        if let Some(p) = &http_tls_provider {
            v.push(LabeledTlsProvider {
                label: "http",
                provider: Arc::clone(p),
            });
        }
        if let Some(p) = &tcp_tls_provider {
            v.push(LabeledTlsProvider {
                label: "tcp",
                provider: Arc::clone(p),
            });
        }
        v
    };

    // Shared handle for topics in the reap grace period.
    // Created here so that AppState and every RetentionTask respawn share the same Arc.
    let grace_pending: GracePendingHandle =
        Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new()));

    // Start HTTP server BEFORE engine.init() so health probes work during startup.
    let deploy_config = config.deploy.clone();
    let http_state = AppState {
        manager: Arc::clone(&manager),
        auth: Arc::clone(&auth),
        message_id_counter: Arc::clone(&message_id_counter),
        prom_handle,
        log_reload_handle: Arc::new(log_reload_handle),
        lifecycle: Arc::clone(&lifecycle),
        admin_api_key: deploy_config.admin_api_key.clone(),
        drain_trigger: Some(drain_tx),
        deploy_config,
        shared_config: Arc::clone(&shared_cfg),
        config_path: config_path_arc.clone(),
        reload_state: reload_state.clone(),
        grace_pending: Arc::clone(&grace_pending),
    };
    let http_server = HttpServer::new(config.http.clone(), http_state, http_tls_provider)
        .with_tls_providers(TlsProviders(tls_providers.clone()));
    let mut http_handle = tokio::spawn(async move { http_server.run().await });

    // Now run the slow initialization (storage + engine replay).
    // HTTP /health/live already returns 200; /health/ready returns 503 "initializing".
    // TCP accept loop waits for lifecycle.is_initialized().
    if let Err(e) = storage.init().await {
        tracing::error!("storage init failed: {e}");
        process::exit(1);
    }
    if let Err(e) = engine.init().await {
        tracing::error!("queue engine init failed: {e}");
        process::exit(1);
    }
    tracing::info!("storage and queue engine initialized");
    lifecycle.set_initialized();

    let tcp_server = TcpServer::new(
        config.tcp.clone(),
        config.deploy.clone(),
        Arc::clone(&shared_cfg),
        Arc::clone(&manager),
        Arc::clone(&auth),
        Arc::clone(&message_id_counter),
        Arc::clone(&lifecycle),
        tcp_tls_provider,
    );

    // -- Spawn servers + await shutdown signal --

    let mut tcp_handle = tokio::spawn(async move { tcp_server.run().await });

    // -- Background task spawn helpers --
    //
    // Each returns a JoinHandle so the main loop can monitor and respawn on panic.

    let spawn_retention = {
        let storage = Arc::clone(&storage);
        let manager = Arc::clone(&manager);
        let engine = Arc::clone(&engine);
        let shared_cfg = Arc::clone(&shared_cfg);
        let lifecycle = Arc::clone(&lifecycle);
        let grace_pending = Arc::clone(&grace_pending);
        move || {
            let task = RetentionTask::new(
                Arc::clone(&storage),
                Arc::clone(&manager),
                Arc::clone(&engine),
                Arc::clone(&shared_cfg),
            )
            .with_grace_pending(Arc::clone(&grace_pending));
            tokio::spawn(task.run(lifecycle.subscribe()))
        }
    };

    let spawn_cleanup = {
        let storage = Arc::clone(&storage);
        let manager = Arc::clone(&manager);
        let engine = Arc::clone(&engine);
        let shared_cfg = Arc::clone(&shared_cfg);
        let lifecycle = Arc::clone(&lifecycle);
        move || {
            let storage = Arc::clone(&storage);
            let manager = Arc::clone(&manager);
            let engine = Arc::clone(&engine);
            let shared_cfg = Arc::clone(&shared_cfg);
            let mut shutdown_rx = lifecycle.subscribe();
            tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_secs(60));
                interval.tick().await; // skip first immediate tick
                loop {
                    tokio::select! {
                        _ = shutdown_rx.changed() => break,
                        _ = interval.tick() => {
                            let max_segments = shared_cfg.load().max_segments_per_topic;
                            let topics = match storage.list_topics().await {
                                Ok(t) => t,
                                Err(e) => {
                                    tracing::warn!(error = %e, "cleanup: list_topics failed");
                                    continue;
                                }
                            };
                            let mut seen_ns = std::collections::HashSet::new();

                            for (ns, topic) in &topics {
                                seen_ns.insert(ns.clone());
                                let safe_below = manager.min_live_offset(ns, topic).await;

                                if let Some(floor) = safe_below
                                    && let Err(e) = engine.compact(ns, topic, floor).await {
                                        tracing::warn!(namespace = %ns, topic = %topic,
                                            error = %e, "cleanup: compact failed");
                                        continue;
                                    }

                                match storage.delete_segments(ns, topic, None, Some(max_segments), safe_below).await {
                                    Ok(n) if n > 0 => {
                                        metrics::counter!("fafb_segments_deleted_total",
                                            "namespace" => ns.0.clone(), "topic" => topic.0.clone()
                                        ).increment(n as u64);
                                        tracing::info!(namespace = %ns, topic = %topic, count = n,
                                            "segments cleaned up");
                                    }
                                    Err(e) => tracing::warn!(namespace = %ns, topic = %topic,
                                        error = %e, "cleanup failed"),
                                    _ => {}
                                }
                            }

                            for ns in &seen_ns {
                                let bytes = engine.namespace_disk_bytes(ns).await;
                                metrics::gauge!("fafb_namespace_disk_bytes",
                                    "namespace" => ns.0.clone()
                                ).set(bytes as f64);
                            }
                        }
                    }
                }
            })
        }
    };

    let spawn_compaction = {
        let storage = Arc::clone(&storage);
        let manager = Arc::clone(&manager);
        let shared_cfg = Arc::clone(&shared_cfg);
        let lifecycle = Arc::clone(&lifecycle);
        move || {
            let storage = Arc::clone(&storage);
            let manager = Arc::clone(&manager);
            let shared_cfg = Arc::clone(&shared_cfg);
            let mut shutdown_rx = lifecycle.subscribe();
            tokio::spawn(async move {
                let cfg = shared_cfg.load();
                let merge_threshold = cfg.segment_merge_threshold_bytes;
                let interval_secs = cfg.compaction_interval_secs;
                drop(cfg);
                if merge_threshold == 0 {
                    tracing::info!("compaction task: merge disabled (threshold=0), exiting");
                    return;
                }
                tracing::info!(interval_secs, merge_threshold, "compaction task started");
                let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
                interval.tick().await; // skip first immediate tick
                loop {
                    tokio::select! {
                        _ = shutdown_rx.changed() => break,
                        _ = interval.tick() => {
                            // Re-read merge threshold on each tick for hot-reload.
                            let merge_threshold = shared_cfg.load().segment_merge_threshold_bytes;
                            if merge_threshold == 0 {
                                continue; // compaction dynamically disabled
                            }
                            let sweep_start = std::time::Instant::now();
                            let topics = match storage.list_topics().await {
                                Ok(t) => t,
                                Err(e) => {
                                    tracing::warn!(error = %e, "compaction: list_topics failed");
                                    continue;
                                }
                            };
                            let mut total_merges = 0u64;
                            for (ns, topic) in &topics {
                                let safe_below = manager.min_live_offset(ns, topic).await;
                                match storage.compact_segments(ns, topic, merge_threshold, safe_below).await {
                                    Ok(n) if n > 0 => {
                                        metrics::counter!("fafb_segments_merged_total",
                                            "namespace" => ns.0.clone(),
                                            "topic" => topic.0.clone()
                                        ).increment(n as u64);
                                        total_merges += n as u64;
                                    }
                                    Err(e) => tracing::warn!(
                                        namespace = %ns, topic = %topic,
                                        error = %e, "compaction failed"
                                    ),
                                    _ => {}
                                }
                            }
                            let elapsed = sweep_start.elapsed();
                            metrics::histogram!("fafb_compaction_sweep_duration_seconds")
                                .record(elapsed.as_secs_f64());
                            if total_merges > 0 {
                                tracing::info!(
                                    merges = total_merges,
                                    elapsed_ms = elapsed.as_millis() as u64,
                                    "compaction sweep complete"
                                );
                            }
                        }
                    }
                }
                tracing::info!("compaction task stopped");
            })
        }
    };

    // -- Initial spawns --

    let mut retention_handle: Option<tokio::task::JoinHandle<()>> = Some(spawn_retention());
    let mut cleanup_handle: Option<tokio::task::JoinHandle<()>> = Some(spawn_cleanup());
    let mut compaction_handle: Option<tokio::task::JoinHandle<()>> = Some(spawn_compaction());

    // Set initial TLS cert expiry gauge per listener.
    for lp in &tls_providers {
        metrics::gauge!("fafb_tls_cert_expiry_seconds", "listener" => lp.label)
            .set(lp.provider.cert_ttl_secs() as f64);
    }

    // Background cert reload task — checks for cert file changes on a timer.
    let spawn_cert_reload = {
        let tls_providers = tls_providers.clone();
        let reload_interval = config.tls.reload_interval_secs;
        let lifecycle = Arc::clone(&lifecycle);
        move || {
            let tls_providers = tls_providers.clone();
            let mut shutdown_rx = lifecycle.subscribe();
            tokio::spawn(async move {
                if reload_interval == 0 || tls_providers.is_empty() {
                    tracing::info!("cert reload task: disabled (interval=0 or TLS off), exiting");
                    return;
                }
                tracing::info!(interval_secs = reload_interval, "cert reload task started");
                let mut interval = tokio::time::interval(Duration::from_secs(reload_interval));
                interval.tick().await; // skip first immediate tick
                loop {
                    tokio::select! {
                        _ = shutdown_rx.changed() => break,
                        _ = interval.tick() => {
                            reload_tls_providers(&tls_providers);
                        }
                    }
                }
                tracing::info!("cert reload task stopped");
            })
        }
    };
    let mut cert_reload_handle: Option<tokio::task::JoinHandle<()>> = Some(spawn_cert_reload());

    // Spawn config file watcher — monitors config.toml for changes with debounce.
    // Only runs when hot_reload.watch_file is true AND a config file was loaded.
    let spawn_config_watcher = {
        let reload_state = reload_state.clone();
        let debounce_secs = config.hot_reload.debounce_secs;
        let lifecycle = Arc::clone(&lifecycle);
        move || -> Option<tokio::task::JoinHandle<()>> {
            let rs = reload_state.as_ref()?.clone();
            Some(fafb_q::config::spawn_config_watcher(
                rs,
                debounce_secs,
                Arc::clone(&lifecycle),
            ))
        }
    };
    let mut config_watcher_handle: Option<tokio::task::JoinHandle<()>> =
        if config.hot_reload.watch_file {
            spawn_config_watcher()
        } else {
            None
        };

    // Spawn log file cleanup task — enforces max_age_days and max_total_size_mb.
    if config.logging.file_enabled {
        let log_dir = config.logging.file_dir.clone();
        let log_prefix = config.logging.file_prefix.clone();
        let max_age_days = config.logging.max_age_days;
        let max_total_bytes = config.logging.max_total_size_mb * 1024 * 1024;

        // Run once immediately on startup to clean up stale logs from a crash.
        {
            let dir = log_dir.clone();
            let prefix = log_prefix.clone();
            if let Err(e) = tokio::task::spawn_blocking(move || {
                cleanup_log_files(&dir, &prefix, max_age_days, max_total_bytes)
            })
            .await
            .unwrap_or_else(|e| Err(std::io::Error::other(e)))
            {
                tracing::warn!(error = %e, "initial log file cleanup failed");
            }
        }

        let mut log_cleanup_shutdown_rx = lifecycle.subscribe();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(3600));
            interval.tick().await; // skip first immediate tick
            loop {
                tokio::select! {
                    _ = log_cleanup_shutdown_rx.changed() => break,
                    _ = interval.tick() => {
                        let dir = log_dir.clone();
                        let prefix = log_prefix.clone();
                        let result = tokio::task::spawn_blocking(move || {
                            cleanup_log_files(&dir, &prefix, max_age_days, max_total_bytes)
                        }).await;
                        match result {
                            Ok(Err(e)) => tracing::warn!(error = %e, "log file cleanup failed"),
                            Err(e) => tracing::warn!(error = %e, "log file cleanup task panicked"),
                            _ => {}
                        }
                    }
                }
            }
        });
    }

    tracing::info!(
        tcp = %format!("{}:{}", config.tcp.bind_address, config.tcp.port),
        http = %format!("{}:{}", config.http.bind_address, config.http.port),
        namespaces = config.namespaces.len(),
        "server ready — accepting connections"
    );

    // -- Main monitoring loop --
    //
    // Signals and server failures trigger shutdown. Background task panics
    // are logged and the task is respawned with exponential backoff.

    let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
        .expect("failed to register SIGTERM handler");
    let mut sighup = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())
        .expect("failed to register SIGHUP handler");

    // Track consecutive panics for backoff (resets on clean ticks).
    let mut retention_failures: u32 = 0;
    let mut cleanup_failures: u32 = 0;
    let mut compaction_failures: u32 = 0;
    let mut cert_reload_failures: u32 = 0;
    let mut config_watcher_failures: u32 = 0;

    // Drain override state — set when drain is triggered via HTTP admin API.
    let mut drain_override_timeout: Option<u64> = None;
    let mut drain_response_tx: Option<tokio::sync::oneshot::Sender<Result<DrainResponse, String>>> =
        None;

    let exit_code = loop {
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {
                tracing::info!("received SIGINT, shutting down");
                break 0;
            }
            _ = sigterm.recv() => {
                tracing::info!("received SIGTERM, shutting down");
                break 0;
            }
            _ = sighup.recv() => {
                tracing::info!("received SIGHUP, reloading TLS certificates");
                reload_tls_providers(&tls_providers);
            }
            Some(req) = drain_rx.recv() => {
                tracing::info!("received drain request via HTTP admin API");
                if lifecycle.is_draining() {
                    let _ = req.response.send(Err("already draining".to_owned()));
                    continue;
                }
                drain_override_timeout = req.timeout_secs;
                drain_response_tx = Some(req.response);
                break 0;
            }
            result = &mut tcp_handle => {
                match result {
                    Ok(Err(e)) => tracing::error!("TCP server error: {e}"),
                    Err(e) => tracing::error!("TCP server task panicked: {e}"),
                    _ => {}
                }
                tracing::warn!("TCP server exited unexpectedly");
                break 1;
            }
            result = &mut http_handle => {
                match result {
                    Ok(Err(e)) => tracing::error!("HTTP server error: {e}"),
                    Err(e) => tracing::error!("HTTP server task panicked: {e}"),
                    _ => {}
                }
                tracing::warn!("HTTP server exited unexpectedly");
                break 1;
            }
            Some(result) = option_join(&mut retention_handle) => {
                retention_handle = None;
                match result {
                    Err(e) => {
                        retention_failures += 1;
                        let delay = backoff_delay(retention_failures);
                        tracing::error!(
                            error = %e,
                            attempt = retention_failures,
                            respawn_delay_secs = delay.as_secs(),
                            "retention task panicked, respawning"
                        );
                        tokio::time::sleep(delay).await;
                        retention_handle = Some(spawn_retention());
                    }
                    Ok(()) => {
                        tracing::info!("retention task exited cleanly");
                    }
                }
            }
            Some(result) = option_join(&mut cleanup_handle) => {
                cleanup_handle = None;
                match result {
                    Err(e) => {
                        cleanup_failures += 1;
                        let delay = backoff_delay(cleanup_failures);
                        tracing::error!(
                            error = %e,
                            attempt = cleanup_failures,
                            respawn_delay_secs = delay.as_secs(),
                            "cleanup task panicked, respawning"
                        );
                        tokio::time::sleep(delay).await;
                        cleanup_handle = Some(spawn_cleanup());
                    }
                    Ok(()) => {
                        tracing::info!("cleanup task exited cleanly");
                    }
                }
            }
            Some(result) = option_join(&mut compaction_handle) => {
                compaction_handle = None;
                match result {
                    Err(e) => {
                        compaction_failures += 1;
                        let delay = backoff_delay(compaction_failures);
                        tracing::error!(
                            error = %e,
                            attempt = compaction_failures,
                            respawn_delay_secs = delay.as_secs(),
                            "compaction task panicked, respawning"
                        );
                        tokio::time::sleep(delay).await;
                        compaction_handle = Some(spawn_compaction());
                    }
                    Ok(()) => {
                        // Normal exit — e.g. merge disabled (threshold=0).
                        tracing::info!("compaction task exited cleanly");
                    }
                }
            }
            Some(result) = option_join(&mut cert_reload_handle) => {
                cert_reload_handle = None;
                match result {
                    Err(e) => {
                        cert_reload_failures += 1;
                        let delay = backoff_delay(cert_reload_failures);
                        tracing::error!(
                            error = %e,
                            attempt = cert_reload_failures,
                            respawn_delay_secs = delay.as_secs(),
                            "cert reload task panicked, respawning"
                        );
                        tokio::time::sleep(delay).await;
                        cert_reload_handle = Some(spawn_cert_reload());
                    }
                    Ok(()) => {
                        tracing::info!("cert reload task exited cleanly");
                    }
                }
            }
            Some(result) = option_join(&mut config_watcher_handle) => {
                // Config watcher exited — respawn with backoff if it panicked.
                config_watcher_handle = None;
                match result {
                    Err(e) => {
                        config_watcher_failures += 1;
                        let delay = backoff_delay(config_watcher_failures);
                        tracing::error!(
                            error = %e,
                            attempt = config_watcher_failures,
                            respawn_delay_secs = delay.as_secs(),
                            "config file watcher panicked, respawning"
                        );
                        tokio::time::sleep(delay).await;
                        config_watcher_handle = spawn_config_watcher();
                    }
                    Ok(()) => {
                        tracing::info!("config file watcher exited cleanly");
                    }
                }
            }
        }
    };

    // -- Graceful shutdown: force-flush → Draining → wait → ShuttingDown → exit --

    let drain_timeout_secs = drain_override_timeout.unwrap_or(config.deploy.drain_timeout_secs);
    let drain_timeout = Duration::from_secs(drain_timeout_secs);
    let shutdown_timeout = Duration::from_secs(config.deploy.shutdown_timeout_secs);

    // 1. Force-flush storage batch buffer for durability before clients disconnect.
    if let Err(e) = engine.force_flush().await {
        tracing::error!("pre-drain force flush failed: {e}");
    }

    // 2. Transition to Draining — TCP accept loop stops, /health/ready → 503.
    lifecycle.begin_drain(drain_timeout_secs);
    tracing::info!(
        active_connections = lifecycle.active_connections(),
        drain_timeout_secs = drain_timeout.as_secs(),
        "entering drain mode"
    );

    // Accelerate inflight timeout so unacked messages resolve before shutdown.
    let accelerated = drain_timeout.as_secs().saturating_div(2).clamp(1, 30);
    manager
        .set_inflight_timeout(Duration::from_secs(accelerated))
        .await;

    // If drain was triggered via HTTP, send the response back to the handler.
    if let Some(resp_tx) = drain_response_tx.take() {
        let _ = resp_tx.send(Ok(DrainResponse {
            active_connections: lifecycle.active_connections(),
            drain_started_wall: lifecycle.drain_started_wall().unwrap_or(0),
            drain_deadline_wall: lifecycle.drain_deadline_wall().unwrap_or(0),
        }));
    }

    // 3. Wait for drain, record metric, transition to ShuttingDown.
    lifecycle.run_drain_wait(drain_timeout).await;

    if tokio::time::timeout(shutdown_timeout, async {
        if let Err(e) = manager.shutdown().await {
            tracing::error!("namespace manager shutdown error: {e}");
        }
        if let Err(e) = engine.shutdown().await {
            tracing::error!("queue engine shutdown error: {e}");
        }
    })
    .await
    .is_err()
    {
        tracing::error!(
            "graceful shutdown timed out after {}s, forcing exit",
            shutdown_timeout.as_secs()
        );
        process::exit(1);
    }

    tracing::info!("shutdown complete");
    process::exit(exit_code);
}

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

    #[test]
    fn backoff_delay_starts_at_one_second() {
        assert_eq!(backoff_delay(1), Duration::from_secs(1));
    }

    #[test]
    fn backoff_delay_doubles() {
        assert_eq!(backoff_delay(2), Duration::from_secs(2));
        assert_eq!(backoff_delay(3), Duration::from_secs(4));
        assert_eq!(backoff_delay(4), Duration::from_secs(8));
    }

    #[test]
    fn backoff_delay_caps_at_60s() {
        assert_eq!(backoff_delay(7), Duration::from_secs(60));
        assert_eq!(backoff_delay(10), Duration::from_secs(60));
        assert_eq!(backoff_delay(100), Duration::from_secs(60));
    }

    #[test]
    fn backoff_delay_zero_failures() {
        // 0 failures should still produce a valid duration
        let d = backoff_delay(0);
        assert!(d.as_secs() <= 60);
    }

    #[test]
    fn backoff_delay_overflow_safe() {
        // u32::MAX should not panic
        let d = backoff_delay(u32::MAX);
        assert_eq!(d, Duration::from_secs(60));
    }

    #[test]
    fn cleanup_log_files_keeps_recent_files() {
        let tmp = tempfile::TempDir::new().unwrap();
        let dir = tmp.path();

        // Create recent log files (mtime = now, well within 14-day limit)
        let path_a = dir.join("fafb-q.2099-01-01.log");
        let path_b = dir.join("fafb-q.2099-01-02.log");
        fs::write(&path_a, "log data a").unwrap();
        fs::write(&path_b, "log data b").unwrap();

        let result = cleanup_log_files(dir.to_str().unwrap(), "fafb-q", 14, 1024 * 1024);
        assert!(result.is_ok());

        // Both should still exist (they're recent)
        assert!(path_a.exists());
        assert!(path_b.exists());
    }

    #[test]
    fn cleanup_log_files_enforces_size_limit() {
        let tmp = tempfile::TempDir::new().unwrap();
        let dir = tmp.path();

        // Create 3 log files, each 100 bytes
        for i in 0..3 {
            let path = dir.join(format!("fafb-q.2099-01-0{i}.log"));
            fs::write(&path, vec![b'x'; 100]).unwrap();
        }

        // Limit to 150 bytes — should delete oldest until under limit
        let result = cleanup_log_files(dir.to_str().unwrap(), "fafb-q", 365, 150);
        assert!(result.is_ok());

        let remaining: Vec<_> = fs::read_dir(dir)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| {
                e.path()
                    .file_name()
                    .and_then(|n| n.to_str())
                    .is_some_and(|n| n.ends_with(".log"))
            })
            .collect();

        // At most 1-2 files should remain (150 bytes fits at most 1 file of 100)
        assert!(remaining.len() <= 2);
    }

    #[test]
    fn cleanup_log_files_ignores_non_matching_files() {
        let tmp = tempfile::TempDir::new().unwrap();
        let dir = tmp.path();

        // Create a non-matching file
        let other_path = dir.join("other.txt");
        fs::write(&other_path, "not a log").unwrap();

        // Create a matching file
        let log_path = dir.join("fafb-q.2099-01-01.log");
        fs::write(&log_path, "log data").unwrap();

        let result = cleanup_log_files(dir.to_str().unwrap(), "fafb-q", 365, 1024 * 1024);
        assert!(result.is_ok());

        // Both should still exist (non-matching file untouched, log file within limits)
        assert!(other_path.exists());
        assert!(log_path.exists());
    }

    #[test]
    fn cleanup_log_files_handles_missing_dir() {
        let result = cleanup_log_files("/nonexistent/dir", "fafb-q", 14, 1024);
        assert!(result.is_err());
    }

    #[test]
    fn completions_generates_nonempty_output() {
        use clap::CommandFactory;

        for shell in [
            clap_complete::Shell::Bash,
            clap_complete::Shell::Zsh,
            clap_complete::Shell::Fish,
        ] {
            let mut buf = Vec::new();
            let mut cmd = Cli::command();
            clap_complete::generate(shell, &mut cmd, "fafb-q", &mut buf);
            assert!(
                !buf.is_empty(),
                "completions for {shell:?} should not be empty"
            );
            // Completions should contain the binary name.
            let output = String::from_utf8(buf).expect("completions should be valid UTF-8");
            assert!(
                output.contains("fafb-q"),
                "completions for {shell:?} should contain 'fafb-q'"
            );
        }
    }

    #[test]
    fn color_flag_parses_all_variants() {
        use clap::Parser;

        // --color=auto (default) — global flags go after the subcommand
        // due to `args_conflicts_with_subcommands`.
        let cli = Cli::try_parse_from(["fafb-q", "health", "--color", "auto"]).unwrap();
        assert_eq!(cli.global.color, fafb_q::cli::ColorMode::Auto);

        // --color=always
        let cli = Cli::try_parse_from(["fafb-q", "health", "--color", "always"]).unwrap();
        assert_eq!(cli.global.color, fafb_q::cli::ColorMode::Always);

        // --color=never
        let cli = Cli::try_parse_from(["fafb-q", "health", "--color", "never"]).unwrap();
        assert_eq!(cli.global.color, fafb_q::cli::ColorMode::Never);
    }
}
