use std::str::FromStr; use chrono::Utc; use lampada::{Connected, SupervisorSender}; use stanza::{ client::{ Stanza, iq::{self, Iq, IqType}, }, stanza_error::Error as StanzaError, xep_0030, }; use tracing::{debug, error, info, warn}; use uuid::Uuid; use crate::{ UpdateMessage, chat::{Body, Message}, error::{DatabaseError, Error, IqError, MessageRecvError, PresenceError, RosterError}, presence::{Offline, Online, Presence, PresenceType, Show}, roster::Contact, }; use super::ClientLogic; pub async fn handle_stanza(logic: ClientLogic, stanza: Stanza, connection: Connected) { let result = process_stanza(logic.clone(), stanza, connection).await; match result { Ok(u) => match u { _ => { if let Some(u) = u { logic.handle_update(u).await } } }, Err(e) => logic.handle_error(e).await, } } pub async fn recv_message( logic: ClientLogic, stanza_message: stanza::client::message::Message, ) -> Result, MessageRecvError> { if let Some(mut from) = stanza_message.from { // TODO: don't ignore delay from. xep says SHOULD send error if incorrect. let timestamp = stanza_message .delay .map(|delay| delay.stamp) .unwrap_or_else(|| Utc::now()); // TODO: group chat messages let mut message = Message { id: stanza_message .id // TODO: proper id storage .map(|id| Uuid::from_str(&id).unwrap_or_else(|_| Uuid::new_v4())) .unwrap_or_else(|| Uuid::new_v4()), from: from.clone(), timestamp, body: Body { // TODO: should this be an option? body: stanza_message .body .map(|body| body.body) .unwrap_or_default() .unwrap_or_default(), }, }; // TODO: can this be more efficient? logic .db() .create_message_with_user_resource_and_chat(message.clone(), from.clone()) .await .map_err(|e| DatabaseError(e.into()))?; message.from = message.from.as_bare(); from = from.as_bare(); Ok(Some(UpdateMessage::Message { to: from, message })) } else { Err(MessageRecvError::MissingFrom) } } pub async fn recv_presence( presence: stanza::client::presence::Presence, ) -> Result, PresenceError> { if let Some(from) = presence.from { match presence.r#type { Some(r#type) => match r#type { // error processing a presence from somebody 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"), )) } // should not happen (error to server) stanza::client::presence::PresenceType::Probe => { // TODO: should probably write an error and restart stream Err(PresenceError::Unsupported) } 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 Ok(Some(UpdateMessage::SubscriptionRequest(from))) } stanza::client::presence::PresenceType::Unavailable => { let offline = Offline { status: presence.status.map(|status| status.status.0), }; let timestamp = presence .delay .map(|delay| delay.stamp) .unwrap_or_else(|| Utc::now()); Ok(Some(UpdateMessage::Presence { from, presence: Presence { timestamp, presence: PresenceType::Offline(offline), }, })) } // for now, do nothing, as these are simply informational. will receive roster push from the server regarding the changes to do with them. stanza::client::presence::PresenceType::Subscribed => Ok(None), stanza::client::presence::PresenceType::Unsubscribe => Ok(None), stanza::client::presence::PresenceType::Unsubscribed => Ok(None), }, None => { let online = Online { show: presence.show.map(|show| match show { stanza::client::presence::Show::Away => Show::Away, stanza::client::presence::Show::Chat => Show::Chat, stanza::client::presence::Show::Dnd => Show::DoNotDisturb, stanza::client::presence::Show::Xa => Show::ExtendedAway, }), status: presence.status.map(|status| status.status.0), priority: presence.priority.map(|priority| priority.0), }; let timestamp = presence .delay .map(|delay| delay.stamp) .unwrap_or_else(|| Utc::now()); Ok(Some(UpdateMessage::Presence { from, presence: Presence { timestamp, presence: PresenceType::Online(online), }, })) } } } else { Err(PresenceError::MissingFrom) } } pub async fn recv_iq( logic: ClientLogic, connection: Connected, iq: Iq, ) -> Result, IqError> { if let Some(to) = &iq.to { if *to == *connection.jid() { } else { return Err(IqError::IncorrectAddressee(to.clone())); } } match iq.r#type { stanza::client::iq::IqType::Error | stanza::client::iq::IqType::Result => { let from = iq .from .clone() .unwrap_or_else(|| connection.server().clone()); let id = iq.id.clone(); debug!("received iq result with id `{}` from {}", id, from); logic.pending().respond(Stanza::Iq(iq), id).await?; Ok(None) } stanza::client::iq::IqType::Get => { let from = iq .from .clone() .unwrap_or_else(|| connection.server().clone()); if let Some(query) = iq.query { match query { stanza::client::iq::Query::DiscoInfo(_query) => { // TODO: should this only be replied to server? info!("received disco#info request from {}", from); let disco = xep_0030::info::Query { node: None, features: vec![xep_0030::info::Feature { var: "http://jabber.org/protocol/disco#info".to_string(), }], identities: vec![xep_0030::info::Identity { category: "client".to_string(), name: Some("filamento".to_string()), r#type: "pc".to_string(), }], }; let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Result, lang: None, query: Some(iq::Query::DiscoInfo(disco)), errors: vec![], }; connection.write_handle().write(Stanza::Iq(iq)).await?; 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()), id: iq.id, to: iq.from, r#type: IqType::Error, lang: None, query: None, errors: vec![StanzaError::ServiceUnavailable.into()], }; connection.write_handle().write(Stanza::Iq(iq)).await?; warn!("replied to unsupported iq get from {}", from); Ok(None) } // stanza::client::iq::Query::Bind(bind) => todo!(), // stanza::client::iq::Query::DiscoItems(query) => todo!(), // stanza::client::iq::Query::Ping(ping) => todo!(), // stanza::client::iq::Query::Roster(query) => todo!(), // stanza::client::iq::Query::Unsupported => todo!(), } } else { info!("received malformed iq query from {}", from); let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Error, lang: None, query: None, errors: vec![StanzaError::BadRequest.into()], }; connection.write_handle().write(Stanza::Iq(iq)).await?; info!("replied to malformed iq query from {}", from); Ok(None) } } stanza::client::iq::IqType::Set => { let from = iq .from .clone() .unwrap_or_else(|| connection.server().clone()); if let Some(query) = iq.query { match query { stanza::client::iq::Query::Roster(mut query) => { // TODO: should only have one, otherwise send error // if let Some(item) = query.items.pop() && query.items.len() == 1 { if let Some(item) = query.items.pop() { match item.subscription { Some(stanza::roster::Subscription::Remove) => { if let Err(e) = logic.db().delete_contact(item.jid.clone()).await { logic .handle_error(RosterError::Cache(e.into()).into()) .await; } Ok(Some(UpdateMessage::RosterDelete(item.jid))) } _ => { let contact: Contact = item.into(); if let Err(e) = logic.db().upsert_contact(contact.clone()).await { logic .handle_error(RosterError::Cache(e.into()).into()) .await; } let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Result, lang: None, query: None, errors: vec![], }; if let Err(e) = connection.write_handle().write(Stanza::Iq(iq)).await { logic .handle_error(RosterError::PushReply(e.into()).into()) .await; } Ok(Some(UpdateMessage::RosterUpdate(contact))) } } } else { warn!("received malformed roster push"); let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Error, lang: None, query: None, errors: vec![StanzaError::NotAcceptable.into()], }; connection.write_handle().write(Stanza::Iq(iq)).await?; Ok(None) } } // TODO: send unsupported to server _ => { warn!("received unsupported iq set from {}", from); let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Error, lang: None, query: None, errors: vec![StanzaError::ServiceUnavailable.into()], }; connection.write_handle().write(Stanza::Iq(iq)).await?; warn!("replied to unsupported iq set from {}", from); Ok(None) } } } else { warn!("received malformed iq set from {}", from); let iq = Iq { from: Some(connection.jid().clone()), id: iq.id, to: iq.from, r#type: IqType::Error, lang: None, query: None, errors: vec![StanzaError::NotAcceptable.into()], }; connection.write_handle().write(Stanza::Iq(iq)).await?; Ok(None) } } } } pub async fn process_stanza( logic: ClientLogic, stanza: Stanza, connection: Connected, ) -> Result, Error> { 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 Stanza::Error(error) => { unreachable!() } // should this cause a stream restart? Stanza::OtherContent(content) => { Err(Error::UnrecognizedContent) // TODO: send error to write_thread } }; update }