buzzert/Kordophone
Private
Public Access
1
0
Fork 0
Code Packages Activity

daemon: incorporate update monitor in daemon activities

Browse Source
This commit is contained in:
James Magahern
2025-05-01 20:36:43 -07:00
parent 1c2f09e81b
commit 2314713bb4
8 changed files with 196 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
kordophoned/src/daemon/events.rs
View File

@@ -9,6 +9,9 @@ pub enum Event {
/// Get the version of the daemon.
GetVersion(Reply<String>),
/// Asynchronous event for syncing the conversation list with the server.
SyncConversationList(Reply<()>),
/// Asynchronous event for syncing all conversations with the server.
SyncAllConversations(Reply<()>),

61
kordophoned/src/daemon/mod.rs
View File

@@ -30,6 +30,9 @@ use kordophone::api::{
AuthenticationStore,
};
mod update_monitor;
use update_monitor::UpdateMonitor;
#[derive(Debug, Error)]
pub enum DaemonError {
#[error("Client Not Configured")]
@@ -87,11 +90,11 @@ impl AuthenticationStore for DatabaseAuthenticationStore {
}
}
mod target {
pub mod target {
pub static SYNC: &str = "sync";
pub static EVENT: &str = "event";
pub static UPDATES: &str = "updates";
}
pub struct Daemon {
pub event_sender: Sender<Event>,
event_receiver: Receiver<Event>,
@@ -139,6 +142,13 @@ impl Daemon {
log::info!("Starting daemon version {}", self.version);
log::debug!("Debug logging enabled.");
let mut update_monitor = UpdateMonitor::new(self.database.clone(), self.event_sender.clone());
tokio::spawn(async move {
log::info!(target: target::UPDATES, "Starting update monitor");
update_monitor.run().await; // should run indefinitely
});
while let Some(event) = self.event_receiver.recv().await {
log::debug!(target: target::EVENT, "Received event: {:?}", event);
self.handle_event(event).await;
@@ -151,6 +161,20 @@ impl Daemon {
reply.send(self.version.clone()).unwrap();
},
Event::SyncConversationList(reply) => {
let mut db_clone = self.database.clone();
let signal_sender = self.signal_sender.clone();
self.runtime.spawn(async move {
let result = Self::sync_conversation_list(&mut db_clone, &signal_sender).await;
if let Err(e) = result {
log::error!("Error handling sync event: {}", e);
}
});
// This is a background operation, so return right away.
reply.send(()).unwrap();
},
Event::SyncAllConversations(reply) => {
let mut db_clone = self.database.clone();
let signal_sender = self.signal_sender.clone();
@@ -231,8 +255,32 @@ impl Daemon {
self.database.lock().await.with_repository(|r| r.get_messages_for_conversation(&conversation_id).unwrap()).await
}
async fn sync_conversation_list(database: &mut Arc<Mutex<Database>>, signal_sender: &Sender<Signal>) -> Result<()> {
log::info!(target: target::SYNC, "Starting list conversation sync");
let mut client = Self::get_client_impl(database).await?;
// Fetch conversations from server
let fetched_conversations = client.get_conversations().await?;
let db_conversations: Vec<kordophone_db::models::Conversation> = fetched_conversations.into_iter()
.map(kordophone_db::models::Conversation::from)
.collect();
// Insert each conversation
let num_conversations = db_conversations.len();
for conversation in db_conversations {
database.with_repository(|r| r.insert_conversation(conversation)).await?;
}
// Send conversations updated signal
signal_sender.send(Signal::ConversationsUpdated).await?;
log::info!(target: target::SYNC, "Synchronized {} conversations", num_conversations);
Ok(())
}
async fn sync_all_conversations_impl(database: &mut Arc<Mutex<Database>>, signal_sender: &Sender<Signal>) -> Result<()> {
log::info!(target: target::SYNC, "Starting conversation sync");
log::info!(target: target::SYNC, "Starting full conversation sync");
let mut client = Self::get_client_impl(database).await?;
@@ -266,6 +314,13 @@ impl Daemon {
let mut client = Self::get_client_impl(database).await?;
// Check if conversation exists in database.
let conversation = database.with_repository(|r| r.get_conversation_by_guid(&conversation_id)).await?;
if conversation.is_none() {
// If the conversation doesn't exist, first do a conversation list sync.
Self::sync_conversation_list(database, signal_sender).await?;
}
// Fetch and sync messages for this conversation
let last_message_id = database.with_repository(|r| -> Option<String> {
r.get_last_message_for_conversation(&conversation_id)

98
kordophoned/src/daemon/update_monitor.rs Normal file
View File

@@ -0,0 +1,98 @@
use crate::daemon::{
Daemon,
DaemonResult,
events::{Event, Reply},
target,
};
use kordophone::APIInterface;
use kordophone::api::event_socket::EventSocket;
use kordophone::model::event::Event as UpdateEvent;
use kordophone_db::database::Database;
use tokio::sync::mpsc::Sender;
use std::sync::Arc;
use tokio::sync::Mutex;
pub struct UpdateMonitor {
database: Arc<Mutex<Database>>,
event_sender: Sender<Event>,
}
impl UpdateMonitor {
pub fn new(database: Arc<Mutex<Database>>, event_sender: Sender<Event>) -> Self {
Self { database, event_sender }
}
pub async fn send_event<T>(
&self,
make_event: impl FnOnce(Reply<T>) -> Event,
) -> DaemonResult<T> {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
self.event_sender.send(make_event(reply_tx))
.await
.map_err(|_| "Failed to send event")?;
reply_rx.await.map_err(|_| "Failed to receive reply".into())
}
async fn handle_update(&mut self, update: UpdateEvent) {
match update {
UpdateEvent::ConversationChanged(conversation) => {
log::info!(target: target::UPDATES, "Conversation changed: {:?}", conversation);
log::info!(target: target::UPDATES, "Triggering conversation list sync");
self.send_event(Event::SyncConversationList).await
.unwrap_or_else(|e| {
log::error!("Failed to send daemon event: {}", e);
});
}
UpdateEvent::MessageReceived(conversation, message) => {
log::info!(target: target::UPDATES, "Message received: msgid:{:?}, convid:{:?}", message.guid, conversation.guid);
log::info!(target: target::UPDATES, "Triggering message sync 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);
});
}
}
}
pub async fn run(&mut self) {
use futures_util::stream::StreamExt;
log::info!(target: target::UPDATES, "Starting update monitor");
loop {
log::debug!(target: target::UPDATES, "Creating client");
let mut client = match Daemon::get_client_impl(&mut self.database).await {
Ok(client) => client,
Err(e) => {
log::error!("Failed to get client: {}", e);
log::warn!("Retrying in 5 seconds...");
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
continue;
}
};
log::debug!(target: target::UPDATES, "Opening event socket");
let socket = match client.open_event_socket().await {
Ok(events) => events,
Err(e) => {
log::warn!("Failed to open event socket: {}", e);
log::warn!("Retrying in 5 seconds...");
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
continue;
}
};
log::debug!(target: target::UPDATES, "Starting event stream");
let mut event_stream = socket.events().await;
while let Some(Ok(event)) = event_stream.next().await {
self.handle_update(event).await;
}
}
}
}