pub mod settings; use settings::Settings; pub mod events; use events::*; pub mod signals; use signals::*; use anyhow::Result; use directories::ProjectDirs; use std::error::Error; use std::path::PathBuf; use thiserror::Error; use tokio::sync::mpsc::{Sender, Receiver}; use std::sync::Arc; use tokio::sync::Mutex; use async_trait::async_trait; use kordophone_db::{ database::{Database, DatabaseAccess}, models::{Conversation, Message}, }; use kordophone::model::JwtToken; use kordophone::api::{ http_client::{Credentials, HTTPAPIClient}, APIInterface, TokenStore, }; #[derive(Debug, Error)] pub enum DaemonError { #[error("Client Not Configured")] ClientNotConfigured, } pub type DaemonResult = Result>; struct DatabaseTokenStore { database: Arc>, } #[async_trait] impl TokenStore for DatabaseTokenStore { async fn get_token(&mut self) -> Option { self.database.lock().await.get_token().await } async fn set_token(&mut self, token: JwtToken) { self.database.lock().await.set_token(token).await; } } mod target { pub static SYNC: &str = "sync"; pub static EVENT: &str = "event"; } pub struct Daemon { pub event_sender: Sender, event_receiver: Receiver, signal_receiver: Option>, signal_sender: Sender, version: String, database: Arc>, runtime: tokio::runtime::Runtime, } impl Daemon { pub fn new() -> Result { let database_path = Self::get_database_path(); log::info!("Database path: {}", database_path.display()); // Create the database directory if it doesn't exist let database_dir = database_path.parent().unwrap(); std::fs::create_dir_all(database_dir)?; // Create event channels let (event_sender, event_receiver) = tokio::sync::mpsc::channel(100); let (signal_sender, signal_receiver) = tokio::sync::mpsc::channel(100); // Create background task runtime let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() .build() .unwrap(); let database_impl = Database::new(&database_path.to_string_lossy())?; let database = Arc::new(Mutex::new(database_impl)); Ok(Self { version: "0.1.0".to_string(), database, event_receiver, event_sender, signal_receiver: Some(signal_receiver), signal_sender, runtime }) } pub async fn run(&mut self) { log::info!("Starting daemon version {}", self.version); log::debug!("Debug logging enabled."); while let Some(event) = self.event_receiver.recv().await { log::debug!(target: target::EVENT, "Received event: {:?}", event); self.handle_event(event).await; } } async fn handle_event(&mut self, event: Event) { match event { Event::GetVersion(reply) => { reply.send(self.version.clone()).unwrap(); }, Event::SyncAllConversations(reply) => { let mut db_clone = self.database.clone(); let signal_sender = self.signal_sender.clone(); self.runtime.spawn(async move { let result = Self::sync_all_conversations_impl(&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::GetAllConversations(reply) => { let conversations = self.get_conversations().await; reply.send(conversations).unwrap(); }, Event::GetAllSettings(reply) => { let settings = self.get_settings().await .unwrap_or_else(|e| { log::error!("Failed to get settings: {:#?}", e); Settings::default() }); reply.send(settings).unwrap(); }, Event::UpdateSettings(settings, reply) => { self.update_settings(&settings).await .unwrap_or_else(|e| { log::error!("Failed to update settings: {}", e); }); reply.send(()).unwrap(); }, Event::GetMessages(conversation_id, last_message_id, reply) => { let messages = self.get_messages(conversation_id, last_message_id).await; reply.send(messages).unwrap(); }, } } /// Panics if the signal receiver has already been taken. pub fn obtain_signal_receiver(&mut self) -> Receiver { self.signal_receiver.take().unwrap() } async fn get_conversations(&mut self) -> Vec { self.database.lock().await.with_repository(|r| r.all_conversations().unwrap()).await } async fn get_messages(&mut self, conversation_id: String, last_message_id: Option) -> Vec { self.database.lock().await.with_repository(|r| r.get_messages_for_conversation(&conversation_id).unwrap()).await } async fn sync_all_conversations_impl(database: &mut Arc>, signal_sender: &Sender) -> Result<()> { log::info!(target: target::SYNC, "Starting 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 = fetched_conversations.into_iter() .map(|c| kordophone_db::models::Conversation::from(c)) .collect(); // Process each conversation let num_conversations = db_conversations.len(); for conversation in db_conversations { let conversation_id = conversation.guid.clone(); // Insert the conversation database.with_repository(|r| r.insert_conversation(conversation)).await?; // Fetch and sync messages for this conversation let last_message_id = database.with_repository(|r| -> Option { r.get_last_message_for_conversation(&conversation_id) .unwrap_or(None) .map(|m| m.id) }).await; log::debug!(target: target::SYNC, "Fetching messages for conversation {}", conversation_id); log::debug!(target: target::SYNC, "Last message id: {:?}", last_message_id); let messages = client.get_messages(&conversation_id, None, None, last_message_id).await?; let db_messages: Vec = messages.into_iter() .map(|m| kordophone_db::models::Message::from(m)) .collect(); // Insert each message log::debug!(target: target::SYNC, "Inserting {} messages for conversation {}", db_messages.len(), conversation_id); database.with_repository(|r| r.insert_messages(&conversation_id, db_messages)).await?; } // Send conversations updated signal. signal_sender.send(Signal::ConversationsUpdated).await?; log::info!(target: target::SYNC, "Synchronized {} conversations", num_conversations); Ok(()) } async fn get_settings(&mut self) -> Result { let settings = self.database.with_settings(|s| Settings::from_db(s) ).await?; Ok(settings) } async fn update_settings(&mut self, settings: &Settings) -> Result<()> { self.database.with_settings(|s| settings.save(s)).await } async fn get_client(&mut self) -> Result> { Self::get_client_impl(&mut self.database).await } async fn get_client_impl(database: &mut Arc>) -> Result> { let settings = database.with_settings(|s| Settings::from_db(s) ).await?; let server_url = settings.server_url .ok_or(DaemonError::ClientNotConfigured)?; let client = HTTPAPIClient::new( server_url.parse().unwrap(), match (settings.username, settings.credential_item) { (Some(username), Some(password)) => Some( Credentials { username, password, } ), _ => None, }, DatabaseTokenStore { database: database.clone() } ); Ok(client) } fn get_database_path() -> PathBuf { if let Some(proj_dirs) = ProjectDirs::from("net", "buzzert", "kordophonecd") { let data_dir = proj_dirs.data_dir(); data_dir.join("database.db") } else { // Fallback to a local path if we can't get the system directories PathBuf::from("database.db") } } }