//
// Copyright (c) 2017, 2020 ADLINK Technology Inc.
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//

//! A Unique Hybrid Logical Clock.
//!
//! This library is an implementation of an
//! [Hybrid Logical Clock (HLC)](https://cse.buffalo.edu/tech-reports/2014-04.pdf)
//! associated to a unique identifier.
//! Thus, it is able to generate timestamps that are unique across a distributed system,
//! without the need of a centralized time source.
//!
//! # Quick Start
//!
//! ```
//! use uhlc::HLC;
//!
//! # async_std::task::block_on(async {
//! // create an HLC with a generated UUID and relying on SystemTime::now()
//! let hlc = HLC::default();
//!
//! // generate timestamps
//! let ts1 = hlc.new_timestamp().await;
//! let ts2 = hlc.new_timestamp().await;
//! assert!(ts2 > ts1);
//!
//! // update the HLC with a timestamp incoming from another HLC
//! // (typically remote, but not in this example...)
//! let hlc2 = HLC::default();
//! let other_ts = hlc2.new_timestamp().await;
//!
//! if ! hlc.update_with_timestamp(&other_ts).await.is_ok() {
//!     println!(r#"The incoming timestamp would make this HLC
//!              to drift too much. You should refuse it!"#);
//! }
//!
//! let ts3 = hlc.new_timestamp().await;
//! assert!(ts3 > ts2);
//! assert!(ts3 > other_ts);
//! # })
//! ```

#![doc(
    html_logo_url = "https://www.rust-lang.org/logos/rust-logo-128x128-blk.png",
    html_favicon_url = "https://www.rust-lang.org/favicon.ico",
    html_root_url = "https://atolab.github.io/uhlc-rs/"
)]

use async_std::sync::Mutex;
use log::warn;
use std::cmp;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

mod id;
pub use id::*;

mod ntp64;
pub use ntp64::*;

mod timestamp;
pub use timestamp::*;

/// The size of counter part in [`NTP64`] (in bits)
pub const CSIZE: u8 = 4u8;
// Bit-mask of the counter part within the 64 bits time
const CMASK: u64 = (1u64 << CSIZE) - 1u64;
// Bit-mask of the logical clock part within the 64 bits time
const LMASK: u64 = !CMASK;

/// HLC Delta in milliseconds: maximum accepted drift for an external timestamp.
///
/// I.e.: if an incoming timestamp has a time > now() + delta, then the HLC is not updated.
pub const DELTA_MS: u64 = 100;

/// An Hybric Logical Clock generating [`Timestamp`]s
pub struct HLC {
    id: ID,
    clock: fn() -> NTP64,
    delta: NTP64,
    last_time: Mutex<NTP64>,
}

macro_rules! asynclock {
    ($var:expr) => {
        if let Some(g) = $var.try_lock() {
            g
        } else {
            $var.lock().await
        }
    };
}

impl HLC {
    /// Create a new [`HLC`] with an `id` (must be unique) and a `clock`
    /// (a function returning the current physical time as an [`NTP64`]).
    ///
    /// # Examples
    ///
    /// ```
    /// use uhlc::HLC;
    ///
    /// # async_std::task::block_on(async {
    /// let hlc = HLC::with_clock(
    ///     uuid::Uuid::new_v4().into(),
    ///     uhlc::system_time_clock);
    /// println!("{}", hlc.new_timestamp().await);
    /// # })
    /// ```
    pub fn with_clock(id: ID, clock: fn() -> NTP64) -> HLC {
        let delta = NTP64::from(Duration::from_millis(DELTA_MS));
        HLC {
            id,
            clock,
            delta,
            last_time: Default::default(),
        }
    }

    /// Create a new [`HLC`] with an `id` (must be unique) and using
    /// [`system_time_clock()`] as physical clock.
    ///
    /// # Examples
    ///
    /// ```
    /// use uhlc::HLC;
    ///
    /// # async_std::task::block_on(async {
    /// let hlc = HLC::with_system_time(uuid::Uuid::new_v4().into());
    /// println!("{}", hlc.new_timestamp().await);
    /// # })
    /// ```
    pub fn with_system_time(id: ID) -> HLC {
        HLC::with_clock(id, system_time_clock)
    }

    /// Generate a new [`Timestamp`].
    ///
    /// This timestamp is unique in the system and is always greater
    /// than the latest timestamp generated by the HLC and than the
    /// latest incoming timestamp that was used to update this [`HLC`]
    /// (using [`HLC::update_with_timestamp()`]).
    ///
    /// # Examples
    ///
    /// ```
    /// use uhlc::HLC;
    ///
    /// # async_std::task::block_on(async {
    /// let hlc = HLC::default();
    /// let ts1 =  hlc.new_timestamp().await;
    /// let ts2 =  hlc.new_timestamp().await;
    /// assert!(ts2 > ts1);
    /// # })
    /// ```
    pub async fn new_timestamp(&self) -> Timestamp {
        let mut now = (self.clock)();
        now.0 &= LMASK;
        let mut last_time = asynclock!(self.last_time);
        if now.0 > (last_time.0 & LMASK) {
            *last_time = now
        } else {
            *last_time += 1;
        }
        Timestamp::new(*last_time, self.id.clone())
    }

    /// Update this [`HLC`] with a [`Timestamp`].
    ///
    /// Typically, this timestamp should have been generated by another HLC.
    /// If the timestamp exceeds the current time of this HLC by more than [`DELTA_MS`]
    /// an [`Err`] is returned.
    ///
    /// # Examples
    ///
    /// ```
    /// use uhlc::HLC;
    ///
    /// # async_std::task::block_on(async {
    /// let hlc1 = HLC::default();
    ///
    /// // update the HLC with a timestamp incoming from another HLC
    /// // (typically remote, but not in this example...)
    /// let hlc2 = HLC::default();
    /// let other_ts = hlc2.new_timestamp().await;
    /// if ! hlc1.update_with_timestamp(&other_ts).await.is_ok() {
    ///     println!(r#"The incoming timestamp would make this HLC
    ///              to drift too much. You should refuse it!"#);
    /// }
    ///
    /// let ts = hlc1.new_timestamp().await;
    /// assert!(ts > other_ts);
    /// # })
    /// ```
    pub async fn update_with_timestamp(&self, timestamp: &Timestamp) -> Result<(), String> {
        let mut now = (self.clock)();
        now.0 &= LMASK;
        let msg_time = timestamp.get_time();
        if *msg_time > now && *msg_time - now > self.delta {
            let err_msg = format!(
                "incoming timestamp from {} exceeding delta {}ms is rejected: {} vs. now: {}",
                timestamp.get_id(),
                self.delta.to_duration().as_millis(),
                msg_time,
                now
            );
            warn!("{}", err_msg);
            Err(err_msg)
        } else {
            let mut last_time = asynclock!(self.last_time);
            let max_time = cmp::max(cmp::max(now, *msg_time), *last_time);
            if max_time == now {
                *last_time = now;
            } else if max_time == *msg_time {
                *last_time = *msg_time + 1;
            } else {
                *last_time += 1;
            }
            Ok(())
        }
    }
}

impl Default for HLC {
    /// Create a new [`HLC`] with a generated UUID and using
    /// [`system_time_clock()`] as physical clock.
    fn default() -> Self {
        HLC::with_system_time(uuid::Uuid::new_v4().into())
    }
}

/// A physical clock relying on std::time::SystemTime::now().
///
/// It returns a NTP64 relative to std::time::UNIX_EPOCH (1st Jan 1970).
///
/// # Examples
///
/// ```
/// # async_std::task::block_on(async {
/// let hlc = uhlc::HLC::with_clock(uuid::Uuid::new_v4().into(), uhlc::system_time_clock);
/// println!("{}", hlc.new_timestamp().await);
/// # })
/// ```
#[inline]
pub fn system_time_clock() -> NTP64 {
    NTP64::from(SystemTime::now().duration_since(UNIX_EPOCH).unwrap())
}

#[cfg(test)]
mod tests {
    use crate::*;
    use async_std::sync::Arc;
    use async_std::task;
    use futures::join;
    use std::convert::TryFrom;
    use std::time::Duration;

    fn is_sorted(vec: &[Timestamp]) -> bool {
        let mut it = vec.iter();
        let mut ts = it.next().unwrap();
        for next in it {
            if next <= ts {
                return false;
            };
            ts = next;
        }
        true
    }

    #[test]
    fn hlc_parallel() {
        task::block_on(async {
            let id0: ID = ID::try_from(vec![0x00].as_ref()).unwrap();
            let id1: ID = ID::try_from(vec![0x01].as_ref()).unwrap();
            let id2: ID = ID::try_from(vec![0x02].as_ref()).unwrap();
            let id3: ID = ID::try_from(vec![0x03].as_ref()).unwrap();
            let hlc0 = Arc::new(HLC::with_system_time(id0.clone()));
            let hlc1 = Arc::new(HLC::with_system_time(id1.clone()));
            let hlc2 = Arc::new(HLC::with_system_time(id2.clone()));
            let hlc3 = Arc::new(HLC::with_system_time(id3.clone()));

            // Make 4 tasks to generate 10000 timestamps each with distinct HLCs,
            // and also to update each other HLCs
            const NB_TIME: usize = 10000;
            let t0 = {
                let hlc0 = hlc0.clone();
                let hlc1 = hlc1.clone();
                task::spawn(async move {
                    let mut times: Vec<Timestamp> = Vec::with_capacity(10000);
                    for _ in 0..NB_TIME {
                        let ts = hlc0.new_timestamp().await;
                        assert!(hlc1.update_with_timestamp(&ts).await.is_ok());
                        times.push(ts)
                    }
                    times
                })
            };
            let t1 = {
                let hlc1 = hlc1.clone();
                let hlc2 = hlc2.clone();
                task::spawn(async move {
                    let mut times: Vec<Timestamp> = Vec::with_capacity(10000);
                    for _ in 0..NB_TIME {
                        let ts = hlc1.new_timestamp().await;
                        assert!(hlc2.update_with_timestamp(&ts).await.is_ok());
                        times.push(ts)
                    }
                    times
                })
            };
            let t2 = {
                let hlc2 = hlc3.clone();
                let hlc3 = hlc3.clone();
                task::spawn(async move {
                    let mut times: Vec<Timestamp> = Vec::with_capacity(10000);
                    for _ in 0..NB_TIME {
                        let ts = hlc2.new_timestamp().await;
                        assert!(hlc3.update_with_timestamp(&ts).await.is_ok());
                        times.push(ts)
                    }
                    times
                })
            };
            let t3 = {
                let hlc3 = hlc3.clone();
                let hlc0 = hlc0.clone();
                task::spawn(async move {
                    let mut times: Vec<Timestamp> = Vec::with_capacity(10000);
                    for _ in 0..NB_TIME {
                        let ts = hlc3.new_timestamp().await;
                        assert!(hlc0.update_with_timestamp(&ts).await.is_ok());
                        times.push(ts)
                    }
                    times
                })
            };
            let vecs = join!(t0, t1, t2, t3);

            // test that each timeseries is sorted (i.e. monotonic time)
            assert!(is_sorted(&vecs.0));
            assert!(is_sorted(&vecs.1));
            assert!(is_sorted(&vecs.2));
            assert!(is_sorted(&vecs.3));

            // test that there is no duplicate amongst all timestamps
            let mut all_times: Vec<Timestamp> = vecs
                .0
                .into_iter()
                .chain(vecs.1.into_iter())
                .chain(vecs.2.into_iter())
                .chain(vecs.3.into_iter())
                .collect::<Vec<Timestamp>>();
            assert_eq!(NB_TIME * 4, all_times.len());
            all_times.sort();
            all_times.dedup();
            assert_eq!(NB_TIME * 4, all_times.len());
        });
    }

    #[test]
    fn hlc_update_with_timestamp() {
        task::block_on(async {
            let id: ID = ID::from(uuid::Uuid::new_v4());
            let hlc = HLC::with_system_time(id.clone());

            // Test that updating with an old Timestamp don't break the HLC
            let past_ts = Timestamp::new(Default::default(), id.clone());
            let now_ts = hlc.new_timestamp().await;
            assert!(hlc.update_with_timestamp(&past_ts).await.is_ok());
            assert!(hlc.new_timestamp().await > now_ts);

            // Test that updating with a Timestamp exceeding the delta is refused
            let now_ts = hlc.new_timestamp().await;
            let future_time = now_ts.get_time() + NTP64::from(Duration::from_millis(500));
            let future_ts = Timestamp::new(future_time, id.clone());
            assert!(hlc.update_with_timestamp(&future_ts).await.is_err())
        });
    }
}