diff options
Diffstat (limited to 'filamento/src/logic')
| -rw-r--r-- | filamento/src/logic/abort.rs | 4 | ||||
| -rw-r--r-- | filamento/src/logic/connect.rs | 8 | ||||
| -rw-r--r-- | filamento/src/logic/connection_error.rs | 7 | ||||
| -rw-r--r-- | filamento/src/logic/disconnect.rs | 7 | ||||
| -rw-r--r-- | filamento/src/logic/local_only.rs | 71 | ||||
| -rw-r--r-- | filamento/src/logic/mod.rs | 35 | ||||
| -rw-r--r-- | filamento/src/logic/offline.rs | 107 | ||||
| -rw-r--r-- | filamento/src/logic/online.rs | 565 | ||||
| -rw-r--r-- | filamento/src/logic/process_stanza.rs | 558 |
9 files changed, 1110 insertions, 252 deletions
diff --git a/filamento/src/logic/abort.rs b/filamento/src/logic/abort.rs index df82655..3588b13 100644 --- a/filamento/src/logic/abort.rs +++ b/filamento/src/logic/abort.rs @@ -1,7 +1,9 @@ use lampada::error::ReadError; +use crate::files::FileStore; + use super::ClientLogic; -pub async fn on_abort(logic: ClientLogic) { +pub async fn on_abort<Fs: FileStore + Clone>(logic: ClientLogic<Fs>) { logic.pending().drain().await; } diff --git a/filamento/src/logic/connect.rs b/filamento/src/logic/connect.rs index dc05448..6e392f1 100644 --- a/filamento/src/logic/connect.rs +++ b/filamento/src/logic/connect.rs @@ -5,17 +5,21 @@ use tracing::debug; use crate::{ Command, UpdateMessage, error::{ConnectionJobError, Error, RosterError}, + files::FileStore, presence::{Online, PresenceType}, }; use super::ClientLogic; -pub async fn handle_connect(logic: ClientLogic, connection: Connected) { +pub async fn handle_connect<Fs: FileStore + Clone + Send + Sync + 'static>( + logic: ClientLogic<Fs>, + connection: Connected, +) { let (send, recv) = oneshot::channel(); debug!("getting roster"); logic .clone() - .handle_online(Command::GetRoster(send), connection.clone()) + .handle_online(Command::GetRosterWithUsers(send), connection.clone()) .await; debug!("sent roster req"); let roster = recv.await; diff --git a/filamento/src/logic/connection_error.rs b/filamento/src/logic/connection_error.rs index 081900b..36c1cef 100644 --- a/filamento/src/logic/connection_error.rs +++ b/filamento/src/logic/connection_error.rs @@ -1,7 +1,12 @@ use lampada::error::ConnectionError; +use crate::files::FileStore; + use super::ClientLogic; -pub async fn handle_connection_error(logic: ClientLogic, error: ConnectionError) { +pub async fn handle_connection_error<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, + error: ConnectionError, +) { logic.handle_error(error.into()).await; } diff --git a/filamento/src/logic/disconnect.rs b/filamento/src/logic/disconnect.rs index 241c3e6..ebcfd4f 100644 --- a/filamento/src/logic/disconnect.rs +++ b/filamento/src/logic/disconnect.rs @@ -1,11 +1,14 @@ use lampada::Connected; use stanza::client::Stanza; -use crate::{UpdateMessage, presence::Offline}; +use crate::{UpdateMessage, files::FileStore, presence::Offline}; use super::ClientLogic; -pub async fn handle_disconnect(logic: ClientLogic, connection: Connected) { +pub async fn handle_disconnect<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, + connection: Connected, +) { // TODO: be able to set offline status message let offline_presence: stanza::client::presence::Presence = Offline::default().into_stanza(None); let stanza = Stanza::Presence(offline_presence); diff --git a/filamento/src/logic/local_only.rs b/filamento/src/logic/local_only.rs index 3f6fe8d..7f3a2e6 100644 --- a/filamento/src/logic/local_only.rs +++ b/filamento/src/logic/local_only.rs @@ -1,47 +1,94 @@ -use jid::JID; +use jid::{BareJID, JID}; use uuid::Uuid; use crate::{ chat::{Chat, Message}, error::DatabaseError, + files::FileStore, user::User, }; use super::ClientLogic; -pub async fn handle_get_chats(logic: &ClientLogic) -> Result<Vec<Chat>, DatabaseError> { +pub async fn handle_get_chats<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, +) -> Result<Vec<Chat>, DatabaseError> { Ok(logic.db().read_chats().await?) } -pub async fn handle_get_chats_ordered(logic: &ClientLogic) -> Result<Vec<Chat>, DatabaseError> { +pub async fn handle_get_chats_ordered<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, +) -> Result<Vec<Chat>, DatabaseError> { Ok(logic.db().read_chats_ordered().await?) } -pub async fn handle_get_chats_ordered_with_latest_messages( - logic: &ClientLogic, +pub async fn handle_get_chats_ordered_with_latest_messages<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, ) -> Result<Vec<(Chat, Message)>, DatabaseError> { Ok(logic.db().read_chats_ordered_with_latest_messages().await?) } -pub async fn handle_get_chat(logic: &ClientLogic, jid: JID) -> Result<Chat, DatabaseError> { +pub async fn handle_get_chats_ordered_with_latest_messages_and_users<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, +) -> Result<Vec<((Chat, User), (Message, User))>, DatabaseError> { + Ok(logic + .db() + .read_chats_ordered_with_latest_messages_and_users() + .await?) +} + +pub async fn handle_get_chat<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, +) -> Result<Chat, DatabaseError> { Ok(logic.db().read_chat(jid).await?) } -pub async fn handle_get_messages( - logic: &ClientLogic, - jid: JID, +pub async fn handle_get_chat_and_user<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, +) -> Result<(Chat, User), DatabaseError> { + Ok(logic.db().read_chat_and_user(jid).await?) +} + +pub async fn handle_get_message<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + id: Uuid, +) -> Result<Message, DatabaseError> { + Ok(logic.db().read_message(id).await?) +} + +pub async fn handle_get_messages<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, ) -> Result<Vec<Message>, DatabaseError> { Ok(logic.db().read_message_history(jid).await?) } -pub async fn handle_delete_chat(logic: &ClientLogic, jid: JID) -> Result<(), DatabaseError> { +pub async fn handle_get_messages_with_users<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, +) -> Result<Vec<(Message, User)>, DatabaseError> { + Ok(logic.db().read_message_history_with_users(jid).await?) +} + +pub async fn handle_delete_chat<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, +) -> Result<(), DatabaseError> { Ok(logic.db().delete_chat(jid).await?) } -pub async fn handle_delete_messaage(logic: &ClientLogic, uuid: Uuid) -> Result<(), DatabaseError> { +pub async fn handle_delete_messaage<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + uuid: Uuid, +) -> Result<(), DatabaseError> { Ok(logic.db().delete_message(uuid).await?) } -pub async fn handle_get_user(logic: &ClientLogic, jid: JID) -> Result<User, DatabaseError> { +pub async fn handle_get_user<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + jid: BareJID, +) -> Result<User, DatabaseError> { Ok(logic.db().read_user(jid).await?) } diff --git a/filamento/src/logic/mod.rs b/filamento/src/logic/mod.rs index 1ddd7d3..ddf0343 100644 --- a/filamento/src/logic/mod.rs +++ b/filamento/src/logic/mod.rs @@ -1,15 +1,16 @@ use std::{collections::HashMap, sync::Arc}; -use jid::JID; +use jid::{BareJID, JID}; use lampada::{Connected, Logic, error::ReadError}; use stanza::client::Stanza; use tokio::sync::{Mutex, mpsc, oneshot}; -use tracing::{error, info, warn}; +use tracing::{error, info}; use crate::{ Client, Command, UpdateMessage, db::Db, error::{Error, IqRequestError, ResponseError}, + files::FileStore, }; mod abort; @@ -22,12 +23,13 @@ mod online; mod process_stanza; #[derive(Clone)] -pub struct ClientLogic { - client: Client, - bare_jid: JID, +pub struct ClientLogic<Fs: FileStore> { + client: Client<Fs>, + jid: BareJID, db: Db, pending: Pending, update_sender: mpsc::Sender<UpdateMessage>, + file_store: Fs, } #[derive(Clone)] @@ -75,23 +77,25 @@ impl Pending { } } -impl ClientLogic { +impl<Fs: FileStore> ClientLogic<Fs> { pub fn new( - client: Client, - bare_jid: JID, + client: Client<Fs>, + jid: BareJID, db: Db, update_sender: mpsc::Sender<UpdateMessage>, + file_store: Fs, ) -> Self { Self { db, pending: Pending::new(), update_sender, client, - bare_jid, + jid, + file_store, } } - pub fn client(&self) -> &Client { + pub fn client(&self) -> &Client<Fs> { &self.client } @@ -103,6 +107,10 @@ impl ClientLogic { &self.pending } + pub fn file_store(&self) -> &Fs { + &self.file_store + } + pub fn update_sender(&self) -> &mpsc::Sender<UpdateMessage> { &self.update_sender } @@ -113,13 +121,14 @@ impl ClientLogic { self.update_sender().send(update).await; } - pub async fn handle_error(&self, e: Error) { + // TODO: delete this + pub async fn handle_error(&self, e: Error<Fs>) { error!("{}", e); } } -impl Logic for ClientLogic { - type Cmd = Command; +impl<Fs: FileStore + Clone + Send + Sync + 'static> Logic for ClientLogic<Fs> { + type Cmd = Command<Fs>; // pub async fn handle_stream_error(self, error) {} // stanza errors (recoverable) diff --git a/filamento/src/logic/offline.rs b/filamento/src/logic/offline.rs index 6399cf7..aa84f3d 100644 --- a/filamento/src/logic/offline.rs +++ b/filamento/src/logic/offline.rs @@ -1,28 +1,35 @@ +use std::process::id; + use chrono::Utc; +use jid::FullJID; use lampada::error::WriteError; +use tracing::error; use uuid::Uuid; use crate::{ Command, chat::{Delivery, Message}, error::{ - DatabaseError, DiscoError, Error, IqRequestError, MessageSendError, NickError, RosterError, - StatusError, + AvatarPublishError, DatabaseError, DiscoError, Error, IqRequestError, MessageSendError, + NickError, PEPError, RosterError, StatusError, }, + files::FileStore, presence::Online, roster::Contact, + user::User, }; use super::{ ClientLogic, local_only::{ - handle_delete_chat, handle_delete_messaage, handle_get_chat, handle_get_chats, - handle_get_chats_ordered, handle_get_chats_ordered_with_latest_messages, - handle_get_messages, handle_get_user, + handle_delete_chat, handle_delete_messaage, handle_get_chat, handle_get_chat_and_user, + handle_get_chats, handle_get_chats_ordered, handle_get_chats_ordered_with_latest_messages, + handle_get_chats_ordered_with_latest_messages_and_users, handle_get_message, + handle_get_messages, handle_get_messages_with_users, handle_get_user, }, }; -pub async fn handle_offline(logic: ClientLogic, command: Command) { +pub async fn handle_offline<Fs: FileStore + Clone>(logic: ClientLogic<Fs>, command: Command<Fs>) { let result = handle_offline_result(&logic, command).await; match result { Ok(_) => {} @@ -30,21 +37,39 @@ pub async fn handle_offline(logic: ClientLogic, command: Command) { } } -pub async fn handle_set_status(logic: &ClientLogic, online: Online) -> Result<(), StatusError> { +pub async fn handle_set_status<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + online: Online, +) -> Result<(), StatusError> { logic.db().upsert_cached_status(online).await?; Ok(()) } -pub async fn handle_get_roster(logic: &ClientLogic) -> Result<Vec<Contact>, RosterError> { +pub async fn handle_get_roster<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, +) -> Result<Vec<Contact>, RosterError> { Ok(logic.db().read_cached_roster().await?) } -pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Result<(), Error> { +pub async fn handle_get_roster_with_users<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, +) -> Result<Vec<(Contact, User)>, RosterError> { + Ok(logic.db().read_cached_roster_with_users().await?) +} + +pub async fn handle_offline_result<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + command: Command<Fs>, +) -> Result<(), Error<Fs>> { match command { Command::GetRoster(sender) => { let roster = handle_get_roster(logic).await; sender.send(roster); } + Command::GetRosterWithUsers(sender) => { + let roster = handle_get_roster_with_users(logic).await; + sender.send(roster); + } Command::GetChats(sender) => { let chats = handle_get_chats(logic).await; sender.send(chats); @@ -57,14 +82,30 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res let chats = handle_get_chats_ordered_with_latest_messages(logic).await; sender.send(chats); } + Command::GetChatsOrderedWithLatestMessagesAndUsers(sender) => { + let chats = handle_get_chats_ordered_with_latest_messages_and_users(logic).await; + sender.send(chats); + } Command::GetChat(jid, sender) => { let chats = handle_get_chat(logic, jid).await; sender.send(chats); } + Command::GetChatAndUser(jid, sender) => { + let chat = handle_get_chat_and_user(logic, jid).await; + let _ = sender.send(chat); + } + Command::GetMessage(id, sender) => { + let message = handle_get_message(logic, id).await; + let _ = sender.send(message); + } Command::GetMessages(jid, sender) => { let messages = handle_get_messages(logic, jid).await; sender.send(messages); } + Command::GetMessagesWithUsers(jid, sender) => { + let messages = handle_get_messages_with_users(logic, jid).await; + sender.send(messages); + } Command::DeleteChat(jid, sender) => { let result = handle_delete_chat(logic, jid).await; sender.send(result); @@ -77,7 +118,6 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res let user = handle_get_user(logic, jid).await; sender.send(user); } - // TODO: offline queue to modify roster Command::AddContact(_jid, sender) => { sender.send(Err(RosterError::Write(WriteError::Disconnected))); } @@ -112,14 +152,13 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res let result = handle_set_status(logic, online).await; sender.send(result); } - // TODO: offline message queue Command::SendMessage(jid, body) => { let id = Uuid::new_v4(); let timestamp = Utc::now(); let message = Message { id, - from: logic.bare_jid.clone(), + from: logic.jid.clone(), // TODO: failure reason delivery: Some(Delivery::Failed), timestamp, @@ -129,11 +168,15 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res // TODO: mark these as potentially failed upon client launch if let Err(e) = logic .db() - .create_message_with_self_resource( + // TODO: can create message without a resource.... + .create_message_with_user_resource( message.clone(), jid.clone(), // TODO: when message is queued and sent, the from must also be updated with the correct resource - logic.bare_jid.clone(), + FullJID { + bare_jid: logic.jid.clone(), + resourcepart: "unsent".to_string(), + }, ) .await { @@ -142,11 +185,25 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res .handle_error(MessageSendError::MessageHistory(e.into()).into()) .await; } + + let from = match logic.db().read_user(logic.jid.clone()).await { + Ok(u) => u, + Err(e) => { + error!("{}", e); + User { + jid: logic.jid.clone(), + nick: None, + avatar: None, + } + } + }; + logic .update_sender() .send(crate::UpdateMessage::Message { - to: jid.as_bare(), + to: jid, message, + from, }) .await; } @@ -159,7 +216,7 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res Command::DiscoItems(_jid, _node, sender) => { sender.send(Err(DiscoError::Write(WriteError::Disconnected))); } - Command::Publish { + Command::PublishPEPItem { item: _, node: _, sender, @@ -169,6 +226,24 @@ pub async fn handle_offline_result(logic: &ClientLogic, command: Command) -> Res Command::ChangeNick(_, sender) => { sender.send(Err(NickError::Disconnected)); } + Command::ChangeAvatar(_items, sender) => { + sender.send(Err(AvatarPublishError::Disconnected)); + } + Command::DeletePEPNode { node: _, sender } => { + sender.send(Err(PEPError::IqResponse(IqRequestError::Write( + WriteError::Disconnected, + )))); + } + Command::GetPEPItem { + node: _, + sender, + jid: _, + id: _, + } => { + sender.send(Err(PEPError::IqResponse(IqRequestError::Write( + WriteError::Disconnected, + )))); + } } Ok(()) } diff --git a/filamento/src/logic/online.rs b/filamento/src/logic/online.rs index b069f59..b36f9a9 100644 --- a/filamento/src/logic/online.rs +++ b/filamento/src/logic/online.rs @@ -1,35 +1,35 @@ +use std::{io::Cursor, time::Duration}; + +use base64::{prelude::BASE64_STANDARD, Engine}; use chrono::Utc; -use jid::JID; +use image::ImageReader; +use jid::{BareJID, JID}; use lampada::{Connected, WriteMessage, error::WriteError}; +use sha1::{Digest, Sha1}; use stanza::{ client::{ iq::{self, Iq, IqType, Query}, Stanza - }, - xep_0030::{info, items}, - xep_0060::pubsub::{self, Pubsub}, - xep_0172::{self, Nick}, - xep_0203::Delay, + }, xep_0030::{info, items}, xep_0060::{self, owner, pubsub::{self, Pubsub}}, xep_0084, xep_0172::{self, Nick}, xep_0203::Delay }; -use tokio::sync::oneshot; +use tokio::{sync::oneshot, task::spawn_blocking}; +#[cfg(target_arch = "wasm32")] +use tokio_with_wasm::alias as tokio; use tracing::{debug, error, info}; use uuid::Uuid; use crate::{ - chat::{Body, Chat, Delivery, Message}, disco::{Info, Items}, error::{ - DatabaseError, DiscoError, Error, IqRequestError, MessageSendError, NickError, PublishError, RosterError, StatusError, SubscribeError - }, pep, presence::{Online, Presence, PresenceType}, roster::{Contact, ContactUpdate}, Command, UpdateMessage + avatar, chat::{Body, Chat, Delivery, Message}, disco::{Info, Items}, error::{ + AvatarPublishError, DatabaseError, DiscoError, Error, IqRequestError, MessageSendError, NickError, PEPError, RosterError, StatusError, SubscribeError + }, files::FileStore, pep, presence::{Online, Presence, PresenceType}, roster::{Contact, ContactUpdate}, user::User, Command, UpdateMessage }; use super::{ - ClientLogic, local_only::{ - handle_delete_chat, handle_delete_messaage, handle_get_chat, handle_get_chats, - handle_get_chats_ordered, handle_get_chats_ordered_with_latest_messages, - handle_get_messages, handle_get_user, - }, + handle_delete_chat, handle_delete_messaage, handle_get_chat, handle_get_chat_and_user, handle_get_chats, handle_get_chats_ordered, handle_get_chats_ordered_with_latest_messages, handle_get_chats_ordered_with_latest_messages_and_users, handle_get_message, handle_get_messages, handle_get_messages_with_users, handle_get_user + }, ClientLogic }; -pub async fn handle_online(logic: ClientLogic, command: Command, connection: Connected) { +pub async fn handle_online<Fs: FileStore + Clone + 'static>(logic: ClientLogic<Fs>, command: Command<Fs>, connection: Connected) { let result = handle_online_result(&logic, command, connection).await; match result { Ok(_) => {} @@ -37,13 +37,13 @@ pub async fn handle_online(logic: ClientLogic, command: Command, connection: Con } } -pub async fn handle_get_roster( - logic: &ClientLogic, +pub async fn handle_get_roster<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, ) -> Result<Vec<Contact>, RosterError> { let iq_id = Uuid::new_v4().to_string(); let stanza = Stanza::Iq(Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq_id.to_string(), to: None, r#type: IqType::Get, @@ -96,14 +96,79 @@ pub async fn handle_get_roster( } } -pub async fn handle_add_contact( - logic: &ClientLogic, +// this can't query the client... otherwise there is a hold-up and the connection can't complete +pub async fn handle_get_roster_with_users<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + connection: Connected, +) -> Result<Vec<(Contact, User)>, RosterError> { + let iq_id = Uuid::new_v4().to_string(); + let stanza = Stanza::Iq(Iq { + from: Some(connection.jid().clone().into()), + id: iq_id.to_string(), + to: None, + r#type: IqType::Get, + lang: None, + query: Some(iq::Query::Roster(stanza::roster::Query { + ver: None, + items: Vec::new(), + })), + errors: Vec::new(), + }); + let response = logic + .pending() + .request(&connection, stanza, iq_id.clone()) + .await?; + // TODO: timeout + match response { + Stanza::Iq(Iq { + from: _, + id, + to: _, + r#type, + lang: _, + query: Some(iq::Query::Roster(stanza::roster::Query { ver: _, items })), + errors: _, + }) if id == iq_id && r#type == IqType::Result => { + let contacts: Vec<Contact> = items.into_iter().map(|item| item.into()).collect(); + if let Err(e) = logic.db().replace_cached_roster(contacts.clone()).await { + logic + .handle_error(Error::Roster(RosterError::Cache(e.into()))) + .await; + }; + let mut users = Vec::new(); + for contact in &contacts { + let user = logic.db().read_user(contact.user_jid.clone()).await?; + users.push(user); + } + Ok(contacts.into_iter().zip(users).collect()) + } + ref s @ Stanza::Iq(Iq { + from: _, + ref id, + to: _, + r#type, + lang: _, + query: _, + ref errors, + }) if *id == iq_id && r#type == IqType::Error => { + if let Some(error) = errors.first() { + Err(RosterError::StanzaError(error.clone())) + } else { + Err(RosterError::UnexpectedStanza(s.clone())) + } + } + s => Err(RosterError::UnexpectedStanza(s)), + } +} + +pub async fn handle_add_contact<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), RosterError> { let iq_id = Uuid::new_v4().to_string(); let set_stanza = Stanza::Iq(Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq_id.clone(), to: None, r#type: IqType::Set, @@ -154,12 +219,13 @@ pub async fn handle_add_contact( } } -pub async fn handle_buddy_request( - logic: &ClientLogic, +pub async fn handle_buddy_request<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), SubscribeError> { - let client_user = logic.db.read_user(logic.bare_jid.clone()).await?; + let jid: JID = jid.into(); + let client_user = logic.db.read_user(logic.jid.clone()).await?; let nick = client_user.nick.map(|nick| Nick(nick)); let presence = Stanza::Presence(stanza::client::presence::Presence { to: Some(jid.clone()), @@ -177,16 +243,16 @@ pub async fn handle_buddy_request( Ok(()) } -pub async fn handle_subscription_request( - logic: &ClientLogic, +pub async fn handle_subscription_request<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), SubscribeError> { // TODO: i should probably have builders - let client_user = logic.db.read_user(logic.bare_jid.clone()).await?; + let client_user = logic.db.read_user(logic.jid.clone()).await?; let nick = client_user.nick.map(|nick| Nick(nick)); let presence = Stanza::Presence(stanza::client::presence::Presence { - to: Some(jid), + to: Some(jid.into()), r#type: Some(stanza::client::presence::PresenceType::Subscribe), nick, ..Default::default() @@ -195,18 +261,19 @@ pub async fn handle_subscription_request( Ok(()) } -pub async fn handle_accept_buddy_request( - logic: &ClientLogic, +pub async fn handle_accept_buddy_request<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), SubscribeError> { + let jid: JID = jid.into(); let presence = Stanza::Presence(stanza::client::presence::Presence { to: Some(jid.clone()), r#type: Some(stanza::client::presence::PresenceType::Subscribed), ..Default::default() }); connection.write_handle().write(presence).await?; - let client_user = logic.db.read_user(logic.bare_jid.clone()).await?; + let client_user = logic.db.read_user(logic.jid.clone()).await?; let nick = client_user.nick.map(|nick| Nick(nick)); let presence = Stanza::Presence(stanza::client::presence::Presence { to: Some(jid), @@ -218,17 +285,14 @@ pub async fn handle_accept_buddy_request( Ok(()) } -pub async fn handle_accept_subscription_request( - logic: &ClientLogic, +pub async fn handle_accept_subscription_request<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), SubscribeError> { - let client_user = logic.db.read_user(logic.bare_jid.clone()).await?; - let nick = client_user.nick.map(|nick| Nick(nick)); let presence = Stanza::Presence(stanza::client::presence::Presence { - to: Some(jid), - r#type: Some(stanza::client::presence::PresenceType::Subscribe), - nick, + to: Some(jid.into()), + r#type: Some(stanza::client::presence::PresenceType::Subscribed), ..Default::default() }); connection.write_handle().write(presence).await?; @@ -237,10 +301,10 @@ pub async fn handle_accept_subscription_request( pub async fn handle_unsubscribe_from_contact( connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), WriteError> { let presence = Stanza::Presence(stanza::client::presence::Presence { - to: Some(jid), + to: Some(jid.into()), r#type: Some(stanza::client::presence::PresenceType::Unsubscribe), ..Default::default() }); @@ -248,9 +312,9 @@ pub async fn handle_unsubscribe_from_contact( Ok(()) } -pub async fn handle_unsubscribe_contact(connection: Connected, jid: JID) -> Result<(), WriteError> { +pub async fn handle_unsubscribe_contact(connection: Connected, jid: BareJID) -> Result<(), WriteError> { let presence = Stanza::Presence(stanza::client::presence::Presence { - to: Some(jid), + to: Some(jid.into()), r#type: Some(stanza::client::presence::PresenceType::Unsubscribed), ..Default::default() }); @@ -258,7 +322,8 @@ pub async fn handle_unsubscribe_contact(connection: Connected, jid: JID) -> Resu Ok(()) } -pub async fn handle_unfriend_contact(connection: Connected, jid: JID) -> Result<(), WriteError> { +pub async fn handle_unfriend_contact(connection: Connected, jid: BareJID) -> Result<(), WriteError> { + let jid: JID = jid.into(); let presence = Stanza::Presence(stanza::client::presence::Presence { to: Some(jid.clone()), r#type: Some(stanza::client::presence::PresenceType::Unsubscribe), @@ -274,14 +339,14 @@ pub async fn handle_unfriend_contact(connection: Connected, jid: JID) -> Result< Ok(()) } -pub async fn handle_delete_contact( - logic: &ClientLogic, +pub async fn handle_delete_contact<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, ) -> Result<(), RosterError> { let iq_id = Uuid::new_v4().to_string(); let set_stanza = Stanza::Iq(Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq_id.clone(), to: None, r#type: IqType::Set, @@ -333,10 +398,10 @@ pub async fn handle_delete_contact( } } -pub async fn handle_update_contact( - logic: &ClientLogic, +pub async fn handle_update_contact<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, - jid: JID, + jid: BareJID, contact_update: ContactUpdate, ) -> Result<(), RosterError> { let iq_id = Uuid::new_v4().to_string(); @@ -347,7 +412,8 @@ pub async fn handle_update_contact( .map(|group| stanza::roster::Group(Some(group))), ); let set_stanza = Stanza::Iq(Iq { - from: Some(connection.jid().clone()), + // TODO: these clones could technically be avoided? + from: Some(connection.jid().clone().into()), id: iq_id.clone(), to: None, r#type: IqType::Set, @@ -398,8 +464,8 @@ pub async fn handle_update_contact( } } -pub async fn handle_set_status( - logic: &ClientLogic, +pub async fn handle_set_status<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, online: Online, ) -> Result<(), StatusError> { @@ -411,14 +477,23 @@ pub async fn handle_set_status( Ok(()) } -pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid: JID, body: Body) { +pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, connection: Connected, jid: BareJID, body: Body) { // upsert the chat and user the message will be delivered to. if there is a conflict, it will return whatever was there, otherwise it will return false by default. - let have_chatted = logic.db().upsert_chat_and_user(&jid).await.unwrap_or(false); + // let have_chatted = logic.db().upsert_chat_and_user(&jid).await.unwrap_or(false); + let have_chatted = match logic.db().upsert_chat_and_user(jid.clone()).await { + Ok(have_chatted) => { + have_chatted + }, + Err(e) => { + error!("{}", e); + false + }, + }; let nick; let mark_chat_as_chatted; if have_chatted == false { - match logic.db.read_user(logic.bare_jid.clone()).await { + match logic.db.read_user(logic.jid.clone()).await { Ok(u) => { nick = u.nick.map(|nick| Nick(nick)); mark_chat_as_chatted = true; @@ -441,7 +516,7 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid let timestamp = Utc::now(); let message = Message { id, - from: connection.jid().as_bare(), + from: connection.jid().to_bare(), body: body.clone(), timestamp, delivery: Some(Delivery::Sending), @@ -451,7 +526,7 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid // TODO: mark these as potentially failed upon client launch if let Err(e) = logic .db() - .create_message_with_self_resource(message.clone(), jid.clone(), connection.jid().clone()) + .create_message_with_user_resource(message.clone(), jid.clone(), connection.jid().clone()) .await { // TODO: should these really be handle_error or just the error macro? @@ -460,20 +535,33 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid .await; } + let from = match logic.db().read_user(logic.jid.clone()).await { + Ok(u) => u, + Err(e) => { + error!("{}", e); + User { + jid: logic.jid.clone(), + nick: None, + avatar: None, + } + }, + }; + // tell the client a message is being sent logic .update_sender() .send(UpdateMessage::Message { - to: jid.as_bare(), + to: jid.clone(), message, + from, }) .await; // prepare the message stanza let message_stanza = Stanza::Message(stanza::client::message::Message { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: Some(id.to_string()), - to: Some(jid.clone()), + to: Some(jid.clone().into()), // TODO: specify message type r#type: stanza::client::message::MessageType::Chat, // TODO: lang ? @@ -498,14 +586,19 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid match result { Ok(_) => { info!("sent message: {:?}", message_stanza); + if let Err(e) = logic.db().update_message_delivery(id, Delivery::Written).await { + error!("updating message delivery: {}", e); + } logic .update_sender() .send(UpdateMessage::MessageDelivery { id, delivery: Delivery::Written, + chat: jid.clone(), }) .await; if mark_chat_as_chatted { + debug!("marking chat as chatted"); if let Err(e) = logic.db.mark_chat_as_chatted(jid).await { logic .handle_error(MessageSendError::MarkChatAsChatted(e.into()).into()) @@ -519,6 +612,7 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid .send(UpdateMessage::MessageDelivery { id, delivery: Delivery::Failed, + chat: jid, }) .await; logic.handle_error(MessageSendError::Write(e).into()).await; @@ -540,15 +634,15 @@ pub async fn handle_send_presence( Ok(()) } -pub async fn handle_disco_info( - logic: &ClientLogic, +pub async fn handle_disco_info<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, jid: Option<JID>, node: Option<String>, ) -> Result<Info, DiscoError> { let id = Uuid::new_v4().to_string(); let request = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: id.clone(), to: jid.clone(), r#type: IqType::Get, @@ -576,7 +670,7 @@ pub async fn handle_disco_info( }) if r#type == IqType::Result || r#type == IqType::Error => { if from == jid || { if jid == None { - from == Some(connection.jid().as_bare()) + from == Some(connection.jid().to_bare().into()) } else { false } @@ -603,7 +697,7 @@ pub async fn handle_disco_info( } } else { Err(DiscoError::IncorrectEntity( - from.unwrap_or_else(|| connection.jid().as_bare()), + from.unwrap_or_else(|| connection.jid().to_bare().into()), )) } } @@ -611,15 +705,15 @@ pub async fn handle_disco_info( } } -pub async fn handle_disco_items( - logic: &ClientLogic, +pub async fn handle_disco_items<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, jid: Option<JID>, node: Option<String>, ) -> Result<Items, DiscoError> { let id = Uuid::new_v4().to_string(); let request = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: id.clone(), to: jid.clone(), r#type: IqType::Get, @@ -645,7 +739,7 @@ pub async fn handle_disco_items( }) if r#type == IqType::Result || r#type == IqType::Error => { if from == jid || { if jid == None { - from == Some(connection.jid().as_bare()) + from == Some(connection.jid().to_bare().into()) } else { false } @@ -672,7 +766,7 @@ pub async fn handle_disco_items( } } else { Err(DiscoError::IncorrectEntity( - from.unwrap_or_else(|| connection.jid().as_bare()), + from.unwrap_or_else(|| connection.jid().to_bare().into()), )) } } @@ -680,24 +774,64 @@ pub async fn handle_disco_items( } } -pub async fn handle_publish( - logic: &ClientLogic, +pub async fn handle_publish_pep_item<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, connection: Connected, item: pep::Item, node: String, -) -> Result<(), PublishError> { +) -> Result<(), PEPError> { let id = Uuid::new_v4().to_string(); let publish = match item { - pep::Item::Nick(n) => pubsub::Publish { - node, - items: vec![pubsub::Item { - item: Some(pubsub::Content::Nick(Nick(n))), - ..Default::default() - }], + pep::Item::Nick(n) => { + if let Some(n) = n { + pubsub::Publish { + node, + items: vec![pubsub::Item { + item: Some(pubsub::Content::Nick(Nick(n))), + ..Default::default() + }], + } + } else { + pubsub::Publish { + node, + items: vec![pubsub::Item { + item: Some(pubsub::Content::Nick(Nick("".to_string()))), + ..Default::default() + }] + } + } + }, + pep::Item::AvatarMetadata(metadata) => { + if let Some(metadata) = metadata { + pubsub::Publish { node, items: vec![pubsub::Item { + item: Some(pubsub::Content::AvatarMetadata(xep_0084::Metadata { info: vec![xep_0084::Info { bytes: metadata.bytes, height: None, id: metadata.hash.clone(), r#type: metadata.r#type, url: None, width: None }], pointers: Vec::new() })), + id: Some(metadata.hash), + ..Default::default() + }]} + } else { + pubsub::Publish { node, items: vec![pubsub::Item { + item: Some(pubsub::Content::AvatarMetadata(xep_0084::Metadata { info: Vec::new(), pointers: Vec::new() })), + ..Default::default() + }]} + } + }, + pep::Item::AvatarData(data) => { + if let Some(data) = data { + pubsub::Publish { node, items: vec![pubsub::Item { + item: Some(pubsub::Content::AvatarData(xep_0084::Data(data.data_b64))), + id: Some(data.hash), + ..Default::default() + }] } + } else { + pubsub::Publish { node, items: vec![pubsub::Item { + item: Some(pubsub::Content::AvatarData(xep_0084::Data("".to_string()))), + ..Default::default() + }]} + } }, }; let request = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: id.clone(), to: None, r#type: IqType::Set, @@ -719,50 +853,251 @@ pub async fn handle_publish( // TODO: maybe abstract a bunch of these different errors related to iqs into an iq error thing? as in like call iq.result(), get the query from inside, error otherwise. }) if r#type == IqType::Result || r#type == IqType::Error => { if from == None || - from == Some(connection.jid().as_bare()) + from == Some(connection.jid().to_bare().into()) { match r#type { IqType::Result => { if let Some(query) = query { match query { Query::Pubsub(_) => Ok(()), - q => Err(PublishError::MismatchedQuery(q)), + q => Err(PEPError::MismatchedQuery(q)), + } + } else { + Err(PEPError::MissingQuery) + } + } + IqType::Error => { + Err(PEPError::StanzaErrors(errors)) + } + _ => unreachable!(), + } + } else { + Err(PEPError::IncorrectEntity( + from.unwrap_or_else(|| connection.jid().to_bare().into()), + )) + } + } + s => Err(PEPError::UnexpectedStanza(s)), + } +} + +pub async fn handle_get_pep_item<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, connection: Connected, jid: Option<BareJID>, node: String, id: String) -> Result<pep::Item, PEPError> { + let jid = jid.map(|jid| Into::<JID>::into(jid)); + let stanza_id = Uuid::new_v4().to_string(); + let request = Iq { + from: Some(connection.jid().clone().into()), + id: stanza_id.clone(), + to: jid.clone(), + r#type: IqType::Get, + lang: None, + query: Some(Query::Pubsub(Pubsub::Items(pubsub::Items { + max_items: None, + node, + subid: None, + items: vec![pubsub::Item { id: Some(id.clone()), publisher: None, item: None }], + }))), + errors: Vec::new(), + }; + match logic + .pending() + .request(&connection, Stanza::Iq(request), stanza_id) + .await? { + + Stanza::Iq(Iq { + from, + r#type, + query, + errors, + .. + // TODO: maybe abstract a bunch of these different errors related to iqs into an iq error thing? as in like call iq.result(), get the query from inside, error otherwise. + }) if r#type == IqType::Result || r#type == IqType::Error => { + if from == jid || { + if jid == None { + from == Some(connection.jid().to_bare().into()) + } else { + false + } + } { + match r#type { + IqType::Result => { + if let Some(query) = query { + match query { + Query::Pubsub(Pubsub::Items(mut items)) => { + if let Some(item) = items.items.pop() { + if item.id == Some(id.clone()) { + match item.item.ok_or(PEPError::MissingItem)? { + pubsub::Content::Nick(nick) => { + if nick.0.is_empty() { + Ok(pep::Item::Nick(None)) + } else { + Ok(pep::Item::Nick(Some(nick.0))) + + } + }, + pubsub::Content::AvatarData(data) => Ok(pep::Item::AvatarData(Some(avatar::Data { hash: id, data_b64: data.0 }))), + pubsub::Content::AvatarMetadata(metadata) => Ok(pep::Item::AvatarMetadata(metadata.info.into_iter().find(|info| info.url.is_none()).map(|info| info.into()))), + pubsub::Content::Unknown(_element) => Err(PEPError::UnsupportedItem), + } + } else { + Err(PEPError::IncorrectItemID(id, item.id.unwrap_or_else(|| "missing id".to_string()))) + } + } else { + Err(PEPError::MissingItem) + } + }, + q => Err(PEPError::MismatchedQuery(q)), } } else { - Err(PublishError::MissingQuery) + Err(PEPError::MissingQuery) } } IqType::Error => { - Err(PublishError::StanzaErrors(errors)) + Err(PEPError::StanzaErrors(errors)) } _ => unreachable!(), } } else { - Err(PublishError::IncorrectEntity( - from.unwrap_or_else(|| connection.jid().as_bare()), + // TODO: include expected entity + Err(PEPError::IncorrectEntity( + from.unwrap_or_else(|| connection.jid().to_bare().into()), )) } } - s => Err(PublishError::UnexpectedStanza(s)), + s => Err(PEPError::UnexpectedStanza(s)), } } -pub async fn handle_change_nick(logic: &ClientLogic, nick: String) -> Result<(), NickError> { +pub async fn handle_change_nick<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, nick: Option<String>) -> Result<(), NickError> { logic.client().publish(pep::Item::Nick(nick), xep_0172::XMLNS.to_string()).await?; Ok(()) } +pub async fn handle_change_avatar<Fs: FileStore + Clone + 'static>(logic: &ClientLogic<Fs>, img_data: Option<Vec<u8>>) -> Result<(), AvatarPublishError<Fs>> { + match img_data { + // set avatar + Some(data) => { + let (bytes, hash, data_png, data_b64) = spawn_blocking(move || -> Result<_, _> { + // load the image data and guess the format + let image = ImageReader::new(Cursor::new(data)).with_guessed_format()?.decode()?; + + // convert the image to png; + let mut data_png = Vec::new(); + let image = image.resize(192, 192, image::imageops::FilterType::Nearest); + image.write_to(&mut Cursor::new(&mut data_png), image::ImageFormat::Jpeg)?; + + // calculate the length of the data in bytes. + let bytes = data_png.len().try_into()?; + + // calculate sha1 hash of the data + let mut sha1 = Sha1::new(); + sha1.update(&data_png); + let sha1_result = sha1.finalize(); + let hash = hex::encode(sha1_result); + + // encode the image data as base64 + let data_b64 = BASE64_STANDARD.encode(data_png.clone()); + + Ok::<(u32, String, Vec<u8>, String), AvatarPublishError<Fs>>((bytes, hash, data_png, data_b64)) + }).await.unwrap()?; + + // publish the data to the data node + logic.client().publish(pep::Item::AvatarData(Some(avatar::Data { hash: hash.clone(), data_b64 })), "urn:xmpp:avatar:data".to_string()).await?; + + // publish the metadata to the metadata node + logic.client().publish(pep::Item::AvatarMetadata(Some(avatar::Metadata { bytes, hash: hash.clone(), r#type: "image/jpeg".to_string() })), "urn:xmpp:avatar:metadata".to_string()).await?; + + // if everything went well, save the data to the disk. + + if !logic.file_store().is_stored(&hash).await.map_err(|err| AvatarPublishError::FileStore(err))? { + logic.file_store().store(&hash, &data_png).await.map_err(|err| AvatarPublishError::FileStore(err))? + } + // when the client receives the updated metadata notification from the pep node, it will already have it saved on the disk so will not require a retrieval. + // TODO: should the node be purged? + + Ok(()) + }, + // remove avatar + None => { + logic.client().delete_pep_node("urn:xmpp:avatar:data".to_string()).await?; + logic.client().publish(pep::Item::AvatarMetadata(None), "urn:xmpp:avatar:metadata".to_string(), ).await?; + Ok(()) + }, + } +} + +pub async fn handle_delete_pep_node<Fs: FileStore + Clone>( + logic: &ClientLogic<Fs>, + connection: Connected, + node: String, +) -> Result<(), PEPError> { + let id = Uuid::new_v4().to_string(); + let request = Iq { + from: Some(connection.jid().clone().into()), + id: id.clone(), + to: None, + r#type: IqType::Set, + lang: None, + query: Some(Query::PubsubOwner(xep_0060::owner::Pubsub::Delete(owner::Delete{ node, redirect: None }))), + errors: Vec::new(), + }; + match logic + .pending() + .request(&connection, Stanza::Iq(request), id) + .await? { + + Stanza::Iq(Iq { + from, + r#type, + query, + errors, + .. + // TODO: maybe abstract a bunch of these different errors related to iqs into an iq error thing? as in like call iq.result(), get the query from inside, error otherwise. + }) if r#type == IqType::Result || r#type == IqType::Error => { + if from == None || + from == Some(connection.jid().to_bare().into()) + { + match r#type { + IqType::Result => { + if let Some(query) = query { + match query { + Query::PubsubOwner(_) => Ok(()), + q => Err(PEPError::MismatchedQuery(q)), + } + } else { + // Err(PEPError::MissingQuery) + Ok(()) + } + } + IqType::Error => { + Err(PEPError::StanzaErrors(errors)) + } + _ => unreachable!(), + } + } else { + Err(PEPError::IncorrectEntity( + from.unwrap_or_else(|| connection.jid().to_bare().into()), + )) + } + } + s => Err(PEPError::UnexpectedStanza(s)), + } +} + // TODO: could probably macro-ise? -pub async fn handle_online_result( - logic: &ClientLogic, - command: Command, +pub async fn handle_online_result<Fs: FileStore + Clone + 'static>( + logic: &ClientLogic<Fs>, + command: Command<Fs>, connection: Connected, -) -> Result<(), Error> { +) -> Result<(), Error<Fs>> { match command { Command::GetRoster(result_sender) => { let roster = handle_get_roster(logic, connection).await; let _ = result_sender.send(roster); } + Command::GetRosterWithUsers(result_sender) => { + let roster = handle_get_roster_with_users(logic, connection).await; + let _ = result_sender.send(roster); + } Command::GetChats(sender) => { let chats = handle_get_chats(logic).await; let _ = sender.send(chats); @@ -775,14 +1110,30 @@ pub async fn handle_online_result( let chats = handle_get_chats_ordered_with_latest_messages(logic).await; let _ = sender.send(chats); } + Command::GetChatsOrderedWithLatestMessagesAndUsers(sender) => { + let chats = handle_get_chats_ordered_with_latest_messages_and_users(logic).await; + sender.send(chats); + } Command::GetChat(jid, sender) => { let chat = handle_get_chat(logic, jid).await; let _ = sender.send(chat); } + Command::GetChatAndUser(jid, sender) => { + let chat = handle_get_chat_and_user(logic, jid).await; + let _ = sender.send(chat); + } + Command::GetMessage(id, sender) => { + let message = handle_get_message(logic, id).await; + let _ = sender.send(message); + } Command::GetMessages(jid, sender) => { let messages = handle_get_messages(logic, jid).await; let _ = sender.send(messages); } + Command::GetMessagesWithUsers(jid, sender) => { + let messages = handle_get_messages_with_users(logic, jid).await; + sender.send(messages); + } Command::DeleteChat(jid, sender) => { let result = handle_delete_chat(logic, jid).await; let _ = sender.send(result); @@ -854,14 +1205,26 @@ pub async fn handle_online_result( let result = handle_disco_items(logic, connection, jid, node).await; let _ = sender.send(result); } - Command::Publish { item, node, sender } => { - let result = handle_publish(logic, connection, item, node).await; + Command::PublishPEPItem { item, node, sender } => { + let result = handle_publish_pep_item(logic, connection, item, node).await; let _ = sender.send(result); } Command::ChangeNick(nick, sender) => { let result = handle_change_nick(logic, nick).await; let _ = sender.send(result); } + Command::ChangeAvatar(img_data, sender) => { + let result = handle_change_avatar(logic, img_data).await; + let _ = sender.send(result); + }, + Command::DeletePEPNode { node, sender } => { + let result = handle_delete_pep_node(logic, connection, node).await; + let _ = sender.send(result); + }, + Command::GetPEPItem { node, sender, jid, id } => { + let result = handle_get_pep_item(logic, connection, jid, node, id).await; + let _ = sender.send(result); + }, } Ok(()) } diff --git a/filamento/src/logic/process_stanza.rs b/filamento/src/logic/process_stanza.rs index 11d7588..67b0d3f 100644 --- a/filamento/src/logic/process_stanza.rs +++ b/filamento/src/logic/process_stanza.rs @@ -1,7 +1,9 @@ use std::str::FromStr; +use base64::{Engine, prelude::BASE64_STANDARD}; use chrono::Utc; use lampada::{Connected, SupervisorSender}; +use sha1::{Digest, Sha1}; use stanza::{ client::{ Stanza, @@ -17,14 +19,23 @@ use uuid::Uuid; use crate::{ UpdateMessage, caps, chat::{Body, Message}, - error::{DatabaseError, Error, IqError, MessageRecvError, PresenceError, RosterError}, + error::{ + AvatarUpdateError, DatabaseError, Error, IqError, IqProcessError, MessageRecvError, + PresenceError, RosterError, + }, + files::FileStore, presence::{Offline, Online, Presence, PresenceType, Show}, roster::Contact, + user::User, }; use super::ClientLogic; -pub async fn handle_stanza(logic: ClientLogic, stanza: Stanza, connection: Connected) { +pub async fn handle_stanza<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, + stanza: Stanza, + connection: Connected, +) { let result = process_stanza(logic.clone(), stanza, connection).await; match result { Ok(u) => match u { @@ -38,10 +49,10 @@ pub async fn handle_stanza(logic: ClientLogic, stanza: Stanza, connection: Conne } } -pub async fn recv_message( - logic: ClientLogic, +pub async fn recv_message<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, stanza_message: stanza::client::message::Message, -) -> Result<Option<UpdateMessage>, MessageRecvError> { +) -> Result<Option<UpdateMessage>, MessageRecvError<Fs>> { if let Some(from) = stanza_message.from { // TODO: don't ignore delay from. xep says SHOULD send error if incorrect. let timestamp = stanza_message @@ -50,6 +61,7 @@ pub async fn recv_message( .unwrap_or_else(|| Utc::now()); // TODO: group chat messages + // body MUST be before user changes in order to avoid race condition where you e.g. get a nick update before the user is in the client state. // if there is a body, should create chat message if let Some(body) = stanza_message.body { let message = Message { @@ -58,101 +70,398 @@ pub async fn recv_message( // TODO: proper id xep .map(|id| Uuid::from_str(&id).unwrap_or_else(|_| Uuid::new_v4())) .unwrap_or_else(|| Uuid::new_v4()), - from: from.as_bare(), + from: from.to_bare(), timestamp, body: Body { body: body.body.unwrap_or_default(), }, delivery: None, }; + // TODO: process message type="error" // save the message to the database - logic.db().upsert_chat_and_user(&from).await?; - if let Err(e) = logic - .db() - .create_message_with_user_resource(message.clone(), from.clone(), from.clone()) - .await - { - logic - .handle_error(Error::MessageRecv(MessageRecvError::MessageHistory(e))) - .await; - } + match logic.db().upsert_chat_and_user(from.to_bare()).await { + Ok(_) => match from.as_full() { + Ok(from) => { + if let Err(e) = logic + .db() + .create_message_with_user_resource( + message.clone(), + from.to_bare(), + from.clone(), + ) + .await + { + error!("failed to create message: {}", e); + } + } + Err(e) => error!("failed to create message: {}", e), + }, + Err(e) => { + error!("failed to upsert chat and user: {}", e); + } + }; + + let from_user = match logic.db().read_user(from.to_bare()).await { + Ok(u) => u, + Err(e) => { + error!("{}", e); + User { + jid: from.to_bare(), + nick: None, + avatar: None, + } + } + }; // update the client with the new message logic .update_sender() .send(UpdateMessage::Message { - to: from.as_bare(), + to: from.to_bare(), + from: from_user, message, }) .await; } if let Some(nick) = stanza_message.nick { - if let Err(e) = logic - .db() - .upsert_user_nick(from.as_bare(), nick.0.clone()) - .await - { - logic - .handle_error(Error::MessageRecv(MessageRecvError::NickUpdate(e))) - .await; + let nick = nick.0; + if nick.is_empty() { + match logic.db().delete_user_nick(from.to_bare()).await { + Ok(changed) => { + if changed { + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: None, + }) + .await; + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv(MessageRecvError::NickUpdate(e))) + .await; + // if failed, send user update anyway + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: None, + }) + .await; + } + } + } else { + match logic + .db() + .upsert_user_nick(from.to_bare(), nick.clone()) + .await + { + Ok(changed) => { + if changed { + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: Some(nick), + }) + .await; + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv(MessageRecvError::NickUpdate(e))) + .await; + // if failed, send user update anyway + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: Some(nick), + }) + .await; + } + } } - - logic - .update_sender() - .send(UpdateMessage::NickChanged { - jid: from.as_bare(), - nick: nick.0, - }) - .await; } if let Some(event) = stanza_message.event { match event { - Event::Items(items) => match items.node.as_str() { - "http://jabber.org/protocol/nick" => match items.items { - ItemsType::Item(items) => { - if let Some(item) = items.first() { - match &item.item { - Some(c) => match c { - Content::Nick(nick) => { - if let Err(e) = logic - .db() - .upsert_user_nick(from.as_bare(), nick.0.clone()) - .await - { - logic - .handle_error(Error::MessageRecv( - MessageRecvError::NickUpdate(e), - )) - .await; + Event::Items(items) => { + match items.node.as_str() { + "http://jabber.org/protocol/nick" => match items.items { + ItemsType::Item(items) => { + if let Some(item) = items.first() { + match &item.item { + Some(c) => match c { + Content::Nick(nick) => { + let nick = nick.0.clone(); + if nick.is_empty() { + match logic + .db() + .delete_user_nick(from.to_bare()) + .await + { + Ok(changed) => { + if changed { + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: None, + }) + .await; + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv( + MessageRecvError::NickUpdate(e), + )) + .await; + // if failed, send user update anyway + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: None, + }) + .await; + } + } + } else { + match logic + .db() + .upsert_user_nick( + from.to_bare(), + nick.clone(), + ) + .await + { + Ok(changed) => { + if changed { + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: Some(nick), + }) + .await; + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv( + MessageRecvError::NickUpdate(e), + )) + .await; + // if failed, send user update anyway + logic + .update_sender() + .send(UpdateMessage::NickChanged { + jid: from.to_bare(), + nick: Some(nick), + }) + .await; + } + } + } } + _ => {} + }, + None => {} + } + } + } + _ => {} + }, + "urn:xmpp:avatar:metadata" => { + match items.items { + ItemsType::Item(items) => { + if let Some(item) = items.first() { + debug!("found item"); + match &item.item { + Some(Content::AvatarMetadata(metadata)) => { + debug!("found metadata"); + // check if user avatar has been deleted + if let Some(metadata) = metadata + .info + .iter() + .find(|info| info.url.is_none()) + { + debug!("checking if user avatar has changed"); + // check if user avatar has changed + match logic + .db() + .upsert_user_avatar( + from.to_bare(), + metadata.id.clone(), + ) + .await + { + Ok((changed, old_avatar)) => { + if changed { + if let Some(old_avatar) = old_avatar + { + if let Err(e) = logic + .file_store() + .delete(&old_avatar) + .await.map_err(|err| AvatarUpdateError::FileStore(err)) { + logic.handle_error(MessageRecvError::AvatarUpdate(e).into()).await; + } + } + } - logic - .update_sender() - .send(UpdateMessage::NickChanged { - jid: from.as_bare(), - nick: nick.0.clone(), - }) - .await; + match logic + .file_store() + .is_stored(&metadata.id) + .await + .map_err(|err| { + AvatarUpdateError::<Fs>::FileStore( + err, + ) + }) { + Ok(false) => { + // get data + let pep_item = logic.client().get_pep_item(Some(from.to_bare()), "urn:xmpp:avatar:data".to_string(), metadata.id.clone()).await.map_err(|err| Into::<AvatarUpdateError<Fs>>::into(err))?; + match pep_item { + crate::pep::Item::AvatarData(data) => { + let data = data.map(|data| data.data_b64).unwrap_or_default().replace("\n", ""); + // TODO: these should all be in a separate avatarupdate function + debug!("got avatar data"); + match BASE64_STANDARD.decode(data) { + Ok(data) => { + let mut hasher = Sha1::new(); + hasher.update(&data); + let received_data_hash = hex::encode(hasher.finalize()); + debug!("received_data_hash: {}, metadata_id: {}", received_data_hash, metadata.id); + if received_data_hash.to_lowercase() == metadata.id.to_lowercase() { + if let Err(e) = logic.file_store().store(&received_data_hash, &data).await { + logic.handle_error(Error::MessageRecv(MessageRecvError::AvatarUpdate(AvatarUpdateError::FileStore(e)))).await; + } + if changed { + logic + .update_sender() + .send( + UpdateMessage::AvatarChanged { + jid: from.to_bare(), + id: Some( + metadata.id.clone(), + ), + }, + ) + .await; + } + } + }, + Err(e) => { + logic.handle_error(Error::MessageRecv(MessageRecvError::AvatarUpdate(AvatarUpdateError::Base64(e)))).await; + }, + } + }, + _ => { + logic.handle_error(Error::MessageRecv(MessageRecvError::AvatarUpdate(AvatarUpdateError::MissingData))).await; + } + } + } + Ok(true) => { + // just send the update + if changed { + logic + .update_sender() + .send( + UpdateMessage::AvatarChanged { + jid: from.to_bare(), + id: Some( + metadata.id.clone(), + ), + }, + ) + .await; + } + } + Err(e) => { + logic.handle_error(Error::MessageRecv(MessageRecvError::AvatarUpdate(e))).await; + } + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv( + MessageRecvError::AvatarUpdate( + AvatarUpdateError::Database( + e, + ), + ), + )) + .await; + } + } + } else { + // delete avatar + match logic + .db() + .delete_user_avatar(from.to_bare()) + .await + { + Ok((changed, old_avatar)) => { + if changed { + if let Some(old_avatar) = old_avatar + { + if let Err(e) = logic + .file_store() + .delete(&old_avatar) + .await.map_err(|err| AvatarUpdateError::FileStore(err)) { + logic.handle_error(MessageRecvError::AvatarUpdate(e).into()).await; + } + } + logic + .update_sender() + .send( + UpdateMessage::AvatarChanged { + jid: from.to_bare(), + id: None, + }, + ) + .await; + } + } + Err(e) => { + logic + .handle_error(Error::MessageRecv( + MessageRecvError::AvatarUpdate( + AvatarUpdateError::Database( + e, + ), + ), + )) + .await; + } + } + } + // check if the new file is in the file store + // if not, retrieve from server and save in the file store (remember to check if the hash matches) + // send the avatar update + } + _ => {} } - Content::Unknown(element) => {} - }, - None => {} + } } + _ => {} } } - ItemsType::Retract(retracts) => {} - }, - _ => {} - }, + _ => {} + } + } // Event::Collection(collection) => todo!(), // Event::Configuration(configuration) => todo!(), // Event::Delete(delete) => todo!(), // Event::Purge(purge) => todo!(), // Event::Subscription(subscription) => todo!(), - _ => {} + _ => {} // TODO: catch these catch-alls in some way } } @@ -173,13 +482,7 @@ pub async fn recv_presence( stanza::client::presence::PresenceType::Error => { // TODO: is there any other information that should go with the error? also MUST have an error, otherwise it's a different error. maybe it shoulnd't be an option. // TODO: ughhhhhhhhhhhhh these stanza errors should probably just have an option, and custom display - Err(PresenceError::StanzaError( - presence - .errors - .first() - .cloned() - .expect("error MUST have error"), - )) + Err(PresenceError::StanzaError(presence.errors.first().cloned())) } // should not happen (error to server) stanza::client::presence::PresenceType::Probe => { @@ -188,6 +491,7 @@ pub async fn recv_presence( } stanza::client::presence::PresenceType::Subscribe => { // may get a subscription request from somebody who is not a contact!!! therefore should be its own kind of event + let from = from.try_into()?; Ok(Some(UpdateMessage::SubscriptionRequest(from))) } stanza::client::presence::PresenceType::Unavailable => { @@ -240,15 +544,16 @@ pub async fn recv_presence( } } -pub async fn recv_iq( - logic: ClientLogic, +pub async fn recv_iq<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, connection: Connected, iq: Iq, -) -> Result<Option<UpdateMessage>, IqError> { +) -> Result<Option<UpdateMessage>, IqProcessError> { if let Some(to) = &iq.to { - if *to == *connection.jid() { + // TODO: this clone could mayb b avoided + if *to == connection.jid().clone().into() { } else { - return Err(IqError::IncorrectAddressee(to.clone())); + return Err(IqProcessError::Iq(IqError::IncorrectAddressee(to.clone()))); } } match iq.r#type { @@ -256,17 +561,24 @@ pub async fn recv_iq( let from = iq .from .clone() - .unwrap_or_else(|| connection.server().clone()); + // TODO: maybe actually store the server in the connection again LOL + // .unwrap_or_else(|| connection.server().clone()); + .unwrap_or_else(|| connection.jid().domainpart.parse().unwrap()); let id = iq.id.clone(); debug!("received iq result with id `{}` from {}", id, from); - logic.pending().respond(Stanza::Iq(iq), id).await?; + logic + .pending() + .respond(Stanza::Iq(iq), id) + .await + .map_err(|e| Into::<IqError>::into(e))?; Ok(None) } stanza::client::iq::IqType::Get => { let from = iq .from .clone() - .unwrap_or_else(|| connection.server().clone()); + // .unwrap_or_else(|| connection.server().clone()); + .unwrap_or_else(|| connection.jid().domainpart.parse().unwrap()); if let Some(query) = iq.query { match query { stanza::client::iq::Query::DiscoInfo(query) => { @@ -280,7 +592,7 @@ pub async fn recv_iq( } else { match logic .db() - .read_capabilities(&query.node.clone().unwrap()) + .read_capabilities(query.node.clone().unwrap()) .await { Ok(c) => match caps::decode_info_base64(c) { @@ -290,7 +602,7 @@ pub async fn recv_iq( } Err(_e) => { let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -299,14 +611,18 @@ pub async fn recv_iq( errors: vec![StanzaError::ItemNotFound.into()], }; // TODO: log error - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; info!("replied to disco#info request from {}", from); return Ok(None); } }, Err(_e) => { let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -315,14 +631,18 @@ pub async fn recv_iq( errors: vec![StanzaError::ItemNotFound.into()], }; // TODO: log error - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; info!("replied to disco#info request from {}", from); return Ok(None); } } }; let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Result, @@ -330,14 +650,18 @@ pub async fn recv_iq( query: Some(iq::Query::DiscoInfo(disco)), errors: vec![], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; info!("replied to disco#info request from {}", from); Ok(None) } _ => { warn!("received unsupported iq get from {}", from); let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -345,7 +669,11 @@ pub async fn recv_iq( query: None, errors: vec![StanzaError::ServiceUnavailable.into()], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; warn!("replied to unsupported iq get from {}", from); Ok(None) } // stanza::client::iq::Query::Bind(bind) => todo!(), @@ -357,7 +685,7 @@ pub async fn recv_iq( } else { info!("received malformed iq query from {}", from); let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -365,7 +693,11 @@ pub async fn recv_iq( query: None, errors: vec![StanzaError::BadRequest.into()], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; info!("replied to malformed iq query from {}", from); Ok(None) } @@ -374,7 +706,8 @@ pub async fn recv_iq( let from = iq .from .clone() - .unwrap_or_else(|| connection.server().clone()); + // .unwrap_or_else(|| connection.server().clone()); + .unwrap_or_else(|| connection.jid().domainpart.parse().unwrap()); if let Some(query) = iq.query { match query { stanza::client::iq::Query::Roster(mut query) => { @@ -401,7 +734,7 @@ pub async fn recv_iq( .await; } let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Result, @@ -416,13 +749,18 @@ pub async fn recv_iq( .handle_error(RosterError::PushReply(e.into()).into()) .await; } - Ok(Some(UpdateMessage::RosterUpdate(contact))) + let user = logic + .db() + .read_user(contact.user_jid.clone()) + .await + .map_err(|e| Into::<RosterError>::into(e))?; + Ok(Some(UpdateMessage::RosterUpdate(contact, user))) } } } else { warn!("received malformed roster push"); let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -430,7 +768,11 @@ pub async fn recv_iq( query: None, errors: vec![StanzaError::NotAcceptable.into()], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; Ok(None) } } @@ -438,7 +780,7 @@ pub async fn recv_iq( _ => { warn!("received unsupported iq set from {}", from); let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -446,7 +788,11 @@ pub async fn recv_iq( query: None, errors: vec![StanzaError::ServiceUnavailable.into()], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; warn!("replied to unsupported iq set from {}", from); Ok(None) } @@ -454,7 +800,7 @@ pub async fn recv_iq( } else { warn!("received malformed iq set from {}", from); let iq = Iq { - from: Some(connection.jid().clone()), + from: Some(connection.jid().clone().into()), id: iq.id, to: iq.from, r#type: IqType::Error, @@ -462,24 +808,28 @@ pub async fn recv_iq( query: None, errors: vec![StanzaError::NotAcceptable.into()], }; - connection.write_handle().write(Stanza::Iq(iq)).await?; + connection + .write_handle() + .write(Stanza::Iq(iq)) + .await + .map_err(|e| Into::<IqError>::into(e))?; Ok(None) } } } } -pub async fn process_stanza( - logic: ClientLogic, +pub async fn process_stanza<Fs: FileStore + Clone>( + logic: ClientLogic<Fs>, stanza: Stanza, connection: Connected, -) -> Result<Option<UpdateMessage>, Error> { +) -> Result<Option<UpdateMessage>, Error<Fs>> { let update = match stanza { Stanza::Message(stanza_message) => Ok(recv_message(logic, stanza_message).await?), Stanza::Presence(presence) => Ok(recv_presence(presence).await?), Stanza::Iq(iq) => Ok(recv_iq(logic, connection.clone(), iq).await?), // unreachable, always caught by lampada - // TODO: make cleaner than this in some way + // TODO: make cleaner than this in some way, by just reexporting a stanza enum from lampada ig. Stanza::Error(error) => { unreachable!() } |