inline_sdk/
realtime.rs

1//! Realtime WebSocket RPC transport for Inline protocol calls.
2
3use futures_util::{SinkExt, StreamExt};
4use prost::Message;
5use std::collections::HashMap;
6use std::fmt;
7use std::future::Future;
8use std::sync::Arc;
9use std::time::Duration;
10use tokio::sync::{Semaphore, broadcast, mpsc, oneshot, watch};
11use tokio_tungstenite::connect_async;
12use tokio_tungstenite::tungstenite::Message as WsMessage;
13use tokio_tungstenite::tungstenite::client::IntoClientRequest;
14use tokio_tungstenite::tungstenite::http::HeaderValue;
15use url::Url;
16
17use crate::client_info::{self, ClientIdentity};
18use inline_protocol::proto;
19
20/// Default timeout for opening a realtime connection.
21pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
22/// Default timeout for a single realtime RPC invocation.
23pub const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(60);
24/// Default number of queued commands accepted by a multiplexed realtime session.
25pub const DEFAULT_SESSION_COMMAND_CAPACITY: usize = 64;
26/// Default maximum number of RPCs awaiting responses on one realtime session.
27pub const DEFAULT_SESSION_MAX_IN_FLIGHT_RPCS: usize = 64;
28/// Default number of pushed events retained for each realtime session subscriber.
29pub const DEFAULT_SESSION_EVENT_CAPACITY: usize = 256;
30/// Default interval between protocol heartbeat pings.
31pub const DEFAULT_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
32/// Default deadline for a matching protocol pong.
33pub const DEFAULT_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(12);
34
35/// Error returned by realtime connection and RPC operations.
36#[derive(thiserror::Error)]
37#[non_exhaustive]
38pub enum RealtimeError {
39    /// Invalid realtime WebSocket URL supplied to a realtime client builder.
40    #[error("invalid realtime URL: {message}")]
41    InvalidUrl {
42        /// Original URL value supplied by the caller.
43        url: String,
44        /// Human-readable validation failure.
45        message: String,
46    },
47    /// WebSocket transport error.
48    #[error("websocket error: {0}")]
49    WebSocket(Box<tokio_tungstenite::tungstenite::Error>),
50    /// A validated client identity could not be represented as a realtime header.
51    #[error("{field} contains characters that are invalid in realtime headers")]
52    InvalidHeaderValue {
53        /// Name of the invalid realtime header field.
54        field: &'static str,
55    },
56    /// Protobuf decode error for a server message.
57    #[error("protocol error: {0}")]
58    Protocol(#[from] prost::DecodeError),
59    /// The server returned an RPC result envelope without a result body.
60    #[error("missing rpc result")]
61    MissingResult,
62    /// The server returned a result oneof that does not match the requested method.
63    #[error("unexpected rpc result for {method}: expected {expected}, got {actual}")]
64    UnexpectedResult {
65        /// Requested RPC method.
66        method: &'static str,
67        /// Expected result oneof variant.
68        expected: &'static str,
69        /// Actual result oneof variant.
70        actual: &'static str,
71    },
72    /// A realtime operation exceeded its configured timeout.
73    #[error("realtime {operation} timed out after {timeout:?}")]
74    Timeout {
75        /// Operation that timed out.
76        operation: &'static str,
77        /// Configured timeout.
78        timeout: Duration,
79    },
80    /// The realtime server rejected the connection.
81    #[error("{friendly}")]
82    ConnectionError {
83        /// Numeric connection-error reason.
84        reason: i32,
85        /// Stable protobuf enum name for the reason.
86        reason_name: String,
87        /// Human-readable formatted error.
88        friendly: String,
89    },
90    /// The realtime connection closed before the requested operation completed.
91    #[error("realtime connection closed")]
92    ConnectionClosed,
93    /// A multiplexed event subscriber could not keep up with pushed events.
94    #[error("realtime event subscriber lagged and skipped {skipped} events")]
95    EventLagged {
96        /// Number of events dropped for this subscriber.
97        skipped: u64,
98    },
99    /// The realtime server returned an RPC error for a request.
100    #[error("{friendly}")]
101    RpcError {
102        /// Transport-level status code.
103        code: i32,
104        /// Inline RPC error code.
105        error_code: i32,
106        /// Stable protobuf enum name for the RPC error.
107        error_name: String,
108        /// Server-provided error message.
109        message: String,
110        /// Human-readable formatted error.
111        friendly: String,
112    },
113}
114
115impl fmt::Debug for RealtimeError {
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        match self {
118            RealtimeError::InvalidUrl { url, message } => f
119                .debug_struct("InvalidUrl")
120                .field("url", &realtime_url_for_debug(url))
121                .field("message", message)
122                .finish(),
123            RealtimeError::WebSocket(error) => f.debug_tuple("WebSocket").field(error).finish(),
124            RealtimeError::InvalidHeaderValue { field } => f
125                .debug_struct("InvalidHeaderValue")
126                .field("field", field)
127                .finish(),
128            RealtimeError::Protocol(error) => f.debug_tuple("Protocol").field(error).finish(),
129            RealtimeError::MissingResult => f.debug_struct("MissingResult").finish(),
130            RealtimeError::UnexpectedResult {
131                method,
132                expected,
133                actual,
134            } => f
135                .debug_struct("UnexpectedResult")
136                .field("method", method)
137                .field("expected", expected)
138                .field("actual", actual)
139                .finish(),
140            RealtimeError::Timeout { operation, timeout } => f
141                .debug_struct("Timeout")
142                .field("operation", operation)
143                .field("timeout", timeout)
144                .finish(),
145            RealtimeError::ConnectionError {
146                reason,
147                reason_name,
148                friendly,
149            } => f
150                .debug_struct("ConnectionError")
151                .field("reason", reason)
152                .field("reason_name", reason_name)
153                .field("friendly", friendly)
154                .finish(),
155            RealtimeError::ConnectionClosed => f.debug_struct("ConnectionClosed").finish(),
156            RealtimeError::EventLagged { skipped } => f
157                .debug_struct("EventLagged")
158                .field("skipped", skipped)
159                .finish(),
160            RealtimeError::RpcError {
161                code,
162                error_code,
163                error_name,
164                message,
165                friendly,
166            } => f
167                .debug_struct("RpcError")
168                .field("code", code)
169                .field("error_code", error_code)
170                .field("error_name", error_name)
171                .field("message", message)
172                .field("friendly", friendly)
173                .finish(),
174        }
175    }
176}
177
178impl From<tokio_tungstenite::tungstenite::Error> for RealtimeError {
179    fn from(error: tokio_tungstenite::tungstenite::Error) -> Self {
180        Self::WebSocket(Box::new(error))
181    }
182}
183
184/// Stateful realtime connection for issuing Inline RPC calls.
185pub struct RealtimeClient {
186    ws: tokio_tungstenite::WebSocketStream<
187        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
188    >,
189    seq: u32,
190    id_gen: IdGenerator,
191    rpc_timeout: Option<Duration>,
192    heartbeat_interval: Option<Duration>,
193    heartbeat_timeout: Duration,
194    max_in_flight_rpcs: usize,
195}
196
197/// Server-pushed realtime event received outside a direct RPC result.
198#[derive(Clone, Debug, PartialEq)]
199#[non_exhaustive]
200pub enum RealtimeEvent {
201    /// Inline protocol updates pushed by the server.
202    Updates(Vec<proto::Update>),
203    /// Server acknowledgement for a previously sent client message.
204    Ack {
205        /// Client message ID acknowledged by the server.
206        msg_id: u64,
207    },
208    /// Server pong response.
209    Pong {
210        /// Pong nonce.
211        nonce: u64,
212    },
213}
214
215/// Cloneable realtime session that multiplexes concurrent RPCs and pushed
216/// events over one WebSocket connection.
217///
218/// Create an event receiver with [`RealtimeSession::subscribe`] before issuing
219/// RPCs that may cause the server to push update hints.
220#[derive(Clone)]
221pub struct RealtimeSession {
222    commands: mpsc::Sender<SessionCommand>,
223    events: broadcast::Sender<RealtimeEvent>,
224    closed: watch::Receiver<bool>,
225    rpc_timeout: Option<Duration>,
226    heartbeat_interval: Option<Duration>,
227    heartbeat_timeout: Duration,
228    rpc_permits: Arc<Semaphore>,
229}
230
231impl fmt::Debug for RealtimeSession {
232    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
233        f.debug_struct("RealtimeSession")
234            .field("rpc_timeout", &self.rpc_timeout)
235            .field("heartbeat_interval", &self.heartbeat_interval)
236            .field("heartbeat_timeout", &self.heartbeat_timeout)
237            .field(
238                "available_rpc_permits",
239                &self.rpc_permits.available_permits(),
240            )
241            .field("closed", &*self.closed.borrow())
242            .finish_non_exhaustive()
243    }
244}
245
246/// Receiver for pushed events from a multiplexed [`RealtimeSession`].
247pub struct RealtimeEventReceiver {
248    events: broadcast::Receiver<RealtimeEvent>,
249    closed: watch::Receiver<bool>,
250}
251
252impl fmt::Debug for RealtimeEventReceiver {
253    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
254        f.debug_struct("RealtimeEventReceiver")
255            .field("closed", &*self.closed.borrow())
256            .finish_non_exhaustive()
257    }
258}
259
260enum SessionCommand {
261    Invoke {
262        method: proto::Method,
263        input: proto::rpc_call::Input,
264        response: oneshot::Sender<Result<proto::rpc_result::Result, RealtimeError>>,
265    },
266}
267
268/// Builder for [`RealtimeClient`].
269#[must_use]
270#[derive(Clone)]
271pub struct RealtimeClientBuilder {
272    url: String,
273    token: String,
274    identity: ClientIdentity,
275    connect_timeout: Option<Duration>,
276    rpc_timeout: Option<Duration>,
277    heartbeat_interval: Option<Duration>,
278    heartbeat_timeout: Duration,
279    max_in_flight_rpcs: usize,
280}
281
282impl fmt::Debug for RealtimeClientBuilder {
283    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
284        f.debug_struct("RealtimeClientBuilder")
285            .field("url", &realtime_url_for_debug(&self.url))
286            .field("token", &"<redacted>")
287            .field("identity", &self.identity)
288            .field("connect_timeout", &self.connect_timeout)
289            .field("rpc_timeout", &self.rpc_timeout)
290            .field("heartbeat_interval", &self.heartbeat_interval)
291            .field("heartbeat_timeout", &self.heartbeat_timeout)
292            .field("max_in_flight_rpcs", &self.max_in_flight_rpcs)
293            .finish()
294    }
295}
296
297/// Typed Inline realtime RPC request.
298///
299/// This trait is implemented for generated protocol input types so callers can
300/// use [`RealtimeClient::call`] without manually pairing each input with its
301/// [`proto::Method`] and result oneof.
302pub trait RpcRequest: Sized {
303    /// Typed response returned by this request.
304    type Response;
305
306    /// RPC method associated with this input type.
307    const METHOD: proto::Method;
308
309    /// Converts this request into the generated RPC input oneof.
310    fn into_rpc_input(self) -> proto::rpc_call::Input;
311
312    /// Extracts the typed response from the generated RPC result oneof.
313    fn response_from_rpc_result(
314        result: proto::rpc_result::Result,
315    ) -> Result<Self::Response, RealtimeError>;
316}
317
318impl RealtimeClient {
319    /// Starts a realtime client builder.
320    pub fn builder(url: impl Into<String>, token: impl Into<String>) -> RealtimeClientBuilder {
321        RealtimeClientBuilder::new(url, token)
322    }
323
324    /// Connects to realtime using the default SDK identity.
325    pub async fn connect(url: &str, token: &str) -> Result<Self, RealtimeError> {
326        Self::builder(url, token).connect().await
327    }
328
329    /// Connects to realtime using a caller-provided client identity.
330    pub async fn connect_with_identity(
331        url: &str,
332        token: &str,
333        identity: ClientIdentity,
334    ) -> Result<Self, RealtimeError> {
335        Self::builder(url, token).identity(identity).connect().await
336    }
337}
338
339impl RealtimeSession {
340    /// Connects a multiplexed realtime session using the default SDK identity.
341    pub async fn connect(url: &str, token: &str) -> Result<Self, RealtimeError> {
342        RealtimeClient::builder(url, token).connect_session().await
343    }
344
345    /// Connects a multiplexed realtime session with an explicit client identity.
346    pub async fn connect_with_identity(
347        url: &str,
348        token: &str,
349        identity: ClientIdentity,
350    ) -> Result<Self, RealtimeError> {
351        RealtimeClient::builder(url, token)
352            .identity(identity)
353            .connect_session()
354            .await
355    }
356
357    /// Subscribes to server-pushed events routed by this session.
358    pub fn subscribe(&self) -> RealtimeEventReceiver {
359        RealtimeEventReceiver {
360            events: self.events.subscribe(),
361            closed: self.closed.clone(),
362        }
363    }
364
365    /// Returns whether the session transport task has stopped.
366    pub fn is_closed(&self) -> bool {
367        *self.closed.borrow()
368    }
369
370    /// Invokes a typed Inline RPC while the same transport continues routing
371    /// pushed events to subscribers.
372    pub async fn call<R>(&self, request: R) -> Result<R::Response, RealtimeError>
373    where
374        R: RpcRequest,
375    {
376        let result = self.invoke(R::METHOD, request.into_rpc_input()).await?;
377        R::response_from_rpc_result(result)
378    }
379
380    /// Invokes an Inline RPC while the same transport continues routing pushed
381    /// events and other concurrent RPC results.
382    pub async fn invoke(
383        &self,
384        method: proto::Method,
385        input: proto::rpc_call::Input,
386    ) -> Result<proto::rpc_result::Result, RealtimeError> {
387        if self.is_closed() {
388            return Err(RealtimeError::ConnectionClosed);
389        }
390        let (response_tx, response_rx) = oneshot::channel();
391        let commands = self.commands.clone();
392        let permits = self.rpc_permits.clone();
393        let response = async move {
394            let _permit = permits
395                .acquire_owned()
396                .await
397                .map_err(|_| RealtimeError::ConnectionClosed)?;
398            commands
399                .send(SessionCommand::Invoke {
400                    method,
401                    input,
402                    response: response_tx,
403                })
404                .await
405                .map_err(|_| RealtimeError::ConnectionClosed)?;
406            response_rx
407                .await
408                .map_err(|_| RealtimeError::ConnectionClosed)?
409        };
410        with_optional_timeout("rpc", self.rpc_timeout, response).await
411    }
412
413    fn from_client(client: RealtimeClient) -> Self {
414        let rpc_timeout = client.rpc_timeout;
415        let heartbeat_interval = client.heartbeat_interval;
416        let heartbeat_timeout = client.heartbeat_timeout;
417        let max_in_flight_rpcs = client.max_in_flight_rpcs;
418        let (command_tx, command_rx) = mpsc::channel(DEFAULT_SESSION_COMMAND_CAPACITY);
419        let (event_tx, _) = broadcast::channel(DEFAULT_SESSION_EVENT_CAPACITY);
420        let (closed_tx, closed_rx) = watch::channel(false);
421        tokio::spawn(run_realtime_session(
422            client,
423            command_rx,
424            event_tx.clone(),
425            closed_tx,
426        ));
427        Self {
428            commands: command_tx,
429            events: event_tx,
430            closed: closed_rx,
431            rpc_timeout,
432            heartbeat_interval,
433            heartbeat_timeout,
434            rpc_permits: Arc::new(Semaphore::new(max_in_flight_rpcs)),
435        }
436    }
437}
438
439impl RealtimeEventReceiver {
440    /// Waits for the next pushed event or a terminal session failure.
441    pub async fn recv(&mut self) -> Result<RealtimeEvent, RealtimeError> {
442        loop {
443            if *self.closed.borrow() && self.events.is_empty() {
444                return Err(RealtimeError::ConnectionClosed);
445            }
446
447            tokio::select! {
448                result = self.events.recv() => return match result {
449                    Ok(event) => Ok(event),
450                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
451                        Err(RealtimeError::EventLagged { skipped })
452                    }
453                    Err(broadcast::error::RecvError::Closed) => {
454                        Err(RealtimeError::ConnectionClosed)
455                    }
456                },
457                changed = self.closed.changed() => {
458                    match changed {
459                        Ok(()) => continue,
460                        Err(_) => return Err(RealtimeError::ConnectionClosed),
461                    }
462                }
463            }
464        }
465    }
466}
467
468impl RealtimeClientBuilder {
469    /// Creates a realtime client builder with the default SDK identity.
470    pub fn new(url: impl Into<String>, token: impl Into<String>) -> Self {
471        Self {
472            url: url.into(),
473            token: token.into(),
474            identity: ClientIdentity::sdk(),
475            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
476            rpc_timeout: Some(DEFAULT_RPC_TIMEOUT),
477            heartbeat_interval: Some(DEFAULT_HEARTBEAT_INTERVAL),
478            heartbeat_timeout: DEFAULT_HEARTBEAT_TIMEOUT,
479            max_in_flight_rpcs: DEFAULT_SESSION_MAX_IN_FLIGHT_RPCS,
480        }
481    }
482
483    /// Sets the client identity used in realtime headers and connection init.
484    pub fn identity(mut self, identity: ClientIdentity) -> Self {
485        self.identity = identity;
486        self
487    }
488
489    /// Sets the timeout for opening the WebSocket and receiving `ConnectionOpen`.
490    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
491        self.connect_timeout = Some(timeout);
492        self
493    }
494
495    /// Disables the connect timeout.
496    pub fn without_connect_timeout(mut self) -> Self {
497        self.connect_timeout = None;
498        self
499    }
500
501    /// Sets the timeout for each RPC invocation.
502    pub fn rpc_timeout(mut self, timeout: Duration) -> Self {
503        self.rpc_timeout = Some(timeout);
504        self
505    }
506
507    /// Disables the per-RPC timeout.
508    pub fn without_rpc_timeout(mut self) -> Self {
509        self.rpc_timeout = None;
510        self
511    }
512
513    /// Sets the maximum number of RPCs awaiting responses on one session.
514    pub fn max_in_flight_rpcs(mut self, maximum: usize) -> Self {
515        self.max_in_flight_rpcs = maximum.max(1);
516        self
517    }
518
519    /// Configures protocol heartbeat interval and pong deadline.
520    pub fn heartbeat(mut self, interval: Duration, timeout: Duration) -> Self {
521        self.heartbeat_interval = Some(interval.max(Duration::from_millis(1)));
522        self.heartbeat_timeout = timeout.max(Duration::from_millis(1));
523        self
524    }
525
526    /// Disables multiplexed-session protocol heartbeat pings.
527    pub fn without_heartbeat(mut self) -> Self {
528        self.heartbeat_interval = None;
529        self
530    }
531
532    /// Opens the WebSocket connection and waits for `ConnectionOpen`.
533    pub async fn connect(self) -> Result<RealtimeClient, RealtimeError> {
534        let url = normalize_realtime_url(self.url)?;
535        log::debug!(
536            target: "inline_sdk::realtime",
537            "opening realtime websocket url={} identity_type={} connect_timeout={:?} rpc_timeout={:?}",
538            realtime_url_for_log(&url),
539            self.identity.client_type(),
540            self.connect_timeout,
541            self.rpc_timeout
542        );
543        let mut request = url.as_str().into_client_request()?;
544        request.headers_mut().insert(
545            client_info::CLIENT_TYPE_HEADER,
546            realtime_header_value("client_type", self.identity.client_type())?,
547        );
548        request.headers_mut().insert(
549            client_info::CLIENT_VERSION_HEADER,
550            realtime_header_value("client_version", self.identity.client_version())?,
551        );
552        request.headers_mut().insert(
553            "user-agent",
554            realtime_header_value("user_agent", &client_info::user_agent_for(&self.identity))?,
555        );
556
557        let (ws, _) =
558            with_optional_timeout("connect", self.connect_timeout, connect_async(request)).await?;
559        log::debug!(target: "inline_sdk::realtime", "websocket connected");
560        let mut client = RealtimeClient {
561            ws,
562            seq: 0,
563            id_gen: IdGenerator::new(),
564            rpc_timeout: self.rpc_timeout,
565            heartbeat_interval: self.heartbeat_interval,
566            heartbeat_timeout: self.heartbeat_timeout,
567            max_in_flight_rpcs: self.max_in_flight_rpcs,
568        };
569
570        with_optional_timeout(
571            "connection_init",
572            self.connect_timeout,
573            client.send_connection_init(&self.token, &self.identity),
574        )
575        .await?;
576        log::trace!(target: "inline_sdk::realtime", "connection init sent");
577        with_optional_timeout(
578            "connection_open",
579            self.connect_timeout,
580            client.wait_for_connection_open(),
581        )
582        .await?;
583        log::debug!(target: "inline_sdk::realtime", "realtime protocol open");
584        Ok(client)
585    }
586
587    /// Opens one WebSocket and starts a multiplexed session for concurrent RPC
588    /// calls and pushed events.
589    pub async fn connect_session(self) -> Result<RealtimeSession, RealtimeError> {
590        self.connect().await.map(RealtimeSession::from_client)
591    }
592}
593
594impl RealtimeClient {
595    /// Invokes a typed Inline RPC request.
596    pub async fn call<R>(&mut self, request: R) -> Result<R::Response, RealtimeError>
597    where
598        R: RpcRequest,
599    {
600        log::trace!(
601            target: "inline_sdk::realtime",
602            "calling typed rpc method={}",
603            R::METHOD.as_str_name()
604        );
605        let result = self.invoke(R::METHOD, request.into_rpc_input()).await?;
606        R::response_from_rpc_result(result)
607    }
608
609    /// Invokes an Inline RPC method and waits for the matching result.
610    pub async fn invoke(
611        &mut self,
612        method: proto::Method,
613        input: proto::rpc_call::Input,
614    ) -> Result<proto::rpc_result::Result, RealtimeError> {
615        let message_id = self.send_rpc_call(method, input).await?;
616
617        with_optional_timeout(
618            "rpc",
619            self.rpc_timeout,
620            self.wait_for_rpc_result(message_id),
621        )
622        .await
623    }
624
625    async fn send_rpc_call(
626        &mut self,
627        method: proto::Method,
628        input: proto::rpc_call::Input,
629    ) -> Result<u64, RealtimeError> {
630        let message_id = self.next_id();
631        log::trace!(
632            target: "inline_sdk::realtime",
633            "sending rpc method={} msg_id={message_id}",
634            method.as_str_name()
635        );
636        let message = proto::ClientMessage {
637            id: message_id,
638            seq: self.next_seq(),
639            body: Some(proto::client_message::Body::RpcCall(proto::RpcCall {
640                method: method as i32,
641                input: Some(input),
642            })),
643        };
644        self.send_client_message(message).await?;
645        Ok(message_id)
646    }
647
648    /// Returns the configured per-RPC timeout.
649    pub fn rpc_timeout(&self) -> Option<Duration> {
650        self.rpc_timeout
651    }
652
653    /// Waits for the next server-pushed realtime event.
654    ///
655    /// This reads the same Inline realtime protocol stream used for RPC calls.
656    /// Callers that need both request/response RPCs and long-lived pushed
657    /// updates should either serialize access to one [`RealtimeClient`] or use
658    /// a separate realtime connection for the event receiver.
659    pub async fn next_event(&mut self) -> Result<RealtimeEvent, RealtimeError> {
660        loop {
661            let message = self.read_server_message().await?;
662            if let Some(event) = self.event_from_server_message(message).await? {
663                return Ok(event);
664            }
665        }
666    }
667
668    async fn send_connection_init(
669        &mut self,
670        token: &str,
671        identity: &ClientIdentity,
672    ) -> Result<(), RealtimeError> {
673        let init = connection_init_for_token(token, identity);
674        let message = proto::ClientMessage {
675            id: self.next_id(),
676            seq: self.next_seq(),
677            body: Some(proto::client_message::Body::ConnectionInit(init)),
678        };
679
680        self.send_client_message(message).await
681    }
682
683    async fn wait_for_connection_open(&mut self) -> Result<(), RealtimeError> {
684        loop {
685            let message = self.read_server_message().await?;
686            let message_id = message.id;
687            match message.body {
688                Some(proto::server_protocol_message::Body::ConnectionOpen(_)) => return Ok(()),
689                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
690                    log::warn!(
691                        target: "inline_sdk::realtime",
692                        "connection open rejected reason={}",
693                        proto::connection_error::Reason::try_from(error.reason)
694                            .map(|reason| reason.as_str_name())
695                            .unwrap_or("UNKNOWN")
696                    );
697                    return Err(connection_error_from_proto(error));
698                }
699                Some(proto::server_protocol_message::Body::Message(message)) => {
700                    let _ = self.server_payload_event(message_id, message).await?;
701                }
702                _ => {}
703            }
704        }
705    }
706
707    async fn wait_for_rpc_result(
708        &mut self,
709        message_id: u64,
710    ) -> Result<proto::rpc_result::Result, RealtimeError> {
711        loop {
712            let message = self.read_server_message().await?;
713            match message.body {
714                Some(proto::server_protocol_message::Body::RpcResult(result))
715                    if result.req_msg_id == message_id =>
716                {
717                    log::trace!(
718                        target: "inline_sdk::realtime",
719                        "received rpc result msg_id={message_id}"
720                    );
721                    return result.result.ok_or(RealtimeError::MissingResult);
722                }
723                Some(proto::server_protocol_message::Body::RpcError(error))
724                    if error.req_msg_id == message_id =>
725                {
726                    log::warn!(
727                        target: "inline_sdk::realtime",
728                        "received rpc error msg_id={message_id} error={} status={}",
729                        rpc_error_code_name(error.error_code),
730                        error.code,
731                    );
732                    return Err(rpc_error_from_proto(error));
733                }
734                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
735                    log::warn!(
736                        target: "inline_sdk::realtime",
737                        "connection error while waiting for rpc msg_id={message_id} reason={}",
738                        proto::connection_error::Reason::try_from(error.reason)
739                            .map(|reason| reason.as_str_name())
740                            .unwrap_or("UNKNOWN")
741                    );
742                    return Err(connection_error_from_proto(error));
743                }
744                Some(proto::server_protocol_message::Body::Message(server_message)) => {
745                    if let Some(event) = self
746                        .server_payload_event(message.id, server_message)
747                        .await?
748                    {
749                        log::trace!(
750                            target: "inline_sdk::realtime",
751                            "received pushed realtime event while waiting for rpc msg_id={message_id}: {}",
752                            realtime_event_kind(&event)
753                        );
754                    }
755                }
756                _ => {}
757            }
758        }
759    }
760
761    async fn event_from_server_message(
762        &mut self,
763        message: proto::ServerProtocolMessage,
764    ) -> Result<Option<RealtimeEvent>, RealtimeError> {
765        match message.body {
766            Some(proto::server_protocol_message::Body::Message(server_message)) => {
767                self.server_payload_event(message.id, server_message).await
768            }
769            Some(proto::server_protocol_message::Body::Ack(ack)) => {
770                Ok(Some(RealtimeEvent::Ack { msg_id: ack.msg_id }))
771            }
772            Some(proto::server_protocol_message::Body::Pong(pong)) => {
773                Ok(Some(RealtimeEvent::Pong { nonce: pong.nonce }))
774            }
775            Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
776                Err(connection_error_from_proto(error))
777            }
778            Some(proto::server_protocol_message::Body::ConnectionOpen(_))
779            | Some(proto::server_protocol_message::Body::RpcResult(_))
780            | Some(proto::server_protocol_message::Body::RpcError(_))
781            | None => Ok(None),
782        }
783    }
784
785    async fn server_payload_event(
786        &mut self,
787        message_id: u64,
788        message: proto::ServerMessage,
789    ) -> Result<Option<RealtimeEvent>, RealtimeError> {
790        match message.payload {
791            Some(proto::server_message::Payload::Update(payload)) => {
792                self.send_ack(message_id).await?;
793                Ok(Some(RealtimeEvent::Updates(payload.updates)))
794            }
795            None => Ok(None),
796        }
797    }
798
799    async fn send_ack(&mut self, msg_id: u64) -> Result<(), RealtimeError> {
800        let message = proto::ClientMessage {
801            id: self.next_id(),
802            seq: self.next_seq(),
803            body: Some(proto::client_message::Body::Ack(proto::Ack { msg_id })),
804        };
805        self.send_client_message(message).await
806    }
807
808    async fn send_ping(&mut self, nonce: u64) -> Result<(), RealtimeError> {
809        let message = proto::ClientMessage {
810            id: self.next_id(),
811            seq: self.next_seq(),
812            body: Some(proto::client_message::Body::Ping(proto::Ping { nonce })),
813        };
814        self.send_client_message(message).await
815    }
816
817    async fn send_client_message(
818        &mut self,
819        message: proto::ClientMessage,
820    ) -> Result<(), RealtimeError> {
821        let bytes = message.encode_to_vec();
822        self.ws.send(WsMessage::Binary(bytes.into())).await?;
823        Ok(())
824    }
825
826    async fn read_server_message(&mut self) -> Result<proto::ServerProtocolMessage, RealtimeError> {
827        loop {
828            let message = self
829                .ws
830                .next()
831                .await
832                .ok_or(RealtimeError::ConnectionClosed)??;
833            match message {
834                WsMessage::Binary(data) => {
835                    return Ok(proto::ServerProtocolMessage::decode(&*data)?);
836                }
837                WsMessage::Text(_) => continue,
838                WsMessage::Close(_) => return Err(RealtimeError::ConnectionClosed),
839                WsMessage::Ping(_) | WsMessage::Pong(_) => continue,
840                _ => continue,
841            }
842        }
843    }
844
845    fn next_seq(&mut self) -> u32 {
846        self.seq = self.seq.wrapping_add(1);
847        self.seq
848    }
849
850    fn next_id(&mut self) -> u64 {
851        self.id_gen.next_id()
852    }
853}
854
855async fn run_realtime_session(
856    mut client: RealtimeClient,
857    mut commands: mpsc::Receiver<SessionCommand>,
858    events: broadcast::Sender<RealtimeEvent>,
859    closed: watch::Sender<bool>,
860) {
861    let heartbeat_interval = client.heartbeat_interval;
862    let heartbeat_timeout = client.heartbeat_timeout;
863    let mut heartbeat =
864        tokio::time::interval(heartbeat_interval.unwrap_or(Duration::from_secs(24 * 60 * 60)));
865    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
866    heartbeat.tick().await;
867    let mut pending_ping: Option<(u64, tokio::time::Instant)> = None;
868    let mut pending =
869        HashMap::<u64, oneshot::Sender<Result<proto::rpc_result::Result, RealtimeError>>>::new();
870
871    loop {
872        // A timed-out or cancelled caller drops its receiver. Remove those
873        // entries before accepting more work so an unresponsive server cannot
874        // grow the pending map indefinitely.
875        pending.retain(|_, response| !response.is_closed());
876        let heartbeat_deadline = pending_ping
877            .map(|(_, deadline)| deadline)
878            .unwrap_or_else(tokio::time::Instant::now);
879        tokio::select! {
880            _ = heartbeat.tick(), if heartbeat_interval.is_some() => {
881                if pending_ping.is_none() {
882                    let nonce = client.next_id();
883                    if let Err(error) = client.send_ping(nonce).await {
884                        log::warn!(
885                            target: "inline_sdk::realtime",
886                            "failed to send realtime heartbeat: {error}"
887                        );
888                        break;
889                    }
890                    pending_ping = Some((nonce, tokio::time::Instant::now() + heartbeat_timeout));
891                }
892            }
893            _ = tokio::time::sleep_until(heartbeat_deadline), if pending_ping.is_some() => {
894                let nonce = pending_ping.map(|(nonce, _)| nonce).unwrap_or_default();
895                log::warn!(
896                    target: "inline_sdk::realtime",
897                    "realtime heartbeat timed out nonce={nonce} timeout={heartbeat_timeout:?}"
898                );
899                break;
900            }
901            command = commands.recv() => {
902                let Some(command) = command else {
903                    break;
904                };
905                match command {
906                    SessionCommand::Invoke { method, input, response } => {
907                        match client.send_rpc_call(method, input).await {
908                            Ok(message_id) => {
909                                pending.insert(message_id, response);
910                            }
911                            Err(error) => {
912                                let _ = response.send(Err(error));
913                                break;
914                            }
915                        }
916                    }
917                }
918            }
919            message = client.read_server_message() => {
920                let message = match message {
921                    Ok(message) => message,
922                    Err(error) => {
923                        log::warn!(
924                            target: "inline_sdk::realtime",
925                            "multiplexed realtime session stopped: {error}"
926                        );
927                        break;
928                    }
929                };
930                match message.body {
931                    Some(proto::server_protocol_message::Body::RpcResult(result)) => {
932                        if let Some(response) = pending.remove(&result.req_msg_id) {
933                            let value = result.result.ok_or(RealtimeError::MissingResult);
934                            let _ = response.send(value);
935                        } else {
936                            log::debug!(
937                                target: "inline_sdk::realtime",
938                                "received result for unknown or timed-out rpc msg_id={}",
939                                result.req_msg_id
940                            );
941                        }
942                    }
943                    Some(proto::server_protocol_message::Body::RpcError(error)) => {
944                        if let Some(response) = pending.remove(&error.req_msg_id) {
945                            let _ = response.send(Err(rpc_error_from_proto(error)));
946                        } else {
947                            log::debug!(
948                                target: "inline_sdk::realtime",
949                                "received error for unknown or timed-out rpc msg_id={}",
950                                error.req_msg_id
951                            );
952                        }
953                    }
954                    Some(proto::server_protocol_message::Body::Message(server_message)) => {
955                        match client.server_payload_event(message.id, server_message).await {
956                            Ok(Some(event)) => {
957                                let _ = events.send(event);
958                            }
959                            Ok(None) => {}
960                            Err(error) => {
961                                log::warn!(
962                                    target: "inline_sdk::realtime",
963                                    "failed to route pushed realtime event: {error}"
964                                );
965                                break;
966                            }
967                        }
968                    }
969                    Some(proto::server_protocol_message::Body::Ack(ack)) => {
970                        let _ = events.send(RealtimeEvent::Ack { msg_id: ack.msg_id });
971                    }
972                    Some(proto::server_protocol_message::Body::Pong(pong)) => {
973                        if pending_ping.is_some_and(|(nonce, _)| nonce == pong.nonce) {
974                            pending_ping = None;
975                        }
976                        let _ = events.send(RealtimeEvent::Pong { nonce: pong.nonce });
977                    }
978                    Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
979                        let error = connection_error_from_proto(error);
980                        log::warn!(
981                            target: "inline_sdk::realtime",
982                            "multiplexed realtime session rejected: {error}"
983                        );
984                        break;
985                    }
986                    Some(proto::server_protocol_message::Body::ConnectionOpen(_)) | None => {}
987                }
988            }
989        }
990    }
991
992    for (_, response) in pending {
993        let _ = response.send(Err(RealtimeError::ConnectionClosed));
994    }
995    let _ = closed.send(true);
996}
997
998macro_rules! rpc_requests {
999    ($(($input_ty:ident, $method:ident, $input_variant:ident, $result_ty:ident, $result_variant:ident)),+ $(,)?) => {
1000        $(
1001            impl RpcRequest for proto::$input_ty {
1002                type Response = proto::$result_ty;
1003
1004                const METHOD: proto::Method = proto::Method::$method;
1005
1006                fn into_rpc_input(self) -> proto::rpc_call::Input {
1007                    proto::rpc_call::Input::$input_variant(self)
1008                }
1009
1010                fn response_from_rpc_result(
1011                    result: proto::rpc_result::Result,
1012                ) -> Result<Self::Response, RealtimeError> {
1013                    match result {
1014                        proto::rpc_result::Result::$result_variant(response) => Ok(response),
1015                        other => {
1016                            let actual = rpc_result_variant_name(&other);
1017                            log::warn!(
1018                                target: "inline_sdk::realtime",
1019                                "unexpected rpc result method={} expected={} actual={actual}",
1020                                Self::METHOD.as_str_name(),
1021                                stringify!($result_variant)
1022                            );
1023                            Err(RealtimeError::UnexpectedResult {
1024                                method: Self::METHOD.as_str_name(),
1025                                expected: stringify!($result_variant),
1026                                actual,
1027                            })
1028                        }
1029                    }
1030                }
1031            }
1032        )+
1033
1034        fn rpc_result_variant_name(result: &proto::rpc_result::Result) -> &'static str {
1035            match result {
1036                $(proto::rpc_result::Result::$result_variant(_) => stringify!($result_variant),)+
1037            }
1038        }
1039    };
1040}
1041
1042rpc_requests!(
1043    (GetMeInput, GetMe, GetMe, GetMeResult, GetMe),
1044    (
1045        GetPeerPhotoInput,
1046        GetPeerPhoto,
1047        GetPeerPhoto,
1048        GetPeerPhotoResult,
1049        GetPeerPhoto
1050    ),
1051    (
1052        DeleteMessagesInput,
1053        DeleteMessages,
1054        DeleteMessages,
1055        DeleteMessagesResult,
1056        DeleteMessages
1057    ),
1058    (
1059        SendMessageInput,
1060        SendMessage,
1061        SendMessage,
1062        SendMessageResult,
1063        SendMessage
1064    ),
1065    (
1066        GetChatHistoryInput,
1067        GetChatHistory,
1068        GetChatHistory,
1069        GetChatHistoryResult,
1070        GetChatHistory
1071    ),
1072    (
1073        AddReactionInput,
1074        AddReaction,
1075        AddReaction,
1076        AddReactionResult,
1077        AddReaction
1078    ),
1079    (
1080        DeleteReactionInput,
1081        DeleteReaction,
1082        DeleteReaction,
1083        DeleteReactionResult,
1084        DeleteReaction
1085    ),
1086    (
1087        EditMessageInput,
1088        EditMessage,
1089        EditMessage,
1090        EditMessageResult,
1091        EditMessage
1092    ),
1093    (
1094        CreateChatInput,
1095        CreateChat,
1096        CreateChat,
1097        CreateChatResult,
1098        CreateChat
1099    ),
1100    (
1101        GetSpaceMembersInput,
1102        GetSpaceMembers,
1103        GetSpaceMembers,
1104        GetSpaceMembersResult,
1105        GetSpaceMembers
1106    ),
1107    (
1108        DeleteChatInput,
1109        DeleteChat,
1110        DeleteChat,
1111        DeleteChatResult,
1112        DeleteChat
1113    ),
1114    (
1115        InviteToSpaceInput,
1116        InviteToSpace,
1117        InviteToSpace,
1118        InviteToSpaceResult,
1119        InviteToSpace
1120    ),
1121    (
1122        GetChatParticipantsInput,
1123        GetChatParticipants,
1124        GetChatParticipants,
1125        GetChatParticipantsResult,
1126        GetChatParticipants
1127    ),
1128    (
1129        AddChatParticipantInput,
1130        AddChatParticipant,
1131        AddChatParticipant,
1132        AddChatParticipantResult,
1133        AddChatParticipant
1134    ),
1135    (
1136        RemoveChatParticipantInput,
1137        RemoveChatParticipant,
1138        RemoveChatParticipant,
1139        RemoveChatParticipantResult,
1140        RemoveChatParticipant
1141    ),
1142    (
1143        TranslateMessagesInput,
1144        TranslateMessages,
1145        TranslateMessages,
1146        TranslateMessagesResult,
1147        TranslateMessages
1148    ),
1149    (GetChatsInput, GetChats, GetChats, GetChatsResult, GetChats),
1150    (
1151        UpdateUserSettingsInput,
1152        UpdateUserSettings,
1153        UpdateUserSettings,
1154        UpdateUserSettingsResult,
1155        UpdateUserSettings
1156    ),
1157    (
1158        GetUserSettingsInput,
1159        GetUserSettings,
1160        GetUserSettings,
1161        GetUserSettingsResult,
1162        GetUserSettings
1163    ),
1164    (
1165        SendComposeActionInput,
1166        SendComposeAction,
1167        SendComposeAction,
1168        SendComposeActionResult,
1169        SendComposeAction
1170    ),
1171    (
1172        CreateBotInput,
1173        CreateBot,
1174        CreateBot,
1175        CreateBotResult,
1176        CreateBot
1177    ),
1178    (
1179        DeleteMemberInput,
1180        DeleteMember,
1181        DeleteMember,
1182        DeleteMemberResult,
1183        DeleteMember
1184    ),
1185    (
1186        MarkAsUnreadInput,
1187        MarkAsUnread,
1188        MarkAsUnread,
1189        MarkAsUnreadResult,
1190        MarkAsUnread
1191    ),
1192    (
1193        GetUpdatesStateInput,
1194        GetUpdatesState,
1195        GetUpdatesState,
1196        GetUpdatesStateResult,
1197        GetUpdatesState
1198    ),
1199    (GetChatInput, GetChat, GetChat, GetChatResult, GetChat),
1200    (
1201        GetUpdatesInput,
1202        GetUpdates,
1203        GetUpdates,
1204        GetUpdatesResult,
1205        GetUpdates
1206    ),
1207    (
1208        UpdateMemberAccessInput,
1209        UpdateMemberAccess,
1210        UpdateMemberAccess,
1211        UpdateMemberAccessResult,
1212        UpdateMemberAccess
1213    ),
1214    (
1215        SearchMessagesInput,
1216        SearchMessages,
1217        SearchMessages,
1218        SearchMessagesResult,
1219        SearchMessages
1220    ),
1221    (
1222        ForwardMessagesInput,
1223        ForwardMessages,
1224        ForwardMessages,
1225        ForwardMessagesResult,
1226        ForwardMessages
1227    ),
1228    (
1229        UpdateChatVisibilityInput,
1230        UpdateChatVisibility,
1231        UpdateChatVisibility,
1232        UpdateChatVisibilityResult,
1233        UpdateChatVisibility
1234    ),
1235    (
1236        PinMessageInput,
1237        PinMessage,
1238        PinMessage,
1239        PinMessageResult,
1240        PinMessage
1241    ),
1242    (
1243        UpdateChatInfoInput,
1244        UpdateChatInfo,
1245        UpdateChatInfo,
1246        UpdateChatInfoResult,
1247        UpdateChatInfo
1248    ),
1249    (ListBotsInput, ListBots, ListBots, ListBotsResult, ListBots),
1250    (
1251        RevealBotTokenInput,
1252        RevealBotToken,
1253        RevealBotToken,
1254        RevealBotTokenResult,
1255        RevealBotToken
1256    ),
1257    (
1258        MoveThreadInput,
1259        MoveThread,
1260        MoveThread,
1261        MoveThreadResult,
1262        MoveThread
1263    ),
1264    (
1265        RotateBotTokenInput,
1266        RotateBotToken,
1267        RotateBotToken,
1268        RotateBotTokenResult,
1269        RotateBotToken
1270    ),
1271    (
1272        UpdateBotProfileInput,
1273        UpdateBotProfile,
1274        UpdateBotProfile,
1275        UpdateBotProfileResult,
1276        UpdateBotProfile
1277    ),
1278    (
1279        GetMessagesInput,
1280        GetMessages,
1281        GetMessages,
1282        GetMessagesResult,
1283        GetMessages
1284    ),
1285    (
1286        UpdateDialogNotificationSettingsInput,
1287        UpdateDialogNotificationSettings,
1288        UpdateDialogNotificationSettings,
1289        UpdateDialogNotificationSettingsResult,
1290        UpdateDialogNotificationSettings
1291    ),
1292    (
1293        ReadMessagesInput,
1294        ReadMessages,
1295        ReadMessages,
1296        ReadMessagesResult,
1297        ReadMessages
1298    ),
1299    (
1300        UpdatePushNotificationDetailsInput,
1301        UpdatePushNotificationDetails,
1302        UpdatePushNotificationDetails,
1303        UpdatePushNotificationDetailsResult,
1304        UpdatePushNotificationDetails
1305    ),
1306    (
1307        CreateSubthreadInput,
1308        CreateSubthread,
1309        CreateSubthread,
1310        CreateSubthreadResult,
1311        CreateSubthread
1312    ),
1313    (
1314        GetBotCommandsInput,
1315        GetBotCommands,
1316        GetBotCommands,
1317        GetBotCommandsResult,
1318        GetBotCommands
1319    ),
1320    (
1321        SetBotCommandsInput,
1322        SetBotCommands,
1323        SetBotCommands,
1324        SetBotCommandsResult,
1325        SetBotCommands
1326    ),
1327    (
1328        GetPeerBotCommandsInput,
1329        GetPeerBotCommands,
1330        GetPeerBotCommands,
1331        GetPeerBotCommandsResult,
1332        GetPeerBotCommands
1333    ),
1334    (
1335        ShowInChatListInput,
1336        ShowInChatList,
1337        ShowInChatList,
1338        ShowInChatListResult,
1339        ShowInChatList
1340    ),
1341    (
1342        ReserveChatIdsInput,
1343        ReserveChatIds,
1344        ReserveChatIds,
1345        ReserveChatIdsResult,
1346        ReserveChatIds
1347    ),
1348    (
1349        InvokeMessageActionInput,
1350        InvokeMessageAction,
1351        InvokeMessageAction,
1352        InvokeMessageActionResult,
1353        InvokeMessageAction
1354    ),
1355    (
1356        AnswerMessageActionInput,
1357        AnswerMessageAction,
1358        AnswerMessageAction,
1359        AnswerMessageActionResult,
1360        AnswerMessageAction
1361    ),
1362    (
1363        RevokeSessionInput,
1364        RevokeSession,
1365        RevokeSession,
1366        RevokeSessionResult,
1367        RevokeSession
1368    ),
1369    (
1370        UpdateDialogOpenInput,
1371        UpdateDialogOpen,
1372        UpdateDialogOpen,
1373        UpdateDialogOpenResult,
1374        UpdateDialogOpen
1375    ),
1376    (
1377        UpdateDialogOrderInput,
1378        UpdateDialogOrder,
1379        UpdateDialogOrder,
1380        UpdateDialogOrderResult,
1381        UpdateDialogOrder
1382    ),
1383    (
1384        ClearChatHistoryInput,
1385        ClearChatHistory,
1386        ClearChatHistory,
1387        ClearChatHistoryResult,
1388        ClearChatHistory
1389    ),
1390    (
1391        DeleteBotInput,
1392        DeleteBot,
1393        DeleteBot,
1394        DeleteBotResult,
1395        DeleteBot
1396    ),
1397    (
1398        DeleteMessageAttachmentInput,
1399        DeleteMessageAttachment,
1400        DeleteMessageAttachment,
1401        DeleteMessageAttachmentResult,
1402        DeleteMessageAttachment
1403    ),
1404    (
1405        SetBotAvatarInput,
1406        SetBotAvatar,
1407        SetBotAvatar,
1408        SetBotAvatarResult,
1409        SetBotAvatar
1410    ),
1411    (
1412        ClearBotAvatarInput,
1413        ClearBotAvatar,
1414        ClearBotAvatar,
1415        ClearBotAvatarResult,
1416        ClearBotAvatar
1417    ),
1418    (
1419        GetBotPresenceInput,
1420        GetBotPresence,
1421        GetBotPresence,
1422        GetBotPresenceResult,
1423        GetBotPresence
1424    ),
1425    (
1426        SetBotPresenceStateInput,
1427        SetBotPresenceState,
1428        SetBotPresenceState,
1429        SetBotPresenceStateResult,
1430        SetBotPresenceState
1431    ),
1432    (
1433        UpdateDialogFollowModeInput,
1434        UpdateDialogFollowMode,
1435        UpdateDialogFollowMode,
1436        UpdateDialogFollowModeResult,
1437        UpdateDialogFollowMode
1438    ),
1439    (
1440        GetSessionsInput,
1441        GetSessions,
1442        GetSessions,
1443        GetSessionsResult,
1444        GetSessions
1445    ),
1446    (
1447        CheckUsernameInput,
1448        CheckUsername,
1449        CheckUsername,
1450        CheckUsernameResult,
1451        CheckUsername
1452    ),
1453    (
1454        ChangeUsernameInput,
1455        ChangeUsername,
1456        ChangeUsername,
1457        ChangeUsernameResult,
1458        ChangeUsername
1459    ),
1460    (
1461        UpdateProfileInput,
1462        UpdateProfile,
1463        UpdateProfile,
1464        UpdateProfileResult,
1465        UpdateProfile
1466    ),
1467    (
1468        GetSpaceUrlPreviewExclusionsInput,
1469        GetSpaceUrlPreviewExclusions,
1470        GetSpaceUrlPreviewExclusions,
1471        GetSpaceUrlPreviewExclusionsResult,
1472        GetSpaceUrlPreviewExclusions
1473    ),
1474    (
1475        AddSpaceUrlPreviewExclusionInput,
1476        AddSpaceUrlPreviewExclusion,
1477        AddSpaceUrlPreviewExclusion,
1478        AddSpaceUrlPreviewExclusionResult,
1479        AddSpaceUrlPreviewExclusion
1480    ),
1481    (
1482        RemoveSpaceUrlPreviewExclusionInput,
1483        RemoveSpaceUrlPreviewExclusion,
1484        RemoveSpaceUrlPreviewExclusion,
1485        RemoveSpaceUrlPreviewExclusionResult,
1486        RemoveSpaceUrlPreviewExclusion
1487    ),
1488    (
1489        GetUserGroupsInput,
1490        GetUserGroups,
1491        GetUserGroups,
1492        GetUserGroupsResult,
1493        GetUserGroups
1494    ),
1495    (
1496        CreateUserGroupInput,
1497        CreateUserGroup,
1498        CreateUserGroup,
1499        CreateUserGroupResult,
1500        CreateUserGroup
1501    ),
1502    (
1503        UpdateUserGroupInput,
1504        UpdateUserGroup,
1505        UpdateUserGroup,
1506        UpdateUserGroupResult,
1507        UpdateUserGroup
1508    ),
1509    (
1510        DeleteUserGroupInput,
1511        DeleteUserGroup,
1512        DeleteUserGroup,
1513        DeleteUserGroupResult,
1514        DeleteUserGroup
1515    ),
1516    (
1517        GetSpaceSettingsInput,
1518        GetSpaceSettings,
1519        GetSpaceSettings,
1520        GetSpaceSettingsResult,
1521        GetSpaceSettings
1522    ),
1523    (
1524        ToggleSpaceGridInput,
1525        ToggleSpaceGrid,
1526        ToggleSpaceGrid,
1527        ToggleSpaceGridResult,
1528        ToggleSpaceGrid
1529    ),
1530    (
1531        GetThreadReferencesInput,
1532        GetThreadReferences,
1533        GetThreadReferences,
1534        GetThreadReferencesResult,
1535        GetThreadReferences
1536    ),
1537    (
1538        GetThreadSubthreadsInput,
1539        GetThreadSubthreads,
1540        GetThreadSubthreads,
1541        GetThreadSubthreadsResult,
1542        GetThreadSubthreads
1543    ),
1544);
1545
1546fn connection_init_for_token(token: &str, identity: &ClientIdentity) -> proto::ConnectionInit {
1547    proto::ConnectionInit {
1548        token: token.to_string(),
1549        build_number: None,
1550        layer: None,
1551        client_version: Some(identity.client_version().to_string()),
1552        os_version: client_info::current_os_version(),
1553    }
1554}
1555
1556fn rpc_error_code_name(error_code: i32) -> String {
1557    proto::rpc_error::Code::try_from(error_code)
1558        .map(|code| code.as_str_name())
1559        .unwrap_or("UNKNOWN")
1560        .to_string()
1561}
1562
1563fn rpc_error_from_proto(error: proto::RpcError) -> RealtimeError {
1564    let error_name = rpc_error_code_name(error.error_code);
1565    let friendly = format_rpc_error(error.error_code, &error_name, &error.message, error.code);
1566    RealtimeError::RpcError {
1567        code: error.code,
1568        error_code: error.error_code,
1569        error_name,
1570        message: error.message,
1571        friendly,
1572    }
1573}
1574
1575fn format_rpc_error(error_code: i32, error_name: &str, message: &str, status_code: i32) -> String {
1576    let label = match error_name {
1577        "UNKNOWN" => "Unknown RPC error",
1578        "BAD_REQUEST" => "Bad request",
1579        "UNAUTHENTICATED" => "Not authenticated",
1580        "RATE_LIMIT" => "Rate limited",
1581        "INTERNAL_ERROR" => "Internal server error",
1582        "PEER_ID_INVALID" => "Invalid peer (chat/user id)",
1583        "MESSAGE_ID_INVALID" => "Invalid message id",
1584        "USER_ID_INVALID" => "Invalid user id",
1585        "USER_ALREADY_MEMBER" => "User already in chat/space",
1586        "SPACE_ID_INVALID" => "Invalid space id",
1587        "CHAT_ID_INVALID" => "Invalid chat id",
1588        "EMAIL_INVALID" => "Invalid email address",
1589        "PHONE_NUMBER_INVALID" => "Invalid phone number",
1590        "SPACE_ADMIN_REQUIRED" => "Space admin required",
1591        "SPACE_OWNER_REQUIRED" => "Space owner required",
1592        "USERNAME_INVALID" => "Invalid username",
1593        "USERNAME_TAKEN" => "Username already taken",
1594        "FIRST_NAME_INVALID" => "Invalid first name",
1595        _ => "Unknown RPC error",
1596    };
1597
1598    let mut formatted = String::from(label);
1599    if error_name == "UNKNOWN" && error_code != 0 {
1600        formatted.push_str(&format!(" {error_code}"));
1601    }
1602    if !message.is_empty() && !message.eq_ignore_ascii_case(label) {
1603        formatted.push_str(": ");
1604        formatted.push_str(message);
1605    }
1606    if status_code != 0 {
1607        formatted.push_str(&format!(" (HTTP {status_code})"));
1608    }
1609    formatted
1610}
1611
1612fn connection_error_from_proto(error: proto::ConnectionError) -> RealtimeError {
1613    let reason = error.reason;
1614    let reason_name = proto::connection_error::Reason::try_from(reason)
1615        .map(|reason| reason.as_str_name())
1616        .unwrap_or("UNKNOWN")
1617        .to_string();
1618    let friendly = format_connection_error(reason, &reason_name);
1619    RealtimeError::ConnectionError {
1620        reason,
1621        reason_name,
1622        friendly,
1623    }
1624}
1625
1626fn format_connection_error(reason: i32, reason_name: &str) -> String {
1627    match reason_name {
1628        "UNAUTHORIZED" => "Realtime connection unauthorized".to_string(),
1629        "INVALID_AUTH" => "Realtime auth token is invalid".to_string(),
1630        "SESSION_REVOKED" => "Realtime session was revoked".to_string(),
1631        "REASON_UNSPECIFIED" => "Realtime connection rejected".to_string(),
1632        _ => format!("Realtime connection rejected: unknown reason {reason}"),
1633    }
1634}
1635
1636fn normalize_realtime_url(url: impl Into<String>) -> Result<Url, RealtimeError> {
1637    let original = url.into();
1638    let normalized = original.trim().to_string();
1639    if normalized.is_empty() {
1640        return Err(RealtimeError::InvalidUrl {
1641            url: original,
1642            message: "realtime URL cannot be empty".to_string(),
1643        });
1644    }
1645
1646    let parsed = Url::parse(&normalized).map_err(|err| RealtimeError::InvalidUrl {
1647        url: normalized.clone(),
1648        message: err.to_string(),
1649    })?;
1650
1651    if !matches!(parsed.scheme(), "ws" | "wss") {
1652        return Err(RealtimeError::InvalidUrl {
1653            url: normalized,
1654            message: "scheme must be ws or wss".to_string(),
1655        });
1656    }
1657
1658    if parsed.host_str().is_none() {
1659        return Err(RealtimeError::InvalidUrl {
1660            url: normalized,
1661            message: "host is required".to_string(),
1662        });
1663    }
1664
1665    if !parsed.username().is_empty() || parsed.password().is_some() {
1666        return Err(RealtimeError::InvalidUrl {
1667            url: normalized,
1668            message: "credentials are not valid in the realtime URL".to_string(),
1669        });
1670    }
1671
1672    if parsed.fragment().is_some() {
1673        return Err(RealtimeError::InvalidUrl {
1674            url: normalized,
1675            message: "fragments are not valid in the realtime URL".to_string(),
1676        });
1677    }
1678
1679    Ok(parsed)
1680}
1681
1682fn realtime_url_for_log(url: &Url) -> String {
1683    let host = url.host_str().unwrap_or("<missing-host>");
1684    let port = url
1685        .port()
1686        .map(|port| format!(":{port}"))
1687        .unwrap_or_default();
1688    format!("{}://{}{}{}", url.scheme(), host, port, url.path())
1689}
1690
1691fn realtime_url_for_debug(raw_url: &str) -> String {
1692    Url::parse(raw_url.trim())
1693        .map(|url| realtime_url_for_log(&url))
1694        .unwrap_or_else(|_| "<invalid>".to_string())
1695}
1696
1697fn realtime_header_value(field: &'static str, value: &str) -> Result<HeaderValue, RealtimeError> {
1698    HeaderValue::from_str(value).map_err(|_| RealtimeError::InvalidHeaderValue { field })
1699}
1700
1701fn realtime_event_kind(event: &RealtimeEvent) -> &'static str {
1702    match event {
1703        RealtimeEvent::Updates(_) => "updates",
1704        RealtimeEvent::Ack { .. } => "ack",
1705        RealtimeEvent::Pong { .. } => "pong",
1706    }
1707}
1708
1709async fn with_optional_timeout<T, E, Fut>(
1710    operation: &'static str,
1711    timeout: Option<Duration>,
1712    future: Fut,
1713) -> Result<T, RealtimeError>
1714where
1715    Fut: Future<Output = Result<T, E>>,
1716    RealtimeError: From<E>,
1717{
1718    match timeout {
1719        Some(timeout) => tokio::time::timeout(timeout, future)
1720            .await
1721            .map_err(|_| RealtimeError::Timeout { operation, timeout })?
1722            .map_err(RealtimeError::from),
1723        None => future.await.map_err(RealtimeError::from),
1724    }
1725}
1726
1727struct IdGenerator {
1728    last_timestamp: u64,
1729    sequence: u32,
1730}
1731
1732impl IdGenerator {
1733    fn new() -> Self {
1734        Self {
1735            last_timestamp: 0,
1736            sequence: 0,
1737        }
1738    }
1739
1740    fn next_id(&mut self) -> u64 {
1741        let timestamp = current_epoch_seconds().saturating_sub(EPOCH_SECONDS);
1742        if timestamp == self.last_timestamp {
1743            self.sequence = self.sequence.wrapping_add(1);
1744        } else {
1745            self.sequence = 0;
1746            self.last_timestamp = timestamp;
1747        }
1748
1749        (timestamp << 32) | self.sequence as u64
1750    }
1751}
1752
1753const EPOCH_SECONDS: u64 = 1_735_689_600; // 2025-01-01T00:00:00Z
1754
1755fn current_epoch_seconds() -> u64 {
1756    use std::time::{SystemTime, UNIX_EPOCH};
1757    SystemTime::now()
1758        .duration_since(UNIX_EPOCH)
1759        .unwrap_or_default()
1760        .as_secs()
1761}
1762
1763#[cfg(test)]
1764mod tests {
1765    use super::*;
1766    use tokio::net::{TcpListener, TcpStream};
1767    use tokio_tungstenite::WebSocketStream;
1768    use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
1769    use tokio_tungstenite::{accept_async, accept_hdr_async};
1770
1771    #[test]
1772    fn rpc_error_code_name_uses_stable_proto_name() {
1773        assert_eq!(
1774            rpc_error_code_name(proto::rpc_error::Code::PeerIdInvalid as i32),
1775            "PEER_ID_INVALID"
1776        );
1777        assert_eq!(rpc_error_code_name(999), "UNKNOWN");
1778    }
1779
1780    #[test]
1781    fn rpc_error_formatter_covers_new_profile_codes() {
1782        assert_eq!(
1783            format_rpc_error(
1784                proto::rpc_error::Code::UsernameInvalid as i32,
1785                "USERNAME_INVALID",
1786                "",
1787                400
1788            ),
1789            "Invalid username (HTTP 400)"
1790        );
1791        assert_eq!(
1792            format_rpc_error(
1793                proto::rpc_error::Code::UsernameTaken as i32,
1794                "USERNAME_TAKEN",
1795                "handle exists",
1796                409
1797            ),
1798            "Username already taken: handle exists (HTTP 409)"
1799        );
1800        assert_eq!(
1801            format_rpc_error(
1802                proto::rpc_error::Code::FirstNameInvalid as i32,
1803                "FIRST_NAME_INVALID",
1804                "",
1805                400
1806            ),
1807            "Invalid first name (HTTP 400)"
1808        );
1809    }
1810
1811    #[test]
1812    fn connection_error_preserves_proto_reason() {
1813        let err = connection_error_from_proto(proto::ConnectionError {
1814            reason: proto::connection_error::Reason::InvalidAuth as i32,
1815        });
1816
1817        match err {
1818            RealtimeError::ConnectionError {
1819                reason,
1820                reason_name,
1821                friendly,
1822            } => {
1823                assert_eq!(reason, 2);
1824                assert_eq!(reason_name, "INVALID_AUTH");
1825                assert_eq!(friendly, "Realtime auth token is invalid");
1826            }
1827            other => panic!("expected connection error, got {other:?}"),
1828        }
1829    }
1830
1831    #[test]
1832    fn unknown_connection_error_reason_is_preserved() {
1833        let err = connection_error_from_proto(proto::ConnectionError { reason: 999 });
1834
1835        match err {
1836            RealtimeError::ConnectionError {
1837                reason,
1838                reason_name,
1839                friendly,
1840            } => {
1841                assert_eq!(reason, 999);
1842                assert_eq!(reason_name, "UNKNOWN");
1843                assert_eq!(friendly, "Realtime connection rejected: unknown reason 999");
1844            }
1845            other => panic!("expected connection error, got {other:?}"),
1846        }
1847    }
1848
1849    #[test]
1850    fn connection_init_uses_custom_client_identity() {
1851        let init =
1852            connection_init_for_token("token-1", &ClientIdentity::new("integration-test", "9.9.9"));
1853
1854        assert_eq!(init.token, "token-1");
1855        assert_eq!(init.client_version.as_deref(), Some("9.9.9"));
1856        assert!(init.build_number.is_none());
1857        assert!(init.layer.is_none());
1858    }
1859
1860    #[tokio::test]
1861    #[allow(clippy::result_large_err)]
1862    async fn realtime_client_connects_and_calls_get_me_against_local_server() {
1863        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1864        let addr = listener.local_addr().unwrap();
1865        let server = tokio::spawn(async move {
1866            let (stream, _) = listener.accept().await.unwrap();
1867            let mut ws = accept_hdr_async(stream, |request: &Request, response: Response| {
1868                assert_eq!(
1869                    request
1870                        .headers()
1871                        .get(client_info::CLIENT_TYPE_HEADER)
1872                        .and_then(|value| value.to_str().ok()),
1873                    Some("transport-test")
1874                );
1875                assert_eq!(
1876                    request
1877                        .headers()
1878                        .get(client_info::CLIENT_VERSION_HEADER)
1879                        .and_then(|value| value.to_str().ok()),
1880                    Some("1.2.3")
1881                );
1882                Ok(response)
1883            })
1884            .await
1885            .unwrap();
1886
1887            let init = read_test_client_message(&mut ws).await;
1888            match init.body {
1889                Some(proto::client_message::Body::ConnectionInit(init)) => {
1890                    assert_eq!(init.token, "token-1");
1891                    assert_eq!(init.client_version.as_deref(), Some("1.2.3"));
1892                    assert!(init.os_version.is_some());
1893                }
1894                other => panic!("expected connection init, got {other:?}"),
1895            }
1896            send_test_server_message(
1897                &mut ws,
1898                proto::ServerProtocolMessage {
1899                    id: 1,
1900                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1901                        proto::ConnectionOpen {},
1902                    )),
1903                },
1904            )
1905            .await;
1906
1907            let rpc = read_test_client_message(&mut ws).await;
1908            match &rpc.body {
1909                Some(proto::client_message::Body::RpcCall(call)) => {
1910                    assert_eq!(call.method, proto::Method::GetMe as i32);
1911                    assert!(matches!(call.input, Some(proto::rpc_call::Input::GetMe(_))));
1912                }
1913                other => panic!("expected getMe rpc call, got {other:?}"),
1914            }
1915            send_test_server_message(
1916                &mut ws,
1917                proto::ServerProtocolMessage {
1918                    id: 2,
1919                    body: Some(proto::server_protocol_message::Body::RpcResult(
1920                        proto::RpcResult {
1921                            req_msg_id: rpc.id,
1922                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
1923                                user: Some(proto::User {
1924                                    id: 42,
1925                                    first_name: Some("Ada".to_string()),
1926                                    ..Default::default()
1927                                }),
1928                            })),
1929                        },
1930                    )),
1931                },
1932            )
1933            .await;
1934        });
1935
1936        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1937            .identity(ClientIdentity::new("transport-test", "1.2.3"))
1938            .without_connect_timeout()
1939            .without_rpc_timeout()
1940            .connect()
1941            .await
1942            .unwrap();
1943        let result = client.call(proto::GetMeInput {}).await.unwrap();
1944
1945        assert_eq!(result.user.unwrap().id, 42);
1946        server.await.unwrap();
1947    }
1948
1949    #[tokio::test]
1950    async fn realtime_client_receives_update_events_and_acks_them() {
1951        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1952        let addr = listener.local_addr().unwrap();
1953        let server = tokio::spawn(async move {
1954            let (stream, _) = listener.accept().await.unwrap();
1955            let mut ws = accept_async(stream).await.unwrap();
1956
1957            let init = read_test_client_message(&mut ws).await;
1958            assert!(matches!(
1959                init.body,
1960                Some(proto::client_message::Body::ConnectionInit(_))
1961            ));
1962            send_test_server_message(
1963                &mut ws,
1964                proto::ServerProtocolMessage {
1965                    id: 1,
1966                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1967                        proto::ConnectionOpen {},
1968                    )),
1969                },
1970            )
1971            .await;
1972            send_test_server_message(&mut ws, test_update_server_message(9, 77)).await;
1973
1974            let ack = read_test_client_message(&mut ws).await;
1975            match ack.body {
1976                Some(proto::client_message::Body::Ack(ack)) => {
1977                    assert_eq!(ack.msg_id, 9);
1978                }
1979                other => panic!("expected ack for pushed update, got {other:?}"),
1980            }
1981        });
1982
1983        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1984            .without_connect_timeout()
1985            .without_rpc_timeout()
1986            .connect()
1987            .await
1988            .unwrap();
1989        let event = client.next_event().await.unwrap();
1990
1991        match event {
1992            RealtimeEvent::Updates(updates) => {
1993                assert_eq!(updates.len(), 1);
1994                match updates[0].update.as_ref() {
1995                    Some(proto::update::Update::NewMessage(update)) => {
1996                        assert_eq!(update.message.as_ref().unwrap().id, 77);
1997                    }
1998                    other => panic!("expected new message update, got {other:?}"),
1999                }
2000            }
2001            other => panic!("expected updates event, got {other:?}"),
2002        }
2003        server.await.unwrap();
2004    }
2005
2006    #[tokio::test]
2007    async fn realtime_rpc_wait_acks_pushed_updates_before_matching_result() {
2008        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2009        let addr = listener.local_addr().unwrap();
2010        let server = tokio::spawn(async move {
2011            let (stream, _) = listener.accept().await.unwrap();
2012            let mut ws = accept_async(stream).await.unwrap();
2013
2014            let init = read_test_client_message(&mut ws).await;
2015            assert!(matches!(
2016                init.body,
2017                Some(proto::client_message::Body::ConnectionInit(_))
2018            ));
2019            send_test_server_message(
2020                &mut ws,
2021                proto::ServerProtocolMessage {
2022                    id: 1,
2023                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2024                        proto::ConnectionOpen {},
2025                    )),
2026                },
2027            )
2028            .await;
2029
2030            let rpc = read_test_client_message(&mut ws).await;
2031            send_test_server_message(&mut ws, test_update_server_message(10, 88)).await;
2032
2033            let ack = read_test_client_message(&mut ws).await;
2034            match ack.body {
2035                Some(proto::client_message::Body::Ack(ack)) => {
2036                    assert_eq!(ack.msg_id, 10);
2037                }
2038                other => panic!("expected ack while waiting for rpc result, got {other:?}"),
2039            }
2040
2041            send_test_server_message(
2042                &mut ws,
2043                proto::ServerProtocolMessage {
2044                    id: 2,
2045                    body: Some(proto::server_protocol_message::Body::RpcResult(
2046                        proto::RpcResult {
2047                            req_msg_id: rpc.id,
2048                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2049                                user: Some(proto::User {
2050                                    id: 42,
2051                                    ..Default::default()
2052                                }),
2053                            })),
2054                        },
2055                    )),
2056                },
2057            )
2058            .await;
2059        });
2060
2061        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2062            .without_connect_timeout()
2063            .without_rpc_timeout()
2064            .connect()
2065            .await
2066            .unwrap();
2067        let result = client.call(proto::GetMeInput {}).await.unwrap();
2068
2069        assert_eq!(result.user.unwrap().id, 42);
2070        server.await.unwrap();
2071    }
2072
2073    #[tokio::test]
2074    async fn realtime_session_routes_pushed_updates_during_rpc() {
2075        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2076        let addr = listener.local_addr().unwrap();
2077        let server = tokio::spawn(async move {
2078            let (stream, _) = listener.accept().await.unwrap();
2079            let mut ws = accept_async(stream).await.unwrap();
2080
2081            let init = read_test_client_message(&mut ws).await;
2082            assert!(matches!(
2083                init.body,
2084                Some(proto::client_message::Body::ConnectionInit(_))
2085            ));
2086            send_test_server_message(
2087                &mut ws,
2088                proto::ServerProtocolMessage {
2089                    id: 1,
2090                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2091                        proto::ConnectionOpen {},
2092                    )),
2093                },
2094            )
2095            .await;
2096
2097            let rpc = read_test_client_message(&mut ws).await;
2098            send_test_server_message(&mut ws, test_update_server_message(10, 88)).await;
2099            let ack = read_test_client_message(&mut ws).await;
2100            assert!(matches!(
2101                ack.body,
2102                Some(proto::client_message::Body::Ack(proto::Ack { msg_id: 10 }))
2103            ));
2104            send_test_server_message(
2105                &mut ws,
2106                proto::ServerProtocolMessage {
2107                    id: 2,
2108                    body: Some(proto::server_protocol_message::Body::RpcResult(
2109                        proto::RpcResult {
2110                            req_msg_id: rpc.id,
2111                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2112                                user: Some(proto::User {
2113                                    id: 42,
2114                                    ..Default::default()
2115                                }),
2116                            })),
2117                        },
2118                    )),
2119                },
2120            )
2121            .await;
2122        });
2123
2124        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2125            .without_connect_timeout()
2126            .without_rpc_timeout()
2127            .connect_session()
2128            .await
2129            .unwrap();
2130        let mut events = session.subscribe();
2131        let (result, event) = tokio::join!(session.call(proto::GetMeInput {}), events.recv());
2132
2133        assert_eq!(result.unwrap().user.unwrap().id, 42);
2134        match event.unwrap() {
2135            RealtimeEvent::Updates(updates) => match updates[0].update.as_ref() {
2136                Some(proto::update::Update::NewMessage(update)) => {
2137                    assert_eq!(update.message.as_ref().unwrap().id, 88);
2138                }
2139                other => panic!("expected new message update, got {other:?}"),
2140            },
2141            other => panic!("expected updates event, got {other:?}"),
2142        }
2143        server.await.unwrap();
2144    }
2145
2146    #[tokio::test]
2147    async fn realtime_session_matches_concurrent_rpc_results_by_request_id() {
2148        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2149        let addr = listener.local_addr().unwrap();
2150        let server = tokio::spawn(async move {
2151            let (stream, _) = listener.accept().await.unwrap();
2152            let mut ws = accept_async(stream).await.unwrap();
2153            let _ = read_test_client_message(&mut ws).await;
2154            send_test_server_message(
2155                &mut ws,
2156                proto::ServerProtocolMessage {
2157                    id: 1,
2158                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2159                        proto::ConnectionOpen {},
2160                    )),
2161                },
2162            )
2163            .await;
2164
2165            let first = read_test_client_message(&mut ws).await;
2166            let second = read_test_client_message(&mut ws).await;
2167            for (id, request, user_id) in [(2, &second, 2), (3, &first, 1)] {
2168                send_test_server_message(
2169                    &mut ws,
2170                    proto::ServerProtocolMessage {
2171                        id,
2172                        body: Some(proto::server_protocol_message::Body::RpcResult(
2173                            proto::RpcResult {
2174                                req_msg_id: request.id,
2175                                result: Some(proto::rpc_result::Result::GetMe(
2176                                    proto::GetMeResult {
2177                                        user: Some(proto::User {
2178                                            id: user_id,
2179                                            ..Default::default()
2180                                        }),
2181                                    },
2182                                )),
2183                            },
2184                        )),
2185                    },
2186                )
2187                .await;
2188            }
2189        });
2190
2191        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2192            .without_connect_timeout()
2193            .without_rpc_timeout()
2194            .connect_session()
2195            .await
2196            .unwrap();
2197        let (first, second) = tokio::join!(
2198            session.call(proto::GetMeInput {}),
2199            session.call(proto::GetMeInput {})
2200        );
2201
2202        assert_eq!(first.unwrap().user.unwrap().id, 1);
2203        assert_eq!(second.unwrap().user.unwrap().id, 2);
2204        server.await.unwrap();
2205    }
2206
2207    #[tokio::test]
2208    async fn realtime_session_bounds_in_flight_rpcs() {
2209        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2210        let addr = listener.local_addr().unwrap();
2211        let server = tokio::spawn(async move {
2212            let (stream, _) = listener.accept().await.unwrap();
2213            let mut ws = accept_async(stream).await.unwrap();
2214            let _ = read_test_client_message(&mut ws).await;
2215            send_test_server_message(
2216                &mut ws,
2217                proto::ServerProtocolMessage {
2218                    id: 1,
2219                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2220                        proto::ConnectionOpen {},
2221                    )),
2222                },
2223            )
2224            .await;
2225
2226            let first = read_test_client_message(&mut ws).await;
2227            assert!(
2228                tokio::time::timeout(Duration::from_millis(50), read_test_client_message(&mut ws),)
2229                    .await
2230                    .is_err(),
2231                "session issued a second RPC above its configured in-flight limit"
2232            );
2233            send_test_server_message(&mut ws, get_me_result_message(2, first.id, 1)).await;
2234            let second = read_test_client_message(&mut ws).await;
2235            send_test_server_message(&mut ws, get_me_result_message(3, second.id, 2)).await;
2236        });
2237
2238        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2239            .without_connect_timeout()
2240            .without_rpc_timeout()
2241            .max_in_flight_rpcs(1)
2242            .connect_session()
2243            .await
2244            .unwrap();
2245        let (first, second) = tokio::join!(
2246            session.call(proto::GetMeInput {}),
2247            session.call(proto::GetMeInput {})
2248        );
2249
2250        assert_eq!(first.unwrap().user.unwrap().id, 1);
2251        assert_eq!(second.unwrap().user.unwrap().id, 2);
2252        server.await.unwrap();
2253    }
2254
2255    #[tokio::test]
2256    async fn realtime_session_drops_timed_out_rpc_and_remains_usable() {
2257        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2258        let addr = listener.local_addr().unwrap();
2259        let server = tokio::spawn(async move {
2260            let (stream, _) = listener.accept().await.unwrap();
2261            let mut ws = accept_async(stream).await.unwrap();
2262            let _ = read_test_client_message(&mut ws).await;
2263            send_test_server_message(
2264                &mut ws,
2265                proto::ServerProtocolMessage {
2266                    id: 1,
2267                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2268                        proto::ConnectionOpen {},
2269                    )),
2270                },
2271            )
2272            .await;
2273
2274            let timed_out = read_test_client_message(&mut ws).await;
2275            tokio::time::sleep(Duration::from_millis(40)).await;
2276            send_test_server_message(
2277                &mut ws,
2278                proto::ServerProtocolMessage {
2279                    id: 2,
2280                    body: Some(proto::server_protocol_message::Body::RpcResult(
2281                        proto::RpcResult {
2282                            req_msg_id: timed_out.id,
2283                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2284                                user: Some(proto::User {
2285                                    id: 1,
2286                                    ..Default::default()
2287                                }),
2288                            })),
2289                        },
2290                    )),
2291                },
2292            )
2293            .await;
2294
2295            let recovered = read_test_client_message(&mut ws).await;
2296            send_test_server_message(
2297                &mut ws,
2298                proto::ServerProtocolMessage {
2299                    id: 3,
2300                    body: Some(proto::server_protocol_message::Body::RpcResult(
2301                        proto::RpcResult {
2302                            req_msg_id: recovered.id,
2303                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2304                                user: Some(proto::User {
2305                                    id: 2,
2306                                    ..Default::default()
2307                                }),
2308                            })),
2309                        },
2310                    )),
2311                },
2312            )
2313            .await;
2314        });
2315
2316        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2317            .without_connect_timeout()
2318            .rpc_timeout(Duration::from_millis(25))
2319            .connect_session()
2320            .await
2321            .unwrap();
2322
2323        assert!(matches!(
2324            session.call(proto::GetMeInput {}).await,
2325            Err(RealtimeError::Timeout {
2326                operation: "rpc",
2327                ..
2328            })
2329        ));
2330        let recovered = session.call(proto::GetMeInput {}).await.unwrap();
2331        assert_eq!(recovered.user.unwrap().id, 2);
2332        server.await.unwrap();
2333    }
2334
2335    #[tokio::test]
2336    async fn realtime_session_closes_when_heartbeat_pong_is_missing() {
2337        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2338        let addr = listener.local_addr().unwrap();
2339        let server = tokio::spawn(async move {
2340            let (stream, _) = listener.accept().await.unwrap();
2341            let mut ws = accept_async(stream).await.unwrap();
2342            let _ = read_test_client_message(&mut ws).await;
2343            send_test_server_message(
2344                &mut ws,
2345                proto::ServerProtocolMessage {
2346                    id: 1,
2347                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2348                        proto::ConnectionOpen {},
2349                    )),
2350                },
2351            )
2352            .await;
2353            let ping = read_test_client_message(&mut ws).await;
2354            assert!(matches!(
2355                ping.body,
2356                Some(proto::client_message::Body::Ping(_))
2357            ));
2358            tokio::time::sleep(Duration::from_millis(100)).await;
2359        });
2360
2361        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2362            .without_connect_timeout()
2363            .heartbeat(Duration::from_millis(20), Duration::from_millis(30))
2364            .connect_session()
2365            .await
2366            .unwrap();
2367        let mut events = session.subscribe();
2368
2369        let error = tokio::time::timeout(Duration::from_millis(150), events.recv())
2370            .await
2371            .expect("heartbeat deadline should close the session")
2372            .unwrap_err();
2373        assert!(matches!(error, RealtimeError::ConnectionClosed));
2374        assert!(session.is_closed());
2375        server.await.unwrap();
2376    }
2377
2378    #[test]
2379    fn realtime_url_validation_accepts_ws_urls() {
2380        let url = normalize_realtime_url(" wss://api.inline.chat/realtime?edge=iad ").unwrap();
2381        assert_eq!(url.as_str(), "wss://api.inline.chat/realtime?edge=iad");
2382    }
2383
2384    #[test]
2385    fn realtime_url_validation_rejects_invalid_urls() {
2386        let err = normalize_realtime_url("inline.test").unwrap_err();
2387        match err {
2388            RealtimeError::InvalidUrl { url, message } => {
2389                assert_eq!(url, "inline.test");
2390                assert!(message.contains("relative URL without a base"));
2391            }
2392            other => panic!("expected invalid URL, got {other:?}"),
2393        }
2394
2395        let err = normalize_realtime_url("https://api.inline.chat/realtime").unwrap_err();
2396        match err {
2397            RealtimeError::InvalidUrl { message, .. } => {
2398                assert_eq!(message, "scheme must be ws or wss");
2399            }
2400            other => panic!("expected invalid URL, got {other:?}"),
2401        }
2402
2403        let err = normalize_realtime_url("wss://user:secret@api.inline.chat/realtime").unwrap_err();
2404        match &err {
2405            RealtimeError::InvalidUrl { message, .. } => {
2406                assert_eq!(message, "credentials are not valid in the realtime URL");
2407                assert!(!err.to_string().contains("secret"));
2408            }
2409            other => panic!("expected invalid URL, got {other:?}"),
2410        }
2411
2412        let err = normalize_realtime_url("wss://api.inline.chat/realtime#token").unwrap_err();
2413        match err {
2414            RealtimeError::InvalidUrl { message, .. } => {
2415                assert_eq!(message, "fragments are not valid in the realtime URL");
2416            }
2417            other => panic!("expected invalid URL, got {other:?}"),
2418        }
2419    }
2420
2421    #[test]
2422    fn realtime_invalid_url_debug_redacts_unsafe_url_parts() {
2423        let err = normalize_realtime_url(
2424            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret#frag",
2425        )
2426        .unwrap_err();
2427        let debug = format!("{err:?}");
2428
2429        assert!(debug.contains("wss://api.inline.chat/realtime"));
2430        assert!(!debug.contains("url-secret"));
2431        assert!(!debug.contains("query-secret"));
2432    }
2433
2434    #[test]
2435    fn realtime_url_for_log_omits_query_and_fragment() {
2436        let url = Url::parse("wss://api.inline.chat/realtime?token=secret#frag").unwrap();
2437        assert_eq!(realtime_url_for_log(&url), "wss://api.inline.chat/realtime");
2438    }
2439
2440    #[test]
2441    fn realtime_header_value_rejects_invalid_values() {
2442        let err = realtime_header_value("user_agent", "bad\nvalue").unwrap_err();
2443        match err {
2444            RealtimeError::InvalidHeaderValue { field } => {
2445                assert_eq!(field, "user_agent");
2446            }
2447            other => panic!("expected invalid header value, got {other:?}"),
2448        }
2449    }
2450
2451    #[test]
2452    fn realtime_builder_keeps_custom_identity_and_default_timeouts() {
2453        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
2454            .identity(ClientIdentity::new("agent", "0.2.0"));
2455
2456        assert_eq!(builder.url, "wss://api.inline.chat/realtime");
2457        assert_eq!(builder.token, "token-1");
2458        assert_eq!(builder.identity.client_type(), "agent");
2459        assert_eq!(builder.identity.client_version(), "0.2.0");
2460        assert_eq!(builder.connect_timeout, Some(DEFAULT_CONNECT_TIMEOUT));
2461        assert_eq!(builder.rpc_timeout, Some(DEFAULT_RPC_TIMEOUT));
2462        assert_eq!(builder.heartbeat_interval, Some(DEFAULT_HEARTBEAT_INTERVAL));
2463        assert_eq!(builder.heartbeat_timeout, DEFAULT_HEARTBEAT_TIMEOUT);
2464        assert_eq!(
2465            builder.max_in_flight_rpcs,
2466            DEFAULT_SESSION_MAX_IN_FLIGHT_RPCS
2467        );
2468    }
2469
2470    #[test]
2471    fn realtime_builder_debug_redacts_token() {
2472        let builder = RealtimeClient::builder(
2473            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret",
2474            "secret-token-1",
2475        );
2476        let debug = format!("{builder:?}");
2477
2478        assert!(debug.contains("RealtimeClientBuilder"));
2479        assert!(debug.contains("wss://api.inline.chat/realtime"));
2480        assert!(debug.contains("<redacted>"));
2481        assert!(!debug.contains("secret-token-1"));
2482        assert!(!debug.contains("url-secret"));
2483        assert!(!debug.contains("query-secret"));
2484    }
2485
2486    #[test]
2487    fn realtime_builder_can_override_or_disable_timeouts() {
2488        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
2489            .connect_timeout(Duration::from_secs(5))
2490            .rpc_timeout(Duration::from_secs(10))
2491            .max_in_flight_rpcs(7);
2492
2493        assert_eq!(builder.connect_timeout, Some(Duration::from_secs(5)));
2494        assert_eq!(builder.rpc_timeout, Some(Duration::from_secs(10)));
2495        assert_eq!(builder.max_in_flight_rpcs, 7);
2496
2497        let builder = builder
2498            .without_connect_timeout()
2499            .without_rpc_timeout()
2500            .without_heartbeat();
2501        assert_eq!(builder.connect_timeout, None);
2502        assert_eq!(builder.rpc_timeout, None);
2503        assert_eq!(builder.heartbeat_interval, None);
2504    }
2505
2506    #[test]
2507    fn typed_rpc_request_maps_method_input_and_result() {
2508        assert_eq!(
2509            <proto::GetChatsInput as RpcRequest>::METHOD,
2510            proto::Method::GetChats
2511        );
2512
2513        let input = proto::GetChatsInput {};
2514        match input.into_rpc_input() {
2515            proto::rpc_call::Input::GetChats(_) => {}
2516            other => panic!("expected GetChats input, got {other:?}"),
2517        }
2518
2519        let response = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
2520            proto::rpc_result::Result::GetChats(proto::GetChatsResult::default()),
2521        )
2522        .unwrap();
2523        assert!(response.dialogs.is_empty());
2524    }
2525
2526    #[test]
2527    fn typed_rpc_request_rejects_unexpected_result_variant() {
2528        let err = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
2529            proto::rpc_result::Result::GetMe(proto::GetMeResult::default()),
2530        )
2531        .unwrap_err();
2532
2533        match err {
2534            RealtimeError::UnexpectedResult {
2535                method,
2536                expected,
2537                actual,
2538            } => {
2539                assert_eq!(method, "GET_CHATS");
2540                assert_eq!(expected, "GetChats");
2541                assert_eq!(actual, "GetMe");
2542            }
2543            other => panic!("expected unexpected result error, got {other:?}"),
2544        }
2545    }
2546
2547    #[tokio::test]
2548    async fn optional_timeout_reports_elapsed_operation() {
2549        let err = with_optional_timeout("test", Some(Duration::from_millis(1)), async {
2550            tokio::time::sleep(Duration::from_millis(20)).await;
2551            Ok::<(), RealtimeError>(())
2552        })
2553        .await
2554        .unwrap_err();
2555
2556        match err {
2557            RealtimeError::Timeout { operation, timeout } => {
2558                assert_eq!(operation, "test");
2559                assert_eq!(timeout, Duration::from_millis(1));
2560            }
2561            other => panic!("expected timeout, got {other:?}"),
2562        }
2563    }
2564
2565    #[tokio::test]
2566    async fn optional_timeout_can_be_disabled() {
2567        with_optional_timeout("test", None, async { Ok::<_, RealtimeError>("done") })
2568            .await
2569            .unwrap();
2570    }
2571
2572    async fn read_test_client_message(ws: &mut WebSocketStream<TcpStream>) -> proto::ClientMessage {
2573        loop {
2574            match ws.next().await.unwrap().unwrap() {
2575                WsMessage::Binary(bytes) => {
2576                    return proto::ClientMessage::decode(&*bytes).unwrap();
2577                }
2578                WsMessage::Ping(_) | WsMessage::Pong(_) | WsMessage::Text(_) => continue,
2579                other => panic!("unexpected websocket message: {other:?}"),
2580            }
2581        }
2582    }
2583
2584    async fn send_test_server_message(
2585        ws: &mut WebSocketStream<TcpStream>,
2586        message: proto::ServerProtocolMessage,
2587    ) {
2588        ws.send(WsMessage::Binary(message.encode_to_vec().into()))
2589            .await
2590            .unwrap();
2591    }
2592
2593    fn get_me_result_message(
2594        id: u64,
2595        request_id: u64,
2596        user_id: i64,
2597    ) -> proto::ServerProtocolMessage {
2598        proto::ServerProtocolMessage {
2599            id,
2600            body: Some(proto::server_protocol_message::Body::RpcResult(
2601                proto::RpcResult {
2602                    req_msg_id: request_id,
2603                    result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2604                        user: Some(proto::User {
2605                            id: user_id,
2606                            ..Default::default()
2607                        }),
2608                    })),
2609                },
2610            )),
2611        }
2612    }
2613
2614    fn test_update_server_message(
2615        message_id: u64,
2616        inline_message_id: i64,
2617    ) -> proto::ServerProtocolMessage {
2618        proto::ServerProtocolMessage {
2619            id: message_id,
2620            body: Some(proto::server_protocol_message::Body::Message(
2621                proto::ServerMessage {
2622                    payload: Some(proto::server_message::Payload::Update(
2623                        proto::UpdatesPayload {
2624                            updates: vec![proto::Update {
2625                                seq: Some(1),
2626                                date: Some(1),
2627                                update: Some(proto::update::Update::NewMessage(
2628                                    proto::UpdateNewMessage {
2629                                        message: Some(proto::Message {
2630                                            id: inline_message_id,
2631                                            from_id: 42,
2632                                            peer_id: Some(proto::Peer {
2633                                                r#type: Some(proto::peer::Type::Chat(
2634                                                    proto::PeerChat { chat_id: 7 },
2635                                                )),
2636                                            }),
2637                                            chat_id: 7,
2638                                            message: Some("hello".to_owned()),
2639                                            date: 1,
2640                                            ..Default::default()
2641                                        }),
2642                                    },
2643                                )),
2644                            }],
2645                        },
2646                    )),
2647                },
2648            )),
2649        }
2650    }
2651}
inline_sdk/realtime.rs

inline_sdk/
realtime.rs
