use crate::error::{DbError, Result};
use crate::util::timestamp;
use std::sync::Mutex;
use std::time::{Duration, SystemTime};
pub const DEFAULT_FUTURE_STAMP_TOLERANCE: Duration = Duration::from_secs(24 * 60 * 60);
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum FutureStampPolicy {
#[default]
Default,
Tolerance(Duration),
Allow,
}
impl FutureStampPolicy {
fn tolerance(self) -> Option<Duration> {
match self {
Self::Default => Some(DEFAULT_FUTURE_STAMP_TOLERANCE),
Self::Tolerance(d) => Some(d),
Self::Allow => None,
}
}
}
pub trait Clock: Send + Sync {
fn now(&self) -> String;
fn raise_floor(&self, floor: SystemTime);
}
pub(crate) async fn recorded_at_floor(
conn: &libsql::Connection,
policy: FutureStampPolicy,
) -> Result<Option<SystemTime>> {
let max_ts: Option<String> = conn
.query(
"SELECT MAX(recorded_at) FROM (
SELECT MAX(recorded_at) AS recorded_at FROM concepts
UNION ALL
SELECT MAX(recorded_at) AS recorded_at FROM links
)",
(),
)
.await?
.next()
.await?
.and_then(|row| row.get(0).ok());
Ok(match max_ts {
Some(ts) => match parse_iso8601_utc(&ts) {
Ok(t) => {
if let Some(tolerance) = policy.tolerance() {
let limit = SystemTime::now() + tolerance;
if t > limit {
return Err(DbError::FutureRecordedAt {
stamp: ts,
limit: format_iso8601_utc(limit),
});
}
}
Some(t)
}
Err(e) => {
tracing::warn!(
"clock: failed to parse MAX(recorded_at)={:?}: {}; no floor applied",
ts,
e
);
None
}
},
None => None,
})
}
pub struct SystemClock {
last_issued: Mutex<SystemTime>,
}
impl SystemClock {
pub async fn new(conn: &libsql::Connection, policy: FutureStampPolicy) -> Result<Self> {
let floor = recorded_at_floor(conn, policy)
.await?
.unwrap_or_else(SystemTime::now);
Ok(Self {
last_issued: Mutex::new(floor),
})
}
}
impl Clock for SystemClock {
fn now(&self) -> String {
let mut guard = self.last_issued.lock().unwrap();
let wall = SystemTime::now();
let next = if wall > *guard {
wall
} else {
*guard + std::time::Duration::from_micros(1)
};
*guard = next;
format_iso8601_utc(next)
}
fn raise_floor(&self, floor: SystemTime) {
let mut guard = self.last_issued.lock().unwrap();
if floor > *guard {
*guard = floor;
}
}
}
pub struct FakeClock {
current: Mutex<SystemTime>,
}
impl FakeClock {
pub fn new(initial: SystemTime) -> Self {
Self {
current: Mutex::new(initial),
}
}
pub fn advance(&self, duration: std::time::Duration) {
let mut guard = self.current.lock().unwrap();
*guard += duration;
}
pub fn peek(&self) -> String {
format_iso8601_utc(*self.current.lock().unwrap())
}
}
impl Clock for FakeClock {
fn now(&self) -> String {
let mut guard = self.current.lock().unwrap();
let res = format_iso8601_utc(*guard);
*guard += std::time::Duration::from_micros(1);
res
}
fn raise_floor(&self, floor: SystemTime) {
let mut guard = self.current.lock().unwrap();
if floor > *guard {
*guard = floor + std::time::Duration::from_micros(1);
}
}
}
pub fn parse_iso8601_utc(s: &str) -> Result<SystemTime> {
timestamp::parse(s)
}
pub fn format_iso8601_utc(st: SystemTime) -> String {
timestamp::format(st)
}