Update Rust crate base64 to 0.23.0

Cargo.lock

@@ -2240,7 +2240,7 @@ name = "iota-storage"
 version = "0.1.0"
 dependencies = [
  "arc-swap",
- "base64 0.23.1",
+ "base64 0.22.1",
  "iota-logger",
  "iota-paths",
  "iota-util",
@@ -2283,7 +2283,7 @@ dependencies = [
 name = "iota-util"
 version = "0.1.0"
 dependencies = [
- "base64 0.23.1",
+ "base64 0.22.1",
  "hex",
  "iota-paths",
  "mtp",
@@ -2996,7 +2996,7 @@ name = "omikron-connector"
 version = "0.1.0"
 dependencies = [
  "async-trait",
- "base64 0.23.1",
+ "base64 0.22.1",
  "dashmap",
  "iota-connection",
  "iota-logger",

iota-connection/src/message_handlers.rs

@@ -115,9 +115,6 @@ 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)
@@ -128,48 +125,8 @@ 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::Content, &context.type_map)
+            let message = relay_string(&content.content, DataType::AppContent, &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())
@@ -179,35 +136,20 @@ 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());
-            let relay_message_id = relay_string(
+            chat_files::add_message(
-                &content.content,
+                u128::try_from(send_time)
-                DataType::RelayMessageId,
+                    .map_err(|_| "Relay MessageSend has a negative SendTime".to_string())?,
-                &context.type_map,
+                false,
-            )
+                recipient_id,
-            .ok_or_else(|| "Relay MessageSend is missing RelayMessageId".to_string())?;
+                sender_id,
-            if relay_message_id != context.message_id {
+                message,
-                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::Content, &context.type_map)
+            let message = relay_string(&content.content, DataType::AppContent, &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())
@@ -275,7 +217,7 @@ pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue {
         Ok(mutation) => mutation,
         Err(response) => return response,
     };
-    let Some(content) = cv.get_data(DataType::Content).as_str() else {
+    let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
         return error_response(cv, CommunicationType::ErrorInvalidData);
     };
 
@@ -323,9 +265,6 @@ 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),
     }
 }
@@ -349,7 +288,7 @@ fn stored_message_fields(
 ) -> Vec<(DataType, DataValue)> {
     let mut fields = vec![
         (
-            DataType::MessageId,
+            DataType::AppMessageId,
             DataValue::SignedNumber(message.id as i128),
         ),
         (
@@ -357,7 +296,7 @@ fn stored_message_fields(
             DataValue::SignedNumber(message.message_time as i128),
         ),
         (
-            DataType::Content,
+            DataType::AppContent,
             DataValue::Str(message.content.clone()),
         ),
         (
@@ -377,25 +316,6 @@ 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)));
     }
@@ -478,7 +398,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue {
                 DataValue::Str(record.wrapping_scheme),
             )
             .add_typed_default(
-                DataType::CreatedAt,
+                DataType::AppCreatedAt,
                 DataValue::SignedNumber(record.created_at as i128),
             )
             .add_typed_default(
@@ -840,18 +760,10 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue {
         Err(response) => return response,
     };
 
-    let (message, offset) = match partner_id {
+    let message = match chat_files::get_message(owner, send_time, partner_id) {
-        Some(partner_id) => match chat_files::get_message_with_offset(owner, partner_id, send_time)
+        Ok(Some(message)) => message,
-        {
+        Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
-            Ok(Some((message, offset))) => (message, Some(offset)),
+        Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
-            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)
@@ -860,10 +772,6 @@ 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,6 +2,8 @@ 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,4 +12,8 @@ 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,7 +7,11 @@ edition = "2024"
 iota-logger = { path = "../iota-logger" }
 iota-util = { path = "../iota-util" }
 iota-paths = { path = "../iota-paths" }
-base64 = "0.23.0"
+<<<<<<< HEAD
+
+=======
+>>>>>>> refs/remotes/origin/main
+base64 = "0.22.1"
 json = "*"
 arc-swap = "1"
 once_cell = "1.21.3"

iota-storage/src/storage_error.rs

@@ -8,8 +8,6 @@ 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,9 +2,7 @@ use crate::storage_error::StorageError;
 use crate::util::db;
 use crate::util::sync::{self, EntityType, Operation};
 use iota_logger::log;
-use rusqlite::{OptionalExtension, params};
+use rusqlite::params;
-
-pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10;
 
 #[derive(PartialEq, Debug, Clone)]
 pub enum MessageState {
@@ -50,16 +48,7 @@ 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,
@@ -69,22 +58,6 @@ 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,
@@ -176,7 +149,7 @@ fn update_message_content(
 
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
-            .unwrap_or_default()
+            .unwrap()
             .as_millis() as i64;
 
         let tx = conn.unchecked_transaction()?;
@@ -378,8 +351,8 @@ pub fn add_reaction(
     user_id: i64,
     reaction: &str,
 ) -> Result<(), StorageError> {
-    db::with_immediate_transaction(|tx| {
+    db::with_db(|conn| {
-        let msg_id: i64 = tx.query_row(
+        let msg_id: i64 = conn.query_row(
             r#"
             SELECT id FROM messages
             WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
@@ -389,46 +362,28 @@ 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_or_default()
+            .unwrap()
             .as_millis() as i64;
 
-        let inserted = tx.execute(
+        let tx = conn.unchecked_transaction()?;
+        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],
         )?;
-        if inserted > 0 {
+        sync::record_event(
-            sync::record_event(
+            &tx,
-                tx,
+            storage_owner,
-                storage_owner,
+            EntityType::Message,
-                EntityType::Message,
+            msg_id,
-                msg_id,
+            Operation::Upsert,
-                Operation::Upsert,
+        )?;
-            )?;
+        tx.commit()?;
-            Ok(())
+        Ok(())
-        } else {
-            Ok(())
-        }
     })
 }
 
@@ -467,46 +422,45 @@ pub fn remove_reaction(
     })
 }
 
-pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
+pub fn add_message(
-    let NewMessage {
+    send_time: u128,
-        relay_signer_id,
+    storage_owner_is_sender: bool,
-        relay_message_id,
+    storage_owner: i64,
-        authored_at,
+    external_user: i64,
-        send_time,
+    message: &str,
-        storage_owner,
+    height: i64,
-        external_user,
+    reply_to: Option<i64>,
-        sent_by_self,
+) {
-        content,
+    let message_time = match i64::try_from(send_time) {
-        height,
+        Ok(v) => v,
-        reply_to,
+        Err(_) => {
-        origin_iota_received_at,
+            log!("Failed to store message: send_time out of range for i64 ({send_time})");
-        destination_iota_received_at,
+            return;
-        initial_state,
+        }
-    } = message;
+    };
-    let msg_id = db::with_db(|conn| {
+
+    if let Err(e) = 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,
+                storage_owner, external_user, message_time, content,
-                message_state, height, reply_to, relay_signer_id, relay_message_id,
+                sent_by_self, message_state, height, reply_to
-                authored_at, origin_iota_received_at, destination_iota_received_at
+            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
-            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
             "#,
             params![
                 storage_owner,
                 external_user,
-                send_time,
+                message_time,
-                content,
+                message,
-                i64::from(sent_by_self),
+                if storage_owner_is_sender {
-                initial_state.as_str(),
+                    1_i64
+                } else {
+                    0_i64
+                },
+                MessageState::Sending.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();
@@ -518,122 +472,15 @@ pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
             Operation::Upsert,
         )?;
         tx.commit()?;
-        Ok(msg_id)
+        Ok(())
-    })?;
+    }) {
+        log!("Failed to insert message into sqlite: {}", e);
+        return;
+    }
 
     let mut contact = crate::users::contact::Contact::new(external_user);
-    contact.set_last_message_at(send_time);
+    contact.set_last_message_at(message_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(
@@ -727,15 +574,7 @@ fn load_reactions(
             ))
         }) {
             for row in rows.flatten() {
-                let reactions = map.entry(row.0).or_default();
+                map.entry(row.0).or_default().push(row.1);
-                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);
-                }
             }
         }
     }
@@ -755,11 +594,7 @@ pub fn get_messages(
     match db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
-            SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
+            SELECT id, message_time, content, sent_by_self, message_state, height, reply_to, edited_count
-                   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
@@ -773,22 +608,13 @@ pub fn get_messages(
                 Ok(StoredMessage {
                     id: row.get(0)?,
                     external_user,
-                    relay_signer_id: row.get(1)?,
+                    message_time: row.get(1)?,
-                    relay_message_id: row.get(2)?,
+                    content: row.get(2)?,
-                    message_time: row.get(3)?,
+                    sent_by_self: row.get::<_, i64>(3)? != 0,
-                    authored_at: row.get(4)?,
+                    message_state: row.get(4)?,
-                    origin_iota_received_at: row.get(5)?,
+                    height: row.get(5).unwrap_or(0),
-                    destination_iota_received_at: row.get(6)?,
+                    reply_to: row.get(6).ok().flatten(),
-                    client_received_at: row.get(7)?,
+                    edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
-                    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(),
                 })
             },
@@ -826,10 +652,7 @@ pub fn get_message(
     db::with_db(|conn| {
         let mut stmt = conn.prepare(
             r#"
-            SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
+            SELECT id, message_time, content, sent_by_self, message_state, height,
-                   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
@@ -843,23 +666,14 @@ pub fn get_message(
         let rows = stmt.query_map(params![storage_owner, message_time, external_user], |row| {
             Ok(StoredMessage {
                 id: row.get(0)?,
-                relay_signer_id: row.get(1)?,
+                message_time: row.get(1)?,
-                relay_message_id: row.get(2)?,
+                content: row.get(2)?,
-                message_time: row.get(3)?,
+                sent_by_self: row.get::<_, i64>(3)? != 0,
-                authored_at: row.get(4)?,
+                message_state: row.get(4)?,
-                origin_iota_received_at: row.get(5)?,
+                height: row.get(5).unwrap_or(0),
-                destination_iota_received_at: row.get(6)?,
+                reply_to: row.get(6).ok().flatten(),
-                client_received_at: row.get(7)?,
+                edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
-                client_received_recorded_at: row.get(8)?,
+                external_user: 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(),
             })
         })?;
@@ -886,40 +700,6 @@ 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();
@@ -928,28 +708,19 @@ 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, 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 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 rows = stmt.query_map([storage_owner], |row| {
-            let external_user: i64 = row.get(17)?;
+            let external_user: i64 = row.get(8)?;
             Ok(StoredMessage {
                 id: row.get(0)?,
                 external_user,
-                relay_signer_id: row.get(1)?,
+                message_time: row.get(1)?,
-                relay_message_id: row.get(2)?,
+                content: row.get(2)?,
-                message_time: row.get(3)?,
+                sent_by_self: row.get::<_, i64>(3)? != 0,
-                authored_at: row.get(4)?,
+                message_state: row.get(4)?,
-                origin_iota_received_at: row.get(5)?,
+                height: row.get(5).unwrap_or(0),
-                destination_iota_received_at: row.get(6)?,
+                reply_to: row.get(6).ok().flatten(),
-                client_received_at: row.get(7)?,
+                edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
-                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, Transaction};
+use rusqlite::Connection;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
@@ -55,17 +55,6 @@ 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. */
@@ -399,71 +388,6 @@ 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(())
 }
 
@@ -533,7 +457,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, 11);
+        assert_eq!(version, 9);
         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")?;
@@ -546,13 +470,11 @@ mod tests {
     #[test]
     fn migrates_version_five_once_and_is_idempotent() -> Result<(), StorageError> {
         let conn = Connection::open_in_memory()?;
-        conn.execute_batch(
+        conn.execute_batch("PRAGMA user_version = 5;")?;
-            "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, 11);
+        assert_eq!(version, 9);
         for table in [
             "sync_heads",
             "sync_events",

iota-storage/src/util/relay_replay.rs

@@ -12,7 +12,6 @@ pub fn reserve(
     signer_id: u64,
     message_id: &str,
     created_at: u64,
-    accepted_at: i64,
     destination_id: u64,
     frame: &[u8],
     frame_id: u32,
@@ -27,12 +26,11 @@ pub fn reserve(
 
     db::with_db(|connection| {
         let inserted = connection.execute(
-            "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')",
+            "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')",
             params![
                 signer_id,
                 message_id,
                 created_at,
-                accepted_at,
                 destination_id,
                 frame,
                 i64::from(frame_id),
@@ -69,64 +67,41 @@ 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', downstream_acked_at = COALESCE(downstream_acked_at, ?3) WHERE destination_id = ?1 AND frame_id = ?2",
+            "UPDATE relay_inbox SET state = 'delivered' WHERE destination_id = ?1 AND frame_id = ?2",
-            params![destination_id, i64::from(frame_id), delivered_at],
+            params![destination_id, i64::from(frame_id)],
         )?;
         Ok(())
     })
 }
 
-fn mark_transition(
+pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), StorageError> {
-    signer_id: u64,
+    if !matches!(
-    message_id: &str,
+        state,
-    state: &str,
+        "received" | "applied" | "queued" | "delivered" | "rejected"
-    column: &str,
+    ) {
-) -> Result<(), StorageError> {
+        return Err(StorageError::Other("invalid relay inbox state".into()));
+    }
     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(
-            &format!("UPDATE relay_inbox SET state = ?3, {column} = COALESCE({column}, ?4) WHERE signer_id = ?1 AND message_id = ?2"),
+            "UPDATE relay_inbox SET state = ?3 WHERE signer_id = ?1 AND message_id = ?2",
-            params![signer_id, message_id, state, timestamp],
+            params![signer_id, message_id, state],
         )?;
         Ok(())
     })
 }
 
-pub fn mark_applied(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
+pub fn prune_completed(before_created_at: i64) -> 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 COALESCE(downstream_acked_at, rejected_at) < ?1 AND state IN ('delivered', 'rejected')",
+            "DELETE FROM relay_inbox WHERE created_at < ?1 AND state IN ('delivered', 'rejected')",
-            params![before_terminal_at],
+            params![before_created_at],
         )?;
         connection.execute(
-            "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)",
+            "DELETE FROM relay_replay WHERE created_at < ?1",
-            [],
+            params![before_created_at],
         )?;
         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 = 2;
+pub const CACHE_SCHEMA_VERSION: i64 = 1;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EntityType {

iota-updater/Cargo.toml

@@ -5,6 +5,10 @@ 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,10 +6,17 @@ 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> {
@@ -18,10 +25,17 @@ 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 f3c5037b0a099ae5389486eec77471bd5addab4a
+Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e

omikron-connector/src/omikron_connection.rs

@@ -68,6 +68,18 @@ 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)
@@ -597,8 +609,46 @@ 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
     }
 
     // -------------------------------------------------------------------------
@@ -671,8 +721,12 @@ 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)) => {
@@ -698,6 +752,20 @@ 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) {
@@ -887,44 +955,6 @@ 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");
@@ -968,7 +998,6 @@ 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)
@@ -1006,7 +1035,6 @@ 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,
@@ -1053,99 +1081,7 @@ 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
@@ -1188,9 +1124,10 @@ impl OmikronConnection {
                     .await;
                 return;
             }
-            if let Err(error) = relay_replay::mark_queued(
+            if let Err(error) = relay_replay::mark_state(
                 verified.context.signer_id,
                 &verified.context.message_id,
+                "queued",
             ) {
                 log!("Relay origin state update failed: {}", error);
             }
@@ -1199,63 +1136,21 @@ 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_downstream_acked(
+                    if let Err(error) = relay_replay::mark_state(
                         verified.context.signer_id,
                         &verified.context.message_id,
+                        "delivered",
                     ) {
                         log!("Relay origin delivery state update failed: {}", error);
                     }
-                    let response = response
+                    self.send_relay_response(frame.id(), CommunicationType::Success)
-                        .add_typed_default(
+                        .await;
-                            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());
@@ -1328,9 +1223,10 @@ impl OmikronConnection {
                             queue_error
                         );
                     }
-                    let _ = relay_replay::mark_rejected(
+                    let _ = relay_replay::mark_state(
                         verified.context.signer_id,
                         &verified.context.message_id,
+                        "rejected",
                     );
                     self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
                         .await;
@@ -1338,24 +1234,7 @@ impl OmikronConnection {
                 }
             };
             if let Err(error) =
-                message_handlers::apply_verified_relay_content(
+                message_handlers::apply_verified_relay_content(&verified.context, &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) =
@@ -1363,17 +1242,19 @@ impl OmikronConnection {
                 {
                     log!("Relay application queue cleanup failed: {}", queue_error);
                 }
-                let _ = relay_replay::mark_rejected(
+                let _ = relay_replay::mark_state(
                     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_applied(
+            if let Err(error) = relay_replay::mark_state(
                 verified.context.signer_id,
                 &verified.context.message_id,
+                "applied",
             ) {
                 log!("Relay application state update failed: {}", error);
                 self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
@@ -1382,20 +1263,15 @@ impl OmikronConnection {
             }
         }
 
-        if let Err(error) = relay_replay::mark_queued(
+        if let Err(error) = relay_replay::mark_state(
             verified.context.signer_id,
             &verified.context.message_id,
+            "queued",
         ) {
             log!("Relay queue state update failed: {}", error);
         }
-        self.send_relay_success(
+        self.send_relay_response(frame.id(), CommunicationType::Success)
-            frame.id(),
+            .await;
-            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);
         }
@@ -1831,7 +1707,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::Content).as_str() else {
+        let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
             return;
         };
         if chat_files::apply_remote_edit(receiver_id, sender_id, send_time, sender_id, content)
@@ -1914,7 +1790,7 @@ impl OmikronConnection {
                 .await;
             return;
         };
-        let Some(content) = cv.get_data(DataType::Content).as_str() else {
+        let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
             let _ = self
                 .send_message(&error_response(cv, CommunicationType::ErrorInvalidData))
                 .await;
@@ -1924,7 +1800,7 @@ impl OmikronConnection {
             CommunicationType::MessageEditLive,
             cv,
             &mutation,
-            vec![(DataType::Content, DataValue::Str(content.to_string()))],
+            vec![(DataType::AppContent, DataValue::Str(content.to_string()))],
         );
         if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some()
             && chat_files::apply_remote_edit(
@@ -2361,7 +2237,11 @@ 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()
@@ -2530,9 +2410,29 @@ 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}-{}-{}",
@@ -2603,5 +2503,6 @@ mod tests {
         }
 
         assert!(jittered_reconnect_delay(Duration::from_secs(600)) <= MAX_RECONNECT_DELAY);
+>>>>>>> refs/remotes/origin/main
     }
 }