Merge branch 'refs/heads/main' into cluster

# Conflicts:
#	src/http.rs

crates/lavina-core/src/auth.rs

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

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/repo/mod.rs

@@ -132,7 +132,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/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;
@@ -58,9 +57,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 +72,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 +121,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 +212,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 +226,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 +243,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 +274,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 +322,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 +341,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 +350,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 +386,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 +403,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<()> {
+async fn auth_user(authenticator: &Authenticator, login: &str, plain_password: &str) -> Result<()> {
-    let verdict = Authenticator::new(storage).authenticate(login, plain_password).await?;
+    let verdict = authenticator.authenticate(login, plain_password).await?;
     // TODO properly map these onto protocol messages
     match verdict {
         Verdict::Authenticated => Ok(()),
@@ -417,7 +415,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 +751,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()),
@@ -860,7 +856,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 +992,7 @@ impl RunningServer {
     }
 }
 
-pub async fn launch(
+pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result<RunningServer> {
-    config: ServerConfig,
-    core: LavinaCore,
-    metrics: MetricsRegistry,
-    storage: Storage,
-) -> 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 +1033,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/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);
@@ -401,7 +401,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 +441,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 +480,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 +518,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 +562,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 +587,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/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::auth::Verdict;
-use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, PlayerRegistry, StopReason};
+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(
+pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result<RunningServer> {
-    config: ServerConfig,
-    core: LavinaCore,
-    metrics: MetricsRegistry,
-    storage: Storage,
-) -> 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,
+    core: LavinaCore,
-    mut storage: Storage,
     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/tests/lib.rs

@@ -158,7 +158,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 +182,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);
@@ -250,7 +250,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);
@@ -305,7 +305,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);
@@ -351,7 +351,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);

crates/proto-irc/src/server.rs

@@ -1,5 +1,3 @@
-use std::sync::Arc;
-
 use nonempty::NonEmpty;
 use tokio::io::AsyncWrite;
 use tokio::io::AsyncWriteExt;

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::UpdatePasswordResult;
-use lavina_core::auth::{Authenticator, 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 svc_fn = service_fn(|r| route(&metrics, &core, &storage, r));
-                    let core = core.clone();
+                    let server = http1::Builder::new().serve_connection(stream, svc_fn);
-                    let storage = storage.clone();
-                    let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), core.clone(), storage.clone(), r)));
                     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,
+    registry: &MetricsRegistry,
-    core: LavinaCore,
+    core: &LavinaCore,
-    storage: Storage,
+    storage: &Storage,
     request: Request<hyper::body::Incoming>,
 ) -> HttpResult<Response<Full<Bytes>>> {
-    struct HttpReqExtractor<'a, T> {
+    propagade_span_from_headers(&request);
-        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);
-
     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::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::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 irc = projection_irc::launch(irc_config, core.clone(), metrics.clone()).await?;
-    let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone(), storage.clone()).await?;
+    let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone()).await?;
     tracing::info!("Started");
 
     sleep.await;