diff --git a/src/base.rs b/src/base.rs index 55ca8584..1a179d7e 100644 --- a/src/base.rs +++ b/src/base.rs @@ -13,6 +13,7 @@ use std::time::{Duration, Instant}; use emojis::Emoji; use matrix_sdk::ruma::events::receipt::ReceiptThread; +use matrix_sdk::ruma::OwnedTransactionId; use ratatui::{ buffer::Buffer, layout::{Alignment, Rect}, @@ -731,6 +732,10 @@ pub enum IambError { #[error("Matrix client error: {0}")] Matrix(#[from] matrix_sdk::Error), + /// A failure when sending a message. + #[error("Send queue error: {0}")] + SendQueue(#[from] matrix_sdk::send_queue::RoomSendQueueError), + /// A failure in the sled storage. #[error("Matrix client storage error: {0}")] Store(#[from] matrix_sdk::StoreError), @@ -862,6 +867,19 @@ impl EventLocation { } } +/// Indicates where a local echo lives in the [`ChatStore`]. +#[derive(Debug, Clone)] +pub enum EchoLocation { + /// The [`OwnedTransactionId`] belongs to a message. + /// + /// If the first argument is [`None`], then it's part of the main scrollback. When [`Some`], it + /// specifies which thread it's in reply to. + Message(Option, MessageKey), + + /// The local echo has been replaced by an event with this event id. + Replaced(OwnedEventId), +} + #[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct UnreadInfo { pub(crate) unread: bool, @@ -888,6 +906,7 @@ pub struct RoomInfo { /// A map of event IDs to where they are stored in this struct. pub keys: HashMap, + pub echo_keys: HashMap, /// The messages loaded for this room. messages: Messages, @@ -933,6 +952,7 @@ impl Default for RoomInfo { name: Default::default(), tags: Default::default(), keys: Default::default(), + echo_keys: Default::default(), event_receipts: Default::default(), user_receipts: Default::default(), reactions: Default::default(), @@ -1115,7 +1135,7 @@ impl RoomInfo { MessageEvent::Original(orig) => { orig.content.apply_replacement(new_msgtype); }, - MessageEvent::Local(_, content) => { + MessageEvent::Local(_, _, content) => { content.apply_replacement(new_msgtype); }, MessageEvent::Redacted(_, _) | @@ -1131,7 +1151,10 @@ impl RoomInfo { pub fn insert_any_state(&mut self, msg: AnySyncStateEvent) { let event_id = msg.event_id().to_owned(); - let key = (msg.origin_server_ts().into(), event_id.clone()); + let key = MessageKey { + ts: msg.origin_server_ts().into(), + id: event_id.clone().into(), + }; let loc = EventLocation::State(key.clone()); self.keys.insert(event_id, loc); @@ -1167,13 +1190,13 @@ impl RoomInfo { let last_receipt = std::cmp::max(last_receipt, last_unthreaded); match (last_message, last_receipt) { - (Some(((ts, _), _)), Some((read_ts, _))) => { - UnreadInfo { unread: ts > read_ts, latest: Some(*ts) } + (Some((key, _)), Some(read_key)) => { + UnreadInfo { unread: key.ts > read_key.ts, latest: Some(key.ts) } }, - (Some(((ts, _), _)), None) => { + (Some((key, _)), None) => { // If we've never loaded/generated a room's receipt (example, // a newly joined but never viewed room), show it as unread. - UnreadInfo { unread: true, latest: Some(*ts) } + UnreadInfo { unread: true, latest: Some(key.ts) } }, (None, _) => UnreadInfo::default(), } @@ -1182,7 +1205,10 @@ impl RoomInfo { /// Inserts events that couldn't be decrypted into the scrollback. pub fn insert_encrypted(&mut self, msg: RoomEncryptedEvent) { let event_id = msg.event_id().to_owned(); - let key = (msg.origin_server_ts().into(), event_id.clone()); + let key = MessageKey { + ts: msg.origin_server_ts().into(), + id: event_id.clone().into(), + }; self.keys.insert(event_id, EventLocation::Message(None, key.clone())); self.messages.insert(key, msg.into()); @@ -1191,7 +1217,10 @@ impl RoomInfo { /// Insert a new message. pub fn insert_message(&mut self, msg: RoomMessageEvent) { let event_id = msg.event_id().to_owned(); - let key = (msg.origin_server_ts().into(), event_id.clone()); + let key = MessageKey { + ts: msg.origin_server_ts().into(), + id: event_id.clone().into(), + }; let loc = EventLocation::Message(None, key.clone()); self.keys.insert(event_id, loc); @@ -1200,7 +1229,10 @@ impl RoomInfo { fn insert_thread(&mut self, msg: RoomMessageEvent, thread_root: OwnedEventId) { let event_id = msg.event_id().to_owned(); - let key = (msg.origin_server_ts().into(), event_id.clone()); + let key = MessageKey { + ts: msg.origin_server_ts().into(), + id: event_id.clone().into(), + }; let replies = self .threads @@ -1302,11 +1334,12 @@ impl RoomInfo { } pub fn fully_read(&mut self, user_id: &UserId) { - let Some(((_, event_id), _)) = self.messages.last_key_value() else { + let Some(event_id) = self.messages.iter().rev().find_map(|(key, _)| key.id.as_origin()) + else { return; }; - self.set_receipt(ReceiptThread::Main, user_id.to_owned(), event_id.clone()); + self.set_receipt(ReceiptThread::Main, user_id.to_owned(), event_id.to_owned()); let newest = self .threads @@ -1315,13 +1348,15 @@ impl RoomInfo { let thread = ReceiptThread::Thread(thread_id.to_owned()); messages - .last_key_value() - .map(|((_, event_id), _)| (thread, event_id.to_owned())) + .iter() + .rev() + .find_map(|(key, _)| key.id.as_origin()) + .map(|id| (thread, id.to_owned())) }) .collect::>(); for (thread, event_id) in newest.into_iter() { - self.set_receipt(thread, user_id.to_owned(), event_id.clone()); + self.set_receipt(thread, user_id.to_owned(), event_id); } } diff --git a/src/message/mod.rs b/src/message/mod.rs index 05bc9d26..a4fed1a7 100644 --- a/src/message/mod.rs +++ b/src/message/mod.rs @@ -11,6 +11,8 @@ use std::ops::{Deref, DerefMut}; use chrono::{DateTime, Local as LocalTz}; use humansize::{format_size, DECIMAL}; use matrix_sdk::ruma::events::receipt::ReceiptThread; +use matrix_sdk::ruma::OwnedTransactionId; +use matrix_sdk::send_queue::SendHandle; use unicode_width::UnicodeWidthStr; use matrix_sdk::ruma::{ @@ -73,7 +75,11 @@ pub use html::TreeGenState; type ProtocolPreview<'a> = (&'a Protocol, u16, u16); -pub type MessageKey = (MessageTimeStamp, OwnedEventId); +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)] +pub struct MessageKey { + pub ts: MessageTimeStamp, + pub id: MessageId, +} pub struct Messages(BTreeMap, pub ReceiptThread); @@ -105,14 +111,9 @@ impl Messages { } pub fn insert_message(&mut self, key: MessageKey, msg: impl Into) { - let event_id = key.1.clone(); let msg = msg.into(); self.0.insert(key, msg); - - // Remove any echo. - let key = (MessageTimeStamp::LocalEcho, event_id); - let _ = self.0.remove(&key); } } @@ -155,10 +156,10 @@ fn hash_finish_usize(hasher: DefaultHasher) -> Option { } } -/// Hash an [EventId] into a [usize]. -fn hash_event_id(event_id: &EventId) -> Option { +/// Hash an [`MessageId`] into a [`usize`]. +fn hash_message_id(id: &MessageId) -> Option { let mut hasher = DefaultHasher::new(); - event_id.hash(&mut hasher); + id.hash(&mut hasher); hash_finish_usize(hasher) } @@ -194,11 +195,25 @@ fn placeholder_frame( Some(placeholder) } -#[inline] -fn millis_to_datetime(ms: UInt) -> DateTime { - let time = i64::from(ms) / 1000; - let time = DateTime::from_timestamp(time, 0).unwrap_or_default(); - time.into() +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash)] +pub enum MessageId { + Origin(OwnedEventId), + Local(OwnedTransactionId), +} + +impl MessageId { + pub fn as_origin(&self) -> Option<&EventId> { + match self { + Self::Origin(id) => Some(id), + _ => None, + } + } +} + +impl From for MessageId { + fn from(value: OwnedEventId) -> Self { + Self::Origin(value) + } } #[derive(thiserror::Error, Debug)] @@ -210,83 +225,40 @@ pub enum TimeStampIntError { UIntError(>::Error), } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum MessageTimeStamp { - OriginServer(UInt), - LocalEcho, -} +#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord)] +pub struct MessageTimeStamp(pub MilliSecondsSinceUnixEpoch); impl MessageTimeStamp { - fn as_datetime(&self) -> DateTime { - match self { - MessageTimeStamp::OriginServer(ms) => millis_to_datetime(*ms), - MessageTimeStamp::LocalEcho => LocalTz::now(), - } + fn as_datetime(self) -> DateTime { + let time = i64::from(self.0 .0) / 1000; + let time = DateTime::from_timestamp(time, 0).unwrap_or_default(); + time.into() } - fn same_day(&self, other: &Self) -> bool { + fn same_day(self, other: Self) -> bool { let dt1 = self.as_datetime(); let dt2 = other.as_datetime(); dt1.date_naive() == dt2.date_naive() } - fn show_date(&self) -> Option> { + fn show_date(self) -> Option> { let time = self.as_datetime().format("%A, %B %d %Y").to_string(); Span::styled(time, BOLD_STYLE).into() } - fn show_time(&self) -> Option> { - match self { - MessageTimeStamp::OriginServer(ms) => { - let time = millis_to_datetime(*ms).format("%T"); - let time = format!(" [{time}]"); - - Span::raw(time).into() - }, - MessageTimeStamp::LocalEcho => None, - } - } - - fn is_local_echo(&self) -> bool { - matches!(self, MessageTimeStamp::LocalEcho) - } - - pub fn as_millis(&self) -> Option { - match self { - MessageTimeStamp::OriginServer(ms) => MilliSecondsSinceUnixEpoch(*ms).into(), - MessageTimeStamp::LocalEcho => None, - } - } -} - -impl Ord for MessageTimeStamp { - fn cmp(&self, other: &Self) -> Ordering { - match (self, other) { - (MessageTimeStamp::OriginServer(_), MessageTimeStamp::LocalEcho) => Ordering::Less, - (MessageTimeStamp::OriginServer(a), MessageTimeStamp::OriginServer(b)) => a.cmp(b), - (MessageTimeStamp::LocalEcho, MessageTimeStamp::OriginServer(_)) => Ordering::Greater, - (MessageTimeStamp::LocalEcho, MessageTimeStamp::LocalEcho) => Ordering::Equal, - } - } -} - -impl PartialOrd for MessageTimeStamp { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} + fn show_time(self) -> Option> { + let time = self.as_datetime().format("%T"); + let time = format!(" [{time}]"); -impl From for MessageTimeStamp { - fn from(millis: UInt) -> Self { - MessageTimeStamp::OriginServer(millis) + Span::raw(time).into() } } impl From for MessageTimeStamp { fn from(millis: MilliSecondsSinceUnixEpoch) -> Self { - MessageTimeStamp::OriginServer(millis.0) + Self(millis) } } @@ -294,10 +266,7 @@ impl TryFrom<&MessageTimeStamp> for usize { type Error = TimeStampIntError; fn try_from(ts: &MessageTimeStamp) -> Result { - let n = match ts { - MessageTimeStamp::LocalEcho => 0, - MessageTimeStamp::OriginServer(u) => usize::try_from(u64::from(*u))?, - }; + let n = usize::try_from(u64::from(ts.0 .0))?; Ok(n) } @@ -307,14 +276,10 @@ impl TryFrom for MessageTimeStamp { type Error = TimeStampIntError; fn try_from(u: usize) -> Result { - if u == 0 { - Ok(MessageTimeStamp::LocalEcho) - } else { - let n = u64::try_from(u)?; - let n = UInt::try_from(n).map_err(TimeStampIntError::UIntError)?; + let n = u64::try_from(u)?; + let n = UInt::try_from(n).map_err(TimeStampIntError::UIntError)?; - Ok(MessageTimeStamp::from(n)) - } + Ok(MessageTimeStamp::from(MilliSecondsSinceUnixEpoch(n))) } } @@ -348,33 +313,30 @@ impl MessageCursor { pub fn from_cursor(cursor: &Cursor, thread: &Messages) -> Option { let ev_hash = cursor.get_x(); - let ev_term = OwnedEventId::try_from("$").ok()?; + let ev_term = OwnedEventId::try_from("$").ok()?.into(); let ts_start = MessageTimeStamp::try_from(cursor.get_y()).ok()?; - let start = (ts_start, ev_term); + let start = MessageKey { ts: ts_start, id: ev_term }; - for ((ts, event_id), _) in thread.range(&start..) { - if hash_event_id(event_id)? == ev_hash { - return Self::from((*ts, event_id.clone())).into(); + for (key, _) in thread.range(&start..) { + if hash_message_id(&key.id)? == ev_hash { + return Self::from(key.clone()).into(); } - if ts > &ts_start { + if key.ts > ts_start { break; } } // If we can't find the cursor, then go to the nearest timestamp. - thread - .range(start..) - .next() - .map(|((ts, ev), _)| Self::from((*ts, ev.clone()))) + thread.range(start..).next().map(|(key, _)| Self::from(key.clone())) } pub fn to_cursor(&self, thread: &Messages) -> Option { - let (ts, event_id) = self.to_key(thread)?; + let key = self.to_key(thread)?; - let y = usize::try_from(ts).ok()?; - let x = hash_event_id(event_id)?; + let y = usize::try_from(&key.ts).ok()?; + let x = hash_message_id(&key.id)?; Cursor::new(y, x).into() } @@ -429,19 +391,19 @@ pub enum MessageEvent { Original(Box), Redacted(OwnedEventId, Option), State(Box), - Local(OwnedEventId, Box), + Local(OwnedTransactionId, SendHandle, Box), } impl MessageEvent { - pub fn event_id(&self) -> &EventId { - match self { + pub fn event_id(&self) -> Option<&EventId> { + Some(match self { MessageEvent::EncryptedOriginal(ev) => ev.event_id.as_ref(), MessageEvent::EncryptedRedacted(ev) => ev.event_id.as_ref(), MessageEvent::Original(ev) => ev.event_id.as_ref(), MessageEvent::Redacted(event_id, _) => event_id.as_ref(), MessageEvent::State(ev) => ev.event_id(), - MessageEvent::Local(event_id, _) => event_id.as_ref(), - } + MessageEvent::Local(..) => return None, + }) } pub fn content(&self) -> Option<&RoomMessageEventContent> { @@ -451,7 +413,7 @@ impl MessageEvent { MessageEvent::EncryptedRedacted(_) => None, MessageEvent::Redacted(_, _) => None, MessageEvent::State(_) => None, - MessageEvent::Local(_, content) => Some(content), + MessageEvent::Local(_, _, content) => Some(content), } } @@ -471,7 +433,7 @@ impl MessageEvent { }, MessageEvent::Redacted(_, reason) => body_cow_reason(reason.as_deref()), MessageEvent::State(ev) => body_cow_state(ev), - MessageEvent::Local(_, content) => body_cow_content(content), + MessageEvent::Local(_, _, content) => body_cow_content(content), } } @@ -482,7 +444,7 @@ impl MessageEvent { MessageEvent::Original(ev) => &ev.content, MessageEvent::Redacted(_, _) => return None, MessageEvent::State(ev) => return Some(html_state(ev)), - MessageEvent::Local(_, content) => content, + MessageEvent::Local(_, _, content) => content, }; if let MessageType::Text(content) = &content.msgtype { @@ -502,7 +464,7 @@ impl MessageEvent { MessageEvent::EncryptedRedacted(_) => return, MessageEvent::Redacted(_, _) => return, MessageEvent::State(_) => return, - MessageEvent::Local(_, _) => return, + MessageEvent::Local(..) => return, MessageEvent::Original(ev) => { let event_id = ev.event_id.to_owned(); let reason = redaction_reason_event(redaction); @@ -857,7 +819,7 @@ impl Message { let content = match &self.event { MessageEvent::EncryptedOriginal(_) => return None, MessageEvent::EncryptedRedacted(_) => return None, - MessageEvent::Local(_, content) => content, + MessageEvent::Local(_, _, content) => content, MessageEvent::Original(ev) => &ev.content, MessageEvent::Redacted(_, _) => return None, MessageEvent::State(_) => return None, @@ -878,7 +840,7 @@ impl Message { let content = match &self.event { MessageEvent::EncryptedOriginal(_) => return None, MessageEvent::EncryptedRedacted(_) => return None, - MessageEvent::Local(_, content) => content, + MessageEvent::Local(_, _, content) => content, MessageEvent::Original(ev) => &ev.content, MessageEvent::Redacted(_, _) => return None, MessageEvent::State(_) => return None, @@ -902,7 +864,7 @@ impl Message { style = style.add_modifier(StyleModifier::REVERSED) } - if self.timestamp.is_local_echo() { + if matches!(self.event, MessageEvent::Local(..)) { style = style.add_modifier(StyleModifier::ITALIC); } @@ -923,7 +885,7 @@ impl Message { ) -> MessageFormatter<'a> { let orig = width; let date = match &prev { - Some(prev) if prev.timestamp.same_day(&self.timestamp) => None, + Some(prev) if prev.timestamp.same_day(self.timestamp) => None, _ => self.timestamp.show_date(), }; let user_gutter = settings.tunables.user_gutter_width; @@ -938,7 +900,7 @@ impl Message { let read = info .event_receipts .values() - .filter_map(|receipts| receipts.get(self.event.event_id())) + .filter_map(|receipts| self.event.event_id().and_then(|id| receipts.get(id))) .flat_map(|read| read.iter()) .map(|user_id| user_id.to_owned()) .collect(); @@ -1020,11 +982,12 @@ impl Message { } if settings.tunables.reaction_display { - let reactions = info.get_reactions(self.event.event_id()); + let reactions = + self.event.event_id().map(|id| info.get_reactions(id)).unwrap_or_default(); fmt.push_reactions(reactions, style, &mut text); } - if let Some(thread) = info.get_thread(Some(self.event.event_id())) { + if let Some(thread) = self.event.event_id().and_then(|id| info.get_thread(Some(id))) { fmt.push_thread_reply_count(thread.len(), &mut text); } @@ -1099,7 +1062,7 @@ impl Message { ) -> Option> { if let Some(prev) = prev { if self.sender == prev.sender && - self.timestamp.same_day(&prev.timestamp) && + self.timestamp.same_day(prev.timestamp) && !self.event.is_emote() { return None; diff --git a/src/tests.rs b/src/tests.rs index 9213d247..7523bc97 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -1,6 +1,7 @@ use std::path::PathBuf; use std::{collections::HashMap, iter::FromIterator as _}; +use matrix_sdk::ruma::MilliSecondsSinceUnixEpoch; use matrix_sdk::ruma::{ event_id, events::room::message::RoomMessageEventContent, @@ -20,6 +21,7 @@ use serde_json::{Map, Value}; use tokio::sync::mpsc::unbounded_channel; use url::Url; +use crate::message::MessageTimeStamp; use crate::{ base::{ChatStore, EventLocation, ProgramStore, RoomInfo}, config::{ @@ -36,13 +38,7 @@ use crate::{ UserDisplayStyle, UserDisplayTunables, }, - message::{ - Message, - MessageEvent, - MessageKey, - MessageTimeStamp::{LocalEcho, OriginServer}, - Messages, - }, + message::{Message, MessageEvent, MessageKey, Messages}, worker::Requester, }; @@ -63,11 +59,27 @@ lazy_static! { pub static ref MSG4_EVID: OwnedEventId = event_id!("$JP6qFV7WyXk5ZnexM3:example.com").to_owned(); pub static ref MSG5_EVID: OwnedEventId = EventId::new(server_name!("example.com")); - pub static ref MSG1_KEY: MessageKey = (LocalEcho, MSG1_EVID.clone()); - pub static ref MSG2_KEY: MessageKey = (OriginServer(UInt::new(1).unwrap()), MSG2_EVID.clone()); - pub static ref MSG3_KEY: MessageKey = (OriginServer(UInt::new(2).unwrap()), MSG3_EVID.clone()); - pub static ref MSG4_KEY: MessageKey = (OriginServer(UInt::new(2).unwrap()), MSG4_EVID.clone()); - pub static ref MSG5_KEY: MessageKey = (OriginServer(UInt::new(8).unwrap()), MSG5_EVID.clone()); + pub static ref MSG1_KEY: MessageKey = MessageKey { + // 2000-01-01T00:00:00 + ts: MessageTimeStamp(MilliSecondsSinceUnixEpoch(UInt::new(946681200).unwrap())), + id: MSG1_EVID.clone().into() + }; + pub static ref MSG2_KEY: MessageKey = MessageKey { + ts: MessageTimeStamp(MilliSecondsSinceUnixEpoch(UInt::new(1).unwrap())), + id: MSG2_EVID.clone().into() + }; + pub static ref MSG3_KEY: MessageKey = MessageKey { + ts: MessageTimeStamp(MilliSecondsSinceUnixEpoch(UInt::new(2).unwrap())), + id: MSG3_EVID.clone().into() + }; + pub static ref MSG4_KEY: MessageKey = MessageKey { + ts: MessageTimeStamp(MilliSecondsSinceUnixEpoch(UInt::new(2).unwrap())), + id: MSG4_EVID.clone().into() + }; + pub static ref MSG5_KEY: MessageKey = MessageKey { + ts: MessageTimeStamp(MilliSecondsSinceUnixEpoch(UInt::new(8).unwrap())), + id: MSG5_EVID.clone().into() + }; } pub fn user_style(user: &str) -> Style { @@ -79,13 +91,13 @@ pub fn mock_room1_message( sender: OwnedUserId, key: MessageKey, ) -> Message { - let timestamp = key.0.as_millis().unwrap(); - let event_id = key.1; + let timestamp = key.ts.0; + let event_id = key.id.as_origin().unwrap(); let event = serde_json::from_value(Value::Object(Map::from_iter([ ("type".to_owned(), Value::String("m.room.message".into())), ("content".to_owned(), serde_json::to_value(&content).unwrap()), - ("event_id".to_owned(), serde_json::to_value(&event_id).unwrap()), + ("event_id".to_owned(), serde_json::to_value(event_id).unwrap()), ("sender".to_owned(), serde_json::to_value(&sender).unwrap()), ("origin_server_ts".to_owned(), serde_json::to_value(timestamp).unwrap()), ("room_id".to_owned(), serde_json::to_value(&*TEST_ROOM1_ID).unwrap()), @@ -97,9 +109,8 @@ pub fn mock_room1_message( pub fn mock_message1() -> Message { let content = RoomMessageEventContent::text_plain("writhe"); - let content = MessageEvent::Local(MSG1_EVID.clone(), content.into()); - Message::new(content, TEST_USER1.clone(), MSG1_KEY.0) + mock_room1_message(content, TEST_USER1.clone(), MSG1_KEY.clone()) } pub fn mock_message2() -> Message { @@ -129,7 +140,7 @@ pub fn mock_message5() -> Message { pub fn mock_keys() -> HashMap { let mut keys = HashMap::new(); - keys.insert(MSG1_EVID.clone(), EventLocation::Message(None, MSG1_KEY.clone())); + keys.insert(MSG1_EVID.clone(), EventLocation::Message(None, MSG2_KEY.clone())); keys.insert(MSG2_EVID.clone(), EventLocation::Message(None, MSG2_KEY.clone())); keys.insert(MSG3_EVID.clone(), EventLocation::Message(None, MSG3_KEY.clone())); keys.insert(MSG4_EVID.clone(), EventLocation::Message(None, MSG4_KEY.clone())); diff --git a/src/windows/mod.rs b/src/windows/mod.rs index bb10415b..cf465489 100644 --- a/src/windows/mod.rs +++ b/src/windows/mod.rs @@ -1623,7 +1623,7 @@ impl Promptable for MemberItem { #[cfg(test)] mod tests { use super::*; - use matrix_sdk::ruma::{room_alias_id, server_name}; + use matrix_sdk::ruma::{room_alias_id, server_name, MilliSecondsSinceUnixEpoch}; #[derive(Debug, Eq, PartialEq)] struct TestRoomItem { @@ -1753,7 +1753,7 @@ mod tests { name: "Room 2", unread: UnreadInfo { unread: false, - latest: Some(MessageTimeStamp::OriginServer(40u32.into())), + latest: Some(MessageTimeStamp(MilliSecondsSinceUnixEpoch(40u32.into()))), }, invite: false, }; @@ -1765,7 +1765,7 @@ mod tests { name: "Room 3", unread: UnreadInfo { unread: false, - latest: Some(MessageTimeStamp::OriginServer(20u32.into())), + latest: Some(MessageTimeStamp(MilliSecondsSinceUnixEpoch(20u32.into()))), }, invite: false, }; diff --git a/src/windows/room/chat.rs b/src/windows/room/chat.rs index 7d26a1fe..0a1db27d 100644 --- a/src/windows/room/chat.rs +++ b/src/windows/room/chat.rs @@ -7,6 +7,7 @@ use std::path::{Path, PathBuf}; use edit::edit_with_builder as external_edit; use edit::Builder; +use matrix_sdk::send_queue::RoomSendQueueError; use matrix_sdk::EncryptionState; use modalkit::editing::store::RegisterError; use ratatui::style::{Color, Style}; @@ -28,8 +29,6 @@ use matrix_sdk::{ OriginalRoomMessageEvent, Relation, ReplyWithinThread, - RoomMessageEventContent, - TextMessageEventContent, }, OwnedEventId, OwnedRoomId, @@ -74,6 +73,7 @@ use modalkit::prelude::*; use crate::base::{ DownloadFlags, + EchoLocation, IambAction, IambBufferId, IambError, @@ -88,14 +88,7 @@ use crate::base::{ SendAction, }; -use crate::message::{ - text_to_message, - Message, - MessageEvent, - MessageKey, - MessageTimeStamp, - TreeGenState, -}; +use crate::message::{text_to_message, MessageEvent, MessageId, MessageKey, TreeGenState}; use crate::worker::Requester; use super::scrollback::{Scrollback, ScrollbackState}; @@ -343,7 +336,7 @@ impl ChatState { let ev = match &msg.event { MessageEvent::Original(ev) => &ev.content, - MessageEvent::Local(_, ev) => ev.deref(), + MessageEvent::Local(_, _, ev) => ev.deref(), _ => { let msg = "Cannot edit a redacted message"; let err = UIError::Failure(msg.into()); @@ -390,7 +383,14 @@ impl ChatState { MessageEvent::EncryptedOriginal(ev) => ev.event_id.clone(), MessageEvent::EncryptedRedacted(ev) => ev.event_id.clone(), MessageEvent::Original(ev) => ev.event_id.clone(), - MessageEvent::Local(event_id, _) => event_id.clone(), + MessageEvent::Local(..) => { + // XXX: Implement reactions for local echos + + let msg = "Cannot react to a local echo"; + let err = UIError::Failure(msg.into()); + + return Err(err); + }, MessageEvent::State(ev) => ev.event_id().to_owned(), MessageEvent::Redacted(_, _) => { let msg = "Cannot react to a redacted message"; @@ -409,7 +409,8 @@ impl ChatState { let reaction = Annotation::new(event_id, emoji); let msg = ReactionEventContent::new(reaction); - let _ = room.send(msg).await.map_err(IambError::from)?; + + room.send_queue().send(msg.into()).await.map_err(IambError::from)?; Ok(None) }, @@ -428,7 +429,22 @@ impl ChatState { MessageEvent::EncryptedOriginal(ev) => ev.event_id.clone(), MessageEvent::EncryptedRedacted(ev) => ev.event_id.clone(), MessageEvent::Original(ev) => ev.event_id.clone(), - MessageEvent::Local(event_id, _) => event_id.clone(), + MessageEvent::Local(_, handle, _) => { + let succeeded = handle + .abort() + .await + .map_err(RoomSendQueueError::from) + .map_err(IambError::from)?; + + if !succeeded { + let msg = "local echo was already sent; please retry"; + let err = UIError::Failure(msg.into()); + + return Err(err); + } + + return Ok(None); + }, MessageEvent::State(ev) => ev.event_id().to_owned(), MessageEvent::Redacted(_, _) => { let msg = "Cannot redact already redacted message"; @@ -491,7 +507,12 @@ impl ChatState { MessageEvent::EncryptedOriginal(ev) => ev.event_id.clone(), MessageEvent::EncryptedRedacted(ev) => ev.event_id.clone(), MessageEvent::Original(ev) => ev.event_id.clone(), - MessageEvent::Local(event_id, _) => event_id.clone(), + MessageEvent::Local(..) => { + let msg = "Cannot unreact to a local echo"; + let err = UIError::Failure(msg.into()); + + return Err(err); + }, MessageEvent::State(ev) => ev.event_id().to_owned(), MessageEvent::Redacted(_, _) => { let msg = "Cannot unreact to a redacted message"; @@ -539,9 +560,8 @@ impl ChatState { ) -> IambResult { let room = self.get_joined(&store.application.worker)?; let info = store.application.rooms.get_or_default(self.id().to_owned()); - let mut show_echo = true; - let (event_id, msg) = match act { + match act { SendAction::Submit | SendAction::SubmitFromEditor => { let msg = self.tbox.get(); @@ -564,13 +584,48 @@ impl ChatState { let mut msg = text_to_message(msg); - if let Some((_, event_id)) = &self.editing { + if let Some(key) = &self.editing { + let id = match &key.id { + MessageId::Origin(id) => id, + MessageId::Local(transaction_id) => { + match info.echo_keys.get(transaction_id) { + Some(EchoLocation::Replaced(id)) => id, + Some(EchoLocation::Message(thread, orig_key)) => { + let Some(MessageEvent::Local(_, handle, _)) = info + .get_thread(thread.as_deref()) + .and_then(|thread| thread.get(orig_key)) + .map(|msg| &msg.event) + else { + let msg = "local echo not found in store"; + return Err(UIError::Failure(msg.into())); + }; + + let succeeded = handle + .edit(msg.into()) + .await + .map_err(RoomSendQueueError::from) + .map_err(IambError::from)?; + + if !succeeded { + let msg = "local echo was already sent; please retry"; + return Err(UIError::Failure(msg.into())); + } + + self.reset(); + return Ok(None); + }, + None => { + let msg = "local echo not found in store"; + return Err(UIError::Failure(msg.into())); + }, + } + }, + }; + msg.relates_to = Some(Relation::Replacement(Replacement::new( - event_id.clone(), + id.to_owned(), msg.msgtype.clone().into(), ))); - - show_echo = false; } else if let Some(thread_root) = self.scrollback.thread() { if let Some(m) = self.get_reply_to(info) { msg = msg.make_for_thread(m, ReplyWithinThread::Yes, AddMentions::No); @@ -583,15 +638,10 @@ impl ChatState { msg = msg.make_reply_to(m, ForwardThread::Yes, AddMentions::No); } - // XXX: second parameter can be a locally unique transaction id. - // Useful for doing retries. - let resp = room.send(msg.clone()).await.map_err(IambError::from)?; - let event_id = resp.event_id; + room.send_queue().send(msg.into()).await.map_err(IambError::from)?; // Reset message bar state now that it's been sent. self.reset(); - - (event_id, msg) }, SendAction::Upload(file) => { let path = Path::new(file.as_str()); @@ -604,17 +654,10 @@ impl ChatState { .unwrap_or_else(|| Cow::from("Attachment")); let config = AttachmentConfig::new(); - let resp = room - .send_attachment(name.as_ref(), &mime, bytes, config) + room.send_queue() + .send_attachment(name.as_ref(), mime, bytes, config) .await .map_err(IambError::from)?; - - // Mock up the local echo message for the scrollback. - let msg = TextMessageEventContent::plain(format!("[Attached File: {name}]")); - let msg = MessageType::Text(msg); - let msg = RoomMessageEventContent::new(msg); - - (resp.event_id, msg) }, SendAction::UploadImage(width, height, bytes) => { // Convert to png because arboard does not give us the mime type. @@ -633,27 +676,11 @@ impl ChatState { let name = "Clipboard.png"; let config = AttachmentConfig::new(); - let resp = room - .send_attachment(name, &mime, bytes, config) + room.send_queue() + .send_attachment(name, mime, bytes, config) .await .map_err(IambError::from)?; - - // Mock up the local echo message for the scrollback. - let msg = TextMessageEventContent::plain(format!("[Attached File: {name}]")); - let msg = MessageType::Text(msg); - let msg = RoomMessageEventContent::new(msg); - - (resp.event_id, msg) }, - }; - - if show_echo { - let user = store.application.settings.profile.user_id.clone(); - let key = (MessageTimeStamp::LocalEcho, event_id.clone()); - let msg = MessageEvent::Local(event_id, msg.into()); - let msg = Message::new(msg, user, MessageTimeStamp::LocalEcho); - let thread = self.scrollback.get_thread_mut(info); - thread.insert(key, msg); } // Jump to the end of the scrollback to show the message. diff --git a/src/windows/room/scrollback.rs b/src/windows/room/scrollback.rs index 02340573..2c3a2c70 100644 --- a/src/windows/room/scrollback.rs +++ b/src/windows/room/scrollback.rs @@ -1035,9 +1035,13 @@ impl Promptable for ScrollbackState { let err = EditError::Failure(msg.into()); Err(err) } else { - let root = key.1.clone(); + let Some(root) = key.id.as_origin() else { + let msg = "Cannot create thread for local echo."; + let err = EditError::Failure(msg.into()); + return Err(err); + }; let room_id = self.room_id.clone(); - let id = IambId::Room(room_id, Some(root)); + let id = IambId::Room(room_id, Some(root.to_owned())); let open = WindowAction::Switch(OpenTarget::Application(id)); Ok(vec![(open.into(), ctx.clone())]) } @@ -1388,8 +1392,8 @@ impl StatefulWidget for Scrollback<'_> { let _ = lines.drain(..n); } - if let Some(((ts, event_id), row, _, _)) = lines.first() { - state.viewctx.corner.timestamp = Some((*ts, event_id.clone())); + if let Some((key, row, _, _)) = lines.first() { + state.viewctx.corner.timestamp = Some((*key).clone()); state.viewctx.corner.text_row = *row; } @@ -1397,7 +1401,7 @@ impl StatefulWidget for Scrollback<'_> { let x = area.left(); let mut image_previews = vec![]; - for ((_, _), _, txt, line_preview) in lines.into_iter() { + for (_, _, txt, line_preview) in lines.into_iter() { let _ = buf.set_line(x, y, &txt, area.width); if let Some((backend, msg_x, _)) = line_preview { image_previews.push((x + msg_x, y, backend)); @@ -1423,8 +1427,12 @@ impl StatefulWidget for Scrollback<'_> { state.cursor.timestamp.is_none() { // If the cursor is at the last message, then update the read marker. - if let Some((k, _)) = thread.last_key_value() { - info.set_receipt(thread.1.clone(), settings.profile.user_id.clone(), k.1.clone()); + if let Some(event_id) = thread.iter().rev().find_map(|(key, _)| key.id.as_origin()) { + info.set_receipt( + thread.1.clone(), + settings.profile.user_id.clone(), + event_id.to_owned(), + ); } } diff --git a/src/worker.rs b/src/worker.rs index 01fd4055..9d701f68 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -13,6 +13,10 @@ use std::time::{Duration, Instant}; use futures::{stream::FuturesUnordered, StreamExt}; use gethostname::gethostname; +use matrix_sdk::ruma::events::relation::Thread; +use matrix_sdk::ruma::events::room::message::Relation; +use matrix_sdk::ruma::events::AnyMessageLikeEventContent; +use matrix_sdk::send_queue::{LocalEcho, LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate}; use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; use tokio::sync::Semaphore; use tokio::task::JoinHandle; @@ -88,7 +92,8 @@ use matrix_sdk::{ use modalkit::errors::UIError; use modalkit::prelude::{EditInfo, InfoMessage}; -use crate::base::MessageNeed; +use crate::base::{EchoLocation, MessageNeed}; +use crate::message::{Message, MessageEvent, MessageId, MessageKey}; use crate::notifications::register_notifications; use crate::{ base::{ @@ -556,6 +561,119 @@ async fn send_receipts_forever(client: &Client, store: &AsyncProgramStore) { } } +fn insert_local_echo( + own_user_id: OwnedUserId, + info: &mut RoomInfo, + echo: LocalEcho, +) -> Result<(), serde_json::Error> { + let LocalEcho { transaction_id, content } = echo; + + match content { + LocalEchoContent::Event { serialized_event, send_handle, .. } => { + let content = serialized_event.deserialize()?; + let AnyMessageLikeEventContent::RoomMessage(msg) = content else { + // XXX: Handle other event types + return Ok(()); + }; + + let thread = match msg.relates_to.as_ref() { + Some(Relation::Replacement(..)) => { + // XXX: Show echo on edited message + return Ok(()); + }, + Some(Relation::Thread(Thread { event_id, .. })) => Some(event_id.to_owned()), + _ => None, + }; + + let ts = send_handle.created_at.into(); + let key = MessageKey { ts, id: MessageId::Local(transaction_id.clone()) }; + let msg = MessageEvent::Local(transaction_id.clone(), send_handle, msg.into()); + let msg = Message::new(msg, own_user_id, ts); + + info.echo_keys + .insert(transaction_id, EchoLocation::Message(thread.clone(), key.clone())); + + let thread = info.get_thread_mut(thread); + thread.insert(key, msg); + }, + LocalEchoContent::React { .. } => { + // XXX: Handle reactions to local echos + }, + } + Ok(()) +} + +async fn subscribe_sendqueue_forever(client: &Client, store: &AsyncProgramStore) { + let own_user_id = client.user_id().unwrap(); + let mut receiver = client.send_queue().subscribe(); + + // load unsent requests + if let Ok(room_echos) = client.send_queue().local_echoes().await { + let mut locked = store.lock().await; + for (room_id, echos) in room_echos { + let info = locked.application.get_room_info(room_id); + for echo in echos { + let _ = insert_local_echo(own_user_id.to_owned(), info, echo); + } + } + } + + while let Ok(SendQueueUpdate { room_id, update }) = receiver.recv().await { + let mut locked = store.lock().await; + let info = locked.application.get_room_info(room_id); + match update { + RoomSendQueueUpdate::NewLocalEvent(echo) => { + let _ = insert_local_echo(own_user_id.to_owned(), info, echo); + }, + RoomSendQueueUpdate::ReplacedLocalEvent { transaction_id, new_content } => { + let Some(EchoLocation::Message(thread, key)) = + info.echo_keys.get(&transaction_id).cloned() + else { + continue; + }; + + let Ok(content) = new_content.deserialize() else { + continue; + }; + let AnyMessageLikeEventContent::RoomMessage(new_content) = content else { + // XXX: Handle other event types + continue; + }; + + let Some(msg) = info.get_thread_mut(thread).get_mut(&key) else { + continue; + }; + + let MessageEvent::Local(_, _, msg) = &mut msg.event else { + continue; + }; + + *msg = new_content.into(); + }, + + RoomSendQueueUpdate::SendError { .. } => { + // XXX: Show the error to the user + }, + RoomSendQueueUpdate::CancelledLocalEvent { transaction_id } => { + info.echo_keys.remove(&transaction_id); + }, + RoomSendQueueUpdate::SentEvent { transaction_id, event_id } => { + if let Some(location) = info.echo_keys.get_mut(&transaction_id) { + let location = std::mem::replace(location, EchoLocation::Replaced(event_id)); + + if let EchoLocation::Message(thread, key) = location { + info.get_thread_mut(thread).remove(&key); + } + } + }, + + RoomSendQueueUpdate::RetryEvent { .. } | RoomSendQueueUpdate::MediaUpload { .. } => { + // Ignore these events + }, + } + } +} + pub async fn do_first_sync(client: &Client, store: &AsyncProgramStore) -> Result<(), MatrixError> { // Perform an initial, lazily-loaded sync. let mut room = RoomEventFilter::default(); @@ -571,6 +689,8 @@ pub async fn do_first_sync(client: &Client, store: &AsyncProgramStore) -> Result client.sync_once(settings).await?; + client.send_queue().respawn_tasks_for_rooms_with_unsent_requests().await; + // Populate sync_info with our initial set of rooms/dms/spaces. refresh_rooms(client, store).await; @@ -1260,7 +1380,8 @@ impl ClientWorker { let rcpt = send_receipts_forever(&client, &store); let room = refresh_rooms_forever(&client, &store); let notifications = register_notifications(&client, &settings, &store); - let ((), (), (), ()) = tokio::join!(load, rcpt, room, notifications); + let sendqueue = subscribe_sendqueue_forever(&client, &store); + let ((), (), (), (), ()) = tokio::join!(load, rcpt, room, notifications, sendqueue); } }) .into();