Skip to main content

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
1490fn connection_init_for_token(token: &str, identity: &ClientIdentity) -> proto::ConnectionInit {
1491    proto::ConnectionInit {
1492        token: token.to_string(),
1493        build_number: None,
1494        layer: None,
1495        client_version: Some(identity.client_version().to_string()),
1496        os_version: client_info::current_os_version(),
1497    }
1498}
1499
1500fn rpc_error_code_name(error_code: i32) -> String {
1501    proto::rpc_error::Code::try_from(error_code)
1502        .map(|code| code.as_str_name())
1503        .unwrap_or("UNKNOWN")
1504        .to_string()
1505}
1506
1507fn rpc_error_from_proto(error: proto::RpcError) -> RealtimeError {
1508    let error_name = rpc_error_code_name(error.error_code);
1509    let friendly = format_rpc_error(error.error_code, &error_name, &error.message, error.code);
1510    RealtimeError::RpcError {
1511        code: error.code,
1512        error_code: error.error_code,
1513        error_name,
1514        message: error.message,
1515        friendly,
1516    }
1517}
1518
1519fn format_rpc_error(error_code: i32, error_name: &str, message: &str, status_code: i32) -> String {
1520    let label = match error_name {
1521        "UNKNOWN" => "Unknown RPC error",
1522        "BAD_REQUEST" => "Bad request",
1523        "UNAUTHENTICATED" => "Not authenticated",
1524        "RATE_LIMIT" => "Rate limited",
1525        "INTERNAL_ERROR" => "Internal server error",
1526        "PEER_ID_INVALID" => "Invalid peer (chat/user id)",
1527        "MESSAGE_ID_INVALID" => "Invalid message id",
1528        "USER_ID_INVALID" => "Invalid user id",
1529        "USER_ALREADY_MEMBER" => "User already in chat/space",
1530        "SPACE_ID_INVALID" => "Invalid space id",
1531        "CHAT_ID_INVALID" => "Invalid chat id",
1532        "EMAIL_INVALID" => "Invalid email address",
1533        "PHONE_NUMBER_INVALID" => "Invalid phone number",
1534        "SPACE_ADMIN_REQUIRED" => "Space admin required",
1535        "SPACE_OWNER_REQUIRED" => "Space owner required",
1536        "USERNAME_INVALID" => "Invalid username",
1537        "USERNAME_TAKEN" => "Username already taken",
1538        "FIRST_NAME_INVALID" => "Invalid first name",
1539        _ => "Unknown RPC error",
1540    };
1541
1542    let mut formatted = String::from(label);
1543    if error_name == "UNKNOWN" && error_code != 0 {
1544        formatted.push_str(&format!(" {error_code}"));
1545    }
1546    if !message.is_empty() && !message.eq_ignore_ascii_case(label) {
1547        formatted.push_str(": ");
1548        formatted.push_str(message);
1549    }
1550    if status_code != 0 {
1551        formatted.push_str(&format!(" (HTTP {status_code})"));
1552    }
1553    formatted
1554}
1555
1556fn connection_error_from_proto(error: proto::ConnectionError) -> RealtimeError {
1557    let reason = error.reason;
1558    let reason_name = proto::connection_error::Reason::try_from(reason)
1559        .map(|reason| reason.as_str_name())
1560        .unwrap_or("UNKNOWN")
1561        .to_string();
1562    let friendly = format_connection_error(reason, &reason_name);
1563    RealtimeError::ConnectionError {
1564        reason,
1565        reason_name,
1566        friendly,
1567    }
1568}
1569
1570fn format_connection_error(reason: i32, reason_name: &str) -> String {
1571    match reason_name {
1572        "UNAUTHORIZED" => "Realtime connection unauthorized".to_string(),
1573        "INVALID_AUTH" => "Realtime auth token is invalid".to_string(),
1574        "SESSION_REVOKED" => "Realtime session was revoked".to_string(),
1575        "REASON_UNSPECIFIED" => "Realtime connection rejected".to_string(),
1576        _ => format!("Realtime connection rejected: unknown reason {reason}"),
1577    }
1578}
1579
1580fn normalize_realtime_url(url: impl Into<String>) -> Result<Url, RealtimeError> {
1581    let original = url.into();
1582    let normalized = original.trim().to_string();
1583    if normalized.is_empty() {
1584        return Err(RealtimeError::InvalidUrl {
1585            url: original,
1586            message: "realtime URL cannot be empty".to_string(),
1587        });
1588    }
1589
1590    let parsed = Url::parse(&normalized).map_err(|err| RealtimeError::InvalidUrl {
1591        url: normalized.clone(),
1592        message: err.to_string(),
1593    })?;
1594
1595    if !matches!(parsed.scheme(), "ws" | "wss") {
1596        return Err(RealtimeError::InvalidUrl {
1597            url: normalized,
1598            message: "scheme must be ws or wss".to_string(),
1599        });
1600    }
1601
1602    if parsed.host_str().is_none() {
1603        return Err(RealtimeError::InvalidUrl {
1604            url: normalized,
1605            message: "host is required".to_string(),
1606        });
1607    }
1608
1609    if !parsed.username().is_empty() || parsed.password().is_some() {
1610        return Err(RealtimeError::InvalidUrl {
1611            url: normalized,
1612            message: "credentials are not valid in the realtime URL".to_string(),
1613        });
1614    }
1615
1616    if parsed.fragment().is_some() {
1617        return Err(RealtimeError::InvalidUrl {
1618            url: normalized,
1619            message: "fragments are not valid in the realtime URL".to_string(),
1620        });
1621    }
1622
1623    Ok(parsed)
1624}
1625
1626fn realtime_url_for_log(url: &Url) -> String {
1627    let host = url.host_str().unwrap_or("<missing-host>");
1628    let port = url
1629        .port()
1630        .map(|port| format!(":{port}"))
1631        .unwrap_or_default();
1632    format!("{}://{}{}{}", url.scheme(), host, port, url.path())
1633}
1634
1635fn realtime_url_for_debug(raw_url: &str) -> String {
1636    Url::parse(raw_url.trim())
1637        .map(|url| realtime_url_for_log(&url))
1638        .unwrap_or_else(|_| "<invalid>".to_string())
1639}
1640
1641fn realtime_header_value(field: &'static str, value: &str) -> Result<HeaderValue, RealtimeError> {
1642    HeaderValue::from_str(value).map_err(|_| RealtimeError::InvalidHeaderValue { field })
1643}
1644
1645fn realtime_event_kind(event: &RealtimeEvent) -> &'static str {
1646    match event {
1647        RealtimeEvent::Updates(_) => "updates",
1648        RealtimeEvent::Ack { .. } => "ack",
1649        RealtimeEvent::Pong { .. } => "pong",
1650    }
1651}
1652
1653async fn with_optional_timeout<T, E, Fut>(
1654    operation: &'static str,
1655    timeout: Option<Duration>,
1656    future: Fut,
1657) -> Result<T, RealtimeError>
1658where
1659    Fut: Future<Output = Result<T, E>>,
1660    RealtimeError: From<E>,
1661{
1662    match timeout {
1663        Some(timeout) => tokio::time::timeout(timeout, future)
1664            .await
1665            .map_err(|_| RealtimeError::Timeout { operation, timeout })?
1666            .map_err(RealtimeError::from),
1667        None => future.await.map_err(RealtimeError::from),
1668    }
1669}
1670
1671struct IdGenerator {
1672    last_timestamp: u64,
1673    sequence: u32,
1674}
1675
1676impl IdGenerator {
1677    fn new() -> Self {
1678        Self {
1679            last_timestamp: 0,
1680            sequence: 0,
1681        }
1682    }
1683
1684    fn next_id(&mut self) -> u64 {
1685        let timestamp = current_epoch_seconds().saturating_sub(EPOCH_SECONDS);
1686        if timestamp == self.last_timestamp {
1687            self.sequence = self.sequence.wrapping_add(1);
1688        } else {
1689            self.sequence = 0;
1690            self.last_timestamp = timestamp;
1691        }
1692
1693        (timestamp << 32) | self.sequence as u64
1694    }
1695}
1696
1697const EPOCH_SECONDS: u64 = 1_735_689_600; // 2025-01-01T00:00:00Z
1698
1699fn current_epoch_seconds() -> u64 {
1700    use std::time::{SystemTime, UNIX_EPOCH};
1701    SystemTime::now()
1702        .duration_since(UNIX_EPOCH)
1703        .unwrap_or_default()
1704        .as_secs()
1705}
1706
1707#[cfg(test)]
1708mod tests {
1709    use super::*;
1710    use tokio::net::{TcpListener, TcpStream};
1711    use tokio_tungstenite::WebSocketStream;
1712    use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
1713    use tokio_tungstenite::{accept_async, accept_hdr_async};
1714
1715    #[test]
1716    fn rpc_error_code_name_uses_stable_proto_name() {
1717        assert_eq!(
1718            rpc_error_code_name(proto::rpc_error::Code::PeerIdInvalid as i32),
1719            "PEER_ID_INVALID"
1720        );
1721        assert_eq!(rpc_error_code_name(999), "UNKNOWN");
1722    }
1723
1724    #[test]
1725    fn rpc_error_formatter_covers_new_profile_codes() {
1726        assert_eq!(
1727            format_rpc_error(
1728                proto::rpc_error::Code::UsernameInvalid as i32,
1729                "USERNAME_INVALID",
1730                "",
1731                400
1732            ),
1733            "Invalid username (HTTP 400)"
1734        );
1735        assert_eq!(
1736            format_rpc_error(
1737                proto::rpc_error::Code::UsernameTaken as i32,
1738                "USERNAME_TAKEN",
1739                "handle exists",
1740                409
1741            ),
1742            "Username already taken: handle exists (HTTP 409)"
1743        );
1744        assert_eq!(
1745            format_rpc_error(
1746                proto::rpc_error::Code::FirstNameInvalid as i32,
1747                "FIRST_NAME_INVALID",
1748                "",
1749                400
1750            ),
1751            "Invalid first name (HTTP 400)"
1752        );
1753    }
1754
1755    #[test]
1756    fn connection_error_preserves_proto_reason() {
1757        let err = connection_error_from_proto(proto::ConnectionError {
1758            reason: proto::connection_error::Reason::InvalidAuth as i32,
1759        });
1760
1761        match err {
1762            RealtimeError::ConnectionError {
1763                reason,
1764                reason_name,
1765                friendly,
1766            } => {
1767                assert_eq!(reason, 2);
1768                assert_eq!(reason_name, "INVALID_AUTH");
1769                assert_eq!(friendly, "Realtime auth token is invalid");
1770            }
1771            other => panic!("expected connection error, got {other:?}"),
1772        }
1773    }
1774
1775    #[test]
1776    fn unknown_connection_error_reason_is_preserved() {
1777        let err = connection_error_from_proto(proto::ConnectionError { reason: 999 });
1778
1779        match err {
1780            RealtimeError::ConnectionError {
1781                reason,
1782                reason_name,
1783                friendly,
1784            } => {
1785                assert_eq!(reason, 999);
1786                assert_eq!(reason_name, "UNKNOWN");
1787                assert_eq!(friendly, "Realtime connection rejected: unknown reason 999");
1788            }
1789            other => panic!("expected connection error, got {other:?}"),
1790        }
1791    }
1792
1793    #[test]
1794    fn connection_init_uses_custom_client_identity() {
1795        let init =
1796            connection_init_for_token("token-1", &ClientIdentity::new("integration-test", "9.9.9"));
1797
1798        assert_eq!(init.token, "token-1");
1799        assert_eq!(init.client_version.as_deref(), Some("9.9.9"));
1800        assert!(init.build_number.is_none());
1801        assert!(init.layer.is_none());
1802    }
1803
1804    #[tokio::test]
1805    #[allow(clippy::result_large_err)]
1806    async fn realtime_client_connects_and_calls_get_me_against_local_server() {
1807        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1808        let addr = listener.local_addr().unwrap();
1809        let server = tokio::spawn(async move {
1810            let (stream, _) = listener.accept().await.unwrap();
1811            let mut ws = accept_hdr_async(stream, |request: &Request, response: Response| {
1812                assert_eq!(
1813                    request
1814                        .headers()
1815                        .get(client_info::CLIENT_TYPE_HEADER)
1816                        .and_then(|value| value.to_str().ok()),
1817                    Some("transport-test")
1818                );
1819                assert_eq!(
1820                    request
1821                        .headers()
1822                        .get(client_info::CLIENT_VERSION_HEADER)
1823                        .and_then(|value| value.to_str().ok()),
1824                    Some("1.2.3")
1825                );
1826                Ok(response)
1827            })
1828            .await
1829            .unwrap();
1830
1831            let init = read_test_client_message(&mut ws).await;
1832            match init.body {
1833                Some(proto::client_message::Body::ConnectionInit(init)) => {
1834                    assert_eq!(init.token, "token-1");
1835                    assert_eq!(init.client_version.as_deref(), Some("1.2.3"));
1836                    assert!(init.os_version.is_some());
1837                }
1838                other => panic!("expected connection init, got {other:?}"),
1839            }
1840            send_test_server_message(
1841                &mut ws,
1842                proto::ServerProtocolMessage {
1843                    id: 1,
1844                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1845                        proto::ConnectionOpen {},
1846                    )),
1847                },
1848            )
1849            .await;
1850
1851            let rpc = read_test_client_message(&mut ws).await;
1852            match &rpc.body {
1853                Some(proto::client_message::Body::RpcCall(call)) => {
1854                    assert_eq!(call.method, proto::Method::GetMe as i32);
1855                    assert!(matches!(call.input, Some(proto::rpc_call::Input::GetMe(_))));
1856                }
1857                other => panic!("expected getMe rpc call, got {other:?}"),
1858            }
1859            send_test_server_message(
1860                &mut ws,
1861                proto::ServerProtocolMessage {
1862                    id: 2,
1863                    body: Some(proto::server_protocol_message::Body::RpcResult(
1864                        proto::RpcResult {
1865                            req_msg_id: rpc.id,
1866                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
1867                                user: Some(proto::User {
1868                                    id: 42,
1869                                    first_name: Some("Ada".to_string()),
1870                                    ..Default::default()
1871                                }),
1872                            })),
1873                        },
1874                    )),
1875                },
1876            )
1877            .await;
1878        });
1879
1880        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1881            .identity(ClientIdentity::new("transport-test", "1.2.3"))
1882            .without_connect_timeout()
1883            .without_rpc_timeout()
1884            .connect()
1885            .await
1886            .unwrap();
1887        let result = client.call(proto::GetMeInput {}).await.unwrap();
1888
1889        assert_eq!(result.user.unwrap().id, 42);
1890        server.await.unwrap();
1891    }
1892
1893    #[tokio::test]
1894    async fn realtime_client_receives_update_events_and_acks_them() {
1895        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1896        let addr = listener.local_addr().unwrap();
1897        let server = tokio::spawn(async move {
1898            let (stream, _) = listener.accept().await.unwrap();
1899            let mut ws = accept_async(stream).await.unwrap();
1900
1901            let init = read_test_client_message(&mut ws).await;
1902            assert!(matches!(
1903                init.body,
1904                Some(proto::client_message::Body::ConnectionInit(_))
1905            ));
1906            send_test_server_message(
1907                &mut ws,
1908                proto::ServerProtocolMessage {
1909                    id: 1,
1910                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1911                        proto::ConnectionOpen {},
1912                    )),
1913                },
1914            )
1915            .await;
1916            send_test_server_message(&mut ws, test_update_server_message(9, 77)).await;
1917
1918            let ack = read_test_client_message(&mut ws).await;
1919            match ack.body {
1920                Some(proto::client_message::Body::Ack(ack)) => {
1921                    assert_eq!(ack.msg_id, 9);
1922                }
1923                other => panic!("expected ack for pushed update, got {other:?}"),
1924            }
1925        });
1926
1927        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1928            .without_connect_timeout()
1929            .without_rpc_timeout()
1930            .connect()
1931            .await
1932            .unwrap();
1933        let event = client.next_event().await.unwrap();
1934
1935        match event {
1936            RealtimeEvent::Updates(updates) => {
1937                assert_eq!(updates.len(), 1);
1938                match updates[0].update.as_ref() {
1939                    Some(proto::update::Update::NewMessage(update)) => {
1940                        assert_eq!(update.message.as_ref().unwrap().id, 77);
1941                    }
1942                    other => panic!("expected new message update, got {other:?}"),
1943                }
1944            }
1945            other => panic!("expected updates event, got {other:?}"),
1946        }
1947        server.await.unwrap();
1948    }
1949
1950    #[tokio::test]
1951    async fn realtime_rpc_wait_acks_pushed_updates_before_matching_result() {
1952        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1953        let addr = listener.local_addr().unwrap();
1954        let server = tokio::spawn(async move {
1955            let (stream, _) = listener.accept().await.unwrap();
1956            let mut ws = accept_async(stream).await.unwrap();
1957
1958            let init = read_test_client_message(&mut ws).await;
1959            assert!(matches!(
1960                init.body,
1961                Some(proto::client_message::Body::ConnectionInit(_))
1962            ));
1963            send_test_server_message(
1964                &mut ws,
1965                proto::ServerProtocolMessage {
1966                    id: 1,
1967                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1968                        proto::ConnectionOpen {},
1969                    )),
1970                },
1971            )
1972            .await;
1973
1974            let rpc = read_test_client_message(&mut ws).await;
1975            send_test_server_message(&mut ws, test_update_server_message(10, 88)).await;
1976
1977            let ack = read_test_client_message(&mut ws).await;
1978            match ack.body {
1979                Some(proto::client_message::Body::Ack(ack)) => {
1980                    assert_eq!(ack.msg_id, 10);
1981                }
1982                other => panic!("expected ack while waiting for rpc result, got {other:?}"),
1983            }
1984
1985            send_test_server_message(
1986                &mut ws,
1987                proto::ServerProtocolMessage {
1988                    id: 2,
1989                    body: Some(proto::server_protocol_message::Body::RpcResult(
1990                        proto::RpcResult {
1991                            req_msg_id: rpc.id,
1992                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
1993                                user: Some(proto::User {
1994                                    id: 42,
1995                                    ..Default::default()
1996                                }),
1997                            })),
1998                        },
1999                    )),
2000                },
2001            )
2002            .await;
2003        });
2004
2005        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2006            .without_connect_timeout()
2007            .without_rpc_timeout()
2008            .connect()
2009            .await
2010            .unwrap();
2011        let result = client.call(proto::GetMeInput {}).await.unwrap();
2012
2013        assert_eq!(result.user.unwrap().id, 42);
2014        server.await.unwrap();
2015    }
2016
2017    #[tokio::test]
2018    async fn realtime_session_routes_pushed_updates_during_rpc() {
2019        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2020        let addr = listener.local_addr().unwrap();
2021        let server = tokio::spawn(async move {
2022            let (stream, _) = listener.accept().await.unwrap();
2023            let mut ws = accept_async(stream).await.unwrap();
2024
2025            let init = read_test_client_message(&mut ws).await;
2026            assert!(matches!(
2027                init.body,
2028                Some(proto::client_message::Body::ConnectionInit(_))
2029            ));
2030            send_test_server_message(
2031                &mut ws,
2032                proto::ServerProtocolMessage {
2033                    id: 1,
2034                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2035                        proto::ConnectionOpen {},
2036                    )),
2037                },
2038            )
2039            .await;
2040
2041            let rpc = read_test_client_message(&mut ws).await;
2042            send_test_server_message(&mut ws, test_update_server_message(10, 88)).await;
2043            let ack = read_test_client_message(&mut ws).await;
2044            assert!(matches!(
2045                ack.body,
2046                Some(proto::client_message::Body::Ack(proto::Ack { msg_id: 10 }))
2047            ));
2048            send_test_server_message(
2049                &mut ws,
2050                proto::ServerProtocolMessage {
2051                    id: 2,
2052                    body: Some(proto::server_protocol_message::Body::RpcResult(
2053                        proto::RpcResult {
2054                            req_msg_id: rpc.id,
2055                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2056                                user: Some(proto::User {
2057                                    id: 42,
2058                                    ..Default::default()
2059                                }),
2060                            })),
2061                        },
2062                    )),
2063                },
2064            )
2065            .await;
2066        });
2067
2068        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2069            .without_connect_timeout()
2070            .without_rpc_timeout()
2071            .connect_session()
2072            .await
2073            .unwrap();
2074        let mut events = session.subscribe();
2075        let (result, event) = tokio::join!(session.call(proto::GetMeInput {}), events.recv());
2076
2077        assert_eq!(result.unwrap().user.unwrap().id, 42);
2078        match event.unwrap() {
2079            RealtimeEvent::Updates(updates) => match updates[0].update.as_ref() {
2080                Some(proto::update::Update::NewMessage(update)) => {
2081                    assert_eq!(update.message.as_ref().unwrap().id, 88);
2082                }
2083                other => panic!("expected new message update, got {other:?}"),
2084            },
2085            other => panic!("expected updates event, got {other:?}"),
2086        }
2087        server.await.unwrap();
2088    }
2089
2090    #[tokio::test]
2091    async fn realtime_session_matches_concurrent_rpc_results_by_request_id() {
2092        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2093        let addr = listener.local_addr().unwrap();
2094        let server = tokio::spawn(async move {
2095            let (stream, _) = listener.accept().await.unwrap();
2096            let mut ws = accept_async(stream).await.unwrap();
2097            let _ = read_test_client_message(&mut ws).await;
2098            send_test_server_message(
2099                &mut ws,
2100                proto::ServerProtocolMessage {
2101                    id: 1,
2102                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2103                        proto::ConnectionOpen {},
2104                    )),
2105                },
2106            )
2107            .await;
2108
2109            let first = read_test_client_message(&mut ws).await;
2110            let second = read_test_client_message(&mut ws).await;
2111            for (id, request, user_id) in [(2, &second, 2), (3, &first, 1)] {
2112                send_test_server_message(
2113                    &mut ws,
2114                    proto::ServerProtocolMessage {
2115                        id,
2116                        body: Some(proto::server_protocol_message::Body::RpcResult(
2117                            proto::RpcResult {
2118                                req_msg_id: request.id,
2119                                result: Some(proto::rpc_result::Result::GetMe(
2120                                    proto::GetMeResult {
2121                                        user: Some(proto::User {
2122                                            id: user_id,
2123                                            ..Default::default()
2124                                        }),
2125                                    },
2126                                )),
2127                            },
2128                        )),
2129                    },
2130                )
2131                .await;
2132            }
2133        });
2134
2135        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2136            .without_connect_timeout()
2137            .without_rpc_timeout()
2138            .connect_session()
2139            .await
2140            .unwrap();
2141        let (first, second) = tokio::join!(
2142            session.call(proto::GetMeInput {}),
2143            session.call(proto::GetMeInput {})
2144        );
2145
2146        assert_eq!(first.unwrap().user.unwrap().id, 1);
2147        assert_eq!(second.unwrap().user.unwrap().id, 2);
2148        server.await.unwrap();
2149    }
2150
2151    #[tokio::test]
2152    async fn realtime_session_bounds_in_flight_rpcs() {
2153        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2154        let addr = listener.local_addr().unwrap();
2155        let server = tokio::spawn(async move {
2156            let (stream, _) = listener.accept().await.unwrap();
2157            let mut ws = accept_async(stream).await.unwrap();
2158            let _ = read_test_client_message(&mut ws).await;
2159            send_test_server_message(
2160                &mut ws,
2161                proto::ServerProtocolMessage {
2162                    id: 1,
2163                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2164                        proto::ConnectionOpen {},
2165                    )),
2166                },
2167            )
2168            .await;
2169
2170            let first = read_test_client_message(&mut ws).await;
2171            assert!(
2172                tokio::time::timeout(Duration::from_millis(50), read_test_client_message(&mut ws),)
2173                    .await
2174                    .is_err(),
2175                "session issued a second RPC above its configured in-flight limit"
2176            );
2177            send_test_server_message(&mut ws, get_me_result_message(2, first.id, 1)).await;
2178            let second = read_test_client_message(&mut ws).await;
2179            send_test_server_message(&mut ws, get_me_result_message(3, second.id, 2)).await;
2180        });
2181
2182        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2183            .without_connect_timeout()
2184            .without_rpc_timeout()
2185            .max_in_flight_rpcs(1)
2186            .connect_session()
2187            .await
2188            .unwrap();
2189        let (first, second) = tokio::join!(
2190            session.call(proto::GetMeInput {}),
2191            session.call(proto::GetMeInput {})
2192        );
2193
2194        assert_eq!(first.unwrap().user.unwrap().id, 1);
2195        assert_eq!(second.unwrap().user.unwrap().id, 2);
2196        server.await.unwrap();
2197    }
2198
2199    #[tokio::test]
2200    async fn realtime_session_drops_timed_out_rpc_and_remains_usable() {
2201        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2202        let addr = listener.local_addr().unwrap();
2203        let server = tokio::spawn(async move {
2204            let (stream, _) = listener.accept().await.unwrap();
2205            let mut ws = accept_async(stream).await.unwrap();
2206            let _ = read_test_client_message(&mut ws).await;
2207            send_test_server_message(
2208                &mut ws,
2209                proto::ServerProtocolMessage {
2210                    id: 1,
2211                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2212                        proto::ConnectionOpen {},
2213                    )),
2214                },
2215            )
2216            .await;
2217
2218            let timed_out = read_test_client_message(&mut ws).await;
2219            tokio::time::sleep(Duration::from_millis(40)).await;
2220            send_test_server_message(
2221                &mut ws,
2222                proto::ServerProtocolMessage {
2223                    id: 2,
2224                    body: Some(proto::server_protocol_message::Body::RpcResult(
2225                        proto::RpcResult {
2226                            req_msg_id: timed_out.id,
2227                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2228                                user: Some(proto::User {
2229                                    id: 1,
2230                                    ..Default::default()
2231                                }),
2232                            })),
2233                        },
2234                    )),
2235                },
2236            )
2237            .await;
2238
2239            let recovered = read_test_client_message(&mut ws).await;
2240            send_test_server_message(
2241                &mut ws,
2242                proto::ServerProtocolMessage {
2243                    id: 3,
2244                    body: Some(proto::server_protocol_message::Body::RpcResult(
2245                        proto::RpcResult {
2246                            req_msg_id: recovered.id,
2247                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2248                                user: Some(proto::User {
2249                                    id: 2,
2250                                    ..Default::default()
2251                                }),
2252                            })),
2253                        },
2254                    )),
2255                },
2256            )
2257            .await;
2258        });
2259
2260        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2261            .without_connect_timeout()
2262            .rpc_timeout(Duration::from_millis(25))
2263            .connect_session()
2264            .await
2265            .unwrap();
2266
2267        assert!(matches!(
2268            session.call(proto::GetMeInput {}).await,
2269            Err(RealtimeError::Timeout {
2270                operation: "rpc",
2271                ..
2272            })
2273        ));
2274        let recovered = session.call(proto::GetMeInput {}).await.unwrap();
2275        assert_eq!(recovered.user.unwrap().id, 2);
2276        server.await.unwrap();
2277    }
2278
2279    #[tokio::test]
2280    async fn realtime_session_closes_when_heartbeat_pong_is_missing() {
2281        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
2282        let addr = listener.local_addr().unwrap();
2283        let server = tokio::spawn(async move {
2284            let (stream, _) = listener.accept().await.unwrap();
2285            let mut ws = accept_async(stream).await.unwrap();
2286            let _ = read_test_client_message(&mut ws).await;
2287            send_test_server_message(
2288                &mut ws,
2289                proto::ServerProtocolMessage {
2290                    id: 1,
2291                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
2292                        proto::ConnectionOpen {},
2293                    )),
2294                },
2295            )
2296            .await;
2297            let ping = read_test_client_message(&mut ws).await;
2298            assert!(matches!(
2299                ping.body,
2300                Some(proto::client_message::Body::Ping(_))
2301            ));
2302            tokio::time::sleep(Duration::from_millis(100)).await;
2303        });
2304
2305        let session = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
2306            .without_connect_timeout()
2307            .heartbeat(Duration::from_millis(20), Duration::from_millis(30))
2308            .connect_session()
2309            .await
2310            .unwrap();
2311        let mut events = session.subscribe();
2312
2313        let error = tokio::time::timeout(Duration::from_millis(150), events.recv())
2314            .await
2315            .expect("heartbeat deadline should close the session")
2316            .unwrap_err();
2317        assert!(matches!(error, RealtimeError::ConnectionClosed));
2318        assert!(session.is_closed());
2319        server.await.unwrap();
2320    }
2321
2322    #[test]
2323    fn realtime_url_validation_accepts_ws_urls() {
2324        let url = normalize_realtime_url(" wss://api.inline.chat/realtime?edge=iad ").unwrap();
2325        assert_eq!(url.as_str(), "wss://api.inline.chat/realtime?edge=iad");
2326    }
2327
2328    #[test]
2329    fn realtime_url_validation_rejects_invalid_urls() {
2330        let err = normalize_realtime_url("inline.test").unwrap_err();
2331        match err {
2332            RealtimeError::InvalidUrl { url, message } => {
2333                assert_eq!(url, "inline.test");
2334                assert!(message.contains("relative URL without a base"));
2335            }
2336            other => panic!("expected invalid URL, got {other:?}"),
2337        }
2338
2339        let err = normalize_realtime_url("https://api.inline.chat/realtime").unwrap_err();
2340        match err {
2341            RealtimeError::InvalidUrl { message, .. } => {
2342                assert_eq!(message, "scheme must be ws or wss");
2343            }
2344            other => panic!("expected invalid URL, got {other:?}"),
2345        }
2346
2347        let err = normalize_realtime_url("wss://user:secret@api.inline.chat/realtime").unwrap_err();
2348        match &err {
2349            RealtimeError::InvalidUrl { message, .. } => {
2350                assert_eq!(message, "credentials are not valid in the realtime URL");
2351                assert!(!err.to_string().contains("secret"));
2352            }
2353            other => panic!("expected invalid URL, got {other:?}"),
2354        }
2355
2356        let err = normalize_realtime_url("wss://api.inline.chat/realtime#token").unwrap_err();
2357        match err {
2358            RealtimeError::InvalidUrl { message, .. } => {
2359                assert_eq!(message, "fragments are not valid in the realtime URL");
2360            }
2361            other => panic!("expected invalid URL, got {other:?}"),
2362        }
2363    }
2364
2365    #[test]
2366    fn realtime_invalid_url_debug_redacts_unsafe_url_parts() {
2367        let err = normalize_realtime_url(
2368            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret#frag",
2369        )
2370        .unwrap_err();
2371        let debug = format!("{err:?}");
2372
2373        assert!(debug.contains("wss://api.inline.chat/realtime"));
2374        assert!(!debug.contains("url-secret"));
2375        assert!(!debug.contains("query-secret"));
2376    }
2377
2378    #[test]
2379    fn realtime_url_for_log_omits_query_and_fragment() {
2380        let url = Url::parse("wss://api.inline.chat/realtime?token=secret#frag").unwrap();
2381        assert_eq!(realtime_url_for_log(&url), "wss://api.inline.chat/realtime");
2382    }
2383
2384    #[test]
2385    fn realtime_header_value_rejects_invalid_values() {
2386        let err = realtime_header_value("user_agent", "bad\nvalue").unwrap_err();
2387        match err {
2388            RealtimeError::InvalidHeaderValue { field } => {
2389                assert_eq!(field, "user_agent");
2390            }
2391            other => panic!("expected invalid header value, got {other:?}"),
2392        }
2393    }
2394
2395    #[test]
2396    fn realtime_builder_keeps_custom_identity_and_default_timeouts() {
2397        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
2398            .identity(ClientIdentity::new("agent", "0.2.0"));
2399
2400        assert_eq!(builder.url, "wss://api.inline.chat/realtime");
2401        assert_eq!(builder.token, "token-1");
2402        assert_eq!(builder.identity.client_type(), "agent");
2403        assert_eq!(builder.identity.client_version(), "0.2.0");
2404        assert_eq!(builder.connect_timeout, Some(DEFAULT_CONNECT_TIMEOUT));
2405        assert_eq!(builder.rpc_timeout, Some(DEFAULT_RPC_TIMEOUT));
2406        assert_eq!(builder.heartbeat_interval, Some(DEFAULT_HEARTBEAT_INTERVAL));
2407        assert_eq!(builder.heartbeat_timeout, DEFAULT_HEARTBEAT_TIMEOUT);
2408        assert_eq!(
2409            builder.max_in_flight_rpcs,
2410            DEFAULT_SESSION_MAX_IN_FLIGHT_RPCS
2411        );
2412    }
2413
2414    #[test]
2415    fn realtime_builder_debug_redacts_token() {
2416        let builder = RealtimeClient::builder(
2417            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret",
2418            "secret-token-1",
2419        );
2420        let debug = format!("{builder:?}");
2421
2422        assert!(debug.contains("RealtimeClientBuilder"));
2423        assert!(debug.contains("wss://api.inline.chat/realtime"));
2424        assert!(debug.contains("<redacted>"));
2425        assert!(!debug.contains("secret-token-1"));
2426        assert!(!debug.contains("url-secret"));
2427        assert!(!debug.contains("query-secret"));
2428    }
2429
2430    #[test]
2431    fn realtime_builder_can_override_or_disable_timeouts() {
2432        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
2433            .connect_timeout(Duration::from_secs(5))
2434            .rpc_timeout(Duration::from_secs(10))
2435            .max_in_flight_rpcs(7);
2436
2437        assert_eq!(builder.connect_timeout, Some(Duration::from_secs(5)));
2438        assert_eq!(builder.rpc_timeout, Some(Duration::from_secs(10)));
2439        assert_eq!(builder.max_in_flight_rpcs, 7);
2440
2441        let builder = builder
2442            .without_connect_timeout()
2443            .without_rpc_timeout()
2444            .without_heartbeat();
2445        assert_eq!(builder.connect_timeout, None);
2446        assert_eq!(builder.rpc_timeout, None);
2447        assert_eq!(builder.heartbeat_interval, None);
2448    }
2449
2450    #[test]
2451    fn typed_rpc_request_maps_method_input_and_result() {
2452        assert_eq!(
2453            <proto::GetChatsInput as RpcRequest>::METHOD,
2454            proto::Method::GetChats
2455        );
2456
2457        let input = proto::GetChatsInput {};
2458        match input.into_rpc_input() {
2459            proto::rpc_call::Input::GetChats(_) => {}
2460            other => panic!("expected GetChats input, got {other:?}"),
2461        }
2462
2463        let response = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
2464            proto::rpc_result::Result::GetChats(proto::GetChatsResult::default()),
2465        )
2466        .unwrap();
2467        assert!(response.dialogs.is_empty());
2468    }
2469
2470    #[test]
2471    fn typed_rpc_request_rejects_unexpected_result_variant() {
2472        let err = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
2473            proto::rpc_result::Result::GetMe(proto::GetMeResult::default()),
2474        )
2475        .unwrap_err();
2476
2477        match err {
2478            RealtimeError::UnexpectedResult {
2479                method,
2480                expected,
2481                actual,
2482            } => {
2483                assert_eq!(method, "GET_CHATS");
2484                assert_eq!(expected, "GetChats");
2485                assert_eq!(actual, "GetMe");
2486            }
2487            other => panic!("expected unexpected result error, got {other:?}"),
2488        }
2489    }
2490
2491    #[tokio::test]
2492    async fn optional_timeout_reports_elapsed_operation() {
2493        let err = with_optional_timeout("test", Some(Duration::from_millis(1)), async {
2494            tokio::time::sleep(Duration::from_millis(20)).await;
2495            Ok::<(), RealtimeError>(())
2496        })
2497        .await
2498        .unwrap_err();
2499
2500        match err {
2501            RealtimeError::Timeout { operation, timeout } => {
2502                assert_eq!(operation, "test");
2503                assert_eq!(timeout, Duration::from_millis(1));
2504            }
2505            other => panic!("expected timeout, got {other:?}"),
2506        }
2507    }
2508
2509    #[tokio::test]
2510    async fn optional_timeout_can_be_disabled() {
2511        with_optional_timeout("test", None, async { Ok::<_, RealtimeError>("done") })
2512            .await
2513            .unwrap();
2514    }
2515
2516    async fn read_test_client_message(ws: &mut WebSocketStream<TcpStream>) -> proto::ClientMessage {
2517        loop {
2518            match ws.next().await.unwrap().unwrap() {
2519                WsMessage::Binary(bytes) => {
2520                    return proto::ClientMessage::decode(&*bytes).unwrap();
2521                }
2522                WsMessage::Ping(_) | WsMessage::Pong(_) | WsMessage::Text(_) => continue,
2523                other => panic!("unexpected websocket message: {other:?}"),
2524            }
2525        }
2526    }
2527
2528    async fn send_test_server_message(
2529        ws: &mut WebSocketStream<TcpStream>,
2530        message: proto::ServerProtocolMessage,
2531    ) {
2532        ws.send(WsMessage::Binary(message.encode_to_vec().into()))
2533            .await
2534            .unwrap();
2535    }
2536
2537    fn get_me_result_message(
2538        id: u64,
2539        request_id: u64,
2540        user_id: i64,
2541    ) -> proto::ServerProtocolMessage {
2542        proto::ServerProtocolMessage {
2543            id,
2544            body: Some(proto::server_protocol_message::Body::RpcResult(
2545                proto::RpcResult {
2546                    req_msg_id: request_id,
2547                    result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
2548                        user: Some(proto::User {
2549                            id: user_id,
2550                            ..Default::default()
2551                        }),
2552                    })),
2553                },
2554            )),
2555        }
2556    }
2557
2558    fn test_update_server_message(
2559        message_id: u64,
2560        inline_message_id: i64,
2561    ) -> proto::ServerProtocolMessage {
2562        proto::ServerProtocolMessage {
2563            id: message_id,
2564            body: Some(proto::server_protocol_message::Body::Message(
2565                proto::ServerMessage {
2566                    payload: Some(proto::server_message::Payload::Update(
2567                        proto::UpdatesPayload {
2568                            updates: vec![proto::Update {
2569                                seq: Some(1),
2570                                date: Some(1),
2571                                update: Some(proto::update::Update::NewMessage(
2572                                    proto::UpdateNewMessage {
2573                                        message: Some(proto::Message {
2574                                            id: inline_message_id,
2575                                            from_id: 42,
2576                                            peer_id: Some(proto::Peer {
2577                                                r#type: Some(proto::peer::Type::Chat(
2578                                                    proto::PeerChat { chat_id: 7 },
2579                                                )),
2580                                            }),
2581                                            chat_id: 7,
2582                                            message: Some("hello".to_owned()),
2583                                            date: 1,
2584                                            ..Default::default()
2585                                        }),
2586                                    },
2587                                )),
2588                            }],
2589                        },
2590                    )),
2591                },
2592            )),
2593        }
2594    }
2595}