use crate::metrics;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use environment::RuntimeContext;
use eth2_libp2p::NetworkGlobals;
use futures::{FutureExt, StreamExt, TryFutureExt};
use parking_lot::Mutex;
use slog::{debug, error, info, warn};
use slot_clock::SlotClock;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::{interval_at, Instant};
use types::{EthSpec, Slot};

/// Create a warning log whenever the peer count is at or below this value.
pub const WARN_PEER_COUNT: usize = 1;

const SECS_PER_MINUTE: f64 = 60.0;
const SECS_PER_HOUR: f64 = 3600.0;
const SECS_PER_DAY: f64 = 86400.0; // non-leap
const SECS_PER_WEEK: f64 = 604_800.0; // non-leap
const DAYS_PER_WEEK: f64 = 7.0;
const HOURS_PER_DAY: f64 = 24.0;
const MINUTES_PER_HOUR: f64 = 60.0;

/// The number of historical observations that should be used to determine the average sync time.
const SPEEDO_OBSERVATIONS: usize = 4;

/// Spawns a notifier service which periodically logs information about the node.
pub fn spawn_notifier<T: BeaconChainTypes>(
    context: RuntimeContext<T::EthSpec>,
    beacon_chain: Arc<BeaconChain<T>>,
    network: Arc<NetworkGlobals<T::EthSpec>>,
    milliseconds_per_slot: u64,
) -> Result<tokio::sync::oneshot::Sender<()>, String> {
    let log = context.log.clone();

    let slot_duration = Duration::from_millis(milliseconds_per_slot);
    let duration_to_next_slot = beacon_chain
        .slot_clock
        .duration_to_next_slot()
        .ok_or_else(|| "slot_notifier unable to determine time to next slot")?;

    // Run this half way through each slot.
    let start_instant = Instant::now() + duration_to_next_slot + (slot_duration / 2);

    // Run this each slot.
    let interval_duration = slot_duration;

    let speedo = Mutex::new(Speedo::default());

    // Note: `interval_at` panics when interval_duration is 0
    // TODO: `Return type of closure passed to `for_each` is restricted to `Future<Output = ()>`
    // Hence, shifting the .then() error logs into the `for_each` closure.
    // Can be solved with `TryStreamExt::try_for_each` if `Interval` implemented `TryStream`.
    // Check if this can be refactored better.
    let interval_future = interval_at(start_instant, interval_duration).for_each(|_| {
        let connected_peer_count = network.connected_peers();

        let head_info = match beacon_chain.head_info() {
            Ok(head) => head,
            Err(e) => {
                error!(
                    log,
                    "Notifier failed to notify, Failed to get beacon chain head info";
                    "error" => format!("{:?}", e)
                );
                return futures::future::ready(());
            }
        };

        let head_slot = head_info.slot;
        let head_epoch = head_slot.epoch(T::EthSpec::slots_per_epoch());
        let current_slot = match beacon_chain.slot() {
            Ok(slot) => slot,
            Err(e) => {
                error!(
                    log,
                    "Notify failed to notify, Unable to read current slot";
                    "error" => format!("{:?}", e)
                );
                return futures::future::ready(());
            }
        };

        let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
        let finalized_epoch = head_info.finalized_checkpoint.epoch;
        let finalized_root = head_info.finalized_checkpoint.root;
        let head_root = head_info.block_root;

        let mut speedo = speedo.lock();
        speedo.observe(head_slot, Instant::now());

        metrics::set_gauge(
            &metrics::SYNC_SLOTS_PER_SECOND,
            speedo.slots_per_second().unwrap_or_else(|| 0_f64) as i64,
        );

        // The next two lines take advantage of saturating subtraction on `Slot`.
        let head_distance = current_slot - head_slot;

        if connected_peer_count <= WARN_PEER_COUNT {
            warn!(log, "Low peer count"; "peer_count" => peer_count_pretty(connected_peer_count));
        }

        debug!(
            log,
            "Slot timer";
            "peers" => peer_count_pretty(connected_peer_count),
            "finalized_root" => format!("{}", finalized_root),
            "finalized_epoch" => finalized_epoch,
            "head_block" => format!("{}", head_root),
            "head_slot" => head_slot,
            "current_slot" => current_slot,
        );

        if head_epoch + 1 < current_epoch {
            let distance = format!(
                "{} slots ({})",
                head_distance.as_u64(),
                slot_distance_pretty(head_distance, slot_duration)
            );

            info!(
                log,
                "Syncing";
                "peers" => peer_count_pretty(connected_peer_count),
                "distance" => distance,
                "speed" => sync_speed_pretty(speedo.slots_per_second()),
                "est_time" => estimated_time_pretty(speedo.estimated_time_till_slot(current_slot)),
            );

            return futures::future::ready(());
        };

        macro_rules! not_quite_synced_log {
                ($message: expr) => {
                    info!(
                        log,
                        $message;
                        "peers" => peer_count_pretty(connected_peer_count),
                        "finalized_root" => format!("{}", finalized_root),
                        "finalized_epoch" => finalized_epoch,
                        "head_slot" => head_slot,
                        "current_slot" => current_slot,
                    );
                }
            }

        if head_epoch + 1 == current_epoch {
            not_quite_synced_log!("Synced to previous epoch")
        } else if head_slot != current_slot {
            not_quite_synced_log!("Synced to current epoch")
        } else {
            info!(
                log,
                "Synced";
                "peers" => peer_count_pretty(connected_peer_count),
                "finalized_root" => format!("{}", finalized_root),
                "finalized_epoch" => finalized_epoch,
                "epoch" => current_epoch,
                "slot" => current_slot,
            );
        };

        futures::future::ready(())
    });

    let (exit_signal, exit) = tokio::sync::oneshot::channel();

    let future = futures::future::select(interval_future, exit.map_err(|_| ()).map(|_| ()));

    // TODO: check if the runtime handle should spawn this future.
    tokio::task::spawn(future);

    Ok(exit_signal)
}

/// Returns the peer count, returning something helpful if it's `usize::max_value` (effectively a
/// `None` value).
fn peer_count_pretty(peer_count: usize) -> String {
    if peer_count == usize::max_value() {
        String::from("--")
    } else {
        format!("{}", peer_count)
    }
}

/// Returns a nicely formatted string describing the rate of slot imports per second.
fn sync_speed_pretty(slots_per_second: Option<f64>) -> String {
    if let Some(slots_per_second) = slots_per_second {
        format!("{:.2} slots/sec", slots_per_second)
    } else {
        "--".into()
    }
}

/// Returns a nicely formatted string how long will we reach the target slot.
fn estimated_time_pretty(seconds_till_slot: Option<f64>) -> String {
    if let Some(seconds_till_slot) = seconds_till_slot {
        seconds_pretty(seconds_till_slot)
    } else {
        "--".into()
    }
}

/// Returns a nicely formatted string describing the `slot_span` in terms of weeks, days, hours
/// and/or minutes.
fn slot_distance_pretty(slot_span: Slot, slot_duration: Duration) -> String {
    if slot_duration == Duration::from_secs(0) {
        return String::from("Unknown");
    }

    let secs = (slot_duration * slot_span.as_u64() as u32).as_secs();
    seconds_pretty(secs as f64)
}

/// Returns a nicely formatted string describing the `slot_span` in terms of weeks, days, hours
/// and/or minutes.
fn seconds_pretty(secs: f64) -> String {
    if secs <= 0.0 {
        return "--".into();
    }

    let weeks = secs / SECS_PER_WEEK;
    let days = secs / SECS_PER_DAY;
    let hours = secs / SECS_PER_HOUR;
    let minutes = secs / SECS_PER_MINUTE;

    if weeks.floor() > 0.0 {
        format!(
            "{:.0} weeks {:.0} days",
            weeks,
            (days % DAYS_PER_WEEK).round()
        )
    } else if days.floor() > 0.0 {
        format!(
            "{:.0} days {:.0} hrs",
            days,
            (hours % HOURS_PER_DAY).round()
        )
    } else if hours.floor() > 0.0 {
        format!(
            "{:.0} hrs {:.0} mins",
            hours,
            (minutes % MINUTES_PER_HOUR).round()
        )
    } else {
        format!("{:.0} mins", minutes.round())
    }
}

/// "Speedo" is Australian for speedometer. This struct observes syncing times.
#[derive(Default)]
pub struct Speedo(Vec<(Slot, Instant)>);

impl Speedo {
    /// Observe that we were at some `slot` at the given `instant`.
    pub fn observe(&mut self, slot: Slot, instant: Instant) {
        if self.0.len() > SPEEDO_OBSERVATIONS {
            self.0.remove(0);
        }

        self.0.push((slot, instant));
    }

    /// Returns the average of the speeds between each observation.
    ///
    /// Does not gracefully handle slots that are above `u32::max_value()`.
    pub fn slots_per_second(&self) -> Option<f64> {
        let speeds = self
            .0
            .windows(2)
            .filter_map(|windows| {
                let (slot_a, instant_a) = windows[0];
                let (slot_b, instant_b) = windows[1];

                // Taking advantage of saturating subtraction on `Slot`.
                let distance = f64::from((slot_b - slot_a).as_u64() as u32);

                let seconds = f64::from((instant_b - instant_a).as_millis() as u32) / 1_000.0;

                if seconds > 0.0 {
                    Some(distance / seconds)
                } else {
                    None
                }
            })
            .collect::<Vec<f64>>();

        let count = speeds.len();
        let sum: f64 = speeds.iter().sum();

        if count > 0 {
            Some(sum / f64::from(count as u32))
        } else {
            None
        }
    }

    /// Returns the time we should reach the given `slot`, judging by the latest observation and
    /// historical average syncing time.
    ///
    /// Returns `None` if the slot is prior to our latest observed slot or we have not made any
    /// observations.
    pub fn estimated_time_till_slot(&self, target_slot: Slot) -> Option<f64> {
        let (prev_slot, _) = self.0.last()?;
        let slots_per_second = self.slots_per_second()?;

        if target_slot > *prev_slot && slots_per_second > 0.0 {
            let distance = (target_slot - *prev_slot).as_u64() as f64;
            Some(distance / slots_per_second)
        } else {
            None
        }
    }
}