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    AuthStartRequest, AuthStartResult, AuthVerifyRequest, AuthVerifyResult, ChatParticipantRecord,
20    ChatParticipantsPage, ChatParticipantsRequest, ClientErrorCategory, ClientEvent, ClientFailure,
21    ClientStatus, ClientStatusSnapshot, ConnectRequest, CreateDmRequest, CreateReplyThreadRequest,
22    CreateThreadRequest, CreatedChat, DeleteMessageRequest, DialogRecord, DialogsPage,
23    DialogsRequest, EditMessageRequest, HistoryPage, HistoryRequest, InlineId, MessageContent,
24    MessageMutation, MessageRecord, RandomId, ReactRequest, ReadRequest, SendTextRequest,
25    TransactionEvent, TransactionId, TransactionIdentity, TransactionState, TypingRequest,
26    UploadRequest,
27};
28
29/// Result type returned by client backends.
30pub type BackendResult<T> = Result<T, BackendError>;
31
32/// Redacted backend error.
33#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
34#[error("{category:?}: {message}")]
35pub struct BackendError {
36    /// Stable error category.
37    pub category: ClientErrorCategory,
38    /// Redacted message suitable for hosts.
39    pub message: String,
40}
41
42impl BackendError {
43    /// Creates a backend error.
44    pub fn new(category: ClientErrorCategory, message: impl Into<String>) -> Self {
45        Self {
46            category,
47            message: message.into(),
48        }
49    }
50}
51
52impl From<BackendError> for ClientFailure {
53    fn from(error: BackendError) -> Self {
54        Self::new(error.category, error.message)
55    }
56}
57
58/// Outcome from a text-send operation.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct SendTextOutcome {
61    /// Client mutation acknowledgement.
62    pub mutation: MessageMutation,
63    /// Chat containing the message.
64    pub chat_id: InlineId,
65    /// Final message ID, when known.
66    pub message_id: Option<InlineId>,
67    /// Message record applied to the client store, when known.
68    pub message: Option<MessageRecord>,
69    /// Durable transaction state after the send attempt.
70    pub state: TransactionState,
71    /// Redacted failure, when the transaction failed.
72    pub failure: Option<ClientFailure>,
73}
74
75/// Outcome from a side-effect operation.
76#[derive(Clone, Debug, Default, PartialEq, Eq)]
77pub struct OperationOutcome {
78    /// Committed client events caused by the operation.
79    pub events: Vec<ClientEvent>,
80}
81
82impl OperationOutcome {
83    /// Creates an outcome with no committed events.
84    pub fn empty() -> Self {
85        Self::default()
86    }
87
88    /// Creates an outcome with committed events.
89    pub fn with_events(events: Vec<ClientEvent>) -> Self {
90        Self { events }
91    }
92}
93
94impl SendTextOutcome {
95    /// Creates a text-send outcome.
96    pub fn new(mutation: MessageMutation, chat_id: InlineId, message_id: Option<InlineId>) -> Self {
97        Self::with_state(
98            mutation,
99            chat_id,
100            message_id,
101            None,
102            TransactionState::Completed,
103            None,
104        )
105    }
106
107    /// Creates a text-send outcome with an explicit transaction state.
108    pub fn with_state(
109        mutation: MessageMutation,
110        chat_id: InlineId,
111        message_id: Option<InlineId>,
112        message: Option<MessageRecord>,
113        state: TransactionState,
114        failure: Option<ClientFailure>,
115    ) -> Self {
116        Self {
117            mutation,
118            chat_id,
119            message_id,
120            message,
121            state,
122            failure,
123        }
124    }
125
126    /// Returns a transaction event for this send outcome.
127    pub fn transaction_event(&self) -> TransactionEvent {
128        TransactionEvent {
129            identity: self.mutation.transaction.clone(),
130            state: self.state,
131            failure: self.failure.clone(),
132        }
133    }
134}
135
136/// Async backend boundary for the client runner.
137pub trait ClientBackend: fmt::Debug + Send + Sync + 'static {
138    /// Sends an Inline login code.
139    fn auth_start(
140        &self,
141        request: AuthStartRequest,
142    ) -> BoxFuture<'static, BackendResult<AuthStartResult>>;
143
144    /// Verifies an Inline login code and persists the resulting session.
145    fn auth_verify(
146        &self,
147        request: AuthVerifyRequest,
148    ) -> BoxFuture<'static, BackendResult<AuthVerifyResult>>;
149
150    /// Resumes a previously stored session, if available.
151    fn resume_session(&self) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>>;
152
153    /// Connects or reconnects the client.
154    fn connect(
155        &self,
156        request: ConnectRequest,
157    ) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>>;
158
159    /// Logs out the current account.
160    fn logout(&self) -> BoxFuture<'static, BackendResult<()>>;
161
162    /// Lists dialogs.
163    fn dialogs(&self, request: DialogsRequest) -> BoxFuture<'static, BackendResult<DialogsPage>>;
164
165    /// Fetches chat history.
166    fn history(&self, request: HistoryRequest) -> BoxFuture<'static, BackendResult<HistoryPage>>;
167
168    /// Fetches chat participants.
169    fn chat_participants(
170        &self,
171        request: ChatParticipantsRequest,
172    ) -> BoxFuture<'static, BackendResult<ChatParticipantsPage>>;
173
174    /// Creates or opens a direct message chat.
175    fn create_dm(&self, request: CreateDmRequest)
176    -> BoxFuture<'static, BackendResult<CreatedChat>>;
177
178    /// Creates a regular Inline thread chat.
179    fn create_thread(
180        &self,
181        request: CreateThreadRequest,
182    ) -> BoxFuture<'static, BackendResult<CreatedChat>>;
183
184    /// Creates a child/reply Inline thread chat.
185    fn create_reply_thread(
186        &self,
187        request: CreateReplyThreadRequest,
188    ) -> BoxFuture<'static, BackendResult<CreatedChat>>;
189
190    /// Sends a text message.
191    fn send_text(
192        &self,
193        request: SendTextRequest,
194    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>>;
195
196    /// Uploads and sends a media message.
197    fn send_media(
198        &self,
199        request: UploadRequest,
200        bytes: Vec<u8>,
201    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>>;
202
203    /// Edits a text message.
204    fn edit_message(
205        &self,
206        request: EditMessageRequest,
207    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
208
209    /// Deletes or unsends a message.
210    fn delete_message(
211        &self,
212        request: DeleteMessageRequest,
213    ) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
214
215    /// Adds or removes a reaction.
216    fn react(&self, request: ReactRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
217
218    /// Marks messages read.
219    fn read(&self, request: ReadRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>>;
220
221    /// Sends a typing state.
222    fn typing(&self, request: TypingRequest)
223    -> BoxFuture<'static, BackendResult<OperationOutcome>>;
224
225    /// Receives the next batch of server-pushed client events.
226    fn receive_events(&self) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>>;
227}
228
229/// In-memory backend for client, bridge-adapter, and runtime tests.
230#[derive(Clone, Debug, Default)]
231pub struct InMemoryBackend {
232    state: Arc<Mutex<InMemoryBackendState>>,
233}
234
235#[derive(Debug)]
236struct InMemoryBackendState {
237    connected: bool,
238    account_namespace: Option<String>,
239    dialogs: Vec<DialogRecord>,
240    participants: HashMap<i64, Vec<ChatParticipantRecord>>,
241    messages: HashMap<i64, Vec<MessageRecord>>,
242    event_batches: VecDeque<BackendResult<Vec<ClientEvent>>>,
243    next_chat_id: i64,
244    next_message_id: i64,
245    next_random_id: i64,
246    next_transaction_id: u64,
247}
248
249impl Default for InMemoryBackendState {
250    fn default() -> Self {
251        Self {
252            connected: false,
253            account_namespace: None,
254            dialogs: Vec::new(),
255            participants: HashMap::new(),
256            messages: HashMap::new(),
257            event_batches: VecDeque::new(),
258            next_chat_id: 10_000,
259            next_message_id: 1,
260            next_random_id: 1,
261            next_transaction_id: 1,
262        }
263    }
264}
265
266impl InMemoryBackend {
267    /// Creates an empty in-memory backend.
268    pub fn new() -> Self {
269        Self::default()
270    }
271
272    /// Inserts or replaces a dialog.
273    pub fn upsert_dialog(&self, dialog: DialogRecord) {
274        let mut state = self.state.lock().expect("in-memory backend poisoned");
275        if let Some(existing) = state
276            .dialogs
277            .iter_mut()
278            .find(|existing| existing.chat_id == dialog.chat_id)
279        {
280            *existing = dialog;
281            return;
282        }
283        state.dialogs.push(dialog);
284    }
285
286    /// Replaces the participant snapshot for a chat.
287    pub fn set_chat_participants(
288        &self,
289        chat_id: InlineId,
290        participants: Vec<ChatParticipantRecord>,
291    ) {
292        let mut state = self.state.lock().expect("in-memory backend poisoned");
293        state.participants.insert(chat_id.get(), participants);
294    }
295
296    /// Inserts a message into the in-memory history.
297    pub fn insert_message(&self, message: MessageRecord) {
298        let mut state = self.state.lock().expect("in-memory backend poisoned");
299        let chat_id = message.chat_id.get();
300        state.messages.entry(chat_id).or_default().push(message);
301        if let Some(messages) = state.messages.get_mut(&chat_id) {
302            messages.sort_by_key(|message| (message.timestamp, message.message_id.get()));
303        }
304    }
305
306    /// Queues a server-pushed event batch for [`ClientBackend::receive_events`].
307    pub fn push_event_batch(&self, events: Vec<ClientEvent>) {
308        self.state
309            .lock()
310            .expect("in-memory backend poisoned")
311            .event_batches
312            .push_back(Ok(events));
313    }
314
315    /// Queues a server-pushed receive error for [`ClientBackend::receive_events`].
316    pub fn push_event_error(&self, error: BackendError) {
317        self.state
318            .lock()
319            .expect("in-memory backend poisoned")
320            .event_batches
321            .push_back(Err(error));
322    }
323
324    /// Returns whether the backend is connected.
325    pub fn is_connected(&self) -> bool {
326        self.state
327            .lock()
328            .expect("in-memory backend poisoned")
329            .connected
330    }
331
332    fn connect_now(&self, request: ConnectRequest) -> BackendResult<ClientStatusSnapshot> {
333        let mut state = self.state.lock().expect("in-memory backend poisoned");
334        state.connected = true;
335        state.account_namespace = request.account_namespace;
336        Ok(ClientStatusSnapshot::current(ClientStatus::Connected))
337    }
338
339    fn auth_start_now(&self, request: AuthStartRequest) -> BackendResult<AuthStartResult> {
340        if request.contact.trim().is_empty() {
341            return Err(BackendError::new(
342                ClientErrorCategory::InvalidInput,
343                "auth contact must not be empty",
344            ));
345        }
346        Ok(AuthStartResult {
347            existing_user: true,
348            needs_invite_code: false,
349            challenge_token: None,
350        })
351    }
352
353    fn auth_verify_now(&self, request: AuthVerifyRequest) -> BackendResult<AuthVerifyResult> {
354        if request.contact.trim().is_empty() {
355            return Err(BackendError::new(
356                ClientErrorCategory::InvalidInput,
357                "auth contact must not be empty",
358            ));
359        }
360        if request.code.trim().is_empty() {
361            return Err(BackendError::new(
362                ClientErrorCategory::InvalidInput,
363                "verification code must not be empty",
364            ));
365        }
366        let account_namespace = request
367            .account_namespace
368            .map(|namespace| namespace.trim().to_owned())
369            .filter(|namespace| !namespace.is_empty())
370            .unwrap_or_else(|| "1".to_owned());
371        let mut state = self.state.lock().expect("in-memory backend poisoned");
372        state.connected = true;
373        state.account_namespace = Some(account_namespace.clone());
374        Ok(AuthVerifyResult {
375            user_id: InlineId::new(1),
376            account_namespace,
377            status: ClientStatusSnapshot::current(ClientStatus::Connected),
378        })
379    }
380
381    fn resume_session_now(&self) -> BackendResult<ClientStatusSnapshot> {
382        let mut state = self.state.lock().expect("in-memory backend poisoned");
383        if state.connected || state.account_namespace.is_some() {
384            state.connected = true;
385            Ok(ClientStatusSnapshot::current(ClientStatus::Connected))
386        } else {
387            Ok(ClientStatusSnapshot::current(ClientStatus::AuthRequired))
388        }
389    }
390
391    fn logout_now(&self) -> BackendResult<()> {
392        let mut state = self.state.lock().expect("in-memory backend poisoned");
393        state.connected = false;
394        Ok(())
395    }
396
397    fn dialogs_now(&self, request: DialogsRequest) -> BackendResult<DialogsPage> {
398        self.require_connected()?;
399        let state = self.state.lock().expect("in-memory backend poisoned");
400        let start = parse_cursor(request.cursor.as_deref())?;
401        let limit = request.limit.unwrap_or(50).max(1) as usize;
402        let dialogs = state
403            .dialogs
404            .iter()
405            .skip(start)
406            .take(limit)
407            .map(|dialog| {
408                let mut dialog = dialog.clone();
409                dialog.synced_through_message_id =
410                    max_message_id_from_backend(&state.messages, dialog.chat_id);
411                dialog
412            })
413            .collect::<Vec<_>>();
414        let next = start + dialogs.len();
415        Ok(DialogsPage {
416            dialogs,
417            users: Vec::new(),
418            next_cursor: (next < state.dialogs.len()).then(|| next.to_string()),
419        })
420    }
421
422    fn history_now(&self, request: HistoryRequest) -> BackendResult<HistoryPage> {
423        self.require_connected()?;
424        let state = self.state.lock().expect("in-memory backend poisoned");
425        let mut messages = state
426            .messages
427            .get(&request.chat_id.get())
428            .cloned()
429            .unwrap_or_default();
430        messages.sort_by_key(|message| (message.timestamp, message.message_id.get()));
431        if request.before_message_id.is_some() && request.after_message_id.is_some() {
432            return Err(BackendError::new(
433                ClientErrorCategory::InvalidInput,
434                "history request cannot specify both before_message_id and after_message_id",
435            ));
436        }
437        if let Some(before) = request.before_message_id {
438            messages.retain(|message| message.message_id.get() < before.get());
439        }
440        if let Some(after) = request.after_message_id {
441            messages.retain(|message| message.message_id.get() > after.get());
442        }
443        let limit = request.limit.unwrap_or(50).max(1) as usize;
444        let has_more = messages.len() > limit;
445        if has_more {
446            if request.after_message_id.is_some() {
447                messages.truncate(limit);
448            } else {
449                let start = messages.len() - limit;
450                messages = messages[start..].to_vec();
451            }
452        }
453        Ok(HistoryPage {
454            messages,
455            users: Vec::new(),
456            has_more,
457            next_cursor: None,
458        })
459    }
460
461    fn send_text_now(&self, request: SendTextRequest) -> BackendResult<SendTextOutcome> {
462        self.require_connected()?;
463        if request.text.trim().is_empty() {
464            return Err(BackendError::new(
465                ClientErrorCategory::InvalidInput,
466                "message text must not be empty",
467            ));
468        }
469
470        let mut state = self.state.lock().expect("in-memory backend poisoned");
471        let chat_id = chat_id_for_peer(request.peer);
472        let message_id = InlineId::new(state.next_message_id);
473        state.next_message_id += 1;
474        let random_id = request.random_id.unwrap_or_else(|| {
475            let id = RandomId::new(state.next_random_id);
476            state.next_random_id += 1;
477            id
478        });
479        let transaction_id = TransactionId::try_new(format!("mem-{}", state.next_transaction_id))
480            .expect("generated transaction id is valid");
481        state.next_transaction_id += 1;
482
483        let transaction = TransactionIdentity::new(transaction_id, request.external_id, random_id)
484            .with_final_message_id(message_id);
485        let message = MessageRecord {
486            chat_id,
487            message_id,
488            sender_id: InlineId::new(0),
489            timestamp: message_id.get(),
490            is_outgoing: true,
491            content: MessageContent::Text { text: request.text },
492            reply_to_message_id: request.reply_to_message_id,
493            transaction: Some(transaction.clone()),
494        };
495        state
496            .messages
497            .entry(chat_id.get())
498            .or_default()
499            .push(message.clone());
500        crate::store::upsert_dialog_last_message(&mut state.dialogs, chat_id, message_id);
501
502        Ok(SendTextOutcome::with_state(
503            MessageMutation {
504                transaction,
505                message_id: Some(message_id),
506            },
507            chat_id,
508            Some(message_id),
509            Some(message),
510            TransactionState::Completed,
511            None,
512        ))
513    }
514
515    fn create_chat_now(
516        &self,
517        title: Option<String>,
518        parent_chat_id: Option<InlineId>,
519        parent_message_id: Option<InlineId>,
520        participants: Vec<ChatParticipantRecord>,
521    ) -> BackendResult<CreatedChat> {
522        self.require_connected()?;
523        let title = title
524            .map(|title| title.trim().to_owned())
525            .filter(|title| !title.is_empty());
526        let mut state = self.state.lock().expect("in-memory backend poisoned");
527        let chat_id = InlineId::new(state.next_chat_id);
528        state.next_chat_id += 1;
529        state.dialogs.push(DialogRecord {
530            chat_id,
531            peer_user_id: None,
532            title: title.clone(),
533            last_message_id: None,
534            synced_through_message_id: None,
535            unread_count: Some(0),
536        });
537        if !participants.is_empty() {
538            state.participants.insert(chat_id.get(), participants);
539        }
540        Ok(CreatedChat {
541            chat_id,
542            title,
543            parent_chat_id,
544            parent_message_id,
545        })
546    }
547
548    fn send_media_now(
549        &self,
550        request: UploadRequest,
551        bytes: Vec<u8>,
552    ) -> BackendResult<SendTextOutcome> {
553        self.require_connected()?;
554        if bytes.is_empty() {
555            return Err(BackendError::new(
556                ClientErrorCategory::InvalidInput,
557                "media bytes must not be empty",
558            ));
559        }
560
561        let mut state = self.state.lock().expect("in-memory backend poisoned");
562        let chat_id = chat_id_for_peer(request.peer);
563        let message_id = InlineId::new(state.next_message_id);
564        state.next_message_id += 1;
565        let random_id = request.random_id.unwrap_or_else(|| {
566            let id = RandomId::new(state.next_random_id);
567            state.next_random_id += 1;
568            id
569        });
570        let transaction_id =
571            TransactionId::try_new(format!("mem-upload-{}", state.next_transaction_id))
572                .expect("generated transaction id is valid");
573        state.next_transaction_id += 1;
574
575        let transaction =
576            TransactionIdentity::new(transaction_id, request.external_id.clone(), random_id)
577                .with_final_message_id(message_id);
578        let message = MessageRecord {
579            chat_id,
580            message_id,
581            sender_id: InlineId::new(0),
582            timestamp: message_id.get(),
583            is_outgoing: true,
584            content: MessageContent::Media {
585                kind: request.kind,
586                file_id: format!("mem-file-{}", message_id.get()),
587                url: None,
588                mime_type: request.mime_type.clone(),
589                file_name: request.file_name.clone(),
590                caption: request.caption.clone(),
591                size_bytes: request.size_bytes.or(Some(bytes.len() as u64)),
592                width: request.width,
593                height: request.height,
594                duration_ms: request.duration_ms,
595            },
596            reply_to_message_id: request.reply_to_message_id,
597            transaction: Some(transaction.clone()),
598        };
599        state
600            .messages
601            .entry(chat_id.get())
602            .or_default()
603            .push(message.clone());
604        crate::store::upsert_dialog_last_message(&mut state.dialogs, chat_id, message_id);
605
606        Ok(SendTextOutcome::with_state(
607            MessageMutation {
608                transaction,
609                message_id: Some(message_id),
610            },
611            chat_id,
612            Some(message_id),
613            Some(message),
614            TransactionState::Completed,
615            None,
616        ))
617    }
618
619    fn require_connected(&self) -> BackendResult<()> {
620        if self
621            .state
622            .lock()
623            .expect("in-memory backend poisoned")
624            .connected
625        {
626            Ok(())
627        } else {
628            Err(BackendError::new(
629                ClientErrorCategory::AuthRequired,
630                "client is not connected",
631            ))
632        }
633    }
634}
635
636impl ClientBackend for InMemoryBackend {
637    fn auth_start(
638        &self,
639        request: AuthStartRequest,
640    ) -> BoxFuture<'static, BackendResult<AuthStartResult>> {
641        let backend = self.clone();
642        Box::pin(async move { backend.auth_start_now(request) })
643    }
644
645    fn auth_verify(
646        &self,
647        request: AuthVerifyRequest,
648    ) -> BoxFuture<'static, BackendResult<AuthVerifyResult>> {
649        let backend = self.clone();
650        Box::pin(async move { backend.auth_verify_now(request) })
651    }
652
653    fn resume_session(&self) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
654        let backend = self.clone();
655        Box::pin(async move { backend.resume_session_now() })
656    }
657
658    fn connect(
659        &self,
660        request: ConnectRequest,
661    ) -> BoxFuture<'static, BackendResult<ClientStatusSnapshot>> {
662        let backend = self.clone();
663        Box::pin(async move { backend.connect_now(request) })
664    }
665
666    fn logout(&self) -> BoxFuture<'static, BackendResult<()>> {
667        let backend = self.clone();
668        Box::pin(async move { backend.logout_now() })
669    }
670
671    fn dialogs(&self, request: DialogsRequest) -> BoxFuture<'static, BackendResult<DialogsPage>> {
672        let backend = self.clone();
673        Box::pin(async move { backend.dialogs_now(request) })
674    }
675
676    fn history(&self, request: HistoryRequest) -> BoxFuture<'static, BackendResult<HistoryPage>> {
677        let backend = self.clone();
678        Box::pin(async move { backend.history_now(request) })
679    }
680
681    fn chat_participants(
682        &self,
683        request: ChatParticipantsRequest,
684    ) -> BoxFuture<'static, BackendResult<ChatParticipantsPage>> {
685        let backend = self.clone();
686        Box::pin(async move {
687            backend.require_connected()?;
688            let state = backend.state.lock().expect("in-memory backend poisoned");
689            Ok(ChatParticipantsPage {
690                participants: state
691                    .participants
692                    .get(&request.chat_id.get())
693                    .cloned()
694                    .unwrap_or_default(),
695                users: Vec::new(),
696            })
697        })
698    }
699
700    fn create_dm(
701        &self,
702        request: CreateDmRequest,
703    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
704        let backend = self.clone();
705        Box::pin(async move {
706            if request.user_id.get() <= 0 {
707                return Err(BackendError::new(
708                    ClientErrorCategory::InvalidInput,
709                    "user_id must be positive",
710                ));
711            }
712            backend.create_chat_now(
713                Some(format!("DM {}", request.user_id.get())),
714                None,
715                None,
716                vec![
717                    ChatParticipantRecord {
718                        user_id: InlineId::new(0),
719                        date: None,
720                    },
721                    ChatParticipantRecord {
722                        user_id: request.user_id,
723                        date: None,
724                    },
725                ],
726            )
727        })
728    }
729
730    fn create_thread(
731        &self,
732        request: CreateThreadRequest,
733    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
734        let backend = self.clone();
735        Box::pin(async move {
736            let participants = request
737                .participants
738                .into_iter()
739                .map(|participant| ChatParticipantRecord {
740                    user_id: participant.user_id,
741                    date: None,
742                })
743                .collect();
744            backend.create_chat_now(request.title, None, None, participants)
745        })
746    }
747
748    fn create_reply_thread(
749        &self,
750        request: CreateReplyThreadRequest,
751    ) -> BoxFuture<'static, BackendResult<CreatedChat>> {
752        let backend = self.clone();
753        Box::pin(async move {
754            if request.parent_chat_id.get() <= 0 {
755                return Err(BackendError::new(
756                    ClientErrorCategory::InvalidInput,
757                    "parent_chat_id must be positive",
758                ));
759            }
760            let participants = request
761                .participants
762                .into_iter()
763                .map(|participant| ChatParticipantRecord {
764                    user_id: participant.user_id,
765                    date: None,
766                })
767                .collect();
768            backend.create_chat_now(
769                request.title,
770                Some(request.parent_chat_id),
771                request.parent_message_id,
772                participants,
773            )
774        })
775    }
776
777    fn send_text(
778        &self,
779        request: SendTextRequest,
780    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
781        let backend = self.clone();
782        Box::pin(async move { backend.send_text_now(request) })
783    }
784
785    fn send_media(
786        &self,
787        request: UploadRequest,
788        bytes: Vec<u8>,
789    ) -> BoxFuture<'static, BackendResult<SendTextOutcome>> {
790        let backend = self.clone();
791        Box::pin(async move { backend.send_media_now(request, bytes) })
792    }
793
794    fn edit_message(
795        &self,
796        request: EditMessageRequest,
797    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
798        let backend = self.clone();
799        Box::pin(async move {
800            backend.require_connected()?;
801            if request.text.trim().is_empty() {
802                return Err(BackendError::new(
803                    ClientErrorCategory::InvalidInput,
804                    "message text must not be empty",
805                ));
806            }
807            let mut state = backend.state.lock().expect("in-memory backend poisoned");
808            let messages = state.messages.entry(request.chat_id.get()).or_default();
809            if let Some(message) = messages
810                .iter_mut()
811                .find(|message| message.message_id == request.message_id)
812            {
813                message.content = MessageContent::Text { text: request.text };
814                return Ok(OperationOutcome::with_events(vec![
815                    ClientEvent::MessageStored {
816                        message: message.clone(),
817                    },
818                ]));
819            }
820            Err(BackendError::new(
821                ClientErrorCategory::InvalidInput,
822                "message not found",
823            ))
824        })
825    }
826
827    fn delete_message(
828        &self,
829        request: DeleteMessageRequest,
830    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
831        let backend = self.clone();
832        Box::pin(async move {
833            backend.require_connected()?;
834            let mut state = backend.state.lock().expect("in-memory backend poisoned");
835            let messages = state.messages.entry(request.chat_id.get()).or_default();
836            messages.retain(|message| message.message_id != request.message_id);
837            Ok(OperationOutcome::with_events(vec![
838                ClientEvent::MessageDeleted {
839                    chat_id: request.chat_id,
840                    message_id: request.message_id,
841                },
842            ]))
843        })
844    }
845
846    fn react(&self, request: ReactRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
847        let backend = self.clone();
848        Box::pin(async move {
849            backend.require_connected()?;
850            if request.reaction.trim().is_empty() {
851                return Err(BackendError::new(
852                    ClientErrorCategory::InvalidInput,
853                    "reaction must not be empty",
854                ));
855            }
856            Ok(OperationOutcome::with_events(vec![
857                ClientEvent::ReactionChanged {
858                    chat_id: request.chat_id,
859                    message_id: request.message_id,
860                    user_id: InlineId::new(0),
861                    reaction: request.reaction,
862                    removed: request.remove,
863                },
864            ]))
865        })
866    }
867
868    fn read(&self, request: ReadRequest) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
869        let backend = self.clone();
870        Box::pin(async move {
871            backend.require_connected()?;
872            Ok(OperationOutcome::with_events(vec![
873                ClientEvent::ReadStateChanged {
874                    chat_id: request.chat_id,
875                },
876            ]))
877        })
878    }
879
880    fn typing(
881        &self,
882        request: TypingRequest,
883    ) -> BoxFuture<'static, BackendResult<OperationOutcome>> {
884        let backend = self.clone();
885        Box::pin(async move {
886            backend.require_connected()?;
887            Ok(OperationOutcome::with_events(vec![ClientEvent::Typing {
888                chat_id: request.chat_id,
889                user_id: InlineId::new(0),
890                is_typing: request.is_typing,
891            }]))
892        })
893    }
894
895    fn receive_events(&self) -> BoxFuture<'static, BackendResult<Vec<ClientEvent>>> {
896        let backend = self.clone();
897        Box::pin(async move {
898            loop {
899                {
900                    let mut state = backend.state.lock().expect("in-memory backend poisoned");
901                    if !state.connected {
902                        return Err(BackendError::new(
903                            ClientErrorCategory::AuthRequired,
904                            "client is not connected",
905                        ));
906                    }
907                    if let Some(events) = state.event_batches.pop_front() {
908                        return events;
909                    }
910                }
911                tokio::time::sleep(Duration::from_millis(10)).await;
912            }
913        })
914    }
915}
916
917fn parse_cursor(cursor: Option<&str>) -> BackendResult<usize> {
918    match cursor {
919        Some(cursor) if !cursor.trim().is_empty() => cursor.parse::<usize>().map_err(|_| {
920            BackendError::new(
921                ClientErrorCategory::InvalidInput,
922                "invalid pagination cursor",
923            )
924        }),
925        _ => Ok(0),
926    }
927}
928
929fn max_message_id_from_backend(
930    messages: &HashMap<i64, Vec<MessageRecord>>,
931    chat_id: InlineId,
932) -> Option<InlineId> {
933    messages
934        .get(&chat_id.get())?
935        .iter()
936        .map(|message| message.message_id.get())
937        .max()
938        .map(InlineId::new)
939}
940
941fn chat_id_for_peer(peer: crate::PeerRef) -> InlineId {
942    match peer {
943        crate::PeerRef::User { user_id } => user_id,
944        crate::PeerRef::Chat { chat_id } => chat_id,
945        crate::PeerRef::Thread { thread_id } => thread_id,
946    }
947}
948
949#[cfg(test)]
950mod tests {
951    use crate::{PeerRef, SendTextRequest};
952
953    use super::*;
954
955    fn token_connect() -> ConnectRequest {
956        ConnectRequest::new(crate::AuthCredential::AccessToken {
957            token: crate::AuthToken::try_new("token").unwrap(),
958        })
959    }
960
961    #[tokio::test]
962    async fn in_memory_backend_requires_connect_for_dialogs() {
963        let backend = InMemoryBackend::new();
964
965        let err = backend
966            .dialogs(DialogsRequest::default())
967            .await
968            .expect_err("dialogs should require connect");
969
970        assert_eq!(err.category, ClientErrorCategory::AuthRequired);
971    }
972
973    #[tokio::test]
974    async fn in_memory_backend_lists_dialogs_with_cursor() {
975        let backend = InMemoryBackend::new();
976        backend.upsert_dialog(DialogRecord {
977            chat_id: InlineId::new(1),
978            peer_user_id: None,
979            title: Some("one".to_owned()),
980            last_message_id: None,
981            synced_through_message_id: None,
982            unread_count: Some(0),
983        });
984        backend.upsert_dialog(DialogRecord {
985            chat_id: InlineId::new(2),
986            peer_user_id: Some(InlineId::new(3)),
987            title: Some("two".to_owned()),
988            last_message_id: None,
989            synced_through_message_id: None,
990            unread_count: Some(0),
991        });
992        backend.connect(token_connect()).await.unwrap();
993
994        let first = backend
995            .dialogs(DialogsRequest {
996                limit: Some(1),
997                cursor: None,
998            })
999            .await
1000            .unwrap();
1001        assert_eq!(first.dialogs[0].chat_id, InlineId::new(1));
1002        assert_eq!(first.next_cursor.as_deref(), Some("1"));
1003
1004        let second = backend
1005            .dialogs(DialogsRequest {
1006                limit: Some(1),
1007                cursor: first.next_cursor,
1008            })
1009            .await
1010            .unwrap();
1011        assert_eq!(second.dialogs[0].chat_id, InlineId::new(2));
1012        assert_eq!(second.next_cursor, None);
1013    }
1014
1015    #[tokio::test]
1016    async fn in_memory_backend_returns_chat_participants() {
1017        let backend = InMemoryBackend::new();
1018        backend.set_chat_participants(
1019            InlineId::new(7),
1020            vec![ChatParticipantRecord {
1021                user_id: InlineId::new(42),
1022                date: Some(100),
1023            }],
1024        );
1025        backend.connect(token_connect()).await.unwrap();
1026
1027        let page = backend
1028            .chat_participants(ChatParticipantsRequest {
1029                chat_id: InlineId::new(7),
1030            })
1031            .await
1032            .unwrap();
1033
1034        assert_eq!(page.participants.len(), 1);
1035        assert_eq!(page.participants[0].user_id, InlineId::new(42));
1036        assert_eq!(page.participants[0].date, Some(100));
1037    }
1038
1039    #[tokio::test]
1040    async fn in_memory_backend_sends_text_and_records_history() {
1041        let backend = InMemoryBackend::new();
1042        backend.connect(token_connect()).await.unwrap();
1043
1044        let outcome = backend
1045            .send_text(SendTextRequest::new(
1046                PeerRef::Chat {
1047                    chat_id: InlineId::new(7),
1048                },
1049                "hello",
1050            ))
1051            .await
1052            .unwrap();
1053
1054        assert_eq!(outcome.chat_id, InlineId::new(7));
1055        assert_eq!(outcome.message_id, Some(InlineId::new(1)));
1056
1057        let history = backend
1058            .history(HistoryRequest {
1059                chat_id: InlineId::new(7),
1060                limit: Some(10),
1061                before_message_id: None,
1062                after_message_id: None,
1063            })
1064            .await
1065            .unwrap();
1066        assert_eq!(history.messages.len(), 1);
1067        assert_eq!(history.messages[0].message_id, InlineId::new(1));
1068    }
1069
1070    #[tokio::test]
1071    async fn in_memory_backend_rejects_empty_text() {
1072        let backend = InMemoryBackend::new();
1073        backend.connect(token_connect()).await.unwrap();
1074
1075        let err = backend
1076            .send_text(SendTextRequest::new(
1077                PeerRef::Chat {
1078                    chat_id: InlineId::new(7),
1079                },
1080                " ",
1081            ))
1082            .await
1083            .expect_err("empty message should fail");
1084
1085        assert_eq!(err.category, ClientErrorCategory::InvalidInput);
1086    }
1087}
inline_client/backend.rs

inline_client/
backend.rs
