Update Rust crate rand_core to 0.10

iota-connection/src/message_handlers.rs

@@ -115,6 +115,9 @@ fn validate_relay_identity(
 pub fn apply_verified_relay_content(
     context: &VerifiedRelayContext,
     content: &VerifiedRelayContent,
+    accepted_at: i64,
+    storage_owner: i64,
+    sent_by_self: bool,
 ) -> Result<(), String> {
     validate_relay_identity(context, &content.content)?;
     let sender_id = i64::try_from(context.signer_id)
@@ -125,8 +128,48 @@ pub fn apply_verified_relay_content(
         .map_err(|_| "Relay creation time exceeds the local storage range".to_string())?;
 
     match content.message_type.as_str() {
+        "MessageState" => {
+            let partner_id = relay_number(
+                &content.content,
+                DataType::ChatPartnerId,
+                &context.type_map,
+            )
+            .and_then(|value| i64::try_from(value).ok())
+            .filter(|id| *id == recipient_id)
+            .ok_or_else(|| "Relay MessageState has an invalid ChatPartnerId".to_string())?;
+            let relay_message_id = relay_string(
+                &content.content,
+                DataType::RelayMessageId,
+                &context.type_map,
+            )
+            .ok_or_else(|| "Relay MessageState is missing RelayMessageId".to_string())?;
+            let event_at = relay_number(&content.content, DataType::EventAt, &context.type_map)
+                .and_then(|value| i64::try_from(value).ok())
+                .ok_or_else(|| "Relay MessageState is missing EventAt".to_string())?;
+            let state = relay_string(
+                &content.content,
+                DataType::MessageState,
+                &context.type_map,
+            )
+            .map(MessageState::from_str)
+            .filter(|state| matches!(state, MessageState::Received | MessageState::Read))
+            .ok_or_else(|| "Relay MessageState has an invalid state".to_string())?;
+            chat_files::record_message_receipt(
+                storage_owner,
+                recipient_id,
+                relay_message_id,
+                sender_id,
+                &context.message_id,
+                state,
+                event_at,
+                now_millis_i64(),
+            )
+            .map_err(|error| error.to_string())?;
+            let _ = partner_id;
+            Ok(())
+        }
         "MessageSend" => {
-            let message = relay_string(&content.content, DataType::AppContent, &context.type_map)
+            let message = relay_string(&content.content, DataType::Content, &context.type_map)
                 .ok_or_else(|| "Relay MessageSend is missing Content".to_string())?;
             let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map)
                 .and_then(|value| i64::try_from(value).ok())
@@ -136,20 +179,35 @@ pub fn apply_verified_relay_content(
                 .unwrap_or_default();
             let reply_to = relay_number(&content.content, DataType::ReplyId, &context.type_map)
                 .and_then(|value| i64::try_from(value).ok());
-            chat_files::add_message(
-                u128::try_from(send_time)
-                    .map_err(|_| "Relay MessageSend has a negative SendTime".to_string())?,
-                false,
-                recipient_id,
-                sender_id,
-                message,
+            let relay_message_id = relay_string(
+                &content.content,
+                DataType::RelayMessageId,
+                &context.type_map,
+            )
+            .ok_or_else(|| "Relay MessageSend is missing RelayMessageId".to_string())?;
+            if relay_message_id != context.message_id {
+                return Err("Relay MessageSend identity does not match its protected message ID".into());
+            }
+            chat_files::add_message(chat_files::NewMessage {
+                relay_signer_id: sender_id,
+                relay_message_id,
+                authored_at: created_at,
+                send_time,
+                storage_owner,
+                external_user: if sent_by_self { recipient_id } else { sender_id },
+                sent_by_self,
+                content: message,
                 height,
                 reply_to,
-            );
+                origin_iota_received_at: sent_by_self.then_some(accepted_at),
+                destination_iota_received_at: (!sent_by_self).then_some(accepted_at),
+                initial_state: MessageState::Sent,
+            })
+            .map_err(|error| error.to_string())?;
             Ok(())
         }
         "MessageEdit" => {
-            let message = relay_string(&content.content, DataType::AppContent, &context.type_map)
+            let message = relay_string(&content.content, DataType::Content, &context.type_map)
                 .ok_or_else(|| "Relay MessageEdit is missing Content".to_string())?;
             let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map)
                 .and_then(|value| i64::try_from(value).ok())
@@ -217,7 +275,7 @@ pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue {
         Ok(mutation) => mutation,
         Err(response) => return response,
     };
-    let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
+    let Some(content) = cv.get_data(DataType::Content).as_str() else {
         return error_response(cv, CommunicationType::ErrorInvalidData);
     };
 
@@ -265,6 +323,9 @@ pub fn handle_message_reaction(cv: &CommunicationValue, add: bool) -> Communicat
 
     match result {
         Ok(()) => success_response(cv),
+        Err(iota_storage::storage_error::StorageError::ReactionLimitReached) => {
+            error_response(cv, CommunicationType::ErrorInvalidData)
+        }
         Err(_) => error_response(cv, CommunicationType::ErrorNotFound),
     }
 }
@@ -288,7 +349,7 @@ fn stored_message_fields(
 ) -> Vec<(DataType, DataValue)> {
     let mut fields = vec![
         (
-            DataType::AppMessageId,
+            DataType::MessageId,
             DataValue::SignedNumber(message.id as i128),
         ),
         (
@@ -296,7 +357,7 @@ fn stored_message_fields(
             DataValue::SignedNumber(message.message_time as i128),
         ),
         (
-            DataType::AppContent,
+            DataType::Content,
             DataValue::Str(message.content.clone()),
         ),
         (
@@ -316,6 +377,25 @@ fn stored_message_fields(
     if let Ok(sender_id) = u128::try_from(sender_id) {
         fields.push((DataType::SenderId, DataValue::UnsignedNumber(sender_id)));
     }
+    if let Some(relay_message_id) = &message.relay_message_id {
+        fields.push((
+            DataType::RelayMessageId,
+            DataValue::Str(relay_message_id.clone()),
+        ));
+    }
+    for (data_type, timestamp) in [
+        (DataType::AuthoredAt, message.authored_at),
+        (DataType::OriginIotaReceivedAt, message.origin_iota_received_at),
+        (DataType::DestinationIotaReceivedAt, message.destination_iota_received_at),
+        (DataType::ClientReceivedAt, message.client_received_at),
+        (DataType::ClientReceivedRecordedAt, message.client_received_recorded_at),
+        (DataType::ReadAt, message.read_at),
+        (DataType::ReadRecordedAt, message.read_recorded_at),
+    ] {
+        if let Some(timestamp) = timestamp {
+            fields.push((data_type, DataValue::SignedNumber(timestamp.into())));
+        }
+    }
     if message.edited {
         fields.push((DataType::Edited, DataValue::Bool(true)));
     }
@@ -398,7 +478,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue {
                 DataValue::Str(record.wrapping_scheme),
             )
             .add_typed_default(
-                DataType::AppCreatedAt,
+                DataType::CreatedAt,
                 DataValue::SignedNumber(record.created_at as i128),
             )
             .add_typed_default(
@@ -760,10 +840,18 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue {
         Err(response) => return response,
     };
 
-    let message = match chat_files::get_message(owner, send_time, partner_id) {
-        Ok(Some(message)) => message,
-        Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
-        Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
+    let (message, offset) = match partner_id {
+        Some(partner_id) => match chat_files::get_message_with_offset(owner, partner_id, send_time)
+        {
+            Ok(Some((message, offset))) => (message, Some(offset)),
+            Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
+            Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
+        },
+        None => match chat_files::get_message(owner, send_time, None) {
+            Ok(Some(message)) => (message, None),
+            Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
+            Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
+        },
     };
 
     let mut response = CommunicationValue::new(CommunicationType::MessageGet)
@@ -772,6 +860,10 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue {
     for (data_type, value) in stored_message_fields(&message, owner, message.external_user) {
         response = response.add_typed_default(data_type, value);
     }
+    if let Some(offset) = offset {
+        response =
+            response.add_typed_default(DataType::Offset, DataValue::SignedNumber(offset as i128));
+    }
     response
 }
 

iota-connection/src/relay.rs

@@ -2,8 +2,6 @@ use iota_util::route_target::RouteTarget;
 use mtp::codec::{
     CommunicationValue, ProtectionPolicy, RelayError, RelayOpenOptions, SignaturePolicy, TypeMap,
     VerifiedRelayContent, VerifiedRelayMetadata, forward_relay_frame,
-    open_relay_content_with_keyrings, open_relay_metadata_with_without_replay,
-    relay_metadata_claimed_signer_id,
     open_relay_content_with_limits_without_replay, open_relay_metadata_with_without_replay,
     relay_metadata_claimed_signer_id_with_options,
 };

iota-state/Cargo.toml

@@ -12,8 +12,4 @@ dashmap = "6.1.0"
 once_cell = "1.21.3"
 tokio = { version = "1.50.0", features = ["full"] }
 json = "*"
-<<<<<<< HEAD
 sysinfo = "0.38.0"
-=======
-sysinfo = "0.39.0"
->>>>>>> refs/remotes/origin/main

iota-storage/Cargo.toml

@@ -7,10 +7,6 @@ edition = "2024"
 iota-logger = { path = "../iota-logger" }
 iota-util = { path = "../iota-util" }
 iota-paths = { path = "../iota-paths" }
-<<<<<<< HEAD
-
-=======
->>>>>>> refs/remotes/origin/main
 base64 = "0.22.1"
 json = "*"
 arc-swap = "1"

iota-storage/src/storage_error.rs

@@ -8,6 +8,8 @@ pub enum StorageError {
     Pool(String),
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
+    #[error("message has reached the unique reaction limit")]
+    ReactionLimitReached,
     #[error("{0}")]
     Other(String),
 }

iota-storage/src/util/chat_files.rs

@@ -2,7 +2,9 @@ use crate::storage_error::StorageError;
 use crate::util::db;
 use crate::util::sync::{self, EntityType, Operation};
 use iota_logger::log;
-use rusqlite::params;
+use rusqlite::{OptionalExtension, params};
+
+pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10;
 
 #[derive(PartialEq, Debug, Clone)]
 pub enum MessageState {
@@ -48,7 +50,16 @@ impl MessageState {
 pub struct StoredMessage {
     pub id: i64,
     pub external_user: i64,
+    pub relay_signer_id: Option<i64>,
+    pub relay_message_id: Option<String>,
     pub message_time: i64,
+    pub authored_at: Option<i64>,
+    pub origin_iota_received_at: Option<i64>,
+    pub destination_iota_received_at: Option<i64>,
+    pub client_received_at: Option<i64>,
+    pub client_received_recorded_at: Option<i64>,
+    pub read_at: Option<i64>,
+    pub read_recorded_at: Option<i64>,
     pub content: String,
     pub edited: bool,
     pub sent_by_self: bool,
@@ -58,6 +69,22 @@ pub struct StoredMessage {
     pub reactions: Vec<StoredReaction>,
 }
 
+pub struct NewMessage<'a> {
+    pub relay_signer_id: i64,
+    pub relay_message_id: &'a str,
+    pub authored_at: i64,
+    pub send_time: i64,
+    pub storage_owner: i64,
+    pub external_user: i64,
+    pub sent_by_self: bool,
+    pub content: &'a str,
+    pub height: i64,
+    pub reply_to: Option<i64>,
+    pub origin_iota_received_at: Option<i64>,
+    pub destination_iota_received_at: Option<i64>,
+    pub initial_state: MessageState,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct StoredReaction {
     pub reaction: String,
@@ -149,7 +176,7 @@ fn update_message_content(
 
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
-            .unwrap()
+            .unwrap_or_default()
             .as_millis() as i64;
 
         let tx = conn.unchecked_transaction()?;
@@ -351,8 +378,8 @@ pub fn add_reaction(
     user_id: i64,
     reaction: &str,
 ) -> Result<(), StorageError> {
-    db::with_db(|conn| {
-        let msg_id: i64 = conn.query_row(
+    db::with_immediate_transaction(|tx| {
+        let msg_id: i64 = tx.query_row(
             r#"
             SELECT id FROM messages
             WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
@@ -362,28 +389,46 @@ pub fn add_reaction(
             |row| row.get(0),
         )?;
 
+        let reaction_exists: bool = tx.query_row(
+            "SELECT EXISTS(SELECT 1 FROM reactions WHERE message_id = ?1 AND reaction = ?2)",
+            params![msg_id, reaction],
+            |row| row.get(0),
+        )?;
+        if !reaction_exists {
+            let unique_reactions: i64 = tx.query_row(
+                "SELECT COUNT(DISTINCT reaction) FROM reactions WHERE message_id = ?1",
+                [msg_id],
+                |row| row.get(0),
+            )?;
+            if unique_reactions >= MAX_UNIQUE_REACTIONS_PER_MESSAGE as i64 {
+                return Err(StorageError::ReactionLimitReached);
+            }
+        }
+
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
-            .unwrap()
+            .unwrap_or_default()
             .as_millis() as i64;
 
-        let tx = conn.unchecked_transaction()?;
-        tx.execute(
+        let inserted = tx.execute(
             r#"
             INSERT OR IGNORE INTO reactions (message_id, user_id, reaction, created_at)
             VALUES (?1, ?2, ?3, ?4)
             "#,
             params![msg_id, user_id, reaction, now],
         )?;
-        sync::record_event(
-            &tx,
-            storage_owner,
-            EntityType::Message,
-            msg_id,
-            Operation::Upsert,
-        )?;
-        tx.commit()?;
-        Ok(())
+        if inserted > 0 {
+            sync::record_event(
+                tx,
+                storage_owner,
+                EntityType::Message,
+                msg_id,
+                Operation::Upsert,
+            )?;
+            Ok(())
+        } else {
+            Ok(())
+        }
     })
 }
 
@@ -422,45 +467,46 @@ pub fn remove_reaction(
     })
 }
 
-pub fn add_message(
-    send_time: u128,
-    storage_owner_is_sender: bool,
-    storage_owner: i64,
-    external_user: i64,
-    message: &str,
-    height: i64,
-    reply_to: Option<i64>,
-) {
-    let message_time = match i64::try_from(send_time) {
-        Ok(v) => v,
-        Err(_) => {
-            log!("Failed to store message: send_time out of range for i64 ({send_time})");
-            return;
-        }
-    };
-
-    if let Err(e) = db::with_db(|conn| {
+pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
+    let NewMessage {
+        relay_signer_id,
+        relay_message_id,
+        authored_at,
+        send_time,
+        storage_owner,
+        external_user,
+        sent_by_self,
+        content,
+        height,
+        reply_to,
+        origin_iota_received_at,
+        destination_iota_received_at,
+        initial_state,
+    } = message;
+    let msg_id = db::with_db(|conn| {
         let tx = conn.unchecked_transaction()?;
         tx.execute(
             r#"
             INSERT INTO messages (
-                storage_owner, external_user, message_time, content,
-                sent_by_self, message_state, height, reply_to
-            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+                storage_owner, external_user, message_time, content, sent_by_self,
+                message_state, height, reply_to, relay_signer_id, relay_message_id,
+                authored_at, origin_iota_received_at, destination_iota_received_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
             "#,
             params![
                 storage_owner,
                 external_user,
-                message_time,
-                message,
-                if storage_owner_is_sender {
-                    1_i64
-                } else {
-                    0_i64
-                },
-                MessageState::Sending.as_str(),
+                send_time,
+                content,
+                i64::from(sent_by_self),
+                initial_state.as_str(),
                 height,
                 reply_to,
+                relay_signer_id,
+                relay_message_id,
+                authored_at,
+                origin_iota_received_at,
+                destination_iota_received_at,
             ],
         )?;
         let msg_id = tx.last_insert_rowid();
@@ -472,15 +518,122 @@ pub fn add_message(
             Operation::Upsert,
         )?;
         tx.commit()?;
-        Ok(())
-    }) {
-        log!("Failed to insert message into sqlite: {}", e);
-        return;
-    }
+        Ok(msg_id)
+    })?;
 
     let mut contact = crate::users::contact::Contact::new(external_user);
-    contact.set_last_message_at(message_time);
+    contact.set_last_message_at(send_time);
     crate::util::chats_util::mod_user(storage_owner, &contact);
+    Ok(msg_id)
+}
+
+pub fn change_message_state_by_relay_id(
+    storage_owner: i64,
+    relay_signer_id: i64,
+    relay_message_id: &str,
+    new_state: MessageState,
+) -> Result<(), StorageError> {
+    db::with_db(|conn| {
+        let tx = conn.unchecked_transaction()?;
+        let Some((msg_id, current)) = tx
+            .query_row(
+                "SELECT id, message_state FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
+                params![storage_owner, relay_signer_id, relay_message_id],
+                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
+            )
+            .optional()?
+        else {
+            return Ok(());
+        };
+        let state = MessageState::from_str(&current).upgrade(new_state).as_str();
+        tx.execute("UPDATE messages SET message_state = ?1 WHERE id = ?2", params![state, msg_id])?;
+        sync::record_event(&tx, storage_owner, EntityType::Message, msg_id, Operation::Upsert)?;
+        tx.commit()?;
+        Ok(())
+    })
+}
+
+pub fn record_message_receipt(
+    storage_owner: i64,
+    target_signer_id: i64,
+    target_message_id: &str,
+    receipt_signer_id: i64,
+    receipt_message_id: &str,
+    receipt_type: MessageState,
+    event_at: i64,
+    recorded_at: i64,
+) -> Result<(), StorageError> {
+    let receipt_type = match receipt_type {
+        MessageState::Received => "received",
+        MessageState::Read => "read",
+        _ => return Err(StorageError::Other("invalid message receipt state".into())),
+    };
+    db::with_db(|conn| {
+        let tx = conn.unchecked_transaction()?;
+        let Some((message_id, external_user)) = tx
+            .query_row(
+                "SELECT id, external_user FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
+                params![storage_owner, target_signer_id, target_message_id],
+                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
+            )
+            .optional()?
+        else {
+            return Err(StorageError::Other("message receipt target was not found".into()));
+        };
+        if external_user != receipt_signer_id {
+            return Err(StorageError::Other("message receipt signer is not the chat partner".into()));
+        }
+        tx.execute(
+            "INSERT OR IGNORE INTO message_receipts (storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+            params![storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at],
+        )?;
+        let (state_column, recorded_column) = if receipt_type == "read" {
+            ("read_at", "read_recorded_at")
+        } else {
+            ("client_received_at", "client_received_recorded_at")
+        };
+        let state = MessageState::from_str(
+            &tx.query_row("SELECT message_state FROM messages WHERE id = ?1", [message_id], |row| row.get::<_, String>(0))?,
+        )
+        .upgrade(MessageState::from_str(receipt_type))
+        .as_str()
+        .to_string();
+        tx.execute(
+            &format!("UPDATE messages SET {state_column} = COALESCE({state_column}, ?1), {recorded_column} = COALESCE({recorded_column}, ?2), message_state = ?3 WHERE id = ?4"),
+            params![event_at, recorded_at, state, message_id],
+        )?;
+        sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
+        tx.commit()?;
+        Ok(())
+    })
+}
+
+pub fn record_destination_iota_received(
+    storage_owner: i64,
+    relay_signer_id: i64,
+    relay_message_id: &str,
+    accepted_at: i64,
+) -> Result<(), StorageError> {
+    db::with_db(|conn| {
+        let tx = conn.unchecked_transaction()?;
+        let Some(message_id) = tx
+            .query_row(
+                "SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
+                params![storage_owner, relay_signer_id, relay_message_id],
+                |row| row.get::<_, i64>(0),
+            )
+            .optional()?
+        else {
+            return Ok(());
+        };
+        tx.execute(
+            "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2",
+            params![accepted_at, message_id],
+        )?;
+        sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
+        tx.commit()?;
+        Ok(())
+    })
 }
 
 pub fn change_message_state(
@@ -574,7 +727,15 @@ fn load_reactions(
             ))
         }) {
             for row in rows.flatten() {
-                map.entry(row.0).or_default().push(row.1);
+                let reactions = map.entry(row.0).or_default();
+                if reactions
+                    .iter()
+                    .any(|stored: &StoredReaction| stored.reaction == row.1.reaction)
+                {
+                    reactions.push(row.1);
+                } else if reactions.len() < MAX_UNIQUE_REACTIONS_PER_MESSAGE {
+                    reactions.push(row.1);
+                }
             }
         }
     }
@@ -594,7 +755,11 @@ pub fn get_messages(
     match db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
-            SELECT id, message_time, content, sent_by_self, message_state, height, reply_to, edited_count
+            SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
+                   origin_iota_received_at, destination_iota_received_at,
+                   client_received_at, client_received_recorded_at, read_at,
+                   read_recorded_at, content, sent_by_self, message_state, height,
+                   reply_to, edited_count
             FROM messages
             WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0
             ORDER BY message_time DESC, id DESC
@@ -608,13 +773,22 @@ pub fn get_messages(
                 Ok(StoredMessage {
                     id: row.get(0)?,
                     external_user,
-                    message_time: row.get(1)?,
-                    content: row.get(2)?,
-                    sent_by_self: row.get::<_, i64>(3)? != 0,
-                    message_state: row.get(4)?,
-                    height: row.get(5).unwrap_or(0),
-                    reply_to: row.get(6).ok().flatten(),
-                    edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
+                    relay_signer_id: row.get(1)?,
+                    relay_message_id: row.get(2)?,
+                    message_time: row.get(3)?,
+                    authored_at: row.get(4)?,
+                    origin_iota_received_at: row.get(5)?,
+                    destination_iota_received_at: row.get(6)?,
+                    client_received_at: row.get(7)?,
+                    client_received_recorded_at: row.get(8)?,
+                    read_at: row.get(9)?,
+                    read_recorded_at: row.get(10)?,
+                    content: row.get(11)?,
+                    sent_by_self: row.get::<_, i64>(12)? != 0,
+                    message_state: row.get(13)?,
+                    height: row.get(14).unwrap_or(0),
+                    reply_to: row.get(15).ok().flatten(),
+                    edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
                     reactions: Vec::new(),
                 })
             },
@@ -652,7 +826,10 @@ pub fn get_message(
     db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
-            SELECT id, message_time, content, sent_by_self, message_state, height,
+            SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
+                   origin_iota_received_at, destination_iota_received_at,
+                   client_received_at, client_received_recorded_at, read_at,
+                   read_recorded_at, content, sent_by_self, message_state, height,
                    reply_to, edited_count, external_user
             FROM messages
             WHERE storage_owner = ?1
@@ -666,14 +843,23 @@ pub fn get_message(
         let rows = stmt.query_map(params![storage_owner, message_time, external_user], |row| {
             Ok(StoredMessage {
                 id: row.get(0)?,
-                message_time: row.get(1)?,
-                content: row.get(2)?,
-                sent_by_self: row.get::<_, i64>(3)? != 0,
-                message_state: row.get(4)?,
-                height: row.get(5).unwrap_or(0),
-                reply_to: row.get(6).ok().flatten(),
-                edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
-                external_user: row.get(8)?,
+                relay_signer_id: row.get(1)?,
+                relay_message_id: row.get(2)?,
+                message_time: row.get(3)?,
+                authored_at: row.get(4)?,
+                origin_iota_received_at: row.get(5)?,
+                destination_iota_received_at: row.get(6)?,
+                client_received_at: row.get(7)?,
+                client_received_recorded_at: row.get(8)?,
+                read_at: row.get(9)?,
+                read_recorded_at: row.get(10)?,
+                content: row.get(11)?,
+                sent_by_self: row.get::<_, i64>(12)? != 0,
+                message_state: row.get(13)?,
+                height: row.get(14).unwrap_or(0),
+                reply_to: row.get(15).ok().flatten(),
+                edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
+                external_user: row.get(17)?,
                 reactions: Vec::new(),
             })
         })?;
@@ -700,6 +886,40 @@ pub fn get_message(
     })
 }
 
+pub fn get_message_with_offset(
+    storage_owner: i64,
+    external_user: i64,
+    message_time: i64,
+) -> Result<Option<(StoredMessage, i64)>, StorageError> {
+    let Some(message) = get_message(storage_owner, message_time, Some(external_user))? else {
+        return Ok(None);
+    };
+    let offset = db::with_db(|conn| {
+        conn.query_row(
+            r#"
+            SELECT COUNT(*)
+            FROM messages
+            WHERE storage_owner = ?1
+              AND external_user = ?2
+              AND deleted_by_external = 0
+              AND (
+                   message_time > ?3
+                   OR (message_time = ?3 AND id > ?4)
+              )
+            "#,
+            params![
+                storage_owner,
+                external_user,
+                message.message_time,
+                message.id
+            ],
+            |row| row.get(0),
+        )
+        .map_err(StorageError::from)
+    })?;
+    Ok(Some((message, offset)))
+}
+
 pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage> {
     if ids.is_empty() {
         return Vec::new();
@@ -708,19 +928,28 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage
     // A journal id uniquely identifies a row. Load all messages for this owner and retain only
     // those ids; this keeps reaction hydration identical to normal message loading.
     match db::with_db(|conn| {
-        let mut stmt = conn.prepare("SELECT id, message_time, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0")?;
+        let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0")?;
         let rows = stmt.query_map([storage_owner], |row| {
-            let external_user: i64 = row.get(8)?;
+            let external_user: i64 = row.get(17)?;
             Ok(StoredMessage {
                 id: row.get(0)?,
                 external_user,
-                message_time: row.get(1)?,
-                content: row.get(2)?,
-                sent_by_self: row.get::<_, i64>(3)? != 0,
-                message_state: row.get(4)?,
-                height: row.get(5).unwrap_or(0),
-                reply_to: row.get(6).ok().flatten(),
-                edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
+                relay_signer_id: row.get(1)?,
+                relay_message_id: row.get(2)?,
+                message_time: row.get(3)?,
+                authored_at: row.get(4)?,
+                origin_iota_received_at: row.get(5)?,
+                destination_iota_received_at: row.get(6)?,
+                client_received_at: row.get(7)?,
+                client_received_recorded_at: row.get(8)?,
+                read_at: row.get(9)?,
+                read_recorded_at: row.get(10)?,
+                content: row.get(11)?,
+                sent_by_self: row.get::<_, i64>(12)? != 0,
+                message_state: row.get(13)?,
+                height: row.get(14).unwrap_or(0),
+                reply_to: row.get(15).ok().flatten(),
+                edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
                 reactions: Vec::new(),
             })
         })?;

iota-storage/src/util/db.rs

@@ -1,6 +1,6 @@
 use once_cell::sync::Lazy;
 use r2d2::ManageConnection;
-use rusqlite::Connection;
+use rusqlite::{Connection, Transaction};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
@@ -55,6 +55,17 @@ where
     f(&conn)
 }
 
+pub fn with_immediate_transaction<T, F>(f: F) -> Result<T, StorageError>
+where
+    F: FnOnce(&Transaction<'_>) -> Result<T, StorageError>,
+{
+    let mut conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?;
+    let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+    let value = f(&tx)?;
+    tx.commit()?;
+    Ok(value)
+}
+
 /* Verify the persistent database before the pool is initialized. A corrupt
  * database is moved aside rather than opened again, preserving material for
  * operator recovery while allowing the daemon to report the failed storage. */
@@ -388,6 +399,71 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
         )?;
     }
 
+    if current_version < 10 {
+        conn.execute_batch(
+            r#"
+            CREATE INDEX IF NOT EXISTS idx_messages_history
+                ON messages (
+                    storage_owner,
+                    external_user,
+                    deleted_by_external,
+                    message_time DESC,
+                    id DESC
+                );
+            PRAGMA user_version = 10;
+            "#,
+        )?;
+    }
+
+    if current_version < 11 {
+        for (column, definition) in [
+            ("relay_signer_id", "relay_signer_id INTEGER"),
+            ("relay_message_id", "relay_message_id TEXT"),
+            ("authored_at", "authored_at INTEGER"),
+            ("origin_iota_received_at", "origin_iota_received_at INTEGER"),
+            ("destination_iota_received_at", "destination_iota_received_at INTEGER"),
+            ("client_received_at", "client_received_at INTEGER"),
+            ("client_received_recorded_at", "client_received_recorded_at INTEGER"),
+            ("read_at", "read_at INTEGER"),
+            ("read_recorded_at", "read_recorded_at INTEGER"),
+        ] {
+            add_column_if_missing(conn, column, definition)?;
+        }
+        for (column, definition) in [
+            ("accepted_at", "accepted_at INTEGER"),
+            ("applied_at", "applied_at INTEGER"),
+            ("queued_at", "queued_at INTEGER"),
+            ("downstream_acked_at", "downstream_acked_at INTEGER"),
+            ("rejected_at", "rejected_at INTEGER"),
+        ] {
+            add_table_column_if_missing(conn, "relay_inbox", column, definition)?;
+        }
+        conn.execute_batch(
+            r#"
+            CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_relay_identity
+                ON messages (storage_owner, relay_signer_id, relay_message_id)
+                WHERE relay_message_id IS NOT NULL;
+            CREATE TABLE IF NOT EXISTS message_receipts (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                storage_owner INTEGER NOT NULL,
+                target_signer_id INTEGER NOT NULL,
+                target_message_id TEXT NOT NULL,
+                receipt_signer_id INTEGER NOT NULL,
+                receipt_message_id TEXT NOT NULL,
+                receipt_type TEXT NOT NULL CHECK (receipt_type IN ('received', 'read')),
+                event_at INTEGER NOT NULL,
+                recorded_at INTEGER NOT NULL,
+                UNIQUE(receipt_signer_id, receipt_message_id),
+                UNIQUE(storage_owner, target_signer_id, target_message_id,
+                       receipt_signer_id, receipt_type)
+            );
+            CREATE INDEX IF NOT EXISTS idx_message_receipts_target
+                ON message_receipts (storage_owner, target_signer_id, target_message_id);
+            PRAGMA user_version = 11;
+            "#,
+        )?;
+    }
+
     Ok(())
 }
 
@@ -457,7 +533,7 @@ mod tests {
         run_migrations_on_connection(&conn)?;
 
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 9);
+        assert_eq!(version, 11);
         for column in ["height", "reply_to", "edited_count", "deleted_by_external"] {
             let mut statement =
                 conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
@@ -470,11 +546,13 @@ mod tests {
     #[test]
     fn migrates_version_five_once_and_is_idempotent() -> Result<(), StorageError> {
         let conn = Connection::open_in_memory()?;
-        conn.execute_batch("PRAGMA user_version = 5;")?;
+        conn.execute_batch(
+            "CREATE TABLE messages (id INTEGER PRIMARY KEY, storage_owner INTEGER NOT NULL, external_user INTEGER NOT NULL, message_time INTEGER NOT NULL, content TEXT NOT NULL, sent_by_self INTEGER NOT NULL, message_state TEXT NOT NULL, height INTEGER NOT NULL DEFAULT 0, reply_to INTEGER, edited_count INTEGER NOT NULL DEFAULT 0, deleted_by_external INTEGER NOT NULL DEFAULT 0); PRAGMA user_version = 5;",
+        )?;
         run_migrations_on_connection(&conn)?;
         run_migrations_on_connection(&conn)?;
         let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
-        assert_eq!(version, 9);
+        assert_eq!(version, 11);
         for table in [
             "sync_heads",
             "sync_events",

iota-storage/src/util/relay_replay.rs

@@ -12,6 +12,7 @@ pub fn reserve(
     signer_id: u64,
     message_id: &str,
     created_at: u64,
+    accepted_at: i64,
     destination_id: u64,
     frame: &[u8],
     frame_id: u32,
@@ -26,11 +27,12 @@ pub fn reserve(
 
     db::with_db(|connection| {
         let inserted = connection.execute(
-            "INSERT OR IGNORE INTO relay_inbox (signer_id, message_id, created_at, destination_id, frame, frame_id, type_map_version, state) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'received')",
+            "INSERT OR IGNORE INTO relay_inbox (signer_id, message_id, created_at, accepted_at, destination_id, frame, frame_id, type_map_version, state) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'received')",
             params![
                 signer_id,
                 message_id,
                 created_at,
+                accepted_at,
                 destination_id,
                 frame,
                 i64::from(frame_id),
@@ -67,41 +69,64 @@ pub fn mark_delivered_for_frame(destination_id: u64, frame_id: u32) -> Result<()
     let destination_id = i64::try_from(destination_id)
         .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
     db::with_db(|connection| {
+        let delivered_at = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis() as i64;
         connection.execute(
-            "UPDATE relay_inbox SET state = 'delivered' WHERE destination_id = ?1 AND frame_id = ?2",
-            params![destination_id, i64::from(frame_id)],
+            "UPDATE relay_inbox SET state = 'delivered', downstream_acked_at = COALESCE(downstream_acked_at, ?3) WHERE destination_id = ?1 AND frame_id = ?2",
+            params![destination_id, i64::from(frame_id), delivered_at],
         )?;
         Ok(())
     })
 }
 
-pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), StorageError> {
-    if !matches!(
-        state,
-        "received" | "applied" | "queued" | "delivered" | "rejected"
-    ) {
-        return Err(StorageError::Other("invalid relay inbox state".into()));
-    }
+fn mark_transition(
+    signer_id: u64,
+    message_id: &str,
+    state: &str,
+    column: &str,
+) -> Result<(), StorageError> {
     let signer_id = i64::try_from(signer_id)
         .map_err(|_| StorageError::Other("relay signer ID exceeds SQLite range".into()))?;
+    let timestamp = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64;
     db::with_db(|connection| {
         connection.execute(
-            "UPDATE relay_inbox SET state = ?3 WHERE signer_id = ?1 AND message_id = ?2",
-            params![signer_id, message_id, state],
+            &format!("UPDATE relay_inbox SET state = ?3, {column} = COALESCE({column}, ?4) WHERE signer_id = ?1 AND message_id = ?2"),
+            params![signer_id, message_id, state, timestamp],
         )?;
         Ok(())
     })
 }
 
-pub fn prune_completed(before_created_at: i64) -> Result<(), StorageError> {
+pub fn mark_applied(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
+    mark_transition(signer_id, message_id, "applied", "applied_at")
+}
+
+pub fn mark_queued(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
+    mark_transition(signer_id, message_id, "queued", "queued_at")
+}
+
+pub fn mark_downstream_acked(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
+    mark_transition(signer_id, message_id, "delivered", "downstream_acked_at")
+}
+
+pub fn mark_rejected(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
+    mark_transition(signer_id, message_id, "rejected", "rejected_at")
+}
+
+pub fn prune_completed(before_terminal_at: i64) -> Result<(), StorageError> {
     db::with_db(|connection| {
         connection.execute(
-            "DELETE FROM relay_inbox WHERE created_at < ?1 AND state IN ('delivered', 'rejected')",
-            params![before_created_at],
+            "DELETE FROM relay_inbox WHERE COALESCE(downstream_acked_at, rejected_at) < ?1 AND state IN ('delivered', 'rejected')",
+            params![before_terminal_at],
         )?;
         connection.execute(
-            "DELETE FROM relay_replay WHERE created_at < ?1",
-            params![before_created_at],
+            "DELETE FROM relay_replay WHERE NOT EXISTS (SELECT 1 FROM relay_inbox WHERE relay_inbox.signer_id = relay_replay.signer_id AND relay_inbox.message_id = relay_replay.message_id)",
+            [],
         )?;
         Ok(())
     })

iota-storage/src/util/sync.rs

@@ -4,7 +4,7 @@ use crate::util::db;
 use rusqlite::{Transaction, params};
 use std::collections::BTreeMap;
 
-pub const CACHE_SCHEMA_VERSION: i64 = 1;
+pub const CACHE_SCHEMA_VERSION: i64 = 2;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EntityType {

iota-updater/Cargo.toml

@@ -5,10 +5,6 @@ edition = "2024"
 
 [dependencies]
 iota-paths = { path = "../iota-paths" }
-<<<<<<< HEAD
-
-=======
->>>>>>> refs/remotes/origin/main
 tokio = { version = "1.50.0", features = ["full"] }
 sha2 = "0.11.0"
 hex = "*"

iota-util/src/crypto_helper.rs

@@ -6,17 +6,10 @@ pub fn generate_keyring() -> Keyring {
 }
 
 pub fn keyring_to_base64(keyring: &Keyring) -> String {
-<<<<<<< HEAD
     keyring
         .try_to_bytes()
         .map(|bytes| STANDARD.encode(bytes))
         .unwrap_or_default()
-=======
-    let bytes = keyring
-        .try_to_bytes()
-        .expect("keyring fields must fit the wire format");
-    STANDARD.encode(bytes)
->>>>>>> refs/remotes/origin/main
 }
 
 pub fn keyring_from_base64(s: &str) -> Option<Keyring> {
@@ -25,17 +18,10 @@ pub fn keyring_from_base64(s: &str) -> Option<Keyring> {
 }
 
 pub fn public_key_bundle_to_base64(bundle: &PublicKeyBundle) -> String {
-<<<<<<< HEAD
     bundle
         .try_as_bytes()
         .map(|bytes| STANDARD.encode(bytes))
         .unwrap_or_default()
-=======
-    let bytes = bundle
-        .try_as_bytes()
-        .expect("public key bundle fields must fit the wire format");
-    STANDARD.encode(bytes)
->>>>>>> refs/remotes/origin/main
 }
 
 pub fn public_key_bundle_from_base64(s: &str) -> Option<PublicKeyBundle> {

mtp-type-maps

@@ -1 +1 @@
-Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e
+Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a

omikron-connector/src/omikron_connection.rs

@@ -68,18 +68,6 @@ fn omikron_public_key_path() -> &'static Path {
         .unwrap_or_else(|| Path::new("omikron.mpkb"))
 }
 
-fn save_keyring(keyring: &Keyring, path: &Path) -> Result<(), String> {
-    let temporary = serialization_path(path)?;
-    mtp::files::save_keyring_raw(keyring, &temporary)
-        .map_err(|error| format!("serialize keyring: {error}"))?;
-    let bytes =
-        std::fs::read(&temporary).map_err(|error| format!("read serialized keyring: {error}"));
-    let _ = std::fs::remove_file(&temporary);
-    let bytes = bytes?;
-    iota_util::atomic_file::replace_private(path, &bytes, 3)
-        .map_err(|error| format!("write {}: {error}", path.display()))
-}
-
 fn save_omikron_public_key(key: &PublicKeyBundle, path: &Path) -> Result<(), String> {
     let temporary = serialization_path(path)?;
     mtp::files::save_public_key_bundle(key, &temporary)
@@ -609,46 +597,8 @@ impl OmikronConnection {
     // Identity (own Keyring, migrated from the legacy base64-in-config format)
     // -------------------------------------------------------------------------
 
-<<<<<<< HEAD
-    /*
-     * `iota.mk` is now the source of truth for this Iota's identity. A
-     * pre-existing base64 keyring in config.json (from before the MTP auth
-     * migration) is imported once so already-registered Iotas keep their
-     * identity, and mirrored back into config.json for older code paths
-     * that still read it directly.
-     */
-    async fn load_or_migrate_keyring(&self) -> Keyring {
-        let path = identity_path();
-        if let Ok(kr) = mtp::files::load_keyring_raw(path) {
-            return kr;
-        }
-
-        let legacy = CONFIG.load().keyring.clone();
-        let keyring = legacy
-            .and_then(|b64| keyring_from_base64(&b64))
-            .unwrap_or_else(|| {
-                log!(
-                    "WARNING: No existing keyring found. Neither {} nor config.json \
-                     contain a keyring; generating a new identity. If you already had \
-                     an Iota identity, restore {} from a backup to avoid losing access.",
-                    path.display(),
-                    path.display()
-                );
-                crypto_helper::generate_keyring()
-            });
-
-        if let Some(parent) = path.parent() {
-            let _ = std::fs::create_dir_all(parent);
-        }
-        if let Err(e) = save_keyring(&keyring, path) {
-            log!("Failed to persist {}: {}", path.display(), e);
-        }
-
-        keyring
-=======
     async fn load_or_migrate_keyring(&self, passphrase: &[u8]) -> Result<Keyring, IdentityError> {
         load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone(), passphrase)
->>>>>>> refs/remotes/origin/main
     }
 
     // -------------------------------------------------------------------------
@@ -721,12 +671,8 @@ impl OmikronConnection {
         };
 
         let (host, port, public_key) = if let Some(endpoint) = discovered {
-            let discovered_key_bytes = endpoint.public_key.try_as_bytes().map_err(|error| {
-                format!("Failed to serialize discovered Omikron public key: {error}")
-            })?;
             match &cached_key {
                 Some(cached) => {
-<<<<<<< HEAD
                     let keys_match =
                         match (cached.try_as_bytes(), endpoint.public_key.try_as_bytes()) {
                             (Ok(cached_bytes), Ok(discovered_bytes)) => {
@@ -752,20 +698,6 @@ impl OmikronConnection {
                     } else {
                         (endpoint.host, endpoint.port, cached.clone())
                     }
-=======
-                    let cached_key_bytes = cached.try_as_bytes().map_err(|error| {
-                        format!("Failed to serialize cached Omikron public key: {error}")
-                    })?;
-                    if cached_key_bytes != discovered_key_bytes {
-                        log!(
-                            "Fetched Omikron public key differs from the cached {} - keeping the \
-                             cached key. Delete {} manually if this is an expected key rotation.",
-                            OMIKRON_PUBLIC_KEY_PATH,
-                            OMIKRON_PUBLIC_KEY_PATH
-                        );
-                    }
-                    (endpoint.host, endpoint.port, cached.clone())
->>>>>>> refs/remotes/origin/main
                 }
                 None => {
                     if let Err(e) = save_omikron_public_key(&endpoint.public_key, key_path) {
@@ -955,6 +887,44 @@ impl OmikronConnection {
         }
     }
 
+    async fn send_relay_success(
+        &self,
+        frame_id: Option<u32>,
+        iota_id: u64,
+        relay_message_id: &str,
+        accepted_at: i64,
+        include_origin_timestamp: bool,
+    ) {
+        let Some(frame_id) = frame_id else { return };
+        let response = CommunicationValue::new(CommunicationType::Success)
+            .with_id(frame_id)
+            .add_typed_default(DataType::IotaId, DataValue::UnsignedNumber(iota_id.into()))
+            .add_typed_default(
+                DataType::RelayMessageId,
+                DataValue::Str(relay_message_id.to_string()),
+            )
+            .add_typed_default(
+                DataType::RelayAcceptedAt,
+                DataValue::SignedNumber(accepted_at.into()),
+            );
+        let response = if include_origin_timestamp {
+            response
+                .add_typed_default(
+                    DataType::OriginIotaReceivedAt,
+                    DataValue::SignedNumber(accepted_at.into()),
+                )
+                .add_typed_default(
+                    DataType::DestinationIotaReceivedAt,
+                    DataValue::SignedNumber(accepted_at.into()),
+                )
+        } else {
+            response
+        };
+        if let Err(error) = self.send_message(&response).await {
+            log!("Relay response could not be sent: {}", error);
+        }
+    }
+
     async fn handle_relay(self: Arc<Self>, frame: CommunicationValue) {
         let Some(incoming_frame_id) = frame.id() else {
             log!("Rejecting Relay without a message id");
@@ -998,6 +968,7 @@ impl OmikronConnection {
                 return;
             }
         };
+        let accepted_at = now_millis_i64();
         let signer_is_local = i64::try_from(verified.context.signer_id)
             .ok()
             .and_then(iota_storage::users::user_manager::get_user)
@@ -1035,6 +1006,7 @@ impl OmikronConnection {
             verified.context.signer_id,
             &verified.context.message_id,
             verified.context.created_at,
+            accepted_at,
             verified.context.final_recipient_id,
             &frame_bytes,
             frame_id,
@@ -1081,7 +1053,99 @@ impl OmikronConnection {
             relay_replay::RelayReservation::Existing { .. } => false,
         };
 
+        /* A shared Iota owns both independent replicas before delivering to its
+         * local recipient. The destination path below writes the recipient copy. */
+        if signer_is_local && recipient_is_local && !already_applied {
+            let content = match open_verified_relay_content(
+                &verified,
+                &[&keyring],
+                verified.context.signer_id,
+            ) {
+                Ok(value) => value,
+                Err(error) => {
+                    log!("Relay shared-Iota origin content verification failed: {}", error);
+                    let _ = relay_replay::mark_rejected(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                    );
+                    self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                        .await;
+                    return;
+                }
+            };
+            let owner = match i64::try_from(verified.context.signer_id) {
+                Ok(value) => value,
+                Err(_) => {
+                    self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                        .await;
+                    return;
+                }
+            };
+            if let Err(error) = message_handlers::apply_verified_relay_content(
+                &verified.context,
+                &content,
+                accepted_at,
+                owner,
+                true,
+            ) {
+                log!("Relay shared-Iota origin application failed: {}", error);
+                let _ = relay_replay::mark_rejected(
+                    verified.context.signer_id,
+                    &verified.context.message_id,
+                );
+                self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                    .await;
+                return;
+            }
+            if let Err(error) = chat_files::record_destination_iota_received(
+                owner,
+                owner,
+                &verified.context.message_id,
+                accepted_at,
+            ) {
+                log!("Relay shared-Iota destination timestamp storage failed: {}", error);
+                self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
+                    .await;
+                return;
+            }
+        }
+
         if signer_is_local && !recipient_is_local {
+            if !already_applied {
+                let content = match open_verified_relay_content(
+                    &verified,
+                    &[&keyring],
+                    verified.context.signer_id,
+                ) {
+                    Ok(value) => value,
+                    Err(error) => {
+                        log!("Relay origin content verification failed: {}", error);
+                        let _ = relay_replay::mark_rejected(
+                            verified.context.signer_id,
+                            &verified.context.message_id,
+                        );
+                        self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                            .await;
+                        return;
+                    }
+                };
+                if let Err(error) = message_handlers::apply_verified_relay_content(
+                    &verified.context,
+                    &content,
+                    accepted_at,
+                    i64::try_from(verified.context.signer_id).unwrap_or_default(),
+                    true,
+                ) {
+                    log!("Relay origin application failed: {}", error);
+                    let _ = relay_replay::mark_rejected(
+                        verified.context.signer_id,
+                        &verified.context.message_id,
+                    );
+                    self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                        .await;
+                    return;
+                }
+            }
             let router = match self
                 .hosting_iota_for_user(verified.context.final_recipient_id)
                 .await
@@ -1124,10 +1188,9 @@ impl OmikronConnection {
                     .await;
                 return;
             }
-            if let Err(error) = relay_replay::mark_state(
+            if let Err(error) = relay_replay::mark_queued(
                 verified.context.signer_id,
                 &verified.context.message_id,
-                "queued",
             ) {
                 log!("Relay origin state update failed: {}", error);
             }
@@ -1136,21 +1199,63 @@ impl OmikronConnection {
                 .await
             {
                 Ok(response) if response.is_type(CommunicationType::Success) => {
+                    let returned_id = response
+                        .get_data(DataType::RelayMessageId)
+                        .as_str();
+                    let destination_accepted_at = response
+                        .get_data(DataType::RelayAcceptedAt)
+                        .as_number()
+                        .and_then(|value| i64::try_from(value).ok());
+                    if returned_id != Some(verified.context.message_id.as_str()) {
+                        log!("Relay acknowledgement returned a different RelayMessageId");
+                        self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                            .await;
+                        return;
+                    }
+                    let Some(destination_accepted_at) = destination_accepted_at else {
+                        log!("Relay acknowledgement is missing RelayAcceptedAt");
+                        self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
+                            .await;
+                        return;
+                    };
+                    if let (Ok(owner), Ok(signer)) = (
+                        i64::try_from(verified.context.signer_id),
+                        i64::try_from(verified.context.signer_id),
+                    ) {
+                        if let Err(error) = chat_files::record_destination_iota_received(
+                            owner,
+                            signer,
+                            &verified.context.message_id,
+                            destination_accepted_at,
+                        ) {
+                            log!("Relay destination acknowledgement storage failed: {}", error);
+                        }
+                    }
                     if let Err(error) = relay_queue::acknowledge_iota(router, frame_id) {
                         log!(
                             "Relay origin acknowledgement could not clear the queue: {}",
                             error
                         );
                     }
-                    if let Err(error) = relay_replay::mark_state(
+                    if let Err(error) = relay_replay::mark_downstream_acked(
                         verified.context.signer_id,
                         &verified.context.message_id,
-                        "delivered",
                     ) {
                         log!("Relay origin delivery state update failed: {}", error);
                     }
-                    self.send_relay_response(frame.id(), CommunicationType::Success)
-                        .await;
+                    let response = response
+                        .add_typed_default(
+                            DataType::OriginIotaReceivedAt,
+                            DataValue::SignedNumber(accepted_at.into()),
+                        )
+                        .add_typed_default(
+                            DataType::DestinationIotaReceivedAt,
+                            DataValue::SignedNumber(destination_accepted_at.into()),
+                        )
+                        .with_id(frame_id);
+                    if let Err(error) = self.send_message(&response).await {
+                        log!("Relay response could not be sent: {}", error);
+                    }
                 }
                 Ok(response) => {
                     log!("Relay origin route returned {}", response.get_type());
@@ -1223,10 +1328,9 @@ impl OmikronConnection {
                             queue_error
                         );
                     }
-                    let _ = relay_replay::mark_state(
+                    let _ = relay_replay::mark_rejected(
                         verified.context.signer_id,
                         &verified.context.message_id,
-                        "rejected",
                     );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
                         .await;
@@ -1234,7 +1338,24 @@ impl OmikronConnection {
                 }
             };
             if let Err(error) =
-                message_handlers::apply_verified_relay_content(&verified.context, &content)
+                message_handlers::apply_verified_relay_content(
+                    &verified.context,
+                    &content,
+                    accepted_at,
+                    match i64::try_from(destination) {
+                        Ok(value) => value,
+                        Err(_) => {
+                            log!("Relay destination ID exceeds storage range");
+                            self.send_relay_response(
+                                frame.id(),
+                                CommunicationType::ErrorInvalidData,
+                            )
+                            .await;
+                            return;
+                        }
+                    },
+                    false,
+                )
             {
                 log!("Relay application dispatch failed: {}", error);
                 if let Err(queue_error) =
@@ -1242,19 +1363,17 @@ impl OmikronConnection {
                 {
                     log!("Relay application queue cleanup failed: {}", queue_error);
                 }
-                let _ = relay_replay::mark_state(
+                let _ = relay_replay::mark_rejected(
                     verified.context.signer_id,
                     &verified.context.message_id,
-                    "rejected",
                 );
                 self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
                     .await;
                 return;
             }
-            if let Err(error) = relay_replay::mark_state(
+            if let Err(error) = relay_replay::mark_applied(
                 verified.context.signer_id,
                 &verified.context.message_id,
-                "applied",
             ) {
                 log!("Relay application state update failed: {}", error);
                 self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
@@ -1263,15 +1382,20 @@ impl OmikronConnection {
             }
         }
 
-        if let Err(error) = relay_replay::mark_state(
+        if let Err(error) = relay_replay::mark_queued(
             verified.context.signer_id,
             &verified.context.message_id,
-            "queued",
         ) {
             log!("Relay queue state update failed: {}", error);
         }
-        self.send_relay_response(frame.id(), CommunicationType::Success)
-            .await;
+        self.send_relay_success(
+            frame.id(),
+            local_iota_id,
+            &verified.context.message_id,
+            accepted_at,
+            signer_is_local,
+        )
+        .await;
         if let Err(error) = self.send_message(&forwarded).await {
             log!("Relay delivery to local client failed: {}", error);
         }
@@ -1707,7 +1831,7 @@ impl OmikronConnection {
         let Some(send_time) = data_i64(cv, DataType::SendTime).filter(|time| *time > 0) else {
             return;
         };
-        let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
+        let Some(content) = cv.get_data(DataType::Content).as_str() else {
             return;
         };
         if chat_files::apply_remote_edit(receiver_id, sender_id, send_time, sender_id, content)
@@ -1790,7 +1914,7 @@ impl OmikronConnection {
                 .await;
             return;
         };
-        let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
+        let Some(content) = cv.get_data(DataType::Content).as_str() else {
             let _ = self
                 .send_message(&error_response(cv, CommunicationType::ErrorInvalidData))
                 .await;
@@ -1800,7 +1924,7 @@ impl OmikronConnection {
             CommunicationType::MessageEditLive,
             cv,
             &mutation,
-            vec![(DataType::AppContent, DataValue::Str(content.to_string()))],
+            vec![(DataType::Content, DataValue::Str(content.to_string()))],
         );
         if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some()
             && chat_files::apply_remote_edit(
@@ -2237,11 +2361,7 @@ impl OmikronConnection {
                 ))
             })?;
         }
-<<<<<<< HEAD
-        save_keyring(&keyring, path).map_err(|error| {
-=======
         save_protected_keyring_verified(&keyring, path, &identity_secret).map_err(|error| {
->>>>>>> refs/remotes/origin/main
             OmikronError::Internal(format!(
                 "could not save new identity {}: {error}",
                 path.display()
@@ -2410,29 +2530,9 @@ impl OmikronClient for OmikronConnection {
 }
 
 #[cfg(test)]
-<<<<<<< HEAD
 mod tests {
     use super::*;
 
-    #[test]
-    fn durable_keyring_save_preserves_mtp_format() {
-        let directory = std::env::temp_dir().join(format!("iota-keyring-test-{}", Uuid::new_v4()));
-        std::fs::create_dir_all(&directory).unwrap();
-        let path = directory.join(IOTA_KEYRING_PATH);
-        let keyring = crypto_helper::generate_keyring();
-
-        save_keyring(&keyring, &path).unwrap();
-
-        let loaded = mtp::files::load_keyring_raw(&path).unwrap();
-        assert_eq!(
-            keyring.try_to_bytes().unwrap(),
-            loaded.try_to_bytes().unwrap()
-        );
-        std::fs::remove_dir_all(directory).unwrap();
-=======
-mod identity_tests {
-    use super::*;
-
     fn test_path(name: &str) -> PathBuf {
         std::env::temp_dir().join(format!(
             "iota-identity-{name}-{}-{}",
@@ -2503,6 +2603,5 @@ mod identity_tests {
         }
 
         assert!(jittered_reconnect_delay(Duration::from_secs(600)) <= MAX_RECONNECT_DELAY);
->>>>>>> refs/remotes/origin/main
     }
 }