scuriolus 0.1.0

use std::{
    cmp::min,
    future::Future,
    pin::Pin,
    sync::{
        atomic::{AtomicI64, Ordering},
        Arc,
    },
};

use anyhow::Error;
use chrono::{DateTime, TimeDelta, Utc};
use tokio::{select, sync::Notify, time::sleep};

use super::{CheatClockFactory, CheatClockRemote, Clock, ClockBase, SAFETY_REFRESH_DELAY};

trait RunningCheatClockBase {
    fn offset_ms(&self) -> &Arc<AtomicI64>;
    fn running_now(&self) -> DateTime<Utc> {
        let now = Utc::now();
        let offset = self.offset_ms().load(Ordering::Relaxed);
        now - TimeDelta::milliseconds(offset)
    }
}

/// The [`RunningCheatClock`] is a [`Clock`] that mimic a real clock but starting in the past and that can be advanced.
#[derive(Clone, Debug, Default)]
pub struct RunningCheatClock {
    offset_ms: Arc<AtomicI64>,
    remote_offset_ms: Arc<AtomicI64>,
    notify: Arc<Notify>,
}

impl RunningCheatClockBase for RunningCheatClock {
    fn offset_ms(&self) -> &Arc<AtomicI64> {
        &self.offset_ms
    }
}

impl ClockBase for RunningCheatClock {
    fn now(&self) -> DateTime<Utc> {
        self.running_now()
    }
}

impl Clock for RunningCheatClock {
    fn clone_box(&self) -> Box<dyn Clock> {
        Box::new(self.clone())
    }

    fn sleep(&self, delay: TimeDelta) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        self.sleep_until(self.now() + delay)
    }

    fn sleep_until(&self, date: DateTime<Utc>) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            loop {
                let offset = self.offset_ms.load(Ordering::Relaxed);
                let remote_offset = self.remote_offset_ms.load(Ordering::Relaxed);
                let potential_new_offset = (Utc::now() - date).num_milliseconds();

                #[cfg(any(test, debug_assertions))]
                assert!(offset >= remote_offset);

                if offset != remote_offset && offset != potential_new_offset {
                    if potential_new_offset <= remote_offset {
                        self.offset_ms.store(remote_offset, Ordering::Relaxed);
                        self.notify.notify_waiters();
                    } else {
                        self.offset_ms
                            .store(potential_new_offset, Ordering::Relaxed);
                    }
                }

                let now = self.now();
                if now >= date {
                    break;
                }

                select! {
                    _ = sleep(min(TimeDelta::to_std(&(date-now)).unwrap_or_else(|_| {
                        tracing::error!("Failed to sleep, conversion error from TimeDelta to Duration");
                        SAFETY_REFRESH_DELAY
                    }), SAFETY_REFRESH_DELAY)) => {},
                    _ = self.notify.notified() => {}
                }
            }
        })
    }

    fn synchronize(&self) {
        self.offset_ms.store(
            self.remote_offset_ms.load(Ordering::Relaxed),
            Ordering::Relaxed,
        );
    }
}

/// A [`CheatClockRemote`] for [`RunningCheatClock`].
#[derive(Clone, Debug)]
pub struct RunningCheatClockRemote {
    offset_ms: Arc<AtomicI64>,
    remote_offset_ms: Arc<AtomicI64>,
    notify: Arc<Notify>,
}

impl RunningCheatClockBase for RunningCheatClockRemote {
    fn offset_ms(&self) -> &Arc<AtomicI64> {
        &self.offset_ms
    }
}

impl ClockBase for RunningCheatClockRemote {
    fn now(&self) -> DateTime<Utc> {
        self.running_now()
    }
}

impl CheatClockRemote for RunningCheatClockRemote {
    #[cfg(test)]
    fn not_blocking(&self) {
        self.remote_offset_ms.store(0, Ordering::Relaxed);
    }

    fn advance_toward(&self, date: DateTime<Utc>) {
        self.remote_offset_ms
            .store((Utc::now() - date).num_milliseconds(), Ordering::Relaxed);
        self.notify.notify_waiters();
    }

    fn remote_now(&self) -> DateTime<Utc> {
        let now = Utc::now();
        let offset = self.remote_offset_ms.load(Ordering::Relaxed);
        now - TimeDelta::milliseconds(offset)
    }

    fn notify(&self) -> &Notify {
        &self.notify
    }

    fn is_sync(&self) -> bool {
        self.offset_ms.load(Ordering::Relaxed) == self.remote_offset_ms.load(Ordering::Relaxed)
    }

    fn clone_box(&self) -> Box<dyn CheatClockRemote> {
        Box::new(self.clone())
    }
}

/// A [`CheatClockFactory`] for [`RunningCheatClock`].
#[derive(Clone, Debug)]
pub struct RunningCheatClockFactory {}

impl CheatClockFactory for RunningCheatClockFactory {
    fn get_clock(
        &self,
        date: DateTime<Utc>,
    ) -> Result<(Box<dyn CheatClockRemote>, Box<dyn Clock>), anyhow::Error> {
        let now = Utc::now();
        if now < date {
            return Err(Error::msg("Can't set time in the future"));
        }

        let offset_base = (now - date).num_milliseconds();

        let offset_ms = Arc::new(AtomicI64::new(offset_base));
        let remote_offset_ms = Arc::new(AtomicI64::new(offset_base));
        let notify = Arc::new(Notify::new());
        Ok((
            Box::new(RunningCheatClockRemote {
                offset_ms: offset_ms.clone(),
                remote_offset_ms: remote_offset_ms.clone(),
                notify: notify.clone(),
            }),
            Box::new(RunningCheatClock {
                offset_ms,
                remote_offset_ms,
                notify,
            }),
        ))
    }
}

#[cfg(test)]
mod tests {
    use std::{
        panic,
        str::FromStr as _,
        sync::{
            atomic::{AtomicI16, Ordering},
            Arc,
        },
    };

    use chrono::{DateTime, TimeDelta, Utc};
    use tokio::time::sleep as tokio_sleep;

    use super::{super::REMOTE_REFRESH_RATE, CheatClockFactory as _, RunningCheatClockFactory};

    const TOLERANCE: TimeDelta = TimeDelta::milliseconds(10);
    const DELAY: TimeDelta = TimeDelta::milliseconds(100);

    #[tokio::test]
    async fn sleep() {
        let date = DateTime::from_str("2012-12-12 00:00:00Z").unwrap();
        let (remote, clock) = RunningCheatClockFactory {}.get_clock(date).unwrap();

        remote.not_blocking();

        assert!(clock.now() - date < TOLERANCE);

        let before = Utc::now();
        clock.sleep(DELAY).await;
        let after = Utc::now();

        assert!(after - before < TOLERANCE);

        assert!(clock.now() - date > DELAY - TOLERANCE);
        assert!(clock.now() - date < DELAY + TOLERANCE);
    }

    #[tokio::test]
    async fn sleep_until() {
        let date = DateTime::from_str("2012-12-12 00:00:00Z").unwrap();
        let (remote, clock) = RunningCheatClockFactory {}.get_clock(date).unwrap();

        remote.not_blocking();

        assert!(clock.now() - date < TOLERANCE);

        let before = Utc::now();
        clock.sleep_until(clock.now() + DELAY).await;
        let after = Utc::now();

        assert!(after - before < TOLERANCE);

        assert!(clock.now() - date > DELAY - TOLERANCE);
        assert!(clock.now() - date < DELAY + TOLERANCE);
    }

    #[tokio::test]
    async fn remote() {
        let date = DateTime::from_str("2012-12-12 00:00:00Z").unwrap();
        let (remote, clock) = RunningCheatClockFactory {}.get_clock(date).unwrap();

        assert!(clock.now() - date < TOLERANCE);
        assert!(remote.now() - date < TOLERANCE);

        let marker = Arc::new(AtomicI16::new(0));
        let marker_clone = marker.clone();

        tokio::spawn(async move {
            clock.sleep(DELAY).await;
            marker_clone.store(1, Ordering::Relaxed);
            clock.sleep(DELAY).await;
            marker_clone.store(2, Ordering::Relaxed);
            clock
                .sleep_until(DateTime::from_str("2020-01-01 00:00:00Z").unwrap())
                .await;
            panic!("should not reach here");
        });

        assert_eq!(marker.load(Ordering::Relaxed), 0);
        tokio_sleep(DELAY.to_std().unwrap()).await;
        //now wait for the clock to auto-update
        tokio_sleep(REMOTE_REFRESH_RATE + TOLERANCE.to_std().unwrap()).await;
        assert_eq!(marker.load(Ordering::Relaxed), 1);

        remote.advance_toward(date + DELAY + DELAY);

        //now wait for the arc to update
        tokio_sleep(TOLERANCE.to_std().unwrap()).await;

        assert_eq!(marker.load(Ordering::Relaxed), 2);
    }
}