From 430c12e139a5927e56ace6cd8099bbbfe9df8327 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:17:57 +0200 Subject: [PATCH] [Add] Client settings --- iota-connection/src/message_handlers.rs | 631 +++++++++++++++++- iota-daemon-lib/src/command_router.rs | 6 + iota-daemon/src/main.rs | 62 ++ iota-storage/src/storage_error.rs | 4 + iota-storage/src/users/user_manager.rs | 39 ++ iota-storage/src/util/blocked_users.rs | 104 +++ iota-storage/src/util/chat_files.rs | 225 +++++-- .../src/util/client_message_delivery.rs | 93 +++ .../src/util/client_relay_delivery.rs | 39 ++ iota-storage/src/util/db.rs | 141 +++- iota-storage/src/util/downstream_relay.rs | 69 ++ iota-storage/src/util/message_retention.rs | 28 + .../src/util/message_storage_policy.rs | 149 +++++ iota-storage/src/util/mod.rs | 9 + iota-storage/src/util/outgoing_relay.rs | 61 ++ iota-storage/src/util/receipt_policy.rs | 64 ++ iota-storage/src/util/relay_queue.rs | 151 ++++- iota-storage/src/util/relay_replay.rs | 59 +- iota-storage/src/util/sync.rs | 122 +++- iota-storage/src/util/user_blobs.rs | 256 +++++++ mtp-type-maps | 2 +- omikron-connector/src/omikron_connection.rs | 592 ++++++++++++++-- 22 files changed, 2779 insertions(+), 127 deletions(-) create mode 100644 iota-storage/src/util/blocked_users.rs create mode 100644 iota-storage/src/util/client_message_delivery.rs create mode 100644 iota-storage/src/util/client_relay_delivery.rs create mode 100644 iota-storage/src/util/downstream_relay.rs create mode 100644 iota-storage/src/util/message_retention.rs create mode 100644 iota-storage/src/util/message_storage_policy.rs create mode 100644 iota-storage/src/util/outgoing_relay.rs create mode 100644 iota-storage/src/util/receipt_policy.rs create mode 100644 iota-storage/src/util/user_blobs.rs diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 7da9a1d..438e744 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -5,12 +5,21 @@ use iota_storage::util::communities_util::CommunitiesUtil; use iota_storage::util::e2ee_storage::{self, ChatSecretQuery}; use iota_storage::util::settings; use iota_storage::util::synced_settings::{self, SettingScope, SyncedSetting}; +use iota_storage::util::user_blobs::{self, UserBlob}; +use iota_storage::util::{blocked_users, message_storage_policy, receipt_policy}; use mtp::codec::{ CommunicationType, CommunicationValue, DataType, DataValue, TypeMap, VerifiedRelayContent, }; use crate::relay::VerifiedRelayContext; use iota_storage::storage_error::StorageError; +use std::sync::atomic::{AtomicU32, Ordering}; + +static NEXT_NOTIFICATION_ID: AtomicU32 = AtomicU32::new(1); + +fn next_notification_id() -> u32 { + NEXT_NOTIFICATION_ID.fetch_add(1, Ordering::Relaxed).max(1) +} #[derive(Debug)] pub struct MessageMutation { @@ -25,6 +34,18 @@ pub struct SettingMutation { pub changed: Option, } +#[derive(Debug)] +pub struct BlobMutation { + pub response: CommunicationValue, + pub changed: Option, +} + +#[derive(Debug)] +pub struct PolicyMutation { + pub response: CommunicationValue, + pub changed: Option, +} + struct SettingLocator { scope: SettingScope, scope_key: String, @@ -174,6 +195,15 @@ pub fn apply_verified_relay_content( .map(MessageState::from_str) .filter(|state| matches!(state, MessageState::Received | MessageState::Read)) .ok_or_else(|| "Relay MessageState has an invalid state".to_string())?; + if sent_by_self { + return chat_files::change_message_state_by_relay_id( + storage_owner, + recipient_id, + relay_message_id, + state, + ) + .map_err(|error| error.to_string()); + } chat_files::record_message_receipt( storage_owner, recipient_id, @@ -323,6 +353,32 @@ pub fn apply_verified_relay_content( } } +/* Local receipt disclosure is checked before the relay enters durable state. */ +pub fn validate_outgoing_receipt_policy( + sender_id: i64, + context: &VerifiedRelayContext, + content: &VerifiedRelayContent, +) -> Result<(), String> { + if content.message_type != "MessageState" { + return Ok(()); + } + 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(|| "local MessageState has an invalid state".to_string())?; + let policy = receipt_policy::get(sender_id).map_err(|error| error.to_string())?; + let allowed = match state { + MessageState::Read => policy.send_read_receipts, + MessageState::Received => policy.send_received_receipts, + MessageState::Sent | MessageState::Sending => false, + }; + if allowed { + Ok(()) + } else { + Err("local receipt disclosure is disabled by policy".into()) + } +} + pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue { let mutation = match message_mutation(cv) { Ok(mutation) => mutation, @@ -777,9 +833,11 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { Err(_) => return sync_error(cv), }; let known_session = sync::has_session(user_id, session_id).unwrap_or(false); + let acknowledged_version = sync::acknowledged_version(user_id, session_id).unwrap_or(None); let full = !cache_valid || reported_version == 0 || !known_session + || acknowledged_version.is_some_and(|version| reported_version < version) || reported_version > head || schema != CACHE_SCHEMA_VERSION; let (contacts, messages, settings, deleted_messages, deleted_contacts, deleted_settings, mode) = @@ -845,11 +903,59 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { .iter() .map(|message| stored_message_value(message, user_id, message.external_user)) .collect(); + if iota_storage::util::client_message_delivery::record_sync_delivery( + user_id, + session_id, + head, + messages.iter().map(|message| message.id), + ) + .is_err() + { + return sync_error(cv); + } + let (blobs, deleted_blobs) = if mode == "delta" { + match sync::delta(user_id, reported_version, head) { + Ok(delta) => { + let blobs = match user_blobs::list_by_ids(user_id, &delta.blob_upserts) { + Ok(blobs) => blobs, + Err(_) => return sync_error(cv), + }; + let deleted = + match user_blobs::list_deleted_by_ids(user_id, &delta.deleted_blob_ids) { + Ok(blobs) => blobs, + Err(_) => return sync_error(cv), + }; + (blobs, deleted) + } + Err(_) => return sync_error(cv), + } + } else { + match user_blobs::list(user_id) { + Ok(blobs) => (blobs, Vec::new()), + Err(_) => return sync_error(cv), + } + }; + let blocked_users = match blocked_users::list(user_id) { + Ok(users) => users, + Err(_) => return sync_error(cv), + }; + let receipt_policy = match receipt_policy::get(user_id) { + Ok(policy) => policy, + Err(_) => return sync_error(cv), + }; + let message_storage_policy = match message_storage_policy::get(user_id) { + Ok(policy) => policy, + Err(_) => return sync_error(cv), + }; let contact_ids = match current_contact_ids(user_id) { Ok(contact_ids) => contact_ids, Err(_) => return error_response(cv, CommunicationType::ErrorInternal), }; - CommunicationValue::new(CommunicationType::ClientStateSync) + let retention_duration = match message_storage_policy.retention { + message_storage_policy::MessageRetention::Forever => None, + message_storage_policy::MessageRetention::Duration { duration_ms } => Some(duration_ms), + }; + let mut response = CommunicationValue::new(CommunicationType::ClientStateSync) .with_request_id(cv) .with_receiver(sender_wire_id(user_id)) .add_typed_default( @@ -879,6 +985,45 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { DataType::Settings, DataValue::Array(settings.iter().map(synced_setting_value).collect()), ) + .add_typed_default( + DataType::Blobs, + DataValue::Array(blobs.iter().map(blob_value).collect()), + ) + .add_typed_default( + DataType::DeletedBlobIds, + DataValue::Array( + deleted_blobs + .iter() + .map(|blob| DataValue::Str(blob.blob_id.clone())) + .collect(), + ), + ) + .add_typed_default( + DataType::BlockedUserIds, + DataValue::Array( + blocked_users + .into_iter() + .map(|id| DataValue::SignedNumber(id.into())) + .collect(), + ), + ) + .add_typed_default( + DataType::SendReadReceipts, + DataValue::Bool(receipt_policy.send_read_receipts), + ) + .add_typed_default( + DataType::SendReceivedReceipts, + DataValue::Bool(receipt_policy.send_received_receipts), + ) + .add_typed_default( + DataType::MessageHistoryMode, + DataValue::Str(match message_storage_policy.history_mode { + message_storage_policy::MessageHistoryMode::Retain => "retain".into(), + message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery => { + "delete_after_client_delivery".into() + } + }), + ) .add_typed_default( DataType::Communities, DataValue::Array(community_values(user_id)), @@ -911,11 +1056,18 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { ), ) .add_typed_default(DataType::UserIds, contact_ids) - .add_typed_default(DataType::Calls, DataValue::Array(Vec::new())) + .add_typed_default(DataType::Calls, DataValue::Array(Vec::new())); + if let Some(duration_ms) = retention_duration { + response = response.add_typed_default( + DataType::MessageRetentionDuration, + DataValue::SignedNumber(duration_ms.into()), + ); + } + response } pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { - use iota_storage::util::sync::{self, CACHE_SCHEMA_VERSION}; + use iota_storage::util::sync::CACHE_SCHEMA_VERSION; let user_id = match required_sender_id(cv) { Ok(id) if id > 0 => id, _ => return sync_error(cv), @@ -928,7 +1080,14 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { Some(version) if version >= 0 => version, _ => return sync_error(cv), }; - if sync::acknowledge(user_id, session_id, version, CACHE_SCHEMA_VERSION).is_err() { + if iota_storage::util::client_message_delivery::acknowledge_client_state( + user_id, + session_id, + version, + CACHE_SCHEMA_VERSION, + ) + .is_err() + { return sync_error(cv); } success_response(cv) @@ -1483,6 +1642,66 @@ fn setting_response( ) } +fn blob_value(blob: &UserBlob) -> DataValue { + typed_container(vec![ + (DataType::BlobId, DataValue::Str(blob.blob_id.clone())), + (DataType::Blob, DataValue::Bytes(blob.blob.clone())), + ( + DataType::VersionNumber, + DataValue::SignedNumber(blob.revision.into()), + ), + ( + DataType::UpdatedAt, + DataValue::SignedNumber(blob.updated_at.into()), + ), + ]) +} + +fn blob_response( + cv: &CommunicationValue, + ty: CommunicationType, + blob: &UserBlob, +) -> CommunicationValue { + CommunicationValue::new(ty) + .with_request_id(cv) + .with_receiver(sender_wire_id(blob.user_id)) + .add_typed_default(DataType::BlobId, DataValue::Str(blob.blob_id.clone())) + .add_typed_default(DataType::Blob, DataValue::Bytes(blob.blob.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(blob.revision.into()), + ) + .add_typed_default( + DataType::UpdatedAt, + DataValue::SignedNumber(blob.updated_at.into()), + ) +} + +fn blob_changed(user_id: i64, blob_id: String, revision: i64, deleted: bool) -> CommunicationValue { + CommunicationValue::new(CommunicationType::UserBlobChanged) + .with_id(next_notification_id()) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default(DataType::BlobId, DataValue::Str(blob_id)) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(revision.into()), + ) + .add_typed_default(DataType::Deleted, DataValue::Bool(deleted)) +} + +fn blob_request(cv: &CommunicationValue) -> Result<(i64, String), CommunicationValue> { + let user_id = required_sender_id(cv)?; + if user_id <= 0 { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + } + let id = cv + .get_data(DataType::BlobId) + .and_then(DataValue::as_str) + .filter(|id| !id.is_empty()) + .ok_or_else(|| error_response(cv, CommunicationType::ErrorInvalidData))?; + Ok((user_id, id.to_owned())) +} + fn setting_changed(setting: &SyncedSetting) -> CommunicationValue { CommunicationValue::new(CommunicationType::SyncedSettingChanged) .with_receiver(sender_wire_id(setting.user_id)) @@ -1703,6 +1922,410 @@ pub fn handle_synced_settings_list(cv: &CommunicationValue) -> CommunicationValu } } +pub fn handle_user_blob_put(cv: &CommunicationValue) -> BlobMutation { + let Ok((user_id, blob_id)) = blob_request(cv) else { + return BlobMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let Some(blob) = cv.get_data(DataType::Blob).and_then(DataValue::as_bytes) else { + return BlobMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let expected_revision = data_i64(cv, DataType::ExpectedRevision); + match user_blobs::put(user_id, &blob_id, &blob, expected_revision) { + Ok(stored) => BlobMutation { + response: blob_response(cv, CommunicationType::UserBlobPut, &stored), + changed: Some(blob_changed( + user_id, + stored.blob_id.clone(), + stored.revision, + false, + )), + }, + Err(iota_storage::storage_error::StorageError::RevisionConflict) => BlobMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }, + Err(_) => BlobMutation { + response: error_response(cv, CommunicationType::ErrorInternal), + changed: None, + }, + } +} + +pub fn handle_user_blob_get(cv: &CommunicationValue) -> CommunicationValue { + let Ok((user_id, blob_id)) = blob_request(cv) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match user_blobs::get(user_id, &blob_id) { + Ok(Some(blob)) => blob_response(cv, CommunicationType::UserBlobGet, &blob), + Ok(None) => error_response(cv, CommunicationType::ErrorNotFound), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +pub fn handle_user_blob_delete(cv: &CommunicationValue) -> BlobMutation { + let Ok((user_id, blob_id)) = blob_request(cv) else { + return BlobMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + match user_blobs::delete(user_id, &blob_id, data_i64(cv, DataType::ExpectedRevision)) { + Ok(Some(deleted)) => BlobMutation { + response: CommunicationValue::new(CommunicationType::UserBlobDelete) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default(DataType::BlobId, DataValue::Str(blob_id.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(deleted.revision.into()), + ), + changed: deleted + .changed + .then(|| blob_changed(user_id, blob_id, deleted.revision, true)), + }, + Ok(None) => BlobMutation { + response: CommunicationValue::new(CommunicationType::UserBlobDelete) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)), + changed: None, + }, + Err(iota_storage::storage_error::StorageError::RevisionConflict) => BlobMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }, + Err(_) => BlobMutation { + response: error_response(cv, CommunicationType::ErrorInternal), + changed: None, + }, + } +} + +pub fn handle_user_blob_list(cv: &CommunicationValue) -> CommunicationValue { + let Ok(user_id) = required_sender_id(cv) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + if user_id <= 0 { + return error_response(cv, CommunicationType::ErrorInvalidData); + } + match user_blobs::list(user_id) { + Ok(blobs) => CommunicationValue::new(CommunicationType::UserBlobList) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::Blobs, + DataValue::Array(blobs.iter().map(blob_value).collect()), + ), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +fn authenticated_user(cv: &CommunicationValue) -> Result { + let user_id = required_sender_id(cv)?; + if user_id <= 0 { + return Err(error_response(cv, CommunicationType::ErrorInvalidData)); + } + Ok(user_id) +} + +pub fn handle_user_block(cv: &CommunicationValue) -> PolicyMutation { + let Ok(user_id) = authenticated_user(cv) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let Some(blocked_user_id) = data_i64(cv, DataType::BlockedUserId).filter(|id| *id > 0) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + match blocked_users::block(user_id, blocked_user_id) { + Ok(record) => PolicyMutation { + response: CommunicationValue::new(CommunicationType::UserBlock) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::BlockedUserId, + DataValue::SignedNumber(blocked_user_id.into()), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(record.revision.into()), + ), + changed: Some( + CommunicationValue::new(CommunicationType::BlockedUsersChanged) + .with_id(next_notification_id()) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::BlockedUserId, + DataValue::SignedNumber(blocked_user_id.into()), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(record.revision.into()), + ) + .add_typed_default(DataType::Deleted, DataValue::Bool(record.deleted)), + ), + }, + Err(_) => PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInternal), + changed: None, + }, + } +} + +pub fn handle_user_unblock(cv: &CommunicationValue) -> PolicyMutation { + let Ok(user_id) = authenticated_user(cv) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let Some(blocked_user_id) = data_i64(cv, DataType::BlockedUserId).filter(|id| *id > 0) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + match blocked_users::unblock(user_id, blocked_user_id) { + Ok(mutation) => PolicyMutation { + response: CommunicationValue::new(CommunicationType::UserUnblock) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::BlockedUserId, + DataValue::SignedNumber(blocked_user_id.into()), + ), + changed: mutation.map(|mutation| { + CommunicationValue::new(CommunicationType::BlockedUsersChanged) + .with_id(next_notification_id()) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::BlockedUserId, + DataValue::SignedNumber(blocked_user_id.into()), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(mutation.revision.into()), + ) + .add_typed_default(DataType::Deleted, DataValue::Bool(mutation.deleted)) + }), + }, + Err(_) => PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInternal), + changed: None, + }, + } +} + +pub fn handle_blocked_users_get(cv: &CommunicationValue) -> CommunicationValue { + let Ok(user_id) = authenticated_user(cv) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match blocked_users::list(user_id) { + Ok(users) => CommunicationValue::new(CommunicationType::BlockedUsersGet) + .with_request_id(cv) + .with_receiver(sender_wire_id(user_id)) + .add_typed_default( + DataType::BlockedUserIds, + DataValue::Array( + users + .into_iter() + .map(|id| DataValue::SignedNumber(id.into())) + .collect(), + ), + ), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +fn receipt_response( + cv: &CommunicationValue, + ty: CommunicationType, + policy: receipt_policy::ReceiptPolicy, +) -> CommunicationValue { + CommunicationValue::new(ty) + .with_request_id(cv) + .with_receiver(sender_wire_id(policy.user_id)) + .add_typed_default( + DataType::SendReadReceipts, + DataValue::Bool(policy.send_read_receipts), + ) + .add_typed_default( + DataType::SendReceivedReceipts, + DataValue::Bool(policy.send_received_receipts), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(policy.revision.into()), + ) +} +pub fn handle_receipt_policy_get(cv: &CommunicationValue) -> CommunicationValue { + let Ok(user_id) = authenticated_user(cv) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match receipt_policy::get(user_id) { + Ok(policy) => receipt_response(cv, CommunicationType::ReceiptPolicyGet, policy), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} +pub fn handle_receipt_policy_set(cv: &CommunicationValue) -> PolicyMutation { + let Ok(user_id) = authenticated_user(cv) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let (Some(read), Some(received)) = ( + cv.get_data(DataType::SendReadReceipts) + .and_then(DataValue::as_bool), + cv.get_data(DataType::SendReceivedReceipts) + .and_then(DataValue::as_bool), + ) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + match receipt_policy::set(user_id, read, received) { + Ok(policy) => { + let response = receipt_response(cv, CommunicationType::ReceiptPolicySet, policy); + let changed = receipt_response(cv, CommunicationType::ReceiptPolicyChanged, policy) + .with_id(next_notification_id()); + PolicyMutation { + response, + changed: Some(changed), + } + } + Err(_) => PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInternal), + changed: None, + }, + } +} + +fn storage_policy_response( + cv: &CommunicationValue, + ty: CommunicationType, + policy: message_storage_policy::MessageStoragePolicy, +) -> CommunicationValue { + let (history_mode, duration) = match policy.history_mode { + message_storage_policy::MessageHistoryMode::Retain => ("retain", None), + message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery => { + ("delete_after_client_delivery", None) + } + }; + let retention_duration = match policy.retention { + message_storage_policy::MessageRetention::Forever => None, + message_storage_policy::MessageRetention::Duration { duration_ms } => Some(duration_ms), + }; + let mut response = CommunicationValue::new(ty) + .with_request_id(cv) + .with_receiver(sender_wire_id(policy.user_id)) + .add_typed_default( + DataType::MessageHistoryMode, + DataValue::Str(history_mode.into()), + ) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(policy.revision.into()), + ); + if let Some(duration_ms) = retention_duration.or(duration) { + response = response.add_typed_default( + DataType::MessageRetentionDuration, + DataValue::SignedNumber(duration_ms.into()), + ); + } + response +} + +pub fn handle_message_storage_policy_get(cv: &CommunicationValue) -> CommunicationValue { + let Ok(user_id) = authenticated_user(cv) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match message_storage_policy::get(user_id) { + Ok(policy) => { + storage_policy_response(cv, CommunicationType::MessageStoragePolicyGet, policy) + } + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + +pub fn handle_message_storage_policy_set(cv: &CommunicationValue) -> PolicyMutation { + let Ok(user_id) = authenticated_user(cv) else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let Some(history_mode) = cv + .get_data(DataType::MessageHistoryMode) + .and_then(DataValue::as_str) + else { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + }; + let history_mode = match history_mode { + "retain" => message_storage_policy::MessageHistoryMode::Retain, + "delete_after_client_delivery" => { + message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery + } + _ => { + return PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }; + } + }; + let retention = match data_i64(cv, DataType::MessageRetentionDuration) { + Some(duration_ms) => message_storage_policy::MessageRetention::Duration { duration_ms }, + None => message_storage_policy::MessageRetention::Forever, + }; + match message_storage_policy::set(user_id, history_mode, retention) { + Ok(policy) => { + let response = + storage_policy_response(cv, CommunicationType::MessageStoragePolicySet, policy); + let changed = + storage_policy_response(cv, CommunicationType::MessageStoragePolicyChanged, policy) + .with_id(next_notification_id()); + PolicyMutation { + response, + changed: Some(changed), + } + } + Err(_) => PolicyMutation { + response: error_response(cv, CommunicationType::ErrorInvalidData), + changed: None, + }, + } +} + +/* This request is issued over Omikron's trusted Iota connection. Clients have + * no route to it, preventing arbitrary block-relationship disclosure. */ +pub fn handle_user_block_check(cv: &CommunicationValue) -> CommunicationValue { + let Some(sender_id) = data_i64(cv, DataType::SenderId).filter(|id| *id > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + let Some(receiver_id) = data_i64(cv, DataType::ReceiverId).filter(|id| *id > 0) else { + return error_response(cv, CommunicationType::ErrorInvalidData); + }; + match blocked_users::is_blocked(receiver_id, sender_id) { + Ok(blocked) => CommunicationValue::new(CommunicationType::UserBlockCheck) + .with_request_id(cv) + .add_typed_default(DataType::IsBlocked, DataValue::Bool(blocked)), + Err(_) => error_response(cv, CommunicationType::ErrorInternal), + } +} + #[cfg(test)] mod synced_settings_tests { use super::{handle_synced_setting_get, handle_synced_setting_set, parse_setting_locator}; diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs index 2e3c3e0..59a87f9 100644 --- a/iota-daemon-lib/src/command_router.rs +++ b/iota-daemon-lib/src/command_router.rs @@ -238,6 +238,12 @@ impl CommandRouter { LocalRequest::PurgeUserData { user_id } => match user_manager::purge_user_data(user_id) { Ok(()) => ResponseResult::Ok(ResponsePayload::UserDataPurged { user_id }), + Err(iota_storage::storage_error::StorageError::PendingRelayOwnershipUnknown) => { + log!( + "User data purge is waiting for pending relay ownership classification for {user_id}" + ); + ResponseResult::Error(IpcErrorCode::StorageFailure) + } Err(error) => { log!("User data purge failed for {user_id}: {error}"); ResponseResult::Error(IpcErrorCode::StorageFailure) diff --git a/iota-daemon/src/main.rs b/iota-daemon/src/main.rs index d2ff2c6..4ebe682 100644 --- a/iota-daemon/src/main.rs +++ b/iota-daemon/src/main.rs @@ -9,6 +9,9 @@ use std::process::ExitCode; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{broadcast, watch}; + +const MESSAGE_RETENTION_INTERVAL: Duration = Duration::from_secs(60); +const SYNC_COMPACTION_INTERVAL: Duration = Duration::from_secs(60 * 60); #[tokio::main(flavor = "multi_thread")] async fn main() -> ExitCode { let scope = match std::env::var("IOTA_DEPLOYMENT_MODE").ok().as_deref() { @@ -357,6 +360,65 @@ async fn main() -> ExitCode { runtime.set_startup_phase(StartupPhase::Ready); log!("iota-daemon started (phase: Ready)"); + let retention_runtime = runtime.clone(); + runtime + .tasks + .spawn_tracked("message-retention", async move { + loop { + let purge = tokio::task::spawn_blocking(|| { + iota_storage::util::message_retention::purge_expired_messages( + iota_storage::util::sync::now_millis(), + ) + }) + .await; + match purge { + Ok(Ok(result)) if result.deleted_messages > 0 => { + log!("purged {} expired messages", result.deleted_messages); + } + Ok(Ok(_)) => {} + Ok(Err(error)) => log!("message retention cleanup failed: {}", error), + Err(error) => log!("message retention task failed: {}", error), + } + tokio::select! { + _ = tokio::time::sleep(MESSAGE_RETENTION_INTERVAL) => {}, + _ = retention_runtime.cancellation.cancelled() => break, + } + } + Ok(()) + }) + .await; + + let compaction_runtime = runtime.clone(); + runtime + .tasks + .spawn_tracked("sync-compaction", async move { + loop { + let compact = + tokio::task::spawn_blocking(iota_storage::util::sync::compact_all_sync_state) + .await; + match compact { + Ok(Ok(result)) + if result.removed_events > 0 || result.removed_blob_tombstones > 0 => + { + log!( + "compacted {} sync events and {} blob tombstones", + result.removed_events, + result.removed_blob_tombstones + ); + } + Ok(Ok(_)) => {} + Ok(Err(error)) => log!("sync compaction failed: {}", error), + Err(error) => log!("sync compaction task failed: {}", error), + } + tokio::select! { + _ = tokio::time::sleep(SYNC_COMPACTION_INTERVAL) => {}, + _ = compaction_runtime.cancellation.cancelled() => break, + } + } + Ok(()) + }) + .await; + // --- Main lifecycle loop --- let signal = async { #[cfg(unix)] diff --git a/iota-storage/src/storage_error.rs b/iota-storage/src/storage_error.rs index 86a0c53..f407b7e 100644 --- a/iota-storage/src/storage_error.rs +++ b/iota-storage/src/storage_error.rs @@ -10,6 +10,10 @@ pub enum StorageError { Io(#[from] std::io::Error), #[error("message has reached the unique reaction limit")] ReactionLimitReached, + #[error("revision conflict")] + RevisionConflict, + #[error("pending relay ownership is unknown")] + PendingRelayOwnershipUnknown, #[error("{0}")] Other(String), } diff --git a/iota-storage/src/users/user_manager.rs b/iota-storage/src/users/user_manager.rs index 90068f4..43b2ed4 100644 --- a/iota-storage/src/users/user_manager.rs +++ b/iota-storage/src/users/user_manager.rs @@ -255,8 +255,15 @@ pub fn release_user(user_id: i64) -> Result<(), crate::storage_error::StorageErr /// Authoritative hosted-data erasure used by local purge and future Omega /// erasure delivery. Management metadata and credentials are left intact. pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::StorageError> { + if crate::util::relay_queue::has_unclassified_relays()? { + return Err(crate::storage_error::StorageError::PendingRelayOwnershipUnknown); + } db::with_db(|conn| { let tx = conn.unchecked_transaction()?; + tx.execute( + "DELETE FROM message_receipts WHERE storage_owner = ?1", + params![user_id], + )?; tx.execute("DELETE FROM message_edits WHERE message_id IN (SELECT id FROM messages WHERE storage_owner = ?1)", params![user_id])?; tx.execute("DELETE FROM reactions WHERE message_id IN (SELECT id FROM messages WHERE storage_owner = ?1)", params![user_id])?; tx.execute( @@ -276,6 +283,22 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage "DELETE FROM synced_settings WHERE user_id = ?1", params![user_id], )?; + tx.execute( + "DELETE FROM user_blobs WHERE user_id = ?1", + params![user_id], + )?; + tx.execute( + "DELETE FROM blocked_users WHERE user_id = ?1", + params![user_id], + )?; + tx.execute( + "DELETE FROM user_receipt_policy WHERE user_id = ?1", + params![user_id], + )?; + tx.execute( + "DELETE FROM user_message_storage_policy WHERE user_id = ?1", + params![user_id], + )?; tx.execute( "DELETE FROM sync_events WHERE user_id = ?1", params![user_id], @@ -288,10 +311,26 @@ pub fn purge_user_data(user_id: i64) -> Result<(), crate::storage_error::Storage "DELETE FROM client_sync_state WHERE user_id = ?1", params![user_id], )?; + tx.execute( + "DELETE FROM client_message_deliveries WHERE user_id = ?1", + params![user_id], + )?; tx.execute( "DELETE FROM trusted_apps WHERE user_id = ?1", params![user_id], )?; + tx.execute( + "DELETE FROM pending_relays WHERE relay_signer_id = ?1 OR relay_destination_user_id = ?1", + params![user_id], + )?; + tx.execute( + "DELETE FROM relay_replay WHERE EXISTS (SELECT 1 FROM relay_inbox WHERE relay_inbox.signer_id = relay_replay.signer_id AND relay_inbox.message_id = relay_replay.message_id AND (relay_inbox.signer_id = ?1 OR relay_inbox.destination_id = ?1))", + params![user_id], + )?; + tx.execute( + "DELETE FROM relay_inbox WHERE signer_id = ?1 OR destination_id = ?1", + params![user_id], + )?; tx.execute( "UPDATE user_residency SET data_state = 'empty', updated_at = ?2 WHERE user_id = ?1", params![user_id, now_millis()], diff --git a/iota-storage/src/util/blocked_users.rs b/iota-storage/src/util/blocked_users.rs new file mode 100644 index 0000000..7e22634 --- /dev/null +++ b/iota-storage/src/util/blocked_users.rs @@ -0,0 +1,104 @@ +/* Blocks are user policy independent of contacts or conversations. */ +use crate::storage_error::StorageError; +use crate::util::{db, sync}; +use rusqlite::{OptionalExtension, params}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BlockMutation { + pub blocked_user_id: i64, + pub revision: i64, + pub deleted: bool, +} +fn validate(user_id: i64, blocked_user_id: i64) -> Result<(), StorageError> { + if user_id <= 0 || blocked_user_id <= 0 || user_id == blocked_user_id { + Err(StorageError::Other("invalid block relationship".into())) + } else { + Ok(()) + } +} +pub fn block(user_id: i64, blocked_user_id: i64) -> Result { + validate(user_id, blocked_user_id)?; + db::with_immediate_transaction(|tx| { + let prior = tx.query_row("SELECT id, revision FROM blocked_users WHERE user_id = ?1 AND blocked_user_id = ?2", params![user_id, blocked_user_id], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?))).optional()?; + if let Some((_, revision)) = prior { + return Ok(BlockMutation { + blocked_user_id, + revision, + deleted: false, + }); + } + let created_at = sync::now_millis(); + tx.execute("INSERT INTO blocked_users (user_id, blocked_user_id, revision, created_at) VALUES (?1, ?2, 0, ?3)", params![user_id, blocked_user_id, created_at])?; + let id = tx.last_insert_rowid(); + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::BlockedUser, + id, + sync::Operation::Upsert, + )?; + tx.execute( + "UPDATE blocked_users SET revision = ?2 WHERE id = ?1", + params![id, revision], + )?; + Ok(BlockMutation { + blocked_user_id, + revision, + deleted: false, + }) + }) +} +pub fn unblock(user_id: i64, blocked_user_id: i64) -> Result, StorageError> { + validate(user_id, blocked_user_id)?; + db::with_immediate_transaction(|tx| { + let id = tx + .query_row( + "SELECT id FROM blocked_users WHERE user_id = ?1 AND blocked_user_id = ?2", + params![user_id, blocked_user_id], + |r| r.get(0), + ) + .optional()?; + let Some(id) = id else { + return Ok(None); + }; + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::BlockedUser, + id, + sync::Operation::Delete, + )?; + tx.execute("DELETE FROM blocked_users WHERE id = ?1", [id])?; + Ok(Some(BlockMutation { + blocked_user_id, + revision, + deleted: true, + })) + }) +} +pub fn is_blocked(user_id: i64, candidate_sender_id: i64) -> Result { + validate(user_id, candidate_sender_id)?; + db::with_db(|conn| { + Ok(conn + .query_row( + "SELECT 1 FROM blocked_users WHERE user_id = ?1 AND blocked_user_id = ?2", + params![user_id, candidate_sender_id], + |_| Ok(()), + ) + .optional()? + .is_some()) + }) +} +pub fn list(user_id: i64) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid block owner".into())); + } + db::with_db(|conn| { + let mut stmt = conn.prepare( + "SELECT blocked_user_id FROM blocked_users WHERE user_id = ?1 ORDER BY blocked_user_id", + )?; + stmt.query_map([user_id], |r| r.get(0))? + .collect::, _>>() + .map_err(StorageError::from) + }) +} diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 051557b..e04406a 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/iota-storage/src/util/chat_files.rs @@ -1,8 +1,9 @@ use crate::storage_error::StorageError; use crate::util::db; +use crate::util::message_storage_policy::{self, MessageRetention}; use crate::util::sync::{self, EntityType, Operation}; use iota_logger::log; -use rusqlite::{OptionalExtension, params}; +use rusqlite::{OptionalExtension, Transaction, params}; pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10; @@ -149,7 +150,7 @@ fn update_message_content( db::with_db(|conn| { let msg = conn.query_row( r#" - SELECT id, content, sent_by_self + SELECT id, content, sent_by_self, history_deleted FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3 ORDER BY id DESC LIMIT 1 @@ -160,11 +161,12 @@ fn update_message_content( row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, i64>(2)?, + row.get::<_, i64>(3)?, )) }, )?; - let (msg_id, old_content, sent_by_self) = msg; + let (msg_id, old_content, sent_by_self, history_deleted) = msg; if require_sent_by_self && sent_by_self != 1 { return Err(StorageError::Other( "Only the original sender can edit this message".into(), @@ -176,6 +178,9 @@ fn update_message_content( )); } + if history_deleted != 0 { + return Ok(()); + } let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -217,46 +222,144 @@ pub fn hard_delete_message( message_time: i64, ) -> Result<(), StorageError> { db::with_db(|conn| { - let msg_id: i64 = conn.query_row( - r#" - SELECT id FROM messages + let msg_id = conn + .query_row( + r#" + SELECT id, history_deleted FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3 ORDER BY id DESC LIMIT 1 "#, - params![storage_owner, external_user, message_time], - |row| row.get(0), - )?; + params![storage_owner, external_user, message_time], + |row| row.get::<_, i64>(0), + ) + .optional()?; + let Some(msg_id) = msg_id else { + return Ok(()); + }; let tx = conn.unchecked_transaction()?; - tx.execute( - "DELETE FROM message_edits WHERE message_id = ?1", - params![msg_id], - )?; - tx.execute( - "DELETE FROM reactions WHERE message_id = ?1", - params![msg_id], - )?; - tx.execute("DELETE FROM messages WHERE id = ?1", params![msg_id])?; - sync::record_event( - &tx, - storage_owner, - EntityType::Message, - msg_id, - Operation::Delete, - )?; + purge_message_in_tx(&tx, storage_owner, msg_id)?; tx.commit()?; Ok(()) }) } +pub fn purge_message(storage_owner: i64, message_id: i64) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| purge_message_in_tx(tx, storage_owner, message_id)) +} + +pub fn remove_message_history(storage_owner: i64, message_id: i64) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| remove_message_history_in_tx(tx, storage_owner, message_id)) +} + +/* Resolves the protocol identity retained in a tombstone before removing visible history. */ +pub fn remove_message_history_by_relay_identity( + storage_owner: i64, + relay_signer_id: i64, + relay_message_id: &str, +) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + let 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()?; + if let Some(message_id) = message_id { + remove_message_history_in_tx(tx, storage_owner, message_id)?; + } + Ok(()) + }) +} + +pub fn remove_message_history_in_tx( + tx: &Transaction<'_>, + storage_owner: i64, + message_id: i64, +) -> Result<(), StorageError> { + let message: Option<(i64, i64)> = tx + .query_row( + "SELECT external_user, history_deleted FROM messages WHERE id = ?1 AND storage_owner = ?2", + params![message_id, storage_owner], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + let Some((external_user, history_deleted)) = message else { + return Ok(()); + }; + if history_deleted != 0 { + return Ok(()); + } + tx.execute( + "DELETE FROM message_edits WHERE message_id = ?1", + [message_id], + )?; + tx.execute("DELETE FROM reactions WHERE message_id = ?1", [message_id])?; + tx.execute("DELETE FROM message_receipts WHERE storage_owner = ?1 AND EXISTS (SELECT 1 FROM messages WHERE id = ?2 AND relay_signer_id = message_receipts.target_signer_id AND relay_message_id = message_receipts.target_message_id)", params![storage_owner, message_id])?; + tx.execute("UPDATE messages SET content = '', history_deleted = 1, history_deleted_at = ?2, expires_at = NULL, client_received_at = NULL, client_received_recorded_at = NULL, read_at = NULL, read_recorded_at = NULL WHERE id = ?1", params![message_id, sync::now_millis()])?; + sync::record_event( + tx, + storage_owner, + EntityType::Message, + message_id, + Operation::Delete, + )?; + update_contact_last_message_in_tx(tx, storage_owner, external_user)?; + Ok(()) +} + +fn update_contact_last_message_in_tx( + tx: &Transaction<'_>, + storage_owner: i64, + external_user: i64, +) -> Result<(), StorageError> { + tx.execute("UPDATE contacts SET last_message_at = (SELECT MAX(COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, message_time)) FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0 AND history_deleted = 0) WHERE storage_owner = ?1 AND user_id = ?2", params![storage_owner, external_user])?; + Ok(()) +} + +pub fn purge_message_in_tx( + tx: &Transaction<'_>, + storage_owner: i64, + message_id: i64, +) -> Result<(), StorageError> { + let (external_user, relay_signer_id, relay_message_id) = tx.query_row( + "SELECT external_user, relay_signer_id, relay_message_id FROM messages WHERE id = ?1 AND storage_owner = ?2", + params![message_id, storage_owner], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option>(1)?, row.get::<_, Option>(2)?)), + )?; + if let (Some(relay_signer_id), Some(relay_message_id)) = (relay_signer_id, relay_message_id) { + tx.execute( + "DELETE FROM message_receipts WHERE storage_owner = ?1 AND target_signer_id = ?2 AND target_message_id = ?3", + params![storage_owner, relay_signer_id, relay_message_id], + )?; + } + tx.execute( + "DELETE FROM message_edits WHERE message_id = ?1", + [message_id], + )?; + tx.execute("DELETE FROM reactions WHERE message_id = ?1", [message_id])?; + tx.execute("DELETE FROM messages WHERE id = ?1", [message_id])?; + sync::record_event( + tx, + storage_owner, + EntityType::Message, + message_id, + Operation::Delete, + )?; + update_contact_last_message_in_tx(tx, storage_owner, external_user)?; + Ok(()) +} + /* Deletes a message from the sender's local copy after checking ownership. */ pub fn delete_message( storage_owner: i64, external_user: i64, message_time: i64, ) -> Result<(), StorageError> { - ensure_message_direction(storage_owner, external_user, message_time, true)?; - hard_delete_message(storage_owner, external_user, message_time) + match ensure_message_direction(storage_owner, external_user, message_time, true) { + Ok(()) => hard_delete_message(storage_owner, external_user, message_time), + Err(StorageError::Db(rusqlite::Error::QueryReturnedNoRows)) => Ok(()), + Err(error) => Err(error), + } } /* Flags the recipient's local copy after validating its sender, preserving its history. */ @@ -271,8 +374,11 @@ pub fn apply_remote_delete( "Remote sender does not match chat partner".into(), )); } - ensure_message_direction(storage_owner, external_user, message_time, false)?; - flag_deleted_by_external(storage_owner, external_user, message_time) + match ensure_message_direction(storage_owner, external_user, message_time, false) { + Ok(()) => flag_deleted_by_external(storage_owner, external_user, message_time), + Err(StorageError::Db(rusqlite::Error::QueryReturnedNoRows)) => Ok(()), + Err(error) => Err(error), + } } fn ensure_message_direction( @@ -330,6 +436,7 @@ pub fn flag_deleted_by_external( msg_id, Operation::Delete, )?; + update_contact_last_message_in_tx(&tx, storage_owner, external_user)?; tx.commit()?; Ok(()) }) @@ -346,15 +453,18 @@ pub fn delete_edit_history( message_time: i64, ) -> Result<(), StorageError> { db::with_db(|conn| { - let msg_id: i64 = conn.query_row( + let (msg_id, history_deleted): (i64, i64) = conn.query_row( r#" - SELECT id FROM messages + SELECT id, history_deleted FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3 ORDER BY id DESC LIMIT 1 "#, params![storage_owner, external_user, message_time], - |row| row.get(0), + |row| Ok((row.get(0)?, row.get(1)?)), )?; + if history_deleted != 0 { + return Ok(()); + } let tx = conn.unchecked_transaction()?; tx.execute( @@ -381,15 +491,18 @@ pub fn add_reaction( reaction: &str, ) -> Result<(), StorageError> { db::with_immediate_transaction(|tx| { - let msg_id: i64 = tx.query_row( + let (msg_id, history_deleted): (i64, i64) = tx.query_row( r#" - SELECT id FROM messages + SELECT id, history_deleted FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3 ORDER BY id DESC LIMIT 1 "#, params![storage_owner, external_user, message_time], - |row| row.get(0), + |row| Ok((row.get(0)?, row.get(1)?)), )?; + if history_deleted != 0 { + return Ok(()); + } let reaction_exists: bool = tx.query_row( "SELECT EXISTS(SELECT 1 FROM reactions WHERE message_id = ?1 AND reaction = ?2)", @@ -442,15 +555,18 @@ pub fn remove_reaction( reaction: &str, ) -> Result<(), StorageError> { db::with_db(|conn| { - let msg_id: i64 = conn.query_row( + let (msg_id, history_deleted): (i64, i64) = conn.query_row( r#" - SELECT id FROM messages + SELECT id, history_deleted FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3 ORDER BY id DESC LIMIT 1 "#, params![storage_owner, external_user, message_time], - |row| row.get(0), + |row| Ok((row.get(0)?, row.get(1)?)), )?; + if history_deleted != 0 { + return Ok(()); + } let tx = conn.unchecked_transaction()?; tx.execute( @@ -485,6 +601,16 @@ pub fn add_message(message: NewMessage<'_>) -> Result { destination_iota_received_at, initial_state, } = message; + let stored_at = sync::now_millis(); + let policy = message_storage_policy::get(storage_owner)?; + let expires_at = match policy.retention { + MessageRetention::Forever => None, + MessageRetention::Duration { duration_ms } => Some( + stored_at + .checked_add(duration_ms) + .ok_or_else(|| StorageError::Other("message expiry overflow".into()))?, + ), + }; db::with_db(|conn| { let tx = conn.unchecked_transaction()?; tx.execute( @@ -492,8 +618,8 @@ pub fn add_message(message: NewMessage<'_>) -> Result { 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) + authored_at, origin_iota_received_at, destination_iota_received_at, stored_at, expires_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15) "#, params![ storage_owner, @@ -509,6 +635,8 @@ pub fn add_message(message: NewMessage<'_>) -> Result { authored_at, origin_iota_received_at, destination_iota_received_at, + stored_at, + expires_at, ], )?; let msg_id = tx.last_insert_rowid(); @@ -583,11 +711,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, authored_at, history_deleted)) = 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, authored_at, history_deleted 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)?, row.get::<_, Option>(2)?, row.get::<_, i64>(3)?)), ) .optional()? else { @@ -605,6 +733,9 @@ pub fn record_message_receipt( "message receipt event time is outside the accepted clock range".into(), )); } + if history_deleted != 0 { + return Ok(()); + } 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], @@ -832,7 +963,7 @@ pub fn get_messages( read_recorded_at, delivery_failed_at, delivery_failure, 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 + WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0 AND history_deleted = 0 ORDER BY COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) DESC, id DESC LIMIT ?3 OFFSET ?4 "#, @@ -908,6 +1039,7 @@ pub fn get_message( WHERE storage_owner = ?1 AND message_time = ?2 AND deleted_by_external = 0 + AND history_deleted = 0 AND (?3 IS NULL OR external_user = ?3) ORDER BY id DESC "#, @@ -977,6 +1109,7 @@ pub fn get_message_with_offset( WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0 + AND history_deleted = 0 AND ( COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) > COALESCE(?3, ?4) @@ -1005,7 +1138,7 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec Vec Vec { let ids = match db::with_db(|conn| { let mut stmt = conn.prepare( - "SELECT id FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0", + "SELECT id FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0 AND history_deleted = 0", )?; Ok(stmt .query_map([storage_owner], |row| row.get::<_, i64>(0))? diff --git a/iota-storage/src/util/client_message_delivery.rs b/iota-storage/src/util/client_message_delivery.rs new file mode 100644 index 0000000..baa4f5f --- /dev/null +++ b/iota-storage/src/util/client_message_delivery.rs @@ -0,0 +1,93 @@ +/* Delivery records bind a specific state-sync response to the messages it contained. */ +use crate::storage_error::StorageError; +use crate::util::{chat_files, db, message_storage_policy, sync}; +use rusqlite::{OptionalExtension, params}; + +pub fn record_sync_delivery( + user_id: i64, + session_id: i64, + version: i64, + message_ids: impl IntoIterator, +) -> Result<(), StorageError> { + if user_id <= 0 || session_id <= 0 || version < 0 { + return Err(StorageError::Other( + "invalid message delivery record".into(), + )); + } + db::with_immediate_transaction(|tx| { + for message_id in message_ids { + tx.execute("INSERT OR IGNORE INTO client_message_deliveries (user_id, session_id, sync_version, message_id, created_at) VALUES (?1, ?2, ?3, ?4, ?5)", params![user_id, session_id, version, message_id, sync::now_millis()])?; + } + Ok(()) + }) +} + +pub fn acknowledge_sync_delivery( + user_id: i64, + session_id: i64, + version: i64, +) -> Result, StorageError> { + db::with_immediate_transaction(|tx| { + let mut statement = tx.prepare("SELECT DISTINCT message_id FROM client_message_deliveries WHERE user_id = ?1 AND session_id = ?2 AND sync_version <= ?3")?; + let ids = statement + .query_map(params![user_id, session_id, version], |row| row.get(0))? + .collect::, _>>() + .map_err(StorageError::from)?; + tx.execute("DELETE FROM client_message_deliveries WHERE user_id = ?1 AND session_id = ?2 AND sync_version <= ?3", params![user_id, session_id, version])?; + Ok(ids) + }) +} + +/* Acknowledge a state-sync response and delete only its recorded visible messages atomically. */ +pub fn acknowledge_client_state( + user_id: i64, + session_id: i64, + version: i64, + cache_schema_version: i64, +) -> Result<(), StorageError> { + if user_id <= 0 || session_id <= 0 || version < 0 { + return Err(StorageError::Other("invalid sync acknowledgement".into())); + } + db::with_immediate_transaction(|tx| { + let head = tx + .query_row( + "SELECT version FROM sync_heads WHERE user_id = ?1", + [user_id], + |row| row.get::<_, i64>(0), + ) + .optional()? + .unwrap_or_default(); + if version > head { + return Err(StorageError::Other( + "acknowledgement is ahead of head".into(), + )); + } + let ids = { + let mut statement = tx.prepare("SELECT DISTINCT message_id FROM client_message_deliveries WHERE user_id = ?1 AND session_id = ?2 AND sync_version <= ?3")?; + statement + .query_map(params![user_id, session_id, version], |row| { + row.get::<_, i64>(0) + })? + .collect::, _>>()? + }; + tx.execute("INSERT INTO client_sync_state (user_id, session_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, session_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, session_id, version, cache_schema_version, sync::now_millis()])?; + if message_storage_policy::get_in_tx(tx, user_id)?.history_mode + == message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery + { + for message_id in ids { + chat_files::remove_message_history_in_tx(tx, user_id, message_id)?; + } + } + tx.execute("DELETE FROM client_message_deliveries WHERE user_id = ?1 AND session_id = ?2 AND sync_version <= ?3", params![user_id, session_id, version])?; + Ok(()) + }) +} + +pub fn purge_session_deliveries(user_id: i64, session_id: i64) -> Result { + db::with_db(|conn| { + Ok(conn.execute( + "DELETE FROM client_message_deliveries WHERE user_id = ?1 AND session_id = ?2", + params![user_id, session_id], + )?) + }) +} diff --git a/iota-storage/src/util/client_relay_delivery.rs b/iota-storage/src/util/client_relay_delivery.rs new file mode 100644 index 0000000..bd72208 --- /dev/null +++ b/iota-storage/src/util/client_relay_delivery.rs @@ -0,0 +1,39 @@ +/* Commits client delivery, relay state, and retention cleanup as one operation. */ +use crate::storage_error::StorageError; +use crate::util::{chat_files, db, message_storage_policy, relay_queue, relay_replay}; +use rusqlite::OptionalExtension; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClientRelayDeliveryResult { + NotFound, + Acknowledged, +} + +pub fn acknowledge_client_delivery( + destination_id: i64, + frame_id: u32, +) -> Result { + db::with_immediate_transaction(|tx| { + let Some((pending_id, relay)) = + relay_queue::find_user_delivery_in_tx(tx, destination_id, frame_id)? + else { + return Ok(ClientRelayDeliveryResult::NotFound); + }; + let policy = message_storage_policy::get_in_tx(tx, destination_id)?; + relay_replay::mark_delivered_for_frame_in_tx(tx, destination_id, frame_id)?; + if policy.history_mode + == message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery + { + let message_id = tx.query_row( + "SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", + rusqlite::params![destination_id, relay.signer_id, relay.message_id], + |row| row.get::<_, i64>(0), + ).optional()?; + if let Some(message_id) = message_id { + chat_files::remove_message_history_in_tx(tx, destination_id, message_id)?; + } + } + relay_queue::acknowledge_in_tx(tx, pending_id)?; + Ok(ClientRelayDeliveryResult::Acknowledged) + }) +} diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index e8cbf6e..c983324 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -579,6 +579,137 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { conn.pragma_update(None, "user_version", 16)?; } + if current_version < 17 { + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS user_blobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + blob_id TEXT NOT NULL, + blob BLOB NOT NULL, + revision INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0 CHECK (deleted IN (0, 1)), + updated_at INTEGER NOT NULL, + UNIQUE(user_id, blob_id) + ); + CREATE INDEX IF NOT EXISTS idx_user_blobs_owner + ON user_blobs (user_id, deleted); + CREATE TABLE IF NOT EXISTS blocked_users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL, + blocked_user_id INTEGER NOT NULL, + revision INTEGER NOT NULL, + created_at INTEGER NOT NULL, + UNIQUE(user_id, blocked_user_id) + ); + CREATE INDEX IF NOT EXISTS idx_blocked_users_owner ON blocked_users (user_id); + CREATE TABLE IF NOT EXISTS user_receipt_policy ( + user_id INTEGER PRIMARY KEY, + send_read_receipts INTEGER NOT NULL CHECK (send_read_receipts IN (0, 1)), + send_received_receipts INTEGER NOT NULL CHECK (send_received_receipts IN (0, 1)), + revision INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS user_message_storage_policy ( + user_id INTEGER PRIMARY KEY, + history_mode TEXT NOT NULL CHECK (history_mode IN ('retain', 'delete_after_client_delivery')), + retention_mode TEXT NOT NULL CHECK (retention_mode IN ('forever', 'duration')), + retention_duration_ms INTEGER, + revision INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + PRAGMA user_version = 17; + "#, + )?; + } + + if current_version < 18 { + 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, "stored_at", "stored_at INTEGER")?; + add_column_if_missing(conn, "expires_at", "expires_at INTEGER")?; + conn.execute_batch( + "CREATE INDEX IF NOT EXISTS idx_messages_expiry ON messages (storage_owner, expires_at) WHERE expires_at IS NOT NULL;", + )?; + } + conn.pragma_update(None, "user_version", 18)?; + } + + if current_version < 19 { + 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( + "UPDATE messages SET stored_at = COALESCE(destination_iota_received_at, origin_iota_received_at) WHERE stored_at IS NULL", + [], + )?; + } + conn.pragma_update(None, "user_version", 19)?; + } + + if current_version < 20 { + 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, + "history_deleted", + "history_deleted INTEGER NOT NULL DEFAULT 0 CHECK (history_deleted IN (0, 1))", + )?; + add_column_if_missing(conn, "history_deleted_at", "history_deleted_at INTEGER")?; + } + let pending_relays_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'pending_relays')", + [], + |row| row.get(0), + )?; + if pending_relays_exist { + add_table_column_if_missing( + conn, + "pending_relays", + "relay_signer_id", + "relay_signer_id INTEGER", + )?; + add_table_column_if_missing( + conn, + "pending_relays", + "relay_destination_user_id", + "relay_destination_user_id INTEGER", + )?; + add_table_column_if_missing( + conn, + "pending_relays", + "relay_message_id", + "relay_message_id TEXT", + )?; + } + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS client_message_deliveries ( + user_id INTEGER NOT NULL, + session_id INTEGER NOT NULL, + sync_version INTEGER NOT NULL, + message_id INTEGER NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (user_id, session_id, sync_version, message_id) + ); + CREATE INDEX IF NOT EXISTS idx_client_message_deliveries_session + ON client_message_deliveries (user_id, session_id, sync_version); + PRAGMA user_version = 20; + "#, + )?; + } + Ok(()) } @@ -648,7 +779,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, 20); 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 +798,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, 20); for table in [ "sync_heads", "sync_events", @@ -678,6 +809,10 @@ mod tests { "relay_inbox", "synced_settings", "pending_user_operations", + "user_blobs", + "blocked_users", + "user_receipt_policy", + "user_message_storage_policy", ] { let exists: i64 = conn.query_row( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1", @@ -702,7 +837,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, 20); for column in [ "id", "user_id", diff --git a/iota-storage/src/util/downstream_relay.rs b/iota-storage/src/util/downstream_relay.rs new file mode 100644 index 0000000..b1eddf2 --- /dev/null +++ b/iota-storage/src/util/downstream_relay.rs @@ -0,0 +1,69 @@ +/* Completes or rejects an Iota-to-Iota relay without leaving retry state behind. */ +use crate::storage_error::StorageError; +use crate::util::{db, sync}; +use rusqlite::{OptionalExtension, params}; + +pub fn acknowledge_iota_delivery( + destination_iota: u64, + frame_id: u32, + signer_id: i64, + relay_message_id: &str, + destination_accepted_at: i64, +) -> Result<(), StorageError> { + let destination_iota = i64::try_from(destination_iota) + .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?; + db::with_immediate_transaction(|tx| { + let pending_id = tx.query_row("SELECT id FROM pending_relays WHERE destination_id = ?1 AND target_kind = 1 AND frame_id = ?2", params![destination_iota, i64::from(frame_id)], |row| row.get::<_, i64>(0)).optional()?; + let Some(pending_id) = pending_id else { + return Ok(()); + }; + let message_id = tx.query_row("SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?1 AND relay_message_id = ?2", params![signer_id, relay_message_id], |row| row.get::<_, i64>(0)).optional()?; + if let Some(message_id) = message_id { + 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", params![destination_accepted_at, message_id])?; + sync::record_event( + tx, + signer_id, + sync::EntityType::Message, + message_id, + sync::Operation::Upsert, + )?; + } + tx.execute("UPDATE relay_inbox SET state = 'delivered', downstream_acked_at = COALESCE(downstream_acked_at, ?3) WHERE signer_id = ?1 AND message_id = ?2", params![signer_id, relay_message_id, sync::now_millis()])?; + tx.execute("DELETE FROM pending_relays WHERE id = ?1", [pending_id])?; + Ok(()) + }) +} + +pub fn reject_iota_delivery( + destination_iota: u64, + frame_id: u32, + signer_id: i64, + relay_message_id: &str, + failure: &str, +) -> Result<(), StorageError> { + let destination_iota = i64::try_from(destination_iota) + .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?; + db::with_immediate_transaction(|tx| { + let pending_id = tx.query_row("SELECT id FROM pending_relays WHERE destination_id = ?1 AND target_kind = 1 AND frame_id = ?2", params![destination_iota, i64::from(frame_id)], |row| row.get::<_, i64>(0)).optional()?; + let Some(pending_id) = pending_id else { + return Ok(()); + }; + let message_id = tx.query_row("SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?1 AND relay_message_id = ?2", params![signer_id, relay_message_id], |row| row.get::<_, i64>(0)).optional()?; + if let Some(message_id) = message_id { + tx.execute( + "UPDATE messages SET delivery_failed_at = ?1, delivery_failure = ?2 WHERE id = ?3", + params![sync::now_millis(), failure, message_id], + )?; + sync::record_event( + tx, + signer_id, + sync::EntityType::Message, + message_id, + sync::Operation::Upsert, + )?; + } + tx.execute("UPDATE relay_inbox SET state = 'rejected', rejected_at = COALESCE(rejected_at, ?3) WHERE signer_id = ?1 AND message_id = ?2", params![signer_id, relay_message_id, sync::now_millis()])?; + tx.execute("DELETE FROM pending_relays WHERE id = ?1", [pending_id])?; + Ok(()) + }) +} diff --git a/iota-storage/src/util/message_retention.rs b/iota-storage/src/util/message_retention.rs new file mode 100644 index 0000000..032e54b --- /dev/null +++ b/iota-storage/src/util/message_retention.rs @@ -0,0 +1,28 @@ +/* Retention cleanup removes only stored history; pending transport is retained. */ +use crate::storage_error::StorageError; +use crate::util::{chat_files, db}; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct RetentionResult { + pub deleted_messages: usize, +} + +pub fn purge_expired_messages(now: i64) -> Result { + let expired = db::with_db(|conn| { + let mut statement = conn.prepare( + "SELECT id, storage_owner FROM messages WHERE expires_at IS NOT NULL AND expires_at <= ?1 ORDER BY expires_at ASC", + )?; + statement + .query_map([now], |row| { + Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)) + })? + .collect::, _>>() + .map_err(StorageError::from) + })?; + let mut result = RetentionResult::default(); + for (message_id, storage_owner) in expired { + chat_files::purge_message(storage_owner, message_id)?; + result.deleted_messages += 1; + } + Ok(result) +} diff --git a/iota-storage/src/util/message_storage_policy.rs b/iota-storage/src/util/message_storage_policy.rs new file mode 100644 index 0000000..0d9568f --- /dev/null +++ b/iota-storage/src/util/message_storage_policy.rs @@ -0,0 +1,149 @@ +/* Retention policy is normalized so invalid combinations cannot reach cleanup logic. */ +use crate::storage_error::StorageError; +use crate::util::{db, sync}; +use rusqlite::{OptionalExtension, Transaction}; + +pub const MIN_RETENTION_DURATION_MS: i64 = 60_000; +pub const MAX_RETENTION_DURATION_MS: i64 = 31_536_000_000; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MessageHistoryMode { + Retain, + DeleteAfterClientDelivery, +} +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MessageRetention { + Forever, + Duration { duration_ms: i64 }, +} +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MessageStoragePolicy { + pub user_id: i64, + pub history_mode: MessageHistoryMode, + pub retention: MessageRetention, + pub revision: i64, + pub updated_at: i64, +} +pub fn default_policy() -> MessageStoragePolicy { + MessageStoragePolicy { + user_id: 0, + history_mode: MessageHistoryMode::Retain, + retention: MessageRetention::Forever, + revision: 0, + updated_at: 0, + } +} +fn validate(user_id: i64, retention: MessageRetention) -> Result<(), StorageError> { + if user_id <= 0 { + return Err(StorageError::Other( + "invalid message storage policy owner".into(), + )); + } + if let MessageRetention::Duration { duration_ms } = retention { + if !(MIN_RETENTION_DURATION_MS..=MAX_RETENTION_DURATION_MS).contains(&duration_ms) { + return Err(StorageError::Other("invalid retention duration".into())); + } + } + Ok(()) +} +fn history_name(value: MessageHistoryMode) -> &'static str { + match value { + MessageHistoryMode::Retain => "retain", + MessageHistoryMode::DeleteAfterClientDelivery => "delete_after_client_delivery", + } +} +fn parse_history(value: &str) -> Result { + match value { + "retain" => Ok(MessageHistoryMode::Retain), + "delete_after_client_delivery" => Ok(MessageHistoryMode::DeleteAfterClientDelivery), + _ => Err(StorageError::Other( + "invalid stored message history mode".into(), + )), + } +} +pub fn get(user_id: i64) -> Result { + validate(user_id, MessageRetention::Forever)?; + db::with_db(|conn| { + let found = conn.query_row("SELECT history_mode, retention_mode, retention_duration_ms, revision, updated_at FROM user_message_storage_policy WHERE user_id = ?1", [user_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, Option>(2)?, r.get::<_, i64>(3)?, r.get::<_, i64>(4)?))).optional()?; + policy_from_found(user_id, found) + }) +} + +pub fn get_in_tx(tx: &Transaction<'_>, user_id: i64) -> Result { + validate(user_id, MessageRetention::Forever)?; + let found = tx.query_row("SELECT history_mode, retention_mode, retention_duration_ms, revision, updated_at FROM user_message_storage_policy WHERE user_id = ?1", [user_id], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, Option>(2)?, r.get::<_, i64>(3)?, r.get::<_, i64>(4)?))).optional()?; + policy_from_found(user_id, found) +} + +fn policy_from_found( + user_id: i64, + found: Option<(String, String, Option, i64, i64)>, +) -> Result { + match found { + None => Ok(MessageStoragePolicy { + user_id, + ..default_policy() + }), + Some((history, mode, duration, revision, updated_at)) => { + let retention = match (mode.as_str(), duration) { + ("forever", None) => MessageRetention::Forever, + ("duration", Some(duration_ms)) => MessageRetention::Duration { duration_ms }, + _ => { + return Err(StorageError::Other( + "invalid stored retention policy".into(), + )); + } + }; + validate(user_id, retention)?; + Ok(MessageStoragePolicy { + user_id, + history_mode: parse_history(&history)?, + retention, + revision, + updated_at, + }) + } + } +} +pub fn set( + user_id: i64, + history_mode: MessageHistoryMode, + retention: MessageRetention, +) -> Result { + validate(user_id, retention)?; + db::with_immediate_transaction(|tx| { + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::MessageStoragePolicy, + user_id, + sync::Operation::Upsert, + )?; + let updated_at = sync::now_millis(); + let (retention_mode, duration): (&str, Option) = match retention { + MessageRetention::Forever => ("forever", None), + MessageRetention::Duration { duration_ms } => ("duration", Some(duration_ms)), + }; + tx.execute("INSERT INTO user_message_storage_policy (user_id, history_mode, retention_mode, retention_duration_ms, revision, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(user_id) DO UPDATE SET history_mode = excluded.history_mode, retention_mode = excluded.retention_mode, retention_duration_ms = excluded.retention_duration_ms, revision = excluded.revision, updated_at = excluded.updated_at", rusqlite::params![user_id, history_name(history_mode), retention_mode, duration, revision, updated_at])?; + match retention { + MessageRetention::Forever => { + tx.execute( + "UPDATE messages SET expires_at = NULL WHERE storage_owner = ?1", + [user_id], + )?; + } + MessageRetention::Duration { duration_ms } => { + tx.execute( + "UPDATE messages SET expires_at = stored_at + ?2 WHERE storage_owner = ?1 AND stored_at IS NOT NULL", + rusqlite::params![user_id, duration_ms], + )?; + } + } + Ok(MessageStoragePolicy { + user_id, + history_mode, + retention, + revision, + updated_at, + }) + }) +} diff --git a/iota-storage/src/util/mod.rs b/iota-storage/src/util/mod.rs index 0731077..bcf195e 100644 --- a/iota-storage/src/util/mod.rs +++ b/iota-storage/src/util/mod.rs @@ -1,11 +1,20 @@ +pub mod blocked_users; pub mod chat_files; pub mod chats_util; +pub mod client_message_delivery; +pub mod client_relay_delivery; pub mod communities_util; pub mod config_util; pub mod db; +pub mod downstream_relay; pub mod e2ee_storage; +pub mod message_retention; +pub mod message_storage_policy; +pub mod outgoing_relay; +pub mod receipt_policy; pub mod relay_queue; pub mod relay_replay; pub mod settings; pub mod sync; pub mod synced_settings; +pub mod user_blobs; diff --git a/iota-storage/src/util/outgoing_relay.rs b/iota-storage/src/util/outgoing_relay.rs new file mode 100644 index 0000000..0ea7e17 --- /dev/null +++ b/iota-storage/src/util/outgoing_relay.rs @@ -0,0 +1,61 @@ +/* Accepts an outgoing relay only when its retry record and retention state commit together. */ +use crate::storage_error::StorageError; +use crate::util::{chat_files, db, message_storage_policy, relay_queue, relay_replay}; +use iota_util::route_target::RouteTarget; +use rusqlite::OptionalExtension; + +pub struct OutgoingRelay<'a> { + pub target: RouteTarget, + pub identity: &'a relay_queue::RelayIdentity, + pub frame: &'a [u8], + pub created_at: i64, + pub frame_id: u32, + pub type_map_version: &'a str, +} + +pub fn commit_outgoing_relay(relay: OutgoingRelay<'_>) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + relay_queue::enqueue_in_tx( + tx, + relay.target, + relay.identity, + relay.frame, + relay.created_at, + relay.frame_id, + relay.type_map_version, + )?; + if message_storage_policy::get_in_tx(tx, relay.identity.signer_id)?.history_mode + == message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery + { + let message_id = tx.query_row("SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?1 AND relay_message_id = ?2", rusqlite::params![relay.identity.signer_id, relay.identity.message_id], |row| row.get::<_, i64>(0)).optional()?; + if let Some(message_id) = message_id { + chat_files::remove_message_history_in_tx(tx, relay.identity.signer_id, message_id)?; + } + } + relay_replay::mark_queued_in_tx(tx, relay.identity.signer_id, &relay.identity.message_id)?; + Ok(()) + }) +} + +pub fn apply_outgoing_history_policy( + storage_owner: i64, + relay_signer_id: i64, + relay_message_id: &str, +) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + if message_storage_policy::get_in_tx(tx, storage_owner)?.history_mode + != message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery + { + return Ok(()); + } + let message_id = tx.query_row( + "SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", + rusqlite::params![storage_owner, relay_signer_id, relay_message_id], + |row| row.get::<_, i64>(0), + ).optional()?; + if let Some(message_id) = message_id { + chat_files::remove_message_history_in_tx(tx, storage_owner, message_id)?; + } + Ok(()) + }) +} diff --git a/iota-storage/src/util/receipt_policy.rs b/iota-storage/src/util/receipt_policy.rs new file mode 100644 index 0000000..934037e --- /dev/null +++ b/iota-storage/src/util/receipt_policy.rs @@ -0,0 +1,64 @@ +/* Receipt disclosure is typed policy because Iota enforces it when relaying states. */ +use crate::storage_error::StorageError; +use crate::util::{db, sync}; +use rusqlite::OptionalExtension; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReceiptPolicy { + pub user_id: i64, + pub send_read_receipts: bool, + pub send_received_receipts: bool, + pub revision: i64, + pub updated_at: i64, +} +pub fn default_policy() -> ReceiptPolicy { + ReceiptPolicy { + user_id: 0, + send_read_receipts: true, + send_received_receipts: false, + revision: 0, + updated_at: 0, + } +} +fn validate(user_id: i64) -> Result<(), StorageError> { + if user_id > 0 { + Ok(()) + } else { + Err(StorageError::Other("invalid receipt policy owner".into())) + } +} +pub fn get(user_id: i64) -> Result { + validate(user_id)?; + db::with_db(|conn| { + let found = conn.query_row("SELECT send_read_receipts, send_received_receipts, revision, updated_at FROM user_receipt_policy WHERE user_id = ?1", [user_id], |r| Ok(ReceiptPolicy { user_id, send_read_receipts: r.get(0)?, send_received_receipts: r.get(1)?, revision: r.get(2)?, updated_at: r.get(3)? })).optional()?; + Ok(found.unwrap_or(ReceiptPolicy { + user_id, + ..default_policy() + })) + }) +} +pub fn set( + user_id: i64, + send_read_receipts: bool, + send_received_receipts: bool, +) -> Result { + validate(user_id)?; + db::with_immediate_transaction(|tx| { + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::ReceiptPolicy, + user_id, + sync::Operation::Upsert, + )?; + let updated_at = sync::now_millis(); + tx.execute("INSERT INTO user_receipt_policy (user_id, send_read_receipts, send_received_receipts, revision, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id) DO UPDATE SET send_read_receipts = excluded.send_read_receipts, send_received_receipts = excluded.send_received_receipts, revision = excluded.revision, updated_at = excluded.updated_at", rusqlite::params![user_id, send_read_receipts, send_received_receipts, revision, updated_at])?; + Ok(ReceiptPolicy { + user_id, + send_read_receipts, + send_received_receipts, + revision, + updated_at, + }) + }) +} diff --git a/iota-storage/src/util/relay_queue.rs b/iota-storage/src/util/relay_queue.rs index 835ebcd..75accbb 100644 --- a/iota-storage/src/util/relay_queue.rs +++ b/iota-storage/src/util/relay_queue.rs @@ -1,7 +1,7 @@ use crate::storage_error::StorageError; use crate::util::db; use iota_util::route_target::RouteTarget; -use rusqlite::params; +use rusqlite::{OptionalExtension, Transaction, params}; #[derive(Clone, Debug, PartialEq, Eq)] pub struct PendingRelay { @@ -11,10 +11,43 @@ pub struct PendingRelay { pub created_at: i64, pub frame_id: u32, pub type_map_version: String, + pub relay_signer_id: Option, + pub relay_destination_user_id: Option, + pub relay_message_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RelayIdentity { + pub signer_id: i64, + pub destination_user_id: i64, + pub message_id: String, } pub fn enqueue( target: RouteTarget, + relay: &RelayIdentity, + frame: &[u8], + created_at: i64, + frame_id: u32, + type_map_version: &str, +) -> Result<(), StorageError> { + db::with_immediate_transaction(|tx| { + enqueue_in_tx( + tx, + target, + relay, + frame, + created_at, + frame_id, + type_map_version, + ) + }) +} + +pub fn enqueue_in_tx( + tx: &Transaction<'_>, + target: RouteTarget, + relay: &RelayIdentity, frame: &[u8], created_at: i64, frame_id: u32, @@ -26,26 +59,27 @@ pub fn enqueue( RouteTarget::User(_) => 0_i64, RouteTarget::Iota(_) => 1_i64, }; - db::with_db(|connection| { - connection.execute( - "INSERT OR IGNORE INTO pending_relays (destination_id, target_kind, frame, created_at, frame_id, type_map_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + tx.execute( + "INSERT OR IGNORE INTO pending_relays (destination_id, target_kind, relay_signer_id, relay_destination_user_id, relay_message_id, frame, created_at, frame_id, type_map_version) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", params![ destination_id, target_kind, + relay.signer_id, + relay.destination_user_id, + relay.message_id, frame, created_at, i64::from(frame_id), type_map_version ], - )?; - Ok(()) - }) + )?; + Ok(()) } pub fn list(limit: i64) -> Result, StorageError> { db::with_db(|connection| { let mut statement = connection.prepare( - "SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version FROM pending_relays ORDER BY id LIMIT ?1", + "SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id FROM pending_relays ORDER BY id LIMIT ?1", )?; let rows = statement.query_map(params![limit.clamp(1, 500)], |row| { let destination_id = row.get::<_, i64>(1)?; @@ -81,12 +115,113 @@ pub fn list(limit: i64) -> Result, StorageError> { ) })?, type_map_version: row.get(6)?, + relay_signer_id: row.get(7)?, + relay_destination_user_id: row.get(8)?, + relay_message_id: row.get(9)?, }) })?; rows.collect::, _>>().map_err(Into::into) }) } +pub fn list_without_relay_identity() -> Result, StorageError> { + list_without_relay_identity_after(0, i64::MAX) +} + +pub fn list_without_relay_identity_after( + after_id: i64, + limit: i64, +) -> Result, StorageError> { + db::with_db(|connection| { + let mut statement = connection.prepare("SELECT id, destination_id, target_kind, frame, created_at, frame_id, type_map_version, relay_signer_id, relay_destination_user_id, relay_message_id FROM pending_relays WHERE id > ?1 AND (relay_signer_id IS NULL OR relay_destination_user_id IS NULL OR relay_message_id IS NULL) ORDER BY id LIMIT ?2")?; + let rows = statement.query_map( + params![after_id, limit.clamp(1, 500)], + pending_relay_from_row, + )?; + rows.collect::, _>>().map_err(Into::into) + }) +} + +fn pending_relay_from_row(row: &rusqlite::Row<'_>) -> Result { + let destination_id = u64::try_from(row.get::<_, i64>(1)?) + .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(1, 0))?; + let target = match row.get::<_, i64>(2)? { + 0 => RouteTarget::User(destination_id), + 1 => RouteTarget::Iota(destination_id), + _ => return Err(rusqlite::Error::IntegralValueOutOfRange(2, 0)), + }; + Ok(PendingRelay { + id: row.get(0)?, + target, + frame: row.get(3)?, + created_at: row.get(4)?, + frame_id: u32::try_from(row.get::<_, i64>(5)?) + .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(5, 0))?, + type_map_version: row.get(6)?, + relay_signer_id: row.get(7)?, + relay_destination_user_id: row.get(8)?, + relay_message_id: row.get(9)?, + }) +} + +pub fn set_relay_identity(id: i64, relay: &RelayIdentity) -> Result<(), StorageError> { + db::with_db(|connection| { + connection.execute("UPDATE pending_relays SET relay_signer_id = ?2, relay_destination_user_id = ?3, relay_message_id = ?4 WHERE id = ?1", params![id, relay.signer_id, relay.destination_user_id, relay.message_id])?; + Ok(()) + }) +} + +pub fn has_unclassified_relays() -> Result { + db::with_db(|connection| { + connection + .query_row("SELECT EXISTS(SELECT 1 FROM pending_relays WHERE relay_signer_id IS NULL OR relay_destination_user_id IS NULL OR relay_message_id IS NULL)", [], |row| row.get(0)) + .map_err(Into::into) + }) +} + +pub fn find_user_delivery_in_tx( + tx: &Transaction<'_>, + destination_id: i64, + frame_id: u32, +) -> Result, StorageError> { + tx.query_row( + "SELECT id, relay_signer_id, relay_destination_user_id, relay_message_id FROM pending_relays WHERE destination_id = ?1 AND target_kind = 0 AND frame_id = ?2", + params![destination_id, i64::from(frame_id)], + |row| { + let signer_id = row.get::<_, Option>(1)?; + let destination_user_id = row.get::<_, Option>(2)?; + let message_id = row.get::<_, Option>(3)?; + match (signer_id, destination_user_id, message_id) { + (Some(signer_id), Some(destination_user_id), Some(message_id)) => Ok((row.get(0)?, RelayIdentity { signer_id, destination_user_id, message_id })), + _ => Err(rusqlite::Error::InvalidQuery), + } + }, + ).optional().map_err(|error| match error { + rusqlite::Error::InvalidQuery => StorageError::PendingRelayOwnershipUnknown, + error => StorageError::Db(error), + }) +} + +pub fn acknowledge_in_tx(tx: &Transaction<'_>, id: i64) -> Result<(), StorageError> { + tx.execute("DELETE FROM pending_relays WHERE id = ?1", [id])?; + Ok(()) +} + +pub fn reject_outgoing_relay( + destination_iota: u64, + frame_id: u32, + signer_id: i64, + relay_message_id: &str, +) -> Result<(), StorageError> { + let destination_iota = i64::try_from(destination_iota) + .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?; + db::with_immediate_transaction(|tx| { + tx.execute("DELETE FROM pending_relays WHERE destination_id = ?1 AND target_kind = 1 AND frame_id = ?2", params![destination_iota, i64::from(frame_id)])?; + tx.execute("UPDATE relay_inbox SET state = 'rejected', rejected_at = COALESCE(rejected_at, ?3) WHERE signer_id = ?1 AND message_id = ?2", params![signer_id, relay_message_id, crate::util::sync::now_millis()])?; + Ok(()) + }) +} + pub fn acknowledge(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()))?; diff --git a/iota-storage/src/util/relay_replay.rs b/iota-storage/src/util/relay_replay.rs index 0c5bddc..0b7d03a 100644 --- a/iota-storage/src/util/relay_replay.rs +++ b/iota-storage/src/util/relay_replay.rs @@ -1,6 +1,13 @@ use crate::storage_error::StorageError; use crate::util::db; -use rusqlite::params; +use rusqlite::{OptionalExtension, params}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeliveredRelay { + pub signer_id: i64, + pub message_id: String, + pub destination_id: i64, +} #[derive(Debug, Clone, PartialEq, Eq)] pub enum RelayReservation { @@ -65,22 +72,52 @@ pub fn reserve( }) } -pub fn mark_delivered_for_frame(destination_id: u64, frame_id: u32) -> Result<(), StorageError> { +pub fn mark_delivered_for_frame( + destination_id: u64, + frame_id: u32, +) -> Result, StorageError> { let destination_id = i64::try_from(destination_id) .map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?; - db::with_db(|connection| { + db::with_immediate_transaction(|tx| { let delivered_at = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as i64; - connection.execute( + let relay = tx + .query_row( + "SELECT signer_id, message_id, destination_id FROM relay_inbox WHERE destination_id = ?1 AND frame_id = ?2", + params![destination_id, i64::from(frame_id)], + |row| Ok(DeliveredRelay { signer_id: row.get(0)?, message_id: row.get(1)?, destination_id: row.get(2)? }), + ) + .optional()?; + if relay.is_none() { + return Ok(None); + } + tx.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], )?; - Ok(()) + Ok(relay) }) } +pub fn mark_delivered_for_frame_in_tx( + tx: &rusqlite::Transaction<'_>, + destination_id: i64, + frame_id: u32, +) -> Result<(), StorageError> { + let changed = tx.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), crate::util::sync::now_millis()], + )?; + if changed == 0 { + return Err(StorageError::Other( + "pending relay has no relay inbox record".into(), + )); + } + Ok(()) +} + fn mark_transition( signer_id: u64, message_id: &str, @@ -110,6 +147,18 @@ pub fn mark_queued(signer_id: u64, message_id: &str) -> Result<(), StorageError> mark_transition(signer_id, message_id, "queued", "queued_at") } +pub fn mark_queued_in_tx( + tx: &rusqlite::Transaction<'_>, + signer_id: i64, + message_id: &str, +) -> Result<(), StorageError> { + tx.execute( + "UPDATE relay_inbox SET state = 'queued', queued_at = COALESCE(queued_at, ?3) WHERE signer_id = ?1 AND message_id = ?2", + params![signer_id, message_id, crate::util::sync::now_millis()], + )?; + Ok(()) +} + pub fn mark_downstream_acked(signer_id: u64, message_id: &str) -> Result<(), StorageError> { mark_transition(signer_id, message_id, "delivered", "downstream_acked_at") } diff --git a/iota-storage/src/util/sync.rs b/iota-storage/src/util/sync.rs index 8ef8eb2..78cb6d6 100644 --- a/iota-storage/src/util/sync.rs +++ b/iota-storage/src/util/sync.rs @@ -1,16 +1,21 @@ /* Durable per-user state journal used by device cache synchronization. */ use crate::storage_error::StorageError; use crate::util::db; -use rusqlite::{Transaction, params}; +use rusqlite::{OptionalExtension, Transaction, params}; use std::collections::BTreeMap; -pub const CACHE_SCHEMA_VERSION: i64 = 3; +pub const CACHE_SCHEMA_VERSION: i64 = 4; +pub const STALE_CLIENT_SYNC_STATE_MS: i64 = 90 * 24 * 60 * 60 * 1_000; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EntityType { Message, Contact, Setting, + UserBlob, + BlockedUser, + ReceiptPolicy, + MessageStoragePolicy, } impl EntityType { fn as_str(self) -> &'static str { @@ -18,6 +23,10 @@ impl EntityType { Self::Message => "message", Self::Contact => "contact", Self::Setting => "setting", + Self::UserBlob => "user_blob", + Self::BlockedUser => "blocked_user", + Self::ReceiptPolicy => "receipt_policy", + Self::MessageStoragePolicy => "message_storage_policy", } } } @@ -44,6 +53,19 @@ pub struct Delta { pub deleted_contact_ids: Vec, pub setting_upserts: Vec, pub deleted_setting_ids: Vec, + pub blob_upserts: Vec, + pub deleted_blob_ids: Vec, + pub blocked_user_upserts: Vec, + pub deleted_blocked_user_ids: Vec, + pub receipt_policy_changed: bool, + pub message_storage_policy_changed: bool, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct SyncCompactionResult { + pub removed_stale_clients: usize, + pub removed_events: usize, + pub removed_blob_tombstones: usize, } pub fn now_millis() -> i64 { @@ -104,6 +126,18 @@ pub fn has_session(user_id: i64, session_id: i64) -> Result }) } +pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result, StorageError> { + db::with_db(|conn| { + conn.query_row( + "SELECT acknowledged_version FROM client_sync_state WHERE user_id = ?1 AND session_id = ?2", + params![user_id, session_id], + |row| row.get(0), + ) + .optional() + .map_err(StorageError::from) + }) +} + pub fn acknowledge( user_id: i64, session_id: i64, @@ -131,6 +165,70 @@ pub fn acknowledge( }) } +/* Retain deltas until every non-stale client has acknowledged them. */ +pub fn compact_user_sync_state(user_id: i64) -> Result { + if user_id <= 0 { + return Err(StorageError::Other("invalid sync owner".into())); + } + let stale_before = now_millis().saturating_sub(STALE_CLIENT_SYNC_STATE_MS); + db::with_immediate_transaction(|tx| { + tx.execute( + "DELETE FROM client_message_deliveries WHERE user_id = ?1 AND session_id IN (SELECT session_id FROM client_sync_state WHERE user_id = ?1 AND updated_at < ?2)", + params![user_id, stale_before], + )?; + let removed_stale_clients = tx.execute( + "DELETE FROM client_sync_state WHERE user_id = ?1 AND updated_at < ?2", + params![user_id, stale_before], + )?; + let minimum_acknowledged: Option = tx.query_row( + "SELECT MIN(acknowledged_version) FROM client_sync_state WHERE user_id = ?1", + [user_id], + |row| row.get(0), + )?; + let revision = match minimum_acknowledged { + Some(revision) => revision, + None => tx + .query_row( + "SELECT version FROM sync_heads WHERE user_id = ?1", + [user_id], + |row| row.get(0), + ) + .unwrap_or(0), + }; + let removed_events = tx.execute( + "DELETE FROM sync_events WHERE user_id = ?1 AND version <= ?2", + params![user_id, revision], + )?; + let removed_blob_tombstones = tx.execute( + "DELETE FROM user_blobs WHERE user_id = ?1 AND deleted = 1 AND revision <= ?2", + params![user_id, revision], + )?; + Ok(SyncCompactionResult { + removed_stale_clients, + removed_events, + removed_blob_tombstones, + }) + }) +} + +pub fn compact_all_sync_state() -> Result { + let users = db::with_db(|conn| { + let mut statement = conn.prepare("SELECT user_id FROM sync_heads")?; + statement + .query_map([], |row| row.get::<_, i64>(0))? + .collect::, _>>() + .map_err(StorageError::from) + })?; + let mut total = SyncCompactionResult::default(); + for user_id in users { + let result = compact_user_sync_state(user_id)?; + total.removed_stale_clients += result.removed_stale_clients; + total.removed_events += result.removed_events; + total.removed_blob_tombstones += result.removed_blob_tombstones; + } + Ok(total) +} + /// Returns the final operation for each entity after `from_version`. pub fn delta(user_id: i64, from_version: i64, captured_head: i64) -> Result { if from_version < 0 || from_version > captured_head { @@ -163,6 +261,12 @@ fn reduce_events(events: BTreeMap<(String, i64), String>) -> Delta { ("contact", _) => out.contact_upserts.push(id), ("setting", "delete") => out.deleted_setting_ids.push(id), ("setting", _) => out.setting_upserts.push(id), + ("user_blob", "delete") => out.deleted_blob_ids.push(id), + ("user_blob", _) => out.blob_upserts.push(id), + ("blocked_user", "delete") => out.deleted_blocked_user_ids.push(id), + ("blocked_user", _) => out.blocked_user_upserts.push(id), + ("receipt_policy", _) => out.receipt_policy_changed = true, + ("message_storage_policy", _) => out.message_storage_policy_changed = true, _ => {} } } @@ -215,6 +319,20 @@ mod tests { assert!(delta.setting_upserts.is_empty()); } + #[test] + fn blob_delete_is_included_in_delta() { + let mut events = BTreeMap::new(); + events.insert( + (EntityType::UserBlob.as_str().to_string(), 7), + Operation::Delete.as_str().to_string(), + ); + + let delta = reduce_events(events); + + assert_eq!(delta.deleted_blob_ids, vec![7]); + assert!(delta.blob_upserts.is_empty()); + } + #[test] fn final_setting_operation_wins() { let mut events = BTreeMap::new(); diff --git a/iota-storage/src/util/user_blobs.rs b/iota-storage/src/util/user_blobs.rs new file mode 100644 index 0000000..34d1cb3 --- /dev/null +++ b/iota-storage/src/util/user_blobs.rs @@ -0,0 +1,256 @@ +/* Opaque client-owned data is stored without interpreting its encrypted bytes. */ +use crate::storage_error::StorageError; +use crate::util::{db, sync}; +use rusqlite::{Connection, OptionalExtension, Row, Transaction, params}; + +pub const MAX_BLOB_ID_BYTES: usize = 256; +pub const MAX_BLOB_BYTES: usize = 1_048_576; +pub const MAX_USER_BLOB_BYTES: usize = 16_777_216; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UserBlob { + pub id: i64, + pub user_id: i64, + pub blob_id: String, + pub blob: Vec, + pub revision: i64, + pub updated_at: i64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeletedUserBlob { + pub id: i64, + pub blob_id: String, + pub revision: i64, + pub changed: bool, +} + +fn validate(user_id: i64, blob_id: &str, blob: Option<&[u8]>) -> Result<(), StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid blob owner".into())); + } + if blob_id.is_empty() || blob_id.len() > MAX_BLOB_ID_BYTES { + return Err(StorageError::Other("invalid blob id".into())); + } + if let Some(blob) = blob { + if blob.len() > MAX_BLOB_BYTES { + return Err(StorageError::Other("blob exceeds size limit".into())); + } + } + Ok(()) +} + +fn from_row(row: &Row<'_>) -> rusqlite::Result { + Ok(UserBlob { + id: row.get(0)?, + user_id: row.get(1)?, + blob_id: row.get(2)?, + blob: row.get(3)?, + revision: row.get(4)?, + updated_at: row.get(5)?, + }) +} + +fn load(tx: &Transaction<'_>, id: i64) -> Result { + Ok(tx.query_row("SELECT id, user_id, blob_id, blob, revision, updated_at FROM user_blobs WHERE id = ?1 AND deleted = 0", [id], from_row)?) +} + +pub fn put( + user_id: i64, + blob_id: &str, + blob: &[u8], + expected_revision: Option, +) -> Result { + validate(user_id, blob_id, Some(blob))?; + db::with_immediate_transaction(|tx| { + let existing = tx + .query_row( + "SELECT id, revision, deleted, length(blob) FROM user_blobs WHERE user_id = ?1 AND blob_id = ?2", + params![user_id, blob_id], + |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, bool>(2)?, + r.get::<_, i64>(3)?, + )) + }, + ) + .optional()?; + let existing_id = if let Some((id, revision, deleted, _)) = existing { + if deleted { + if !matches!(expected_revision, Some(0)) { + return Err(StorageError::RevisionConflict); + } + } else if expected_revision != Some(revision) { + return Err(StorageError::RevisionConflict); + } + Some(id) + } else { + if !matches!(expected_revision, None | Some(0)) { + return Err(StorageError::RevisionConflict); + } + None + }; + let current_size = existing + .filter(|(_, _, deleted, _)| !*deleted) + .map(|(_, _, _, size)| size) + .unwrap_or(0); + let aggregate: i64 = tx.query_row("SELECT COALESCE(SUM(length(blob)), 0) FROM user_blobs WHERE user_id = ?1 AND deleted = 0", [user_id], |r| r.get(0))?; + let proposed_size = aggregate + .checked_sub(current_size) + .and_then(|size| size.checked_add(blob.len() as i64)) + .ok_or_else(|| StorageError::Other("user blob storage quota overflow".into()))?; + if proposed_size > MAX_USER_BLOB_BYTES as i64 { + return Err(StorageError::Other( + "user blob storage limit exceeded".into(), + )); + } + let id = match existing_id { + Some(id) => id, + None => { + tx.execute("INSERT INTO user_blobs (user_id, blob_id, blob, revision, deleted, updated_at) VALUES (?1, ?2, ?3, 0, 0, ?4)", params![user_id, blob_id, blob, sync::now_millis()])?; + tx.last_insert_rowid() + } + }; + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::UserBlob, + id, + sync::Operation::Upsert, + )?; + tx.execute("UPDATE user_blobs SET blob = ?2, revision = ?3, deleted = 0, updated_at = ?4 WHERE id = ?1", params![id, blob, revision, sync::now_millis()])?; + load(tx, id) + }) +} + +pub fn get(user_id: i64, blob_id: &str) -> Result, StorageError> { + validate(user_id, blob_id, None)?; + db::with_db(|conn| get_on(conn, user_id, blob_id)) +} +fn get_on( + conn: &Connection, + user_id: i64, + blob_id: &str, +) -> Result, StorageError> { + Ok(conn.query_row("SELECT id, user_id, blob_id, blob, revision, updated_at FROM user_blobs WHERE user_id = ?1 AND blob_id = ?2 AND deleted = 0", params![user_id, blob_id], from_row).optional()?) +} +pub fn list(user_id: i64) -> Result, StorageError> { + list_by_query( + user_id, + "SELECT id, user_id, blob_id, blob, revision, updated_at FROM user_blobs WHERE user_id = ?1 AND deleted = 0", + &[], + ) +} +pub fn list_by_ids(user_id: i64, ids: &[i64]) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid blob owner".into())); + } + if ids.is_empty() { + return Ok(Vec::new()); + } + db::with_db(|conn| { + let mut out = Vec::new(); + for id in ids { + if let Some(blob) = conn.query_row("SELECT id, user_id, blob_id, blob, revision, updated_at FROM user_blobs WHERE user_id = ?1 AND id = ?2 AND deleted = 0", params![user_id, id], from_row).optional()? { out.push(blob); } + } + Ok(out) + }) +} + +pub fn list_deleted_by_ids( + user_id: i64, + ids: &[i64], +) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid blob owner".into())); + } + if ids.is_empty() { + return Ok(Vec::new()); + } + db::with_db(|conn| { + let mut deleted = Vec::new(); + for id in ids { + if let Some(blob) = conn.query_row( + "SELECT id, blob_id, revision FROM user_blobs WHERE user_id = ?1 AND id = ?2 AND deleted = 1", + params![user_id, id], + |row| { + Ok(DeletedUserBlob { + id: row.get(0)?, + blob_id: row.get(1)?, + revision: row.get(2)?, + changed: true, + }) + }, + ).optional()? { + deleted.push(blob); + } + } + Ok(deleted) + }) +} +fn list_by_query(user_id: i64, query: &str, _: &[i64]) -> Result, StorageError> { + if user_id <= 0 { + return Err(StorageError::Other("invalid blob owner".into())); + } + db::with_db(|conn| { + let mut statement = conn.prepare(query)?; + let rows = statement.query_map([user_id], from_row)?; + rows.collect::, _>>() + .map_err(StorageError::from) + }) +} +pub fn delete( + user_id: i64, + blob_id: &str, + expected_revision: Option, +) -> Result, StorageError> { + validate(user_id, blob_id, None)?; + db::with_immediate_transaction(|tx| { + let existing = tx + .query_row( + "SELECT id, revision, deleted FROM user_blobs WHERE user_id = ?1 AND blob_id = ?2", + params![user_id, blob_id], + |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, bool>(2)?, + )) + }, + ) + .optional()?; + let Some((id, current, deleted)) = existing else { + return Ok(None); + }; + if deleted { + return Ok(Some(DeletedUserBlob { + id, + blob_id: blob_id.into(), + revision: current, + changed: false, + })); + } + if expected_revision != Some(current) { + return Err(StorageError::RevisionConflict); + } + let revision = sync::record_event( + tx, + user_id, + sync::EntityType::UserBlob, + id, + sync::Operation::Delete, + )?; + tx.execute( + "UPDATE user_blobs SET blob = X'', deleted = 1, revision = ?2, updated_at = ?3 WHERE id = ?1", + params![id, revision, sync::now_millis()], + )?; + Ok(Some(DeletedUserBlob { + id, + blob_id: blob_id.into(), + revision, + changed: true, + })) + }) +} diff --git a/mtp-type-maps b/mtp-type-maps index f4e45aa..4b82f4f 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c +Subproject commit 4b82f4f8139ed9aa74fa86f73ba8f0d565703c86 diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index adb18ee..263910f 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -1,9 +1,9 @@ use dashmap::{DashMap, DashSet}; use iota_logger::{log, log_cv_in, log_cv_out, log_t}; use iota_state::AppState; -use iota_storage::util::chat_files; use iota_storage::util::config_util::{CONFIG, modify_config}; -use iota_storage::util::{relay_queue, relay_replay}; +use iota_storage::util::relay_replay; +use iota_storage::util::{chat_files, client_relay_delivery, outgoing_relay, relay_queue}; use iota_util::crypto_helper::{self, keyring_from_base64}; use iota_util::crypto_util::{self}; use mtp::client::{Client, ClientConfig, MTPConnection, Policy, SendMode, Sender}; @@ -510,6 +510,9 @@ impl OmikronConnection { log_t!("omikron_authenticated"); + self.classify_legacy_pending_relays().await; + self.flush_pending_relays().await; + let maintenance_self = self.clone(); let maintenance_handle = tokio::spawn(async move { maintenance_self.maintenance_loop(connection).await; @@ -753,6 +756,7 @@ impl OmikronConnection { self.app.lock().unwrap().push_ping_val(ping_ms as f64); } + self.classify_legacy_pending_relays().await; self.flush_pending_relays().await; if let Err(error) = relay_replay::prune_completed( now_millis_i64().saturating_sub(RELAY_RETENTION_MILLIS), @@ -980,6 +984,74 @@ impl OmikronConnection { return; } + /* Evaluate recipient policy before reserving relay replay state or + * persisting the frame, so blocked traffic leaves no durable trace. */ + if recipient_is_local { + match iota_storage::util::blocked_users::is_blocked(recipient_id, signer_id) { + Ok(true) => { + log!( + "Rejecting Relay from blocked signer {} to recipient {}", + signer_id, + recipient_id + ); + self.send_relay_response( + Some(incoming_frame_id), + CommunicationType::ErrorNotAuthenticated, + ) + .await; + return; + } + Ok(false) => {} + Err(error) => { + log!( + "Relay block policy lookup failed for recipient {}: {}", + recipient_id, + error + ); + self.send_relay_response( + Some(incoming_frame_id), + CommunicationType::ErrorInternal, + ) + .await; + return; + } + } + } + + /* An origin Iota is authoritative for receipt disclosure. Inspect + * local-origin relay content before reserving or queuing the frame. */ + if signer_is_local { + let content = match open_verified_relay_content( + &verified, + &[&keyring], + verified.context.signer_id, + ) { + Ok(content) => content, + Err(error) => { + log!("Relay origin content verification failed: {}", error); + self.send_relay_response( + Some(incoming_frame_id), + CommunicationType::ErrorInvalidData, + ) + .await; + return; + } + }; + if let Err(error) = message_handlers::validate_outgoing_receipt_policy( + signer_id, + &verified.context, + &content, + ) { + log!("Rejecting local receipt relay: {}", error); + self.send_relay_response( + Some(incoming_frame_id), + CommunicationType::ErrorInvalidData, + ) + .await; + return; + } + } + let frame_bytes = match frame.clone().without_id().to_bytes() { Ok(bytes) => bytes, Err(error) => { @@ -1189,13 +1261,32 @@ impl OmikronConnection { return; } }; - if let Err(error) = relay_queue::enqueue( - RouteTarget::Iota(router), - &bytes, - now_millis_i64(), - frame_id, - &type_map_version, + let relay_identity = match ( + i64::try_from(verified.context.signer_id), + i64::try_from(verified.context.final_recipient_id), ) { + (Ok(signer_id), Ok(destination_user_id)) => relay_queue::RelayIdentity { + signer_id, + destination_user_id, + message_id: verified.context.message_id.clone(), + }, + _ => { + log!("Relay identity exceeds the storage range"); + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + .await; + return; + } + }; + if let Err(error) = + outgoing_relay::commit_outgoing_relay(outgoing_relay::OutgoingRelay { + target: RouteTarget::Iota(router), + identity: &relay_identity, + frame: &bytes, + created_at: now_millis_i64(), + frame_id, + type_map_version: &type_map_version, + }) + { log!("Relay origin retry queue failed: {}", error); record_origin_delivery_failure( verified.context.signer_id, @@ -1206,13 +1297,8 @@ impl OmikronConnection { .await; return; } - if let Err(error) = - relay_replay::mark_queued(verified.context.signer_id, &verified.context.message_id) - { - log!("Relay origin state update failed: {}", error); - } match self - .await_response(&forwarded, Some(Duration::from_secs(20))) + .await_relay_response(&forwarded, Duration::from_secs(20)) .await { Ok(response) if response.is_type(CommunicationType::Success) => { @@ -1233,33 +1319,27 @@ impl OmikronConnection { .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, + let Ok(signer_id) = i64::try_from(verified.context.signer_id) else { + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + .await; + return; + }; + if let Err(error) = + iota_storage::util::downstream_relay::acknowledge_iota_delivery( + router, + frame_id, + signer_id, &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: {}", + "Relay destination acknowledgement storage failed: {}", error ); - } - if let Err(error) = relay_replay::mark_downstream_acked( - verified.context.signer_id, - &verified.context.message_id, - ) { - log!("Relay origin delivery state update failed: {}", error); + self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) + .await; + return; } let response = response .add_typed_default( @@ -1277,11 +1357,30 @@ 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", - ); + if response.is_type(CommunicationType::ErrorInternal) { + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "destination_internal_error", + ); + } else if let Ok(signer_id) = i64::try_from(verified.context.signer_id) { + if let Err(error) = + iota_storage::util::downstream_relay::reject_iota_delivery( + router, + frame_id, + signer_id, + &verified.context.message_id, + "destination_rejected", + ) + { + log!("Relay rejection cleanup failed: {}", error); + } + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "destination_rejected", + ); + } self.send_relay_response( frame.id(), response @@ -1326,8 +1425,25 @@ impl OmikronConnection { return; } }; + let relay_identity = match ( + i64::try_from(verified.context.signer_id), + i64::try_from(verified.context.final_recipient_id), + ) { + (Ok(signer_id), Ok(destination_user_id)) => relay_queue::RelayIdentity { + signer_id, + destination_user_id, + message_id: verified.context.message_id.clone(), + }, + _ => { + log!("Relay identity exceeds the storage range"); + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + .await; + return; + } + }; if let Err(error) = relay_queue::enqueue( RouteTarget::User(destination), + &relay_identity, &bytes, now_millis_i64(), frame_id, @@ -1404,6 +1520,24 @@ impl OmikronConnection { } } + if signer_is_local && recipient_is_local { + let Ok(owner) = i64::try_from(verified.context.signer_id) else { + self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) + .await; + return; + }; + if let Err(error) = outgoing_relay::apply_outgoing_history_policy( + owner, + owner, + &verified.context.message_id, + ) { + log!("Shared-Iota outgoing history policy failed: {}", error); + self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) + .await; + return; + } + } + if let Err(error) = relay_replay::mark_queued(verified.context.signer_id, &verified.context.message_id) { @@ -1422,11 +1556,117 @@ impl OmikronConnection { } } + async fn classify_legacy_pending_relays(&self) { + let mut after_id = 0; + loop { + let records = match relay_queue::list_without_relay_identity_after(after_id, 100) { + Ok(records) => records, + Err(error) => { + log!("Pending relay ownership query failed: {}", error); + return; + } + }; + if records.is_empty() { + return; + } + for record in records { + after_id = record.id; + let Some(version) = mtp::type_map::Version::parse(&record.type_map_version) else { + log!( + "Deleting pending Relay {} with invalid type-map version", + record.id + ); + let _ = relay_queue::delete(record.id); + continue; + }; + let type_map = mtp::codec::TypeMap::new(version); + let Ok(frame) = CommunicationValue::from_bytes_with(&record.frame, &type_map) + else { + log!("Deleting pending Relay {} with invalid frame", record.id); + let _ = relay_queue::delete(record.id); + continue; + }; + let Some(local_iota_id) = CONFIG.load().iota_id else { + return; + }; + let Some(keyring) = self.keyring.read().await.as_ref().cloned() else { + return; + }; + let verified = verify_relay_metadata( + &frame, + local_iota_id, + &keyring, + |signer_id| async move { self.resolve_relay_signing_keys(signer_id).await }, + ) + .await; + let verified = match verified { + Ok(verified) => verified, + Err(RelayValidationError::KeyLookup(error)) => { + log!( + "Deferring pending Relay {} ownership lookup: {}", + record.id, + error + ); + continue; + } + Err(RelayValidationError::MissingSigningKeys(signer_id)) => { + log!( + "Deferring pending Relay {} until signer {} keys are available", + record.id, + signer_id + ); + continue; + } + Err(error) => { + log!( + "Deleting structurally invalid pending Relay {}: {}", + record.id, + error + ); + let _ = relay_queue::delete(record.id); + continue; + } + }; + let identity = match ( + i64::try_from(verified.context.signer_id), + i64::try_from(verified.context.final_recipient_id), + ) { + (Ok(signer_id), Ok(destination_user_id)) => relay_queue::RelayIdentity { + signer_id, + destination_user_id, + message_id: verified.context.message_id, + }, + _ => { + let _ = relay_queue::delete(record.id); + continue; + } + }; + if let Err(error) = relay_queue::set_relay_identity(record.id, &identity) { + log!( + "Pending Relay {} ownership backfill failed: {}", + record.id, + error + ); + } + } + } + } + async fn flush_pending_relays(&self) { let Ok(records) = relay_queue::list(100) else { return; }; for record in records { + if record.relay_signer_id.is_none() + || record.relay_destination_user_id.is_none() + || record.relay_message_id.is_none() + { + log!( + "Skipping pending Relay {} until ownership is classified", + record.id + ); + continue; + } let Some(version) = mtp::type_map::Version::parse(&record.type_map_version) else { log!( "Retaining pending Relay {} with invalid type-map version {}", @@ -1440,6 +1680,72 @@ impl OmikronConnection { log!("Retaining pending Relay {} with invalid frame", record.id); continue; }; + if record.relay_signer_id.is_none() + || record.relay_destination_user_id.is_none() + || record.relay_message_id.is_none() + { + let Some(local_iota_id) = CONFIG.load().iota_id else { + log!( + "Retaining pending Relay {} until the Iota identity is available", + record.id + ); + continue; + }; + let Some(keyring) = self.keyring.read().await.as_ref().cloned() else { + log!( + "Retaining pending Relay {} until the Iota keyring is available", + record.id + ); + continue; + }; + let resolver_connection = self; + let verified = verify_relay_metadata( + &frame, + local_iota_id, + &keyring, + move |signer_id| async move { + resolver_connection + .resolve_relay_signing_keys(signer_id) + .await + }, + ) + .await; + let Ok(verified) = verified else { + log!("Deleting unverifiable pending Relay {}", record.id); + if let Err(error) = relay_queue::delete(record.id) { + log!("Pending Relay {} cleanup failed: {}", record.id, error); + } + continue; + }; + let relay_identity = match ( + i64::try_from(verified.context.signer_id), + i64::try_from(verified.context.final_recipient_id), + ) { + (Ok(signer_id), Ok(destination_user_id)) => relay_queue::RelayIdentity { + signer_id, + destination_user_id, + message_id: verified.context.message_id, + }, + _ => { + log!( + "Deleting pending Relay {} with an out-of-range identity", + record.id + ); + if let Err(error) = relay_queue::delete(record.id) { + log!("Pending Relay {} cleanup failed: {}", record.id, error); + } + continue; + } + }; + if let Err(error) = relay_queue::set_relay_identity(record.id, &relay_identity) { + log!( + "Pending Relay {} ownership backfill failed: {}", + record.id, + error + ); + continue; + } + } let Ok(forwarded) = forward_verified_relay(&frame, record.target) else { log!( "Retaining pending Relay {} with invalid route target", @@ -1450,22 +1756,63 @@ impl OmikronConnection { match record.target { RouteTarget::Iota(destination_iota) => { match self - .await_response(&forwarded, Some(Duration::from_secs(20))) + .await_relay_response(&forwarded, Duration::from_secs(20)) .await { Ok(response) if response.is_type(CommunicationType::Success) => { - if let Err(error) = - relay_queue::acknowledge_iota(destination_iota, record.frame_id) + let accepted_at = response + .get_data(DataType::RelayAcceptedAt) + .as_number() + .and_then(|value| i64::try_from(value).ok()); + let relay_id = response.get_data(DataType::RelayMessageId).as_str(); + if let (Some(accepted_at), Some(message_id), Some(signer_id)) = ( + accepted_at, + record.relay_message_id.as_deref(), + record.relay_signer_id, + ) && relay_id == Some(message_id) { - log!( - "Pending Relay {} acknowledgement could not clear the queue: {}", - record.id, - error - ); + if let Err(error) = + iota_storage::util::downstream_relay::acknowledge_iota_delivery( + destination_iota, + record.frame_id, + signer_id, + message_id, + accepted_at, + ) + { + log!( + "Pending Relay {} acknowledgement failed: {}", + record.id, + error + ); + } + } else { + log!("Pending Relay {} returned malformed Success", record.id); + } + } + Ok(response) if !response.is_type(CommunicationType::ErrorInternal) => { + if let (Some(message_id), Some(signer_id)) = + (record.relay_message_id.as_deref(), record.relay_signer_id) + { + if let Err(error) = + iota_storage::util::downstream_relay::reject_iota_delivery( + destination_iota, + record.frame_id, + signer_id, + message_id, + "destination_rejected", + ) + { + log!( + "Pending Relay {} rejection cleanup failed: {}", + record.id, + error + ); + } } } Ok(response) => log!( - "Pending Relay {} route returned {}", + "Pending Relay {} route returned retryable {}", record.id, response.get_type() ), @@ -1497,16 +1844,15 @@ impl OmikronConnection { .as_number() .and_then(|value| u64::try_from(value).ok()) { - match relay_queue::acknowledge(destination_id, frame_id) { - Ok(true) => { - if let Err(error) = - relay_replay::mark_delivered_for_frame(destination_id, frame_id) - { - log!("Relay delivery state update failed: {}", error); - } + let Ok(destination_id) = i64::try_from(destination_id) else { + log!("Relay destination ID exceeds storage range"); + return; + }; + match client_relay_delivery::acknowledge_client_delivery(destination_id, frame_id) { + Ok(client_relay_delivery::ClientRelayDeliveryResult::Acknowledged) => { return; } - Ok(false) => {} + Ok(client_relay_delivery::ClientRelayDeliveryResult::NotFound) => {} Err(error) => log!("Relay delivery acknowledgement failed: {}", error), } } @@ -1587,6 +1933,18 @@ impl OmikronConnection { dispatch!(SyncedSettingGet, handle_synced_setting_get); dispatch!(SyncedSettingDelete, handle_synced_setting_delete); dispatch!(SyncedSettingsList, handle_synced_settings_list); + dispatch!(UserBlobPut, handle_user_blob_put); + dispatch!(UserBlobGet, handle_user_blob_get); + dispatch!(UserBlobDelete, handle_user_blob_delete); + dispatch!(UserBlobList, handle_user_blob_list); + dispatch!(UserBlock, handle_user_block); + dispatch!(UserUnblock, handle_user_unblock); + dispatch!(BlockedUsersGet, handle_blocked_users_get); + dispatch!(ReceiptPolicyGet, handle_receipt_policy_get); + dispatch!(ReceiptPolicySet, handle_receipt_policy_set); + dispatch!(MessageStoragePolicyGet, handle_message_storage_policy_get); + dispatch!(MessageStoragePolicySet, handle_message_storage_policy_set); + dispatch!(UserBlockCheck, handle_user_block_check); dispatch!(EraseHostedUserData, handle_erase_hosted_user_data); } @@ -2232,6 +2590,90 @@ impl OmikronConnection { .await; } + async fn handle_user_blob_put(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_user_blob_put(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_user_blob_get(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_user_blob_get(cv)) + .await; + } + + async fn handle_user_blob_delete(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_user_blob_delete(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_user_blob_list(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_user_blob_list(cv)) + .await; + } + + async fn handle_user_block(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_user_block(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_user_unblock(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_user_unblock(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_blocked_users_get(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_blocked_users_get(cv)) + .await; + } + + async fn handle_receipt_policy_get(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_receipt_policy_get(cv)) + .await; + } + + async fn handle_receipt_policy_set(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_receipt_policy_set(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_message_storage_policy_get(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_message_storage_policy_get(cv)) + .await; + } + + async fn handle_message_storage_policy_set(self: Arc, cv: &CommunicationValue) { + let mutation = message_handlers::handle_message_storage_policy_set(cv); + let _ = self.send_message(&mutation.response).await; + if let Some(changed) = mutation.changed { + let _ = self.send_message(&changed).await; + } + } + + async fn handle_user_block_check(self: Arc, cv: &CommunicationValue) { + let _ = self + .send_message(&message_handlers::handle_user_block_check(cv)) + .await; + } + // ------------------------------------------------------------------------- // Public API // ------------------------------------------------------------------------- @@ -2365,6 +2807,40 @@ impl OmikronConnection { } } + async fn await_relay_response( + &self, + cv: &CommunicationValue, + timeout: Duration, + ) -> Result { + let (tx, rx) = oneshot::channel(); + let msg_id = cv.require_id().map_err(|error| error.to_string())?; + WAITING_TASKS.insert( + msg_id, + WaitingTask { + task: Box::new(move |response| { + let _ = tx.send(response); + true + }), + inserted_at: Instant::now(), + }, + ); + if let Err(error) = self.send_message(cv).await { + WAITING_TASKS.remove(&msg_id); + return Err(format!("Relay send failed: {error}")); + } + match tokio::time::timeout(timeout, rx).await { + Ok(Ok(response)) => Ok(response), + Ok(Err(_)) => { + WAITING_TASKS.remove(&msg_id); + Err("Relay response channel closed".into()) + } + Err(_) => { + WAITING_TASKS.remove(&msg_id); + Err("Relay response timed out".into()) + } + } + } + pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { let mut rx = self.state_watch_tx.subscribe(); if rx.borrow().is_connected() {