mxaccessgw/clients/rust/src/client.rs

//! High-level wrapper around the generated `MxAccessGateway` gRPC client.
//!
//! [`GatewayClient::connect`] builds an authenticated `tonic` channel using
//! the supplied [`ClientOptions`], applies the bearer-token interceptor, and
//! exposes typed methods for the unary and streaming RPCs. Most application
//! code should prefer [`GatewayClient::open_session`] and the [`Session`]
//! handle it returns, rather than the `*_raw` methods.

use tonic::codegen::InterceptedService;
use tonic::transport::Channel;
use tonic::Request;

use crate::auth::AuthInterceptor;
use crate::error::{
    ensure_command_success, ensure_mxaccess_success, ensure_protocol_success, Error,
};
use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGatewayClient;
use crate::generated::mxaccess_gateway::v1::{
    AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AlarmFeedMessage,
    CloseSessionReply, CloseSessionRequest, MxCommandReply, MxCommandRequest, MxEvent,
    OpenSessionReply, OpenSessionRequest, QueryActiveAlarmsRequest, ReplayGap, StreamAlarmsRequest,
    StreamEventsRequest,
};
use crate::options::{build_tls_config, ClientOptions};
use crate::session::Session;

/// Generated gateway client wrapped in the auth interceptor that
/// [`GatewayClient`] uses internally.
pub type RawGatewayClient = MxAccessGatewayClient<InterceptedService<Channel, AuthInterceptor>>;

/// One item yielded by the per-session event stream returned by
/// [`GatewayClient::stream_events`].
///
/// Almost every item is an ordinary MXAccess event ([`EventItem::Event`]).
/// The one exception is the reconnect-replay gap sentinel
/// ([`EventItem::ReplayGap`]): the gateway emits it at most once, at the head
/// of a stream that was *resumed* via
/// [`Session::events_after`](crate::session::Session::events_after)
/// (`StreamEventsRequest.after_worker_sequence`) when the requested sequence is
/// older than the oldest event still retained in the session replay ring — i.e.
/// events were evicted and cannot be replayed.
///
/// The client does **not** synthesize this signal: it faithfully forwards the
/// gateway's sentinel `MxEvent` (whose `replay_gap` field is set), and only
/// makes it a distinct, typed variant so consumers can `match` on it instead of
/// inspecting a field on a value that otherwise looks like a normal event.
///
/// # Reacting to a gap
///
/// A [`EventItem::ReplayGap`] means "you missed events — discard any local
/// state and re-snapshot." The events in the open interval
/// `(requested_after_sequence, oldest_available_sequence)` are gone. To resume
/// the stream without provoking another gap, reconnect with
/// [`Session::events_after`](crate::session::Session::events_after) passing
/// `oldest_available_sequence - 1`, which replays starting at the first still
/// retained event (`oldest_available_sequence`):
///
/// ```no_run
/// # use zb_mom_ww_mxgateway_client::{EventItem, Session};
/// # use futures_util::StreamExt;
/// # async fn run(session: Session, cursor: u64) -> Result<(), zb_mom_ww_mxgateway_client::Error> {
/// let mut stream = session.events_after(cursor).await?;
/// while let Some(item) = stream.next().await {
///     match item? {
///         EventItem::Event(event) => {
///             let _ = event; // apply the change
///         }
///         EventItem::ReplayGap(gap) => {
///             // Local state is stale — re-snapshot, then resume without a gap.
///             let resume_cursor = gap.oldest_available_sequence.saturating_sub(1);
///             stream = session.events_after(resume_cursor).await?;
///         }
///     }
/// }
/// # Ok(())
/// # }
/// ```
// The `Event` variant is the hot path (nearly every stream item) and is the
// large one; the rare `ReplayGap` sentinel is small. Boxing `Event` to equalize
// the variants would add a heap allocation to every streamed event — a
// regression versus the prior `Result<MxEvent, Error>` surface, which already
// moved `MxEvent` by value. Keep the common path allocation-free.
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, PartialEq)]
pub enum EventItem {
    /// A normal MXAccess event forwarded from the worker.
    Event(MxEvent),
    /// The reconnect-replay gap sentinel — recent event history was evicted
    /// before this resumed stream could replay it. See [`EventItem`] for how
    /// to react.
    ReplayGap(ReplayGap),
}

impl EventItem {
    /// Classify an incoming `MxEvent` into the typed stream item.
    ///
    /// A present `replay_gap` promotes the event to [`EventItem::ReplayGap`];
    /// otherwise it is an ordinary [`EventItem::Event`]. The sentinel is never
    /// dropped and never surfaced as a normal event.
    fn from_event(mut event: MxEvent) -> Self {
        match event.replay_gap.take() {
            Some(gap) => EventItem::ReplayGap(gap),
            None => EventItem::Event(event),
        }
    }

    /// Borrow the inner [`MxEvent`] when this item is a normal event, or
    /// `None` when it is the [`EventItem::ReplayGap`] sentinel.
    #[must_use]
    pub fn as_event(&self) -> Option<&MxEvent> {
        match self {
            EventItem::Event(event) => Some(event),
            EventItem::ReplayGap(_) => None,
        }
    }

    /// Borrow the [`ReplayGap`] when this item is the reconnect-replay gap
    /// sentinel, or `None` for a normal event.
    #[must_use]
    pub fn replay_gap(&self) -> Option<&ReplayGap> {
        match self {
            EventItem::ReplayGap(gap) => Some(gap),
            EventItem::Event(_) => None,
        }
    }

    /// Consume the item and return the inner [`MxEvent`] when it is a normal
    /// event, or `None` for the [`EventItem::ReplayGap`] sentinel.
    #[must_use]
    pub fn into_event(self) -> Option<MxEvent> {
        match self {
            EventItem::Event(event) => Some(event),
            EventItem::ReplayGap(_) => None,
        }
    }
}

/// Pinned, boxed [`EventItem`] stream returned by
/// [`GatewayClient::stream_events`]. Each item is either a normal
/// [`EventItem::Event`] or the [`EventItem::ReplayGap`] reconnect-replay
/// sentinel. Errors are pre-mapped from `tonic::Status` to [`Error`]; dropping
/// the stream cancels the call.
pub type EventStream =
    std::pin::Pin<Box<dyn futures_core::Stream<Item = Result<EventItem, Error>> + Send + 'static>>;

/// Pinned, boxed [`ActiveAlarmSnapshot`] stream returned by
/// [`GatewayClient::query_active_alarms`]. Errors are pre-mapped from
/// `tonic::Status` to [`Error`]; dropping the stream cancels the call.
pub type ActiveAlarmStream = std::pin::Pin<
    Box<dyn futures_core::Stream<Item = Result<ActiveAlarmSnapshot, Error>> + Send + 'static>,
>;

/// Pinned, boxed [`AlarmFeedMessage`] stream returned by
/// [`GatewayClient::stream_alarms`]. Errors are pre-mapped from
/// `tonic::Status` to [`Error`]; dropping the stream cancels the call.
pub type AlarmFeedStream = std::pin::Pin<
    Box<dyn futures_core::Stream<Item = Result<AlarmFeedMessage, Error>> + Send + 'static>,
>;

/// Thin async wrapper around the generated gateway client.
///
/// The wrapper is `Clone`: every clone shares the underlying tonic channel
/// (cheap, reference-counted) and the same call/stream timeouts. It is
/// designed to be cheap enough to clone per request handler.
#[derive(Clone)]
pub struct GatewayClient {
    inner: RawGatewayClient,
    call_timeout: std::time::Duration,
    stream_timeout: Option<std::time::Duration>,
}

impl GatewayClient {
    /// Connect to the gateway endpoint described by `options` and return a
    /// ready-to-use client.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InvalidEndpoint`] if the endpoint URL or CA file is
    /// malformed, and [`Error::Transport`] if the TCP/TLS handshake fails.
    pub async fn connect(options: ClientOptions) -> Result<Self, Error> {
        let mut endpoint =
            Channel::from_shared(options.endpoint().to_owned()).map_err(|source| {
                Error::InvalidEndpoint {
                    endpoint: options.endpoint().to_owned(),
                    detail: source.to_string(),
                }
            })?;
        endpoint = endpoint.connect_timeout(options.connect_timeout());

        if let Some(tls) = build_tls_config(&options)? {
            endpoint = endpoint.tls_config(tls)?;
        }

        let channel = endpoint.connect().await?;
        let interceptor = AuthInterceptor::new(options.api_key().cloned());
        let max_grpc_message_bytes = options.max_grpc_message_bytes();

        Ok(Self {
            inner: MxAccessGatewayClient::with_interceptor(channel, interceptor)
                .max_decoding_message_size(max_grpc_message_bytes)
                .max_encoding_message_size(max_grpc_message_bytes),
            call_timeout: options.call_timeout(),
            stream_timeout: options.stream_timeout(),
        })
    }

    /// Borrow the underlying generated client. Use this only when you need
    /// access to RPCs not surfaced by the wrapper.
    pub fn raw_client(&mut self) -> &mut RawGatewayClient {
        &mut self.inner
    }

    /// Consume the wrapper and return the underlying generated client.
    pub fn into_inner(self) -> RawGatewayClient {
        self.inner
    }

    /// Build a [`Session`] handle from a previously opened session id. No
    /// RPC is performed — this is the cheap counterpart to
    /// [`GatewayClient::open_session`] for callers that already own the id.
    pub fn session(&self, session_id: impl Into<String>) -> Session {
        Session::new(session_id, self.clone())
    }

    /// Issue an `OpenSession` RPC and return the raw reply without
    /// validating its `protocol_status`.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`].
    pub async fn open_session_raw(
        &self,
        request: OpenSessionRequest,
    ) -> Result<OpenSessionReply, Error> {
        let mut client = self.inner.clone();
        let response = client.open_session(self.unary_request(request)).await?;
        Ok(response.into_inner())
    }

    /// Open a session, validate its `protocol_status`, and return a typed
    /// [`Session`] handle bound to this client.
    ///
    /// # Errors
    ///
    /// Returns [`Error::ProtocolStatus`] if the gateway accepts the call
    /// but reports a non-OK protocol status, plus any of the
    /// [`Error`] variants produced by [`open_session_raw`](Self::open_session_raw).
    pub async fn open_session(&self, request: OpenSessionRequest) -> Result<Session, Error> {
        let reply = self.open_session_raw(request).await?;
        ensure_protocol_success("open session", reply.protocol_status.as_ref())?;
        Ok(Session::new(reply.session_id, self.clone()))
    }

    /// Issue a `CloseSession` RPC and return the raw reply.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`].
    pub async fn close_session_raw(
        &self,
        request: CloseSessionRequest,
    ) -> Result<CloseSessionReply, Error> {
        let mut client = self.inner.clone();
        let response = client.close_session(self.unary_request(request)).await?;
        Ok(response.into_inner())
    }

    /// Issue an `Invoke` RPC and return the raw reply, even when the
    /// command-level protocol status is non-OK.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`].
    pub async fn invoke_raw(&self, request: MxCommandRequest) -> Result<MxCommandReply, Error> {
        let mut client = self.inner.clone();
        let response = client.invoke(self.unary_request(request)).await?;
        Ok(response.into_inner())
    }

    /// Issue an `Invoke` RPC and surface a failing reply as a typed error.
    ///
    /// The reply is validated twice: [`ensure_command_success`] rejects a
    /// non-OK protocol envelope as [`Error::Command`], then
    /// [`ensure_mxaccess_success`] rejects an MXAccess-level failure (negative
    /// `hresult` or a non-success `MXSTATUS_PROXY` entry) as
    /// [`Error::MxAccess`], preserving MXAccess parity. Callers that need the
    /// unvalidated reply should use [`invoke_raw`](Self::invoke_raw) instead.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Command`] when the reply's `protocol_status` is not
    /// `Ok`, [`Error::MxAccess`] when the reply reports an MXAccess-level
    /// failure, plus any errors propagated by
    /// [`invoke_raw`](Self::invoke_raw).
    pub async fn invoke(&self, request: MxCommandRequest) -> Result<MxCommandReply, Error> {
        let reply = ensure_command_success(self.invoke_raw(request).await?)?;
        ensure_mxaccess_success(reply)
    }

    /// Open the server-streaming `StreamEvents` RPC.
    ///
    /// The returned [`EventStream`] yields [`EventItem`] values as the worker
    /// produces them: ordinary MXAccess events as [`EventItem::Event`], and the
    /// gateway's reconnect-replay gap sentinel — set only on resumed streams
    /// whose requested sequence predates the retained replay history — as
    /// [`EventItem::ReplayGap`]. Dropping the stream cancels the gRPC call
    /// cooperatively.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`] if the
    /// server rejects the subscription.
    pub async fn stream_events(&self, request: StreamEventsRequest) -> Result<EventStream, Error> {
        let mut client = self.inner.clone();
        let response = client.stream_events(self.stream_request(request)).await?;
        let stream = futures_util::StreamExt::map(response.into_inner(), |result| {
            result.map(EventItem::from_event).map_err(Error::from)
        });

        Ok(Box::pin(stream))
    }

    /// Acknowledge an active MXAccess alarm condition through the gateway.
    ///
    /// The gateway authenticates the request against the API key's
    /// `invoke:alarm-ack` scope and forwards the acknowledge to the worker's
    /// MXAccess session; the resulting native MxStatus is returned in the
    /// reply. Acks are idempotent at the MxAccess layer.
    ///
    /// # Errors
    ///
    /// Returns [`Error::ProtocolStatus`] when the gateway accepts the call but
    /// reports a non-OK protocol status, plus any of the [`Error`] variants
    /// produced by transport failures.
    pub async fn acknowledge_alarm(
        &self,
        request: AcknowledgeAlarmRequest,
    ) -> Result<AcknowledgeAlarmReply, Error> {
        let mut client = self.inner.clone();
        let response = client
            .acknowledge_alarm(self.unary_request(request))
            .await?;
        let reply = response.into_inner();
        ensure_protocol_success("acknowledge alarm", reply.protocol_status.as_ref())?;
        Ok(reply)
    }

    /// Open the server-streaming `QueryActiveAlarms` RPC — the gateway's
    /// ConditionRefresh equivalent.
    ///
    /// The returned [`ActiveAlarmStream`] yields one [`ActiveAlarmSnapshot`]
    /// per currently-active alarm. Dropping the stream cancels the gRPC call
    /// cooperatively. Optional alarm-reference prefix scoping
    /// (`request.alarm_filter_prefix`) limits the stream to a sub-tree.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`] if the
    /// server rejects the request.
    pub async fn query_active_alarms(
        &self,
        request: QueryActiveAlarmsRequest,
    ) -> Result<ActiveAlarmStream, Error> {
        let mut client = self.inner.clone();
        let response = client
            .query_active_alarms(self.stream_request(request))
            .await?;
        let stream = futures_util::StreamExt::map(response.into_inner(), |result| {
            result.map_err(Error::from)
        });

        Ok(Box::pin(stream))
    }

    /// Attach to the gateway's central `StreamAlarms` feed.
    ///
    /// The returned [`AlarmFeedStream`] opens with one [`AlarmFeedMessage`]
    /// per currently-active alarm (the ConditionRefresh snapshot), then a
    /// single `snapshot_complete`, then a `transition` for every subsequent
    /// raise / acknowledge / clear. It is served by the gateway's always-on
    /// alarm monitor — no worker session is opened — so any number of clients
    /// may attach. Dropping the stream cancels the gRPC call cooperatively.
    /// Optional alarm-reference prefix scoping (`request.alarm_filter_prefix`)
    /// limits the stream to a sub-tree.
    ///
    /// # Errors
    ///
    /// Returns the `tonic::Status` mapped through [`Error::from`] if the
    /// server rejects the request.
    pub async fn stream_alarms(
        &self,
        request: StreamAlarmsRequest,
    ) -> Result<AlarmFeedStream, Error> {
        let mut client = self.inner.clone();
        let response = client.stream_alarms(self.stream_request(request)).await?;
        let stream = futures_util::StreamExt::map(response.into_inner(), |result| {
            result.map_err(Error::from)
        });

        Ok(Box::pin(stream))
    }

    fn unary_request<T>(&self, message: T) -> Request<T> {
        let mut request = Request::new(message);
        request.set_timeout(self.call_timeout);
        request
    }

    fn stream_request<T>(&self, message: T) -> Request<T> {
        let mut request = Request::new(message);
        if let Some(timeout) = self.stream_timeout {
            request.set_timeout(timeout);
        }

        request
    }
}