inline_sdk/
realtime.rs

1//! Realtime WebSocket RPC transport for Inline protocol calls.
2
3use futures_util::{SinkExt, StreamExt};
4use prost::Message;
5use std::fmt;
6use std::future::Future;
7use std::time::Duration;
8use tokio_tungstenite::connect_async;
9use tokio_tungstenite::tungstenite::Message as WsMessage;
10use tokio_tungstenite::tungstenite::client::IntoClientRequest;
11use tokio_tungstenite::tungstenite::http::HeaderValue;
12use url::Url;
13
14use crate::client_info::{self, ClientIdentity};
15use inline_protocol::proto;
16
17/// Default timeout for opening a realtime connection.
18pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
19/// Default timeout for a single realtime RPC invocation.
20pub const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(60);
21
22/// Error returned by realtime connection and RPC operations.
23#[derive(thiserror::Error)]
24#[non_exhaustive]
25pub enum RealtimeError {
26    /// Invalid realtime WebSocket URL supplied to a realtime client builder.
27    #[error("invalid realtime URL: {message}")]
28    InvalidUrl {
29        /// Original URL value supplied by the caller.
30        url: String,
31        /// Human-readable validation failure.
32        message: String,
33    },
34    /// WebSocket transport error.
35    #[error("websocket error: {0}")]
36    WebSocket(Box<tokio_tungstenite::tungstenite::Error>),
37    /// A validated client identity could not be represented as a realtime header.
38    #[error("{field} contains characters that are invalid in realtime headers")]
39    InvalidHeaderValue {
40        /// Name of the invalid realtime header field.
41        field: &'static str,
42    },
43    /// Protobuf decode error for a server message.
44    #[error("protocol error: {0}")]
45    Protocol(#[from] prost::DecodeError),
46    /// The server returned an RPC result envelope without a result body.
47    #[error("missing rpc result")]
48    MissingResult,
49    /// The server returned a result oneof that does not match the requested method.
50    #[error("unexpected rpc result for {method}: expected {expected}, got {actual}")]
51    UnexpectedResult {
52        /// Requested RPC method.
53        method: &'static str,
54        /// Expected result oneof variant.
55        expected: &'static str,
56        /// Actual result oneof variant.
57        actual: &'static str,
58    },
59    /// A realtime operation exceeded its configured timeout.
60    #[error("realtime {operation} timed out after {timeout:?}")]
61    Timeout {
62        /// Operation that timed out.
63        operation: &'static str,
64        /// Configured timeout.
65        timeout: Duration,
66    },
67    /// The realtime server rejected the connection.
68    #[error("{friendly}")]
69    ConnectionError {
70        /// Numeric connection-error reason.
71        reason: i32,
72        /// Stable protobuf enum name for the reason.
73        reason_name: String,
74        /// Human-readable formatted error.
75        friendly: String,
76    },
77    /// The realtime connection closed before the requested operation completed.
78    #[error("realtime connection closed")]
79    ConnectionClosed,
80    /// The realtime server returned an RPC error for a request.
81    #[error("{friendly}")]
82    RpcError {
83        /// Transport-level status code.
84        code: i32,
85        /// Inline RPC error code.
86        error_code: i32,
87        /// Stable protobuf enum name for the RPC error.
88        error_name: String,
89        /// Server-provided error message.
90        message: String,
91        /// Human-readable formatted error.
92        friendly: String,
93    },
94}
95
96impl fmt::Debug for RealtimeError {
97    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98        match self {
99            RealtimeError::InvalidUrl { url, message } => f
100                .debug_struct("InvalidUrl")
101                .field("url", &realtime_url_for_debug(url))
102                .field("message", message)
103                .finish(),
104            RealtimeError::WebSocket(error) => f.debug_tuple("WebSocket").field(error).finish(),
105            RealtimeError::InvalidHeaderValue { field } => f
106                .debug_struct("InvalidHeaderValue")
107                .field("field", field)
108                .finish(),
109            RealtimeError::Protocol(error) => f.debug_tuple("Protocol").field(error).finish(),
110            RealtimeError::MissingResult => f.debug_struct("MissingResult").finish(),
111            RealtimeError::UnexpectedResult {
112                method,
113                expected,
114                actual,
115            } => f
116                .debug_struct("UnexpectedResult")
117                .field("method", method)
118                .field("expected", expected)
119                .field("actual", actual)
120                .finish(),
121            RealtimeError::Timeout { operation, timeout } => f
122                .debug_struct("Timeout")
123                .field("operation", operation)
124                .field("timeout", timeout)
125                .finish(),
126            RealtimeError::ConnectionError {
127                reason,
128                reason_name,
129                friendly,
130            } => f
131                .debug_struct("ConnectionError")
132                .field("reason", reason)
133                .field("reason_name", reason_name)
134                .field("friendly", friendly)
135                .finish(),
136            RealtimeError::ConnectionClosed => f.debug_struct("ConnectionClosed").finish(),
137            RealtimeError::RpcError {
138                code,
139                error_code,
140                error_name,
141                message,
142                friendly,
143            } => f
144                .debug_struct("RpcError")
145                .field("code", code)
146                .field("error_code", error_code)
147                .field("error_name", error_name)
148                .field("message", message)
149                .field("friendly", friendly)
150                .finish(),
151        }
152    }
153}
154
155impl From<tokio_tungstenite::tungstenite::Error> for RealtimeError {
156    fn from(error: tokio_tungstenite::tungstenite::Error) -> Self {
157        Self::WebSocket(Box::new(error))
158    }
159}
160
161/// Stateful realtime connection for issuing Inline RPC calls.
162pub struct RealtimeClient {
163    ws: tokio_tungstenite::WebSocketStream<
164        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
165    >,
166    seq: u32,
167    id_gen: IdGenerator,
168    rpc_timeout: Option<Duration>,
169}
170
171/// Server-pushed realtime event received outside a direct RPC result.
172#[derive(Clone, Debug, PartialEq)]
173#[non_exhaustive]
174pub enum RealtimeEvent {
175    /// Inline protocol updates pushed by the server.
176    Updates(Vec<proto::Update>),
177    /// Server acknowledgement for a previously sent client message.
178    Ack {
179        /// Client message ID acknowledged by the server.
180        msg_id: u64,
181    },
182    /// Server pong response.
183    Pong {
184        /// Pong nonce.
185        nonce: u64,
186    },
187}
188
189/// Builder for [`RealtimeClient`].
190#[must_use]
191#[derive(Clone)]
192pub struct RealtimeClientBuilder {
193    url: String,
194    token: String,
195    identity: ClientIdentity,
196    connect_timeout: Option<Duration>,
197    rpc_timeout: Option<Duration>,
198}
199
200impl fmt::Debug for RealtimeClientBuilder {
201    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
202        f.debug_struct("RealtimeClientBuilder")
203            .field("url", &realtime_url_for_debug(&self.url))
204            .field("token", &"<redacted>")
205            .field("identity", &self.identity)
206            .field("connect_timeout", &self.connect_timeout)
207            .field("rpc_timeout", &self.rpc_timeout)
208            .finish()
209    }
210}
211
212/// Typed Inline realtime RPC request.
213///
214/// This trait is implemented for generated protocol input types so callers can
215/// use [`RealtimeClient::call`] without manually pairing each input with its
216/// [`proto::Method`] and result oneof.
217pub trait RpcRequest: Sized {
218    /// Typed response returned by this request.
219    type Response;
220
221    /// RPC method associated with this input type.
222    const METHOD: proto::Method;
223
224    /// Converts this request into the generated RPC input oneof.
225    fn into_rpc_input(self) -> proto::rpc_call::Input;
226
227    /// Extracts the typed response from the generated RPC result oneof.
228    fn response_from_rpc_result(
229        result: proto::rpc_result::Result,
230    ) -> Result<Self::Response, RealtimeError>;
231}
232
233impl RealtimeClient {
234    /// Starts a realtime client builder.
235    pub fn builder(url: impl Into<String>, token: impl Into<String>) -> RealtimeClientBuilder {
236        RealtimeClientBuilder::new(url, token)
237    }
238
239    /// Connects to realtime using the default SDK identity.
240    pub async fn connect(url: &str, token: &str) -> Result<Self, RealtimeError> {
241        Self::builder(url, token).connect().await
242    }
243
244    /// Connects to realtime using a caller-provided client identity.
245    pub async fn connect_with_identity(
246        url: &str,
247        token: &str,
248        identity: ClientIdentity,
249    ) -> Result<Self, RealtimeError> {
250        Self::builder(url, token).identity(identity).connect().await
251    }
252}
253
254impl RealtimeClientBuilder {
255    /// Creates a realtime client builder with the default SDK identity.
256    pub fn new(url: impl Into<String>, token: impl Into<String>) -> Self {
257        Self {
258            url: url.into(),
259            token: token.into(),
260            identity: ClientIdentity::sdk(),
261            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
262            rpc_timeout: Some(DEFAULT_RPC_TIMEOUT),
263        }
264    }
265
266    /// Sets the client identity used in realtime headers and connection init.
267    pub fn identity(mut self, identity: ClientIdentity) -> Self {
268        self.identity = identity;
269        self
270    }
271
272    /// Sets the timeout for opening the WebSocket and receiving `ConnectionOpen`.
273    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
274        self.connect_timeout = Some(timeout);
275        self
276    }
277
278    /// Disables the connect timeout.
279    pub fn without_connect_timeout(mut self) -> Self {
280        self.connect_timeout = None;
281        self
282    }
283
284    /// Sets the timeout for each RPC invocation.
285    pub fn rpc_timeout(mut self, timeout: Duration) -> Self {
286        self.rpc_timeout = Some(timeout);
287        self
288    }
289
290    /// Disables the per-RPC timeout.
291    pub fn without_rpc_timeout(mut self) -> Self {
292        self.rpc_timeout = None;
293        self
294    }
295
296    /// Opens the WebSocket connection and waits for `ConnectionOpen`.
297    pub async fn connect(self) -> Result<RealtimeClient, RealtimeError> {
298        let url = normalize_realtime_url(self.url)?;
299        log::debug!(
300            target: "inline_sdk::realtime",
301            "opening realtime websocket url={} identity_type={} connect_timeout={:?} rpc_timeout={:?}",
302            realtime_url_for_log(&url),
303            self.identity.client_type(),
304            self.connect_timeout,
305            self.rpc_timeout
306        );
307        let mut request = url.into_client_request()?;
308        request.headers_mut().insert(
309            client_info::CLIENT_TYPE_HEADER,
310            realtime_header_value("client_type", self.identity.client_type())?,
311        );
312        request.headers_mut().insert(
313            client_info::CLIENT_VERSION_HEADER,
314            realtime_header_value("client_version", self.identity.client_version())?,
315        );
316        request.headers_mut().insert(
317            "user-agent",
318            realtime_header_value("user_agent", &client_info::user_agent_for(&self.identity))?,
319        );
320
321        let (ws, _) =
322            with_optional_timeout("connect", self.connect_timeout, connect_async(request)).await?;
323        log::debug!(target: "inline_sdk::realtime", "websocket connected");
324        let mut client = RealtimeClient {
325            ws,
326            seq: 0,
327            id_gen: IdGenerator::new(),
328            rpc_timeout: self.rpc_timeout,
329        };
330
331        with_optional_timeout(
332            "connection_init",
333            self.connect_timeout,
334            client.send_connection_init(&self.token, &self.identity),
335        )
336        .await?;
337        log::trace!(target: "inline_sdk::realtime", "connection init sent");
338        with_optional_timeout(
339            "connection_open",
340            self.connect_timeout,
341            client.wait_for_connection_open(),
342        )
343        .await?;
344        log::debug!(target: "inline_sdk::realtime", "realtime protocol open");
345        Ok(client)
346    }
347}
348
349impl RealtimeClient {
350    /// Invokes a typed Inline RPC request.
351    pub async fn call<R>(&mut self, request: R) -> Result<R::Response, RealtimeError>
352    where
353        R: RpcRequest,
354    {
355        log::trace!(
356            target: "inline_sdk::realtime",
357            "calling typed rpc method={}",
358            R::METHOD.as_str_name()
359        );
360        let result = self.invoke(R::METHOD, request.into_rpc_input()).await?;
361        R::response_from_rpc_result(result)
362    }
363
364    /// Invokes an Inline RPC method and waits for the matching result.
365    pub async fn invoke(
366        &mut self,
367        method: proto::Method,
368        input: proto::rpc_call::Input,
369    ) -> Result<proto::rpc_result::Result, RealtimeError> {
370        let rpc_call = proto::RpcCall {
371            method: method as i32,
372            input: Some(input),
373        };
374        let message_id = self.next_id();
375        log::trace!(
376            target: "inline_sdk::realtime",
377            "sending rpc method={} msg_id={message_id}",
378            method.as_str_name()
379        );
380        let message = proto::ClientMessage {
381            id: message_id,
382            seq: self.next_seq(),
383            body: Some(proto::client_message::Body::RpcCall(rpc_call)),
384        };
385
386        self.send_client_message(message).await?;
387
388        with_optional_timeout(
389            "rpc",
390            self.rpc_timeout,
391            self.wait_for_rpc_result(message_id),
392        )
393        .await
394    }
395
396    /// Returns the configured per-RPC timeout.
397    pub fn rpc_timeout(&self) -> Option<Duration> {
398        self.rpc_timeout
399    }
400
401    /// Waits for the next server-pushed realtime event.
402    ///
403    /// This reads the same Inline realtime protocol stream used for RPC calls.
404    /// Callers that need both request/response RPCs and long-lived pushed
405    /// updates should either serialize access to one [`RealtimeClient`] or use
406    /// a separate realtime connection for the event receiver.
407    pub async fn next_event(&mut self) -> Result<RealtimeEvent, RealtimeError> {
408        loop {
409            let message = self.read_server_message().await?;
410            if let Some(event) = self.event_from_server_message(message).await? {
411                return Ok(event);
412            }
413        }
414    }
415
416    async fn send_connection_init(
417        &mut self,
418        token: &str,
419        identity: &ClientIdentity,
420    ) -> Result<(), RealtimeError> {
421        let init = connection_init_for_token(token, identity);
422        let message = proto::ClientMessage {
423            id: self.next_id(),
424            seq: self.next_seq(),
425            body: Some(proto::client_message::Body::ConnectionInit(init)),
426        };
427
428        self.send_client_message(message).await
429    }
430
431    async fn wait_for_connection_open(&mut self) -> Result<(), RealtimeError> {
432        loop {
433            let message = self.read_server_message().await?;
434            let message_id = message.id;
435            match message.body {
436                Some(proto::server_protocol_message::Body::ConnectionOpen(_)) => return Ok(()),
437                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
438                    log::warn!(
439                        target: "inline_sdk::realtime",
440                        "connection open rejected reason={}",
441                        proto::connection_error::Reason::try_from(error.reason)
442                            .map(|reason| reason.as_str_name())
443                            .unwrap_or("UNKNOWN")
444                    );
445                    return Err(connection_error_from_proto(error));
446                }
447                Some(proto::server_protocol_message::Body::Message(message)) => {
448                    let _ = self.server_payload_event(message_id, message).await?;
449                }
450                _ => {}
451            }
452        }
453    }
454
455    async fn wait_for_rpc_result(
456        &mut self,
457        message_id: u64,
458    ) -> Result<proto::rpc_result::Result, RealtimeError> {
459        loop {
460            let message = self.read_server_message().await?;
461            match message.body {
462                Some(proto::server_protocol_message::Body::RpcResult(result))
463                    if result.req_msg_id == message_id =>
464                {
465                    log::trace!(
466                        target: "inline_sdk::realtime",
467                        "received rpc result msg_id={message_id}"
468                    );
469                    return result.result.ok_or(RealtimeError::MissingResult);
470                }
471                Some(proto::server_protocol_message::Body::RpcError(error))
472                    if error.req_msg_id == message_id =>
473                {
474                    let message = error.message;
475                    let error_name = rpc_error_code_name(error.error_code);
476                    let friendly =
477                        format_rpc_error(error.error_code, &error_name, &message, error.code);
478                    log::warn!(
479                        target: "inline_sdk::realtime",
480                        "received rpc error msg_id={message_id} error={error_name} status={}",
481                        error.code
482                    );
483                    return Err(RealtimeError::RpcError {
484                        code: error.code,
485                        error_code: error.error_code,
486                        error_name,
487                        message,
488                        friendly,
489                    });
490                }
491                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
492                    log::warn!(
493                        target: "inline_sdk::realtime",
494                        "connection error while waiting for rpc msg_id={message_id} reason={}",
495                        proto::connection_error::Reason::try_from(error.reason)
496                            .map(|reason| reason.as_str_name())
497                            .unwrap_or("UNKNOWN")
498                    );
499                    return Err(connection_error_from_proto(error));
500                }
501                Some(proto::server_protocol_message::Body::Message(server_message)) => {
502                    if let Some(event) = self
503                        .server_payload_event(message.id, server_message)
504                        .await?
505                    {
506                        log::trace!(
507                            target: "inline_sdk::realtime",
508                            "received pushed realtime event while waiting for rpc msg_id={message_id}: {}",
509                            realtime_event_kind(&event)
510                        );
511                    }
512                }
513                _ => {}
514            }
515        }
516    }
517
518    async fn event_from_server_message(
519        &mut self,
520        message: proto::ServerProtocolMessage,
521    ) -> Result<Option<RealtimeEvent>, RealtimeError> {
522        match message.body {
523            Some(proto::server_protocol_message::Body::Message(server_message)) => {
524                self.server_payload_event(message.id, server_message).await
525            }
526            Some(proto::server_protocol_message::Body::Ack(ack)) => {
527                Ok(Some(RealtimeEvent::Ack { msg_id: ack.msg_id }))
528            }
529            Some(proto::server_protocol_message::Body::Pong(pong)) => {
530                Ok(Some(RealtimeEvent::Pong { nonce: pong.nonce }))
531            }
532            Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
533                Err(connection_error_from_proto(error))
534            }
535            Some(proto::server_protocol_message::Body::ConnectionOpen(_))
536            | Some(proto::server_protocol_message::Body::RpcResult(_))
537            | Some(proto::server_protocol_message::Body::RpcError(_))
538            | None => Ok(None),
539        }
540    }
541
542    async fn server_payload_event(
543        &mut self,
544        message_id: u64,
545        message: proto::ServerMessage,
546    ) -> Result<Option<RealtimeEvent>, RealtimeError> {
547        match message.payload {
548            Some(proto::server_message::Payload::Update(payload)) => {
549                self.send_ack(message_id).await?;
550                Ok(Some(RealtimeEvent::Updates(payload.updates)))
551            }
552            None => Ok(None),
553        }
554    }
555
556    async fn send_ack(&mut self, msg_id: u64) -> Result<(), RealtimeError> {
557        let message = proto::ClientMessage {
558            id: self.next_id(),
559            seq: self.next_seq(),
560            body: Some(proto::client_message::Body::Ack(proto::Ack { msg_id })),
561        };
562        self.send_client_message(message).await
563    }
564
565    async fn send_client_message(
566        &mut self,
567        message: proto::ClientMessage,
568    ) -> Result<(), RealtimeError> {
569        let bytes = message.encode_to_vec();
570        self.ws.send(WsMessage::Binary(bytes)).await?;
571        Ok(())
572    }
573
574    async fn read_server_message(&mut self) -> Result<proto::ServerProtocolMessage, RealtimeError> {
575        loop {
576            let message = self
577                .ws
578                .next()
579                .await
580                .ok_or(RealtimeError::ConnectionClosed)??;
581            match message {
582                WsMessage::Binary(data) => {
583                    return Ok(proto::ServerProtocolMessage::decode(&*data)?);
584                }
585                WsMessage::Text(_) => continue,
586                WsMessage::Close(_) => return Err(RealtimeError::ConnectionClosed),
587                WsMessage::Ping(_) | WsMessage::Pong(_) => continue,
588                _ => continue,
589            }
590        }
591    }
592
593    fn next_seq(&mut self) -> u32 {
594        self.seq = self.seq.wrapping_add(1);
595        self.seq
596    }
597
598    fn next_id(&mut self) -> u64 {
599        self.id_gen.next_id()
600    }
601}
602
603macro_rules! rpc_requests {
604    ($(($input_ty:ident, $method:ident, $input_variant:ident, $result_ty:ident, $result_variant:ident)),+ $(,)?) => {
605        $(
606            impl RpcRequest for proto::$input_ty {
607                type Response = proto::$result_ty;
608
609                const METHOD: proto::Method = proto::Method::$method;
610
611                fn into_rpc_input(self) -> proto::rpc_call::Input {
612                    proto::rpc_call::Input::$input_variant(self)
613                }
614
615                fn response_from_rpc_result(
616                    result: proto::rpc_result::Result,
617                ) -> Result<Self::Response, RealtimeError> {
618                    match result {
619                        proto::rpc_result::Result::$result_variant(response) => Ok(response),
620                        other => {
621                            let actual = rpc_result_variant_name(&other);
622                            log::warn!(
623                                target: "inline_sdk::realtime",
624                                "unexpected rpc result method={} expected={} actual={actual}",
625                                Self::METHOD.as_str_name(),
626                                stringify!($result_variant)
627                            );
628                            Err(RealtimeError::UnexpectedResult {
629                                method: Self::METHOD.as_str_name(),
630                                expected: stringify!($result_variant),
631                                actual,
632                            })
633                        }
634                    }
635                }
636            }
637        )+
638
639        fn rpc_result_variant_name(result: &proto::rpc_result::Result) -> &'static str {
640            match result {
641                $(proto::rpc_result::Result::$result_variant(_) => stringify!($result_variant),)+
642            }
643        }
644    };
645}
646
647rpc_requests!(
648    (GetMeInput, GetMe, GetMe, GetMeResult, GetMe),
649    (
650        GetPeerPhotoInput,
651        GetPeerPhoto,
652        GetPeerPhoto,
653        GetPeerPhotoResult,
654        GetPeerPhoto
655    ),
656    (
657        DeleteMessagesInput,
658        DeleteMessages,
659        DeleteMessages,
660        DeleteMessagesResult,
661        DeleteMessages
662    ),
663    (
664        SendMessageInput,
665        SendMessage,
666        SendMessage,
667        SendMessageResult,
668        SendMessage
669    ),
670    (
671        GetChatHistoryInput,
672        GetChatHistory,
673        GetChatHistory,
674        GetChatHistoryResult,
675        GetChatHistory
676    ),
677    (
678        AddReactionInput,
679        AddReaction,
680        AddReaction,
681        AddReactionResult,
682        AddReaction
683    ),
684    (
685        DeleteReactionInput,
686        DeleteReaction,
687        DeleteReaction,
688        DeleteReactionResult,
689        DeleteReaction
690    ),
691    (
692        EditMessageInput,
693        EditMessage,
694        EditMessage,
695        EditMessageResult,
696        EditMessage
697    ),
698    (
699        CreateChatInput,
700        CreateChat,
701        CreateChat,
702        CreateChatResult,
703        CreateChat
704    ),
705    (
706        GetSpaceMembersInput,
707        GetSpaceMembers,
708        GetSpaceMembers,
709        GetSpaceMembersResult,
710        GetSpaceMembers
711    ),
712    (
713        DeleteChatInput,
714        DeleteChat,
715        DeleteChat,
716        DeleteChatResult,
717        DeleteChat
718    ),
719    (
720        InviteToSpaceInput,
721        InviteToSpace,
722        InviteToSpace,
723        InviteToSpaceResult,
724        InviteToSpace
725    ),
726    (
727        GetChatParticipantsInput,
728        GetChatParticipants,
729        GetChatParticipants,
730        GetChatParticipantsResult,
731        GetChatParticipants
732    ),
733    (
734        AddChatParticipantInput,
735        AddChatParticipant,
736        AddChatParticipant,
737        AddChatParticipantResult,
738        AddChatParticipant
739    ),
740    (
741        RemoveChatParticipantInput,
742        RemoveChatParticipant,
743        RemoveChatParticipant,
744        RemoveChatParticipantResult,
745        RemoveChatParticipant
746    ),
747    (
748        TranslateMessagesInput,
749        TranslateMessages,
750        TranslateMessages,
751        TranslateMessagesResult,
752        TranslateMessages
753    ),
754    (GetChatsInput, GetChats, GetChats, GetChatsResult, GetChats),
755    (
756        UpdateUserSettingsInput,
757        UpdateUserSettings,
758        UpdateUserSettings,
759        UpdateUserSettingsResult,
760        UpdateUserSettings
761    ),
762    (
763        GetUserSettingsInput,
764        GetUserSettings,
765        GetUserSettings,
766        GetUserSettingsResult,
767        GetUserSettings
768    ),
769    (
770        SendComposeActionInput,
771        SendComposeAction,
772        SendComposeAction,
773        SendComposeActionResult,
774        SendComposeAction
775    ),
776    (
777        CreateBotInput,
778        CreateBot,
779        CreateBot,
780        CreateBotResult,
781        CreateBot
782    ),
783    (
784        DeleteMemberInput,
785        DeleteMember,
786        DeleteMember,
787        DeleteMemberResult,
788        DeleteMember
789    ),
790    (
791        MarkAsUnreadInput,
792        MarkAsUnread,
793        MarkAsUnread,
794        MarkAsUnreadResult,
795        MarkAsUnread
796    ),
797    (
798        GetUpdatesStateInput,
799        GetUpdatesState,
800        GetUpdatesState,
801        GetUpdatesStateResult,
802        GetUpdatesState
803    ),
804    (GetChatInput, GetChat, GetChat, GetChatResult, GetChat),
805    (
806        GetUpdatesInput,
807        GetUpdates,
808        GetUpdates,
809        GetUpdatesResult,
810        GetUpdates
811    ),
812    (
813        UpdateMemberAccessInput,
814        UpdateMemberAccess,
815        UpdateMemberAccess,
816        UpdateMemberAccessResult,
817        UpdateMemberAccess
818    ),
819    (
820        SearchMessagesInput,
821        SearchMessages,
822        SearchMessages,
823        SearchMessagesResult,
824        SearchMessages
825    ),
826    (
827        ForwardMessagesInput,
828        ForwardMessages,
829        ForwardMessages,
830        ForwardMessagesResult,
831        ForwardMessages
832    ),
833    (
834        UpdateChatVisibilityInput,
835        UpdateChatVisibility,
836        UpdateChatVisibility,
837        UpdateChatVisibilityResult,
838        UpdateChatVisibility
839    ),
840    (
841        PinMessageInput,
842        PinMessage,
843        PinMessage,
844        PinMessageResult,
845        PinMessage
846    ),
847    (
848        UpdateChatInfoInput,
849        UpdateChatInfo,
850        UpdateChatInfo,
851        UpdateChatInfoResult,
852        UpdateChatInfo
853    ),
854    (ListBotsInput, ListBots, ListBots, ListBotsResult, ListBots),
855    (
856        RevealBotTokenInput,
857        RevealBotToken,
858        RevealBotToken,
859        RevealBotTokenResult,
860        RevealBotToken
861    ),
862    (
863        MoveThreadInput,
864        MoveThread,
865        MoveThread,
866        MoveThreadResult,
867        MoveThread
868    ),
869    (
870        RotateBotTokenInput,
871        RotateBotToken,
872        RotateBotToken,
873        RotateBotTokenResult,
874        RotateBotToken
875    ),
876    (
877        UpdateBotProfileInput,
878        UpdateBotProfile,
879        UpdateBotProfile,
880        UpdateBotProfileResult,
881        UpdateBotProfile
882    ),
883    (
884        GetMessagesInput,
885        GetMessages,
886        GetMessages,
887        GetMessagesResult,
888        GetMessages
889    ),
890    (
891        UpdateDialogNotificationSettingsInput,
892        UpdateDialogNotificationSettings,
893        UpdateDialogNotificationSettings,
894        UpdateDialogNotificationSettingsResult,
895        UpdateDialogNotificationSettings
896    ),
897    (
898        ReadMessagesInput,
899        ReadMessages,
900        ReadMessages,
901        ReadMessagesResult,
902        ReadMessages
903    ),
904    (
905        UpdatePushNotificationDetailsInput,
906        UpdatePushNotificationDetails,
907        UpdatePushNotificationDetails,
908        UpdatePushNotificationDetailsResult,
909        UpdatePushNotificationDetails
910    ),
911    (
912        CreateSubthreadInput,
913        CreateSubthread,
914        CreateSubthread,
915        CreateSubthreadResult,
916        CreateSubthread
917    ),
918    (
919        GetBotCommandsInput,
920        GetBotCommands,
921        GetBotCommands,
922        GetBotCommandsResult,
923        GetBotCommands
924    ),
925    (
926        SetBotCommandsInput,
927        SetBotCommands,
928        SetBotCommands,
929        SetBotCommandsResult,
930        SetBotCommands
931    ),
932    (
933        GetPeerBotCommandsInput,
934        GetPeerBotCommands,
935        GetPeerBotCommands,
936        GetPeerBotCommandsResult,
937        GetPeerBotCommands
938    ),
939    (
940        ShowInChatListInput,
941        ShowInChatList,
942        ShowInChatList,
943        ShowInChatListResult,
944        ShowInChatList
945    ),
946    (
947        ReserveChatIdsInput,
948        ReserveChatIds,
949        ReserveChatIds,
950        ReserveChatIdsResult,
951        ReserveChatIds
952    ),
953    (
954        InvokeMessageActionInput,
955        InvokeMessageAction,
956        InvokeMessageAction,
957        InvokeMessageActionResult,
958        InvokeMessageAction
959    ),
960    (
961        AnswerMessageActionInput,
962        AnswerMessageAction,
963        AnswerMessageAction,
964        AnswerMessageActionResult,
965        AnswerMessageAction
966    ),
967    (
968        RevokeSessionInput,
969        RevokeSession,
970        RevokeSession,
971        RevokeSessionResult,
972        RevokeSession
973    ),
974    (
975        UpdateDialogOpenInput,
976        UpdateDialogOpen,
977        UpdateDialogOpen,
978        UpdateDialogOpenResult,
979        UpdateDialogOpen
980    ),
981    (
982        UpdateDialogOrderInput,
983        UpdateDialogOrder,
984        UpdateDialogOrder,
985        UpdateDialogOrderResult,
986        UpdateDialogOrder
987    ),
988    (
989        ClearChatHistoryInput,
990        ClearChatHistory,
991        ClearChatHistory,
992        ClearChatHistoryResult,
993        ClearChatHistory
994    ),
995    (
996        DeleteBotInput,
997        DeleteBot,
998        DeleteBot,
999        DeleteBotResult,
1000        DeleteBot
1001    ),
1002    (
1003        DeleteMessageAttachmentInput,
1004        DeleteMessageAttachment,
1005        DeleteMessageAttachment,
1006        DeleteMessageAttachmentResult,
1007        DeleteMessageAttachment
1008    ),
1009    (
1010        SetBotAvatarInput,
1011        SetBotAvatar,
1012        SetBotAvatar,
1013        SetBotAvatarResult,
1014        SetBotAvatar
1015    ),
1016    (
1017        ClearBotAvatarInput,
1018        ClearBotAvatar,
1019        ClearBotAvatar,
1020        ClearBotAvatarResult,
1021        ClearBotAvatar
1022    ),
1023    (
1024        GetBotPresenceInput,
1025        GetBotPresence,
1026        GetBotPresence,
1027        GetBotPresenceResult,
1028        GetBotPresence
1029    ),
1030    (
1031        SetBotPresenceStateInput,
1032        SetBotPresenceState,
1033        SetBotPresenceState,
1034        SetBotPresenceStateResult,
1035        SetBotPresenceState
1036    ),
1037    (
1038        UpdateDialogFollowModeInput,
1039        UpdateDialogFollowMode,
1040        UpdateDialogFollowMode,
1041        UpdateDialogFollowModeResult,
1042        UpdateDialogFollowMode
1043    ),
1044    (
1045        GetSessionsInput,
1046        GetSessions,
1047        GetSessions,
1048        GetSessionsResult,
1049        GetSessions
1050    ),
1051    (
1052        CheckUsernameInput,
1053        CheckUsername,
1054        CheckUsername,
1055        CheckUsernameResult,
1056        CheckUsername
1057    ),
1058    (
1059        ChangeUsernameInput,
1060        ChangeUsername,
1061        ChangeUsername,
1062        ChangeUsernameResult,
1063        ChangeUsername
1064    ),
1065    (
1066        UpdateProfileInput,
1067        UpdateProfile,
1068        UpdateProfile,
1069        UpdateProfileResult,
1070        UpdateProfile
1071    ),
1072    (
1073        GetSpaceUrlPreviewExclusionsInput,
1074        GetSpaceUrlPreviewExclusions,
1075        GetSpaceUrlPreviewExclusions,
1076        GetSpaceUrlPreviewExclusionsResult,
1077        GetSpaceUrlPreviewExclusions
1078    ),
1079    (
1080        AddSpaceUrlPreviewExclusionInput,
1081        AddSpaceUrlPreviewExclusion,
1082        AddSpaceUrlPreviewExclusion,
1083        AddSpaceUrlPreviewExclusionResult,
1084        AddSpaceUrlPreviewExclusion
1085    ),
1086    (
1087        RemoveSpaceUrlPreviewExclusionInput,
1088        RemoveSpaceUrlPreviewExclusion,
1089        RemoveSpaceUrlPreviewExclusion,
1090        RemoveSpaceUrlPreviewExclusionResult,
1091        RemoveSpaceUrlPreviewExclusion
1092    ),
1093);
1094
1095fn connection_init_for_token(token: &str, identity: &ClientIdentity) -> proto::ConnectionInit {
1096    proto::ConnectionInit {
1097        token: token.to_string(),
1098        build_number: None,
1099        layer: None,
1100        client_version: Some(identity.client_version().to_string()),
1101        os_version: client_info::current_os_version(),
1102    }
1103}
1104
1105fn rpc_error_code_name(error_code: i32) -> String {
1106    proto::rpc_error::Code::try_from(error_code)
1107        .map(|code| code.as_str_name())
1108        .unwrap_or("UNKNOWN")
1109        .to_string()
1110}
1111
1112fn format_rpc_error(error_code: i32, error_name: &str, message: &str, status_code: i32) -> String {
1113    let label = match error_name {
1114        "UNKNOWN" => "Unknown RPC error",
1115        "BAD_REQUEST" => "Bad request",
1116        "UNAUTHENTICATED" => "Not authenticated",
1117        "RATE_LIMIT" => "Rate limited",
1118        "INTERNAL_ERROR" => "Internal server error",
1119        "PEER_ID_INVALID" => "Invalid peer (chat/user id)",
1120        "MESSAGE_ID_INVALID" => "Invalid message id",
1121        "USER_ID_INVALID" => "Invalid user id",
1122        "USER_ALREADY_MEMBER" => "User already in chat/space",
1123        "SPACE_ID_INVALID" => "Invalid space id",
1124        "CHAT_ID_INVALID" => "Invalid chat id",
1125        "EMAIL_INVALID" => "Invalid email address",
1126        "PHONE_NUMBER_INVALID" => "Invalid phone number",
1127        "SPACE_ADMIN_REQUIRED" => "Space admin required",
1128        "SPACE_OWNER_REQUIRED" => "Space owner required",
1129        "USERNAME_INVALID" => "Invalid username",
1130        "USERNAME_TAKEN" => "Username already taken",
1131        "FIRST_NAME_INVALID" => "Invalid first name",
1132        _ => "Unknown RPC error",
1133    };
1134
1135    let mut formatted = String::from(label);
1136    if error_name == "UNKNOWN" && error_code != 0 {
1137        formatted.push_str(&format!(" {error_code}"));
1138    }
1139    if !message.is_empty() && !message.eq_ignore_ascii_case(label) {
1140        formatted.push_str(": ");
1141        formatted.push_str(message);
1142    }
1143    if status_code != 0 {
1144        formatted.push_str(&format!(" (HTTP {status_code})"));
1145    }
1146    formatted
1147}
1148
1149fn connection_error_from_proto(error: proto::ConnectionError) -> RealtimeError {
1150    let reason = error.reason;
1151    let reason_name = proto::connection_error::Reason::try_from(reason)
1152        .map(|reason| reason.as_str_name())
1153        .unwrap_or("UNKNOWN")
1154        .to_string();
1155    let friendly = format_connection_error(reason, &reason_name);
1156    RealtimeError::ConnectionError {
1157        reason,
1158        reason_name,
1159        friendly,
1160    }
1161}
1162
1163fn format_connection_error(reason: i32, reason_name: &str) -> String {
1164    match reason_name {
1165        "UNAUTHORIZED" => "Realtime connection unauthorized".to_string(),
1166        "INVALID_AUTH" => "Realtime auth token is invalid".to_string(),
1167        "SESSION_REVOKED" => "Realtime session was revoked".to_string(),
1168        "REASON_UNSPECIFIED" => "Realtime connection rejected".to_string(),
1169        _ => format!("Realtime connection rejected: unknown reason {reason}"),
1170    }
1171}
1172
1173fn normalize_realtime_url(url: impl Into<String>) -> Result<Url, RealtimeError> {
1174    let original = url.into();
1175    let normalized = original.trim().to_string();
1176    if normalized.is_empty() {
1177        return Err(RealtimeError::InvalidUrl {
1178            url: original,
1179            message: "realtime URL cannot be empty".to_string(),
1180        });
1181    }
1182
1183    let parsed = Url::parse(&normalized).map_err(|err| RealtimeError::InvalidUrl {
1184        url: normalized.clone(),
1185        message: err.to_string(),
1186    })?;
1187
1188    if !matches!(parsed.scheme(), "ws" | "wss") {
1189        return Err(RealtimeError::InvalidUrl {
1190            url: normalized,
1191            message: "scheme must be ws or wss".to_string(),
1192        });
1193    }
1194
1195    if parsed.host_str().is_none() {
1196        return Err(RealtimeError::InvalidUrl {
1197            url: normalized,
1198            message: "host is required".to_string(),
1199        });
1200    }
1201
1202    if !parsed.username().is_empty() || parsed.password().is_some() {
1203        return Err(RealtimeError::InvalidUrl {
1204            url: normalized,
1205            message: "credentials are not valid in the realtime URL".to_string(),
1206        });
1207    }
1208
1209    if parsed.fragment().is_some() {
1210        return Err(RealtimeError::InvalidUrl {
1211            url: normalized,
1212            message: "fragments are not valid in the realtime URL".to_string(),
1213        });
1214    }
1215
1216    Ok(parsed)
1217}
1218
1219fn realtime_url_for_log(url: &Url) -> String {
1220    let host = url.host_str().unwrap_or("<missing-host>");
1221    let port = url
1222        .port()
1223        .map(|port| format!(":{port}"))
1224        .unwrap_or_default();
1225    format!("{}://{}{}{}", url.scheme(), host, port, url.path())
1226}
1227
1228fn realtime_url_for_debug(raw_url: &str) -> String {
1229    Url::parse(raw_url.trim())
1230        .map(|url| realtime_url_for_log(&url))
1231        .unwrap_or_else(|_| "<invalid>".to_string())
1232}
1233
1234fn realtime_header_value(field: &'static str, value: &str) -> Result<HeaderValue, RealtimeError> {
1235    HeaderValue::from_str(value).map_err(|_| RealtimeError::InvalidHeaderValue { field })
1236}
1237
1238fn realtime_event_kind(event: &RealtimeEvent) -> &'static str {
1239    match event {
1240        RealtimeEvent::Updates(_) => "updates",
1241        RealtimeEvent::Ack { .. } => "ack",
1242        RealtimeEvent::Pong { .. } => "pong",
1243    }
1244}
1245
1246async fn with_optional_timeout<T, E, Fut>(
1247    operation: &'static str,
1248    timeout: Option<Duration>,
1249    future: Fut,
1250) -> Result<T, RealtimeError>
1251where
1252    Fut: Future<Output = Result<T, E>>,
1253    RealtimeError: From<E>,
1254{
1255    match timeout {
1256        Some(timeout) => tokio::time::timeout(timeout, future)
1257            .await
1258            .map_err(|_| RealtimeError::Timeout { operation, timeout })?
1259            .map_err(RealtimeError::from),
1260        None => future.await.map_err(RealtimeError::from),
1261    }
1262}
1263
1264struct IdGenerator {
1265    last_timestamp: u64,
1266    sequence: u32,
1267}
1268
1269impl IdGenerator {
1270    fn new() -> Self {
1271        Self {
1272            last_timestamp: 0,
1273            sequence: 0,
1274        }
1275    }
1276
1277    fn next_id(&mut self) -> u64 {
1278        let timestamp = current_epoch_seconds().saturating_sub(EPOCH_SECONDS);
1279        if timestamp == self.last_timestamp {
1280            self.sequence = self.sequence.wrapping_add(1);
1281        } else {
1282            self.sequence = 0;
1283            self.last_timestamp = timestamp;
1284        }
1285
1286        (timestamp << 32) | self.sequence as u64
1287    }
1288}
1289
1290const EPOCH_SECONDS: u64 = 1_735_689_600; // 2025-01-01T00:00:00Z
1291
1292fn current_epoch_seconds() -> u64 {
1293    use std::time::{SystemTime, UNIX_EPOCH};
1294    SystemTime::now()
1295        .duration_since(UNIX_EPOCH)
1296        .unwrap_or_default()
1297        .as_secs()
1298}
1299
1300#[cfg(test)]
1301mod tests {
1302    use super::*;
1303    use tokio::net::{TcpListener, TcpStream};
1304    use tokio_tungstenite::WebSocketStream;
1305    use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
1306    use tokio_tungstenite::{accept_async, accept_hdr_async};
1307
1308    #[test]
1309    fn rpc_error_code_name_uses_stable_proto_name() {
1310        assert_eq!(
1311            rpc_error_code_name(proto::rpc_error::Code::PeerIdInvalid as i32),
1312            "PEER_ID_INVALID"
1313        );
1314        assert_eq!(rpc_error_code_name(999), "UNKNOWN");
1315    }
1316
1317    #[test]
1318    fn rpc_error_formatter_covers_new_profile_codes() {
1319        assert_eq!(
1320            format_rpc_error(
1321                proto::rpc_error::Code::UsernameInvalid as i32,
1322                "USERNAME_INVALID",
1323                "",
1324                400
1325            ),
1326            "Invalid username (HTTP 400)"
1327        );
1328        assert_eq!(
1329            format_rpc_error(
1330                proto::rpc_error::Code::UsernameTaken as i32,
1331                "USERNAME_TAKEN",
1332                "handle exists",
1333                409
1334            ),
1335            "Username already taken: handle exists (HTTP 409)"
1336        );
1337        assert_eq!(
1338            format_rpc_error(
1339                proto::rpc_error::Code::FirstNameInvalid as i32,
1340                "FIRST_NAME_INVALID",
1341                "",
1342                400
1343            ),
1344            "Invalid first name (HTTP 400)"
1345        );
1346    }
1347
1348    #[test]
1349    fn connection_error_preserves_proto_reason() {
1350        let err = connection_error_from_proto(proto::ConnectionError {
1351            reason: proto::connection_error::Reason::InvalidAuth as i32,
1352        });
1353
1354        match err {
1355            RealtimeError::ConnectionError {
1356                reason,
1357                reason_name,
1358                friendly,
1359            } => {
1360                assert_eq!(reason, 2);
1361                assert_eq!(reason_name, "INVALID_AUTH");
1362                assert_eq!(friendly, "Realtime auth token is invalid");
1363            }
1364            other => panic!("expected connection error, got {other:?}"),
1365        }
1366    }
1367
1368    #[test]
1369    fn unknown_connection_error_reason_is_preserved() {
1370        let err = connection_error_from_proto(proto::ConnectionError { reason: 999 });
1371
1372        match err {
1373            RealtimeError::ConnectionError {
1374                reason,
1375                reason_name,
1376                friendly,
1377            } => {
1378                assert_eq!(reason, 999);
1379                assert_eq!(reason_name, "UNKNOWN");
1380                assert_eq!(friendly, "Realtime connection rejected: unknown reason 999");
1381            }
1382            other => panic!("expected connection error, got {other:?}"),
1383        }
1384    }
1385
1386    #[test]
1387    fn connection_init_uses_custom_client_identity() {
1388        let init =
1389            connection_init_for_token("token-1", &ClientIdentity::new("integration-test", "9.9.9"));
1390
1391        assert_eq!(init.token, "token-1");
1392        assert_eq!(init.client_version.as_deref(), Some("9.9.9"));
1393        assert!(init.build_number.is_none());
1394        assert!(init.layer.is_none());
1395    }
1396
1397    #[tokio::test]
1398    async fn realtime_client_connects_and_calls_get_me_against_local_server() {
1399        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1400        let addr = listener.local_addr().unwrap();
1401        let server = tokio::spawn(async move {
1402            let (stream, _) = listener.accept().await.unwrap();
1403            let mut ws = accept_hdr_async(stream, |request: &Request, response: Response| {
1404                assert_eq!(
1405                    request
1406                        .headers()
1407                        .get(client_info::CLIENT_TYPE_HEADER)
1408                        .and_then(|value| value.to_str().ok()),
1409                    Some("transport-test")
1410                );
1411                assert_eq!(
1412                    request
1413                        .headers()
1414                        .get(client_info::CLIENT_VERSION_HEADER)
1415                        .and_then(|value| value.to_str().ok()),
1416                    Some("1.2.3")
1417                );
1418                Ok(response)
1419            })
1420            .await
1421            .unwrap();
1422
1423            let init = read_test_client_message(&mut ws).await;
1424            match init.body {
1425                Some(proto::client_message::Body::ConnectionInit(init)) => {
1426                    assert_eq!(init.token, "token-1");
1427                    assert_eq!(init.client_version.as_deref(), Some("1.2.3"));
1428                    assert!(init.os_version.is_some());
1429                }
1430                other => panic!("expected connection init, got {other:?}"),
1431            }
1432            send_test_server_message(
1433                &mut ws,
1434                proto::ServerProtocolMessage {
1435                    id: 1,
1436                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1437                        proto::ConnectionOpen {},
1438                    )),
1439                },
1440            )
1441            .await;
1442
1443            let rpc = read_test_client_message(&mut ws).await;
1444            match &rpc.body {
1445                Some(proto::client_message::Body::RpcCall(call)) => {
1446                    assert_eq!(call.method, proto::Method::GetMe as i32);
1447                    assert!(matches!(call.input, Some(proto::rpc_call::Input::GetMe(_))));
1448                }
1449                other => panic!("expected getMe rpc call, got {other:?}"),
1450            }
1451            send_test_server_message(
1452                &mut ws,
1453                proto::ServerProtocolMessage {
1454                    id: 2,
1455                    body: Some(proto::server_protocol_message::Body::RpcResult(
1456                        proto::RpcResult {
1457                            req_msg_id: rpc.id,
1458                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
1459                                user: Some(proto::User {
1460                                    id: 42,
1461                                    first_name: Some("Ada".to_string()),
1462                                    ..Default::default()
1463                                }),
1464                            })),
1465                        },
1466                    )),
1467                },
1468            )
1469            .await;
1470        });
1471
1472        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1473            .identity(ClientIdentity::new("transport-test", "1.2.3"))
1474            .without_connect_timeout()
1475            .without_rpc_timeout()
1476            .connect()
1477            .await
1478            .unwrap();
1479        let result = client.call(proto::GetMeInput {}).await.unwrap();
1480
1481        assert_eq!(result.user.unwrap().id, 42);
1482        server.await.unwrap();
1483    }
1484
1485    #[tokio::test]
1486    async fn realtime_client_receives_update_events_and_acks_them() {
1487        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1488        let addr = listener.local_addr().unwrap();
1489        let server = tokio::spawn(async move {
1490            let (stream, _) = listener.accept().await.unwrap();
1491            let mut ws = accept_async(stream).await.unwrap();
1492
1493            let init = read_test_client_message(&mut ws).await;
1494            assert!(matches!(
1495                init.body,
1496                Some(proto::client_message::Body::ConnectionInit(_))
1497            ));
1498            send_test_server_message(
1499                &mut ws,
1500                proto::ServerProtocolMessage {
1501                    id: 1,
1502                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1503                        proto::ConnectionOpen {},
1504                    )),
1505                },
1506            )
1507            .await;
1508            send_test_server_message(&mut ws, test_update_server_message(9, 77)).await;
1509
1510            let ack = read_test_client_message(&mut ws).await;
1511            match ack.body {
1512                Some(proto::client_message::Body::Ack(ack)) => {
1513                    assert_eq!(ack.msg_id, 9);
1514                }
1515                other => panic!("expected ack for pushed update, got {other:?}"),
1516            }
1517        });
1518
1519        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1520            .without_connect_timeout()
1521            .without_rpc_timeout()
1522            .connect()
1523            .await
1524            .unwrap();
1525        let event = client.next_event().await.unwrap();
1526
1527        match event {
1528            RealtimeEvent::Updates(updates) => {
1529                assert_eq!(updates.len(), 1);
1530                match updates[0].update.as_ref() {
1531                    Some(proto::update::Update::NewMessage(update)) => {
1532                        assert_eq!(update.message.as_ref().unwrap().id, 77);
1533                    }
1534                    other => panic!("expected new message update, got {other:?}"),
1535                }
1536            }
1537            other => panic!("expected updates event, got {other:?}"),
1538        }
1539        server.await.unwrap();
1540    }
1541
1542    #[tokio::test]
1543    async fn realtime_rpc_wait_acks_pushed_updates_before_matching_result() {
1544        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1545        let addr = listener.local_addr().unwrap();
1546        let server = tokio::spawn(async move {
1547            let (stream, _) = listener.accept().await.unwrap();
1548            let mut ws = accept_async(stream).await.unwrap();
1549
1550            let init = read_test_client_message(&mut ws).await;
1551            assert!(matches!(
1552                init.body,
1553                Some(proto::client_message::Body::ConnectionInit(_))
1554            ));
1555            send_test_server_message(
1556                &mut ws,
1557                proto::ServerProtocolMessage {
1558                    id: 1,
1559                    body: Some(proto::server_protocol_message::Body::ConnectionOpen(
1560                        proto::ConnectionOpen {},
1561                    )),
1562                },
1563            )
1564            .await;
1565
1566            let rpc = read_test_client_message(&mut ws).await;
1567            send_test_server_message(&mut ws, test_update_server_message(10, 88)).await;
1568
1569            let ack = read_test_client_message(&mut ws).await;
1570            match ack.body {
1571                Some(proto::client_message::Body::Ack(ack)) => {
1572                    assert_eq!(ack.msg_id, 10);
1573                }
1574                other => panic!("expected ack while waiting for rpc result, got {other:?}"),
1575            }
1576
1577            send_test_server_message(
1578                &mut ws,
1579                proto::ServerProtocolMessage {
1580                    id: 2,
1581                    body: Some(proto::server_protocol_message::Body::RpcResult(
1582                        proto::RpcResult {
1583                            req_msg_id: rpc.id,
1584                            result: Some(proto::rpc_result::Result::GetMe(proto::GetMeResult {
1585                                user: Some(proto::User {
1586                                    id: 42,
1587                                    ..Default::default()
1588                                }),
1589                            })),
1590                        },
1591                    )),
1592                },
1593            )
1594            .await;
1595        });
1596
1597        let mut client = RealtimeClient::builder(format!("ws://{addr}/realtime"), "token-1")
1598            .without_connect_timeout()
1599            .without_rpc_timeout()
1600            .connect()
1601            .await
1602            .unwrap();
1603        let result = client.call(proto::GetMeInput {}).await.unwrap();
1604
1605        assert_eq!(result.user.unwrap().id, 42);
1606        server.await.unwrap();
1607    }
1608
1609    #[test]
1610    fn realtime_url_validation_accepts_ws_urls() {
1611        let url = normalize_realtime_url(" wss://api.inline.chat/realtime?edge=iad ").unwrap();
1612        assert_eq!(url.as_str(), "wss://api.inline.chat/realtime?edge=iad");
1613    }
1614
1615    #[test]
1616    fn realtime_url_validation_rejects_invalid_urls() {
1617        let err = normalize_realtime_url("inline.test").unwrap_err();
1618        match err {
1619            RealtimeError::InvalidUrl { url, message } => {
1620                assert_eq!(url, "inline.test");
1621                assert!(message.contains("relative URL without a base"));
1622            }
1623            other => panic!("expected invalid URL, got {other:?}"),
1624        }
1625
1626        let err = normalize_realtime_url("https://api.inline.chat/realtime").unwrap_err();
1627        match err {
1628            RealtimeError::InvalidUrl { message, .. } => {
1629                assert_eq!(message, "scheme must be ws or wss");
1630            }
1631            other => panic!("expected invalid URL, got {other:?}"),
1632        }
1633
1634        let err = normalize_realtime_url("wss://user:secret@api.inline.chat/realtime").unwrap_err();
1635        match &err {
1636            RealtimeError::InvalidUrl { message, .. } => {
1637                assert_eq!(message, "credentials are not valid in the realtime URL");
1638                assert!(!err.to_string().contains("secret"));
1639            }
1640            other => panic!("expected invalid URL, got {other:?}"),
1641        }
1642
1643        let err = normalize_realtime_url("wss://api.inline.chat/realtime#token").unwrap_err();
1644        match err {
1645            RealtimeError::InvalidUrl { message, .. } => {
1646                assert_eq!(message, "fragments are not valid in the realtime URL");
1647            }
1648            other => panic!("expected invalid URL, got {other:?}"),
1649        }
1650    }
1651
1652    #[test]
1653    fn realtime_invalid_url_debug_redacts_unsafe_url_parts() {
1654        let err = normalize_realtime_url(
1655            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret#frag",
1656        )
1657        .unwrap_err();
1658        let debug = format!("{err:?}");
1659
1660        assert!(debug.contains("wss://api.inline.chat/realtime"));
1661        assert!(!debug.contains("url-secret"));
1662        assert!(!debug.contains("query-secret"));
1663    }
1664
1665    #[test]
1666    fn realtime_url_for_log_omits_query_and_fragment() {
1667        let url = Url::parse("wss://api.inline.chat/realtime?token=secret#frag").unwrap();
1668        assert_eq!(realtime_url_for_log(&url), "wss://api.inline.chat/realtime");
1669    }
1670
1671    #[test]
1672    fn realtime_header_value_rejects_invalid_values() {
1673        let err = realtime_header_value("user_agent", "bad\nvalue").unwrap_err();
1674        match err {
1675            RealtimeError::InvalidHeaderValue { field } => {
1676                assert_eq!(field, "user_agent");
1677            }
1678            other => panic!("expected invalid header value, got {other:?}"),
1679        }
1680    }
1681
1682    #[test]
1683    fn realtime_builder_keeps_custom_identity_and_default_timeouts() {
1684        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
1685            .identity(ClientIdentity::new("agent", "0.2.0"));
1686
1687        assert_eq!(builder.url, "wss://api.inline.chat/realtime");
1688        assert_eq!(builder.token, "token-1");
1689        assert_eq!(builder.identity.client_type(), "agent");
1690        assert_eq!(builder.identity.client_version(), "0.2.0");
1691        assert_eq!(builder.connect_timeout, Some(DEFAULT_CONNECT_TIMEOUT));
1692        assert_eq!(builder.rpc_timeout, Some(DEFAULT_RPC_TIMEOUT));
1693    }
1694
1695    #[test]
1696    fn realtime_builder_debug_redacts_token() {
1697        let builder = RealtimeClient::builder(
1698            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret",
1699            "secret-token-1",
1700        );
1701        let debug = format!("{builder:?}");
1702
1703        assert!(debug.contains("RealtimeClientBuilder"));
1704        assert!(debug.contains("wss://api.inline.chat/realtime"));
1705        assert!(debug.contains("<redacted>"));
1706        assert!(!debug.contains("secret-token-1"));
1707        assert!(!debug.contains("url-secret"));
1708        assert!(!debug.contains("query-secret"));
1709    }
1710
1711    #[test]
1712    fn realtime_builder_can_override_or_disable_timeouts() {
1713        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
1714            .connect_timeout(Duration::from_secs(5))
1715            .rpc_timeout(Duration::from_secs(10));
1716
1717        assert_eq!(builder.connect_timeout, Some(Duration::from_secs(5)));
1718        assert_eq!(builder.rpc_timeout, Some(Duration::from_secs(10)));
1719
1720        let builder = builder.without_connect_timeout().without_rpc_timeout();
1721        assert_eq!(builder.connect_timeout, None);
1722        assert_eq!(builder.rpc_timeout, None);
1723    }
1724
1725    #[test]
1726    fn typed_rpc_request_maps_method_input_and_result() {
1727        assert_eq!(
1728            <proto::GetChatsInput as RpcRequest>::METHOD,
1729            proto::Method::GetChats
1730        );
1731
1732        let input = proto::GetChatsInput {};
1733        match input.into_rpc_input() {
1734            proto::rpc_call::Input::GetChats(_) => {}
1735            other => panic!("expected GetChats input, got {other:?}"),
1736        }
1737
1738        let response = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
1739            proto::rpc_result::Result::GetChats(proto::GetChatsResult::default()),
1740        )
1741        .unwrap();
1742        assert!(response.dialogs.is_empty());
1743    }
1744
1745    #[test]
1746    fn typed_rpc_request_rejects_unexpected_result_variant() {
1747        let err = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
1748            proto::rpc_result::Result::GetMe(proto::GetMeResult::default()),
1749        )
1750        .unwrap_err();
1751
1752        match err {
1753            RealtimeError::UnexpectedResult {
1754                method,
1755                expected,
1756                actual,
1757            } => {
1758                assert_eq!(method, "GET_CHATS");
1759                assert_eq!(expected, "GetChats");
1760                assert_eq!(actual, "GetMe");
1761            }
1762            other => panic!("expected unexpected result error, got {other:?}"),
1763        }
1764    }
1765
1766    #[tokio::test]
1767    async fn optional_timeout_reports_elapsed_operation() {
1768        let err = with_optional_timeout("test", Some(Duration::from_millis(1)), async {
1769            tokio::time::sleep(Duration::from_millis(20)).await;
1770            Ok::<(), RealtimeError>(())
1771        })
1772        .await
1773        .unwrap_err();
1774
1775        match err {
1776            RealtimeError::Timeout { operation, timeout } => {
1777                assert_eq!(operation, "test");
1778                assert_eq!(timeout, Duration::from_millis(1));
1779            }
1780            other => panic!("expected timeout, got {other:?}"),
1781        }
1782    }
1783
1784    #[tokio::test]
1785    async fn optional_timeout_can_be_disabled() {
1786        with_optional_timeout("test", None, async { Ok::<_, RealtimeError>("done") })
1787            .await
1788            .unwrap();
1789    }
1790
1791    async fn read_test_client_message(ws: &mut WebSocketStream<TcpStream>) -> proto::ClientMessage {
1792        loop {
1793            match ws.next().await.unwrap().unwrap() {
1794                WsMessage::Binary(bytes) => {
1795                    return proto::ClientMessage::decode(&*bytes).unwrap();
1796                }
1797                WsMessage::Ping(_) | WsMessage::Pong(_) | WsMessage::Text(_) => continue,
1798                other => panic!("unexpected websocket message: {other:?}"),
1799            }
1800        }
1801    }
1802
1803    async fn send_test_server_message(
1804        ws: &mut WebSocketStream<TcpStream>,
1805        message: proto::ServerProtocolMessage,
1806    ) {
1807        ws.send(WsMessage::Binary(message.encode_to_vec()))
1808            .await
1809            .unwrap();
1810    }
1811
1812    fn test_update_server_message(
1813        message_id: u64,
1814        inline_message_id: i64,
1815    ) -> proto::ServerProtocolMessage {
1816        proto::ServerProtocolMessage {
1817            id: message_id,
1818            body: Some(proto::server_protocol_message::Body::Message(
1819                proto::ServerMessage {
1820                    payload: Some(proto::server_message::Payload::Update(
1821                        proto::UpdatesPayload {
1822                            updates: vec![proto::Update {
1823                                seq: Some(1),
1824                                date: Some(1),
1825                                update: Some(proto::update::Update::NewMessage(
1826                                    proto::UpdateNewMessage {
1827                                        message: Some(proto::Message {
1828                                            id: inline_message_id,
1829                                            from_id: 42,
1830                                            peer_id: Some(proto::Peer {
1831                                                r#type: Some(proto::peer::Type::Chat(
1832                                                    proto::PeerChat { chat_id: 7 },
1833                                                )),
1834                                            }),
1835                                            chat_id: 7,
1836                                            message: Some("hello".to_owned()),
1837                                            date: 1,
1838                                            ..Default::default()
1839                                        }),
1840                                    },
1841                                )),
1842                            }],
1843                        },
1844                    )),
1845                },
1846            )),
1847        }
1848    }
1849}
inline_sdk/realtime.rs

inline_sdk/
realtime.rs
