Merge branch 'refs/heads/main' into cluster

# Conflicts:
#	crates/projection-xmpp/tests/lib.rs
#	src/http.rs

crates/lavina-core/src/auth.rs

@@ -17,11 +17,12 @@ pub enum UpdatePasswordResult {
     UserNotFound,
 }
 
-pub struct Authenticator<'a> {
-    storage: &'a Storage,
+#[derive(Clone)]
+pub struct Authenticator {
+    storage: Storage,
 }
-impl<'a> Authenticator<'a> {
-    pub fn new(storage: &'a Storage) -> Self {
+impl Authenticator {
+    pub fn new(storage: Storage) -> Self {
         Self { storage }
     }
 

crates/lavina-core/src/clustering.rs

@@ -39,6 +39,13 @@ pub struct SendMessageReq<'a> {
     pub created_at: &'a str,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct SetRoomTopicReq<'a> {
+    pub room_id: &'a str,
+    pub player_id: &'a str,
+    pub topic: &'a str,
+}
+
 impl LavinaClient {
     pub fn new(addresses: Addresses) -> Self {
         let client = ClientBuilder::new(Client::new()).with(TracingMiddleware::<DefaultSpanBackend>::new()).build();
@@ -66,4 +73,22 @@ impl LavinaClient {
             }
         }
     }
+
+    pub async fn set_room_topic(&self, node_id: u32, req: SetRoomTopicReq<'_>) -> Result<()> {
+        tracing::info!("Setting the topic of a room on a remote node");
+        let Some(address) = self.addresses.get(node_id as usize) else {
+            tracing::error!("Failed");
+            return Err(anyhow!("Unknown node"));
+        };
+        match self.client.post(format!("http://{}/cluster/rooms/set_topic", address)).json(&req).send().await {
+            Ok(_) => {
+                tracing::info!("Room topic set");
+                Ok(())
+            }
+            Err(e) => {
+                tracing::error!("Failed to set room topic: {e:?}");
+                Err(e.into())
+            }
+        }
+    }
 }

crates/lavina-core/src/lib.rs

@@ -4,6 +4,7 @@ use anyhow::Result;
 use prometheus::Registry as MetricsRegistry;
 use std::sync::Arc;
 
+use crate::auth::Authenticator;
 use crate::dialog::DialogRegistry;
 use crate::player::PlayerRegistry;
 use crate::repo::Storage;
@@ -25,6 +26,7 @@ pub struct LavinaCore {
     pub players: PlayerRegistry,
     pub rooms: RoomRegistry,
     pub dialogs: DialogRegistry,
+    pub authenticator: Authenticator,
 }
 
 impl LavinaCore {
@@ -46,10 +48,12 @@ impl LavinaCore {
             client,
         )?;
         dialogs.set_players(players.clone()).await;
+        let authenticator = Authenticator::new(storage.clone());
         Ok(LavinaCore {
             players,
             rooms,
             dialogs,
+            authenticator,
         })
     }
 

crates/lavina-core/src/player.rs

@@ -8,6 +8,7 @@
 //! A player actor is a serial handler of commands from a single player. It is preferable to run all per-player validations in the player actor,
 //! so that they don't overload the room actor.
 use std::collections::{HashMap, HashSet};
+use std::ops::Deref;
 use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
@@ -17,7 +18,7 @@ use tokio::sync::mpsc::{channel, Receiver, Sender};
 use tokio::sync::RwLock;
 use tracing::{Instrument, Span};
 
-use crate::clustering::{ClusterMetadata, LavinaClient, SendMessageReq};
+use crate::clustering::{ClusterMetadata, LavinaClient, SendMessageReq, SetRoomTopicReq};
 use crate::dialog::DialogRegistry;
 use crate::prelude::*;
 use crate::repo::Storage;
@@ -60,7 +61,7 @@ pub struct PlayerConnection {
     player_handle: PlayerHandle,
 }
 impl PlayerConnection {
-    /// Handled in [Player::send_message].
+    /// Handled in [Player::send_room_message].
     #[tracing::instrument(skip(self, body), name = "PlayerConnection::send_message")]
     pub async fn send_message(&mut self, room_id: RoomId, body: Str) -> Result<SendMessageResult> {
         let (promise, deferred) = oneshot();
@@ -78,7 +79,7 @@ impl PlayerConnection {
         Ok(deferred.await?)
     }
 
-    /// Handled in [Player::change_topic].
+    /// Handled in [Player::change_room_topic].
     #[tracing::instrument(skip(self, new_topic), name = "PlayerConnection::change_topic")]
     pub async fn change_topic(&mut self, room_id: RoomId, new_topic: Str) -> Result<()> {
         let (promise, deferred) = oneshot();
@@ -125,6 +126,15 @@ impl PlayerConnection {
         self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
         Ok(deferred.await?)
     }
+
+    /// Handler in [Player::check_user_existence].
+    #[tracing::instrument(skip(self), name = "PlayerConnection::check_user_existence")]
+    pub async fn check_user_existence(&self, recipient: PlayerId) -> Result<GetInfoResult> {
+        let (promise, deferred) = oneshot();
+        let cmd = ClientCommand::GetInfo { recipient, promise };
+        self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
+        Ok(deferred.await?)
+    }
 }
 
 /// Handle to a player actor.
@@ -201,6 +211,15 @@ pub enum ClientCommand {
         body: Str,
         promise: Promise<()>,
     },
+    GetInfo {
+        recipient: PlayerId,
+        promise: Promise<GetInfoResult>,
+    },
+}
+
+pub enum GetInfoResult {
+    UserExists,
+    UserDoesntExist,
 }
 
 pub enum JoinResult {
@@ -413,23 +432,31 @@ impl Player {
         (handle_clone, fiber)
     }
 
-    async fn main_loop(mut self) -> Self {
-        let rooms = self.storage.get_rooms_of_a_user(self.storage_id).await.unwrap();
-        for room_id in rooms {
+    fn room_location(&self, room_id: &RoomId) -> Option<u32> {
         let node = match &**room_id.as_inner() {
             "aaaaa" => self.cluster_metadata.test_owner,
             "test" => self.cluster_metadata.test2_owner,
             _ => self.cluster_metadata.main_owner,
         };
         if node == self.cluster_metadata.node_id {
+            None
+        } else {
+            Some(node)
+        }
+    }
+
+    async fn main_loop(mut self) -> Self {
+        let rooms = self.storage.get_rooms_of_a_user(self.storage_id).await.unwrap();
+        for room_id in rooms {
+            if let Some(remote_node) = self.room_location(&room_id) {
+                self.my_rooms.insert(room_id, RoomRef::Remote { node_id: remote_node });
+            } else {
                 let room = self.rooms.get_room(&room_id).await;
                 if let Some(room) = room {
                     self.my_rooms.insert(room_id, RoomRef::Local(room));
                 } else {
                     tracing::error!("Room #{room_id:?} not found");
                 }
-            } else {
-                self.my_rooms.insert(room_id, RoomRef::Remote { node_id: node });
             }
         }
         while let Some(cmd) = self.rx.recv().await {
@@ -506,7 +533,7 @@ impl Player {
                 let _ = promise.send(());
             }
             ClientCommand::SendMessage { room_id, body, promise } => {
-                let result = self.send_message(connection_id, room_id, body).await;
+                let result = self.send_room_message(connection_id, room_id, body).await;
                 let _ = promise.send(result);
             }
             ClientCommand::ChangeTopic {
@@ -514,7 +541,7 @@ impl Player {
                 new_topic,
                 promise,
             } => {
-                self.change_topic(connection_id, room_id, new_topic).await;
+                self.change_room_topic(connection_id, room_id, new_topic).await;
                 let _ = promise.send(());
             }
             ClientCommand::GetRooms { promise } => {
@@ -529,6 +556,10 @@ impl Player {
                 self.send_dialog_message(connection_id, recipient, body).await;
                 let _ = promise.send(());
             }
+            ClientCommand::GetInfo { recipient, promise } => {
+                let result = self.check_user_existence(recipient).await;
+                let _ = promise.send(result);
+            }
         }
     }
 
@@ -541,6 +572,9 @@ impl Player {
             return JoinResult::AlreadyJoined;
         }
 
+        if let Some(remote_node) = self.room_location(&room_id) {
+            todo!()
+        } else {
             let room = match self.rooms.get_or_create_room(room_id.clone()).await {
                 Ok(room) => room,
                 Err(e) => {
@@ -550,8 +584,7 @@ impl Player {
             };
             room.add_member(&self.player_id, self.storage_id).await;
             room.subscribe(&self.player_id, self.handle.clone()).await;
-        // self.my_rooms.insert(room_id.clone(), room.clone());
-        panic!();
+            self.my_rooms.insert(room_id.clone(), RoomRef::Local(room.clone()));
             let room_info = room.get_room_info().await;
             let update = Updates::RoomJoined {
                 room_id,
@@ -560,6 +593,7 @@ impl Player {
             self.broadcast_update(update, connection_id).await;
             JoinResult::Success(room_info)
         }
+    }
 
     #[tracing::instrument(skip(self), name = "Player::leave_room")]
     async fn leave_room(&mut self, connection_id: ConnectionId, room_id: RoomId) {
@@ -576,8 +610,13 @@ impl Player {
         self.broadcast_update(update, connection_id).await;
     }
 
-    #[tracing::instrument(skip(self, body), name = "Player::send_message")]
-    async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) -> SendMessageResult {
+    #[tracing::instrument(skip(self, body), name = "Player::send_room_message")]
+    async fn send_room_message(
+        &mut self,
+        connection_id: ConnectionId,
+        room_id: RoomId,
+        body: Str,
+    ) -> SendMessageResult {
         let Some(room) = self.my_rooms.get(&room_id) else {
             tracing::info!("no room found");
             return SendMessageResult::NoSuchRoom;
@@ -607,14 +646,25 @@ impl Player {
         SendMessageResult::Success(created_at)
     }
 
-    #[tracing::instrument(skip(self, new_topic), name = "Player::change_topic")]
-    async fn change_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) {
+    #[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");
             return;
         };
-        // room.set_topic(&self.player_id, new_topic.clone()).await;
-        todo!();
+        match room {
+            RoomRef::Local(room) => {
+                room.set_topic(&self.player_id, new_topic.clone()).await;
+            }
+            RoomRef::Remote { node_id } => {
+                let req = SetRoomTopicReq {
+                    room_id: room_id.as_inner(),
+                    player_id: self.player_id.as_inner(),
+                    topic: &*new_topic,
+                };
+                self.cluster_client.set_room_topic(*node_id, req).await.unwrap();
+            }
+        }
         let update = Updates::RoomTopicChanged { room_id, new_topic };
         self.broadcast_update(update, connection_id).await;
     }
@@ -651,6 +701,15 @@ impl Player {
         self.broadcast_update(update, connection_id).await;
     }
 
+    #[tracing::instrument(skip(self), name = "Player::check_user_existence")]
+    async fn check_user_existence(&self, recipient: PlayerId) -> GetInfoResult {
+        if self.storage.check_user_existence(recipient.as_inner().as_ref()).await.unwrap() {
+            GetInfoResult::UserExists
+        } else {
+            GetInfoResult::UserDoesntExist
+        }
+    }
+
     /// Broadcasts an update to all connections except the one with the given id.
     ///
     /// This is called after handling a client command.

crates/lavina-core/src/repo/mod.rs

@@ -7,7 +7,7 @@ use anyhow::anyhow;
 use chrono::{DateTime, Utc};
 use serde::Deserialize;
 use sqlx::sqlite::SqliteConnectOptions;
-use sqlx::{ConnectOptions, Connection, FromRow, Sqlite, SqliteConnection, Transaction};
+use sqlx::{ConnectOptions, Connection, Execute, FromRow, Sqlite, SqliteConnection, Transaction};
 use tokio::sync::Mutex;
 
 use crate::prelude::*;
@@ -56,6 +56,17 @@ impl Storage {
         Ok(res)
     }
 
+    #[tracing::instrument(skip(self), name = "Storage::check_user_existence")]
+    pub async fn check_user_existence(&self, username: &str) -> Result<bool> {
+        let mut executor = self.conn.lock().await;
+        let result: Option<(String,)> = sqlx::query_as("select name from users where name = ?;")
+            .bind(username)
+            .fetch_optional(&mut *executor)
+            .await?;
+
+        Ok(result.is_some())
+    }
+
     #[tracing::instrument(skip(self), name = "Storage::retrieve_room_by_name")]
     pub async fn retrieve_room_by_name(&self, name: &str) -> Result<Option<StoredRoom>> {
         let mut executor = self.conn.lock().await;
@@ -132,7 +143,7 @@ impl Storage {
     }
 
     #[tracing::instrument(skip(self), name = "Storage::create_user")]
-    pub async fn create_user(&mut self, name: &str) -> Result<()> {
+    pub async fn create_user(&self, name: &str) -> Result<()> {
         let query = sqlx::query(
             "insert into users(name)
             values (?);",

crates/projection-irc/src/handler.rs

@@ -0,0 +1,18 @@
+use lavina_core::{player::PlayerConnection, prelude::Str, LavinaCore};
+use std::future::Future;
+use tokio::io::AsyncWrite;
+
+pub struct IrcConnection<'a, T: AsyncWrite + Unpin> {
+    pub server_name: Str,
+    /// client is nick of requester
+    pub client: Str,
+    pub writer: &'a mut T,
+    pub player_connection: &'a mut PlayerConnection,
+}
+
+pub trait Handler<T>
+where
+    T: AsyncWrite + Unpin,
+{
+    fn handle(&self, arg: IrcConnection<T>) -> impl Future<Output = anyhow::Result<()>>;
+}

crates/projection-irc/src/lib.rs

@@ -17,7 +17,6 @@ use tokio::sync::mpsc::channel;
 use lavina_core::auth::{Authenticator, Verdict};
 use lavina_core::player::*;
 use lavina_core::prelude::*;
-use lavina_core::repo::Storage;
 use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
 use lavina_core::terminator::Terminator;
 use lavina_core::LavinaCore;
@@ -28,11 +27,14 @@ use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
 use proto_irc::user::PrefixedNick;
 use proto_irc::{Chan, Recipient, Tag};
 use sasl::AuthBody;
-
 mod cap;
+use handler::Handler;
+mod whois;
 
 use crate::cap::Capabilities;
 
+mod handler;
+
 pub const APP_VERSION: &str = concat!("lavina", "_", env!("CARGO_PKG_VERSION"));
 
 #[derive(Deserialize, Debug, Clone)]
@@ -58,9 +60,8 @@ async fn handle_socket(
     config: ServerConfig,
     mut stream: TcpStream,
     socket_addr: &SocketAddr,
-    mut core: LavinaCore,
+    core: LavinaCore,
     termination: Deferred<()>, // TODO use it to stop the connection gracefully
-    mut storage: Storage,
 ) -> Result<()> {
     log::info!("Received an IRC connection from {socket_addr}");
     let (reader, writer) = stream.split();
@@ -74,7 +75,7 @@ async fn handle_socket(
             log::info!("Socket handling was terminated");
             return Ok(())
         },
-        registered_user = handle_registration(&mut reader, &mut writer, &mut storage, &config) =>
+        registered_user = handle_registration(&mut reader, &mut writer, &core, &config) =>
             match registered_user {
                 Ok(user) => {
                     log::debug!("User registered");
@@ -123,7 +124,7 @@ impl RegistrationState {
         &mut self,
         msg: ClientMessage,
         writer: &mut BufWriter<WriteHalf<'_>>,
-        storage: &mut Storage,
+        core: &LavinaCore,
         config: &ServerConfig,
     ) -> Result<Option<RegisteredUser>> {
         match msg {
@@ -214,7 +215,7 @@ impl RegistrationState {
                         realname,
                         enabled_capabilities: self.enabled_capabilities,
                     };
-                    self.finalize_auth(candidate_user, writer, storage, config).await
+                    self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
                 }
             },
             ClientMessage::Nick { nickname } => {
@@ -228,7 +229,7 @@ impl RegistrationState {
                         realname: realname.clone(),
                         enabled_capabilities: self.enabled_capabilities,
                     };
-                    self.finalize_auth(candidate_user, writer, storage, config).await
+                    self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
                 } else {
                     self.future_nickname = Some(nickname);
                     Ok(None)
@@ -245,7 +246,7 @@ impl RegistrationState {
                         realname,
                         enabled_capabilities: self.enabled_capabilities,
                     };
-                    self.finalize_auth(candidate_user, writer, storage, config).await
+                    self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
                 } else {
                     self.future_username = Some((username, realname));
                     Ok(None)
@@ -276,7 +277,7 @@ impl RegistrationState {
                     }
                 } else {
                     let body = AuthBody::from_str(body.as_bytes())?;
-                    if let Err(e) = auth_user(storage, &body.login, &body.password).await {
+                    if let Err(e) = auth_user(&core.authenticator, &body.login, &body.password).await {
                         tracing::warn!("Authentication failed: {:?}", e);
                         let target = self.future_nickname.clone().unwrap_or_else(|| "*".into());
                         sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into())
@@ -324,7 +325,7 @@ impl RegistrationState {
         &mut self,
         candidate_user: RegisteredUser,
         writer: &mut BufWriter<WriteHalf<'_>>,
-        storage: &mut Storage,
+        authenticator: &Authenticator,
         config: &ServerConfig,
     ) -> Result<Option<RegisteredUser>> {
         if self.enabled_capabilities.contains(Capabilities::Sasl)
@@ -343,7 +344,7 @@ impl RegistrationState {
                 writer.flush().await?;
                 return Ok(None);
             };
-            auth_user(storage, &*candidate_user.nickname, &*candidate_password).await?;
+            auth_user(authenticator, &*candidate_user.nickname, &*candidate_password).await?;
             Ok(Some(candidate_user))
         }
     }
@@ -352,7 +353,7 @@ impl RegistrationState {
 async fn handle_registration<'a>(
     reader: &mut BufReader<ReadHalf<'a>>,
     writer: &mut BufWriter<WriteHalf<'a>>,
-    storage: &mut Storage,
+    core: &LavinaCore,
     config: &ServerConfig,
 ) -> Result<RegisteredUser> {
     let mut buffer = vec![];
@@ -388,7 +389,7 @@ async fn handle_registration<'a>(
             }
         };
         tracing::debug!("Incoming IRC message: {msg:?}");
-        if let Some(user) = state.handle_msg(msg, writer, storage, config).await? {
+        if let Some(user) = state.handle_msg(msg, writer, core, config).await? {
             break Ok(user);
         }
         buffer.clear();
@@ -405,8 +406,8 @@ fn sasl_fail_message(sender: Str, nick: Str, text: Str) -> ServerMessage {
     }
 }
 
-async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) -> Result<()> {
-    let verdict = Authenticator::new(storage).authenticate(login, plain_password).await?;
+async fn auth_user(authenticator: &Authenticator, login: &str, plain_password: &str) -> Result<()> {
+    let verdict = authenticator.authenticate(login, plain_password).await?;
     // TODO properly map these onto protocol messages
     match verdict {
         Verdict::Authenticated => Ok(()),
@@ -417,7 +418,7 @@ async fn auth_user(storage: &mut Storage, login: &str, plain_password: &str) ->
 
 async fn handle_registered_socket<'a>(
     config: ServerConfig,
-    mut players: PlayerRegistry,
+    players: PlayerRegistry,
     rooms: RoomRegistry,
     reader: &mut BufReader<ReadHalf<'a>>,
     writer: &mut BufWriter<WriteHalf<'a>>,
@@ -753,8 +754,6 @@ async fn handle_incoming_message(
             ClientMessage::Who { target } => match &target {
                 Recipient::Nick(nick) => {
                     // TODO handle non-existing user
-                    let mut username = format!("~{nick}");
-                    let mut host = format!("user/{nick}");
                     ServerMessage {
                         tags: vec![],
                         sender: Some(config.server_name.clone()),
@@ -806,6 +805,17 @@ async fn handle_incoming_message(
                     log::warn!("Local chans not supported");
                 }
             },
+            ClientMessage::Whois { arg } => {
+                arg.handle(handler::IrcConnection {
+                    server_name: config.server_name.clone(),
+                    client: user.nickname.clone(),
+                    writer,
+                    player_connection: user_handle,
+                })
+                .await?;
+
+                writer.flush().await?;
+            }
             ClientMessage::Mode { target } => {
                 match target {
                     Recipient::Nick(nickname) => {
@@ -860,7 +870,7 @@ fn user_to_who_msg(config: &ServerConfig, requestor: &RegisteredUser, target_use
     let username = format!("~{target_user_nickname}").into();
 
     // User's host is not public, replace it with `user/<nickname>` pattern
-    let mut host = format!("user/{target_user_nickname}").into();
+    let host = format!("user/{target_user_nickname}").into();
 
     ServerMessageBody::N352WhoReply {
         client: requestor.nickname.clone(),
@@ -996,12 +1006,7 @@ impl RunningServer {
     }
 }
 
-pub async fn launch(
-    config: ServerConfig,
-    core: LavinaCore,
-    metrics: MetricsRegistry,
-    storage: Storage,
-) -> Result<RunningServer> {
+pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result<RunningServer> {
     log::info!("Starting IRC projection");
     let (stopped_tx, mut stopped_rx) = channel(32);
     let current_connections = IntGauge::new("irc_current_connections", "Open and alive TCP connections")?;
@@ -1042,9 +1047,8 @@ pub async fn launch(
                                 let core = core.clone();
                                 let current_connections_clone = current_connections.clone();
                                 let stopped_tx = stopped_tx.clone();
-                                let storage = storage.clone();
                                 async move {
-                                    match handle_socket(config, stream, &socket_addr, core, termination, storage).await {
+                                    match handle_socket(config, stream, &socket_addr, core, termination).await {
                                         Ok(_) => log::info!("Connection terminated"),
                                         Err(err) => log::warn!("Connection failed: {err}"),
                                     }

crates/projection-irc/src/whois.rs

@@ -0,0 +1,67 @@
+use lavina_core::{
+    player::{GetInfoResult, PlayerId},
+    prelude::Str,
+};
+use proto_irc::{
+    client::command_args::Whois,
+    commands::whois::{
+        error::{ErrNoNicknameGiven431, ErrNoSuchNick401},
+        response::RplEndOfWhois318,
+    },
+    response::{IrcResponseMessage, WriteResponse},
+};
+use tokio::io::AsyncWrite;
+
+use crate::handler::{Handler, IrcConnection};
+
+impl<T: AsyncWrite + Unpin> Handler<T> for Whois {
+    async fn handle(&self, body: IrcConnection<'_, T>) -> anyhow::Result<()> {
+        match self {
+            Whois::Nick(nick) => handle_nick_target(nick.clone(), body).await?,
+            Whois::TargetNick(_, nick) => handle_nick_target(nick.clone(), body).await?,
+            Whois::EmptyArgs => {
+                let IrcConnection {
+                    server_name,
+                    mut writer,
+                    ..
+                } = body;
+                IrcResponseMessage::empty_tags(
+                    Some(server_name.clone()),
+                    ErrNoNicknameGiven431::new(server_name.clone()),
+                )
+                .write_response(&mut writer)
+                .await?
+            }
+        }
+        Ok(())
+    }
+}
+
+async fn handle_nick_target(nick: Str, body: IrcConnection<'_, impl AsyncWrite + Unpin>) -> anyhow::Result<()> {
+    let IrcConnection {
+        server_name,
+        mut writer,
+        client,
+        player_connection,
+    } = body;
+
+    if let GetInfoResult::UserDoesntExist =
+        player_connection.check_user_existence(PlayerId::from(nick.clone())?).await?
+    {
+        IrcResponseMessage::empty_tags(
+            Some(server_name.clone()),
+            ErrNoSuchNick401::new(client.clone(), nick.clone()),
+        )
+        .write_response(&mut writer)
+        .await?
+    }
+
+    IrcResponseMessage::empty_tags(
+        Some(server_name.clone()),
+        RplEndOfWhois318::new(client.clone(), nick.clone()),
+    )
+    .write_response(&mut writer)
+    .await?;
+
+    Ok(())
+}

crates/projection-irc/tests/lib.rs

@@ -119,7 +119,7 @@ impl TestServer {
         })
         .await?;
         let core = LavinaCore::new(metrics.clone(), storage.clone()).await?;
-        let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap();
+        let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
         Ok(TestServer {
             metrics,
             storage,
@@ -143,7 +143,7 @@ impl TestServer {
         core.shutdown().await?;
         let metrics = MetricsRegistry::new();
         let core = LavinaCore::new(metrics.clone(), storage.clone()).await?;
-        let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap();
+        let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
         Ok(TestServer {
             metrics,
             storage,
@@ -167,7 +167,7 @@ async fn scenario_basic() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -196,7 +196,7 @@ async fn scenario_join_and_reboot() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -266,7 +266,7 @@ async fn scenario_force_join_msg() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream1 = TcpStream::connect(server.server.addr).await?;
     let mut s1 = TestScope::new(&mut stream1);
@@ -332,9 +332,9 @@ async fn scenario_two_users() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester1").await?;
-    Authenticator::new(&server.storage).set_password("tester1", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester1", "password").await?;
     server.storage.create_user("tester2").await?;
-    Authenticator::new(&server.storage).set_password("tester2", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester2", "password").await?;
 
     let mut stream1 = TcpStream::connect(server.server.addr).await?;
     let mut s1 = TestScope::new(&mut stream1);
@@ -383,7 +383,14 @@ async fn scenario_two_users() -> Result<()> {
     // The second user should receive the PART message
     s2.expect(":tester1 PART #test").await?;
 
+    s1.send("WHOIS tester2").await?;
+    s1.expect(":testserver 318 tester1 tester2 :End of /WHOIS list").await?;
+
     stream1.shutdown().await?;
+    s2.send("WHOIS tester3").await?;
+    s2.expect(":testserver 401 tester2 tester3 :No such nick/channel").await?;
+    s2.expect(":testserver 318 tester2 tester3 :End of /WHOIS list").await?;
+
     stream2.shutdown().await?;
 
     server.shutdown().await?;
@@ -401,7 +408,7 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -441,7 +448,7 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -480,7 +487,7 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -518,7 +525,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -562,7 +569,7 @@ async fn terminate_socket_scenario() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -587,7 +594,7 @@ async fn server_time_capability() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);

crates/projection-xmpp/src/iq.rs

@@ -4,16 +4,16 @@ use quick_xml::events::Event;
 
 use lavina_core::room::{RoomId, RoomRegistry};
 use proto_xmpp::bind::{BindResponse, Jid, Name, Server};
-use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType};
+use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType, Message, MessageType};
 use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery};
+use proto_xmpp::mam::{Fin, Set};
 use proto_xmpp::roster::RosterQuery;
 use proto_xmpp::session::Session;
+use proto_xmpp::xml::ToXml;
 
 use crate::proto::IqClientBody;
 use crate::XmppConnection;
 
-use proto_xmpp::xml::ToXml;
-
 impl<'a> XmppConnection<'a> {
     pub async fn handle_iq(&self, output: &mut Vec<Event<'static>>, iq: Iq<IqClientBody>) {
         match iq.body {
@@ -87,6 +87,18 @@ impl<'a> XmppConnection<'a> {
                 };
                 req.serialize(output);
             }
+            IqClientBody::MessageArchiveRequest(_) => {
+                let response = Iq {
+                    from: iq.to,
+                    id: iq.id,
+                    to: None,
+                    r#type: IqType::Result,
+                    body: Fin {
+                        set: Set { count: Some(0) },
+                    },
+                };
+                response.serialize(output);
+            }
             _ => {
                 let req = Iq {
                     from: None,

crates/projection-xmpp/src/lib.rs

@@ -22,10 +22,9 @@ use tokio::sync::mpsc::channel;
 use tokio_rustls::rustls::{Certificate, PrivateKey};
 use tokio_rustls::TlsAcceptor;
 
-use lavina_core::auth::{Authenticator, Verdict};
-use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, PlayerRegistry, StopReason};
+use lavina_core::auth::Verdict;
+use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, StopReason};
 use lavina_core::prelude::*;
-use lavina_core::repo::Storage;
 use lavina_core::room::RoomRegistry;
 use lavina_core::terminator::Terminator;
 use lavina_core::LavinaCore;
@@ -81,12 +80,7 @@ impl RunningServer {
     }
 }
 
-pub async fn launch(
-    config: ServerConfig,
-    core: LavinaCore,
-    metrics: MetricsRegistry,
-    storage: Storage,
-) -> Result<RunningServer> {
+pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result<RunningServer> {
     log::info!("Starting XMPP projection");
 
     let certs = certs(&mut SyncBufReader::new(File::open(config.cert)?))?;
@@ -126,13 +120,12 @@ pub async fn launch(
                                 continue;
                             }
                             let core = core.clone();
-                            let storage = storage.clone();
                             let hostname = config.hostname.clone();
                             let terminator = Terminator::spawn(|termination| {
                                 let stopped_tx = stopped_tx.clone();
                                 let loaded_config = loaded_config.clone();
                                 async move {
-                                    match handle_socket(loaded_config, stream, &socket_addr, core, storage, hostname, termination).await {
+                                    match handle_socket(loaded_config, stream, &socket_addr, core, hostname, termination).await {
                                         Ok(_) => log::info!("Connection terminated"),
                                         Err(err) => log::warn!("Connection failed: {err}"),
                                     }
@@ -170,8 +163,7 @@ async fn handle_socket(
     cert_config: Arc<LoadedConfig>,
     mut stream: TcpStream,
     socket_addr: &SocketAddr,
-    mut core: LavinaCore,
-    mut storage: Storage,
+    core: LavinaCore,
     hostname: Str,
     termination: Deferred<()>, // TODO use it to stop the connection gracefully
 ) -> Result<()> {
@@ -205,7 +197,7 @@ async fn handle_socket(
             log::info!("Socket handling was terminated");
             return Ok(())
         },
-        authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &mut storage, &hostname) => {
+        authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &core, &hostname) => {
             match authenticated {
                 Ok(authenticated) => {
                     let mut connection = core.players.connect_to_player(&authenticated.player_id).await;
@@ -272,7 +264,7 @@ async fn socket_auth(
     xml_reader: &mut NsReader<(impl AsyncBufRead + Unpin)>,
     xml_writer: &mut Writer<(impl AsyncWrite + Unpin)>,
     reader_buf: &mut Vec<u8>,
-    storage: &mut Storage,
+    core: &LavinaCore,
     hostname: &Str,
 ) -> Result<Authenticated> {
     // TODO validate the server hostname received in the stream start
@@ -301,7 +293,7 @@ async fn socket_auth(
     match AuthBody::from_str(&auth.body) {
         Ok(logopass) => {
             let name = &logopass.login;
-            let verdict = Authenticator::new(storage).authenticate(name, &logopass.password).await?;
+            let verdict = core.authenticator.authenticate(name, &logopass.password).await?;
             match verdict {
                 Verdict::Authenticated => {
                     proto_xmpp::sasl::Success.write_xml(xml_writer).await?;

crates/projection-xmpp/src/proto.rs

@@ -7,6 +7,7 @@ use lavina_core::prelude::*;
 use proto_xmpp::bind::BindRequest;
 use proto_xmpp::client::{Iq, Message, Presence};
 use proto_xmpp::disco::{InfoQuery, ItemQuery};
+use proto_xmpp::mam::MessageArchiveRequest;
 use proto_xmpp::roster::RosterQuery;
 use proto_xmpp::session::Session;
 use proto_xmpp::xml::*;
@@ -18,6 +19,7 @@ pub enum IqClientBody {
     Roster(RosterQuery),
     DiscoInfo(InfoQuery),
     DiscoItem(ItemQuery),
+    MessageArchiveRequest(MessageArchiveRequest),
     Unknown(Ignore),
 }
 
@@ -38,6 +40,7 @@ impl FromXml for IqClientBody {
                 RosterQuery,
                 InfoQuery,
                 ItemQuery,
+                MessageArchiveRequest,
                 {
                     delegate_parsing!(Ignore, namespace, event).into()
                 }

crates/projection-xmpp/tests/lib.rs

@@ -1,4 +1,5 @@
 use std::io::ErrorKind;
+use std::str::from_utf8;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -6,6 +7,7 @@ use anyhow::Result;
 use assert_matches::*;
 use prometheus::Registry as MetricsRegistry;
 use quick_xml::events::Event;
+use quick_xml::name::LocalName;
 use quick_xml::NsReader;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::io::{ReadHalf as GenericReadHalf, WriteHalf as GenericWriteHalf};
@@ -22,6 +24,10 @@ use lavina_core::LavinaCore;
 use projection_xmpp::{launch, RunningServer, ServerConfig};
 use proto_xmpp::xml::{Continuation, FromXml, Parser};
 
+fn element_name<'a>(local_name: &LocalName<'a>) -> &'a str {
+    from_utf8(local_name.into_inner()).unwrap()
+}
+
 pub async fn read_irc_message(reader: &mut BufReader<ReadHalf<'_>>, buf: &mut Vec<u8>) -> Result<usize> {
     let mut size = 0;
     let res = reader.read_until(b'\n', buf).await?;
@@ -35,10 +41,6 @@ struct TestScope<'a> {
     buffer: Vec<u8>,
 }
 
-fn element_name<'a>(event: &quick_xml::events::BytesStart<'a>) -> &'a str {
-    std::str::from_utf8(event.local_name().into_inner()).unwrap()
-}
-
 impl<'a> TestScope<'a> {
     fn new(stream: &mut TcpStream) -> TestScope<'_> {
         let (reader, writer) = stream.split();
@@ -60,11 +62,11 @@ impl<'a> TestScope<'a> {
     }
 
     async fn expect_starttls_required(&mut self) -> Result<()> {
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b), "features"));
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"starttls"));
-        assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"required"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"starttls"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "starttls"));
+        assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "required"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "starttls"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features"));
         Ok(())
     }
 }
@@ -97,20 +99,20 @@ impl<'a> TestScopeTls<'a> {
     }
 
     async fn expect_auth_mechanisms(&mut self) -> Result<()> {
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features"));
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"mechanisms"));
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"mechanism"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "mechanisms"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "mechanism"));
         assert_matches!(self.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"PLAIN"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"mechanism"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"mechanisms"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "mechanism"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "mechanisms"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features"));
         Ok(())
     }
 
     async fn expect_bind_feature(&mut self) -> Result<()> {
-        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"features"));
-        assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"bind"));
-        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"features"));
+        assert_matches!(self.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "features"));
+        assert_matches!(self.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "bind"));
+        assert_matches!(self.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "features"));
         Ok(())
     }
 
@@ -123,6 +125,7 @@ impl<'a> TestScopeTls<'a> {
 }
 
 struct IgnoreCertVerification;
+
 impl ServerCertVerifier for IgnoreCertVerification {
     fn verify_server_cert(
         &self,
@@ -143,6 +146,7 @@ struct TestServer {
     core: LavinaCore,
     server: RunningServer,
 }
+
 impl TestServer {
     async fn start() -> Result<TestServer> {
         let _ = tracing_subscriber::fmt::try_init();
@@ -158,7 +162,7 @@ impl TestServer {
         })
         .await?;
         let core = LavinaCore::new(metrics.clone(), storage.clone()).await?;
-        let server = launch(config, core.clone(), metrics.clone(), storage.clone()).await.unwrap();
+        let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
         Ok(TestServer {
             metrics,
             storage,
@@ -182,7 +186,7 @@ async fn scenario_basic() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -191,10 +195,10 @@ async fn scenario_basic() -> Result<()> {
     s.send(r#"<?xml version="1.0"?>"#).await?;
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_starttls_required().await?;
     s.send(r#"<starttls/>"#).await?;
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed"));
     let buffer = s.buffer;
     tracing::info!("TLS feature negotiation complete");
 
@@ -213,26 +217,26 @@ async fn scenario_basic() -> Result<()> {
     s.send(r#"<?xml version="1.0"?>"#).await?;
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_auth_mechanisms().await?;
 
-    // base64-encoded b"\x00tester\x00password"
+    // base64-encoded "\x00tester\x00password"
     s.send(r#"<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">AHRlc3RlcgBwYXNzd29yZA==</auth>"#)
         .await?;
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"success"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "success"));
     s.send(r#"<?xml version="1.0"?>"#).await?;
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_bind_feature().await?;
     s.send(r#"<iq id="bind_1" type="set"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><resource>kek</resource></bind></iq>"#).await?;
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"iq"));
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"bind"));
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"jid"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "iq"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "bind"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "jid"));
     assert_matches!(s.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"tester@localhost/tester"));
-    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"jid"));
-    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"bind"));
-    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"iq"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "jid"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "bind"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "iq"));
     s.send(r#"<presence xmlns="jabber:client" type="unavailable"><status>Logged out</status></presence>"#).await?;
 
     stream.shutdown().await?;
@@ -250,7 +254,7 @@ async fn scenario_wrong_password() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -259,10 +263,10 @@ async fn scenario_wrong_password() -> Result<()> {
     s.send(r#"<?xml version="1.0"?>"#).await?;
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_starttls_required().await?;
     s.send(r#"<starttls/>"#).await?;
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed"));
     let buffer = s.buffer;
     tracing::info!("TLS feature negotiation complete");
 
@@ -281,14 +285,14 @@ async fn scenario_wrong_password() -> Result<()> {
     s.send(r#"<?xml version="1.0"?>"#).await?;
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_auth_mechanisms().await?;
-    // base64-encoded b"\x00tester\x00password2"
+    // base64-encoded "\x00tester\x00password2"
     s.send(r#"<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">AHRlc3RlcgBwYXNzd29yZDI=</auth>"#)
         .await?;
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"failure"));
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"not-authorized"));
-    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(b.local_name().into_inner(), b"failure"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "failure"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "not-authorized"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "failure"));
 
     let _ = stream.shutdown().await;
 
@@ -305,7 +309,7 @@ async fn scenario_basic_without_headers() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -313,10 +317,10 @@ async fn scenario_basic_without_headers() -> Result<()> {
 
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_starttls_required().await?;
     s.send(r#"<starttls/>"#).await?;
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed"));
     let buffer = s.buffer;
     tracing::info!("TLS feature negotiation complete");
 
@@ -334,7 +338,7 @@ async fn scenario_basic_without_headers() -> Result<()> {
 
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
 
     stream.shutdown().await?;
 
@@ -351,7 +355,7 @@ async fn terminate_socket() -> Result<()> {
     // test scenario
 
     server.storage.create_user("tester").await?;
-    Authenticator::new(&server.storage).set_password("tester", "password").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
 
     let mut stream = TcpStream::connect(server.server.addr).await?;
     let mut s = TestScope::new(&mut stream);
@@ -361,10 +365,10 @@ async fn terminate_socket() -> Result<()> {
 
     s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
     assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
-    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(b.local_name().into_inner(), b"stream"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
     s.expect_starttls_required().await?;
     s.send(r#"<starttls/>"#).await?;
-    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(b.local_name().into_inner(), b"proceed"));
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed"));
 
     let connector = TlsConnector::from(Arc::new(
         ClientConfig::builder()
@@ -383,3 +387,89 @@ async fn terminate_socket() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_message_archive_request() -> Result<()> {
+    let mut server = TestServer::start().await?;
+
+    // test scenario
+
+    server.storage.create_user("tester").await?;
+    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
+
+    let mut stream = TcpStream::connect(server.server.addr).await?;
+    let mut s = TestScope::new(&mut stream);
+    tracing::info!("TCP connection established");
+
+    s.send(r#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
+    s.expect_starttls_required().await?;
+    s.send(r#"<starttls/>"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "proceed"));
+    let buffer = s.buffer;
+    tracing::info!("TLS feature negotiation complete");
+
+    let connector = TlsConnector::from(Arc::new(
+        ClientConfig::builder()
+            .with_safe_defaults()
+            .with_custom_certificate_verifier(Arc::new(IgnoreCertVerification))
+            .with_no_client_auth(),
+    ));
+    tracing::info!("Initiating TLS connection...");
+    let mut stream = connector.connect(ServerName::IpAddress(server.server.addr.ip()), stream).await?;
+    tracing::info!("TLS connection established");
+
+    let mut s = TestScopeTls::new(&mut stream, buffer);
+
+    s.send(r#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
+    s.expect_auth_mechanisms().await?;
+
+    // base64-encoded "\x00tester\x00password"
+    s.send(r#"<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">AHRlc3RlcgBwYXNzd29yZA==</auth>"#)
+        .await?;
+    assert_matches!(s.next_xml_event().await?, Event::Empty(b) => assert_eq!(element_name(&b.local_name()), "success"));
+    s.send(r#"<?xml version="1.0"?>"#).await?;
+    s.send(r#"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="127.0.0.1" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace" xmlns="jabber:client" version="1.0">"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Decl(_) => {});
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "stream"));
+    s.expect_bind_feature().await?;
+    s.send(r#"<iq id="bind_1" type="set"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><resource>kek</resource></bind></iq>"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "iq"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "bind"));
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => assert_eq!(element_name(&b.local_name()), "jid"));
+    assert_matches!(s.next_xml_event().await?, Event::Text(b) => assert_eq!(&*b, b"tester@localhost/tester"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "jid"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "bind"));
+    assert_matches!(s.next_xml_event().await?, Event::End(b) => assert_eq!(element_name(&b.local_name()), "iq"));
+
+    s.send(r#"<iq type='get' id='juliet1'><query xmlns='urn:xmpp:mam:2' queryid='f27'/></iq>"#).await?;
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => {
+        assert_eq!(element_name(&b.local_name()), "iq")
+    });
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => {
+        assert_eq!(element_name(&b.local_name()), "fin")
+    });
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => {
+        assert_eq!(element_name(&b.local_name()), "set")
+    });
+    assert_matches!(s.next_xml_event().await?, Event::Start(b) => {
+        assert_eq!(element_name(&b.local_name()), "count")
+    });
+    assert_matches!(s.next_xml_event().await?, Event::Text(b) => {
+        assert_eq!(&*b, b"0")
+    });
+
+    s.send(r#"<presence xmlns="jabber:client" type="unavailable"><status>Logged out</status></presence>"#).await?;
+
+    stream.shutdown().await?;
+
+    // wrap up
+
+    server.shutdown().await?;
+    Ok(())
+}

crates/proto-irc/src/client.rs

@@ -1,9 +1,9 @@
-use super::*;
-
 use anyhow::{anyhow, Result};
 use nom::combinator::{all_consuming, opt};
 use nonempty::NonEmpty;
 
+use super::*;
+
 /// Client-to-server command.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum ClientMessage {
@@ -42,6 +42,10 @@ pub enum ClientMessage {
     Who {
         target: Recipient, // aka mask
     },
+    /// WHOIS [<target>] <nick>
+    Whois {
+        arg: command_args::Whois,
+    },
     /// `TOPIC <chan> :<topic>`
     Topic {
         chan: Chan,
@@ -63,6 +67,17 @@ pub enum ClientMessage {
     Authenticate(Str),
 }
 
+pub mod command_args {
+    use crate::prelude::Str;
+
+    #[derive(Clone, Debug, PartialEq, Eq)]
+    pub enum Whois {
+        Nick(Str),
+        TargetNick(Str, Str),
+        EmptyArgs,
+    }
+}
+
 pub fn client_message(input: &str) -> Result<ClientMessage> {
     let res = all_consuming(alt((
         client_message_capability,
@@ -74,6 +89,7 @@ pub fn client_message(input: &str) -> Result<ClientMessage> {
         client_message_join,
         client_message_mode,
         client_message_who,
+        client_message_whois,
         client_message_topic,
         client_message_part,
         client_message_privmsg,
@@ -177,6 +193,31 @@ fn client_message_who(input: &str) -> IResult<&str, ClientMessage> {
     Ok((input, ClientMessage::Who { target }))
 }
 
+fn client_message_whois(input: &str) -> IResult<&str, ClientMessage> {
+    let (input, _) = tag("WHOIS ")(input)?;
+    let args: Vec<_> = input.split_whitespace().collect();
+    match args.as_slice()[..] {
+        [nick] => Ok((
+            "",
+            ClientMessage::Whois {
+                arg: command_args::Whois::Nick(nick.into()),
+            },
+        )),
+        [target, nick, ..] => Ok((
+            "",
+            ClientMessage::Whois {
+                arg: command_args::Whois::TargetNick(target.into(), nick.into()),
+            },
+        )),
+        [] => Ok((
+            "",
+            ClientMessage::Whois {
+                arg: command_args::Whois::EmptyArgs,
+            },
+        )),
+    }
+}
+
 fn client_message_topic(input: &str) -> IResult<&str, ClientMessage> {
     let (input, _) = tag("TOPIC ")(input)?;
     let (input, chan) = chan(input)?;
@@ -311,6 +352,7 @@ mod test {
     use nonempty::nonempty;
 
     use super::*;
+
     #[test]
     fn test_client_message_cap_ls() {
         let input = "CAP LS 302";
@@ -360,6 +402,66 @@ mod test {
         assert_matches!(result, Ok(result) => assert_eq!(expected, result));
     }
     #[test]
+    fn test_client_message_whois() {
+        let test_user = "WHOIS val";
+        let test_user_user = "WHOIS val val";
+        let test_server_user = "WHOIS com.test.server user";
+        let test_user_server = "WHOIS user com.test.server";
+        let test_users_list = "WHOIS user_1,user_2,user_3";
+        let test_server_users_list = "WHOIS com.test.server user_1,user_2,user_3";
+        let test_more_than_two_params = "WHOIS test.server user_1,user_2,user_3 whatever spam";
+        let test_none_none_params = "WHOIS   ";
+
+        let res_one_arg = client_message(test_user);
+        let res_user_user = client_message(test_user_user);
+
+        let res_server_user = client_message(test_server_user);
+        let res_user_server = client_message(test_user_server);
+        let res_users_list = client_message(test_users_list);
+        let res_server_users_list = client_message(test_server_users_list);
+        let res_more_than_two_params = client_message(test_more_than_two_params);
+        let res_none_none_params = client_message(test_none_none_params);
+
+        let expected_arg = ClientMessage::Whois {
+            arg: command_args::Whois::Nick("val".into()),
+        };
+        let expected_user_user = ClientMessage::Whois {
+            arg: command_args::Whois::TargetNick("val".into(), "val".into()),
+        };
+
+        let expected_server_user = ClientMessage::Whois {
+            arg: command_args::Whois::TargetNick("com.test.server".into(), "user".into()),
+        };
+
+        let expected_user_server = ClientMessage::Whois {
+            arg: command_args::Whois::TargetNick("user".into(), "com.test.server".into()),
+        };
+
+        let expected_user_list = ClientMessage::Whois {
+            arg: command_args::Whois::Nick("user_1,user_2,user_3".into()),
+        };
+        let expected_server_user_list = ClientMessage::Whois {
+            arg: command_args::Whois::TargetNick("com.test.server".into(), "user_1,user_2,user_3".into()),
+        };
+
+        let expected_more_than_two_params = ClientMessage::Whois {
+            arg: command_args::Whois::TargetNick("test.server".into(), "user_1,user_2,user_3".into()),
+        };
+
+        let expected_none_none_params = ClientMessage::Whois {
+            arg: command_args::Whois::EmptyArgs,
+        };
+
+        assert_matches!(res_one_arg, Ok(result) => assert_eq!(expected_arg, result));
+        assert_matches!(res_user_user, Ok(result) => assert_eq!(expected_user_user, result));
+        assert_matches!(res_server_user, Ok(result) => assert_eq!(expected_server_user, result));
+        assert_matches!(res_user_server, Ok(result) => assert_eq!(expected_user_server, result));
+        assert_matches!(res_users_list, Ok(result) => assert_eq!(expected_user_list, result));
+        assert_matches!(res_server_users_list, Ok(result) => assert_eq!(expected_server_user_list, result));
+        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";
         let expected = ClientMessage::User {

crates/proto-irc/src/commands/mod.rs

@@ -0,0 +1 @@
+pub mod whois;

crates/proto-irc/src/commands/whois/error.rs

@@ -0,0 +1,67 @@
+use tokio::io::{AsyncWrite, AsyncWriteExt};
+
+use crate::{prelude::Str, response::WriteResponse};
+
+///  ErrNoSuchNick401  
+pub struct ErrNoSuchNick401 {
+    client: Str,
+    nick: Str,
+}
+
+impl ErrNoSuchNick401 {
+    pub fn new(client: Str, nick: Str) -> Self {
+        ErrNoSuchNick401 { client, nick }
+    }
+}
+
+/// ErrNoSuchServer402
+struct ErrNoSuchServer402 {
+    client: Str,
+    /// target parameter in WHOIS
+    /// example:  `/whois <target> <nick>`
+    server_name: Str,
+}
+
+///  ErrNoNicknameGiven431
+pub struct ErrNoNicknameGiven431 {
+    client: Str,
+}
+impl ErrNoNicknameGiven431 {
+    pub fn new(client: Str) -> Self {
+        ErrNoNicknameGiven431 { client }
+    }
+}
+
+impl WriteResponse for ErrNoSuchNick401 {
+    async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
+        writer.write_all(b"401 ").await?;
+        writer.write_all(self.client.as_bytes()).await?;
+        writer.write_all(b" ").await?;
+        writer.write_all(self.nick.as_bytes()).await?;
+        writer.write_all(b" :").await?;
+        writer.write_all("No such nick/channel".as_bytes()).await?;
+        Ok(())
+    }
+}
+
+impl WriteResponse for ErrNoNicknameGiven431 {
+    async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
+        writer.write_all(b"431").await?;
+        writer.write_all(self.client.as_bytes()).await?;
+        writer.write_all(b" :").await?;
+        writer.write_all("No nickname given".as_bytes()).await?;
+        Ok(())
+    }
+}
+
+impl WriteResponse for ErrNoSuchServer402 {
+    async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
+        writer.write_all(b"402 ").await?;
+        writer.write_all(self.client.as_bytes()).await?;
+        writer.write_all(b" ").await?;
+        writer.write_all(self.server_name.as_bytes()).await?;
+        writer.write_all(b" :").await?;
+        writer.write_all("No such server".as_bytes()).await?;
+        Ok(())
+    }
+}

crates/proto-irc/src/commands/whois/mod.rs

@@ -0,0 +1,2 @@
+pub mod error;
+pub mod response;

crates/proto-irc/src/commands/whois/response.rs

@@ -0,0 +1,24 @@
+use tokio::io::{AsyncWrite, AsyncWriteExt};
+
+use crate::{prelude::Str, response::WriteResponse};
+
+pub struct RplEndOfWhois318 {
+    client: Str,
+    nick: Str,
+}
+impl RplEndOfWhois318 {
+    pub fn new(client: Str, nick: Str) -> Self {
+        RplEndOfWhois318 { client, nick }
+    }
+}
+impl WriteResponse for RplEndOfWhois318 {
+    async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
+        writer.write_all(b"318 ").await?;
+        writer.write_all(self.client.as_bytes()).await?;
+        writer.write_all(b" ").await?;
+        writer.write_all(self.nick.as_bytes()).await?;
+        writer.write_all(b" :").await?;
+        writer.write_all("End of /WHOIS list".as_bytes()).await?;
+        Ok(())
+    }
+}

crates/proto-irc/src/lib.rs

@@ -1,6 +1,8 @@
 //! Client-to-Server IRC protocol.
 pub mod client;
+pub mod commands;
 mod prelude;
+pub mod response;
 pub mod server;
 #[cfg(test)]
 mod testkit;

crates/proto-irc/src/response.rs

@@ -0,0 +1,47 @@
+use std::future::Future;
+
+use tokio::io::{AsyncWrite, AsyncWriteExt};
+
+use crate::prelude::Str;
+use crate::Tag;
+
+pub trait WriteResponse {
+    fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> impl Future<Output = std::io::Result<()>>;
+}
+
+/// Server-to-client enum agnostic message
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct IrcResponseMessage<T> {
+    /// Optional tags section, prefixed with `@`
+    pub tags: Vec<Tag>,
+    /// Optional server name, prefixed with `:`.
+    pub sender: Option<Str>,
+    pub body: T,
+}
+
+impl<T> IrcResponseMessage<T> {
+    pub fn empty_tags(sender: Option<Str>, body: T) -> Self {
+        IrcResponseMessage {
+            tags: vec![],
+            sender,
+            body,
+        }
+    }
+
+    pub fn new(tags: Vec<Tag>, sender: Option<Str>, body: T) -> Self {
+        IrcResponseMessage { tags, sender, body }
+    }
+}
+
+impl<T: WriteResponse> WriteResponse for IrcResponseMessage<T> {
+    async fn write_response(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
+        if let Some(sender) = &self.sender {
+            writer.write_all(b":").await?;
+            writer.write_all(sender.as_bytes()).await?;
+            writer.write_all(b" ").await?;
+        }
+        self.body.write_response(writer).await?;
+        writer.write_all(b"\r\n").await?;
+        Ok(())
+    }
+}

crates/proto-irc/src/server.rs

@@ -1,12 +1,11 @@
-use std::sync::Arc;
-
 use nonempty::NonEmpty;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;
 
-use super::*;
 use crate::user::PrefixedNick;
 
+use super::*;
+
 /// Server-to-client message.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ServerMessage {
@@ -114,6 +113,12 @@ pub enum ServerMessageBody {
         /// Usually `b"End of WHO list"`
         msg: Str,
     },
+    N318EndOfWhois {
+        client: Str,
+        nick: Str,
+        /// Usually `b"End of /WHOIS list"`
+        msg: Str,
+    },
     N332Topic {
         client: Str,
         chat: Chan,
@@ -143,6 +148,10 @@ pub enum ServerMessageBody {
         client: Str,
         chan: Chan,
     },
+    N431ErrNoNicknameGiven {
+        client: Str,
+        message: Str,
+    },
     N474BannedFromChan {
         client: Str,
         chan: Chan,
@@ -280,6 +289,14 @@ impl ServerMessageBody {
                 writer.write_all(b" :").await?;
                 writer.write_all(msg.as_bytes()).await?;
             }
+            ServerMessageBody::N318EndOfWhois { client, nick, msg } => {
+                writer.write_all(b"318 ").await?;
+                writer.write_all(client.as_bytes()).await?;
+                writer.write_all(b" ").await?;
+                writer.write_all(nick.as_bytes()).await?;
+                writer.write_all(b" :").await?;
+                writer.write_all(msg.as_bytes()).await?;
+            }
             ServerMessageBody::N332Topic { client, chat, topic } => {
                 writer.write_all(b"332 ").await?;
                 writer.write_all(client.as_bytes()).await?;
@@ -342,6 +359,12 @@ impl ServerMessageBody {
                 chan.write_async(writer).await?;
                 writer.write_all(b" :End of /NAMES list").await?;
             }
+            ServerMessageBody::N431ErrNoNicknameGiven { client, message } => {
+                writer.write_all(b"431").await?;
+                writer.write_all(client.as_bytes()).await?;
+                writer.write_all(b" :").await?;
+                writer.write_all(message.as_bytes()).await?;
+            }
             ServerMessageBody::N474BannedFromChan { client, chan, message } => {
                 writer.write_all(b"474 ").await?;
                 writer.write_all(client.as_bytes()).await?;
@@ -470,9 +493,10 @@ fn server_message_body_cap(input: &str) -> IResult<&str, ServerMessageBody> {
 mod test {
     use assert_matches::*;
 
-    use super::*;
     use crate::testkit::*;
 
+    use super::*;
+
     #[test]
     fn test_server_message_notice() {
         let input = "NOTICE * :*** Looking up your hostname...\r\n";

crates/proto-xmpp/src/bind.rs

@@ -74,8 +74,8 @@ impl Jid {
 pub struct BindRequest(pub Resource);
 
 impl FromXmlTag for BindRequest {
-    const NS: &'static str = XMLNS;
     const NAME: &'static str = "bind";
+    const NS: &'static str = XMLNS;
 }
 
 impl FromXml for BindRequest {

crates/proto-xmpp/src/client.rs

@@ -658,7 +658,7 @@ mod tests {
 
     #[tokio::test]
     async fn parse_message() {
-        let input = r#"<message id="aacea" type="chat" to="nikita@vlnv.dev"><subject>daa</subject><body>bbb</body><unknown-stuff></unknown-stuff></message>"#;
+        let input = r#"<message id="aacea" type="chat" to="chelik@xmpp.ru"><subject>daa</subject><body>bbb</body><unknown-stuff></unknown-stuff></message>"#;
         let result: Message<Ignore> = crate::xml::parse(input).unwrap();
         assert_eq!(
             result,
@@ -666,8 +666,8 @@ mod tests {
                 from: None,
                 id: Some("aacea".to_string()),
                 to: Some(Jid {
-                    name: Some(Name("nikita".into())),
-                    server: Server("vlnv.dev".into()),
+                    name: Some(Name("chelik".into())),
+                    server: Server("xmpp.ru".into()),
                     resource: None
                 }),
                 r#type: MessageType::Chat,
@@ -681,7 +681,7 @@ mod tests {
 
     #[tokio::test]
     async fn parse_message_empty_custom() {
-        let input = r#"<message id="aacea" type="chat" to="nikita@vlnv.dev"><subject>daa</subject><body>bbb</body><unknown-stuff/></message>"#;
+        let input = r#"<message id="aacea" type="chat" to="chelik@xmpp.ru"><subject>daa</subject><body>bbb</body><unknown-stuff/></message>"#;
         let result: Message<Ignore> = crate::xml::parse(input).unwrap();
         assert_eq!(
             result,
@@ -689,8 +689,8 @@ mod tests {
                 from: None,
                 id: Some("aacea".to_string()),
                 to: Some(Jid {
-                    name: Some(Name("nikita".into())),
-                    server: Server("vlnv.dev".into()),
+                    name: Some(Name("chelik".into())),
+                    server: Server("xmpp.ru".into()),
                     resource: None
                 }),
                 r#type: MessageType::Chat,

crates/proto-xmpp/src/lib.rs

@@ -3,6 +3,7 @@
 pub mod bind;
 pub mod client;
 pub mod disco;
+pub mod mam;
 pub mod muc;
 mod prelude;
 pub mod roster;

crates/proto-xmpp/src/mam.rs

@@ -0,0 +1,225 @@
+use anyhow::{anyhow, Result};
+use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event};
+use quick_xml::name::{Namespace, ResolveResult};
+use std::io::Read;
+
+use crate::xml::*;
+
+pub const MAM_XMLNS: &'static str = "urn:xmpp:mam:2";
+pub const DATA_XMLNS: &'static str = "jabber:x:data";
+pub const RESULT_SET_XMLNS: &'static str = "http://jabber.org/protocol/rsm";
+
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub struct MessageArchiveRequest {
+    pub x: Option<X>,
+}
+
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub struct X {
+    pub fields: Vec<Field>,
+}
+
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub struct Field {
+    pub values: Vec<String>,
+}
+
+// Message archive response styled as a result set.
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub struct Fin {
+    pub set: Set,
+}
+
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub struct Set {
+    pub count: Option<i32>,
+}
+
+impl ToXml for Fin {
+    fn serialize(&self, events: &mut Vec<Event<'static>>) {
+        let fin_bytes = BytesStart::new(format!(r#"fin xmlns="{}" complete=True"#, MAM_XMLNS));
+        let set_bytes = BytesStart::new(format!(r#"set xmlns="{}""#, RESULT_SET_XMLNS));
+        events.push(Event::Start(fin_bytes));
+        events.push(Event::Start(set_bytes));
+
+        if let &Some(count) = &self.set.count {
+            events.push(Event::Start(BytesStart::new("count")));
+            events.push(Event::Text(BytesText::new(count.to_string().as_str()).into_owned()));
+            events.push(Event::End(BytesEnd::new("count")));
+        }
+        events.push(Event::End(BytesEnd::new("set")));
+        events.push(Event::End(BytesEnd::new("fin")));
+    }
+}
+
+impl FromXmlTag for X {
+    const NAME: &'static str = "x";
+    const NS: &'static str = DATA_XMLNS;
+}
+
+impl FromXmlTag for MessageArchiveRequest {
+    const NAME: &'static str = "query";
+    const NS: &'static str = MAM_XMLNS;
+}
+
+impl FromXml for X {
+    type P = impl Parser<Output = Result<Self>>;
+
+    fn parse() -> Self::P {
+        |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
+            println!("X::parse {:?}", event);
+
+            let bytes = match event {
+                Event::Start(bytes) if bytes.name().0 == X::NAME.as_bytes() => bytes,
+                Event::Empty(bytes) if bytes.name().0 == X::NAME.as_bytes() => return Ok(X { fields: vec![] }),
+                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+            };
+            let mut fields = vec![];
+            loop {
+                (namespace, event) = yield;
+                match event {
+                    Event::Start(_) => {
+                        // start of <field>
+                        let mut values = vec![];
+                        loop {
+                            (namespace, event) = yield;
+                            match event {
+                                Event::Start(bytes) if bytes.name().0 == b"value" => {
+                                    // start of <value>
+                                }
+                                Event::End(bytes) if bytes.name().0 == b"field" => {
+                                    // end of </field>
+                                    break;
+                                }
+                                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+                            }
+                            (namespace, event) = yield;
+                            let text: String = match event {
+                                Event::Text(bytes) => {
+                                    // text inside <value></value>
+                                    String::from_utf8(bytes.to_vec())?
+                                }
+                                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+                            };
+                            (namespace, event) = yield;
+                            match event {
+                                Event::End(bytes) if bytes.name().0 == b"value" => {
+                                    // end of </value>
+                                }
+                                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+                            }
+                            values.push(text);
+                        }
+                        fields.push(Field { values })
+                    }
+                    Event::End(bytes) if bytes.name().0 == X::NAME.as_bytes() => {
+                        // end of <x/>
+                        return Ok(X { fields });
+                    }
+                    _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+                }
+            }
+        }
+    }
+}
+
+impl FromXml for MessageArchiveRequest {
+    type P = impl Parser<Output = Result<Self>>;
+
+    fn parse() -> Self::P {
+        |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
+            println!("MessageArchiveRequest::parse {:?}", event);
+
+            let bytes = match event {
+                Event::Empty(_) => return Ok(MessageArchiveRequest { x: None }),
+                Event::Start(bytes) => bytes,
+                _ => return Err(anyhow!("Unexpected XML event: {event:?}")),
+            };
+            if bytes.name().0 != MessageArchiveRequest::NAME.as_bytes() {
+                return Err(anyhow!("Unexpected XML tag: {:?}", bytes.name()));
+            }
+            let ResolveResult::Bound(Namespace(ns)) = namespace else {
+                return Err(anyhow!("No namespace provided"));
+            };
+            if ns != MAM_XMLNS.as_bytes() {
+                return Err(anyhow!("Incorrect namespace"));
+            }
+            (namespace, event) = yield;
+            match event {
+                Event::End(bytes) if bytes.name().0 == MessageArchiveRequest::NAME.as_bytes() => {
+                    Ok(MessageArchiveRequest { x: None })
+                }
+                Event::Start(bytes) | Event::Empty(bytes) if bytes.name().0 == X::NAME.as_bytes() => {
+                    let x = delegate_parsing!(X, namespace, event)?;
+                    Ok(MessageArchiveRequest { x: Some(x) })
+                }
+                _ => Err(anyhow!("Unexpected XML event: {event:?}")),
+            }
+        }
+    }
+}
+
+impl MessageArchiveRequest {}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bind::{Jid, Name, Server};
+    use crate::client::{Iq, IqType};
+
+    #[test]
+    fn test_parse_archive_query() {
+        let input = r#"<iq to='pubsub.shakespeare.lit' type='set' id='juliet1'><query xmlns='urn:xmpp:mam:2' queryid='f28'/></iq>"#;
+
+        let result: Iq<MessageArchiveRequest> = parse(input).unwrap();
+        assert_eq!(
+            result,
+            Iq {
+                from: None,
+                id: "juliet1".to_string(),
+                to: Option::from(Jid {
+                    name: None,
+                    server: Server("pubsub.shakespeare.lit".into()),
+                    resource: None,
+                }),
+                r#type: IqType::Set,
+                body: MessageArchiveRequest { x: None },
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_query_messages_from_jid() {
+        let input = r#"<iq type='set' id='juliet1'><query xmlns='urn:xmpp:mam:2'><x xmlns='jabber:x:data' type='submit'><field var='FORM_TYPE' type='hidden'><value>value1</value></field><field var='with'><value>juliet@capulet.lit</value></field></x></query></iq>"#;
+
+        let result: Iq<MessageArchiveRequest> = parse(input).unwrap();
+        assert_eq!(
+            result,
+            Iq {
+                from: None,
+                id: "juliet1".to_string(),
+                to: None,
+                r#type: IqType::Set,
+                body: MessageArchiveRequest {
+                    x: Some(X {
+                        fields: vec![
+                            Field {
+                                values: vec!["value1".to_string()],
+                            },
+                            Field {
+                                values: vec!["juliet@capulet.lit".to_string()],
+                            },
+                        ]
+                    })
+                },
+            }
+        );
+    }
+
+    #[test]
+    fn test_parse_query_messages_from_jid_with_unclosed_tag() {
+        let input = r#"<iq type='set' id='juliet1'><query xmlns='urn:xmpp:mam:2'><x xmlns='jabber:x:data' type='submit'><field var='FORM_TYPE' type='hidden'><value>value1</value></field><field var='with'><value>juliet@capulet.lit</value></field></query></iq>"#;
+
+        assert!(parse::<Iq<MessageArchiveRequest>>(input).is_err())
+    }
+}

crates/proto-xmpp/src/roster.rs

@@ -2,7 +2,7 @@ use quick_xml::events::{BytesStart, Event};
 
 use crate::xml::*;
 use anyhow::{anyhow, Result};
-use quick_xml::name::ResolveResult;
+use quick_xml::name::{Namespace, ResolveResult};
 
 pub const XMLNS: &'static str = "jabber:iq:roster";
 
@@ -14,6 +14,9 @@ impl FromXml for RosterQuery {
 
     fn parse() -> Self::P {
         |(mut namespace, mut event): (ResolveResult<'static>, &'static Event<'static>)| -> Result<Self> {
+            let ResolveResult::Bound(Namespace(ns)) = namespace else {
+                return Err(anyhow!("No namespace provided"));
+            };
             match event {
                 Event::Start(_) => (),
                 Event::Empty(_) => return Ok(RosterQuery),
@@ -38,3 +41,39 @@ impl ToXml for RosterQuery {
         events.push(Event::Empty(BytesStart::new(format!(r#"query xmlns="{}""#, XMLNS))));
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bind::{Jid, Name, Resource, Server};
+    use crate::client::{Iq, IqType};
+
+    #[test]
+    fn test_parse() {
+        let input =
+            r#"<iq from='juliet@example.com/balcony' id='bv1bs71f' type='get'><query xmlns='jabber:iq:roster'/></iq>"#;
+
+        let result: Iq<RosterQuery> = parse(input).unwrap();
+        assert_eq!(
+            result,
+            Iq {
+                from: Option::from(Jid {
+                    name: Option::from(Name("juliet".into())),
+                    server: Server("example.com".into()),
+                    resource: Option::from(Resource("balcony".into())),
+                }),
+                id: "bv1bs71f".to_string(),
+                to: None,
+                r#type: IqType::Get,
+                body: RosterQuery,
+            }
+        )
+    }
+
+    #[test]
+    fn test_missing_namespace() {
+        let input = r#"<iq from='juliet@example.com/balcony' id='bv1bs71f' type='get'><query/></iq>"#;
+
+        assert!(parse::<Iq<RosterQuery>>(input).is_err());
+    }
+}

crates/proto-xmpp/src/stream.rs

@@ -170,14 +170,14 @@ mod test {
 
     #[tokio::test]
     async fn client_stream_start_correct_parse() {
-        let input = r###"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="vlnv.dev" version="1.0" xmlns="jabber:client" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace">"###;
+        let input = r###"<stream:stream xmlns:stream="http://etherx.jabber.org/streams" to="xmpp.ru" version="1.0" xmlns="jabber:client" xml:lang="en" xmlns:xml="http://www.w3.org/XML/1998/namespace">"###;
         let mut reader = NsReader::from_reader(input.as_bytes());
         let mut buf = vec![];
         let res = ClientStreamStart::parse(&mut reader, &mut buf).await.unwrap();
         assert_eq!(
             res,
             ClientStreamStart {
-                to: "vlnv.dev".to_owned(),
+                to: "xmpp.ru".to_owned(),
                 lang: Some("en".to_owned()),
                 version: "1.0".to_owned()
             }
@@ -187,12 +187,12 @@ mod test {
     #[tokio::test]
     async fn server_stream_start_write() {
         let input = ServerStreamStart {
-            from: "vlnv.dev".to_owned(),
+            from: "xmpp.ru".to_owned(),
             lang: "en".to_owned(),
             id: "stream_id".to_owned(),
             version: "1.0".to_owned(),
         };
-        let expected = r###"<stream:stream from="vlnv.dev" version="1.0" xmlns="jabber:client" xmlns:stream="http://etherx.jabber.org/streams" xml:lang="en" id="stream_id">"###;
+        let expected = r###"<stream:stream from="xmpp.ru" version="1.0" xmlns="jabber:client" xmlns:stream="http://etherx.jabber.org/streams" xml:lang="en" id="stream_id">"###;
         let mut output: Vec<u8> = vec![];
         let mut writer = Writer::new(&mut output);
         input.write_xml(&mut writer).await.unwrap();

src/http.rs

@@ -1,4 +1,3 @@
-use chrono::Utc;
 use std::convert::Infallible;
 use std::net::SocketAddr;
 
@@ -9,15 +8,11 @@ use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::{Method, Request, Response, StatusCode};
 use hyper_util::rt::TokioIo;
-use opentelemetry::propagation::Extractor;
 use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
-use tracing::Span;
-use tracing_opentelemetry::OpenTelemetrySpanExt;
 
-use lavina_core::auth::UpdatePasswordResult::PasswordUpdated;
-use lavina_core::auth::{Authenticator, UpdatePasswordResult};
+use lavina_core::auth::UpdatePasswordResult;
 use lavina_core::clustering::SendMessageReq;
 use lavina_core::player::{PlayerId, PlayerRegistry, SendMessageResult};
 use lavina_core::prelude::*;
@@ -67,10 +62,8 @@ async fn main_loop(
                 let core = core.clone();
                 let storage = storage.clone();
                 tokio::task::spawn(async move {
-                    let registry = metrics.clone();
-                    let core = core.clone();
-                    let storage = storage.clone();
-                    let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), core.clone(), storage.clone(), r)));
+                    let svc_fn = service_fn(|r| route(&metrics, &core, &storage, r));
+                    let server = http1::Builder::new().serve_connection(stream, svc_fn);
                     if let Err(err) = server.await {
                         tracing::error!("Error serving connection: {:?}", err);
                     }
@@ -82,35 +75,20 @@ async fn main_loop(
     Ok(())
 }
 
-#[tracing::instrument(skip_all, name = "route")]
+#[tracing::instrument(skip_all)]
 async fn route(
-    registry: MetricsRegistry,
-    core: LavinaCore,
-    storage: Storage,
+    registry: &MetricsRegistry,
+    core: &LavinaCore,
+    storage: &Storage,
     request: Request<hyper::body::Incoming>,
 ) -> HttpResult<Response<Full<Bytes>>> {
-    struct HttpReqExtractor<'a, T> {
-        req: &'a Request<T>,
-    }
-    impl<'a, T> Extractor for HttpReqExtractor<'a, T> {
-        fn get(&self, key: &str) -> Option<&str> {
-            self.req.headers().get(key).and_then(|v| v.to_str().ok())
-        }
-
-        fn keys(&self) -> Vec<&str> {
-            self.req.headers().keys().map(|k| k.as_str()).collect()
-        }
-    }
-
-    let ctx = opentelemetry::global::get_text_map_propagator(|pp| pp.extract(&HttpReqExtractor { req: &request }));
-    Span::current().set_parent(ctx);
-
+    propagade_span_from_headers(&request);
     let res = match (request.method(), request.uri().path()) {
         (&Method::GET, "/metrics") => endpoint_metrics(registry),
-        (&Method::GET, "/rooms") => endpoint_rooms(core.rooms).await,
+        (&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await,
         (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
-        (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core.players).await.or5xx(),
-        (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, storage).await.or5xx(),
+        (&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, &core.players).await.or5xx(),
+        (&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, core).await.or5xx(),
         (&Method::POST, rooms::paths::SEND_MESSAGE) => endpoint_send_room_message(request, core).await.or5xx(),
         (&Method::POST, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(),
         (&Method::POST, "/cluster/rooms/add_message") => endpoint_cluster_add_message(request, core).await.or5xx(),
@@ -119,7 +97,7 @@ async fn route(
     Ok(res)
 }
 
-fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> {
+fn endpoint_metrics(registry: &MetricsRegistry) -> Response<Full<Bytes>> {
     let mf = registry.gather();
     let mut buffer = vec![];
     TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail");
@@ -127,7 +105,7 @@ fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> {
 }
 
 #[tracing::instrument(skip_all)]
-async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> {
+async fn endpoint_rooms(rooms: &RoomRegistry) -> Response<Full<Bytes>> {
     // TODO introduce management API types independent from core-domain types
     // TODO remove `Serialize` implementations from all core-domain types
     let room_list = rooms.get_all_rooms().await.to_body();
@@ -137,7 +115,7 @@ async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> {
 #[tracing::instrument(skip_all)]
 async fn endpoint_create_player(
     request: Request<hyper::body::Incoming>,
-    mut storage: Storage,
+    storage: &Storage,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
@@ -153,7 +131,7 @@ async fn endpoint_create_player(
 #[tracing::instrument(skip_all)]
 async fn endpoint_stop_player(
     request: Request<hyper::body::Incoming>,
-    players: PlayerRegistry,
+    players: &PlayerRegistry,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else {
@@ -171,13 +149,13 @@ async fn endpoint_stop_player(
 #[tracing::instrument(skip_all)]
 async fn endpoint_set_password(
     request: Request<hyper::body::Incoming>,
-    storage: Storage,
+    core: &LavinaCore,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
         return Ok(malformed_request());
     };
-    let verdict = Authenticator::new(&storage).set_password(&res.player_name, &res.password).await?;
+    let verdict = core.authenticator.set_password(&res.player_name, &res.password).await?;
     match verdict {
         UpdatePasswordResult::PasswordUpdated => {}
         UpdatePasswordResult::UserNotFound => {
@@ -187,10 +165,10 @@ async fn endpoint_set_password(
     Ok(empty_204_request())
 }
 
-#[tracing::instrument(skip_all, name = "LavinaClient::endpoint_send_room_message")]
+#[tracing::instrument(skip_all)]
 async fn endpoint_send_room_message(
     request: Request<hyper::body::Incoming>,
-    mut core: LavinaCore,
+    core: &LavinaCore,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(req) = serde_json::from_slice::<rooms::SendMessageReq>(&str[..]) else {
@@ -210,9 +188,10 @@ async fn endpoint_send_room_message(
     }
 }
 
+#[tracing::instrument(skip_all)]
 async fn endpoint_set_room_topic(
     request: Request<hyper::body::Incoming>,
-    core: LavinaCore,
+    core: &LavinaCore,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(req) = serde_json::from_slice::<rooms::SetTopicReq>(&str[..]) else {
@@ -232,7 +211,7 @@ async fn endpoint_set_room_topic(
 #[tracing::instrument(skip_all, name = "endpoint_cluster_add_message")]
 async fn endpoint_cluster_add_message(
     request: Request<hyper::body::Incoming>,
-    core: LavinaCore,
+    core: &LavinaCore,
 ) -> Result<Response<Full<Bytes>>> {
     let str = request.collect().await?.to_bytes();
     let Ok(req) = serde_json::from_slice::<SendMessageReq>(&str[..]) else {
@@ -339,3 +318,24 @@ where
         Full::new(Bytes::from(buffer))
     }
 }
+
+fn propagade_span_from_headers<T>(req: &Request<T>) {
+    use opentelemetry::propagation::Extractor;
+    use tracing::Span;
+    use tracing_opentelemetry::OpenTelemetrySpanExt;
+    struct HttpReqExtractor<'a, T> {
+        req: &'a Request<T>,
+    }
+    impl<'a, T> Extractor for HttpReqExtractor<'a, T> {
+        fn get(&self, key: &str) -> Option<&str> {
+            self.req.headers().get(key).and_then(|v| v.to_str().ok())
+        }
+
+        fn keys(&self) -> Vec<&str> {
+            self.req.headers().keys().map(|k| k.as_str()).collect()
+        }
+    }
+
+    let ctx = opentelemetry::global::get_text_map_propagator(|pp| pp.extract(&HttpReqExtractor { req }));
+    Span::current().set_parent(ctx);
+}

src/main.rs

@@ -73,8 +73,8 @@ async fn main() -> Result<()> {
     let storage = Storage::open(storage_config).await?;
     let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
     let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), core.clone(), storage.clone()).await?;
-    let irc = projection_irc::launch(irc_config, core.clone(), metrics.clone(), storage.clone()).await?;
-    let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone(), storage.clone()).await?;
+    let irc = projection_irc::launch(irc_config, core.clone(), metrics.clone()).await?;
+    let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone()).await?;
     tracing::info!("Started");
 
     sleep.await;