// Copyright 2024 MaidSafe.net limited.
//
// This SAFE Network Software is licensed to you under The General Public License (GPL), version 3.
// Unless required by applicable law or agreed to in writing, the SAFE Network Software distributed
// under the GPL Licence is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. Please review the Licences for the specific language governing
// permissions and limitations relating to use of the SAFE Network Software.

// Allow expect and panic in binaries - to be refactored
#![allow(clippy::expect_used)]
#![allow(clippy::panic)]

#[macro_use]
extern crate tracing;

mod log;
mod rpc_service;
mod subcommands;
#[cfg(unix)]
mod upgrade;

use crate::log::{reset_critical_failure, set_critical_failure};
use crate::subcommands::EvmNetworkCommand;
use ant_bootstrap::BootstrapConfig;
use ant_bootstrap::InitialPeersConfig;
use ant_bootstrap::bootstrap::Bootstrap;
use ant_evm::{EvmNetwork, RewardsAddress, get_evm_network};
use ant_logging::metrics::init_metrics;
use ant_logging::{Level, LogFormat, LogOutputDest, ReloadHandle};
use ant_node::utils::{get_antnode_root_dir, get_root_dir_and_keypair};
use ant_node::{Marker, NodeBuilder, NodeEvent, NodeEventsReceiver};
use ant_protocol::{
    node_rpc::{NodeCtrl, StopResult},
    version,
};
use clap::Parser;
use color_eyre::{Result, eyre::eyre};
use const_hex::traits::FromHex;
use libp2p::PeerId;
#[cfg(unix)]
use rand::Rng;
use std::{
    env,
    net::{IpAddr, Ipv4Addr, SocketAddr},
    path::{Path, PathBuf},
    process::Command,
    time::Duration,
};
use tokio::{
    runtime::Runtime,
    sync::{broadcast::error::RecvError, mpsc},
    time::sleep,
};
use tracing_appender::non_blocking::WorkerGuard;

#[derive(Debug, Clone)]
pub enum LogOutputDestArg {
    Stdout,
    DataDir,
    Path(PathBuf),
}

impl std::fmt::Display for LogOutputDestArg {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            LogOutputDestArg::Stdout => write!(f, "stdout"),
            LogOutputDestArg::DataDir => write!(f, "data-dir"),
            LogOutputDestArg::Path(path) => write!(f, "{}", path.display()),
        }
    }
}

pub fn parse_log_output(val: &str) -> Result<LogOutputDestArg> {
    match val {
        "stdout" => Ok(LogOutputDestArg::Stdout),
        "data-dir" => Ok(LogOutputDestArg::DataDir),
        // The path should be a directory, but we can't use something like `is_dir` to check
        // because the path doesn't need to exist. We can create it for the user.
        value => Ok(LogOutputDestArg::Path(PathBuf::from(value))),
    }
}

// Please do not remove the blank lines in these doc comments.
// They are used for inserting line breaks when the help menu is rendered in the UI.
#[derive(Parser, Debug)]
#[command(disable_version_flag = true)]
#[clap(name = "antnode cli", version = env!("CARGO_PKG_VERSION"))]
struct Opt {
    /// Set to connect to the alpha network.
    #[clap(long)]
    alpha: bool,

    /// Print the crate version.
    #[clap(long)]
    crate_version: bool,

    #[cfg(feature = "open-metrics")]
    /// Start the metrics server.
    ///
    /// This is automatically enabled if `metrics_server_port` is specified.
    #[clap(
        long,
        default_value_t = false,
        required_if_eq("metrics_server_port", "0")
    )]
    enable_metrics_server: bool,

    /// Specify the EVM network to use.
    /// The network can either be a pre-configured one or a custom network.
    /// When setting a custom network, you must specify the RPC URL to a fully synced node and
    /// the addresses of the network token and chunk payments contracts.
    #[command(subcommand)]
    evm_network: Option<EvmNetworkCommand>,

    /// Specify the IP to listen on.
    ///
    /// The special value `0.0.0.0` binds to all network interfaces available.
    #[clap(long, default_value_t = IpAddr::V4(Ipv4Addr::UNSPECIFIED))]
    ip: IpAddr,

    /// Specify the logging output destination.
    ///
    /// Valid values are "stdout", "data-dir", or a custom path.
    ///
    /// `data-dir` is the default value.
    ///
    /// The data directory location is platform specific:
    ///  - Linux: $HOME/.local/share/autonomi/node/<peer-id>/logs
    ///  - macOS: $HOME/Library/Application Support/autonomi/node/<peer-id>/logs
    ///  - Windows: C:\Users\<username>\AppData\Roaming\autonomi\node\<peer-id>\logs
    #[expect(rustdoc::invalid_html_tags)]
    #[clap(long, default_value_t = LogOutputDestArg::DataDir, value_parser = parse_log_output, verbatim_doc_comment)]
    log_output_dest: LogOutputDestArg,

    /// Specify the logging format.
    ///
    /// Valid values are "default" or "json".
    ///
    /// If the argument is not used, the default format will be applied.
    #[clap(long, value_parser = LogFormat::parse_from_str, verbatim_doc_comment)]
    log_format: Option<LogFormat>,

    /// Specify the maximum number of uncompressed log files to store.
    ///
    /// This argument is ignored if `log_output_dest` is set to "stdout"
    ///
    /// After reaching this limit, the older files are archived to save space.
    /// You can also specify the maximum number of archived log files to keep.
    #[clap(long, verbatim_doc_comment)]
    max_log_files: Option<usize>,

    /// Specify the maximum number of archived log files to store.
    ///
    /// This argument is ignored if `log_output_dest` is set to "stdout"
    ///
    /// After reaching this limit, the older archived files are deleted.
    #[clap(long, verbatim_doc_comment)]
    max_archived_log_files: Option<usize>,

    #[cfg(feature = "open-metrics")]
    /// Specify the port for the OpenMetrics server.
    ///
    /// If set, `--enable-metrics-server` will automatically be set to true.
    /// If not set, you must manually specify `--enable-metrics-server` and a port will be selected at random.
    #[clap(long, default_value_t = 0)]
    metrics_server_port: u16,

    /// Specify the network ID to use. This will allow you to run the node on a different network.
    ///
    /// By default, the network ID is set to 1, which represents the mainnet.
    #[clap(long, verbatim_doc_comment)]
    network_id: Option<u8>,

    /// By default, a node would use UPnP to open a port in the home router and allow incoming connections.
    ///
    /// Set to 'true' to disable UPnP.
    #[clap(long, default_value_t = false)]
    no_upnp: bool,

    /// Print the package version.
    #[cfg(not(feature = "nightly"))]
    #[clap(long)]
    package_version: bool,

    #[command(flatten)]
    peers: InitialPeersConfig,

    /// Print the network protocol version.
    #[clap(long)]
    protocol_version: bool,

    /// Specify the port to listen on.
    ///
    /// The special value `0` will cause the OS to assign a random port.
    #[clap(long, default_value_t = 0)]
    port: u16,

    /// Specify the rewards address.
    /// The rewards address is the address that will receive the rewards for the node.
    /// It should be a valid EVM address.
    #[clap(long)]
    rewards_address: Option<String>,

    /// Enable the mode to run as a relay client if it is behind a NAT and is not externally reachable.
    #[clap(long, default_value_t = false)]
    relay: bool,

    /// Specify the node's data directory.
    ///
    /// If not provided, the default location is platform specific:
    ///  - Linux: $HOME/.local/share/autonomi/node/<peer-id>
    ///  - macOS: $HOME/Library/Application Support/autonomi/node/<peer-id>
    ///  - Windows: C:\Users\<username>\AppData\Roaming\autonomi\node\<peer-id>
    #[expect(rustdoc::invalid_html_tags)]
    #[clap(long, verbatim_doc_comment)]
    root_dir: Option<PathBuf>,

    /// Enable the admin/control RPC service by providing an IP and port for it to listen on.
    ///
    /// The RPC service can be used for querying information about the running node.
    #[clap(long)]
    rpc: Option<SocketAddr>,

    /// Print version information.
    #[clap(long)]
    version: bool,

    /// Set this to true if you want the node to write the cache files in the older formats.
    #[clap(long, default_value_t = false)]
    write_older_cache_files: bool,

    /// Stop the node instead of restarting after a successful upgrade.
    ///
    /// Useful when running under a service manager that handles restarts.
    #[clap(long, default_value_t = false)]
    stop_on_upgrade: bool,
}

fn main() -> Result<()> {
    color_eyre::install()?;
    let opt = Opt::parse();

    let network_id = if let Some(network_id) = opt.network_id {
        network_id
    } else if opt.alpha {
        2
    } else {
        1
    };
    version::set_network_id(network_id);

    let identify_protocol_str = version::IDENTIFY_PROTOCOL_STR
        .read()
        .expect("Failed to obtain read lock for IDENTIFY_PROTOCOL_STR");
    if opt.version {
        println!(
            "{}",
            ant_build_info::version_string(
                "Autonomi Node",
                env!("CARGO_PKG_VERSION"),
                Some(&identify_protocol_str)
            )
        );
        return Ok(());
    }

    let rewards_address = RewardsAddress::from_hex(opt.rewards_address.as_ref().expect(
        "the following required arguments were not provided: --rewards-address <REWARDS_ADDRESS>",
    ))?;

    if opt.crate_version {
        println!("Crate version: {}", env!("CARGO_PKG_VERSION"));
        return Ok(());
    }

    if opt.protocol_version {
        println!("Network version: {identify_protocol_str}");
        return Ok(());
    }

    #[cfg(not(feature = "nightly"))]
    if opt.package_version {
        println!("Package version: {}", ant_build_info::package_version());
        return Ok(());
    }

    let evm_network: EvmNetwork = match opt.evm_network.as_ref() {
        Some(evm_network) => Ok(evm_network.clone().into()),
        None => match get_evm_network(opt.peers.local, Some(network_id)) {
            Ok(net) => Ok(net),
            Err(_) => Err(eyre!(
                "EVM network not specified. Please specify a network using the subcommand or by setting the `EVM_NETWORK` environment variable."
            )),
        },
    }?;

    println!("EVM network: {evm_network:?}");

    let node_socket_addr = SocketAddr::new(opt.ip, opt.port);
    let (root_dir, keypair) = get_root_dir_and_keypair(&opt.root_dir)?;

    let (log_output_dest, log_reload_handle, _log_appender_guard) =
        init_logging(&opt, keypair.public().to_peer_id())?;

    // Create a tokio runtime per `run_node` attempt, this ensures
    // any spawned tasks are closed before we would attempt to run
    // another process with these args.
    let rt = Runtime::new()?;

    let bootstrap_config = BootstrapConfig::try_from(&opt.peers)?
        .with_backwards_compatible_writes(opt.write_older_cache_files);
    let bootstrap = rt.block_on(Bootstrap::new(bootstrap_config))?;

    let msg = format!(
        "Running {} v{}",
        env!("CARGO_BIN_NAME"),
        env!("CARGO_PKG_VERSION")
    );
    info!("\n{}\n{}", msg, "=".repeat(msg.len()));

    ant_build_info::log_version_info(env!("CARGO_PKG_VERSION"), &identify_protocol_str);
    debug!(
        "antnode built with git version: {}",
        ant_build_info::git_info()
    );

    if opt.peers.local {
        rt.spawn(init_metrics(std::process::id()));
    }
    let restart_options = rt.block_on(async move {
        let mut node_builder = NodeBuilder::new(
            keypair,
            bootstrap,
            rewards_address,
            evm_network,
            node_socket_addr,
            root_dir,
        );
        node_builder.local(opt.peers.local);
        node_builder.no_upnp(opt.no_upnp);
        node_builder.relay_client(opt.relay);
        #[cfg(feature = "open-metrics")]
        let mut node_builder = node_builder;
        // if enable flag is provided or only if the port is specified then enable the server by setting Some()
        #[cfg(feature = "open-metrics")]
        let metrics_server_port = if opt.enable_metrics_server || opt.metrics_server_port != 0 {
            Some(opt.metrics_server_port)
        } else {
            None
        };
        #[cfg(feature = "open-metrics")]
        node_builder.metrics_server_port(metrics_server_port);
        let restart_options = run_node(
            node_builder,
            opt.rpc,
            &log_output_dest,
            log_reload_handle,
            opt.stop_on_upgrade,
        )
        .await?;

        Ok::<_, eyre::Report>(restart_options)
    })?;

    // actively shut down the runtime
    rt.shutdown_timeout(Duration::from_secs(2));

    // Restart only if we received a restart command.
    if let Some((root_dir, port)) = restart_options {
        start_new_node_process(root_dir, port);
        println!("A new node process has been started successfully.");
    } else {
        println!("The node process has been stopped.");
    }

    Ok(())
}

/// Start a node with the given configuration.
/// Returns:
/// - `Ok(Some((root_dir, port)))` if we receive a restart request.
/// - `Ok(None)` if we want to shutdown the node.
/// - `Err(_)` if we want to shutdown the node with an error.
async fn run_node(
    node_builder: NodeBuilder,
    rpc: Option<SocketAddr>,
    log_output_dest: &str,
    log_reload_handle: ReloadHandle,
    #[cfg_attr(not(unix), allow(unused_variables))] stop_on_upgrade: bool,
) -> Result<Option<(PathBuf, u16)>> {
    let started_instant = std::time::Instant::now();

    reset_critical_failure(log_output_dest);

    info!("Starting node ...");
    let running_node = node_builder.build_and_run()?;

    println!(
        "
Node started

PeerId is {}
You can check your reward balance by running:
`safe wallet balance --peer-id={}`
    ",
        running_node.peer_id(),
        running_node.peer_id()
    );

    // write the PID to the root dir
    let pid = std::process::id();
    let pid_file = running_node.root_dir_path().join("antnode.pid");
    std::fs::write(pid_file, pid.to_string().as_bytes())?;

    // Channel to receive node ctrl cmds from RPC service (if enabled), and events monitoring task
    let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<NodeCtrl>(5);

    // Monitor `NodeEvents`
    let node_events_rx = running_node.node_events_channel().subscribe();
    monitor_node_events(node_events_rx, ctrl_tx.clone());

    // Monitor ctrl-c
    let ctrl_tx_clone = ctrl_tx.clone();
    tokio::spawn(async move {
        if let Err(err) = tokio::signal::ctrl_c().await {
            // I/O error, ignore/print the error, but continue to handle as if ctrl-c was received
            warn!("Listening to ctrl-c error: {err}");
        }
        if let Err(err) = ctrl_tx_clone
            .send(NodeCtrl::Stop {
                delay: Duration::from_secs(1),
                result: StopResult::Error(eyre!("Ctrl-C received!")),
            })
            .await
        {
            error!("Failed to send node control msg to antnode bin main thread: {err}");
        }
    });

    // Start up gRPC interface if enabled by user
    let ctrl_tx_rpc = ctrl_tx.clone();
    if let Some(addr) = rpc {
        rpc_service::start_rpc_service(
            addr,
            log_output_dest,
            running_node.clone(),
            ctrl_tx_rpc,
            started_instant,
            log_reload_handle,
        );
    }

    #[cfg(unix)]
    {
        let ctrl_tx_restart = ctrl_tx.clone();
        let running_node_clone = running_node.clone();
        tokio::spawn(async move {
            // 72 hours (259200 seconds) with ±5% randomization to prevent simultaneous upgrades
            let base_delay = 259200;
            let variance = rand::thread_rng().gen_range(-12960..=12960);
            let upgrade_check_delay_secs = (base_delay + variance) as u64;
            loop {
                let upgrade_check_wake_time =
                    chrono::Utc::now() + chrono::Duration::seconds(upgrade_check_delay_secs as i64);
                info!(
                    "Next upgrade check scheduled for {}",
                    upgrade_check_wake_time
                );
                sleep(Duration::from_secs(upgrade_check_delay_secs)).await;

                match upgrade::perform_upgrade().await {
                    Ok(()) => {
                        let node_restart_delay =
                            upgrade::calculate_restart_delay(&running_node_clone).await;
                        let node_restart_wake_time = chrono::Utc::now() + node_restart_delay;
                        info!("Node will stop/restart for upgrade at {node_restart_wake_time}");
                        sleep(node_restart_delay).await;

                        let node_ctrl = if stop_on_upgrade {
                            NodeCtrl::Stop {
                                delay: Duration::from_secs(0),
                                result: StopResult::Success(
                                    "Upgrade completed successfully".to_string(),
                                ),
                            }
                        } else {
                            NodeCtrl::Restart {
                                delay: Duration::from_secs(0),
                                retain_peer_id: true,
                            }
                        };

                        if let Err(err) = ctrl_tx_restart.send(node_ctrl).await {
                            error!("Failed to send control command: {err}");
                        }
                        break;
                    }
                    Err(upgrade::UpgradeError::AlreadyLatest) => {
                        info!("Already running latest version");
                    }
                    Err(e) => {
                        warn!("Error during upgrade process: {e}");
                    }
                }
            }
        });
    }

    // Keep the node and gRPC service (if enabled) running.
    // We'll monitor any NodeCtrl cmd to restart/stop/update,
    loop {
        match ctrl_rx.recv().await {
            Some(NodeCtrl::Restart {
                delay,
                retain_peer_id: _,
            }) => {
                let root_dir = running_node.root_dir_path();
                let node_port = running_node.get_node_listening_port().await?;

                let msg = format!("Node is restarting in {delay:?}...");
                info!("{msg}");
                println!("{msg} Node path: {log_output_dest}");
                sleep(delay).await;

                return Ok(Some((root_dir, node_port)));
            }
            Some(NodeCtrl::Stop { delay, result }) => {
                let msg = format!("Node is stopping in {delay:?}...");
                info!("{msg}");
                println!("{msg} Node log path: {log_output_dest}");
                sleep(delay).await;
                match result {
                    StopResult::Success(message) => {
                        info!("Node stopped successfully: {}", message);
                        return Ok(None);
                    }
                    StopResult::Error(cause) => {
                        error!("Node stopped with error: {}", cause);
                        set_critical_failure(log_output_dest, &cause.to_string());
                        return Err(cause);
                    }
                }
            }
            Some(NodeCtrl::Update(_delay)) => {
                // TODO: implement self-update once antnode app releases are published again
                println!("No self-update supported yet.");
            }
            None => {
                info!("Internal node ctrl cmds channel has been closed, restarting node");
                break Err(eyre!("Internal node ctrl cmds channel has been closed"));
            }
        }
    }
}

fn monitor_node_events(mut node_events_rx: NodeEventsReceiver, ctrl_tx: mpsc::Sender<NodeCtrl>) {
    let _handle = tokio::spawn(async move {
        loop {
            match node_events_rx.recv().await {
                Ok(NodeEvent::ConnectedToNetwork) => Marker::NodeConnectedToNetwork.log(),
                Ok(NodeEvent::ChannelClosed) | Err(RecvError::Closed) => {
                    if let Err(err) = ctrl_tx
                        .send(NodeCtrl::Stop {
                            delay: Duration::from_secs(1),
                            result: StopResult::Error(eyre!("Node events channel closed!")),
                        })
                        .await
                    {
                        error!("Failed to send node control msg to antnode bin main thread: {err}");
                        break;
                    }
                }
                Ok(NodeEvent::TerminateNode(reason)) => {
                    if let Err(err) = ctrl_tx
                        .send(NodeCtrl::Stop {
                            delay: Duration::from_secs(1),
                            result: StopResult::Error(eyre!("Node terminated due to: {reason:?}")),
                        })
                        .await
                    {
                        error!("Failed to send node control msg to antnode bin main thread: {err}");
                        break;
                    }
                }
                Ok(event) => {
                    /* we ignore other events */
                    debug!("Currently ignored node event {event:?}");
                }
                Err(RecvError::Lagged(n)) => {
                    warn!("Skipped {n} node events!");
                    continue;
                }
            }
        }
    });
}

fn init_logging(opt: &Opt, peer_id: PeerId) -> Result<(String, ReloadHandle, Option<WorkerGuard>)> {
    let logging_targets = vec![
        ("ant_bootstrap".to_string(), Level::INFO),
        ("ant_build_info".to_string(), Level::DEBUG),
        ("ant_evm".to_string(), Level::DEBUG),
        ("ant_logging".to_string(), Level::DEBUG),
        ("ant_node".to_string(), Level::INFO),
        ("ant_protocol".to_string(), Level::DEBUG),
        ("antnode".to_string(), Level::INFO),
        ("evmlib".to_string(), Level::DEBUG),
    ];

    let output_dest = match &opt.log_output_dest {
        LogOutputDestArg::Stdout => LogOutputDest::Stdout,
        LogOutputDestArg::DataDir => {
            let path = get_antnode_root_dir(peer_id)?.join("logs");
            LogOutputDest::Path(path)
        }
        LogOutputDestArg::Path(path) => LogOutputDest::Path(path.clone()),
    };

    #[cfg(not(feature = "otlp"))]
    let (reload_handle, log_appender_guard) = {
        let mut log_builder = ant_logging::LogBuilder::new(logging_targets);
        log_builder.output_dest(output_dest.clone());
        log_builder.format(opt.log_format.unwrap_or(LogFormat::Default));
        if let Some(files) = opt.max_log_files {
            log_builder.max_log_files(files);
        }
        if let Some(files) = opt.max_archived_log_files {
            log_builder.max_archived_log_files(files);
        }

        log_builder.initialize()?
    };

    #[cfg(feature = "otlp")]
    let (_rt, reload_handle, log_appender_guard) = {
        // init logging in a separate runtime if we are sending traces to an opentelemetry server
        let rt = Runtime::new()?;
        let (reload_handle, log_appender_guard) = rt.block_on(async {
            let mut log_builder = ant_logging::LogBuilder::new(logging_targets);
            log_builder.output_dest(output_dest.clone());
            log_builder.format(opt.log_format.unwrap_or(LogFormat::Default));
            if let Some(files) = opt.max_log_files {
                log_builder.max_log_files(files);
            }
            if let Some(files) = opt.max_archived_log_files {
                log_builder.max_archived_log_files(files);
            }
            log_builder.initialize()
        })?;
        (rt, reload_handle, log_appender_guard)
    };

    Ok((output_dest.to_string(), reload_handle, log_appender_guard))
}

/// Starts a new process running the binary with the same args as the current process.
/// Injects `--root-dir` and `--port` at the beginning if they are not already present,
/// to ensure the node retains its peer ID on restart.
fn start_new_node_process(root_dir: PathBuf, port: u16) {
    // The invoked path is the path the process was actually launched with. In the case where a
    // symlink was used, this path will be different from the current exe path, which is the source
    // of the symlink, or the canonical path. The invoked path will be preferred for the automatic
    // upgrading process, to support the use of symlinked binaries.
    let invoked_path = env::args().next().expect("failed to get invoked path");
    let current_exe_path = env::current_exe().expect("could not get current executable path");
    let executable_path = if Path::new(&invoked_path).exists() {
        invoked_path
    } else {
        current_exe_path.to_string_lossy().to_string()
    };

    let mut args: Vec<String> = env::args().collect();
    debug!("Arguments used for the initial process: {args:?}");

    // Handle "(deleted)" suffix
    let executable_path = if executable_path.contains(" (deleted)") {
        warn!(
            "The executable path contains ' (deleted)', which indicates the binary was replaced.
             This has been removed from the exe location string"
        );
        executable_path.replace(" (deleted)", "")
    } else {
        executable_path
    };

    // Remove the --first flag: it's only valid for the initial start.
    args.retain(|arg| arg != "--first");

    let has_root_dir = args.iter().any(|arg| arg == "--root-dir");
    let has_port = args.iter().any(|arg| arg == "--port");
    let mut final_args = Vec::new();

    // To have the restarted node retain its peer ID, inject --root-dir and --port at the beginning
    // if they are not already present.
    // The arguments must be at the beginning rather than appended at the end because the EVM
    // options must be at the end.
    if !has_root_dir {
        final_args.push("--root-dir".to_string());
        final_args.push(root_dir.to_string_lossy().to_string());
    }
    if !has_port {
        final_args.push("--port".to_string());
        final_args.push(port.to_string());
    }

    final_args.extend_from_slice(&args[1..]);

    let mut cmd = Command::new(&executable_path);
    cmd.args(&final_args);

    info!("A new process will be launched: {cmd:?}");
    let _handle = match cmd.spawn() {
        Ok(status) => status,
        Err(e) => {
            // Do not return an error as this isn't a critical failure.
            // The current node can continue.
            eprintln!("Failed to execute restart: {e:?}");
            error!("Failed to execute restart: {e:?}");
            return;
        }
    };
}
