nodedb_lite/

runtime.rs

//! Platform-specific async runtime abstractions.
//!
//! NodeDB-Lite compiles for native (Tokio) and WASM (`wasm-bindgen-futures`).
//! This module provides a thin abstraction over the differences so engine
//! code doesn't need `#[cfg]` everywhere.
//!
//! **Native (iOS/Android/Desktop):** Tokio — `spawn`, `spawn_blocking`, `sleep`, `interval`.
//! **WASM (Browser):** `wasm-bindgen-futures` + `gloo-timers` — `spawn_local`, no blocking
//! threads, timer-backed sleep and interval.

use std::future::Future;
use std::time::Duration;

/// Spawn a future on the runtime, returning a handle to the running task.
///
/// - Native: `tokio::spawn` (runs on Tokio thread pool, requires `Send`).
/// - WASM: `wasm_bindgen_futures::spawn_local` (runs on the microtask queue,
///   no `Send` requirement).
///
/// Ignoring the handle detaches the task: nothing can then stop it or wait for
/// it. Register it with [`crate::tasks::TaskRegistry`] instead, so database
/// shutdown can stop it before the runtime goes away.
#[cfg(not(target_arch = "wasm32"))]
pub fn spawn<F>(future: F) -> TaskHandle
where
    F: Future<Output = ()> + Send + 'static,
{
    TaskHandle {
        inner: tokio::spawn(future),
    }
}

#[cfg(target_arch = "wasm32")]
pub fn spawn<F>(future: F) -> TaskHandle
where
    F: Future<Output = ()> + 'static,
{
    wasm_bindgen_futures::spawn_local(future);
    TaskHandle
}

/// Handle to a task started by [`spawn`].
///
/// Native wraps a Tokio `JoinHandle`, so the task can be joined and, as a last
/// resort, aborted. WASM has neither primitive — `spawn_local` returns nothing
/// — so the handle is inert there and shutdown rests entirely on the
/// cooperative stop signal.
#[cfg(not(target_arch = "wasm32"))]
pub struct TaskHandle {
    inner: tokio::task::JoinHandle<()>,
}

#[cfg(target_arch = "wasm32")]
pub struct TaskHandle;

impl TaskHandle {
    /// Cancel the task at its next await point.
    ///
    /// The backstop for a task that does not observe its stop signal. Prefer
    /// the signal: abort drops the future wherever it happens to be suspended,
    /// which can be mid-write.
    pub fn abort(&self) {
        #[cfg(not(target_arch = "wasm32"))]
        {
            self.inner.abort();
        }
    }

    /// Wait for the task to finish, up to `timeout`.
    ///
    /// Returns `true` when the task finished, `false` on timeout — leaving the
    /// caller free to [`abort`](Self::abort) it. Always `true` on WASM, which
    /// has nothing to join.
    pub async fn join_within(&mut self, timeout: Duration) -> bool {
        #[cfg(not(target_arch = "wasm32"))]
        {
            tokio::select! {
                _ = &mut self.inner => true,
                _ = tokio::time::sleep(timeout) => false,
            }
        }
        #[cfg(target_arch = "wasm32")]
        {
            let _ = timeout;
            true
        }
    }
}

/// Run a blocking closure off the async runtime.
///
/// - Native: `tokio::task::spawn_blocking` — moves closure to the blocking pool.
/// - WASM: Runs synchronously (WASM has no blocking pool; callers must
///   ensure the closure is fast or use the async StorageEngine path).
#[cfg(not(target_arch = "wasm32"))]
pub async fn spawn_blocking<F, T>(f: F) -> Result<T, crate::error::LiteError>
where
    F: FnOnce() -> T + Send + 'static,
    T: Send + 'static,
{
    tokio::task::spawn_blocking(f)
        .await
        .map_err(|e| crate::error::LiteError::JoinError {
            detail: e.to_string(),
        })
}

#[cfg(target_arch = "wasm32")]
pub async fn spawn_blocking<F, T>(f: F) -> Result<T, crate::error::LiteError>
where
    F: FnOnce() -> T,
{
    // No blocking pool on WASM — run synchronously.
    // This is acceptable because:
    // 1. SQLite WASM operations are fast (in-memory or OPFS sync access)
    // 2. HNSW/CSR operations are CPU-bound but sub-millisecond for edge datasets
    Ok(f())
}

/// Sleep for a duration.
///
/// - Native: `tokio::time::sleep`.
/// - WASM: `gloo_timers::future::sleep` (backed by JS `setTimeout`).
#[cfg(not(target_arch = "wasm32"))]
pub async fn sleep(duration: Duration) {
    tokio::time::sleep(duration).await;
}

#[cfg(target_arch = "wasm32")]
pub async fn sleep(duration: Duration) {
    gloo_timers::future::sleep(duration).await;
}

/// A recurring interval timer.
///
/// Obtain one via [`interval`]. Call `.tick().await` to wait for each period.
///
/// On native the first `tick()` returns immediately (matches Tokio semantics).
/// On WASM the first `tick()` waits one full period. The primary consumer
/// (sync keepalive) tolerates either behaviour.
pub struct Interval {
    #[cfg(not(target_arch = "wasm32"))]
    inner: tokio::time::Interval,
    #[cfg(target_arch = "wasm32")]
    period: Duration,
}

impl Interval {
    /// Measure the next period from when this tick is consumed rather than
    /// from the schedule, so a slow consumer does not come back to a burst of
    /// catch-up ticks.
    ///
    /// Tokio's default replays every missed tick immediately. For a periodic
    /// task whose work can outlast its own period — a flush, say — that turns
    /// one slow pass into back-to-back passes with no gap between them, and the
    /// task never yields long enough for anything else to make progress. On
    /// WASM each tick already sleeps a full period after the previous one, so
    /// this is the behaviour there either way.
    pub fn delay_missed_ticks(&mut self) {
        #[cfg(not(target_arch = "wasm32"))]
        {
            self.inner
                .set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        }
    }

    /// Wait until the next tick.
    pub async fn tick(&mut self) {
        #[cfg(not(target_arch = "wasm32"))]
        {
            self.inner.tick().await;
        }
        #[cfg(target_arch = "wasm32")]
        {
            gloo_timers::future::sleep(self.period).await;
        }
    }
}

/// Create a recurring interval timer that ticks every `period`.
///
/// - Native: wraps `tokio::time::interval`; first tick is immediate.
/// - WASM: backed by `gloo_timers`; first tick waits one period.
pub fn interval(period: Duration) -> Interval {
    #[cfg(not(target_arch = "wasm32"))]
    {
        Interval {
            inner: tokio::time::interval(period),
        }
    }
    #[cfg(target_arch = "wasm32")]
    {
        Interval { period }
    }
}

/// Get the current timestamp in milliseconds since Unix epoch.
///
/// Platform-independent — works on native and WASM.
pub fn now_millis() -> u64 {
    #[cfg(not(target_arch = "wasm32"))]
    {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64
    }
    #[cfg(target_arch = "wasm32")]
    {
        // js_sys::Date::now() returns milliseconds since epoch as f64.
        js_sys::Date::now() as u64
    }
}

/// Get the current timestamp in milliseconds since Unix epoch, as `i64`.
///
/// Same clock as [`now_millis`] but signed, for the system/valid-time fields
/// used by the bitemporal engines. Platform-independent — works on native and
/// WASM.
pub fn now_millis_i64() -> i64 {
    now_millis() as i64
}

/// Last system-time millisecond handed out by [`monotonic_millis_i64`].
static LAST_SYS_MS: std::sync::atomic::AtomicI64 = std::sync::atomic::AtomicI64::new(0);

/// Get a strictly-monotonic system timestamp in milliseconds since Unix epoch.
///
/// Like [`now_millis_i64`] but guarantees the returned value is strictly greater
/// than any value previously returned **within this process**. When the
/// wall-clock has not advanced since the last call (or moves backwards), the
/// previous value `+ 1` is returned instead.
///
/// The bitemporal document history keys each version by `system_from_ms`
/// (`{collection}:{doc_id}\0{system_from_ms:020}`). Two writes landing in the
/// same wall-clock millisecond — e.g. a `put` immediately followed by a
/// valid-time close — would otherwise derive the identical version key and the
/// second write would silently overwrite the first (the storage layer is plain
/// last-write-wins KV). This clock gives every version a distinct, ordered key.
/// The sub-millisecond forward skew under burst is bounded by the write rate and
/// self-corrects once wall-clock time catches up.
pub fn monotonic_millis_i64() -> i64 {
    use std::sync::atomic::Ordering;
    let now = now_millis_i64();
    loop {
        let last = LAST_SYS_MS.load(Ordering::Relaxed);
        let next = if now > last { now } else { last + 1 };
        match LAST_SYS_MS.compare_exchange_weak(last, next, Ordering::Relaxed, Ordering::Relaxed) {
            Ok(_) => return next,
            Err(_) => continue,
        }
    }
}

/// Get the current timestamp in seconds since Unix epoch.
pub fn now_secs() -> u64 {
    now_millis() / 1000
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn spawn_blocking_works() {
        let result = spawn_blocking(|| 42).await.unwrap();
        assert_eq!(result, 42);
    }

    #[tokio::test]
    async fn spawn_blocking_string() {
        let result = spawn_blocking(|| "hello".to_string()).await.unwrap();
        assert_eq!(result, "hello");
    }

    #[tokio::test]
    async fn sleep_returns() {
        // Just verify it doesn't hang.
        sleep(Duration::from_millis(1)).await;
    }

    #[test]
    fn now_millis_nonzero() {
        let ts = now_millis();
        assert!(ts > 0, "timestamp should be nonzero on native");
    }

    #[test]
    fn now_secs_reasonable() {
        let ts = now_secs();
        // Should be after 2024-01-01 (1704067200).
        assert!(ts > 1_704_067_200, "timestamp {ts} seems too old");
    }

    #[tokio::test]
    async fn interval_ticks_twice() {
        let mut iv = interval(Duration::from_millis(1));
        // First tick is immediate on native (Tokio semantics).
        iv.tick().await;
        // Second tick waits one period — should still resolve promptly.
        iv.tick().await;
    }

    #[tokio::test]
    async fn spawn_fires() {
        let (tx, rx) = tokio::sync::oneshot::channel();
        spawn(async move {
            let _ = tx.send(42);
        });
        let val = rx.await.unwrap();
        assert_eq!(val, 42);
    }
}