diff --git a/kordophoned/src/daemon/update_monitor.rs b/kordophoned/src/daemon/update_monitor.rs index 7632540..1888ad9 100644 --- a/kordophoned/src/daemon/update_monitor.rs +++ b/kordophoned/src/daemon/update_monitor.rs @@ -16,15 +16,22 @@ use kordophone_db::database::DatabaseAccess; use tokio::sync::mpsc::Sender; use std::sync::Arc; use tokio::sync::Mutex; +use std::collections::HashMap; +use std::time::{Duration, Instant}; pub struct UpdateMonitor { database: Arc>, event_sender: Sender, + last_sync_times: HashMap, } impl UpdateMonitor { pub fn new(database: Arc>, event_sender: Sender) -> Self { - Self { database, event_sender } + Self { + database, + event_sender, + last_sync_times: HashMap::new(), + } } pub async fn send_event( @@ -44,27 +51,33 @@ impl UpdateMonitor { UpdateEvent::ConversationChanged(conversation) => { log::info!(target: target::UPDATES, "Conversation changed: {:?}", conversation); - // Weird. We can get in a loop because calling getMessages triggers a conversation changed - // event for some reason. Check to see if the change event says the last message id is the same - // as the last message id in the database. If so, skip the sync. + // Check if we've synced this conversation recently (within 5 seconds) + // This is currently a hack/workaround to prevent an infinite loop of sync events, because for some reason + // imagent will post a conversation changed notification when we call getMessages. + if let Some(last_sync) = self.last_sync_times.get(&conversation.guid) { + if last_sync.elapsed() < Duration::from_secs(5) { + log::info!(target: target::UPDATES, "Skipping sync for conversation id: {}. Last sync was {} seconds ago.", + conversation.guid, last_sync.elapsed().as_secs_f64()); + return; + } + } + + // This is the non-hacky path once we can reason about chat items with associatedMessageGUIDs (e.g., reactions). let last_message = self.database.with_repository(|r| r.get_last_message_for_conversation(&conversation.guid)).await.unwrap_or_default(); - let should_sync_conversation = match (&last_message, &conversation.last_message) { + match (&last_message, &conversation.last_message) { (Some(message), Some(conversation_message)) => { if message.id == conversation_message.guid { - false - } else { - true + log::info!(target: target::UPDATES, "Skipping sync for conversation id: {}. We already have this message.", conversation.guid); + return; } } - _ => true + _ => {} }; - if !should_sync_conversation { - log::info!(target: target::UPDATES, "Skipping sync for conversation id: {}. We already have this message.", conversation.guid); - return; - } + // Update the last sync time and proceed with sync + self.last_sync_times.insert(conversation.guid.clone(), Instant::now()); - log::info!(target: target::UPDATES, "Syncing new messages for conversation id: {}, last message: {:?}", conversation.guid, last_message); + log::info!(target: target::UPDATES, "Syncing new messages for conversation id: {}", conversation.guid); self.send_event(|r| Event::SyncConversation(conversation.guid, r)).await .unwrap_or_else(|e| { log::error!("Failed to send daemon event: {}", e);