openrtc 1.0.4

//! Phase 5 of the OpenRTC auth + connection remediation plan
//! (`docs/plans/openrtc-auth-connection-remediation-plan.md`).
//!
//! Rust mirror of `src/services/runtime/authReadiness.ts`. This is a
//! single observable readiness gate combining the four auth legs the
//! TS bridge pushes (`filesAuth`, `plutoRtcAuth`, `runtimeAuth`,
//! `firestore`). The drive-grant connection actor consults the gate
//! before its first dial so a Firestore rendezvous read does not race
//! a not-yet-resolved auth token.
//!
//! Both the TS and Rust stores follow the same shape:
//!
//!   - Each leg starts `Pending`.
//!   - Setters mutate one leg at a time and notify subscribers.
//!   - `firestore` is auto-derived from `plutoRtcAuth + runtimeAuth`
//!     (both `Ready` ⇒ `Ready`; either `Error` ⇒ `Error`; otherwise
//!     `Pending`). Callers should not push `firestore` directly unless
//!     a future phase tracks explicit Firestore staleness.
//!   - `token_epoch` increments on a `runtimeAuth` non-ready→ready
//!     transition so consumers can invalidate cached reads.
//!
//! ## How `DriveGrantConnectionActor` uses this
//!
//! Phase 5 wires `DriveGrantConnectionActor::ensure_ready` to call
//! `AuthReadinessStore::wait_until_ready` *before* the first dial.
//! Production code that needs a fully ready auth context (drive-share
//! rendezvous reads, drive-grant Firestore listeners) calls
//! `wait_until_ready` directly.

#![cfg(not(target_arch = "wasm32"))]

use std::time::Duration;

use tokio::sync::watch;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthLegState {
    Pending,
    Ready,
    Error,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthLeg {
    FilesAuth,
    PlutoRtcAuth,
    RuntimeAuth,
    Firestore,
}

const ALL_LEGS: &[AuthLeg] = &[
    AuthLeg::FilesAuth,
    AuthLeg::PlutoRtcAuth,
    AuthLeg::RuntimeAuth,
    AuthLeg::Firestore,
];

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthReadinessSnapshot {
    pub files_auth: AuthLegState,
    pub pluto_rtc_auth: AuthLegState,
    pub runtime_auth: AuthLegState,
    pub firestore: AuthLegState,
    pub token_epoch: u64,
    pub last_error: Option<String>,
}

impl AuthReadinessSnapshot {
    fn pending() -> Self {
        Self {
            files_auth: AuthLegState::Pending,
            pluto_rtc_auth: AuthLegState::Pending,
            runtime_auth: AuthLegState::Pending,
            firestore: AuthLegState::Pending,
            token_epoch: 0,
            last_error: None,
        }
    }

    pub fn leg(&self, leg: AuthLeg) -> AuthLegState {
        match leg {
            AuthLeg::FilesAuth => self.files_auth,
            AuthLeg::PlutoRtcAuth => self.pluto_rtc_auth,
            AuthLeg::RuntimeAuth => self.runtime_auth,
            AuthLeg::Firestore => self.firestore,
        }
    }

    pub fn legs_ready(&self, legs: &[AuthLeg]) -> bool {
        legs.iter().all(|leg| self.leg(*leg) == AuthLegState::Ready)
    }

    pub fn is_ready(&self) -> bool {
        self.legs_ready(ALL_LEGS)
    }
}

#[derive(Debug, Clone)]
pub enum AuthReadinessWaitError {
    Timeout {
        remaining: Vec<AuthLeg>,
        elapsed: Duration,
    },
    Closed,
}

impl std::fmt::Display for AuthReadinessWaitError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AuthReadinessWaitError::Timeout { remaining, elapsed } => write!(
                f,
                "auth-readiness: timed out after {:?} waiting for legs: {:?}",
                elapsed, remaining
            ),
            AuthReadinessWaitError::Closed => {
                write!(f, "auth-readiness: store dropped before legs reached ready")
            }
        }
    }
}

impl std::error::Error for AuthReadinessWaitError {}

/// Thread-safe readiness store. Cheap to clone (`watch::Sender` is
/// internally `Arc`-backed); production wires one shared instance into
/// the `Client`.
#[derive(Clone)]
pub struct AuthReadinessStore {
    tx: watch::Sender<AuthReadinessSnapshot>,
}

impl AuthReadinessStore {
    pub fn new() -> Self {
        let (tx, _rx) = watch::channel(AuthReadinessSnapshot::pending());
        Self { tx }
    }

    pub fn snapshot(&self) -> AuthReadinessSnapshot {
        self.tx.borrow().clone()
    }

    pub fn subscribe(&self) -> watch::Receiver<AuthReadinessSnapshot> {
        self.tx.subscribe()
    }

    pub fn mark_files_auth(&self, state: AuthLegState, error: Option<String>) {
        self.apply(AuthLeg::FilesAuth, state, error);
    }
    pub fn mark_pluto_rtc_auth(&self, state: AuthLegState, error: Option<String>) {
        self.apply(AuthLeg::PlutoRtcAuth, state, error);
    }
    pub fn mark_runtime_auth(&self, state: AuthLegState, error: Option<String>) {
        self.apply(AuthLeg::RuntimeAuth, state, error);
    }
    pub fn mark_firestore(&self, state: AuthLegState, error: Option<String>) {
        self.apply(AuthLeg::Firestore, state, error);
    }

    /// Reset to the initial pending state without bumping `token_epoch`.
    /// Used on sign-out so subsequent `wait_until_ready` waiters block
    /// until the next sign-in.
    pub fn reset(&self) {
        self.tx.send_modify(|snapshot| {
            let epoch = snapshot.token_epoch;
            *snapshot = AuthReadinessSnapshot::pending();
            snapshot.token_epoch = epoch;
        });
    }

    /// Wait until every leg in `legs` reaches `Ready`. Resolves
    /// immediately if already ready. If `timeout` is `Some(_)` and
    /// elapses before all legs are ready, returns
    /// `AuthReadinessWaitError::Timeout` with the laggards. If the
    /// store is dropped, returns `AuthReadinessWaitError::Closed`.
    pub async fn wait_until_ready(
        &self,
        legs: &[AuthLeg],
        timeout: Option<Duration>,
    ) -> Result<(), AuthReadinessWaitError> {
        let legs: Vec<AuthLeg> = if legs.is_empty() {
            ALL_LEGS.to_vec()
        } else {
            legs.to_vec()
        };
        if self.tx.borrow().legs_ready(&legs) {
            return Ok(());
        }

        let mut rx = self.tx.subscribe();
        #[cfg(target_arch = "wasm32")]
        let started_ms = js_sys::Date::now();
        #[cfg(not(target_arch = "wasm32"))]
        let started = std::time::Instant::now();
        let wait = async {
            loop {
                if rx.borrow().legs_ready(&legs) {
                    return Ok::<(), AuthReadinessWaitError>(());
                }
                if rx.changed().await.is_err() {
                    return Err(AuthReadinessWaitError::Closed);
                }
            }
        };

        match timeout {
            Some(duration) => match tokio::time::timeout(duration, wait).await {
                Ok(result) => result,
                Err(_) => {
                    let snapshot = self.tx.borrow();
                    let remaining = legs
                        .iter()
                        .copied()
                        .filter(|leg| snapshot.leg(*leg) != AuthLegState::Ready)
                        .collect();
                    #[cfg(target_arch = "wasm32")]
                    let elapsed = Duration::from_millis(
                        (js_sys::Date::now().saturating_sub(started_ms)).max(0.0) as u64,
                    );
                    #[cfg(not(target_arch = "wasm32"))]
                    let elapsed = started.elapsed();
                    Err(AuthReadinessWaitError::Timeout { remaining, elapsed })
                }
            },
            None => wait.await,
        }
    }

    fn apply(&self, leg: AuthLeg, next: AuthLegState, error: Option<String>) {
        self.tx.send_modify(|snapshot| {
            let was_runtime_ready = snapshot.runtime_auth == AuthLegState::Ready;
            let previous_state = snapshot.leg(leg);
            let previous_error = snapshot.last_error.clone();

            if previous_state == next && previous_error == error {
                return;
            }

            match leg {
                AuthLeg::FilesAuth => snapshot.files_auth = next,
                AuthLeg::PlutoRtcAuth => snapshot.pluto_rtc_auth = next,
                AuthLeg::RuntimeAuth => snapshot.runtime_auth = next,
                AuthLeg::Firestore => snapshot.firestore = next,
            }

            snapshot.last_error = match next {
                AuthLegState::Error => error.or(snapshot.last_error.clone()),
                _ => None,
            };

            // Bump token epoch on runtimeAuth non-ready→ready edge.
            if leg == AuthLeg::RuntimeAuth && next == AuthLegState::Ready && !was_runtime_ready {
                snapshot.token_epoch = snapshot.token_epoch.saturating_add(1);
            }

            // Auto-derive `firestore` once a non-firestore leg moves.
            if leg != AuthLeg::Firestore {
                let derived = match (snapshot.pluto_rtc_auth, snapshot.runtime_auth) {
                    (AuthLegState::Ready, AuthLegState::Ready) => AuthLegState::Ready,
                    (AuthLegState::Error, _) | (_, AuthLegState::Error) => AuthLegState::Error,
                    _ => AuthLegState::Pending,
                };
                if snapshot.firestore != derived {
                    snapshot.firestore = derived;
                }
            }
        });
    }
}

impl Default for AuthReadinessStore {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn starts_pending_across_all_legs() {
        let store = AuthReadinessStore::new();
        let snapshot = store.snapshot();

        assert_eq!(snapshot.files_auth, AuthLegState::Pending);
        assert_eq!(snapshot.pluto_rtc_auth, AuthLegState::Pending);
        assert_eq!(snapshot.runtime_auth, AuthLegState::Pending);
        assert_eq!(snapshot.firestore, AuthLegState::Pending);
        assert_eq!(snapshot.token_epoch, 0);
        assert_eq!(snapshot.last_error, None);
        assert!(!snapshot.is_ready());
    }

    #[test]
    fn firestore_is_derived_from_pluto_rtc_and_runtime_auth() {
        let store = AuthReadinessStore::new();

        store.mark_pluto_rtc_auth(AuthLegState::Ready, None);
        assert_eq!(store.snapshot().firestore, AuthLegState::Pending);

        store.mark_runtime_auth(AuthLegState::Ready, None);
        assert_eq!(store.snapshot().firestore, AuthLegState::Ready);

        store.mark_pluto_rtc_auth(AuthLegState::Error, Some("expired".to_string()));
        let snapshot = store.snapshot();
        assert_eq!(snapshot.firestore, AuthLegState::Error);
        assert_eq!(snapshot.last_error.as_deref(), Some("expired"));
    }

    #[test]
    fn token_epoch_bumps_on_runtime_ready_edges() {
        let store = AuthReadinessStore::new();

        store.mark_runtime_auth(AuthLegState::Ready, None);
        assert_eq!(store.snapshot().token_epoch, 1);

        store.mark_runtime_auth(AuthLegState::Ready, None);
        assert_eq!(store.snapshot().token_epoch, 1);

        store.mark_runtime_auth(AuthLegState::Pending, None);
        store.mark_runtime_auth(AuthLegState::Ready, None);
        assert_eq!(store.snapshot().token_epoch, 2);
    }

    #[test]
    fn reset_restores_pending_while_preserving_token_epoch() {
        let store = AuthReadinessStore::new();

        store.mark_runtime_auth(AuthLegState::Ready, None);
        let epoch = store.snapshot().token_epoch;
        assert_eq!(epoch, 1);

        store.reset();
        let snapshot = store.snapshot();
        assert_eq!(snapshot.files_auth, AuthLegState::Pending);
        assert_eq!(snapshot.pluto_rtc_auth, AuthLegState::Pending);
        assert_eq!(snapshot.runtime_auth, AuthLegState::Pending);
        assert_eq!(snapshot.firestore, AuthLegState::Pending);
        assert_eq!(snapshot.token_epoch, epoch);
    }

    #[tokio::test]
    async fn wait_until_ready_resolves_after_legs_ready_in_any_order() {
        let store = AuthReadinessStore::new();
        let waiter = {
            let store = store.clone();
            tokio::spawn(async move { store.wait_until_ready(&[], None).await })
        };

        store.mark_runtime_auth(AuthLegState::Ready, None);
        store.mark_files_auth(AuthLegState::Ready, None);
        store.mark_pluto_rtc_auth(AuthLegState::Ready, None);

        waiter.await.expect("join").expect("ready");
    }

    #[tokio::test]
    async fn wait_until_ready_honors_requested_subset() {
        let store = AuthReadinessStore::new();
        let waiter = {
            let store = store.clone();
            tokio::spawn(async move {
                store
                    .wait_until_ready(
                        &[AuthLeg::RuntimeAuth, AuthLeg::Firestore],
                        Some(Duration::from_secs(1)),
                    )
                    .await
            })
        };

        store.mark_files_auth(AuthLegState::Error, Some("ignored".to_string()));
        store.mark_pluto_rtc_auth(AuthLegState::Ready, None);
        store.mark_runtime_auth(AuthLegState::Ready, None);

        waiter.await.expect("join").expect("subset ready");
    }

    #[tokio::test]
    async fn wait_until_ready_times_out_with_remaining_legs() {
        let store = AuthReadinessStore::new();
        let err = store
            .wait_until_ready(
                &[AuthLeg::FilesAuth, AuthLeg::RuntimeAuth],
                Some(Duration::from_millis(10)),
            )
            .await
            .expect_err("timeout");

        match err {
            AuthReadinessWaitError::Timeout { remaining, .. } => {
                assert_eq!(remaining, vec![AuthLeg::FilesAuth, AuthLeg::RuntimeAuth]);
            }
            other => panic!("expected timeout, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn subscribe_observes_state_transitions() {
        let store = AuthReadinessStore::new();
        let mut rx = store.subscribe();

        assert_eq!(rx.borrow().files_auth, AuthLegState::Pending);
        store.mark_files_auth(AuthLegState::Ready, None);
        rx.changed().await.expect("changed");
        assert_eq!(rx.borrow().files_auth, AuthLegState::Ready);
    }
}