cnsprcy 0.3.2

use std::time::{
	Duration,
	SystemTime,
	UNIX_EPOCH
};
use std::sync::atomic::Ordering;

#[cfg(target_has_atomic = "64")]
use std::sync::atomic::AtomicU64;

#[cfg(not(target_has_atomic = "64"))]
use portable_atomic::AtomicU64;

use liter::Value;
use serde::{Serialize, Deserialize};
use fixed::types::U32F32;
use fixed::traits::FromFixed;

use super::NodeID;

// /// Seconds between NTP Epoch (1900-01-01) and Unix epoch (1970-01-01)
// const SECONDS_BETWEEN_EPOCHS: f64 = 2208988800f64;
/// Bitmask to get only the 16th least significant bit (1000000000000000)
const ONLY_LS_BIT_16: u64 = 1 << 15;
/// Bitmask to get only the 48 most significant bits
const MS_48_BITS_SET: u64 = 0xFF_FF_FF_FF_FF_FF_00_00;

//const NTP_EPOCH: SystemTime = UNIX_EPOCH + Duration::from_secs(2208988800u);

#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq, Value)]
pub enum Status {
	Active,
	Suspicious,
	Unreachable,
	Indirect,
	Quit
}

#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PeerState {
	pub id: NodeID,
	pub clk: Clock,
	pub state: Status
}

impl PeerState {

	pub fn has_id(&self, id: NodeID) -> bool {
		self.id == id
	}
	pub fn is_active(&self) -> bool {
		matches!(self.state, Status::Active)
	}
}

impl std::fmt::Debug for PeerState {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		write!(f, "{}@{}: {:?}", self.id, self.clk, self.state)
	}
}

pub fn elapsed(time: Option<SystemTime>) -> Option<Duration> {
	time?
		.elapsed()
		.map_err(|e| error!("time is {:?} in the future!", e.duration()))
		.ok()
}

/// `until - elapsed(from)`
///
/// Returns None if:
/// - `from` is None,
/// - `from` is Some, but in the future,
/// - `from` is Some(value) that was longer than (or exactly) `until` ago
///
/// Otherwise: the amount of time remaining until `from` was `until` ago
pub fn remaining(from: Option<SystemTime>, until: Duration)
	-> Option<Duration>
{
	let elapsed_time = from?
		.elapsed()
		.map_err(|e| error!(time_error = %e, "elapsed() from the past"))
		.ok()?;
	until
		.checked_sub(elapsed_time)
		.filter(|time| !time.is_zero())
}

/**
A *hybrid logical clock* (HLC) timestamp, consisting of a modified NTP timestamp 
rounded to 48 bits and a 16 bit counter value.
Instead of the NTP epoch (1900-01-01), this uses the Unix epoch (1970-01-01).

The NTP timestamp is an unsigned 64b fixed point number
- upper 32b are seconds since 1900 (here: 1970)
- lower 32b are fractional seconds

The HLC Timestamp is the same as an NTP timestamp, except for the 16 least
significant bits. These represent the "counter" value for correctly ordering
events where the system clock has not increased since the last timestamp
(this includes "simultaneous" events and events occuring after the system clock
is set backwards).

To illustrate:

```text
0000000000000000000000000000000000000000000000000000000000000000 | bits
[------------seconds-----------][-----fractional seconds-------] | NTP
[--------------rounded timestamp---------------][---counter----] | HLC
```

Taken from ["Logical Physical Clocks and Consistent Snapshots in Globally
Distributed Databases"](https://cse.buffalo.edu/tech-reports/2014-04.pdf) by
Kulkarni, Demirbas, Madeppa, Avva, and Leone.
**However**, unlike the implementation described there, this one is **not**
backwards compatible with NTP because it uses the Unix epoch instead of the NTP
epoch.
*/
#[derive(
	Clone,
	Copy,
	Debug,
	PartialEq,
	Eq,
	PartialOrd,
	Ord,
	Serialize,
	Deserialize,
	Value
)]
pub struct Clock(u64);

impl Clock {
	/// Generate the first ever HLC
	pub fn null_clock() -> Self { Self(0) }

	/// Convert the time component to a SystemTime
	pub fn get_time(self) -> SystemTime {
		let seconds = f64::unwrapped_from_fixed(
			U32F32::from_bits(self.0 & MS_48_BITS_SET)
		);
		UNIX_EPOCH
			// - Duration::from_secs_f64(SECONDS_BETWEEN_EPOCHS)
			+ Duration::from_secs_f64(seconds)
	}
}

impl std::fmt::Display for Clock {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		let unix_time = f64::unwrapped_from_fixed(
			U32F32::from_bits(self.0 & MS_48_BITS_SET)
		);
		// let unix_time = seconds - SECONDS_BETWEEN_EPOCHS;
		let counter: u16 = (self.0 & !MS_48_BITS_SET)
			.try_into()
			.unwrap();
		write!(f, "{unix_time}/{counter}")
	}
}

/**
HLC implementation using [`AtomicU64`] and [`SystemTime`]

See the documentation for [`Clock`] for more details.
```
# use cnsprcy::peer::status::AtomicClock;
# use cnsprcy::peer::Clock;
let clk = AtomicClock::null_clock();

assert_ne!(clk.next(), Clock::null_clock());
assert!(clk.next() < clk.next());
```
*/
#[derive(Debug)]
pub struct AtomicClock(AtomicU64);

impl AtomicClock {

	/// Generate the first ever HLC
	pub const fn null_clock() -> Self { Self(AtomicU64::new(0)) }

	/// Load a stored [`Clock`]
	pub const fn load(prev: Clock) -> Self { Self(AtomicU64::new(prev.0)) }

	fn next_with(&self, mut now: u64) -> Clock {
		#[cfg(debug_assertions)]
		let mut retries = 0;

		let mut prev = self.0.fetch_max(now, Ordering::AcqRel);
		while prev >= now {
			let (time, counter) = Self::split(prev);
			let counter = counter
				.checked_add(1)
				.expect("too much clock drift / too many HLCs created");

			// Fuse the timestamp and the counter
			now = time | (counter as u64);
			prev = self.0.fetch_max(now, Ordering::AcqRel);
			#[cfg(debug_assertions)] {
				retries += 1;
				assert!(retries <= 100);
			}
		}
		Clock(now)
	}

	/// Get the next timestamp: `now > prev ? (now, 0) else (prev, prev.c + 1)`
	pub fn next(&self) -> Clock {
		self.next_with(Self::get_timestamp_now())
	}

	/// Get the currently stored value, which is the latest [`Clock`] generated
	pub fn prev(&self) -> Clock {
		Clock(self.0.load(Ordering::Relaxed))
	}

	/// Get current 64bit timestamp with the lowest 16 bits set to 0
	fn get_timestamp_now() -> u64 {
		// Complete Timestamp as f64
		let time = SystemTime::now()
			.duration_since(UNIX_EPOCH)
			.expect("Unable to get local time!")
			.as_secs_f64();
			// + SECONDS_BETWEEN_EPOCHS;
		// transmute floating point to fixed point (32, 32)
		let mut t = U32F32::from_num(time);
		// determine whether to round up
		let only_bit_16 = t.to_bits() & ONLY_LS_BIT_16;
		// round up
		t += U32F32::from_bits(only_bit_16);
		// return as u64 with the counter bits set to 0
		t.to_bits() & MS_48_BITS_SET
	}

	/// Split the timestamp and counter
	fn split(clock: u64) -> (u64, u16) {
		// zero out the leading 48 bits, then convert u64 to u16
		let counter: u16 = (clock & !MS_48_BITS_SET)
			.try_into()
			.unwrap();
		let time = clock & MS_48_BITS_SET;
		(time, counter)
	}
}

#[test]
fn atomic_clock() {
	let clk = AtomicClock::null_clock();

	// DODGY TEST:
	// 0 and 1 as .next_with is dodgy because they are in the counter part of
	// the HLC, they both have a time component of 0.

	let tests = [
		//(Clock::null_clock(), "clock is zero now"),
		(clk.next_with(0), "clock is the same"),
		(clk.next_with(1), "clock increased"),
		(clk.next_with(0), "clock went back"),
		(clk.next_with(0), "clock is the same"),
		(clk.next(), "clock increased a lot"),
		(clk.next_with(0), "clock went back a lot"),
		(clk.next_with(0), "clock is the same"),
		(clk.next(), "clock increased a lot again"),
		(clk.next_with(0), "clock went back a lot again")
	];

	let mut prev = Clock::null_clock();
	for (i, (val, msg)) in tests.into_iter().enumerate() {
		assert!(prev < val, "{prev} IS NOT < {val} @ {msg} (test {i})");
		prev = val;
	}
}