Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e25aebaec4 | |||
|
3bfec96848 |
||
|
|
29d46d1c95 |
16 changed files with 819 additions and 344 deletions
142
Cargo.lock
generated
142
Cargo.lock
generated
|
|
@ -51,7 +51,7 @@ dependencies = [
|
|||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"rand 0.10.2",
|
||||
"sha1 0.11.0",
|
||||
"sha1",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
|
|
@ -224,13 +224,13 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "aes"
|
||||
version = "0.8.4"
|
||||
version = "0.9.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0"
|
||||
checksum = "35f0f96ce78e38c3dc6d8948aa8163d06385be74000f3c7a95bf1eef35d3ea32"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cipher",
|
||||
"cpufeatures 0.2.17",
|
||||
"cipher 0.5.2",
|
||||
"cpubits",
|
||||
"cpufeatures 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -337,15 +337,6 @@ dependencies = [
|
|||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arbitrary"
|
||||
version = "1.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1"
|
||||
dependencies = [
|
||||
"derive_arbitrary",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arc-swap"
|
||||
version = "1.9.2"
|
||||
|
|
@ -541,6 +532,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
|
||||
dependencies = [
|
||||
"hybrid-array",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -652,7 +644,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cipher",
|
||||
"cipher 0.4.4",
|
||||
"cpufeatures 0.2.17",
|
||||
]
|
||||
|
||||
|
|
@ -675,7 +667,7 @@ checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35"
|
|||
dependencies = [
|
||||
"aead",
|
||||
"chacha20 0.9.1",
|
||||
"cipher",
|
||||
"cipher 0.4.4",
|
||||
"poly1305",
|
||||
"zeroize",
|
||||
]
|
||||
|
|
@ -700,10 +692,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
|
||||
dependencies = [
|
||||
"crypto-common 0.1.7",
|
||||
"inout",
|
||||
"inout 0.1.4",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cipher"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e8cf2a2c93cd704877c0858356ed03480ff301ee950b43f1cbe4573b088bfa6c"
|
||||
dependencies = [
|
||||
"crypto-common 0.2.2",
|
||||
"inout 0.2.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "clap"
|
||||
version = "4.6.6"
|
||||
|
|
@ -817,9 +819,9 @@ checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
|
|||
|
||||
[[package]]
|
||||
name = "constant_time_eq"
|
||||
version = "0.3.1"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c74b8349d32d297c9134b8c88677813a227df8f779daa29bfc29c183fe3dca6"
|
||||
checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b"
|
||||
|
||||
[[package]]
|
||||
name = "convert_case"
|
||||
|
|
@ -867,6 +869,12 @@ version = "0.8.7"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
|
||||
|
||||
[[package]]
|
||||
name = "cpubits"
|
||||
version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "15b85f9c39137c3a891689859392b1bd49812121d0d61c9caf00d46ed5ce06ae"
|
||||
|
||||
[[package]]
|
||||
name = "cpufeatures"
|
||||
version = "0.2.17"
|
||||
|
|
@ -885,21 +893,6 @@ dependencies = [
|
|||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crc"
|
||||
version = "3.4.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d"
|
||||
dependencies = [
|
||||
"crc-catalog",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crc-catalog"
|
||||
version = "2.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
|
||||
|
||||
[[package]]
|
||||
name = "crc32fast"
|
||||
version = "1.5.0"
|
||||
|
|
@ -1139,17 +1132,6 @@ version = "0.5.8"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
|
||||
|
||||
[[package]]
|
||||
name = "derive_arbitrary"
|
||||
version = "1.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "derive_more"
|
||||
version = "2.1.1"
|
||||
|
|
@ -1194,6 +1176,7 @@ dependencies = [
|
|||
"const-oid 0.10.2",
|
||||
"crypto-common 0.2.2",
|
||||
"ctutils",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -1722,16 +1705,7 @@ version = "0.13.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018"
|
||||
dependencies = [
|
||||
"hmac 0.13.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hmac"
|
||||
version = "0.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
|
||||
dependencies = [
|
||||
"digest 0.10.7",
|
||||
"hmac",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -2051,6 +2025,15 @@ dependencies = [
|
|||
"generic-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "inout"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4250ce6452e92010fdf7268ccc5d14faa80bb12fc741938534c58f16804e03c7"
|
||||
dependencies = [
|
||||
"hybrid-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "instability"
|
||||
version = "0.3.13"
|
||||
|
|
@ -2579,12 +2562,11 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
|
|||
|
||||
[[package]]
|
||||
name = "lzma-rust2"
|
||||
version = "0.13.0"
|
||||
version = "0.16.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c60a23ffb90d527e23192f1246b14746e2f7f071cb84476dd879071696c18a4a"
|
||||
checksum = "ca93e534d1142d1d0dcca6d25fe302508a5dfb40b302802904577725ea0b695b"
|
||||
dependencies = [
|
||||
"crc",
|
||||
"sha2 0.10.9",
|
||||
"sha2 0.11.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -3128,12 +3110,12 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "pbkdf2"
|
||||
version = "0.12.2"
|
||||
version = "0.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2"
|
||||
checksum = "112d82ceb8c5bf524d9af484d4e4970c9fd5a0cc15ba14ad93dccd28873b0629"
|
||||
dependencies = [
|
||||
"digest 0.10.7",
|
||||
"hmac 0.12.1",
|
||||
"digest 0.11.3",
|
||||
"hmac",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -4084,17 +4066,6 @@ dependencies = [
|
|||
"unsafe-libyaml",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha1"
|
||||
version = "0.10.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.2.17",
|
||||
"digest 0.10.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha1"
|
||||
version = "0.11.0"
|
||||
|
|
@ -4433,7 +4404,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.3",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.61.2",
|
||||
|
|
@ -4562,6 +4533,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134"
|
||||
dependencies = [
|
||||
"deranged",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"num-conv",
|
||||
"num_threads",
|
||||
|
|
@ -4759,6 +4731,12 @@ version = "0.2.5"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
|
||||
|
||||
[[package]]
|
||||
name = "typed-path"
|
||||
version = "0.12.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e"
|
||||
|
||||
[[package]]
|
||||
name = "typenum"
|
||||
version = "1.20.1"
|
||||
|
|
@ -5550,26 +5528,26 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "zip"
|
||||
version = "6.0.0"
|
||||
version = "8.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "eb2a05c7c36fde6c09b08576c9f7fb4cda705990f73b58fe011abf7dfb24168b"
|
||||
checksum = "2d04a6b5381502aa6087c94c669499eb1602eb9c5e8198e534de571f7154809b"
|
||||
dependencies = [
|
||||
"aes",
|
||||
"arbitrary",
|
||||
"bzip2",
|
||||
"constant_time_eq",
|
||||
"crc32fast",
|
||||
"deflate64",
|
||||
"flate2",
|
||||
"getrandom 0.3.4",
|
||||
"hmac 0.12.1",
|
||||
"getrandom 0.4.3",
|
||||
"hmac",
|
||||
"indexmap",
|
||||
"lzma-rust2",
|
||||
"memchr",
|
||||
"pbkdf2",
|
||||
"ppmd-rust",
|
||||
"sha1 0.10.7",
|
||||
"sha1",
|
||||
"time",
|
||||
"typed-path",
|
||||
"zeroize",
|
||||
"zopfli",
|
||||
"zstd",
|
||||
|
|
|
|||
|
|
@ -115,6 +115,9 @@ fn validate_relay_identity(
|
|||
pub fn apply_verified_relay_content(
|
||||
context: &VerifiedRelayContext,
|
||||
content: &VerifiedRelayContent,
|
||||
accepted_at: i64,
|
||||
storage_owner: i64,
|
||||
sent_by_self: bool,
|
||||
) -> Result<(), String> {
|
||||
validate_relay_identity(context, &content.content)?;
|
||||
let sender_id = i64::try_from(context.signer_id)
|
||||
|
|
@ -125,8 +128,48 @@ pub fn apply_verified_relay_content(
|
|||
.map_err(|_| "Relay creation time exceeds the local storage range".to_string())?;
|
||||
|
||||
match content.message_type.as_str() {
|
||||
"MessageState" => {
|
||||
let partner_id = relay_number(
|
||||
&content.content,
|
||||
DataType::ChatPartnerId,
|
||||
&context.type_map,
|
||||
)
|
||||
.and_then(|value| i64::try_from(value).ok())
|
||||
.filter(|id| *id == recipient_id)
|
||||
.ok_or_else(|| "Relay MessageState has an invalid ChatPartnerId".to_string())?;
|
||||
let relay_message_id = relay_string(
|
||||
&content.content,
|
||||
DataType::RelayMessageId,
|
||||
&context.type_map,
|
||||
)
|
||||
.ok_or_else(|| "Relay MessageState is missing RelayMessageId".to_string())?;
|
||||
let event_at = relay_number(&content.content, DataType::EventAt, &context.type_map)
|
||||
.and_then(|value| i64::try_from(value).ok())
|
||||
.ok_or_else(|| "Relay MessageState is missing EventAt".to_string())?;
|
||||
let state = relay_string(
|
||||
&content.content,
|
||||
DataType::MessageState,
|
||||
&context.type_map,
|
||||
)
|
||||
.map(MessageState::from_str)
|
||||
.filter(|state| matches!(state, MessageState::Received | MessageState::Read))
|
||||
.ok_or_else(|| "Relay MessageState has an invalid state".to_string())?;
|
||||
chat_files::record_message_receipt(
|
||||
storage_owner,
|
||||
recipient_id,
|
||||
relay_message_id,
|
||||
sender_id,
|
||||
&context.message_id,
|
||||
state,
|
||||
event_at,
|
||||
now_millis_i64(),
|
||||
)
|
||||
.map_err(|error| error.to_string())?;
|
||||
let _ = partner_id;
|
||||
Ok(())
|
||||
}
|
||||
"MessageSend" => {
|
||||
let message = relay_string(&content.content, DataType::AppContent, &context.type_map)
|
||||
let message = relay_string(&content.content, DataType::Content, &context.type_map)
|
||||
.ok_or_else(|| "Relay MessageSend is missing Content".to_string())?;
|
||||
let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map)
|
||||
.and_then(|value| i64::try_from(value).ok())
|
||||
|
|
@ -136,20 +179,35 @@ pub fn apply_verified_relay_content(
|
|||
.unwrap_or_default();
|
||||
let reply_to = relay_number(&content.content, DataType::ReplyId, &context.type_map)
|
||||
.and_then(|value| i64::try_from(value).ok());
|
||||
chat_files::add_message(
|
||||
u128::try_from(send_time)
|
||||
.map_err(|_| "Relay MessageSend has a negative SendTime".to_string())?,
|
||||
false,
|
||||
recipient_id,
|
||||
sender_id,
|
||||
message,
|
||||
let relay_message_id = relay_string(
|
||||
&content.content,
|
||||
DataType::RelayMessageId,
|
||||
&context.type_map,
|
||||
)
|
||||
.ok_or_else(|| "Relay MessageSend is missing RelayMessageId".to_string())?;
|
||||
if relay_message_id != context.message_id {
|
||||
return Err("Relay MessageSend identity does not match its protected message ID".into());
|
||||
}
|
||||
chat_files::add_message(chat_files::NewMessage {
|
||||
relay_signer_id: sender_id,
|
||||
relay_message_id,
|
||||
authored_at: created_at,
|
||||
send_time,
|
||||
storage_owner,
|
||||
external_user: if sent_by_self { recipient_id } else { sender_id },
|
||||
sent_by_self,
|
||||
content: message,
|
||||
height,
|
||||
reply_to,
|
||||
);
|
||||
origin_iota_received_at: sent_by_self.then_some(accepted_at),
|
||||
destination_iota_received_at: (!sent_by_self).then_some(accepted_at),
|
||||
initial_state: MessageState::Sent,
|
||||
})
|
||||
.map_err(|error| error.to_string())?;
|
||||
Ok(())
|
||||
}
|
||||
"MessageEdit" => {
|
||||
let message = relay_string(&content.content, DataType::AppContent, &context.type_map)
|
||||
let message = relay_string(&content.content, DataType::Content, &context.type_map)
|
||||
.ok_or_else(|| "Relay MessageEdit is missing Content".to_string())?;
|
||||
let send_time = relay_number(&content.content, DataType::SendTime, &context.type_map)
|
||||
.and_then(|value| i64::try_from(value).ok())
|
||||
|
|
@ -217,7 +275,7 @@ pub fn handle_message_edit(cv: &CommunicationValue) -> CommunicationValue {
|
|||
Ok(mutation) => mutation,
|
||||
Err(response) => return response,
|
||||
};
|
||||
let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
|
||||
let Some(content) = cv.get_data(DataType::Content).as_str() else {
|
||||
return error_response(cv, CommunicationType::ErrorInvalidData);
|
||||
};
|
||||
|
||||
|
|
@ -265,6 +323,9 @@ pub fn handle_message_reaction(cv: &CommunicationValue, add: bool) -> Communicat
|
|||
|
||||
match result {
|
||||
Ok(()) => success_response(cv),
|
||||
Err(iota_storage::storage_error::StorageError::ReactionLimitReached) => {
|
||||
error_response(cv, CommunicationType::ErrorInvalidData)
|
||||
}
|
||||
Err(_) => error_response(cv, CommunicationType::ErrorNotFound),
|
||||
}
|
||||
}
|
||||
|
|
@ -288,7 +349,7 @@ fn stored_message_fields(
|
|||
) -> Vec<(DataType, DataValue)> {
|
||||
let mut fields = vec![
|
||||
(
|
||||
DataType::AppMessageId,
|
||||
DataType::MessageId,
|
||||
DataValue::SignedNumber(message.id as i128),
|
||||
),
|
||||
(
|
||||
|
|
@ -296,7 +357,7 @@ fn stored_message_fields(
|
|||
DataValue::SignedNumber(message.message_time as i128),
|
||||
),
|
||||
(
|
||||
DataType::AppContent,
|
||||
DataType::Content,
|
||||
DataValue::Str(message.content.clone()),
|
||||
),
|
||||
(
|
||||
|
|
@ -316,6 +377,25 @@ fn stored_message_fields(
|
|||
if let Ok(sender_id) = u128::try_from(sender_id) {
|
||||
fields.push((DataType::SenderId, DataValue::UnsignedNumber(sender_id)));
|
||||
}
|
||||
if let Some(relay_message_id) = &message.relay_message_id {
|
||||
fields.push((
|
||||
DataType::RelayMessageId,
|
||||
DataValue::Str(relay_message_id.clone()),
|
||||
));
|
||||
}
|
||||
for (data_type, timestamp) in [
|
||||
(DataType::AuthoredAt, message.authored_at),
|
||||
(DataType::OriginIotaReceivedAt, message.origin_iota_received_at),
|
||||
(DataType::DestinationIotaReceivedAt, message.destination_iota_received_at),
|
||||
(DataType::ClientReceivedAt, message.client_received_at),
|
||||
(DataType::ClientReceivedRecordedAt, message.client_received_recorded_at),
|
||||
(DataType::ReadAt, message.read_at),
|
||||
(DataType::ReadRecordedAt, message.read_recorded_at),
|
||||
] {
|
||||
if let Some(timestamp) = timestamp {
|
||||
fields.push((data_type, DataValue::SignedNumber(timestamp.into())));
|
||||
}
|
||||
}
|
||||
if message.edited {
|
||||
fields.push((DataType::Edited, DataValue::Bool(true)));
|
||||
}
|
||||
|
|
@ -398,7 +478,7 @@ pub fn handle_get_chat_secret(cv: &CommunicationValue) -> CommunicationValue {
|
|||
DataValue::Str(record.wrapping_scheme),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::AppCreatedAt,
|
||||
DataType::CreatedAt,
|
||||
DataValue::SignedNumber(record.created_at as i128),
|
||||
)
|
||||
.add_typed_default(
|
||||
|
|
@ -760,10 +840,18 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue {
|
|||
Err(response) => return response,
|
||||
};
|
||||
|
||||
let message = match chat_files::get_message(owner, send_time, partner_id) {
|
||||
Ok(Some(message)) => message,
|
||||
Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
|
||||
Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
|
||||
let (message, offset) = match partner_id {
|
||||
Some(partner_id) => match chat_files::get_message_with_offset(owner, partner_id, send_time)
|
||||
{
|
||||
Ok(Some((message, offset))) => (message, Some(offset)),
|
||||
Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
|
||||
Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
|
||||
},
|
||||
None => match chat_files::get_message(owner, send_time, None) {
|
||||
Ok(Some(message)) => (message, None),
|
||||
Ok(None) => return error_response(cv, CommunicationType::ErrorNotFound),
|
||||
Err(_) => return error_response(cv, CommunicationType::ErrorInvalidData),
|
||||
},
|
||||
};
|
||||
|
||||
let mut response = CommunicationValue::new(CommunicationType::MessageGet)
|
||||
|
|
@ -772,6 +860,10 @@ pub fn handle_message_get(cv: &CommunicationValue) -> CommunicationValue {
|
|||
for (data_type, value) in stored_message_fields(&message, owner, message.external_user) {
|
||||
response = response.add_typed_default(data_type, value);
|
||||
}
|
||||
if let Some(offset) = offset {
|
||||
response =
|
||||
response.add_typed_default(DataType::Offset, DataValue::SignedNumber(offset as i128));
|
||||
}
|
||||
response
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,8 +2,6 @@ use iota_util::route_target::RouteTarget;
|
|||
use mtp::codec::{
|
||||
CommunicationValue, ProtectionPolicy, RelayError, RelayOpenOptions, SignaturePolicy, TypeMap,
|
||||
VerifiedRelayContent, VerifiedRelayMetadata, forward_relay_frame,
|
||||
open_relay_content_with_keyrings, open_relay_metadata_with_without_replay,
|
||||
relay_metadata_claimed_signer_id,
|
||||
open_relay_content_with_limits_without_replay, open_relay_metadata_with_without_replay,
|
||||
relay_metadata_claimed_signer_id_with_options,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -6,6 +6,6 @@ edition = "2024"
|
|||
[dependencies]
|
||||
anyhow = "1"
|
||||
tempfile = "3"
|
||||
zip = "6"
|
||||
zip = "8"
|
||||
serde_json = "1"
|
||||
iota-paths = { path = "../iota-paths" }
|
||||
|
|
|
|||
|
|
@ -12,8 +12,4 @@ dashmap = "6.1.0"
|
|||
once_cell = "1.21.3"
|
||||
tokio = { version = "1.50.0", features = ["full"] }
|
||||
json = "*"
|
||||
<<<<<<< HEAD
|
||||
sysinfo = "0.38.0"
|
||||
=======
|
||||
sysinfo = "0.39.0"
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
|
|
|
|||
|
|
@ -7,10 +7,6 @@ edition = "2024"
|
|||
iota-logger = { path = "../iota-logger" }
|
||||
iota-util = { path = "../iota-util" }
|
||||
iota-paths = { path = "../iota-paths" }
|
||||
<<<<<<< HEAD
|
||||
|
||||
=======
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
base64 = "0.22.1"
|
||||
json = "*"
|
||||
arc-swap = "1"
|
||||
|
|
|
|||
|
|
@ -8,6 +8,8 @@ pub enum StorageError {
|
|||
Pool(String),
|
||||
#[error("IO error: {0}")]
|
||||
Io(#[from] std::io::Error),
|
||||
#[error("message has reached the unique reaction limit")]
|
||||
ReactionLimitReached,
|
||||
#[error("{0}")]
|
||||
Other(String),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,7 +2,9 @@ use crate::storage_error::StorageError;
|
|||
use crate::util::db;
|
||||
use crate::util::sync::{self, EntityType, Operation};
|
||||
use iota_logger::log;
|
||||
use rusqlite::params;
|
||||
use rusqlite::{OptionalExtension, params};
|
||||
|
||||
pub const MAX_UNIQUE_REACTIONS_PER_MESSAGE: usize = 10;
|
||||
|
||||
#[derive(PartialEq, Debug, Clone)]
|
||||
pub enum MessageState {
|
||||
|
|
@ -48,7 +50,16 @@ impl MessageState {
|
|||
pub struct StoredMessage {
|
||||
pub id: i64,
|
||||
pub external_user: i64,
|
||||
pub relay_signer_id: Option<i64>,
|
||||
pub relay_message_id: Option<String>,
|
||||
pub message_time: i64,
|
||||
pub authored_at: Option<i64>,
|
||||
pub origin_iota_received_at: Option<i64>,
|
||||
pub destination_iota_received_at: Option<i64>,
|
||||
pub client_received_at: Option<i64>,
|
||||
pub client_received_recorded_at: Option<i64>,
|
||||
pub read_at: Option<i64>,
|
||||
pub read_recorded_at: Option<i64>,
|
||||
pub content: String,
|
||||
pub edited: bool,
|
||||
pub sent_by_self: bool,
|
||||
|
|
@ -58,6 +69,22 @@ pub struct StoredMessage {
|
|||
pub reactions: Vec<StoredReaction>,
|
||||
}
|
||||
|
||||
pub struct NewMessage<'a> {
|
||||
pub relay_signer_id: i64,
|
||||
pub relay_message_id: &'a str,
|
||||
pub authored_at: i64,
|
||||
pub send_time: i64,
|
||||
pub storage_owner: i64,
|
||||
pub external_user: i64,
|
||||
pub sent_by_self: bool,
|
||||
pub content: &'a str,
|
||||
pub height: i64,
|
||||
pub reply_to: Option<i64>,
|
||||
pub origin_iota_received_at: Option<i64>,
|
||||
pub destination_iota_received_at: Option<i64>,
|
||||
pub initial_state: MessageState,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct StoredReaction {
|
||||
pub reaction: String,
|
||||
|
|
@ -149,7 +176,7 @@ fn update_message_content(
|
|||
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i64;
|
||||
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
|
|
@ -351,8 +378,8 @@ pub fn add_reaction(
|
|||
user_id: i64,
|
||||
reaction: &str,
|
||||
) -> Result<(), StorageError> {
|
||||
db::with_db(|conn| {
|
||||
let msg_id: i64 = conn.query_row(
|
||||
db::with_immediate_transaction(|tx| {
|
||||
let msg_id: i64 = tx.query_row(
|
||||
r#"
|
||||
SELECT id FROM messages
|
||||
WHERE storage_owner = ?1 AND external_user = ?2 AND message_time = ?3
|
||||
|
|
@ -362,28 +389,46 @@ pub fn add_reaction(
|
|||
|row| row.get(0),
|
||||
)?;
|
||||
|
||||
let reaction_exists: bool = tx.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM reactions WHERE message_id = ?1 AND reaction = ?2)",
|
||||
params![msg_id, reaction],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
if !reaction_exists {
|
||||
let unique_reactions: i64 = tx.query_row(
|
||||
"SELECT COUNT(DISTINCT reaction) FROM reactions WHERE message_id = ?1",
|
||||
[msg_id],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
if unique_reactions >= MAX_UNIQUE_REACTIONS_PER_MESSAGE as i64 {
|
||||
return Err(StorageError::ReactionLimitReached);
|
||||
}
|
||||
}
|
||||
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i64;
|
||||
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
tx.execute(
|
||||
let inserted = tx.execute(
|
||||
r#"
|
||||
INSERT OR IGNORE INTO reactions (message_id, user_id, reaction, created_at)
|
||||
VALUES (?1, ?2, ?3, ?4)
|
||||
"#,
|
||||
params![msg_id, user_id, reaction, now],
|
||||
)?;
|
||||
sync::record_event(
|
||||
&tx,
|
||||
storage_owner,
|
||||
EntityType::Message,
|
||||
msg_id,
|
||||
Operation::Upsert,
|
||||
)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
if inserted > 0 {
|
||||
sync::record_event(
|
||||
tx,
|
||||
storage_owner,
|
||||
EntityType::Message,
|
||||
msg_id,
|
||||
Operation::Upsert,
|
||||
)?;
|
||||
Ok(())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -422,45 +467,46 @@ pub fn remove_reaction(
|
|||
})
|
||||
}
|
||||
|
||||
pub fn add_message(
|
||||
send_time: u128,
|
||||
storage_owner_is_sender: bool,
|
||||
storage_owner: i64,
|
||||
external_user: i64,
|
||||
message: &str,
|
||||
height: i64,
|
||||
reply_to: Option<i64>,
|
||||
) {
|
||||
let message_time = match i64::try_from(send_time) {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
log!("Failed to store message: send_time out of range for i64 ({send_time})");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = db::with_db(|conn| {
|
||||
pub fn add_message(message: NewMessage<'_>) -> Result<i64, StorageError> {
|
||||
let NewMessage {
|
||||
relay_signer_id,
|
||||
relay_message_id,
|
||||
authored_at,
|
||||
send_time,
|
||||
storage_owner,
|
||||
external_user,
|
||||
sent_by_self,
|
||||
content,
|
||||
height,
|
||||
reply_to,
|
||||
origin_iota_received_at,
|
||||
destination_iota_received_at,
|
||||
initial_state,
|
||||
} = message;
|
||||
let msg_id = db::with_db(|conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
tx.execute(
|
||||
r#"
|
||||
INSERT INTO messages (
|
||||
storage_owner, external_user, message_time, content,
|
||||
sent_by_self, message_state, height, reply_to
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
|
||||
storage_owner, external_user, message_time, content, sent_by_self,
|
||||
message_state, height, reply_to, relay_signer_id, relay_message_id,
|
||||
authored_at, origin_iota_received_at, destination_iota_received_at
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
|
||||
"#,
|
||||
params![
|
||||
storage_owner,
|
||||
external_user,
|
||||
message_time,
|
||||
message,
|
||||
if storage_owner_is_sender {
|
||||
1_i64
|
||||
} else {
|
||||
0_i64
|
||||
},
|
||||
MessageState::Sending.as_str(),
|
||||
send_time,
|
||||
content,
|
||||
i64::from(sent_by_self),
|
||||
initial_state.as_str(),
|
||||
height,
|
||||
reply_to,
|
||||
relay_signer_id,
|
||||
relay_message_id,
|
||||
authored_at,
|
||||
origin_iota_received_at,
|
||||
destination_iota_received_at,
|
||||
],
|
||||
)?;
|
||||
let msg_id = tx.last_insert_rowid();
|
||||
|
|
@ -472,15 +518,122 @@ pub fn add_message(
|
|||
Operation::Upsert,
|
||||
)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}) {
|
||||
log!("Failed to insert message into sqlite: {}", e);
|
||||
return;
|
||||
}
|
||||
Ok(msg_id)
|
||||
})?;
|
||||
|
||||
let mut contact = crate::users::contact::Contact::new(external_user);
|
||||
contact.set_last_message_at(message_time);
|
||||
contact.set_last_message_at(send_time);
|
||||
crate::util::chats_util::mod_user(storage_owner, &contact);
|
||||
Ok(msg_id)
|
||||
}
|
||||
|
||||
pub fn change_message_state_by_relay_id(
|
||||
storage_owner: i64,
|
||||
relay_signer_id: i64,
|
||||
relay_message_id: &str,
|
||||
new_state: MessageState,
|
||||
) -> Result<(), StorageError> {
|
||||
db::with_db(|conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
let Some((msg_id, current)) = tx
|
||||
.query_row(
|
||||
"SELECT id, message_state FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
|
||||
params![storage_owner, relay_signer_id, relay_message_id],
|
||||
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
|
||||
)
|
||||
.optional()?
|
||||
else {
|
||||
return Ok(());
|
||||
};
|
||||
let state = MessageState::from_str(¤t).upgrade(new_state).as_str();
|
||||
tx.execute("UPDATE messages SET message_state = ?1 WHERE id = ?2", params![state, msg_id])?;
|
||||
sync::record_event(&tx, storage_owner, EntityType::Message, msg_id, Operation::Upsert)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn record_message_receipt(
|
||||
storage_owner: i64,
|
||||
target_signer_id: i64,
|
||||
target_message_id: &str,
|
||||
receipt_signer_id: i64,
|
||||
receipt_message_id: &str,
|
||||
receipt_type: MessageState,
|
||||
event_at: i64,
|
||||
recorded_at: i64,
|
||||
) -> Result<(), StorageError> {
|
||||
let receipt_type = match receipt_type {
|
||||
MessageState::Received => "received",
|
||||
MessageState::Read => "read",
|
||||
_ => return Err(StorageError::Other("invalid message receipt state".into())),
|
||||
};
|
||||
db::with_db(|conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
let Some((message_id, external_user)) = tx
|
||||
.query_row(
|
||||
"SELECT id, external_user FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
|
||||
params![storage_owner, target_signer_id, target_message_id],
|
||||
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
|
||||
)
|
||||
.optional()?
|
||||
else {
|
||||
return Err(StorageError::Other("message receipt target was not found".into()));
|
||||
};
|
||||
if external_user != receipt_signer_id {
|
||||
return Err(StorageError::Other("message receipt signer is not the chat partner".into()));
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT OR IGNORE INTO message_receipts (storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
|
||||
params![storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at],
|
||||
)?;
|
||||
let (state_column, recorded_column) = if receipt_type == "read" {
|
||||
("read_at", "read_recorded_at")
|
||||
} else {
|
||||
("client_received_at", "client_received_recorded_at")
|
||||
};
|
||||
let state = MessageState::from_str(
|
||||
&tx.query_row("SELECT message_state FROM messages WHERE id = ?1", [message_id], |row| row.get::<_, String>(0))?,
|
||||
)
|
||||
.upgrade(MessageState::from_str(receipt_type))
|
||||
.as_str()
|
||||
.to_string();
|
||||
tx.execute(
|
||||
&format!("UPDATE messages SET {state_column} = COALESCE({state_column}, ?1), {recorded_column} = COALESCE({recorded_column}, ?2), message_state = ?3 WHERE id = ?4"),
|
||||
params![event_at, recorded_at, state, message_id],
|
||||
)?;
|
||||
sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn record_destination_iota_received(
|
||||
storage_owner: i64,
|
||||
relay_signer_id: i64,
|
||||
relay_message_id: &str,
|
||||
accepted_at: i64,
|
||||
) -> Result<(), StorageError> {
|
||||
db::with_db(|conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
let Some(message_id) = tx
|
||||
.query_row(
|
||||
"SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3",
|
||||
params![storage_owner, relay_signer_id, relay_message_id],
|
||||
|row| row.get::<_, i64>(0),
|
||||
)
|
||||
.optional()?
|
||||
else {
|
||||
return Ok(());
|
||||
};
|
||||
tx.execute(
|
||||
"UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2",
|
||||
params![accepted_at, message_id],
|
||||
)?;
|
||||
sync::record_event(&tx, storage_owner, EntityType::Message, message_id, Operation::Upsert)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn change_message_state(
|
||||
|
|
@ -574,7 +727,15 @@ fn load_reactions(
|
|||
))
|
||||
}) {
|
||||
for row in rows.flatten() {
|
||||
map.entry(row.0).or_default().push(row.1);
|
||||
let reactions = map.entry(row.0).or_default();
|
||||
if reactions
|
||||
.iter()
|
||||
.any(|stored: &StoredReaction| stored.reaction == row.1.reaction)
|
||||
{
|
||||
reactions.push(row.1);
|
||||
} else if reactions.len() < MAX_UNIQUE_REACTIONS_PER_MESSAGE {
|
||||
reactions.push(row.1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -594,7 +755,11 @@ pub fn get_messages(
|
|||
match db::with_db(|conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
r#"
|
||||
SELECT id, message_time, content, sent_by_self, message_state, height, reply_to, edited_count
|
||||
SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
|
||||
origin_iota_received_at, destination_iota_received_at,
|
||||
client_received_at, client_received_recorded_at, read_at,
|
||||
read_recorded_at, content, sent_by_self, message_state, height,
|
||||
reply_to, edited_count
|
||||
FROM messages
|
||||
WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0
|
||||
ORDER BY message_time DESC, id DESC
|
||||
|
|
@ -608,13 +773,22 @@ pub fn get_messages(
|
|||
Ok(StoredMessage {
|
||||
id: row.get(0)?,
|
||||
external_user,
|
||||
message_time: row.get(1)?,
|
||||
content: row.get(2)?,
|
||||
sent_by_self: row.get::<_, i64>(3)? != 0,
|
||||
message_state: row.get(4)?,
|
||||
height: row.get(5).unwrap_or(0),
|
||||
reply_to: row.get(6).ok().flatten(),
|
||||
edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
|
||||
relay_signer_id: row.get(1)?,
|
||||
relay_message_id: row.get(2)?,
|
||||
message_time: row.get(3)?,
|
||||
authored_at: row.get(4)?,
|
||||
origin_iota_received_at: row.get(5)?,
|
||||
destination_iota_received_at: row.get(6)?,
|
||||
client_received_at: row.get(7)?,
|
||||
client_received_recorded_at: row.get(8)?,
|
||||
read_at: row.get(9)?,
|
||||
read_recorded_at: row.get(10)?,
|
||||
content: row.get(11)?,
|
||||
sent_by_self: row.get::<_, i64>(12)? != 0,
|
||||
message_state: row.get(13)?,
|
||||
height: row.get(14).unwrap_or(0),
|
||||
reply_to: row.get(15).ok().flatten(),
|
||||
edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
|
||||
reactions: Vec::new(),
|
||||
})
|
||||
},
|
||||
|
|
@ -652,7 +826,10 @@ pub fn get_message(
|
|||
db::with_db(|conn| {
|
||||
let mut stmt = conn.prepare(
|
||||
r#"
|
||||
SELECT id, message_time, content, sent_by_self, message_state, height,
|
||||
SELECT id, relay_signer_id, relay_message_id, message_time, authored_at,
|
||||
origin_iota_received_at, destination_iota_received_at,
|
||||
client_received_at, client_received_recorded_at, read_at,
|
||||
read_recorded_at, content, sent_by_self, message_state, height,
|
||||
reply_to, edited_count, external_user
|
||||
FROM messages
|
||||
WHERE storage_owner = ?1
|
||||
|
|
@ -666,14 +843,23 @@ pub fn get_message(
|
|||
let rows = stmt.query_map(params![storage_owner, message_time, external_user], |row| {
|
||||
Ok(StoredMessage {
|
||||
id: row.get(0)?,
|
||||
message_time: row.get(1)?,
|
||||
content: row.get(2)?,
|
||||
sent_by_self: row.get::<_, i64>(3)? != 0,
|
||||
message_state: row.get(4)?,
|
||||
height: row.get(5).unwrap_or(0),
|
||||
reply_to: row.get(6).ok().flatten(),
|
||||
edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
|
||||
external_user: row.get(8)?,
|
||||
relay_signer_id: row.get(1)?,
|
||||
relay_message_id: row.get(2)?,
|
||||
message_time: row.get(3)?,
|
||||
authored_at: row.get(4)?,
|
||||
origin_iota_received_at: row.get(5)?,
|
||||
destination_iota_received_at: row.get(6)?,
|
||||
client_received_at: row.get(7)?,
|
||||
client_received_recorded_at: row.get(8)?,
|
||||
read_at: row.get(9)?,
|
||||
read_recorded_at: row.get(10)?,
|
||||
content: row.get(11)?,
|
||||
sent_by_self: row.get::<_, i64>(12)? != 0,
|
||||
message_state: row.get(13)?,
|
||||
height: row.get(14).unwrap_or(0),
|
||||
reply_to: row.get(15).ok().flatten(),
|
||||
edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
|
||||
external_user: row.get(17)?,
|
||||
reactions: Vec::new(),
|
||||
})
|
||||
})?;
|
||||
|
|
@ -700,6 +886,40 @@ pub fn get_message(
|
|||
})
|
||||
}
|
||||
|
||||
pub fn get_message_with_offset(
|
||||
storage_owner: i64,
|
||||
external_user: i64,
|
||||
message_time: i64,
|
||||
) -> Result<Option<(StoredMessage, i64)>, StorageError> {
|
||||
let Some(message) = get_message(storage_owner, message_time, Some(external_user))? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let offset = db::with_db(|conn| {
|
||||
conn.query_row(
|
||||
r#"
|
||||
SELECT COUNT(*)
|
||||
FROM messages
|
||||
WHERE storage_owner = ?1
|
||||
AND external_user = ?2
|
||||
AND deleted_by_external = 0
|
||||
AND (
|
||||
message_time > ?3
|
||||
OR (message_time = ?3 AND id > ?4)
|
||||
)
|
||||
"#,
|
||||
params![
|
||||
storage_owner,
|
||||
external_user,
|
||||
message.message_time,
|
||||
message.id
|
||||
],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.map_err(StorageError::from)
|
||||
})?;
|
||||
Ok(Some((message, offset)))
|
||||
}
|
||||
|
||||
pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec<StoredMessage> {
|
||||
if ids.is_empty() {
|
||||
return Vec::new();
|
||||
|
|
@ -708,19 +928,28 @@ 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, message_time, 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, 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 rows = stmt.query_map([storage_owner], |row| {
|
||||
let external_user: i64 = row.get(8)?;
|
||||
let external_user: i64 = row.get(17)?;
|
||||
Ok(StoredMessage {
|
||||
id: row.get(0)?,
|
||||
external_user,
|
||||
message_time: row.get(1)?,
|
||||
content: row.get(2)?,
|
||||
sent_by_self: row.get::<_, i64>(3)? != 0,
|
||||
message_state: row.get(4)?,
|
||||
height: row.get(5).unwrap_or(0),
|
||||
reply_to: row.get(6).ok().flatten(),
|
||||
edited: row.get::<_, i64>(7).unwrap_or(0) > 0,
|
||||
relay_signer_id: row.get(1)?,
|
||||
relay_message_id: row.get(2)?,
|
||||
message_time: row.get(3)?,
|
||||
authored_at: row.get(4)?,
|
||||
origin_iota_received_at: row.get(5)?,
|
||||
destination_iota_received_at: row.get(6)?,
|
||||
client_received_at: row.get(7)?,
|
||||
client_received_recorded_at: row.get(8)?,
|
||||
read_at: row.get(9)?,
|
||||
read_recorded_at: row.get(10)?,
|
||||
content: row.get(11)?,
|
||||
sent_by_self: row.get::<_, i64>(12)? != 0,
|
||||
message_state: row.get(13)?,
|
||||
height: row.get(14).unwrap_or(0),
|
||||
reply_to: row.get(15).ok().flatten(),
|
||||
edited: row.get::<_, i64>(16).unwrap_or(0) > 0,
|
||||
reactions: Vec::new(),
|
||||
})
|
||||
})?;
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use once_cell::sync::Lazy;
|
||||
use r2d2::ManageConnection;
|
||||
use rusqlite::Connection;
|
||||
use rusqlite::{Connection, Transaction};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
|
@ -55,6 +55,17 @@ where
|
|||
f(&conn)
|
||||
}
|
||||
|
||||
pub fn with_immediate_transaction<T, F>(f: F) -> Result<T, StorageError>
|
||||
where
|
||||
F: FnOnce(&Transaction<'_>) -> Result<T, StorageError>,
|
||||
{
|
||||
let mut conn = POOL.get().map_err(|e| StorageError::Pool(e.to_string()))?;
|
||||
let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
|
||||
let value = f(&tx)?;
|
||||
tx.commit()?;
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
/* Verify the persistent database before the pool is initialized. A corrupt
|
||||
* database is moved aside rather than opened again, preserving material for
|
||||
* operator recovery while allowing the daemon to report the failed storage. */
|
||||
|
|
@ -388,6 +399,71 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> {
|
|||
)?;
|
||||
}
|
||||
|
||||
if current_version < 10 {
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_history
|
||||
ON messages (
|
||||
storage_owner,
|
||||
external_user,
|
||||
deleted_by_external,
|
||||
message_time DESC,
|
||||
id DESC
|
||||
);
|
||||
PRAGMA user_version = 10;
|
||||
"#,
|
||||
)?;
|
||||
}
|
||||
|
||||
if current_version < 11 {
|
||||
for (column, definition) in [
|
||||
("relay_signer_id", "relay_signer_id INTEGER"),
|
||||
("relay_message_id", "relay_message_id TEXT"),
|
||||
("authored_at", "authored_at INTEGER"),
|
||||
("origin_iota_received_at", "origin_iota_received_at INTEGER"),
|
||||
("destination_iota_received_at", "destination_iota_received_at INTEGER"),
|
||||
("client_received_at", "client_received_at INTEGER"),
|
||||
("client_received_recorded_at", "client_received_recorded_at INTEGER"),
|
||||
("read_at", "read_at INTEGER"),
|
||||
("read_recorded_at", "read_recorded_at INTEGER"),
|
||||
] {
|
||||
add_column_if_missing(conn, column, definition)?;
|
||||
}
|
||||
for (column, definition) in [
|
||||
("accepted_at", "accepted_at INTEGER"),
|
||||
("applied_at", "applied_at INTEGER"),
|
||||
("queued_at", "queued_at INTEGER"),
|
||||
("downstream_acked_at", "downstream_acked_at INTEGER"),
|
||||
("rejected_at", "rejected_at INTEGER"),
|
||||
] {
|
||||
add_table_column_if_missing(conn, "relay_inbox", column, definition)?;
|
||||
}
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_relay_identity
|
||||
ON messages (storage_owner, relay_signer_id, relay_message_id)
|
||||
WHERE relay_message_id IS NOT NULL;
|
||||
CREATE TABLE IF NOT EXISTS message_receipts (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
storage_owner INTEGER NOT NULL,
|
||||
target_signer_id INTEGER NOT NULL,
|
||||
target_message_id TEXT NOT NULL,
|
||||
receipt_signer_id INTEGER NOT NULL,
|
||||
receipt_message_id TEXT NOT NULL,
|
||||
receipt_type TEXT NOT NULL CHECK (receipt_type IN ('received', 'read')),
|
||||
event_at INTEGER NOT NULL,
|
||||
recorded_at INTEGER NOT NULL,
|
||||
UNIQUE(receipt_signer_id, receipt_message_id),
|
||||
UNIQUE(storage_owner, target_signer_id, target_message_id,
|
||||
receipt_signer_id, receipt_type)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_message_receipts_target
|
||||
ON message_receipts (storage_owner, target_signer_id, target_message_id);
|
||||
PRAGMA user_version = 11;
|
||||
"#,
|
||||
)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
@ -457,7 +533,7 @@ mod tests {
|
|||
run_migrations_on_connection(&conn)?;
|
||||
|
||||
let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
|
||||
assert_eq!(version, 9);
|
||||
assert_eq!(version, 11);
|
||||
for column in ["height", "reply_to", "edited_count", "deleted_by_external"] {
|
||||
let mut statement =
|
||||
conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?;
|
||||
|
|
@ -470,11 +546,13 @@ mod tests {
|
|||
#[test]
|
||||
fn migrates_version_five_once_and_is_idempotent() -> Result<(), StorageError> {
|
||||
let conn = Connection::open_in_memory()?;
|
||||
conn.execute_batch("PRAGMA user_version = 5;")?;
|
||||
conn.execute_batch(
|
||||
"CREATE TABLE messages (id INTEGER PRIMARY KEY, storage_owner INTEGER NOT NULL, external_user INTEGER NOT NULL, message_time INTEGER NOT NULL, content TEXT NOT NULL, sent_by_self INTEGER NOT NULL, message_state TEXT NOT NULL, height INTEGER NOT NULL DEFAULT 0, reply_to INTEGER, edited_count INTEGER NOT NULL DEFAULT 0, deleted_by_external INTEGER NOT NULL DEFAULT 0); PRAGMA user_version = 5;",
|
||||
)?;
|
||||
run_migrations_on_connection(&conn)?;
|
||||
run_migrations_on_connection(&conn)?;
|
||||
let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?;
|
||||
assert_eq!(version, 9);
|
||||
assert_eq!(version, 11);
|
||||
for table in [
|
||||
"sync_heads",
|
||||
"sync_events",
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ pub fn reserve(
|
|||
signer_id: u64,
|
||||
message_id: &str,
|
||||
created_at: u64,
|
||||
accepted_at: i64,
|
||||
destination_id: u64,
|
||||
frame: &[u8],
|
||||
frame_id: u32,
|
||||
|
|
@ -26,11 +27,12 @@ pub fn reserve(
|
|||
|
||||
db::with_db(|connection| {
|
||||
let inserted = connection.execute(
|
||||
"INSERT OR IGNORE INTO relay_inbox (signer_id, message_id, created_at, destination_id, frame, frame_id, type_map_version, state) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'received')",
|
||||
"INSERT OR IGNORE INTO relay_inbox (signer_id, message_id, created_at, accepted_at, destination_id, frame, frame_id, type_map_version, state) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'received')",
|
||||
params![
|
||||
signer_id,
|
||||
message_id,
|
||||
created_at,
|
||||
accepted_at,
|
||||
destination_id,
|
||||
frame,
|
||||
i64::from(frame_id),
|
||||
|
|
@ -67,41 +69,64 @@ pub fn mark_delivered_for_frame(destination_id: u64, frame_id: u32) -> Result<()
|
|||
let destination_id = i64::try_from(destination_id)
|
||||
.map_err(|_| StorageError::Other("relay destination ID exceeds SQLite range".into()))?;
|
||||
db::with_db(|connection| {
|
||||
let delivered_at = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i64;
|
||||
connection.execute(
|
||||
"UPDATE relay_inbox SET state = 'delivered' WHERE destination_id = ?1 AND frame_id = ?2",
|
||||
params![destination_id, i64::from(frame_id)],
|
||||
"UPDATE relay_inbox SET state = 'delivered', downstream_acked_at = COALESCE(downstream_acked_at, ?3) WHERE destination_id = ?1 AND frame_id = ?2",
|
||||
params![destination_id, i64::from(frame_id), delivered_at],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn mark_state(signer_id: u64, message_id: &str, state: &str) -> Result<(), StorageError> {
|
||||
if !matches!(
|
||||
state,
|
||||
"received" | "applied" | "queued" | "delivered" | "rejected"
|
||||
) {
|
||||
return Err(StorageError::Other("invalid relay inbox state".into()));
|
||||
}
|
||||
fn mark_transition(
|
||||
signer_id: u64,
|
||||
message_id: &str,
|
||||
state: &str,
|
||||
column: &str,
|
||||
) -> Result<(), StorageError> {
|
||||
let signer_id = i64::try_from(signer_id)
|
||||
.map_err(|_| StorageError::Other("relay signer ID exceeds SQLite range".into()))?;
|
||||
let timestamp = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i64;
|
||||
db::with_db(|connection| {
|
||||
connection.execute(
|
||||
"UPDATE relay_inbox SET state = ?3 WHERE signer_id = ?1 AND message_id = ?2",
|
||||
params![signer_id, message_id, state],
|
||||
&format!("UPDATE relay_inbox SET state = ?3, {column} = COALESCE({column}, ?4) WHERE signer_id = ?1 AND message_id = ?2"),
|
||||
params![signer_id, message_id, state, timestamp],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
}
|
||||
|
||||
pub fn prune_completed(before_created_at: i64) -> Result<(), StorageError> {
|
||||
pub fn mark_applied(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
|
||||
mark_transition(signer_id, message_id, "applied", "applied_at")
|
||||
}
|
||||
|
||||
pub fn mark_queued(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
|
||||
mark_transition(signer_id, message_id, "queued", "queued_at")
|
||||
}
|
||||
|
||||
pub fn mark_downstream_acked(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
|
||||
mark_transition(signer_id, message_id, "delivered", "downstream_acked_at")
|
||||
}
|
||||
|
||||
pub fn mark_rejected(signer_id: u64, message_id: &str) -> Result<(), StorageError> {
|
||||
mark_transition(signer_id, message_id, "rejected", "rejected_at")
|
||||
}
|
||||
|
||||
pub fn prune_completed(before_terminal_at: i64) -> Result<(), StorageError> {
|
||||
db::with_db(|connection| {
|
||||
connection.execute(
|
||||
"DELETE FROM relay_inbox WHERE created_at < ?1 AND state IN ('delivered', 'rejected')",
|
||||
params![before_created_at],
|
||||
"DELETE FROM relay_inbox WHERE COALESCE(downstream_acked_at, rejected_at) < ?1 AND state IN ('delivered', 'rejected')",
|
||||
params![before_terminal_at],
|
||||
)?;
|
||||
connection.execute(
|
||||
"DELETE FROM relay_replay WHERE created_at < ?1",
|
||||
params![before_created_at],
|
||||
"DELETE FROM relay_replay WHERE NOT EXISTS (SELECT 1 FROM relay_inbox WHERE relay_inbox.signer_id = relay_replay.signer_id AND relay_inbox.message_id = relay_replay.message_id)",
|
||||
[],
|
||||
)?;
|
||||
Ok(())
|
||||
})
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ use crate::util::db;
|
|||
use rusqlite::{Transaction, params};
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
pub const CACHE_SCHEMA_VERSION: i64 = 1;
|
||||
pub const CACHE_SCHEMA_VERSION: i64 = 2;
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum EntityType {
|
||||
|
|
|
|||
|
|
@ -5,10 +5,6 @@ edition = "2024"
|
|||
|
||||
[dependencies]
|
||||
iota-paths = { path = "../iota-paths" }
|
||||
<<<<<<< HEAD
|
||||
|
||||
=======
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
tokio = { version = "1.50.0", features = ["full"] }
|
||||
sha2 = "0.11.0"
|
||||
hex = "*"
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ tokio = { version = "1.50.0", features = ["full"] }
|
|||
sysinfo = "0.38.0"
|
||||
uuid = { version = "*", features = ["v4"] }
|
||||
walkdir = "2.5.0"
|
||||
zip = "6.0.0"
|
||||
zip = "8.0.0"
|
||||
base64 = "0.22.1"
|
||||
hex = "*"
|
||||
|
||||
|
|
|
|||
|
|
@ -6,17 +6,10 @@ pub fn generate_keyring() -> Keyring {
|
|||
}
|
||||
|
||||
pub fn keyring_to_base64(keyring: &Keyring) -> String {
|
||||
<<<<<<< HEAD
|
||||
keyring
|
||||
.try_to_bytes()
|
||||
.map(|bytes| STANDARD.encode(bytes))
|
||||
.unwrap_or_default()
|
||||
=======
|
||||
let bytes = keyring
|
||||
.try_to_bytes()
|
||||
.expect("keyring fields must fit the wire format");
|
||||
STANDARD.encode(bytes)
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
}
|
||||
|
||||
pub fn keyring_from_base64(s: &str) -> Option<Keyring> {
|
||||
|
|
@ -25,17 +18,10 @@ pub fn keyring_from_base64(s: &str) -> Option<Keyring> {
|
|||
}
|
||||
|
||||
pub fn public_key_bundle_to_base64(bundle: &PublicKeyBundle) -> String {
|
||||
<<<<<<< HEAD
|
||||
bundle
|
||||
.try_as_bytes()
|
||||
.map(|bytes| STANDARD.encode(bytes))
|
||||
.unwrap_or_default()
|
||||
=======
|
||||
let bytes = bundle
|
||||
.try_as_bytes()
|
||||
.expect("public key bundle fields must fit the wire format");
|
||||
STANDARD.encode(bytes)
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
}
|
||||
|
||||
pub fn public_key_bundle_from_base64(s: &str) -> Option<PublicKeyBundle> {
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e
|
||||
Subproject commit f3c5037b0a099ae5389486eec77471bd5addab4a
|
||||
|
|
@ -68,18 +68,6 @@ fn omikron_public_key_path() -> &'static Path {
|
|||
.unwrap_or_else(|| Path::new("omikron.mpkb"))
|
||||
}
|
||||
|
||||
fn save_keyring(keyring: &Keyring, path: &Path) -> Result<(), String> {
|
||||
let temporary = serialization_path(path)?;
|
||||
mtp::files::save_keyring_raw(keyring, &temporary)
|
||||
.map_err(|error| format!("serialize keyring: {error}"))?;
|
||||
let bytes =
|
||||
std::fs::read(&temporary).map_err(|error| format!("read serialized keyring: {error}"));
|
||||
let _ = std::fs::remove_file(&temporary);
|
||||
let bytes = bytes?;
|
||||
iota_util::atomic_file::replace_private(path, &bytes, 3)
|
||||
.map_err(|error| format!("write {}: {error}", path.display()))
|
||||
}
|
||||
|
||||
fn save_omikron_public_key(key: &PublicKeyBundle, path: &Path) -> Result<(), String> {
|
||||
let temporary = serialization_path(path)?;
|
||||
mtp::files::save_public_key_bundle(key, &temporary)
|
||||
|
|
@ -609,46 +597,8 @@ impl OmikronConnection {
|
|||
// Identity (own Keyring, migrated from the legacy base64-in-config format)
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
<<<<<<< HEAD
|
||||
/*
|
||||
* `iota.mk` is now the source of truth for this Iota's identity. A
|
||||
* pre-existing base64 keyring in config.json (from before the MTP auth
|
||||
* migration) is imported once so already-registered Iotas keep their
|
||||
* identity, and mirrored back into config.json for older code paths
|
||||
* that still read it directly.
|
||||
*/
|
||||
async fn load_or_migrate_keyring(&self) -> Keyring {
|
||||
let path = identity_path();
|
||||
if let Ok(kr) = mtp::files::load_keyring_raw(path) {
|
||||
return kr;
|
||||
}
|
||||
|
||||
let legacy = CONFIG.load().keyring.clone();
|
||||
let keyring = legacy
|
||||
.and_then(|b64| keyring_from_base64(&b64))
|
||||
.unwrap_or_else(|| {
|
||||
log!(
|
||||
"WARNING: No existing keyring found. Neither {} nor config.json \
|
||||
contain a keyring; generating a new identity. If you already had \
|
||||
an Iota identity, restore {} from a backup to avoid losing access.",
|
||||
path.display(),
|
||||
path.display()
|
||||
);
|
||||
crypto_helper::generate_keyring()
|
||||
});
|
||||
|
||||
if let Some(parent) = path.parent() {
|
||||
let _ = std::fs::create_dir_all(parent);
|
||||
}
|
||||
if let Err(e) = save_keyring(&keyring, path) {
|
||||
log!("Failed to persist {}: {}", path.display(), e);
|
||||
}
|
||||
|
||||
keyring
|
||||
=======
|
||||
async fn load_or_migrate_keyring(&self, passphrase: &[u8]) -> Result<Keyring, IdentityError> {
|
||||
load_or_migrate_keyring_at(identity_path(), CONFIG.load().keyring.clone(), passphrase)
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
|
|
@ -721,12 +671,8 @@ impl OmikronConnection {
|
|||
};
|
||||
|
||||
let (host, port, public_key) = if let Some(endpoint) = discovered {
|
||||
let discovered_key_bytes = endpoint.public_key.try_as_bytes().map_err(|error| {
|
||||
format!("Failed to serialize discovered Omikron public key: {error}")
|
||||
})?;
|
||||
match &cached_key {
|
||||
Some(cached) => {
|
||||
<<<<<<< HEAD
|
||||
let keys_match =
|
||||
match (cached.try_as_bytes(), endpoint.public_key.try_as_bytes()) {
|
||||
(Ok(cached_bytes), Ok(discovered_bytes)) => {
|
||||
|
|
@ -752,20 +698,6 @@ impl OmikronConnection {
|
|||
} else {
|
||||
(endpoint.host, endpoint.port, cached.clone())
|
||||
}
|
||||
=======
|
||||
let cached_key_bytes = cached.try_as_bytes().map_err(|error| {
|
||||
format!("Failed to serialize cached Omikron public key: {error}")
|
||||
})?;
|
||||
if cached_key_bytes != discovered_key_bytes {
|
||||
log!(
|
||||
"Fetched Omikron public key differs from the cached {} - keeping the \
|
||||
cached key. Delete {} manually if this is an expected key rotation.",
|
||||
OMIKRON_PUBLIC_KEY_PATH,
|
||||
OMIKRON_PUBLIC_KEY_PATH
|
||||
);
|
||||
}
|
||||
(endpoint.host, endpoint.port, cached.clone())
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
}
|
||||
None => {
|
||||
if let Err(e) = save_omikron_public_key(&endpoint.public_key, key_path) {
|
||||
|
|
@ -955,6 +887,44 @@ impl OmikronConnection {
|
|||
}
|
||||
}
|
||||
|
||||
async fn send_relay_success(
|
||||
&self,
|
||||
frame_id: Option<u32>,
|
||||
iota_id: u64,
|
||||
relay_message_id: &str,
|
||||
accepted_at: i64,
|
||||
include_origin_timestamp: bool,
|
||||
) {
|
||||
let Some(frame_id) = frame_id else { return };
|
||||
let response = CommunicationValue::new(CommunicationType::Success)
|
||||
.with_id(frame_id)
|
||||
.add_typed_default(DataType::IotaId, DataValue::UnsignedNumber(iota_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::RelayMessageId,
|
||||
DataValue::Str(relay_message_id.to_string()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::RelayAcceptedAt,
|
||||
DataValue::SignedNumber(accepted_at.into()),
|
||||
);
|
||||
let response = if include_origin_timestamp {
|
||||
response
|
||||
.add_typed_default(
|
||||
DataType::OriginIotaReceivedAt,
|
||||
DataValue::SignedNumber(accepted_at.into()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::DestinationIotaReceivedAt,
|
||||
DataValue::SignedNumber(accepted_at.into()),
|
||||
)
|
||||
} else {
|
||||
response
|
||||
};
|
||||
if let Err(error) = self.send_message(&response).await {
|
||||
log!("Relay response could not be sent: {}", error);
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_relay(self: Arc<Self>, frame: CommunicationValue) {
|
||||
let Some(incoming_frame_id) = frame.id() else {
|
||||
log!("Rejecting Relay without a message id");
|
||||
|
|
@ -998,6 +968,7 @@ impl OmikronConnection {
|
|||
return;
|
||||
}
|
||||
};
|
||||
let accepted_at = now_millis_i64();
|
||||
let signer_is_local = i64::try_from(verified.context.signer_id)
|
||||
.ok()
|
||||
.and_then(iota_storage::users::user_manager::get_user)
|
||||
|
|
@ -1035,6 +1006,7 @@ impl OmikronConnection {
|
|||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
verified.context.created_at,
|
||||
accepted_at,
|
||||
verified.context.final_recipient_id,
|
||||
&frame_bytes,
|
||||
frame_id,
|
||||
|
|
@ -1081,7 +1053,99 @@ impl OmikronConnection {
|
|||
relay_replay::RelayReservation::Existing { .. } => false,
|
||||
};
|
||||
|
||||
/* A shared Iota owns both independent replicas before delivering to its
|
||||
* local recipient. The destination path below writes the recipient copy. */
|
||||
if signer_is_local && recipient_is_local && !already_applied {
|
||||
let content = match open_verified_relay_content(
|
||||
&verified,
|
||||
&[&keyring],
|
||||
verified.context.signer_id,
|
||||
) {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
log!("Relay shared-Iota origin content verification failed: {}", error);
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let owner = match i64::try_from(verified.context.signer_id) {
|
||||
Ok(value) => value,
|
||||
Err(_) => {
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
if let Err(error) = message_handlers::apply_verified_relay_content(
|
||||
&verified.context,
|
||||
&content,
|
||||
accepted_at,
|
||||
owner,
|
||||
true,
|
||||
) {
|
||||
log!("Relay shared-Iota origin application failed: {}", error);
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if let Err(error) = chat_files::record_destination_iota_received(
|
||||
owner,
|
||||
owner,
|
||||
&verified.context.message_id,
|
||||
accepted_at,
|
||||
) {
|
||||
log!("Relay shared-Iota destination timestamp storage failed: {}", error);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if signer_is_local && !recipient_is_local {
|
||||
if !already_applied {
|
||||
let content = match open_verified_relay_content(
|
||||
&verified,
|
||||
&[&keyring],
|
||||
verified.context.signer_id,
|
||||
) {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
log!("Relay origin content verification failed: {}", error);
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
if let Err(error) = message_handlers::apply_verified_relay_content(
|
||||
&verified.context,
|
||||
&content,
|
||||
accepted_at,
|
||||
i64::try_from(verified.context.signer_id).unwrap_or_default(),
|
||||
true,
|
||||
) {
|
||||
log!("Relay origin application failed: {}", error);
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
let router = match self
|
||||
.hosting_iota_for_user(verified.context.final_recipient_id)
|
||||
.await
|
||||
|
|
@ -1124,10 +1188,9 @@ impl OmikronConnection {
|
|||
.await;
|
||||
return;
|
||||
}
|
||||
if let Err(error) = relay_replay::mark_state(
|
||||
if let Err(error) = relay_replay::mark_queued(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"queued",
|
||||
) {
|
||||
log!("Relay origin state update failed: {}", error);
|
||||
}
|
||||
|
|
@ -1136,21 +1199,63 @@ impl OmikronConnection {
|
|||
.await
|
||||
{
|
||||
Ok(response) if response.is_type(CommunicationType::Success) => {
|
||||
let returned_id = response
|
||||
.get_data(DataType::RelayMessageId)
|
||||
.as_str();
|
||||
let destination_accepted_at = response
|
||||
.get_data(DataType::RelayAcceptedAt)
|
||||
.as_number()
|
||||
.and_then(|value| i64::try_from(value).ok());
|
||||
if returned_id != Some(verified.context.message_id.as_str()) {
|
||||
log!("Relay acknowledgement returned a different RelayMessageId");
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
let Some(destination_accepted_at) = destination_accepted_at else {
|
||||
log!("Relay acknowledgement is missing RelayAcceptedAt");
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
};
|
||||
if let (Ok(owner), Ok(signer)) = (
|
||||
i64::try_from(verified.context.signer_id),
|
||||
i64::try_from(verified.context.signer_id),
|
||||
) {
|
||||
if let Err(error) = chat_files::record_destination_iota_received(
|
||||
owner,
|
||||
signer,
|
||||
&verified.context.message_id,
|
||||
destination_accepted_at,
|
||||
) {
|
||||
log!("Relay destination acknowledgement storage failed: {}", error);
|
||||
}
|
||||
}
|
||||
if let Err(error) = relay_queue::acknowledge_iota(router, frame_id) {
|
||||
log!(
|
||||
"Relay origin acknowledgement could not clear the queue: {}",
|
||||
error
|
||||
);
|
||||
}
|
||||
if let Err(error) = relay_replay::mark_state(
|
||||
if let Err(error) = relay_replay::mark_downstream_acked(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"delivered",
|
||||
) {
|
||||
log!("Relay origin delivery state update failed: {}", error);
|
||||
}
|
||||
self.send_relay_response(frame.id(), CommunicationType::Success)
|
||||
.await;
|
||||
let response = response
|
||||
.add_typed_default(
|
||||
DataType::OriginIotaReceivedAt,
|
||||
DataValue::SignedNumber(accepted_at.into()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::DestinationIotaReceivedAt,
|
||||
DataValue::SignedNumber(destination_accepted_at.into()),
|
||||
)
|
||||
.with_id(frame_id);
|
||||
if let Err(error) = self.send_message(&response).await {
|
||||
log!("Relay response could not be sent: {}", error);
|
||||
}
|
||||
}
|
||||
Ok(response) => {
|
||||
log!("Relay origin route returned {}", response.get_type());
|
||||
|
|
@ -1223,10 +1328,9 @@ impl OmikronConnection {
|
|||
queue_error
|
||||
);
|
||||
}
|
||||
let _ = relay_replay::mark_state(
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"rejected",
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
|
|
@ -1234,7 +1338,24 @@ impl OmikronConnection {
|
|||
}
|
||||
};
|
||||
if let Err(error) =
|
||||
message_handlers::apply_verified_relay_content(&verified.context, &content)
|
||||
message_handlers::apply_verified_relay_content(
|
||||
&verified.context,
|
||||
&content,
|
||||
accepted_at,
|
||||
match i64::try_from(destination) {
|
||||
Ok(value) => value,
|
||||
Err(_) => {
|
||||
log!("Relay destination ID exceeds storage range");
|
||||
self.send_relay_response(
|
||||
frame.id(),
|
||||
CommunicationType::ErrorInvalidData,
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
},
|
||||
false,
|
||||
)
|
||||
{
|
||||
log!("Relay application dispatch failed: {}", error);
|
||||
if let Err(queue_error) =
|
||||
|
|
@ -1242,19 +1363,17 @@ impl OmikronConnection {
|
|||
{
|
||||
log!("Relay application queue cleanup failed: {}", queue_error);
|
||||
}
|
||||
let _ = relay_replay::mark_state(
|
||||
let _ = relay_replay::mark_rejected(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"rejected",
|
||||
);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if let Err(error) = relay_replay::mark_state(
|
||||
if let Err(error) = relay_replay::mark_applied(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"applied",
|
||||
) {
|
||||
log!("Relay application state update failed: {}", error);
|
||||
self.send_relay_response(frame.id(), CommunicationType::ErrorInternal)
|
||||
|
|
@ -1263,15 +1382,20 @@ impl OmikronConnection {
|
|||
}
|
||||
}
|
||||
|
||||
if let Err(error) = relay_replay::mark_state(
|
||||
if let Err(error) = relay_replay::mark_queued(
|
||||
verified.context.signer_id,
|
||||
&verified.context.message_id,
|
||||
"queued",
|
||||
) {
|
||||
log!("Relay queue state update failed: {}", error);
|
||||
}
|
||||
self.send_relay_response(frame.id(), CommunicationType::Success)
|
||||
.await;
|
||||
self.send_relay_success(
|
||||
frame.id(),
|
||||
local_iota_id,
|
||||
&verified.context.message_id,
|
||||
accepted_at,
|
||||
signer_is_local,
|
||||
)
|
||||
.await;
|
||||
if let Err(error) = self.send_message(&forwarded).await {
|
||||
log!("Relay delivery to local client failed: {}", error);
|
||||
}
|
||||
|
|
@ -1707,7 +1831,7 @@ impl OmikronConnection {
|
|||
let Some(send_time) = data_i64(cv, DataType::SendTime).filter(|time| *time > 0) else {
|
||||
return;
|
||||
};
|
||||
let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
|
||||
let Some(content) = cv.get_data(DataType::Content).as_str() else {
|
||||
return;
|
||||
};
|
||||
if chat_files::apply_remote_edit(receiver_id, sender_id, send_time, sender_id, content)
|
||||
|
|
@ -1790,7 +1914,7 @@ impl OmikronConnection {
|
|||
.await;
|
||||
return;
|
||||
};
|
||||
let Some(content) = cv.get_data(DataType::AppContent).as_str() else {
|
||||
let Some(content) = cv.get_data(DataType::Content).as_str() else {
|
||||
let _ = self
|
||||
.send_message(&error_response(cv, CommunicationType::ErrorInvalidData))
|
||||
.await;
|
||||
|
|
@ -1800,7 +1924,7 @@ impl OmikronConnection {
|
|||
CommunicationType::MessageEditLive,
|
||||
cv,
|
||||
&mutation,
|
||||
vec![(DataType::AppContent, DataValue::Str(content.to_string()))],
|
||||
vec![(DataType::Content, DataValue::Str(content.to_string()))],
|
||||
);
|
||||
if iota_storage::users::user_manager::get_user(mutation.partner_id).is_some()
|
||||
&& chat_files::apply_remote_edit(
|
||||
|
|
@ -2237,11 +2361,7 @@ impl OmikronConnection {
|
|||
))
|
||||
})?;
|
||||
}
|
||||
<<<<<<< HEAD
|
||||
save_keyring(&keyring, path).map_err(|error| {
|
||||
=======
|
||||
save_protected_keyring_verified(&keyring, path, &identity_secret).map_err(|error| {
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
OmikronError::Internal(format!(
|
||||
"could not save new identity {}: {error}",
|
||||
path.display()
|
||||
|
|
@ -2410,29 +2530,9 @@ impl OmikronClient for OmikronConnection {
|
|||
}
|
||||
|
||||
#[cfg(test)]
|
||||
<<<<<<< HEAD
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn durable_keyring_save_preserves_mtp_format() {
|
||||
let directory = std::env::temp_dir().join(format!("iota-keyring-test-{}", Uuid::new_v4()));
|
||||
std::fs::create_dir_all(&directory).unwrap();
|
||||
let path = directory.join(IOTA_KEYRING_PATH);
|
||||
let keyring = crypto_helper::generate_keyring();
|
||||
|
||||
save_keyring(&keyring, &path).unwrap();
|
||||
|
||||
let loaded = mtp::files::load_keyring_raw(&path).unwrap();
|
||||
assert_eq!(
|
||||
keyring.try_to_bytes().unwrap(),
|
||||
loaded.try_to_bytes().unwrap()
|
||||
);
|
||||
std::fs::remove_dir_all(directory).unwrap();
|
||||
=======
|
||||
mod identity_tests {
|
||||
use super::*;
|
||||
|
||||
fn test_path(name: &str) -> PathBuf {
|
||||
std::env::temp_dir().join(format!(
|
||||
"iota-identity-{name}-{}-{}",
|
||||
|
|
@ -2503,6 +2603,5 @@ mod identity_tests {
|
|||
}
|
||||
|
||||
assert!(jittered_reconnect_delay(Duration::from_secs(600)) <= MAX_RECONNECT_DELAY);
|
||||
>>>>>>> refs/remotes/origin/main
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue