1use std::{
9 collections::HashMap,
10 fmt,
11 sync::Arc,
12 time::{SystemTime, UNIX_EPOCH},
13};
14
15use futures_util::future::BoxFuture;
16use inline_sdk::{
17 ApiClient, ApiError, AuthMetadata, ClientIdentity, RealtimeClient, RealtimeError,
18 UploadFileBytesInput, UploadFileResult, UploadFileType, UploadVideoMetadata, proto,
19};
20use serde_json::Value;
21
22use crate::{
23 AuthContactKind, AuthCredential, AuthStartRequest, AuthStartResult, AuthToken,
24 AuthVerifyRequest, AuthVerifyResult, BackendError, BackendResult, ChatCreateParticipant,
25 ChatParticipantRecord, ChatParticipantsPage, ChatParticipantsRequest, ClientBackend,
26 ClientErrorCategory, ClientEvent, ClientStatusSnapshot, ClientStore, ConnectRequest,
27 CreateDmRequest, CreateReplyThreadRequest, CreateThreadRequest, CreatedChat,
28 DeleteMessageRequest, DialogRecord, DialogsPage, DialogsRequest, EditMessageRequest,
29 HistoryPage, HistoryRequest, InMemoryStore, InlineId, MediaKind, MessageContent,
30 MessageMutation, MessageRecord, OperationOutcome, PeerRef, RandomId, ReactRequest, ReadRequest,
31 RealtimeConnectRequest, RealtimeConnector, SdkRealtimeConnector, SendTextOutcome,
32 SendTextRequest, StoreError, StoredSession, StoredTransaction, TransactionId,
33 TransactionIdentity, TransactionState, TypingRequest, UploadRequest, UserRecord, VERSION,
34};
35
36const DEFAULT_API_BASE_URL: &str = "https://api.inline.chat/v1";
37const DEFAULT_REALTIME_URL: &str = "wss://api.inline.chat/realtime";
38
39#[derive(Debug, thiserror::Error)]
41#[non_exhaustive]
42pub enum SdkBackendBuildError {
43 #[error("failed to build Inline API client: {0}")]
45 Api(#[from] ApiError),
46}
47
48#[derive(Clone)]
50pub struct SdkBackendBuilder {
51 api_base_url: String,
52 realtime_url: String,
53 identity: ClientIdentity,
54 store: Arc<dyn ClientStore>,
55 realtime_connector: Option<Arc<dyn RealtimeConnector>>,
56}
57
58impl fmt::Debug for SdkBackendBuilder {
59 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60 f.debug_struct("SdkBackendBuilder")
61 .field("api_base_url", &self.api_base_url)
62 .field("realtime_url", &redacted_url_for_debug(&self.realtime_url))
63 .field("identity", &self.identity)
64 .field("store", &"<client-store>")
65 .field(
66 "realtime_connector",
67 &self
68 .realtime_connector
69 .as_ref()
70 .map(|_| "<realtime-connector>"),
71 )
72 .finish()
73 }
74}
75
76impl Default for SdkBackendBuilder {
77 fn default() -> Self {
78 Self {
79 api_base_url: DEFAULT_API_BASE_URL.to_owned(),
80 realtime_url: DEFAULT_REALTIME_URL.to_owned(),
81 identity: ClientIdentity::new("inline-client", VERSION),
82 store: Arc::new(InMemoryStore::new()),
83 realtime_connector: None,
84 }
85 }
86}
87
88impl SdkBackendBuilder {
89 pub fn api_base_url(mut self, api_base_url: impl Into<String>) -> Self {
91 self.api_base_url = api_base_url.into();
92 self
93 }
94
95 pub fn realtime_url(mut self, realtime_url: impl Into<String>) -> Self {
97 self.realtime_url = realtime_url.into();
98 self
99 }
100
101 pub fn identity(mut self, identity: ClientIdentity) -> Self {
103 self.identity = identity;
104 self
105 }
106
107 pub fn store(mut self, store: impl ClientStore) -> Self {
109 self.store = Arc::new(store);
110 self
111 }
112
113 pub fn shared_store(mut self, store: Arc<dyn ClientStore>) -> Self {
115 self.store = store;
116 self
117 }
118
119 pub fn enable_realtime_handshake(mut self) -> Self {
121 self.realtime_connector = Some(Arc::new(SdkRealtimeConnector::new()));
122 self
123 }
124
125 pub fn realtime_connector(mut self, connector: impl RealtimeConnector) -> Self {
127 self.realtime_connector = Some(Arc::new(connector));
128 self
129 }
130
131 pub fn shared_realtime_connector(mut self, connector: Arc<dyn RealtimeConnector>) -> Self {
133 self.realtime_connector = Some(connector);
134 self
135 }
136
137 pub fn without_realtime_handshake(mut self) -> Self {
139 self.realtime_connector = None;
140 self
141 }
142
143 pub fn build(self) -> Result<SdkBackend, SdkBackendBuildError> {
145 let api = ApiClient::try_new_with_identity(self.api_base_url, self.identity.clone())?;
146 Ok(SdkBackend {
147 api,
148 realtime_url: self.realtime_url,
149 identity: self.identity,
150 store: self.store,
151 realtime_connector: self.realtime_connector,
152 })
153 }
154}
155
156#[derive(Clone)]
158pub struct SdkBackend {
159 api: ApiClient,
160 realtime_url: String,
161 identity: ClientIdentity,
162 store: Arc<dyn ClientStore>,
163 realtime_connector: Option<Arc<dyn RealtimeConnector>>,
164}
165
166impl fmt::Debug for SdkBackend {
167 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
168 f.debug_struct("SdkBackend")
169 .field("api", &self.api)
170 .field("realtime_url", &redacted_url_for_debug(&self.realtime_url))
171 .field("identity", &self.identity)
172 .field("store", &"<client-store>")
173 .field(
174 "realtime_connector",
175 &self
176 .realtime_connector
177 .as_ref()
178 .map(|_| "<realtime-connector>"),
179 )
180 .finish()
181 }
182}
183
184impl SdkBackend {
185 pub fn builder() -> SdkBackendBuilder {
187 SdkBackendBuilder::default()
188 }
189
190 pub fn api_client(&self) -> &ApiClient {
192 &self.api
193 }
194
195 pub fn realtime_url(&self) -> &str {
197 &self.realtime_url
198 }
199
200 pub fn identity(&self) -> &ClientIdentity {
202 &self.identity
203 }
204
205 pub fn store(&self) -> Arc<dyn ClientStore> {
207 self.store.clone()
208 }
209
210 pub fn realtime_handshake_enabled(&self) -> bool {
212 self.realtime_connector.is_some()
213 }
214
215 async fn require_session(&self) -> BackendResult<StoredSession> {
216 self.store
217 .load_session()
218 .await
219 .map_err(store_error_to_backend)?
220 .ok_or_else(|| {
221 BackendError::new(ClientErrorCategory::AuthRequired, "client is not connected")
222 })
223 }
224
225 async fn connect_realtime(&self, session: &StoredSession) -> BackendResult<RealtimeClient> {
226 RealtimeClient::connect_with_identity(
227 &self.realtime_url,
228 session.auth.access_token().expose_secret(),
229 self.identity.clone(),
230 )
231 .await
232 .map_err(realtime_error_to_backend)
233 }
234
235 async fn connect_with_auth(
236 &self,
237 request: ConnectRequest,
238 ) -> BackendResult<ClientStatusSnapshot> {
239 if let Some(connector) = &self.realtime_connector {
240 connector
241 .connect(RealtimeConnectRequest::new(
242 self.realtime_url.clone(),
243 request.auth.access_token().clone(),
244 self.identity.clone(),
245 ))
246 .await?;
247 }
248 self.store
249 .save_session(StoredSession {
250 auth: request.auth,
251 account_namespace: request.account_namespace,
252 })
253 .await
254 .map_err(store_error_to_backend)?;
255 Ok(ClientStatusSnapshot::current(
256 crate::ClientStatus::Connected,
257 ))
258 }
259
260 fn auth_metadata(
261 &self,
262 kind: AuthContactKind,
263 contact: &str,
264 device_name: Option<&str>,
265 ) -> AuthMetadata {
266 let device_id = format!(
267 "inline-client-{:016x}",
268 stable_hash(&format!("{}:{contact}", kind.as_str()))
269 );
270 let metadata = AuthMetadata::new(device_id, self.identity.clone());
271 match device_name
272 .map(str::trim)
273 .filter(|device_name| !device_name.is_empty())
274 {
275 Some(device_name) => metadata.with_device_name(device_name),
276 None => metadata.with_device_name(self.identity.client_type()),
277 }
278 }
279
280 async fn resume_stored_session(&self) -> BackendResult<ClientStatusSnapshot> {
281 let Some(session) = self
282 .store
283 .load_session()
284 .await
285 .map_err(store_error_to_backend)?
286 else {
287 return Ok(ClientStatusSnapshot::current(
288 crate::ClientStatus::AuthRequired,
289 ));
290 };
291
292 if let Some(connector) = &self.realtime_connector {
293 connector
294 .connect(RealtimeConnectRequest::new(
295 self.realtime_url.clone(),
296 session.auth.access_token().clone(),
297 self.identity.clone(),
298 ))
299 .await?;
300 }
301 Ok(ClientStatusSnapshot::current(
302 crate::ClientStatus::Connected,
303 ))
304 }
305}
306
307impl ClientBackend for SdkBackend {
308 fn auth_start(
309 &self,
310 request: AuthStartRequest,
311 ) -> BoxFuture<'static, BackendResult<AuthStartResult>> {
312 let backend = self.clone();
313 Box::pin(async move {
314 let contact = request.contact.trim().to_owned();
315 if contact.is_empty() {
316 return Err(BackendError::new(
317 ClientErrorCategory::InvalidInput,
318 "auth contact must not be empty",
319 ));
320 }
321 let metadata =
322 backend.auth_metadata(request.kind, &contact, request.device_name.as_deref());
323 let result = match request.kind {
324 AuthContactKind::Email => backend
325 .api
326 .send_email_code(&contact, &metadata)
327 .await
328 .map_err(api_error_to_backend)?,
329 AuthContactKind::Phone => backend
330 .api
331 .send_sms_code(&contact, &metadata)
332 .await
333 .map_err(api_error_to_backend)?,
334 };
335 Ok(AuthStartResult {
336 existing_user: result.existing_user,
337 needs_invite_code: result.needs_invite_code,
338 challenge_token: result.challenge_token,
339 })
340 })
341 }
342
343 fn auth_verify(
344 &self,
345 request: AuthVerifyRequest,
346 ) -> BoxFuture<'static, BackendResult<AuthVerifyResult>> {
347 let backend = self.clone();
348 Box::pin(async move {
349 let contact = request.contact.trim().to_owned();
350 let code = request.code.trim().to_owned();
351 if contact.is_empty() {
352 return Err(BackendError::new(
353 ClientErrorCategory::InvalidInput,
354 "auth contact must not be empty",
355 ));
356 }
357 if code.is_empty() {
358 return Err(BackendError::new(
359 ClientErrorCategory::InvalidInput,
360 "verification code must not be empty",
361 ));
362 }
363
364 let metadata =
365 backend.auth_metadata(request.kind, &contact, request.device_name.as_deref());
366 let result = match request.kind {
367 AuthContactKind::Email => backend
368 .api
369 .verify_email_code(
370 &contact,
371 &code,
372 request.challenge_token.as_deref(),
373 &metadata,
374 )
375 .await
376 .map_err(api_error_to_backend)?,
377 AuthContactKind::Phone => backend
378 .api
379 .verify_sms_code(&contact, &code, &metadata)
380 .await
381 .map_err(api_error_to_backend)?,
382 };
383 let token = AuthToken::try_new(result.token).map_err(|error| {
384 BackendError::new(ClientErrorCategory::ProtocolMismatch, error.to_string())
385 })?;
386 let account_namespace = request
387 .account_namespace
388 .map(|namespace| namespace.trim().to_owned())
389 .filter(|namespace| !namespace.is_empty())
390 .unwrap_or_else(|| result.user_id.to_string());
391 let status = backend
392 .connect_with_auth(
393 ConnectRequest::new(AuthCredential::AccessToken { token })
394 .with_account_namespace(account_namespace.clone()),
395 )
396 .await?;
397 Ok(AuthVerifyResult {
398 user_id: InlineId::new(result.user_id),
399 account_namespace,
400 status,
401 })
402 })
403 }
404
405 fn resume_session(&self) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
406 let backend = self.clone();
407 Box::pin(async move { backend.resume_stored_session().await })
408 }
409
410 fn connect(
411 &self,
412 request: ConnectRequest,
413 ) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
414 let backend = self.clone();
415 Box::pin(async move { backend.connect_with_auth(request).await })
416 }
417
418 fn logout(&self) -> BoxFuture<'static, BackendResult<()>> {
419 let backend = self.clone();
420 Box::pin(async move {
421 backend
422 .store
423 .clear_session()
424 .await
425 .map_err(store_error_to_backend)
426 })
427 }
428
429 fn dialogs(&self, request: DialogsRequest) -> BoxFuture<'static, BackendResult<DialogsPage>> {
430 let backend = self.clone();
431 Box::pin(async move {
432 let session = backend.require_session().await?;
433 let live: BackendResult<DialogsPage> = async {
434 let mut realtime = backend.connect_realtime(&session).await?;
435 let result = realtime
436 .call(proto::GetChatsInput {})
437 .await
438 .map_err(realtime_error_to_backend)?;
439 let page = dialogs_page_from_get_chats(&result, request.clone())?;
440 for dialog in &page.dialogs {
441 backend
442 .store
443 .record_dialog(dialog.clone())
444 .await
445 .map_err(store_error_to_backend)?;
446 }
447 backend
448 .store
449 .record_users(page.users.clone())
450 .await
451 .map_err(store_error_to_backend)?;
452 for message in result.messages {
453 let record = message_record_from_proto_message(message, None, None);
454 backend
455 .store
456 .record_message(record)
457 .await
458 .map_err(store_error_to_backend)?;
459 }
460 backend
461 .store
462 .dialogs(request.clone())
463 .await
464 .map_err(store_error_to_backend)
465 }
466 .await;
467 match live {
468 Ok(page) => Ok(page),
469 Err(error) => {
470 log::warn!(
471 "Inline realtime dialog sync failed; serving cached dialogs: {}",
472 error
473 );
474 backend
475 .store
476 .dialogs(request)
477 .await
478 .map_err(store_error_to_backend)
479 }
480 }
481 })
482 }
483
484 fn history(&self, request: HistoryRequest) -> BoxFuture<'static, BackendResult<HistoryPage>> {
485 let backend = self.clone();
486 Box::pin(async move {
487 let session = backend.require_session().await?;
488 if request.before_message_id.is_some() && request.after_message_id.is_some() {
489 return Err(BackendError::new(
490 ClientErrorCategory::InvalidInput,
491 "history request cannot specify both before_message_id and after_message_id",
492 ));
493 }
494 let limit = request.limit.unwrap_or(50).max(1);
495 let live: BackendResult<HistoryPage> = async {
496 let mut realtime = backend.connect_realtime(&session).await?;
497 let fetch_limit = limit.saturating_add(1).min(i32::MAX as u32) as i32;
498 let result = realtime
499 .call(history_input_for_request(&request, fetch_limit))
500 .await
501 .map_err(realtime_error_to_backend)?;
502 let mut records = result
503 .messages
504 .into_iter()
505 .map(|message| {
506 message_record_from_proto_message(message, Some(request.chat_id), None)
507 })
508 .collect::<Vec<_>>();
509 records.sort_by_key(|message| (message.timestamp, message.message_id.get()));
510 let has_more = records.len() > limit as usize;
511 if has_more {
512 records.truncate(limit as usize);
513 }
514 for message in &records {
515 backend
516 .store
517 .record_message(message.clone())
518 .await
519 .map_err(store_error_to_backend)?;
520 }
521 Ok(HistoryPage {
522 messages: records,
523 users: Vec::new(),
524 has_more,
525 next_cursor: None,
526 })
527 }
528 .await;
529 match live {
530 Ok(page) => Ok(page),
531 Err(error) => {
532 log::warn!(
533 "Inline realtime history sync failed; serving cached history for chat {}: {}",
534 request.chat_id.get(),
535 error
536 );
537 backend
538 .store
539 .history(request)
540 .await
541 .map_err(store_error_to_backend)
542 }
543 }
544 })
545 }
546
547 fn chat_participants(
548 &self,
549 request: ChatParticipantsRequest,
550 ) -> BoxFuture<'static, BackendResult<ChatParticipantsPage>> {
551 let backend = self.clone();
552 Box::pin(async move {
553 let session = backend.require_session().await?;
554 let mut realtime = backend.connect_realtime(&session).await?;
555 let result = realtime
556 .call(proto::GetChatParticipantsInput {
557 chat_id: request.chat_id.get(),
558 })
559 .await
560 .map_err(realtime_error_to_backend)?;
561 let page = chat_participants_page_from_proto(result);
562 backend
563 .store
564 .record_users(page.users.clone())
565 .await
566 .map_err(store_error_to_backend)?;
567 Ok(page)
568 })
569 }
570
571 fn create_dm(
572 &self,
573 request: CreateDmRequest,
574 ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
575 let backend = self.clone();
576 Box::pin(async move {
577 if request.user_id.get() <= 0 {
578 return Err(BackendError::new(
579 ClientErrorCategory::InvalidInput,
580 "user_id must be positive",
581 ));
582 }
583 let session = backend.require_session().await?;
584 let result = backend
585 .api
586 .create_private_chat(
587 session.auth.access_token().expose_secret(),
588 request.user_id.get(),
589 )
590 .await
591 .map_err(api_error_to_backend)?;
592 let (created, user) = created_chat_from_private_chat_result(
593 result.chat,
594 result.dialog,
595 result.user,
596 request.user_id,
597 )?;
598 backend
599 .store
600 .record_dialog(DialogRecord {
601 chat_id: created.chat_id,
602 title: created.title.clone(),
603 last_message_id: None,
604 synced_through_message_id: None,
605 unread_count: Some(0),
606 })
607 .await
608 .map_err(store_error_to_backend)?;
609 if let Some(user) = user {
610 backend
611 .store
612 .record_users(vec![user])
613 .await
614 .map_err(store_error_to_backend)?;
615 }
616 Ok(created)
617 })
618 }
619
620 fn create_thread(
621 &self,
622 request: CreateThreadRequest,
623 ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
624 let backend = self.clone();
625 Box::pin(async move {
626 validate_create_participants(&request.participants)?;
627 let session = backend.require_session().await?;
628 let mut realtime = backend.connect_realtime(&session).await?;
629 let result = realtime
630 .call(proto::CreateChatInput {
631 title: trimmed_option(request.title),
632 space_id: request.space_id.map(InlineId::get),
633 description: trimmed_option(request.description),
634 emoji: trimmed_option(request.emoji),
635 is_public: request.is_public,
636 participants: request
637 .participants
638 .into_iter()
639 .map(|participant| proto::InputChatParticipant {
640 user_id: participant.user_id.get(),
641 })
642 .collect(),
643 reserved_chat_id: None,
644 })
645 .await
646 .map_err(realtime_error_to_backend)?;
647 let created = created_chat_from_proto(result.chat, result.dialog, None, None, None)?;
648 backend
649 .store
650 .record_dialog(DialogRecord {
651 chat_id: created.chat_id,
652 title: created.title.clone(),
653 last_message_id: None,
654 synced_through_message_id: None,
655 unread_count: Some(0),
656 })
657 .await
658 .map_err(store_error_to_backend)?;
659 Ok(created)
660 })
661 }
662
663 fn create_reply_thread(
664 &self,
665 request: CreateReplyThreadRequest,
666 ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
667 let backend = self.clone();
668 Box::pin(async move {
669 if request.parent_chat_id.get() <= 0 {
670 return Err(BackendError::new(
671 ClientErrorCategory::InvalidInput,
672 "parent_chat_id must be positive",
673 ));
674 }
675 validate_create_participants(&request.participants)?;
676 let session = backend.require_session().await?;
677 let mut realtime = backend.connect_realtime(&session).await?;
678 let result = realtime
679 .call(proto::CreateSubthreadInput {
680 parent_chat_id: request.parent_chat_id.get(),
681 parent_message_id: request.parent_message_id.map(InlineId::get),
682 title: trimmed_option(request.title),
683 description: trimmed_option(request.description),
684 emoji: trimmed_option(request.emoji),
685 participants: request
686 .participants
687 .into_iter()
688 .map(|participant| proto::InputChatParticipant {
689 user_id: participant.user_id.get(),
690 })
691 .collect(),
692 })
693 .await
694 .map_err(realtime_error_to_backend)?;
695 let created = created_chat_from_proto(
696 result.chat,
697 result.dialog,
698 result.anchor_message,
699 Some(request.parent_chat_id),
700 request.parent_message_id,
701 )?;
702 backend
703 .store
704 .record_dialog(DialogRecord {
705 chat_id: created.chat_id,
706 title: created.title.clone(),
707 last_message_id: None,
708 synced_through_message_id: None,
709 unread_count: Some(0),
710 })
711 .await
712 .map_err(store_error_to_backend)?;
713 Ok(created)
714 })
715 }
716
717 fn send_text(
718 &self,
719 request: SendTextRequest,
720 ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
721 let backend = self.clone();
722 Box::pin(async move {
723 if request.text.trim().is_empty() {
724 return Err(BackendError::new(
725 ClientErrorCategory::InvalidInput,
726 "message text must not be empty",
727 ));
728 }
729
730 let session = backend.require_session().await?;
731 let initial_chat_id = chat_id_for_peer(request.peer);
732 let random_id = request
733 .random_id
734 .unwrap_or_else(|| random_id_for_request(&request));
735 let transaction_id = transaction_id_for_send(&request, random_id);
736 let identity = TransactionIdentity::new(
737 transaction_id.clone(),
738 request.external_id.clone(),
739 random_id,
740 );
741
742 if let Some(existing) = backend
743 .store
744 .transaction(transaction_id.clone())
745 .await
746 .map_err(store_error_to_backend)?
747 {
748 return Ok(outcome_from_stored_transaction(existing, initial_chat_id));
749 }
750
751 backend
752 .store
753 .record_transaction(
754 StoredTransaction::new(identity.clone(), TransactionState::Queued)
755 .with_chat_id(initial_chat_id),
756 )
757 .await
758 .map_err(store_error_to_backend)?;
759
760 let mut realtime = match RealtimeClient::connect_with_identity(
761 &backend.realtime_url,
762 session.auth.access_token().expose_secret(),
763 backend.identity.clone(),
764 )
765 .await
766 {
767 Ok(realtime) => realtime,
768 Err(error) => {
769 let backend_error = realtime_error_to_backend(error);
770 backend
771 .record_failed_transaction(identity, initial_chat_id, backend_error.clone())
772 .await?;
773 return Err(backend_error);
774 }
775 };
776
777 backend
778 .store
779 .record_transaction(
780 StoredTransaction::new(identity.clone(), TransactionState::Sent)
781 .with_chat_id(initial_chat_id),
782 )
783 .await
784 .map_err(store_error_to_backend)?;
785
786 let input = proto::SendMessageInput {
787 peer_id: Some(input_peer_for_client_peer(request.peer)),
788 message: Some(request.text.clone()),
789 reply_to_msg_id: request.reply_to_message_id.map(InlineId::get),
790 random_id: Some(random_id.get()),
791 media: None,
792 temporary_send_date: Some(now_seconds()),
793 is_sticker: None,
794 has_link: None,
795 entities: None,
796 parse_markdown: Some(false),
797 send_mode: None,
798 actions: None,
799 };
800 let send_result = match realtime.call(input).await {
801 Ok(result) => result,
802 Err(error) => {
803 let backend_error = realtime_error_to_backend(error);
804 backend
805 .record_failed_transaction(identity, initial_chat_id, backend_error.clone())
806 .await?;
807 return Err(backend_error);
808 }
809 };
810
811 let applied =
812 apply_send_message_updates(&request, identity, initial_chat_id, send_result);
813 if let Some(message) = &applied.message {
814 backend
815 .store
816 .record_message(message.clone())
817 .await
818 .map_err(store_error_to_backend)?;
819 }
820 backend
821 .store
822 .record_transaction(applied.transaction.clone())
823 .await
824 .map_err(store_error_to_backend)?;
825
826 Ok(SendTextOutcome::with_state(
827 MessageMutation {
828 transaction: applied.transaction.identity,
829 message_id: applied.message_id,
830 },
831 applied.chat_id,
832 applied.message_id,
833 applied.message,
834 applied.transaction.state,
835 applied.transaction.failure,
836 ))
837 })
838 }
839
840 fn send_media(
841 &self,
842 request: UploadRequest,
843 bytes: Vec<u8>,
844 ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
845 let backend = self.clone();
846 Box::pin(async move {
847 if bytes.is_empty() {
848 return Err(BackendError::new(
849 ClientErrorCategory::InvalidInput,
850 "media bytes must not be empty",
851 ));
852 }
853
854 let session = backend.require_session().await?;
855 let initial_chat_id = chat_id_for_peer(request.peer);
856 let random_id = request
857 .random_id
858 .unwrap_or_else(|| random_id_for_upload_request(&request, bytes.len()));
859 let transaction_id = transaction_id_for_upload(&request, random_id);
860 let identity = TransactionIdentity::new(
861 transaction_id.clone(),
862 request.external_id.clone(),
863 random_id,
864 );
865
866 if let Some(existing) = backend
867 .store
868 .transaction(transaction_id.clone())
869 .await
870 .map_err(store_error_to_backend)?
871 {
872 return Ok(outcome_from_stored_transaction(existing, initial_chat_id));
873 }
874
875 backend
876 .store
877 .record_transaction(
878 StoredTransaction::new(identity.clone(), TransactionState::Queued)
879 .with_chat_id(initial_chat_id),
880 )
881 .await
882 .map_err(store_error_to_backend)?;
883
884 let upload_input = upload_input_for_request(&request, bytes);
885 let upload = match backend
886 .api
887 .upload_file_bytes(session.auth.access_token().expose_secret(), upload_input)
888 .await
889 {
890 Ok(upload) => upload,
891 Err(error) => {
892 let backend_error = api_error_to_backend(error);
893 backend
894 .record_failed_transaction(identity, initial_chat_id, backend_error.clone())
895 .await?;
896 return Err(backend_error);
897 }
898 };
899 let media = match input_media_from_upload(&upload) {
900 Ok(media) => media,
901 Err(error) => {
902 backend
903 .record_failed_transaction(identity, initial_chat_id, error.clone())
904 .await?;
905 return Err(error);
906 }
907 };
908
909 let mut realtime = match backend.connect_realtime(&session).await {
910 Ok(realtime) => realtime,
911 Err(error) => {
912 backend
913 .record_failed_transaction(identity, initial_chat_id, error.clone())
914 .await?;
915 return Err(error);
916 }
917 };
918
919 backend
920 .store
921 .record_transaction(
922 StoredTransaction::new(identity.clone(), TransactionState::Sent)
923 .with_chat_id(initial_chat_id),
924 )
925 .await
926 .map_err(store_error_to_backend)?;
927
928 let input = proto::SendMessageInput {
929 peer_id: Some(input_peer_for_client_peer(request.peer)),
930 message: request.caption.clone(),
931 reply_to_msg_id: request.reply_to_message_id.map(InlineId::get),
932 random_id: Some(random_id.get()),
933 media: Some(media),
934 temporary_send_date: Some(now_seconds()),
935 is_sticker: None,
936 has_link: None,
937 entities: None,
938 parse_markdown: Some(false),
939 send_mode: None,
940 actions: None,
941 };
942 let send_result = match realtime.call(input).await {
943 Ok(result) => result,
944 Err(error) => {
945 let backend_error = realtime_error_to_backend(error);
946 backend
947 .record_failed_transaction(identity, initial_chat_id, backend_error.clone())
948 .await?;
949 return Err(backend_error);
950 }
951 };
952
953 let applied =
954 apply_upload_message_updates(&request, identity, initial_chat_id, send_result);
955 if let Some(message) = &applied.message {
956 backend
957 .store
958 .record_message(message.clone())
959 .await
960 .map_err(store_error_to_backend)?;
961 }
962 backend
963 .store
964 .record_transaction(applied.transaction.clone())
965 .await
966 .map_err(store_error_to_backend)?;
967
968 Ok(SendTextOutcome::with_state(
969 MessageMutation {
970 transaction: applied.transaction.identity,
971 message_id: applied.message_id,
972 },
973 applied.chat_id,
974 applied.message_id,
975 applied.message,
976 applied.transaction.state,
977 applied.transaction.failure,
978 ))
979 })
980 }
981
982 fn edit_message(
983 &self,
984 request: EditMessageRequest,
985 ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
986 let backend = self.clone();
987 Box::pin(async move {
988 if request.text.trim().is_empty() {
989 return Err(BackendError::new(
990 ClientErrorCategory::InvalidInput,
991 "message text must not be empty",
992 ));
993 }
994 let session = backend.require_session().await?;
995 let mut realtime = backend.connect_realtime(&session).await?;
996 let result = realtime
997 .call(proto::EditMessageInput {
998 message_id: request.message_id.get(),
999 peer_id: Some(input_peer_for_chat(request.chat_id)),
1000 text: request.text,
1001 entities: None,
1002 parse_markdown: Some(false),
1003 actions: None,
1004 })
1005 .await
1006 .map_err(realtime_error_to_backend)?;
1007 let events = backend
1008 .apply_updates(result.updates, Some(request.chat_id), None)
1009 .await?;
1010 Ok(OperationOutcome::with_events(events))
1011 })
1012 }
1013
1014 fn delete_message(
1015 &self,
1016 request: DeleteMessageRequest,
1017 ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1018 let backend = self.clone();
1019 Box::pin(async move {
1020 let session = backend.require_session().await?;
1021 let mut realtime = backend.connect_realtime(&session).await?;
1022 let result = realtime
1023 .call(proto::DeleteMessagesInput {
1024 message_ids: vec![request.message_id.get()],
1025 peer_id: Some(input_peer_for_chat(request.chat_id)),
1026 })
1027 .await
1028 .map_err(realtime_error_to_backend)?;
1029 let mut events = backend
1030 .apply_updates(result.updates, Some(request.chat_id), None)
1031 .await?;
1032 if !events.iter().any(|event| {
1033 matches!(
1034 event,
1035 ClientEvent::MessageDeleted {
1036 chat_id,
1037 message_id
1038 } if *chat_id == request.chat_id && *message_id == request.message_id
1039 )
1040 }) {
1041 events.push(ClientEvent::MessageDeleted {
1042 chat_id: request.chat_id,
1043 message_id: request.message_id,
1044 });
1045 }
1046 Ok(OperationOutcome::with_events(events))
1047 })
1048 }
1049
1050 fn react(&self, request: ReactRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1051 let backend = self.clone();
1052 Box::pin(async move {
1053 let reaction = request.reaction.trim().to_owned();
1054 if reaction.is_empty() {
1055 return Err(BackendError::new(
1056 ClientErrorCategory::InvalidInput,
1057 "reaction must not be empty",
1058 ));
1059 }
1060 let session = backend.require_session().await?;
1061 let mut realtime = backend.connect_realtime(&session).await?;
1062 let updates = if request.remove {
1063 realtime
1064 .call(proto::DeleteReactionInput {
1065 emoji: reaction.clone(),
1066 peer_id: Some(input_peer_for_chat(request.chat_id)),
1067 message_id: request.message_id.get(),
1068 })
1069 .await
1070 .map_err(realtime_error_to_backend)?
1071 .updates
1072 } else {
1073 realtime
1074 .call(proto::AddReactionInput {
1075 emoji: reaction.clone(),
1076 message_id: request.message_id.get(),
1077 peer_id: Some(input_peer_for_chat(request.chat_id)),
1078 })
1079 .await
1080 .map_err(realtime_error_to_backend)?
1081 .updates
1082 };
1083 let mut events = backend
1084 .apply_updates(updates, Some(request.chat_id), None)
1085 .await?;
1086 if !events.iter().any(|event| {
1087 matches!(
1088 event,
1089 ClientEvent::ReactionChanged {
1090 chat_id,
1091 message_id,
1092 ..
1093 } if *chat_id == request.chat_id && *message_id == request.message_id
1094 )
1095 }) {
1096 events.push(ClientEvent::ReactionChanged {
1097 chat_id: request.chat_id,
1098 message_id: request.message_id,
1099 user_id: InlineId::new(0),
1100 reaction,
1101 removed: request.remove,
1102 });
1103 }
1104 Ok(OperationOutcome::with_events(events))
1105 })
1106 }
1107
1108 fn read(&self, request: ReadRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1109 let backend = self.clone();
1110 Box::pin(async move {
1111 let session = backend.require_session().await?;
1112 let mut realtime = backend.connect_realtime(&session).await?;
1113 let result = realtime
1114 .call(proto::ReadMessagesInput {
1115 peer_id: Some(input_peer_for_chat(request.chat_id)),
1116 max_id: request.max_message_id.map(InlineId::get),
1117 })
1118 .await
1119 .map_err(realtime_error_to_backend)?;
1120 let mut events = backend
1121 .apply_updates(result.updates, Some(request.chat_id), None)
1122 .await?;
1123 if !events.iter().any(|event| {
1124 matches!(event, ClientEvent::ReadStateChanged { chat_id } if *chat_id == request.chat_id)
1125 }) {
1126 events.push(ClientEvent::ReadStateChanged {
1127 chat_id: request.chat_id,
1128 });
1129 }
1130 Ok(OperationOutcome::with_events(events))
1131 })
1132 }
1133
1134 fn typing(
1135 &self,
1136 request: TypingRequest,
1137 ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1138 let backend = self.clone();
1139 Box::pin(async move {
1140 let session = backend.require_session().await?;
1141 let mut realtime = backend.connect_realtime(&session).await?;
1142 realtime
1143 .call(proto::SendComposeActionInput {
1144 peer_id: Some(input_peer_for_chat(request.chat_id)),
1145 action: request
1146 .is_typing
1147 .then_some(proto::update_compose_action::ComposeAction::Typing as i32),
1148 })
1149 .await
1150 .map_err(realtime_error_to_backend)?;
1151 Ok(OperationOutcome::empty())
1152 })
1153 }
1154}
1155
1156impl SdkBackend {
1157 async fn record_failed_transaction(
1158 &self,
1159 identity: TransactionIdentity,
1160 chat_id: InlineId,
1161 error: BackendError,
1162 ) -> BackendResult<()> {
1163 self.store
1164 .record_transaction(
1165 StoredTransaction::new(identity, TransactionState::Failed)
1166 .with_chat_id(chat_id)
1167 .with_failure(error.into()),
1168 )
1169 .await
1170 .map_err(store_error_to_backend)
1171 }
1172
1173 async fn apply_updates(
1174 &self,
1175 updates: Vec<proto::Update>,
1176 fallback_chat_id: Option<InlineId>,
1177 transaction: Option<TransactionIdentity>,
1178 ) -> BackendResult<Vec<ClientEvent>> {
1179 let mut events = Vec::new();
1180 for update in updates {
1181 match update.update {
1182 Some(proto::update::Update::NewMessage(update)) => {
1183 if let Some(message) = update.message {
1184 let record =
1185 message_record_from_proto_message(message, fallback_chat_id, None);
1186 self.store
1187 .record_message(record.clone())
1188 .await
1189 .map_err(store_error_to_backend)?;
1190 events.push(ClientEvent::MessageStored { message: record });
1191 }
1192 }
1193 Some(proto::update::Update::EditMessage(update)) => {
1194 if let Some(message) = update.message {
1195 let record =
1196 message_record_from_proto_message(message, fallback_chat_id, None);
1197 self.store
1198 .record_message(record.clone())
1199 .await
1200 .map_err(store_error_to_backend)?;
1201 events.push(ClientEvent::MessageStored { message: record });
1202 }
1203 }
1204 Some(proto::update::Update::DeleteMessages(update)) => {
1205 let chat_id = update
1206 .peer_id
1207 .as_ref()
1208 .and_then(chat_id_from_peer)
1209 .or(fallback_chat_id);
1210 if let Some(chat_id) = chat_id {
1211 for message_id in update.message_ids {
1212 events.push(ClientEvent::MessageDeleted {
1213 chat_id,
1214 message_id: InlineId::new(message_id),
1215 });
1216 }
1217 }
1218 }
1219 Some(proto::update::Update::UpdateReaction(update)) => {
1220 if let Some(reaction) = update.reaction {
1221 events.push(ClientEvent::ReactionChanged {
1222 chat_id: InlineId::new(reaction.chat_id),
1223 message_id: InlineId::new(reaction.message_id),
1224 user_id: InlineId::new(reaction.user_id),
1225 reaction: reaction.emoji,
1226 removed: false,
1227 });
1228 }
1229 }
1230 Some(proto::update::Update::DeleteReaction(update)) => {
1231 events.push(ClientEvent::ReactionChanged {
1232 chat_id: InlineId::new(update.chat_id),
1233 message_id: InlineId::new(update.message_id),
1234 user_id: InlineId::new(update.user_id),
1235 reaction: update.emoji,
1236 removed: true,
1237 });
1238 }
1239 Some(proto::update::Update::UpdateReadMaxId(update)) => {
1240 let chat_id = update
1241 .peer_id
1242 .as_ref()
1243 .and_then(chat_id_from_peer)
1244 .or(fallback_chat_id);
1245 if let Some(chat_id) = chat_id {
1246 events.push(ClientEvent::ReadStateChanged { chat_id });
1247 }
1248 }
1249 Some(proto::update::Update::UpdateComposeAction(update)) => {
1250 let chat_id = update
1251 .peer_id
1252 .as_ref()
1253 .and_then(chat_id_from_peer)
1254 .or(fallback_chat_id);
1255 if let Some(chat_id) = chat_id {
1256 events.push(ClientEvent::Typing {
1257 chat_id,
1258 user_id: InlineId::new(update.user_id),
1259 is_typing: update.action
1260 == proto::update_compose_action::ComposeAction::Typing as i32,
1261 });
1262 }
1263 }
1264 Some(proto::update::Update::UpdateMessageId(update)) => {
1265 if let (Some(chat_id), Some(transaction)) =
1266 (fallback_chat_id, transaction.as_ref())
1267 && update.random_id == transaction.random_id.get()
1268 {
1269 events.push(ClientEvent::MessageUpserted {
1270 chat_id,
1271 message_id: InlineId::new(update.message_id),
1272 });
1273 }
1274 }
1275 _ => {}
1276 }
1277 }
1278 Ok(events)
1279 }
1280}
1281
1282fn store_error_to_backend(error: StoreError) -> BackendError {
1283 BackendError::new(error.category, error.message)
1284}
1285
1286fn dialogs_page_from_get_chats(
1287 result: &proto::GetChatsResult,
1288 request: DialogsRequest,
1289) -> BackendResult<DialogsPage> {
1290 let users_by_id = result
1291 .users
1292 .iter()
1293 .map(|user| (user.id, user))
1294 .collect::<HashMap<_, _>>();
1295 let start = request
1296 .cursor
1297 .as_deref()
1298 .map(str::trim)
1299 .filter(|cursor| !cursor.is_empty())
1300 .map(|cursor| {
1301 cursor.parse::<usize>().map_err(|_| {
1302 BackendError::new(
1303 ClientErrorCategory::InvalidInput,
1304 "invalid pagination cursor",
1305 )
1306 })
1307 })
1308 .transpose()?
1309 .unwrap_or(0);
1310 let limit = request.limit.unwrap_or(100).max(1) as usize;
1311
1312 let mut dialogs = result
1313 .chats
1314 .iter()
1315 .map(|chat| {
1316 let dialog = dialog_for_chat(result, chat);
1317 DialogRecord {
1318 chat_id: InlineId::new(chat.id),
1319 title: Some(chat_display_name_from_proto(chat, &users_by_id)),
1320 last_message_id: chat.last_msg_id.map(InlineId::new),
1321 synced_through_message_id: None,
1322 unread_count: dialog
1323 .and_then(|dialog| dialog.unread_count)
1324 .and_then(|count| u32::try_from(count).ok()),
1325 }
1326 })
1327 .collect::<Vec<_>>();
1328 let total = dialogs.len();
1329 if start >= dialogs.len() {
1330 dialogs.clear();
1331 } else {
1332 dialogs = dialogs.into_iter().skip(start).take(limit).collect();
1333 }
1334 Ok(DialogsPage {
1335 dialogs,
1336 users: result.users.iter().map(user_record_from_proto).collect(),
1337 next_cursor: (start + limit < total).then(|| (start + limit).to_string()),
1338 })
1339}
1340
1341fn dialog_for_chat<'a>(
1342 result: &'a proto::GetChatsResult,
1343 chat: &proto::Chat,
1344) -> Option<&'a proto::Dialog> {
1345 result.dialogs.iter().find(|dialog| {
1346 dialog.chat_id == Some(chat.id)
1347 || dialog
1348 .peer
1349 .as_ref()
1350 .zip(chat.peer_id.as_ref())
1351 .is_some_and(|(dialog_peer, chat_peer)| dialog_peer == chat_peer)
1352 })
1353}
1354
1355fn chat_display_name_from_proto(
1356 chat: &proto::Chat,
1357 users_by_id: &HashMap<i64, &proto::User>,
1358) -> String {
1359 let mut name = chat
1360 .peer_id
1361 .as_ref()
1362 .and_then(user_id_from_peer)
1363 .and_then(|user_id| users_by_id.get(&user_id))
1364 .and_then(|user| user_display_name_from_proto(user))
1365 .unwrap_or_else(|| {
1366 let title = chat.title.trim();
1367 if title.is_empty() {
1368 format!("Chat {}", chat.id)
1369 } else {
1370 title.to_owned()
1371 }
1372 });
1373 if let Some(emoji) = chat
1374 .emoji
1375 .as_deref()
1376 .map(str::trim)
1377 .filter(|value| !value.is_empty())
1378 {
1379 name = format!("{emoji} {name}");
1380 }
1381 name
1382}
1383
1384fn user_record_from_proto(user: &proto::User) -> UserRecord {
1385 UserRecord {
1386 user_id: InlineId::new(user.id),
1387 display_name: user_display_name_from_proto(user),
1388 username: non_empty_option(user.username.as_deref()),
1389 first_name: non_empty_option(user.first_name.as_deref()),
1390 last_name: non_empty_option(user.last_name.as_deref()),
1391 avatar_url: user
1392 .profile_photo
1393 .as_ref()
1394 .and_then(|photo| non_empty_option(photo.cdn_url.as_deref()))
1395 .or_else(|| {
1396 user.bot_avatar
1397 .as_ref()
1398 .and_then(|avatar| non_empty_option(avatar.cdn_url.as_deref()))
1399 }),
1400 is_bot: user.bot,
1401 }
1402}
1403
1404fn chat_participants_page_from_proto(
1405 result: proto::GetChatParticipantsResult,
1406) -> ChatParticipantsPage {
1407 let mut seen = std::collections::HashSet::new();
1408 let mut participants = Vec::new();
1409 for participant in result.participants {
1410 if seen.insert(participant.user_id) {
1411 participants.push(ChatParticipantRecord {
1412 user_id: InlineId::new(participant.user_id),
1413 date: Some(participant.date),
1414 });
1415 }
1416 }
1417 participants.sort_by_key(|participant| participant.user_id.get());
1418 ChatParticipantsPage {
1419 participants,
1420 users: result.users.iter().map(user_record_from_proto).collect(),
1421 }
1422}
1423
1424fn validate_create_participants(participants: &[ChatCreateParticipant]) -> BackendResult<()> {
1425 if participants
1426 .iter()
1427 .any(|participant| participant.user_id.get() <= 0)
1428 {
1429 return Err(BackendError::new(
1430 ClientErrorCategory::InvalidInput,
1431 "participant user_id must be positive",
1432 ));
1433 }
1434 Ok(())
1435}
1436
1437fn created_chat_from_private_chat_result(
1438 chat: Value,
1439 dialog: Value,
1440 user: Value,
1441 user_id: InlineId,
1442) -> BackendResult<(CreatedChat, Option<UserRecord>)> {
1443 let chat_id = api_i64_field(&chat, &["id", "chatId", "chat_id"])
1444 .or_else(|| api_i64_field(&dialog, &["chatId", "chat_id"]))
1445 .ok_or_else(|| {
1446 BackendError::new(
1447 ClientErrorCategory::ProtocolMismatch,
1448 "create_private_chat result did not include a chat id",
1449 )
1450 })?;
1451 let user = user_record_from_api_value(&user).or(Some(UserRecord {
1452 user_id,
1453 display_name: None,
1454 username: None,
1455 first_name: None,
1456 last_name: None,
1457 avatar_url: None,
1458 is_bot: None,
1459 }));
1460 let title = api_string_field(&chat, &["title"])
1461 .or_else(|| user.as_ref().and_then(user_display_name_from_record));
1462 Ok((
1463 CreatedChat {
1464 chat_id: InlineId::new(chat_id),
1465 title,
1466 parent_chat_id: None,
1467 parent_message_id: None,
1468 },
1469 user,
1470 ))
1471}
1472
1473fn created_chat_from_proto(
1474 chat: Option<proto::Chat>,
1475 dialog: Option<proto::Dialog>,
1476 anchor_message: Option<proto::Message>,
1477 parent_chat_id: Option<InlineId>,
1478 parent_message_id: Option<InlineId>,
1479) -> BackendResult<CreatedChat> {
1480 let chat_id = chat
1481 .as_ref()
1482 .map(|chat| chat.id)
1483 .or_else(|| dialog.as_ref().and_then(|dialog| dialog.chat_id))
1484 .ok_or_else(|| {
1485 BackendError::new(
1486 ClientErrorCategory::ProtocolMismatch,
1487 "create chat result did not include a chat id",
1488 )
1489 })?;
1490 let title = chat
1491 .as_ref()
1492 .and_then(|chat| trimmed_option(Some(chat.title.clone())));
1493 let parent_chat_id = parent_chat_id.or_else(|| {
1494 chat.as_ref()
1495 .and_then(|chat| chat.parent_chat_id)
1496 .map(InlineId::new)
1497 });
1498 let parent_message_id = parent_message_id
1499 .or_else(|| {
1500 chat.as_ref()
1501 .and_then(|chat| chat.parent_message_id)
1502 .map(InlineId::new)
1503 })
1504 .or_else(|| anchor_message.map(|message| InlineId::new(message.id)));
1505
1506 Ok(CreatedChat {
1507 chat_id: InlineId::new(chat_id),
1508 title,
1509 parent_chat_id,
1510 parent_message_id,
1511 })
1512}
1513
1514fn user_record_from_api_value(value: &Value) -> Option<UserRecord> {
1515 let user_id = api_i64_field(value, &["id", "userId", "user_id"])?;
1516 Some(UserRecord {
1517 user_id: InlineId::new(user_id),
1518 display_name: api_string_field(value, &["displayName", "display_name", "name"]),
1519 username: api_string_field(value, &["username"]),
1520 first_name: api_string_field(value, &["firstName", "first_name"]),
1521 last_name: api_string_field(value, &["lastName", "last_name"]),
1522 avatar_url: api_string_field(value, &["avatarUrl", "avatar_url"])
1523 .or_else(|| api_nested_string_field(value, "profilePhoto", &["cdnUrl", "cdn_url"]))
1524 .or_else(|| api_nested_string_field(value, "profile_photo", &["cdnUrl", "cdn_url"]))
1525 .or_else(|| api_nested_string_field(value, "botAvatar", &["cdnUrl", "cdn_url"]))
1526 .or_else(|| api_nested_string_field(value, "bot_avatar", &["cdnUrl", "cdn_url"])),
1527 is_bot: api_bool_field(value, &["isBot", "is_bot", "bot"]),
1528 })
1529}
1530
1531fn user_display_name_from_record(user: &UserRecord) -> Option<String> {
1532 user.display_name
1533 .clone()
1534 .or_else(|| {
1535 let first = user.first_name.as_deref().unwrap_or_default().trim();
1536 let last = user.last_name.as_deref().unwrap_or_default().trim();
1537 let full = [first, last]
1538 .into_iter()
1539 .filter(|part| !part.is_empty())
1540 .collect::<Vec<_>>()
1541 .join(" ");
1542 (!full.is_empty()).then_some(full)
1543 })
1544 .or_else(|| user.username.clone())
1545}
1546
1547fn api_i64_field(value: &Value, fields: &[&str]) -> Option<i64> {
1548 fields
1549 .iter()
1550 .find_map(|field| value.get(*field).and_then(value_as_i64))
1551}
1552
1553fn api_bool_field(value: &Value, fields: &[&str]) -> Option<bool> {
1554 fields
1555 .iter()
1556 .find_map(|field| value.get(*field).and_then(Value::as_bool))
1557}
1558
1559fn api_string_field(value: &Value, fields: &[&str]) -> Option<String> {
1560 fields
1561 .iter()
1562 .find_map(|field| value.get(*field).and_then(value_as_string))
1563}
1564
1565fn api_nested_string_field(value: &Value, field: &str, nested_fields: &[&str]) -> Option<String> {
1566 value
1567 .get(field)
1568 .and_then(|nested| api_string_field(nested, nested_fields))
1569}
1570
1571fn value_as_i64(value: &Value) -> Option<i64> {
1572 value
1573 .as_i64()
1574 .or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
1575 .or_else(|| value.as_str()?.trim().parse().ok())
1576}
1577
1578fn value_as_string(value: &Value) -> Option<String> {
1579 value
1580 .as_str()
1581 .map(str::trim)
1582 .filter(|value| !value.is_empty())
1583 .map(ToOwned::to_owned)
1584}
1585
1586fn trimmed_option(value: Option<String>) -> Option<String> {
1587 value
1588 .map(|value| value.trim().to_owned())
1589 .filter(|value| !value.is_empty())
1590}
1591
1592fn user_display_name_from_proto(user: &proto::User) -> Option<String> {
1593 if let Some(bot_name) = user
1594 .bot_avatar
1595 .as_ref()
1596 .and_then(|avatar| non_empty_option(Some(&avatar.display_name)))
1597 {
1598 return Some(bot_name);
1599 }
1600 let first = user.first_name.as_deref().unwrap_or_default().trim();
1601 let last = user.last_name.as_deref().unwrap_or_default().trim();
1602 let full = [first, last]
1603 .into_iter()
1604 .filter(|part| !part.is_empty())
1605 .collect::<Vec<_>>()
1606 .join(" ");
1607 if !full.is_empty() {
1608 return Some(full);
1609 }
1610 non_empty_option(user.username.as_deref())
1611}
1612
1613fn user_id_from_peer(peer: &proto::Peer) -> Option<i64> {
1614 match &peer.r#type {
1615 Some(proto::peer::Type::User(user)) => Some(user.user_id),
1616 _ => None,
1617 }
1618}
1619
1620fn non_empty_option(value: Option<&str>) -> Option<String> {
1621 value
1622 .map(str::trim)
1623 .filter(|value| !value.is_empty())
1624 .map(ToOwned::to_owned)
1625}
1626
1627fn realtime_error_to_backend(error: RealtimeError) -> BackendError {
1628 match error {
1629 RealtimeError::InvalidUrl { message, .. } => {
1630 BackendError::new(ClientErrorCategory::InvalidInput, message)
1631 }
1632 RealtimeError::Timeout { .. } => {
1633 BackendError::new(ClientErrorCategory::Timeout, error.to_string())
1634 }
1635 RealtimeError::ConnectionError { .. } | RealtimeError::RpcError { .. } => {
1636 BackendError::new(ClientErrorCategory::AuthExpired, error.to_string())
1637 }
1638 RealtimeError::ConnectionClosed | RealtimeError::WebSocket(_) => {
1639 BackendError::new(ClientErrorCategory::Network, error.to_string())
1640 }
1641 RealtimeError::InvalidHeaderValue { .. }
1642 | RealtimeError::Protocol(_)
1643 | RealtimeError::MissingResult
1644 | RealtimeError::UnexpectedResult { .. } => {
1645 BackendError::new(ClientErrorCategory::ProtocolMismatch, error.to_string())
1646 }
1647 _ => BackendError::new(ClientErrorCategory::Internal, error.to_string()),
1648 }
1649}
1650
1651fn api_error_to_backend(error: ApiError) -> BackendError {
1652 let rendered = error.to_string();
1653 match error {
1654 ApiError::InvalidBaseUrl { message, .. } | ApiError::InvalidInput { message } => {
1655 BackendError::new(ClientErrorCategory::InvalidInput, message)
1656 }
1657 ApiError::Status { status, .. } if status == 401 || status == 403 => {
1658 BackendError::new(ClientErrorCategory::AuthExpired, rendered)
1659 }
1660 ApiError::Api {
1661 status: Some(401) | Some(403),
1662 ..
1663 } => BackendError::new(ClientErrorCategory::AuthExpired, rendered),
1664 ApiError::Http(_) | ApiError::Status { .. } => {
1665 BackendError::new(ClientErrorCategory::Network, rendered)
1666 }
1667 ApiError::Json(_) => BackendError::new(ClientErrorCategory::ProtocolMismatch, rendered),
1668 ApiError::Io(_) | ApiError::Api { .. } => {
1669 BackendError::new(ClientErrorCategory::Internal, rendered)
1670 }
1671 _ => BackendError::new(ClientErrorCategory::Internal, rendered),
1672 }
1673}
1674
1675fn input_peer_for_client_peer(peer: PeerRef) -> proto::InputPeer {
1676 proto::InputPeer {
1677 r#type: Some(match peer {
1678 PeerRef::User { user_id } => proto::input_peer::Type::User(proto::InputPeerUser {
1679 user_id: user_id.get(),
1680 }),
1681 PeerRef::Chat { chat_id } => proto::input_peer::Type::Chat(proto::InputPeerChat {
1682 chat_id: chat_id.get(),
1683 }),
1684 PeerRef::Thread { thread_id } => proto::input_peer::Type::Chat(proto::InputPeerChat {
1685 chat_id: thread_id.get(),
1686 }),
1687 }),
1688 }
1689}
1690
1691fn input_peer_for_chat(chat_id: InlineId) -> proto::InputPeer {
1692 proto::InputPeer {
1693 r#type: Some(proto::input_peer::Type::Chat(proto::InputPeerChat {
1694 chat_id: chat_id.get(),
1695 })),
1696 }
1697}
1698
1699fn history_input_for_request(
1700 request: &HistoryRequest,
1701 fetch_limit: i32,
1702) -> proto::GetChatHistoryInput {
1703 if let Some(after) = request.after_message_id {
1704 return proto::GetChatHistoryInput {
1705 peer_id: Some(input_peer_for_chat(request.chat_id)),
1706 offset_id: None,
1707 limit: Some(fetch_limit),
1708 mode: Some(proto::GetChatHistoryMode::HistoryModeNewer as i32),
1709 anchor_id: None,
1710 before_id: None,
1711 after_id: Some(after.get()),
1712 before_limit: None,
1713 after_limit: None,
1714 include_anchor: None,
1715 };
1716 }
1717
1718 proto::GetChatHistoryInput {
1719 peer_id: Some(input_peer_for_chat(request.chat_id)),
1720 offset_id: request.before_message_id.map(InlineId::get),
1721 limit: Some(fetch_limit),
1722 mode: None,
1723 anchor_id: None,
1724 before_id: None,
1725 after_id: None,
1726 before_limit: None,
1727 after_limit: None,
1728 include_anchor: None,
1729 }
1730}
1731
1732fn chat_id_from_peer(peer: &proto::Peer) -> Option<InlineId> {
1733 match &peer.r#type {
1734 Some(proto::peer::Type::Chat(chat)) => Some(InlineId::new(chat.chat_id)),
1735 Some(proto::peer::Type::User(user)) => Some(InlineId::new(user.user_id)),
1736 None => None,
1737 }
1738}
1739
1740fn chat_id_for_peer(peer: PeerRef) -> InlineId {
1741 match peer {
1742 PeerRef::User { user_id } => user_id,
1743 PeerRef::Chat { chat_id } => chat_id,
1744 PeerRef::Thread { thread_id } => thread_id,
1745 }
1746}
1747
1748fn transaction_id_for_send(request: &SendTextRequest, random_id: RandomId) -> TransactionId {
1749 let stable = request
1750 .external_id
1751 .as_ref()
1752 .map(|external| stable_hash(&format!("{}:{}", external.source(), external.id())))
1753 .unwrap_or_else(|| random_id.get() as u64);
1754 TransactionId::try_new(format!("sdk-send-{stable:016x}"))
1755 .expect("generated transaction ID should be valid")
1756}
1757
1758fn random_id_for_request(request: &SendTextRequest) -> RandomId {
1759 let seed = request
1760 .external_id
1761 .as_ref()
1762 .map(|external| format!("{}:{}:{}", external.source(), external.id(), request.text))
1763 .unwrap_or_else(|| format!("{}:{}", now_seconds(), request.text));
1764 RandomId::new((stable_hash(&seed) & 0x7fff_ffff_ffff_ffff) as i64)
1765}
1766
1767fn transaction_id_for_upload(request: &UploadRequest, random_id: RandomId) -> TransactionId {
1768 let stable = request
1769 .external_id
1770 .as_ref()
1771 .map(|external| stable_hash(&format!("{}:{}", external.source(), external.id())))
1772 .unwrap_or_else(|| random_id.get() as u64);
1773 TransactionId::try_new(format!("sdk-upload-{stable:016x}"))
1774 .expect("generated transaction ID should be valid")
1775}
1776
1777fn random_id_for_upload_request(request: &UploadRequest, size_bytes: usize) -> RandomId {
1778 let seed = request
1779 .external_id
1780 .as_ref()
1781 .map(|external| {
1782 format!(
1783 "{}:{}:{}",
1784 external.source(),
1785 external.id(),
1786 request.caption.as_deref().unwrap_or_default()
1787 )
1788 })
1789 .unwrap_or_else(|| {
1790 format!(
1791 "{}:{}:{}:{}",
1792 now_seconds(),
1793 request.file_name.as_deref().unwrap_or_default(),
1794 request.caption.as_deref().unwrap_or_default(),
1795 size_bytes
1796 )
1797 });
1798 RandomId::new((stable_hash(&seed) & 0x7fff_ffff_ffff_ffff) as i64)
1799}
1800
1801fn upload_input_for_request(request: &UploadRequest, bytes: Vec<u8>) -> UploadFileBytesInput {
1802 let file_name = request
1803 .file_name
1804 .as_deref()
1805 .map(str::trim)
1806 .filter(|value| !value.is_empty())
1807 .map(ToOwned::to_owned)
1808 .unwrap_or_else(|| default_upload_file_name(request));
1809 let file_type = upload_file_type_for_request(request);
1810 let mut input = UploadFileBytesInput::new(bytes, file_name, file_type);
1811 if let Some(mime_type) = request
1812 .mime_type
1813 .as_deref()
1814 .map(str::trim)
1815 .filter(|value| !value.is_empty())
1816 {
1817 input = input.with_mime_type(mime_type);
1818 }
1819 if file_type == UploadFileType::Video
1820 && let Some(metadata) = upload_video_metadata_for_request(request)
1821 {
1822 input = input.with_video_metadata(metadata);
1823 }
1824 input
1825}
1826
1827fn upload_file_type_for_request(request: &UploadRequest) -> UploadFileType {
1828 match request.kind {
1829 MediaKind::Photo => UploadFileType::Photo,
1830 MediaKind::Video if upload_video_metadata_for_request(request).is_some() => {
1831 UploadFileType::Video
1832 }
1833 MediaKind::Video | MediaKind::Document | MediaKind::Voice => UploadFileType::Document,
1834 }
1835}
1836
1837fn upload_video_metadata_for_request(request: &UploadRequest) -> Option<UploadVideoMetadata> {
1838 let width = i32::try_from(request.width?)
1839 .ok()
1840 .filter(|value| *value > 0)?;
1841 let height = i32::try_from(request.height?)
1842 .ok()
1843 .filter(|value| *value > 0)?;
1844 let duration = duration_ms_to_seconds_i32(request.duration_ms?)?;
1845 Some(UploadVideoMetadata::new(width, height, duration))
1846}
1847
1848fn duration_ms_to_seconds_i32(duration_ms: u64) -> Option<i32> {
1849 if duration_ms == 0 {
1850 return None;
1851 }
1852 let seconds = duration_ms.saturating_add(999) / 1_000;
1853 i32::try_from(seconds).ok().filter(|value| *value > 0)
1854}
1855
1856fn default_upload_file_name(request: &UploadRequest) -> String {
1857 let extension = match request.kind {
1858 MediaKind::Photo => extension_for_mime(request.mime_type.as_deref(), "jpg"),
1859 MediaKind::Video => extension_for_mime(request.mime_type.as_deref(), "mp4"),
1860 MediaKind::Voice => extension_for_mime(request.mime_type.as_deref(), "ogg"),
1861 MediaKind::Document => "bin",
1862 };
1863 format!("inline-upload.{}", extension.trim_start_matches('.'))
1864}
1865
1866fn extension_for_mime(mime_type: Option<&str>, fallback: &'static str) -> &'static str {
1867 match mime_type
1868 .unwrap_or_default()
1869 .trim()
1870 .to_ascii_lowercase()
1871 .as_str()
1872 {
1873 "image/png" => "png",
1874 "image/gif" => "gif",
1875 "image/webp" => "webp",
1876 "video/webm" => "webm",
1877 "audio/mpeg" => "mp3",
1878 "audio/mp4" | "audio/aac" => "m4a",
1879 _ => fallback,
1880 }
1881}
1882
1883fn input_media_from_upload(upload: &UploadFileResult) -> BackendResult<proto::InputMedia> {
1884 if let Some(photo_id) = upload.photo_id {
1885 return Ok(proto::InputMedia {
1886 media: Some(proto::input_media::Media::Photo(proto::InputMediaPhoto {
1887 photo_id,
1888 })),
1889 });
1890 }
1891 if let Some(video_id) = upload.video_id {
1892 return Ok(proto::InputMedia {
1893 media: Some(proto::input_media::Media::Video(proto::InputMediaVideo {
1894 video_id,
1895 })),
1896 });
1897 }
1898 if let Some(document_id) = upload.document_id {
1899 return Ok(proto::InputMedia {
1900 media: Some(proto::input_media::Media::Document(
1901 proto::InputMediaDocument { document_id },
1902 )),
1903 });
1904 }
1905 Err(BackendError::new(
1906 ClientErrorCategory::ProtocolMismatch,
1907 "uploadFile response did not include a media id",
1908 ))
1909}
1910
1911fn stable_hash(value: &str) -> u64 {
1912 let mut hash = 0xcbf2_9ce4_8422_2325u64;
1913 for byte in value.as_bytes() {
1914 hash ^= u64::from(*byte);
1915 hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
1916 }
1917 hash
1918}
1919
1920#[derive(Debug)]
1921struct AppliedSendMessage {
1922 transaction: StoredTransaction,
1923 message: Option<MessageRecord>,
1924 chat_id: InlineId,
1925 message_id: Option<InlineId>,
1926}
1927
1928fn apply_send_message_updates(
1929 request: &SendTextRequest,
1930 identity: TransactionIdentity,
1931 fallback_chat_id: InlineId,
1932 result: proto::SendMessageResult,
1933) -> AppliedSendMessage {
1934 let mut final_message: Option<proto::Message> = None;
1935 let mut final_message_id: Option<InlineId> = None;
1936 for update in result.updates {
1937 match update.update {
1938 Some(proto::update::Update::NewMessage(update)) => {
1939 if let Some(message) = update.message {
1940 final_message_id = Some(InlineId::new(message.id));
1941 final_message = Some(message);
1942 }
1943 }
1944 Some(proto::update::Update::UpdateMessageId(update))
1945 if update.random_id == identity.random_id.get() =>
1946 {
1947 final_message_id = Some(InlineId::new(update.message_id));
1948 }
1949 _ => {}
1950 }
1951 }
1952
1953 let state = if final_message_id.is_some() {
1954 TransactionState::Completed
1955 } else {
1956 TransactionState::Acked
1957 };
1958 let mut identity = identity;
1959 if let Some(message_id) = final_message_id {
1960 identity = identity.with_final_message_id(message_id);
1961 }
1962 let message =
1963 final_message.map(|message| message_record_from_proto(message, request, identity.clone()));
1964 let chat_id = message
1965 .as_ref()
1966 .map(|message| message.chat_id)
1967 .unwrap_or(fallback_chat_id);
1968 let mut transaction = StoredTransaction::new(identity, state).with_chat_id(chat_id);
1969 if let Some(message_id) = final_message_id {
1970 transaction = transaction.with_message_id(message_id);
1971 }
1972
1973 AppliedSendMessage {
1974 transaction,
1975 message,
1976 chat_id,
1977 message_id: final_message_id,
1978 }
1979}
1980
1981fn apply_upload_message_updates(
1982 request: &UploadRequest,
1983 identity: TransactionIdentity,
1984 fallback_chat_id: InlineId,
1985 result: proto::SendMessageResult,
1986) -> AppliedSendMessage {
1987 let mut final_message: Option<proto::Message> = None;
1988 let mut final_message_id: Option<InlineId> = None;
1989 for update in result.updates {
1990 match update.update {
1991 Some(proto::update::Update::NewMessage(update)) => {
1992 if let Some(message) = update.message {
1993 final_message_id = Some(InlineId::new(message.id));
1994 final_message = Some(message);
1995 }
1996 }
1997 Some(proto::update::Update::UpdateMessageId(update))
1998 if update.random_id == identity.random_id.get() =>
1999 {
2000 final_message_id = Some(InlineId::new(update.message_id));
2001 }
2002 _ => {}
2003 }
2004 }
2005
2006 let state = if final_message_id.is_some() {
2007 TransactionState::Completed
2008 } else {
2009 TransactionState::Acked
2010 };
2011 let mut identity = identity;
2012 if let Some(message_id) = final_message_id {
2013 identity = identity.with_final_message_id(message_id);
2014 }
2015 let message = final_message.map(|message| {
2016 message_record_from_proto_message(
2017 message,
2018 Some(chat_id_for_peer(request.peer)),
2019 Some(identity.clone()),
2020 )
2021 });
2022 let chat_id = message
2023 .as_ref()
2024 .map(|message| message.chat_id)
2025 .unwrap_or(fallback_chat_id);
2026 let mut transaction = StoredTransaction::new(identity, state).with_chat_id(chat_id);
2027 if let Some(message_id) = final_message_id {
2028 transaction = transaction.with_message_id(message_id);
2029 }
2030
2031 AppliedSendMessage {
2032 transaction,
2033 message,
2034 chat_id,
2035 message_id: final_message_id,
2036 }
2037}
2038
2039fn message_record_from_proto(
2040 message: proto::Message,
2041 request: &SendTextRequest,
2042 transaction: TransactionIdentity,
2043) -> MessageRecord {
2044 message_record_from_proto_message(
2045 message,
2046 Some(chat_id_for_peer(request.peer)),
2047 Some(transaction),
2048 )
2049}
2050
2051fn message_record_from_proto_message(
2052 message: proto::Message,
2053 fallback_chat_id: Option<InlineId>,
2054 transaction: Option<TransactionIdentity>,
2055) -> MessageRecord {
2056 let chat_id = if message.chat_id != 0 {
2057 InlineId::new(message.chat_id)
2058 } else if let Some(peer) = &message.peer_id {
2059 chat_id_from_peer(peer).unwrap_or_else(|| fallback_chat_id.unwrap_or(InlineId::new(0)))
2060 } else {
2061 fallback_chat_id.unwrap_or(InlineId::new(0))
2062 };
2063 let content = content_from_proto_message(&message);
2064 MessageRecord {
2065 chat_id,
2066 message_id: InlineId::new(message.id),
2067 sender_id: InlineId::new(message.from_id),
2068 timestamp: message.date,
2069 is_outgoing: message.out,
2070 content,
2071 reply_to_message_id: message.reply_to_msg_id.map(InlineId::new),
2072 transaction,
2073 }
2074}
2075
2076fn content_from_proto_message(message: &proto::Message) -> MessageContent {
2077 match &message.media {
2078 None => match &message.message {
2079 Some(text) => MessageContent::Text { text: text.clone() },
2080 None => MessageContent::Unsupported {
2081 reason: "empty message content".to_owned(),
2082 },
2083 },
2084 Some(media) => media_content_from_proto(media, message.message.clone()),
2085 }
2086}
2087
2088fn media_content_from_proto(
2089 media: &proto::MessageMedia,
2090 caption: Option<String>,
2091) -> MessageContent {
2092 match &media.media {
2093 Some(proto::message_media::Media::Photo(photo)) => {
2094 if let Some(photo) = &photo.photo {
2095 let (url, size_bytes, width, height) = best_photo_size(photo);
2096 MessageContent::Media {
2097 kind: MediaKind::Photo,
2098 file_id: photo.id.to_string(),
2099 url,
2100 mime_type: photo_mime_type(photo),
2101 file_name: None,
2102 caption,
2103 size_bytes,
2104 width,
2105 height,
2106 duration_ms: None,
2107 }
2108 } else {
2109 empty_media_content(MediaKind::Photo, caption)
2110 }
2111 }
2112 Some(proto::message_media::Media::Video(video)) => {
2113 if let Some(video) = &video.video {
2114 MessageContent::Media {
2115 kind: MediaKind::Video,
2116 file_id: video.id.to_string(),
2117 url: video.cdn_url.clone(),
2118 mime_type: Some("video/mp4".to_owned()),
2119 file_name: None,
2120 caption,
2121 size_bytes: positive_u64(video.size),
2122 width: positive_u32(video.w),
2123 height: positive_u32(video.h),
2124 duration_ms: seconds_to_milliseconds(video.duration),
2125 }
2126 } else {
2127 empty_media_content(MediaKind::Video, caption)
2128 }
2129 }
2130 Some(proto::message_media::Media::Document(document)) => {
2131 if let Some(document) = &document.document {
2132 MessageContent::Media {
2133 kind: MediaKind::Document,
2134 file_id: document.id.to_string(),
2135 url: document.cdn_url.clone(),
2136 mime_type: non_empty_string(&document.mime_type),
2137 file_name: non_empty_string(&document.file_name),
2138 caption,
2139 size_bytes: positive_u64(document.size),
2140 width: None,
2141 height: None,
2142 duration_ms: None,
2143 }
2144 } else {
2145 empty_media_content(MediaKind::Document, caption)
2146 }
2147 }
2148 Some(proto::message_media::Media::Voice(voice)) => {
2149 if let Some(voice) = &voice.voice {
2150 MessageContent::Media {
2151 kind: MediaKind::Voice,
2152 file_id: voice.id.to_string(),
2153 url: voice.cdn_url.clone(),
2154 mime_type: non_empty_string(&voice.mime_type),
2155 file_name: None,
2156 caption,
2157 size_bytes: positive_u64(voice.size),
2158 width: None,
2159 height: None,
2160 duration_ms: seconds_to_milliseconds(voice.duration),
2161 }
2162 } else {
2163 empty_media_content(MediaKind::Voice, caption)
2164 }
2165 }
2166 Some(proto::message_media::Media::Nudge(_)) => MessageContent::Unsupported {
2167 reason: "nudge messages are not supported by inline-client yet".to_owned(),
2168 },
2169 None => MessageContent::Unsupported {
2170 reason: "empty media message".to_owned(),
2171 },
2172 }
2173}
2174
2175fn empty_media_content(kind: MediaKind, caption: Option<String>) -> MessageContent {
2176 MessageContent::Media {
2177 kind,
2178 file_id: String::new(),
2179 url: None,
2180 mime_type: None,
2181 file_name: None,
2182 caption,
2183 size_bytes: None,
2184 width: None,
2185 height: None,
2186 duration_ms: None,
2187 }
2188}
2189
2190fn best_photo_size(
2191 photo: &proto::Photo,
2192) -> (Option<String>, Option<u64>, Option<u32>, Option<u32>) {
2193 let mut best: Option<(&proto::PhotoSize, i64)> = None;
2194 for size in &photo.sizes {
2195 if size.cdn_url.is_none() {
2196 continue;
2197 }
2198 let area = i64::from(size.w) * i64::from(size.h);
2199 if best.is_none_or(|(_, best_area)| area > best_area) {
2200 best = Some((size, area));
2201 }
2202 }
2203 if let Some((size, _)) = best {
2204 return (
2205 size.cdn_url.clone(),
2206 positive_u64(size.size),
2207 positive_u32(size.w),
2208 positive_u32(size.h),
2209 );
2210 }
2211 (None, None, None, None)
2212}
2213
2214fn photo_mime_type(photo: &proto::Photo) -> Option<String> {
2215 match photo.format {
2216 1 => Some("image/jpeg".to_owned()),
2217 2 => Some("image/png".to_owned()),
2218 _ => None,
2219 }
2220}
2221
2222fn positive_u64(value: i32) -> Option<u64> {
2223 u64::try_from(value).ok().filter(|value| *value > 0)
2224}
2225
2226fn positive_u32(value: i32) -> Option<u32> {
2227 u32::try_from(value).ok().filter(|value| *value > 0)
2228}
2229
2230fn seconds_to_milliseconds(seconds: i32) -> Option<u64> {
2231 positive_u64(seconds).map(|seconds| seconds.saturating_mul(1_000))
2232}
2233
2234fn non_empty_string(value: &str) -> Option<String> {
2235 (!value.is_empty()).then(|| value.to_owned())
2236}
2237
2238fn outcome_from_stored_transaction(
2239 transaction: StoredTransaction,
2240 fallback_chat_id: InlineId,
2241) -> SendTextOutcome {
2242 let message_id = transaction
2243 .message_id
2244 .or(transaction.identity.final_message_id);
2245 SendTextOutcome::with_state(
2246 MessageMutation {
2247 transaction: transaction.identity.clone(),
2248 message_id,
2249 },
2250 transaction.chat_id.unwrap_or(fallback_chat_id),
2251 message_id,
2252 None,
2253 transaction.state,
2254 transaction.failure,
2255 )
2256}
2257
2258fn now_seconds() -> i64 {
2259 SystemTime::now()
2260 .duration_since(UNIX_EPOCH)
2261 .map(|duration| duration.as_secs() as i64)
2262 .unwrap_or_default()
2263}
2264
2265fn redacted_url_for_debug(url: &str) -> String {
2266 let without_fragment = url.split('#').next().unwrap_or(url);
2267 let without_query = without_fragment
2268 .split('?')
2269 .next()
2270 .unwrap_or(without_fragment);
2271 match without_query.split_once("://") {
2272 Some((scheme, rest)) => {
2273 let host_and_path = rest.rsplit_once('@').map_or(rest, |(_, tail)| tail);
2274 format!("{scheme}://{host_and_path}")
2275 }
2276 None => without_query.to_owned(),
2277 }
2278}
2279
2280#[cfg(test)]
2281mod tests {
2282 use crate::{
2283 AuthCredential, AuthToken, ClientBackend, DialogRecord, DialogsRequest, ExternalId,
2284 FakeRealtimeConnector, HistoryRequest, InlineId, RandomId, SendTextRequest,
2285 StoredTransaction,
2286 };
2287
2288 use super::*;
2289
2290 fn connect_request() -> ConnectRequest {
2291 ConnectRequest::new(AuthCredential::AccessToken {
2292 token: AuthToken::try_new("secret-token").unwrap(),
2293 })
2294 .with_account_namespace("team")
2295 }
2296
2297 #[test]
2298 fn sdk_backend_debug_redacts_realtime_url_credentials() {
2299 let backend = SdkBackend::builder()
2300 .realtime_url("wss://user:secret@api.inline.chat/realtime?token=secret#frag")
2301 .build()
2302 .unwrap();
2303
2304 let rendered = format!("{backend:?}");
2305 assert!(rendered.contains("wss://api.inline.chat/realtime"));
2306 assert!(!rendered.contains("secret"));
2307 assert!(!rendered.contains("token="));
2308 }
2309
2310 #[test]
2311 fn sdk_backend_defaults_use_production_endpoints() {
2312 let backend = SdkBackend::builder().build().unwrap();
2313
2314 assert_eq!(
2315 backend.api_client().base_url(),
2316 "https://api.inline.chat/v1"
2317 );
2318 assert_eq!(backend.realtime_url(), "wss://api.inline.chat/realtime");
2319 }
2320
2321 #[tokio::test]
2322 async fn sdk_backend_connect_persists_session() {
2323 let store = InMemoryStore::new();
2324 let backend = SdkBackend::builder().store(store.clone()).build().unwrap();
2325
2326 let status = backend.connect(connect_request()).await.unwrap();
2327
2328 assert_eq!(status.status, crate::ClientStatus::Connected);
2329 let session = store.load_session().await.unwrap().unwrap();
2330 assert_eq!(session.account_namespace.as_deref(), Some("team"));
2331 let rendered = format!("{session:?}");
2332 assert!(!rendered.contains("secret-token"));
2333 }
2334
2335 #[tokio::test]
2336 async fn sdk_backend_connect_can_perform_realtime_handshake() {
2337 let store = InMemoryStore::new();
2338 let realtime = FakeRealtimeConnector::new();
2339 let backend = SdkBackend::builder()
2340 .store(store.clone())
2341 .realtime_connector(realtime.clone())
2342 .realtime_url("wss://api.inline.chat/realtime")
2343 .build()
2344 .unwrap();
2345
2346 assert!(backend.realtime_handshake_enabled());
2347 backend.connect(connect_request()).await.unwrap();
2348
2349 let attempts = realtime.attempts();
2350 assert_eq!(attempts.len(), 1);
2351 assert_eq!(attempts[0].realtime_url, "wss://api.inline.chat/realtime");
2352 assert!(store.load_session().await.unwrap().is_some());
2353 }
2354
2355 #[tokio::test]
2356 async fn sdk_backend_does_not_persist_session_when_realtime_handshake_fails() {
2357 let store = InMemoryStore::new();
2358 let backend = SdkBackend::builder()
2359 .store(store.clone())
2360 .realtime_connector(FakeRealtimeConnector::failing(BackendError::new(
2361 ClientErrorCategory::Network,
2362 "offline",
2363 )))
2364 .build()
2365 .unwrap();
2366
2367 let err = backend.connect(connect_request()).await.unwrap_err();
2368
2369 assert_eq!(err.category, ClientErrorCategory::Network);
2370 assert!(store.load_session().await.unwrap().is_none());
2371 }
2372
2373 #[tokio::test]
2374 async fn sdk_backend_resume_without_session_reports_auth_required() {
2375 let backend = SdkBackend::builder().build().unwrap();
2376
2377 let status = backend.resume_session().await.unwrap();
2378
2379 assert_eq!(status.status, crate::ClientStatus::AuthRequired);
2380 }
2381
2382 #[tokio::test]
2383 async fn sdk_backend_resume_uses_stored_session() {
2384 let store = InMemoryStore::new();
2385 store.save_session(connect_session()).await.unwrap();
2386 let realtime = FakeRealtimeConnector::new();
2387 let backend = SdkBackend::builder()
2388 .store(store)
2389 .realtime_connector(realtime.clone())
2390 .realtime_url("wss://api.inline.chat/realtime")
2391 .build()
2392 .unwrap();
2393
2394 let status = backend.resume_session().await.unwrap();
2395
2396 assert_eq!(status.status, crate::ClientStatus::Connected);
2397 let attempts = realtime.attempts();
2398 assert_eq!(attempts.len(), 1);
2399 assert_eq!(attempts[0].realtime_url, "wss://api.inline.chat/realtime");
2400 }
2401
2402 #[tokio::test]
2403 async fn sdk_backend_reads_dialogs_from_store_after_connect() {
2404 let store = InMemoryStore::new();
2405 store.upsert_dialog(DialogRecord {
2406 chat_id: InlineId::new(9),
2407 title: Some("general".to_owned()),
2408 last_message_id: None,
2409 synced_through_message_id: None,
2410 unread_count: Some(0),
2411 });
2412 let backend = SdkBackend::builder()
2413 .store(store)
2414 .realtime_url("not-a-websocket-url")
2415 .build()
2416 .unwrap();
2417
2418 backend.connect(connect_request()).await.unwrap();
2419 let dialogs = backend.dialogs(DialogsRequest::default()).await.unwrap();
2420
2421 assert_eq!(dialogs.dialogs.len(), 1);
2422 assert_eq!(dialogs.dialogs[0].chat_id, InlineId::new(9));
2423 }
2424
2425 #[test]
2426 fn dialogs_page_from_get_chats_uses_user_records() {
2427 let result = proto::GetChatsResult {
2428 dialogs: vec![proto::Dialog {
2429 peer: Some(proto::Peer {
2430 r#type: Some(proto::peer::Type::User(proto::PeerUser { user_id: 42 })),
2431 }),
2432 chat_id: Some(7),
2433 unread_count: Some(3),
2434 ..Default::default()
2435 }],
2436 chats: vec![proto::Chat {
2437 id: 7,
2438 title: "Direct chat fallback".to_owned(),
2439 last_msg_id: Some(99),
2440 emoji: Some("*".to_owned()),
2441 peer_id: Some(proto::Peer {
2442 r#type: Some(proto::peer::Type::User(proto::PeerUser { user_id: 42 })),
2443 }),
2444 ..Default::default()
2445 }],
2446 users: vec![proto::User {
2447 id: 42,
2448 first_name: Some("Ada".to_owned()),
2449 last_name: Some("Lovelace".to_owned()),
2450 username: Some("ada".to_owned()),
2451 profile_photo: Some(proto::UserProfilePhoto {
2452 cdn_url: Some("https://cdn.inline.test/ada.jpg".to_owned()),
2453 ..Default::default()
2454 }),
2455 bot: Some(false),
2456 ..Default::default()
2457 }],
2458 ..Default::default()
2459 };
2460
2461 let page = dialogs_page_from_get_chats(
2462 &result,
2463 DialogsRequest {
2464 limit: Some(10),
2465 cursor: None,
2466 },
2467 )
2468 .unwrap();
2469
2470 assert_eq!(page.dialogs.len(), 1);
2471 assert_eq!(page.dialogs[0].chat_id, InlineId::new(7));
2472 assert_eq!(page.dialogs[0].title.as_deref(), Some("* Ada Lovelace"));
2473 assert_eq!(page.dialogs[0].last_message_id, Some(InlineId::new(99)));
2474 assert_eq!(page.dialogs[0].unread_count, Some(3));
2475 assert_eq!(page.users.len(), 1);
2476 assert_eq!(page.users[0].user_id, InlineId::new(42));
2477 assert_eq!(page.users[0].display_name.as_deref(), Some("Ada Lovelace"));
2478 assert_eq!(
2479 page.users[0].avatar_url.as_deref(),
2480 Some("https://cdn.inline.test/ada.jpg")
2481 );
2482 assert_eq!(page.users[0].is_bot, Some(false));
2483 }
2484
2485 #[test]
2486 fn chat_participants_page_uses_direct_participants() {
2487 let page = chat_participants_page_from_proto(proto::GetChatParticipantsResult {
2488 participants: vec![proto::ChatParticipant {
2489 user_id: 10,
2490 date: 100,
2491 }],
2492 users: vec![proto::User {
2493 id: 10,
2494 first_name: Some("Ada".to_owned()),
2495 ..Default::default()
2496 }],
2497 });
2498
2499 assert_eq!(page.participants.len(), 1);
2500 assert_eq!(page.participants[0].user_id, InlineId::new(10));
2501 assert_eq!(page.participants[0].date, Some(100));
2502 assert_eq!(page.users.len(), 1);
2503 }
2504
2505 #[tokio::test]
2506 async fn sdk_backend_requires_session_for_history() {
2507 let backend = SdkBackend::builder().build().unwrap();
2508
2509 let err = backend
2510 .history(HistoryRequest {
2511 chat_id: InlineId::new(1),
2512 limit: Some(10),
2513 before_message_id: None,
2514 after_message_id: None,
2515 })
2516 .await
2517 .expect_err("history should require connect");
2518
2519 assert_eq!(err.category, ClientErrorCategory::AuthRequired);
2520 }
2521
2522 #[tokio::test]
2523 async fn sdk_backend_send_text_requires_session() {
2524 let backend = SdkBackend::builder().build().unwrap();
2525
2526 let err = backend
2527 .send_text(SendTextRequest::new(
2528 crate::PeerRef::Chat {
2529 chat_id: InlineId::new(1),
2530 },
2531 "hello",
2532 ))
2533 .await
2534 .expect_err("send_text should require connect");
2535
2536 assert_eq!(err.category, ClientErrorCategory::AuthRequired);
2537 }
2538
2539 #[tokio::test]
2540 async fn sdk_backend_send_text_rejects_empty_text_before_network() {
2541 let backend = SdkBackend::builder().build().unwrap();
2542
2543 let err = backend
2544 .send_text(SendTextRequest::new(
2545 crate::PeerRef::Chat {
2546 chat_id: InlineId::new(1),
2547 },
2548 " ",
2549 ))
2550 .await
2551 .expect_err("empty message should fail before auth or network");
2552
2553 assert_eq!(err.category, ClientErrorCategory::InvalidInput);
2554 }
2555
2556 #[tokio::test]
2557 async fn sdk_backend_send_text_returns_existing_transaction_without_network() {
2558 let store = InMemoryStore::new();
2559 let request = SendTextRequest {
2560 peer: crate::PeerRef::Chat {
2561 chat_id: InlineId::new(7),
2562 },
2563 text: "hello".to_owned(),
2564 external_id: Some(ExternalId::try_new("host-event", "event-1").unwrap()),
2565 random_id: Some(RandomId::new(99)),
2566 reply_to_message_id: None,
2567 };
2568 let transaction_id = transaction_id_for_send(&request, request.random_id.unwrap());
2569 let identity = TransactionIdentity::new(
2570 transaction_id.clone(),
2571 request.external_id.clone(),
2572 request.random_id.unwrap(),
2573 )
2574 .with_final_message_id(InlineId::new(11));
2575 store.save_session(connect_session()).await.unwrap();
2576 store
2577 .record_transaction(
2578 StoredTransaction::new(identity, TransactionState::Completed)
2579 .with_chat_id(InlineId::new(7))
2580 .with_message_id(InlineId::new(11)),
2581 )
2582 .await
2583 .unwrap();
2584 let backend = SdkBackend::builder().store(store).build().unwrap();
2585
2586 let outcome = backend.send_text(request).await.unwrap();
2587
2588 assert_eq!(outcome.message_id, Some(InlineId::new(11)));
2589 assert_eq!(outcome.state, TransactionState::Completed);
2590 assert_eq!(outcome.mutation.transaction.transaction_id, transaction_id);
2591 }
2592
2593 #[test]
2594 fn media_content_from_proto_preserves_document_descriptor() {
2595 let content = media_content_from_proto(
2596 &proto::MessageMedia {
2597 media: Some(proto::message_media::Media::Document(
2598 proto::MessageDocument {
2599 document: Some(proto::Document {
2600 id: 55,
2601 file_name: "report.pdf".to_owned(),
2602 mime_type: "application/pdf".to_owned(),
2603 size: 12_345,
2604 cdn_url: Some("https://cdn.inline.test/report.pdf".to_owned()),
2605 ..Default::default()
2606 }),
2607 },
2608 )),
2609 },
2610 Some("quarterly report".to_owned()),
2611 );
2612
2613 match content {
2614 MessageContent::Media {
2615 kind,
2616 file_id,
2617 url,
2618 mime_type,
2619 file_name,
2620 caption,
2621 size_bytes,
2622 width,
2623 height,
2624 duration_ms,
2625 } => {
2626 assert_eq!(kind, MediaKind::Document);
2627 assert_eq!(file_id, "55");
2628 assert_eq!(url.as_deref(), Some("https://cdn.inline.test/report.pdf"));
2629 assert_eq!(mime_type.as_deref(), Some("application/pdf"));
2630 assert_eq!(file_name.as_deref(), Some("report.pdf"));
2631 assert_eq!(caption.as_deref(), Some("quarterly report"));
2632 assert_eq!(size_bytes, Some(12_345));
2633 assert_eq!(width, None);
2634 assert_eq!(height, None);
2635 assert_eq!(duration_ms, None);
2636 }
2637 other => panic!("expected media content, got {other:?}"),
2638 }
2639 }
2640
2641 #[test]
2642 fn media_content_from_proto_picks_largest_photo_descriptor() {
2643 let content = media_content_from_proto(
2644 &proto::MessageMedia {
2645 media: Some(proto::message_media::Media::Photo(proto::MessagePhoto {
2646 photo: Some(proto::Photo {
2647 id: 9,
2648 format: 2,
2649 sizes: vec![
2650 proto::PhotoSize {
2651 w: 20,
2652 h: 20,
2653 size: 100,
2654 cdn_url: Some("https://cdn.inline.test/small.png".to_owned()),
2655 ..Default::default()
2656 },
2657 proto::PhotoSize {
2658 w: 200,
2659 h: 200,
2660 size: 500,
2661 cdn_url: None,
2662 ..Default::default()
2663 },
2664 proto::PhotoSize {
2665 w: 50,
2666 h: 50,
2667 size: 200,
2668 cdn_url: Some("https://cdn.inline.test/large.png".to_owned()),
2669 ..Default::default()
2670 },
2671 ],
2672 ..Default::default()
2673 }),
2674 })),
2675 },
2676 None,
2677 );
2678
2679 match content {
2680 MessageContent::Media {
2681 kind,
2682 file_id,
2683 url,
2684 mime_type,
2685 size_bytes,
2686 width,
2687 height,
2688 ..
2689 } => {
2690 assert_eq!(kind, MediaKind::Photo);
2691 assert_eq!(file_id, "9");
2692 assert_eq!(url.as_deref(), Some("https://cdn.inline.test/large.png"));
2693 assert_eq!(mime_type.as_deref(), Some("image/png"));
2694 assert_eq!(size_bytes, Some(200));
2695 assert_eq!(width, Some(50));
2696 assert_eq!(height, Some(50));
2697 }
2698 other => panic!("expected media content, got {other:?}"),
2699 }
2700 }
2701
2702 #[test]
2703 fn upload_input_for_video_without_complete_metadata_falls_back_to_document() {
2704 let request = UploadRequest {
2705 peer: crate::PeerRef::Chat {
2706 chat_id: InlineId::new(7),
2707 },
2708 kind: MediaKind::Video,
2709 file_name: Some("clip.mp4".to_owned()),
2710 mime_type: Some("video/mp4".to_owned()),
2711 size_bytes: Some(4),
2712 caption: None,
2713 width: Some(640),
2714 height: None,
2715 duration_ms: Some(1_500),
2716 external_id: None,
2717 random_id: None,
2718 reply_to_message_id: None,
2719 };
2720
2721 let input = upload_input_for_request(&request, vec![1, 2, 3, 4]);
2722
2723 assert_eq!(input.file_type, UploadFileType::Document);
2724 assert_eq!(input.file_name, "clip.mp4");
2725 assert_eq!(input.mime_type.as_deref(), Some("video/mp4"));
2726 assert_eq!(input.video_metadata, None);
2727 }
2728
2729 #[test]
2730 fn upload_input_for_video_with_complete_metadata_uses_video() {
2731 let request = UploadRequest {
2732 peer: crate::PeerRef::Chat {
2733 chat_id: InlineId::new(7),
2734 },
2735 kind: MediaKind::Video,
2736 file_name: Some("clip.mp4".to_owned()),
2737 mime_type: Some("video/mp4".to_owned()),
2738 size_bytes: Some(4),
2739 caption: None,
2740 width: Some(640),
2741 height: Some(480),
2742 duration_ms: Some(1_500),
2743 external_id: None,
2744 random_id: None,
2745 reply_to_message_id: None,
2746 };
2747
2748 let input = upload_input_for_request(&request, vec![1, 2, 3, 4]);
2749
2750 assert_eq!(input.file_type, UploadFileType::Video);
2751 assert_eq!(
2752 input.video_metadata,
2753 Some(UploadVideoMetadata::new(640, 480, 2))
2754 );
2755 }
2756
2757 #[test]
2758 fn history_input_for_after_message_id_uses_newer_mode() {
2759 let input = history_input_for_request(
2760 &HistoryRequest {
2761 chat_id: InlineId::new(7),
2762 limit: Some(20),
2763 before_message_id: None,
2764 after_message_id: Some(InlineId::new(10)),
2765 },
2766 21,
2767 );
2768
2769 assert_eq!(input.after_id, Some(10));
2770 assert_eq!(
2771 input.mode,
2772 Some(proto::GetChatHistoryMode::HistoryModeNewer as i32)
2773 );
2774 assert_eq!(input.offset_id, None);
2775 assert_eq!(input.limit, Some(21));
2776 }
2777
2778 #[test]
2779 fn apply_send_message_updates_extracts_final_message() {
2780 let request = SendTextRequest {
2781 peer: crate::PeerRef::Chat {
2782 chat_id: InlineId::new(7),
2783 },
2784 text: "hello".to_owned(),
2785 external_id: Some(ExternalId::try_new("host-event", "event-1").unwrap()),
2786 random_id: Some(RandomId::new(99)),
2787 reply_to_message_id: None,
2788 };
2789 let identity = TransactionIdentity::new(
2790 TransactionId::try_new("txn").unwrap(),
2791 request.external_id.clone(),
2792 request.random_id.unwrap(),
2793 );
2794 let result = proto::SendMessageResult {
2795 updates: vec![proto::Update {
2796 seq: Some(1),
2797 date: Some(1),
2798 update: Some(proto::update::Update::NewMessage(proto::UpdateNewMessage {
2799 message: Some(proto::Message {
2800 id: 11,
2801 from_id: 2,
2802 peer_id: None,
2803 chat_id: 7,
2804 message: Some("hello".to_owned()),
2805 out: true,
2806 date: 123,
2807 mentioned: None,
2808 reply_to_msg_id: None,
2809 media: None,
2810 edit_date: None,
2811 grouped_id: None,
2812 attachments: None,
2813 reactions: None,
2814 is_sticker: None,
2815 has_link: None,
2816 entities: None,
2817 send_mode: None,
2818 fwd_from: None,
2819 replies: None,
2820 actions: None,
2821 rev: None,
2822 }),
2823 })),
2824 }],
2825 };
2826
2827 let applied = apply_send_message_updates(&request, identity, InlineId::new(7), result);
2828
2829 assert_eq!(applied.message_id, Some(InlineId::new(11)));
2830 assert_eq!(applied.transaction.state, TransactionState::Completed);
2831 assert_eq!(applied.message.unwrap().message_id, InlineId::new(11));
2832 }
2833
2834 fn connect_session() -> StoredSession {
2835 StoredSession {
2836 auth: AuthCredential::AccessToken {
2837 token: AuthToken::try_new("secret-token").unwrap(),
2838 },
2839 account_namespace: Some("team".to_owned()),
2840 }
2841 }
2842}