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

Started working on attachment store

Browse Source
This commit is contained in:
James Magahern
2025-05-15 20:11:10 -07:00
parent 77177e07aa
commit 0d4c2e5104
10 changed files with 721 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,6 @@
pub mod settings;
use settings::Settings;
use settings::keys as SettingsKey;
use settings::Settings;
pub mod events;
use events::*;
@@ -11,13 +11,13 @@ use signals::*;
use anyhow::Result;
use directories::ProjectDirs;
use std::error::Error;
use std::path::PathBuf;
use std::collections::HashMap;
use std::error::Error;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use thiserror::Error;
use tokio::sync::mpsc::{Sender, Receiver};
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::Mutex;
use uuid::Uuid;
@@ -26,8 +26,8 @@ use kordophone_db::{
models::{Conversation, Message},
};
use kordophone::api::APIInterface;
use kordophone::api::http_client::HTTPAPIClient;
use kordophone::api::APIInterface;
use kordophone::model::outgoing_message::OutgoingMessage;
use kordophone::model::ConversationID;
@@ -38,8 +38,12 @@ mod auth_store;
use auth_store::DatabaseAuthenticationStore;
mod post_office;
use post_office::PostOffice;
use post_office::Event as PostOfficeEvent;
use post_office::PostOffice;
mod attachment_store;
pub use attachment_store::Attachment;
use attachment_store::AttachmentStore;
#[derive(Debug, Error)]
pub enum DaemonError {
@@ -49,6 +53,8 @@ pub enum DaemonError {
pub type DaemonResult<T> = Result<T, Box<dyn Error + Send + Sync>>;
type DaemonClient = HTTPAPIClient<DatabaseAuthenticationStore>;
pub mod target {
pub static SYNC: &str = "sync";
pub static EVENT: &str = "event";
@@ -68,6 +74,8 @@ pub struct Daemon {
outgoing_messages: HashMap<ConversationID, Vec<OutgoingMessage>>,
attachment_store: AttachmentStore,
version: String,
database: Arc<Mutex<Database>>,
runtime: tokio::runtime::Runtime,
@@ -87,7 +95,7 @@ impl Daemon {
let (signal_sender, signal_receiver) = tokio::sync::mpsc::channel(100);
let (post_office_sink, post_office_source) = tokio::sync::mpsc::channel(100);
// Create background task runtime
// Create background task runtime
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
@@ -95,17 +103,22 @@ impl Daemon {
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,
let data_path = Self::get_data_dir().expect("Unable to get data path");
let attachment_store = AttachmentStore::new(&data_path);
Ok(Self {
version: "0.1.0".to_string(),
database,
event_receiver,
event_sender,
signal_receiver: Some(signal_receiver),
signal_sender,
signal_sender,
post_office_sink,
post_office_source: Some(post_office_source),
outgoing_messages: HashMap::new(),
runtime
attachment_store: attachment_store,
runtime,
})
}
@@ -114,7 +127,8 @@ impl Daemon {
log::debug!("Debug logging enabled.");
// Update monitor
let mut update_monitor = UpdateMonitor::new(self.database.clone(), self.event_sender.clone());
let mut update_monitor =
UpdateMonitor::new(self.database.clone(), self.event_sender.clone());
tokio::spawn(async move {
update_monitor.run().await; // should run indefinitely
});
@@ -125,7 +139,10 @@ impl Daemon {
let event_sender = self.event_sender.clone();
let post_office_source = self.post_office_source.take().unwrap();
tokio::spawn(async move {
let mut post_office = PostOffice::new(post_office_source, event_sender, async move || Self::get_client_impl(&mut database).await );
let mut post_office =
PostOffice::new(post_office_source, event_sender, async move || {
Self::get_client_impl(&mut database).await
});
post_office.run().await;
});
}
@@ -140,7 +157,7 @@ impl Daemon {
match event {
Event::GetVersion(reply) => {
reply.send(self.version.clone()).unwrap();
},
}
Event::SyncConversationList(reply) => {
let mut db_clone = self.database.clone();
@@ -152,132 +169,166 @@ impl Daemon {
}
});
// This is a background operation, so return right away.
// 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();
self.runtime.spawn(async move {
let result = Self::sync_all_conversations_impl(&mut db_clone, &signal_sender).await;
let result =
Self::sync_all_conversations_impl(&mut db_clone, &signal_sender).await;
if let Err(e) = result {
log::error!(target: target::SYNC, "Error handling sync event: {}", e);
}
});
// This is a background operation, so return right away.
// 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;
let result = Self::sync_conversation_impl(
&mut db_clone,
&signal_sender,
conversation_id,
)
.await;
if let Err(e) = result {
log::error!(target: target::SYNC, "Error handling sync event: {}", e);
}
});
reply.send(()).unwrap();
},
}
Event::GetAllConversations(limit, offset, reply) => {
let conversations = self.get_conversations_limit_offset(limit, offset).await;
reply.send(conversations).unwrap();
},
}
Event::GetAllSettings(reply) => {
let settings = self.get_settings().await
.unwrap_or_else(|e| {
log::error!(target: target::SETTINGS, "Failed to get settings: {:#?}", e);
Settings::default()
});
let settings = self.get_settings().await.unwrap_or_else(|e| {
log::error!(target: target::SETTINGS, "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!(target: target::SETTINGS, "Failed to update settings: {}", e);
});
self.update_settings(&settings).await.unwrap_or_else(|e| {
log::error!(target: target::SETTINGS, "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!(target: target::SYNC, "Failed to delete all conversations: {}", e);
});
self.delete_all_conversations().await.unwrap_or_else(|e| {
log::error!(target: target::SYNC, "Failed to delete all conversations: {}", e);
});
reply.send(()).unwrap();
},
}
Event::SendMessage(conversation_id, text, reply) => {
let conversation_id = conversation_id.clone();
let uuid = self.enqueue_outgoing_message(text, conversation_id.clone()).await;
let uuid = self
.enqueue_outgoing_message(text, conversation_id.clone())
.await;
reply.send(uuid).unwrap();
// Send message updated signal, we have a placeholder message we will return.
self.signal_sender.send(Signal::MessagesUpdated(conversation_id.clone())).await.unwrap();
},
// Send message updated signal, we have a placeholder message we will return.
self.signal_sender
.send(Signal::MessagesUpdated(conversation_id.clone()))
.await
.unwrap();
}
Event::MessageSent(message, outgoing_message, conversation_id) => {
log::info!(target: target::EVENT, "Daemon: message sent: {}", message.id);
// Insert the message into the database.
// Insert the message into the database.
log::debug!(target: target::EVENT, "inserting sent message into database: {}", message.id);
self.database.lock().await
.with_repository(|r|
r.insert_message( &conversation_id, message)
).await.unwrap();
self.database
.lock()
.await
.with_repository(|r| r.insert_message(&conversation_id, message))
.await
.unwrap();
// Remove from outgoing messages.
log::debug!(target: target::EVENT, "Removing message from outgoing messages: {}", outgoing_message.guid);
self.outgoing_messages.get_mut(&conversation_id)
self.outgoing_messages
.get_mut(&conversation_id)
.map(|messages| messages.retain(|m| m.guid != outgoing_message.guid));
// Send message updated signal.
self.signal_sender.send(Signal::MessagesUpdated(conversation_id)).await.unwrap();
},
self.signal_sender
.send(Signal::MessagesUpdated(conversation_id))
.await
.unwrap();
}
}
}
/// Panics if the signal receiver has already been taken.
pub fn obtain_signal_receiver(&mut self) -> Receiver<Signal> {
/// 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(i32::MAX, 0).unwrap()).await
self.database
.lock()
.await
.with_repository(|r| r.all_conversations(i32::MAX, 0).unwrap())
.await
}
async fn get_conversations_limit_offset(&mut self, limit: i32, offset: i32) -> Vec<Conversation> {
self.database.lock().await.with_repository(|r| r.all_conversations(limit, offset).unwrap()).await
async fn get_conversations_limit_offset(
&mut self,
limit: i32,
offset: i32,
) -> Vec<Conversation> {
self.database
.lock()
.await
.with_repository(|r| r.all_conversations(limit, offset).unwrap())
.await
}
async fn get_messages(&mut self, conversation_id: String, last_message_id: Option<String>) -> Vec<Message> {
// Get outgoing messages for this conversation.
async fn get_messages(
&mut self,
conversation_id: String,
last_message_id: Option<String>,
) -> Vec<Message> {
// Get outgoing messages for this conversation.
let empty_vec: Vec<OutgoingMessage> = vec![];
let outgoing_messages: &Vec<OutgoingMessage> = self.outgoing_messages.get(&conversation_id)
let outgoing_messages: &Vec<OutgoingMessage> = self
.outgoing_messages
.get(&conversation_id)
.unwrap_or(&empty_vec);
self.database.lock().await
.with_repository(|r|
self.database
.lock()
.await
.with_repository(|r| {
r.get_messages_for_conversation(&conversation_id)
.unwrap()
.into_iter()
.chain(outgoing_messages.into_iter().map(|m| m.into()))
.collect()
)
.unwrap()
.into_iter()
.chain(outgoing_messages.into_iter().map(|m| m.into()))
.collect()
})
.await
}
@@ -289,31 +340,41 @@ impl Daemon {
.build();
// Keep a record of this so we can provide a consistent model to the client.
self.outgoing_messages.entry(conversation_id)
self.outgoing_messages
.entry(conversation_id)
.or_insert(vec![])
.push(outgoing_message.clone());
let guid = outgoing_message.guid.clone();
self.post_office_sink.send(PostOfficeEvent::EnqueueOutgoingMessage(outgoing_message)).await.unwrap();
self.post_office_sink
.send(PostOfficeEvent::EnqueueOutgoingMessage(outgoing_message))
.await
.unwrap();
guid
}
async fn sync_conversation_list(database: &mut Arc<Mutex<Database>>, signal_sender: &Sender<Signal>) -> Result<()> {
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()
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?;
database
.with_repository(|r| r.insert_conversation(conversation))
.await?;
}
// Send conversations updated signal
@@ -323,25 +384,31 @@ impl Daemon {
Ok(())
}
async fn sync_all_conversations_impl(database: &mut Arc<Mutex<Database>>, signal_sender: &Sender<Signal>) -> Result<()> {
async fn sync_all_conversations_impl(
database: &mut Arc<Mutex<Database>>,
signal_sender: &Sender<Signal>,
) -> Result<()> {
log::info!(target: target::SYNC, "Starting full 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()
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?;
database
.with_repository(|r| r.insert_conversation(conversation))
.await?;
// Sync individual conversation.
Self::sync_conversation_impl(database, signal_sender, conversation_id).await?;
}
@@ -351,44 +418,59 @@ impl Daemon {
log::info!(target: target::SYNC, "Full sync complete, {} conversations processed", num_conversations);
Ok(())
}
}
async fn sync_conversation_impl(database: &mut Arc<Mutex<Database>>, signal_sender: &Sender<Signal>, conversation_id: String) -> Result<()> {
async fn sync_conversation_impl(
database: &mut Arc<Mutex<Database>>,
signal_sender: &Sender<Signal>,
conversation_id: String,
) -> Result<()> {
log::debug!(target: target::SYNC, "Starting conversation sync for {}", conversation_id);
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?;
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.
// If the conversation doesn't exist, first do a conversation list sync.
log::warn!(target: target::SYNC, "Conversation {} not found, performing list sync", conversation_id);
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)
.unwrap_or(None)
.map(|m| m.id)
}).await;
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()
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?;
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?;
signal_sender
.send(Signal::MessagesUpdated(conversation_id.clone()))
.await?;
}
log::debug!(target: target::SYNC, "Synchronized {} messages for conversation {}", num_messages, &conversation_id);
@@ -408,35 +490,45 @@ impl Daemon {
Self::get_client_impl(&mut self.database).await
}
async fn get_client_impl(database: &mut Arc<Mutex<Database>>) -> Result<HTTPAPIClient<DatabaseAuthenticationStore>> {
async fn get_client_impl(
database: &mut Arc<Mutex<Database>>,
) -> Result<HTTPAPIClient<DatabaseAuthenticationStore>> {
let settings = database.with_settings(Settings::from_db).await?;
let server_url = settings.server_url
let server_url = settings
.server_url
.ok_or(DaemonError::ClientNotConfigured)?;
let client = HTTPAPIClient::new(
server_url.parse().unwrap(),
DatabaseAuthenticationStore::new(database.clone())
DatabaseAuthenticationStore::new(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.database
.with_repository(|r| -> Result<()> {
r.delete_all_conversations()?;
r.delete_all_messages()?;
Ok(())
})
.await?;
self.signal_sender.send(Signal::ConversationsUpdated).await?;
self.signal_sender
.send(Signal::ConversationsUpdated)
.await?;
Ok(())
}
fn get_data_dir() -> Option<PathBuf> {
ProjectDirs::from("net", "buzzert", "kordophonecd").map(|p| PathBuf::from(p.data_dir()))
}
fn get_database_path() -> PathBuf {
if let Some(proj_dirs) = ProjectDirs::from("net", "buzzert", "kordophonecd") {
let data_dir = proj_dirs.data_dir();
if let Some(data_dir) = Self::get_data_dir() {
data_dir.join("database.db")
} else {
// Fallback to a local path if we can't get the system directories