use crate::crypto::account::OlmAccountState;
use crate::crypto::cross_signing::CrossSigningState;
use crate::crypto::device_tracker::{DeviceTracker, StoredDevice};
use crate::crypto::group_sessions::{GroupDecryptError, GroupSessionManager, RoomEventPlaintext};
use crate::crypto::olm_sessions::{DecryptedToDevice, OlmDecryptError, OlmSessionManager};
use crate::error::MessengerError;
use crate::ids::{DeviceId, EventId, RequestId, RoomId, TxnId, UserId};
use crate::outgoing_queue::{Jitter, Lane, OutgoingQueue, PendingRequest, ResponseOutcome};
use crate::persist::{FlushBatch, SealedRecord};
use crate::room::state::RoomState;
use crate::room::timeline::{
interpret_content, ForwardRefusal, ItemContent, SendState, Timeline, KEY_FORWARDED, KEY_FORWARDED_FROM,
};
use crate::room::RoomKind;
use crate::store::sealed::SealedRecordCodec;
use crate::store::{CryptoStore, RecordCodec, StateStore, Store};
use crate::wire::events::{
DirectContent, ForwardedRoomKeyContent, InReplyTo, MegolmEncryptedContent, Membership, RawEvent, ReceiptContent,
RelatesTo, RoomEncryptedContent, RoomKeyContent, RoomKeyRequestAction, RoomKeyRequestBody, RoomKeyRequestContent,
RoomKeyWithheldContent, StrippedStateEvent, TagContent, TagInfo, TextLikeMessageContent, ToDeviceEvent,
TypingContent, Unsigned,
};
use crate::wire::sync::{parse_sync_response, InvitedRoom, JoinedRoom, LeftRoom};
use crate::wire::{percent_decode_segment, HttpResponseDescriptor, OutgoingRequest, OutgoingRequestKind};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use zeroize::Zeroizing;
const SYNC_TIMEOUT_MS: u64 = 30_000;
const LOAD_OLDER_PAGE_SIZE: u32 = 50;
const TYPING_TIMEOUT_MS: u64 = 30_000;
const UNKNOWN_SENDER_RETRY_CAPACITY: usize = 256;
const TYPING_DEBOUNCE_MS: i64 = 4_000;
const ROOM_KEY_SHARE_EVENT_TYPE: &str = "m.room.encrypted";
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Counters {
pub next_request_id: u64,
pub next_txn_id: u64,
}
fn fresh_txn_seed() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_micros() as u64)
.unwrap_or(0)
}
fn restore_counter(persisted_next: u64, max_used_seed: Option<u64>) -> u64 {
match max_used_seed {
Some(seed) => persisted_next.max(seed.saturating_add(1)),
None => persisted_next,
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CoreConfig {
pub user_id: UserId,
pub device_id: DeviceId,
pub server_name: String,
}
#[derive(Default)]
pub struct CoreSecrets {
pub store_seal_key: Option<Zeroizing<[u8; 32]>>,
pub backup_key: Option<Zeroizing<[u8; 32]>>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MessageKind {
Text,
Notice,
Emote,
}
impl MessageKind {
fn msgtype(self) -> &'static str {
match self {
MessageKind::Text => "m.text",
MessageKind::Notice => "m.notice",
MessageKind::Emote => "m.emote",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OutgoingMessage {
pub kind: MessageKind,
pub body: String,
pub reply_to: Option<EventId>,
pub edit_of: Option<EventId>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CreateRoomKind {
Dm {
peer: UserId,
},
Group {
name: String,
invite: Vec<UserId>,
members_can_invite: bool,
},
Channel {
name: String,
topic: Option<String>,
},
}
#[derive(Clone, Debug, PartialEq)]
pub enum MessengerCommand {
SendMessage {
room_id: RoomId,
message: OutgoingMessage,
txn_id: Option<TxnId>,
},
Forward {
from_room: RoomId,
event_id: EventId,
to_room: RoomId,
txn_id: Option<TxnId>,
},
React {
room_id: RoomId,
target: EventId,
key: String,
},
Redact {
room_id: RoomId,
target: EventId,
reason: Option<String>,
},
RetrySend {
room_id: RoomId,
txn_id: TxnId,
},
CreateRoom {
kind: CreateRoomKind,
},
JoinRoom {
room_id: RoomId,
},
LeaveRoom {
room_id: RoomId,
},
Invite {
room_id: RoomId,
user_id: UserId,
},
Kick {
room_id: RoomId,
user_id: UserId,
reason: Option<String>,
},
SetTag {
room_id: RoomId,
tag: String,
order: Option<f64>,
},
RemoveTag {
room_id: RoomId,
tag: String,
},
MarkRead {
room_id: RoomId,
event_id: EventId,
},
SetTyping {
room_id: RoomId,
typing: bool,
},
RetryDecryption {
room_id: RoomId,
event_id: EventId,
},
LoadOlder {
room_id: RoomId,
},
SetAccountData {
event_type: String,
content: serde_json::Value,
},
SetRoomAccountData {
room_id: RoomId,
event_type: String,
content: serde_json::Value,
},
SearchPublicRooms {
term: String,
},
SearchUsers {
term: String,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MessengerEvent {
RoomsChanged,
TimelineChanged {
room_id: RoomId,
},
TypingChanged {
room_id: RoomId,
},
ReceiptsChanged {
room_id: RoomId,
},
UnreadChanged {
room_id: RoomId,
},
DeviceKeyChanged {
user_id: UserId,
device_id: DeviceId,
},
PublicRoomsChanged,
UserSearchChanged,
}
#[derive(Clone)]
struct PendingDecryptItem {
session_id: String,
sender: UserId,
origin_server_ts: i64,
content: MegolmEncryptedContent,
}
#[derive(Deserialize)]
struct RoomMessagesResponseBody {
#[serde(default)]
chunk: Vec<RawEvent>,
#[serde(default)]
end: Option<String>,
}
#[derive(Deserialize)]
struct SendResponseBody {
event_id: EventId,
}
#[derive(Deserialize)]
struct CreateRoomResponseBody {
room_id: RoomId,
}
#[derive(Clone, Debug, PartialEq)]
pub struct PublicRoomsResultEntry {
pub room_id: RoomId,
pub name: Option<String>,
pub topic: Option<String>,
pub num_joined_members: u64,
}
#[derive(Deserialize)]
struct PublicRoomsResponseBody {
#[serde(default)]
chunk: Vec<PublicRoomsChunkEntry>,
}
#[derive(Deserialize)]
struct PublicRoomsChunkEntry {
room_id: RoomId,
#[serde(default)]
name: Option<String>,
#[serde(default)]
topic: Option<String>,
#[serde(default)]
num_joined_members: u64,
}
#[derive(Clone, Debug, PartialEq)]
pub struct UserDirectoryResultEntry {
pub user_id: UserId,
pub display_name: Option<String>,
}
#[derive(Deserialize)]
struct UserDirectorySearchResponseBody {
#[serde(default)]
results: Vec<UserDirectorySearchResultEntry>,
}
#[derive(Deserialize)]
struct UserDirectorySearchResultEntry {
user_id: UserId,
#[serde(default)]
display_name: Option<String>,
}
#[derive(Clone)]
enum SendPayload {
Message(OutgoingMessage),
Event {
event_type: String,
content: serde_json::Value,
},
Reaction {
target: EventId,
key: String,
},
Redaction {
target: EventId,
reason: Option<String>,
},
}
struct RoomEventWire {
event_type: String,
content: serde_json::Value,
relates_to: Option<RelatesTo>,
}
impl SendPayload {
fn room_event(&self) -> Option<RoomEventWire> {
match self {
SendPayload::Message(message) => Some(RoomEventWire {
event_type: "m.room.message".to_string(),
content: message_wire_content(message),
relates_to: message_relates_to(message),
}),
SendPayload::Event { event_type, content } => {
Some(RoomEventWire { event_type: event_type.clone(), content: content.clone(), relates_to: None })
}
SendPayload::Reaction { .. } | SendPayload::Redaction { .. } => None,
}
}
fn is_room_event(&self) -> bool {
matches!(self, SendPayload::Message(_) | SendPayload::Event { .. })
}
}
#[derive(Clone)]
enum WireSend {
Event {
event_type: String,
content: serde_json::Value,
},
Redact {
target: EventId,
reason: Option<String>,
},
}
#[derive(Clone)]
enum SendPhase {
Queued,
AwaitingKeysQuery,
AwaitingKeysClaim {
devices: Vec<StoredDevice>,
},
AwaitingKeyShare {
pending: BTreeSet<RequestId>,
},
AwaitingSend,
Failed {
errcode: String,
},
}
#[derive(Clone)]
struct PendingSend {
room_id: RoomId,
payload: SendPayload,
phase: SendPhase,
wire: Option<WireSend>,
}
type AccountDataKey = (Option<RoomId>, String);
struct AccountDataGuard {
pending: u32,
server_value: Option<serde_json::Value>,
}
fn account_data_key_from_path(path: &str) -> Option<AccountDataKey> {
let segments: Vec<&str> = path.split('/').collect();
let n = segments.len();
if n < 4 || segments[n - 2] != "account_data" {
return None;
}
let event_type = percent_decode_segment(segments[n - 1])?;
if segments[n - 4] == "user" {
return Some((None, event_type));
}
if n >= 6 && segments[n - 4] == "rooms" && segments[n - 6] == "user" {
let room_id = RoomId::parse(percent_decode_segment(segments[n - 3])?).ok()?;
return Some((Some(room_id), event_type));
}
None
}
pub struct MessengerCore<C: RecordCodec> {
config: CoreConfig,
store: Store<C>,
account: OlmAccountState,
xsign: Option<CrossSigningState>,
xsign_inflight: bool,
xsign_attempts: u8,
outgoing: OutgoingQueue,
counters: Counters,
jitter: Box<dyn Jitter>,
rooms: BTreeMap<RoomId, RoomState>,
timelines: BTreeMap<RoomId, Timeline>,
direct_account_data: Option<DirectContent>,
global_account_data: BTreeMap<String, serde_json::Value>,
room_account_data: BTreeMap<RoomId, BTreeMap<String, serde_json::Value>>,
typing: BTreeMap<RoomId, Vec<UserId>>,
receipts: BTreeMap<RoomId, ReceiptContent>,
public_rooms_result: Vec<PublicRoomsResultEntry>,
latest_public_rooms_request: Option<RequestId>,
user_search_result: Vec<UserDirectoryResultEntry>,
latest_user_search_request: Option<RequestId>,
account_data_guard: BTreeMap<AccountDataKey, AccountDataGuard>,
account_data_write_keys: BTreeMap<RequestId, AccountDataKey>,
pending_decrypt: BTreeMap<RoomId, BTreeMap<EventId, PendingDecryptItem>>,
key_requests_sent: BTreeSet<(RoomId, String)>,
history_claims: BTreeMap<RequestId, Vec<StoredDevice>>,
history_claim_tried: BTreeSet<(UserId, DeviceId)>,
withheld_sessions: BTreeSet<(RoomId, String)>,
key_requests_answered: BTreeSet<(UserId, DeviceId, String)>,
unknown_sender_retry: VecDeque<ToDeviceEvent>,
typing_debounce: BTreeMap<RoomId, i64>,
pending_sends: BTreeMap<RoomId, VecDeque<TxnId>>,
send_state: BTreeMap<TxnId, PendingSend>,
send_request_owner: BTreeMap<RequestId, TxnId>,
pending_create_room: BTreeMap<RequestId, CreateRoomKind>,
pending_kinds: BTreeMap<RequestId, OutgoingRequestKind>,
pending_room_messages: BTreeMap<RequestId, RoomId>,
sync_request_id: Option<RequestId>,
sync_caught_up: bool,
pending_flush_batches: VecDeque<FlushBatch>,
events: Vec<MessengerEvent>,
change_counter: u64,
ingest_error: Option<String>,
}
impl<C: RecordCodec> MessengerCore<C> {
pub fn open(
records: impl IntoIterator<Item = SealedRecord>,
codec: C,
config: CoreConfig,
secrets: CoreSecrets,
_now_ms: i64,
jitter: Box<dyn Jitter>,
) -> Result<Self, MessengerError> {
let _ = secrets;
let mut store = Store::load(records, codec, config.device_id.clone())?;
let mut outgoing = OutgoingQueue::load(&store)?;
let account = OlmAccountState::load_or_create(&mut store)?;
let mut rooms = BTreeMap::new();
let mut timelines = BTreeMap::new();
for room_id in store.room_ids()? {
if let Some(bytes) = store.room_state(&room_id)? {
let state: RoomState = serde_json::from_slice(bytes)
.map_err(|source| MessengerError::Crypto(format!("decode room state for {room_id}: {source}")))?;
rooms.insert(room_id.clone(), state);
timelines.insert(room_id, Timeline::new());
}
}
let pending_requests = store.pending_requests()?;
let max_pending_request_seed = pending_requests.iter().filter_map(|req| req.request.id.as_seed()).max();
outgoing.discard_kind(&mut store, OutgoingRequestKind::Sync)?;
let persisted_counters = store.counters()?.unwrap_or_default();
let mut counters = Counters {
next_request_id: restore_counter(persisted_counters.next_request_id, max_pending_request_seed),
next_txn_id: restore_counter(persisted_counters.next_txn_id, None),
};
if counters != persisted_counters {
store.save_counters(counters)?;
}
counters.next_txn_id = counters.next_txn_id.max(fresh_txn_seed());
let mut core = Self {
config,
store,
account,
outgoing,
counters,
jitter,
rooms,
timelines,
direct_account_data: None,
global_account_data: BTreeMap::new(),
room_account_data: BTreeMap::new(),
typing: BTreeMap::new(),
receipts: BTreeMap::new(),
public_rooms_result: Vec::new(),
latest_public_rooms_request: None,
user_search_result: Vec::new(),
latest_user_search_request: None,
account_data_guard: BTreeMap::new(),
account_data_write_keys: BTreeMap::new(),
pending_decrypt: BTreeMap::new(),
key_requests_sent: BTreeSet::new(),
history_claims: BTreeMap::new(),
history_claim_tried: BTreeSet::new(),
withheld_sessions: BTreeSet::new(),
key_requests_answered: BTreeSet::new(),
unknown_sender_retry: VecDeque::new(),
typing_debounce: BTreeMap::new(),
pending_sends: BTreeMap::new(),
send_state: BTreeMap::new(),
send_request_owner: BTreeMap::new(),
pending_create_room: BTreeMap::new(),
pending_kinds: BTreeMap::new(),
pending_room_messages: BTreeMap::new(),
sync_request_id: None,
sync_caught_up: false,
pending_flush_batches: VecDeque::new(),
events: Vec::new(),
change_counter: 0,
ingest_error: None,
xsign: None,
xsign_inflight: false,
xsign_attempts: 0,
};
core.rebuild_account_data_guard(&pending_requests);
Ok(core)
}
fn rebuild_account_data_guard(&mut self, pending_requests: &[PendingRequest]) {
let mut writes: Vec<&PendingRequest> = pending_requests
.iter()
.filter(|record| matches!(record.request.kind, OutgoingRequestKind::AccountData | OutgoingRequestKind::RoomAccountData))
.collect();
writes.sort_by(|a, b| a.request.id.cmp(&b.request.id));
for record in writes {
let Some(key) = account_data_key_from_path(&record.request.path) else { continue };
let Some(body) = record.request.body.clone() else { continue };
let guard = self.account_data_guard.entry(key.clone()).or_insert(AccountDataGuard { pending: 0, server_value: None });
guard.pending += 1;
self.store_account_data_cache(&key, Some(body));
self.account_data_write_keys.insert(record.request.id.clone(), key);
}
}
fn next_request_id(&mut self) -> Result<RequestId, MessengerError> {
let seed = self.counters.next_request_id;
self.counters.next_request_id = seed.saturating_add(1);
self.store.save_counters(self.counters)?;
Ok(RequestId::next(seed))
}
fn next_txn_id(&mut self) -> Result<TxnId, MessengerError> {
let seed = self.counters.next_txn_id;
self.counters.next_txn_id = seed.saturating_add(1);
self.store.save_counters(self.counters)?;
Ok(TxnId::new(seed))
}
fn enqueue_request(&mut self, request: OutgoingRequest, lane: Lane) -> Result<(), MessengerError> {
let id = request.id.clone();
let kind = request.kind;
if let Some(batch) = self.outgoing.enqueue(&mut self.store, request, lane)? {
self.pending_flush_batches.push_back(batch);
}
self.pending_kinds.insert(id, kind);
Ok(())
}
fn emit(&mut self, event: MessengerEvent) {
self.change_counter = self.change_counter.wrapping_add(1);
self.events.push(event);
}
fn drain_events(&mut self) -> Vec<MessengerEvent> {
std::mem::take(&mut self.events)
}
fn start_new_send(
&mut self,
room_id: RoomId,
payload: SendPayload,
caller_txn_id: Option<TxnId>,
now_ms: i64,
) -> Result<(), MessengerError> {
let txn_id = match caller_txn_id {
Some(txn_id) => txn_id,
None => self.next_txn_id()?,
};
if let Some(wire) = payload.room_event() {
if self.typing_debounce.remove(&room_id).is_some() {
if let Ok(request_id) = self.next_request_id() {
let request = OutgoingRequest::typing(request_id, &room_id, &self.config.user_id, false, None);
let _ = self.enqueue_request(request, Lane::Other);
}
}
let content = match &payload {
SendPayload::Message(message) => message_local_echo_content(message),
_ => interpret_content(&wire.event_type, None, &wire.content),
};
let timeline = self.timelines.entry(room_id.clone()).or_default();
timeline.push_local_echo(
txn_id.clone(),
self.config.user_id.clone(),
now_ms,
&wire.event_type,
content,
wire.content,
);
timeline.mark_send_state(&txn_id, SendState::Sending);
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
self.send_state.insert(txn_id.clone(), PendingSend { room_id: room_id.clone(), payload, phase: SendPhase::Queued, wire: None });
let queue = self.pending_sends.entry(room_id).or_default();
let was_empty = queue.is_empty();
queue.push_back(txn_id.clone());
if was_empty {
self.start_send_pipeline(&txn_id, now_ms)?;
}
Ok(())
}
fn retry_send(&mut self, room_id: &RoomId, txn_id: &TxnId, now_ms: i64) -> Result<(), MessengerError> {
let Some(pending) = self.send_state.get(txn_id) else { return Ok(()) };
if pending.room_id != *room_id || !matches!(pending.phase, SendPhase::Failed { .. }) {
return Ok(());
}
self.set_send_phase(txn_id, SendPhase::Queued);
let queue = self.pending_sends.entry(room_id.clone()).or_default();
let was_empty = queue.is_empty();
queue.push_back(txn_id.clone());
if was_empty {
self.start_send_pipeline(txn_id, now_ms)?;
}
Ok(())
}
fn set_send_phase(&mut self, txn_id: &TxnId, phase: SendPhase) {
if let Some(entry) = self.send_state.get_mut(txn_id) {
entry.phase = phase;
}
}
fn start_send_pipeline(&mut self, txn_id: &TxnId, now_ms: i64) -> Result<(), MessengerError> {
let Some(pending) = self.send_state.get(txn_id) else { return Ok(()) };
let room_id = pending.room_id.clone();
if pending.wire.is_some() {
return self.begin_awaiting_send(txn_id, &room_id);
}
let is_room_event = pending.payload.is_room_event();
let encrypted_room = self.rooms.get(&room_id).is_some_and(|room| room.encryption.is_some());
if is_room_event && encrypted_room {
self.begin_key_pipeline(txn_id, &room_id, now_ms)
} else {
self.build_plaintext_wire_body(txn_id);
self.begin_awaiting_send(txn_id, &room_id)
}
}
fn encrypted_room_member_ids(&self, room_id: &RoomId) -> BTreeSet<UserId> {
let mut ids: BTreeSet<UserId> = self
.rooms
.get(room_id)
.map(|state| {
state
.members
.iter()
.filter(|(_, member)| matches!(member.membership, Membership::Join | Membership::Invite))
.map(|(user_id, _)| user_id.clone())
.collect()
})
.unwrap_or_default();
ids.insert(self.config.user_id.clone());
ids
}
fn mark_outdated_or_untracked(&mut self, member_ids: &BTreeSet<UserId>) -> Result<Vec<UserId>, MessengerError> {
let tracked: BTreeMap<UserId, bool> = self.store.tracked_users()?.into_iter().collect();
let mut outdated = Vec::new();
let mut newly_tracked = Vec::new();
for user_id in member_ids {
match tracked.get(user_id) {
Some(true) => outdated.push(user_id.clone()),
Some(false) => {}
None => newly_tracked.push(user_id.clone()),
}
}
if !newly_tracked.is_empty() {
DeviceTracker::on_device_lists(&mut self.store, &newly_tracked, &[])?;
outdated.extend(newly_tracked);
}
Ok(outdated)
}
fn known_devices_for(&self, member_ids: &BTreeSet<UserId>) -> Result<Vec<StoredDevice>, MessengerError> {
let mut all = Vec::new();
for user_id in member_ids {
all.extend(DeviceTracker::devices_for_user(&self.store, user_id)?);
}
Ok(all)
}
fn candidate_recipient_devices(&self, member_ids: &BTreeSet<UserId>, known_devices: &[StoredDevice]) -> Vec<StoredDevice> {
known_devices
.iter()
.filter(|device| member_ids.contains(&device.user_id) && !device.blocked)
.filter(|device| !(device.user_id == self.config.user_id && device.device_id == self.config.device_id))
.cloned()
.collect()
}
fn begin_key_pipeline(&mut self, txn_id: &TxnId, room_id: &RoomId, now_ms: i64) -> Result<(), MessengerError> {
let members = self.encrypted_room_member_ids(room_id);
let outdated = self.mark_outdated_or_untracked(&members)?;
if !outdated.is_empty() {
let request_id = self.next_request_id()?;
if let Some(request) = DeviceTracker::keys_query_request(&self.store, request_id.clone())? {
self.send_request_owner.insert(request_id.clone(), txn_id.clone());
self.enqueue_request(request, Lane::Other)?;
self.set_send_phase(txn_id, SendPhase::AwaitingKeysQuery);
return Ok(());
}
}
self.begin_keys_claim_or_share(txn_id, room_id, now_ms)
}
fn begin_keys_claim_or_share(&mut self, txn_id: &TxnId, room_id: &RoomId, now_ms: i64) -> Result<(), MessengerError> {
let members = self.encrypted_room_member_ids(room_id);
let known_devices = self.known_devices_for(&members)?;
let candidates = self.candidate_recipient_devices(&members, &known_devices);
let missing: Vec<StoredDevice> =
OlmSessionManager::sessions_missing_for(&self.store, &candidates)?.into_iter().cloned().collect();
if !missing.is_empty() {
let request_id = self.next_request_id()?;
let refs: Vec<&StoredDevice> = missing.iter().collect();
if let Some(request) = OlmSessionManager::keys_claim_request(request_id.clone(), &refs) {
self.send_request_owner.insert(request_id.clone(), txn_id.clone());
self.enqueue_request(request, Lane::Other)?;
self.set_send_phase(txn_id, SendPhase::AwaitingKeysClaim { devices: missing });
return Ok(());
}
}
self.begin_share_or_encrypt(txn_id, room_id, now_ms)
}
fn begin_share_or_encrypt(&mut self, txn_id: &TxnId, room_id: &RoomId, now_ms: i64) -> Result<(), MessengerError> {
let Some(encryption) = self.rooms.get(room_id).and_then(|room| room.encryption.clone()) else {
self.build_plaintext_wire_body(txn_id);
return self.begin_awaiting_send(txn_id, room_id);
};
let members = self.encrypted_room_member_ids(room_id);
let known_devices = self.known_devices_for(&members)?;
let update = GroupSessionManager::ensure_outbound_session(
&mut self.store,
room_id,
&self.config.user_id,
&self.config.device_id,
&encryption,
&members,
&known_devices,
now_ms,
)?;
let own_keys = self.account.identity_keys();
GroupSessionManager::adopt_own_outbound_session(
&mut self.store,
room_id,
&self.config.user_id,
own_keys.curve25519,
own_keys.ed25519,
&update,
)?;
if let Some(room_key_content) = update.room_key_content.clone() {
let missing_ids: BTreeSet<(UserId, DeviceId)> = OlmSessionManager::sessions_missing_for(&self.store, &update.new_recipients)?
.into_iter()
.map(|device| (device.user_id.clone(), device.device_id.clone()))
.collect();
let sessioned: Vec<StoredDevice> = update
.new_recipients
.into_iter()
.filter(|device| !missing_ids.contains(&(device.user_id.clone(), device.device_id.clone())))
.collect();
let mut pending_ids = BTreeSet::new();
for chunk in GroupSessionManager::chunk_recipients_for_send_to_device(&sessioned) {
let body = GroupSessionManager::build_room_key_send_to_device_body(
&mut self.store,
&self.account,
&self.config.user_id,
&self.config.device_id,
chunk,
&room_key_content,
)?;
let request_id = self.next_request_id()?;
let share_txn = self.next_txn_id()?;
let request = OutgoingRequest::send_to_device(request_id.clone(), ROOM_KEY_SHARE_EVENT_TYPE, &share_txn, body);
self.send_request_owner.insert(request_id.clone(), txn_id.clone());
self.enqueue_request(request, Lane::ToDevice)?;
pending_ids.insert(request_id);
}
if !pending_ids.is_empty() {
self.set_send_phase(txn_id, SendPhase::AwaitingKeyShare { pending: pending_ids });
return Ok(());
}
}
self.finish_encrypt_and_send(txn_id, room_id)?;
self.begin_awaiting_send(txn_id, room_id)
}
fn finish_encrypt_and_send(&mut self, txn_id: &TxnId, room_id: &RoomId) -> Result<(), MessengerError> {
let Some(wire) = self.send_state.get(txn_id).and_then(|pending| pending.payload.room_event()) else {
return Ok(());
};
let sender_curve = self.account.identity_keys().curve25519.to_base64();
let encrypted = GroupSessionManager::encrypt_event(
&mut self.store,
room_id,
&sender_curve,
&self.config.device_id,
&wire.event_type,
wire.content,
wire.relates_to,
)?;
let wire_content = serde_json::to_value(RoomEncryptedContent::Megolm(encrypted))?;
if let Some(entry) = self.send_state.get_mut(txn_id) {
entry.wire = Some(WireSend::Event { event_type: "m.room.encrypted".to_string(), content: wire_content });
}
Ok(())
}
fn build_plaintext_wire_body(&mut self, txn_id: &TxnId) {
let Some(pending) = self.send_state.get(txn_id) else { return };
let wire = match &pending.payload {
SendPayload::Message(_) | SendPayload::Event { .. } => match pending.payload.room_event() {
Some(room_event) => WireSend::Event { event_type: room_event.event_type, content: room_event.content },
None => return,
},
SendPayload::Reaction { target, key } => WireSend::Event {
event_type: "m.reaction".to_string(),
content: serde_json::json!({
"m.relates_to": { "rel_type": "m.annotation", "event_id": target.as_str(), "key": key }
}),
},
SendPayload::Redaction { target, reason } => WireSend::Redact { target: target.clone(), reason: reason.clone() },
};
if let Some(entry) = self.send_state.get_mut(txn_id) {
entry.wire = Some(wire);
}
}
fn begin_awaiting_send(&mut self, txn_id: &TxnId, room_id: &RoomId) -> Result<(), MessengerError> {
let Some(wire) = self.send_state.get(txn_id).and_then(|pending| pending.wire.clone()) else { return Ok(()) };
let request_id = self.next_request_id()?;
let request = match &wire {
WireSend::Event { event_type, content } => {
OutgoingRequest::room_send(request_id.clone(), room_id, event_type, txn_id, content.clone())
}
WireSend::Redact { target, reason } => {
let body = match reason {
Some(reason) => serde_json::json!({ "reason": reason }),
None => serde_json::json!({}),
};
OutgoingRequest::room_redact(request_id.clone(), room_id, target, txn_id, body)
}
};
self.send_request_owner.insert(request_id, txn_id.clone());
self.enqueue_request(request, Lane::Room(room_id.clone()))?;
self.set_send_phase(txn_id, SendPhase::AwaitingSend);
Ok(())
}
fn advance_send_on_success(
&mut self,
txn_id: &TxnId,
request_id: &RequestId,
kind: Option<OutgoingRequestKind>,
response: &HttpResponseDescriptor,
now_ms: i64,
) {
match kind {
Some(OutgoingRequestKind::KeysQuery) => {
let Some(room_id) = self.send_state.get(txn_id).map(|pending| pending.room_id.clone()) else { return };
let _ = self.begin_keys_claim_or_share(txn_id, &room_id, now_ms);
}
Some(OutgoingRequestKind::KeysClaim) => {
let Some(pending) = self.send_state.get(txn_id) else { return };
let room_id = pending.room_id.clone();
if let SendPhase::AwaitingKeysClaim { devices, .. } = &pending.phase {
let devices = devices.clone();
let refs: Vec<&StoredDevice> = devices.iter().collect();
let _ = OlmSessionManager::on_keys_claim_response(&mut self.store, &self.account, &refs, &response.body);
}
let _ = self.begin_share_or_encrypt(txn_id, &room_id, now_ms);
}
Some(OutgoingRequestKind::SendToDevice) => {
let mut remaining = None;
if let Some(entry) = self.send_state.get_mut(txn_id) {
if let SendPhase::AwaitingKeyShare { pending } = &mut entry.phase {
pending.remove(request_id);
remaining = Some(pending.len());
}
}
if remaining == Some(0) {
let Some(room_id) = self.send_state.get(txn_id).map(|pending| pending.room_id.clone()) else { return };
let _ = self.finish_encrypt_and_send(txn_id, &room_id);
let _ = self.begin_awaiting_send(txn_id, &room_id);
}
}
Some(OutgoingRequestKind::RoomSend) | Some(OutgoingRequestKind::RoomRedact) => {
self.complete_send_success(txn_id, response, now_ms);
}
_ => {}
}
}
fn complete_send_success(&mut self, txn_id: &TxnId, response: &HttpResponseDescriptor, now_ms: i64) {
let Some(pending) = self.send_state.get(txn_id) else { return };
let room_id = pending.room_id.clone();
if pending.payload.is_room_event() {
if let Ok(body) = serde_json::from_slice::<SendResponseBody>(&response.body) {
if let Some(timeline) = self.timelines.get_mut(&room_id) {
timeline.set_echo_event_id(txn_id, body.event_id);
}
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
}
self.send_state.remove(txn_id);
self.finish_active_send(&room_id, txn_id, now_ms);
}
fn fail_send(&mut self, txn_id: &TxnId, errcode: String, now_ms: i64) {
let Some(pending) = self.send_state.get(txn_id) else { return };
let room_id = pending.room_id.clone();
if pending.payload.is_room_event() {
if let Some(timeline) = self.timelines.get_mut(&room_id) {
timeline.mark_send_state(txn_id, SendState::Failed { reason: errcode.clone() });
}
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
self.set_send_phase(txn_id, SendPhase::Failed { errcode });
self.finish_active_send(&room_id, txn_id, now_ms);
}
fn finish_active_send(&mut self, room_id: &RoomId, txn_id: &TxnId, now_ms: i64) {
if let Some(queue) = self.pending_sends.get_mut(room_id) {
if queue.front() == Some(txn_id) {
queue.pop_front();
} else {
queue.retain(|id| id != txn_id);
}
}
let next = self.pending_sends.get(room_id).and_then(|queue| queue.front().cloned());
if let Some(next_txn) = next {
let _ = self.start_send_pipeline(&next_txn, now_ms);
}
}
pub fn releasable_requests(&mut self, now_ms: i64) -> Vec<OutgoingRequest> {
self.ensure_sync_enqueued();
self.outgoing.releasable(&self.store, now_ms)
}
fn ensure_sync_enqueued(&mut self) {
if self.sync_request_id.is_some() {
return;
}
let Ok(request_id) = self.next_request_id() else { return };
let since = self.store.sync_token().ok().flatten().map(str::to_string);
let timeout_ms = if since.is_some() && self.sync_caught_up { SYNC_TIMEOUT_MS } else { 0 };
let request = OutgoingRequest::sync(request_id.clone(), since.as_deref(), Some(timeout_ms));
if self.enqueue_request(request, Lane::Sync).is_ok() {
self.sync_request_id = Some(request_id);
}
}
pub fn on_response(&mut self, request_id: RequestId, resp: HttpResponseDescriptor, now_ms: i64) -> Vec<MessengerEvent> {
let kind = self.pending_kinds.get(&request_id).copied();
let room_messages_room = self.pending_room_messages.get(&request_id).cloned();
let create_room_kind = self.pending_create_room.get(&request_id).cloned();
let send_owner = self.send_request_owner.get(&request_id).cloned();
let outcome = self.outgoing.on_response(&mut self.store, &request_id, resp, now_ms, self.jitter.as_mut());
let (is_terminal, terminal_response, failed_errcode) = match outcome {
Ok(Some(ResponseOutcome::Done(response))) => (true, Some(response), None),
Ok(Some(ResponseOutcome::Failed { errcode, .. })) => (true, None, Some(errcode)),
_ => (false, None, None),
};
if is_terminal {
self.pending_kinds.remove(&request_id);
if matches!(kind, Some(OutgoingRequestKind::Sync)) {
self.sync_request_id = None;
if terminal_response.is_some() {
self.sync_caught_up = true;
}
}
if room_messages_room.is_some() {
self.pending_room_messages.remove(&request_id);
}
self.pending_create_room.remove(&request_id);
self.send_request_owner.remove(&request_id);
if let Some(key) = self.account_data_write_keys.remove(&request_id) {
self.settle_account_data_write(&key, terminal_response.is_some());
}
}
if let Some(response) = &terminal_response {
if let Err(error) = self.handle_terminal_success(&request_id, kind, room_messages_room, create_room_kind, response) {
self.ingest_error = Some(match kind {
Some(kind) => format!("{kind:?} response not ingested: {error}"),
None => format!("response not ingested: {error}"),
});
}
}
if is_terminal
&& terminal_response.is_none()
&& matches!(kind, Some(OutgoingRequestKind::SigningKeysUpload) | Some(OutgoingRequestKind::SignaturesUpload))
{
self.xsign_inflight = false;
}
if let Some(txn_id) = send_owner {
if let Some(response) = &terminal_response {
self.advance_send_on_success(&txn_id, &request_id, kind, response, now_ms);
} else if let Some(errcode) = failed_errcode.clone() {
self.fail_send(&txn_id, errcode, now_ms);
}
}
if matches!(kind, Some(OutgoingRequestKind::KeysUpload)) {
if let Some(errcode) = failed_errcode {
self.ingest_error = Some(format!("keys/upload failed: {errcode}"));
}
}
self.drain_events()
}
pub fn on_transport_error(&mut self, request_id: &RequestId, now_ms: i64) {
let _ = self.outgoing.on_transport_error(request_id, now_ms, self.jitter.as_mut());
}
fn handle_terminal_success(
&mut self,
request_id: &RequestId,
kind: Option<OutgoingRequestKind>,
room_messages_room: Option<RoomId>,
create_room_kind: Option<CreateRoomKind>,
response: &HttpResponseDescriptor,
) -> Result<(), MessengerError> {
match kind {
Some(OutgoingRequestKind::Sync) => {
let parsed = parse_sync_response(&response.body)?;
self.ingest_sync(&parsed)?;
}
Some(OutgoingRequestKind::KeysUpload) => {
self.account.on_keys_upload_response(&mut self.store)?;
}
Some(OutgoingRequestKind::KeysClaim) if self.history_claims.contains_key(request_id) => {
if let Some(devices) = self.history_claims.remove(request_id) {
let refs: Vec<&StoredDevice> = devices.iter().collect();
let _ = OlmSessionManager::on_keys_claim_response(&mut self.store, &self.account, &refs, &response.body);
}
self.retry_missing_room_keys();
}
Some(OutgoingRequestKind::SigningKeysUpload) => {
self.xsign_inflight = false;
if let Some(state) = self.xsign.as_mut() {
state.uploaded = true;
let snapshot = state.clone();
snapshot.save(&mut self.store)?;
}
self.maybe_enqueue_cross_signing()?;
}
Some(OutgoingRequestKind::SignaturesUpload) => {
self.xsign_inflight = false;
let failures_empty = serde_json::from_slice::<serde_json::Value>(&response.body)
.ok()
.and_then(|v| v.get("failures").and_then(|f| f.as_object()).map(|f| f.is_empty()))
.unwrap_or(true);
if failures_empty {
if let Some(state) = self.xsign.as_mut() {
state.device_signed = true;
let snapshot = state.clone();
snapshot.save(&mut self.store)?;
}
}
}
Some(OutgoingRequestKind::KeysQuery) => {
let outcome = DeviceTracker::on_keys_query_response(&mut self.store, &response.body)?;
if !outcome.key_changes.is_empty() {
let room_ids: Vec<RoomId> = self.rooms.keys().cloned().collect();
for change in &outcome.key_changes {
let _ = GroupSessionManager::forget_shared_device(
&mut self.store,
&room_ids,
&change.user_id,
&change.device_id,
);
self.emit(MessengerEvent::DeviceKeyChanged {
user_id: change.user_id.clone(),
device_id: change.device_id.clone(),
});
}
}
self.retry_unknown_sender_queue()?;
}
Some(OutgoingRequestKind::RoomMessages) => {
if let Some(room_id) = room_messages_room {
self.ingest_room_messages_response(&room_id, &response.body)?;
}
}
Some(OutgoingRequestKind::CreateRoom) => {
if let Some(kind) = create_room_kind {
self.on_create_room_response(kind, response)?;
}
}
Some(OutgoingRequestKind::PublicRooms) if self.latest_public_rooms_request.as_ref() == Some(request_id) => {
let body: PublicRoomsResponseBody = serde_json::from_slice(&response.body)?;
self.public_rooms_result = body
.chunk
.into_iter()
.map(|entry| PublicRoomsResultEntry {
room_id: entry.room_id,
name: entry.name,
topic: entry.topic,
num_joined_members: entry.num_joined_members,
})
.collect();
self.emit(MessengerEvent::PublicRoomsChanged);
}
Some(OutgoingRequestKind::UserDirectorySearch) if self.latest_user_search_request.as_ref() == Some(request_id) => {
let body: UserDirectorySearchResponseBody = serde_json::from_slice(&response.body)?;
self.user_search_result = body
.results
.into_iter()
.map(|entry| UserDirectoryResultEntry { user_id: entry.user_id, display_name: entry.display_name })
.collect();
self.emit(MessengerEvent::UserSearchChanged);
}
_ => {}
}
Ok(())
}
fn on_create_room_response(&mut self, kind: CreateRoomKind, response: &HttpResponseDescriptor) -> Result<(), MessengerError> {
let CreateRoomKind::Dm { peer } = kind else { return Ok(()) };
let body: CreateRoomResponseBody = serde_json::from_slice(&response.body)?;
self.merge_direct_account_data(body.room_id, peer)
}
fn merge_direct_account_data(&mut self, room_id: RoomId, peer: UserId) -> Result<(), MessengerError> {
let mut direct = self.direct_account_data.clone().unwrap_or_default();
let rooms_for_peer = direct.0.entry(peer).or_default();
if !rooms_for_peer.contains(&room_id) {
rooms_for_peer.push(room_id);
}
let value = serde_json::to_value(&direct)?;
self.write_account_data(None, "m.direct".to_string(), value)
}
fn cached_account_data(&self, key: &AccountDataKey) -> Option<&serde_json::Value> {
match &key.0 {
Some(room_id) => self.room_account_data(room_id, &key.1),
None => self.global_account_data(&key.1),
}
}
fn store_account_data_cache(&mut self, key: &AccountDataKey, value: Option<serde_json::Value>) {
if key.0.is_none() && key.1 == "m.direct" {
self.direct_account_data = value.as_ref().and_then(|v| serde_json::from_value::<DirectContent>(v.clone()).ok());
}
match (&key.0, value) {
(None, Some(v)) => {
self.global_account_data.insert(key.1.clone(), v);
}
(None, None) => {
self.global_account_data.remove(&key.1);
}
(Some(room_id), Some(v)) => {
self.room_account_data.entry(room_id.clone()).or_default().insert(key.1.clone(), v);
}
(Some(room_id), None) => {
if let Some(by_type) = self.room_account_data.get_mut(room_id) {
by_type.remove(&key.1);
}
}
}
}
fn write_account_data(&mut self, room_id: Option<RoomId>, event_type: String, value: serde_json::Value) -> Result<(), MessengerError> {
let request_id = self.next_request_id()?;
let request = match &room_id {
Some(room_id) => OutgoingRequest::room_account_data(request_id.clone(), &self.config.user_id, room_id, &event_type, value.clone()),
None => OutgoingRequest::account_data(request_id.clone(), &self.config.user_id, &event_type, value.clone()),
};
let key: AccountDataKey = (room_id, event_type);
let known = self.cached_account_data(&key).cloned();
let guard = self.account_data_guard.entry(key.clone()).or_insert(AccountDataGuard { pending: 0, server_value: known });
guard.pending += 1;
self.store_account_data_cache(&key, Some(value));
match self.enqueue_request(request, Lane::AccountData) {
Ok(()) => {
self.account_data_write_keys.insert(request_id, key);
Ok(())
}
Err(error) => {
self.settle_account_data_write(&key, false);
Err(error)
}
}
}
fn settle_account_data_write(&mut self, key: &AccountDataKey, succeeded: bool) {
let Some(guard) = self.account_data_guard.get_mut(key) else { return };
guard.pending = guard.pending.saturating_sub(1);
if guard.pending > 0 {
return;
}
let Some(guard) = self.account_data_guard.remove(key) else { return };
if !succeeded {
self.store_account_data_cache(key, guard.server_value);
self.emit(MessengerEvent::RoomsChanged);
}
}
fn apply_synced_account_data(&mut self, room_id: Option<&RoomId>, event_type: &str, content: serde_json::Value) {
let key: AccountDataKey = (room_id.cloned(), event_type.to_string());
if let Some(guard) = self.account_data_guard.get_mut(&key) {
guard.server_value = Some(content);
return;
}
self.store_account_data_cache(&key, Some(content));
}
pub fn take_flush_batch(&mut self) -> Option<FlushBatch> {
if let Some(batch) = self.pending_flush_batches.pop_front() {
return Some(batch);
}
self.store.take_flush_batch()
}
pub fn ack_flush(&mut self, id: u64) {
self.store.ack_flush(id);
}
pub fn events(&mut self) -> Vec<MessengerEvent> {
self.drain_events()
}
pub fn take_ingest_error(&mut self) -> Option<String> {
self.ingest_error.take()
}
pub fn change_counter(&self) -> u64 {
self.change_counter
}
pub fn room_ids(&self) -> impl Iterator<Item = &RoomId> {
self.rooms.keys()
}
pub fn room_state(&self, room_id: &RoomId) -> Option<&RoomState> {
self.rooms.get(room_id)
}
pub fn timeline(&self, room_id: &RoomId) -> Option<&Timeline> {
self.timelines.get(room_id)
}
pub fn room_kind(&self, room_id: &RoomId) -> Option<RoomKind> {
self.rooms.get(room_id).map(|state| state.derive_room_kind(room_id, self.direct_account_data.as_ref()))
}
pub fn typing_users(&self, room_id: &RoomId) -> &[UserId] {
self.typing.get(room_id).map(Vec::as_slice).unwrap_or(&[])
}
pub fn receipts(&self, room_id: &RoomId) -> Option<&ReceiptContent> {
self.receipts.get(room_id)
}
pub fn room_account_data(&self, room_id: &RoomId, event_type: &str) -> Option<&serde_json::Value> {
self.room_account_data.get(room_id).and_then(|by_type| by_type.get(event_type))
}
pub fn global_account_data(&self, event_type: &str) -> Option<&serde_json::Value> {
self.global_account_data.get(event_type)
}
pub fn key_request_count(&self) -> usize {
self.key_requests_sent.len()
}
pub fn user_id(&self) -> &UserId {
&self.config.user_id
}
pub fn has_tracked_device(&self, user_id: &UserId) -> bool {
DeviceTracker::devices_for_user(&self.store, user_id).map(|devices| !devices.is_empty()).unwrap_or(false)
}
pub fn public_rooms_result(&self) -> &[PublicRoomsResultEntry] {
&self.public_rooms_result
}
pub fn user_search_result(&self) -> &[UserDirectoryResultEntry] {
&self.user_search_result
}
pub fn send_failure_reason(&self, txn_id: &TxnId) -> Option<&str> {
match self.send_state.get(txn_id).map(|pending| &pending.phase) {
Some(SendPhase::Failed { errcode }) => Some(errcode.as_str()),
_ => None,
}
}
pub fn dispatch(&mut self, cmd: MessengerCommand, now_ms: i64) -> Result<(), MessengerError> {
match cmd {
MessengerCommand::SendMessage { room_id, message, txn_id } => {
if let Some(edit_of) = &message.edit_of {
let owned = self
.timelines
.get(&room_id)
.and_then(|timeline| timeline.item_by_event_id(edit_of))
.is_some_and(|item| item.sender == self.config.user_id);
if !owned {
return Err(MessengerError::EditNotOwned { event_id: edit_of.clone() });
}
}
self.start_new_send(room_id, SendPayload::Message(message), txn_id, now_ms)?;
}
MessengerCommand::Forward { from_room, event_id, to_room, txn_id } => {
let payload = self.build_forward_payload(&from_room, &event_id, &to_room)?;
self.start_new_send(to_room, payload, txn_id, now_ms)?;
}
MessengerCommand::React { room_id, target, key } => {
self.start_new_send(room_id, SendPayload::Reaction { target, key }, None, now_ms)?;
}
MessengerCommand::Redact { room_id, target, reason } => {
self.start_new_send(room_id, SendPayload::Redaction { target, reason }, None, now_ms)?;
}
MessengerCommand::RetrySend { room_id, txn_id } => {
self.retry_send(&room_id, &txn_id, now_ms)?;
}
MessengerCommand::CreateRoom { kind } => {
self.dispatch_create_room(kind)?;
}
MessengerCommand::JoinRoom { room_id } => {
let request_id = self.next_request_id()?;
let request = OutgoingRequest::join_room(request_id, room_id.as_str());
self.enqueue_request(request, Lane::Other)?;
let dm_peer = self.rooms.get(&room_id).and_then(|room| {
let invited_as_direct = room.members.get(&self.config.user_id).is_some_and(|member| member.is_direct);
if !invited_as_direct {
return None;
}
room.members.keys().find(|user_id| **user_id != self.config.user_id).cloned()
});
if let Some(peer) = dm_peer {
self.merge_direct_account_data(room_id, peer)?;
}
}
MessengerCommand::LeaveRoom { room_id } => {
let request_id = self.next_request_id()?;
let request = OutgoingRequest::leave_room(request_id, &room_id);
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::Invite { room_id, user_id } => {
let request_id = self.next_request_id()?;
let request = OutgoingRequest::invite(request_id, &room_id, &user_id);
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::Kick { room_id, user_id, reason } => {
let request_id = self.next_request_id()?;
let request = OutgoingRequest::kick(request_id, &room_id, &user_id, reason.as_deref());
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::SetTag { room_id, tag, order } => {
let mut content = self.current_tag_content(&room_id);
content.tags.insert(tag, TagInfo { order });
self.enqueue_tag_update(room_id, content)?;
}
MessengerCommand::RemoveTag { room_id, tag } => {
let mut content = self.current_tag_content(&room_id);
content.tags.remove(&tag);
self.enqueue_tag_update(room_id, content)?;
}
MessengerCommand::MarkRead { room_id, event_id } => {
let request_id = self.next_request_id()?;
let body = serde_json::json!({ "m.fully_read": event_id.as_str(), "m.read": event_id.as_str() });
let request = OutgoingRequest::read_markers(request_id, &room_id, body);
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::SetTyping { room_id, typing } => {
if typing {
let should_send = match self.typing_debounce.get(&room_id) {
Some(&last_sent_ms) => now_ms.saturating_sub(last_sent_ms) >= TYPING_DEBOUNCE_MS,
None => true,
};
if !should_send {
return Ok(());
}
self.typing_debounce.insert(room_id.clone(), now_ms);
} else {
self.typing_debounce.remove(&room_id);
}
let request_id = self.next_request_id()?;
let timeout_ms = typing.then_some(TYPING_TIMEOUT_MS);
let request = OutgoingRequest::typing(request_id, &room_id, &self.config.user_id, typing, timeout_ms);
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::RetryDecryption { room_id, event_id } => {
if self.retry_decrypt_one(&room_id, &event_id) {
self.emit(MessengerEvent::TimelineChanged { room_id });
}
}
MessengerCommand::LoadOlder { room_id } => {
let from = self
.timelines
.get(&room_id)
.and_then(|timeline| {
timeline
.older_token()
.map(str::to_string)
.or_else(|| timeline.gap().and_then(|gap| gap.prev_batch.clone()))
})
.or_else(|| {
if self.rooms.contains_key(&room_id) {
self.store.sync_token().ok().flatten().map(str::to_string)
} else {
None
}
});
let Some(from) = from else { return Ok(()) };
let request_id = self.next_request_id()?;
let request =
OutgoingRequest::room_messages(request_id.clone(), &room_id, &from, "b", Some(LOAD_OLDER_PAGE_SIZE));
self.enqueue_request(request, Lane::Room(room_id.clone()))?;
self.pending_room_messages.insert(request_id, room_id);
}
MessengerCommand::SetAccountData { event_type, content } => {
self.write_account_data(None, event_type, content)?;
}
MessengerCommand::SetRoomAccountData { room_id, event_type, content } => {
self.write_account_data(Some(room_id), event_type, content)?;
}
MessengerCommand::SearchPublicRooms { term } => {
let request_id = self.next_request_id()?;
let body = serde_json::json!({ "filter": { "generic_search_term": term }, "limit": 50 });
let request = OutgoingRequest::public_rooms(request_id.clone(), body);
self.latest_public_rooms_request = Some(request_id);
self.enqueue_request(request, Lane::Other)?;
}
MessengerCommand::SearchUsers { term } => {
let request_id = self.next_request_id()?;
let body = serde_json::json!({ "search_term": term, "limit": 20 });
let request = OutgoingRequest::user_directory_search(request_id.clone(), body);
self.latest_user_search_request = Some(request_id);
self.enqueue_request(request, Lane::Other)?;
}
}
Ok(())
}
fn dispatch_create_room(&mut self, kind: CreateRoomKind) -> Result<(), MessengerError> {
let body = match &kind {
CreateRoomKind::Dm { peer } => serde_json::json!({
"is_direct": true,
"invite": [peer.as_str()],
}),
CreateRoomKind::Group { name, invite, members_can_invite } => serde_json::json!({
"visibility": "private",
"is_direct": false,
"name": name,
"invite": invite.iter().map(UserId::as_str).collect::<Vec<_>>(),
"members_can_invite": members_can_invite,
}),
CreateRoomKind::Channel { name, topic } => {
let mut value = serde_json::json!({
"visibility": "public",
"is_direct": false,
"name": name,
});
if let Some(topic) = topic {
value["topic"] = serde_json::Value::String(topic.clone());
}
value
}
};
let request_id = self.next_request_id()?;
let request = OutgoingRequest::create_room(request_id.clone(), body);
self.pending_create_room.insert(request_id.clone(), kind);
self.enqueue_request(request, Lane::Other)?;
Ok(())
}
fn current_tag_content(&self, room_id: &RoomId) -> TagContent {
self.room_account_data(room_id, "m.tag")
.and_then(|value| serde_json::from_value(value.clone()).ok())
.unwrap_or_default()
}
fn enqueue_tag_update(&mut self, room_id: RoomId, content: TagContent) -> Result<(), MessengerError> {
let value = serde_json::to_value(&content)?;
self.write_account_data(Some(room_id), "m.tag".to_string(), value)
}
fn build_forward_payload(
&self,
from_room: &RoomId,
event_id: &EventId,
to_room: &RoomId,
) -> Result<SendPayload, MessengerError> {
let refuse = |reason: &dyn std::fmt::Display| {
MessengerError::IntentRefused(format!("cannot forward {event_id} from {from_room}: {reason}"))
};
if !self.rooms.contains_key(to_room) {
return Err(refuse(&format!("the target room {to_room} is unknown")));
}
let source = match self.timelines.get(from_room) {
Some(timeline) => timeline.forward_source(event_id),
None => Err(ForwardRefusal::Missing),
}
.map_err(|reason| refuse(&reason))?;
let (marker_key, marker) = self.forward_marker(from_room);
Ok(SendPayload::Event {
event_type: source.event_type,
content: forwardable_content(source.content, marker_key, marker),
})
}
fn forward_marker(&self, from_room: &RoomId) -> (&'static str, serde_json::Value) {
let public_channel = self.room_kind(from_room) == Some(RoomKind::Channel);
if public_channel {
let room_name = self.forward_room_name(from_room);
(KEY_FORWARDED_FROM, serde_json::json!({ "room_id": from_room.as_str(), "room_name": room_name }))
} else {
(KEY_FORWARDED, serde_json::Value::Bool(true))
}
}
fn forward_room_name(&self, room_id: &RoomId) -> String {
if let Some(name) = self.rooms.get(room_id).and_then(|room| room.name.clone()) {
return name;
}
match self.room_kind(room_id) {
Some(RoomKind::Dm) => "Direct",
Some(RoomKind::Channel) => "Channel",
Some(RoomKind::Group) | None => "Group",
}
.to_string()
}
fn retry_decrypt_one(&mut self, room_id: &RoomId, event_id: &EventId) -> bool {
let Some(item) = self.pending_decrypt.get(room_id).and_then(|by_event| by_event.get(event_id)).cloned() else {
return false;
};
match GroupSessionManager::decrypt_event(
&mut self.store,
room_id,
event_id,
item.origin_server_ts,
&item.sender,
&item.content,
) {
Ok(plaintext) => {
self.apply_decrypted_plaintext(room_id, event_id, &item.sender, item.origin_server_ts, &plaintext);
}
Err(GroupDecryptError::MissingSession { .. }) => {
let _ = self.request_missing_room_key(room_id, &item.sender, &item.content);
return false;
}
Err(_) => return false,
}
if let Some(by_event) = self.pending_decrypt.get_mut(room_id) {
by_event.remove(event_id);
}
true
}
pub fn pull_latest_page(&mut self, room_id: RoomId) -> Result<(), MessengerError> {
let request_id = self.next_request_id()?;
let request = OutgoingRequest::room_messages_latest(request_id.clone(), &room_id, 20);
self.enqueue_request(request, Lane::Room(room_id.clone()))?;
self.pending_room_messages.insert(request_id, room_id);
Ok(())
}
pub fn decrypt_loaded_timeline(&mut self) {
let pending: Vec<(RoomId, Vec<EventId>)> = self
.pending_decrypt
.iter()
.map(|(room_id, by_event)| (room_id.clone(), by_event.keys().cloned().collect()))
.collect();
for (room_id, event_ids) in pending {
for event_id in event_ids {
self.retry_decrypt_one(&room_id, &event_id);
}
}
let rooms: Vec<RoomId> = self.timelines.keys().cloned().collect();
for room_id in rooms {
let sealed: Vec<(EventId, i64, UserId, MegolmEncryptedContent)> = {
let Some(timeline) = self.timelines.get(&room_id) else {
continue;
};
timeline
.items()
.iter()
.filter_map(|item| {
let event_id = item.event_id.clone()?;
let (session_id, ciphertext) = match &item.content {
ItemContent::Encrypted { session_id, ciphertext_b64 } => {
(session_id.clone(), ciphertext_b64.clone())
}
_ => return None,
};
Some((
event_id,
item.origin_server_ts,
item.sender.clone(),
MegolmEncryptedContent {
ciphertext,
session_id,
sender_key: None,
device_id: None,
relates_to: None,
},
))
})
.collect()
};
for (event_id, origin_server_ts, sender, content) in sealed {
let plaintext = match GroupSessionManager::decrypt_event(
&mut self.store,
&room_id,
&event_id,
origin_server_ts,
&sender,
&content,
) {
Ok(plaintext) => plaintext,
Err(GroupDecryptError::MissingSession { .. }) => {
let _ = self.request_missing_room_key(&room_id, &sender, &content);
continue;
}
Err(_) => continue,
};
self.apply_decrypted_plaintext(
&room_id,
&event_id,
&sender,
origin_server_ts,
&plaintext,
);
}
}
}
fn apply_decrypted_plaintext(
&mut self,
room_id: &RoomId,
event_id: &EventId,
sender: &UserId,
origin_server_ts: i64,
plaintext: &RoomEventPlaintext,
) {
let Some(timeline) = self.timelines.get_mut(room_id) else { return };
if let Some(relation) = RelatesTo::from_content(&plaintext.content) {
if matches!(relation, RelatesTo::Annotation { .. } | RelatesTo::Replace { .. }) {
let new_content = plaintext.content.get("m.new_content").cloned();
timeline.apply_decrypted_relation(event_id, sender.clone(), origin_server_ts, relation, new_content);
return;
}
}
timeline.set_decrypted_event(event_id, &plaintext.event_type, &plaintext.content);
}
fn retry_pending_for_session(&mut self, room_id: &RoomId, session_id: &str) {
let Some(by_event) = self.pending_decrypt.get(room_id) else { return };
let candidates: Vec<EventId> =
by_event.iter().filter(|(_, item)| item.session_id == session_id).map(|(event_id, _)| event_id.clone()).collect();
let mut changed = false;
for event_id in candidates {
if self.retry_decrypt_one(room_id, &event_id) {
changed = true;
}
}
if changed {
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
}
pub fn republish_device_keys(&mut self) {
self.account.forget_published_device_keys();
}
fn ingest_sync(&mut self, response: &crate::wire::sync::SyncResponse) -> Result<(), MessengerError> {
let mut newly_received_sessions: Vec<(RoomId, String)> = Vec::new();
for event in &response.to_device.events {
self.process_to_device_event(event, &mut newly_received_sessions)?;
}
DeviceTracker::on_device_lists(&mut self.store, &response.device_lists.changed, &response.device_lists.left)?;
let published_otk_count = response.device_one_time_keys_count.get("signed_curve25519").copied().unwrap_or(0);
self.account.on_sync_counts(&mut self.store, published_otk_count, &response.device_unused_fallback_key_types)?;
let keys_upload_request_id = self.next_request_id()?;
if let Some(request) =
self.account.keys_upload_request(keys_upload_request_id, &self.config.user_id, &self.config.device_id)?
{
self.enqueue_request(request, Lane::Other)?;
}
self.maybe_enqueue_cross_signing()?;
let outdated_before_rooms = self.outdated_tracked_users()?;
if !outdated_before_rooms.is_empty() {
self.enqueue_keys_query()?;
}
let rooms_touched =
!response.rooms.join.is_empty() || !response.rooms.invite.is_empty() || !response.rooms.leave.is_empty();
for (room_id, joined) in &response.rooms.join {
self.ingest_joined_room(room_id, joined)?;
}
for (room_id, invited) in &response.rooms.invite {
self.ingest_invited_room(room_id, invited)?;
}
for (room_id, left) in &response.rooms.leave {
self.ingest_left_room(room_id, left)?;
}
if rooms_touched {
self.emit(MessengerEvent::RoomsChanged);
}
if !self.outdated_tracked_users()?.is_subset(&outdated_before_rooms) {
self.enqueue_keys_query()?;
}
for event in &response.account_data.events {
self.apply_synced_account_data(None, &event.event_type, event.content.clone());
}
for (room_id, session_id) in &newly_received_sessions {
self.retry_pending_for_session(room_id, session_id);
}
self.store.save_sync_token(response.next_batch.clone())?;
Ok(())
}
fn process_to_device_event(
&mut self,
event: &ToDeviceEvent,
newly_received_sessions: &mut Vec<(RoomId, String)>,
) -> Result<(), MessengerError> {
if event.event_type != "m.room.encrypted" {
return Ok(());
}
let Ok(RoomEncryptedContent::Olm(content)) = serde_json::from_value::<RoomEncryptedContent>(event.content.clone())
else {
return Ok(());
};
match OlmSessionManager::decrypt_to_device(&mut self.store, &mut self.account, &self.config.user_id, &event.sender, &content) {
Ok(decrypted) => self.route_decrypted_to_device(decrypted, newly_received_sessions),
Err(OlmDecryptError::UnknownSenderDevice { .. }) => {
self.queue_unknown_sender_retry(event.clone());
Ok(())
}
Err(_) => Ok(()),
}
}
fn route_decrypted_to_device(
&mut self,
decrypted: DecryptedToDevice,
newly_received_sessions: &mut Vec<(RoomId, String)>,
) -> Result<(), MessengerError> {
match decrypted.event_type.as_str() {
"m.room_key" => {
if let Ok(content) = serde_json::from_value::<RoomKeyContent>(decrypted.content.clone()) {
if content.algorithm == "m.megolm.v1.aes-sha2" {
GroupSessionManager::receive_room_key(
&mut self.store,
&decrypted.sender,
decrypted.sender_device_curve25519,
decrypted.sender_ed25519,
&content,
)?;
newly_received_sessions.push((content.room_id, content.session_id));
}
}
}
"m.forwarded_room_key" => {
if let Ok(content) = serde_json::from_value::<ForwardedRoomKeyContent>(decrypted.content.clone()) {
if let Some((user_id, curve, ed)) = self.forward_binding(&decrypted, &content)? {
let stored = GroupSessionManager::receive_forwarded_room_key(
&mut self.store,
&user_id,
curve,
ed,
&content,
)?;
if stored {
newly_received_sessions.push((content.room_id, content.session_id));
}
}
}
}
"m.room_key_request" => {
if let Ok(content) = serde_json::from_value::<RoomKeyRequestContent>(decrypted.content.clone()) {
self.answer_room_key_request(&decrypted, &content)?;
}
}
"m.room_key.withheld" => {
if let Ok(content) = serde_json::from_value::<RoomKeyWithheldContent>(decrypted.content.clone()) {
if let (Some(room_id), Some(session_id)) = (content.room_id, content.session_id) {
self.withheld_sessions.insert((room_id, session_id));
}
}
}
_ => {}
}
Ok(())
}
fn forward_binding(
&self,
decrypted: &DecryptedToDevice,
content: &ForwardedRoomKeyContent,
) -> Result<Option<(UserId, mail4agent_vodozemac::Curve25519PublicKey, mail4agent_vodozemac::Ed25519PublicKey)>, MessengerError>
{
let members = self.encrypted_room_member_ids(&content.room_id);
let forwarder_is_us = decrypted.sender == self.config.user_id;
if !forwarder_is_us && !members.contains(&decrypted.sender) {
return Ok(None);
}
for user_id in &members {
for device in DeviceTracker::devices_for_user(&self.store, user_id)? {
if device.blocked || device.curve25519.to_base64() != content.sender_key {
continue;
}
let same_user = decrypted.sender == device.user_id;
if same_user || forwarder_is_us {
return Ok(Some((device.user_id, device.curve25519, device.ed25519)));
}
}
}
if decrypted.sender_device_curve25519.to_base64() == content.sender_key {
return Ok(Some((
decrypted.sender.clone(),
decrypted.sender_device_curve25519,
decrypted.sender_ed25519,
)));
}
Ok(None)
}
fn answer_room_key_request(
&mut self,
decrypted: &DecryptedToDevice,
content: &RoomKeyRequestContent,
) -> Result<(), MessengerError> {
if matches!(content.action, RoomKeyRequestAction::RequestCancellation) {
self.key_requests_answered.remove(&(
decrypted.sender.clone(),
content.requesting_device_id.clone(),
content.request_id.clone(),
));
return Ok(());
}
if !matches!(content.action, RoomKeyRequestAction::Request) {
return Ok(());
}
let Some(body) = content.body.as_ref() else { return Ok(()) };
if body.algorithm != "m.megolm.v1.aes-sha2" {
return Ok(());
}
let members = self.encrypted_room_member_ids(&body.room_id);
if !members.contains(&decrypted.sender) {
return Ok(());
}
if decrypted.sender != self.config.user_id && !self.history_forwarding_allowed(&body.room_id) {
return Ok(());
}
if decrypted.sender == self.config.user_id && content.requesting_device_id == self.config.device_id {
return Ok(());
}
let dedup = (decrypted.sender.clone(), content.requesting_device_id.clone(), content.request_id.clone());
if self.key_requests_answered.contains(&dedup) {
return Ok(());
}
let Some(target) = self.requester_device(decrypted, &content.requesting_device_id)? else {
return Ok(());
};
let our_curve = self.account.identity_keys().curve25519.to_base64();
let (inner_type, payload) =
if let Some(forwarded) = GroupSessionManager::forwarded_room_key_content(
&self.store,
&body.room_id,
&body.session_id,
&our_curve,
)? {
("m.forwarded_room_key", serde_json::to_value(&forwarded)?)
} else if let Some(room_key) =
GroupSessionManager::outbound_room_key_content(&self.store, &body.room_id, &body.session_id)?
{
("m.room_key", serde_json::to_value(&room_key)?)
} else {
return Ok(());
};
let sent = self.enqueue_olm_to_devices(std::slice::from_ref(&target), inner_type, payload)?;
if sent > 0 {
self.key_requests_answered.insert(dedup);
}
Ok(())
}
fn requester_device(
&self,
decrypted: &DecryptedToDevice,
requesting_device_id: &DeviceId,
) -> Result<Option<StoredDevice>, MessengerError> {
let devices = DeviceTracker::devices_for_user(&self.store, &decrypted.sender)?;
if let Some(device) = devices.iter().find(|device| &device.device_id == requesting_device_id) {
return Ok((!device.blocked).then(|| device.clone()));
}
let curve = decrypted.sender_device_curve25519.to_base64();
if let Some(device) = devices.iter().find(|device| device.curve25519.to_base64() == curve) {
return Ok((!device.blocked).then(|| device.clone()));
}
Ok(None)
}
fn request_missing_room_key(
&mut self,
room_id: &RoomId,
sender: &UserId,
content: &MegolmEncryptedContent,
) -> Result<(), MessengerError> {
let key = (room_id.clone(), content.session_id.clone());
if self.key_requests_sent.contains(&key) || self.withheld_sessions.contains(&key) {
return Ok(());
}
let sender_key = match content.sender_key.clone() {
Some(sender_key) => sender_key,
None => {
let devices = DeviceTracker::devices_for_user(&self.store, sender)?;
let found = content
.device_id
.as_ref()
.and_then(|device_id| devices.iter().find(|device| &device.device_id == device_id))
.or_else(|| devices.iter().find(|device| !device.blocked));
match found {
Some(device) => device.curve25519.to_base64(),
None => return Ok(()),
}
}
};
let mut targets = DeviceTracker::devices_for_user(&self.store, sender)?;
targets.extend(DeviceTracker::devices_for_user(&self.store, &self.config.user_id)?);
if targets.is_empty() {
return Ok(());
}
let missing: Vec<StoredDevice> = OlmSessionManager::sessions_missing_for(&self.store, &targets)?
.into_iter()
.filter(|device| {
!device.blocked
&& !(device.user_id == self.config.user_id && device.device_id == self.config.device_id)
&& !self.history_claim_tried.contains(&(device.user_id.clone(), device.device_id.clone()))
})
.cloned()
.collect();
if !missing.is_empty() {
let claim_id = self.next_request_id()?;
let refs: Vec<&StoredDevice> = missing.iter().collect();
if let Some(request) = OlmSessionManager::keys_claim_request(claim_id.clone(), &refs) {
for device in &missing {
self.history_claim_tried.insert((device.user_id.clone(), device.device_id.clone()));
}
self.history_claims.insert(claim_id, missing);
self.enqueue_request(request, Lane::Other)?;
return Ok(());
}
}
let request_id = self.next_txn_id()?;
let body = RoomKeyRequestContent {
action: RoomKeyRequestAction::Request,
body: Some(RoomKeyRequestBody {
algorithm: "m.megolm.v1.aes-sha2".to_string(),
room_id: room_id.clone(),
session_id: content.session_id.clone(),
sender_key,
}),
request_id: request_id.as_str().to_string(),
requesting_device_id: self.config.device_id.clone(),
};
let sent = self.enqueue_olm_to_devices(&targets, "m.room_key_request", serde_json::to_value(&body)?)?;
if sent > 0 {
self.key_requests_sent.insert(key);
}
Ok(())
}
fn enqueue_olm_to_devices(
&mut self,
devices: &[StoredDevice],
inner_type: &str,
content: serde_json::Value,
) -> Result<usize, MessengerError> {
let mut unique = Vec::new();
for device in devices {
if device.blocked || (device.user_id == self.config.user_id && device.device_id == self.config.device_id) {
continue;
}
if unique.iter().any(|have: &StoredDevice| have.user_id == device.user_id && have.device_id == device.device_id)
{
continue;
}
unique.push(device.clone());
}
let missing = OlmSessionManager::sessions_missing_for(&self.store, &unique)?;
let missing_ids: BTreeSet<(UserId, DeviceId)> =
missing.into_iter().map(|device| (device.user_id.clone(), device.device_id.clone())).collect();
let ready: Vec<StoredDevice> = unique
.into_iter()
.filter(|device| !missing_ids.contains(&(device.user_id.clone(), device.device_id.clone())))
.collect();
let mut sent = 0;
for chunk in GroupSessionManager::chunk_recipients_for_send_to_device(&ready) {
let body = GroupSessionManager::build_encrypted_to_device_body(
&mut self.store,
&self.account,
&self.config.user_id,
&self.config.device_id,
chunk,
inner_type,
content.clone(),
)?;
let request_id = self.next_request_id()?;
let share_txn = self.next_txn_id()?;
let request = OutgoingRequest::send_to_device(request_id, ROOM_KEY_SHARE_EVENT_TYPE, &share_txn, body);
self.enqueue_request(request, Lane::ToDevice)?;
sent += chunk.len();
}
Ok(sent)
}
fn queue_unknown_sender_retry(&mut self, event: ToDeviceEvent) {
self.unknown_sender_retry.push_back(event);
while self.unknown_sender_retry.len() > UNKNOWN_SENDER_RETRY_CAPACITY {
self.unknown_sender_retry.pop_front();
}
}
fn retry_unknown_sender_queue(&mut self) -> Result<(), MessengerError> {
let pending: Vec<ToDeviceEvent> = std::mem::take(&mut self.unknown_sender_retry).into_iter().collect();
let mut newly_received_sessions = Vec::new();
for event in pending {
self.process_to_device_event(&event, &mut newly_received_sessions)?;
}
for (room_id, session_id) in &newly_received_sessions {
self.retry_pending_for_session(room_id, session_id);
}
Ok(())
}
fn outdated_tracked_users(&self) -> Result<BTreeSet<UserId>, MessengerError> {
Ok(self.store.tracked_users()?.into_iter().filter_map(|(user_id, outdated)| outdated.then_some(user_id)).collect())
}
fn enqueue_keys_query(&mut self) -> Result<(), MessengerError> {
let request_id = self.next_request_id()?;
if let Some(request) = DeviceTracker::keys_query_request(&self.store, request_id)? {
self.enqueue_request(request, Lane::Other)?;
}
Ok(())
}
fn retry_missing_room_keys(&mut self) {
let work: Vec<(RoomId, EventId)> = self
.pending_decrypt
.iter()
.flat_map(|(room, by_event)| by_event.keys().map(move |event| (room.clone(), event.clone())))
.collect();
for (room_id, event_id) in work {
let _ = self.retry_decrypt_one(&room_id, &event_id);
}
}
fn history_forwarding_allowed(&self, room_id: &RoomId) -> bool {
!matches!(
self.rooms.get(room_id).and_then(|room| room.history_visibility.clone()),
Some(crate::wire::events::HistoryVisibility::Joined) | Some(crate::wire::events::HistoryVisibility::Invited)
)
}
fn maybe_enqueue_cross_signing(&mut self) -> Result<(), MessengerError> {
if self.xsign_inflight || self.xsign_attempts >= 4 {
return Ok(());
}
if self.xsign.is_none() {
self.xsign = Some(CrossSigningState::load_or_create(&mut self.store)?);
}
let Some(state) = self.xsign.clone() else { return Ok(()) };
let user_id = self.config.user_id.clone();
let request_id;
let request = if !state.uploaded {
request_id = self.next_request_id()?;
OutgoingRequest::signing_keys_upload(request_id, state.upload_body(&user_id)?)
} else if !state.device_signed {
let device_id = self.config.device_id.clone();
let device_keys = self.account.device_keys_json(&user_id, &device_id)?;
request_id = self.next_request_id()?;
OutgoingRequest::signatures_upload(request_id, state.device_signature_body(&user_id, &device_id, device_keys)?)
} else {
return Ok(());
};
self.xsign_attempts += 1;
self.xsign_inflight = true;
self.enqueue_request(request, Lane::Other)
}
fn track_joined_members<'a>(&mut self, room_id: &RoomId, events: impl Iterator<Item = &'a RawEvent>) -> Result<(), MessengerError> {
if self.rooms.get(room_id).is_none_or(|room| room.encryption.is_none()) {
return Ok(());
}
let joined: Vec<UserId> = events
.filter(|event| event.event_type == "m.room.member")
.filter(|event| event.content.get("membership").and_then(serde_json::Value::as_str) == Some("join"))
.filter_map(|event| event.state_key.as_deref().and_then(|state_key| UserId::parse(state_key).ok()))
.filter(|user_id| *user_id != self.config.user_id)
.collect();
if joined.is_empty() {
return Ok(());
}
DeviceTracker::on_device_lists(&mut self.store, &joined, &[])
}
fn ingest_joined_room(&mut self, room_id: &RoomId, joined: &JoinedRoom) -> Result<(), MessengerError> {
for event in joined.state.events.iter().chain(joined.timeline.events.iter()) {
self.rooms.entry(room_id.clone()).or_default().apply_state_event(event)?;
}
self.track_joined_members(room_id, joined.state.events.iter().chain(joined.timeline.events.iter()))?;
let unread_before = self.rooms.get(room_id).map(RoomState::unread_count);
self.rooms.entry(room_id.clone()).or_default().apply_summary(&joined.summary, &joined.unread_notifications);
let unread_after = self.rooms.get(room_id).map(RoomState::unread_count);
if unread_before != unread_after {
self.emit(MessengerEvent::UnreadChanged { room_id: room_id.clone() });
}
for event in &joined.account_data.events {
self.apply_synced_account_data(Some(room_id), &event.event_type, event.content.clone());
}
if let Some(typing_event) = joined.ephemeral.events.iter().find(|event| event.event_type == "m.typing") {
if let Ok(typing) = serde_json::from_value::<TypingContent>(typing_event.content.clone()) {
self.typing.insert(room_id.clone(), typing.user_ids);
self.emit(MessengerEvent::TypingChanged { room_id: room_id.clone() });
}
}
if let Some(receipt_event) = joined.ephemeral.events.iter().find(|event| event.event_type == "m.receipt") {
if let Ok(receipts) = serde_json::from_value::<ReceiptContent>(receipt_event.content.clone()) {
self.receipts.insert(room_id.clone(), receipts);
self.emit(MessengerEvent::ReceiptsChanged { room_id: room_id.clone() });
}
}
if !joined.timeline.events.is_empty() {
self.timelines.entry(room_id.clone()).or_default().apply_timeline_batch(
&joined.timeline.events,
joined.timeline.limited,
joined.timeline.prev_batch.clone(),
);
self.decrypt_new_timeline_events(room_id, &joined.timeline.events);
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
self.persist_room_state(room_id)
}
fn ingest_invited_room(&mut self, room_id: &RoomId, invited: &InvitedRoom) -> Result<(), MessengerError> {
for stripped in &invited.invite_state.events {
let synthetic = stripped_state_to_raw_event(stripped);
self.rooms.entry(room_id.clone()).or_default().apply_state_event(&synthetic)?;
}
self.persist_room_state(room_id)
}
fn ingest_left_room(&mut self, room_id: &RoomId, left: &LeftRoom) -> Result<(), MessengerError> {
for event in left.state.events.iter().chain(left.timeline.events.iter()) {
self.rooms.entry(room_id.clone()).or_default().apply_state_event(event)?;
}
for event in &left.account_data.events {
self.apply_synced_account_data(Some(room_id), &event.event_type, event.content.clone());
}
if !left.timeline.events.is_empty() {
self.timelines.entry(room_id.clone()).or_default().apply_timeline_batch(
&left.timeline.events,
left.timeline.limited,
left.timeline.prev_batch.clone(),
);
self.decrypt_new_timeline_events(room_id, &left.timeline.events);
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
}
self.persist_room_state(room_id)
}
fn persist_room_state(&mut self, room_id: &RoomId) -> Result<(), MessengerError> {
let Some(state) = self.rooms.get(room_id) else { return Ok(()) };
let bytes = serde_json::to_vec(state)
.map_err(|source| MessengerError::Crypto(format!("encode room state for {room_id}: {source}")))?;
self.store.save_room_state(room_id, bytes)?;
Ok(())
}
fn decrypt_new_timeline_events(&mut self, room_id: &RoomId, events: &[RawEvent]) {
for event in events {
if event.event_type != "m.room.encrypted" || event.state_key.is_some() || event.unsigned.redacted_because.is_some() {
continue;
}
let Ok(RoomEncryptedContent::Megolm(content)) = serde_json::from_value::<RoomEncryptedContent>(event.content.clone())
else {
continue;
};
match GroupSessionManager::decrypt_event(
&mut self.store,
room_id,
&event.event_id,
event.origin_server_ts,
&event.sender,
&content,
) {
Ok(plaintext) => {
self.apply_decrypted_plaintext(room_id, &event.event_id, &event.sender, event.origin_server_ts, &plaintext);
if let Some(by_event) = self.pending_decrypt.get_mut(room_id) {
by_event.remove(&event.event_id);
}
}
Err(err) => {
if matches!(err, GroupDecryptError::MissingSession { .. }) {
let _ = self.request_missing_room_key(room_id, &event.sender, &content);
}
let reason = format!("{:?}", err.utd_reason());
if let Some(timeline) = self.timelines.get_mut(room_id) {
timeline.set_decrypted(&event.event_id, ItemContent::Undecryptable { reason });
}
self.pending_decrypt.entry(room_id.clone()).or_default().insert(
event.event_id.clone(),
PendingDecryptItem {
session_id: content.session_id.clone(),
sender: event.sender.clone(),
origin_server_ts: event.origin_server_ts,
content,
},
);
}
}
}
}
fn ingest_room_messages_response(&mut self, room_id: &RoomId, body: &[u8]) -> Result<(), MessengerError> {
let parsed: RoomMessagesResponseBody = serde_json::from_slice(body)?;
let mut events = parsed.chunk;
events.reverse();
self.timelines.entry(room_id.clone()).or_default().prepend_back_page(&events, parsed.end);
self.decrypt_new_timeline_events(room_id, &events);
self.emit(MessengerEvent::TimelineChanged { room_id: room_id.clone() });
self.persist_room_state(room_id)
}
}
impl MessengerCore<SealedRecordCodec> {
pub fn open_sealed(
records: impl IntoIterator<Item = SealedRecord>,
secrets: CoreSecrets,
config: CoreConfig,
now_ms: i64,
jitter: Box<dyn Jitter>,
) -> Result<Self, MessengerError> {
let seal_key = secrets.store_seal_key.clone().ok_or_else(|| {
MessengerError::Crypto(
"MessengerCore::open_sealed requires CoreSecrets::store_seal_key"
.to_string(),
)
})?;
let codec = SealedRecordCodec::new(seal_key);
Self::open(records, codec, config, secrets, now_ms, jitter)
}
}
fn stripped_placeholder_event_id() -> EventId {
EventId::parse("$stripped-preview").expect("this literal is a well-formed, hardcoded $-prefixed opaque id")
}
fn stripped_state_to_raw_event(stripped: &StrippedStateEvent) -> RawEvent {
RawEvent {
event_id: stripped_placeholder_event_id(),
event_type: stripped.event_type.clone(),
sender: stripped.sender.clone(),
origin_server_ts: 0,
state_key: Some(stripped.state_key.clone()),
content: stripped.content.clone(),
unsigned: Unsigned::default(),
}
}
fn forwardable_content(mut content: serde_json::Value, marker_key: &str, marker: serde_json::Value) -> serde_json::Value {
if let Some(object) = content.as_object_mut() {
for key in ["m.relates_to", "m.mentions", "m.new_content", KEY_FORWARDED, KEY_FORWARDED_FROM] {
object.remove(key);
}
object.insert(marker_key.to_string(), marker);
}
content
}
fn message_local_echo_content(message: &OutgoingMessage) -> ItemContent {
let body = TextLikeMessageContent { body: message.body.clone(), format: None, formatted_body: None };
match message.kind {
MessageKind::Text => ItemContent::Text(body),
MessageKind::Notice => ItemContent::Notice(body),
MessageKind::Emote => ItemContent::Emote(body),
}
}
fn message_wire_content(message: &OutgoingMessage) -> serde_json::Value {
let msgtype = message.kind.msgtype();
let new_content = serde_json::json!({ "msgtype": msgtype, "body": message.body });
let mut content = new_content.clone();
if let Some(edit_of) = &message.edit_of {
content["body"] = serde_json::Value::String(format!("* {}", message.body));
content["m.new_content"] = new_content;
content["m.relates_to"] = serde_json::json!({ "rel_type": "m.replace", "event_id": edit_of.as_str() });
} else if let Some(reply_to) = &message.reply_to {
content["m.relates_to"] = serde_json::json!({ "m.in_reply_to": { "event_id": reply_to.as_str() } });
}
content
}
fn message_relates_to(message: &OutgoingMessage) -> Option<RelatesTo> {
if let Some(edit_of) = &message.edit_of {
Some(RelatesTo::Replace { event_id: edit_of.clone() })
} else {
message.reply_to.clone().map(|event_id| RelatesTo::InReplyTo(InReplyTo { event_id }))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::room::timeline::Forwarded;
use crate::store::InsecurePlainCodecForTests;
use crate::wire::HttpMethod;
struct FixedJitter(f64);
impl Jitter for FixedJitter {
fn next_unit(&mut self) -> f64 {
self.0
}
}
fn device_id() -> DeviceId {
DeviceId::parse("DEV1").expect("valid device id")
}
fn user_id() -> UserId {
UserId::parse("@alice:example.org").expect("valid user id")
}
fn test_config() -> CoreConfig {
CoreConfig { user_id: user_id(), device_id: device_id(), server_name: "example.org".to_string() }
}
fn open_fresh_core() -> MessengerCore<InsecurePlainCodecForTests> {
MessengerCore::open(Vec::new(), InsecurePlainCodecForTests, test_config(), CoreSecrets::default(), 0, Box::new(FixedJitter(0.0)))
.expect("open succeeds on a brand-new device")
}
fn flush_and_ack(core: &mut MessengerCore<InsecurePlainCodecForTests>) {
while let Some(batch) = core.take_flush_batch() {
core.ack_flush(batch.id);
}
}
#[test]
fn counters_survive_restart_and_never_repeat_txn_ids() {
let device = device_id();
let mut store = Store::new(device.clone(), InsecurePlainCodecForTests);
store.save_counters(Counters { next_request_id: 3, next_txn_id: 7 }).expect("save");
let batch = store.take_flush_batch().expect("counters dirtied the store");
store.ack_flush(batch.id);
let reloaded = Store::load(batch.records, InsecurePlainCodecForTests, device).expect("reload");
let restored = reloaded.counters().expect("no error").expect("counters were persisted");
assert_eq!(restored.next_txn_id, 7, "a restart resumes the txn-id sequence exactly where it left off");
assert_eq!(restored.next_request_id, 3);
assert_eq!(restore_counter(3, Some(10)), 11, "resumes past the highest pending id, never repeating it");
assert_eq!(restore_counter(3, None), 3, "nothing pending: the persisted value alone is authoritative");
}
#[test]
fn open_restores_the_request_counter_past_a_still_pending_requests_id() {
let device = device_id();
let mut store = Store::new(device.clone(), InsecurePlainCodecForTests);
let mut queue = OutgoingQueue::new();
let pending = OutgoingRequest::sync(RequestId::next(5), None, None);
queue.enqueue(&mut store, pending, Lane::Sync).expect("enqueue");
store.save_counters(Counters { next_request_id: 2, next_txn_id: 0 }).expect("save");
let batch = store.take_flush_batch().expect("pending");
store.ack_flush(batch.id);
let core = MessengerCore::open(
batch.records,
InsecurePlainCodecForTests,
test_config(),
CoreSecrets::default(),
0,
Box::new(FixedJitter(0.0)),
)
.expect("open succeeds");
assert_eq!(
core.counters.next_request_id, 6,
"resumes past the highest id already in flight, not the stale persisted value"
);
}
#[test]
fn sync_token_persisted_after_everything_it_covers() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.releasable_requests(0);
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let sync_request = requests.iter().find(|r| r.kind == OutgoingRequestKind::Sync).expect("a sync request was enqueued");
assert!(core.store.sync_token().expect("no error").is_none(), "no token before the first response");
let body = serde_json::json!({
"next_batch": "s1",
"device_one_time_keys_count": { "signed_curve25519": 0 },
"device_unused_fallback_key_types": []
})
.to_string();
core.on_response(sync_request.id.clone(), HttpResponseDescriptor { status: 200, body: body.into_bytes() }, 0);
assert_eq!(core.store.sync_token().expect("no error"), Some("s1"));
}
#[test]
fn only_one_sync_in_flight() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.releasable_requests(0);
flush_and_ack(&mut core);
let first = core.releasable_requests(0);
assert_eq!(first.iter().filter(|r| r.kind == OutgoingRequestKind::Sync).count(), 1);
let second = core.releasable_requests(0);
assert!(second.iter().all(|r| r.kind != OutgoingRequestKind::Sync), "no second sync request released");
}
#[test]
fn first_sync_of_a_session_never_long_polls_then_the_loop_does() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.releasable_requests(0);
flush_and_ack(&mut core);
let first = core.releasable_requests(0);
let first_sync = first.iter().find(|r| r.kind == OutgoingRequestKind::Sync).expect("a sync request was enqueued");
assert!(first_sync.query.contains(&("timeout".to_string(), "0".to_string())), "the initial sync must answer at once");
assert!(first_sync.query.iter().all(|(key, _)| key != "since"), "a fresh device has no since token");
let body = serde_json::json!({
"next_batch": "s1",
"device_one_time_keys_count": { "signed_curve25519": 0 },
"device_unused_fallback_key_types": []
})
.to_string();
core.on_response(first_sync.id.clone(), HttpResponseDescriptor { status: 200, body: body.into_bytes() }, 0);
flush_and_ack(&mut core);
core.releasable_requests(0);
flush_and_ack(&mut core);
let second = core.releasable_requests(0);
let second_sync = second.iter().find(|r| r.kind == OutgoingRequestKind::Sync).expect("the follow-up sync was enqueued");
assert!(second_sync.query.contains(&("since".to_string(), "s1".to_string())));
assert!(second_sync.query.contains(&("timeout".to_string(), SYNC_TIMEOUT_MS.to_string())), "once caught up the loop long-polls");
}
#[test]
fn room_key_arrival_retries_undecryptable_items() {
use crate::crypto::group_sessions::GroupSessionManager;
use mail4agent_vodozemac::megolm::{GroupSession, SessionConfig};
use mail4agent_vodozemac::olm::Account;
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
let sender = UserId::parse("@bob:example.org").expect("valid user id");
let identity = Account::new().identity_keys();
let mut group_session = GroupSession::new(SessionConfig::version_1());
let session_id = group_session.session_id();
let session_key = group_session.session_key();
let plaintext = crate::crypto::group_sessions::RoomEventPlaintext {
event_type: "m.room.message".to_string(),
content: serde_json::json!({ "msgtype": "m.text", "body": "hi" }),
room_id: room_id.clone(),
};
let message = group_session.encrypt(serde_json::to_vec(&plaintext).expect("valid JSON"));
let event: RawEvent = serde_json::from_value(serde_json::json!({
"event_id": "$evt:example.org",
"type": "m.room.encrypted",
"sender": sender.as_str(),
"origin_server_ts": 1,
"content": {
"algorithm": "m.megolm.v1.aes-sha2",
"ciphertext": message.to_base64(),
"session_id": session_id,
}
}))
.expect("valid event");
let mut timeline = Timeline::new();
timeline.apply_timeline_batch(std::slice::from_ref(&event), false, None);
core.timelines.insert(room_id.clone(), timeline);
core.decrypt_new_timeline_events(&room_id, std::slice::from_ref(&event));
{
let timeline = core.timeline(&room_id).expect("timeline exists");
let item = timeline.item_by_event_id(&EventId::parse("$evt:example.org").expect("valid event id")).expect("item present");
assert!(matches!(item.content, ItemContent::Undecryptable { .. }));
}
let content = RoomKeyContent {
algorithm: "m.megolm.v1.aes-sha2".to_string(),
room_id: room_id.clone(),
session_id: session_id.clone(),
session_key: session_key.to_base64(),
};
GroupSessionManager::receive_room_key(&mut core.store, &sender, identity.curve25519, identity.ed25519, &content)
.expect("receive room key");
core.retry_pending_for_session(&room_id, &session_id);
let timeline = core.timeline(&room_id).expect("timeline exists");
let item = timeline.item_by_event_id(&EventId::parse("$evt:example.org").expect("valid event id")).expect("item present");
match &item.content {
ItemContent::Text(text) => assert_eq!(text.body, "hi"),
other => panic!("expected the retried decrypt to succeed, got {other:?}"),
}
}
#[test]
fn redelivered_to_device_after_restart_is_dropped_silently() {
use crate::crypto::olm_sessions::OlmSessionManager;
let alice_user = UserId::parse("@alice:example.org").expect("valid user id");
let alice_device = DeviceId::parse("ALICEDEV").expect("valid device id");
let mut alice_store = Store::new(alice_device.clone(), InsecurePlainCodecForTests);
let alice_account = OlmAccountState::load_or_create(&mut alice_store).expect("create alice's account");
let alice_device_keys =
alice_account.device_keys_json(&alice_user, &alice_device).expect("sign alice's device_keys");
let mut bob_store = Store::new(device_id(), InsecurePlainCodecForTests);
let mut bob_account = OlmAccountState::load_or_create(&mut bob_store).expect("create bob's account");
bob_account.on_sync_counts(&mut bob_store, 0, &["signed_curve25519".to_string()]).expect("top up bob's OTKs");
let upload = bob_account
.keys_upload_request(RequestId::next(0), &user_id(), &device_id())
.expect("builds a request")
.expect("something to upload");
let one_time_keys = upload.body.expect("upload has a body");
let one_time_keys = one_time_keys.get("one_time_keys").and_then(serde_json::Value::as_object).expect("OTKs present");
let (key_id, key_value) = one_time_keys.iter().next().expect("at least one OTK");
let claim_body = serde_json::to_vec(&serde_json::json!({
"one_time_keys": { user_id().as_str(): { device_id().as_str(): { key_id: key_value } } }
}))
.expect("valid JSON");
let bob_device_stored = crate::crypto::device_tracker::StoredDevice {
user_id: user_id(),
device_id: device_id(),
curve25519: bob_account.identity_keys().curve25519,
ed25519: bob_account.identity_keys().ed25519,
algorithms: vec!["m.olm.v1.curve25519-aes-sha2".to_string(), "m.megolm.v1.aes-sha2".to_string()],
display_name: None,
verified: false,
blocked: false,
};
OlmSessionManager::on_keys_claim_response(&mut alice_store, &alice_account, &[&bob_device_stored], &claim_body)
.expect("alice establishes a session with bob");
DeviceTracker::on_keys_query_response(
&mut bob_store,
&serde_json::to_vec(&serde_json::json!({
"device_keys": { alice_user.as_str(): { alice_device.as_str(): alice_device_keys } }
}))
.expect("valid JSON"),
)
.expect("bob learns alice's device");
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
let group_session = mail4agent_vodozemac::megolm::GroupSession::new(mail4agent_vodozemac::megolm::SessionConfig::version_1());
let room_key_content = RoomKeyContent {
algorithm: "m.megolm.v1.aes-sha2".to_string(),
room_id,
session_id: group_session.session_id(),
session_key: group_session.session_key().to_base64(),
};
let encrypted = OlmSessionManager::encrypt_to_device(
&mut alice_store,
&alice_account,
&alice_user,
&alice_device,
&bob_device_stored,
"m.room_key",
serde_json::to_value(&room_key_content).expect("valid JSON"),
)
.expect("encrypt to bob");
let to_device_event = ToDeviceEvent {
sender: alice_user.clone(),
event_type: "m.room.encrypted".to_string(),
content: serde_json::to_value(RoomEncryptedContent::Olm(encrypted)).expect("valid JSON"),
};
let mut core =
MessengerCore::open(Vec::new(), InsecurePlainCodecForTests, test_config(), CoreSecrets::default(), 0, Box::new(FixedJitter(0.0)))
.expect("open succeeds");
core.store = bob_store;
core.account = bob_account;
let mut sessions = Vec::new();
core.process_to_device_event(&to_device_event, &mut sessions).expect("first decrypt succeeds");
assert_eq!(
sessions,
vec![(room_key_content.room_id.clone(), room_key_content.session_id.clone())],
"the first, genuine delivery is fully validated and accepted"
);
let batch = core.store.take_flush_batch().expect("decrypting dirtied the store");
core.store.ack_flush(batch.id);
let reloaded_store = Store::load(batch.records, InsecurePlainCodecForTests, device_id()).expect("reload succeeds");
core.store = reloaded_store;
core.account = OlmAccountState::load_or_create(&mut core.store).expect("reload reuses the persisted account");
let mut sessions_after_restart = Vec::new();
let result = core.process_to_device_event(&to_device_event, &mut sessions_after_restart);
assert!(result.is_ok(), "a redelivered to-device event is dropped, not surfaced as an ingestion error");
assert!(sessions_after_restart.is_empty(), "the replay never produces a second accepted room key");
}
#[test]
fn unknown_sender_device_is_retried_after_keys_query() {
use crate::crypto::olm_sessions::OlmSessionManager;
let alice_user = UserId::parse("@alice:example.org").expect("valid user id");
let alice_device = DeviceId::parse("ALICEDEV").expect("valid device id");
let mut alice_store = Store::new(alice_device.clone(), InsecurePlainCodecForTests);
let alice_account = OlmAccountState::load_or_create(&mut alice_store).expect("create alice's account");
let alice_device_keys =
alice_account.device_keys_json(&alice_user, &alice_device).expect("sign alice's device_keys");
let bob_user = UserId::parse("@bob:example.org").expect("valid user id");
let bob_device = DeviceId::parse("BOBDEV").expect("valid device id");
let mut bob_store = Store::new(bob_device.clone(), InsecurePlainCodecForTests);
let mut bob_account = OlmAccountState::load_or_create(&mut bob_store).expect("create bob's account");
bob_account.on_sync_counts(&mut bob_store, 0, &["signed_curve25519".to_string()]).expect("top up bob's OTKs");
let upload = bob_account
.keys_upload_request(RequestId::next(0), &bob_user, &bob_device)
.expect("builds a request")
.expect("something to upload");
let one_time_keys = upload.body.expect("upload has a body");
let one_time_keys = one_time_keys.get("one_time_keys").and_then(serde_json::Value::as_object).expect("OTKs present");
let (key_id, key_value) = one_time_keys.iter().next().expect("at least one OTK");
let claim_body = serde_json::to_vec(&serde_json::json!({
"one_time_keys": { bob_user.as_str(): { bob_device.as_str(): { key_id: key_value } } }
}))
.expect("valid JSON");
let bob_device_stored = crate::crypto::device_tracker::StoredDevice {
user_id: bob_user.clone(),
device_id: bob_device.clone(),
curve25519: bob_account.identity_keys().curve25519,
ed25519: bob_account.identity_keys().ed25519,
algorithms: vec!["m.olm.v1.curve25519-aes-sha2".to_string(), "m.megolm.v1.aes-sha2".to_string()],
display_name: None,
verified: false,
blocked: false,
};
OlmSessionManager::on_keys_claim_response(&mut alice_store, &alice_account, &[&bob_device_stored], &claim_body)
.expect("alice establishes a session with bob");
let encrypted = OlmSessionManager::encrypt_to_device(
&mut alice_store,
&alice_account,
&alice_user,
&alice_device,
&bob_device_stored,
"m.dummy",
serde_json::json!({}),
)
.expect("encrypt to bob");
let to_device_event = ToDeviceEvent {
sender: alice_user.clone(),
event_type: "m.room.encrypted".to_string(),
content: serde_json::to_value(RoomEncryptedContent::Olm(encrypted)).expect("valid JSON"),
};
let bob_config = CoreConfig { user_id: bob_user.clone(), device_id: bob_device.clone(), server_name: "example.org".to_string() };
let mut core =
MessengerCore::open(Vec::new(), InsecurePlainCodecForTests, bob_config, CoreSecrets::default(), 0, Box::new(FixedJitter(0.0)))
.expect("open succeeds");
core.store = bob_store;
core.account = bob_account;
let mut sessions = Vec::new();
core.process_to_device_event(&to_device_event, &mut sessions).expect("queues for retry, does not error");
assert_eq!(core.unknown_sender_retry.len(), 1, "queued for a retry after the next keys/query");
let alice_curve = alice_account.identity_keys().curve25519.to_base64();
assert!(
core.store.olm_sessions_for_device(&alice_curve).expect("no error").is_empty(),
"the refused attempt created no Olm session (it would have consumed the one-time key)"
);
let query_body = serde_json::to_vec(&serde_json::json!({
"device_keys": { alice_user.as_str(): { alice_device.as_str(): alice_device_keys } }
}))
.expect("valid JSON");
DeviceTracker::on_keys_query_response(&mut core.store, &query_body).expect("bob learns alice's device");
core.retry_unknown_sender_queue().expect("no error");
assert!(core.unknown_sender_retry.is_empty(), "the queued entry is drained by the retry attempt");
assert_eq!(
core.store.olm_sessions_for_device(&alice_curve).expect("no error").len(),
1,
"the retry decrypted the very same pre-key message and established the inbound session"
);
}
#[test]
fn dispatch_load_older_produces_an_outgoing_request() {
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
let mut timeline = Timeline::new();
let event: RawEvent = serde_json::from_value(serde_json::json!({
"event_id": "$1:example.org",
"type": "m.room.message",
"sender": "@alice:example.org",
"origin_server_ts": 1,
"content": { "msgtype": "m.text", "body": "hi" }
}))
.expect("valid event");
timeline.apply_timeline_batch(&[event], true, Some("t1".to_string()));
core.timelines.insert(room_id.clone(), timeline);
flush_and_ack(&mut core);
core.dispatch(MessengerCommand::LoadOlder { room_id: room_id.clone() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let load_older = requests.iter().find(|r| r.kind == OutgoingRequestKind::RoomMessages);
assert!(load_older.is_some(), "a RoomMessages request was enqueued using the room's own gap token");
}
#[test]
fn load_older_on_a_resumed_room_pages_back_from_the_sync_token() {
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
core.store.save_sync_token("s42_7".to_string()).expect("save the sync token");
flush_and_ack(&mut core);
core.dispatch(MessengerCommand::LoadOlder { room_id: room_id.clone() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
assert!(core.releasable_requests(0).iter().all(|r| r.kind != OutgoingRequestKind::RoomMessages));
core.rooms.insert(room_id.clone(), RoomState::default());
core.dispatch(MessengerCommand::LoadOlder { room_id: room_id.clone() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let load_older = requests.iter().find(|r| r.kind == OutgoingRequestKind::RoomMessages).expect("a RoomMessages request was enqueued");
assert!(load_older.query.contains(&("from".to_string(), "s42_7".to_string())), "pages back from the sync token: {:?}", load_older.query);
}
fn bob_joins_room_body(encrypted: bool) -> serde_json::Value {
let mut state = Vec::new();
if encrypted {
state.push(serde_json::json!({
"event_id": "$enc:example.org", "type": "m.room.encryption", "sender": "@alice:example.org",
"origin_server_ts": 1, "state_key": "", "content": { "algorithm": "m.megolm.v1.aes-sha2" }
}));
}
serde_json::json!({
"next_batch": "s1",
"rooms": { "join": { "!r:example.org": {
"state": { "events": state },
"timeline": { "events": [{
"event_id": "$join:example.org", "type": "m.room.member", "sender": "@bob:example.org",
"origin_server_ts": 2, "state_key": "@bob:example.org", "content": { "membership": "join" }
}] }
} } }
})
}
#[test]
fn a_member_joining_an_encrypted_room_is_tracked_and_queried_right_away() {
let mut core = open_fresh_core();
let bob = UserId::parse("@bob:example.org").expect("valid user id");
deliver_sync(&mut core, bob_joins_room_body(true));
let tracked = core.store.tracked_users().expect("no error");
assert!(tracked.iter().any(|(user_id, outdated)| *user_id == bob && *outdated), "bob is tracked with an outdated device list: {tracked:?}");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let query = requests.iter().find(|r| r.kind == OutgoingRequestKind::KeysQuery).expect("a /keys/query was enqueued for the new member");
assert!(query.body.as_ref().is_some_and(|body| body["device_keys"].get(bob.as_str()).is_some()), "the query names bob: {:?}", query.body);
}
#[test]
fn a_member_joining_a_plaintext_room_is_not_tracked() {
let mut core = open_fresh_core();
deliver_sync(&mut core, bob_joins_room_body(false));
assert!(core.store.tracked_users().expect("no error").is_empty());
flush_and_ack(&mut core);
assert!(core.releasable_requests(0).iter().all(|r| r.kind != OutgoingRequestKind::KeysQuery));
}
#[test]
fn dispatch_mark_read_and_set_typing_produce_outgoing_requests() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
let event_id = EventId::parse("$1:example.org").expect("valid event id");
core.dispatch(MessengerCommand::MarkRead { room_id: room_id.clone(), event_id }, 0).expect("dispatch succeeds");
core.dispatch(MessengerCommand::SetTyping { room_id: room_id.clone(), typing: true }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
assert!(requests.iter().any(|r| r.kind == OutgoingRequestKind::ReadMarkers));
assert!(requests.iter().any(|r| r.kind == OutgoingRequestKind::Typing));
}
#[test]
fn snapshot_reflects_the_latest_applied_sync() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.releasable_requests(0);
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let sync_request = requests.iter().find(|r| r.kind == OutgoingRequestKind::Sync).expect("sync enqueued");
let body = serde_json::json!({
"next_batch": "s1",
"rooms": {
"join": {
"!r:example.org": {
"state": { "events": [] },
"timeline": {
"events": [
{
"event_id": "$1:example.org",
"type": "m.room.message",
"sender": "@alice:example.org",
"origin_server_ts": 1,
"content": { "msgtype": "m.text", "body": "hi" }
}
]
},
"unread_notifications": { "notification_count": 1, "highlight_count": 0 }
}
}
}
})
.to_string();
let events = core.on_response(sync_request.id.clone(), HttpResponseDescriptor { status: 200, body: body.into_bytes() }, 0);
assert!(events.contains(&MessengerEvent::RoomsChanged));
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
assert!(events.contains(&MessengerEvent::TimelineChanged { room_id: room_id.clone() }));
assert!(events.contains(&MessengerEvent::UnreadChanged { room_id: room_id.clone() }));
let timeline = core.timeline(&room_id).expect("timeline present");
assert_eq!(timeline.items().len(), 1);
assert_eq!(core.room_state(&room_id).expect("room present").unread_count(), 1);
assert_eq!(core.change_counter(), events.len() as u64);
assert!(core.events().is_empty());
}
#[test]
fn dispatch_set_account_data_updates_optimistically_and_sends_a_put() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
let content = serde_json::json!({ "note": "keep" });
core.dispatch(
MessengerCommand::SetAccountData { event_type: "com.example.prefs".to_string(), content: content.clone() },
0,
)
.expect("dispatch succeeds");
assert_eq!(
core.global_account_data("com.example.prefs"),
Some(&content),
"SetAccountData updates the core's own cache before any round trip"
);
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let request = requests.iter().find(|r| r.kind == OutgoingRequestKind::AccountData).expect("an AccountData request was enqueued");
assert_eq!(request.method, HttpMethod::Put);
assert_eq!(request.path, "/_matrix/client/v3/user/%40alice%3Aexample.org/account_data/com.example.prefs");
assert_eq!(request.body.as_ref(), Some(&content));
}
#[test]
fn dispatch_set_room_account_data_updates_optimistically_and_sends_a_put() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
let content = serde_json::json!({ "flag": true });
core.dispatch(
MessengerCommand::SetRoomAccountData {
room_id: room_id.clone(),
event_type: "com.example.flag".to_string(),
content: content.clone(),
},
0,
)
.expect("dispatch succeeds");
assert_eq!(
core.room_account_data(&room_id, "com.example.flag"),
Some(&content),
"SetRoomAccountData updates the core's own cache before any round trip"
);
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let request =
requests.iter().find(|r| r.kind == OutgoingRequestKind::RoomAccountData).expect("a RoomAccountData request was enqueued");
assert_eq!(request.method, HttpMethod::Put);
assert_eq!(
request.path,
"/_matrix/client/v3/user/%40alice%3Aexample.org/rooms/%21r%3Aexample.org/account_data/com.example.flag"
);
assert_eq!(request.body.as_ref(), Some(&content));
}
#[test]
fn dispatch_search_public_rooms_sends_the_expected_request() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.dispatch(MessengerCommand::SearchPublicRooms { term: "trading".to_string() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let request = requests.iter().find(|r| r.kind == OutgoingRequestKind::PublicRooms).expect("a PublicRooms request was enqueued");
assert_eq!(request.method, HttpMethod::Post);
assert_eq!(request.path, "/_matrix/client/v3/publicRooms");
assert_eq!(request.body, Some(serde_json::json!({ "filter": { "generic_search_term": "trading" }, "limit": 50 })));
}
#[test]
fn dispatch_search_users_sends_the_expected_request() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.dispatch(MessengerCommand::SearchUsers { term: "bob".to_string() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let requests = core.releasable_requests(0);
let request =
requests.iter().find(|r| r.kind == OutgoingRequestKind::UserDirectorySearch).expect("a UserDirectorySearch request was enqueued");
assert_eq!(request.method, HttpMethod::Post);
assert_eq!(request.path, "/_matrix/client/v3/user_directory/search");
assert_eq!(request.body, Some(serde_json::json!({ "search_term": "bob", "limit": 20 })));
}
#[test]
fn public_rooms_response_superseded_by_a_later_search_is_dropped() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
core.dispatch(MessengerCommand::SearchPublicRooms { term: "first".to_string() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let first_request = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::PublicRooms)
.expect("the first search's own request");
core.dispatch(MessengerCommand::SearchPublicRooms { term: "second".to_string() }, 0).expect("dispatch succeeds");
flush_and_ack(&mut core);
let second_request = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::PublicRooms)
.expect("the second search's own request");
let stale_body = serde_json::json!({ "chunk": [{ "room_id": "!stale:example.org", "num_joined_members": 1 }] });
core.on_response(first_request.id, HttpResponseDescriptor { status: 200, body: serde_json::to_vec(&stale_body).expect("json") }, 0);
assert!(core.public_rooms_result().is_empty(), "a response to a superseded search is dropped");
let fresh_body = serde_json::json!({ "chunk": [{ "room_id": "!fresh:example.org", "num_joined_members": 2 }] });
core.on_response(second_request.id, HttpResponseDescriptor { status: 200, body: serde_json::to_vec(&fresh_body).expect("json") }, 0);
assert_eq!(core.public_rooms_result().len(), 1);
assert_eq!(core.public_rooms_result()[0].room_id, RoomId::parse("!fresh:example.org").expect("valid room id"));
}
fn is_account_data_write(request: &OutgoingRequest) -> bool {
matches!(request.kind, OutgoingRequestKind::AccountData | OutgoingRequestKind::RoomAccountData)
}
#[test]
fn pin_then_archive_back_to_back_keeps_both_tags() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
for tag in ["m.favourite", "u.example.archived"] {
core.dispatch(MessengerCommand::SetTag { room_id: room_id.clone(), tag: tag.to_string(), order: None }, 0)
.expect("dispatch succeeds");
}
let cached = core.room_account_data(&room_id, "m.tag").expect("m.tag cached");
assert!(cached["tags"].get("m.favourite").is_some());
assert!(cached["tags"].get("u.example.archived").is_some());
flush_and_ack(&mut core);
let first_release = core.releasable_requests(0);
let first_writes: Vec<&OutgoingRequest> = first_release.iter().filter(|r| is_account_data_write(r)).collect();
assert_eq!(first_writes.len(), 1, "the second PUT is not released while the first is in flight");
let first_body = first_writes[0].body.clone().expect("a body");
assert!(first_body["tags"].get("m.favourite").is_some());
assert!(first_body["tags"].get("u.example.archived").is_none(), "the first PUT carries only what existed when it was built");
core.on_response(first_writes[0].id.clone(), HttpResponseDescriptor { status: 200, body: b"{}".to_vec() }, 0);
flush_and_ack(&mut core);
let second_release = core.releasable_requests(0);
let second = second_release.iter().find(|r| is_account_data_write(r)).expect("the second PUT is released after the first completed");
let second_body = second.body.clone().expect("a body");
assert!(second_body["tags"].get("m.favourite").is_some(), "the later PUT still carries the earlier tag");
assert!(second_body["tags"].get("u.example.archived").is_some());
}
#[test]
fn account_data_lane_is_fifo_single_flight() {
let mut core = open_fresh_core();
flush_and_ack(&mut core);
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
core.dispatch(MessengerCommand::SetAccountData { event_type: "org.t.one".to_string(), content: serde_json::json!({}) }, 0)
.expect("dispatch succeeds");
core.dispatch(
MessengerCommand::SetRoomAccountData {
room_id: room_id.clone(),
event_type: "org.t.two".to_string(),
content: serde_json::json!({}),
},
0,
)
.expect("dispatch succeeds");
core.dispatch(MessengerCommand::SetAccountData { event_type: "org.t.three".to_string(), content: serde_json::json!({}) }, 0)
.expect("dispatch succeeds");
core.dispatch(MessengerCommand::RemoveTag { room_id: room_id.clone(), tag: "m.favourite".to_string() }, 0)
.expect("dispatch succeeds");
flush_and_ack(&mut core);
let mut released_paths = Vec::new();
for _ in 0..4 {
let release = core.releasable_requests(0);
let writes: Vec<&OutgoingRequest> = release.iter().filter(|r| is_account_data_write(r)).collect();
assert_eq!(writes.len(), 1, "exactly one account-data write in flight at a time");
released_paths.push(writes[0].path.clone());
core.on_response(writes[0].id.clone(), HttpResponseDescriptor { status: 200, body: b"{}".to_vec() }, 0);
flush_and_ack(&mut core);
}
assert!(released_paths[0].ends_with("/account_data/org.t.one"));
assert!(released_paths[1].ends_with("/account_data/org.t.two"));
assert!(released_paths[2].ends_with("/account_data/org.t.three"));
assert!(released_paths[3].ends_with("/account_data/m.tag"));
assert!(core.releasable_requests(0).iter().all(|r| !is_account_data_write(r)), "the lane is drained");
}
fn deliver_sync(core: &mut MessengerCore<InsecurePlainCodecForTests>, body: serde_json::Value) -> Vec<OutgoingRequest> {
let mut others = Vec::new();
for _ in 0..3 {
flush_and_ack(core);
let mut sync_request = None;
for request in core.releasable_requests(0) {
if request.kind == OutgoingRequestKind::Sync && sync_request.is_none() {
sync_request = Some(request);
} else {
others.push(request);
}
}
if let Some(sync_request) = sync_request {
let body = serde_json::to_vec(&body).expect("valid JSON");
core.on_response(sync_request.id, HttpResponseDescriptor { status: 200, body }, 0);
return others;
}
}
panic!("no sync request became releasable");
}
fn tag_sync_body(room_id: &str, tags: serde_json::Value, next_batch: &str) -> serde_json::Value {
let mut join = serde_json::Map::new();
join.insert(
room_id.to_string(),
serde_json::json!({ "account_data": { "events": [{ "type": "m.tag", "content": { "tags": tags } }] } }),
);
serde_json::json!({ "next_batch": next_batch, "rooms": { "join": serde_json::Value::Object(join) } })
}
fn ok_response() -> HttpResponseDescriptor {
HttpResponseDescriptor { status: 200, body: b"{}".to_vec() }
}
#[test]
fn sync_echo_of_an_older_tag_write_does_not_clobber_a_newer_optimistic_value() {
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
for tag in ["m.favourite", "u.example.archived"] {
core.dispatch(MessengerCommand::SetTag { room_id: room_id.clone(), tag: tag.to_string(), order: None }, 0)
.expect("dispatch succeeds");
}
let released = deliver_sync(&mut core, tag_sync_body("!r:example.org", serde_json::json!({ "m.favourite": {} }), "s1"));
let cached = core.room_account_data(&room_id, "m.tag").expect("m.tag cached");
assert!(cached["tags"].get("u.example.archived").is_some(), "the older echo must not clobber the newer optimistic value");
assert!(cached["tags"].get("m.favourite").is_some());
let key = (Some(room_id.clone()), "m.tag".to_string());
assert_eq!(
core.account_data_guard.get(&key).expect("writes are outstanding").server_value,
Some(serde_json::json!({ "tags": { "m.favourite": {} } })),
"the echo is remembered as the server's value"
);
let first = released.iter().find(|r| r.kind == OutgoingRequestKind::RoomAccountData).expect("first write released");
core.on_response(first.id.clone(), ok_response(), 0);
flush_and_ack(&mut core);
let second = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::RoomAccountData)
.expect("second write released after the first");
core.on_response(second.id, ok_response(), 0);
assert!(core.account_data_guard.is_empty());
let cached = core.room_account_data(&room_id, "m.tag").expect("m.tag cached");
assert!(cached["tags"].get("m.favourite").is_some() && cached["tags"].get("u.example.archived").is_some());
deliver_sync(
&mut core,
tag_sync_body("!r:example.org", serde_json::json!({ "m.favourite": {}, "u.example.archived": {} }), "s2"),
);
let cached = core.room_account_data(&room_id, "m.tag").expect("m.tag cached");
assert!(cached["tags"].get("u.example.archived").is_some());
}
#[test]
fn terminal_account_data_failure_restores_the_server_value() {
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
deliver_sync(&mut core, tag_sync_body("!r:example.org", serde_json::json!({ "m.favourite": {} }), "s1"));
core.dispatch(MessengerCommand::SetTag { room_id: room_id.clone(), tag: "u.example.archived".to_string(), order: None }, 0)
.expect("dispatch succeeds");
assert!(core.room_account_data(&room_id, "m.tag").expect("cached")["tags"].get("u.example.archived").is_some());
flush_and_ack(&mut core);
let write = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::RoomAccountData)
.expect("the write is released");
let counter_before = core.change_counter();
let forbidden =
HttpResponseDescriptor { status: 403, body: br#"{"errcode":"M_FORBIDDEN","error":"nope"}"#.to_vec() };
let events = core.on_response(write.id, forbidden, 0);
assert!(core.account_data_guard.is_empty());
let restored = core.room_account_data(&room_id, "m.tag").expect("the server's value is back");
assert!(restored["tags"].get("u.example.archived").is_none(), "the failed write's optimistic value is gone");
assert!(restored["tags"].get("m.favourite").is_some());
assert!(events.contains(&MessengerEvent::RoomsChanged));
assert!(core.change_counter() > counter_before, "the UI is told to re-render");
for tag in ["u.a", "u.b"] {
core.dispatch(MessengerCommand::SetTag { room_id: room_id.clone(), tag: tag.to_string(), order: None }, 0)
.expect("dispatch succeeds");
}
flush_and_ack(&mut core);
let first = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::RoomAccountData)
.expect("first write released");
let forbidden =
HttpResponseDescriptor { status: 403, body: br#"{"errcode":"M_FORBIDDEN","error":"nope"}"#.to_vec() };
core.on_response(first.id, forbidden, 0);
let cached = core.room_account_data(&room_id, "m.tag").expect("cached");
assert!(cached["tags"].get("u.a").is_some() && cached["tags"].get("u.b").is_some(), "cache untouched while a write is pending");
flush_and_ack(&mut core);
let second = core
.releasable_requests(0)
.into_iter()
.find(|r| r.kind == OutgoingRequestKind::RoomAccountData)
.expect("second write released");
core.on_response(second.id, ok_response(), 0);
let cached = core.room_account_data(&room_id, "m.tag").expect("cached");
assert!(cached["tags"].get("u.b").is_some(), "the last write succeeded: its optimistic value stands");
}
#[test]
fn pending_account_data_writes_rebuild_their_guard_after_a_restart() {
let mut core = open_fresh_core();
let room_id = RoomId::parse("!r:example.org").expect("valid room id");
core.dispatch(MessengerCommand::SetTag { room_id: room_id.clone(), tag: "m.favourite".to_string(), order: None }, 0)
.expect("dispatch succeeds");
core.dispatch(
MessengerCommand::SetAccountData { event_type: "org.t.one".to_string(), content: serde_json::json!({ "n": 1 }) },
0,
)
.expect("dispatch succeeds");
let mut records = Vec::new();
while let Some(batch) = core.take_flush_batch() {
records.extend(batch.records.clone());
core.ack_flush(batch.id);
}
let reopened = MessengerCore::open(
records,
InsecurePlainCodecForTests,
test_config(),
CoreSecrets::default(),
0,
Box::new(FixedJitter(0.0)),
)
.expect("reopen succeeds");
assert_eq!(reopened.account_data_guard.len(), 2, "one guard per key with a pending write");
assert!(reopened.account_data_guard.values().all(|guard| guard.pending == 1 && guard.server_value.is_none()));
assert_eq!(reopened.global_account_data("org.t.one"), Some(&serde_json::json!({ "n": 1 })));
assert!(reopened.room_account_data(&room_id, "m.tag").expect("cached")["tags"].get("m.favourite").is_some());
}
type TestCore = MessengerCore<InsecurePlainCodecForTests>;
fn room(id: &str) -> RoomId {
RoomId::parse(id).expect("valid room id")
}
fn event(id: &str) -> EventId {
EventId::parse(id).expect("valid event id")
}
fn add_room(core: &mut TestCore, room_id: &RoomId, encrypted: bool, name: Option<&str>) {
let mut state = RoomState::new();
if encrypted {
state.encryption = Some(crate::wire::events::RoomEncryptionContent {
algorithm: "m.megolm.v1.aes-sha2".to_string(),
rotation_period_ms: 604_800_000,
rotation_period_msgs: 100,
});
}
state.name = name.map(str::to_string);
core.rooms.insert(room_id.clone(), state);
core.timelines.entry(room_id.clone()).or_default();
}
fn mark_as_channel(core: &mut TestCore, room_id: &RoomId) {
let Some(state) = core.rooms.get_mut(room_id) else { return };
state
.apply_state_event(&state_event_raw(
"m.room.join_rules",
"",
"@alice:example.org",
serde_json::json!({ "join_rule": "public" }),
))
.expect("join_rules");
state
.apply_state_event(&state_event_raw(
"m.room.power_levels",
"",
"@alice:example.org",
serde_json::json!({ "events_default": 50, "users_default": 0 }),
))
.expect("power_levels");
}
fn state_event_raw(event_type: &str, state_key: &str, sender: &str, content: serde_json::Value) -> RawEvent {
serde_json::from_value(serde_json::json!({
"event_id": format!("${event_type}:example.org"),
"type": event_type,
"sender": sender,
"origin_server_ts": 1,
"state_key": state_key,
"content": content,
}))
.expect("valid raw event")
}
fn raw_event(event_id: &str, event_type: &str, sender: &str, content: serde_json::Value) -> RawEvent {
serde_json::from_value(serde_json::json!({
"event_id": event_id,
"type": event_type,
"sender": sender,
"origin_server_ts": 1,
"content": content,
}))
.expect("valid raw event")
}
fn seed_plain(core: &mut TestCore, room_id: &RoomId, event_id: &str, sender: &str, event_type: &str, content: serde_json::Value) {
let raw = raw_event(event_id, event_type, sender, content);
core.timelines.entry(room_id.clone()).or_default().apply_timeline_batch(&[raw], false, None);
}
fn seed_sealed(core: &mut TestCore, room_id: &RoomId, event_id: &str, sender: &str) {
let content = serde_json::json!({ "algorithm": "m.megolm.v1.aes-sha2", "ciphertext": "AAAA", "session_id": "s1" });
seed_plain(core, room_id, event_id, sender, "m.room.encrypted", content);
}
fn seed_decrypted(
core: &mut TestCore,
room_id: &RoomId,
event_id_str: &str,
sender: &str,
event_type: &str,
content: serde_json::Value,
) {
seed_sealed(core, room_id, event_id_str, sender);
if let Some(timeline) = core.timelines.get_mut(room_id) {
timeline.set_decrypted_event(&event(event_id_str), event_type, &content);
}
}
fn sent_room_event(core: &mut TestCore) -> (String, serde_json::Value) {
flush_and_ack(core);
let request = core
.releasable_requests(0)
.into_iter()
.find(|request| request.kind == OutgoingRequestKind::RoomSend)
.expect("a room send is releasable");
let confirmed = format!("$sent{}:example.org", core.counters.next_request_id);
let body = serde_json::json!({ "event_id": confirmed }).to_string().into_bytes();
core.on_response(request.id.clone(), HttpResponseDescriptor { status: 200, body }, 0);
(request.path, request.body.expect("a room send has a body"))
}
fn forward(core: &mut TestCore, from: &RoomId, event_id: &str, to: &RoomId) -> Result<(), MessengerError> {
core.dispatch(
MessengerCommand::Forward { from_room: from.clone(), event_id: event(event_id), to_room: to.clone(), txn_id: None },
0,
)
}
fn last_item(core: &TestCore, room_id: &RoomId) -> crate::room::timeline::TimelineItem {
core.timeline(room_id).and_then(|timeline| timeline.items().last()).expect("the room has an item").clone()
}
#[test]
fn forward_from_encrypted_room_carries_only_the_forwarded_flag() {
let mut core = open_fresh_core();
let dm = room("!dm:example.org");
let target = room("!target:example.org");
add_room(&mut core, &dm, true, Some("Secret DM name"));
add_room(&mut core, &target, false, None);
seed_decrypted(
&mut core,
&dm,
"$src:example.org",
"@bob:example.org",
"m.room.message",
serde_json::json!({
"msgtype": "m.text",
"body": "quarterly numbers",
"m.relates_to": { "m.in_reply_to": { "event_id": "$parent:example.org" } },
"m.mentions": { "user_ids": ["@carol:example.org"] },
}),
);
forward(&mut core, &dm, "$src:example.org", &target).expect("a decrypted message is forwardable");
let (path, body) = sent_room_event(&mut core);
assert!(path.contains("/send/m.room.message/"), "the event type is kept: {path}");
assert_eq!(
body,
serde_json::json!({ "msgtype": "m.text", "body": "quarterly numbers", "forwarded": true }),
"the text plus the bare flag; the reply pointer and the mentions are stripped"
);
let wire = body.to_string();
for leaked in ["bob", "carol", "!dm", "Secret DM name", "$parent"] {
assert!(!wire.contains(leaked), "{leaked:?} must not travel with a forward out of an encrypted room: {wire}");
}
let echo = last_item(&core, &target);
assert_eq!(echo.forwarded, Some(Forwarded::Hidden), "the forwarder's own echo shows the same marker");
}
#[test]
fn forward_from_public_channel_carries_room_id_and_name_no_sender() {
let mut core = open_fresh_core();
let channel = room("!chan:example.org");
let target = room("!target:example.org");
add_room(&mut core, &channel, true, Some("Announcements"));
mark_as_channel(&mut core, &channel);
add_room(&mut core, &target, false, None);
seed_plain(
&mut core,
&channel,
"$post:example.org",
"@dave:example.org",
"m.room.message",
serde_json::json!({
"msgtype": "m.text",
"body": "BTC breaks out",
"m.mentions": { "room": true },
"m.relates_to": { "m.in_reply_to": { "event_id": "$older:example.org" } },
}),
);
forward(&mut core, &channel, "$post:example.org", &target).expect("a channel post is forwardable");
let (_, body) = sent_room_event(&mut core);
assert_eq!(
body,
serde_json::json!({
"msgtype": "m.text",
"body": "BTC breaks out",
"forwarded_from": { "room_id": "!chan:example.org", "room_name": "Announcements" },
})
);
assert!(!body.to_string().contains("dave"), "the original sender never travels");
assert_eq!(
last_item(&core, &target).forwarded,
Some(Forwarded::Channel { room_id: channel.clone(), room_name: "Announcements".to_string() })
);
}
#[test]
fn forward_from_an_unnamed_room_never_lists_member_names() {
let mut core = open_fresh_core();
let channel = room("!chan:example.org");
let target = room("!target:example.org");
add_room(&mut core, &channel, true, None);
mark_as_channel(&mut core, &channel);
if let Some(state) = core.rooms.get_mut(&channel) {
state.members.insert(
UserId::parse("@erin:example.org").expect("valid user id"),
crate::room::state::MemberState {
membership: Membership::Join,
displayname: Some("Erin Private".to_string()),
is_direct: false,
},
);
}
add_room(&mut core, &target, false, None);
seed_plain(&mut core, &channel, "$post:example.org", "@erin:example.org", "m.room.message", serde_json::json!({ "msgtype": "m.text", "body": "hi" }));
forward(&mut core, &channel, "$post:example.org", &target).expect("forwardable");
let (_, body) = sent_room_event(&mut core);
let name = body["forwarded_from"]["room_name"].as_str().expect("a name is sent");
assert!(!name.contains("Erin"), "an unnamed room is attributed by a generic label, not by its members: {name}");
}
#[test]
fn forward_replaces_the_sources_own_forward_marker() {
let mut core = open_fresh_core();
let dm = room("!dm:example.org");
let target = room("!target:example.org");
add_room(&mut core, &dm, true, None);
add_room(&mut core, &target, false, None);
seed_decrypted(
&mut core,
&dm,
"$src:example.org",
"@bob:example.org",
"m.room.message",
serde_json::json!({
"msgtype": "m.text",
"body": "second hop",
"forwarded_from": { "room_id": "!older:example.org", "room_name": "Older channel" },
}),
);
forward(&mut core, &dm, "$src:example.org", &target).expect("forwardable");
let (_, body) = sent_room_event(&mut core);
assert_eq!(body, serde_json::json!({ "msgtype": "m.text", "body": "second hop", "forwarded": true }));
}
#[test]
fn forward_of_edited_item_sends_the_latest_text() {
let mut core = open_fresh_core();
let channel = room("!chan:example.org");
let dm = room("!dm:example.org");
let target = room("!target:example.org");
add_room(&mut core, &channel, false, Some("Announcements"));
add_room(&mut core, &dm, true, None);
add_room(&mut core, &target, false, None);
seed_plain(&mut core, &channel, "$orig:example.org", "@dave:example.org", "m.room.message", serde_json::json!({ "msgtype": "m.text", "body": "typo text" }));
seed_plain(
&mut core,
&channel,
"$edit:example.org",
"@dave:example.org",
"m.room.message",
serde_json::json!({
"msgtype": "m.text",
"body": "* fixed text",
"m.new_content": { "msgtype": "m.text", "body": "fixed text" },
"m.relates_to": { "rel_type": "m.replace", "event_id": "$orig:example.org" },
}),
);
forward(&mut core, &channel, "$orig:example.org", &target).expect("forwardable");
let (_, body) = sent_room_event(&mut core);
assert_eq!(body["body"], "fixed text", "the latest edit's text, not the original and not the `* ` fallback");
assert!(body.get("m.new_content").is_none() && body.get("m.relates_to").is_none());
seed_decrypted(&mut core, &dm, "$secret:example.org", "@bob:example.org", "m.room.message", serde_json::json!({ "msgtype": "m.text", "body": "typo secret" }));
seed_sealed(&mut core, &dm, "$secret-edit:example.org", "@bob:example.org");
let applied = core.timelines.get_mut(&dm).expect("timeline").apply_decrypted_relation(
&event("$secret-edit:example.org"),
UserId::parse("@bob:example.org").expect("valid user id"),
2,
RelatesTo::Replace { event_id: event("$secret:example.org") },
Some(serde_json::json!({ "msgtype": "m.text", "body": "fixed secret" })),
);
assert!(applied, "the decrypted edit folds onto its target");
forward(&mut core, &dm, "$secret:example.org", &target).expect("forwardable");
let (_, body) = sent_room_event(&mut core);
assert_eq!(body, serde_json::json!({ "msgtype": "m.text", "body": "fixed secret", "forwarded": true }));
}
#[test]
fn forward_of_unknown_msgtype_keeps_its_fields() {
let mut core = open_fresh_core();
let dm = room("!dm:example.org");
let target = room("!target:example.org");
add_room(&mut core, &dm, true, None);
add_room(&mut core, &target, false, None);
let card = serde_json::json!({
"msgtype": "com.example.card",
"body": "opaque",
"note": "kept raw",
});
seed_decrypted(&mut core, &dm, "$card:example.org", "@bob:example.org", "m.room.message", card.clone());
forward(&mut core, &dm, "$card:example.org", &target).expect("an unrecognized msgtype is still an m.room.message");
let (path, body) = sent_room_event(&mut core);
assert!(path.contains("/send/m.room.message/"));
let mut expected = card;
expected["forwarded"] = serde_json::Value::Bool(true);
assert_eq!(body, expected, "unrecognized msgtype fields pass through plus the marker");
assert!(matches!(last_item(&core, &target).content, ItemContent::Unknown), "the echo does not invent a card type");
}
#[test]
fn forward_refuses_redacted_sealed_unknown_and_missing() {
let mut core = open_fresh_core();
let dm = room("!dm:example.org");
let target = room("!target:example.org");
add_room(&mut core, &dm, true, None);
add_room(&mut core, &target, false, None);
seed_decrypted(&mut core, &dm, "$redacted:example.org", "@bob:example.org", "m.room.message", serde_json::json!({ "msgtype": "m.text", "body": "gone" }));
core.timelines.get_mut(&dm).expect("timeline").apply_redaction(&event("$redacted:example.org"));
seed_sealed(&mut core, &dm, "$sealed:example.org", "@bob:example.org");
seed_sealed(&mut core, &dm, "$utd:example.org", "@bob:example.org");
core.timelines
.get_mut(&dm)
.expect("timeline")
.set_decrypted(&event("$utd:example.org"), ItemContent::Undecryptable { reason: "MissingSession".to_string() });
seed_plain(&mut core, &dm, "$call:example.org", "@bob:example.org", "m.call.invite", serde_json::json!({ "call_id": "c1" }));
for id in ["$redacted:example.org", "$sealed:example.org", "$utd:example.org", "$call:example.org", "$missing:example.org"] {
match forward(&mut core, &dm, id, &target) {
Err(MessengerError::IntentRefused(reason)) => {
assert!(reason.contains(id), "the refusal names the event {id}: {reason}");
}
other => panic!("forwarding {id} must be refused, got {other:?}"),
}
}
match forward(&mut core, &room("!nowhere:example.org"), "$redacted:example.org", &target) {
Err(MessengerError::IntentRefused(reason)) => assert!(reason.contains("$redacted:example.org"), "{reason}"),
other => panic!("an unknown source room means a missing event, got {other:?}"),
}
match forward(&mut core, &dm, "$sealed:example.org", &room("!nowhere:example.org")) {
Err(MessengerError::IntentRefused(reason)) => assert!(reason.contains("unknown"), "{reason}"),
other => panic!("an unknown target room is refused, got {other:?}"),
}
assert!(core.send_state.is_empty(), "a refused forward queues nothing");
assert!(core.timeline(&target).is_some_and(|timeline| timeline.items().is_empty()), "and leaves no echo");
}
}