From 9af194451d1139c585e4cfa2153f1c6db8c98509 Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Fri, 28 Aug 2026 15:01:14 +0300 Subject: [PATCH 1/4] Update Rust crate base64 to 0.23.0 --- iota-util/Cargo.toml | 2 +- omikron-connector/Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/iota-util/Cargo.toml b/iota-util/Cargo.toml index 676d02f..658edd5 100644 --- a/iota-util/Cargo.toml +++ b/iota-util/Cargo.toml @@ -15,7 +15,7 @@ sysinfo = "0.38.0" uuid = { version = "*", features = ["v4"] } walkdir = "2.5.0" zip = "6.0.0" -base64 = "0.22.1" +base64 = "0.23.0" hex = "*" [dev-dependencies] diff --git a/omikron-connector/Cargo.toml b/omikron-connector/Cargo.toml index c4773f7..676f034 100644 --- a/omikron-connector/Cargo.toml +++ b/omikron-connector/Cargo.toml @@ -22,5 +22,5 @@ reqwest = "0.13.2" tokio = { version = "1.50.0", features = ["full"] } tokio-util = { version = "0.7", features = ["rt"] } uuid = { version = "*", features = ["v4"] } -base64 = "0.22.1" +base64 = "0.23.0" rand_core = { version = "0.6", features = ["getrandom", "std"] } From 29d46d1c954c340782adaf2314853900a5e52d2c Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Fri, 28 Aug 2026 16:21:11 +0200 Subject: [PATCH 2/4] [Add] Replyjumps --- iota-connection/src/message_handlers.rs | 125 ++++++- iota-connection/src/relay.rs | 2 - iota-state/Cargo.toml | 4 - iota-storage/Cargo.toml | 4 - iota-storage/src/storage_error.rs | 2 + iota-storage/src/util/chat_files.rs | 385 ++++++++++++++++---- iota-storage/src/util/db.rs | 86 ++++- iota-storage/src/util/relay_replay.rs | 38 +- iota-updater/Cargo.toml | 4 - iota-util/src/crypto_helper.rs | 14 - mtp-type-maps | 2 +- omikron-connector/src/omikron_connection.rs | 224 +++++++----- 12 files changed, 650 insertions(+), 240 deletions(-) diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 4be17f6..1f247aa 100644 --- a/iota-connection/src/message_handlers.rs +++ b/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,32 @@ 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())?; + 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 +272,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 +320,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 +346,7 @@ fn stored_message_fields( ) -> Vec<(DataType, DataValue)> { let mut fields = vec![ ( - DataType::AppMessageId, + DataType::MessageId, DataValue::SignedNumber(message.id as i128), ), ( @@ -296,7 +354,7 @@ fn stored_message_fields( DataValue::SignedNumber(message.message_time as i128), ), ( - DataType::AppContent, + DataType::Content, DataValue::Str(message.content.clone()), ), ( @@ -316,6 +374,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 +475,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 +837,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 +857,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 } diff --git a/iota-connection/src/relay.rs b/iota-connection/src/relay.rs index f9ef1fe..161fb93 100644 --- a/iota-connection/src/relay.rs +++ b/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, }; diff --git a/iota-state/Cargo.toml b/iota-state/Cargo.toml index e4acfc2..4db4406 100644 --- a/iota-state/Cargo.toml +++ b/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 diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index ac8783e..9f91c52 100644 --- a/iota-storage/Cargo.toml +++ b/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" diff --git a/iota-storage/src/storage_error.rs b/iota-storage/src/storage_error.rs index 100bf08..86a0c53 100644 --- a/iota-storage/src/storage_error.rs +++ b/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), } diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 024c302..76695e5 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/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, + pub relay_message_id: Option, pub message_time: i64, + pub authored_at: Option, + pub origin_iota_received_at: Option, + pub destination_iota_received_at: Option, + pub client_received_at: Option, + pub client_received_recorded_at: Option, + pub read_at: Option, + pub read_recorded_at: Option, pub content: String, pub edited: bool, pub sent_by_self: bool, @@ -58,6 +69,22 @@ pub struct StoredMessage { pub reactions: Vec, } +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, + pub origin_iota_received_at: Option, + pub destination_iota_received_at: Option, + 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, -) { - 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 { + 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(¤t).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, 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 { if ids.is_empty() { return Vec::new(); @@ -708,19 +928,28 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec(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(), }) })?; diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 46363cf..260c667 100644 --- a/iota-storage/src/util/db.rs +++ b/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(f: F) -> Result +where + F: FnOnce(&Transaction<'_>) -> Result, +{ + 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", diff --git a/iota-storage/src/util/relay_replay.rs b/iota-storage/src/util/relay_replay.rs index bc3bda7..db3cf0e 100644 --- a/iota-storage/src/util/relay_replay.rs +++ b/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,9 +69,13 @@ 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(()) }) @@ -84,24 +90,36 @@ pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), S } 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| { + let column = match state { + "applied" => "applied_at", + "queued" => "queued_at", + "delivered" => "downstream_acked_at", + "rejected" => "rejected_at", + "received" => "accepted_at", + _ => return Err(StorageError::Other("invalid relay inbox state".into())), + }; 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 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(()) }) diff --git a/iota-updater/Cargo.toml b/iota-updater/Cargo.toml index 9e19437..1fbbef9 100644 --- a/iota-updater/Cargo.toml +++ b/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 = "*" diff --git a/iota-util/src/crypto_helper.rs b/iota-util/src/crypto_helper.rs index 1931da7..339dd1f 100644 --- a/iota-util/src/crypto_helper.rs +++ b/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 { @@ -25,17 +18,10 @@ pub fn keyring_from_base64(s: &str) -> Option { } 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 { diff --git a/mtp-type-maps b/mtp-type-maps index 486541b..f3c5037 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e +Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 9599cb6..0a3a6a3 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/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 { 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,30 @@ impl OmikronConnection { } } + async fn send_relay_success( + &self, + frame_id: Option, + iota_id: u64, + relay_message_id: &str, + accepted_at: i64, + ) { + 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()), + ); + if let Err(error) = self.send_message(&response).await { + log!("Relay response could not be sent: {}", error); + } + } + async fn handle_relay(self: Arc, frame: CommunicationValue) { let Some(incoming_frame_id) = frame.id() else { log!("Rejecting Relay without a message id"); @@ -998,6 +954,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 +992,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, @@ -1082,6 +1040,43 @@ impl OmikronConnection { }; 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_state( + verified.context.signer_id, + &verified.context.message_id, + "rejected", + ); + 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_state( + verified.context.signer_id, + &verified.context.message_id, + "rejected", + ); + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + .await; + return; + } + } let router = match self .hosting_iota_for_user(verified.context.final_recipient_id) .await @@ -1136,6 +1131,34 @@ impl OmikronConnection { .await { Ok(response) if response.is_type(CommunicationType::Success) => { + let returned_id = response + .get_data(DataType::RelayMessageId) + .as_str(); + let 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; + } + if let Some(accepted_at) = accepted_at { + 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, + 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: {}", @@ -1149,8 +1172,10 @@ impl OmikronConnection { ) { log!("Relay origin delivery state update failed: {}", error); } - self.send_relay_response(frame.id(), CommunicationType::Success) - .await; + let response = response.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()); @@ -1234,7 +1259,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) = @@ -1270,8 +1312,13 @@ impl OmikronConnection { ) { 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, + ) + .await; if let Err(error) = self.send_message(&forwarded).await { log!("Relay delivery to local client failed: {}", error); } @@ -1707,7 +1754,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 +1837,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 +1847,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 +2284,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 +2453,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 +2526,5 @@ mod identity_tests { } assert!(jittered_reconnect_delay(Duration::from_secs(600)) <= MAX_RECONNECT_DELAY); ->>>>>>> refs/remotes/origin/main } } From 3bfec968485d82e329f1d53b03e1c96c0de7a2b7 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:00:44 +0200 Subject: [PATCH 3/4] [Add] timestamping --- iota-connection/src/message_handlers.rs | 3 + iota-storage/src/util/relay_replay.rs | 37 +++--- iota-storage/src/util/sync.rs | 2 +- omikron-connector/src/omikron_connection.rs | 137 +++++++++++++++----- 4 files changed, 133 insertions(+), 46 deletions(-) diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 1f247aa..d692be4 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -185,6 +185,9 @@ pub fn apply_verified_relay_content( &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, diff --git a/iota-storage/src/util/relay_replay.rs b/iota-storage/src/util/relay_replay.rs index db3cf0e..0c5bddc 100644 --- a/iota-storage/src/util/relay_replay.rs +++ b/iota-storage/src/util/relay_replay.rs @@ -81,13 +81,12 @@ pub fn mark_delivered_for_frame(destination_id: u64, frame_id: u32) -> Result<() }) } -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() @@ -95,14 +94,6 @@ pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), S .unwrap_or_default() .as_millis() as i64; db::with_db(|connection| { - let column = match state { - "applied" => "applied_at", - "queued" => "queued_at", - "delivered" => "downstream_acked_at", - "rejected" => "rejected_at", - "received" => "accepted_at", - _ => return Err(StorageError::Other("invalid relay inbox state".into())), - }; connection.execute( &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], @@ -111,6 +102,22 @@ pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), S }) } +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( diff --git a/iota-storage/src/util/sync.rs b/iota-storage/src/util/sync.rs index b84c94a..a358a1d 100644 --- a/iota-storage/src/util/sync.rs +++ b/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 { diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 0a3a6a3..0c37013 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -893,6 +893,7 @@ impl OmikronConnection { 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) @@ -906,6 +907,19 @@ impl OmikronConnection { 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); } @@ -1039,6 +1053,63 @@ 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( @@ -1049,10 +1120,9 @@ impl OmikronConnection { Ok(value) => value, Err(error) => { log!("Relay origin content verification failed: {}", 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; @@ -1067,10 +1137,9 @@ impl OmikronConnection { true, ) { log!("Relay origin application failed: {}", 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; @@ -1119,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); } @@ -1134,7 +1202,7 @@ impl OmikronConnection { let returned_id = response .get_data(DataType::RelayMessageId) .as_str(); - let accepted_at = response + let destination_accepted_at = response .get_data(DataType::RelayAcceptedAt) .as_number() .and_then(|value| i64::try_from(value).ok()); @@ -1144,19 +1212,23 @@ impl OmikronConnection { .await; return; } - if let Some(accepted_at) = accepted_at { - if let (Ok(owner), Ok(signer)) = ( - i64::try_from(verified.context.signer_id), - i64::try_from(verified.context.signer_id), + 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, ) { - if let Err(error) = chat_files::record_destination_iota_received( - owner, - signer, - &verified.context.message_id, - accepted_at, - ) { - log!("Relay destination acknowledgement storage failed: {}", error); - } + log!("Relay destination acknowledgement storage failed: {}", error); } } if let Err(error) = relay_queue::acknowledge_iota(router, frame_id) { @@ -1165,14 +1237,22 @@ impl OmikronConnection { 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); } - let response = response.with_id(frame_id); + 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); } @@ -1248,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; @@ -1284,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) @@ -1305,10 +1382,9 @@ 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); } @@ -1317,6 +1393,7 @@ impl OmikronConnection { local_iota_id, &verified.context.message_id, accepted_at, + signer_is_local, ) .await; if let Err(error) = self.send_message(&forwarded).await { From c2b15b9890b8a4b1387f4cb0b7716d596f78bff4 Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Fri, 28 Aug 2026 18:01:00 +0300 Subject: [PATCH 4/4] Update Rust crate base64 to 0.23.0 --- Cargo.lock | 6 +++--- iota-storage/Cargo.toml | 2 +- iota-util/Cargo.toml | 2 +- omikron-connector/Cargo.toml | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index baa33b5..6e552b2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2240,7 +2240,7 @@ name = "iota-storage" version = "0.1.0" dependencies = [ "arc-swap", - "base64 0.22.1", + "base64 0.23.1", "iota-logger", "iota-paths", "iota-util", @@ -2283,7 +2283,7 @@ dependencies = [ name = "iota-util" version = "0.1.0" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "hex", "iota-paths", "mtp", @@ -2996,7 +2996,7 @@ name = "omikron-connector" version = "0.1.0" dependencies = [ "async-trait", - "base64 0.22.1", + "base64 0.23.1", "dashmap", "iota-connection", "iota-logger", diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index 9f91c52..ebbf514 100644 --- a/iota-storage/Cargo.toml +++ b/iota-storage/Cargo.toml @@ -7,7 +7,7 @@ edition = "2024" iota-logger = { path = "../iota-logger" } iota-util = { path = "../iota-util" } iota-paths = { path = "../iota-paths" } -base64 = "0.22.1" +base64 = "0.23.0" json = "*" arc-swap = "1" once_cell = "1.21.3" diff --git a/iota-util/Cargo.toml b/iota-util/Cargo.toml index 676d02f..658edd5 100644 --- a/iota-util/Cargo.toml +++ b/iota-util/Cargo.toml @@ -15,7 +15,7 @@ sysinfo = "0.38.0" uuid = { version = "*", features = ["v4"] } walkdir = "2.5.0" zip = "6.0.0" -base64 = "0.22.1" +base64 = "0.23.0" hex = "*" [dev-dependencies] diff --git a/omikron-connector/Cargo.toml b/omikron-connector/Cargo.toml index c4773f7..676f034 100644 --- a/omikron-connector/Cargo.toml +++ b/omikron-connector/Cargo.toml @@ -22,5 +22,5 @@ reqwest = "0.13.2" tokio = { version = "1.50.0", features = ["full"] } tokio-util = { version = "0.7", features = ["rt"] } uuid = { version = "*", features = ["v4"] } -base64 = "0.22.1" +base64 = "0.23.0" rand_core = { version = "0.6", features = ["getrandom", "std"] }