Kordophone/kordophoned/src/daemon/mod.rs

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<T> = Result<T, Box<dyn Error + Send + Sync>>;

struct DatabaseTokenStore {
    database: Arc<Mutex<Database>>,
}

#[async_trait]
impl TokenStore for DatabaseTokenStore {
    async fn get_token(&mut self) -> Option<JwtToken> {
        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>,
    event_receiver: Receiver<Event>,

    signal_receiver: Option<Receiver<Signal>>,
    signal_sender: Sender<Signal>,

    version: String,
    database: Arc<Mutex<Database>>,
    runtime: tokio::runtime::Runtime,
}

impl Daemon {
    pub fn new() -> Result<Self> {
        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();
            },
        }
    }

    /// Panics if the signal receiver has already been taken. 
    pub fn obtain_signal_receiver(&mut self) -> Receiver<Signal> { 
        self.signal_receiver.take().unwrap()
    }

    async fn get_conversations(&mut self) -> Vec<Conversation> {
        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<String>) -> Vec<Message> {
        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<Mutex<Database>>, signal_sender: &Sender<Signal>) -> 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<kordophone_db::models::Conversation> = 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<Mutex<Database>>, signal_sender: &Sender<Signal>, 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<String> {
            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<kordophone_db::models::Message> = 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<Settings> {
        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<HTTPAPIClient<DatabaseTokenStore>> {
        Self::get_client_impl(&mut self.database).await
    }

    async fn get_client_impl(database: &mut Arc<Mutex<Database>>) -> Result<HTTPAPIClient<DatabaseTokenStore>> {
        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(),

            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")
        }
    }
}