// Copyright 2020. The Tari Project
//
// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
// following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
// disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
// following disclaimer in the documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
// products derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#![allow(clippy::mutex_atomic)]
#![allow(clippy::indexing_slicing)]
use std::{
    collections::HashMap,
    fmt,
    iter,
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

use futures::future;
use once_cell::sync::Lazy;
use rand::{RngExt, distr};
use tari_common_sqlite::connection::{DbConnection, DbConnectionUrl};
use tari_comms::{
    CommsBuilder,
    CommsNode,
    PeerConnection,
    RefKind,
    backoff::ConstantBackoff,
    connection_manager::{ConnectionDirection, ConnectionManagerEvent},
    peer_manager::{
        NodeId,
        NodeIdentity,
        Peer,
        PeerFeatures,
        database::{MIGRATIONS, PeerDatabaseSql},
    },
    pipeline::{self, SinkService},
    protocol::{
        ProtocolId,
        messaging::{MessagingEvent, MessagingEventReceiver, MessagingEventSender, MessagingProtocolExtension},
        rpc::RpcServer,
    },
    transports::MemoryTransport,
    types::CommsDatabase,
};
use tari_comms_dht::{
    Dht,
    DhtConfig,
    domain_message::OutboundDomainMessage,
    envelope::NodeDestination,
    inbound::DecryptedDhtMessage,
    outbound::OutboundEncryption,
};
use tari_shutdown::{Shutdown, ShutdownSignal};
use tari_test_utils::streams::convert_unbounded_mpsc_to_stream;
use tokio::{
    runtime,
    sync::{broadcast, mpsc},
    task,
    time,
};
use tower::ServiceBuilder;

use crate::memory_net::{DrainBurst, create_test_peer};

pub static MEMORYNET_MSG_PROTOCOL_ID: ProtocolId = ProtocolId::from_static(b"t/msg/1.0");
pub type NodeEventRx = mpsc::UnboundedReceiver<(NodeId, NodeId)>;
pub type NodeEventTx = mpsc::UnboundedSender<(NodeId, NodeId)>;

#[macro_export]
macro_rules! banner {
    ($($arg: tt)*) => {
        println!();
        println!("----------------------------------------------------------");
        println!($($arg)*);
        println!("----------------------------------------------------------");
        println!();
    }
}

static NAME_MAP: Lazy<Mutex<HashMap<NodeId, String>>> = Lazy::new(|| Mutex::new(HashMap::new()));
static NAME_POS: Lazy<Mutex<usize>> = Lazy::new(|| Mutex::new(0));

pub fn register_name(node_id: NodeId, name: String) {
    NAME_MAP.lock().unwrap().insert(node_id, name);
}

pub fn get_name(node_id: &NodeId) -> String {
    NAME_MAP
        .lock()
        .unwrap()
        .get(node_id)
        .map(|name| format!("{} ({})", name, node_id.short_str()))
        .unwrap_or_else(|| format!("NoName ({})", node_id.short_str()))
}

pub fn get_short_name(node_id: &NodeId) -> String {
    NAME_MAP
        .lock()
        .unwrap()
        .get(node_id)
        .map(|name| name.to_string())
        .unwrap_or_else(|| format!("NoName ({})", node_id.short_str()))
}

pub fn get_next_name() -> String {
    let pos = {
        let mut i = NAME_POS.lock().unwrap();
        *i += 1;
        *i - 1
    };

    format!("Node{pos}")
}

pub async fn shutdown_all(nodes: Vec<TestNode>) {
    let tasks = nodes.into_iter().map(|node| node.shutdown());
    future::join_all(tasks).await;
}

#[allow(clippy::similar_names)]
pub async fn discovery(wallets: &[TestNode], messaging_events_rx: &mut NodeEventRx) -> (usize, usize, usize) {
    let mut successes = 0;
    let mut total_messages = 0;
    let mut total_time = Duration::from_secs(0);
    for i in 0..wallets.len() - 1 {
        let wallet1 = wallets.get(i).unwrap();
        let wallet2 = wallets.get(i + 1).unwrap();

        banner!("🌎 '{wallet1}' is going to try discover '{wallet2}'.");

        let start = Instant::now();
        let discovery_result = wallet1
            .dht
            .discovery_service_requester()
            .discover_peer(
                wallet2.node_identity().public_key().clone(),
                wallet2.node_identity().public_key().clone().into(),
            )
            .await;

        match discovery_result {
            Ok(peer) => {
                successes += 1;
                total_time += start.elapsed();
                banner!(
                    "⚡️🎉😎 '{}' discovered peer '{}' ({}) in {:.2?}",
                    wallet1,
                    get_name(&peer.node_id),
                    peer,
                    start.elapsed()
                );

                time::sleep(Duration::from_secs(5)).await;
                total_messages += drain_messaging_events(messaging_events_rx, false).await;
            },
            Err(err) => {
                banner!(
                    "💩 '{}' failed to discover '{}' after {:.2?} because '{}'",
                    wallet1,
                    wallet2,
                    start.elapsed(),
                    err
                );

                time::sleep(Duration::from_secs(5)).await;
                total_messages += drain_messaging_events(messaging_events_rx, false).await;
            },
        }
    }

    banner!(
        "✨ The set of discoveries succeeded {} out of {} times and took a total of {:.1}s with {} messages sent.",
        successes,
        wallets.len() - 1,
        total_time.as_secs_f32(),
        total_messages
    );
    (total_messages, successes, wallets.len() - 1)
}

pub async fn network_peer_list_stats(nodes: &[TestNode], wallets: &[TestNode]) {
    let mut stats = HashMap::<String, usize>::with_capacity(wallets.len());
    for wallet in wallets {
        let mut num_known = 0;
        for node in nodes {
            if node
                .comms
                .peer_manager()
                .exists_node_id(wallet.node_identity().node_id())
                .await
                .unwrap()
            {
                num_known += 1;
            }
        }
        stats.insert(get_name(wallet.node_identity().node_id()), num_known);
    }

    let mut avg = Vec::with_capacity(wallets.len());
    for (n, v) in stats {
        let perc = v as f32 / nodes.len() as f32;
        avg.push(perc);
        println!(
            "{} is known by {} out of {} nodes ({:.2}%)",
            n,
            v,
            nodes.len(),
            perc * 100.0
        );
    }
    println!(
        "Average {:.2}%",
        avg.into_iter().sum::<f32>() / wallets.len() as f32 * 100.0
    );
}

pub async fn network_connectivity_stats(nodes: &[TestNode], wallets: &[TestNode], quiet_mode: bool) {
    pub async fn display(nodes: &[TestNode], quiet_mode: bool) -> (usize, usize) {
        let mut total = 0;
        let mut avg = Vec::new();
        for node in nodes {
            let conns = node.comms.connectivity().get_active_connections().await.unwrap();
            total += conns.len();
            avg.push(conns.len());

            println!("{} connected to {} nodes", node, conns.len());
            if !quiet_mode {
                for c in conns {
                    println!("  {} ({})", get_name(c.peer_node_id()), c.direction());
                }
            }
        }
        (total, avg.into_iter().sum())
    }
    let (mut total, mut avg) = display(nodes, quiet_mode).await;
    let (t, a) = display(wallets, quiet_mode).await;
    total += t;
    avg += a;
    println!(
        "{} total connections on the network. ({} per peer on average)",
        total,
        avg / (wallets.len() + nodes.len())
    );
}

#[allow(clippy::too_many_lines)]
pub async fn do_network_wide_propagation(nodes: &mut [TestNode], origin_node_index: Option<usize>) -> (usize, usize) {
    let random_node = match origin_node_index {
        Some(n) if n < nodes.len() => &nodes[n],
        Some(_) | None => &nodes[rand::rng().random_range(0..nodes.len() - 1)],
    };

    let random_node_id = random_node.comms.node_identity().node_id().clone();
    const PUBLIC_MESSAGE: &str = "This is something you're all interested in!";

    banner!("🌎 {} is going to broadcast a message to the network", random_node);
    let send_states = random_node
        .dht
        .outbound_requester()
        .broadcast(
            NodeDestination::Unknown,
            OutboundEncryption::ClearText,
            vec![],
            OutboundDomainMessage::new(&0i32, PUBLIC_MESSAGE.to_string()),
            "Memory net example".to_string(),
        )
        .await
        .unwrap();
    let num_connections = random_node
        .comms
        .connectivity()
        .get_active_connections()
        .await
        .unwrap()
        .len();
    let (success, failed) = send_states.wait_all().await;
    println!(
        "🦠 {} broadcast to {}/{} peer(s) ({} connection(s))",
        random_node.name,
        success.len(),
        success.len() + failed.len(),
        num_connections
    );

    let start_global = Instant::now();
    // Spawn task for each peer that will read the message and propagate it on
    let tasks = nodes
        .iter_mut()
        .filter(|n| n.comms.node_identity().node_id() != &random_node_id)
        .enumerate()
        .map(|(idx, node)| {
            let mut outbound_req = node.dht.outbound_requester();
            let mut connectivity = node.comms.connectivity();
            let mut ims_rx = node.ims_rx.take().unwrap();
            let start = Instant::now();
            let node_name = node.name.clone();

            task::spawn(async move {
                let result = time::timeout(Duration::from_secs(30), ims_rx.recv()).await;
                let mut is_success = false;
                match result {
                    Ok(Some(msg)) => {
                        let public_msg = msg
                            .decryption_result
                            .unwrap()
                            .decode_part::<String>(1)
                            .unwrap()
                            .unwrap();
                        println!(
                            "📬 {} got public message '{}' (t={:.0?})",
                            node_name,
                            public_msg,
                            start_global.elapsed(),
                        );
                        is_success = true;
                        let send_states = outbound_req
                            .propagate(
                                NodeDestination::Unknown,
                                OutboundEncryption::ClearText,
                                vec![msg.source_peer.node_id.clone()],
                                OutboundDomainMessage::new(&0i32, public_msg),
                                "Memory net example".to_string(),
                            )
                            .await
                            .unwrap();
                        let num_connections = connectivity.get_active_connections().await.unwrap().len();
                        let (success, failed) = send_states.wait_all().await;
                        println!(
                            "🦠 {} propagated to {}/{} peer(s) ({} connection(s))",
                            node_name,
                            success.len(),
                            success.len() + failed.len(),
                            num_connections
                        );
                    },
                    Err(_) | Ok(None) => {
                        banner!(
                            "💩 {} failed to receive network message after {:.2?}",
                            node_name,
                            start.elapsed(),
                        );
                    },
                }

                (idx, ims_rx, is_success)
            })
        });

    // Put the ims_rxs back
    let ims_rxs = future::join_all(tasks).await;
    let mut num_successes = 0;
    for ims in ims_rxs {
        let (idx, ims_rx, is_success) = ims.unwrap();
        nodes[idx].ims_rx = Some(ims_rx);
        if is_success {
            num_successes += 1;
        }
    }

    banner!(
        "🙌 Finished propagation test. {} out of {} nodes received the message",
        num_successes,
        nodes.len() - 1
    );
    (num_successes, nodes.len() - 1)
}

pub async fn drain_messaging_events(messaging_rx: &mut NodeEventRx, show_logs: bool) -> usize {
    let stream = convert_unbounded_mpsc_to_stream(messaging_rx);
    tokio::pin!(stream);
    let drain_fut = DrainBurst::new(&mut stream);
    if show_logs {
        let messages = drain_fut.await;
        let num_messages = messages.len();
        let mut node_id_buf = Vec::new();
        let mut last_from_node = None;
        for (from_node, to_node) in &messages {
            match &last_from_node {
                Some(node_id) if *node_id == from_node => {
                    node_id_buf.push(to_node);
                },
                Some(_) => {
                    println!(
                        "📨 {} sent {} messages to {}.️",
                        get_short_name(last_from_node.take().unwrap()),
                        node_id_buf.len(),
                        node_id_buf.drain(..).map(get_short_name).collect::<Vec<_>>().join(", ")
                    );

                    last_from_node = Some(from_node);
                    node_id_buf.push(to_node)
                },
                None => {
                    last_from_node = Some(from_node);
                    node_id_buf.push(to_node)
                },
            }
        }
        println!("{num_messages} messages sent between nodes");
        num_messages
    } else {
        let len = drain_fut.await.len();
        println!("📨 {len} messages exchanged");
        len
    }
}

fn connection_manager_logger(
    node_id: &NodeId,
    quiet_mode: bool,
) -> impl FnMut(Arc<ConnectionManagerEvent>) -> Arc<ConnectionManagerEvent> {
    let node_name = get_name(node_id);
    move |event| {
        if quiet_mode {
            return event;
        }
        #[allow(clippy::enum_glob_use)]
        use ConnectionManagerEvent::*;
        print!("EVENT: ");
        match &*event {
            PeerConnected(conn) => match conn.direction() {
                ConnectionDirection::Inbound => {
                    // println!(
                    //     "'{}' got inbound connection from '{}'",
                    //     node_name,
                    //     get_name(conn.peer_node_id()),
                    // );
                },
                ConnectionDirection::Outbound => {
                    println!("'{}' connected to '{}'", node_name, get_name(conn.peer_node_id()),);
                },
            },
            PeerDisconnected(_, node_id, minimized) => {
                println!(
                    "'{}' disconnected from '{}', {:?}",
                    get_name(node_id),
                    node_name,
                    minimized
                );
            },
            PeerConnectFailed(node_id, err) => {
                println!(
                    "'{}' failed to connect to '{}' because '{:?}'",
                    node_name,
                    get_name(node_id),
                    err
                );
            },
            PeerInboundConnectFailed(err) => {
                println!("'{node_name}' failed to accept inbound connection because '{err:?}'");
            },
            NewInboundSubstream(node_id, protocol, _) => {
                println!(
                    "'{}' negotiated protocol '{}' to '{}'",
                    get_name(node_id),
                    String::from_utf8_lossy(protocol),
                    node_name
                );
            },
            PeerViolation { peer_node_id, details } => {
                println!(
                    "'{}' violated protocol with '{}' because '{}'",
                    node_name,
                    get_name(peer_node_id),
                    details
                );
            },
        }
        event
    }
}

pub struct TestNode {
    pub name: String,
    pub comms: CommsNode,
    pub seed_peers: Vec<Peer>,
    pub dht: Dht,
    pub conn_man_events_rx: mpsc::Receiver<Arc<ConnectionManagerEvent>>,
    pub ims_rx: Option<mpsc::Receiver<DecryptedDhtMessage>>,
    pub messaging_events: MessagingEventSender,
    pub shutdown: Shutdown,
}

impl TestNode {
    pub fn new(
        comms: CommsNode,
        dht: Dht,
        seed_peers: Vec<Peer>,
        ims_rx: mpsc::Receiver<DecryptedDhtMessage>,
        node_messsage_tx: NodeEventTx,
        messaging_events: MessagingEventSender,
        quiet_mode: bool,
        shutdown: Shutdown,
    ) -> Self {
        let name = get_next_name();
        register_name(comms.node_identity().node_id().clone(), name.clone());

        let (conn_man_events_tx, events_rx) = mpsc::channel(100);
        Self::spawn_event_monitor(
            &comms,
            messaging_events.subscribe(),
            conn_man_events_tx,
            node_messsage_tx,
            quiet_mode,
        );

        Self {
            name,
            seed_peers,
            comms,
            dht,
            ims_rx: Some(ims_rx),
            conn_man_events_rx: events_rx,
            messaging_events,
            shutdown,
        }
    }

    fn spawn_event_monitor(
        comms: &CommsNode,
        mut messaging_events: MessagingEventReceiver,
        events_tx: mpsc::Sender<Arc<ConnectionManagerEvent>>,
        messaging_events_tx: NodeEventTx,
        quiet_mode: bool,
    ) {
        let mut conn_man_event_sub = comms.subscribe_connection_manager_events();
        let executor = runtime::Handle::current();

        let node_id = comms.node_identity().node_id().clone();
        executor.spawn(async move {
            let mut logger = connection_manager_logger(&node_id, quiet_mode);
            loop {
                match conn_man_event_sub.recv().await {
                    Ok(event) => {
                        let _result = events_tx.send(logger(event)).await;
                    },
                    Err(broadcast::error::RecvError::Closed) => break,
                    Err(err) => log::error!("{err}"),
                }
            }
        });

        let node_id = comms.node_identity().node_id().clone();
        executor.spawn(async move {
            loop {
                let event = messaging_events.recv().await;
                use MessagingEvent::MessageReceived;
                match event {
                    Ok(MessageReceived(peer_node_id, _)) => {
                        messaging_events_tx.send((peer_node_id, node_id.clone())).unwrap();
                    },
                    Err(broadcast::error::RecvError::Closed) => {
                        break;
                    },
                    _ => {},
                }
            }
        });
    }

    pub fn node_identity(&self) -> Arc<NodeIdentity> {
        self.comms.node_identity()
    }

    pub fn to_peer(&self) -> Peer {
        self.comms.node_identity().to_peer()
    }

    #[allow(dead_code)]
    pub async fn expect_peer_connection(&mut self, node_id: &NodeId) -> Option<PeerConnection> {
        if let Some(conn) = self
            .comms
            .connectivity()
            .get_connection(node_id.clone(), RefKind::Weak)
            .await
            .unwrap()
        {
            return Some(conn);
        }
        use ConnectionManagerEvent::PeerConnected;
        loop {
            let event = time::timeout(Duration::from_secs(30), self.conn_man_events_rx.recv())
                .await
                .ok()??;

            match &*event {
                PeerConnected(conn) if conn.peer_node_id() == node_id => {
                    break Some(*conn.clone());
                },
                _ => {},
            }
        }
    }

    pub async fn shutdown(mut self) {
        self.shutdown.trigger();
        self.comms.wait_until_shutdown().await;
    }
}

impl AsRef<TestNode> for TestNode {
    fn as_ref(&self) -> &TestNode {
        self
    }
}

impl fmt::Display for TestNode {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "{} ({})",
            self.name,
            self.comms.node_identity().node_id().short_str()
        )
    }
}

pub fn make_node_identity(features: PeerFeatures) -> Arc<NodeIdentity> {
    let port = MemoryTransport::acquire_next_memsocket_port();
    Arc::new(NodeIdentity::random(
        &mut rand::rng(),
        format!("/memory/{port}").parse().unwrap(),
        features,
    ))
}

fn create_peer_storage() -> PeerDatabaseSql {
    let db_connection = DbConnection::connect_temp_file_and_migrate(MIGRATIONS).unwrap();
    PeerDatabaseSql::new(db_connection, &create_test_peer()).unwrap()
}

pub async fn make_node(
    features: PeerFeatures,
    peer_identities: Vec<Arc<NodeIdentity>>,
    node_event_tx: NodeEventTx,
    num_neighbouring_nodes: usize,
    num_random_nodes: usize,
    propagation_factor: usize,
    quiet_mode: bool,
) -> TestNode {
    let node_identity = make_node_identity(features);
    make_node_from_node_identities(
        node_identity,
        peer_identities,
        node_event_tx,
        num_neighbouring_nodes,
        num_random_nodes,
        propagation_factor,
        quiet_mode,
    )
    .await
}

pub async fn make_node_from_node_identities(
    node_identity: Arc<NodeIdentity>,
    peer_identities: Vec<Arc<NodeIdentity>>,
    node_events_tx: NodeEventTx,
    num_neighbouring_nodes: usize,
    num_random_nodes: usize,
    propagation_factor: usize,
    quiet_mode: bool,
) -> TestNode {
    let (tx, ims_rx) = mpsc::channel(1);
    let seed_peers = peer_identities.iter().map(|n| n.to_peer()).collect::<Vec<_>>();
    let shutdown = Shutdown::new();
    let (comms, dht, messaging_events) = setup_comms_dht(
        node_identity,
        create_peer_storage(),
        tx,
        num_neighbouring_nodes,
        num_random_nodes,
        propagation_factor,
        seed_peers.clone(),
        shutdown.to_signal(),
    )
    .await;

    TestNode::new(
        comms,
        dht,
        seed_peers,
        ims_rx,
        node_events_tx,
        messaging_events,
        quiet_mode,
        shutdown,
    )
}

async fn setup_comms_dht(
    node_identity: Arc<NodeIdentity>,
    storage: CommsDatabase,
    inbound_tx: mpsc::Sender<DecryptedDhtMessage>,
    num_neighbouring_nodes: usize,
    num_random_nodes: usize,
    propagation_factor: usize,
    seed_peers: Vec<Peer>,
    shutdown_signal: ShutdownSignal,
) -> (CommsNode, Dht, MessagingEventSender) {
    // Create inbound and outbound channels
    let (outbound_tx, outbound_rx) = mpsc::unbounded_channel();

    let comms = CommsBuilder::new()
        .allow_test_addresses()
        // In this case the listener address and the public address are the same (/memory/...)
        .with_listener_address(node_identity.first_public_address().unwrap())
        .with_shutdown_signal(shutdown_signal)
        .with_node_identity(node_identity)
        .with_min_connectivity(1)
        .with_peer_storage(storage)
        .with_dial_backoff(ConstantBackoff::new(Duration::from_millis(1000)))
        .build()
        .unwrap();
    for peer in seed_peers {
        comms.peer_manager().add_or_update_peer(peer).await.unwrap();
    }

    let db_name = iter::repeat(())
        .map(|_| rand::rng().sample(distr::Alphanumeric) as char)
        .take(8)
        .collect::<String>();

    let dht = Dht::builder()
        .with_config(DhtConfig {
            // Use MemoryShared because a Memory connection only seems to work on MacOs
            database_url: DbConnectionUrl::MemoryShared(db_name),
            auto_join: false,
            discovery_request_timeout: Duration::from_secs(15),
            num_neighbouring_nodes,
            num_random_nodes,
            propagation_factor,
            network_discovery: Default::default(),
            ..DhtConfig::default_local_test()
        })
        .with_outbound_sender(outbound_tx)
        .build(
            comms.node_identity(),
            comms.peer_manager(),
            comms.connectivity(),
            comms.shutdown_signal(),
        )
        .await
        .unwrap();

    let dht_outbound_layer = dht.outbound_middleware_layer();
    let pipeline = pipeline::Builder::new()
        .with_outbound_pipeline(outbound_rx, |sink| {
            ServiceBuilder::new().layer(dht_outbound_layer).service(sink)
        })
        .max_concurrent_inbound_tasks(10)
        .with_inbound_pipeline(
            ServiceBuilder::new()
                .layer(dht.inbound_middleware_layer())
                .service(SinkService::new(inbound_tx)),
        )
        .build();

    let (messaging_events_tx, _) = broadcast::channel(100);
    let comms = comms
        .add_rpc_server(RpcServer::new().add_service(dht.rpc_service()))
        .add_protocol_extension(
            MessagingProtocolExtension::new(MEMORYNET_MSG_PROTOCOL_ID.clone(), messaging_events_tx.clone(), pipeline)
                .enable_message_received_event(),
        )
        .spawn_with_transport(MemoryTransport)
        .await
        .unwrap();

    (comms, dht, messaging_events_tx)
}

pub async fn take_a_break(num_nodes: usize) {
    banner!("Taking a break for a few seconds to let things settle...");
    time::sleep(Duration::from_millis(num_nodes as u64 * 100)).await;
}
