feat(filamento): user avatar publishing and processing

1 files changed, 308 insertions, 54 deletions

diff --git a/filamento/src/logic/online.rs b/filamento/src/logic/online.rs
index b069f59..745adc1 100644
--- a/filamento/src/logic/online.rs
+++ b/filamento/src/logic/online.rs
@@ -1,23 +1,24 @@
+use std::io::Cursor;
+
+use base64::{prelude::BASE64_STANDARD, Engine};
 use chrono::Utc;
+use image::ImageReader;
 use jid::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 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}, Command, UpdateMessage
 };
 
 use super::{
@@ -29,7 +30,7 @@ use super::{
     },
 };
 
-pub async fn handle_online(logic: ClientLogic, command: Command, connection: Connected) {
+pub async fn handle_online<Fs: FileStore + Clone>(logic: ClientLogic<Fs>, command: Command<Fs>, connection: Connected) {
     let result = handle_online_result(&logic, command, connection).await;
     match result {
         Ok(_) => {}
@@ -37,8 +38,8 @@ 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();
@@ -96,8 +97,8 @@ pub async fn handle_get_roster(
     }
 }
 
-pub async fn handle_add_contact(
-    logic: &ClientLogic,
+pub async fn handle_add_contact<Fs: FileStore + Clone>(
+    logic: &ClientLogic<Fs>,
     connection: Connected,
     jid: JID,
 ) -> Result<(), RosterError> {
@@ -154,8 +155,8 @@ 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,
 ) -> Result<(), SubscribeError> {
@@ -177,8 +178,8 @@ 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,
 ) -> Result<(), SubscribeError> {
@@ -195,8 +196,8 @@ 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,
 ) -> Result<(), SubscribeError> {
@@ -218,8 +219,8 @@ 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,
 ) -> Result<(), SubscribeError> {
@@ -274,8 +275,8 @@ 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,
 ) -> Result<(), RosterError> {
@@ -333,8 +334,8 @@ 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,
     contact_update: ContactUpdate,
@@ -398,8 +399,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,9 +412,18 @@ 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: JID, 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).await {
+        Ok(have_chatted) => {
+            have_chatted
+        },
+        Err(e) => {
+            error!("{}", e);
+            false
+        },
+    };
 
     let nick;
     let mark_chat_as_chatted;
@@ -506,6 +516,7 @@ pub async fn handle_send_message(logic: &ClientLogic, connection: Connected, jid
                 })
                 .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())
@@ -540,8 +551,8 @@ 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>,
@@ -611,8 +622,8 @@ 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>,
@@ -680,20 +691,60 @@ 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 {
@@ -726,38 +777,229 @@ pub async fn handle_publish(
                         if let Some(query) = query {
                             match query {
                                 Query::Pubsub(_) => Ok(()),
-                                q => Err(PublishError::MismatchedQuery(q)),
+                                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(
+                Err(PEPError::IncorrectEntity(
                     from.unwrap_or_else(|| connection.jid().as_bare()),
                 ))
             }
         }
-        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_get_pep_item<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, connection: Connected, jid: Option<JID>, node: String, id: String) -> Result<pep::Item, PEPError> {
+    let stanza_id = Uuid::new_v4().to_string();
+    let request = Iq {
+        from: Some(connection.jid().clone()),
+        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().as_bare())
+                } 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(PEPError::MissingQuery)
+                        }
+                    }
+                    IqType::Error => {
+                        Err(PEPError::StanzaErrors(errors))
+                    }
+                    _ => unreachable!(),
+                }
+            } else {
+                // TODO: include expected entity
+                Err(PEPError::IncorrectEntity(
+                    from.unwrap_or_else(|| connection.jid().as_bare()),
+                ))
+            }
+        }
+        s => Err(PEPError::UnexpectedStanza(s)),
+    }
+}
+
+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>(logic: &ClientLogic<Fs>, img_data: Option<Vec<u8>>) -> Result<(), AvatarPublishError<Fs>> {
+    match img_data {
+        // set avatar
+        Some(data) => {
+            // 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();
+            image.write_to(&mut Cursor::new(&mut data_png), image::ImageFormat::Png)?;
+
+            // 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 = BASE64_STANDARD.encode(sha1_result);
+
+            // encode the image data as base64
+            let data_b64 = BASE64_STANDARD.encode(data_png.clone());
+
+            // 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/png".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()),
+        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().as_bare())
+             {
+                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().as_bare()),
+                ))
+            }
+        }
+        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>(
+    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;
@@ -854,14 +1096,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(())
 }