aboutsummaryrefslogtreecommitdiffstats
path: root/filamento/src/logic
diff options
context:
space:
mode:
Diffstat (limited to 'filamento/src/logic')
-rw-r--r--filamento/src/logic/connect.rs4
-rw-r--r--filamento/src/logic/local_only.rs40
-rw-r--r--filamento/src/logic/mod.rs10
-rw-r--r--filamento/src/logic/offline.rs62
-rw-r--r--filamento/src/logic/online.rs258
-rw-r--r--filamento/src/logic/process_stanza.rs203
6 files changed, 408 insertions, 169 deletions
diff --git a/filamento/src/logic/connect.rs b/filamento/src/logic/connect.rs
index 37cdad5..6e392f1 100644
--- a/filamento/src/logic/connect.rs
+++ b/filamento/src/logic/connect.rs
@@ -11,7 +11,7 @@ use crate::{
use super::ClientLogic;
-pub async fn handle_connect<Fs: FileStore + Clone + Send + Sync>(
+pub async fn handle_connect<Fs: FileStore + Clone + Send + Sync + 'static>(
logic: ClientLogic<Fs>,
connection: Connected,
) {
@@ -19,7 +19,7 @@ pub async fn handle_connect<Fs: FileStore + Clone + Send + Sync>(
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/local_only.rs b/filamento/src/logic/local_only.rs
index cabbef4..7f3a2e6 100644
--- a/filamento/src/logic/local_only.rs
+++ b/filamento/src/logic/local_only.rs
@@ -1,4 +1,4 @@
-use jid::JID;
+use jid::{BareJID, JID};
use uuid::Uuid;
use crate::{
@@ -28,23 +28,53 @@ pub async fn handle_get_chats_ordered_with_latest_messages<Fs: FileStore + Clone
Ok(logic.db().read_chats_ordered_with_latest_messages().await?)
}
+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: JID,
+ jid: BareJID,
) -> Result<Chat, DatabaseError> {
Ok(logic.db().read_chat(jid).await?)
}
+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: JID,
+ jid: BareJID,
) -> Result<Vec<Message>, DatabaseError> {
Ok(logic.db().read_message_history(jid).await?)
}
+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: JID,
+ jid: BareJID,
) -> Result<(), DatabaseError> {
Ok(logic.db().delete_chat(jid).await?)
}
@@ -58,7 +88,7 @@ pub async fn handle_delete_messaage<Fs: FileStore + Clone>(
pub async fn handle_get_user<Fs: FileStore + Clone>(
logic: &ClientLogic<Fs>,
- jid: JID,
+ 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 5e05dac..ddf0343 100644
--- a/filamento/src/logic/mod.rs
+++ b/filamento/src/logic/mod.rs
@@ -1,6 +1,6 @@
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};
@@ -25,7 +25,7 @@ mod process_stanza;
#[derive(Clone)]
pub struct ClientLogic<Fs: FileStore> {
client: Client<Fs>,
- bare_jid: JID,
+ jid: BareJID,
db: Db,
pending: Pending,
update_sender: mpsc::Sender<UpdateMessage>,
@@ -80,7 +80,7 @@ impl Pending {
impl<Fs: FileStore> ClientLogic<Fs> {
pub fn new(
client: Client<Fs>,
- bare_jid: JID,
+ jid: BareJID,
db: Db,
update_sender: mpsc::Sender<UpdateMessage>,
file_store: Fs,
@@ -90,7 +90,7 @@ impl<Fs: FileStore> ClientLogic<Fs> {
pending: Pending::new(),
update_sender,
client,
- bare_jid,
+ jid,
file_store,
}
}
@@ -127,7 +127,7 @@ impl<Fs: FileStore> ClientLogic<Fs> {
}
}
-impl<Fs: FileStore + Clone + Send + Sync> Logic for ClientLogic<Fs> {
+impl<Fs: FileStore + Clone + Send + Sync + 'static> Logic for ClientLogic<Fs> {
type Cmd = Command<Fs>;
// pub async fn handle_stream_error(self, error) {}
diff --git a/filamento/src/logic/offline.rs b/filamento/src/logic/offline.rs
index 566972c..aa84f3d 100644
--- a/filamento/src/logic/offline.rs
+++ b/filamento/src/logic/offline.rs
@@ -1,7 +1,9 @@
use std::process::id;
use chrono::Utc;
+use jid::FullJID;
use lampada::error::WriteError;
+use tracing::error;
use uuid::Uuid;
use crate::{
@@ -14,14 +16,16 @@ use crate::{
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,
},
};
@@ -47,6 +51,12 @@ pub async fn handle_get_roster<Fs: FileStore + Clone>(
Ok(logic.db().read_cached_roster().await?)
}
+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>,
@@ -56,6 +66,10 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
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);
@@ -68,14 +82,30 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
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);
@@ -128,7 +158,7 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
let message = Message {
id,
- from: logic.bare_jid.clone(),
+ from: logic.jid.clone(),
// TODO: failure reason
delivery: Some(Delivery::Failed),
timestamp,
@@ -138,11 +168,15 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
// 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
{
@@ -151,11 +185,25 @@ pub async fn handle_offline_result<Fs: FileStore + Clone>(
.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;
}
diff --git a/filamento/src/logic/online.rs b/filamento/src/logic/online.rs
index d9441d7..b36f9a9 100644
--- a/filamento/src/logic/online.rs
+++ b/filamento/src/logic/online.rs
@@ -3,7 +3,7 @@ use std::{io::Cursor, time::Duration};
use base64::{prelude::BASE64_STANDARD, Engine};
use chrono::Utc;
use image::ImageReader;
-use jid::JID;
+use jid::{BareJID, JID};
use lampada::{Connected, WriteMessage, error::WriteError};
use sha1::{Digest, Sha1};
use stanza::{
@@ -11,26 +11,25 @@ use stanza::{
iq::{self, Iq, IqType, Query}, Stanza
}, 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::{
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
+ }, 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<Fs: FileStore + Clone>(logic: ClientLogic<Fs>, command: Command<Fs>, 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(_) => {}
@@ -44,7 +43,7 @@ pub async fn handle_get_roster<Fs: FileStore + Clone>(
) -> 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,
@@ -97,14 +96,79 @@ pub async fn handle_get_roster<Fs: FileStore + Clone>(
}
}
+// 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,
@@ -158,9 +222,10 @@ pub async fn handle_add_contact<Fs: FileStore + Clone>(
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()),
@@ -181,13 +246,13 @@ pub async fn handle_buddy_request<Fs: FileStore + Clone>(
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()
@@ -199,15 +264,16 @@ pub async fn handle_subscription_request<Fs: FileStore + Clone>(
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),
@@ -222,14 +288,11 @@ pub async fn handle_accept_buddy_request<Fs: FileStore + Clone>(
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?;
@@ -238,10 +301,10 @@ pub async fn handle_accept_subscription_request<Fs: FileStore + Clone>(
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()
});
@@ -249,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()
});
@@ -259,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),
@@ -278,11 +342,11 @@ pub async fn handle_unfriend_contact(connection: Connected, jid: JID) -> Result<
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,
@@ -337,7 +401,7 @@ pub async fn handle_delete_contact<Fs: FileStore + Clone>(
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();
@@ -348,7 +412,8 @@ pub async fn handle_update_contact<Fs: FileStore + Clone>(
.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,
@@ -412,10 +477,10 @@ pub async fn handle_set_status<Fs: FileStore + Clone>(
Ok(())
}
-pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, 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 = match logic.db().upsert_chat_and_user(&jid).await {
+ let have_chatted = match logic.db().upsert_chat_and_user(jid.clone()).await {
Ok(have_chatted) => {
have_chatted
},
@@ -428,7 +493,7 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
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;
@@ -451,7 +516,7 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
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),
@@ -461,7 +526,7 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
// 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?
@@ -470,20 +535,33 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
.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 ?
@@ -508,11 +586,15 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
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 {
@@ -530,6 +612,7 @@ pub async fn handle_send_message<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
.send(UpdateMessage::MessageDelivery {
id,
delivery: Delivery::Failed,
+ chat: jid,
})
.await;
logic.handle_error(MessageSendError::Write(e).into()).await;
@@ -559,7 +642,7 @@ pub async fn handle_disco_info<Fs: FileStore + Clone>(
) -> 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,
@@ -587,7 +670,7 @@ pub async fn handle_disco_info<Fs: FileStore + Clone>(
}) 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
}
@@ -614,7 +697,7 @@ pub async fn handle_disco_info<Fs: FileStore + Clone>(
}
} else {
Err(DiscoError::IncorrectEntity(
- from.unwrap_or_else(|| connection.jid().as_bare()),
+ from.unwrap_or_else(|| connection.jid().to_bare().into()),
))
}
}
@@ -630,7 +713,7 @@ pub async fn handle_disco_items<Fs: FileStore + Clone>(
) -> 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,
@@ -656,7 +739,7 @@ pub async fn handle_disco_items<Fs: FileStore + Clone>(
}) 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
}
@@ -683,7 +766,7 @@ pub async fn handle_disco_items<Fs: FileStore + Clone>(
}
} else {
Err(DiscoError::IncorrectEntity(
- from.unwrap_or_else(|| connection.jid().as_bare()),
+ from.unwrap_or_else(|| connection.jid().to_bare().into()),
))
}
}
@@ -748,7 +831,7 @@ pub async fn handle_publish_pep_item<Fs: FileStore + Clone>(
},
};
let request = Iq {
- from: Some(connection.jid().clone()),
+ from: Some(connection.jid().clone().into()),
id: id.clone(),
to: None,
r#type: IqType::Set,
@@ -770,7 +853,7 @@ pub async fn handle_publish_pep_item<Fs: FileStore + Clone>(
// 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 => {
@@ -790,7 +873,7 @@ pub async fn handle_publish_pep_item<Fs: FileStore + Clone>(
}
} else {
Err(PEPError::IncorrectEntity(
- from.unwrap_or_else(|| connection.jid().as_bare()),
+ from.unwrap_or_else(|| connection.jid().to_bare().into()),
))
}
}
@@ -798,10 +881,11 @@ pub async fn handle_publish_pep_item<Fs: FileStore + Clone>(
}
}
-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> {
+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()),
+ from: Some(connection.jid().clone().into()),
id: stanza_id.clone(),
to: jid.clone(),
r#type: IqType::Get,
@@ -829,7 +913,7 @@ pub async fn handle_get_pep_item<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
}) 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
}
@@ -875,7 +959,7 @@ pub async fn handle_get_pep_item<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
} else {
// TODO: include expected entity
Err(PEPError::IncorrectEntity(
- from.unwrap_or_else(|| connection.jid().as_bare()),
+ from.unwrap_or_else(|| connection.jid().to_bare().into()),
))
}
}
@@ -888,29 +972,33 @@ pub async fn handle_change_nick<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>,
Ok(())
}
-pub async fn handle_change_avatar<Fs: FileStore + Clone>(logic: &ClientLogic<Fs>, img_data: Option<Vec<u8>>) -> Result<(), AvatarPublishError<Fs>> {
+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) => {
- // load the image data and guess the format
- let image = ImageReader::new(Cursor::new(data)).with_guessed_format()?.decode()?;
+ 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)?;
- // 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 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);
- // 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());
- // 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?;
@@ -944,7 +1032,7 @@ pub async fn handle_delete_pep_node<Fs: FileStore + Clone>(
) -> Result<(), PEPError> {
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: None,
r#type: IqType::Set,
@@ -966,7 +1054,7 @@ pub async fn handle_delete_pep_node<Fs: FileStore + Clone>(
// 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 => {
@@ -987,7 +1075,7 @@ pub async fn handle_delete_pep_node<Fs: FileStore + Clone>(
}
} else {
Err(PEPError::IncorrectEntity(
- from.unwrap_or_else(|| connection.jid().as_bare()),
+ from.unwrap_or_else(|| connection.jid().to_bare().into()),
))
}
}
@@ -996,7 +1084,7 @@ pub async fn handle_delete_pep_node<Fs: FileStore + Clone>(
}
// TODO: could probably macro-ise?
-pub async fn handle_online_result<Fs: FileStore + Clone>(
+pub async fn handle_online_result<Fs: FileStore + Clone + 'static>(
logic: &ClientLogic<Fs>,
command: Command<Fs>,
connection: Connected,
@@ -1006,6 +1094,10 @@ pub async fn handle_online_result<Fs: FileStore + Clone>(
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);
@@ -1018,14 +1110,30 @@ pub async fn handle_online_result<Fs: FileStore + Clone>(
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);
diff --git a/filamento/src/logic/process_stanza.rs b/filamento/src/logic/process_stanza.rs
index 9c49b04..67b0d3f 100644
--- a/filamento/src/logic/process_stanza.rs
+++ b/filamento/src/logic/process_stanza.rs
@@ -20,12 +20,13 @@ use crate::{
UpdateMessage, caps,
chat::{Body, Message},
error::{
- AvatarUpdateError, DatabaseError, Error, IqError, MessageRecvError, PresenceError,
- RosterError,
+ AvatarUpdateError, DatabaseError, Error, IqError, IqProcessError, MessageRecvError,
+ PresenceError, RosterError,
},
files::FileStore,
presence::{Offline, Online, Presence, PresenceType, Show},
roster::Contact,
+ user::User,
};
use super::ClientLogic;
@@ -69,37 +70,47 @@ pub async fn recv_message<Fs: FileStore + Clone>(
// 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
- match logic.db().upsert_chat_and_user(&from).await {
- Ok(_) => {
- 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;
- error!("failed to upsert chat and user")
+ 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) => {
- logic
- .handle_error(Error::MessageRecv(MessageRecvError::MessageHistory(e)))
- .await;
- error!("failed to upsert chat and user")
+ error!("{}", e);
+ User {
+ jid: from.to_bare(),
+ nick: None,
+ avatar: None,
+ }
}
};
@@ -107,7 +118,8 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::Message {
- to: from.as_bare(),
+ to: from.to_bare(),
+ from: from_user,
message,
})
.await;
@@ -116,13 +128,13 @@ pub async fn recv_message<Fs: FileStore + Clone>(
if let Some(nick) = stanza_message.nick {
let nick = nick.0;
if nick.is_empty() {
- match logic.db().delete_user_nick(from.as_bare()).await {
+ match logic.db().delete_user_nick(from.to_bare()).await {
Ok(changed) => {
if changed {
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: None,
})
.await;
@@ -136,7 +148,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: None,
})
.await;
@@ -145,7 +157,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
} else {
match logic
.db()
- .upsert_user_nick(from.as_bare(), nick.clone())
+ .upsert_user_nick(from.to_bare(), nick.clone())
.await
{
Ok(changed) => {
@@ -153,7 +165,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: Some(nick),
})
.await;
@@ -167,7 +179,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: Some(nick),
})
.await;
@@ -190,7 +202,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
if nick.is_empty() {
match logic
.db()
- .delete_user_nick(from.as_bare())
+ .delete_user_nick(from.to_bare())
.await
{
Ok(changed) => {
@@ -198,7 +210,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: None,
})
.await;
@@ -214,7 +226,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: None,
})
.await;
@@ -224,7 +236,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
match logic
.db()
.upsert_user_nick(
- from.as_bare(),
+ from.to_bare(),
nick.clone(),
)
.await
@@ -234,7 +246,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: Some(nick),
})
.await;
@@ -250,7 +262,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
logic
.update_sender()
.send(UpdateMessage::NickChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
nick: Some(nick),
})
.await;
@@ -285,7 +297,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
match logic
.db()
.upsert_user_avatar(
- from.as_bare(),
+ from.to_bare(),
metadata.id.clone(),
)
.await
@@ -314,7 +326,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
}) {
Ok(false) => {
// get data
- let pep_item = logic.client().get_pep_item(Some(from.as_bare()), "urn:xmpp:avatar:data".to_string(), metadata.id.clone()).await.map_err(|err| Into::<AvatarUpdateError<Fs>>::into(err))?;
+ 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", "");
@@ -335,7 +347,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
.update_sender()
.send(
UpdateMessage::AvatarChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
id: Some(
metadata.id.clone(),
),
@@ -362,7 +374,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
.update_sender()
.send(
UpdateMessage::AvatarChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
id: Some(
metadata.id.clone(),
),
@@ -392,7 +404,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
// delete avatar
match logic
.db()
- .delete_user_avatar(from.as_bare())
+ .delete_user_avatar(from.to_bare())
.await
{
Ok((changed, old_avatar)) => {
@@ -410,7 +422,7 @@ pub async fn recv_message<Fs: FileStore + Clone>(
.update_sender()
.send(
UpdateMessage::AvatarChanged {
- jid: from.as_bare(),
+ jid: from.to_bare(),
id: None,
},
)
@@ -470,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 => {
@@ -485,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 => {
@@ -541,11 +548,12 @@ 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 {
@@ -553,17 +561,24 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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) => {
@@ -577,7 +592,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
} 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) {
@@ -587,7 +602,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
}
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,
@@ -596,14 +611,18 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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,
@@ -612,14 +631,18 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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,
@@ -627,14 +650,18 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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,
@@ -642,7 +669,11 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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!(),
@@ -654,7 +685,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
} 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,
@@ -662,7 +693,11 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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)
}
@@ -671,7 +706,8 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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) => {
@@ -698,7 +734,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
.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,
@@ -713,13 +749,18 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
.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,
@@ -727,7 +768,11 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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)
}
}
@@ -735,7 +780,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
_ => {
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,
@@ -743,7 +788,11 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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)
}
@@ -751,7 +800,7 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
} 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,
@@ -759,7 +808,11 @@ pub async fn recv_iq<Fs: FileStore + Clone>(
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)
}
}
@@ -776,7 +829,7 @@ pub async fn process_stanza<Fs: FileStore + Clone>(
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!()
}
generated by cgit (git 2.34.1) at 2025-06-30 21:09:03 +0000