use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Instant;

use mt_account::{apply_proposal, ProposalSettle};
use mt_codec::CanonicalEncode;
use mt_consensus::{proposal_hash, ProposalHeader};
use mt_net::{MsgType, ProtocolMessage};
use mt_crypto::{sign, Hash32, Signature, SIGNATURE_SIZE};
use mt_entry::{
    apply_candidate_expiry, apply_noderegistrations_batch, apply_selection_event,
    candidate_vdf_init, is_selection_window, nodereg_hash, validate_noderegistration,
    NodeRegistration,
};
use mt_genesis::genesis_params;
use mt_lottery::{
    bundle_hash, compute_endpoint, is_cemented, lottery_weight, quorum, reveal_hash,
    seniority_term, validate_bundle, validate_reveal, weighted_ticket_node, BundledConfirmation,
    VdfReveal,
};
use mt_merkle::{empty_internal, SparseMerkleTree, TREE_DEPTH};
use mt_state::compute_state_root;
use mt_store::FsStore;
use mt_timechain::{cemented_bundle_aggregate, next_d, vdf_step};

use crate::clock::{load_current_window, save_current_window};
use crate::identity::{default_data_dir, load_identity, NodeError};
use crate::node_lifecycle::{load_or_init_lifecycle, save_lifecycle, NodeLifecycle, NodePhase};
use crate::state::LocalState;
use crate::timechain_state::{load_or_init_timechain, save_timechain, TimeChainState};

static STOP: AtomicBool = AtomicBool::new(false);

extern "C" fn shutdown_handler(_: libc::c_int) {
    STOP.store(true, Ordering::SeqCst);
}

pub struct StartArgs {
    pub data_dir: Option<PathBuf>,
    pub max_windows: Option<u64>,
    pub d_test_override: Option<u64>,
    /// Multiaddr для libp2p listen (например `/ip4/0.0.0.0/tcp/8444`).
    /// При наличии — узел стартует cross-machine peering thread.
    pub listen_multiaddr: Option<String>,
    /// Путь к genesis-manifest.json с peer list для bootstrap connectivity.
    /// При наличии `--listen` обязателен.
    pub genesis_manifest: Option<PathBuf>,
}

pub fn run(args: StartArgs) -> Result<(), NodeError> {
    let data_dir = args.data_dir.unwrap_or_else(default_data_dir);
    let identity = load_identity(&data_dir)?;
    let params = genesis_params();

    // Cross-machine M8: spawn network thread с собственным tokio runtime.
    // Network событийный loop отделён от consensus loop (VDF compute) —
    // separate OS thread предотвращает блокировку async задач CPU-heavy
    // операциями подсчёта VDF.
    let mut network_handle: Option<NetworkHandle> = None;
    if let (Some(listen_str), Some(manifest_path)) =
        (&args.listen_multiaddr, &args.genesis_manifest)
    {
        network_handle = Some(spawn_network_thread(&identity, listen_str, manifest_path)?);
    }

    let mut state = LocalState::load_or_bootstrap(&data_dir, &identity, params)?;
    let mut current = load_current_window(&data_dir)?;
    let mut timechain = load_or_init_timechain(&data_dir)?;
    let mut lifecycle = load_or_init_lifecycle(&data_dir, &identity, params)?;
    let mut effective_d = args.d_test_override.unwrap_or(timechain.current_d);

    if let Some(d) = args.d_test_override {
        eprintln!(
            "ВНИМАНИЕ: --d-test-override={d} активен. Это test-only режим. \
             В production D = params.d0 = {} итераций (Genesis Decree).",
            params.d0
        );
        timechain.current_d = d;
    }

    install_shutdown_handlers();

    let stop_at = args.max_windows.map(|n| current.saturating_add(n));

    let my_account = identity.account_id();
    let my_node = identity.node_id();
    let initial_balance = state
        .accounts
        .get(&my_account)
        .map(|a| a.balance)
        .unwrap_or(0);

    let is_genesis = NodeLifecycle::is_bootstrap_node(&identity, params);

    // Phase Bootstrap = первая загрузка. Для genesis узла (identity.node_pk
    // совпадает с params.bootstrap_node_pubkey либо placeholder pre-ceremony) —
    // переход прямо в Active. Для candidate узла — переход в CandidateVdf
    // с target_chain_length = τ₂ и w_start = текущее окно + 1.
    if lifecycle.phase == NodePhase::Bootstrap {
        if is_genesis {
            lifecycle.phase = NodePhase::Active;
        } else {
            lifecycle.phase = NodePhase::CandidateVdf;
            lifecycle.target_chain_length = params.tau2_windows;
            lifecycle.w_start = current.saturating_add(1);
            lifecycle.candidate_progress = 0;
            // candidate_endpoint начинается с T_r текущего timechain — это
            // canonical seed для chain старта; на каждом окне ticks через
            // vdf_step_chunked в Active phase code path ниже.
            lifecycle.candidate_endpoint = timechain.t_r;
        }
        save_lifecycle(&data_dir, &lifecycle)?;
    }

    println!("=== montana-node start — узел Montana ===");
    println!();
    println!("data-dir         : {}", data_dir.display());
    println!("account_id       : {}", hex16(&my_account));
    println!("node_id          : {}", hex16(&my_node));
    println!("phase            : {:?}", lifecycle.phase);
    println!("current_window   : {current}");
    println!("D                : {} итераций SHA-256 / окно", effective_d);
    println!("T_r              : {}", hex16(&timechain.t_r));
    println!(
        "balance start    : {} nɈ ({} Ɉ)",
        initial_balance,
        initial_balance / 1_000_000_000
    );
    if let Some(stop) = stop_at {
        println!(
            "stop_at          : окно {stop} (через {} окон)",
            stop - current
        );
    } else {
        println!("stop_at          : Ctrl-C");
    }
    if lifecycle.phase == NodePhase::CandidateVdf {
        let remaining = lifecycle
            .target_chain_length
            .saturating_sub(lifecycle.candidate_progress);
        println!(
            "candidate VDF    : прогресс {}/{}, осталось {} окон до регистрации",
            lifecycle.candidate_progress, lifecycle.target_chain_length, remaining
        );
    }
    println!();
    println!("--- VDF тикает ---");
    println!();

    let session_start = Instant::now();
    let mut session_emitted: u128 = 0;
    let mut session_windows: u64 = 0;
    let mut prev_proposal_hash: Hash32 = [0u8; 32];

    let store = FsStore::open(&data_dir)
        .map_err(|e| NodeError::InvalidArguments(format!("FsStore::open: {e:?}")))?;

    loop {
        // M9: drain incoming envelopes от peer-ов (Proposal + другие), log + (Phase 2) apply
        if let Some(ref mut handle) = network_handle {
            while let Ok(msg) = handle.incoming_rx.try_recv() {
                if msg.msg_type == MsgType::Proposal {
                    eprintln!(
                        "[consensus] inbound Proposal envelope window={} (size={} bytes)                          — M9 Phase 1 (log only; Phase 2 = apply_proposal)",
                        msg.request_id,
                        msg.payload.len()
                    );
                }
            }
        }

        if STOP.load(Ordering::SeqCst) {
            println!();
            println!("[shutdown] получен SIGINT/SIGTERM, сохраняю состояние...");
            break;
        }
        if let Some(stop) = stop_at {
            if current >= stop {
                println!();
                println!("[stop_at] достигнуто целевое окно {stop}");
                break;
            }
        }

        let window_start = Instant::now();
        let next_window = current + 1;

        let next_t_r = vdf_step_chunked(&timechain.t_r, effective_d, "TimeChain VDF", next_window);

        match lifecycle.phase {
            NodePhase::Bootstrap => unreachable!("Bootstrap → CandidateVdf transition выше"),
            NodePhase::CandidateVdf => {
                lifecycle.candidate_endpoint = vdf_step_chunked(
                    &lifecycle.candidate_endpoint,
                    effective_d,
                    "Candidate VDF",
                    next_window,
                );
                lifecycle.candidate_progress = lifecycle
                    .candidate_progress
                    .checked_add(1)
                    .expect("candidate_progress overflow at u64::MAX");

                if lifecycle.candidate_progress >= lifecycle.target_chain_length {
                    let cba_w_start_minus_2 =
                        cemented_bundle_aggregate(lifecycle.w_start.saturating_sub(2), &[]);
                    let proof_endpoint =
                        candidate_vdf_init(&timechain.t_r, &cba_w_start_minus_2, &my_node);

                    let mut nr = NodeRegistration {
                        suite_id: identity.suite_id as u16,
                        node_pubkey: *identity.node_pk.as_bytes(),
                        operator_account_id: my_account,
                        proof_endpoint,
                        w_start: lifecycle.w_start,
                        vdf_chain_length: lifecycle.candidate_progress,
                        signature: Signature::from_array([0u8; SIGNATURE_SIZE]),
                    };
                    let mut scope = Vec::new();
                    nr.encode_signed_scope(&mut scope);
                    nr.signature = sign(&identity.node_sk, &scope).map_err(NodeError::Crypto)?;

                    validate_noderegistration(
                        &nr,
                        &state.nodes,
                        &state.candidates,
                        &state.accounts,
                    )
                    .map_err(|e| {
                        NodeError::InvalidArguments(format!("validate_noderegistration: {e:?}"))
                    })?;

                    let pending_baseline = state.candidates.len() as u64;
                    let active_nodes = state.nodes.len() as u64;
                    let cba_w_p_minus_2 =
                        cemented_bundle_aggregate(next_window.saturating_sub(2), &[]);
                    let outcome = apply_noderegistrations_batch(
                        &mut state.candidates,
                        &[nr.clone()],
                        &timechain.t_r,
                        &cba_w_p_minus_2,
                        pending_baseline,
                        active_nodes,
                        next_window,
                        params,
                    );
                    if outcome.applied.len() != 1 {
                        return Err(NodeError::InvalidArguments(format!(
                            "apply_noderegistrations_batch: applied={}, rejected={}, expected applied=1",
                            outcome.applied.len(),
                            outcome.rejected.len()
                        )));
                    }
                    lifecycle.nodereg_hash = nodereg_hash(&nr);
                    lifecycle.registration_window = next_window;
                    lifecycle.phase = NodePhase::Registered;
                    println!(
                        "[register W={next_window}] nodereg_hash={} | vdf_chain_length={}",
                        hex16(&lifecycle.nodereg_hash),
                        lifecycle.candidate_progress
                    );
                }
            },
            NodePhase::Registered => {},
            NodePhase::Active => {
                let active_chain_length: u64 = state.nodes.iter().map(|n| n.chain_length).sum();
                let cba_w_minus_2 = cemented_bundle_aggregate(current.saturating_sub(2), &[]);
                let endpoint = compute_endpoint(&timechain.t_r, &cba_w_minus_2, &my_node, current);

                let mut reveal = VdfReveal {
                    node_id: my_node,
                    window_index: current,
                    endpoint,
                    signature: Signature::from_array([0u8; SIGNATURE_SIZE]),
                };
                let mut reveal_scope = Vec::new();
                reveal.encode_signed_scope(&mut reveal_scope);
                reveal.signature =
                    sign(&identity.node_sk, &reveal_scope).map_err(NodeError::Crypto)?;
                validate_reveal(
                    &reveal,
                    &state.nodes,
                    &timechain.t_r,
                    &cba_w_minus_2,
                    current,
                )
                .map_err(|e| NodeError::InvalidArguments(format!("validate_reveal: {e:?}")))?;
                let r_hash = reveal_hash(&reveal);

                let mut bc = BundledConfirmation {
                    node_id: my_node,
                    endpoint: timechain.t_r,
                    window_index: current,
                    op_hashes: Vec::new(),
                    reveal_hashes: vec![r_hash],
                    signature: Signature::from_array([0u8; SIGNATURE_SIZE]),
                };
                let mut bc_scope = Vec::new();
                bc.encode_signed_scope(&mut bc_scope);
                bc.signature = sign(&identity.node_sk, &bc_scope).map_err(NodeError::Crypto)?;
                validate_bundle(&bc, &state.nodes, &timechain.t_r)
                    .map_err(|e| NodeError::InvalidArguments(format!("validate_bundle: {e:?}")))?;
                let bc_h = bundle_hash(&bc);

                let my_node_record = state.nodes.get(&my_node).cloned().ok_or_else(|| {
                    NodeError::InvalidArguments("active phase но узел не в NodeTable".into())
                })?;
                let cemented_chain_length = my_node_record.chain_length;
                if !is_cemented(cemented_chain_length, active_chain_length) {
                    return Err(NodeError::InvalidArguments(format!(
                        "singleton cementing: cemented={cemented_chain_length}, active={active_chain_length}, quorum={}",
                        quorum(active_chain_length)
                    )));
                }

                let snapshot = my_node_record.chain_length_snapshot.max(1);
                let _weight = lottery_weight(my_node_record.chain_length, snapshot);
                let _term = seniority_term(my_node_record.chain_length, snapshot);
                let _ticket =
                    weighted_ticket_node(&endpoint, my_node_record.chain_length, snapshot);

                // Proposal-level Merkle roots строятся как Sparse Merkle Tree
                // глубины 256 (тот же primitive что для state-уровня), индексация
                // по natural keys per spec, раздел "Структура proposal-level
                // Merkle roots":
                //   - included_bundles_root: ключ = confirmer_node_id,
                //     значение = (confirmer_node_id || bundle_hash)
                //   - included_reveals_root: ключ = reveal_author_node_id,
                //     значение = (reveal_author_node_id || reveal_hash)
                //   - control_root: ключ = nodereg_hash,
                //     значение = canonical_bytes(control_object)
                // Empty marker для всех трёх = empty_internal(TREE_DEPTH).
                // Singleton: один confirmer = my_node, один reveal = my_node;
                // control_set пустой (нет cemented NodeRegistration в singleton).
                let included_bundles_root = {
                    let mut tree = SparseMerkleTree::new();
                    let mut bundle_meta = Vec::with_capacity(64);
                    bundle_meta.extend_from_slice(&my_node);
                    bundle_meta.extend_from_slice(&bc_h);
                    tree.insert(my_node, &bundle_meta);
                    tree.root()
                };
                let included_reveals_root = {
                    let mut tree = SparseMerkleTree::new();
                    let mut reveal_meta = Vec::with_capacity(64);
                    reveal_meta.extend_from_slice(&my_node);
                    reveal_meta.extend_from_slice(&r_hash);
                    tree.insert(my_node, &reveal_meta);
                    tree.root()
                };
                let control_root = empty_internal(TREE_DEPTH);

                // Header формируется с placeholder values для post-apply полей
                // (state_root / account_root / node_root). Подпись вычисляется
                // ровно один раз — после apply_proposal — над финальным
                // signed_scope с post-apply значениями. Spec, раздел "Proposal":
                // "signature ML-DSA-65 над signed_scope(header) (Правило R1)";
                // все поля канонически вычислимы из cemented set, значит state_root
                // в подписи обязан быть post-apply.
                let mut header = ProposalHeader {
                    prev_proposal_hash,
                    window_index: current,
                    protocol_version: 1,
                    control_root,
                    node_root: [0u8; 32],
                    candidate_root: state.candidates.root(),
                    account_root: [0u8; 32],
                    state_root: [0u8; 32],
                    timechain_value: timechain.t_r,
                    included_bundles_root,
                    included_reveals_root,
                    winner_endpoint: endpoint,
                    winner_id: my_node,
                    proposer_node_id: my_node,
                    target: u128::MAX,
                    fallback_depth: 1,
                    signature: Signature::from_array([0u8; SIGNATURE_SIZE]),
                };

                let settle = ProposalSettle {
                    window_w: current,
                    winner_id: my_node,
                    cemented_confirmers: vec![my_node],
                };
                let post_state_root = apply_proposal(
                    &mut state.accounts,
                    &mut state.nodes,
                    &state.candidates,
                    &settle,
                    params,
                );

                header.state_root = post_state_root;
                header.account_root = state.accounts.root();
                header.node_root = state.nodes.root();
                let mut header_scope = Vec::new();
                header.encode_signed_scope(&mut header_scope);
                header.signature =
                    sign(&identity.node_sk, &header_scope).map_err(NodeError::Crypto)?;

                // M9 Phase 1: broadcast ProposalHeader всем connected peers через
                // network thread. Followers получают envelope с window_index и
                // в M9 Phase 2 будут running apply_proposal locally.
                if let Some(ref handle) = network_handle {
                    let mut header_bytes = Vec::with_capacity(3722);
                    header.encode(&mut header_bytes);
                    let envelope = ProtocolMessage::new(
                        MsgType::Proposal,
                        header.window_index,
                        header_bytes,
                    );
                    if let Err(e) = handle.broadcast_tx.send(envelope) {
                        eprintln!("[consensus] broadcast Proposal w={} failed: {e}", header.window_index);
                    } else {
                        eprintln!(
                            "[consensus] broadcast Proposal window={} → peers",
                            header.window_index
                        );
                    }
                }

                let recomputed = compute_state_root(
                    &state.nodes.root(),
                    &state.candidates.root(),
                    &state.accounts.root(),
                );
                if recomputed != post_state_root {
                    panic!(
                        "state_root self-verify failed: header={:02x?} recomputed={:02x?}",
                        post_state_root, recomputed
                    );
                }

                store
                    .archive_proposal(&header)
                    .map_err(|e| NodeError::InvalidArguments(format!("archive_proposal: {e:?}")))?;
                store.save_meta_last_cemented(current).map_err(|e| {
                    NodeError::InvalidArguments(format!("save_meta_last_cemented: {e:?}"))
                })?;
                prev_proposal_hash = proposal_hash(&header);

                session_emitted = session_emitted.saturating_add(params.emission_moneta);
                session_windows += 1;
            },
        }

        let _ = apply_candidate_expiry(&mut state.candidates, next_window);
        if is_selection_window(next_window, params) {
            let active = state.nodes.len() as u64;
            let cba = cemented_bundle_aggregate(next_window.saturating_sub(2), &[]);
            let activated = apply_selection_event(
                &mut state.candidates,
                &mut state.nodes,
                &mut state.accounts,
                &timechain.t_r,
                &cba,
                active,
                next_window,
                params,
            );
            if !activated.is_empty() {
                println!(
                    "[selection W={next_window}] активировано {} узл(ов)",
                    activated.len()
                );
                if lifecycle.phase == NodePhase::Registered && state.nodes.contains(&my_node) {
                    lifecycle.phase = NodePhase::Active;
                    println!("[active W={next_window}] phase Registered → Active");
                }
            }
        }

        if next_window > 0 && next_window % params.tau2_windows == 0 {
            let median_permille = 1000u32;
            let new_d = next_d(timechain.current_d, median_permille, params);
            if new_d != timechain.current_d {
                println!(
                    "[next_d W={next_window}] D: {} → {} (median_permille={median_permille})",
                    timechain.current_d, new_d
                );
                timechain.current_d = new_d;
                if args.d_test_override.is_none() {
                    effective_d = new_d;
                }
            }
        }

        timechain.t_r = next_t_r;
        timechain.last_window = next_window;
        current = next_window;

        let balance = state
            .accounts
            .get(&my_account)
            .map(|a| a.balance)
            .unwrap_or(0);

        let phase_marker = match lifecycle.phase {
            NodePhase::Bootstrap => "Bootstrap",
            NodePhase::CandidateVdf => "CandidateVdf",
            NodePhase::Registered => "Registered",
            NodePhase::Active => "Active",
        };
        let window_duration = window_start.elapsed();

        // Per-window лог = только progress bar строка от vdf_step_chunked
        // (in-place updates через `\r`, финал с `\n`). T_r/balance/phase
        // не печатаются здесь — оператор видит их через `montana-node status`.
        let _ = phase_marker;
        let _ = window_duration;
        let _ = balance;

        save_progress(&data_dir, &state, &timechain, &lifecycle, current)?;
    }

    let elapsed = session_start.elapsed();
    println!();
    println!("--- сессия завершена ---");
    println!("phase            : {:?}", lifecycle.phase);
    println!(
        "candidate VDF    : {}/{}",
        lifecycle.candidate_progress, lifecycle.target_chain_length
    );
    println!("обработано окон  : {session_windows} (только в Active phase)");
    println!(
        "выплачено        : {} nɈ ({} Ɉ)",
        session_emitted,
        session_emitted / 1_000_000_000
    );
    println!("session wall     : {:.1}s", elapsed.as_secs_f64());
    let final_balance = state
        .accounts
        .get(&my_account)
        .map(|a| a.balance)
        .unwrap_or(0);
    println!(
        "balance final    : {} nɈ ({} Ɉ)",
        final_balance,
        final_balance / 1_000_000_000
    );

    Ok(())
}

fn save_progress(
    data_dir: &std::path::Path,
    state: &LocalState,
    timechain: &TimeChainState,
    lifecycle: &NodeLifecycle,
    window: u64,
) -> Result<(), NodeError> {
    state.save(data_dir)?;
    save_current_window(data_dir, window)?;
    save_timechain(data_dir, timechain)?;
    save_lifecycle(data_dir, lifecycle)?;
    Ok(())
}

fn install_shutdown_handlers() {
    STOP.store(false, Ordering::SeqCst);
    unsafe {
        // SAFETY: shutdown_handler — extern "C" функция с правильной signature
        // (void (*)(int)). libc::signal принимает usize-cast pointer и
        // регистрирует обработчик. Atomic store в глобальный flag —
        // единственный side-effect handler-а, signal-safe (POSIX async-signal-safe).
        // SIGINT — Ctrl-C из терминала. SIGTERM — launchctl unload / systemd stop /
        // kill PID. Оба роутятся на тот же handler для единообразного graceful shutdown.
        libc::signal(libc::SIGINT, shutdown_handler as usize);
        libc::signal(libc::SIGTERM, shutdown_handler as usize);
    }
}

// Декомпозиция vdf_step(prev, d) на chunks с прогрессом в stdout.
// Корректность byte-exact: SHA-256^d ассоциативно по composition,
// vdf_step(vdf_step(x, a), b) = vdf_step(x, a + b) для a + b = d.
//
// Boundaries вычисляются как (d × i) / N — точно делит D на N равных
// долей даже при D не кратном N (последняя chunk может быть на 1 итерацию
// больше из-за rounding, но проценты в выводе всегда точно 10, 20, …, 100).
const VDF_PROGRESS_CHUNKS: u64 = 10;

fn vdf_step_chunked(prev: &Hash32, d: u64, label: &str, window: u64) -> Hash32 {
    if d == 0 {
        return *prev;
    }
    let mut current = *prev;
    let chunk_start = Instant::now();
    let mut prev_boundary: u64 = 0;
    use std::io::Write;
    for i in 1..=VDF_PROGRESS_CHUNKS {
        // Boundary распределяет D ровно: (d × i) / N (overflow безопасен:
        // d ≤ 2^32 typical, × N=10 ≤ 2^36).
        let boundary = d.saturating_mul(i) / VDF_PROGRESS_CHUNKS;
        let this_chunk = boundary - prev_boundary;
        current = vdf_step(&current, this_chunk);
        prev_boundary = boundary;
        let percent = (i * 100) / VDF_PROGRESS_CHUNKS;
        let bar = progress_bar(boundary, d, 30);
        let elapsed = chunk_start.elapsed().as_secs_f64();
        let line = format!(
            "окно {window:>5} {label:<14} {} {:>3}% | {:>4}/{:>4} M | {:>5.1}s",
            bar,
            percent,
            boundary / 1_000_000,
            d / 1_000_000,
            elapsed
        );
        // `\r` в начале — carriage return; bar обновляется в одной logical
        // строке. Финал на 100% chunk завершается `\n`. Работает одинаково
        // в TTY (cursor возврат) и при просмотре через `tail -F` (терминал
        // рендерит `\r` как возврат курсора → animated bar).
        if i == VDF_PROGRESS_CHUNKS {
            println!("\r{line}");
        } else {
            print!("\r{line}");
            let _ = std::io::stdout().flush();
        }
    }
    current
}

fn progress_bar(done: u64, total: u64, width: usize) -> String {
    if total == 0 {
        return "[".to_string() + &"░".repeat(width) + "]";
    }
    let filled = ((done as u128) * (width as u128) / (total as u128)) as usize;
    let filled = filled.min(width);
    let empty = width - filled;
    let mut s = String::with_capacity(width + 2);
    s.push('[');
    for _ in 0..filled {
        s.push('▓');
    }
    for _ in 0..empty {
        s.push('░');
    }
    s.push(']');
    s
}

fn hex16(bytes: &[u8]) -> String {
    let take = bytes.len().min(8);
    let mut s = String::with_capacity(take * 2);
    for b in &bytes[..take] {
        s.push_str(&format!("{b:02x}"));
    }
    s
}

fn spawn_network_thread(
    identity: &crate::identity::Identity,
    listen_str: &str,
    manifest_path: &std::path::Path,
) -> Result<NetworkHandle, NodeError> {
    use std::str::FromStr;

    let manifest_text = std::fs::read_to_string(manifest_path).map_err(|e| {
        NodeError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("genesis-manifest {manifest_path:?}: {e}"),
        ))
    })?;
    let manifest = mt_genesis::GenesisManifest::parse(&manifest_text)
        .map_err(|e| NodeError::Network(format!("parse manifest: {e}")))?;

    let listen_addr = libp2p::Multiaddr::from_str(listen_str)
        .map_err(|e| NodeError::Network(format!("parse --listen {listen_str}: {e}")))?;

    let local_keypair = identity.libp2p_keypair();
    let local_peer_id = identity.libp2p_peer_id();
    let manifest_clone = manifest.clone();

    let (broadcast_tx, broadcast_rx) =
        tokio::sync::mpsc::unbounded_channel::<mt_net::ProtocolMessage>();
    let (incoming_tx, incoming_rx) =
        tokio::sync::mpsc::unbounded_channel::<mt_net::ProtocolMessage>();

    eprintln!(
        "[main] spawning network thread, local_peer_id={local_peer_id},          listen={listen_addr}, peers={n}",
        n = manifest.peers.len()
    );

    std::thread::Builder::new()
        .name("montana-network".into())
        .spawn(move || {
            let runtime = match tokio::runtime::Builder::new_multi_thread()
                .worker_threads(2)
                .enable_all()
                .build()
            {
                Ok(rt) => rt,
                Err(e) => {
                    eprintln!("[network] failed to start tokio runtime: {e}");
                    return;
                },
            };
            if let Err(e) = runtime.block_on(crate::network::run_network_loop(
                local_keypair,
                local_peer_id,
                manifest_clone,
                listen_addr,
                broadcast_rx,
                incoming_tx,
            )) {
                eprintln!("[network] event loop exited with error: {e}");
            }
        })
        .map_err(|e| NodeError::Network(format!("spawn network thread: {e}")))?;

    Ok(NetworkHandle { broadcast_tx, incoming_rx })
}

/// Handle к network thread. Через broadcast_tx consensus loop рассылает
/// envelope-ы всем подключённым peer-ам. Через incoming_rx consensus loop
/// принимает входящие envelope-ы (Proposal, BundledConfirmation, NodeRegistration,
/// ...) от peer-ов для apply.
pub struct NetworkHandle {
    pub broadcast_tx: tokio::sync::mpsc::UnboundedSender<mt_net::ProtocolMessage>,
    pub incoming_rx: tokio::sync::mpsc::UnboundedReceiver<mt_net::ProtocolMessage>,
}