tensamin/iota
Watch
6
0
Fork
You've already forked iota
 0
Code Issues Pull requests 3 Projects Releases 1 Packages Wiki Activity Actions

[Add] Client settings

Browse source
This commit is contained in:
Alex Emmet 2026-09-01 22:17:57 +02:00
commit 430c12e139
No known key found for this signature in database
22 changed files with 2779 additions and 127 deletions
Download patch file Download diff file Expand all files Collapse all files

631
iota-connection/src/message_handlers.rs
View file

@ -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<CommunicationValue>,
}
#[derive(Debug)]
pub struct BlobMutation {
pub response: CommunicationValue,
pub changed: Option<CommunicationValue>,
}
#[derive(Debug)]
pub struct PolicyMutation {
pub response: CommunicationValue,
pub changed: Option<CommunicationValue>,
}
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<i64, CommunicationValue> {
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};

6
iota-daemon-lib/src/command_router.rs
View file

@ -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)

62
iota-daemon/src/main.rs
View file

@ -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)]

4
iota-storage/src/storage_error.rs
View file

@ -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),
}

39
iota-storage/src/users/user_manager.rs
View file

@ -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()],

104
iota-storage/src/util/blocked_users.rs Normal file
View file

@ -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<BlockMutation, StorageError> {
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<Option<BlockMutation>, 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<bool, StorageError> {
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<Vec<i64>, 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::<Result<Vec<_>, _>>()
.map_err(StorageError::from)
})
}

225
iota-storage/src/util/chat_files.rs
View file

@ -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<i64>>(1)?, row.get::<_, Option<String>>(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<i64, StorageError> {
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<i64, StorageError> {
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<i64, StorageError> {
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<i64>>(2)?)),
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, Option<i64>>(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<StoredMessage
// A journal id uniquely identifies a row. Load all messages for this owner and retain only
// those ids; this keeps reaction hydration identical to normal message loading.
match db::with_db(|conn| {
let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0")?;
let mut stmt = conn.prepare("SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 AND deleted_by_external = 0 AND history_deleted = 0")?;
let rows = stmt.query_map([storage_owner], |row| {
let external_user: i64 = row.get(19)?;
Ok(StoredMessage {
@ -1056,7 +1189,7 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage
pub fn get_all_messages(storage_owner: i64) -> Vec<StoredMessage> {
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))?

93
iota-storage/src/util/client_message_delivery.rs Normal file
View file

@ -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<Item = i64>,
) -> 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<Vec<i64>, 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::<Result<Vec<i64>, _>>()
.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::<Result<Vec<_>, _>>()?
};
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<usize, StorageError> {
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],
)?)
})
}

39
iota-storage/src/util/client_relay_delivery.rs Normal file
View file

@ -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<ClientRelayDeliveryResult, StorageError> {
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)
})
}

141
iota-storage/src/util/db.rs
View file

@ -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",

69
iota-storage/src/util/downstream_relay.rs Normal file
View file

@ -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(())
})
}

28
iota-storage/src/util/message_retention.rs Normal file
View file

@ -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<RetentionResult, StorageError> {
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::<Result<Vec<_>, _>>()
.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)
}

149
iota-storage/src/util/message_storage_policy.rs Normal file
View file

@ -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<MessageHistoryMode, StorageError> {
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<MessageStoragePolicy, StorageError> {
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<i64>>(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<MessageStoragePolicy, StorageError> {
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<i64>>(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, i64)>,
) -> Result<MessageStoragePolicy, StorageError> {
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<MessageStoragePolicy, StorageError> {
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<i64>) = 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,
})
})
}

9
iota-storage/src/util/mod.rs
View file

@ -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;

61
iota-storage/src/util/outgoing_relay.rs Normal file
View file

@ -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(())
})
}

64
iota-storage/src/util/receipt_policy.rs Normal file
View file

@ -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<ReceiptPolicy, StorageError> {
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<ReceiptPolicy, StorageError> {
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,
})
})
}

151
iota-storage/src/util/relay_queue.rs
View file

@ -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<i64>,
pub relay_destination_user_id: Option<i64>,
pub relay_message_id: Option<String>,
}
#[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<Vec<PendingRelay>, 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<Vec<PendingRelay>, 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::<Result<Vec<_>, _>>().map_err(Into::into)
})
}
pub fn list_without_relay_identity() -> Result<Vec<PendingRelay>, StorageError> {
list_without_relay_identity_after(0, i64::MAX)
}
pub fn list_without_relay_identity_after(
after_id: i64,
limit: i64,
) -> Result<Vec<PendingRelay>, 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::<Result<Vec<_>, _>>().map_err(Into::into)
})
}
fn pending_relay_from_row(row: &rusqlite::Row<'_>) -> Result<PendingRelay, rusqlite::Error> {
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<bool, StorageError> {
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<Option<(i64, RelayIdentity)>, 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<i64>>(1)?;
let destination_user_id = row.get::<_, Option<i64>>(2)?;
let message_id = row.get::<_, Option<String>>(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<bool, StorageError> {
let destination_id = i64::try_from(destination_id)
.map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;

59
iota-storage/src/util/relay_replay.rs
View file

@ -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<Option<DeliveredRelay>, 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")
}

122
iota-storage/src/util/sync.rs
View file

@ -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<i64>,
pub setting_upserts: Vec<i64>,
pub deleted_setting_ids: Vec<i64>,
pub blob_upserts: Vec<i64>,
pub deleted_blob_ids: Vec<i64>,
pub blocked_user_upserts: Vec<i64>,
pub deleted_blocked_user_ids: Vec<i64>,
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<bool, StorageError>
})
}
pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result<Option<i64>, 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<SyncCompactionResult, StorageError> {
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<i64> = 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<SyncCompactionResult, StorageError> {
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::<Result<Vec<_>, _>>()
.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<Delta, StorageError> {
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();

256
iota-storage/src/util/user_blobs.rs Normal file
View file

@ -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<u8>,
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<UserBlob> {
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<UserBlob, StorageError> {
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<i64>,
) -> Result<UserBlob, StorageError> {
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<Option<UserBlob>, 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<Option<UserBlob>, 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<Vec<UserBlob>, 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<Vec<UserBlob>, 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<Vec<DeletedUserBlob>, 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<Vec<UserBlob>, 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::<Result<Vec<_>, _>>()
.map_err(StorageError::from)
})
}
pub fn delete(
user_id: i64,
blob_id: &str,
expected_revision: Option<i64>,
) -> Result<Option<DeletedUserBlob>, 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,
}))
})
}

2
mtp-type-maps

@ -1 +1 @@
Subproject commit f4e45aa3a3ad0e3c3a257f66857b904a1af7901c
Subproject commit 4b82f4f8139ed9aa74fa86f73ba8f0d565703c86

592
omikron-connector/src/omikron_connection.rs
View file

@ -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<Self>, 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<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_user_blob_get(cv))
.await;
}
async fn handle_user_blob_delete(self: Arc<Self>, 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<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_user_blob_list(cv))
.await;
}
async fn handle_user_block(self: Arc<Self>, 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<Self>, 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<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_blocked_users_get(cv))
.await;
}
async fn handle_receipt_policy_get(self: Arc<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_receipt_policy_get(cv))
.await;
}
async fn handle_receipt_policy_set(self: Arc<Self>, 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<Self>, cv: &CommunicationValue) {
let _ = self
.send_message(&message_handlers::handle_message_storage_policy_get(cv))
.await;
}
async fn handle_message_storage_policy_set(self: Arc<Self>, 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<Self>, 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<CommunicationValue, String> {
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<Duration>) -> Result<(), String> {
let mut rx = self.state_watch_tx.subscribe();
if rx.borrow().is_connected() {