feat: split message to have dedicated to and from fields (#6358)

Description
---
split message to have dedicated to and from fields

Motivation and Context
---
This is needed for #6353 
When a chat client receives a message, it needs to know the complete
address of the receiver, not just a part of it. This allows the user to
get the complete address by saving what the sender said their address is

---------

Co-authored-by: Hansie Odendaal <39146854+hansieodendaal@users.noreply.github.com>

base_layer/chat_ffi/chat.h

@@ -490,6 +490,7 @@ void destroy_conversationalists_vector(struct ConversationalistsVector *ptr);
  * The ```Message``` received should be destroyed after use
  */
 struct Message *create_chat_message(struct TariAddress *receiver,
+                                    struct TariAddress *sender,
                                     const char *message,
                                     int *error_out);
 
@@ -607,7 +608,23 @@ struct ChatByteVector *read_chat_message_body(struct Message *message, int *erro
  * `message` should be destroyed eventually
  * the returned `TariAddress` should be destroyed eventually
  */
-struct TariAddress *read_chat_message_address(struct Message *message, int *error_out);
+struct TariAddress *read_chat_message_sender_address(struct Message *message, int *error_out);
+
+/**
+ * Returns a pointer to a TariAddress
+ *
+ * ## Arguments
+ * `message` - A pointer to a Message
+ * `error_out` - Pointer to an int which will be modified
+ *
+ * ## Returns
+ * `*mut TariAddress` - A ptr to a TariAddress
+ *
+ * ## Safety
+ * `message` should be destroyed eventually
+ * the returned `TariAddress` should be destroyed eventually
+ */
+struct TariAddress *read_chat_message_receiver_address(struct Message *message, int *error_out);
 
 /**
  * Returns a c_uchar representation of the Direction enum

base_layer/chat_ffi/src/callback_handler.rs

@@ -134,7 +134,7 @@ impl CallbackHandler {
         debug!(
             target: LOG_TARGET,
             "Calling MessageReceived callback function for sender {}",
-            message.address,
+            message.sender_address,
         );
 
         unsafe {

base_layer/chat_ffi/src/message.rs

@@ -50,6 +50,7 @@ use crate::{
 #[no_mangle]
 pub unsafe extern "C" fn create_chat_message(
     receiver: *mut TariAddress,
+    sender: *mut TariAddress,
     message: *const c_char,
     error_out: *mut c_int,
 ) -> *mut Message {
@@ -60,6 +61,10 @@ pub unsafe extern "C" fn create_chat_message(
         error = LibChatError::from(InterfaceError::NullError("receiver".to_string())).code;
         ptr::swap(error_out, &mut error as *mut c_int);
     }
+    if sender.is_null() {
+        error = LibChatError::from(InterfaceError::NullError("sender".to_string())).code;
+        ptr::swap(error_out, &mut error as *mut c_int);
+    }
 
     let message_str = match CStr::from_ptr(message).to_str() {
         Ok(str) => str.to_string(),
@@ -71,7 +76,8 @@ pub unsafe extern "C" fn create_chat_message(
     };
 
     let message_out = MessageBuilder::new()
-        .address((*receiver).clone())
+        .receiver_address((*receiver).clone())
+        .sender_address((*sender).clone())
         .message(message_str)
         .build();
 
@@ -306,7 +312,40 @@ pub unsafe extern "C" fn read_chat_message_body(message: *mut Message, error_out
 /// `message` should be destroyed eventually
 /// the returned `TariAddress` should be destroyed eventually
 #[no_mangle]
-pub unsafe extern "C" fn read_chat_message_address(message: *mut Message, error_out: *mut c_int) -> *mut TariAddress {
+pub unsafe extern "C" fn read_chat_message_sender_address(
+    message: *mut Message,
+    error_out: *mut c_int,
+) -> *mut TariAddress {
+    let mut error = 0;
+    ptr::swap(error_out, &mut error as *mut c_int);
+
+    if message.is_null() {
+        error = LibChatError::from(InterfaceError::NullError("message".to_string())).code;
+        ptr::swap(error_out, &mut error as *mut c_int);
+        return ptr::null_mut();
+    }
+
+    let address = (*message).sender_address.clone();
+    Box::into_raw(Box::new(address))
+}
+
+/// Returns a pointer to a TariAddress
+///
+/// ## Arguments
+/// `message` - A pointer to a Message
+/// `error_out` - Pointer to an int which will be modified
+///
+/// ## Returns
+/// `*mut TariAddress` - A ptr to a TariAddress
+///
+/// ## Safety
+/// `message` should be destroyed eventually
+/// the returned `TariAddress` should be destroyed eventually
+#[no_mangle]
+pub unsafe extern "C" fn read_chat_message_receiver_address(
+    message: *mut Message,
+    error_out: *mut c_int,
+) -> *mut TariAddress {
     let mut error = 0;
     ptr::swap(error_out, &mut error as *mut c_int);
 
@@ -316,7 +355,7 @@ pub unsafe extern "C" fn read_chat_message_address(message: *mut Message, error_
         return ptr::null_mut();
     }
 
-    let address = (*message).address.clone();
+    let address = (*message).receiver_address.clone();
     Box::into_raw(Box::new(address))
 }
 
@@ -560,17 +599,22 @@ mod test {
 
     #[test]
     fn test_reading_message_address() {
-        let address =
+        let receiver_address =
             TariAddress::from_hex("0c017c5cd01385f34ac065e3b05948326dc55d2494f120c6f459a07389011b4ec1").unwrap();
-        let message = MessageBuilder::new().address(address.clone()).build();
+        let sender_address =
+            TariAddress::from_hex("3e596f98f6904f0fc1c8685e2274bd8b2c445d5dac284a9398d09a0e9a760436d0").unwrap();
+        let message = MessageBuilder::new()
+            .receiver_address(receiver_address.clone())
+            .sender_address(sender_address.clone())
+            .build();
 
         let message_ptr = Box::into_raw(Box::new(message));
         let error_out = Box::into_raw(Box::new(0));
 
         unsafe {
-            let address_ptr = read_chat_message_address(message_ptr, error_out);
+            let address_ptr = read_chat_message_sender_address(message_ptr, error_out);
 
-            assert_eq!(address.to_bytes(), (*address_ptr).to_bytes());
+            assert_eq!(sender_address.to_bytes(), (*address_ptr).to_bytes());
 
             destroy_chat_message(message_ptr);
             destroy_tari_address(address_ptr);

base_layer/chat_ffi/src/message_metadata.rs

@@ -211,7 +211,8 @@ mod test {
         let message_ptr = Box::into_raw(Box::new(
             MessageBuilder::new()
                 .message("hello".to_string())
-                .address(address)
+                .receiver_address(address.clone())
+                .sender_address(address)
                 .build(),
         ));
         let error_out = Box::into_raw(Box::new(0));

base_layer/contacts/migrations/2024-05-27-145200_add_from_to_fields/up.sql

@@ -0,0 +1,9 @@
+DROP INDEX idx_messages_address;
+
+ALTER TABLE messages DROP address;
+
+ALTER TABLE messages ADD receiver_address BLOB NOT NULL;
+ALTER TABLE messages ADD sender_address BLOB NOT NULL;
+
+CREATE INDEX idx_messages_receiver_address ON messages (receiver_address);
+CREATE INDEX idx_messages_sender_address ON messages (sender_address);

base_layer/contacts/proto/message.proto

@@ -7,9 +7,10 @@ package tari.contacts.chat;
 message Message {
   bytes body = 1;
   repeated MessageMetadata metadata = 2;
-  bytes address = 3;
-  DirectionEnum direction = 4;
-  bytes message_id = 5;
+  bytes receiver_address = 3;
+  bytes sender_address = 4;
+  DirectionEnum direction = 5;
+  bytes message_id = 6;
 }
 
 enum DirectionEnum {

base_layer/contacts/src/chat_client/src/client.rs

@@ -168,6 +168,14 @@ impl ChatClient for Client {
         message
     }
 
+    fn create_message(&self, receiver: &TariAddress, message: String) -> Message {
+        MessageBuilder::new()
+            .receiver_address(receiver.clone())
+            .sender_address(self.address().clone())
+            .message(message)
+            .build()
+    }
+
     async fn check_online_status(&self, address: &TariAddress) -> Result<ContactOnlineStatus, Error> {
         if let Some(mut contacts_service) = self.contacts.clone() {
             let contact = contacts_service.get_contact(address.clone()).await?;
@@ -178,10 +186,6 @@ impl ChatClient for Client {
         Ok(ContactOnlineStatus::Offline)
     }
 
-    fn create_message(&self, receiver: &TariAddress, message: String) -> Message {
-        MessageBuilder::new().address(receiver.clone()).message(message).build()
-    }
-
     async fn get_messages(&self, sender: &TariAddress, limit: u64, page: u64) -> Result<Vec<Message>, Error> {
         let mut messages = vec![];
         if let Some(mut contacts_service) = self.contacts.clone() {
@@ -211,7 +215,7 @@ impl ChatClient for Client {
     async fn send_read_receipt(&self, message: Message) -> Result<(), Error> {
         if let Some(mut contacts_service) = self.contacts.clone() {
             contacts_service
-                .send_read_confirmation(message.address.clone(), message.message_id)
+                .send_read_confirmation(message.receiver_address.clone(), message.message_id)
                 .await?;
         }
 

base_layer/contacts/src/contacts_service/handle.rs

@@ -293,7 +293,10 @@ impl ContactsServiceHandle {
     pub async fn send_message(&mut self, message: Message) -> Result<(), ContactsServiceError> {
         match self
             .request_response_service
-            .call(ContactsServiceRequest::SendMessage(message.address.clone(), message))
+            .call(ContactsServiceRequest::SendMessage(
+                message.receiver_address.clone(),
+                message,
+            ))
             .await??
         {
             ContactsServiceResponse::MessageSent => Ok(()),

base_layer/contacts/src/contacts_service/service.rs

@@ -586,8 +586,10 @@ where T: ContactsBackend + 'static
         message: Message,
         source_public_key: CommsPublicKey,
     ) -> Result<(), ContactsServiceError> {
+        if source_public_key != *message.sender_address.public_key() {
+            return Err(ContactsServiceError::MessageSourceDoesNotMatchOrigin);
+        }
         let our_message = Message {
-            address: TariAddress::from_public_key(&source_public_key, message.address.network()),
             stored_at: EpochTime::now().as_u64(),
             ..message
         };
@@ -618,7 +620,7 @@ where T: ContactsBackend + 'static
         &mut self,
         message: &Message,
     ) -> Result<(), ContactsServiceError> {
-        let address = &message.address;
+        let address = &message.sender_address;
         let confirmation = MessageDispatch::DeliveryConfirmation(Confirmation {
             message_id: message.message_id.clone(),
             timestamp: message.stored_at,

base_layer/contacts/src/contacts_service/storage/types/messages.rs

@@ -40,7 +40,8 @@ use crate::{
 #[diesel(table_name = messages)]
 #[diesel(primary_key(message_id))]
 pub struct MessagesSqlInsert {
-    pub address: Vec<u8>,
+    pub receiver_address: Vec<u8>,
+    pub sender_address: Vec<u8>,
     pub message_id: Vec<u8>,
     pub body: Vec<u8>,
     pub metadata: Vec<u8>,
@@ -53,7 +54,8 @@ pub struct MessagesSqlInsert {
 #[diesel(table_name = messages)]
 #[diesel(primary_key(message_id))]
 pub struct MessagesSql {
-    pub address: Vec<u8>,
+    pub receiver_address: Vec<u8>,
+    pub sender_address: Vec<u8>,
     pub message_id: Vec<u8>,
     pub body: Vec<u8>,
     pub metadata: Vec<u8>,
@@ -83,14 +85,46 @@ impl MessagesSqlInsert {
 
 impl MessagesSql {
     /// Find a particular message by their address, if it exists
+    pub fn find_by_receiver_address(
+        address: &[u8],
+        limit: i64,
+        page: i64,
+        conn: &mut SqliteConnection,
+    ) -> Result<Vec<MessagesSql>, ContactsServiceStorageError> {
+        Ok(messages::table
+            .filter(messages::receiver_address.eq(address))
+            .order(messages::stored_at.desc())
+            .offset(limit * page)
+            .limit(limit)
+            .load::<MessagesSql>(conn)?)
+    }
+
+    pub fn find_by_sender_address(
+        address: &[u8],
+        limit: i64,
+        page: i64,
+        conn: &mut SqliteConnection,
+    ) -> Result<Vec<MessagesSql>, ContactsServiceStorageError> {
+        Ok(messages::table
+            .filter(messages::sender_address.eq(address))
+            .order(messages::stored_at.desc())
+            .offset(limit * page)
+            .limit(limit)
+            .load::<MessagesSql>(conn)?)
+    }
+
     pub fn find_by_address(
         address: &[u8],
         limit: i64,
         page: i64,
         conn: &mut SqliteConnection,
     ) -> Result<Vec<MessagesSql>, ContactsServiceStorageError> {
         Ok(messages::table
-            .filter(messages::address.eq(address))
+            .filter(
+                messages::sender_address
+                    .eq(address)
+                    .or(messages::receiver_address.eq(address)),
+            )
             .order(messages::stored_at.desc())
             .offset(limit * page)
             .limit(limit)
@@ -124,7 +158,11 @@ impl MessagesSql {
     pub fn find_all_conversationlists(
         conn: &mut SqliteConnection,
     ) -> Result<Vec<Vec<u8>>, ContactsServiceStorageError> {
-        Ok(messages::table.select(messages::address).distinct().load(conn)?)
+        Ok(messages::table
+            .select(messages::sender_address)
+            .select(messages::receiver_address)
+            .distinct()
+            .load(conn)?)
     }
 }
 
@@ -134,14 +172,20 @@ impl TryFrom<MessagesSql> for Message {
 
     #[allow(clippy::cast_sign_loss)]
     fn try_from(o: MessagesSql) -> Result<Self, Self::Error> {
-        let address = TariAddress::from_bytes(&o.address).map_err(|_| ContactsServiceStorageError::ConversionError)?;
+        let receiver_address =
+            TariAddress::from_bytes(&o.receiver_address).map_err(|_| ContactsServiceStorageError::ConversionError)?;
+        let sender_address =
+            TariAddress::from_bytes(&o.sender_address).map_err(|_| ContactsServiceStorageError::ConversionError)?;
         let metadata: Vec<MessageMetadata> = serde_json::from_str(
             &String::from_utf8(o.metadata.clone()).map_err(|_| ContactsServiceStorageError::ConversionError)?,
         )
         .map_err(|_| ContactsServiceStorageError::ConversionError)?;
 
         Ok(Self {
-            address,
+            metadata,
+            body: o.body,
+            receiver_address,
+            sender_address,
             direction: Direction::from_byte(
                 u8::try_from(o.direction).map_err(|_| ContactsServiceStorageError::ConversionError)?,
             )
@@ -150,8 +194,6 @@ impl TryFrom<MessagesSql> for Message {
             stored_at: o.stored_at.timestamp() as u64,
             delivery_confirmation_at: Some(o.stored_at.timestamp() as u64),
             read_confirmation_at: Some(o.stored_at.timestamp() as u64),
-            body: o.body,
-            metadata,
             message_id: o.message_id,
         })
     }
@@ -164,9 +206,9 @@ impl TryFrom<Message> for MessagesSqlInsert {
 
     fn try_from(o: Message) -> Result<Self, Self::Error> {
         let metadata = serde_json::to_string(&o.metadata).map_err(|_| ContactsServiceStorageError::ConversionError)?;
-
         Ok(Self {
-            address: o.address.to_bytes().to_vec(),
+            receiver_address: o.receiver_address.to_bytes().to_vec(),
+            sender_address: o.sender_address.to_bytes().to_vec(),
             message_id: o.message_id,
             body: o.body,
             metadata: metadata.into_bytes().to_vec(),