use ruc::*;

use std::fs;
use std::future;
use std::path::Path;
use std::process;
use std::sync::Arc;

use tokio::sync::RwLock;

use clap::{Parser, Subcommand};
use tokio::sync::watch;
use tracing::{Level, error, info};

use hotmint::abci::client::IpcApplicationClient;
use hotmint::api::rpc::ConsensusStatus;
use hotmint::config::{self, GenesisDoc, NodeConfig, NodeKey, NodeMode, PrivValidatorKey};
use hotmint::consensus::application::{Application, NoopApplication, TxValidationResult};
use hotmint::consensus::engine::{ConsensusEngine, EngineConfig};
use hotmint::consensus::pacemaker::PacemakerConfig;
use hotmint::consensus::state::ConsensusState;
use hotmint::consensus::store::{BlockStore, SharedStoreAdapter};
use hotmint::consensus::sync::sync_to_tip;
use hotmint::crypto::{Ed25519Signer, Ed25519Verifier};
use hotmint::mempool::{Mempool, MempoolAdapter};
use hotmint::network::service::{NetworkService, PeerMap};
use hotmint::prelude::*;
use hotmint::storage::block_store::VsdbBlockStore;
use hotmint::storage::consensus_state::PersistentConsensusState;
use hotmint_api::http_rpc::{ChainEvent, HttpRpcServer};
use tokio::sync::broadcast;

#[derive(Parser)]
#[command(name = "hotmint-node", about = "Hotmint BFT consensus node")]
struct Cli {
    /// Path to hotmint home directory.
    #[arg(long, global = true)]
    home: Option<String>,

    #[command(subcommand)]
    command: Command,
}

#[derive(Subcommand)]
enum Command {
    /// Initialize a new node directory.
    Init,
    /// Run the consensus node.
    Node {
        /// Override ABCI application socket path.
        #[arg(long)]
        proxy_app: Option<String>,
        /// Override P2P listen address (multiaddr).
        #[arg(long)]
        p2p_laddr: Option<String>,
        /// Override JSON-RPC listen address.
        #[arg(long)]
        rpc_laddr: Option<String>,
    },
    /// Generate a new validator key (Ed25519 signing keypair).
    GenValidatorKey {
        /// Output file path (default: priv_validator_key.json).
        #[arg(short, long, default_value = "priv_validator_key.json")]
        output: String,
        /// Validator ID to assign (default: 0).
        #[arg(long, default_value_t = 0)]
        validator_id: u64,
    },
    /// Generate a new node key (Ed25519 P2P identity keypair).
    GenNodeKey {
        /// Output file path (default: node_key.json).
        #[arg(short, long, default_value = "node_key.json")]
        output: String,
    },
    /// Display validator info from an existing key file.
    ShowValidator {
        /// Path to priv_validator_key.json.
        #[arg(short, long, default_value = "priv_validator_key.json")]
        file: String,
    },
    /// Display node ID (PeerId) from an existing node key file.
    ShowNodeId {
        /// Path to node_key.json.
        #[arg(short, long, default_value = "node_key.json")]
        file: String,
    },
}

#[tokio::main]
async fn main() {
    let cli = Cli::parse();
    let home = config::resolve_home(cli.home.as_deref());

    match cli.command {
        Command::Init => {
            if let Err(e) = config::init_node_dir(&home) {
                eprintln!("Error: {e}");
                process::exit(1);
            }
        }
        Command::Node {
            proxy_app,
            p2p_laddr,
            rpc_laddr,
        } => {
            tracing_subscriber::fmt()
                .with_max_level(Level::INFO)
                .with_target(false)
                .init();

            if let Err(e) = run_node(&home, proxy_app, p2p_laddr, rpc_laddr).await {
                eprintln!("Fatal: {e}");
                process::exit(1);
            }
        }
        Command::GenValidatorKey {
            output,
            validator_id,
        } => {
            let mut key = PrivValidatorKey::generate();
            key.validator_id = validator_id;
            let path = Path::new(&output);
            if let Err(e) = key.save(path) {
                eprintln!("Error: {e}");
                process::exit(1);
            }
            println!("Generated validator key:");
            println!("  Validator ID: {}", key.validator_id);
            println!("  Public key:   {}", key.public_key);
            println!("  File:         {}", path.display());
        }
        Command::GenNodeKey { output } => {
            let key = NodeKey::generate();
            let path = Path::new(&output);
            if let Err(e) = key.save(path) {
                eprintln!("Error: {e}");
                process::exit(1);
            }
            println!("Generated node key:");
            println!("  Public key: {}", key.public_key);
            match key.peer_id() {
                Ok(pid) => println!("  Peer ID:    {}", pid),
                Err(e) => eprintln!("  Peer ID:    (error: {e})"),
            }
            println!("  File:       {}", path.display());
        }
        Command::ShowValidator { file } => {
            let path = Path::new(&file);
            match PrivValidatorKey::load(path) {
                Ok(key) => {
                    println!("Validator key: {}", path.display());
                    println!("  Validator ID: {}", key.validator_id);
                    println!("  Public key:   {}", key.public_key);
                    match key.peer_id() {
                        Ok(pid) => println!("  Peer ID:      {}", pid),
                        Err(e) => eprintln!("  Peer ID:      (error: {e})"),
                    }
                }
                Err(e) => {
                    eprintln!("Error loading {}: {e}", path.display());
                    process::exit(1);
                }
            }
        }
        Command::ShowNodeId { file } => {
            let path = Path::new(&file);
            match NodeKey::load(path) {
                Ok(key) => {
                    println!("Node key: {}", path.display());
                    println!("  Public key: {}", key.public_key);
                    match key.peer_id() {
                        Ok(pid) => println!("  Peer ID:    {}", pid),
                        Err(e) => eprintln!("  Peer ID:    (error: {e})"),
                    }
                }
                Err(e) => {
                    eprintln!("Error loading {}: {e}", path.display());
                    process::exit(1);
                }
            }
        }
    }
}

async fn run_node(
    home: &Path,
    proxy_app_override: Option<String>,
    p2p_laddr_override: Option<String>,
    rpc_laddr_override: Option<String>,
) -> Result<()> {
    let config_dir = home.join("config");
    let data_dir = home.join("data");

    // 1. Load config
    let mut config = NodeConfig::load(&config_dir.join("config.toml")).c(d!(
        "failed to load config.toml — run `hotmint-node init` first"
    ))?;

    // Apply CLI overrides
    if let Some(pa) = proxy_app_override {
        config.proxy_app = pa;
    }
    if let Some(pl) = p2p_laddr_override {
        config.p2p.laddr = pl;
    }
    if let Some(rl) = rpc_laddr_override {
        config.rpc.laddr = rl;
    }

    // 1b. Validate required config fields before allocating any resources.
    // A validator without proxy_app is always a misconfiguration — fail immediately
    // before opening P2P ports, loading state, or starting the network stack.
    if config.node.mode == NodeMode::Validator && config.proxy_app.is_empty() {
        return Err(eg!("validator mode requires proxy_app to be configured — \
             set proxy_app = \"unix:///path/to/app.sock\" in config.toml, \
             or switch to mode = \"fullnode\" to run without an ABCI backend"));
    }

    // 2. Load validator key (consensus signing) and node key (P2P identity)
    let priv_key = PrivValidatorKey::load(&config_dir.join("priv_validator_key.json"))
        .c(d!("failed to load priv_validator_key.json"))?;
    let signing_key = priv_key.to_signing_key()?;
    let node_key =
        NodeKey::load(&config_dir.join("node_key.json")).c(d!("failed to load node_key.json"))?;
    let litep2p_keypair = node_key.to_litep2p_keypair()?;

    // 3. Load genesis
    let genesis =
        GenesisDoc::load(&config_dir.join("genesis.json")).c(d!("failed to load genesis.json"))?;
    let validator_set = genesis.to_validator_set()?;

    // Validate genesis: no duplicate IDs or public keys.
    {
        let mut seen_ids = std::collections::HashSet::new();
        let mut seen_pks = std::collections::HashSet::new();
        for gv in &genesis.validators {
            assert!(
                seen_ids.insert(gv.id),
                "genesis contains duplicate validator ID: {}",
                gv.id
            );
            assert!(
                seen_pks.insert(&gv.public_key),
                "genesis contains duplicate public key: {}",
                gv.public_key
            );
        }
    }

    // 4. Find our validator ID.
    // If not in genesis, assign a sentinel ID — the node runs as a fullnode
    // (observes consensus, syncs blocks, serves RPC) but does not vote or propose.
    // If later added to the validator set via epoch transition, it automatically
    // begins participating in consensus.
    let our_pk_hex = &priv_key.public_key;
    let is_fullnode;
    let our_vid = if let Some(gv) = genesis
        .validators
        .iter()
        .find(|v| &v.public_key == our_pk_hex)
    {
        is_fullnode = config.node.mode == NodeMode::Fullnode;
        ValidatorId(gv.id)
    } else {
        is_fullnode = true;
        // Sentinel ID for fullnodes — they never sign votes or propose blocks.
        // Verify no real validator uses this ID to prevent confusion.
        let sentinel = ValidatorId(u64::MAX);
        assert!(
            !validator_set.validators().iter().any(|v| v.id == sentinel),
            "genesis contains a validator with ID u64::MAX which collides with the fullnode sentinel"
        );
        sentinel
    };

    if is_fullnode {
        info!(
            node_id = our_vid.0,
            validators = validator_set.validator_count(),
            "starting hotmint fullnode (sync-only, no consensus participation)"
        );
    } else {
        info!(
            validator_id = %our_vid,
            validators = validator_set.validator_count(),
            quorum = validator_set.quorum_threshold(),
            "starting hotmint validator node"
        );
    }

    // 5. Set up persistent storage
    fs::create_dir_all(&data_dir).c(d!("create data dir"))?;
    vsdb::vsdb_set_base_dir(&data_dir).c(d!("set vsdb base dir"))?;

    let store: Arc<parking_lot::RwLock<Box<dyn BlockStore>>> = Arc::new(parking_lot::RwLock::new(
        Box::new(VsdbBlockStore::open(&data_dir).c(d!("open block store"))?),
    ));

    // 6. Restore consensus state
    let pcs = PersistentConsensusState::open(&data_dir).c(d!("open consensus state"))?;
    let mut state =
        ConsensusState::with_chain_id(our_vid, validator_set.clone(), &genesis.chain_id);
    if let Some(view) = pcs.load_current_view() {
        state.current_view = view;
    }
    if let Some(qc) = pcs.load_locked_qc() {
        state.locked_qc = Some(qc);
    }
    if let Some(qc) = pcs.load_highest_qc() {
        state.highest_qc = Some(qc);
    }
    if let Some(h) = pcs.load_last_committed_height() {
        state.last_committed_height = h;
    }
    if let Some(epoch) = pcs.load_current_epoch() {
        state.validator_set = epoch.validator_set.clone();
        state.current_epoch = epoch;
    }
    if let Some(hash) = pcs.load_last_app_hash() {
        state.last_app_hash = hash;
    }
    // A-1: Restore pending epoch transition for crash-safety.
    let restored_pending_epoch = pcs.load_pending_epoch();

    // Check WAL for incomplete commits (crash recovery).
    match hotmint_storage::wal::ConsensusWal::check_recovery(&data_dir) {
        Ok(hotmint_storage::wal::WalRecovery::NeedsReplay { target_height }) => {
            tracing::warn!(
                target_height = target_height.as_u64(),
                last_committed = state.last_committed_height.as_u64(),
                "WAL: detected incomplete commit, node will re-sync missing blocks from peers"
            );
            // The blocks were partially executed but state was not persisted.
            // On restart the node will re-sync from peers using the persisted
            // last_committed_height, which effectively replays the missing blocks.
        }
        Ok(hotmint_storage::wal::WalRecovery::Clean) => {}
        Err(e) => {
            tracing::warn!(%e, "WAL: failed to check recovery status");
        }
    }

    // 7. Parse peers and create network
    let (peer_map, known_addresses) = if config.p2p.persistent_peers.is_empty() {
        (PeerMap::new(), vec![])
    } else {
        config::parse_persistent_peers(&config.p2p.persistent_peers, &genesis)?
    };

    let listen_addr: litep2p::types::multiaddr::Multiaddr = config
        .p2p
        .laddr
        .parse()
        .c(d!("invalid p2p listen address: {}", config.p2p.laddr))?;

    let hotmint::network::service::NetworkServiceHandles {
        service: network_service,
        sink: network_sink,
        msg_rx,
        sync_req_rx,
        mut sync_resp_rx,
        peer_info_rx,
        connected_count_rx: _,
        notif_connected_count_rx: mut notif_count_rx,
        mut mempool_tx_rx,
    } = {
        let peer_book_path = home.join("data").join("peer_book.json");
        let peer_book = hotmint::network::peer::PeerBook::load(&peer_book_path)
            .unwrap_or_else(|_| hotmint::network::peer::PeerBook::new(&peer_book_path));
        let peer_book = Arc::new(RwLock::new(peer_book));
        NetworkService::create(hotmint::network::service::NetworkConfig {
            listen_addr,
            peer_map,
            known_addresses,
            keypair: Some(litep2p_keypair),
            peer_book,
            pex_config: {
                let mut pex = config.pex.clone();
                pex.private_peer_ids = config.p2p.private_peer_ids.clone();
                pex
            },
            relay_consensus: config.node.relay_consensus,
            initial_validators: validator_set
                .validators()
                .iter()
                .map(|v| (v.id, v.public_key.clone()))
                .collect(),
            chain_id_hash: state.chain_id_hash,
        })?
    };

    // 8. Create application (ABCI client or embedded noop)
    // Validators MUST have proxy_app configured — enforced at startup (step 1b) before
    // any resources are allocated. By this point validator+empty proxy_app is impossible.
    let use_abci = !config.proxy_app.is_empty();
    let (app_box, sync_app_box): (Box<dyn Application>, Box<dyn Application>) = if use_abci {
        let proxy_path = config
            .proxy_app
            .strip_prefix("unix://")
            .unwrap_or(&config.proxy_app);
        let ipc_client = IpcApplicationClient::new(proxy_path);
        ipc_client.check_connection().c(d!(
            "ABCI application not reachable at '{}' — start your application first",
            proxy_path
        ))?;
        let ipc_client_for_sync = IpcApplicationClient::new(proxy_path);
        (Box::new(ipc_client), Box::new(ipc_client_for_sync))
    } else {
        info!("fullnode mode: using embedded no-op application");
        (
            Box::new(hotmint::consensus::application::NoopApplication),
            Box::new(hotmint::consensus::application::NoopApplication),
        )
    };
    let mut engine_state_epoch = state.current_epoch.clone();
    let mut engine_state_height = state.last_committed_height;
    let mut engine_state_app_hash = state.last_app_hash;
    // A-1: Track pending epoch through sync for crash-safety.
    let mut engine_state_pending_epoch = restored_pending_epoch;
    let (status_tx, status_rx) = watch::channel(ConsensusStatus::new(
        0,
        state.last_committed_height.as_u64(),
        state.current_epoch.number.as_u64(),
        validator_set.validator_count(),
        state.current_epoch.start_view.as_u64(),
    ));
    let sync_status_rx = status_tx.subscribe();
    // Extra subscriber for the reconnect re-sync watcher (tracks last_committed_height)
    let watcher_status_rx = status_tx.subscribe();

    // Validator set watch channel (updated on epoch transitions via on_commit)
    let initial_vs: Vec<hotmint::api::types::ValidatorInfoResponse> = validator_set
        .validators()
        .iter()
        .map(|v| hotmint::api::types::ValidatorInfoResponse {
            id: v.id.0,
            power: v.power,
            public_key: hex::encode(&v.public_key.0),
        })
        .collect();
    let (vs_tx, vs_rx) = watch::channel(initial_vs);
    let (epoch_tx, epoch_rx) = watch::channel(state.current_epoch.clone());

    // 10. Create mempool (before AppWithStatus so recheck can use it)
    let mempool = Arc::new(Mempool::new(
        config.mempool.max_txs,
        config.mempool.max_tx_bytes,
    ));

    // Channel for chain event broadcasting (WebSocket subscribers).
    let (event_tx, event_rx) = broadcast::channel::<ChainEvent>(256);

    let app: Arc<dyn Application> = Arc::new(AppWithStatus {
        inner: Arc::from(app_box),
        status_tx,
        vs_tx,
        epoch_tx,
        mempool: mempool.clone(),
        event_tx: Some(event_tx),
    });

    // 11. Create RPC server
    let rpc_state = hotmint::api::rpc::RpcState {
        validator_id: our_vid.0,
        mempool: mempool.clone() as Arc<dyn MempoolAdapter>,
        status_rx,
        store: store.clone(),
        peer_info_rx,
        validator_set_rx: vs_rx,
        app: Some(app.clone()),
        network_sink: Some(Arc::new(network_sink.clone())),
        chain_id_hash: state.chain_id_hash,
    };
    // 12. Spawn network + RPC before sync (sync needs the network running).
    // Capture JoinHandles to detect unexpected exits and abort the process.
    // Respect serve_rpc / serve_sync config flags: when disabled, spawn a
    // permanently-pending task to keep the supervisor select! arms typed consistently.
    let network_handle = tokio::spawn(async move { network_service.run().await });
    let rpc_state = Arc::new(rpc_state);
    let rpc_handle: tokio::task::JoinHandle<()> = if config.node.serve_rpc {
        let rpc_server =
            hotmint::api::rpc::RpcServer::bind_arc(&config.rpc.laddr, rpc_state.clone())
                .await
                .c(d!("failed to bind RPC server"))?;
        info!(rpc_addr = %config.rpc.laddr, "RPC server listening");
        tokio::spawn(async move { rpc_server.run().await })
    } else {
        info!("RPC server disabled by config (serve_rpc = false)");
        tokio::spawn(future::pending())
    };

    // P0-1: Spawn HTTP/WebSocket RPC server if configured.
    let http_rpc_handle: tokio::task::JoinHandle<()> = if !config.rpc.http_laddr.is_empty() {
        let http_addr: std::net::SocketAddr = config
            .rpc
            .http_laddr
            .parse()
            .c(d!("invalid http_laddr: {}", config.rpc.http_laddr))?;
        let http_rpc = HttpRpcServer::new(http_addr, rpc_state.clone(), 256);
        let http_event_tx = http_rpc.event_sender();
        // Forward chain events from the broadcast channel to the HTTP WS channel.
        tokio::spawn(async move {
            let mut rx = event_rx;
            while let Ok(ev) = rx.recv().await {
                let _ = http_event_tx.send(ev);
            }
        });
        info!(http_addr = %config.rpc.http_laddr, "HTTP RPC server listening");
        tokio::spawn(async move { http_rpc.run().await })
    } else {
        tokio::spawn(future::pending())
    };

    let sync_sink = network_sink.clone();

    // Mempool gossip: receive txs from peers and add to local mempool.
    // (Outbound gossip is now handled directly via NetworkSink::broadcast_tx in RPC.)
    let gossip_mempool = mempool.clone() as Arc<dyn MempoolAdapter>;
    let gossip_app = app.clone();
    tokio::spawn(async move {
        while let Some(tx_bytes) = mempool_tx_rx.recv().await {
            let result = gossip_app.validate_tx(&tx_bytes, None);
            if !result.valid {
                continue;
            }
            let _ = gossip_mempool
                .add_tx(tx_bytes, result.priority, result.gas_wanted)
                .await;
        }
    });

    // Sync responder: answer incoming sync requests from peers.
    // Gated by config.node.serve_sync; when disabled, spawn a permanently-pending
    // task so the supervisor select! arm has a consistent JoinHandle type.
    let sync_responder_handle: tokio::task::JoinHandle<()> = if config.node.serve_sync {
        let store = store.clone();
        let sync_status_rx = sync_status_rx;
        let mut sync_req_rx = sync_req_rx;
        let sync_sink = sync_sink.clone();
        let responder_app = app.clone();
        tokio::spawn(async move {
            use hotmint_types::sync::{SyncRequest, SyncResponse};
            while let Some(req) = sync_req_rx.recv().await {
                let resp = match req.request {
                    SyncRequest::GetStatus => {
                        let s = *sync_status_rx.borrow();
                        SyncResponse::Status {
                            last_committed_height: Height(s.last_committed_height),
                            current_view: ViewNumber(s.current_view),
                            epoch: EpochNumber(s.epoch_number),
                        }
                    }
                    SyncRequest::GetBlocks {
                        from_height,
                        to_height,
                    } => {
                        // Clamp range to MAX_SYNC_BATCH to prevent DoS
                        let clamped =
                            Height(to_height.as_u64().min(
                                from_height.as_u64() + hotmint_types::sync::MAX_SYNC_BATCH - 1,
                            ));
                        let s = store.read();
                        let blocks = s.get_blocks_in_range(from_height, clamped);
                        let blocks_with_qcs: Vec<_> = blocks
                            .into_iter()
                            .map(|b| {
                                let qc = s.get_commit_qc(b.height);
                                (b, qc)
                            })
                            .collect();
                        drop(s);
                        SyncResponse::Blocks(blocks_with_qcs)
                    }
                    SyncRequest::GetSnapshots => {
                        SyncResponse::Snapshots(responder_app.list_snapshots())
                    }
                    SyncRequest::GetSnapshotChunk {
                        height,
                        chunk_index,
                    } => {
                        let data = responder_app.load_snapshot_chunk(height, chunk_index);
                        SyncResponse::SnapshotChunk {
                            height,
                            chunk_index,
                            data,
                        }
                    }
                };
                sync_sink.send_sync_response(req.request_id, &resp);
            }
        })
    } else {
        info!("sync responder disabled by config (serve_sync = false)");
        tokio::spawn(future::pending())
    };

    // 14. Sync catch-up before starting consensus (if peers configured)
    // Parse sync peers once — used for both initial sync and the reconnect watcher.
    let sync_peers: Vec<(ValidatorId, litep2p::PeerId)> = config
        .p2p
        .persistent_peers
        .iter()
        .filter_map(|p| {
            let (id_str, _) = p.split_once('@')?;
            let vid = ValidatorId(id_str.parse::<u64>().ok()?);
            let peer_id = genesis
                .validators
                .iter()
                .find(|v| v.id == vid.0)
                .and_then(|gv| {
                    let pk = hex::decode(&gv.public_key).ok()?;
                    let lpk = litep2p::crypto::ed25519::PublicKey::try_from_bytes(&pk).ok()?;
                    Some(lpk.to_peer_id())
                })?;
            Some((vid, peer_id))
        })
        .collect();

    if !sync_peers.is_empty() {
        use hotmint_types::sync::SyncRequest;

        info!("waiting for peer connection before sync...");
        let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(15);
        loop {
            if *notif_count_rx.borrow() > 0 {
                // At least one notification substream is open — ready to sync.
                break;
            }
            if tokio::time::Instant::now() >= deadline {
                info!("no peers connected within timeout, skipping sync");
                break;
            }
            let _ = tokio::time::timeout(
                tokio::time::Duration::from_millis(500),
                notif_count_rx.changed(),
            )
            .await;
        }

        if *notif_count_rx.borrow() > 0 {
            // Try syncing from each peer in order until one succeeds.
            // Use the main store so synced blocks are available for:
            // - accurate view estimation after sync
            // - serving sync requests to other nodes
            let mut synced = false;
            for (vid, peer_id) in &sync_peers {
                let bridge_sink = sync_sink.clone();
                let pid = *peer_id;
                let (sync_tx, mut sync_bridge_rx) = tokio::sync::mpsc::channel::<SyncRequest>(16);

                let bridge = tokio::spawn(async move {
                    while let Some(req) = sync_bridge_rx.recv().await {
                        bridge_sink.send_sync_request(pid, &req);
                    }
                });

                // Drain any residual responses from a previous peer iteration
                while sync_resp_rx.try_recv().is_ok() {}

                info!("starting block sync with V{}", vid.0);
                let mut sync_store = SharedStoreAdapter(store.clone());
                let mut sync_state = hotmint::consensus::sync::SyncState {
                    store: &mut sync_store,
                    app: sync_app_box.as_ref(),
                    current_epoch: &mut engine_state_epoch,
                    last_committed_height: &mut engine_state_height,
                    last_app_hash: &mut engine_state_app_hash,
                    chain_id_hash: &state.chain_id_hash,
                    pending_epoch: &mut engine_state_pending_epoch,
                };
                match sync_to_tip(&mut sync_state, &sync_tx, &mut sync_resp_rx).await {
                    Ok(()) => {
                        bridge.abort();
                        synced = true;
                        break;
                    }
                    Err(e) => {
                        info!(%e, peer = vid.0, "sync from peer failed, trying next");
                        bridge.abort();
                    }
                }
            }
            if !synced && !sync_peers.is_empty() {
                info!("all sync peers failed, continuing from current state");
            }
        }
    }

    // Spawn reconnect re-sync watcher: after the engine is running, detect when
    // the node reconnects to peers (notif_count goes 0→>0) and re-sync the store.
    // Uses NoopApplication to avoid double-executing blocks via ABCI — the running
    // engine will fast-forward through the newly stored blocks on the next DC received.
    // Always capture the JoinHandle so the supervisor select! can detect panic/unexpected exit.
    let reconnect_watcher_handle = if !sync_peers.is_empty() {
        let mut watcher_notif_rx = notif_count_rx;
        let watcher_store = store.clone();
        let watcher_sink = sync_sink.clone();
        let watcher_peers = sync_peers.clone();
        let mut watcher_resp_rx = sync_resp_rx;
        let watcher_epoch_rx = epoch_rx;
        // watcher_status_rx provides last_committed_height from the running engine
        let watcher_status_rx = watcher_status_rx;
        let watcher_chain_id_hash = state.chain_id_hash;

        tokio::spawn(async move {
            use hotmint_types::sync::SyncRequest;
            use tracing::warn;

            let mut prev_count = *watcher_notif_rx.borrow();
            loop {
                if watcher_notif_rx.changed().await.is_err() {
                    break;
                }
                let count = *watcher_notif_rx.borrow();
                if prev_count == 0 && count > 0 {
                    info!("reconnected to peers, triggering block re-sync");
                    let current_epoch = watcher_epoch_rx.borrow().clone();
                    // Use last_committed_height from the consensus status, not tip_height,
                    // to avoid skipping uncommitted/proposed blocks that are already stored.
                    let current_height = Height(watcher_status_rx.borrow().last_committed_height);
                    let mut h = current_height;
                    let mut epoch = current_epoch;
                    // NoopApplication: app_hash is carried from blocks, initial value irrelevant
                    let mut app_hash = BlockHash::GENESIS;
                    let mut watcher_pending_epoch: Option<Epoch> = None;

                    for (vid, peer_id) in &watcher_peers {
                        let pid = *peer_id;
                        let bridge_sink = watcher_sink.clone();
                        let (sync_tx, mut sync_bridge_rx) =
                            tokio::sync::mpsc::channel::<SyncRequest>(16);
                        let bridge = tokio::spawn(async move {
                            while let Some(req) = sync_bridge_rx.recv().await {
                                bridge_sink.send_sync_request(pid, &req);
                            }
                        });
                        // Drain stale responses before starting new sync
                        while watcher_resp_rx.try_recv().is_ok() {}
                        let mut store_adapter = SharedStoreAdapter(watcher_store.clone());
                        let noop = NoopApplication;
                        let mut sync_state = hotmint::consensus::sync::SyncState {
                            store: &mut store_adapter,
                            app: &noop,
                            current_epoch: &mut epoch,
                            last_committed_height: &mut h,
                            last_app_hash: &mut app_hash,
                            chain_id_hash: &watcher_chain_id_hash,
                            pending_epoch: &mut watcher_pending_epoch,
                        };
                        match sync_to_tip(&mut sync_state, &sync_tx, &mut watcher_resp_rx).await {
                            Ok(()) => {
                                bridge.abort();
                                break;
                            }
                            Err(e) => {
                                warn!(peer = vid.0, error = %e, "reconnect re-sync failed, trying next");
                                bridge.abort();
                            }
                        }
                    }
                }
                prev_count = count;
            }
        })
    } else {
        // No peers configured: spawn a permanently-pending task so the select! arm
        // has a consistent type without needing special-casing.
        tokio::spawn(future::pending())
    };

    // 15. Write back synced state and create consensus engine
    // (Engine is created after sync so it starts with up-to-date state)
    state.last_committed_height = engine_state_height;
    state.current_epoch = engine_state_epoch;
    state.validator_set = state.current_epoch.validator_set.clone();
    state.last_app_hash = engine_state_app_hash;

    // Advance current_view to match synced state so the engine joins the correct view.
    // Read the actual block.view from the last committed block (accurate even when
    // the network experienced view timeouts where view >> height).
    if engine_state_height > Height::GENESIS {
        let synced_view = {
            let s = store.read();
            s.get_block_by_height(engine_state_height)
                .map(|b| ViewNumber(b.view.as_u64() + 1))
                .unwrap_or(ViewNumber(engine_state_height.as_u64() + 1))
        };
        if synced_view > state.current_view {
            info!(
                synced_view = synced_view.as_u64(),
                "advancing view to match synced state"
            );
            state.current_view = synced_view;
        }
    }

    let signer = Ed25519Signer::new(signing_key, our_vid);
    let pacemaker_config = PacemakerConfig {
        base_timeout_ms: config.consensus.base_timeout_ms,
        max_timeout_ms: config.consensus.max_timeout_ms,
        backoff_multiplier: config.consensus.backoff_multiplier,
    };
    if pacemaker_config.base_timeout_ms == 0 {
        return Err(eg!("consensus.base_timeout_ms must be > 0"));
    }
    if pacemaker_config.max_timeout_ms < pacemaker_config.base_timeout_ms {
        return Err(eg!("consensus.max_timeout_ms must be >= base_timeout_ms"));
    }
    if pacemaker_config.backoff_multiplier < 1.0 {
        return Err(eg!("consensus.backoff_multiplier must be >= 1.0"));
    }
    let engine = ConsensusEngine::new(
        state,
        store.clone(),
        Box::new(network_sink),
        Box::new(ArcApp(app)),
        Box::new(signer),
        msg_rx,
        EngineConfig {
            verifier: Box::new(Ed25519Verifier),
            pacemaker: Some(pacemaker_config),
            persistence: Some(Box::new(pcs)),
            evidence_store: Some(Box::new(
                hotmint_storage::evidence_store::PersistentEvidenceStore::open(&data_dir)
                    .unwrap_or_else(|e| {
                        panic!("failed to open persistent evidence store: {e}");
                    }),
            )),
            wal: Some(Box::new(
                hotmint_storage::wal::ConsensusWal::open(&data_dir)
                    .expect("failed to open consensus WAL"),
            )),
            pending_epoch: engine_state_pending_epoch,
        },
    );

    info!("consensus engine starting");

    // 16. Run consensus engine (blocks forever).
    // Supervise the network, RPC, sync responder, and reconnect watcher tasks:
    // if any exits unexpectedly, log the reason and abort the process so the OS
    // process manager can restart the node.
    tokio::select! {
        () = engine.run() => {},
        res = network_handle => {
            match res {
                Ok(()) => error!("network service exited unexpectedly"),
                Err(e) => error!("network service panicked: {e}"),
            }
            process::exit(1);
        }
        res = rpc_handle => {
            match res {
                Ok(()) => error!("RPC server exited unexpectedly"),
                Err(e) => error!("RPC server panicked: {e}"),
            }
            process::exit(1);
        }
        res = http_rpc_handle => {
            match res {
                Ok(()) => error!("HTTP RPC server exited unexpectedly"),
                Err(e) => error!("HTTP RPC server panicked: {e}"),
            }
            process::exit(1);
        }
        res = sync_responder_handle => {
            match res {
                Ok(()) => error!("sync responder exited unexpectedly"),
                Err(e) => error!("sync responder panicked: {e}"),
            }
            process::exit(1);
        }
        res = reconnect_watcher_handle => {
            match res {
                Ok(()) => error!("reconnect watcher exited unexpectedly"),
                Err(e) => error!("reconnect watcher panicked: {e}"),
            }
            process::exit(1);
        }
        _ = tokio::signal::ctrl_c() => {
            info!("received shutdown signal, exiting...");
        }
        _ = async {
            #[cfg(unix)]
            {
                let mut sigterm = tokio::signal::unix::signal(
                    tokio::signal::unix::SignalKind::terminate()
                ).expect("failed to register SIGTERM handler");
                sigterm.recv().await;
            }
            #[cfg(not(unix))]
            {
                std::future::pending::<()>().await;
            }
        } => {
            info!("received SIGTERM, shutting down...");
        }
    }

    Ok(())
}

/// Wrapper that implements `Application` by delegating to an inner application,
/// while also updating the RPC status watch channel on each commit.
struct AppWithStatus {
    inner: Arc<dyn Application>,
    status_tx: watch::Sender<ConsensusStatus>,
    vs_tx: watch::Sender<Vec<hotmint::api::types::ValidatorInfoResponse>>,
    epoch_tx: watch::Sender<hotmint_types::epoch::Epoch>,
    mempool: Arc<Mempool>,
    event_tx: Option<broadcast::Sender<ChainEvent>>,
}

impl Application for AppWithStatus {
    fn create_payload(&self, ctx: &BlockContext) -> Vec<u8> {
        self.inner.create_payload(ctx)
    }

    fn validate_block(&self, block: &Block, ctx: &BlockContext) -> bool {
        self.inner.validate_block(block, ctx)
    }

    fn validate_tx(
        &self,
        tx: &[u8],
        ctx: Option<&hotmint_types::context::TxContext>,
    ) -> TxValidationResult {
        self.inner.validate_tx(tx, ctx)
    }

    fn execute_block(&self, txs: &[&[u8]], ctx: &BlockContext) -> Result<EndBlockResponse> {
        self.inner.execute_block(txs, ctx)
    }

    fn on_commit(&self, block: &Block, ctx: &BlockContext) -> Result<()> {
        self.inner.on_commit(block, ctx)?;
        let _ = self.status_tx.send(ConsensusStatus::new(
            ctx.view.as_u64(),
            ctx.height.as_u64(),
            ctx.epoch.as_u64(),
            ctx.validator_set.validator_count(),
            ctx.epoch_start_view.as_u64(),
        ));
        // Update validator set for RPC
        let vs: Vec<hotmint::api::types::ValidatorInfoResponse> = ctx
            .validator_set
            .validators()
            .iter()
            .map(|v| hotmint::api::types::ValidatorInfoResponse {
                id: v.id.0,
                power: v.power,
                public_key: hex::encode(&v.public_key.0),
            })
            .collect();
        let _ = self.vs_tx.send(vs);
        // Keep epoch watch current for reconnect re-sync watcher
        let _ = self.epoch_tx.send(hotmint_types::epoch::Epoch::new(
            ctx.epoch,
            ctx.epoch_start_view,
            ctx.validator_set.clone(),
        ));
        // Publish NewBlock event to WebSocket subscribers.
        if let Some(ref tx) = self.event_tx {
            let _ = tx.send(ChainEvent::NewBlock {
                height: block.height.as_u64(),
                hash: format!("{}", block.hash),
                view: block.view.as_u64(),
                proposer: block.proposer.0,
                timestamp: block.timestamp,
            });
            // Emit TxCommitted for each transaction in the block.
            let txs = hotmint::consensus::commit::decode_payload(&block.payload);
            for tx_bytes in &txs {
                let tx_hash = hex::encode(blake3::hash(tx_bytes).as_bytes());
                let _ = tx.send(ChainEvent::TxCommitted {
                    tx_hash,
                    height: block.height.as_u64(),
                });
            }
        }
        // A-5: Mempool re-validation runs in a background task to avoid
        // blocking the commit path. Evicts txs no longer valid after commit.
        let app = self.inner.clone();
        let mempool = self.mempool.clone();
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().spawn(async move {
                mempool
                    .recheck(|tx| {
                        let result = app.validate_tx(tx, None);
                        if result.valid {
                            Some((result.priority, result.gas_wanted))
                        } else {
                            None
                        }
                    })
                    .await;
            });
        });
        Ok(())
    }

    fn on_evidence(&self, proof: &EquivocationProof) -> Result<()> {
        self.inner.on_evidence(proof)
    }

    fn query(&self, path: &str, data: &[u8]) -> Result<hotmint_types::QueryResponse> {
        self.inner.query(path, data)
    }

    fn tracks_app_hash(&self) -> bool {
        self.inner.tracks_app_hash()
    }
}

/// Newtype wrapper to use `Arc<dyn Application>` as `Box<dyn Application>`.
struct ArcApp(Arc<dyn Application>);

impl Application for ArcApp {
    fn create_payload(&self, ctx: &BlockContext) -> Vec<u8> {
        self.0.create_payload(ctx)
    }
    fn validate_block(&self, block: &Block, ctx: &BlockContext) -> bool {
        self.0.validate_block(block, ctx)
    }
    fn validate_tx(
        &self,
        tx: &[u8],
        ctx: Option<&hotmint_types::context::TxContext>,
    ) -> TxValidationResult {
        self.0.validate_tx(tx, ctx)
    }
    fn execute_block(&self, txs: &[&[u8]], ctx: &BlockContext) -> Result<EndBlockResponse> {
        self.0.execute_block(txs, ctx)
    }
    fn on_commit(&self, block: &Block, ctx: &BlockContext) -> Result<()> {
        self.0.on_commit(block, ctx)
    }
    fn on_evidence(&self, proof: &EquivocationProof) -> Result<()> {
        self.0.on_evidence(proof)
    }
    fn query(&self, path: &str, data: &[u8]) -> Result<hotmint_types::QueryResponse> {
        self.0.query(path, data)
    }
    fn tracks_app_hash(&self) -> bool {
        self.0.tracks_app_hash()
    }
}
