heliosdb-nano 3.37.3

//! `HeliosDB` Lite server binary

#![allow(
    clippy::uninlined_format_args,
    clippy::unused_async,
    clippy::too_many_lines,
    clippy::single_match_else,
    clippy::match_wildcard_for_single_variants,
    clippy::manual_string_new
)]

use clap::{Parser, Subcommand};
use heliosdb_nano::{Config, EmbeddedDatabase, Error, Result};
use std::path::PathBuf;
use tracing::info;

/// HA Replication Configuration
#[derive(Debug, Clone)]
struct HAConfig {
    role: String,
    replication_port: u16,
    primary_host: Option<String>,
    standby_hosts: Option<String>,
    observer_hosts: Option<String>,
    sync_mode: String,
    http_port: u16,
    http_listen: Option<String>,
    mcp_token: Option<String>,
    allow_remote_mcp: bool,
    node_id: Option<String>,
}

#[derive(Parser)]
#[command(name = "heliosdb-nano")]
#[command(version = env!("CARGO_PKG_VERSION"))]
#[command(about = "PostgreSQL & MySQL compatible database with vector search and encryption", long_about = None)]
struct Cli {
    #[command(subcommand)]
    command: Commands,
}

#[derive(Subcommand)]
#[allow(clippy::large_enum_variant)]
enum Commands {
    /// Start the database server
    Start {
        /// Data directory (required unless --memory is used)
        #[arg(short, long)]
        data_dir: Option<PathBuf>,

        /// Use in-memory mode
        #[arg(short, long)]
        memory: bool,

        /// Port to listen on
        #[arg(short, long, default_value = "5432")]
        port: u16,

        /// Listen address
        #[arg(long, default_value = "127.0.0.1")]
        listen: String,

        /// Config file
        #[arg(short, long)]
        config: Option<PathBuf>,

        /// Run server in background (daemon mode)
        #[arg(long)]
        daemon: bool,

        /// PID file location (only used with --daemon)
        #[arg(long, default_value = "./heliosdb.pid")]
        pid_file: PathBuf,

        /// Dump on shutdown (only for --memory mode)
        #[arg(long)]
        dump_on_shutdown: bool,

        /// Dump schedule (cron syntax, e.g., "0 */6 * * *")
        #[arg(long)]
        dump_schedule: Option<String>,

        /// TLS certificate file path (PEM format)
        #[arg(long)]
        tls_cert: Option<PathBuf>,

        /// TLS private key file path (PEM format)
        #[arg(long)]
        tls_key: Option<PathBuf>,

        /// Authentication method: trust, password, md5, scram-sha-256
        #[arg(long, default_value = "trust")]
        auth: String,

        /// Password for authentication (required for password/md5/scram-sha-256 auth)
        #[arg(long)]
        password: Option<String>,

        // ========== HA Replication Options ==========
        /// Replication role: standalone, primary, standby, observer
        #[arg(long, default_value = "standalone")]
        replication_role: String,

        /// Replication port for WAL streaming (default: 5433)
        #[arg(long, default_value = "5433")]
        replication_port: u16,

        /// Primary host for standbys to connect to (host:port)
        #[arg(long)]
        primary_host: Option<String>,

        /// Standby hosts for primary to track (comma-separated host:port)
        #[arg(long)]
        standby_hosts: Option<String>,

        /// Observer hosts for split-brain protection (comma-separated host:port)
        #[arg(long)]
        observer_hosts: Option<String>,

        /// Sync mode: async, semi-sync, sync
        #[arg(long, default_value = "async")]
        sync_mode: String,

        /// HTTP API port for health checks and MCP-over-HTTP (default: 8080, 0 disables HTTP)
        #[arg(long, default_value = "8080")]
        http_port: u16,

        /// HTTP API listen address. Defaults to --listen; use 127.0.0.1 for local MCP while PG listens remotely.
        #[arg(long)]
        http_listen: Option<String>,

        /// Bearer token required for MCP-over-HTTP routes. Requires the mcp-endpoint feature.
        #[arg(long)]
        mcp_token: Option<String>,

        /// Mount MCP-over-HTTP routes on a non-loopback HTTP listener without MCP auth. Requires mcp-endpoint; unsafe.
        #[arg(long)]
        allow_remote_mcp: bool,

        /// Node ID (UUID) - auto-generated if not provided
        #[arg(long)]
        node_id: Option<String>,

        // ========== MySQL Protocol Options ==========
        /// Enable MySQL protocol listener
        #[arg(long)]
        mysql: bool,

        /// MySQL listen address (default: 127.0.0.1:3306, localhost-only for security)
        #[arg(long, default_value = "127.0.0.1:3306")]
        mysql_listen: String,

        /// MySQL Unix domain socket path (e.g. /tmp/heliosdb-mysql.sock).
        /// Enables local-only connections for PHP mysqli / WordPress embedded mode.
        #[arg(long)]
        mysql_socket: Option<PathBuf>,

        /// PostgreSQL Unix domain socket directory. If set, listens at
        /// `<dir>/.s.PGSQL.<port>` — the libpq default. Use with psql `-h /tmp`.
        #[arg(long)]
        pg_socket_dir: Option<PathBuf>,

        /// Maximum number of concurrent client connections. The server
        /// rejects new connections once this ceiling is reached. Raise it
        /// for fleets that pool many connections (see NANO-DEFICIENCIES A9).
        #[arg(long, default_value = "100")]
        max_connections: usize,
    },

    /// Stop a running server
    Stop {
        /// PID file location
        #[arg(long, default_value = "./heliosdb.pid")]
        pid_file: PathBuf,
    },

    /// Check server status
    Status {
        /// PID file location
        #[arg(long, default_value = "./heliosdb.pid")]
        pid_file: PathBuf,
    },

    /// Initialize a new database
    Init {
        /// Data directory
        #[arg(default_value = "./heliosdb-data")]
        data_dir: PathBuf,
    },

    /// Run embedded mode (REPL)
    Repl {
        /// Data directory (default: ./heliosdb-data)
        #[arg(short, long, default_value = "./heliosdb-data")]
        data_dir: PathBuf,

        /// Use in-memory database
        #[arg(short, long)]
        memory: bool,

        /// Dump on shutdown (only for --memory mode)
        #[arg(long)]
        dump_on_shutdown: bool,

        /// Dump output file (used with --dump-on-shutdown)
        #[arg(long)]
        dump_file: Option<PathBuf>,
    },

    /// Dump database to file
    Dump {
        /// Output file path
        #[arg(short, long, default_value = "backup.heliodump")]
        output: PathBuf,

        /// Data directory (required unless --connection is used)
        #[arg(short, long)]
        data_dir: Option<PathBuf>,

        /// Append incremental changes
        #[arg(short, long)]
        append: bool,

        /// Compression type (zstd, gzip, brotli, none)
        #[arg(long, default_value = "zstd")]
        compression: String,

        /// Connection string (for server mode)
        #[arg(long)]
        connection: Option<String>,

        /// Verbose output
        #[arg(short, long)]
        verbose: bool,
    },

    /// Restore database from dump file
    Restore {
        /// Input dump file path
        #[arg(short, long)]
        input: PathBuf,

        /// Target data directory
        #[arg(short, long)]
        target: Option<PathBuf>,

        /// Verify dump integrity before restore
        #[arg(long)]
        verify: bool,

        /// Connection string (for server mode)
        #[arg(long)]
        connection: Option<String>,

        /// Verbose output
        #[arg(short, long)]
        verbose: bool,
    },

    /// Code-graph (FR 2+) management. Opt-in on binaries built with
    /// `--features code-graph`.
    #[cfg(feature = "code-graph")]
    #[command(name = "code-graph")]
    CodeGraph {
        #[command(subcommand)]
        action: CodeGraphAction,
    },
}

/// Sub-actions for `heliosdb-nano code-graph`.
#[cfg(feature = "code-graph")]
#[derive(Subcommand)]
enum CodeGraphAction {
    /// Git-hook helper. Reads changed paths from stdin (one per line,
    /// as `git diff-tree --no-commit-id --name-only -r HEAD`
    /// produces), upserts each file's content into the source table
    /// and runs the code-graph indexer.
    Hook {
        /// `.helios-index/heliosdb-data` directory. Empty string ⇒
        /// in-memory (useful for dry-run / smoke tests).
        #[arg(short, long)]
        data_dir: PathBuf,
        /// Root of the repository the paths are relative to.
        #[arg(short, long, default_value = ".")]
        repo_root: PathBuf,
        /// Source-table name (default `src`). The table is created if
        /// it doesn't already exist.
        #[arg(short, long, default_value = "src")]
        source_table: String,
    },
}

#[tokio::main]
async fn main() -> Result<()> {
    // Initialize tracing
    tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .init();

    let cli = Cli::parse();

    match cli.command {
        Commands::Start {
            data_dir,
            memory,
            port,
            listen,
            config,
            daemon,
            pid_file,
            dump_on_shutdown,
            dump_schedule,
            tls_cert,
            tls_key,
            auth,
            password,
            replication_role,
            replication_port,
            primary_host,
            standby_hosts,
            observer_hosts,
            sync_mode,
            http_port,
            http_listen,
            mcp_token,
            allow_remote_mcp,
            node_id,
            mysql,
            mysql_listen,
            mysql_socket,
            pg_socket_dir,
            max_connections,
        } => {
            // Validate that either --data-dir or --memory is specified
            if !memory && data_dir.is_none() {
                return Err(Error::config(
                    "Either --data-dir or --memory must be specified. Use --help for more information.".to_string(),
                ));
            }

            // Validate TLS options
            if tls_cert.is_some() != tls_key.is_some() {
                return Err(Error::config(
                    "Both --tls-cert and --tls-key must be specified together for TLS.".to_string(),
                ));
            }

            // Validate auth options
            let auth_lower = auth.to_lowercase();
            if auth_lower != "trust" && password.is_none() {
                return Err(Error::config(format!(
                    "Authentication method '{}' requires --password to be set.",
                    auth
                )));
            }

            // Note on dump_schedule: scheduled dumps are not yet implemented
            if dump_schedule.is_some() {
                info!("Note: --dump-schedule is not yet implemented");
            }

            let resolved_data_dir = data_dir.unwrap_or_else(|| PathBuf::from("./heliosdb-data"));

            // Build HA configuration
            let ha_config = HAConfig {
                role: replication_role,
                replication_port,
                primary_host,
                standby_hosts,
                observer_hosts,
                sync_mode,
                http_port,
                http_listen,
                mcp_token,
                allow_remote_mcp,
                node_id,
            };

            if daemon {
                start_server_daemon(
                    resolved_data_dir,
                    port,
                    listen,
                    config,
                    pid_file,
                    tls_cert,
                    tls_key,
                    auth,
                    password,
                    ha_config,
                    max_connections,
                )
                .await
            } else {
                start_server(
                    resolved_data_dir,
                    port,
                    listen,
                    config,
                    memory,
                    dump_on_shutdown,
                    tls_cert,
                    tls_key,
                    auth,
                    password,
                    ha_config,
                    mysql,
                    mysql_listen,
                    mysql_socket,
                    pg_socket_dir,
                    max_connections,
                )
                .await
            }
        }
        Commands::Stop { ref pid_file } => stop_server(pid_file),
        Commands::Status { ref pid_file } => check_server_status(pid_file),
        Commands::Init { ref data_dir } => init_database(data_dir),
        Commands::Repl {
            data_dir,
            memory,
            dump_on_shutdown,
            dump_file,
        } => run_repl(data_dir, memory, dump_on_shutdown, dump_file),
        Commands::Dump {
            output,
            data_dir,
            append,
            compression,
            connection,
            verbose,
        } => {
            use heliosdb_nano::cli::DumpCommand;
            let cmd = DumpCommand {
                output,
                append,
                compression,
                connection,
                verbose,
                data_dir,
                memory: false,
            };
            cmd.execute()
        }
        Commands::Restore {
            input,
            target,
            verify,
            connection,
            verbose,
        } => {
            use heliosdb_nano::cli::RestoreCommand;
            let cmd = RestoreCommand {
                input,
                target,
                verify,
                connection,
                verbose,
            };
            cmd.execute()
        }

        #[cfg(feature = "code-graph")]
        Commands::CodeGraph { action } => match action {
            CodeGraphAction::Hook {
                data_dir,
                repo_root,
                source_table,
            } => {
                let stats = heliosdb_nano::code_graph::git_hook::run_from_stdin(&data_dir, &repo_root, &source_table)?;
                println!(
                    "code-graph hook: files_seen={} parsed={} unchanged={} skipped={} symbols={} refs={}",
                    stats.files_seen,
                    stats.files_parsed,
                    stats.files_unchanged,
                    stats.files_skipped,
                    stats.symbols_written,
                    stats.refs_written
                );
                Ok(())
            }
        },
    }
}

#[allow(clippy::too_many_arguments)]
async fn start_server(
    data_dir: PathBuf,
    port: u16,
    listen: String,
    config_path: Option<PathBuf>,
    memory_mode: bool,
    dump_on_shutdown: bool,
    tls_cert: Option<PathBuf>,
    tls_key: Option<PathBuf>,
    auth: String,
    password: Option<String>,
    ha_config: HAConfig,
    mysql_enabled: bool,
    mysql_listen: String,
    mysql_socket: Option<PathBuf>,
    pg_socket_dir: Option<PathBuf>,
    max_connections: usize,
) -> Result<()> {
    use colored::Colorize;
    use heliosdb_nano::protocol::postgres::auth::{AuthManager, AuthMethod};
    use heliosdb_nano::protocol::postgres::server::{PgServer, PgServerConfig};
    use heliosdb_nano::protocol::postgres::ssl::{SslConfig, SslMode};
    use heliosdb_nano::protocol::postgres::{InMemoryPasswordStore, PasswordStore, SharedPasswordStore};
    use heliosdb_nano::storage::{DumpCompressionType, DumpManager, DumpMode, DumpOptions};
    use std::net::SocketAddr;
    use std::sync::Arc;
    use std::time::Instant;

    let startup_time = Instant::now();

    // Print startup banner
    println!();
    println!("╔═══════════════════════════════════════════════════════════════╗");
    println!("║           HeliosDB Nano v{:<32} ║", env!("CARGO_PKG_VERSION"));
    println!("║   PostgreSQL-compatible database with enterprise features     ║");
    println!("╚═══════════════════════════════════════════════════════════════╝");
    println!();

    // Load config
    let _db_config = if let Some(ref path) = config_path {
        println!("[1/4] Loading configuration from {}...", path.display());
        Config::from_file(path.clone())?
    } else {
        println!("[1/4] Using default configuration...");
        Config::default()
    };

    // Open database (in-memory mode avoids disk I/O for all operations)
    let db = if memory_mode {
        println!("[2/4] Initializing in-memory database...");
        Arc::new(EmbeddedDatabase::new_in_memory()?)
    } else {
        println!("[2/4] Initializing database at {}...", data_dir.display());
        Arc::new(EmbeddedDatabase::new(&data_dir)?)
    };
    println!("      Database initialized successfully");

    // Configure PostgreSQL server
    let pg_addr: SocketAddr = format!("{listen}:{port}")
        .parse()
        .map_err(|e| Error::config(format!("Invalid listen address: {e}")))?;

    // Parse authentication method
    let auth_method = match auth.to_lowercase().as_str() {
        "trust" => AuthMethod::Trust,
        "password" => AuthMethod::CleartextPassword,
        "md5" => AuthMethod::Md5,
        "scram-sha-256" | "scram" => AuthMethod::ScramSha256,
        other => {
            return Err(Error::config(format!(
                "Unknown authentication method: '{}'. Use: trust, password, md5, scram-sha-256",
                other
            )))
        }
    };
    let auth_display = match &auth_method {
        AuthMethod::Trust => "Trust (development mode)",
        AuthMethod::CleartextPassword => "Cleartext Password",
        AuthMethod::Md5 => "MD5",
        AuthMethod::ScramSha256 => "SCRAM-SHA-256",
    };

    // Build server config
    let mut pg_config = PgServerConfig::with_address(pg_addr)
        .with_auth_method(auth_method)
        .with_max_connections(max_connections);

    // Configure TLS if specified
    let tls_enabled = tls_cert.is_some();
    if let (Some(cert_path), Some(key_path)) = (&tls_cert, &tls_key) {
        let ssl_config = SslConfig::new(SslMode::Prefer, cert_path, key_path);
        pg_config = pg_config.with_ssl(ssl_config);
    }

    println!("[3/4] Configuring server...");
    println!("      - Listen address: {pg_addr}");
    if ha_config.http_port != 0 {
        let http_listen = ha_config.http_listen.as_deref().unwrap_or(&listen);
        println!("      - HTTP address: {http_listen}:{}", ha_config.http_port);
    }
    println!("      - Max connections: 100");
    println!("      - Authentication: {auth_display}");
    println!("      - SSL/TLS: {}", if tls_enabled { "Enabled" } else { "Disabled" });
    if dump_on_shutdown {
        println!("      - Dump on shutdown: Enabled");
    }

    // HA Configuration
    let ha_role = ha_config.role.to_lowercase();
    if ha_role != "standalone" {
        println!("      - Replication role: {}", ha_config.role);
        println!("      - Replication port: {}", ha_config.replication_port);
        println!("      - Sync mode: {}", ha_config.sync_mode);
        if let Some(ref primary) = ha_config.primary_host {
            println!("      - Primary host: {primary}");
        }
        if let Some(ref standbys) = ha_config.standby_hosts {
            println!("      - Standby hosts: {standbys}");
        }
        if let Some(ref observers) = ha_config.observer_hosts {
            println!("      - Observer hosts: {observers}");
        }
        println!("      - HTTP health port: {}", ha_config.http_port);
    }

    // Create PostgreSQL server with appropriate auth configuration
    let pg_server = if matches!(
        auth_method,
        AuthMethod::CleartextPassword | AuthMethod::Md5 | AuthMethod::ScramSha256
    ) {
        if let Some(ref pwd) = password {
            // Create password store with default users
            let mut store = InMemoryPasswordStore::new();
            store
                .add_user("postgres", pwd)
                .map_err(|e| Error::config(format!("Failed to add user: {e}")))?;
            store
                .add_user("helios", pwd)
                .map_err(|e| Error::config(format!("Failed to add user: {e}")))?;
            let shared_store = SharedPasswordStore::new(store);
            let auth_manager = AuthManager::with_password_store(auth_method, shared_store);
            PgServer::with_auth_manager(pg_config, Arc::clone(&db), auth_manager)?
        } else {
            PgServer::new(pg_config, Arc::clone(&db))?
        }
    } else {
        PgServer::new(pg_config, Arc::clone(&db))?
    };

    println!("[4/4] Starting server...");
    println!();
    println!("════════════════════════════════════════════════════════════════");
    println!(
        "  Server ready! Started in {:.2}s",
        startup_time.elapsed().as_secs_f64()
    );
    println!("════════════════════════════════════════════════════════════════");
    println!();
    println!("  Connect using:");
    println!();
    println!("    psql:       psql -h {listen} -p {port}");
    println!("    Python:     psycopg2.connect(host='{listen}', port={port})");
    println!("    Node.js:    pg.connect({{ host: '{listen}', port: {port} }})");
    println!("    JDBC:       jdbc:postgresql://{listen}:{port}/heliosdb");
    println!();
    println!("    Compatibility notes:");
    println!("      FTS:         docs/compatibility/fts.md");
    println!("      ORM matrix:  https://github.com/HeliosDatabase/HeliosDB-Nano/blob/main/docs/compatibility/orm.md");
    println!("      Known gaps:  SELECT heliosdb_capability_report();");
    if mysql_enabled {
        println!();
        println!(
            "    mysql:      mysql -h {} -P {}",
            mysql_listen.split(':').next().unwrap_or("127.0.0.1"),
            mysql_listen.split(':').nth(1).unwrap_or("3306")
        );
        println!(
            "    PyMySQL:    pymysql.connect(host='{}', port={})",
            mysql_listen.split(':').next().unwrap_or("127.0.0.1"),
            mysql_listen.split(':').nth(1).unwrap_or("3306")
        );
    }
    println!();
    println!(
        "  For REPL mode (single-user):  heliosdb-nano repl -d {}",
        data_dir.display()
    );
    println!();
    println!("  Press Ctrl+C to shut down");
    println!("────────────────────────────────────────────────────────────────");
    println!();

    // Log for tracing subscribers
    info!("HeliosDB Nano server listening on {}", pg_addr);

    // Start HA components if enabled
    #[cfg(feature = "ha-tier1")]
    let _ha_handles = if ha_role != "standalone" {
        start_ha_components(&ha_config, &listen, port, db.storage.clone()).await?
    } else {
        HAHandles::default()
    };

    // Start HTTP endpoint (health checks, plus MCP routes when compiled in).
    //
    // KanttBan bug #2 (v3.27.0): a `--http-port` collision used to take
    // the entire server process down silently — `start_health_server`
    // returned an `Err`, the future completed early, the `tokio::select!`
    // health-server arm fired, and shutdown happened right after the
    // "Server ready!" banner. Now: bind eagerly so we surface the
    // failure at ERROR before the banner, and run the accept loop in a
    // detached task that the main `select!` never observes — so a
    // late-life health failure cannot tear down the database listener.
    // `--http-port 0` opts out (no HTTP endpoint, no listener).
    let http_health_disabled = ha_config.http_port == 0;
    if !http_health_disabled {
        let http_listen = ha_config.http_listen.as_deref().unwrap_or(&listen);
        let http_addr: SocketAddr = format!("{}:{}", http_listen, ha_config.http_port)
            .parse()
            .map_err(|e| Error::config(format!("Invalid HTTP address: {e}")))?;
        match tokio::net::TcpListener::bind(http_addr).await {
            Ok(listener) => {
                info!("HTTP endpoint at http://{} (/health)", http_addr);
                let http_db = Arc::clone(&db);
                let mcp_token = ha_config.mcp_token.clone();
                let allow_remote_mcp = ha_config.allow_remote_mcp;
                tokio::spawn(async move {
                    if let Err(e) = run_http_listener(listener, http_db, mcp_token, allow_remote_mcp).await {
                        tracing::error!("HTTP server accept loop exited: {e}");
                    }
                });
            }
            Err(e) => {
                tracing::error!(
                    "Failed to bind HTTP endpoint on {}: {} (os error {:?}). \
                     The database listener stays up; pass --http-port <free> or \
                     --http-port 0 to silence this.",
                    http_addr,
                    e,
                    e.raw_os_error(),
                );
            }
        }
    } else {
        info!("HTTP endpoint disabled (--http-port 0)");
    }

    // Start MySQL listener if enabled
    let mysql_handle = if mysql_enabled {
        let mysql_addr: SocketAddr = mysql_listen
            .parse()
            .map_err(|e| Error::config(format!("Invalid MySQL listen address '{}': {}", mysql_listen, e)))?;
        let mysql_db = Arc::clone(&db);
        info!("MySQL protocol listening on {}", mysql_addr);
        let conn_counter = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(1));
        Some(tokio::spawn(async move {
            let listener = match tokio::net::TcpListener::bind(mysql_addr).await {
                Ok(l) => l,
                Err(e) => {
                    tracing::error!("Failed to bind MySQL listener on {}: {}", mysql_addr, e);
                    return;
                }
            };
            loop {
                match listener.accept().await {
                    Ok((stream, addr)) => {
                        tracing::debug!("MySQL connection from {}", addr);
                        let db_clone = Arc::clone(&mysql_db);
                        let conn_id = conn_counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        tokio::spawn(async move {
                            if let Err(e) =
                                heliosdb_nano::protocol::mysql::handle_mysql_connection(db_clone, stream, conn_id).await
                            {
                                tracing::debug!("MySQL connection error: {}", e);
                            }
                        });
                    }
                    Err(e) => {
                        tracing::error!("MySQL accept error: {}", e);
                    }
                }
            }
        }))
    } else {
        None
    };

    // Start MySQL Unix domain socket listener if requested (local-only, no TCP)
    // Useful for PHP mysqli / WordPress embedded-mode pointing at
    // /var/run/mysqld/mysqld.sock or equivalent.
    #[cfg(unix)]
    let mysql_unix_handle = if let Some(ref socket_path) = mysql_socket {
        let path = socket_path.clone();
        // Remove stale socket file if it exists (best-effort)
        let _ = std::fs::remove_file(&path);
        let mysql_db = Arc::clone(&db);
        let conn_counter = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(1_000_000));
        info!("MySQL Unix socket listening on {}", path.display());
        println!("    mysql (UDS): mysql --socket={}", path.display());
        Some(tokio::spawn(async move {
            let listener = match tokio::net::UnixListener::bind(&path) {
                Ok(l) => l,
                Err(e) => {
                    tracing::error!("Failed to bind MySQL Unix socket {}: {}", path.display(), e);
                    return;
                }
            };
            // Permissive mode so non-root clients can connect
            let _ = std::fs::set_permissions(&path, std::os::unix::fs::PermissionsExt::from_mode(0o777));
            loop {
                match listener.accept().await {
                    Ok((stream, _)) => {
                        let db_clone = Arc::clone(&mysql_db);
                        let conn_id = conn_counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        tokio::spawn(async move {
                            if let Err(e) = heliosdb_nano::protocol::mysql::handler::handle_mysql_connection_unix(
                                db_clone, stream, conn_id,
                            )
                            .await
                            {
                                tracing::debug!("MySQL UDS connection error: {}", e);
                            }
                        });
                    }
                    Err(e) => {
                        tracing::error!("MySQL UDS accept error: {}", e);
                    }
                }
            }
        }))
    } else {
        None
    };
    #[cfg(not(unix))]
    let mysql_unix_handle: Option<tokio::task::JoinHandle<()>> = None;
    let _ = &mysql_socket; // silence unused on non-unix

    // Start PostgreSQL Unix domain socket listener if requested.
    // libpq uses `<dir>/.s.PGSQL.<port>` when host starts with `/`.
    #[cfg(unix)]
    let pg_unix_handle = if let Some(ref sock_dir) = pg_socket_dir {
        let sock_path = sock_dir.join(format!(".s.PGSQL.{}", port));
        let _ = std::fs::create_dir_all(sock_dir);
        let _ = std::fs::remove_file(&sock_path);
        let pg_db = Arc::clone(&db);
        info!("PostgreSQL Unix socket listening on {}", sock_path.display());
        println!("    psql (UDS):  psql -h {} -p {}", sock_dir.display(), port);
        Some(tokio::spawn(async move {
            let listener = match tokio::net::UnixListener::bind(&sock_path) {
                Ok(l) => l,
                Err(e) => {
                    tracing::error!("Failed to bind PG Unix socket {}: {}", sock_path.display(), e);
                    return;
                }
            };
            let _ = std::fs::set_permissions(&sock_path, std::os::unix::fs::PermissionsExt::from_mode(0o777));
            let conn_counter = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(2_000_000));
            loop {
                match listener.accept().await {
                    Ok((stream, _)) => {
                        let db_clone = Arc::clone(&pg_db);
                        let conn_id = conn_counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        tokio::spawn(async move {
                            if let Err(e) = heliosdb_nano::protocol::postgres::handler::handle_connection_unix(
                                db_clone, stream, conn_id,
                            )
                            .await
                            {
                                tracing::debug!("PG UDS connection error: {}", e);
                            }
                        });
                    }
                    Err(e) => {
                        tracing::error!("PG UDS accept error: {}", e);
                    }
                }
            }
        }))
    } else {
        None
    };
    #[cfg(not(unix))]
    let pg_unix_handle: Option<tokio::task::JoinHandle<()>> = None;
    let _ = &pg_socket_dir;

    // Start server with graceful shutdown handling. The health
    // server runs in its own detached task (see above) so a failure
    // there cannot tear the database listener down.
    tokio::select! {
        result = pg_server.serve() => {
            // Server stopped (error or normal shutdown)
            if let Err(ref e) = result {
                tracing::error!("Server error: {e}");
            }
            result?;
        }
        _ = tokio::signal::ctrl_c() => {
            println!();
            println!("{}", "Received shutdown signal...".yellow());
            if let Some(h) = mysql_handle {
                h.abort();
            }
            if let Some(h) = mysql_unix_handle {
                h.abort();
            }
            if let Some(h) = pg_unix_handle {
                h.abort();
            }
            // Best-effort unlink of Unix socket files on shutdown.
            #[cfg(unix)]
            {
                if let Some(ref p) = mysql_socket { let _ = std::fs::remove_file(p); }
                if let Some(ref d) = pg_socket_dir {
                    let p = d.join(format!(".s.PGSQL.{}", port));
                    let _ = std::fs::remove_file(p);
                }
            }
        }
    }

    // Perform dump on shutdown if enabled
    if dump_on_shutdown {
        println!();
        println!("{}", "Performing database dump before shutdown...".cyan());

        let dump_path = data_dir.join("shutdown_dump.heliodump");
        let dump_manager = DumpManager::new(data_dir.clone(), DumpCompressionType::Zstd);

        let options = DumpOptions {
            output_path: dump_path.clone(),
            mode: DumpMode::Full,
            compression: DumpCompressionType::Zstd,
            append: false,
            tables: None,
            verbose: false,
            connection: None,
            format: heliosdb_nano::storage::DumpOutputFormat::Binary,
        };

        match dump_manager.dump(&options, db.as_ref()) {
            Ok(report) => {
                println!("{}", "Shutdown dump completed successfully!".green().bold());
                println!("  Tables: {}", report.tables_dumped);
                println!("  Rows: {}", report.rows_dumped);
                println!("  Size: {} bytes (compressed)", report.bytes_written);
                println!("  Duration: {} ms", report.duration_ms);
                println!("  Output: {}", dump_path.display().to_string().cyan());
            }
            Err(e) => {
                tracing::error!("Failed to dump database on shutdown: {e}");
            }
        }
    }

    println!();
    println!("Server shutdown complete. Goodbye!");
    Ok(())
}

fn init_database(data_dir: &PathBuf) -> Result<()> {
    println!();
    println!("Initializing new HeliosDB Nano database...");
    println!("  Location: {}", data_dir.display());

    // Create directory
    std::fs::create_dir_all(data_dir)?;

    // Initialize database
    let db = EmbeddedDatabase::new(data_dir)?;

    // Close database
    db.close()?;

    println!();
    println!("Database initialized successfully!");
    println!();
    println!("Next steps:");
    println!("  Start server:    heliosdb-nano start -d {}", data_dir.display());
    println!("  Start REPL:      heliosdb-nano repl -d {}", data_dir.display());
    println!();

    Ok(())
}

fn run_repl(data_dir: PathBuf, memory: bool, dump_on_shutdown: bool, dump_file: Option<PathBuf>) -> Result<()> {
    use colored::Colorize;
    use heliosdb_nano::repl::{ReplConfig, ReplShell};
    use heliosdb_nano::storage::{DumpCompressionType, DumpManager, DumpMode, DumpOptions};

    // Open database with user-friendly output
    let db = if memory {
        println!("Starting REPL with in-memory database...");
        println!("  Note: Data will be lost when you exit.");
        EmbeddedDatabase::new_in_memory()?
    } else {
        println!("Starting REPL with persistent storage at {}...", data_dir.display());
        EmbeddedDatabase::new(&data_dir)?
    };

    // Create REPL configuration
    let config = ReplConfig::default();

    // Create and run REPL
    let mut shell = ReplShell::new(db, config)?;

    // If dump_on_shutdown is requested, handle the shutdown dump after REPL exits
    let result = shell.run();

    if dump_on_shutdown && result.is_ok() {
        let dump_path = dump_file.unwrap_or_else(|| PathBuf::from("heliosdb_dump.heliodump"));
        println!();
        println!("{}", "Dumping database on shutdown...".cyan());

        // Create dump manager (data_dir is used for metadata tracking, not for reading data)
        let dump_manager = DumpManager::new(data_dir, DumpCompressionType::Zstd);

        // Configure dump options
        let options = DumpOptions {
            output_path: dump_path.clone(),
            mode: DumpMode::Full,
            compression: DumpCompressionType::Zstd,
            append: false,
            tables: None,
            verbose: false,
            connection: None,
            format: heliosdb_nano::storage::DumpOutputFormat::Binary,
        };

        // Perform dump using the shell's database reference
        match dump_manager.dump(&options, shell.db()) {
            Ok(report) => {
                println!("{}", "Dump completed successfully!".green().bold());
                println!("  Tables: {}", report.tables_dumped);
                println!("  Rows: {}", report.rows_dumped);
                println!("  Size: {} bytes (compressed)", report.bytes_written);
                println!("  Duration: {} ms", report.duration_ms);
                println!("  Output: {}", dump_path.display().to_string().cyan());
            }
            Err(e) => {
                tracing::error!("Failed to dump database: {e}");
            }
        }
    }

    result
}

#[allow(clippy::too_many_arguments)]
async fn start_server_daemon(
    data_dir: PathBuf,
    port: u16,
    listen: String,
    config_path: Option<PathBuf>,
    pid_file: PathBuf,
    tls_cert: Option<PathBuf>,
    tls_key: Option<PathBuf>,
    auth: String,
    password: Option<String>,
    ha_config: HAConfig,
    max_connections: usize,
) -> Result<()> {
    use std::process::{Command, Stdio};

    // Check if server is already running
    if pid_file.exists() {
        if let Ok(pid_str) = std::fs::read_to_string(&pid_file) {
            if let Ok(pid) = pid_str.trim().parse::<i32>() {
                // Check if process is running
                #[cfg(unix)]
                {
                    if Command::new("kill")
                        .args(["-0", &pid.to_string()])
                        .stdout(Stdio::null())
                        .stderr(Stdio::null())
                        .status()
                        .map(|s| s.success())
                        .unwrap_or(false)
                    {
                        return Err(Error::io(format!(
                            "Server is already running (PID: {pid}). Use 'heliosdb-nano stop' to stop it."
                        )));
                    }
                }
            }
        }
    }

    println!("Starting HeliosDB server in daemon mode...");

    // Prepare command arguments
    let mut args = vec![
        "start".to_string(),
        "--data-dir".to_string(),
        data_dir.display().to_string(),
        "--port".to_string(),
        port.to_string(),
        "--listen".to_string(),
        listen.clone(),
        "--max-connections".to_string(),
        max_connections.to_string(),
    ];

    if let Some(cfg) = config_path {
        args.push("--config".to_string());
        args.push(cfg.display().to_string());
    }

    // Add TLS options
    if let Some(cert) = tls_cert {
        args.push("--tls-cert".to_string());
        args.push(cert.display().to_string());
    }
    if let Some(key) = tls_key {
        args.push("--tls-key".to_string());
        args.push(key.display().to_string());
    }

    // Add auth options
    args.push("--auth".to_string());
    args.push(auth);
    if let Some(pwd) = password {
        args.push("--password".to_string());
        args.push(pwd);
    }

    // Add HA options
    args.push("--replication-role".to_string());
    args.push(ha_config.role.clone());
    args.push("--replication-port".to_string());
    args.push(ha_config.replication_port.to_string());
    args.push("--sync-mode".to_string());
    args.push(ha_config.sync_mode.clone());
    args.push("--http-port".to_string());
    args.push(ha_config.http_port.to_string());
    if let Some(http_listen) = ha_config.http_listen {
        args.push("--http-listen".to_string());
        args.push(http_listen);
    }
    if let Some(mcp_token) = ha_config.mcp_token {
        args.push("--mcp-token".to_string());
        args.push(mcp_token);
    }
    if ha_config.allow_remote_mcp {
        args.push("--allow-remote-mcp".to_string());
    }
    if let Some(primary) = ha_config.primary_host {
        args.push("--primary-host".to_string());
        args.push(primary);
    }
    if let Some(standbys) = ha_config.standby_hosts {
        args.push("--standby-hosts".to_string());
        args.push(standbys);
    }
    if let Some(observers) = ha_config.observer_hosts {
        args.push("--observer-hosts".to_string());
        args.push(observers);
    }
    if let Some(node_id) = ha_config.node_id {
        args.push("--node-id".to_string());
        args.push(node_id);
    }

    // Fork the process
    #[cfg(unix)]
    {
        let exe = std::env::current_exe().map_err(|e| Error::io(format!("Failed to get current executable: {e}")))?;

        let mut child = Command::new(&exe)
            .args(&args)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .map_err(|e| Error::io(format!("Failed to spawn daemon process: {e}")))?;

        let pid = child.id();

        // KanttBan bug #1 (v3.27.0): the parent used to print the
        // "Daemon Started" banner unconditionally and return 0 before
        // the child had bound its port — even when the worker died at
        // startup (e.g. http-port collision, see #2). Now: poll the
        // child's PG port for up to 5 s waiting for it to be
        // accepting connections, OR observe the child's exit status
        // first. If neither happens within the deadline we fail with
        // a non-zero exit so caller scripts (`scripts/heliosdb.sh`)
        // can detect failure without polling `ss` themselves.
        let target = format!("{}:{}", listen, port);
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        let mut child_dead = false;
        loop {
            // Did the child exit prematurely?
            match child.try_wait() {
                Ok(Some(status)) => {
                    child_dead = true;
                    let _ = std::fs::remove_file(&pid_file);
                    return Err(Error::io(format!(
                        "Daemon worker exited at startup (status: {status}). \
                         Check the logs for the bind / config error and retry."
                    )));
                }
                Ok(None) => {} // Still running.
                Err(e) => {
                    tracing::warn!("Failed to poll daemon child: {e}");
                }
            }
            // Try connecting to the PG port.
            if std::net::TcpStream::connect_timeout(
                &target.parse::<std::net::SocketAddr>().unwrap_or_else(|_| {
                    // listen may be 0.0.0.0; probe 127.0.0.1 instead.
                    format!("127.0.0.1:{port}").parse().expect("valid loopback addr")
                }),
                std::time::Duration::from_millis(200),
            )
            .is_ok()
            {
                break;
            }
            if std::time::Instant::now() >= deadline {
                // Don't leave a zombie if we time out — kill the child
                // and fail loudly. Caller can retry.
                let _ = child.kill();
                let _ = std::fs::remove_file(&pid_file);
                return Err(Error::io(format!(
                    "Daemon did not become ready on {target} within 5s. \
                     Most common cause: another listener already on this port \
                     (or on the HTTP health port, see #2 in BUGS_HELIOSDB.md). \
                     Pass --port and --http-port to non-conflicting values."
                )));
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
        let _ = child_dead;

        // Detach so the daemon outlives the launching shell.
        // `child.id()` is preserved in the PID file even after we drop.
        std::mem::forget(child);

        // Write PID file (we already verified the worker is up).
        std::fs::write(&pid_file, pid.to_string()).map_err(|e| Error::io(format!("Failed to write PID file: {e}")))?;

        println!();
        println!("╔═══════════════════════════════════════════════════════════════╗");
        println!("║              HeliosDB Nano Daemon Started                      ║");
        println!("╚═══════════════════════════════════════════════════════════════╝");
        println!();
        println!("  Status:         RUNNING");
        println!("  PID:            {pid}");
        println!("  Address:        {listen}:{port}");
        println!("  Data directory: {}", data_dir.display());
        println!("  PID file:       {}", pid_file.display());
        println!();
        println!("  Connect using:");
        println!("    psql -h {listen} -p {port}");
        println!();
        println!("  Management commands:");
        println!("    heliosdb-nano status     Check server status");
        println!("    heliosdb-nano stop       Stop the server");
        println!();

        Ok(())
    }

    #[cfg(not(unix))]
    {
        Err(Error::io("Daemon mode is only supported on Unix systems"))
    }
}

fn stop_server(pid_file: &PathBuf) -> Result<()> {
    if !pid_file.exists() {
        return Err(Error::io(format!(
            "PID file not found: {}. Server may not be running.",
            pid_file.display()
        )));
    }

    let pid_str = std::fs::read_to_string(pid_file).map_err(|e| Error::io(format!("Failed to read PID file: {e}")))?;

    let pid = pid_str
        .trim()
        .parse::<i32>()
        .map_err(|e| Error::io(format!("Invalid PID in file: {e}")))?;

    println!("Stopping HeliosDB server (PID: {pid})...");

    #[cfg(unix)]
    {
        use std::process::{Command, Stdio};

        // Send SIGTERM to gracefully stop the server
        let status = Command::new("kill")
            .args(["-TERM", &pid.to_string()])
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .map_err(|e| Error::io(format!("Failed to send signal: {e}")))?;

        if status.success() {
            // Wait a bit for graceful shutdown
            std::thread::sleep(std::time::Duration::from_secs(2));

            // Check if process is still running
            let still_running = Command::new("kill")
                .args(["-0", &pid.to_string()])
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .status()
                .map(|s| s.success())
                .unwrap_or(false);

            if still_running {
                println!("Server did not stop gracefully, sending SIGKILL...");
                Command::new("kill")
                    .args(["-KILL", &pid.to_string()])
                    .stdout(Stdio::null())
                    .stderr(Stdio::null())
                    .status()
                    .ok();
            }

            // Remove PID file
            std::fs::remove_file(pid_file).map_err(|e| Error::io(format!("Failed to remove PID file: {e}")))?;

            println!("Server stopped successfully");
            Ok(())
        } else {
            Err(Error::io(format!(
                "Failed to stop server. Process {pid} may not exist."
            )))
        }
    }

    #[cfg(not(unix))]
    {
        Err(Error::io("Server management is only supported on Unix systems"))
    }
}

fn check_server_status(pid_file: &PathBuf) -> Result<()> {
    if !pid_file.exists() {
        println!("Status: NOT RUNNING");
        println!("PID file not found: {}", pid_file.display());
        return Ok(());
    }

    let pid_str = std::fs::read_to_string(pid_file).map_err(|e| Error::io(format!("Failed to read PID file: {e}")))?;

    let pid = pid_str
        .trim()
        .parse::<i32>()
        .map_err(|e| Error::io(format!("Invalid PID in file: {e}")))?;

    #[cfg(unix)]
    {
        use std::process::{Command, Stdio};

        // Check if process is running
        let is_running = Command::new("kill")
            .args(["-0", &pid.to_string()])
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status()
            .map(|s| s.success())
            .unwrap_or(false);

        if is_running {
            println!("Status: RUNNING");
            println!("  PID: {pid}");
            println!("  PID file: {}", pid_file.display());

            // Try to get process info
            if let Ok(output) = Command::new("ps")
                .args(["-p", &pid.to_string(), "-o", "lstart="])
                .output()
            {
                if output.status.success() {
                    let start_time = String::from_utf8_lossy(&output.stdout).trim().to_string();
                    if !start_time.is_empty() {
                        println!("  Started: {start_time}");
                    }
                }
            }
        } else {
            println!("Status: NOT RUNNING");
            println!("  PID file exists but process {pid} is not running");
            println!("  You may need to remove stale PID file: {}", pid_file.display());
        }
    }

    #[cfg(not(unix))]
    {
        println!("Status: UNKNOWN (platform not supported)");
        println!("  PID: {}", pid);
    }

    Ok(())
}

// ========== HA Helper Functions ==========

/// Handles for HA components
#[allow(dead_code)]
#[derive(Default)]
struct HAHandles {
    #[allow(dead_code)]
    replication_handle: Option<tokio::task::JoinHandle<()>>,
}

/// Start HA replication components based on role
#[cfg(feature = "ha-tier1")]
async fn start_ha_components(
    ha_config: &HAConfig,
    listen: &str,
    port: u16,
    storage: std::sync::Arc<heliosdb_nano::storage::StorageEngine>,
) -> Result<HAHandles> {
    use heliosdb_nano::replication::{
        config::PrimaryConfig,
        ha_state::{ha_state, HARole, NodeConfig, SyncMode as HASyncMode},
        streaming::{StreamingClient, StreamingClientConfig, StreamingServer, StreamingServerConfig},
        wal_applicator::WalApplicator,
        wal_store::{WalStore, WalStoreConfig},
        SyncModeConfig,
    };
    use std::sync::Arc;
    use std::time::Duration;
    use uuid::Uuid;

    let node_id = if let Some(ref id) = ha_config.node_id {
        Uuid::parse_str(id).map_err(|e| Error::config(format!("Invalid node ID: {e}")))?
    } else {
        Uuid::new_v4()
    };

    info!("Starting HA components with node ID: {}", node_id);

    // Parse sync mode
    let sync_mode = match ha_config.sync_mode.to_lowercase().as_str() {
        "async" => SyncModeConfig::Async,
        "semi-sync" | "semisync" => SyncModeConfig::SemiSync {
            min_acks: 1,
            timeout_ms: 5000,
        },
        "sync" => SyncModeConfig::Sync {
            min_applied: 1,
            timeout_ms: 10000,
        },
        other => {
            return Err(Error::config(format!(
                "Unknown sync mode: '{}'. Use: async, semi-sync, sync",
                other
            )))
        }
    };

    let role = ha_config.role.to_lowercase();

    // Set HA state configuration for system views
    let ha_role = HARole::from_str(&role);
    let ha_sync_mode = HASyncMode::from_str(&ha_config.sync_mode);

    ha_state().set_config(NodeConfig {
        node_id,
        role: ha_role,
        listen_addr: listen.to_string(),
        port,
        replication_port: ha_config.replication_port,
        sync_mode: ha_sync_mode,
        primary_host: ha_config.primary_host.clone(),
        started_at: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs() as i64,
    });

    match role.as_str() {
        "primary" => {
            // Start streaming server for primary
            let repl_addr = format!("{}:{}", listen, ha_config.replication_port)
                .parse()
                .map_err(|e| Error::config(format!("Invalid replication address: {e}")))?;

            let wal_store = Arc::new(WalStore::new(WalStoreConfig::default()));
            wal_store
                .init()
                .await
                .map_err(|e| Error::io(format!("WAL store init failed: {e}")))?;

            let server_config = StreamingServerConfig {
                listen_addr: repl_addr,
                sync_mode,
                max_standbys: 10,
                heartbeat_interval: Duration::from_secs(1),
                ..Default::default()
            };

            let server = StreamingServer::new(server_config, node_id, wal_store);
            info!("Streaming replication server starting on {}", repl_addr);

            // Spawn server task
            let handle = tokio::spawn(async move {
                if let Err(e) = server.start().await {
                    tracing::error!("Streaming server error: {}", e);
                }
            });

            Ok(HAHandles {
                replication_handle: Some(handle),
            })
        }
        "standby" => {
            // Start streaming client for standby
            let primary_host = ha_config
                .primary_host
                .as_ref()
                .ok_or_else(|| Error::config("--primary-host required for standby role".to_string()))?;

            // Resolve hostname to IP address (supports Docker DNS)
            let primary_addr = tokio::net::lookup_host(primary_host)
                .await
                .map_err(|e| Error::config(format!("Cannot resolve primary host '{}': {}", primary_host, e)))?
                .next()
                .ok_or_else(|| Error::config(format!("No address found for primary host '{}'", primary_host)))?;

            // Initialize query forwarder for transparent write routing (HeliosProxy feature)
            // Extract hostname from primary_host (format: "host:replication_port")
            // Query forwarding connects to primary's postgres port, not replication port
            {
                use heliosdb_nano::replication::query_forwarder::init_query_forwarder;
                let primary_hostname = primary_host.split(':').next().unwrap_or(primary_host);
                // Primary's postgres port - use environment variable or default to 5432
                let primary_pg_port = std::env::var("HELIOSDB_PRIMARY_PG_PORT")
                    .ok()
                    .and_then(|p| p.parse().ok())
                    .unwrap_or(5432u16);
                init_query_forwarder(primary_hostname.to_string(), primary_pg_port);
                info!(
                    "Query forwarder initialized for transparent write routing to {}:{}",
                    primary_hostname, primary_pg_port
                );
            }

            let client_config = StreamingClientConfig {
                node_id,
                primary_addr,
                sync_mode,
                connect_timeout: Duration::from_secs(30),
                reconnect_delay: Duration::from_secs(5),
                max_reconnect_attempts: 0, // Unlimited
            };

            let (client, entry_rx) = StreamingClient::new(client_config);
            info!("Streaming client connecting to primary at {}", primary_host);

            // Create WAL Applicator for applying replicated entries
            let primary_hostname = primary_host.split(':').next().unwrap_or(primary_host);
            let primary_pg_port = std::env::var("HELIOSDB_PRIMARY_PG_PORT")
                .ok()
                .and_then(|p| p.parse().ok())
                .unwrap_or(5432u16);

            let applicator_config = PrimaryConfig {
                host: primary_hostname.to_string(),
                port: primary_pg_port,
                connect_timeout: Duration::from_secs(30),
                use_tls: false,
            };

            let applicator = Arc::new(WalApplicator::new(applicator_config));

            // Start the WAL applicator with storage engine
            applicator
                .start_with_storage(storage.clone())
                .await
                .map_err(|e| Error::io(format!("Failed to start WAL applicator: {}", e)))?;
            info!("WAL Applicator started, ready to apply replicated entries");

            // Get the applicator's queue sender for forwarding entries
            let queue_tx = applicator.get_queue_sender();

            // Spawn task to forward entries from streaming client to WAL applicator
            let _forward_handle = tokio::spawn(async move {
                info!("WAL entry forwarder task started");
                let mut entry_rx = entry_rx;
                while let Some(entry) = entry_rx.recv().await {
                    info!("Forwarder: received entry LSN={}, forwarding to applicator", entry.lsn);
                    if let Err(e) = queue_tx.send(entry).await {
                        tracing::error!("Failed to forward WAL entry to applicator: {}", e);
                        break;
                    }
                    info!("Forwarder: entry forwarded successfully");
                }
                info!("WAL entry forwarder stopped");
            });

            // Spawn streaming client task
            let client_handle = tokio::spawn(async move {
                if let Err(e) = client.start().await {
                    tracing::error!("Streaming client error: {}", e);
                }
            });

            // Keep applicator alive
            let _applicator_ref = applicator;

            Ok(HAHandles {
                replication_handle: Some(client_handle),
            })
        }
        "observer" => {
            info!("Starting as observer node");
            Ok(HAHandles::default())
        }
        _ => Err(Error::config(format!(
            "Unknown replication role: '{}'. Use: standalone, primary, standby, observer",
            ha_config.role
        ))),
    }
}

/// HTTP listener for Docker health checks and, when compiled with
/// `mcp-endpoint`, MCP-over-HTTP routes.
/// Caller owns the
/// already-bound `TcpListener` so bind failures are reported up front
/// (see KanttBan bug #2 / v3.28.0 fix).
async fn run_http_listener(
    listener: tokio::net::TcpListener,
    db: std::sync::Arc<EmbeddedDatabase>,
    mcp_token: Option<String>,
    allow_remote_mcp: bool,
) -> std::io::Result<()> {
    use axum::{routing::get, Json, Router};

    async fn health() -> Json<serde_json::Value> {
        Json(serde_json::json!({ "status": "ok" }))
    }

    let app = Router::new().route("/", get(health)).route("/health", get(health));

    #[cfg(feature = "mcp-endpoint")]
    let app = {
        let http_addr = listener.local_addr()?;
        let state = if let Some(token) = mcp_token {
            heliosdb_nano::mcp::McpState::new(db)
                .with_auth(heliosdb_nano::mcp::McpAuth::BearerToken(std::sync::Arc::from(token)))
        } else {
            heliosdb_nano::mcp::McpState::new(db)
        };
        let bind_result = if allow_remote_mcp {
            Ok(())
        } else {
            heliosdb_nano::mcp::bind_safety_check(http_addr, &state.auth)
        };
        match bind_result {
            Ok(()) => {
                tracing::info!("MCP endpoint mounted at http://{}/mcp", http_addr);
                heliosdb_nano::mcp::attach_mcp_routes(app, state)
            }
            Err(e) => {
                tracing::warn!("{e}; MCP routes are not mounted on this listener");
                app
            }
        }
    };

    #[cfg(not(feature = "mcp-endpoint"))]
    {
        if mcp_token.is_some() || allow_remote_mcp {
            tracing::warn!("MCP HTTP flags were supplied, but this binary was not built with the mcp-endpoint feature");
        }
        let _ = db;
    }

    axum::serve(listener, app).await
}

#[cfg(test)]
mod tests {
    use super::*;
    #[cfg(feature = "mcp-endpoint")]
    use serde_json::json;
    use serde_json::Value;

    #[tokio::test]
    async fn daemon_http_listener_serves_health() {
        let db = std::sync::Arc::new(EmbeddedDatabase::new_in_memory().unwrap());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let handle = tokio::spawn(async move {
            let _ = run_http_listener(listener, db, None, false).await;
        });

        let body: Value = reqwest::get(format!("http://{addr}/health"))
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert_eq!(body["status"], "ok");

        handle.abort();
    }

    #[cfg(feature = "mcp-endpoint")]
    #[tokio::test]
    async fn daemon_http_listener_mounts_mcp_post() {
        let db = std::sync::Arc::new(EmbeddedDatabase::new_in_memory().unwrap());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let handle = tokio::spawn(async move {
            let _ = run_http_listener(listener, db, None, false).await;
        });

        let resp: Value = reqwest::Client::new()
            .post(format!("http://{addr}/mcp"))
            .json(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} }))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert!(
            resp["result"]["tools"]
                .as_array()
                .map_or(false, |tools| !tools.is_empty()),
            "expected MCP tools/list response, got {resp}"
        );

        handle.abort();
    }

    #[cfg(feature = "mcp-endpoint")]
    #[tokio::test]
    async fn daemon_http_listener_does_not_mount_remote_mcp_without_auth() {
        let db = std::sync::Arc::new(EmbeddedDatabase::new_in_memory().unwrap());
        let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let addr = format!("127.0.0.1:{port}");
        let handle = tokio::spawn(async move {
            let _ = run_http_listener(listener, db, None, false).await;
        });

        let health: Value = reqwest::get(format!("http://{addr}/health"))
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert_eq!(health["status"], "ok");

        let status = reqwest::get(format!("http://{addr}/mcp/info")).await.unwrap().status();
        assert_eq!(status, reqwest::StatusCode::NOT_FOUND);

        handle.abort();
    }

    #[cfg(feature = "mcp-endpoint")]
    #[tokio::test]
    async fn daemon_http_listener_mounts_remote_mcp_with_bearer_token() {
        let db = std::sync::Arc::new(EmbeddedDatabase::new_in_memory().unwrap());
        let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let addr = format!("127.0.0.1:{port}");
        let handle = tokio::spawn(async move {
            let _ = run_http_listener(listener, db, Some("td8-secret".to_string()), false).await;
        });

        let client = reqwest::Client::new();
        let unauth = client
            .get(format!("http://{addr}/mcp/info"))
            .send()
            .await
            .unwrap()
            .status();
        assert_eq!(unauth, reqwest::StatusCode::UNAUTHORIZED);

        let info: Value = client
            .get(format!("http://{addr}/mcp/info"))
            .bearer_auth("td8-secret")
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        assert_eq!(info["serverInfo"]["name"], "heliosdb-nano");

        handle.abort();
    }
}