reifydb_runtime/context/clock/

host.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

#![allow(clippy::disallowed_types)]

use std::{
	cmp, fmt, ops,
	sync::{
		Arc,
		atomic::{AtomicU64, Ordering},
	},
	time,
	time::{Duration, SystemTime, UNIX_EPOCH},
};

use reifydb_value::value::{datetime::DateTime, duration::Duration as RvDuration};

use crate::{
	context::clock::{TimerId, TimerWake},
	sync::mutex::Mutex,
};

#[allow(clippy::disallowed_methods)]
#[inline(always)]
fn platform_now_nanos() -> u64 {
	SystemTime::now().duration_since(UNIX_EPOCH).expect("System time is before Unix epoch").as_nanos() as u64
}

#[derive(Clone)]
pub enum Clock {
	Real,

	Mock(MockClock),
}

impl Clock {
	pub fn now(&self) -> DateTime {
		match self {
			Clock::Real => DateTime::from_nanos(
				i64::try_from(platform_now_nanos()).expect("system clock past 2262-04-11"),
			),
			Clock::Mock(mock) => mock.now(),
		}
	}

	#[allow(clippy::disallowed_methods)]
	pub fn instant(&self) -> Instant {
		match self {
			Clock::Real => Instant {
				inner: InstantInner::Real(time::Instant::now()),
			},
			Clock::Mock(mock) => Instant {
				inner: InstantInner::Mock {
					captured_nanos: u64::try_from(mock.now().to_nanos())
						.expect("mock clock is never before the Unix epoch"),
					clock: mock.clone(),
				},
			},
		}
	}

	pub fn testing() -> Self {
		#[cfg(reifydb_dst)]
		return Clock::Mock(MockClock::from_millis(0));
		#[cfg(not(reifydb_dst))]
		return Clock::Real;
	}

	pub fn is_mock(&self) -> bool {
		matches!(self, Clock::Mock(_))
	}

	pub fn as_mock(&self) -> Option<&MockClock> {
		match self {
			Clock::Mock(mock) => Some(mock),
			Clock::Real => None,
		}
	}
}

#[derive(Clone)]
pub struct MockClock {
	inner: Arc<MockClockInner>,
}

struct MockClockInner {
	time_nanos: AtomicU64,
	timers: Mutex<Vec<Timer>>,
	next_timer_id: AtomicU64,
}

struct Timer {
	id: u64,
	deadline_nanos: u64,
	wake: Arc<dyn TimerWake>,
}

impl MockClock {
	pub fn new(initial_nanos: u64) -> Self {
		Self {
			inner: Arc::new(MockClockInner {
				time_nanos: AtomicU64::new(initial_nanos),
				timers: Mutex::new(Vec::new()),
				next_timer_id: AtomicU64::new(0),
			}),
		}
	}

	pub fn from_millis(millis: u64) -> Self {
		Self::new(millis * 1_000_000)
	}

	pub fn now(&self) -> DateTime {
		DateTime::from_nanos(
			i64::try_from(self.inner.time_nanos.load(Ordering::Acquire))
				.expect("mock clock past 2262-04-11"),
		)
	}

	pub fn set_nanos(&self, nanos: u64) {
		self.inner.time_nanos.store(nanos, Ordering::Release);
		self.fire_due(nanos);
	}

	pub fn register_timer(&self, deadline_nanos: u64, wake: Arc<dyn TimerWake>) -> TimerId {
		let id = self.inner.next_timer_id.fetch_add(1, Ordering::Relaxed);
		self.inner.timers.lock().push(Timer {
			id,
			deadline_nanos,
			wake,
		});
		TimerId(id)
	}

	pub fn cancel_timer(&self, timer: TimerId) {
		self.inner.timers.lock().retain(|entry| entry.id != timer.0);
	}

	fn fire_due(&self, now_nanos: u64) {
		let due = {
			let mut timers = self.inner.timers.lock();
			let mut due: Vec<Arc<dyn TimerWake>> = Vec::new();
			timers.retain(|entry| {
				if entry.deadline_nanos <= now_nanos {
					due.push(entry.wake.clone());
					return false;
				}
				true
			});
			due
		};

		for wake in due {
			wake.wake();
		}
	}

	pub fn set_micros(&self, micros: u64) {
		self.set_nanos(micros * 1_000);
	}

	pub fn set_millis(&self, millis: u64) {
		self.set_nanos(millis * 1_000_000);
	}

	pub fn advance_nanos(&self, nanos: u64) {
		let current = u64::try_from(self.now().to_nanos()).expect("mock clock is never before the Unix epoch");
		self.set_nanos(current.saturating_add(nanos));
	}

	pub fn advance_micros(&self, micros: u64) {
		self.advance_nanos(micros * 1_000);
	}

	pub fn advance_millis(&self, millis: u64) {
		self.advance_nanos(millis * 1_000_000);
	}

	pub fn advance_secs(&self, secs: u64) {
		self.advance_nanos(secs * 1_000_000_000);
	}

	pub fn advance_minutes(&self, minutes: u64) {
		self.advance_secs(minutes * 60);
	}

	pub fn advance_hours(&self, hours: u64) {
		self.advance_secs(hours * 3600);
	}

	pub fn advance_days(&self, days: u64) {
		self.advance_secs(days * 86400);
	}
}

#[derive(Clone)]
enum InstantInner {
	Real(time::Instant),
	Mock {
		captured_nanos: u64,
		clock: MockClock,
	},
}

#[derive(Clone)]
pub struct Instant {
	inner: InstantInner,
}

impl Instant {
	#[inline]
	pub fn elapsed(&self) -> Duration {
		match &self.inner {
			InstantInner::Real(instant) => instant.elapsed(),
			InstantInner::Mock {
				captured_nanos,
				clock,
			} => {
				let now = u64::try_from(clock.now().to_nanos())
					.expect("mock clock is never before the Unix epoch");
				let elapsed_nanos = now.saturating_sub(*captured_nanos);
				Duration::from_nanos(elapsed_nanos)
			}
		}
	}

	#[inline]
	pub fn duration_since(&self, earlier: &Instant) -> Duration {
		match (&self.inner, &earlier.inner) {
			(InstantInner::Real(this), InstantInner::Real(other)) => this.duration_since(*other),
			(
				InstantInner::Mock {
					captured_nanos: this_nanos,
					..
				},
				InstantInner::Mock {
					captured_nanos: other_nanos,
					..
				},
			) => {
				let elapsed = this_nanos.saturating_sub(*other_nanos);
				Duration::from_nanos(elapsed)
			}
			_ => panic!("Cannot compare instants from different clock types"),
		}
	}
}

impl PartialEq for Instant {
	fn eq(&self, other: &Self) -> bool {
		match (&self.inner, &other.inner) {
			(InstantInner::Real(a), InstantInner::Real(b)) => a == b,
			(
				InstantInner::Mock {
					captured_nanos: a,
					..
				},
				InstantInner::Mock {
					captured_nanos: b,
					..
				},
			) => a == b,
			_ => panic!("Cannot compare instants from different clock types"),
		}
	}
}

impl Eq for Instant {}

impl PartialOrd for Instant {
	fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
		Some(self.cmp(other))
	}
}

impl Ord for Instant {
	fn cmp(&self, other: &Self) -> cmp::Ordering {
		match (&self.inner, &other.inner) {
			(InstantInner::Real(a), InstantInner::Real(b)) => a.cmp(b),
			(
				InstantInner::Mock {
					captured_nanos: a,
					..
				},
				InstantInner::Mock {
					captured_nanos: b,
					..
				},
			) => a.cmp(b),
			_ => panic!("Cannot compare instants from different clock types"),
		}
	}
}

impl ops::Add<Duration> for Instant {
	type Output = Instant;

	fn add(self, duration: Duration) -> Instant {
		match self.inner {
			InstantInner::Real(instant) => Instant {
				inner: InstantInner::Real(instant + duration),
			},
			InstantInner::Mock {
				captured_nanos,
				clock,
			} => Instant {
				inner: InstantInner::Mock {
					captured_nanos: captured_nanos.saturating_add(duration.as_nanos() as u64),
					clock,
				},
			},
		}
	}
}

impl ops::Add<RvDuration> for Instant {
	type Output = Instant;

	fn add(self, duration: RvDuration) -> Instant {
		self + duration.to_std()
	}
}

impl ops::Sub for &Instant {
	type Output = Duration;

	fn sub(self, other: &Instant) -> Duration {
		self.duration_since(other)
	}
}

impl fmt::Debug for Instant {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		match &self.inner {
			InstantInner::Real(instant) => f.debug_tuple("Instant::Real").field(instant).finish(),
			InstantInner::Mock {
				captured_nanos,
				..
			} => f.debug_tuple("Instant::Mock").field(captured_nanos).finish(),
		}
	}
}