inline_client/
backend.rs

1//! Backend boundary for client runtime operations.
2//!
3//! The runtime talks to this trait instead of directly embedding transport,
4//! sync, or storage behavior. The in-memory backend is useful for client and
5//! bridge-adapter tests; the production backend will compose the real Inline SDK,
6//! realtime transport, store, sync engine, and transaction manager behind the
7//! same shape.
8
9use std::{
10    collections::{HashMap, VecDeque},
11    fmt,
12    sync::{Arc, Mutex},
13    time::Duration,
14};
15
16use futures_util::future::BoxFuture;
17
18use crate::{
19    AccountStateSnapshot, AddChatParticipantRequest, AuthStartRequest, AuthStartResult,
20    AuthVerifyRequest, AuthVerifyResult, ChatParticipantRecord, ChatParticipantsPage,
21    ChatParticipantsRequest, ChatStateSnapshot, ClientErrorCategory, ClientEvent, ClientFailure,
22    ClientStatus, ClientStatusSnapshot, ConnectRequest, CreateDmRequest, CreateReplyThreadRequest,
23    CreateThreadRequest, CreatedChat, DeleteChatRequest, DeleteMessageRequest, DialogRecord,
24    DialogsPage, DialogsRequest, EditMessageRequest, HistoryPage, HistoryRequest, InlineId,
25    MessageContent, MessageMutation, MessageRecord, RandomId, ReactRequest, ReadRequest,
26    RemoveChatParticipantRequest, SendTextRequest, SetMarkedUnreadRequest, TransactionEvent,
27    TransactionId, TransactionIdentity, TransactionState, TypingRequest, UpdateChatInfoRequest,
28    UpdateDialogNotificationsRequest, UploadRequest,
29};
30
31/// Result type returned by client backends.
32pub type BackendResult<T> = Result<T, BackendError>;
33
34/// Redacted backend error.
35#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
36#[error("{category:?}: {message}")]
37#[non_exhaustive]
38pub struct BackendError {
39    /// Stable error category.
40    pub category: ClientErrorCategory,
41    /// Redacted message suitable for hosts.
42    pub message: String,
43    /// Server-requested retry delay, when the failure is rate limited.
44    pub retry_after_seconds: Option<u64>,
45}
46
47impl BackendError {
48    /// Creates a backend error.
49    pub fn new(category: ClientErrorCategory, message: impl Into<String>) -> Self {
50        Self {
51            category,
52            message: message.into(),
53            retry_after_seconds: None,
54        }
55    }
56
57    /// Attaches a server-requested retry delay.
58    pub fn with_retry_after_seconds(mut self, seconds: u64) -> Self {
59        self.retry_after_seconds = (seconds > 0).then_some(seconds);
60        self
61    }
62}
63
64impl From<BackendError> for ClientFailure {
65    fn from(error: BackendError) -> Self {
66        Self::new(error.category, error.message)
67    }
68}
69
70pub(crate) fn retry_after_seconds_from_message(message: &str) -> Option<u64> {
71    let normalized = message.to_ascii_lowercase();
72    for marker in ["retry after ", "retry_after=", "flood_wait_"] {
73        let Some((_, suffix)) = normalized.split_once(marker) else {
74            continue;
75        };
76        let Ok(seconds) = suffix
77            .chars()
78            .take_while(char::is_ascii_digit)
79            .collect::<String>()
80            .parse::<u64>()
81        else {
82            continue;
83        };
84        if seconds > 0 {
85            return Some(seconds);
86        }
87    }
88    None
89}
90
91/// Outcome from a text-send operation.
92#[derive(Clone, Debug, PartialEq, Eq)]
93pub struct SendTextOutcome {
94    /// Client mutation acknowledgement.
95    pub mutation: MessageMutation,
96    /// Chat containing the message.
97    pub chat_id: InlineId,
98    /// Final message ID, when known.
99    pub message_id: Option<InlineId>,
100    /// Message record applied to the client store, when known.
101    pub message: Option<MessageRecord>,
102    /// Durable transaction state after the send attempt.
103    pub state: TransactionState,
104    /// Redacted failure, when the transaction failed.
105    pub failure: Option<ClientFailure>,
106}
107
108/// Outcome from a side-effect operation.
109#[derive(Clone, Debug, Default, PartialEq, Eq)]
110pub struct OperationOutcome {
111    /// Committed client events caused by the operation.
112    pub events: Vec<ClientEvent>,
113}
114
115impl OperationOutcome {
116    /// Creates an outcome with no committed events.
117    pub fn empty() -> Self {
118        Self::default()
119    }
120
121    /// Creates an outcome with committed events.
122    pub fn with_events(events: Vec<ClientEvent>) -> Self {
123        Self { events }
124    }
125}
126
127impl SendTextOutcome {
128    /// Creates a text-send outcome.
129    pub fn new(mutation: MessageMutation, chat_id: InlineId, message_id: Option<InlineId>) -> Self {
130        Self::with_state(
131            mutation,
132            chat_id,
133            message_id,
134            None,
135            TransactionState::Completed,
136            None,
137        )
138    }
139
140    /// Creates a text-send outcome with an explicit transaction state.
141    pub fn with_state(
142        mutation: MessageMutation,
143        chat_id: InlineId,
144        message_id: Option<InlineId>,
145        message: Option<MessageRecord>,
146        state: TransactionState,
147        failure: Option<ClientFailure>,
148    ) -> Self {
149        Self {
150            mutation,
151            chat_id,
152            message_id,
153            message,
154            state,
155            failure,
156        }
157    }
158
159    /// Returns a transaction event for this send outcome.
160    pub fn transaction_event(&self) -> TransactionEvent {
161        TransactionEvent {
162            identity: self.mutation.transaction.clone(),
163            state: self.state,
164            failure: self.failure.clone(),
165        }
166    }
167}
168
169/// Async backend boundary for the client runner.
170pub trait ClientBackend: fmt::Debug + Send + Sync + 'static {
171    /// Sends an Inline login code.
172    fn auth_start(
173        &self,
174        request: AuthStartRequest,
175    ) -> BoxFuture<'static, BackendResult<AuthStartResult>>;
176
177    /// Verifies an Inline login code and persists the resulting session.
178    fn auth_verify(
179        &self,
180        request: AuthVerifyRequest,
181    ) -> BoxFuture<'static, BackendResult<AuthVerifyResult>>;
182
183    /// Resumes a previously stored session, if available.
184    fn resume_session(&self) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>>;
185
186    /// Connects or reconnects the client.
187    fn connect(
188        &self,
189        request: ConnectRequest,
190    ) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>>;
191
192    /// Logs out the current account.
193    fn logout(&self) -> BoxFuture<'static, BackendResult<()>>;
194
195    /// Lists dialogs.
196    fn dialogs(&self, request: DialogsRequest) -> BoxFuture<'static, BackendResult<DialogsPage>>;
197
198    /// Lists only durable cached dialogs without a network request.
199    fn cached_dialogs(
200        &self,
201        request: DialogsRequest,
202    ) -> BoxFuture<'static, BackendResult<DialogsPage>>;
203
204    /// Loads account-level durable state used for consumer reconciliation.
205    fn account_state(&self) -> BoxFuture<'static, BackendResult<AccountStateSnapshot>>;
206
207    /// Loads durable state for one chat used for consumer reconciliation.
208    fn chat_state(&self, chat_id: InlineId)
209    -> BoxFuture<'static, BackendResult<ChatStateSnapshot>>;
210
211    /// Fetches chat history.
212    fn history(&self, request: HistoryRequest) -> BoxFuture<'static, BackendResult<HistoryPage>>;
213
214    /// Fetches only the client's durable cached history without making a
215    /// network request. This is intended for offline consumers and state
216    /// reconciliation; ordinary callers should prefer [`Self::history`].
217    fn cached_history(
218        &self,
219        request: HistoryRequest,
220    ) -> BoxFuture<'static, BackendResult<HistoryPage>>;
221
222    /// Fetches chat participants.
223    fn chat_participants(
224        &self,
225        request: ChatParticipantsRequest,
226    ) -> BoxFuture<'static, BackendResult<ChatParticipantsPage>>;
227
228    /// Adds a user to a chat.
229    fn add_chat_participant(
230        &self,
231        request: AddChatParticipantRequest,
232    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
233
234    /// Removes a user from a chat.
235    fn remove_chat_participant(
236        &self,
237        request: RemoveChatParticipantRequest,
238    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
239
240    /// Updates mutable chat metadata.
241    fn update_chat_info(
242        &self,
243        request: UpdateChatInfoRequest,
244    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
245
246    /// Deletes a chat when permitted by the service.
247    fn delete_chat(
248        &self,
249        request: DeleteChatRequest,
250    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
251
252    /// Creates or opens a direct message chat.
253    fn create_dm(&self, request: CreateDmRequest)
254    -> BoxFuture<'static, BackendResult<CreatedChat>>;
255
256    /// Creates a regular Inline thread chat.
257    fn create_thread(
258        &self,
259        request: CreateThreadRequest,
260    ) -> BoxFuture<'static, BackendResult<CreatedChat>>;
261
262    /// Creates a child/reply Inline thread chat.
263    fn create_reply_thread(
264        &self,
265        request: CreateReplyThreadRequest,
266    ) -> BoxFuture<'static, BackendResult<CreatedChat>>;
267
268    /// Sends a text message.
269    fn send_text(
270        &self,
271        request: SendTextRequest,
272    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>>;
273
274    /// Uploads and sends a media message.
275    fn send_media(
276        &self,
277        request: UploadRequest,
278        bytes: Vec<u8>,
279    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>>;
280
281    /// Edits a text message.
282    fn edit_message(
283        &self,
284        request: EditMessageRequest,
285    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
286
287    /// Deletes or unsends a message.
288    fn delete_message(
289        &self,
290        request: DeleteMessageRequest,
291    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
292
293    /// Adds or removes a reaction.
294    fn react(&self, request: ReactRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
295
296    /// Marks messages read.
297    fn read(&self, request: ReadRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
298
299    /// Sets the explicit marked-unread state for a chat.
300    fn set_marked_unread(
301        &self,
302        request: SetMarkedUnreadRequest,
303    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
304
305    /// Sets or clears a per-dialog notification override.
306    fn update_dialog_notifications(
307        &self,
308        request: UpdateDialogNotificationsRequest,
309    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
310
311    /// Sends a typing state.
312    fn typing(&self, request: TypingRequest)
313    -> BoxFuture<'static, BackendResult<OperationOutcome>>;
314
315    /// Receives the next batch of server-pushed client events.
316    fn receive_events(&self) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>>;
317}
318
319/// In-memory backend for client, bridge-adapter, and runtime tests.
320#[derive(Clone, Debug, Default)]
321pub struct InMemoryBackend {
322    state: Arc<Mutex<InMemoryBackendState>>,
323}
324
325#[derive(Debug)]
326struct InMemoryBackendState {
327    connected: bool,
328    account_namespace: Option<String>,
329    dialogs: Vec<DialogRecord>,
330    participants: HashMap<i64, Vec<ChatParticipantRecord>>,
331    messages: HashMap<i64, Vec<MessageRecord>>,
332    event_batches: VecDeque<BackendResult<Vec<ClientEvent>>>,
333    next_chat_id: i64,
334    next_message_id: i64,
335    next_random_id: i64,
336    next_transaction_id: u64,
337}
338
339impl Default for InMemoryBackendState {
340    fn default() -> Self {
341        Self {
342            connected: false,
343            account_namespace: None,
344            dialogs: Vec::new(),
345            participants: HashMap::new(),
346            messages: HashMap::new(),
347            event_batches: VecDeque::new(),
348            next_chat_id: 10_000,
349            next_message_id: 1,
350            next_random_id: 1,
351            next_transaction_id: 1,
352        }
353    }
354}
355
356impl InMemoryBackend {
357    /// Creates an empty in-memory backend.
358    pub fn new() -> Self {
359        Self::default()
360    }
361
362    /// Inserts or replaces a dialog.
363    pub fn upsert_dialog(&self, dialog: DialogRecord) {
364        let mut state = self.state.lock().expect("in-memory backend poisoned");
365        if let Some(existing) = state
366            .dialogs
367            .iter_mut()
368            .find(|existing| existing.chat_id == dialog.chat_id)
369        {
370            *existing = dialog;
371            return;
372        }
373        state.dialogs.push(dialog);
374    }
375
376    /// Replaces the participant snapshot for a chat.
377    pub fn set_chat_participants(
378        &self,
379        chat_id: InlineId,
380        participants: Vec<ChatParticipantRecord>,
381    ) {
382        let mut state = self.state.lock().expect("in-memory backend poisoned");
383        state.participants.insert(chat_id.get(), participants);
384    }
385
386    /// Inserts a message into the in-memory history.
387    pub fn insert_message(&self, message: MessageRecord) {
388        let mut state = self.state.lock().expect("in-memory backend poisoned");
389        let chat_id = message.chat_id.get();
390        state.messages.entry(chat_id).or_default().push(message);
391        if let Some(messages) = state.messages.get_mut(&chat_id) {
392            messages.sort_by_key(|message| (message.timestamp, message.message_id.get()));
393        }
394    }
395
396    /// Queues a server-pushed event batch for [`ClientBackend::receive_events`].
397    pub fn push_event_batch(&self, events: Vec<ClientEvent>) {
398        self.state
399            .lock()
400            .expect("in-memory backend poisoned")
401            .event_batches
402            .push_back(Ok(events));
403    }
404
405    /// Queues a server-pushed receive error for [`ClientBackend::receive_events`].
406    pub fn push_event_error(&self, error: BackendError) {
407        self.state
408            .lock()
409            .expect("in-memory backend poisoned")
410            .event_batches
411            .push_back(Err(error));
412    }
413
414    /// Returns whether the backend is connected.
415    pub fn is_connected(&self) -> bool {
416        self.state
417            .lock()
418            .expect("in-memory backend poisoned")
419            .connected
420    }
421
422    fn connect_now(&self, request: ConnectRequest) -> BackendResult<ClientStatusSnapshot> {
423        let mut state = self.state.lock().expect("in-memory backend poisoned");
424        state.connected = true;
425        state.account_namespace = request.account_namespace;
426        Ok(ClientStatusSnapshot::current(ClientStatus::Connected))
427    }
428
429    fn auth_start_now(&self, request: AuthStartRequest) -> BackendResult<AuthStartResult> {
430        if request.contact.trim().is_empty() {
431            return Err(BackendError::new(
432                ClientErrorCategory::InvalidInput,
433                "auth contact must not be empty",
434            ));
435        }
436        Ok(AuthStartResult {
437            existing_user: true,
438            needs_invite_code: false,
439            challenge_token: None,
440        })
441    }
442
443    fn auth_verify_now(&self, request: AuthVerifyRequest) -> BackendResult<AuthVerifyResult> {
444        if request.contact.trim().is_empty() {
445            return Err(BackendError::new(
446                ClientErrorCategory::InvalidInput,
447                "auth contact must not be empty",
448            ));
449        }
450        if request.code.trim().is_empty() {
451            return Err(BackendError::new(
452                ClientErrorCategory::InvalidInput,
453                "verification code must not be empty",
454            ));
455        }
456        let account_namespace = request
457            .account_namespace
458            .map(|namespace| namespace.trim().to_owned())
459            .filter(|namespace| !namespace.is_empty())
460            .unwrap_or_else(|| "1".to_owned());
461        let mut state = self.state.lock().expect("in-memory backend poisoned");
462        state.connected = true;
463        state.account_namespace = Some(account_namespace.clone());
464        Ok(AuthVerifyResult {
465            user_id: InlineId::new(1),
466            account_namespace,
467            status: ClientStatusSnapshot::current(ClientStatus::Connected),
468        })
469    }
470
471    fn resume_session_now(&self) -> BackendResult<ClientStatusSnapshot> {
472        let mut state = self.state.lock().expect("in-memory backend poisoned");
473        if state.connected || state.account_namespace.is_some() {
474            state.connected = true;
475            Ok(ClientStatusSnapshot::current(ClientStatus::Connected))
476        } else {
477            Ok(ClientStatusSnapshot::current(ClientStatus::AuthRequired))
478        }
479    }
480
481    fn logout_now(&self) -> BackendResult<()> {
482        let mut state = self.state.lock().expect("in-memory backend poisoned");
483        state.connected = false;
484        Ok(())
485    }
486
487    fn dialogs_now(&self, request: DialogsRequest) -> BackendResult<DialogsPage> {
488        self.require_connected()?;
489        let state = self.state.lock().expect("in-memory backend poisoned");
490        let start = parse_cursor(request.cursor.as_deref())?;
491        let limit = request.limit.unwrap_or(50).max(1) as usize;
492        let dialogs = state
493            .dialogs
494            .iter()
495            .skip(start)
496            .take(limit)
497            .map(|dialog| {
498                let mut dialog = dialog.clone();
499                dialog.synced_through_message_id =
500                    max_message_id_from_backend(&state.messages, dialog.chat_id);
501                dialog
502            })
503            .collect::<Vec<_>>();
504        let next = start + dialogs.len();
505        Ok(DialogsPage {
506            dialogs,
507            users: Vec::new(),
508            next_cursor: (next < state.dialogs.len()).then(|| next.to_string()),
509        })
510    }
511
512    fn history_now(&self, request: HistoryRequest) -> BackendResult<HistoryPage> {
513        self.require_connected()?;
514        let state = self.state.lock().expect("in-memory backend poisoned");
515        let mut messages = state
516            .messages
517            .get(&request.chat_id.get())
518            .cloned()
519            .unwrap_or_default();
520        messages.sort_by_key(|message| (message.timestamp, message.message_id.get()));
521        if request.before_message_id.is_some() && request.after_message_id.is_some() {
522            return Err(BackendError::new(
523                ClientErrorCategory::InvalidInput,
524                "history request cannot specify both before_message_id and after_message_id",
525            ));
526        }
527        if let Some(before) = request.before_message_id {
528            messages.retain(|message| message.message_id.get() < before.get());
529        }
530        if let Some(after) = request.after_message_id {
531            messages.retain(|message| message.message_id.get() > after.get());
532        }
533        let limit = request.limit.unwrap_or(50).max(1) as usize;
534        let has_more = messages.len() > limit;
535        if has_more {
536            if request.after_message_id.is_some() {
537                messages.truncate(limit);
538            } else {
539                let start = messages.len() - limit;
540                messages = messages[start..].to_vec();
541            }
542        }
543        Ok(HistoryPage {
544            messages,
545            users: Vec::new(),
546            has_more,
547            next_cursor: None,
548        })
549    }
550
551    fn send_text_now(&self, request: SendTextRequest) -> BackendResult<SendTextOutcome> {
552        self.require_connected()?;
553        if request.text.trim().is_empty() {
554            return Err(BackendError::new(
555                ClientErrorCategory::InvalidInput,
556                "message text must not be empty",
557            ));
558        }
559
560        let mut state = self.state.lock().expect("in-memory backend poisoned");
561        let chat_id = chat_id_for_peer(request.peer);
562        let message_id = InlineId::new(state.next_message_id);
563        state.next_message_id += 1;
564        let random_id = request.random_id.unwrap_or_else(|| {
565            let id = RandomId::new(state.next_random_id);
566            state.next_random_id += 1;
567            id
568        });
569        let transaction_id = TransactionId::try_new(format!("mem-{}", state.next_transaction_id))
570            .expect("generated transaction id is valid");
571        state.next_transaction_id += 1;
572
573        let transaction = TransactionIdentity::new(transaction_id, request.external_id, random_id)
574            .with_final_message_id(message_id);
575        let message = MessageRecord {
576            chat_id,
577            message_id,
578            sender_id: InlineId::new(0),
579            timestamp: message_id.get(),
580            is_outgoing: true,
581            content: MessageContent::Text { text: request.text },
582            reply_to_message_id: request.reply_to_message_id,
583            transaction: Some(transaction.clone()),
584        };
585        state
586            .messages
587            .entry(chat_id.get())
588            .or_default()
589            .push(message.clone());
590        crate::store::upsert_dialog_last_message(&mut state.dialogs, chat_id, message_id);
591
592        Ok(SendTextOutcome::with_state(
593            MessageMutation {
594                transaction,
595                message_id: Some(message_id),
596                state: None,
597                failure: None,
598            },
599            chat_id,
600            Some(message_id),
601            Some(message),
602            TransactionState::Completed,
603            None,
604        ))
605    }
606
607    fn create_chat_now(
608        &self,
609        title: Option<String>,
610        parent_chat_id: Option<InlineId>,
611        parent_message_id: Option<InlineId>,
612        participants: Vec<ChatParticipantRecord>,
613    ) -> BackendResult<CreatedChat> {
614        self.require_connected()?;
615        let title = title
616            .map(|title| title.trim().to_owned())
617            .filter(|title| !title.is_empty());
618        let mut state = self.state.lock().expect("in-memory backend poisoned");
619        let chat_id = InlineId::new(state.next_chat_id);
620        state.next_chat_id += 1;
621        state.dialogs.push(DialogRecord {
622            chat_id,
623            peer_user_id: None,
624            title: title.clone(),
625            last_message_id: None,
626            synced_through_message_id: None,
627            unread_count: Some(0),
628            ..DialogRecord::new(chat_id)
629        });
630        if !participants.is_empty() {
631            state.participants.insert(chat_id.get(), participants);
632        }
633        Ok(CreatedChat {
634            chat_id,
635            title,
636            parent_chat_id,
637            parent_message_id,
638        })
639    }
640
641    fn send_media_now(
642        &self,
643        request: UploadRequest,
644        bytes: Vec<u8>,
645    ) -> BackendResult<SendTextOutcome> {
646        self.require_connected()?;
647        if bytes.is_empty() {
648            return Err(BackendError::new(
649                ClientErrorCategory::InvalidInput,
650                "media bytes must not be empty",
651            ));
652        }
653
654        let mut state = self.state.lock().expect("in-memory backend poisoned");
655        let chat_id = chat_id_for_peer(request.peer);
656        let message_id = InlineId::new(state.next_message_id);
657        state.next_message_id += 1;
658        let random_id = request.random_id.unwrap_or_else(|| {
659            let id = RandomId::new(state.next_random_id);
660            state.next_random_id += 1;
661            id
662        });
663        let transaction_id =
664            TransactionId::try_new(format!("mem-upload-{}", state.next_transaction_id))
665                .expect("generated transaction id is valid");
666        state.next_transaction_id += 1;
667
668        let transaction =
669            TransactionIdentity::new(transaction_id, request.external_id.clone(), random_id)
670                .with_final_message_id(message_id);
671        let message = MessageRecord {
672            chat_id,
673            message_id,
674            sender_id: InlineId::new(0),
675            timestamp: message_id.get(),
676            is_outgoing: true,
677            content: MessageContent::Media {
678                kind: request.kind,
679                file_id: format!("mem-file-{}", message_id.get()),
680                url: None,
681                mime_type: request.mime_type.clone(),
682                file_name: request.file_name.clone(),
683                caption: request.caption.clone(),
684                size_bytes: request.size_bytes.or(Some(bytes.len() as u64)),
685                width: request.width,
686                height: request.height,
687                duration_ms: request.duration_ms,
688            },
689            reply_to_message_id: request.reply_to_message_id,
690            transaction: Some(transaction.clone()),
691        };
692        state
693            .messages
694            .entry(chat_id.get())
695            .or_default()
696            .push(message.clone());
697        crate::store::upsert_dialog_last_message(&mut state.dialogs, chat_id, message_id);
698
699        Ok(SendTextOutcome::with_state(
700            MessageMutation {
701                transaction,
702                message_id: Some(message_id),
703                state: None,
704                failure: None,
705            },
706            chat_id,
707            Some(message_id),
708            Some(message),
709            TransactionState::Completed,
710            None,
711        ))
712    }
713
714    fn require_connected(&self) -> BackendResult<()> {
715        if self
716            .state
717            .lock()
718            .expect("in-memory backend poisoned")
719            .connected
720        {
721            Ok(())
722        } else {
723            Err(BackendError::new(
724                ClientErrorCategory::AuthRequired,
725                "client is not connected",
726            ))
727        }
728    }
729}
730
731impl ClientBackend for InMemoryBackend {
732    fn auth_start(
733        &self,
734        request: AuthStartRequest,
735    ) -> BoxFuture<'static, BackendResult<AuthStartResult>> {
736        let backend = self.clone();
737        Box::pin(async move { backend.auth_start_now(request) })
738    }
739
740    fn auth_verify(
741        &self,
742        request: AuthVerifyRequest,
743    ) -> BoxFuture<'static, BackendResult<AuthVerifyResult>> {
744        let backend = self.clone();
745        Box::pin(async move { backend.auth_verify_now(request) })
746    }
747
748    fn resume_session(&self) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
749        let backend = self.clone();
750        Box::pin(async move { backend.resume_session_now() })
751    }
752
753    fn connect(
754        &self,
755        request: ConnectRequest,
756    ) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
757        let backend = self.clone();
758        Box::pin(async move { backend.connect_now(request) })
759    }
760
761    fn logout(&self) -> BoxFuture<'static, BackendResult<()>> {
762        let backend = self.clone();
763        Box::pin(async move { backend.logout_now() })
764    }
765
766    fn dialogs(&self, request: DialogsRequest) -> BoxFuture<'static, BackendResult<DialogsPage>> {
767        let backend = self.clone();
768        Box::pin(async move { backend.dialogs_now(request) })
769    }
770
771    fn cached_dialogs(
772        &self,
773        request: DialogsRequest,
774    ) -> BoxFuture<'static, BackendResult<DialogsPage>> {
775        let backend = self.clone();
776        Box::pin(async move { backend.dialogs_now(request) })
777    }
778
779    fn account_state(&self) -> BoxFuture<'static, BackendResult<AccountStateSnapshot>> {
780        let backend = self.clone();
781        Box::pin(async move {
782            backend.require_connected()?;
783            Ok(AccountStateSnapshot::default())
784        })
785    }
786
787    fn chat_state(
788        &self,
789        chat_id: InlineId,
790    ) -> BoxFuture<'static, BackendResult<ChatStateSnapshot>> {
791        let backend = self.clone();
792        Box::pin(async move {
793            backend.require_connected()?;
794            let state = backend.state.lock().expect("in-memory backend poisoned");
795            let dialog = state
796                .dialogs
797                .iter()
798                .find(|dialog| dialog.chat_id == chat_id)
799                .cloned();
800            let participants = state
801                .participants
802                .get(&chat_id.get())
803                .cloned()
804                .unwrap_or_default();
805            Ok(ChatStateSnapshot {
806                chat_id,
807                dialog,
808                deleted: false,
809                deleted_message_ids: Vec::new(),
810                reactions: Vec::new(),
811                reaction_snapshot_message_ids: Vec::new(),
812                read_state: None,
813                participants,
814                participants_complete: true,
815            })
816        })
817    }
818
819    fn history(&self, request: HistoryRequest) -> BoxFuture<'static, BackendResult<HistoryPage>> {
820        let backend = self.clone();
821        Box::pin(async move { backend.history_now(request) })
822    }
823
824    fn cached_history(
825        &self,
826        request: HistoryRequest,
827    ) -> BoxFuture<'static, BackendResult<HistoryPage>> {
828        let backend = self.clone();
829        Box::pin(async move { backend.history_now(request) })
830    }
831
832    fn chat_participants(
833        &self,
834        request: ChatParticipantsRequest,
835    ) -> BoxFuture<'static, BackendResult<ChatParticipantsPage>> {
836        let backend = self.clone();
837        Box::pin(async move {
838            backend.require_connected()?;
839            let state = backend.state.lock().expect("in-memory backend poisoned");
840            Ok(ChatParticipantsPage {
841                participants: state
842                    .participants
843                    .get(&request.chat_id.get())
844                    .cloned()
845                    .unwrap_or_default(),
846                users: Vec::new(),
847            })
848        })
849    }
850
851    fn add_chat_participant(
852        &self,
853        request: AddChatParticipantRequest,
854    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
855        let backend = self.clone();
856        Box::pin(async move {
857            backend.require_connected()?;
858            if request.chat_id.get() <= 0 || request.user_id.get() <= 0 {
859                return Err(BackendError::new(
860                    ClientErrorCategory::InvalidInput,
861                    "chat_id and user_id must be positive",
862                ));
863            }
864            let mut state = backend.state.lock().expect("in-memory backend poisoned");
865            let participants = state.participants.entry(request.chat_id.get()).or_default();
866            if !participants
867                .iter()
868                .any(|participant| participant.user_id == request.user_id)
869            {
870                participants.push(ChatParticipantRecord {
871                    user_id: request.user_id,
872                    date: None,
873                });
874            }
875            Ok(OperationOutcome::with_events(vec![
876                ClientEvent::ChatParticipantsChanged {
877                    chat_id: request.chat_id,
878                },
879            ]))
880        })
881    }
882
883    fn remove_chat_participant(
884        &self,
885        request: RemoveChatParticipantRequest,
886    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
887        let backend = self.clone();
888        Box::pin(async move {
889            backend.require_connected()?;
890            let mut state = backend.state.lock().expect("in-memory backend poisoned");
891            state
892                .participants
893                .entry(request.chat_id.get())
894                .or_default()
895                .retain(|participant| participant.user_id != request.user_id);
896            Ok(OperationOutcome::with_events(vec![
897                ClientEvent::ChatParticipantsChanged {
898                    chat_id: request.chat_id,
899                },
900            ]))
901        })
902    }
903
904    fn update_chat_info(
905        &self,
906        request: UpdateChatInfoRequest,
907    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
908        let backend = self.clone();
909        Box::pin(async move {
910            backend.require_connected()?;
911            if request.title.is_none() && request.emoji.is_none() {
912                return Err(BackendError::new(
913                    ClientErrorCategory::InvalidInput,
914                    "at least one chat info field must be provided",
915                ));
916            }
917            if let Some(title) = request.title {
918                if title.trim().is_empty() {
919                    return Err(BackendError::new(
920                        ClientErrorCategory::InvalidInput,
921                        "chat title must not be empty",
922                    ));
923                }
924                if let Some(dialog) = backend
925                    .state
926                    .lock()
927                    .expect("in-memory backend poisoned")
928                    .dialogs
929                    .iter_mut()
930                    .find(|dialog| dialog.chat_id == request.chat_id)
931                {
932                    dialog.title = Some(title.trim().to_owned());
933                }
934            }
935            Ok(OperationOutcome::with_events(vec![
936                ClientEvent::ChatUpserted {
937                    chat_id: request.chat_id,
938                },
939            ]))
940        })
941    }
942
943    fn delete_chat(
944        &self,
945        request: DeleteChatRequest,
946    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
947        let backend = self.clone();
948        Box::pin(async move {
949            backend.require_connected()?;
950            let mut state = backend.state.lock().expect("in-memory backend poisoned");
951            state
952                .dialogs
953                .retain(|dialog| dialog.chat_id != request.chat_id);
954            state.participants.remove(&request.chat_id.get());
955            state.messages.remove(&request.chat_id.get());
956            Ok(OperationOutcome::with_events(vec![
957                ClientEvent::ChatDeleted {
958                    chat_id: request.chat_id,
959                },
960            ]))
961        })
962    }
963
964    fn create_dm(
965        &self,
966        request: CreateDmRequest,
967    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
968        let backend = self.clone();
969        Box::pin(async move {
970            if request.user_id.get() <= 0 {
971                return Err(BackendError::new(
972                    ClientErrorCategory::InvalidInput,
973                    "user_id must be positive",
974                ));
975            }
976            backend.create_chat_now(
977                Some(format!("DM {}", request.user_id.get())),
978                None,
979                None,
980                vec![
981                    ChatParticipantRecord {
982                        user_id: InlineId::new(0),
983                        date: None,
984                    },
985                    ChatParticipantRecord {
986                        user_id: request.user_id,
987                        date: None,
988                    },
989                ],
990            )
991        })
992    }
993
994    fn create_thread(
995        &self,
996        request: CreateThreadRequest,
997    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
998        let backend = self.clone();
999        Box::pin(async move {
1000            let participants = request
1001                .participants
1002                .into_iter()
1003                .map(|participant| ChatParticipantRecord {
1004                    user_id: participant.user_id,
1005                    date: None,
1006                })
1007                .collect();
1008            backend.create_chat_now(request.title, None, None, participants)
1009        })
1010    }
1011
1012    fn create_reply_thread(
1013        &self,
1014        request: CreateReplyThreadRequest,
1015    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
1016        let backend = self.clone();
1017        Box::pin(async move {
1018            if request.parent_chat_id.get() <= 0 {
1019                return Err(BackendError::new(
1020                    ClientErrorCategory::InvalidInput,
1021                    "parent_chat_id must be positive",
1022                ));
1023            }
1024            let participants = request
1025                .participants
1026                .into_iter()
1027                .map(|participant| ChatParticipantRecord {
1028                    user_id: participant.user_id,
1029                    date: None,
1030                })
1031                .collect();
1032            backend.create_chat_now(
1033                request.title,
1034                Some(request.parent_chat_id),
1035                request.parent_message_id,
1036                participants,
1037            )
1038        })
1039    }
1040
1041    fn send_text(
1042        &self,
1043        request: SendTextRequest,
1044    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
1045        let backend = self.clone();
1046        Box::pin(async move { backend.send_text_now(request) })
1047    }
1048
1049    fn send_media(
1050        &self,
1051        request: UploadRequest,
1052        bytes: Vec<u8>,
1053    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
1054        let backend = self.clone();
1055        Box::pin(async move { backend.send_media_now(request, bytes) })
1056    }
1057
1058    fn edit_message(
1059        &self,
1060        request: EditMessageRequest,
1061    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1062        let backend = self.clone();
1063        Box::pin(async move {
1064            backend.require_connected()?;
1065            if request.text.trim().is_empty() {
1066                return Err(BackendError::new(
1067                    ClientErrorCategory::InvalidInput,
1068                    "message text must not be empty",
1069                ));
1070            }
1071            let mut state = backend.state.lock().expect("in-memory backend poisoned");
1072            let messages = state.messages.entry(request.chat_id.get()).or_default();
1073            if let Some(message) = messages
1074                .iter_mut()
1075                .find(|message| message.message_id == request.message_id)
1076            {
1077                message.content = MessageContent::Text { text: request.text };
1078                return Ok(OperationOutcome::with_events(vec![
1079                    ClientEvent::MessageStored {
1080                        message: message.clone(),
1081                    },
1082                ]));
1083            }
1084            Err(BackendError::new(
1085                ClientErrorCategory::InvalidInput,
1086                "message not found",
1087            ))
1088        })
1089    }
1090
1091    fn delete_message(
1092        &self,
1093        request: DeleteMessageRequest,
1094    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1095        let backend = self.clone();
1096        Box::pin(async move {
1097            backend.require_connected()?;
1098            let mut state = backend.state.lock().expect("in-memory backend poisoned");
1099            let messages = state.messages.entry(request.chat_id.get()).or_default();
1100            messages.retain(|message| message.message_id != request.message_id);
1101            Ok(OperationOutcome::with_events(vec![
1102                ClientEvent::MessageDeleted {
1103                    chat_id: request.chat_id,
1104                    message_id: request.message_id,
1105                },
1106            ]))
1107        })
1108    }
1109
1110    fn react(&self, request: ReactRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1111        let backend = self.clone();
1112        Box::pin(async move {
1113            backend.require_connected()?;
1114            if request.reaction.trim().is_empty() {
1115                return Err(BackendError::new(
1116                    ClientErrorCategory::InvalidInput,
1117                    "reaction must not be empty",
1118                ));
1119            }
1120            Ok(OperationOutcome::with_events(vec![
1121                ClientEvent::ReactionChanged {
1122                    chat_id: request.chat_id,
1123                    message_id: request.message_id,
1124                    user_id: InlineId::new(0),
1125                    reaction: request.reaction,
1126                    removed: request.remove,
1127                },
1128            ]))
1129        })
1130    }
1131
1132    fn read(&self, request: ReadRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1133        let backend = self.clone();
1134        Box::pin(async move {
1135            backend.require_connected()?;
1136            Ok(OperationOutcome::with_events(vec![
1137                ClientEvent::ReadStateChanged {
1138                    chat_id: request.chat_id,
1139                },
1140            ]))
1141        })
1142    }
1143
1144    fn set_marked_unread(
1145        &self,
1146        request: SetMarkedUnreadRequest,
1147    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1148        let backend = self.clone();
1149        Box::pin(async move {
1150            backend.require_connected()?;
1151            Ok(OperationOutcome::with_events(vec![
1152                ClientEvent::ReadStateChanged {
1153                    chat_id: request.chat_id,
1154                },
1155            ]))
1156        })
1157    }
1158
1159    fn update_dialog_notifications(
1160        &self,
1161        request: UpdateDialogNotificationsRequest,
1162    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1163        let backend = self.clone();
1164        Box::pin(async move {
1165            backend.require_connected()?;
1166            if let Some(dialog) = backend
1167                .state
1168                .lock()
1169                .expect("in-memory backend poisoned")
1170                .dialogs
1171                .iter_mut()
1172                .find(|dialog| dialog.chat_id == request.chat_id)
1173            {
1174                dialog.notification_mode = request.mode;
1175            }
1176            Ok(OperationOutcome::with_events(vec![
1177                ClientEvent::ChatUpserted {
1178                    chat_id: request.chat_id,
1179                },
1180            ]))
1181        })
1182    }
1183
1184    fn typing(
1185        &self,
1186        request: TypingRequest,
1187    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
1188        let backend = self.clone();
1189        Box::pin(async move {
1190            backend.require_connected()?;
1191            Ok(OperationOutcome::with_events(vec![ClientEvent::Typing {
1192                chat_id: request.chat_id,
1193                user_id: InlineId::new(0),
1194                is_typing: request.is_typing,
1195            }]))
1196        })
1197    }
1198
1199    fn receive_events(&self) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>> {
1200        let backend = self.clone();
1201        Box::pin(async move {
1202            loop {
1203                {
1204                    let mut state = backend.state.lock().expect("in-memory backend poisoned");
1205                    if !state.connected {
1206                        return Err(BackendError::new(
1207                            ClientErrorCategory::AuthRequired,
1208                            "client is not connected",
1209                        ));
1210                    }
1211                    if let Some(events) = state.event_batches.pop_front() {
1212                        return events;
1213                    }
1214                }
1215                tokio::time::sleep(Duration::from_millis(10)).await;
1216            }
1217        })
1218    }
1219}
1220
1221fn parse_cursor(cursor: Option<&str>) -> BackendResult<usize> {
1222    match cursor {
1223        Some(cursor) if !cursor.trim().is_empty() => cursor.parse::<usize>().map_err(|_| {
1224            BackendError::new(
1225                ClientErrorCategory::InvalidInput,
1226                "invalid pagination cursor",
1227            )
1228        }),
1229        _ => Ok(0),
1230    }
1231}
1232
1233fn max_message_id_from_backend(
1234    messages: &HashMap<i64, Vec<MessageRecord>>,
1235    chat_id: InlineId,
1236) -> Option<InlineId> {
1237    messages
1238        .get(&chat_id.get())?
1239        .iter()
1240        .map(|message| message.message_id.get())
1241        .max()
1242        .map(InlineId::new)
1243}
1244
1245fn chat_id_for_peer(peer: crate::PeerRef) -> InlineId {
1246    match peer {
1247        crate::PeerRef::User { user_id } => user_id,
1248        crate::PeerRef::Chat { chat_id } => chat_id,
1249        crate::PeerRef::Thread { thread_id } => thread_id,
1250    }
1251}
1252
1253#[cfg(test)]
1254mod tests {
1255    use crate::{PeerRef, SendTextRequest};
1256
1257    use super::*;
1258
1259    fn token_connect() -> ConnectRequest {
1260        ConnectRequest::new(crate::AuthCredential::AccessToken {
1261            token: crate::AuthToken::try_new("token").unwrap(),
1262        })
1263    }
1264
1265    #[tokio::test]
1266    async fn in_memory_backend_requires_connect_for_dialogs() {
1267        let backend = InMemoryBackend::new();
1268
1269        let err = backend
1270            .dialogs(DialogsRequest::default())
1271            .await
1272            .expect_err("dialogs should require connect");
1273
1274        assert_eq!(err.category, ClientErrorCategory::AuthRequired);
1275    }
1276
1277    #[tokio::test]
1278    async fn in_memory_backend_lists_dialogs_with_cursor() {
1279        let backend = InMemoryBackend::new();
1280        backend.upsert_dialog(DialogRecord {
1281            chat_id: InlineId::new(1),
1282            peer_user_id: None,
1283            title: Some("one".to_owned()),
1284            last_message_id: None,
1285            synced_through_message_id: None,
1286            unread_count: Some(0),
1287            ..DialogRecord::new(InlineId::new(1))
1288        });
1289        backend.upsert_dialog(DialogRecord {
1290            chat_id: InlineId::new(2),
1291            peer_user_id: Some(InlineId::new(3)),
1292            title: Some("two".to_owned()),
1293            last_message_id: None,
1294            synced_through_message_id: None,
1295            unread_count: Some(0),
1296            ..DialogRecord::new(InlineId::new(2))
1297        });
1298        backend.connect(token_connect()).await.unwrap();
1299
1300        let first = backend
1301            .dialogs(DialogsRequest {
1302                limit: Some(1),
1303                cursor: None,
1304            })
1305            .await
1306            .unwrap();
1307        assert_eq!(first.dialogs[0].chat_id, InlineId::new(1));
1308        assert_eq!(first.next_cursor.as_deref(), Some("1"));
1309
1310        let second = backend
1311            .dialogs(DialogsRequest {
1312                limit: Some(1),
1313                cursor: first.next_cursor,
1314            })
1315            .await
1316            .unwrap();
1317        assert_eq!(second.dialogs[0].chat_id, InlineId::new(2));
1318        assert_eq!(second.next_cursor, None);
1319    }
1320
1321    #[tokio::test]
1322    async fn in_memory_backend_returns_chat_participants() {
1323        let backend = InMemoryBackend::new();
1324        backend.set_chat_participants(
1325            InlineId::new(7),
1326            vec![ChatParticipantRecord {
1327                user_id: InlineId::new(42),
1328                date: Some(100),
1329            }],
1330        );
1331        backend.connect(token_connect()).await.unwrap();
1332
1333        let page = backend
1334            .chat_participants(ChatParticipantsRequest {
1335                chat_id: InlineId::new(7),
1336            })
1337            .await
1338            .unwrap();
1339
1340        assert_eq!(page.participants.len(), 1);
1341        assert_eq!(page.participants[0].user_id, InlineId::new(42));
1342        assert_eq!(page.participants[0].date, Some(100));
1343    }
1344
1345    #[tokio::test]
1346    async fn in_memory_backend_sends_text_and_records_history() {
1347        let backend = InMemoryBackend::new();
1348        backend.connect(token_connect()).await.unwrap();
1349
1350        let outcome = backend
1351            .send_text(SendTextRequest::new(
1352                PeerRef::Chat {
1353                    chat_id: InlineId::new(7),
1354                },
1355                "hello",
1356            ))
1357            .await
1358            .unwrap();
1359
1360        assert_eq!(outcome.chat_id, InlineId::new(7));
1361        assert_eq!(outcome.message_id, Some(InlineId::new(1)));
1362
1363        let history = backend
1364            .history(HistoryRequest {
1365                chat_id: InlineId::new(7),
1366                limit: Some(10),
1367                before_message_id: None,
1368                after_message_id: None,
1369            })
1370            .await
1371            .unwrap();
1372        assert_eq!(history.messages.len(), 1);
1373        assert_eq!(history.messages[0].message_id, InlineId::new(1));
1374    }
1375
1376    #[tokio::test]
1377    async fn in_memory_backend_rejects_empty_text() {
1378        let backend = InMemoryBackend::new();
1379        backend.connect(token_connect()).await.unwrap();
1380
1381        let err = backend
1382            .send_text(SendTextRequest::new(
1383                PeerRef::Chat {
1384                    chat_id: InlineId::new(7),
1385                },
1386                " ",
1387            ))
1388            .await
1389            .expect_err("empty message should fail");
1390
1391        assert_eq!(err.category, ClientErrorCategory::InvalidInput);
1392    }
1393}
inline_client/backend.rs

inline_client/
backend.rs
