Compare commits

...

7 Commits

Author SHA1 Message Date
Mikhail b0789d5457 Send messages on join 2024-05-28 20:31:02 +02:00
Mikhail b3d27e96c2 Parse chathistory command 2024-05-28 19:26:30 +02:00
Mikhail fac20a215d Add chathistory cap 2024-05-28 19:26:30 +02:00
Nikita Vilunov 1a21c05d7d xmpp: add support leaving MUCs via unavailable presence (#71)
Reviewed-on: lavina/lavina#71
2024-05-27 14:24:23 +00:00
Nikita Vilunov 43d105ab23 core: subscribe to rooms on player actor startup 2024-05-26 18:56:28 +02:00
Nikita Vilunov f02971d407 xmpp: add tracing instrumentations 2024-05-26 13:55:28 +02:00
Mikhail 381b5650bc xmpp, core: Send message history on MUC join (#68)
Re-send the entire message history on MUC join. Contributes to #5.

Reviewed-on: lavina/lavina#68
Co-authored-by: Mikhail <mikhail@liamets.dev>
Co-committed-by: Mikhail <mikhail@liamets.dev>
2024-05-26 11:20:26 +00:00
19 changed files with 553 additions and 107 deletions

View File

@ -0,0 +1,2 @@
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};
use crate::room::{RoomHandle, RoomId, RoomInfo, StoredMessage};
use crate::table::{AnonTable, Key as AnonKey};
use crate::LavinaCore;
@ -111,6 +111,18 @@ 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, limit: u32) -> Result<Vec<StoredMessage>> {
let (promise, deferred) = oneshot();
let cmd = ClientCommand::GetRoomHistory {
room_id: room_id.clone(),
promise,
limit,
};
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<()> {
@ -212,6 +224,11 @@ pub enum ClientCommand {
recipient: PlayerId,
promise: Promise<GetInfoResult>,
},
GetRoomHistory {
room_id: RoomId,
promise: Promise<Vec<StoredMessage>>,
limit: u32,
},
}
pub enum GetInfoResult {
@ -402,6 +419,7 @@ impl Player {
} else {
let room = self.services.rooms.get_room(&self.services, &room_id).await;
if let Some(room) = room {
room.subscribe(&self.player_id, self.handle.clone()).await;
self.my_rooms.insert(room_id, RoomRef::Local(room));
} else {
tracing::error!("Room #{room_id:?} not found");
@ -509,6 +527,14 @@ impl Player {
let result = self.check_user_existence(recipient).await;
let _ = promise.send(result);
}
ClientCommand::GetRoomHistory {
room_id,
promise,
limit,
} => {
let result = self.get_room_history(room_id, limit).await;
let _ = promise.send(result);
}
}
}
@ -557,6 +583,23 @@ impl Player {
}
}
#[tracing::instrument(skip(self), name = "Player::retrieve_room_history")]
async fn get_room_history(&mut self, room_id: RoomId, limit: u32) -> 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, limit).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);
@ -593,7 +636,7 @@ impl Player {
body: Str,
) -> SendMessageResult {
let Some(room) = self.my_rooms.get(&room_id) else {
tracing::info!("no room found");
tracing::info!("Room with ID {room_id:?} not found");
return SendMessageResult::NoSuchRoom;
};
let created_at = Utc::now();
@ -632,7 +675,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!("no room found");
tracing::info!("Room with ID {room_id:?} not 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;
use crate::room::{RoomId, StoredMessage};
#[derive(FromRow)]
pub struct StoredRoom {
@ -29,6 +29,37 @@ 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, limit: 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
limit ?;
",
// todo: implement limit
)
.bind(room_id)
.bind(limit)
.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;
@ -71,7 +102,7 @@ impl Storage {
.bind(id)
.bind(content)
.bind(author_id)
.bind(created_at.to_string())
.bind(created_at)
.bind(room_id)
.execute(&mut *executor)
.await?;
@ -174,6 +205,6 @@ impl Storage {
.fetch_all(&mut *executor)
.await?;
res.into_iter().map(|(room_id,)| RoomId::from(room_id)).collect()
res.into_iter().map(|(room_id,)| RoomId::try_from(room_id)).collect()
}
}

View File

@ -5,18 +5,20 @@ 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::Services;
use crate::{LavinaCore, Services};
/// Opaque room id
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
pub struct RoomId(Str);
impl RoomId {
pub fn from(str: impl Into<Str>) -> Result<RoomId> {
pub fn try_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"));
@ -158,6 +160,10 @@ impl RoomHandle {
lock.broadcast_update(update, player_id).await;
}
pub async fn get_message_history(&self, services: &Services, limit: u32) -> Vec<StoredMessage> {
return services.storage.get_room_message_history(self.0.read().await.storage_id, limit).await.unwrap();
}
#[tracing::instrument(skip(self), name = "RoomHandle::unsubscribe")]
pub async fn unsubscribe(&self, player_id: &PlayerId) {
let mut lock = self.0.write().await;
@ -279,3 +285,11 @@ 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

@ -6,5 +6,6 @@ bitflags! {
const None = 0;
const Sasl = 1 << 0;
const ServerTime = 1 << 1;
const ChatHistory = 1 << 2;
}
}

View File

@ -25,7 +25,7 @@ use proto_irc::client::{client_message, ClientMessage};
use proto_irc::server::CapSubBody;
use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
use proto_irc::user::PrefixedNick;
use proto_irc::{Chan, Recipient, Tag};
use proto_irc::{ChannelName, Recipient, Tag};
use sasl::AuthBody;
mod cap;
use handler::Handler;
@ -140,7 +140,7 @@ impl RegistrationState {
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::Cap {
target: self.future_nickname.clone().unwrap_or_else(|| "*".into()),
subcmd: CapSubBody::Ls("sasl=PLAIN server-time".into()),
subcmd: CapSubBody::Ls("sasl=PLAIN server-time draft/chathistory".into()),
},
}
.write_async(writer)
@ -167,6 +167,13 @@ impl RegistrationState {
self.enabled_capabilities |= Capabilities::ServerTime;
}
acked.push(cap);
} else if &*cap.name == "draft/chathistory" {
if cap.to_disable {
self.enabled_capabilities &= !Capabilities::ChatHistory;
} else {
self.enabled_capabilities |= Capabilities::ChatHistory;
}
acked.push(cap);
} else {
naked.push(cap);
}
@ -484,7 +491,14 @@ async fn handle_registered_socket<'a>(
let rooms_list = connection.get_rooms().await?;
for room in &rooms_list {
produce_on_join_cmd_messages(&config, &user, &Chan::Global(room.id.as_inner().clone()), room, writer).await?;
produce_on_join_cmd_messages(
&config,
&user,
&ChannelName::Global(room.id.as_inner().clone()),
room,
writer,
)
.await?;
}
writer.flush().await?;
@ -569,7 +583,7 @@ async fn handle_update(
if player_id == &new_member_id {
if let Some(room) = core.get_room(&room_id).await {
let room_info = room.get_room_info().await;
let chan = Chan::Global(room_id.as_inner().clone());
let chan = ChannelName::Global(room_id.as_inner().clone());
produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?;
writer.flush().await?;
} else {
@ -579,7 +593,7 @@ async fn handle_update(
ServerMessage {
tags: vec![],
sender: Some(new_member_id.as_inner().clone()),
body: ServerMessageBody::Join(Chan::Global(room_id.as_inner().clone())),
body: ServerMessageBody::Join(ChannelName::Global(room_id.as_inner().clone())),
}
.write_async(writer)
.await?;
@ -593,7 +607,7 @@ async fn handle_update(
ServerMessage {
tags: vec![],
sender: Some(former_member_id.as_inner().clone()),
body: ServerMessageBody::Part(Chan::Global(room_id.as_inner().clone())),
body: ServerMessageBody::Part(ChannelName::Global(room_id.as_inner().clone())),
}
.write_async(writer)
.await?;
@ -617,7 +631,7 @@ async fn handle_update(
tags,
sender: Some(author_id.as_inner().clone()),
body: ServerMessageBody::PrivateMessage {
target: Recipient::Chan(Chan::Global(room_id.as_inner().clone())),
target: Recipient::Chan(ChannelName::Global(room_id.as_inner().clone())),
body: body.clone(),
},
}
@ -631,7 +645,7 @@ async fn handle_update(
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
chat: Chan::Global(room_id.as_inner().clone()),
chat: ChannelName::Global(room_id.as_inner().clone()),
topic: new_topic,
},
}
@ -644,7 +658,7 @@ async fn handle_update(
ServerMessage {
tags: vec![],
sender: Some(player_id.as_inner().clone()),
body: ServerMessageBody::Part(Chan::Global(room_id.as_inner().clone())),
body: ServerMessageBody::Part(ChannelName::Global(room_id.as_inner().clone())),
}
.write_async(writer)
.await?;
@ -719,8 +733,8 @@ async fn handle_incoming_message(
handle_part(config, user, user_handle, &chan, writer).await?;
}
ClientMessage::PrivateMessage { recipient, body } => match recipient {
Recipient::Chan(Chan::Global(chan)) => {
let room_id = RoomId::from(chan)?;
Recipient::Chan(ChannelName::Global(chan)) => {
let room_id = RoomId::try_from(chan)?;
user_handle.send_message(room_id, body).await?;
}
Recipient::Nick(nick) => {
@ -731,15 +745,15 @@ async fn handle_incoming_message(
},
ClientMessage::Topic { chan, topic } => {
match chan {
Chan::Global(chan) => {
let room_id = RoomId::from(chan)?;
ChannelName::Global(chan) => {
let room_id = RoomId::try_from(chan)?;
user_handle.change_topic(room_id.clone(), topic.clone()).await?;
ServerMessage {
tags: vec![],
sender: Some(config.server_name.clone()),
body: ServerMessageBody::N332Topic {
client: user.nickname.clone(),
chat: Chan::Global(room_id.as_inner().clone()),
chat: ChannelName::Global(room_id.as_inner().clone()),
topic,
},
}
@ -747,7 +761,7 @@ async fn handle_incoming_message(
.await?;
writer.flush().await?;
}
Chan::Local(_) => {}
ChannelName::Local(_) => {}
};
}
ClientMessage::Who { target } => match &target {
@ -773,8 +787,8 @@ async fn handle_incoming_message(
.await?;
writer.flush().await?;
}
Recipient::Chan(Chan::Global(chan)) => {
let room = core.get_room(&RoomId::from(chan.clone())?).await;
Recipient::Chan(ChannelName::Global(chan)) => {
let room = core.get_room(&RoomId::try_from(chan.clone())?).await;
if let Some(room) = room {
let room_info = room.get_room_info().await;
for member in room_info.members {
@ -800,7 +814,7 @@ async fn handle_incoming_message(
.await?;
writer.flush().await?;
}
Recipient::Chan(Chan::Local(_)) => {
Recipient::Chan(ChannelName::Local(_)) => {
log::warn!("Local chans not supported");
}
},
@ -853,6 +867,38 @@ async fn handle_incoming_message(
log::info!("Received QUIT");
return Ok(HandleResult::Leave);
}
ClientMessage::ChatHistory { chan, limit } => {
let channel_name = match chan.clone() {
ChannelName::Global(chan) => chan,
ChannelName::Local(chan) => chan,
};
let room = core.get_room(&RoomId::try_from(channel_name.clone())?).await;
if let Some(room) = room {
let room_info = room.get_room_info().await;
let messages = user_handle.get_room_message_history(&room_info.id, limit).await?;
for message in messages {
let mut tags = vec![];
if user.enabled_capabilities.contains(Capabilities::ServerTime) {
let tag = Tag {
key: "time".into(),
value: Some(message.created_at.to_rfc3339_opts(SecondsFormat::Millis, true).into()),
};
tags.push(tag);
}
ServerMessage {
tags,
sender: Some(message.author_name.into()),
body: ServerMessageBody::PrivateMessage {
target: Recipient::Chan(chan.clone()),
body: message.content.into(),
},
}
.write_async(writer)
.await?;
}
writer.flush().await?;
}
}
cmd => {
log::warn!("Not implemented handler for client command: {cmd:?}");
}
@ -888,12 +934,12 @@ async fn handle_join(
config: &ServerConfig,
user: &RegisteredUser,
user_handle: &mut PlayerConnection,
chan: &Chan,
chan: &ChannelName,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
match chan {
Chan::Global(chan_name) => {
let room_id = RoomId::from(chan_name.clone())?;
ChannelName::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?;
@ -917,7 +963,7 @@ async fn handle_join(
}
writer.flush().await?;
}
Chan::Local(_) => {
ChannelName::Local(_) => {
// TODO handle join attempts to local chans with an error, we don't support these
}
};
@ -928,16 +974,16 @@ async fn handle_part(
config: &ServerConfig,
user: &RegisteredUser,
user_handle: &mut PlayerConnection,
chan: &Chan,
chan: &ChannelName,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {
if let Chan::Global(chan_name) = chan {
let room_id = RoomId::from(chan_name.clone())?;
if let ChannelName::Global(chan_name) = chan {
let room_id = RoomId::try_from(chan_name.clone())?;
user_handle.leave_room(room_id).await?;
ServerMessage {
tags: vec![],
sender: Some(user.nickname.clone()),
body: ServerMessageBody::Part(Chan::Global(chan_name.clone())),
body: ServerMessageBody::Part(ChannelName::Global(chan_name.clone())),
}
.write_async(writer)
.await?;
@ -951,7 +997,7 @@ async fn handle_part(
async fn produce_on_join_cmd_messages(
config: &ServerConfig,
user: &RegisteredUser,
chan: &Chan,
chan: &ChannelName,
room_info: &RoomInfo,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<()> {

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::from("test").unwrap()).await?;
let res = conn.join_room(RoomId::try_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::from("test").unwrap(), "Hello".into()).await?
let SendMessageResult::Success(res) = conn.send_message(RoomId::try_from("test").unwrap(), "Hello".into()).await?
else {
panic!("Failed to send message");
};

View File

@ -16,6 +16,7 @@ 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) => {
@ -112,6 +113,7 @@ 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()),
@ -120,6 +122,7 @@ 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;
@ -163,7 +166,7 @@ impl<'a> XmppConnection<'a> {
server,
resource: None,
}) if server.0 == self.hostname_rooms => {
let room_id = RoomId::from(room_name.0.clone()).unwrap();
let room_id = RoomId::try_from(room_name.0.clone()).unwrap();
let Some(_) = self.core.get_room(&room_id).await else {
// TODO should return item-not-found
// example:
@ -199,6 +202,7 @@ 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,6 +12,7 @@ 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,7 +22,7 @@ impl<'a> XmppConnection<'a> {
{
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::from(name.0.clone())?, body.clone()).await?;
self.user_handle.send_message(RoomId::try_from(name.0.clone())?, body.clone()).await?;
Message::<()> {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),

View File

@ -4,14 +4,15 @@ use anyhow::Result;
use quick_xml::events::Event;
use lavina_core::room::RoomId;
use proto_xmpp::bind::{Jid, Name, Server};
use proto_xmpp::bind::{Jid, Name, Resource, Server};
use proto_xmpp::client::{Message, MessageType, Presence, Subject};
use proto_xmpp::muc::{Affiliation, Role, XUser, XUserItem};
use proto_xmpp::muc::{Affiliation, Delay, Role, XUser, XUserItem, XmppHistoryMessage};
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 => {
@ -22,30 +23,17 @@ impl<'a> XmppConnection<'a> {
server,
// resources in MUCs are members' personas not implemented (yet?)
resource: Some(_),
}) if server.0 == self.hostname_rooms => {
let mut response = self.muc_presence(&name).await?;
response.id = p.id;
let subject = Message::<()> {
from: Some(Jid {
name: Some(name),
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![],
};
response.serialize(output);
subject.serialize(output);
}
}) 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")
}
},
_ => {
// TODO other presence cases
let response = Presence::<()>::default();
@ -55,6 +43,76 @@ 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 {
id,
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") => {
@ -82,9 +140,9 @@ impl<'a> XmppConnection<'a> {
}
}
// todo: return Presence and serialize on the outside.
async fn muc_presence(&mut self, name: &Name) -> Result<(Presence<XUser>)> {
let a = self.user_handle.join_room(RoomId::from(name.0.clone())?).await?;
#[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 handle bans
let response = Presence {
to: Some(Jid {
@ -114,21 +172,62 @@ impl<'a> XmppConnection<'a> {
};
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.
/// 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, 50).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);
}
}
#[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<()> {
let server = TestServer::start().await.unwrap();
@ -146,7 +245,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.muc_presence(&user.xmpp_name).await.unwrap();
let muc_presence = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence {
to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()),
@ -173,7 +272,7 @@ mod tests {
}],
..Default::default()
};
assert_eq!(expected, response);
assert_eq!(expected, muc_presence);
server.shutdown().await.unwrap();
Ok(())
@ -198,7 +297,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.muc_presence(&user.xmpp_name).await.unwrap();
let response = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence {
to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()),
@ -233,7 +332,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.muc_presence(&user.xmpp_name).await.unwrap();
let response = conn.retrieve_muc_presence(&user.xmpp_name).await.unwrap();
assert_eq!(expected, response);
server.shutdown().await.unwrap();

View File

@ -33,7 +33,7 @@ pub enum ClientMessage {
realname: Str,
},
/// `JOIN <chan>`
Join(Chan),
Join(ChannelName),
/// `MODE <target>`
Mode {
target: Recipient,
@ -48,11 +48,11 @@ pub enum ClientMessage {
},
/// `TOPIC <chan> :<topic>`
Topic {
chan: Chan,
chan: ChannelName,
topic: Str,
},
Part {
chan: Chan,
chan: ChannelName,
message: Option<Str>,
},
/// `PRIVMSG <target> :<msg>`
@ -65,6 +65,10 @@ pub enum ClientMessage {
reason: Str,
},
Authenticate(Str),
ChatHistory {
chan: ChannelName,
limit: u32,
},
}
pub mod command_args {
@ -95,6 +99,7 @@ pub fn client_message(input: &str) -> Result<ClientMessage> {
client_message_privmsg,
client_message_quit,
client_message_authenticate,
client_message_chathistory,
)))(input);
match res {
Ok((_, e)) => Ok(e),
@ -134,6 +139,7 @@ fn client_message_nick(input: &str) -> IResult<&str, ClientMessage> {
},
))
}
fn client_message_pass(input: &str) -> IResult<&str, ClientMessage> {
let (input, _) = tag("PASS ")(input)?;
let (input, r) = opt(tag(":"))(input)?;
@ -172,6 +178,7 @@ fn client_message_user(input: &str) -> IResult<&str, ClientMessage> {
},
))
}
fn client_message_join(input: &str) -> IResult<&str, ClientMessage> {
let (input, _) = tag("JOIN ")(input)?;
let (input, chan) = chan(input)?;
@ -280,6 +287,22 @@ fn client_message_authenticate(input: &str) -> IResult<&str, ClientMessage> {
Ok((input, ClientMessage::Authenticate(body.into())))
}
fn client_message_chathistory(input: &str) -> IResult<&str, ClientMessage> {
let (input, _) = tag("CHATHISTORY LATEST ")(input)?;
let (input, chan) = chan(input)?;
let (input, _) = tag(" * ")(input)?;
let (input, limit) = limit(input)?;
Ok((input, ClientMessage::ChatHistory { chan, limit }))
}
fn limit(input: &str) -> IResult<&str, u32> {
let (input, limit) = receiver(input)?;
let limit = limit.parse().unwrap();
Ok((input, limit))
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum CapabilitySubcommand {
/// CAP LS {code}
@ -383,6 +406,7 @@ mod test {
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_pong() {
let input = "PONG 1337";
@ -391,6 +415,7 @@ mod test {
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_nick() {
let input = "NICK SomeNick";
@ -401,6 +426,7 @@ mod test {
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_whois() {
let test_user = "WHOIS val";
@ -461,6 +487,7 @@ mod test {
assert_matches!(res_more_than_two_params, Ok(result) => assert_eq!(expected_more_than_two_params, result));
assert_matches!(res_none_none_params, Ok(result) => assert_eq!(expected_none_none_params, result))
}
#[test]
fn test_client_message_user() {
let input = "USER SomeNick 8 * :Real Name";
@ -472,28 +499,31 @@ mod test {
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_part() {
let input = "PART #chan :Pokasiki !!!";
let expected = ClientMessage::Part {
chan: Chan::Global("chan".into()),
chan: ChannelName::Global("chan".into()),
message: Some("Pokasiki !!!".into()),
};
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_message_part_empty() {
let input = "PART #chan";
let expected = ClientMessage::Part {
chan: Chan::Global("chan".into()),
chan: ChannelName::Global("chan".into()),
message: None,
};
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_cap_req() {
let input = "CAP REQ :multi-prefix -sasl";
@ -513,4 +543,16 @@ mod test {
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
#[test]
fn test_client_chat_history_latest() {
let input = "CHATHISTORY LATEST #chan * 10";
let expected = ClientMessage::ChatHistory {
chan: ChannelName::Global("chan".into()),
limit: 10,
};
let result = client_message(input);
assert_matches!(result, Ok(result) => assert_eq!(expected, result));
}
}

View File

@ -48,20 +48,20 @@ fn params(input: &str) -> IResult<&str, &str> {
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Chan {
pub enum ChannelName {
/// `#<name>` — network-global channel, available from any server in the network.
Global(Str),
/// `&<name>` — server-local channel, available only to connections to the same server. Rarely used in practice.
Local(Str),
}
impl Chan {
impl ChannelName {
pub async fn write_async(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
match self {
Chan::Global(name) => {
ChannelName::Global(name) => {
writer.write_all(b"#").await?;
writer.write_all(name.as_bytes()).await?;
}
Chan::Local(name) => {
ChannelName::Local(name) => {
writer.write_all(b"&").await?;
writer.write_all(name.as_bytes()).await?;
}
@ -70,17 +70,17 @@ impl Chan {
}
}
fn chan(input: &str) -> IResult<&str, Chan> {
fn chan_global(input: &str) -> IResult<&str, Chan> {
fn chan(input: &str) -> IResult<&str, ChannelName> {
fn chan_global(input: &str) -> IResult<&str, ChannelName> {
let (input, _) = tag("#")(input)?;
let (input, name) = receiver(input)?;
Ok((input, Chan::Global(name.into())))
Ok((input, ChannelName::Global(name.into())))
}
fn chan_local(input: &str) -> IResult<&str, Chan> {
fn chan_local(input: &str) -> IResult<&str, ChannelName> {
let (input, _) = tag("&")(input)?;
let (input, name) = receiver(input)?;
Ok((input, Chan::Local(name.into())))
Ok((input, ChannelName::Local(name.into())))
}
alt((chan_global, chan_local))(input)
@ -89,7 +89,7 @@ fn chan(input: &str) -> IResult<&str, Chan> {
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Recipient {
Nick(Str),
Chan(Chan),
Chan(ChannelName),
}
impl Recipient {
pub async fn write_async(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
@ -125,7 +125,7 @@ mod test {
#[test]
fn test_chan_global() {
let input = "#testchan";
let expected = Chan::Global("testchan".into());
let expected = ChannelName::Global("testchan".into());
let result = chan(input);
assert_matches!(result, Ok((_, result)) => assert_eq!(expected, result));
@ -139,7 +139,7 @@ mod test {
#[test]
fn test_chan_local() {
let input = "&localchan";
let expected = Chan::Local("localchan".into());
let expected = ChannelName::Local("localchan".into());
let result = chan(input);
assert_matches!(result, Ok((_, result)) => assert_eq!(expected, result));

View File

@ -69,8 +69,8 @@ pub enum ServerMessageBody {
target: Recipient,
body: Str,
},
Join(Chan),
Part(Chan),
Join(ChannelName),
Part(ChannelName),
Error {
reason: Str,
},
@ -121,7 +121,7 @@ pub enum ServerMessageBody {
},
N332Topic {
client: Str,
chat: Chan,
chat: ChannelName,
topic: Str,
},
/// A reply to a client's [Who](crate::protos::irc::client::ClientMessage::Who) request.
@ -141,12 +141,12 @@ pub enum ServerMessageBody {
},
N353NamesReply {
client: Str,
chan: Chan,
chan: ChannelName,
members: NonEmpty<PrefixedNick>,
},
N366NamesReplyEnd {
client: Str,
chan: Chan,
chan: ChannelName,
},
N431ErrNoNicknameGiven {
client: Str,
@ -154,7 +154,7 @@ pub enum ServerMessageBody {
},
N474BannedFromChan {
client: Str,
chan: Chan,
chan: ChannelName,
message: Str,
},
N502UsersDontMatch {

View File

@ -15,6 +15,9 @@ 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,14 +1,16 @@
#![allow(unused_variables)]
use quick_xml::events::{BytesEnd, BytesStart, Event};
use quick_xml::name::ResolveResult;
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 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 {
@ -154,6 +156,7 @@ pub struct XUser {
/// 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");
@ -180,6 +183,7 @@ pub struct XUserItem {
pub jid: Jid,
pub role: Role,
}
impl ToXml for XUserItem {
fn serialize(&self, output: &mut Vec<Event<'static>>) {
let mut meg = BytesStart::new("item");
@ -198,6 +202,7 @@ pub enum Affiliation {
Outcast,
None,
}
impl Affiliation {
pub fn from_str(s: &str) -> Option<Self> {
match s {
@ -228,6 +233,7 @@ pub enum Role {
Visitor,
None,
}
impl Role {
pub fn from_str(s: &str) -> Option<Self> {
match s {
@ -249,9 +255,83 @@ impl Role {
}
}
#[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() {
@ -334,4 +414,40 @@ 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

@ -0,0 +1,12 @@
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,3 +67,35 @@ 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::from(req.room_id) else {
let Ok(room_id) = RoomId::try_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::from(req.room_id) else {
let Ok(room_id) = RoomId::try_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::from(req.room_id) else {
let Ok(room_id) = RoomId::try_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::from(req.room_id) else {
let Ok(room_id) = RoomId::try_from(req.room_id) else {
dbg!(&req.room_id);
return Ok(room_not_found());
};