liminal_sdk/remote/

handles.rs

use alloc::string::{String, ToString};
use alloc::sync::Arc;
use alloc::vec::Vec;

use serde::Serialize;
use serde::de::DeserializeOwned;
use spin::Mutex;

#[cfg(feature = "std")]
use crate::DeliveryAck;
use crate::connection::{
    ConnectionEvent, ConnectionLifecycle, ConnectionPool, ConnectionState, ResumeRequest,
    SubscriptionId,
};
use crate::embedded::{
    EmbeddedChannelHandle, EmbeddedConversationHandle, EmptyLifecycleStream, ReadyResult,
    SdkSubscription,
};
use crate::{
    ChannelHandle, ConversationHandle, ConversationId, PressureResponse, SchemaValidate, SdkError,
};

pub use super::participant::RemoteParticipantHandle;

use super::config::SdkConfig;
use super::protocol::{
    RemoteTransport, WireConversationRequest, WirePublishRequest, WireResumeRequest,
    deserialize_payload,
};
use super::{RemoteConfig, ServerAddress};
mod negotiated;

/// The refusal typed `subscribe` returns instead of an instantly-dry stream.
///
/// No ledger row records this gap: `WIRING-LEDGER.md`'s W6 is the BROWSER
/// conversation surface (a TypeScript transport lane), not the Rust typed
/// subscription, and nothing else in `docs/` names it. The refusal says so
/// rather than borrowing a row that does not cover it.
fn unwired_typed_subscribe() -> SdkError {
    SdkError::Unwired {
        surface: "typed subscribe delivery (ChannelHandle::subscribe on a remote handle)"
            .to_string(),
        seam: "no ledger row records this gap; W6 is the browser conversation surface, not this"
            .to_string(),
        alternative: "for real delivery use SubscriptionStream::open plus \
                      SubscriptionStream::recv_timeout, or \
                      WebSocketSubscriptionStream::open on the WebSocket transport"
            .to_string(),
    }
}

#[derive(Debug)]
struct RemoteChannelState {
    lifecycle: ConnectionLifecycle,
    pool: ConnectionPool,
    next_subscription: u64,
}

/// Channel handle that communicates through SDK-internal wire protocol transport.
///
/// The handle inherits whatever transport its [`RemoteConfig`] carries. A config
/// straight out of `RemoteConfig::new` has never been connected, so every
/// operation on a handle built from it refuses with
/// [`SdkError::NotConnected`] rather than reporting a success that reached no
/// wire. Connect the config first.
#[derive(Clone, Debug)]
pub struct RemoteChannelHandle {
    server_address: ServerAddress,
    channel_name: String,
    state: Arc<Mutex<RemoteChannelState>>,
    transport: Arc<dyn RemoteTransport>,
}

impl RemoteChannelHandle {
    /// Creates a remote channel handle from validated configuration.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the connection pool cannot be created.
    pub fn new(config: &RemoteConfig) -> Result<Self, SdkError> {
        Ok(Self {
            server_address: config.server_address.clone(),
            channel_name: config.channel_name.clone(),
            state: Arc::new(Mutex::new(RemoteChannelState {
                lifecycle: ConnectionLifecycle::new(),
                pool: ConnectionPool::new(config.pool_config)?,
                next_subscription: 0,
            })),
            transport: Arc::clone(&config.transport),
        })
    }

    /// Returns current lifecycle state from the SDK-003 state machine.
    #[must_use]
    pub fn connection_state(&self) -> ConnectionState {
        self.state.lock().lifecycle.state().clone()
    }

    /// Records that an externally authorized real reconnect attempt started.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] when the lifecycle rejects the transition.
    pub fn reconnect_started(&self) -> Result<(), SdkError> {
        self.state.lock().lifecycle.reconnect_started()
    }

    /// Marks the remote channel connected and builds subscription resume requests.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if lifecycle transition or recovery state is invalid.
    pub fn connected(&self) -> Result<Vec<ResumeRequest>, SdkError> {
        // Compute the resume requests under the lock, then release it before any
        // transport I/O so concurrent callers do not spin-wait during resume calls.
        let requests = {
            let mut state = self.state.lock();
            let previous = state.lifecycle.state().clone();
            state.lifecycle.connected()?;
            let event = ConnectionEvent::new(previous, state.lifecycle.state().clone());
            state.pool.resume_requests_for_transition(&event)?
        };
        for request in &requests {
            let wire_request = WireResumeRequest::new(*request);
            self.transport.resume(&self.server_address, &wire_request)?;
        }
        Ok(requests)
    }

    /// Marks a subscription active and assigns it to the configured pool.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if a subscription id overflows or pool assignment fails.
    pub fn track_subscription(&self) -> Result<SubscriptionId, SdkError> {
        let mut state = self.state.lock();
        let id = SubscriptionId::new(state.next_subscription);
        state.next_subscription =
            state
                .next_subscription
                .checked_add(1)
                .ok_or_else(|| SdkError::Store {
                    description: "subscription id overflow".to_string(),
                })?;
        state.pool.assign_subscription(id)?;
        Ok(id)
    }

    /// Records an acknowledged sequence for subscription recovery.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the subscription is not active in the pool.
    pub fn acknowledge(
        &self,
        subscription_id: SubscriptionId,
        sequence: u64,
    ) -> Result<(), SdkError> {
        let mut state = self.state.lock();
        state.pool.acknowledge(subscription_id, sequence)
    }

    /// Returns remote server address used by this handle.
    #[must_use]
    pub const fn server_address(&self) -> &ServerAddress {
        &self.server_address
    }
}

impl ChannelHandle for RemoteChannelHandle {
    type Subscription<M>
        = SdkSubscription<M>
    where
        M: DeserializeOwned;

    type ReplyFuture<'a, Resp>
        = ReadyResult<Resp>
    where
        Self: 'a,
        Resp: DeserializeOwned + 'a;

    fn publish<M>(&self, message: M) -> Result<PressureResponse, SdkError>
    where
        M: Serialize + SchemaValidate,
    {
        let request = WirePublishRequest::new(&self.channel_name, &message)?;
        self.transport.publish(&self.server_address, &request)
    }

    /// REFUSES. Typed subscription delivery is not implemented on this handle.
    ///
    /// The returned stream yields exactly one item — `Err(SdkError::Unwired)` —
    /// and then ends. It does not touch the transport, so no `Subscribe` is left
    /// outstanding on the server for a stream that could never deliver.
    ///
    /// What this replaces was worse than unimplemented. The old body performed
    /// the real wire `Subscribe`, registered a subscription in the pool, and
    /// then returned `SdkSubscription::empty()` — a stream whose `poll_next` is
    /// `Ready(None)` immediately and forever. A caller's
    /// `while let Some(m) = sub.next().await` exited on the first poll while the
    /// server pumped `Deliver` frames at a connection nothing was reading.
    ///
    /// For real delivery use the lower-level surface that works today:
    /// `SubscriptionStream::open` plus `SubscriptionStream::recv_timeout` (or
    /// `WebSocketSubscriptionStream::open` on the WebSocket transport).
    fn subscribe<M>(&self) -> Self::Subscription<M>
    where
        M: DeserializeOwned,
    {
        SdkSubscription::error(unwired_typed_subscribe())
    }

    fn request_reply<Req, Resp>(&self, request: Req) -> ReadyResult<Resp>
    where
        Req: Serialize + SchemaValidate,
        Resp: DeserializeOwned,
    {
        ReadyResult::new(self.request_reply_blocking(&request))
    }
}

impl RemoteChannelHandle {
    /// Publishes a message with an idempotency key and returns a genuine delivery
    /// ack.
    ///
    /// The idempotency key drives dedup-on-delivery on the server: a re-publish of
    /// the same key is delivered to subscribers at most once. The returned
    /// [`DeliveryAck`] reports whether this publish was genuinely accepted by a
    /// subscriber ([`DeliveryAck::is_accepted`]), which a caller such as the aion
    /// outbox uses to treat the send as done only on real acceptance. This is
    /// distinct from the backpressure-only [`publish`](ChannelHandle::publish).
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] when the message cannot be serialized, the round trip
    /// fails, or the transport cannot produce a genuine delivery ack.
    ///
    /// Only with std: without it no transport could carry the key, so no keyed
    /// publish is offered rather than one that drops the key (Hermes 7 Oct 08:32).
    #[cfg(feature = "std")]
    pub fn publish_with_idempotency_key<M>(
        &self,
        message: &M,
        idempotency_key: &str,
    ) -> Result<DeliveryAck, SdkError>
    where
        M: Serialize + SchemaValidate,
    {
        let request =
            WirePublishRequest::with_idempotency_key(&self.channel_name, message, idempotency_key)?;
        self.transport
            .publish_with_delivery(&self.server_address, &request)
    }

    /// Publishes `message` as the reply to `request` and returns a genuine
    /// delivery ack (Fable 4c).
    ///
    /// `request` is the id a delivered message carried
    /// ([`DeliveredMessage::message`](crate::DeliveredMessage::message)): the
    /// reply names it as its causal parent, `<boot>:<id>`, so the server's tap
    /// can pair the two and measure the dwell between them. A request that
    /// carried no id (a server before protocol 1.2, or a durable message
    /// replayed from the store) gives a reply with no parent, never an
    /// invented one.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] when the message cannot be serialized, the round trip
    /// fails, or the transport cannot produce a genuine delivery ack.
    #[cfg(feature = "std")]
    pub fn reply_to<M>(
        &self,
        request: Option<liminal::protocol::DeliveredId>,
        message: &M,
    ) -> Result<DeliveryAck, SdkError>
    where
        M: Serialize + SchemaValidate,
    {
        let reply = WirePublishRequest::reply(&self.channel_name, message, request)?;
        self.transport
            .publish_with_delivery(&self.server_address, &reply)
    }

    /// Performs a correlated request-reply round trip over the transport and
    /// deserializes the reply.
    ///
    /// The round trip rides the conversation request-reply path: the channel name
    /// is used as the conversation correlation id and subject, so the reply the
    /// server returns is matched back to this request by `conversation_id` on the
    /// single synchronous connection. Schema validation still runs on the request
    /// so a request-reply enforces the same typing contract as `publish`.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] when the request cannot be serialized, the round trip
    /// fails, or the reply cannot be deserialized into `Resp`.
    fn request_reply_blocking<Req, Resp>(&self, request: &Req) -> Result<Resp, SdkError>
    where
        Req: Serialize + SchemaValidate,
        Resp: DeserializeOwned,
    {
        let conversation_id = ConversationId::new(self.channel_name.clone());
        let wire_request = WireConversationRequest::new(&conversation_id, request)?;
        let reply = self
            .transport
            .request_reply_conversation(&self.server_address, &wire_request)?;
        deserialize_payload(&reply)
    }
}

/// Conversation handle that communicates through SDK-internal wire protocol transport.
///
/// As with [`RemoteChannelHandle`], a handle built from a config that was never
/// connected refuses every send, request, and receive with
/// [`SdkError::NotConnected`] instead of reporting a success that reached no
/// wire.
#[derive(Clone, Debug)]
pub struct RemoteConversationHandle {
    server_address: ServerAddress,
    conversation_id: ConversationId,
    lifecycle: Arc<Mutex<ConnectionLifecycle>>,
    transport: Arc<dyn RemoteTransport>,
    /// Buffered reply bytes from the most recent [`request`](Self::request) round
    /// trip, drained by the next [`receive`](ConversationHandle::receive). The
    /// synchronous transport completes the round trip inside `request`, so the
    /// correlated reply is held here until the caller deserializes it.
    pending_reply: Arc<Mutex<Option<Vec<u8>>>>,
}

impl RemoteConversationHandle {
    /// Creates a remote conversation handle from validated configuration.
    #[must_use]
    pub fn new(config: &RemoteConfig) -> Self {
        Self {
            server_address: config.server_address.clone(),
            conversation_id: config.conversation_id.clone(),
            lifecycle: Arc::new(Mutex::new(ConnectionLifecycle::new())),
            transport: Arc::clone(&config.transport),
            pending_reply: Arc::new(Mutex::new(None)),
        }
    }

    /// Sends a typed request on this conversation and blocks for its correlated
    /// reply, buffering the reply for the next [`receive`](ConversationHandle::receive).
    ///
    /// This is the request leg of the conversation request-reply pattern. It is
    /// kept distinct from [`send`](ConversationHandle::send), which stays
    /// fire-and-forget (the server is silent on success): only `request` sets the
    /// reply-requested flag and waits for the server's correlated answer. The aion
    /// dispatch model (`send` request, then `receive` reply) maps onto a `request`
    /// followed by `receive`.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] when the request cannot be serialized or the round
    /// trip fails.
    pub fn request<Req>(&self, request: Req) -> Result<(), SdkError>
    where
        Req: Serialize,
    {
        let wire_request = WireConversationRequest::new(&self.conversation_id, &request)?;
        let reply = self
            .transport
            .request_reply_conversation(&self.server_address, &wire_request)?;
        *self.pending_reply.lock() = Some(reply);
        Ok(())
    }

    /// Returns current lifecycle state from the SDK-003 state machine.
    #[must_use]
    pub fn connection_state(&self) -> ConnectionState {
        self.lifecycle.lock().state().clone()
    }

    /// Returns remote server address used by this handle.
    #[must_use]
    pub const fn server_address(&self) -> &ServerAddress {
        &self.server_address
    }
}

impl ConversationHandle for RemoteConversationHandle {
    type ReceiveFuture<'a, M>
        = ReadyResult<M>
    where
        Self: 'a,
        M: DeserializeOwned + 'a;

    type LifecycleStream = EmptyLifecycleStream;

    fn send<M>(&self, message: M) -> Result<(), SdkError>
    where
        M: Serialize,
    {
        let request = WireConversationRequest::new(&self.conversation_id, &message)?;
        self.transport
            .send_conversation(&self.server_address, &request)
    }

    fn receive<M>(&self) -> ReadyResult<M>
    where
        M: DeserializeOwned,
    {
        ReadyResult::new(self.receive_blocking())
    }

    fn lifecycle(&self) -> Self::LifecycleStream {
        EmptyLifecycleStream
    }
}

impl RemoteConversationHandle {
    /// Drains and deserializes the correlated reply buffered by the most recent
    /// [`request`](Self::request).
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Conversation`] when no reply is pending (no `request`
    /// has completed since the last `receive`), or [`SdkError::Serialization`]
    /// when the buffered reply cannot be deserialized into `M`.
    fn receive_blocking<M>(&self) -> Result<M, SdkError>
    where
        M: DeserializeOwned,
    {
        let payload = self
            .pending_reply
            .lock()
            .take()
            .ok_or_else(|| SdkError::Conversation {
                conversation_id: self.conversation_id.as_str().to_string(),
                description: "no correlated reply is pending; call request before receive"
                    .to_string(),
            })?;
        deserialize_payload(&payload)
    }
}

/// Runtime-selected channel handle that keeps deployment differences behind configuration.
#[derive(Clone, Debug)]
pub enum SdkChannelHandle {
    /// Embedded direct in-process handle.
    Embedded(EmbeddedChannelHandle),
    /// Remote protocol-backed handle.
    Remote(RemoteChannelHandle),
}

impl SdkChannelHandle {
    /// Creates a channel handle selected by SDK configuration.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the selected handle cannot be initialized.
    pub fn new(config: &SdkConfig) -> Result<Self, SdkError> {
        match config {
            SdkConfig::Embedded(config) => Ok(Self::Embedded(EmbeddedChannelHandle::new(config))),
            SdkConfig::Remote(config) => Ok(Self::Remote(RemoteChannelHandle::new(config)?)),
        }
    }
}

impl ChannelHandle for SdkChannelHandle {
    type Subscription<M>
        = SdkSubscription<M>
    where
        M: DeserializeOwned;

    type ReplyFuture<'a, Resp>
        = ReadyResult<Resp>
    where
        Self: 'a,
        Resp: DeserializeOwned + 'a;

    fn publish<M>(&self, message: M) -> Result<PressureResponse, SdkError>
    where
        M: Serialize + SchemaValidate,
    {
        match self {
            Self::Embedded(handle) => handle.publish(message),
            Self::Remote(handle) => handle.publish(message),
        }
    }

    /// BOTH ARMS REFUSE. Typed subscription delivery is not implemented in
    /// either deployment mode, so whichever mode configuration selected, the
    /// returned stream yields exactly one `Err(SdkError::Unwired)` and then
    /// ends.
    ///
    /// The remote arm's refusal names `SubscriptionStream::open` as the surface
    /// that works. The embedded arm's names the same remote surface, because
    /// embedded mode has no in-process delivery path to offer instead. That arm
    /// used to return a locally empty stream — described here as "what embedded
    /// mode has always meant", which was true of the implementation and
    /// misleading to a caller: an empty stream is indistinguishable from a live
    /// subscription on a quiet channel.
    fn subscribe<M>(&self) -> Self::Subscription<M>
    where
        M: DeserializeOwned,
    {
        match self {
            Self::Embedded(handle) => handle.subscribe(),
            Self::Remote(handle) => handle.subscribe(),
        }
    }

    fn request_reply<Req, Resp>(&self, request: Req) -> ReadyResult<Resp>
    where
        Req: Serialize + SchemaValidate,
        Resp: DeserializeOwned,
    {
        match self {
            Self::Embedded(handle) => handle.request_reply(request),
            Self::Remote(handle) => handle.request_reply(request),
        }
    }
}

/// Runtime-selected conversation handle that keeps deployment differences behind configuration.
#[derive(Clone, Debug)]
pub enum SdkConversationHandle {
    /// Embedded direct in-process handle.
    Embedded(EmbeddedConversationHandle),
    /// Remote protocol-backed handle.
    Remote(RemoteConversationHandle),
}

impl SdkConversationHandle {
    /// Creates a conversation handle selected by SDK configuration.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the selected handle cannot be initialized.
    pub fn new(config: &SdkConfig) -> Result<Self, SdkError> {
        match config {
            SdkConfig::Embedded(config) => {
                Ok(Self::Embedded(EmbeddedConversationHandle::new(config)))
            }
            SdkConfig::Remote(config) => Ok(Self::Remote(RemoteConversationHandle::new(config))),
        }
    }
}

impl ConversationHandle for SdkConversationHandle {
    type ReceiveFuture<'a, M>
        = ReadyResult<M>
    where
        Self: 'a,
        M: DeserializeOwned + 'a;

    type LifecycleStream = EmptyLifecycleStream;

    fn send<M>(&self, message: M) -> Result<(), SdkError>
    where
        M: Serialize,
    {
        match self {
            Self::Embedded(handle) => handle.send(message),
            Self::Remote(handle) => handle.send(message),
        }
    }

    fn receive<M>(&self) -> ReadyResult<M>
    where
        M: DeserializeOwned,
    {
        match self {
            Self::Embedded(handle) => handle.receive(),
            Self::Remote(handle) => handle.receive(),
        }
    }

    fn lifecycle(&self) -> Self::LifecycleStream {
        EmptyLifecycleStream
    }
}