cacheflight 0.1.0

use crate::{
    CacheBackend, CacheMissReason, Error, MetricsHooks, NoopMetrics, RecomputeOutcome,
    RecomputeReason, Result,
};
use std::{
    collections::HashMap,
    fmt,
    future::Future,
    marker::PhantomData,
    pin::Pin,
    sync::Arc,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tokio::sync::{Mutex, watch};

// ── Type-state markers ──────────────────────────────────────────────────────

/// Zero-sized marker: no expiry strategy has been configured yet.
pub struct NoExpiry;

/// Zero-sized marker: a flat TTL has been configured via `.ttl()`.
pub struct HasFlatExpiry;

/// Zero-sized marker: stale-while-revalidate has been configured.
pub struct HasSwrExpiry;

/// Zero-sized marker: probabilistic early expiration is not enabled.
pub struct NoXfetch;

/// Zero-sized marker: probabilistic early expiration is enabled.
pub struct HasXfetch;

// ── Entry wire format ───────────────────────────────────────────────────────

const ENTRY_MAGIC: &[u8; 4] = b"SFG1";
const ENTRY_HEADER_LEN: usize = 28;

// ── Flight coordination ─────────────────────────────────────────────────────

type SharedFlightResult = Result<Arc<Vec<u8>>>;

struct Flight {
    notifier: watch::Sender<Option<SharedFlightResult>>,
}

impl Flight {
    fn new() -> Self {
        let (notifier, _) = watch::channel(None);
        Self { notifier }
    }
}

enum CachedEntryState<'a> {
    Fresh(&'a [u8], f64),
    Stale(&'a [u8], f64),
    Expired,
    Invalid,
}

// ── Expiry configuration (internal) ─────────────────────────────────────────

#[derive(Debug, Clone, Copy)]
enum ExpiryStrategy {
    Flat {
        ttl: Duration,
    },
    Swr {
        fresh_ttl: Duration,
        stale_ttl: Duration,
    },
}

impl ExpiryStrategy {
    fn fresh_ttl(&self) -> Duration {
        match self {
            Self::Flat { ttl } => *ttl,
            Self::Swr { fresh_ttl, .. } => *fresh_ttl,
        }
    }

    fn stale_ttl(&self) -> Duration {
        match self {
            Self::Flat { .. } => Duration::ZERO,
            Self::Swr { stale_ttl, .. } => *stale_ttl,
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct EntryConfig {
    expiry: ExpiryStrategy,
    beta: Option<f64>,
}

// ── Public return types ─────────────────────────────────────────────────────

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum LookupState {
    CacheHit,
    Stale,
    Recomputed,
    Shared,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LookupResult {
    value: Vec<u8>,
    state: LookupState,
}

impl LookupResult {
    fn new(value: Vec<u8>, state: LookupState) -> Self {
        Self { value, state }
    }

    /// Returns a reference to the cached value.
    pub fn value(&self) -> &[u8] {
        &self.value
    }

    /// Consumes the result and returns the owned value.
    pub fn into_value(self) -> Vec<u8> {
        self.value
    }

    /// Returns how this result was served (fresh, stale, recomputed, or shared).
    pub fn state(&self) -> LookupState {
        self.state
    }
}

/// Wraps a computed value so the engine can distinguish successful results
/// from errors during the cache-update path.
///
/// Users return `ComputeValue::new(value)` from their work closure inside
/// a `Result::Ok`. The engine always caches the value on success.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ComputeValue {
    value: Vec<u8>,
}

impl ComputeValue {
    /// Wraps `value` for caching. Always cached.
    pub fn new(value: Vec<u8>) -> Self {
        Self { value }
    }

    fn into_value(self) -> Vec<u8> {
        self.value
    }
}

impl fmt::Display for LookupState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::CacheHit => write!(f, "cache_hit"),
            Self::Stale => write!(f, "stale"),
            Self::Recomputed => write!(f, "recomputed"),
            Self::Shared => write!(f, "shared"),
        }
    }
}

// ── RunBuilder ──────────────────────────────────────────────────────────────

/// Per-call builder returned by [`CacheFlight::run`].
///
/// Chain override methods before `.await` — or just await directly to use
/// the instance-level defaults.
pub struct RunBuilder<'a, B, E, X, F> {
    cf: &'a CacheFlight<B, E, X>,
    key: String,
    work: F,
    fresh_override: Option<Duration>,
    stale_override: Option<Duration>,
    beta_override: Option<f64>,
}

impl<'a, B, X, F> RunBuilder<'a, B, HasFlatExpiry, X, F> {
    /// Overrides the TTL for this single call.
    pub fn ttl(mut self, duration: Duration) -> Self {
        self.fresh_override = Some(duration);
        self
    }
}

impl<'a, B, X, F> RunBuilder<'a, B, HasSwrExpiry, X, F> {
    /// Overrides the fresh window for this single call.
    pub fn fresh_for(mut self, duration: Duration) -> Self {
        self.fresh_override = Some(duration);
        self
    }

    /// Overrides the stale window for this single call.
    pub fn stale_for(mut self, duration: Duration) -> Self {
        self.stale_override = Some(duration);
        self
    }

    /// Overrides both fresh and stale windows for this single call.
    pub fn stale_while_revalidate(mut self, fresh: Duration, stale: Duration) -> Self {
        self.fresh_override = Some(fresh);
        self.stale_override = Some(stale);
        self
    }
}

impl<'a, B, E, F> RunBuilder<'a, B, E, HasXfetch, F> {
    /// Overrides the XFetch beta parameter for this single call.
    pub fn beta(mut self, beta: f64) -> Self {
        self.beta_override = Some(beta);
        self
    }
}

impl<'a, B, E, X, F, Fut> IntoFuture for RunBuilder<'a, B, E, X, F>
where
    B: CacheBackend + 'static,
    E: Send + Sync + 'static,
    X: Send + Sync + 'static,
    F: Fn() -> Fut + Clone + Send + 'static,
    Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    type Output = Result<LookupResult>;
    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move {
            let RunBuilder {
                cf,
                key,
                work,
                fresh_override,
                stale_override,
                beta_override,
            } = self;

            let effective_fresh = fresh_override.unwrap_or_else(|| cf.config.expiry.fresh_ttl());
            let effective_stale = stale_override.unwrap_or_else(|| cf.config.expiry.stale_ttl());
            let effective_beta = beta_override.or(cf.config.beta);
            let total = effective_fresh
                .checked_add(effective_stale)
                .unwrap_or(Duration::MAX);

            let wrapped = {
                let work = work.clone();
                move || {
                    let work = work.clone();
                    async move { work().await.map(ComputeValue::new) }
                }
            };

            match cf.cache.get(&key).await {
                Ok(Some(cached)) => match classify_cached_entry(&cached, now_millis()) {
                    CachedEntryState::Fresh(payload, delta_ema) => {
                        if let Some(beta) = effective_beta {
                            let fresh_until_millis =
                                u64::from_be_bytes(cached[4..12].try_into().unwrap()) as f64;
                            let now = now_millis() as f64;
                            let r = fastrand::f64().max(f64::EPSILON);
                            if now - delta_ema * beta * r.ln() >= fresh_until_millis {
                                cf.metrics.on_xfetch_early_refresh(&key);
                                cf.spawn_stale_refresh(
                                    key.clone(),
                                    wrapped,
                                    total,
                                    effective_fresh,
                                    delta_ema,
                                    RecomputeReason::ProbabilisticEarlyExpiration,
                                )
                                .await;
                                return Ok(LookupResult::new(
                                    payload.to_vec(),
                                    LookupState::CacheHit,
                                ));
                            }
                        }

                        cf.metrics.on_cache_hit(&key);
                        return Ok(LookupResult::new(payload.to_vec(), LookupState::CacheHit));
                    }
                    CachedEntryState::Stale(payload, delta_ema)
                        if effective_stale > Duration::ZERO =>
                    {
                        cf.metrics.on_cache_stale_hit(&key);
                        cf.spawn_stale_refresh(
                            key.clone(),
                            wrapped,
                            total,
                            effective_fresh,
                            delta_ema,
                            RecomputeReason::StaleWhileRevalidate,
                        )
                        .await;
                        return Ok(LookupResult::new(payload.to_vec(), LookupState::Stale));
                    }
                    CachedEntryState::Stale(_, _) | CachedEntryState::Expired => {
                        cf.metrics.on_cache_miss(&key, CacheMissReason::Expired);
                    }
                    CachedEntryState::Invalid => {
                        cf.metrics.on_cache_miss(&key, CacheMissReason::Invalid);
                    }
                },
                Ok(None) => {
                    cf.metrics.on_cache_miss(&key, CacheMissReason::Missing);
                }
                Err(error) => {
                    cf.metrics.on_cache_read_failed(&key, &error);
                    cf.metrics
                        .on_cache_miss(&key, CacheMissReason::BackendError);
                }
            }

            let (flight, joined_existing) = cf.start_or_join_flight(&key).await;

            if joined_existing {
                cf.metrics.on_deduplicated(&key, RecomputeReason::ColdMiss);
                return cf.wait_for_flight(flight, LookupState::Shared).await;
            }

            cf.run_recompute(
                RecomputeParams {
                    key,
                    flight,
                    total_ttl: total,
                    fresh_ttl: effective_fresh,
                    previous_delta_ema: None,
                    reason: RecomputeReason::ColdMiss,
                    state: LookupState::Recomputed,
                },
                wrapped,
            )
            .await
        })
    }
}

// ── CacheFlight ─────────────────────────────────────────────────────────────

/// Deduplicates concurrent recomputation for the same key, with optional
/// stale-while-revalidate and probabilistic early expiration.
pub struct CacheFlight<B, E, X> {
    cache: Arc<B>,
    metrics: Arc<dyn MetricsHooks>,
    config: EntryConfig,
    flight_timeout: Duration,
    flights: Arc<Mutex<HashMap<String, Arc<Flight>>>>,
    _phantom: PhantomData<(E, X)>,
}

impl<B, E, X> Clone for CacheFlight<B, E, X> {
    fn clone(&self) -> Self {
        Self {
            cache: Arc::clone(&self.cache),
            metrics: Arc::clone(&self.metrics),
            config: self.config,
            flight_timeout: self.flight_timeout,
            flights: Arc::clone(&self.flights),
            _phantom: PhantomData,
        }
    }
}

// ── Construction (NoExpiry, NoXfetch) ───────────────────────────────────────

impl<B: CacheBackend> CacheFlight<B, NoExpiry, NoXfetch> {
    /// Creates a new `CacheFlight` instance with a no-op metrics hook.
    pub fn new(cache: B) -> Self {
        Self::with_metrics(cache, NoopMetrics)
    }

    /// Creates a new `CacheFlight` instance with the given metrics hook.
    pub fn with_metrics(cache: B, metrics: impl MetricsHooks) -> Self {
        Self {
            cache: Arc::new(cache),
            metrics: Arc::new(metrics),
            config: EntryConfig {
                expiry: ExpiryStrategy::Flat {
                    ttl: Duration::ZERO,
                },
                beta: None,
            },
            flight_timeout: Duration::from_secs(30),
            flights: Arc::new(Mutex::new(HashMap::new())),
            _phantom: PhantomData,
        }
    }

    /// Sets the maximum time a caller waits for an in-flight recompute
    /// before giving up with a timeout error.
    ///
    /// Defaults to 30 seconds.
    pub fn flight_timeout(mut self, timeout: Duration) -> Self {
        self.flight_timeout = timeout;
        self
    }

    /// Configures a flat TTL (time-to-live) expiry strategy.
    ///
    /// Once the TTL expires, cached entries are treated as stale and a
    /// fresh recompute blocks all concurrent callers.
    pub fn ttl(self, duration: Duration) -> CacheFlight<B, HasFlatExpiry, NoXfetch> {
        CacheFlight {
            cache: self.cache,
            metrics: self.metrics,
            config: EntryConfig {
                expiry: ExpiryStrategy::Flat { ttl: duration },
                beta: None,
            },
            flight_timeout: self.flight_timeout,
            flights: self.flights,
            _phantom: PhantomData,
        }
    }

    /// Configures a stale-while-revalidate expiry strategy.
    ///
    /// Entries are considered fresh for `fresh` and remain usable (stale)
    /// for an additional `stale` during which a background refresh is
    /// triggered without blocking callers.
    pub fn stale_while_revalidate(
        self,
        fresh: Duration,
        stale: Duration,
    ) -> CacheFlight<B, HasSwrExpiry, NoXfetch> {
        CacheFlight {
            cache: self.cache,
            metrics: self.metrics,
            config: EntryConfig {
                expiry: ExpiryStrategy::Swr {
                    fresh_ttl: fresh,
                    stale_ttl: stale,
                },
                beta: None,
            },
            flight_timeout: self.flight_timeout,
            flights: self.flights,
            _phantom: PhantomData,
        }
    }
}

// ── Add XFetch on top of any expiry strategy ───────────────────────────────

impl<B: CacheBackend, X> CacheFlight<B, HasFlatExpiry, X> {
    /// Enables probabilistic early expiration (XFetch) with the given `beta`.
    ///
    /// When `beta > 0`, the engine may trigger a background refresh before
    /// the TTL expires. Higher values make early refreshes more likely.
    pub fn probabilistic_expiry(self, beta: f64) -> CacheFlight<B, HasFlatExpiry, HasXfetch> {
        CacheFlight {
            cache: self.cache,
            metrics: self.metrics,
            config: EntryConfig {
                expiry: self.config.expiry,
                beta: Some(beta),
            },
            flight_timeout: self.flight_timeout,
            flights: self.flights,
            _phantom: PhantomData,
        }
    }
}

impl<B: CacheBackend, X> CacheFlight<B, HasSwrExpiry, X> {
    /// Enables probabilistic early expiration (XFetch) with the given `beta`.
    ///
    /// Works identically to the flat-TTL variant — early refreshes may
    /// fire during the fresh window.
    pub fn probabilistic_expiry(self, beta: f64) -> CacheFlight<B, HasSwrExpiry, HasXfetch> {
        CacheFlight {
            cache: self.cache,
            metrics: self.metrics,
            config: EntryConfig {
                expiry: self.config.expiry,
                beta: Some(beta),
            },
            flight_timeout: self.flight_timeout,
            flights: self.flights,
            _phantom: PhantomData,
        }
    }

    /// Returns the configured fresh TTL.
    pub fn fresh_ttl(&self) -> Duration {
        self.config.expiry.fresh_ttl()
    }

    /// Returns the configured stale TTL.
    pub fn stale_ttl(&self) -> Duration {
        self.config.expiry.stale_ttl()
    }
}

// ── Accessors for Flat TTL ─────────────────────────────────────────────────

impl<B: CacheBackend, X> CacheFlight<B, HasFlatExpiry, X> {
    /// Returns the configured flat TTL duration.
    pub fn ttl_duration(&self) -> Duration {
        match self.config.expiry {
            ExpiryStrategy::Flat { ttl } => ttl,
            _ => unreachable!("type-state guarantees HasFlatExpiry"),
        }
    }
}

// ── Accessors for XFetch ───────────────────────────────────────────────────

impl<B: CacheBackend, E> CacheFlight<B, E, HasXfetch> {
    /// Returns the configured XFetch beta parameter, if any.
    pub fn beta(&self) -> Option<f64> {
        self.config.beta
    }
}

// ── run() — gated on HasFlatExpiry ──────────────────────────────────────────

impl<B: CacheBackend, X> CacheFlight<B, HasFlatExpiry, X> {
    /// Runs the work closure for the given key, deduplicating concurrent
    /// callers and returning the cached or freshly computed value.
    ///
    /// Returns a `RunBuilder` that supports per-call overrides (e.g. `ttl()`).
    pub fn run<F, Fut>(
        &self,
        key: impl Into<String>,
        work: F,
    ) -> RunBuilder<'_, B, HasFlatExpiry, X, F>
    where
        F: Fn() -> Fut + Clone + Send + 'static,
        Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
    {
        RunBuilder {
            cf: self,
            key: key.into(),
            work,
            fresh_override: None,
            stale_override: None,
            beta_override: None,
        }
    }
}

// ── run() — gated on HasSwrExpiry ──────────────────────────────────────────

impl<B: CacheBackend, X> CacheFlight<B, HasSwrExpiry, X> {
    /// Runs the work closure for the given key, deduplicating concurrent
    /// callers and returning the cached or freshly computed value.
    ///
    /// Returns a `RunBuilder` that supports per-call overrides
    /// (e.g. `fresh_for()`, `stale_for()`).
    pub fn run<F, Fut>(
        &self,
        key: impl Into<String>,
        work: F,
    ) -> RunBuilder<'_, B, HasSwrExpiry, X, F>
    where
        F: Fn() -> Fut + Clone + Send + 'static,
        Fut: Future<Output = Result<Vec<u8>>> + Send + 'static,
    {
        RunBuilder {
            cf: self,
            key: key.into(),
            work,
            fresh_override: None,
            stale_override: None,
            beta_override: None,
        }
    }
}

// ── Invalidation API (available on all variants) ─────────────────────────────

impl<B: CacheBackend, E: Send + Sync + 'static, X: Send + Sync + 'static> CacheFlight<B, E, X> {
    /// Removes the cached entry for `key` from the backend.
    ///
    /// The next call to [`run`](Self::run) for this key will result in a
    /// cold miss and trigger a fresh recompute.
    pub async fn invalidate(&self, key: &str) -> Result<()> {
        self.cache.delete(key).await
    }

    /// Removes all cached entries whose key starts with `prefix`.
    ///
    /// Returns the number of entries that were removed.
    pub async fn invalidate_prefix(&self, prefix: &str) -> Result<u64> {
        self.cache.delete_by_prefix(prefix).await
    }
}

// ── Internal helpers (available on all variants) ────────────────────────────

struct RecomputeParams {
    key: String,
    flight: Arc<Flight>,
    total_ttl: Duration,
    fresh_ttl: Duration,
    previous_delta_ema: Option<f64>,
    reason: RecomputeReason,
    state: LookupState,
}

impl<B: CacheBackend, E: Send + Sync + 'static, X: Send + Sync + 'static> CacheFlight<B, E, X> {
    async fn start_or_join_flight(&self, key: &str) -> (Arc<Flight>, bool) {
        let mut flights = self.flights.lock().await;

        if let Some(flight) = flights.get(key) {
            return (flight.clone(), true);
        }

        let flight = Arc::new(Flight::new());
        flights.insert(key.to_owned(), flight.clone());
        (flight, false)
    }

    async fn wait_for_flight(
        &self,
        flight: Arc<Flight>,
        state: LookupState,
    ) -> Result<LookupResult> {
        let mut receiver = flight.notifier.subscribe();

        loop {
            if let Some(result) = receiver.borrow().clone() {
                return result.map(|value| LookupResult::new((*value).clone(), state));
            }

            match tokio::time::timeout(self.flight_timeout, receiver.changed()).await {
                Ok(Ok(())) => continue,
                Ok(Err(_)) => {
                    return Err(Error::internal(
                        "the in-flight leader finished without publishing a result",
                    ));
                }
                Err(_) => {
                    return Err(Error::internal(
                        "timed out waiting for the in-flight recompute to complete",
                    ));
                }
            }
        }
    }

    async fn spawn_stale_refresh<F, Fut>(
        &self,
        key: String,
        work: F,
        total_ttl: Duration,
        fresh_ttl: Duration,
        previous_delta_ema: f64,
        reason: RecomputeReason,
    ) where
        F: Fn() -> Fut + Clone + Send + 'static,
        Fut: Future<Output = Result<ComputeValue>> + Send + 'static,
    {
        let (flight, joined_existing) = self.start_or_join_flight(&key).await;

        if joined_existing {
            self.metrics.on_deduplicated(&key, reason);
            return;
        }

        let this = self.clone();
        tokio::spawn(async move {
            let result = this
                .run_recompute(
                    RecomputeParams {
                        key: key.clone(),
                        flight,
                        total_ttl,
                        fresh_ttl,
                        previous_delta_ema: Some(previous_delta_ema),
                        reason,
                        state: LookupState::Recomputed,
                    },
                    work,
                )
                .await;
            if let Err(error) = result {
                this.metrics
                    .on_background_refresh_failed(&key, reason, &error);
            }
        });
    }

    async fn run_recompute<F, Fut>(&self, p: RecomputeParams, work: F) -> Result<LookupResult>
    where
        F: Fn() -> Fut + Clone + Send + 'static,
        Fut: Future<Output = Result<ComputeValue>> + Send + 'static,
    {
        self.metrics.on_recompute_started(&p.key, p.reason);
        let started_at = Instant::now();

        let result = match work().await {
            Ok(computed) => {
                let value = computed.into_value();

                let elapsed = started_at.elapsed();
                let delta_ema = compute_new_delta_ema(p.previous_delta_ema, elapsed);
                let max_delta = duration_to_millis_f64(p.fresh_ttl).max(1.0) * 2.0;
                let delta_ema = delta_ema.min(max_delta);
                let stale_ttl = p
                    .total_ttl
                    .checked_sub(p.fresh_ttl)
                    .unwrap_or(Duration::ZERO);
                let encoded_entry = encode_cached_entry(&value, p.fresh_ttl, stale_ttl, delta_ema);

                if let Err(error) = self.cache.set(&p.key, encoded_entry, p.total_ttl).await {
                    self.metrics.on_cache_write_failed(&p.key, &error);
                }

                Ok(Arc::new(value))
            }
            Err(error) => Err(error),
        };

        self.metrics.on_recompute_finished(
            &p.key,
            p.reason,
            if result.is_ok() {
                RecomputeOutcome::Success
            } else {
                RecomputeOutcome::Error
            },
            started_at.elapsed(),
        );

        let _ = p.flight.notifier.send(Some(result.clone()));
        self.finish_flight(&p.key).await;

        result.map(|value| LookupResult::new((*value).clone(), p.state))
    }

    async fn finish_flight(&self, key: &str) {
        self.flights.lock().await.remove(key);
    }
}

// ── Encoding / decoding helpers ─────────────────────────────────────────────

fn encode_cached_entry(
    value: &[u8],
    fresh_ttl: Duration,
    stale_ttl: Duration,
    delta_ema_millis: f64,
) -> Vec<u8> {
    let now = now_millis();
    let fresh_until = now.saturating_add(duration_to_millis(fresh_ttl));
    let stale_until = now.saturating_add(duration_to_millis(
        fresh_ttl.checked_add(stale_ttl).unwrap_or(Duration::MAX),
    ));

    let mut bytes = Vec::with_capacity(ENTRY_HEADER_LEN + value.len());
    bytes.extend_from_slice(ENTRY_MAGIC);
    bytes.extend_from_slice(&fresh_until.to_be_bytes());
    bytes.extend_from_slice(&stale_until.to_be_bytes());
    bytes.extend_from_slice(&delta_ema_millis.to_be_bytes());
    bytes.extend_from_slice(value);

    bytes
}

fn classify_cached_entry(bytes: &[u8], now: u64) -> CachedEntryState<'_> {
    if bytes.len() < ENTRY_HEADER_LEN || &bytes[..4] != ENTRY_MAGIC {
        return CachedEntryState::Invalid;
    }

    let fresh_until =
        u64::from_be_bytes(bytes[4..12].try_into().expect("header fresh_until slice"));
    let stale_until =
        u64::from_be_bytes(bytes[12..20].try_into().expect("header stale_until slice"));
    let delta_ema = f64::from_be_bytes(bytes[20..28].try_into().expect("header delta_ema slice"));
    let payload = &bytes[ENTRY_HEADER_LEN..];

    if now < fresh_until {
        CachedEntryState::Fresh(payload, delta_ema)
    } else if now < stale_until {
        CachedEntryState::Stale(payload, delta_ema)
    } else {
        CachedEntryState::Expired
    }
}

fn compute_new_delta_ema(previous: Option<f64>, elapsed: Duration) -> f64 {
    const ALPHA: f64 = 0.2;
    let new_delta = duration_to_millis_f64(elapsed);

    match previous {
        Some(prev) => ALPHA * new_delta + (1.0 - ALPHA) * prev,
        None => new_delta,
    }
}

fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_else(|_| Duration::from_secs(0))
        .as_millis()
        .min(u128::from(u64::MAX)) as u64
}

fn duration_to_millis(d: Duration) -> u64 {
    d.as_millis().min(u128::from(u64::MAX)) as u64
}

fn duration_to_millis_f64(d: Duration) -> f64 {
    d.as_secs_f64() * 1000.0
}