diff --git a/crates/lavina-core/src/auth.rs b/crates/lavina-core/src/auth.rs index 6e6f4ba..7d923de 100644 --- a/crates/lavina-core/src/auth.rs +++ b/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 } } diff --git a/crates/lavina-core/src/lib.rs b/crates/lavina-core/src/lib.rs index b868022..6e42ede 100644 --- a/crates/lavina-core/src/lib.rs +++ b/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, }) } diff --git a/crates/lavina-core/src/repo/mod.rs b/crates/lavina-core/src/repo/mod.rs index f0d210c..d173849 100644 --- a/crates/lavina-core/src/repo/mod.rs +++ b/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 (?);", diff --git a/crates/projection-irc/src/lib.rs b/crates/projection-irc/src/lib.rs index 2a310a1..c153c09 100644 --- a/crates/projection-irc/src/lib.rs +++ b/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>, - storage: &mut Storage, + core: &LavinaCore, config: &ServerConfig, ) -> Result> { 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>, - storage: &mut Storage, + authenticator: &Authenticator, config: &ServerConfig, ) -> Result> { 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>, writer: &mut BufWriter>, - storage: &mut Storage, + core: &LavinaCore, config: &ServerConfig, ) -> Result { 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<()> { - 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 +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>, writer: &mut BufWriter>, @@ -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/` 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( - config: ServerConfig, - core: LavinaCore, - metrics: MetricsRegistry, - storage: Storage, -) -> Result { +pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result { 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}"), } diff --git a/crates/projection-irc/tests/lib.rs b/crates/projection-irc/tests/lib.rs index 2b069ef..193cd4d 100644 --- a/crates/projection-irc/tests/lib.rs +++ b/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); diff --git a/crates/projection-xmpp/src/lib.rs b/crates/projection-xmpp/src/lib.rs index eec6fc3..0cbef87 100644 --- a/crates/projection-xmpp/src/lib.rs +++ b/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 { +pub async fn launch(config: ServerConfig, core: LavinaCore, metrics: MetricsRegistry) -> Result { 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, 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, - storage: &mut Storage, + core: &LavinaCore, hostname: &Str, ) -> Result { // 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?; diff --git a/crates/projection-xmpp/tests/lib.rs b/crates/projection-xmpp/tests/lib.rs index 82c551a..85e917c 100644 --- a/crates/projection-xmpp/tests/lib.rs +++ b/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); diff --git a/crates/proto-irc/src/server.rs b/crates/proto-irc/src/server.rs index 53cc05d..494f456 100644 --- a/crates/proto-irc/src/server.rs +++ b/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; diff --git a/src/http.rs b/src/http.rs index e4cbfb7..f367e51 100644 --- a/src/http.rs +++ b/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, ) -> HttpResult>> { - struct HttpReqExtractor<'a, T> { - req: &'a Request, - } - 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> { +fn endpoint_metrics(registry: &MetricsRegistry) -> Response> { 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> { } #[tracing::instrument(skip_all)] -async fn endpoint_rooms(rooms: RoomRegistry) -> Response> { +async fn endpoint_rooms(rooms: &RoomRegistry) -> Response> { // 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> { #[tracing::instrument(skip_all)] async fn endpoint_create_player( request: Request, - mut storage: Storage, + storage: &Storage, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { @@ -153,7 +131,7 @@ async fn endpoint_create_player( #[tracing::instrument(skip_all)] async fn endpoint_stop_player( request: Request, - players: PlayerRegistry, + players: &PlayerRegistry, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&str[..]) else { @@ -171,13 +149,13 @@ async fn endpoint_stop_player( #[tracing::instrument(skip_all)] async fn endpoint_set_password( request: Request, - storage: Storage, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(res) = serde_json::from_slice::(&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, - mut core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&str[..]) else { @@ -210,9 +188,10 @@ async fn endpoint_send_room_message( } } +#[tracing::instrument(skip_all)] async fn endpoint_set_room_topic( request: Request, - core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&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, - core: LavinaCore, + core: &LavinaCore, ) -> Result>> { let str = request.collect().await?.to_bytes(); let Ok(req) = serde_json::from_slice::(&str[..]) else { @@ -339,3 +318,24 @@ where Full::new(Bytes::from(buffer)) } } + +fn propagade_span_from_headers(req: &Request) { + use opentelemetry::propagation::Extractor; + use tracing::Span; + use tracing_opentelemetry::OpenTelemetrySpanExt; + struct HttpReqExtractor<'a, T> { + req: &'a Request, + } + 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); +} diff --git a/src/main.rs b/src/main.rs index e37636b..90a3159 100644 --- a/src/main.rs +++ b/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;