diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index a43c308..f52f6f1 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -226,6 +226,11 @@ impl ClientConnection { // Direct messages // // ************************************************ // + if cv.is_type(CommunicationType::MessageState) { + message_handlers::handle_message_state(&cv); + return; + } + if cv.is_type(CommunicationType::MessageEdit) { self.send_message(&message_handlers::handle_message_edit(&cv)) .await; diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 7da9a1d..f912259 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -72,6 +72,7 @@ fn add_conversation_for_user( if let Some(name) = name { contact.user_name = Some(name.to_string()); } + contact.set_last_message_at(now_millis_i64()); mod_user(user_id, &contact) } @@ -227,11 +228,7 @@ pub fn apply_verified_relay_content( 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: if sent_by_self { - MessageState::Sending - } else { - MessageState::Sent - }, + initial_state: MessageState::Sent, }) .map_err(|error| error.to_string())?; Ok(()) @@ -433,12 +430,6 @@ fn stored_message_fields( DataValue::Str(relay_message_id.clone()), )); } - if let Some(relay_signer_id) = message.relay_signer_id { - fields.push(( - DataType::SenderId, - DataValue::SignedNumber(relay_signer_id.into()), - )); - } for (data_type, timestamp) in [ (DataType::AuthoredAt, message.authored_at), ( @@ -461,15 +452,6 @@ fn stored_message_fields( fields.push((data_type, DataValue::SignedNumber(timestamp.into()))); } } - if let Some(failed_at) = message.delivery_failed_at { - fields.push(( - DataType::UpdatedAt, - DataValue::SignedNumber(failed_at.into()), - )); - } - if let Some(failure) = &message.delivery_failure { - fields.push((DataType::ErrorType, DataValue::Str(failure.clone()))); - } if message.edited { fields.push((DataType::Edited, DataValue::Bool(true))); } @@ -662,12 +644,6 @@ fn contact_value( if let Some(name) = &contact.user_name { fields.push((DataType::Username, DataValue::Str(name.clone()))); } - if contact.created_at > 0 { - fields.push(( - DataType::CreatedAt, - DataValue::SignedNumber(contact.created_at.into()), - )); - } if let Some(last_message_at) = contact.last_message_at { fields.push(( DataType::LastMessageAt, @@ -942,6 +918,26 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { ) } +pub fn handle_message_state(cv: &CommunicationValue) { + let sender_id = match required_sender_id(cv) { + Ok(sender_id) => sender_id, + Err(_) => return, + }; + let receiver_id = match data_i64(cv, DataType::ChatPartnerId) { + Some(id) if id > 0 => id, + _ => return, + }; + + let timestamp_i64 = data_i64(cv, DataType::SendTime).unwrap_or_else(now_millis_i64); + + let _ = chat_files::change_message_state( + timestamp_i64, + receiver_id, + sender_id, + MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")), + ); +} + pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue { let my_id = match cv.require_sender() { Ok(my_id) => my_id, @@ -1030,12 +1026,6 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue { if let Some(name) = user.user_name { container.push((DataType::Username, DataValue::Str(name))); } - if user.created_at > 0 { - container.push(( - DataType::CreatedAt, - DataValue::SignedNumber(user.created_at.into()), - )); - } if let Some(ts) = user.last_message_at { container.push((DataType::LastMessageAt, DataValue::SignedNumber(ts as i128))); } diff --git a/iota-storage/src/users/contact.rs b/iota-storage/src/users/contact.rs index 4d2ca43..50d224d 100644 --- a/iota-storage/src/users/contact.rs +++ b/iota-storage/src/users/contact.rs @@ -1,32 +1,31 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + #[derive(Debug, Clone)] pub struct Contact { pub user_id: i64, pub user_name: Option, - pub created_at: i64, pub last_message_at: Option, } impl Default for Contact { fn default() -> Self { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64; Contact { user_id: 0, user_name: None, - created_at: 0, - last_message_at: None, + last_message_at: Some(now), } } } impl Contact { pub fn new(user_id: i64) -> Self { - let created_at = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64; Contact { user_id: user_id, user_name: None, - created_at, last_message_at: None, } } diff --git a/iota-storage/src/users/user_profile.rs b/iota-storage/src/users/user_profile.rs index f1aa2a3..5f2abd5 100644 --- a/iota-storage/src/users/user_profile.rs +++ b/iota-storage/src/users/user_profile.rs @@ -27,29 +27,6 @@ impl UserProfile { public_key: String, private_key_hash: String, reset_token: String, - ) -> Self { - Self::new_with_created_at( - user_id, - username, - display_name, - public_key, - private_key_hash, - reset_token, - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i64, - ) - } - - pub fn new_with_created_at( - user_id: i64, - username: String, - display_name: Option, - public_key: String, - private_key_hash: String, - reset_token: String, - created_at: i64, ) -> Self { Self { user_id, @@ -57,7 +34,10 @@ impl UserProfile { display_name, public_key, private_key_hash, - created_at, + created_at: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64, reset_token, trusted_apps: std::collections::HashMap::new(), } diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 051557b..94aefb7 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/iota-storage/src/util/chat_files.rs @@ -60,8 +60,6 @@ pub struct StoredMessage { pub client_received_recorded_at: Option, pub read_at: Option, pub read_recorded_at: Option, - pub delivery_failed_at: Option, - pub delivery_failure: Option, pub content: String, pub edited: bool, pub sent_by_self: bool, @@ -485,7 +483,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result { destination_iota_received_at, initial_state, } = message; - db::with_db(|conn| { + let msg_id = db::with_db(|conn| { let tx = conn.unchecked_transaction()?; tx.execute( r#" @@ -519,16 +517,14 @@ pub fn add_message(message: NewMessage<'_>) -> Result { msg_id, Operation::Upsert, )?; - let mut contact = crate::users::contact::Contact::new(external_user); - contact.set_last_message_at( - destination_iota_received_at - .or(origin_iota_received_at) - .unwrap_or(authored_at), - ); - crate::util::chats_util::upsert_contact(&tx, storage_owner, &contact)?; tx.commit()?; Ok(msg_id) - }) + })?; + + let mut contact = crate::users::contact::Contact::new(external_user); + 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( @@ -583,11 +579,11 @@ pub fn record_message_receipt( }; db::with_db(|conn| { let tx = conn.unchecked_transaction()?; - let Some((message_id, external_user, authored_at)) = tx + let Some((message_id, external_user)) = tx .query_row( - "SELECT id, external_user, authored_at FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", + "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)?, row.get::<_, Option>(2)?)), + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), ) .optional()? else { @@ -598,13 +594,6 @@ pub fn record_message_receipt( "message receipt signer is not the chat partner".into(), )); } - if event_at > recorded_at.saturating_add(5 * 60 * 1000) - || authored_at.is_some_and(|authored_at| event_at < authored_at) - { - return Err(StorageError::Other( - "message receipt event time is outside the accepted clock range".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], @@ -657,7 +646,7 @@ pub fn record_destination_iota_received( return Ok(()); }; tx.execute( - "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), delivery_failed_at = NULL, delivery_failure = NULL, message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2", + "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( @@ -672,41 +661,6 @@ pub fn record_destination_iota_received( }) } -pub fn record_delivery_failure( - storage_owner: i64, - relay_signer_id: i64, - relay_message_id: &str, - failure: &str, - failed_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 delivery_failed_at = ?1, delivery_failure = ?2 WHERE id = ?3 AND destination_iota_received_at IS NULL", - params![failed_at, failure, message_id], - )?; - sync::record_event( - &tx, - storage_owner, - EntityType::Message, - message_id, - Operation::Upsert, - )?; - tx.commit()?; - Ok(()) - }) -} - pub fn change_message_state( timestamp: i64, storage_owner: i64, @@ -829,11 +783,11 @@ pub fn get_messages( 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, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, + 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 COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) DESC, id DESC + ORDER BY message_time DESC, id DESC LIMIT ?3 OFFSET ?4 "#, )?; @@ -854,14 +808,12 @@ pub fn get_messages( client_received_recorded_at: row.get(8)?, read_at: row.get(9)?, read_recorded_at: row.get(10)?, - delivery_failed_at: row.get(11)?, - delivery_failure: row.get(12)?, - content: row.get(13)?, - sent_by_self: row.get::<_, i64>(14)? != 0, - message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, + 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(), }) }, @@ -902,7 +854,7 @@ pub fn get_message( 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, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, + read_recorded_at, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 @@ -926,15 +878,13 @@ pub fn get_message( client_received_recorded_at: row.get(8)?, read_at: row.get(9)?, read_recorded_at: row.get(10)?, - delivery_failed_at: row.get(11)?, - delivery_failure: row.get(12)?, - content: row.get(13)?, - sent_by_self: row.get::<_, i64>(14)? != 0, - message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, - external_user: row.get(19)?, + 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(), }) })?; @@ -978,16 +928,14 @@ pub fn get_message_with_offset( AND external_user = ?2 AND deleted_by_external = 0 AND ( - COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) > - COALESCE(?3, ?4) - OR (COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) = - COALESCE(?3, ?4) AND id > ?4) + message_time > ?3 + OR (message_time = ?3 AND id > ?4) ) "#, params![ storage_owner, external_user, - message.destination_iota_received_at.or(message.origin_iota_received_at).or(message.authored_at), + message.message_time, message.id ], |row| row.get(0), @@ -1005,9 +953,9 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec Vec(14)? != 0, - message_state: row.get(15)?, - height: row.get(16).unwrap_or(0), - reply_to: row.get(17).ok().flatten(), - edited: row.get::<_, i64>(18).unwrap_or(0) > 0, + 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/chats_util.rs b/iota-storage/src/util/chats_util.rs index 3eb4043..1fd3cb0 100644 --- a/iota-storage/src/util/chats_util.rs +++ b/iota-storage/src/util/chats_util.rs @@ -4,42 +4,6 @@ use crate::util::db; use crate::util::sync::{self, EntityType, Operation}; use rusqlite::params; -pub(crate) fn upsert_contact( - tx: &rusqlite::Transaction<'_>, - storage_owner: i64, - contact: &Contact, -) -> Result<(), StorageError> { - tx.execute( - r#" - INSERT INTO contacts (storage_owner, user_id, user_name, created_at, last_message_at) - VALUES (?1, ?2, ?3, ?4, ?5) - ON CONFLICT(storage_owner, user_id) DO UPDATE SET - user_name = COALESCE(excluded.user_name, contacts.user_name), - created_at = MIN(contacts.created_at, excluded.created_at), - last_message_at = CASE - WHEN excluded.last_message_at IS NULL THEN contacts.last_message_at - WHEN contacts.last_message_at IS NULL THEN excluded.last_message_at - ELSE MAX(contacts.last_message_at, excluded.last_message_at) - END - "#, - params![ - storage_owner, - contact.user_id, - contact.user_name, - contact.created_at, - contact.last_message_at, - ], - )?; - sync::record_event( - tx, - storage_owner, - EntityType::Contact, - contact.user_id, - Operation::Upsert, - )?; - Ok(()) -} - pub fn has_user(storage_owner: i64, user_id: i64) -> Result { db::with_db(|conn| { Ok(conn.query_row( @@ -51,7 +15,31 @@ pub fn has_user(storage_owner: i64, user_id: i64) -> Result } pub fn mod_user(storage_owner: i64, contact: &Contact) -> Result<(), StorageError> { - db::with_immediate_transaction(|tx| upsert_contact(tx, storage_owner, contact)) + db::with_immediate_transaction(|tx| { + tx.execute( + r#" + INSERT INTO contacts (storage_owner, user_id, user_name, last_message_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(storage_owner, user_id) DO UPDATE SET + user_name = excluded.user_name, + last_message_at = excluded.last_message_at + "#, + params![ + storage_owner, + contact.user_id, + contact.user_name, + contact.last_message_at, + ], + )?; + sync::record_event( + tx, + storage_owner, + EntityType::Contact, + contact.user_id, + Operation::Upsert, + )?; + Ok(()) + }) } pub fn get_users_by_ids(storage_owner: i64, ids: &[i64]) -> Result, StorageError> { @@ -69,7 +57,7 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result, Sto db::with_db(|conn| { match conn.query_row( r#" - SELECT user_id, user_name, created_at, last_message_at + SELECT user_id, user_name, last_message_at FROM contacts WHERE storage_owner = ?1 AND user_id = ?2 LIMIT 1 @@ -79,8 +67,7 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result, Sto Ok(Contact { user_id: r.get(0)?, user_name: r.get(1)?, - created_at: r.get(2)?, - last_message_at: r.get(3)?, + last_message_at: r.get(2)?, }) }, ) { @@ -95,7 +82,7 @@ pub fn get_users(storage_owner: i64) -> Result, StorageError> { db::with_db(|conn| { let mut stmt = conn.prepare( r#" - SELECT user_id, user_name, created_at, last_message_at + SELECT user_id, user_name, last_message_at FROM contacts WHERE storage_owner = ?1 ORDER BY @@ -109,8 +96,7 @@ pub fn get_users(storage_owner: i64) -> Result, StorageError> { Ok(Contact { user_id: r.get(0)?, user_name: r.get(1)?, - created_at: r.get(2)?, - last_message_at: r.get(3)?, + last_message_at: r.get(2)?, }) })?; diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index e8cbf6e..3e69a6f 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -180,7 +180,6 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { storage_owner INTEGER NOT NULL, user_id INTEGER NOT NULL, user_name TEXT, - created_at INTEGER NOT NULL, last_message_at INTEGER, UNIQUE(storage_owner, user_id) ); @@ -528,57 +527,6 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } - if current_version < 15 { - let messages_exist: bool = conn.query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", - [], - |row| row.get(0), - )?; - if messages_exist { - conn.execute_batch( - r#" - DROP INDEX IF EXISTS idx_messages_history; - CREATE INDEX IF NOT EXISTS idx_messages_history_accepted - ON messages ( - storage_owner, - external_user, - deleted_by_external, - destination_iota_received_at DESC, - origin_iota_received_at DESC, - authored_at DESC, - id DESC - ); - "#, - )?; - } - conn.pragma_update(None, "user_version", 15)?; - } - - if current_version < 16 { - let messages_exist: bool = conn.query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", - [], - |row| row.get(0), - )?; - if messages_exist { - add_column_if_missing(conn, "delivery_failed_at", "delivery_failed_at INTEGER")?; - add_column_if_missing(conn, "delivery_failure", "delivery_failure TEXT")?; - } - let contacts_exist: bool = conn.query_row( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'contacts')", - [], - |row| row.get(0), - )?; - if contacts_exist { - add_table_column_if_missing(conn, "contacts", "created_at", "created_at INTEGER")?; - conn.execute( - "UPDATE contacts SET created_at = COALESCE(created_at, last_message_at, 0)", - [], - )?; - } - conn.pragma_update(None, "user_version", 16)?; - } - Ok(()) } @@ -648,7 +596,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, 16); + assert_eq!(version, 13); 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")?; @@ -667,7 +615,7 @@ mod tests { 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, 16); + assert_eq!(version, 13); for table in [ "sync_heads", "sync_events", @@ -702,7 +650,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, 16); + assert_eq!(version, 13); for column in [ "id", "user_id", diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index adb18ee..5513efa 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -40,21 +40,6 @@ const IOTA_KEYRING_PATH: &str = "iota.mk"; static IDENTITY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); -fn record_origin_delivery_failure(signer_id: u64, relay_message_id: &str, failure: &str) { - let Ok(storage_owner) = i64::try_from(signer_id) else { - return; - }; - if let Err(error) = chat_files::record_delivery_failure( - storage_owner, - storage_owner, - relay_message_id, - failure, - now_millis_i64(), - ) { - log!("Relay delivery failure storage failed: {error}"); - } -} - /* * Keeps identity and pinned Omikron key files independent from the process * working directory, so restarts use the same trusted material. @@ -1151,11 +1136,6 @@ impl OmikronConnection { Ok(destination_iota) => destination_iota, Err(error) => { log!("Relay origin route lookup failed: {}", error); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "destination_iota_not_found", - ); self.send_relay_response(frame.id(), CommunicationType::ErrorNoIota) .await; return; @@ -1165,11 +1145,6 @@ impl OmikronConnection { Ok(value) => value, Err(error) => { log!("Relay origin forwarding validation failed: {}", error); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "forwarding_validation_failed", - ); self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) .await; return; @@ -1179,11 +1154,6 @@ impl OmikronConnection { Ok(bytes) => bytes, Err(error) => { log!("Relay origin retry could not be serialized: {}", error); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "serialization_failed", - ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; return; @@ -1197,11 +1167,6 @@ impl OmikronConnection { &type_map_version, ) { log!("Relay origin retry queue failed: {}", error); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "queue_failed", - ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; return; @@ -1277,11 +1242,6 @@ impl OmikronConnection { } Ok(response) => { log!("Relay origin route returned {}", response.get_type()); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "destination_rejected", - ); self.send_relay_response( frame.id(), response @@ -1292,11 +1252,6 @@ impl OmikronConnection { } Err(error) => { log!("Relay origin forwarding failed: {}", error); - record_origin_delivery_failure( - verified.context.signer_id, - &verified.context.message_id, - "destination_unreachable", - ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; } @@ -1566,6 +1521,7 @@ impl OmikronConnection { dispatch!(DeleteApp, handle_delete_app); dispatch!(ClientConnected, handle_client_connected); dispatch!(ClientStateAck, handle_client_state_ack); + dispatch!(MessageState, handle_message_state); dispatch!(MessageEdit, handle_message_edit); dispatch!(MessageEditLive, handle_message_edit_live); dispatch!(MessageReactionAdd, handle_message_reaction_add); @@ -1815,6 +1771,10 @@ impl OmikronConnection { .await; } + async fn handle_message_state(self: Arc, cv: &CommunicationValue) { + message_handlers::handle_message_state(cv); + } + fn mutation_live_message( ty: CommunicationType, request: &CommunicationValue, diff --git a/omikron-connector/src/user_ops.rs b/omikron-connector/src/user_ops.rs index 5c33416..b140358 100644 --- a/omikron-connector/src/user_ops.rs +++ b/omikron-connector/src/user_ops.rs @@ -84,7 +84,7 @@ fn sign_lifecycle_payload( async fn inspect_credential_account( connection: &dyn OmikronClient, credential: &TuCredential, -) -> Result<(String, String, i64), LifecycleUserError> { +) -> Result<(String, String), LifecycleUserError> { if credential.omega_host != omega_discovery::omega_host() { return Err(LifecycleUserError::OmegaHostMismatch); } @@ -108,16 +108,10 @@ async fn inspect_credential_account( .as_str() .map(str::to_owned) .ok_or(LifecycleUserError::RemoteRejected)?; - let created_at = response - .get_data(DataType::CreatedAt) - .as_signed_number() - .and_then(|value| i64::try_from(value).ok()) - .filter(|value| *value > 0) - .ok_or(LifecycleUserError::RemoteRejected)?; if public_key != public_key_bundle_to_base64(&credential.public_key_bundle()) { return Err(LifecycleUserError::RemoteRejected); } - Ok((username, public_key, created_at)) + Ok((username, public_key)) } async fn credential_proof( @@ -174,17 +168,17 @@ pub async fn attach_user_from_tu( ) -> Result { let credential = TuCredential::parse(contents) .map_err(|error| LifecycleUserError::InvalidCredential(error.to_string()))?; - let (username, public_key, created_at) = - inspect_credential_account(connection, &credential).await?; - let profile = UserProfile::new_with_created_at( + let (username, public_key) = inspect_credential_account(connection, &credential).await?; + let profile = UserProfile::new( credential.user_id, username, None, public_key, hex_hash(contents), String::new(), - created_at, ); + write_user_credential(&profile.username, &credential.to_canonical_string()) + .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; pending_operations::upsert(&PendingUserOperation { user_id: profile.user_id, operation: PendingUserOperationKind::Attach, @@ -197,13 +191,6 @@ pub async fn attach_user_from_tu( created_at: now_millis(), }) .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; - write_user_credential(&profile.username, &credential.to_canonical_string()) - .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; - pending_operations::update_phase( - profile.user_id, - PendingUserOperationPhase::CredentialWritten, - ) - .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; if let Err(error) = credential_proof( connection, &credential,