ppoppo-clock 2.0.0

use crate::{BoxFuture, Clock, Timer};
use jiff::Timestamp;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex, PoisonError};
use std::task::{Context, Poll, Waker};
use std::time::Duration;

// ── FrozenClock ────────────────────────────────────────────────────────────

/// A clock that always returns the same instant. ~80% of test scenarios.
#[derive(Debug)]
pub struct FrozenClock {
    instant: Timestamp,
}

impl FrozenClock {
    /// Parse an RFC 3339 string into a frozen instant. An unparseable string
    /// freezes at the Unix epoch — a test fixture, not an input boundary.
    pub fn at(rfc3339: &str) -> Self {
        let instant = rfc3339.parse().unwrap_or(Timestamp::UNIX_EPOCH);
        Self { instant }
    }

    /// Freeze at `instant`.
    pub fn at_instant(instant: Timestamp) -> Self {
        Self { instant }
    }
}

impl Clock for FrozenClock {
    fn now(&self) -> Timestamp {
        self.instant
    }
}

// ── MockClock ──────────────────────────────────────────────────────────────

type PendingSleeps = Vec<(Timestamp, Alarm)>;

/// What one pending sleep waits on: rung, and the task to wake when it is.
#[derive(Debug, Default)]
struct Slot {
    rung: bool,
    waker: Option<Waker>,
}

/// The clock's side of one pending sleep. Dropping it rings it — the clock
/// drops it when the deadline is reached, and a dropped clock drops them
/// all, so no sleep outlives the clock that would have woken it.
#[derive(Debug)]
struct Alarm(Arc<Mutex<Slot>>);

impl Drop for Alarm {
    fn drop(&mut self) {
        let waker = {
            let mut slot = self.0.lock().unwrap_or_else(PoisonError::into_inner);
            slot.rung = true;
            slot.waker.take()
        };
        if let Some(waker) = waker {
            waker.wake();
        }
    }
}

/// The sleeper's side: ready once its [`Alarm`] has rung.
#[derive(Debug)]
struct Sleep(Arc<Mutex<Slot>>);

impl Future for Sleep {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut slot = self.0.lock().unwrap_or_else(PoisonError::into_inner);
        if slot.rung {
            Poll::Ready(())
        } else {
            slot.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

/// A settable, advanceable clock for time-travel tests.
#[derive(Debug)]
pub struct MockClock {
    inner: Arc<Mutex<Timestamp>>,
    pending: Arc<Mutex<PendingSleeps>>,
}

impl MockClock {
    /// A clock reading `start` until it is set or advanced.
    pub fn new(start: Timestamp) -> Self {
        Self {
            inner: Arc::new(Mutex::new(start)),
            pending: Arc::new(Mutex::new(Vec::new())),
        }
    }

    /// Jump to `instant` and wake every pending sleep whose deadline it
    /// reaches.
    pub fn set_now(&self, instant: Timestamp) {
        let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
        *guard = instant;
        drop(guard);
        self.fire_pending();
    }

    /// Move the clock forward by `dur` and wake every pending sleep whose
    /// deadline it reaches.
    pub fn advance_by(&self, dur: Duration) {
        let new_now = {
            let mut guard = self.inner.lock().unwrap_or_else(PoisonError::into_inner);
            *guard = guard.checked_add(dur).unwrap_or(*guard);
            *guard
        };
        self.fire_ready(new_now);
    }

    /// Advance time and fire all sleeps whose deadline has passed.
    /// Equivalent to `tokio::time::advance` for mock-aware code.
    pub fn advance_and_fire(&self, dur: Duration) {
        self.advance_by(dur);
    }

    fn current(&self) -> Timestamp {
        *self.inner.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn fire_pending(&self) {
        let now = self.current();
        self.fire_ready(now);
    }

    fn fire_ready(&self, now: Timestamp) {
        // Drain matching entries while holding the lock; drop lock before firing.
        // `partition` keeps registration order in both halves.
        let to_fire: PendingSleeps = {
            let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
            let (fired, remaining): (PendingSleeps, PendingSleeps) = std::mem::take(&mut *pending)
                .into_iter()
                .partition(|(wake_at, _)| *wake_at <= now);
            *pending = remaining;
            fired
        };
        // Dropping an alarm rings it.
        drop(to_fire);
    }

    fn register_sleep(&self, wake_at: Timestamp) -> Sleep {
        let slot = Arc::new(Mutex::new(Slot::default()));
        let alarm = Alarm(Arc::clone(&slot));
        if wake_at <= self.current() {
            // Already past — ring now.
            drop(alarm);
        } else {
            let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
            pending.push((wake_at, alarm));
        }
        Sleep(slot)
    }
}

impl Clock for MockClock {
    fn now(&self) -> Timestamp {
        self.current()
    }
}

// ── AdvanceableTimer ───────────────────────────────────────────────────────

/// A timer whose sleeps resolve only when `MockClock::advance_and_fire` is called.
/// Pair with `MockClock` for deterministic time-travel tests.
#[derive(Debug)]
pub struct AdvanceableTimer {
    clock: Arc<MockClock>,
}

impl AdvanceableTimer {
    /// A timer whose sleeps are deadlines on `clock`.
    pub fn new(clock: Arc<MockClock>) -> Self {
        Self { clock }
    }
}

impl Timer for AdvanceableTimer {
    fn sleep(&self, dur: Duration) -> BoxFuture<'static, ()> {
        let now = self.clock.current();
        let wake_at = now.checked_add(dur).unwrap_or(now);
        Box::pin(self.clock.register_sleep(wake_at))
    }

    fn next_tick(&self) -> BoxFuture<'static, ()> {
        Box::pin(std::future::ready(()))
    }
}

// ── Test ergonomics helpers ────────────────────────────────────────────────

/// Returns an `Arc<dyn Clock>` frozen at the given RFC 3339 string.
pub fn frozen_at(rfc3339: &str) -> Arc<dyn Clock> {
    Arc::new(FrozenClock::at(rfc3339))
}

/// Returns a `(Arc<MockClock>, Arc<AdvanceableTimer>)` pair wired together.
pub fn mock_pair(start: Timestamp) -> (Arc<MockClock>, Arc<AdvanceableTimer>) {
    let clock = Arc::new(MockClock::new(start));
    let timer = Arc::new(AdvanceableTimer::new(Arc::clone(&clock)));
    (clock, timer)
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    use super::*;

    fn t0() -> Timestamp {
        "2026-05-10T00:00:00Z".parse().expect("valid RFC 3339")
    }

    // ── FrozenClock tests ──────────────────────────────────────────────

    #[test]
    fn frozen_at_parses_rfc3339() {
        let clock = FrozenClock::at("2026-05-10T00:00:00Z");
        assert_eq!(clock.now(), t0());
    }

    #[test]
    fn frozen_clock_at_instant() {
        let clock = FrozenClock::at_instant(t0());
        assert_eq!(clock.now(), t0());
    }

    #[test]
    fn frozen_clock_never_advances() {
        let clock = FrozenClock::at_instant(t0());
        let a = clock.now();
        let b = clock.now();
        assert_eq!(a, b);
    }

    #[test]
    fn frozen_clock_unix_millis() {
        let clock = FrozenClock::at("1970-01-01T00:00:01Z");
        assert_eq!(clock.now_unix_millis(), 1_000);
    }

    #[test]
    fn frozen_at_helper_trait_dispatch() {
        let c: Arc<dyn Clock> = frozen_at("2026-05-10T00:00:00Z");
        assert_eq!(c.now(), t0());
    }

    // ── MockClock tests ────────────────────────────────────────────────

    #[test]
    fn mock_clock_starts_at_given_instant() {
        let clock = MockClock::new(t0());
        assert_eq!(clock.now(), t0());
    }

    #[test]
    fn mock_clock_advance_by_increases_time() {
        let clock = MockClock::new(t0());
        clock.advance_by(Duration::from_secs(1));
        assert_eq!(clock.now(), t0() + jiff::SignedDuration::from_secs(1));
    }

    #[test]
    fn mock_clock_set_now_updates_instant() {
        let clock = MockClock::new(t0());
        let new_now = t0() + jiff::SignedDuration::from_hours(5);
        clock.set_now(new_now);
        assert_eq!(clock.now(), new_now);
    }

    #[test]
    fn mock_clock_trait_dispatch() {
        let c: Arc<dyn Clock> = Arc::new(MockClock::new(t0()));
        assert_eq!(c.now(), t0());
    }

    // ── AdvanceableTimer tests ─────────────────────────────────────────

    #[tokio::test]
    async fn advanceable_timer_sleep_resolves_after_advance() {
        let (clock, timer) = mock_pair(t0());
        let sleep_fut = timer.sleep(Duration::from_secs(60));
        // Advance past the deadline in a concurrent task
        let clock2 = Arc::clone(&clock);
        tokio::spawn(async move {
            clock2.advance_and_fire(Duration::from_secs(60));
        })
        .await
        .ok();
        sleep_fut.await; // should complete
    }

    #[tokio::test]
    async fn advanceable_timer_partial_advance_fires_only_due_sleeps() {
        let (clock, timer) = mock_pair(t0());
        let sleep_30 = timer.sleep(Duration::from_secs(30));
        let sleep_60 = timer.sleep(Duration::from_secs(60));

        clock.advance_and_fire(Duration::from_secs(45));

        // 30s sleep should be done
        sleep_30.await;

        // 60s sleep should NOT yet be done — fire another advance
        clock.advance_and_fire(Duration::from_secs(15));
        sleep_60.await;
    }

    #[tokio::test]
    async fn a_dropped_clock_wakes_its_pending_sleeps() {
        let (clock, timer) = mock_pair(t0());
        let sleep = timer.sleep(Duration::from_secs(60));
        drop(timer);
        drop(clock);
        sleep.await;
    }

    #[tokio::test]
    async fn advanceable_timer_next_tick_resolves_immediately() {
        let (_, timer) = mock_pair(t0());
        timer.next_tick().await;
    }

    #[test]
    fn mock_pair_helper_returns_linked_pair() {
        let (clock, timer) = mock_pair(t0());
        let _: Arc<MockClock> = clock;
        let _: Arc<AdvanceableTimer> = timer;
    }

    #[test]
    fn frozen_clock_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<FrozenClock>();
    }

    #[test]
    fn mock_clock_send_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<MockClock>();
        assert_send_sync::<AdvanceableTimer>();
    }
}