(fix): fix connection issues

iota-connection/src/message_handlers.rs

@@ -265,7 +265,11 @@ pub fn handle_delete_app(cv: &CommunicationValue) -> CommunicationValue {
         .with_receiver(sender_id as u64)
 }
 
-fn contact_value(contact: &iota_storage::users::contact::Contact) -> DataValue {
+fn contact_value(
+    contact: &iota_storage::users::contact::Contact,
+    messages: &[chat_files::StoredMessage],
+    storage_owner: i64,
+) -> DataValue {
     let mut fields = vec![(
         DataType::UserId,
         DataValue::SignedNumber(contact.user_id as i128),
@@ -279,6 +283,16 @@ fn contact_value(contact: &iota_storage::users::contact::Contact) -> DataValue {
             DataValue::SignedNumber(last_message_at as i128),
         ));
     }
+    fields.push((
+        DataType::Messages,
+        DataValue::Array(
+            messages
+                .iter()
+                .filter(|message| message.external_user == contact.user_id)
+                .map(|message| stored_message_value(message, storage_owner, contact.user_id))
+                .collect(),
+        ),
+    ));
     typed_container(fields)
 }
 
@@ -368,7 +382,12 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue {
         .add_typed_default(DataType::SyncMode, DataValue::Str(mode.into()))
         .add_typed_default(
             DataType::Contacts,
-            DataValue::Array(contacts.iter().map(contact_value).collect()),
+            DataValue::Array(
+                contacts
+                    .iter()
+                    .map(|contact| contact_value(contact, &messages, user_id))
+                    .collect(),
+            ),
         )
         .add_typed_default(DataType::Messages, DataValue::Array(message_values))
         .add_typed_default(

omikron-connector/src/omikron_connection.rs

@@ -6,7 +6,7 @@ use iota_storage::util::config_util::{CONFIG, modify_config};
 use iota_storage::util::e2ee_storage::{self, PendingChatSecretForward, StoredChatSecret};
 use iota_util::crypto_helper::{self, keyring_from_base64};
 use iota_util::crypto_util::{self};
-use mtp::client::{Client, ClientConfig, Policy, Receiver, SendMode, Sender};
+use mtp::client::{Client, ClientConfig, MTPConnection, Policy, SendMode, Sender};
 use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
 use mtp::crypto::{Keyring, PublicKeyBundle};
 use std::env;
@@ -394,16 +394,16 @@ impl OmikronConnection {
             log!("Registered with Iota-ID: {}", connection.client_id);
         }
 
-        let sender_arc = Arc::new(connection.sender);
+        let sender_arc = Arc::new(connection.sender.clone());
         *self.sender.write().await = Some(sender_arc.clone());
         self.set_state(ConnectionState::Connected { identified: true })
             .await;
 
         // Start read loop
-        let mut receiver = connection.receiver;
+        let connection = Arc::new(connection);
         let read_self = self.clone();
         let read_handle = tokio::spawn(async move {
-            read_self.read_loop(&mut receiver).await;
+            read_self.read_loop(connection).await;
         });
 
         log_t!("omikron_authenticated");
@@ -592,9 +592,9 @@ impl OmikronConnection {
     // Read Loop & Heartbeat
     // -------------------------------------------------------------------------
 
-    async fn read_loop(self: Arc<Self>, receiver: &mut Receiver) {
+    async fn read_loop(self: Arc<Self>, connection: Arc<MTPConnection>) {
         loop {
-            let result = receiver.receive().await;
+            let result = connection.receive().await;
             match result {
                 Ok(cv) => {
                     let msg_id = cv.get_id();
@@ -624,7 +624,7 @@ impl OmikronConnection {
                     break;
                 }
             }
-            if !receiver.is_open() {
+            if !connection.receiver.is_open() {
                 self.fail_all_waiting_tasks(format!(
                     "Connection closed (connection_id={}, receiver_open=false)",
                     self.connection_id