pub mod settings; use settings::Settings; use settings::keys as SettingsKey; 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, AuthenticationStore, }; #[derive(Debug, Error)] pub enum DaemonError { #[error("Client Not Configured")] ClientNotConfigured, } pub type DaemonResult = Result>; struct DatabaseAuthenticationStore { database: Arc>, } #[async_trait] impl AuthenticationStore for DatabaseAuthenticationStore { async fn get_credentials(&mut self) -> Option { self.database.lock().await.with_settings(|settings| { let username: Option = settings.get::(SettingsKey::USERNAME) .unwrap_or_else(|e| { log::warn!("error getting username from database: {}", e); None }); // TODO: This would be the point where we map from credential item to password. let password: String = settings.get::(SettingsKey::CREDENTIAL_ITEM) .unwrap_or_else(|e| { log::warn!("error getting password from database: {}", e); None }) .unwrap_or_else(|| { log::warn!("warning: no password in database, [DEBUG] using default password"); "test".to_string() }); if username.is_none() { log::warn!("Username not present in database"); } match (username, password) { (Some(username), password) => Some(Credentials { username, password }), _ => None, } }).await } async fn get_token(&mut self) -> Option { self.database.lock().await .with_settings(|settings| settings.get::(SettingsKey::TOKEN).unwrap_or_default()).await } async fn set_token(&mut self, token: JwtToken) { self.database.lock().await .with_settings(|settings| settings.put(SettingsKey::TOKEN, &token)).await.unwrap_or_else(|e| { log::error!("Failed to set token: {}", e); }); } } 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::SyncConversation(conversation_id, 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_impl(&mut db_clone, &signal_sender, conversation_id).await; if let Err(e) = result { log::error!("Error handling sync event: {}", e); } }); 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(); }, Event::DeleteAllConversations(reply) => { self.delete_all_conversations().await .unwrap_or_else(|e| { log::error!("Failed to delete all conversations: {}", e); }); reply.send(()).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(kordophone_db::models::Conversation::from) .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?; // Sync individual conversation. Self::sync_conversation_impl(database, signal_sender, conversation_id).await?; } // Send conversations updated signal. signal_sender.send(Signal::ConversationsUpdated).await?; log::info!(target: target::SYNC, "Synchronized {} conversations", num_conversations); Ok(()) } async fn sync_conversation_impl(database: &mut Arc>, signal_sender: &Sender, conversation_id: String) -> Result<()> { log::info!(target: target::SYNC, "Starting conversation sync for {}", conversation_id); let mut client = Self::get_client_impl(database).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(kordophone_db::models::Message::from) .collect(); // Insert each message let num_messages = db_messages.len(); log::debug!(target: target::SYNC, "Inserting {} messages for conversation {}", num_messages, &conversation_id); database.with_repository(|r| r.insert_messages(&conversation_id, db_messages)).await?; // Send messages updated signal, if we actually inserted any messages. if num_messages > 0 { signal_sender.send(Signal::MessagesUpdated(conversation_id.clone())).await?; } log::info!(target: target::SYNC, "Synchronized {} messages for conversation {}", num_messages, &conversation_id); Ok(()) } async fn get_settings(&mut self) -> Result { let settings = self.database.with_settings(Settings::from_db).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(Settings::from_db).await?; let server_url = settings.server_url .ok_or(DaemonError::ClientNotConfigured)?; let client = HTTPAPIClient::new( server_url.parse().unwrap(), DatabaseAuthenticationStore { database: database.clone() } ); Ok(client) } async fn delete_all_conversations(&mut self) -> Result<()> { self.database.with_repository(|r| -> Result<()> { r.delete_all_conversations()?; r.delete_all_messages()?; Ok(()) }).await?; self.signal_sender.send(Signal::ConversationsUpdated).await?; Ok(()) } 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") } } }