forked from lavina/lavina
1
0
Fork 0

Compare commits

..

No commits in common. "fe802aca2625ca31e9902afc21eba89a2b216cc9" and "89918d9de1fac1e5d024af7737d3db41afa9e5ab" have entirely different histories.

16 changed files with 89 additions and 639 deletions

View File

@ -1,2 +0,0 @@
alter table messages drop column created_at;
alter table messages add column created_at datetime default "1970-01-01T00:00:00Z";

View File

@ -18,7 +18,7 @@ use tracing::{Instrument, Span};
use crate::clustering::room::*;
use crate::prelude::*;
use crate::room::{RoomHandle, RoomId, RoomInfo, StoredMessage};
use crate::room::{RoomHandle, RoomId, RoomInfo};
use crate::table::{AnonTable, Key as AnonKey};
use crate::LavinaCore;
@ -111,14 +111,6 @@ impl PlayerConnection {
Ok(deferred.await?)
}
#[tracing::instrument(skip(self), name = "PlayerConnection::get_room_message_history")]
pub async fn get_room_message_history(&self, room_id: RoomId) -> Result<Vec<StoredMessage>> {
let (promise, deferred) = oneshot();
let cmd = ClientCommand::GetRoomHistory { room_id, promise };
self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
Ok(deferred.await?)
}
/// Handler in [Player::send_dialog_message].
#[tracing::instrument(skip(self, body), name = "PlayerConnection::send_dialog_message")]
pub async fn send_dialog_message(&self, recipient: PlayerId, body: Str) -> Result<()> {
@ -220,10 +212,6 @@ pub enum ClientCommand {
recipient: PlayerId,
promise: Promise<GetInfoResult>,
},
GetRoomHistory {
room_id: RoomId,
promise: Promise<Vec<StoredMessage>>,
},
}
pub enum GetInfoResult {
@ -521,10 +509,6 @@ impl Player {
let result = self.check_user_existence(recipient).await;
let _ = promise.send(result);
}
ClientCommand::GetRoomHistory { room_id, promise } => {
let result = self.get_room_history(room_id).await;
let _ = promise.send(result);
}
}
}
@ -573,23 +557,6 @@ impl Player {
}
}
#[tracing::instrument(skip(self), name = "Player::retrieve_room_history")]
async fn get_room_history(&mut self, room_id: RoomId) -> Vec<StoredMessage> {
let room = self.my_rooms.get(&room_id);
if let Some(room) = room {
match room {
RoomRef::Local(room) => room.get_message_history(&self.services).await,
RoomRef::Remote { node_id } => {
todo!()
}
}
} else {
tracing::error!("Room with ID {room_id:?} not found");
// todo: return error
todo!()
}
}
#[tracing::instrument(skip(self), name = "Player::leave_room")]
async fn leave_room(&mut self, connection_id: ConnectionId, room_id: RoomId) {
let room = self.my_rooms.remove(&room_id);
@ -626,7 +593,7 @@ impl Player {
body: Str,
) -> SendMessageResult {
let Some(room) = self.my_rooms.get(&room_id) else {
tracing::info!("Room with ID {room_id:?} not found");
tracing::info!("no room found");
return SendMessageResult::NoSuchRoom;
};
let created_at = Utc::now();
@ -665,7 +632,7 @@ impl Player {
#[tracing::instrument(skip(self, new_topic), name = "Player::change_room_topic")]
async fn change_room_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) {
let Some(room) = self.my_rooms.get(&room_id) else {
tracing::info!("Room with ID {room_id:?} not found");
tracing::info!("no room found");
return;
};
match room {

View File

@ -3,7 +3,7 @@ use chrono::{DateTime, Utc};
use sqlx::FromRow;
use crate::repo::Storage;
use crate::room::{RoomId, StoredMessage};
use crate::room::RoomId;
#[derive(FromRow)]
pub struct StoredRoom {
@ -29,34 +29,6 @@ impl Storage {
Ok(res)
}
#[tracing::instrument(skip(self), name = "Storage::retrieve_room_message_history")]
pub async fn get_room_message_history(&self, room_id: u32) -> Result<Vec<StoredMessage>> {
let mut executor = self.conn.lock().await;
let res = sqlx::query_as(
"
select
messages.id as id,
content,
created_at,
users.name as author_name
from
messages
join
users
on messages.author_id = users.id
where
room_id = ?
order by
messages.id;
",
)
.bind(room_id)
.fetch_all(&mut *executor)
.await?;
Ok(res)
}
#[tracing::instrument(skip(self, topic), name = "Storage::create_new_room")]
pub async fn create_new_room(&self, name: &str, topic: &str) -> Result<u32> {
let mut executor = self.conn.lock().await;
@ -99,7 +71,7 @@ impl Storage {
.bind(id)
.bind(content)
.bind(author_id)
.bind(created_at)
.bind(created_at.to_string())
.bind(room_id)
.execute(&mut *executor)
.await?;
@ -202,6 +174,6 @@ impl Storage {
.fetch_all(&mut *executor)
.await?;
res.into_iter().map(|(room_id,)| RoomId::try_from(room_id)).collect()
res.into_iter().map(|(room_id,)| RoomId::from(room_id)).collect()
}
}

View File

@ -5,20 +5,18 @@ use std::{collections::HashMap, hash::Hash, sync::Arc};
use chrono::{DateTime, Utc};
use prometheus::{IntGauge, Registry as MetricRegistry};
use serde::Serialize;
use sqlx::sqlite::SqliteRow;
use sqlx::{FromRow, Row};
use tokio::sync::RwLock as AsyncRwLock;
use crate::player::{PlayerHandle, PlayerId, Updates};
use crate::prelude::*;
use crate::{LavinaCore, Services};
use crate::Services;
/// Opaque room id
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
pub struct RoomId(Str);
impl RoomId {
pub fn try_from(str: impl Into<Str>) -> Result<RoomId> {
pub fn from(str: impl Into<Str>) -> Result<RoomId> {
let bytes = str.into();
if bytes.len() > 32 {
return Err(anyhow::Error::msg("Room name cannot be longer than 32 symbols"));
@ -160,10 +158,6 @@ impl RoomHandle {
lock.broadcast_update(update, player_id).await;
}
pub async fn get_message_history(&self, services: &LavinaCore) -> Vec<StoredMessage> {
return services.storage.get_room_message_history(self.0.read().await.storage_id).await.unwrap();
}
#[tracing::instrument(skip(self), name = "RoomHandle::unsubscribe")]
pub async fn unsubscribe(&self, player_id: &PlayerId) {
let mut lock = self.0.write().await;
@ -285,11 +279,3 @@ pub struct RoomInfo {
pub members: Vec<PlayerId>,
pub topic: Str,
}
#[derive(Debug, FromRow)]
pub struct StoredMessage {
pub id: u32,
pub author_name: String,
pub content: String,
pub created_at: DateTime<Utc>,
}

View File

@ -720,7 +720,7 @@ async fn handle_incoming_message(
}
ClientMessage::PrivateMessage { recipient, body } => match recipient {
Recipient::Chan(Chan::Global(chan)) => {
let room_id = RoomId::try_from(chan)?;
let room_id = RoomId::from(chan)?;
user_handle.send_message(room_id, body).await?;
}
Recipient::Nick(nick) => {
@ -732,7 +732,7 @@ async fn handle_incoming_message(
ClientMessage::Topic { chan, topic } => {
match chan {
Chan::Global(chan) => {
let room_id = RoomId::try_from(chan)?;
let room_id = RoomId::from(chan)?;
user_handle.change_topic(room_id.clone(), topic.clone()).await?;
ServerMessage {
tags: vec![],
@ -774,7 +774,7 @@ async fn handle_incoming_message(
writer.flush().await?;
}
Recipient::Chan(Chan::Global(chan)) => {
let room = core.get_room(&RoomId::try_from(chan.clone())?).await;
let room = core.get_room(&RoomId::from(chan.clone())?).await;
if let Some(room) = room {
let room_info = room.get_room_info().await;
for member in room_info.members {
@ -893,33 +893,25 @@ async fn handle_join(
) -> Result<()> {
match chan {
Chan::Global(chan_name) => {
let room_id = RoomId::try_from(chan_name.clone())?;
match user_handle.join_room(room_id).await? {
JoinResult::Success(room_info) => {
produce_on_join_cmd_messages(&config, &user, chan, &room_info, writer).await?;
}
JoinResult::AlreadyJoined => {
// we do nothing on repeated joins
}
JoinResult::Banned => {
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N474BannedFromChan {
client: user.nickname.clone(),
chan: chan.clone(),
message: "U dun goofed".into(),
},
}
.write_async(writer)
.await?;
let room_id = RoomId::from(chan_name.clone())?;
if let JoinResult::Success(room_info) = user_handle.join_room(room_id).await? {
produce_on_join_cmd_messages(&config, &user, chan, &room_info, writer).await?;
} else {
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N474BannedFromChan {
client: user.nickname.clone(),
chan: chan.clone(),
message: "U dun goofed".into(),
},
}
.write_async(writer)
.await?;
}
writer.flush().await?;
}
Chan::Local(_) => {
// TODO handle join attempts to local chans with an error, we don't support these
}
Chan::Local(_) => {}
};
Ok(())
}
@ -932,7 +924,7 @@ async fn handle_part(
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
if let Chan::Global(chan_name) = chan {
let room_id = RoomId::try_from(chan_name.clone())?;
let room_id = RoomId::from(chan_name.clone())?;
user_handle.leave_room(room_id).await?;
ServerMessage {
tags: vec![],

View File

@ -619,14 +619,14 @@ async fn server_time_capability() -> Result<()> {
server.core.create_player(&PlayerId::from("some_guy")?).await?;
let mut conn = server.core.connect_to_player(&PlayerId::from("some_guy").unwrap()).await;
let res = conn.join_room(RoomId::try_from("test").unwrap()).await?;
let res = conn.join_room(RoomId::from("test").unwrap()).await?;
let JoinResult::Success(_) = res else {
panic!("Failed to join room");
};
s.expect(":some_guy JOIN #test").await?;
let SendMessageResult::Success(res) = conn.send_message(RoomId::try_from("test").unwrap(), "Hello".into()).await?
let SendMessageResult::Success(res) = conn.send_message(RoomId::from("test").unwrap(), "Hello".into()).await?
else {
panic!("Failed to send message");
};

View File

@ -16,7 +16,6 @@ use crate::proto::IqClientBody;
use crate::XmppConnection;
impl<'a> XmppConnection<'a> {
#[tracing::instrument(skip(self, output, iq), name = "XmppConnection::handle_iq")]
pub async fn handle_iq(&self, output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>) {
match iq.body {
IqClientBody::Bind(req) => {
@ -113,7 +112,6 @@ impl<'a> XmppConnection<'a> {
}
}
#[tracing::instrument(skip(self), name = "XmppConnection::bind")]
pub(crate) async fn bind(&self, req: &BindRequest) -> BindResponse {
BindResponse(Jid {
name: Some(self.user.xmpp_name.clone()),
@ -122,7 +120,6 @@ impl<'a> XmppConnection<'a> {
})
}
#[tracing::instrument(skip(self), name = "XmppConnection::disco_info")]
async fn disco_info(&self, to: Option<&Jid>, req: &InfoQuery) -> Result<InfoQuery, IqError> {
let identity;
let feature;
@ -166,7 +163,7 @@ impl<'a> XmppConnection<'a> {
server,
resource: None,
}) if server.0 == self.hostname_rooms => {
let room_id = RoomId::try_from(room_name.0.clone()).unwrap();
let room_id = RoomId::from(room_name.0.clone()).unwrap();
let Some(_) = self.core.get_room(&room_id).await else {
// TODO should return item-not-found
// example:
@ -202,7 +199,6 @@ impl<'a> XmppConnection<'a> {
})
}
#[tracing::instrument(skip(self, core), name = "XmppConnection::disco_items")]
async fn disco_items(&self, to: Option<&Jid>, req: &ItemQuery, core: &LavinaCore) -> ItemQuery {
let item = match to {
Some(Jid {

View File

@ -12,7 +12,6 @@ use proto_xmpp::xml::{Ignore, ToXml};
use crate::XmppConnection;
impl<'a> XmppConnection<'a> {
#[tracing::instrument(skip(self, output, m), name = "XmppConnection::message")]
pub async fn handle_message(&mut self, output: &mut Vec<Event<'static>>, m: Message<Ignore>) -> Result<()> {
if let Some(Jid {
name: Some(name),
@ -21,8 +20,7 @@ impl<'a> XmppConnection<'a> {
}) = m.to
{
if server.0.as_ref() == &*self.hostname_rooms && m.r#type == MessageType::Groupchat {
let Some(body) = &m.body else { return Ok(()) };
self.user_handle.send_message(RoomId::try_from(name.0.clone())?, body.clone()).await?;
self.user_handle.send_message(RoomId::from(name.0.clone())?, m.body.clone().into()).await?;
Message::<()> {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
@ -44,8 +42,7 @@ impl<'a> XmppConnection<'a> {
.serialize(output);
Ok(())
} else if server.0.as_ref() == &*self.hostname && m.r#type == MessageType::Chat {
let Some(body) = &m.body else { return Ok(()) };
self.user_handle.send_dialog_message(PlayerId::from(name.0.clone())?, body.clone()).await?;
self.user_handle.send_dialog_message(PlayerId::from(name.0.clone())?, m.body.clone()).await?;
Ok(())
} else {
todo!()

View File

@ -4,15 +4,13 @@ use anyhow::Result;
use quick_xml::events::Event;
use lavina_core::room::RoomId;
use proto_xmpp::bind::{Jid, Name, Resource, Server};
use proto_xmpp::client::{Message, MessageType, Presence, Subject};
use proto_xmpp::muc::{Affiliation, Delay, Role, XUser, XUserItem, XmppHistoryMessage};
use proto_xmpp::bind::{Jid, Name, Server};
use proto_xmpp::client::Presence;
use proto_xmpp::xml::{Ignore, ToXml};
use crate::XmppConnection;
impl<'a> XmppConnection<'a> {
#[tracing::instrument(skip(self, output, p), name = "XmppConnection::handle_presence")]
pub async fn handle_presence(&mut self, output: &mut Vec<Event<'static>>, p: Presence<Ignore>) -> Result<()> {
match p.to {
None => {
@ -23,17 +21,10 @@ impl<'a> XmppConnection<'a> {
server,
// resources in MUCs are members' personas not implemented (yet?)
resource: Some(_),
}) if server.0 == self.hostname_rooms => match p.r#type.as_deref() {
None => {
self.join_muc(output, p.id, name).await?;
}
Some("unavailable") => {
self.leave_muc(output, p.id, name).await?;
}
_ => {
tracing::error!("Unimplemented case")
}
},
}) if server.0 == self.hostname_rooms => {
let response = self.muc_presence(&name).await?;
response.serialize(output);
}
_ => {
// TODO other presence cases
let response = Presence::<()>::default();
@ -43,75 +34,6 @@ impl<'a> XmppConnection<'a> {
Ok(())
}
async fn join_muc(&mut self, output: &mut Vec<Event<'static>>, id: Option<String>, name: Name) -> Result<()> {
// Response presence
let mut muc_presence = self.retrieve_muc_presence(&name).await?;
muc_presence.id = id;
muc_presence.serialize(output);
// N last messages from the room history before the user joined
let messages = self.retrieve_message_history(&name).await?;
for message in messages {
message.serialize(output)
}
// The subject is the last stanza sent during a MUC join process.
let subject = Message::<()> {
from: Some(Jid {
name: Some(name.clone()),
server: Server(self.hostname_rooms.clone()),
resource: None,
}),
id: None,
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
resource: Some(self.user.xmpp_resource.clone()),
}),
r#type: MessageType::Groupchat,
lang: None,
subject: Some(Subject(None)),
body: None,
custom: vec![],
};
subject.serialize(output);
Ok(())
}
async fn leave_muc(&mut self, output: &mut Vec<Event<'static>>, id: Option<String>, name: Name) -> Result<()> {
self.user_handle.leave_room(RoomId::try_from(name.0.clone())?).await?;
let response = Presence {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
resource: Some(self.user.xmpp_resource.clone()),
}),
from: Some(Jid {
name: Some(name.clone()),
server: Server(self.hostname_rooms.clone()),
resource: Some(self.user.xmpp_muc_name.clone()),
}),
r#type: Some("unavailable".into()),
custom: vec![XUser {
item: XUserItem {
affiliation: Affiliation::Member,
role: Role::None,
jid: Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
resource: Some(self.user.xmpp_resource.clone()),
},
},
self_presence: true,
just_created: false,
}],
..Default::default()
};
response.serialize(output);
Ok(())
}
#[tracing::instrument(skip(self, output, r#type), name = "XmppConnection::self_presence")]
async fn self_presence(&mut self, output: &mut Vec<Event<'static>>, r#type: Option<&str>) {
match r#type {
Some("unavailable") => {
@ -133,17 +55,15 @@ impl<'a> XmppConnection<'a> {
};
response.serialize(output);
}
e => {
tracing::error!("TODO: unknown presence type: {e:?}");
}
_ => todo!(),
}
}
#[tracing::instrument(skip(self), name = "XmppConnection::retrieve_muc_presence")]
async fn retrieve_muc_presence(&mut self, name: &Name) -> Result<Presence<XUser>> {
let _ = self.user_handle.join_room(RoomId::try_from(name.0.clone())?).await?;
// todo: return Presence and serialize on the outside.
async fn muc_presence(&mut self, name: &Name) -> Result<(Presence<()>)> {
let a = self.user_handle.join_room(RoomId::from(name.0.clone())?).await?;
// TODO handle bans
let response = Presence {
let response = Presence::<()> {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
@ -154,78 +74,23 @@ impl<'a> XmppConnection<'a> {
server: Server(self.hostname_rooms.clone()),
resource: Some(self.user.xmpp_muc_name.clone()),
}),
custom: vec![XUser {
item: XUserItem {
affiliation: Affiliation::Member,
role: Role::Participant,
jid: Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
resource: Some(self.user.xmpp_resource.clone()),
},
},
self_presence: true,
just_created: false, // TODO we don't know this for sure at this point
}],
..Default::default()
};
Ok(response)
}
/// Retrieve a room's message history. The output can be serialized into a stream of XML stanzas.
///
/// Example in [XmppHistoryMessage]'s docs.
#[tracing::instrument(skip(self), name = "XmppConnection::retrieve_message_history")]
async fn retrieve_message_history(&self, room_name: &Name) -> Result<Vec<XmppHistoryMessage>> {
let room_id = RoomId::try_from(room_name.0.clone())?;
let history_messages = self.user_handle.get_room_message_history(room_id).await?;
let mut response = vec![];
for history_message in history_messages.into_iter() {
response.push(XmppHistoryMessage {
id: history_message.id.to_string(),
to: Jid {
name: Option::from(Name(self.user.xmpp_muc_name.0.clone().into())),
server: Server(self.hostname.clone()),
resource: None,
},
from: Jid {
name: Option::from(room_name.clone()),
server: Server(self.hostname_rooms.clone()),
resource: Option::from(Resource(history_message.author_name.clone().into())),
},
delay: Delay {
from: Jid {
name: Option::from(Name(history_message.author_name.clone().into())),
server: Server(self.hostname_rooms.clone()),
resource: None,
},
stamp: history_message.created_at.to_rfc3339(),
},
body: history_message.content.clone(),
});
tracing::info!(
"Retrieved message: {:?} {:?}",
history_message.author_name,
history_message.content.clone()
);
}
return Ok(response);
}
}
// todo: set up so that the user has been previously joined.
// todo: first call to muc_presence is OK, next one is OK too.
#[cfg(test)]
mod tests {
use crate::testkit::{expect_user_authenticated, TestServer};
use crate::Authenticated;
use anyhow::Result;
use lavina_core::player::PlayerId;
use proto_xmpp::bind::{Jid, Name, Resource, Server};
use proto_xmpp::client::Presence;
use proto_xmpp::muc::{Affiliation, Role, XUser, XUserItem};
use crate::testkit::{expect_user_authenticated, TestServer};
use crate::Authenticated;
#[tokio::test]
async fn test_muc_joining() -> Result<()> {
@ -244,8 +109,8 @@ mod tests {
let mut player_conn = server.core.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let muc_presence = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence {
let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence::<()> {
to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()),
@ -256,22 +121,9 @@ mod tests {
server: Server(conn.hostname_rooms.clone()),
resource: Some(conn.user.xmpp_muc_name.clone()),
}),
custom: vec![XUser {
item: XUserItem {
affiliation: Affiliation::Member,
role: Role::Participant,
jid: Jid {
name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()),
resource: Some(conn.user.xmpp_resource.clone()),
},
},
self_presence: true,
just_created: false,
}],
..Default::default()
};
assert_eq!(expected, muc_presence);
assert_eq!(expected, response);
server.shutdown().await.unwrap();
Ok(())
@ -296,8 +148,8 @@ mod tests {
let mut player_conn = server.core.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let response = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence {
let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence::<()> {
to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()),
@ -308,19 +160,6 @@ mod tests {
server: Server(conn.hostname_rooms.clone()),
resource: Some(conn.user.xmpp_muc_name.clone()),
}),
custom: vec![XUser {
item: XUserItem {
affiliation: Affiliation::Member,
role: Role::Participant,
jid: Jid {
name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()),
resource: Some(conn.user.xmpp_resource.clone()),
},
},
self_presence: true,
just_created: false,
}],
..Default::default()
};
assert_eq!(expected, response);
@ -331,7 +170,7 @@ mod tests {
let mut player_conn = server.core.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let response = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
assert_eq!(expected, response);
server.shutdown().await.unwrap();

View File

@ -20,8 +20,8 @@ pub struct Message<T> {
// default is Normal
pub r#type: MessageType,
pub lang: Option<Str>,
pub subject: Option<Subject>,
pub body: Option<Str>,
pub subject: Option<Str>,
pub body: Str,
pub custom: Vec<T>,
}
@ -38,20 +38,6 @@ impl<T: FromXml> FromXml for Message<T> {
}
}
#[derive(PartialEq, Eq, Debug)]
pub struct Subject(pub Option<Str>);
impl ToXml for Subject {
fn serialize(&self, events: &mut Vec<Event<'static>>) {
if let Some(ref s) = self.0 {
events.push(Event::Start(BytesStart::new("subject")));
events.push(Event::Text(BytesText::new(s).into_owned()));
events.push(Event::End(BytesEnd::new("subject")));
} else {
events.push(Event::Empty(BytesStart::new("subject")));
}
}
}
#[derive(From)]
struct MessageParser<T: FromXml>(MessageParserInner<T>);
@ -71,7 +57,7 @@ struct MessageParserState<T> {
to: Option<Jid>,
r#type: MessageType,
lang: Option<Str>,
subject: Option<Subject>,
subject: Option<Str>,
body: Option<Str>,
custom: Vec<T>,
}
@ -135,16 +121,22 @@ impl<T: FromXml> Parser for MessageParser<T> {
}
}
}
Event::End(_) => Continuation::Final(Ok(Message {
from: state.from,
id: state.id,
to: state.to,
r#type: state.r#type,
lang: state.lang,
subject: state.subject,
body: state.body,
custom: state.custom,
})),
Event::End(_) => {
if let Some(body) = state.body {
Continuation::Final(Ok(Message {
from: state.from,
id: state.id,
to: state.to,
r#type: state.r#type,
lang: state.lang,
subject: state.subject,
body,
custom: state.custom,
}))
} else {
Continuation::Final(Err(ffail!("Body not found")))
}
}
Event::Empty(_) => {
let parser = T::parse();
match parser.consume(namespace, event) {
@ -161,7 +153,7 @@ impl<T: FromXml> Parser for MessageParser<T> {
InSubject(mut state) => match event {
Event::Text(ref bytes) => {
let subject = fail_fast!(std::str::from_utf8(&*bytes));
state.subject = Some(Subject(Some(subject.into())));
state.subject = Some(subject.into());
Continuation::Continue(InSubject(state).into())
}
Event::End(_) => Continuation::Continue(Outer(state).into()),
@ -216,14 +208,9 @@ impl<T: ToXml> ToXml for Message<T> {
value: self.r#type.as_str().as_bytes().into(),
});
events.push(Event::Start(bytes));
if let Some(subject) = &self.subject {
subject.serialize(events);
}
if let Some(body) = &self.body {
events.push(Event::Start(BytesStart::new("body")));
events.push(Event::Text(BytesText::new(body).into_owned()));
events.push(Event::End(BytesEnd::new("body")));
}
events.push(Event::Start(BytesStart::new("body")));
events.push(Event::Text(BytesText::new(&self.body).into_owned()));
events.push(Event::End(BytesEnd::new("body")));
events.push(Event::End(BytesEnd::new("message")));
}
}
@ -500,7 +487,6 @@ impl<T: ToXml> ToXml for Iq<T> {
#[derive(PartialEq, Eq, Debug)]
pub struct Presence<T> {
pub id: Option<String>,
pub to: Option<Jid>,
pub from: Option<Jid>,
pub priority: Option<PresencePriority>,
@ -513,7 +499,6 @@ pub struct Presence<T> {
impl<T> Default for Presence<T> {
fn default() -> Self {
Self {
id: Default::default(),
to: Default::default(),
from: Default::default(),
priority: Default::default(),
@ -588,10 +573,6 @@ impl<T: FromXml> FromXml for Presence<T> {
let s = std::str::from_utf8(&attr.value)?;
p.r#type = Some(s.into());
}
b"id" => {
let s = std::str::from_utf8(&attr.value)?;
p.id = Option::from(s.to_string());
}
_ => {}
}
}
@ -679,12 +660,6 @@ impl<T: ToXml> ToXml for Presence<T> {
value: from.to_string().as_bytes().into(),
}]);
}
if let Some(ref id) = self.id {
start.extend_attributes([Attribute {
key: QName(b"id"),
value: id.to_string().as_bytes().into(),
}]);
}
events.push(Event::Start(start));
if let Some(ref priority) = self.priority {
let s = priority.0.to_string();
@ -694,9 +669,6 @@ impl<T: ToXml> ToXml for Presence<T> {
Event::End(BytesEnd::new("priority")),
]);
}
for c in &self.custom {
c.serialize(events);
}
events.push(Event::End(BytesEnd::new("presence")));
}
}
@ -723,8 +695,8 @@ mod tests {
}),
r#type: MessageType::Chat,
lang: None,
subject: Some(Subject(Some("daa".into()))),
body: Some("bbb".into()),
subject: Some("daa".into()),
body: "bbb".into(),
custom: vec![Ignore],
}
)
@ -746,8 +718,8 @@ mod tests {
}),
r#type: MessageType::Chat,
lang: None,
subject: Some(Subject(Some("daa".into()))),
body: Some("bbb".into()),
subject: Some("daa".into()),
body: "bbb".into(),
custom: vec![Ignore],
}
)

View File

@ -15,9 +15,6 @@ pub mod streamerror;
pub mod tls;
pub mod xml;
#[cfg(test)]
mod testkit;
// Implemented as a macro instead of a fn due to borrowck limitations
macro_rules! skip_text {
($reader: ident, $buf: ident) => {

View File

@ -1,16 +1,12 @@
#![allow(unused_variables)]
use anyhow::{anyhow, Result};
use quick_xml::events::attributes::Attribute;
use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
use quick_xml::name::{QName, ResolveResult};
use quick_xml::events::Event;
use quick_xml::name::ResolveResult;
use crate::bind::Jid;
use crate::xml::*;
use anyhow::{anyhow, Result};
pub const XMLNS: &'static str = "http://jabber.org/protocol/muc";
pub const XMLNS_USER: &'static str = "http://jabber.org/protocol/muc#user";
pub const XMLNS_DELAY: &'static str = "urn:xmpp:delay";
#[derive(PartialEq, Eq, Debug, Default)]
pub struct History {
@ -147,191 +143,9 @@ impl FromXml for X {
}
}
/// Information about an MUC member. May contain [MUC status codes](https://xmpp.org/registrar/mucstatus.html).
#[derive(Debug, PartialEq, Eq)]
pub struct XUser {
pub item: XUserItem,
/// Code 110. The receiver is the user referred to in the presence stanza.
pub self_presence: bool,
/// Code 201. The room from which the presence stanza was sent was just created.
pub just_created: bool,
}
impl ToXml for XUser {
fn serialize(&self, output: &mut Vec<Event<'static>>) {
let mut tag = BytesStart::new("x");
tag.push_attribute(("xmlns", XMLNS_USER));
output.push(Event::Start(tag));
self.item.serialize(output);
if self.self_presence {
let mut meg = BytesStart::new("status");
meg.push_attribute(("code", "110"));
output.push(Event::Empty(meg));
}
if self.just_created {
let mut meg = BytesStart::new("status");
meg.push_attribute(("code", "201"));
output.push(Event::Empty(meg));
}
output.push(Event::End(BytesEnd::new("x")));
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct XUserItem {
pub affiliation: Affiliation,
pub jid: Jid,
pub role: Role,
}
impl ToXml for XUserItem {
fn serialize(&self, output: &mut Vec<Event<'static>>) {
let mut meg = BytesStart::new("item");
meg.push_attribute(("affiliation", self.affiliation.to_str()));
meg.push_attribute(("role", self.role.to_str()));
meg.push_attribute(("jid", self.jid.to_string().as_str()));
output.push(Event::Empty(meg));
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum Affiliation {
Owner,
Admin,
Member,
Outcast,
None,
}
impl Affiliation {
pub fn from_str(s: &str) -> Option<Self> {
match s {
"owner" => Some(Self::Owner),
"admin" => Some(Self::Admin),
"member" => Some(Self::Member),
"outcast" => Some(Self::Outcast),
"none" => Some(Self::None),
_ => None,
}
}
pub fn to_str(&self) -> &str {
match self {
Self::Owner => "owner",
Self::Admin => "admin",
Self::Member => "member",
Self::Outcast => "outcast",
Self::None => "none",
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum Role {
Moderator,
Participant,
Visitor,
None,
}
impl Role {
pub fn from_str(s: &str) -> Option<Self> {
match s {
"moderator" => Some(Self::Moderator),
"participant" => Some(Self::Participant),
"visitor" => Some(Self::Visitor),
"none" => Some(Self::None),
_ => None,
}
}
pub fn to_str(&self) -> &str {
match self {
Self::Moderator => "moderator",
Self::Participant => "participant",
Self::Visitor => "visitor",
Self::None => "none",
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct Delay {
pub from: Jid,
pub stamp: String,
}
impl ToXml for Delay {
fn serialize(&self, events: &mut Vec<Event>) {
let mut tag = BytesStart::new("delay");
tag.push_attribute(Attribute {
key: QName(b"xmlns"),
value: XMLNS_DELAY.as_bytes().into(),
});
tag.push_attribute(Attribute {
key: QName(b"from"),
value: self.from.to_string().into_bytes().into(),
});
tag.push_attribute(Attribute {
key: QName(b"stamp"),
value: self.stamp.as_bytes().into(),
});
events.push(Event::Empty(tag));
}
}
/// Message-stanza of a historic message.
///
/// Example:
/// ```xml
/// <message from="duqedadi@conference.example.com/misha" xml:lang="en" to="misha@example.com/tux" type="groupchat" id="7ca7cb14-b2af-49c9-bd90-05dabb1113a5">
/// <delay xmlns="urn:xmpp:delay" stamp="2024-05-17T16:05:28.440337Z" from="duqedadi@conference.example.com"/>
/// <body></body>
/// </message>
/// ```
#[derive(Debug, PartialEq, Eq)]
pub struct XmppHistoryMessage {
pub id: String,
pub to: Jid,
pub from: Jid,
pub delay: Delay,
pub body: String,
}
impl ToXml for XmppHistoryMessage {
fn serialize(&self, events: &mut Vec<Event<'static>>) {
let mut message_tag = BytesStart::new("message");
message_tag.push_attribute(Attribute {
key: QName(b"id"),
value: self.id.as_str().as_bytes().into(),
});
message_tag.push_attribute(Attribute {
key: QName(b"to"),
value: self.to.to_string().into_bytes().into(),
});
message_tag.push_attribute(Attribute {
key: QName(b"from"),
value: self.from.to_string().into_bytes().into(),
});
message_tag.push_attribute(Attribute {
key: QName(b"type"),
value: b"groupchat".into(),
});
events.push(Event::Start(message_tag));
self.delay.serialize(events);
let body_tag = BytesStart::new("body");
events.push(Event::Start(body_tag));
events.push(Event::Text(BytesText::new(self.body.to_string().as_str()).into_owned()));
events.push(Event::End(BytesEnd::new("body")));
events.push(Event::End(BytesEnd::new("message")));
}
}
#[cfg(test)]
mod test {
use super::*;
use crate::bind::{Name, Resource, Server};
use crate::testkit::assemble_string_from_event_flow;
#[test]
fn test_history_success_empty() {
@ -414,40 +228,4 @@ mod test {
};
assert_eq!(res, expected);
}
#[test]
fn test_history_message_serialization() {
// Arrange
let history_message = XmppHistoryMessage {
id: "id".to_string(),
to: Jid {
name: Some(Name("sauer@example.com".into())),
server: Server("localhost".into()),
resource: Some(Resource("tester".into())),
},
from: Jid {
name: Some(Name("pepe".into())),
server: Server("rooms.localhost".into()),
resource: Some(Resource("sauer".into())),
},
delay: Delay {
from: Jid {
name: Some(Name("pepe".into())),
server: Server("rooms.localhost".into()),
resource: Some(Resource("tester".into())),
},
stamp: "2021-10-10T10:10:10Z".to_string(),
},
body: "Hello World.".to_string(),
};
let mut events = vec![];
let expected = r#"<message id="id" to="sauer@example.com@localhost/tester" from="pepe@rooms.localhost/sauer" type="groupchat"><delay xmlns="urn:xmpp:delay" from="pepe@rooms.localhost/tester" stamp="2021-10-10T10:10:10Z"/><body>Hello World.</body></message>"#;
// Act
history_message.serialize(&mut events);
let flow = assemble_string_from_event_flow(&events);
// Assert
assert_eq!(flow, expected);
}
}

View File

@ -1,12 +0,0 @@
use quick_xml::events::Event;
use quick_xml::Writer;
use std::io::Cursor;
pub fn assemble_string_from_event_flow(events: &Vec<Event<'_>>) -> String {
let mut writer = Writer::new(Cursor::new(Vec::new()));
for event in events {
writer.write_event(event).unwrap();
}
let result = writer.into_inner().into_inner();
String::from_utf8(result).unwrap()
}

View File

@ -67,35 +67,3 @@ Or you can build it and run manually:
cargo build --release
./target/release/lavina --config config.toml
## Migrations
### Prerequisites
Install sqlx-cli into ~/.local/bin:
cargo install --locked sqlx-cli
### Steps
Migrations run on every application start. For manual run, use sqlx:
sqlx mig run \
--source ./crates/lavina-core/migrations/ \
--database-url sqlite://db.sqlite
To see current status:
sqlx mig info \
--source ./crates/lavina-core/migrations/ \
--database-url sqlite://db.sqlite
sqlx mig info outputs
0/installed first
1/installed msg author
2/installed created at for messages
3/installed dialogs
4/installed new challenges
5/pending message datetime

View File

@ -164,7 +164,7 @@ async fn endpoint_send_room_message(
let Ok(req) = serde_json::from_slice::<rooms::SendMessageReq>(&str[..]) else {
return Ok(malformed_request());
};
let Ok(room_id) = RoomId::try_from(req.room_id) else {
let Ok(room_id) = RoomId::from(req.room_id) else {
return Ok(room_not_found());
};
let Ok(player_id) = PlayerId::from(req.author_id) else {
@ -187,7 +187,7 @@ async fn endpoint_set_room_topic(
let Ok(req) = serde_json::from_slice::<rooms::SetTopicReq>(&str[..]) else {
return Ok(malformed_request());
};
let Ok(room_id) = RoomId::try_from(req.room_id) else {
let Ok(room_id) = RoomId::from(req.room_id) else {
return Ok(room_not_found());
};
let Ok(player_id) = PlayerId::from(req.author_id) else {

View File

@ -29,7 +29,7 @@ async fn endpoint_cluster_join_room(
return Ok(malformed_request());
};
tracing::info!("Incoming request: {:?}", &req);
let Ok(room_id) = RoomId::try_from(req.room_id) else {
let Ok(room_id) = RoomId::from(req.room_id) else {
dbg!(&req.room_id);
return Ok(room_not_found());
};
@ -55,7 +55,7 @@ async fn endpoint_cluster_add_message(
dbg!(&req.created_at);
return Ok(malformed_request());
};
let Ok(room_id) = RoomId::try_from(req.room_id) else {
let Ok(room_id) = RoomId::from(req.room_id) else {
dbg!(&req.room_id);
return Ok(room_not_found());
};