diff --git a/Cargo.lock b/Cargo.lock index 6e552b2..baa33b5 100644 --- a/Cargo.lock +++ b/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", diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index d692be4..4be17f6 100644 --- a/iota-connection/src/message_handlers.rs +++ b/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( - &content.content, - DataType::RelayMessageId, - &context.type_map, - ) - .ok_or_else(|| "Relay MessageSend is missing RelayMessageId".to_string())?; - if relay_message_id != context.message_id { - return Err("Relay MessageSend identity does not match its protected message ID".into()); - } - chat_files::add_message(chat_files::NewMessage { - relay_signer_id: sender_id, - relay_message_id, - authored_at: created_at, - send_time, - storage_owner, - external_user: if sent_by_self { recipient_id } else { sender_id }, - sent_by_self, - content: message, + 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, 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 { - 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 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 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 } diff --git a/iota-connection/src/relay.rs b/iota-connection/src/relay.rs index 161fb93..f9ef1fe 100644 --- a/iota-connection/src/relay.rs +++ b/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, }; diff --git a/iota-state/Cargo.toml b/iota-state/Cargo.toml index 4db4406..e4acfc2 100644 --- a/iota-state/Cargo.toml +++ b/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 diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index ebbf514..ac8783e 100644 --- a/iota-storage/Cargo.toml +++ b/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" diff --git a/iota-storage/src/storage_error.rs b/iota-storage/src/storage_error.rs index 86a0c53..100bf08 100644 --- a/iota-storage/src/storage_error.rs +++ b/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), } diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 76695e5..024c302 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/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}; - -pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10; +use rusqlite::params; #[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, - 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, @@ -69,22 +58,6 @@ 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, @@ -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| { - let msg_id: i64 = tx.query_row( + db::with_db(|conn| { + 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( - tx, - storage_owner, - EntityType::Message, - msg_id, - Operation::Upsert, - )?; - Ok(()) - } else { - Ok(()) - } + sync::record_event( + &tx, + storage_owner, + EntityType::Message, + msg_id, + Operation::Upsert, + )?; + tx.commit()?; + Ok(()) }) } @@ -467,46 +422,45 @@ pub fn remove_reaction( }) } -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| { +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| { 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, 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) + storage_owner, external_user, message_time, content, + sent_by_self, message_state, height, reply_to + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) "#, params![ storage_owner, external_user, - send_time, - content, - i64::from(sent_by_self), - initial_state.as_str(), + message_time, + message, + if storage_owner_is_sender { + 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 { 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(¤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( @@ -727,15 +574,7 @@ fn load_reactions( )) }) { for row in rows.flatten() { - 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); - } + map.entry(row.0).or_default().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, - 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 + SELECT id, message_time, 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)?, - 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, + 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, 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, - 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, + SELECT id, message_time, 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)?, - 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)?, + 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)?, 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, 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(); @@ -928,28 +708,19 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec(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, + 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, reactions: Vec::new(), }) })?; diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 260c667..46363cf 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, 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(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. */ @@ -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( - "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;", - )?; + conn.execute_batch("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", diff --git a/iota-storage/src/util/relay_replay.rs b/iota-storage/src/util/relay_replay.rs index 0c5bddc..bc3bda7 100644 --- a/iota-storage/src/util/relay_replay.rs +++ b/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", - params![destination_id, i64::from(frame_id), delivered_at], + "UPDATE relay_inbox SET state = 'delivered' WHERE destination_id = ?1 AND frame_id = ?2", + params![destination_id, i64::from(frame_id)], )?; Ok(()) }) } -fn mark_transition( - signer_id: u64, - message_id: &str, - state: &str, - column: &str, -) -> Result<(), StorageError> { +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())); + } 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"), - params![signer_id, message_id, state, timestamp], + "UPDATE relay_inbox SET state = ?3 WHERE signer_id = ?1 AND message_id = ?2", + params![signer_id, message_id, state], )?; Ok(()) }) } -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> { +pub fn prune_completed(before_created_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')", - params![before_terminal_at], + "DELETE FROM relay_inbox WHERE created_at < ?1 AND state IN ('delivered', 'rejected')", + 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(()) }) diff --git a/iota-storage/src/util/sync.rs b/iota-storage/src/util/sync.rs index a358a1d..b84c94a 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 = 2; +pub const CACHE_SCHEMA_VERSION: i64 = 1; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EntityType { diff --git a/iota-updater/Cargo.toml b/iota-updater/Cargo.toml index 1fbbef9..9e19437 100644 --- a/iota-updater/Cargo.toml +++ b/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 = "*" diff --git a/iota-util/src/crypto_helper.rs b/iota-util/src/crypto_helper.rs index 339dd1f..1931da7 100644 --- a/iota-util/src/crypto_helper.rs +++ b/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 { @@ -18,10 +25,17 @@ 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 f3c5037..486541b 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a +Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 0c37013..9599cb6 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/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 { 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, - 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, 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 - .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); - } + self.send_relay_response(frame.id(), CommunicationType::Success) + .await; } 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( - &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, - ) + message_handlers::apply_verified_relay_content(&verified.context, &content) { 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( - frame.id(), - local_iota_id, - &verified.context.message_id, - accepted_at, - signer_is_local, - ) - .await; + self.send_relay_response(frame.id(), CommunicationType::Success) + .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 } }