//! Building a node and starting its workers: what `main` does once it has read the
//! configuration and connected to Mycelium, and what the simulator (`transport::sim`)
//! does for every simulated node. Everything here is independent of where the
//! transport and the settings come from.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use automerge::{ActorId, AutoCommit};
use tokio::sync::{RwLock, broadcast, mpsc, watch};
use tracing::trace;

use crate::document::file::DocumentFile;
use crate::document::replica::{Epoch, Replica};
use crate::files::Files;
use crate::remake;
use crate::state::{AppState, NodeLifecycle, Timing};
use crate::swarm::auth::SwarmKey;
use crate::swarm::state_file::StateFile;
use crate::swarm::sync::SyncEvent;
use crate::swarm::{self, peer, sync};
use crate::transport::Transport;
use crate::transport::firewall::{Firewall, FirewallTransport};
use crate::transport::traffic::{CountingTransport, Traffic};

/// What a node is built from.
pub(crate) struct NodeConfig {
    /// The node's transport, without the counting and firewall layers (added here).
    pub(crate) transport: Arc<dyn Transport>,
    pub(crate) initial_epoch: Epoch,
    pub(crate) initial_doc: AutoCommit,
    /// Actors this node wrote with in earlier runs (`--document-file`).
    pub(crate) earlier_actors: Vec<ActorId>,
    /// `Timing::from_keepalive(--keepalive)`.
    pub(crate) timing: Timing,
    pub(crate) neighbours_k: usize,
    pub(crate) alias: Option<String>,
    pub(crate) electorate_window: Duration,
    pub(crate) blob_chunk_bytes: u32,
    pub(crate) swarm_key: Option<Arc<SwarmKey>>,
    pub(crate) protocol_writers: HashSet<String>,
    pub(crate) files: Arc<Files>,
    pub(crate) state_file: Option<Arc<StateFile>>,
    pub(crate) document_file: Option<Arc<DocumentFile>>,
    /// For `GET /admin/config`.
    pub(crate) config_report: serde_json::Value,
}

/// A built node whose workers haven't started yet.
pub(crate) struct Node {
    pub(crate) state: AppState,
    sync_rx: mpsc::UnboundedReceiver<SyncEvent>,
}

impl Node {
    pub(crate) fn new(config: NodeConfig) -> Self {
        let (sync_tx, sync_rx) = mpsc::unbounded_channel();
        let (change_tx, _) = watch::channel(0_u64);
        let (change_events, _) = broadcast::channel(256);
        let (epoch_tx, _) = watch::channel(config.initial_epoch);
        let firewall = Arc::new(Firewall::default());
        let traffic = Arc::new(Traffic::default());
        let initial_actor = config.initial_doc.get_actor().clone();

        let state = AppState {
            doc: Arc::new(RwLock::new(Replica::new(
                config.initial_epoch,
                config.initial_doc,
            ))),
            epoch_tx,
            lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining {
                discovered_swarm_id: None,
            })),
            peer_states: Arc::new(RwLock::new(HashMap::new())),
            // Counted outside the firewall: what it drops outbound counts as sent (it
            // plays a lossy network), what it drops inbound never arrived.
            transport: Arc::new(CountingTransport::new(
                Arc::new(FirewallTransport::new(
                    config.transport,
                    Arc::clone(&firewall),
                )),
                Arc::clone(&traffic),
            )),
            firewall,
            traffic,
            stats: Default::default(),
            sync_events: sync_tx,
            change_tx,
            initial_sync_done: Arc::new(AtomicBool::new(false)),
            change_events,
            timing: config.timing,
            remake: Arc::new(remake::RemakeState::default()),
            swarm_key: config.swarm_key,
            run_id: uuid::Uuid::new_v4().into_bytes(),
            neighbours_k: config.neighbours_k,
            alias: config.alias,
            electorate_window: config.electorate_window,
            blobs: Arc::new(swarm::blob::Blobs::new(config.blob_chunk_bytes)),
            protocol_writers: Arc::new(config.protocol_writers),
            state_file: config.state_file,
            document_file: config.document_file,
            files: config.files,
            config: Arc::new(config.config_report),
            neighbours: Default::default(),
            topology_changed: Default::default(),
            pulling: Default::default(),
            announce_pending: Default::default(),
            rtt: Default::default(),
            throughput: Default::default(),
            stability: Default::default(),
            stale: Default::default(),
            catch_up_source: Default::default(),
            catch_up_preference: Default::default(),
            catch_up_avoided: Default::default(),
        };
        for actor in config.earlier_actors {
            state.remake.add_own_actor(actor);
        }
        state.remake.add_own_actor(initial_actor);
        trace!("Application state initialized");
        Self { state, sync_rx }
    }

    /// Starts what runs from the start: the message consumer and the sync, peer and
    /// remake workers. The remake worker starts before bootstrapping: a node joining a
    /// compacted swarm can only finish its initial sync after fetching the current epoch
    /// from a peer.
    pub(crate) fn spawn_workers(self) -> AppState {
        let Self { state, sync_rx } = self;

        let consumer_state = state.clone();
        tokio::spawn(async move {
            let inner_state = consumer_state.clone();
            let _ = consumer_state
                .transport
                .start_consumer(Box::new(move |sender, msg, _payload_len| {
                    let state_clone = inner_state.clone();
                    Box::pin(async move { peer::handle_message(&state_clone, sender, msg).await })
                }))
                .await;
        });
        trace!("Message consumer task spawned");

        let sync_state = state.clone();
        tokio::spawn(async move {
            sync::run_sync_worker(sync_state, sync_rx).await;
        });
        trace!("Sync worker task spawned");

        let peer_worker_state = state.clone();
        tokio::spawn(async move {
            peer::run_peer_worker(peer_worker_state).await;
        });
        trace!("Peer worker task spawned");

        tokio::spawn(remake::run_worker(state.clone()));
        trace!("Remake worker task spawned");

        tokio::spawn(swarm::topology::run_worker(state.clone()));
        trace!("Topology worker task spawned");

        state
    }
}

/// Starts what runs once bootstrap is done: the change publisher (streams only exist
/// from then on, and it avoids diffing the initial sync) and the document file writer.
pub(crate) fn spawn_ready_workers(state: &AppState) {
    let publisher_state = state.clone();
    tokio::spawn(async move { sync::run_change_publisher(publisher_state).await });
    tokio::spawn(crate::document::file::run(state.clone()));
}