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/// Builder for [`RealtimeClient`].
172#[must_use]
173#[derive(Clone)]
174pub struct RealtimeClientBuilder {
175    url: String,
176    token: String,
177    identity: ClientIdentity,
178    connect_timeout: Option<Duration>,
179    rpc_timeout: Option<Duration>,
180}
181
182impl fmt::Debug for RealtimeClientBuilder {
183    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
184        f.debug_struct("RealtimeClientBuilder")
185            .field("url", &realtime_url_for_debug(&self.url))
186            .field("token", &"<redacted>")
187            .field("identity", &self.identity)
188            .field("connect_timeout", &self.connect_timeout)
189            .field("rpc_timeout", &self.rpc_timeout)
190            .finish()
191    }
192}
193
194/// Typed Inline realtime RPC request.
195///
196/// This trait is implemented for generated protocol input types so callers can
197/// use [`RealtimeClient::call`] without manually pairing each input with its
198/// [`proto::Method`] and result oneof.
199pub trait RpcRequest: Sized {
200    /// Typed response returned by this request.
201    type Response;
202
203    /// RPC method associated with this input type.
204    const METHOD: proto::Method;
205
206    /// Converts this request into the generated RPC input oneof.
207    fn into_rpc_input(self) -> proto::rpc_call::Input;
208
209    /// Extracts the typed response from the generated RPC result oneof.
210    fn response_from_rpc_result(
211        result: proto::rpc_result::Result,
212    ) -> Result<Self::Response, RealtimeError>;
213}
214
215impl RealtimeClient {
216    /// Starts a realtime client builder.
217    pub fn builder(url: impl Into<String>, token: impl Into<String>) -> RealtimeClientBuilder {
218        RealtimeClientBuilder::new(url, token)
219    }
220
221    /// Connects to realtime using the default SDK identity.
222    pub async fn connect(url: &str, token: &str) -> Result<Self, RealtimeError> {
223        Self::builder(url, token).connect().await
224    }
225
226    /// Connects to realtime using a caller-provided client identity.
227    pub async fn connect_with_identity(
228        url: &str,
229        token: &str,
230        identity: ClientIdentity,
231    ) -> Result<Self, RealtimeError> {
232        Self::builder(url, token).identity(identity).connect().await
233    }
234}
235
236impl RealtimeClientBuilder {
237    /// Creates a realtime client builder with the default SDK identity.
238    pub fn new(url: impl Into<String>, token: impl Into<String>) -> Self {
239        Self {
240            url: url.into(),
241            token: token.into(),
242            identity: ClientIdentity::sdk(),
243            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
244            rpc_timeout: Some(DEFAULT_RPC_TIMEOUT),
245        }
246    }
247
248    /// Sets the client identity used in realtime headers and connection init.
249    pub fn identity(mut self, identity: ClientIdentity) -> Self {
250        self.identity = identity;
251        self
252    }
253
254    /// Sets the timeout for opening the WebSocket and receiving `ConnectionOpen`.
255    pub fn connect_timeout(mut self, timeout: Duration) -> Self {
256        self.connect_timeout = Some(timeout);
257        self
258    }
259
260    /// Disables the connect timeout.
261    pub fn without_connect_timeout(mut self) -> Self {
262        self.connect_timeout = None;
263        self
264    }
265
266    /// Sets the timeout for each RPC invocation.
267    pub fn rpc_timeout(mut self, timeout: Duration) -> Self {
268        self.rpc_timeout = Some(timeout);
269        self
270    }
271
272    /// Disables the per-RPC timeout.
273    pub fn without_rpc_timeout(mut self) -> Self {
274        self.rpc_timeout = None;
275        self
276    }
277
278    /// Opens the WebSocket connection and waits for `ConnectionOpen`.
279    pub async fn connect(self) -> Result<RealtimeClient, RealtimeError> {
280        let url = normalize_realtime_url(self.url)?;
281        log::debug!(
282            target: "inline_sdk::realtime",
283            "opening realtime websocket url={} identity_type={} connect_timeout={:?} rpc_timeout={:?}",
284            realtime_url_for_log(&url),
285            self.identity.client_type(),
286            self.connect_timeout,
287            self.rpc_timeout
288        );
289        let mut request = url.into_client_request()?;
290        request.headers_mut().insert(
291            client_info::CLIENT_TYPE_HEADER,
292            realtime_header_value("client_type", self.identity.client_type())?,
293        );
294        request.headers_mut().insert(
295            client_info::CLIENT_VERSION_HEADER,
296            realtime_header_value("client_version", self.identity.client_version())?,
297        );
298        request.headers_mut().insert(
299            "user-agent",
300            realtime_header_value("user_agent", &client_info::user_agent_for(&self.identity))?,
301        );
302
303        let (ws, _) =
304            with_optional_timeout("connect", self.connect_timeout, connect_async(request)).await?;
305        log::debug!(target: "inline_sdk::realtime", "websocket connected");
306        let mut client = RealtimeClient {
307            ws,
308            seq: 0,
309            id_gen: IdGenerator::new(),
310            rpc_timeout: self.rpc_timeout,
311        };
312
313        with_optional_timeout(
314            "connection_init",
315            self.connect_timeout,
316            client.send_connection_init(&self.token, &self.identity),
317        )
318        .await?;
319        log::trace!(target: "inline_sdk::realtime", "connection init sent");
320        with_optional_timeout(
321            "connection_open",
322            self.connect_timeout,
323            client.wait_for_connection_open(),
324        )
325        .await?;
326        log::debug!(target: "inline_sdk::realtime", "realtime protocol open");
327        Ok(client)
328    }
329}
330
331impl RealtimeClient {
332    /// Invokes a typed Inline RPC request.
333    pub async fn call<R>(&mut self, request: R) -> Result<R::Response, RealtimeError>
334    where
335        R: RpcRequest,
336    {
337        log::trace!(
338            target: "inline_sdk::realtime",
339            "calling typed rpc method={}",
340            R::METHOD.as_str_name()
341        );
342        let result = self.invoke(R::METHOD, request.into_rpc_input()).await?;
343        R::response_from_rpc_result(result)
344    }
345
346    /// Invokes an Inline RPC method and waits for the matching result.
347    pub async fn invoke(
348        &mut self,
349        method: proto::Method,
350        input: proto::rpc_call::Input,
351    ) -> Result<proto::rpc_result::Result, RealtimeError> {
352        let rpc_call = proto::RpcCall {
353            method: method as i32,
354            input: Some(input),
355        };
356        let message_id = self.next_id();
357        log::trace!(
358            target: "inline_sdk::realtime",
359            "sending rpc method={} msg_id={message_id}",
360            method.as_str_name()
361        );
362        let message = proto::ClientMessage {
363            id: message_id,
364            seq: self.next_seq(),
365            body: Some(proto::client_message::Body::RpcCall(rpc_call)),
366        };
367
368        self.send_client_message(message).await?;
369
370        with_optional_timeout(
371            "rpc",
372            self.rpc_timeout,
373            self.wait_for_rpc_result(message_id),
374        )
375        .await
376    }
377
378    /// Returns the configured per-RPC timeout.
379    pub fn rpc_timeout(&self) -> Option<Duration> {
380        self.rpc_timeout
381    }
382
383    async fn send_connection_init(
384        &mut self,
385        token: &str,
386        identity: &ClientIdentity,
387    ) -> Result<(), RealtimeError> {
388        let init = connection_init_for_token(token, identity);
389        let message = proto::ClientMessage {
390            id: self.next_id(),
391            seq: self.next_seq(),
392            body: Some(proto::client_message::Body::ConnectionInit(init)),
393        };
394
395        self.send_client_message(message).await
396    }
397
398    async fn wait_for_connection_open(&mut self) -> Result<(), RealtimeError> {
399        loop {
400            let message = self.read_server_message().await?;
401            match message.body {
402                Some(proto::server_protocol_message::Body::ConnectionOpen(_)) => return Ok(()),
403                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
404                    log::warn!(
405                        target: "inline_sdk::realtime",
406                        "connection open rejected reason={}",
407                        proto::connection_error::Reason::try_from(error.reason)
408                            .map(|reason| reason.as_str_name())
409                            .unwrap_or("UNKNOWN")
410                    );
411                    return Err(connection_error_from_proto(error));
412                }
413                _ => {}
414            }
415        }
416    }
417
418    async fn wait_for_rpc_result(
419        &mut self,
420        message_id: u64,
421    ) -> Result<proto::rpc_result::Result, RealtimeError> {
422        loop {
423            let message = self.read_server_message().await?;
424            match message.body {
425                Some(proto::server_protocol_message::Body::RpcResult(result))
426                    if result.req_msg_id == message_id =>
427                {
428                    log::trace!(
429                        target: "inline_sdk::realtime",
430                        "received rpc result msg_id={message_id}"
431                    );
432                    return result.result.ok_or(RealtimeError::MissingResult);
433                }
434                Some(proto::server_protocol_message::Body::RpcError(error))
435                    if error.req_msg_id == message_id =>
436                {
437                    let message = error.message;
438                    let error_name = rpc_error_code_name(error.error_code);
439                    let friendly =
440                        format_rpc_error(error.error_code, &error_name, &message, error.code);
441                    log::warn!(
442                        target: "inline_sdk::realtime",
443                        "received rpc error msg_id={message_id} error={error_name} status={}",
444                        error.code
445                    );
446                    return Err(RealtimeError::RpcError {
447                        code: error.code,
448                        error_code: error.error_code,
449                        error_name,
450                        message,
451                        friendly,
452                    });
453                }
454                Some(proto::server_protocol_message::Body::ConnectionError(error)) => {
455                    log::warn!(
456                        target: "inline_sdk::realtime",
457                        "connection error while waiting for rpc msg_id={message_id} reason={}",
458                        proto::connection_error::Reason::try_from(error.reason)
459                            .map(|reason| reason.as_str_name())
460                            .unwrap_or("UNKNOWN")
461                    );
462                    return Err(connection_error_from_proto(error));
463                }
464                _ => {}
465            }
466        }
467    }
468
469    async fn send_client_message(
470        &mut self,
471        message: proto::ClientMessage,
472    ) -> Result<(), RealtimeError> {
473        let bytes = message.encode_to_vec();
474        self.ws.send(WsMessage::Binary(bytes)).await?;
475        Ok(())
476    }
477
478    async fn read_server_message(&mut self) -> Result<proto::ServerProtocolMessage, RealtimeError> {
479        loop {
480            let message = self
481                .ws
482                .next()
483                .await
484                .ok_or(RealtimeError::ConnectionClosed)??;
485            match message {
486                WsMessage::Binary(data) => {
487                    return Ok(proto::ServerProtocolMessage::decode(&*data)?);
488                }
489                WsMessage::Text(_) => continue,
490                WsMessage::Close(_) => return Err(RealtimeError::ConnectionClosed),
491                WsMessage::Ping(_) | WsMessage::Pong(_) => continue,
492                _ => continue,
493            }
494        }
495    }
496
497    fn next_seq(&mut self) -> u32 {
498        self.seq = self.seq.wrapping_add(1);
499        self.seq
500    }
501
502    fn next_id(&mut self) -> u64 {
503        self.id_gen.next_id()
504    }
505}
506
507macro_rules! rpc_requests {
508    ($(($input_ty:ident, $method:ident, $input_variant:ident, $result_ty:ident, $result_variant:ident)),+ $(,)?) => {
509        $(
510            impl RpcRequest for proto::$input_ty {
511                type Response = proto::$result_ty;
512
513                const METHOD: proto::Method = proto::Method::$method;
514
515                fn into_rpc_input(self) -> proto::rpc_call::Input {
516                    proto::rpc_call::Input::$input_variant(self)
517                }
518
519                fn response_from_rpc_result(
520                    result: proto::rpc_result::Result,
521                ) -> Result<Self::Response, RealtimeError> {
522                    match result {
523                        proto::rpc_result::Result::$result_variant(response) => Ok(response),
524                        other => {
525                            let actual = rpc_result_variant_name(&other);
526                            log::warn!(
527                                target: "inline_sdk::realtime",
528                                "unexpected rpc result method={} expected={} actual={actual}",
529                                Self::METHOD.as_str_name(),
530                                stringify!($result_variant)
531                            );
532                            Err(RealtimeError::UnexpectedResult {
533                                method: Self::METHOD.as_str_name(),
534                                expected: stringify!($result_variant),
535                                actual,
536                            })
537                        }
538                    }
539                }
540            }
541        )+
542
543        fn rpc_result_variant_name(result: &proto::rpc_result::Result) -> &'static str {
544            match result {
545                $(proto::rpc_result::Result::$result_variant(_) => stringify!($result_variant),)+
546            }
547        }
548    };
549}
550
551rpc_requests!(
552    (GetMeInput, GetMe, GetMe, GetMeResult, GetMe),
553    (
554        GetPeerPhotoInput,
555        GetPeerPhoto,
556        GetPeerPhoto,
557        GetPeerPhotoResult,
558        GetPeerPhoto
559    ),
560    (
561        DeleteMessagesInput,
562        DeleteMessages,
563        DeleteMessages,
564        DeleteMessagesResult,
565        DeleteMessages
566    ),
567    (
568        SendMessageInput,
569        SendMessage,
570        SendMessage,
571        SendMessageResult,
572        SendMessage
573    ),
574    (
575        GetChatHistoryInput,
576        GetChatHistory,
577        GetChatHistory,
578        GetChatHistoryResult,
579        GetChatHistory
580    ),
581    (
582        AddReactionInput,
583        AddReaction,
584        AddReaction,
585        AddReactionResult,
586        AddReaction
587    ),
588    (
589        DeleteReactionInput,
590        DeleteReaction,
591        DeleteReaction,
592        DeleteReactionResult,
593        DeleteReaction
594    ),
595    (
596        EditMessageInput,
597        EditMessage,
598        EditMessage,
599        EditMessageResult,
600        EditMessage
601    ),
602    (
603        CreateChatInput,
604        CreateChat,
605        CreateChat,
606        CreateChatResult,
607        CreateChat
608    ),
609    (
610        GetSpaceMembersInput,
611        GetSpaceMembers,
612        GetSpaceMembers,
613        GetSpaceMembersResult,
614        GetSpaceMembers
615    ),
616    (
617        DeleteChatInput,
618        DeleteChat,
619        DeleteChat,
620        DeleteChatResult,
621        DeleteChat
622    ),
623    (
624        InviteToSpaceInput,
625        InviteToSpace,
626        InviteToSpace,
627        InviteToSpaceResult,
628        InviteToSpace
629    ),
630    (
631        GetChatParticipantsInput,
632        GetChatParticipants,
633        GetChatParticipants,
634        GetChatParticipantsResult,
635        GetChatParticipants
636    ),
637    (
638        AddChatParticipantInput,
639        AddChatParticipant,
640        AddChatParticipant,
641        AddChatParticipantResult,
642        AddChatParticipant
643    ),
644    (
645        RemoveChatParticipantInput,
646        RemoveChatParticipant,
647        RemoveChatParticipant,
648        RemoveChatParticipantResult,
649        RemoveChatParticipant
650    ),
651    (
652        TranslateMessagesInput,
653        TranslateMessages,
654        TranslateMessages,
655        TranslateMessagesResult,
656        TranslateMessages
657    ),
658    (GetChatsInput, GetChats, GetChats, GetChatsResult, GetChats),
659    (
660        UpdateUserSettingsInput,
661        UpdateUserSettings,
662        UpdateUserSettings,
663        UpdateUserSettingsResult,
664        UpdateUserSettings
665    ),
666    (
667        GetUserSettingsInput,
668        GetUserSettings,
669        GetUserSettings,
670        GetUserSettingsResult,
671        GetUserSettings
672    ),
673    (
674        SendComposeActionInput,
675        SendComposeAction,
676        SendComposeAction,
677        SendComposeActionResult,
678        SendComposeAction
679    ),
680    (
681        CreateBotInput,
682        CreateBot,
683        CreateBot,
684        CreateBotResult,
685        CreateBot
686    ),
687    (
688        DeleteMemberInput,
689        DeleteMember,
690        DeleteMember,
691        DeleteMemberResult,
692        DeleteMember
693    ),
694    (
695        MarkAsUnreadInput,
696        MarkAsUnread,
697        MarkAsUnread,
698        MarkAsUnreadResult,
699        MarkAsUnread
700    ),
701    (
702        GetUpdatesStateInput,
703        GetUpdatesState,
704        GetUpdatesState,
705        GetUpdatesStateResult,
706        GetUpdatesState
707    ),
708    (GetChatInput, GetChat, GetChat, GetChatResult, GetChat),
709    (
710        GetUpdatesInput,
711        GetUpdates,
712        GetUpdates,
713        GetUpdatesResult,
714        GetUpdates
715    ),
716    (
717        UpdateMemberAccessInput,
718        UpdateMemberAccess,
719        UpdateMemberAccess,
720        UpdateMemberAccessResult,
721        UpdateMemberAccess
722    ),
723    (
724        SearchMessagesInput,
725        SearchMessages,
726        SearchMessages,
727        SearchMessagesResult,
728        SearchMessages
729    ),
730    (
731        ForwardMessagesInput,
732        ForwardMessages,
733        ForwardMessages,
734        ForwardMessagesResult,
735        ForwardMessages
736    ),
737    (
738        UpdateChatVisibilityInput,
739        UpdateChatVisibility,
740        UpdateChatVisibility,
741        UpdateChatVisibilityResult,
742        UpdateChatVisibility
743    ),
744    (
745        PinMessageInput,
746        PinMessage,
747        PinMessage,
748        PinMessageResult,
749        PinMessage
750    ),
751    (
752        UpdateChatInfoInput,
753        UpdateChatInfo,
754        UpdateChatInfo,
755        UpdateChatInfoResult,
756        UpdateChatInfo
757    ),
758    (ListBotsInput, ListBots, ListBots, ListBotsResult, ListBots),
759    (
760        RevealBotTokenInput,
761        RevealBotToken,
762        RevealBotToken,
763        RevealBotTokenResult,
764        RevealBotToken
765    ),
766    (
767        MoveThreadInput,
768        MoveThread,
769        MoveThread,
770        MoveThreadResult,
771        MoveThread
772    ),
773    (
774        RotateBotTokenInput,
775        RotateBotToken,
776        RotateBotToken,
777        RotateBotTokenResult,
778        RotateBotToken
779    ),
780    (
781        UpdateBotProfileInput,
782        UpdateBotProfile,
783        UpdateBotProfile,
784        UpdateBotProfileResult,
785        UpdateBotProfile
786    ),
787    (
788        GetMessagesInput,
789        GetMessages,
790        GetMessages,
791        GetMessagesResult,
792        GetMessages
793    ),
794    (
795        UpdateDialogNotificationSettingsInput,
796        UpdateDialogNotificationSettings,
797        UpdateDialogNotificationSettings,
798        UpdateDialogNotificationSettingsResult,
799        UpdateDialogNotificationSettings
800    ),
801    (
802        ReadMessagesInput,
803        ReadMessages,
804        ReadMessages,
805        ReadMessagesResult,
806        ReadMessages
807    ),
808    (
809        UpdatePushNotificationDetailsInput,
810        UpdatePushNotificationDetails,
811        UpdatePushNotificationDetails,
812        UpdatePushNotificationDetailsResult,
813        UpdatePushNotificationDetails
814    ),
815    (
816        CreateSubthreadInput,
817        CreateSubthread,
818        CreateSubthread,
819        CreateSubthreadResult,
820        CreateSubthread
821    ),
822    (
823        GetBotCommandsInput,
824        GetBotCommands,
825        GetBotCommands,
826        GetBotCommandsResult,
827        GetBotCommands
828    ),
829    (
830        SetBotCommandsInput,
831        SetBotCommands,
832        SetBotCommands,
833        SetBotCommandsResult,
834        SetBotCommands
835    ),
836    (
837        GetPeerBotCommandsInput,
838        GetPeerBotCommands,
839        GetPeerBotCommands,
840        GetPeerBotCommandsResult,
841        GetPeerBotCommands
842    ),
843    (
844        ShowInChatListInput,
845        ShowInChatList,
846        ShowInChatList,
847        ShowInChatListResult,
848        ShowInChatList
849    ),
850    (
851        ReserveChatIdsInput,
852        ReserveChatIds,
853        ReserveChatIds,
854        ReserveChatIdsResult,
855        ReserveChatIds
856    ),
857    (
858        InvokeMessageActionInput,
859        InvokeMessageAction,
860        InvokeMessageAction,
861        InvokeMessageActionResult,
862        InvokeMessageAction
863    ),
864    (
865        AnswerMessageActionInput,
866        AnswerMessageAction,
867        AnswerMessageAction,
868        AnswerMessageActionResult,
869        AnswerMessageAction
870    ),
871    (
872        RevokeSessionInput,
873        RevokeSession,
874        RevokeSession,
875        RevokeSessionResult,
876        RevokeSession
877    ),
878    (
879        UpdateDialogOpenInput,
880        UpdateDialogOpen,
881        UpdateDialogOpen,
882        UpdateDialogOpenResult,
883        UpdateDialogOpen
884    ),
885    (
886        UpdateDialogOrderInput,
887        UpdateDialogOrder,
888        UpdateDialogOrder,
889        UpdateDialogOrderResult,
890        UpdateDialogOrder
891    ),
892    (
893        ClearChatHistoryInput,
894        ClearChatHistory,
895        ClearChatHistory,
896        ClearChatHistoryResult,
897        ClearChatHistory
898    ),
899    (
900        DeleteBotInput,
901        DeleteBot,
902        DeleteBot,
903        DeleteBotResult,
904        DeleteBot
905    ),
906    (
907        DeleteMessageAttachmentInput,
908        DeleteMessageAttachment,
909        DeleteMessageAttachment,
910        DeleteMessageAttachmentResult,
911        DeleteMessageAttachment
912    ),
913    (
914        SetBotAvatarInput,
915        SetBotAvatar,
916        SetBotAvatar,
917        SetBotAvatarResult,
918        SetBotAvatar
919    ),
920    (
921        ClearBotAvatarInput,
922        ClearBotAvatar,
923        ClearBotAvatar,
924        ClearBotAvatarResult,
925        ClearBotAvatar
926    ),
927    (
928        GetBotPresenceInput,
929        GetBotPresence,
930        GetBotPresence,
931        GetBotPresenceResult,
932        GetBotPresence
933    ),
934    (
935        SetBotPresenceStateInput,
936        SetBotPresenceState,
937        SetBotPresenceState,
938        SetBotPresenceStateResult,
939        SetBotPresenceState
940    ),
941    (
942        UpdateDialogFollowModeInput,
943        UpdateDialogFollowMode,
944        UpdateDialogFollowMode,
945        UpdateDialogFollowModeResult,
946        UpdateDialogFollowMode
947    ),
948    (
949        GetSessionsInput,
950        GetSessions,
951        GetSessions,
952        GetSessionsResult,
953        GetSessions
954    ),
955    (
956        CheckUsernameInput,
957        CheckUsername,
958        CheckUsername,
959        CheckUsernameResult,
960        CheckUsername
961    ),
962    (
963        ChangeUsernameInput,
964        ChangeUsername,
965        ChangeUsername,
966        ChangeUsernameResult,
967        ChangeUsername
968    ),
969    (
970        UpdateProfileInput,
971        UpdateProfile,
972        UpdateProfile,
973        UpdateProfileResult,
974        UpdateProfile
975    ),
976    (
977        GetSpaceUrlPreviewExclusionsInput,
978        GetSpaceUrlPreviewExclusions,
979        GetSpaceUrlPreviewExclusions,
980        GetSpaceUrlPreviewExclusionsResult,
981        GetSpaceUrlPreviewExclusions
982    ),
983    (
984        AddSpaceUrlPreviewExclusionInput,
985        AddSpaceUrlPreviewExclusion,
986        AddSpaceUrlPreviewExclusion,
987        AddSpaceUrlPreviewExclusionResult,
988        AddSpaceUrlPreviewExclusion
989    ),
990    (
991        RemoveSpaceUrlPreviewExclusionInput,
992        RemoveSpaceUrlPreviewExclusion,
993        RemoveSpaceUrlPreviewExclusion,
994        RemoveSpaceUrlPreviewExclusionResult,
995        RemoveSpaceUrlPreviewExclusion
996    ),
997);
998
999fn connection_init_for_token(token: &str, identity: &ClientIdentity) -> proto::ConnectionInit {
1000    proto::ConnectionInit {
1001        token: token.to_string(),
1002        build_number: None,
1003        layer: None,
1004        client_version: Some(identity.client_version().to_string()),
1005        os_version: client_info::current_os_version(),
1006    }
1007}
1008
1009fn rpc_error_code_name(error_code: i32) -> String {
1010    proto::rpc_error::Code::try_from(error_code)
1011        .map(|code| code.as_str_name())
1012        .unwrap_or("UNKNOWN")
1013        .to_string()
1014}
1015
1016fn format_rpc_error(error_code: i32, error_name: &str, message: &str, status_code: i32) -> String {
1017    let label = match error_name {
1018        "UNKNOWN" => "Unknown RPC error",
1019        "BAD_REQUEST" => "Bad request",
1020        "UNAUTHENTICATED" => "Not authenticated",
1021        "RATE_LIMIT" => "Rate limited",
1022        "INTERNAL_ERROR" => "Internal server error",
1023        "PEER_ID_INVALID" => "Invalid peer (chat/user id)",
1024        "MESSAGE_ID_INVALID" => "Invalid message id",
1025        "USER_ID_INVALID" => "Invalid user id",
1026        "USER_ALREADY_MEMBER" => "User already in chat/space",
1027        "SPACE_ID_INVALID" => "Invalid space id",
1028        "CHAT_ID_INVALID" => "Invalid chat id",
1029        "EMAIL_INVALID" => "Invalid email address",
1030        "PHONE_NUMBER_INVALID" => "Invalid phone number",
1031        "SPACE_ADMIN_REQUIRED" => "Space admin required",
1032        "SPACE_OWNER_REQUIRED" => "Space owner required",
1033        "USERNAME_INVALID" => "Invalid username",
1034        "USERNAME_TAKEN" => "Username already taken",
1035        "FIRST_NAME_INVALID" => "Invalid first name",
1036        _ => "Unknown RPC error",
1037    };
1038
1039    let mut formatted = String::from(label);
1040    if error_name == "UNKNOWN" && error_code != 0 {
1041        formatted.push_str(&format!(" {error_code}"));
1042    }
1043    if !message.is_empty() && !message.eq_ignore_ascii_case(label) {
1044        formatted.push_str(": ");
1045        formatted.push_str(message);
1046    }
1047    if status_code != 0 {
1048        formatted.push_str(&format!(" (HTTP {status_code})"));
1049    }
1050    formatted
1051}
1052
1053fn connection_error_from_proto(error: proto::ConnectionError) -> RealtimeError {
1054    let reason = error.reason;
1055    let reason_name = proto::connection_error::Reason::try_from(reason)
1056        .map(|reason| reason.as_str_name())
1057        .unwrap_or("UNKNOWN")
1058        .to_string();
1059    let friendly = format_connection_error(reason, &reason_name);
1060    RealtimeError::ConnectionError {
1061        reason,
1062        reason_name,
1063        friendly,
1064    }
1065}
1066
1067fn format_connection_error(reason: i32, reason_name: &str) -> String {
1068    match reason_name {
1069        "UNAUTHORIZED" => "Realtime connection unauthorized".to_string(),
1070        "INVALID_AUTH" => "Realtime auth token is invalid".to_string(),
1071        "SESSION_REVOKED" => "Realtime session was revoked".to_string(),
1072        "REASON_UNSPECIFIED" => "Realtime connection rejected".to_string(),
1073        _ => format!("Realtime connection rejected: unknown reason {reason}"),
1074    }
1075}
1076
1077fn normalize_realtime_url(url: impl Into<String>) -> Result<Url, RealtimeError> {
1078    let original = url.into();
1079    let normalized = original.trim().to_string();
1080    if normalized.is_empty() {
1081        return Err(RealtimeError::InvalidUrl {
1082            url: original,
1083            message: "realtime URL cannot be empty".to_string(),
1084        });
1085    }
1086
1087    let parsed = Url::parse(&normalized).map_err(|err| RealtimeError::InvalidUrl {
1088        url: normalized.clone(),
1089        message: err.to_string(),
1090    })?;
1091
1092    if !matches!(parsed.scheme(), "ws" | "wss") {
1093        return Err(RealtimeError::InvalidUrl {
1094            url: normalized,
1095            message: "scheme must be ws or wss".to_string(),
1096        });
1097    }
1098
1099    if parsed.host_str().is_none() {
1100        return Err(RealtimeError::InvalidUrl {
1101            url: normalized,
1102            message: "host is required".to_string(),
1103        });
1104    }
1105
1106    if !parsed.username().is_empty() || parsed.password().is_some() {
1107        return Err(RealtimeError::InvalidUrl {
1108            url: normalized,
1109            message: "credentials are not valid in the realtime URL".to_string(),
1110        });
1111    }
1112
1113    if parsed.fragment().is_some() {
1114        return Err(RealtimeError::InvalidUrl {
1115            url: normalized,
1116            message: "fragments are not valid in the realtime URL".to_string(),
1117        });
1118    }
1119
1120    Ok(parsed)
1121}
1122
1123fn realtime_url_for_log(url: &Url) -> String {
1124    let host = url.host_str().unwrap_or("<missing-host>");
1125    let port = url
1126        .port()
1127        .map(|port| format!(":{port}"))
1128        .unwrap_or_default();
1129    format!("{}://{}{}{}", url.scheme(), host, port, url.path())
1130}
1131
1132fn realtime_url_for_debug(raw_url: &str) -> String {
1133    Url::parse(raw_url.trim())
1134        .map(|url| realtime_url_for_log(&url))
1135        .unwrap_or_else(|_| "<invalid>".to_string())
1136}
1137
1138fn realtime_header_value(field: &'static str, value: &str) -> Result<HeaderValue, RealtimeError> {
1139    HeaderValue::from_str(value).map_err(|_| RealtimeError::InvalidHeaderValue { field })
1140}
1141
1142async fn with_optional_timeout<T, E, Fut>(
1143    operation: &'static str,
1144    timeout: Option<Duration>,
1145    future: Fut,
1146) -> Result<T, RealtimeError>
1147where
1148    Fut: Future<Output = Result<T, E>>,
1149    RealtimeError: From<E>,
1150{
1151    match timeout {
1152        Some(timeout) => tokio::time::timeout(timeout, future)
1153            .await
1154            .map_err(|_| RealtimeError::Timeout { operation, timeout })?
1155            .map_err(RealtimeError::from),
1156        None => future.await.map_err(RealtimeError::from),
1157    }
1158}
1159
1160struct IdGenerator {
1161    last_timestamp: u64,
1162    sequence: u32,
1163}
1164
1165impl IdGenerator {
1166    fn new() -> Self {
1167        Self {
1168            last_timestamp: 0,
1169            sequence: 0,
1170        }
1171    }
1172
1173    fn next_id(&mut self) -> u64 {
1174        let timestamp = current_epoch_seconds().saturating_sub(EPOCH_SECONDS);
1175        if timestamp == self.last_timestamp {
1176            self.sequence = self.sequence.wrapping_add(1);
1177        } else {
1178            self.sequence = 0;
1179            self.last_timestamp = timestamp;
1180        }
1181
1182        (timestamp << 32) | self.sequence as u64
1183    }
1184}
1185
1186const EPOCH_SECONDS: u64 = 1_735_689_600; // 2025-01-01T00:00:00Z
1187
1188fn current_epoch_seconds() -> u64 {
1189    use std::time::{SystemTime, UNIX_EPOCH};
1190    SystemTime::now()
1191        .duration_since(UNIX_EPOCH)
1192        .unwrap_or_default()
1193        .as_secs()
1194}
1195
1196#[cfg(test)]
1197mod tests {
1198    use super::*;
1199
1200    #[test]
1201    fn rpc_error_code_name_uses_stable_proto_name() {
1202        assert_eq!(
1203            rpc_error_code_name(proto::rpc_error::Code::PeerIdInvalid as i32),
1204            "PEER_ID_INVALID"
1205        );
1206        assert_eq!(rpc_error_code_name(999), "UNKNOWN");
1207    }
1208
1209    #[test]
1210    fn rpc_error_formatter_covers_new_profile_codes() {
1211        assert_eq!(
1212            format_rpc_error(
1213                proto::rpc_error::Code::UsernameInvalid as i32,
1214                "USERNAME_INVALID",
1215                "",
1216                400
1217            ),
1218            "Invalid username (HTTP 400)"
1219        );
1220        assert_eq!(
1221            format_rpc_error(
1222                proto::rpc_error::Code::UsernameTaken as i32,
1223                "USERNAME_TAKEN",
1224                "handle exists",
1225                409
1226            ),
1227            "Username already taken: handle exists (HTTP 409)"
1228        );
1229        assert_eq!(
1230            format_rpc_error(
1231                proto::rpc_error::Code::FirstNameInvalid as i32,
1232                "FIRST_NAME_INVALID",
1233                "",
1234                400
1235            ),
1236            "Invalid first name (HTTP 400)"
1237        );
1238    }
1239
1240    #[test]
1241    fn connection_error_preserves_proto_reason() {
1242        let err = connection_error_from_proto(proto::ConnectionError {
1243            reason: proto::connection_error::Reason::InvalidAuth as i32,
1244        });
1245
1246        match err {
1247            RealtimeError::ConnectionError {
1248                reason,
1249                reason_name,
1250                friendly,
1251            } => {
1252                assert_eq!(reason, 2);
1253                assert_eq!(reason_name, "INVALID_AUTH");
1254                assert_eq!(friendly, "Realtime auth token is invalid");
1255            }
1256            other => panic!("expected connection error, got {other:?}"),
1257        }
1258    }
1259
1260    #[test]
1261    fn unknown_connection_error_reason_is_preserved() {
1262        let err = connection_error_from_proto(proto::ConnectionError { reason: 999 });
1263
1264        match err {
1265            RealtimeError::ConnectionError {
1266                reason,
1267                reason_name,
1268                friendly,
1269            } => {
1270                assert_eq!(reason, 999);
1271                assert_eq!(reason_name, "UNKNOWN");
1272                assert_eq!(friendly, "Realtime connection rejected: unknown reason 999");
1273            }
1274            other => panic!("expected connection error, got {other:?}"),
1275        }
1276    }
1277
1278    #[test]
1279    fn connection_init_uses_custom_client_identity() {
1280        let init =
1281            connection_init_for_token("token-1", &ClientIdentity::new("integration-test", "9.9.9"));
1282
1283        assert_eq!(init.token, "token-1");
1284        assert_eq!(init.client_version.as_deref(), Some("9.9.9"));
1285        assert!(init.build_number.is_none());
1286        assert!(init.layer.is_none());
1287    }
1288
1289    #[test]
1290    fn realtime_url_validation_accepts_ws_urls() {
1291        let url = normalize_realtime_url(" wss://api.inline.chat/realtime?edge=iad ").unwrap();
1292        assert_eq!(url.as_str(), "wss://api.inline.chat/realtime?edge=iad");
1293    }
1294
1295    #[test]
1296    fn realtime_url_validation_rejects_invalid_urls() {
1297        let err = normalize_realtime_url("inline.test").unwrap_err();
1298        match err {
1299            RealtimeError::InvalidUrl { url, message } => {
1300                assert_eq!(url, "inline.test");
1301                assert!(message.contains("relative URL without a base"));
1302            }
1303            other => panic!("expected invalid URL, got {other:?}"),
1304        }
1305
1306        let err = normalize_realtime_url("https://api.inline.chat/realtime").unwrap_err();
1307        match err {
1308            RealtimeError::InvalidUrl { message, .. } => {
1309                assert_eq!(message, "scheme must be ws or wss");
1310            }
1311            other => panic!("expected invalid URL, got {other:?}"),
1312        }
1313
1314        let err = normalize_realtime_url("wss://user:secret@api.inline.chat/realtime").unwrap_err();
1315        match &err {
1316            RealtimeError::InvalidUrl { message, .. } => {
1317                assert_eq!(message, "credentials are not valid in the realtime URL");
1318                assert!(!err.to_string().contains("secret"));
1319            }
1320            other => panic!("expected invalid URL, got {other:?}"),
1321        }
1322
1323        let err = normalize_realtime_url("wss://api.inline.chat/realtime#token").unwrap_err();
1324        match err {
1325            RealtimeError::InvalidUrl { message, .. } => {
1326                assert_eq!(message, "fragments are not valid in the realtime URL");
1327            }
1328            other => panic!("expected invalid URL, got {other:?}"),
1329        }
1330    }
1331
1332    #[test]
1333    fn realtime_invalid_url_debug_redacts_unsafe_url_parts() {
1334        let err = normalize_realtime_url(
1335            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret#frag",
1336        )
1337        .unwrap_err();
1338        let debug = format!("{err:?}");
1339
1340        assert!(debug.contains("wss://api.inline.chat/realtime"));
1341        assert!(!debug.contains("url-secret"));
1342        assert!(!debug.contains("query-secret"));
1343    }
1344
1345    #[test]
1346    fn realtime_url_for_log_omits_query_and_fragment() {
1347        let url = Url::parse("wss://api.inline.chat/realtime?token=secret#frag").unwrap();
1348        assert_eq!(realtime_url_for_log(&url), "wss://api.inline.chat/realtime");
1349    }
1350
1351    #[test]
1352    fn realtime_header_value_rejects_invalid_values() {
1353        let err = realtime_header_value("user_agent", "bad\nvalue").unwrap_err();
1354        match err {
1355            RealtimeError::InvalidHeaderValue { field } => {
1356                assert_eq!(field, "user_agent");
1357            }
1358            other => panic!("expected invalid header value, got {other:?}"),
1359        }
1360    }
1361
1362    #[test]
1363    fn realtime_builder_keeps_custom_identity_and_default_timeouts() {
1364        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
1365            .identity(ClientIdentity::new("agent", "0.2.0"));
1366
1367        assert_eq!(builder.url, "wss://api.inline.chat/realtime");
1368        assert_eq!(builder.token, "token-1");
1369        assert_eq!(builder.identity.client_type(), "agent");
1370        assert_eq!(builder.identity.client_version(), "0.2.0");
1371        assert_eq!(builder.connect_timeout, Some(DEFAULT_CONNECT_TIMEOUT));
1372        assert_eq!(builder.rpc_timeout, Some(DEFAULT_RPC_TIMEOUT));
1373    }
1374
1375    #[test]
1376    fn realtime_builder_debug_redacts_token() {
1377        let builder = RealtimeClient::builder(
1378            "wss://user:url-secret@api.inline.chat/realtime?token=query-secret",
1379            "secret-token-1",
1380        );
1381        let debug = format!("{builder:?}");
1382
1383        assert!(debug.contains("RealtimeClientBuilder"));
1384        assert!(debug.contains("wss://api.inline.chat/realtime"));
1385        assert!(debug.contains("<redacted>"));
1386        assert!(!debug.contains("secret-token-1"));
1387        assert!(!debug.contains("url-secret"));
1388        assert!(!debug.contains("query-secret"));
1389    }
1390
1391    #[test]
1392    fn realtime_builder_can_override_or_disable_timeouts() {
1393        let builder = RealtimeClient::builder("wss://api.inline.chat/realtime", "token-1")
1394            .connect_timeout(Duration::from_secs(5))
1395            .rpc_timeout(Duration::from_secs(10));
1396
1397        assert_eq!(builder.connect_timeout, Some(Duration::from_secs(5)));
1398        assert_eq!(builder.rpc_timeout, Some(Duration::from_secs(10)));
1399
1400        let builder = builder.without_connect_timeout().without_rpc_timeout();
1401        assert_eq!(builder.connect_timeout, None);
1402        assert_eq!(builder.rpc_timeout, None);
1403    }
1404
1405    #[test]
1406    fn typed_rpc_request_maps_method_input_and_result() {
1407        assert_eq!(
1408            <proto::GetChatsInput as RpcRequest>::METHOD,
1409            proto::Method::GetChats
1410        );
1411
1412        let input = proto::GetChatsInput {};
1413        match input.into_rpc_input() {
1414            proto::rpc_call::Input::GetChats(_) => {}
1415            other => panic!("expected GetChats input, got {other:?}"),
1416        }
1417
1418        let response = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
1419            proto::rpc_result::Result::GetChats(proto::GetChatsResult::default()),
1420        )
1421        .unwrap();
1422        assert!(response.dialogs.is_empty());
1423    }
1424
1425    #[test]
1426    fn typed_rpc_request_rejects_unexpected_result_variant() {
1427        let err = <proto::GetChatsInput as RpcRequest>::response_from_rpc_result(
1428            proto::rpc_result::Result::GetMe(proto::GetMeResult::default()),
1429        )
1430        .unwrap_err();
1431
1432        match err {
1433            RealtimeError::UnexpectedResult {
1434                method,
1435                expected,
1436                actual,
1437            } => {
1438                assert_eq!(method, "GET_CHATS");
1439                assert_eq!(expected, "GetChats");
1440                assert_eq!(actual, "GetMe");
1441            }
1442            other => panic!("expected unexpected result error, got {other:?}"),
1443        }
1444    }
1445
1446    #[tokio::test]
1447    async fn optional_timeout_reports_elapsed_operation() {
1448        let err = with_optional_timeout("test", Some(Duration::from_millis(1)), async {
1449            tokio::time::sleep(Duration::from_millis(20)).await;
1450            Ok::<(), RealtimeError>(())
1451        })
1452        .await
1453        .unwrap_err();
1454
1455        match err {
1456            RealtimeError::Timeout { operation, timeout } => {
1457                assert_eq!(operation, "test");
1458                assert_eq!(timeout, Duration::from_millis(1));
1459            }
1460            other => panic!("expected timeout, got {other:?}"),
1461        }
1462    }
1463
1464    #[tokio::test]
1465    async fn optional_timeout_can_be_disabled() {
1466        with_optional_timeout("test", None, async { Ok::<_, RealtimeError>("done") })
1467            .await
1468            .unwrap();
1469    }
1470}
inline_sdk/realtime.rs

inline_sdk/
realtime.rs
