qt-chat-app

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs

network_manager.rs (26939B)


      1 //! CXX-Qt NetworkManager - Full Rust implementation
      2 //!
      3 //! Implements the NetworkManager QObject with:
      4 //! - HTTP client using trailbase-client for type-safe API access
      5 //! - Keychain storage using keyring
      6 //! - SSE for real-time messages using reqwest-eventsource
      7 
      8 use cxx_qt::{CxxQtType, Threading};
      9 use cxx_qt_lib::QString;
     10 use std::pin::Pin;
     11 use std::sync::Arc;
     12 use tokio::sync::Mutex;
     13 
     14 #[cxx_qt::bridge]
     15 mod ffi {
     16     unsafe extern "C++" {
     17         include!("cxx-qt-lib/qstring.h");
     18         type QString = cxx_qt_lib::QString;
     19     }
     20 
     21     unsafe extern "RustQt" {
     22         #[qobject]
     23         #[qproperty(QString, current_user_id)]
     24         #[qproperty(bool, is_authenticated, READ, NOTIFY = authentication_changed)]
     25         type RustNetworkManager = super::RustNetworkManagerRust;
     26 
     27         // ==================== Signals ====================
     28 
     29         #[qsignal]
     30         #[cxx_name = "loginSuccess"]
     31         fn login_success(self: Pin<&mut RustNetworkManager>, auth_token: QString);
     32 
     33         #[qsignal]
     34         #[cxx_name = "loginFailed"]
     35         fn login_failed(self: Pin<&mut RustNetworkManager>, error: QString);
     36 
     37         #[qsignal]
     38         #[cxx_name = "authenticationChanged"]
     39         fn authentication_changed(self: Pin<&mut RustNetworkManager>);
     40 
     41         #[qsignal]
     42         #[cxx_name = "autoLoginSuccess"]
     43         fn auto_login_success(self: Pin<&mut RustNetworkManager>);
     44 
     45         #[qsignal]
     46         #[cxx_name = "autoLoginFailed"]
     47         fn auto_login_failed(self: Pin<&mut RustNetworkManager>, reason: QString);
     48 
     49         #[qsignal]
     50         #[cxx_name = "requestFailed"]
     51         fn request_failed(self: Pin<&mut RustNetworkManager>, error: QString);
     52 
     53         // JSON string signals for QML compatibility
     54         #[qsignal]
     55         #[cxx_name = "chatCreated"]
     56         fn chat_created_json(self: Pin<&mut RustNetworkManager>, chat_json: QString);
     57 
     58         #[qsignal]
     59         #[cxx_name = "chatUpdated"]
     60         fn chat_updated_json(self: Pin<&mut RustNetworkManager>, chat_json: QString);
     61 
     62         #[qsignal]
     63         #[cxx_name = "chatsFetched"]
     64         fn chats_fetched_json(self: Pin<&mut RustNetworkManager>, chats_json: QString, cursor: QString);
     65 
     66         #[qsignal]
     67         #[cxx_name = "messageCreated"]
     68         fn message_created_json(self: Pin<&mut RustNetworkManager>, message_json: QString);
     69 
     70         #[qsignal]
     71         #[cxx_name = "messagesFetched"]
     72         fn messages_fetched_json(self: Pin<&mut RustNetworkManager>, messages_json: QString, cursor: QString);
     73 
     74         #[qsignal]
     75         #[cxx_name = "messageReceived"]
     76         fn message_received_json(self: Pin<&mut RustNetworkManager>, message_json: QString);
     77 
     78         #[qsignal]
     79         #[cxx_name = "sseConnected"]
     80         fn sse_connected(self: Pin<&mut RustNetworkManager>, chat_id: i32);
     81 
     82         #[qsignal]
     83         #[cxx_name = "sseDisconnected"]
     84         fn sse_disconnected(self: Pin<&mut RustNetworkManager>, chat_id: i32, reason: QString);
     85 
     86         #[qsignal]
     87         #[cxx_name = "sseError"]
     88         fn sse_error(self: Pin<&mut RustNetworkManager>, error: QString);
     89 
     90         // ==================== Invokable Methods ====================
     91 
     92         #[qinvokable]
     93         fn login(self: Pin<&mut RustNetworkManager>, email: QString, password: QString);
     94 
     95         #[qinvokable]
     96         fn logout(self: Pin<&mut RustNetworkManager>);
     97 
     98         #[qinvokable]
     99         #[cxx_name = "checkSavedCredentials"]
    100         fn check_saved_credentials(self: Pin<&mut RustNetworkManager>);
    101 
    102         #[qinvokable]
    103         #[cxx_name = "postChat"]
    104         fn post_chat(self: Pin<&mut RustNetworkManager>, name: QString);
    105 
    106         #[qinvokable]
    107         #[cxx_name = "updateChat"]
    108         fn update_chat(self: Pin<&mut RustNetworkManager>, chat_id: i32, title: QString);
    109 
    110         #[qinvokable]
    111         #[cxx_name = "fetchChats"]
    112         fn fetch_chats(self: Pin<&mut RustNetworkManager>, cursor: QString);
    113 
    114         #[qinvokable]
    115         #[cxx_name = "postMessage"]
    116         fn post_message(self: Pin<&mut RustNetworkManager>, chat_id: i32, content: QString);
    117 
    118         #[qinvokable]
    119         #[cxx_name = "fetchMessages"]
    120         fn fetch_messages(self: Pin<&mut RustNetworkManager>, chat_id: i32, cursor: QString);
    121 
    122         #[qinvokable]
    123         #[cxx_name = "subscribeToChat"]
    124         fn subscribe_to_chat(self: Pin<&mut RustNetworkManager>, chat_id: i32);
    125 
    126         #[qinvokable]
    127         #[cxx_name = "subscribeToChats"]
    128         fn subscribe_to_chats(self: Pin<&mut RustNetworkManager>, chat_ids_json: QString);
    129 
    130         #[qinvokable]
    131         #[cxx_name = "unsubscribeFromChat"]
    132         fn unsubscribe_from_chat(self: Pin<&mut RustNetworkManager>);
    133     }
    134 
    135     impl cxx_qt::Threading for RustNetworkManager {}
    136 }
    137 
    138 use crate::http_client;
    139 use crate::keychain;
    140 use crate::message_cache::{DisplayChat, MessageCache};
    141 use crate::sse_client::{SseCommand, SseEvent};
    142 use tokio::runtime::Runtime;
    143 use tokio::sync::mpsc;
    144 use trailbase_client::Client;
    145 
    146 /// Rust implementation of the NetworkManager
    147 pub struct RustNetworkManagerRust {
    148     current_user_id: QString,
    149     is_authenticated: bool,
    150     /// Auth token stored for SSE connections (trailbase-client handles its own token internally)
    151     auth_token: Arc<Mutex<String>>,
    152     runtime: Arc<Runtime>,
    153     /// TrailBase client for type-safe API access (Mutex allows replacing with authenticated client)
    154     client: Arc<Mutex<Client>>,
    155     sse_command_tx: Arc<Mutex<Option<mpsc::Sender<SseCommand>>>>,
    156     /// Message cache with deduplication
    157     message_cache: Arc<Mutex<MessageCache>>,
    158 }
    159 
    160 impl Default for RustNetworkManagerRust {
    161     fn default() -> Self {
    162         // Initialize tracing subscriber (ignore if already initialized)
    163         use tracing_subscriber::{fmt, EnvFilter};
    164         let _ = fmt()
    165             .with_env_filter(EnvFilter::from_default_env())
    166             .try_init();
    167 
    168         let runtime = Runtime::new().expect("Failed to create Tokio runtime");
    169         let client = http_client::create_client();
    170 
    171         Self {
    172             current_user_id: QString::default(),
    173             is_authenticated: false,
    174             auth_token: Arc::new(Mutex::new(String::new())),
    175             runtime: Arc::new(runtime),
    176             client: Arc::new(Mutex::new(client)),
    177             sse_command_tx: Arc::new(Mutex::new(None)),
    178             message_cache: Arc::new(Mutex::new(MessageCache::new())),
    179         }
    180     }
    181 }
    182 
    183 impl ffi::RustNetworkManager {
    184     fn login(mut self: Pin<&mut Self>, email: QString, password: QString) {
    185         let email_str = email.to_string();
    186         let password_str = password.to_string();
    187 
    188         // Validate before making network request
    189         if email_str.trim().is_empty() {
    190             self.as_mut().login_failed(QString::from("Email is required"));
    191             return;
    192         }
    193         if password_str.trim().is_empty() {
    194             self.as_mut().login_failed(QString::from("Password is required"));
    195             return;
    196         }
    197 
    198         let qt_thread = self.qt_thread();
    199         let client = self.rust().client.clone();
    200         let auth_token = self.rust().auth_token.clone();
    201 
    202         self.rust().runtime.spawn(async move {
    203             let client_guard = client.lock().await;
    204             match http_client::login(&client_guard, &email_str, &password_str).await {
    205                 Ok(token) => {
    206                     // Get user ID from the client (extracted from JWT)
    207                     let user_id = client_guard.user().map(|u| u.sub).unwrap_or_default();
    208                     drop(client_guard); // Release lock before storing token
    209 
    210                     // Store token for SSE connections
    211                     {
    212                         let mut auth = auth_token.lock().await;
    213                         *auth = token.clone();
    214                     }
    215 
    216                     // Save to keychain
    217                     if let Err(e) = keychain::save_token(&token) {
    218                         tracing::warn!("Failed to save token to keychain: {}", e);
    219                     }
    220 
    221                     qt_thread
    222                         .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    223                             qobject.as_mut().set_current_user_id(QString::from(&user_id));
    224                             qobject.as_mut().rust_mut().is_authenticated = true;
    225                             qobject.as_mut().authentication_changed();
    226                             qobject.as_mut().login_success(QString::from(&token));
    227                         })
    228                         .unwrap();
    229                 }
    230                 Err(e) => {
    231                     tracing::error!("login error: {:?}", e);
    232                     let error_msg = e.to_string();
    233                     qt_thread
    234                         .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    235                             qobject.as_mut().login_failed(QString::from(&error_msg));
    236                         })
    237                         .unwrap();
    238                 }
    239             }
    240         });
    241     }
    242 
    243     fn logout(mut self: Pin<&mut Self>) {
    244         // Cancel any active SSE subscription
    245         let sse_command_tx = self.as_mut().rust().sse_command_tx.clone();
    246         let auth_token = self.as_mut().rust().auth_token.clone();
    247         let client = self.as_mut().rust().client.clone();
    248         let runtime = self.as_mut().rust().runtime.clone();
    249 
    250         runtime.spawn(async move {
    251             // Cancel SSE
    252             if let Some(tx) = sse_command_tx.lock().await.take() {
    253                 let _ = tx.send(SseCommand::Stop).await;
    254             }
    255 
    256             // Clear auth token
    257             auth_token.lock().await.clear();
    258 
    259             // Logout from the client (clears internal tokens)
    260             {
    261                 let client_guard = client.lock().await;
    262                 let _ = client_guard.logout().await;
    263             }
    264 
    265             // Delete from keychain
    266             if let Err(e) = keychain::delete_token() {
    267                 tracing::warn!("Failed to delete token from keychain: {}", e);
    268             }
    269         });
    270 
    271         self.as_mut().rust_mut().is_authenticated = false;
    272         self.as_mut().authentication_changed();
    273     }
    274 
    275     fn check_saved_credentials(self: Pin<&mut Self>) {
    276         let qt_thread = self.qt_thread();
    277         let auth_token = self.rust().auth_token.clone();
    278         let client = self.rust().client.clone();
    279 
    280         self.rust().runtime.spawn(async move {
    281             match keychain::load_token() {
    282                 Ok(token) if !token.is_empty() => {
    283                     // Replace the client with one that has the token and validate it
    284                     let new_client = http_client::create_client_with_token(&token);
    285                     let user_id = new_client.user().map(|u| u.sub);
    286 
    287                     match user_id {
    288                         Some(user_id) => {
    289                             // Token is valid - store it and update client
    290                             {
    291                                 let mut auth = auth_token.lock().await;
    292                                 *auth = token.clone();
    293                             }
    294                             {
    295                                 let mut client_guard = client.lock().await;
    296                                 *client_guard = new_client;
    297                             }
    298 
    299                             qt_thread
    300                                 .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    301                                     qobject.as_mut().set_current_user_id(QString::from(&user_id));
    302                                     qobject.as_mut().rust_mut().is_authenticated = true;
    303                                     qobject.as_mut().authentication_changed();
    304                                     qobject.as_mut().auto_login_success();
    305                                 })
    306                                 .unwrap();
    307                         }
    308                         None => {
    309                             // Token is invalid/expired - clean it up
    310                             tracing::info!("Saved token is invalid or expired, removing");
    311                             let _ = keychain::delete_token();
    312 
    313                             qt_thread
    314                                 .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    315                                     qobject.as_mut().auto_login_failed(QString::from(
    316                                         "Session expired, please log in again",
    317                                     ));
    318                                 })
    319                                 .unwrap();
    320                         }
    321                     }
    322                 }
    323                 Ok(_) => {
    324                     qt_thread
    325                         .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    326                             qobject.as_mut().auto_login_failed(QString::from("No saved credentials"));
    327                         })
    328                         .unwrap();
    329                 }
    330                 Err(e) => {
    331                     let error_msg = format!("Failed to read saved credentials: {}", e);
    332                     qt_thread
    333                         .queue(move |mut qobject: Pin<&mut ffi::RustNetworkManager>| {
    334                             qobject.as_mut().auto_login_failed(QString::from(&error_msg));
    335                         })
    336                         .unwrap();
    337                 }
    338             }
    339         });
    340     }
    341 
    342     fn post_chat(self: Pin<&mut Self>, name: QString) {
    343         let qt = self.qt_thread();
    344         let client = self.rust().client.clone();
    345         let name_str = name.to_string();
    346 
    347         self.rust().runtime.spawn(async move {
    348             let client_guard = client.lock().await;
    349             match http_client::post_chat(&client_guard, &name_str).await {
    350                 Ok(chat) => {
    351                     // Create chat participation (ignore errors)
    352                     let _ = http_client::create_chat_participation(&client_guard, chat.id).await;
    353 
    354                     let display_chat = DisplayChat::from_chat(&chat);
    355                     let json = serde_json::to_string(&display_chat).unwrap_or_default();
    356                     qt.queue(move |mut q| {
    357                         q.as_mut().chat_created_json(QString::from(&json));
    358                     }).unwrap();
    359                 }
    360                 Err(e) => {
    361                     tracing::error!("post_chat error: {:?}", e);
    362                     let msg = e.to_string();
    363                     qt.queue(move |mut q| {
    364                         q.as_mut().request_failed(QString::from(&msg));
    365                     }).unwrap();
    366                 }
    367             }
    368         });
    369     }
    370 
    371     fn update_chat(self: Pin<&mut Self>, chat_id: i32, title: QString) {
    372         let qt = self.qt_thread();
    373         let client = self.rust().client.clone();
    374         let title_str = title.to_string();
    375 
    376         self.rust().runtime.spawn(async move {
    377             let client_guard = client.lock().await;
    378             match http_client::update_chat(&client_guard, chat_id as i64, &title_str).await {
    379                 Ok(chat) => {
    380                     let display_chat = DisplayChat::from_chat(&chat);
    381                     let json = serde_json::to_string(&display_chat).unwrap_or_default();
    382                     qt.queue(move |mut q| {
    383                         q.as_mut().chat_updated_json(QString::from(&json));
    384                     }).unwrap();
    385                 }
    386                 Err(e) => {
    387                     tracing::error!("update_chat error: {:?}", e);
    388                     let msg = e.to_string();
    389                     qt.queue(move |mut q| {
    390                         q.as_mut().request_failed(QString::from(&msg));
    391                     }).unwrap();
    392                 }
    393             }
    394         });
    395     }
    396 
    397     fn fetch_chats(self: Pin<&mut Self>, cursor: QString) {
    398         let qt = self.qt_thread();
    399         let client = self.rust().client.clone();
    400         let cursor_str = cursor.to_string();
    401 
    402         self.rust().runtime.spawn(async move {
    403             let cursor_opt = if cursor_str.is_empty() { None } else { Some(cursor_str.as_str()) };
    404             let client_guard = client.lock().await;
    405 
    406             match http_client::fetch_chats(&client_guard, cursor_opt).await {
    407                 Ok((chats, next_cursor)) => {
    408                     let display_chats: Vec<DisplayChat> = chats.iter().map(DisplayChat::from_user_chat).collect();
    409                     let json = serde_json::to_string(&display_chats).unwrap_or_default();
    410                     qt.queue(move |mut q| {
    411                         q.as_mut().chats_fetched_json(
    412                             QString::from(&json),
    413                             QString::from(&next_cursor.unwrap_or_default()),
    414                         );
    415                     }).unwrap();
    416                 }
    417                 Err(e) => {
    418                     tracing::error!("fetch_chats error: {:?}", e);
    419                     let msg = e.to_string();
    420                     qt.queue(move |mut q| {
    421                         q.as_mut().request_failed(QString::from(&msg));
    422                     }).unwrap();
    423                 }
    424             }
    425         });
    426     }
    427 
    428     fn post_message(mut self: Pin<&mut Self>, chat_id: i32, content: QString) {
    429         // Validate and trim content
    430         let content_str = content.to_string().trim().to_string();
    431         if content_str.is_empty() {
    432             self.as_mut().request_failed(QString::from("Message cannot be empty"));
    433             return;
    434         }
    435 
    436         let qt = self.qt_thread();
    437         let client = self.rust().client.clone();
    438 
    439         self.rust().runtime.spawn(async move {
    440             let client_guard = client.lock().await;
    441             match http_client::post_message(&client_guard, chat_id as i64, &content_str).await {
    442                 Ok(message) => {
    443                     let json = serde_json::to_string(&message).unwrap_or_default();
    444                     qt.queue(move |mut q| {
    445                         q.as_mut().message_created_json(QString::from(&json));
    446                     }).unwrap();
    447                 }
    448                 Err(e) => {
    449                     tracing::error!("post_message error: {:?}", e);
    450                     let msg = e.to_string();
    451                     qt.queue(move |mut q| {
    452                         q.as_mut().request_failed(QString::from(&msg));
    453                     }).unwrap();
    454                 }
    455             }
    456         });
    457     }
    458 
    459     fn fetch_messages(self: Pin<&mut Self>, chat_id: i32, cursor: QString) {
    460         let qt = self.qt_thread();
    461         let client = self.rust().client.clone();
    462         let message_cache = self.rust().message_cache.clone();
    463         let current_user_id = self.rust().current_user_id.to_string();
    464         let cursor_str = cursor.to_string();
    465         let is_initial_load = cursor_str.is_empty();
    466 
    467         self.rust().runtime.spawn(async move {
    468             let cursor_opt = if is_initial_load { None } else { Some(cursor_str.as_str()) };
    469             let client_guard = client.lock().await;
    470 
    471             match http_client::fetch_messages(&client_guard, chat_id as i64, cursor_opt).await {
    472                 Ok((messages, next_cursor)) => {
    473                     // Transform through cache (handles dedup and adds is_outgoing)
    474                     let display_messages = {
    475                         let mut cache: tokio::sync::MutexGuard<'_, MessageCache> = message_cache.lock().await;
    476                         if is_initial_load {
    477                             cache.set_messages(chat_id as i64, messages, &current_user_id)
    478                         } else {
    479                             cache.prepend_messages(chat_id as i64, messages, &current_user_id)
    480                         }
    481                     };
    482 
    483                     let json = serde_json::to_string(&display_messages).unwrap_or_default();
    484                     qt.queue(move |mut q| {
    485                         q.as_mut().messages_fetched_json(
    486                             QString::from(&json),
    487                             QString::from(&next_cursor.unwrap_or_default()),
    488                         );
    489                     }).unwrap();
    490                 }
    491                 Err(e) => {
    492                     tracing::error!("fetch_messages error: {:?}", e);
    493                     let msg = e.to_string();
    494                     qt.queue(move |mut q| {
    495                         q.as_mut().request_failed(QString::from(&msg));
    496                     }).unwrap();
    497                 }
    498             }
    499         });
    500     }
    501 
    502     fn subscribe_to_chat(self: Pin<&mut Self>, chat_id: i32) {
    503         let qt = self.qt_thread();
    504         let token = self.rust().auth_token.clone();
    505         let sse_command_tx = self.rust().sse_command_tx.clone();
    506         let message_cache = self.rust().message_cache.clone();
    507         let current_user_id = self.rust().current_user_id.to_string();
    508 
    509         // Create channels for SSE client
    510         let (command_tx, command_rx) = mpsc::channel::<SseCommand>(1);
    511         let (event_tx, mut event_rx) = mpsc::channel::<SseEvent>(32);
    512 
    513         self.rust().runtime.spawn(async move {
    514             // Cancel previous subscription if any
    515             {
    516                 let mut cmd_guard = sse_command_tx.lock().await;
    517                 if let Some(old_tx) = cmd_guard.take() {
    518                     let _ = old_tx.send(SseCommand::Stop).await;
    519                 }
    520                 *cmd_guard = Some(command_tx);
    521             }
    522 
    523             let auth_token = token.lock().await.clone();
    524 
    525             // Spawn the SSE client (deduplication is handled inside sse_client)
    526             let sse_handle = tokio::spawn(crate::sse_client::subscribe_to_chat(
    527                 auth_token, chat_id, event_tx, command_rx,
    528             ));
    529 
    530             // Process events from SSE client
    531             while let Some(event) = event_rx.recv().await {
    532                 let qt = qt.clone();
    533                 match event {
    534                     SseEvent::Connected { chat_id } => {
    535                         qt.queue(move |mut qobj| qobj.as_mut().sse_connected(chat_id)).unwrap();
    536                     }
    537                     SseEvent::Message { data } => {
    538                         // Parse Value to Message and transform through cache
    539                         if let Ok(msg) = serde_json::from_value::<crate::generated::Message>(data) {
    540                             let mut cache: tokio::sync::MutexGuard<'_, MessageCache> = message_cache.lock().await;
    541                             if let Some(display_msg) = cache.add_message(msg, &current_user_id) {
    542                                 let json = serde_json::to_string(&display_msg).unwrap_or_default();
    543                                 qt.queue(move |mut qobj| {
    544                                     qobj.as_mut().message_received_json(QString::from(&json));
    545                                 }).unwrap();
    546                             }
    547                             // If None, message was duplicate - don't emit
    548                         }
    549                     }
    550                     SseEvent::Disconnected { chat_id, reason } => {
    551                         qt.queue(move |mut qobj| {
    552                             qobj.as_mut().sse_disconnected(chat_id, QString::from(&reason));
    553                         }).unwrap();
    554                     }
    555                     SseEvent::Error { message } => {
    556                         qt.queue(move |mut qobj| {
    557                             qobj.as_mut().sse_error(QString::from(&message));
    558                         }).unwrap();
    559                     }
    560                 }
    561             }
    562 
    563             let _ = sse_handle.await;
    564         });
    565     }
    566 
    567     fn subscribe_to_chats(self: Pin<&mut Self>, chat_ids_json: QString) {
    568         // Parse JSON array of chat IDs from QML
    569         let chat_ids: Vec<i32> = match serde_json::from_str(&chat_ids_json.to_string()) {
    570             Ok(ids) => ids,
    571             Err(e) => {
    572                 tracing::error!("Failed to parse chat IDs JSON: {}", e);
    573                 return;
    574             }
    575         };
    576 
    577         if chat_ids.is_empty() {
    578             return;
    579         }
    580 
    581         let qt = self.qt_thread();
    582         let token = self.rust().auth_token.clone();
    583         let sse_command_tx = self.rust().sse_command_tx.clone();
    584         let message_cache = self.rust().message_cache.clone();
    585         let current_user_id = self.rust().current_user_id.to_string();
    586 
    587         // Create channels for SSE client
    588         let (command_tx, command_rx) = mpsc::channel::<SseCommand>(1);
    589         let (event_tx, mut event_rx) = mpsc::channel::<SseEvent>(32);
    590 
    591         self.rust().runtime.spawn(async move {
    592             // Cancel previous subscription if any
    593             {
    594                 let mut cmd_guard = sse_command_tx.lock().await;
    595                 if let Some(old_tx) = cmd_guard.take() {
    596                     let _ = old_tx.send(SseCommand::Stop).await;
    597                 }
    598                 *cmd_guard = Some(command_tx);
    599             }
    600 
    601             let auth_token = token.lock().await.clone();
    602 
    603             // Spawn the SSE client for multiple chats
    604             let sse_handle = tokio::spawn(crate::sse_client::subscribe_to_chats(
    605                 auth_token, chat_ids, event_tx, command_rx,
    606             ));
    607 
    608             // Process events from SSE client
    609             while let Some(event) = event_rx.recv().await {
    610                 let qt = qt.clone();
    611                 match event {
    612                     SseEvent::Connected { chat_id } => {
    613                         qt.queue(move |mut qobj| qobj.as_mut().sse_connected(chat_id)).unwrap();
    614                     }
    615                     SseEvent::Message { data } => {
    616                         // Parse Value to Message and transform through cache
    617                         if let Ok(msg) = serde_json::from_value::<crate::generated::Message>(data) {
    618                             let mut cache: tokio::sync::MutexGuard<'_, MessageCache> = message_cache.lock().await;
    619                             if let Some(display_msg) = cache.add_message(msg, &current_user_id) {
    620                                 let json = serde_json::to_string(&display_msg).unwrap_or_default();
    621                                 qt.queue(move |mut qobj| {
    622                                     qobj.as_mut().message_received_json(QString::from(&json));
    623                                 }).unwrap();
    624                             }
    625                         }
    626                     }
    627                     SseEvent::Disconnected { chat_id, reason } => {
    628                         qt.queue(move |mut qobj| {
    629                             qobj.as_mut().sse_disconnected(chat_id, QString::from(&reason));
    630                         }).unwrap();
    631                     }
    632                     SseEvent::Error { message } => {
    633                         qt.queue(move |mut qobj| {
    634                             qobj.as_mut().sse_error(QString::from(&message));
    635                         }).unwrap();
    636                     }
    637                 }
    638             }
    639 
    640             let _ = sse_handle.await;
    641         });
    642     }
    643 
    644     fn unsubscribe_from_chat(self: Pin<&mut Self>) {
    645         let sse_command_tx = self.rust().sse_command_tx.clone();
    646         let runtime = self.rust().runtime.clone();
    647 
    648         runtime.spawn(async move {
    649             if let Some(tx) = sse_command_tx.lock().await.take() {
    650                 let _ = tx.send(SseCommand::Stop).await;
    651             }
    652         });
    653     }
    654 }