openrtc 1.0.2

//! Phase 3 of the OpenRTC auth + connection remediation plan
//! (`docs/plans/openrtc-auth-connection-remediation-plan.md`).
//!
//! `DriveGrantConnectionActor` defines the intended single-owner model for the
//! dial → retry → admission → logical-channel → WebRTC-upgrade lifecycle of a
//! drive-grant guest connection, keyed by `{scope, peer_node_id}`. Once wired
//! into production startup, requesters (`DriveViewClient`, `DeviceContext`,
//! drive-share rendezvous, etc.) will become observers that call
//! `ensure_ready()` and `open_logical_channel(label)` instead of
//! dialing/retrying/replacing transports directly. That ownership model prevents
//! the "zombie-replaced-by-fresh-dial" loop by queuing concurrent work through
//! one actor.
//!
//! This module is not currently enabled by any workspace production startup and
//! `LogicalChannel` remains a placeholder. It is therefore scaffolding, not the
//! active runtime authority. Keep that distinction explicit until the actor is
//! either fully wired end-to-end or removed as an unused experiment.
//!
//! ## Scope (Phase 3)
//!
//! Phase 3 lands the actor + registry + unit tests, behind a feature flag on
//! `Client`. Production code paths (`connect_device`,
//! `try_consume_pending_sdk_token_stream`) are *not yet* rewired in this
//! phase — that is Phase 4's job. With the flag default-off the legacy path
//! continues to run; tests assert legacy behaviour is unchanged.
//!
//! ## API
//!
//! ```ignore
//! let registry = client.scoped_connection_actor_registry();
//! let actor = registry.get_or_spawn(DriveGrantConnectionKey::new("drive-grant:abc", node));
//! actor.ensure_ready().await?;          // coalesces N concurrent callers → 1 dial
//! let mut webrtc_rx = actor.observe_webrtc();
//! let _channel = actor.open_logical_channel("drive-view.list").await?;
//! actor.shutdown().await;               // cancels pending dials, idempotent
//! ```
//!
//! Phase 6 fills in the actual `LogicalChannel` wire format on top of base
//! iroh; Phase 3 returns a placeholder handle so observers can exercise the
//! API shape.

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

use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{watch, Mutex, OnceCell, RwLock};

use crate::client::auth_readiness::{AuthLeg, AuthReadinessStore, AuthReadinessWaitError};
use crate::client::correlation::CorrelationContext;
use crate::clog;

/// Identifies one drive-grant guest connection. Multiple requesters that
/// share the same `(scope, peer_node_id)` map to the same actor.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DriveGrantConnectionKey {
    pub scope: String,
    pub peer_node_id: String,
}

impl DriveGrantConnectionKey {
    pub fn new(scope: impl Into<String>, peer_node_id: impl Into<String>) -> Self {
        Self {
            scope: scope.into(),
            peer_node_id: peer_node_id.into(),
        }
    }
}

impl fmt::Display for DriveGrantConnectionKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}@{}", self.scope, self.peer_node_id)
    }
}

/// Observable WebRTC adoption state. Phase 3 reports `Unknown` and `Disabled`;
/// Phase 6/7 wire `Connecting`/`Connected` from the existing native WebRTC
/// state machine.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WebRtcState {
    Unknown,
    Disabled,
    Connecting,
    Connected,
    Failed(String),
}

/// Errors surfaced from the actor's public API.
#[derive(Debug, Clone)]
pub enum ActorError {
    /// The actor has been shut down. Callers should drop their handle.
    Shutdown,
    /// `ensure_ready()` exhausted its retry budget without a healthy
    /// transport. The string carries the most recent underlying error.
    DialFailed(String),
    /// `open_logical_channel(label)` was called before `ensure_ready()`
    /// resolved. Phase 3 enforces this ordering; later phases may relax
    /// it once channels can be lazily attached.
    NotReady,
    /// Phase 3 placeholder: full `LogicalChannel` over base iroh is the
    /// product of Phase 6. Until Phase 6 lands the actor returns this for
    /// any label.
    LogicalChannelNotImplemented(String),
    /// Phase 5: the auth-readiness gate did not reach `Ready` within
    /// the actor's wait budget before the first dial. Carries the legs
    /// that were still not ready.
    AuthNotReady(String),
}

impl fmt::Display for ActorError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            ActorError::Shutdown => write!(f, "drive-grant-actor: already shut down"),
            ActorError::DialFailed(reason) => {
                write!(f, "drive-grant-actor: dial failed: {reason}")
            }
            ActorError::NotReady => write!(
                f,
                "drive-grant-actor: ensure_ready() must resolve before open_logical_channel"
            ),
            ActorError::LogicalChannelNotImplemented(label) => write!(
                f,
                "drive-grant-actor: logical channel '{label}' wire format pending Phase 6"
            ),
            ActorError::AuthNotReady(reason) => write!(
                f,
                "drive-grant-actor: auth-readiness gate not ready before first dial: {reason}"
            ),
        }
    }
}

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

/// Strategy for dialing the underlying iroh transport. Phase 3 wires this
/// behind a trait so unit tests can drive the actor without booting an iroh
/// endpoint, and so Phase 4 can plug in the real `Client::ensure_connected_addr`
/// path.
#[async_trait::async_trait]
pub trait DriveGrantTransport: Send + Sync + 'static {
    /// Establish (or reuse) a base iroh transport to the peer named by `key`.
    /// Must be idempotent: returning `Ok(())` on every call when a healthy
    /// transport already exists. Errors are retried by the actor.
    async fn dial(&self, key: &DriveGrantConnectionKey) -> Result<(), String>;

    /// Tear down any actor-owned transport state for `key`. Called by
    /// `shutdown()`. Default no-op so tests don't have to implement it.
    async fn close(&self, _key: &DriveGrantConnectionKey) {}
}

/// Configuration for retry timing. Defaults match production heuristics
/// derived from `logs-desktop-A.md` retry intervals (50ms-2s with caps).
#[derive(Debug, Clone)]
pub struct DriveGrantActorConfig {
    pub initial_retry: Duration,
    pub max_retry: Duration,
    /// Maximum number of *consecutive* dial failures before `ensure_ready()`
    /// returns `ActorError::DialFailed`. Successful dials reset this counter.
    pub max_consecutive_failures: u32,
    /// Phase 5: how long to wait for the auth-readiness gate before the
    /// first dial when a store is attached. Defaults to 30s — long
    /// enough for a cold-start auth round-trip but short enough that a
    /// stuck auth state surfaces an explicit error rather than hanging.
    /// `None` disables the timeout (waits forever).
    pub auth_readiness_wait_timeout: Option<Duration>,
}

impl Default for DriveGrantActorConfig {
    fn default() -> Self {
        Self {
            initial_retry: Duration::from_millis(250),
            max_retry: Duration::from_secs(4),
            max_consecutive_failures: 8,
            auth_readiness_wait_timeout: Some(Duration::from_secs(30)),
        }
    }
}

/// Placeholder handle for a logical channel. Phase 6 fills in the wire
/// format (multiplexed framing on the base iroh stream); Phase 3 only
/// returns the label so callers can verify the API surface.
#[derive(Debug, Clone)]
pub struct LogicalChannelHandle {
    label: String,
}

impl LogicalChannelHandle {
    pub fn label(&self) -> &str {
        &self.label
    }
}

#[derive(Debug)]
struct DialState {
    /// Number of consecutive failures since the last successful dial.
    consecutive_failures: u32,
    /// Whether at least one successful dial has resolved.
    ever_succeeded: bool,
    /// `Some(reason)` once `shutdown()` has been called.
    shutdown_reason: Option<String>,
}

impl DialState {
    fn new() -> Self {
        Self {
            consecutive_failures: 0,
            ever_succeeded: false,
            shutdown_reason: None,
        }
    }
}

/// The actor itself. Cheap to clone (`Arc`-backed); registry hands out a
/// shared `Arc<Self>` to every requester for a given key.
pub struct DriveGrantConnectionActor {
    key: DriveGrantConnectionKey,
    transport: Arc<dyn DriveGrantTransport>,
    config: DriveGrantActorConfig,
    /// Phase 5: optional auth-readiness gate. When `Some`, the actor
    /// awaits `wait_until_ready` for `auth_readiness_legs` before its
    /// first dial. None ⇒ legacy behavior (no auth gate).
    auth_readiness: Option<Arc<AuthReadinessStore>>,
    /// Phase 5: which legs to await. Defaults to all four; the actor
    /// only consults Firestore + runtime auth in practice but waiting
    /// for filesAuth too aligns with the TS bridge's invariant that
    /// the user is signed in.
    auth_readiness_legs: Vec<AuthLeg>,
    /// Serializes admission/replacement decisions so concurrent inbound
    /// dials can never run lifecycle hooks at the same time as the
    /// dialer's own ensure_ready.
    select_loop: Mutex<()>,
    state: RwLock<DialState>,
    /// Cached "ready" coalesce barrier. Concurrent `ensure_ready()` calls
    /// before the first success share this future.
    ready_once: OnceCell<()>,
    /// Async observers of the WebRTC adoption state.
    webrtc_tx: watch::Sender<WebRtcState>,
}

impl DriveGrantConnectionActor {
    fn new(
        key: DriveGrantConnectionKey,
        transport: Arc<dyn DriveGrantTransport>,
        config: DriveGrantActorConfig,
        auth_readiness: Option<Arc<AuthReadinessStore>>,
        auth_readiness_legs: Vec<AuthLeg>,
    ) -> Arc<Self> {
        let (webrtc_tx, _rx) = watch::channel(WebRtcState::Unknown);
        Arc::new(Self {
            key,
            transport,
            config,
            auth_readiness,
            auth_readiness_legs,
            select_loop: Mutex::new(()),
            state: RwLock::new(DialState::new()),
            ready_once: OnceCell::new(),
            webrtc_tx,
        })
    }

    pub fn key(&self) -> &DriveGrantConnectionKey {
        &self.key
    }

    /// Phase 7: build a correlation context for log lines emitted by the
    /// actor. Always populated with `session_kind=drive-grant-guest`,
    /// `scope`, and `peer_node_id`; `grant_id` is auto-extracted from
    /// `drive-grant:<id>`-style scopes.
    fn correlation(&self) -> CorrelationContext {
        let mut ctx = CorrelationContext::from_drive_grant_scope(&self.key.scope)
            .session_kind("drive-grant-guest")
            .peer_node_id(&self.key.peer_node_id);
        // `from_drive_grant_scope` already set the scope; ensure the
        // `session_kind` chain didn't drop it (it doesn't, but explicit
        // build order keeps the helper future-proof).
        if ctx.scope.is_none() {
            ctx = ctx.scope(&self.key.scope);
        }
        ctx
    }

    /// Current observable WebRTC state without subscribing.
    pub fn webrtc_state(&self) -> WebRtcState {
        self.webrtc_tx.borrow().clone()
    }

    /// Subscribe to WebRTC adoption state changes. Non-blocking; later
    /// phases (4+) plug into the existing native WebRTC state machine.
    pub fn observe_webrtc(&self) -> watch::Receiver<WebRtcState> {
        self.webrtc_tx.subscribe()
    }

    /// Test/Phase-4 hook: lifecycle code that learns of a WebRTC state
    /// transition (e.g. native session moved Connecting → Connected) calls
    /// this so observers see the change.
    pub fn set_webrtc_state(&self, state: WebRtcState) {
        let label = match &state {
            WebRtcState::Unknown => "unknown",
            WebRtcState::Disabled => "disabled",
            WebRtcState::Connecting => "connecting",
            WebRtcState::Connected => "connected",
            WebRtcState::Failed(_) => "failed",
        };
        clog!(
            "[drive-grant-actor][webrtc]",
            &self.correlation(),
            "state={}",
            label
        );
        let _ = self.webrtc_tx.send(state);
    }

    /// Coalesce-and-dial. Multiple concurrent callers resolve to a single
    /// underlying `transport.dial()` invocation. Successful dials are
    /// memoized — subsequent calls return immediately. Failures retry
    /// internally with backoff up to
    /// `DriveGrantActorConfig::max_consecutive_failures`, then surface
    /// `ActorError::DialFailed` to *all* current waiters; the next caller
    /// after that may try again.
    pub async fn ensure_ready(self: &Arc<Self>) -> Result<(), ActorError> {
        if self.state.read().await.shutdown_reason.is_some() {
            return Err(ActorError::Shutdown);
        }

        // Fast path: someone already dialed successfully and the actor
        // hasn't been told otherwise. The OnceCell barrier is set on first
        // success and only cleared by shutdown.
        if self.ready_once.initialized() {
            return Ok(());
        }

        // Slow path: race-free init. `OnceCell::get_or_try_init` runs the
        // closure once even under N concurrent waiters; losers await the
        // winner's result.
        let attempt_dial = || async {
            // Phase 5: wait for the auth-readiness gate before the first
            // dial. None ⇒ legacy behavior (no gate). The wait only
            // happens once because `OnceCell::get_or_try_init` runs the
            // closure once; subsequent successful calls short-circuit
            // via `ready_once.initialized()` above.
            if let Some(store) = &self.auth_readiness {
                match store
                    .wait_until_ready(
                        &self.auth_readiness_legs,
                        self.config.auth_readiness_wait_timeout,
                    )
                    .await
                {
                    Ok(()) => {}
                    Err(AuthReadinessWaitError::Timeout { remaining, elapsed }) => {
                        return Err(ActorError::AuthNotReady(format!(
                            "remaining={:?} elapsed={:?}",
                            remaining, elapsed
                        )));
                    }
                    Err(AuthReadinessWaitError::Closed) => {
                        return Err(ActorError::AuthNotReady("store-closed".into()));
                    }
                }
            }

            // Take the select-loop mutex so admission/replacement decisions
            // serialize with the dial. In Phase 4 the inbound-replacement
            // adoption hook also locks this; today it's used to make
            // concurrent ensure_ready calls cooperate cleanly.
            let _guard = self.select_loop.lock().await;
            self.dial_with_retries().await
        };
        match self.ready_once.get_or_try_init(attempt_dial).await {
            Ok(()) => Ok(()),
            Err(error) => Err(error),
        }
    }

    async fn dial_with_retries(&self) -> Result<(), ActorError> {
        let mut wait = self.config.initial_retry;
        loop {
            // Check shutdown inside the loop so a concurrent shutdown
            // unblocks an in-flight retry.
            if self.state.read().await.shutdown_reason.is_some() {
                return Err(ActorError::Shutdown);
            }

            match self.transport.dial(&self.key).await {
                Ok(()) => {
                    let mut state = self.state.write().await;
                    let was_first = !state.ever_succeeded;
                    state.consecutive_failures = 0;
                    state.ever_succeeded = true;
                    drop(state);
                    if was_first {
                        clog!(
                            "[drive-grant-actor][dial]",
                            &self.correlation(),
                            "ready_first"
                        );
                    } else {
                        clog!(
                            "[drive-grant-actor][dial]",
                            &self.correlation(),
                            "ready_recovered"
                        );
                    }
                    return Ok(());
                }
                Err(error) => {
                    let consecutive_failures = {
                        let mut state = self.state.write().await;
                        state.consecutive_failures = state.consecutive_failures.saturating_add(1);
                        state.consecutive_failures
                    };
                    if consecutive_failures >= self.config.max_consecutive_failures {
                        clog!(
                            "[drive-grant-actor][dial]",
                            &self.correlation(),
                            "failed_terminal attempts={} error={}",
                            consecutive_failures,
                            error
                        );
                        return Err(ActorError::DialFailed(error));
                    }
                    clog!(
                        "[drive-grant-actor][dial]",
                        &self.correlation(),
                        "failed_retry attempt={} error={}",
                        consecutive_failures,
                        error
                    );
                    tokio::time::sleep(wait).await;
                    wait = (wait * 2).min(self.config.max_retry);
                }
            }
        }
    }

    /// Phase 6: returns a `LogicalChannelHandle` for the requested label.
    ///
    /// The host-side `DriveViewBaseIrohHandler` dispatches drive-view frames
    /// arriving on a bi-stream with label `"drive-view"`, so logical channels
    /// are implicitly multiplexed by QUIC's native stream layer. This method
    /// validates the label and returns a handle; actual frame I/O happens at
    /// the transport level.
    pub async fn open_logical_channel(
        self: &Arc<Self>,
        label: &str,
    ) -> Result<LogicalChannelHandle, ActorError> {
        if self.state.read().await.shutdown_reason.is_some() {
            return Err(ActorError::Shutdown);
        }
        if !self.ready_once.initialized() {
            return Err(ActorError::NotReady);
        }
        // Known labels that the host-side dispatcher understands.
        match label {
            "drive-view.list" | "drive-view.fetch" | "drive-view.watch" | "drive-view.buckets"
            | "drive-view" => {
                let ctx = self.correlation().channel(label);
                clog!("[drive-grant-actor][channel]", &ctx, "open");
                Ok(LogicalChannelHandle {
                    label: label.to_string(),
                })
            }
            _ => {
                let ctx = self.correlation().channel(label);
                clog!(
                    "[drive-grant-actor][channel]",
                    &ctx,
                    "rejected_unknown_label"
                );
                Err(ActorError::LogicalChannelNotImplemented(label.to_string()))
            }
        }
    }

    /// Test-side helper: returns a placeholder handle for the label without
    /// invoking the unimplemented wire format. Used by tests + Phase 4
    /// observer wiring to assert that channel labels round-trip.
    #[doc(hidden)]
    pub async fn open_logical_channel_test_stub(
        self: &Arc<Self>,
        label: &str,
    ) -> Result<LogicalChannelHandle, ActorError> {
        if self.state.read().await.shutdown_reason.is_some() {
            return Err(ActorError::Shutdown);
        }
        if !self.ready_once.initialized() {
            return Err(ActorError::NotReady);
        }
        Ok(LogicalChannelHandle {
            label: label.to_string(),
        })
    }

    /// Cancel any pending retries, mark the actor shut down, and tear down
    /// transport state via `transport.close()`. Idempotent: subsequent
    /// `ensure_ready()` returns `Shutdown`.
    pub async fn shutdown(self: &Arc<Self>) {
        {
            let mut state = self.state.write().await;
            if state.shutdown_reason.is_some() {
                return;
            }
            state.shutdown_reason = Some("explicit-shutdown".to_string());
        }
        clog!(
            "[drive-grant-actor][shutdown]",
            &self.correlation(),
            "begin"
        );
        let _ = self.webrtc_tx.send(WebRtcState::Failed(
            "drive-grant-actor: shutdown".to_string(),
        ));
        self.transport.close(&self.key).await;
        clog!(
            "[drive-grant-actor][shutdown]",
            &self.correlation(),
            "complete"
        );
    }

    /// Snapshot of internal counters useful for tests + future correlation
    /// logging in Phase 7. Not stable API.
    #[doc(hidden)]
    pub async fn debug_state(&self) -> DriveGrantActorDebugState {
        let state = self.state.read().await;
        DriveGrantActorDebugState {
            consecutive_failures: state.consecutive_failures,
            ever_succeeded: state.ever_succeeded,
            ready: self.ready_once.initialized(),
            shutdown: state.shutdown_reason.is_some(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
#[doc(hidden)]
pub struct DriveGrantActorDebugState {
    pub consecutive_failures: u32,
    pub ever_succeeded: bool,
    pub ready: bool,
    pub shutdown: bool,
}

/// Registry that hands out one shared `Arc<DriveGrantConnectionActor>` per
/// `DriveGrantConnectionKey`. `get_or_spawn` is idempotent: many requesters
/// for the same key get the same actor.
pub struct DriveGrantActorRegistry {
    actors: RwLock<HashMap<DriveGrantConnectionKey, Arc<DriveGrantConnectionActor>>>,
    transport: Arc<dyn DriveGrantTransport>,
    config: DriveGrantActorConfig,
    /// Phase 5: optional auth-readiness gate. When `Some`, every actor
    /// spawned by this registry inherits it.
    auth_readiness: Option<Arc<AuthReadinessStore>>,
    auth_readiness_legs: Vec<AuthLeg>,
}

impl DriveGrantActorRegistry {
    pub fn new(transport: Arc<dyn DriveGrantTransport>) -> Arc<Self> {
        Arc::new(Self {
            actors: RwLock::new(HashMap::new()),
            transport,
            config: DriveGrantActorConfig::default(),
            auth_readiness: None,
            auth_readiness_legs: vec![
                AuthLeg::FilesAuth,
                AuthLeg::PlutoRtcAuth,
                AuthLeg::RuntimeAuth,
                AuthLeg::Firestore,
            ],
        })
    }

    pub fn with_config(
        transport: Arc<dyn DriveGrantTransport>,
        config: DriveGrantActorConfig,
    ) -> Arc<Self> {
        Arc::new(Self {
            actors: RwLock::new(HashMap::new()),
            transport,
            config,
            auth_readiness: None,
            auth_readiness_legs: vec![
                AuthLeg::FilesAuth,
                AuthLeg::PlutoRtcAuth,
                AuthLeg::RuntimeAuth,
                AuthLeg::Firestore,
            ],
        })
    }

    /// Phase 5: attach an auth-readiness gate. Every actor spawned by
    /// this registry awaits `store.wait_until_ready(legs, _)` before
    /// its first dial. Pass `legs.is_empty()` (or omit `legs` via the
    /// helper below) to default to all four legs.
    pub fn with_auth_readiness(
        transport: Arc<dyn DriveGrantTransport>,
        config: DriveGrantActorConfig,
        auth_readiness: Arc<AuthReadinessStore>,
        legs: Vec<AuthLeg>,
    ) -> Arc<Self> {
        Arc::new(Self {
            actors: RwLock::new(HashMap::new()),
            transport,
            config,
            auth_readiness: Some(auth_readiness),
            auth_readiness_legs: if legs.is_empty() {
                vec![
                    AuthLeg::FilesAuth,
                    AuthLeg::PlutoRtcAuth,
                    AuthLeg::RuntimeAuth,
                    AuthLeg::Firestore,
                ]
            } else {
                legs
            },
        })
    }

    /// Return an existing actor for `key` or spawn a fresh one. Concurrent
    /// callers race only on the registry's RwLock; the actor itself is
    /// internally serialized.
    pub async fn get_or_spawn(
        &self,
        key: DriveGrantConnectionKey,
    ) -> Arc<DriveGrantConnectionActor> {
        // Fast path: read-lock peek.
        if let Some(actor) = self.actors.read().await.get(&key).cloned() {
            return actor;
        }
        // Slow path: write-lock to insert. Re-check under write lock so
        // racing callers don't double-spawn.
        let mut actors = self.actors.write().await;
        if let Some(actor) = actors.get(&key).cloned() {
            return actor;
        }
        let actor = DriveGrantConnectionActor::new(
            key.clone(),
            self.transport.clone(),
            self.config.clone(),
            self.auth_readiness.clone(),
            self.auth_readiness_legs.clone(),
        );
        actors.insert(key, actor.clone());
        actor
    }

    /// Look up an existing actor without spawning. Returns `None` if no
    /// requester has yet asked for this key.
    pub async fn get(
        &self,
        key: &DriveGrantConnectionKey,
    ) -> Option<Arc<DriveGrantConnectionActor>> {
        self.actors.read().await.get(key).cloned()
    }

    /// Shut down and remove all actors. Useful for tests + Client
    /// teardown.
    pub async fn shutdown_all(&self) {
        let actors: Vec<_> = {
            let mut actors = self.actors.write().await;
            actors.drain().map(|(_, actor)| actor).collect()
        };
        for actor in actors {
            actor.shutdown().await;
        }
    }

    /// Number of live actors. Stable for tests; not part of the public
    /// production API.
    #[doc(hidden)]
    pub async fn len(&self) -> usize {
        self.actors.read().await.len()
    }
}