nikita
/
lavina
forked from lavina/lavina
1
0
Fork 0
Code Pull Requests Activity

Compare commits

base: nikita:b2a06ef9844469e82abdffcbe29c73258115f7e2
Branches Tags
nikita:main
nikita:x-user-and-items
nikita:diagram
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2
..
compare: nikita:d1a72e7978f6dbe441b35619c90aab4c9885def8
Branches Tags
nikita:x-user-and-items
nikita:main
nikita:diagram
lavina:main
lavina:0.0.0
lavina:0.0.1
lavina:0.0.2

No commits in common. "b2a06ef9844469e82abdffcbe29c73258115f7e2" and "d1a72e7978f6dbe441b35619c90aab4c9885def8" have entirely different histories.
b2a06ef984 ... d1a72e7978

22 changed files with 498 additions and 444 deletions
Show Stats Expand all files Collapse all files

25
crates/lavina-core/src/auth.rs
View File

@ -3,7 +3,8 @@ use argon2::password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, Salt
use argon2::Argon2; use argon2::Argon2;
use rand_core::OsRng; use rand_core::OsRng;
use crate::LavinaCore; use crate::prelude::log;
use crate::repo::Storage;
pub enum Verdict { pub enum Verdict {
Authenticated, Authenticated,
@ -16,10 +17,18 @@ pub enum UpdatePasswordResult {
UserNotFound, UserNotFound,
} }
impl LavinaCore { #[derive(Clone)]
#[tracing::instrument(skip(self, provided_password), name = "Services::authenticate")] pub struct Authenticator {
storage: Storage,
}
impl Authenticator {
pub fn new(storage: Storage) -> Self {
Self { storage }
}
#[tracing::instrument(skip(self, provided_password), name = "Authenticator::authenticate")]
pub async fn authenticate(&self, login: &str, provided_password: &str) -> Result<Verdict> { pub async fn authenticate(&self, login: &str, provided_password: &str) -> Result<Verdict> {
let Some(stored_user) = self.services.storage.retrieve_user_by_name(login).await? else { let Some(stored_user) = self.storage.retrieve_user_by_name(login).await? else {
return Ok(Verdict::UserNotFound); return Ok(Verdict::UserNotFound);
}; };
if let Some(argon2_hash) = stored_user.argon2_hash { if let Some(argon2_hash) = stored_user.argon2_hash {
@ -39,9 +48,9 @@ impl LavinaCore {
Ok(Verdict::InvalidPassword) Ok(Verdict::InvalidPassword)
} }
#[tracing::instrument(skip(self, provided_password), name = "Services::set_password")] #[tracing::instrument(skip(self, provided_password), name = "Authenticator::set_password")]
pub async fn set_password(&self, login: &str, provided_password: &str) -> Result<UpdatePasswordResult> { pub async fn set_password(&self, login: &str, provided_password: &str) -> Result<UpdatePasswordResult> {
let Some(u) = self.services.storage.retrieve_user_by_name(login).await? else { let Some(u) = self.storage.retrieve_user_by_name(login).await? else {
return Ok(UpdatePasswordResult::UserNotFound); return Ok(UpdatePasswordResult::UserNotFound);
}; };
@ -51,8 +60,8 @@ impl LavinaCore {
.hash_password(provided_password.as_bytes(), &salt) .hash_password(provided_password.as_bytes(), &salt)
.map_err(|e| anyhow!("Failed to hash password: {e:?}"))?; .map_err(|e| anyhow!("Failed to hash password: {e:?}"))?;
self.services.storage.set_argon2_challenge(u.id, password_hash.to_string().as_str()).await?; self.storage.set_argon2_challenge(u.id, password_hash.to_string().as_str()).await?;
tracing::info!("Password changed for player {login}"); log::info!("Password changed for player {login}");
Ok(UpdatePasswordResult::PasswordUpdated) Ok(UpdatePasswordResult::PasswordUpdated)
} }
} }

14
crates/lavina-core/src/clustering.rs
View File

@ -1,11 +1,11 @@
use std::collections::HashMap;
use std::net::SocketAddr;
use anyhow::{anyhow, Result}; use anyhow::{anyhow, Result};
use reqwest::Client; use reqwest::Client;
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware}; use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_tracing::{DefaultSpanBackend, TracingMiddleware}; use reqwest_tracing::{DefaultSpanBackend, TracingMiddleware};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
pub mod broadcast; pub mod broadcast;
pub mod room; pub mod room;
@ -27,15 +27,19 @@ pub struct ClusterMetadata {
pub rooms: HashMap<String, u32>, pub rooms: HashMap<String, u32>,
} }
#[derive(Clone)]
pub struct LavinaClient { pub struct LavinaClient {
addresses: Addresses, addresses: Arc<Addresses>,
client: ClientWithMiddleware, client: ClientWithMiddleware,
} }
impl LavinaClient { impl LavinaClient {
pub fn new(addresses: Addresses) -> Self { pub fn new(addresses: Addresses) -> Self {
let client = ClientBuilder::new(Client::new()).with(TracingMiddleware::<DefaultSpanBackend>::new()).build(); let client = ClientBuilder::new(Client::new()).with(TracingMiddleware::<DefaultSpanBackend>::new()).build();
Self { addresses, client } Self {
addresses: Arc::new(addresses),
client,
}
} }
async fn send_request(&self, node_id: u32, path: &str, req: impl Serialize) -> Result<()> { async fn send_request(&self, node_id: u32, path: &str, req: impl Serialize) -> Result<()> {

31
crates/lavina-core/src/clustering/broadcast.rs
View File

@ -1,32 +1,41 @@
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use tokio::sync::Mutex; use tokio::sync::Mutex;
use crate::player::{PlayerId, Updates}; use crate::player::{PlayerId, PlayerRegistry, Updates};
use crate::prelude::Str; use crate::prelude::Str;
use crate::room::RoomId; use crate::room::RoomId;
use crate::Services;
/// Receives updates from other nodes and broadcasts them to local player actors. /// Receives updates from other nodes and broadcasts them to local player actors.
struct BroadcastingInner { struct BroadcastingInner {
subscriptions: HashMap<RoomId, HashSet<PlayerId>>, subscriptions: HashMap<RoomId, HashSet<PlayerId>>,
} }
pub struct Broadcasting(Mutex<BroadcastingInner>); impl Broadcasting {}
#[derive(Clone)]
pub struct Broadcasting(Arc<Mutex<BroadcastingInner>>);
impl Broadcasting { impl Broadcasting {
pub fn new() -> Self { pub fn new() -> Self {
let inner = BroadcastingInner { let inner = BroadcastingInner {
subscriptions: HashMap::new(), subscriptions: HashMap::new(),
}; };
Self(Mutex::new(inner)) Self(Arc::new(Mutex::new(inner)))
} }
}
impl Services {
/// Broadcasts the given update to subscribed player actors on local node. /// Broadcasts the given update to subscribed player actors on local node.
#[tracing::instrument(skip(self, message, created_at))] #[tracing::instrument(skip(self, players, message, created_at), name = "Broadcasting::broadcast")]
pub async fn broadcast(&self, room_id: RoomId, author_id: PlayerId, message: Str, created_at: DateTime<Utc>) { pub async fn broadcast(
let inner = self.broadcasting.0.lock().await; &self,
players: &PlayerRegistry,
room_id: RoomId,
author_id: PlayerId,
message: Str,
created_at: DateTime<Utc>,
) {
let inner = self.0.lock().await;
let Some(subscribers) = inner.subscriptions.get(&room_id) else { let Some(subscribers) = inner.subscriptions.get(&room_id) else {
return; return;
}; };
@ -40,7 +49,7 @@ impl Services {
if i == &author_id { if i == &author_id {
continue; continue;
} }
let Some(player) = self.players.get_player(i).await else { let Some(player) = players.get_player(i).await else {
continue; continue;
}; };
player.update(update.clone()).await; player.update(update.clone()).await;
@ -48,6 +57,6 @@ impl Services {
} }
pub async fn subscribe(&self, subscriber: PlayerId, room_id: RoomId) { pub async fn subscribe(&self, subscriber: PlayerId, room_id: RoomId) {
self.broadcasting.0.lock().await.subscriptions.entry(room_id).or_insert_with(HashSet::new).insert(subscriber); self.0.lock().await.subscriptions.entry(room_id).or_insert_with(HashSet::new).insert(subscriber);
} }
} }

37
crates/lavina-core/src/clustering/room.rs
View File

@ -1,11 +1,6 @@
use anyhow::Result;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use crate::clustering::LavinaClient; use crate::clustering::LavinaClient;
use crate::player::PlayerId;
use crate::prelude::Str;
use crate::room::RoomId;
use crate::LavinaCore;
pub mod paths { pub mod paths {
pub const JOIN: &'static str = "/cluster/rooms/join"; pub const JOIN: &'static str = "/cluster/rooms/join";
@ -43,46 +38,22 @@ pub struct SetRoomTopicReq<'a> {
impl LavinaClient { impl LavinaClient {
#[tracing::instrument(skip(self, req), name = "LavinaClient::join_room")] #[tracing::instrument(skip(self, req), name = "LavinaClient::join_room")]
pub async fn join_room(&self, node_id: u32, req: JoinRoomReq<'_>) -> Result<()> { pub async fn join_room(&self, node_id: u32, req: JoinRoomReq<'_>) -> anyhow::Result<()> {
self.send_request(node_id, paths::JOIN, req).await self.send_request(node_id, paths::JOIN, req).await
} }
#[tracing::instrument(skip(self, req), name = "LavinaClient::leave_room")] #[tracing::instrument(skip(self, req), name = "LavinaClient::leave_room")]
pub async fn leave_room(&self, node_id: u32, req: LeaveRoomReq<'_>) -> Result<()> { pub async fn leave_room(&self, node_id: u32, req: LeaveRoomReq<'_>) -> anyhow::Result<()> {
self.send_request(node_id, paths::LEAVE, req).await self.send_request(node_id, paths::LEAVE, req).await
} }
#[tracing::instrument(skip(self, req), name = "LavinaClient::send_room_message")] #[tracing::instrument(skip(self, req), name = "LavinaClient::send_room_message")]
pub async fn send_room_message(&self, node_id: u32, req: SendMessageReq<'_>) -> Result<()> { pub async fn send_room_message(&self, node_id: u32, req: SendMessageReq<'_>) -> anyhow::Result<()> {
self.send_request(node_id, paths::ADD_MESSAGE, req).await self.send_request(node_id, paths::ADD_MESSAGE, req).await
} }
#[tracing::instrument(skip(self, req), name = "LavinaClient::set_room_topic")] #[tracing::instrument(skip(self, req), name = "LavinaClient::set_room_topic")]
pub async fn set_room_topic(&self, node_id: u32, req: SetRoomTopicReq<'_>) -> Result<()> { pub async fn set_room_topic(&self, node_id: u32, req: SetRoomTopicReq<'_>) -> anyhow::Result<()> {
self.send_request(node_id, paths::SET_TOPIC, req).await self.send_request(node_id, paths::SET_TOPIC, req).await
} }
} }
impl LavinaCore {
pub async fn cluster_join_room(&self, room_id: RoomId, player_id: &PlayerId) -> Result<()> {
let room_handle = self.services.rooms.get_or_create_room(&self.services, room_id).await?;
let storage_id =
self.services.storage.create_or_retrieve_user_id_by_name(player_id.as_inner().as_ref()).await?;
room_handle.add_member(&self.services, &player_id, storage_id).await;
Ok(())
}
pub async fn cluster_send_room_message(
&self,
room_id: RoomId,
player_id: &PlayerId,
message: Str,
created_at: chrono::DateTime<chrono::Utc>,
) -> Result<Option<()>> {
let Some(room_handle) = self.services.rooms.get_room(&self.services, &room_id).await else {
return Ok(None);
};
room_handle.send_message(&self.services, &player_id, message, created_at).await;
Ok(Some(()))
}
}

68
crates/lavina-core/src/dialog.rs
View File

@ -4,13 +4,14 @@
//! There are no admins or other roles in dialogs, both participants have equal rights. //! There are no admins or other roles in dialogs, both participants have equal rights.
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::Arc;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use tokio::sync::RwLock as AsyncRwLock; use tokio::sync::RwLock as AsyncRwLock;
use crate::player::{PlayerId, Updates}; use crate::player::{PlayerId, PlayerRegistry, Updates};
use crate::prelude::*; use crate::prelude::*;
use crate::Services; use crate::repo::Storage;
/// Id of a conversation between two players. /// Id of a conversation between two players.
/// ///
@ -44,50 +45,55 @@ struct Dialog {
struct DialogRegistryInner { struct DialogRegistryInner {
dialogs: HashMap<DialogId, AsyncRwLock<Dialog>>, dialogs: HashMap<DialogId, AsyncRwLock<Dialog>>,
players: Option<PlayerRegistry>,
storage: Storage,
} }
pub(crate) struct DialogRegistry(AsyncRwLock<DialogRegistryInner>); #[derive(Clone)]
pub struct DialogRegistry(Arc<AsyncRwLock<DialogRegistryInner>>);
impl Services { impl DialogRegistry {
#[tracing::instrument(skip(self, body, created_at))] pub async fn send_message(
pub async fn send_dialog_message(
&self, &self,
from: PlayerId, from: PlayerId,
to: PlayerId, to: PlayerId,
body: Str, body: Str,
created_at: &DateTime<Utc>, created_at: &DateTime<Utc>,
) -> Result<()> { ) -> Result<()> {
let guard = self.dialogs.0.read().await; let mut guard = self.0.read().await;
let id = DialogId::new(from.clone(), to.clone()); let id = DialogId::new(from.clone(), to.clone());
let dialog = guard.dialogs.get(&id); let dialog = guard.dialogs.get(&id);
if let Some(d) = dialog { if let Some(d) = dialog {
let mut d = d.write().await; let mut d = d.write().await;
self.storage guard
.storage
.insert_dialog_message(d.storage_id, d.message_count, from.as_inner(), &body, created_at) .insert_dialog_message(d.storage_id, d.message_count, from.as_inner(), &body, created_at)
.await?; .await?;
d.message_count += 1; d.message_count += 1;
} else { } else {
drop(guard); drop(guard);
let mut guard2 = self.dialogs.0.write().await; let mut guard2 = self.0.write().await;
// double check in case concurrent access has loaded this dialog // double check in case concurrent access has loaded this dialog
if let Some(d) = guard2.dialogs.get(&id) { if let Some(d) = guard2.dialogs.get(&id) {
let mut d = d.write().await; let mut d = d.write().await;
self.storage guard2
.storage
.insert_dialog_message(d.storage_id, d.message_count, from.as_inner(), &body, created_at) .insert_dialog_message(d.storage_id, d.message_count, from.as_inner(), &body, created_at)
.await?; .await?;
d.message_count += 1; d.message_count += 1;
} else { } else {
let (p1, p2) = id.as_inner(); let (p1, p2) = id.as_inner();
tracing::info!("Dialog {id:?} not found locally, trying to load from storage"); tracing::info!("Dialog {id:?} not found locally, trying to load from storage");
let stored_dialog = match self.storage.retrieve_dialog(p1.as_inner(), p2.as_inner()).await? { let stored_dialog = match guard2.storage.retrieve_dialog(p1.as_inner(), p2.as_inner()).await? {
Some(t) => t, Some(t) => t,
None => { None => {
tracing::info!("Dialog {id:?} does not exist, creating a new one in storage"); tracing::info!("Dialog {id:?} does not exist, creating a new one in storage");
self.storage.initialize_dialog(p1.as_inner(), p2.as_inner(), created_at).await? guard2.storage.initialize_dialog(p1.as_inner(), p2.as_inner(), created_at).await?
} }
}; };
tracing::info!("Dialog {id:?} loaded"); tracing::info!("Dialog {id:?} loaded");
self.storage guard2
.storage
.insert_dialog_message( .insert_dialog_message(
stored_dialog.id, stored_dialog.id,
stored_dialog.message_count, stored_dialog.message_count,
@ -104,10 +110,14 @@ impl Services {
}; };
guard2.dialogs.insert(id.clone(), AsyncRwLock::new(dialog)); guard2.dialogs.insert(id.clone(), AsyncRwLock::new(dialog));
} }
drop(guard2); guard = guard2.downgrade();
} }
// TODO send message to the other player and persist it // TODO send message to the other player and persist it
let Some(player) = self.players.get_player(&to).await else { let Some(players) = &guard.players else {
tracing::error!("No player registry present");
return Ok(());
};
let Some(player) = players.get_player(&to).await else {
tracing::debug!("Player {to:?} not active, not sending message"); tracing::debug!("Player {to:?} not active, not sending message");
return Ok(()); return Ok(());
}; };
@ -123,13 +133,33 @@ impl Services {
} }
impl DialogRegistry { impl DialogRegistry {
pub fn new() -> DialogRegistry { pub fn new(storage: Storage) -> DialogRegistry {
DialogRegistry(AsyncRwLock::new(DialogRegistryInner { DialogRegistry(Arc::new(AsyncRwLock::new(DialogRegistryInner {
dialogs: HashMap::new(), dialogs: HashMap::new(),
})) players: None,
storage,
})))
} }
pub fn shutdown(self) {} pub async fn set_players(&self, players: PlayerRegistry) {
let mut guard = self.0.write().await;
guard.players = Some(players);
}
pub async fn unset_players(&self) {
let mut guard = self.0.write().await;
guard.players = None;
}
pub fn shutdown(self) -> Result<()> {
let res = match Arc::try_unwrap(self.0) {
Ok(e) => e,
Err(_) => return Err(fail("failed to acquire dialogs ownership on shutdown")),
};
let res = res.into_inner();
drop(res);
Ok(())
}
} }
#[cfg(test)] #[cfg(test)]

102
crates/lavina-core/src/lib.rs
View File

@ -1,16 +1,16 @@
//! Domain definitions and implementation of common chat logic. //! Domain definitions and implementation of common chat logic.
use std::ops::Deref;
use std::sync::Arc; use std::sync::Arc;
use anyhow::Result; use anyhow::Result;
use prometheus::Registry as MetricsRegistry; use prometheus::Registry as MetricsRegistry;
use crate::auth::Authenticator;
use crate::clustering::broadcast::Broadcasting; use crate::clustering::broadcast::Broadcasting;
use crate::clustering::{ClusterConfig, ClusterMetadata, LavinaClient}; use crate::clustering::{ClusterConfig, LavinaClient};
use crate::dialog::DialogRegistry; use crate::dialog::DialogRegistry;
use crate::player::{PlayerConnection, PlayerId, PlayerRegistry}; use crate::player::PlayerRegistry;
use crate::repo::Storage; use crate::repo::Storage;
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry}; use crate::room::RoomRegistry;
pub mod auth; pub mod auth;
pub mod clustering; pub mod clustering;
@ -25,88 +25,50 @@ mod table;
#[derive(Clone)] #[derive(Clone)]
pub struct LavinaCore { pub struct LavinaCore {
services: Arc<Services>, pub players: PlayerRegistry,
} pub rooms: RoomRegistry,
pub dialogs: DialogRegistry,
impl Deref for LavinaCore { pub broadcasting: Broadcasting,
type Target = Services; pub authenticator: Authenticator,
fn deref(&self) -> &Self::Target {
&self.services
}
}
impl LavinaCore {
pub async fn connect_to_player(&self, player_id: &PlayerId) -> PlayerConnection {
self.services.players.connect_to_player(&self, player_id).await
}
pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle> {
self.services.rooms.get_room(&self.services, room_id).await
}
pub async fn create_player(&self, player_id: &PlayerId) -> Result<()> {
self.services.storage.create_user(player_id.as_inner()).await
}
pub async fn get_all_rooms(&self) -> Vec<RoomInfo> {
self.services.rooms.get_all_rooms().await
}
pub async fn stop_player(&self, player_id: &PlayerId) -> Result<Option<()>> {
self.services.players.stop_player(player_id).await
}
}
pub struct Services {
pub(crate) players: PlayerRegistry,
pub(crate) rooms: RoomRegistry,
pub(crate) dialogs: DialogRegistry,
pub(crate) broadcasting: Broadcasting,
pub(crate) client: LavinaClient,
pub(crate) storage: Storage,
pub(crate) cluster_metadata: ClusterMetadata,
} }
impl LavinaCore { impl LavinaCore {
pub async fn new( pub async fn new(
metrics: &mut MetricsRegistry, mut metrics: MetricsRegistry,
cluster_config: ClusterConfig, cluster_config: ClusterConfig,
storage: Storage, storage: Storage,
) -> Result<LavinaCore> { ) -> Result<LavinaCore> {
// TODO shutdown all services in reverse order on error // TODO shutdown all services in reverse order on error
let broadcasting = Broadcasting::new(); let broadcasting = Broadcasting::new();
let client = LavinaClient::new(cluster_config.addresses.clone()); let client = LavinaClient::new(cluster_config.addresses.clone());
let rooms = RoomRegistry::new(metrics)?; let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
let dialogs = DialogRegistry::new(); let dialogs = DialogRegistry::new(storage.clone());
let players = PlayerRegistry::empty(metrics)?; let players = PlayerRegistry::empty(
rooms.clone(),
let services = Services { dialogs.clone(),
storage.clone(),
&mut metrics,
Arc::new(cluster_config.metadata),
client,
broadcasting.clone(),
)?;
dialogs.set_players(players.clone()).await;
let authenticator = Authenticator::new(storage.clone());
Ok(LavinaCore {
players, players,
rooms, rooms,
dialogs, dialogs,
broadcasting, broadcasting,
client, authenticator,
storage,
cluster_metadata: cluster_config.metadata,
};
Ok(LavinaCore {
services: Arc::new(services),
}) })
} }
pub async fn shutdown(self) -> Storage { pub async fn shutdown(mut self) -> Result<()> {
let _ = self.players.shutdown_all().await; self.players.shutdown_all().await?;
let services = match Arc::try_unwrap(self.services) { self.dialogs.unset_players().await;
Ok(e) => e, self.players.shutdown()?;
Err(_) => { self.dialogs.shutdown()?;
panic!("failed to acquire services ownership on shutdown"); self.rooms.shutdown()?;
} Ok(())
};
let _ = services.players.shutdown();
let _ = services.dialogs.shutdown();
let _ = services.rooms.shutdown();
services.storage
} }
} }

150
crates/lavina-core/src/player.rs
View File

@ -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, //! 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. //! so that they don't overload the room actor.
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use prometheus::{IntGauge, Registry as MetricsRegistry}; use prometheus::{IntGauge, Registry as MetricsRegistry};
@ -16,11 +17,14 @@ use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tracing::{Instrument, Span}; use tracing::{Instrument, Span};
use crate::clustering::broadcast::Broadcasting;
use crate::clustering::room::*; use crate::clustering::room::*;
use crate::clustering::{ClusterMetadata, LavinaClient};
use crate::dialog::DialogRegistry;
use crate::prelude::*; use crate::prelude::*;
use crate::room::{RoomHandle, RoomId, RoomInfo}; use crate::repo::Storage;
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
use crate::table::{AnonTable, Key as AnonKey}; use crate::table::{AnonTable, Key as AnonKey};
use crate::LavinaCore;
/// Opaque player identifier. Cannot contain spaces, must be shorter than 32. /// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@ -262,21 +266,41 @@ pub enum Updates {
} }
/// Handle to a player registry — a shared data structure containing information about players. /// Handle to a player registry — a shared data structure containing information about players.
pub(crate) struct PlayerRegistry(RwLock<PlayerRegistryInner>); #[derive(Clone)]
pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry { impl PlayerRegistry {
pub fn empty(metrics: &mut MetricsRegistry) -> Result<PlayerRegistry> { pub fn empty(
room_registry: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
metrics: &mut MetricsRegistry,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
) -> Result<PlayerRegistry> {
let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?; let metric_active_players = IntGauge::new("chat_players_active", "Number of alive player actors")?;
metrics.register(Box::new(metric_active_players.clone()))?; metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner { let inner = PlayerRegistryInner {
room_registry,
dialogs,
storage,
cluster_metadata,
cluster_client,
broadcasting,
players: HashMap::new(), players: HashMap::new(),
metric_active_players, metric_active_players,
}; };
Ok(PlayerRegistry(RwLock::new(inner))) Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
} }
pub fn shutdown(self) { pub fn shutdown(self) -> Result<()> {
let res = self.0.into_inner(); let res = match Arc::try_unwrap(self.0) {
Ok(e) => e,
Err(_) => return Err(fail("failed to acquire players ownership on shutdown")),
};
let res = res.into_inner();
drop(res); drop(res);
Ok(())
} }
#[tracing::instrument(skip(self), name = "PlayerRegistry::get_player")] #[tracing::instrument(skip(self), name = "PlayerRegistry::get_player")]
@ -299,8 +323,8 @@ impl PlayerRegistry {
} }
} }
#[tracing::instrument(skip(self, core), name = "PlayerRegistry::get_or_launch_player")] #[tracing::instrument(skip(self), name = "PlayerRegistry::get_or_launch_player")]
pub async fn get_or_launch_player(&self, core: &LavinaCore, id: &PlayerId) -> PlayerHandle { pub async fn get_or_launch_player(&self, id: &PlayerId) -> PlayerHandle {
let inner = self.0.read().await; let inner = self.0.read().await;
if let Some((handle, _)) = inner.players.get(id) { if let Some((handle, _)) = inner.players.get(id) {
handle.clone() handle.clone()
@ -310,7 +334,17 @@ impl PlayerRegistry {
if let Some((handle, _)) = inner.players.get(id) { if let Some((handle, _)) = inner.players.get(id) {
handle.clone() handle.clone()
} else { } else {
let (handle, fiber) = Player::launch(id.clone(), core.clone()).await; let (handle, fiber) = Player::launch(
id.clone(),
self.clone(),
inner.room_registry.clone(),
inner.dialogs.clone(),
inner.cluster_metadata.clone(),
inner.cluster_client.clone(),
inner.broadcasting.clone(),
inner.storage.clone(),
)
.await;
inner.players.insert(id.clone(), (handle.clone(), fiber)); inner.players.insert(id.clone(), (handle.clone(), fiber));
inner.metric_active_players.inc(); inner.metric_active_players.inc();
handle handle
@ -318,13 +352,13 @@ impl PlayerRegistry {
} }
} }
#[tracing::instrument(skip(self, core), name = "PlayerRegistry::connect_to_player")] #[tracing::instrument(skip(self), name = "PlayerRegistry::connect_to_player")]
pub async fn connect_to_player(&self, core: &LavinaCore, id: &PlayerId) -> PlayerConnection { pub async fn connect_to_player(&self, id: &PlayerId) -> PlayerConnection {
let player_handle = self.get_or_launch_player(core, id).await; let player_handle = self.get_or_launch_player(id).await;
player_handle.subscribe().await player_handle.subscribe().await
} }
pub async fn shutdown_all(&self) -> Result<()> { pub async fn shutdown_all(&mut self) -> Result<()> {
let mut inner = self.0.write().await; let mut inner = self.0.write().await;
for (i, (k, j)) in inner.players.drain() { for (i, (k, j)) in inner.players.drain() {
k.send(ActorCommand::Stop).await; k.send(ActorCommand::Stop).await;
@ -339,6 +373,12 @@ impl PlayerRegistry {
/// The player registry state representation. /// The player registry state representation.
struct PlayerRegistryInner { struct PlayerRegistryInner {
room_registry: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
/// Active player actors. /// Active player actors.
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>, players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
metric_active_players: IntGauge, metric_active_players: IntGauge,
@ -358,14 +398,29 @@ struct Player {
banned_from: HashSet<RoomId>, banned_from: HashSet<RoomId>,
rx: Receiver<(ActorCommand, Span)>, rx: Receiver<(ActorCommand, Span)>,
handle: PlayerHandle, handle: PlayerHandle,
services: LavinaCore, players: PlayerRegistry,
rooms: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
} }
impl Player { impl Player {
async fn launch(player_id: PlayerId, core: LavinaCore) -> (PlayerHandle, JoinHandle<Player>) { async fn launch(
player_id: PlayerId,
players: PlayerRegistry,
rooms: RoomRegistry,
dialogs: DialogRegistry,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
storage: Storage,
) -> (PlayerHandle, JoinHandle<Player>) {
let (tx, rx) = channel(32); let (tx, rx) = channel(32);
let handle = PlayerHandle { tx }; let handle = PlayerHandle { tx };
let handle_clone = handle.clone(); let handle_clone = handle.clone();
let storage_id = core.services.storage.retrieve_user_id_by_name(player_id.as_inner()).await.unwrap().unwrap(); let storage_id = storage.retrieve_user_id_by_name(player_id.as_inner()).await.unwrap().unwrap();
let player = Player { let player = Player {
player_id, player_id,
storage_id, storage_id,
@ -377,16 +432,22 @@ impl Player {
banned_from: HashSet::new(), banned_from: HashSet::new(),
rx, rx,
handle, handle,
services: core, players,
rooms,
dialogs,
storage,
cluster_metadata,
cluster_client,
broadcasting,
}; };
let fiber = tokio::task::spawn(player.main_loop()); let fiber = tokio::task::spawn(player.main_loop());
(handle_clone, fiber) (handle_clone, fiber)
} }
fn room_location(&self, room_id: &RoomId) -> Option<u32> { fn room_location(&self, room_id: &RoomId) -> Option<u32> {
let res = self.services.cluster_metadata.rooms.get(room_id.as_inner().as_ref()).copied(); let res = self.cluster_metadata.rooms.get(room_id.as_inner().as_ref()).copied();
let node = res.unwrap_or(self.services.cluster_metadata.main_owner); let node = res.unwrap_or(self.cluster_metadata.main_owner);
if node == self.services.cluster_metadata.node_id { if node == self.cluster_metadata.node_id {
None None
} else { } else {
Some(node) Some(node)
@ -394,13 +455,13 @@ impl Player {
} }
async fn main_loop(mut self) -> Self { async fn main_loop(mut self) -> Self {
let rooms = self.services.storage.get_rooms_of_a_user(self.storage_id).await.unwrap(); let rooms = self.storage.get_rooms_of_a_user(self.storage_id).await.unwrap();
for room_id in rooms { for room_id in rooms {
if let Some(remote_node) = self.room_location(&room_id) { if let Some(remote_node) = self.room_location(&room_id) {
self.my_rooms.insert(room_id.clone(), RoomRef::Remote { node_id: remote_node }); self.my_rooms.insert(room_id.clone(), RoomRef::Remote { node_id: remote_node });
self.services.subscribe(self.player_id.clone(), room_id).await; self.broadcasting.subscribe(self.player_id.clone(), room_id).await;
} else { } else {
let room = self.services.rooms.get_room(&self.services, &room_id).await; let room = self.rooms.get_room(&room_id).await;
if let Some(room) = room { if let Some(room) = room {
self.my_rooms.insert(room_id, RoomRef::Local(room)); self.my_rooms.insert(room_id, RoomRef::Local(room));
} else { } else {
@ -526,10 +587,9 @@ impl Player {
room_id: room_id.as_inner(), room_id: room_id.as_inner(),
player_id: self.player_id.as_inner(), player_id: self.player_id.as_inner(),
}; };
self.services.client.join_room(remote_node, req).await.unwrap(); self.cluster_client.join_room(remote_node, req).await.unwrap();
let room_storage_id = let room_storage_id = self.storage.create_or_retrieve_room_id_by_name(room_id.as_inner()).await.unwrap();
self.services.storage.create_or_retrieve_room_id_by_name(room_id.as_inner()).await.unwrap(); self.storage.add_room_member(room_storage_id, self.storage_id).await.unwrap();
self.services.storage.add_room_member(room_storage_id, self.storage_id).await.unwrap();
self.my_rooms.insert(room_id.clone(), RoomRef::Remote { node_id: remote_node }); self.my_rooms.insert(room_id.clone(), RoomRef::Remote { node_id: remote_node });
JoinResult::Success(RoomInfo { JoinResult::Success(RoomInfo {
id: room_id, id: room_id,
@ -537,14 +597,14 @@ impl Player {
members: vec![], members: vec![],
}) })
} else { } else {
let room = match self.services.rooms.get_or_create_room(&self.services, room_id.clone()).await { let room = match self.rooms.get_or_create_room(room_id.clone()).await {
Ok(room) => room, Ok(room) => room,
Err(e) => { Err(e) => {
log::error!("Failed to get or create room: {e}"); log::error!("Failed to get or create room: {e}");
todo!(); todo!();
} }
}; };
room.add_member(&self.services, &self.player_id, self.storage_id).await; room.add_member(&self.player_id, self.storage_id).await;
room.subscribe(&self.player_id, self.handle.clone()).await; room.subscribe(&self.player_id, self.handle.clone()).await;
self.my_rooms.insert(room_id.clone(), RoomRef::Local(room.clone())); self.my_rooms.insert(room_id.clone(), RoomRef::Local(room.clone()));
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
@ -564,17 +624,17 @@ impl Player {
match room { match room {
RoomRef::Local(room) => { RoomRef::Local(room) => {
room.unsubscribe(&self.player_id).await; room.unsubscribe(&self.player_id).await;
room.remove_member(&self.services, &self.player_id, self.storage_id).await; room.remove_member(&self.player_id, self.storage_id).await;
} }
RoomRef::Remote { node_id } => { RoomRef::Remote { node_id } => {
let req = LeaveRoomReq { let req = LeaveRoomReq {
room_id: room_id.as_inner(), room_id: room_id.as_inner(),
player_id: self.player_id.as_inner(), player_id: self.player_id.as_inner(),
}; };
self.services.client.leave_room(node_id, req).await.unwrap(); self.cluster_client.leave_room(node_id, req).await.unwrap();
let room_storage_id = let room_storage_id =
self.services.storage.create_or_retrieve_room_id_by_name(room_id.as_inner()).await.unwrap(); self.storage.create_or_retrieve_room_id_by_name(room_id.as_inner()).await.unwrap();
self.services.storage.remove_room_member(room_storage_id, self.storage_id).await.unwrap(); self.storage.remove_room_member(room_storage_id, self.storage_id).await.unwrap();
} }
} }
} }
@ -596,10 +656,10 @@ impl Player {
tracing::info!("no room found"); tracing::info!("no room found");
return SendMessageResult::NoSuchRoom; return SendMessageResult::NoSuchRoom;
}; };
let created_at = Utc::now(); let created_at = chrono::Utc::now();
match room { match room {
RoomRef::Local(room) => { RoomRef::Local(room) => {
room.send_message(&self.services, &self.player_id, body.clone(), created_at.clone()).await; room.send_message(&self.player_id, body.clone(), created_at.clone()).await;
} }
RoomRef::Remote { node_id } => { RoomRef::Remote { node_id } => {
let req = SendMessageReq { let req = SendMessageReq {
@ -608,9 +668,10 @@ impl Player {
message: &*body, message: &*body,
created_at: &*created_at.to_rfc3339(), created_at: &*created_at.to_rfc3339(),
}; };
self.services.client.send_room_message(*node_id, req).await.unwrap(); self.cluster_client.send_room_message(*node_id, req).await.unwrap();
self.services self.broadcasting
.broadcast( .broadcast(
&self.players,
room_id.clone(), room_id.clone(),
self.player_id.clone(), self.player_id.clone(),
body.clone(), body.clone(),
@ -637,7 +698,7 @@ impl Player {
}; };
match room { match room {
RoomRef::Local(room) => { RoomRef::Local(room) => {
room.set_topic(&self.services, &self.player_id, new_topic.clone()).await; room.set_topic(&self.player_id, new_topic.clone()).await;
} }
RoomRef::Remote { node_id } => { RoomRef::Remote { node_id } => {
let req = SetRoomTopicReq { let req = SetRoomTopicReq {
@ -645,7 +706,7 @@ impl Player {
player_id: self.player_id.as_inner(), player_id: self.player_id.as_inner(),
topic: &*new_topic, topic: &*new_topic,
}; };
self.services.client.set_room_topic(*node_id, req).await.unwrap(); self.cluster_client.set_room_topic(*node_id, req).await.unwrap();
} }
} }
let update = Updates::RoomTopicChanged { room_id, new_topic }; let update = Updates::RoomTopicChanged { room_id, new_topic };
@ -675,11 +736,8 @@ impl Player {
#[tracing::instrument(skip(self, body), name = "Player::send_dialog_message")] #[tracing::instrument(skip(self, body), name = "Player::send_dialog_message")]
async fn send_dialog_message(&self, connection_id: ConnectionId, recipient: PlayerId, body: Str) { async fn send_dialog_message(&self, connection_id: ConnectionId, recipient: PlayerId, body: Str) {
let created_at = Utc::now(); let created_at = chrono::Utc::now();
self.services self.dialogs.send_message(self.player_id.clone(), recipient.clone(), body.clone(), &created_at).await.unwrap();
.send_dialog_message(self.player_id.clone(), recipient.clone(), body.clone(), &created_at)
.await
.unwrap();
let update = Updates::NewDialogMessage { let update = Updates::NewDialogMessage {
sender: self.player_id.clone(), sender: self.player_id.clone(),
receiver: recipient.clone(), receiver: recipient.clone(),
@ -691,7 +749,7 @@ impl Player {
#[tracing::instrument(skip(self), name = "Player::check_user_existence")] #[tracing::instrument(skip(self), name = "Player::check_user_existence")]
async fn check_user_existence(&self, recipient: PlayerId) -> GetInfoResult { async fn check_user_existence(&self, recipient: PlayerId) -> GetInfoResult {
if self.services.storage.check_user_existence(recipient.as_inner().as_ref()).await.unwrap() { if self.storage.check_user_existence(recipient.as_inner().as_ref()).await.unwrap() {
GetInfoResult::UserExists GetInfoResult::UserExists
} else { } else {
GetInfoResult::UserDoesntExist GetInfoResult::UserDoesntExist

22
crates/lavina-core/src/repo/mod.rs
View File

@ -1,6 +1,7 @@
//! Storage and persistence logic. //! Storage and persistence logic.
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc;
use serde::Deserialize; use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteConnectOptions;
@ -19,8 +20,9 @@ pub struct StorageConfig {
pub db_path: String, pub db_path: String,
} }
#[derive(Clone)]
pub struct Storage { pub struct Storage {
conn: Mutex<SqliteConnection>, conn: Arc<Mutex<SqliteConnection>>,
} }
impl Storage { impl Storage {
pub async fn open(config: StorageConfig) -> Result<Storage> { pub async fn open(config: StorageConfig) -> Result<Storage> {
@ -32,17 +34,17 @@ impl Storage {
migrator.run(&mut conn).await?; migrator.run(&mut conn).await?;
log::info!("Migrations passed"); log::info!("Migrations passed");
let conn = Mutex::new(conn); let conn = Arc::new(Mutex::new(conn));
Ok(Storage { conn }) Ok(Storage { conn })
} }
pub async fn close(self) { pub async fn close(self) -> Result<()> {
let res = self.conn.into_inner(); let res = match Arc::try_unwrap(self.conn) {
match res.close().await { Ok(e) => e,
Ok(_) => {} Err(_) => return Err(fail("failed to acquire DB ownership on shutdown")),
Err(e) => { };
tracing::error!("Failed to close the DB connection: {e:?}"); let res = res.into_inner();
} res.close().await?;
} Ok(())
} }
} }

6
crates/lavina-core/src/repo/room.rs
View File

@ -30,7 +30,7 @@ impl Storage {
} }
#[tracing::instrument(skip(self, topic), name = "Storage::create_new_room")] #[tracing::instrument(skip(self, topic), name = "Storage::create_new_room")]
pub async fn create_new_room(&self, name: &str, topic: &str) -> Result<u32> { pub async fn create_new_room(&mut self, name: &str, topic: &str) -> Result<u32> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
let (id,): (u32,) = sqlx::query_as( let (id,): (u32,) = sqlx::query_as(
"insert into rooms(name, topic) "insert into rooms(name, topic)
@ -47,7 +47,7 @@ impl Storage {
#[tracing::instrument(skip(self, content, created_at), name = "Storage::insert_room_message")] #[tracing::instrument(skip(self, content, created_at), name = "Storage::insert_room_message")]
pub async fn insert_room_message( pub async fn insert_room_message(
&self, &mut self,
room_id: u32, room_id: u32,
id: u32, id: u32,
content: &str, content: &str,
@ -130,7 +130,7 @@ impl Storage {
} }
#[tracing::instrument(skip(self, topic), name = "Storage::set_room_topic")] #[tracing::instrument(skip(self, topic), name = "Storage::set_room_topic")]
pub async fn set_room_topic(&self, id: u32, topic: &str) -> Result<()> { pub async fn set_room_topic(&mut self, id: u32, topic: &str) -> Result<()> {
let mut executor = self.conn.lock().await; let mut executor = self.conn.lock().await;
sqlx::query( sqlx::query(
"update rooms "update rooms

84
crates/lavina-core/src/room.rs
View File

@ -9,7 +9,7 @@ use tokio::sync::RwLock as AsyncRwLock;
use crate::player::{PlayerHandle, PlayerId, Updates}; use crate::player::{PlayerHandle, PlayerId, Updates};
use crate::prelude::*; use crate::prelude::*;
use crate::Services; use crate::repo::Storage;
/// Opaque room id /// Opaque room id
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)] #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@ -35,32 +35,41 @@ impl RoomId {
} }
/// Shared data structure for storing metadata about rooms. /// Shared data structure for storing metadata about rooms.
pub(crate) struct RoomRegistry(AsyncRwLock<RoomRegistryInner>); #[derive(Clone)]
pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
impl RoomRegistry { impl RoomRegistry {
pub fn new(metrics: &mut MetricRegistry) -> Result<RoomRegistry> { pub fn new(metrics: &mut MetricRegistry, storage: Storage) -> Result<RoomRegistry> {
let metric_active_rooms = IntGauge::new("chat_rooms_active", "Number of alive room actors")?; let metric_active_rooms = IntGauge::new("chat_rooms_active", "Number of alive room actors")?;
metrics.register(Box::new(metric_active_rooms.clone()))?; metrics.register(Box::new(metric_active_rooms.clone()))?;
let inner = RoomRegistryInner { let inner = RoomRegistryInner {
rooms: HashMap::new(), rooms: HashMap::new(),
metric_active_rooms, metric_active_rooms,
storage,
}; };
Ok(RoomRegistry(AsyncRwLock::new(inner))) Ok(RoomRegistry(Arc::new(AsyncRwLock::new(inner))))
} }
pub fn shutdown(self) { pub fn shutdown(self) -> Result<()> {
// TODO iterate over rooms and stop them let res = match Arc::try_unwrap(self.0) {
Ok(e) => e,
Err(_) => return Err(fail("failed to acquire rooms ownership on shutdown")),
};
let res = res.into_inner();
// TODO drop all rooms
drop(res);
Ok(())
} }
#[tracing::instrument(skip(self, services), name = "RoomRegistry::get_or_create_room")] #[tracing::instrument(skip(self), name = "RoomRegistry::get_or_create_room")]
pub async fn get_or_create_room(&self, services: &Services, room_id: RoomId) -> Result<RoomHandle> { pub async fn get_or_create_room(&self, room_id: RoomId) -> Result<RoomHandle> {
let mut inner = self.0.write().await; let mut inner = self.0.write().await;
if let Some(room_handle) = inner.get_or_load_room(services, &room_id).await? { if let Some(room_handle) = inner.get_or_load_room(&room_id).await? {
Ok(room_handle.clone()) Ok(room_handle.clone())
} else { } else {
log::debug!("Creating room {}...", &room_id.0); log::debug!("Creating room {}...", &room_id.0);
let topic = "New room"; let topic = "New room";
let id = services.storage.create_new_room(&*room_id.0, &*topic).await?; let id = inner.storage.create_new_room(&*room_id.0, &*topic).await?;
let room = Room { let room = Room {
storage_id: id, storage_id: id,
room_id: room_id.clone(), room_id: room_id.clone(),
@ -68,6 +77,7 @@ impl RoomRegistry {
members: HashSet::new(), members: HashSet::new(),
topic: topic.into(), topic: topic.into(),
message_count: 0, message_count: 0,
storage: inner.storage.clone(),
}; };
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room))); let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
inner.rooms.insert(room_id, room_handle.clone()); inner.rooms.insert(room_id, room_handle.clone());
@ -76,10 +86,10 @@ impl RoomRegistry {
} }
} }
#[tracing::instrument(skip(self, services), name = "RoomRegistry::get_room")] #[tracing::instrument(skip(self), name = "RoomRegistry::get_room")]
pub async fn get_room(&self, services: &Services, room_id: &RoomId) -> Option<RoomHandle> { pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle> {
let mut inner = self.0.write().await; let mut inner = self.0.write().await;
inner.get_or_load_room(services, room_id).await.unwrap() inner.get_or_load_room(room_id).await.unwrap()
} }
#[tracing::instrument(skip(self), name = "RoomRegistry::get_all_rooms")] #[tracing::instrument(skip(self), name = "RoomRegistry::get_all_rooms")]
@ -100,15 +110,16 @@ impl RoomRegistry {
struct RoomRegistryInner { struct RoomRegistryInner {
rooms: HashMap<RoomId, RoomHandle>, rooms: HashMap<RoomId, RoomHandle>,
metric_active_rooms: IntGauge, metric_active_rooms: IntGauge,
storage: Storage,
} }
impl RoomRegistryInner { impl RoomRegistryInner {
#[tracing::instrument(skip(self, services), name = "RoomRegistryInner::get_or_load_room")] #[tracing::instrument(skip(self), name = "RoomRegistryInner::get_or_load_room")]
async fn get_or_load_room(&mut self, services: &Services, room_id: &RoomId) -> Result<Option<RoomHandle>> { async fn get_or_load_room(&mut self, room_id: &RoomId) -> Result<Option<RoomHandle>> {
if let Some(room_handle) = self.rooms.get(room_id) { if let Some(room_handle) = self.rooms.get(room_id) {
log::debug!("Room {} was loaded already", &room_id.0); log::debug!("Room {} was loaded already", &room_id.0);
Ok(Some(room_handle.clone())) Ok(Some(room_handle.clone()))
} else if let Some(stored_room) = services.storage.retrieve_room_by_name(&*room_id.0).await? { } else if let Some(stored_room) = self.storage.retrieve_room_by_name(&*room_id.0).await? {
log::debug!("Loading room {}...", &room_id.0); log::debug!("Loading room {}...", &room_id.0);
let room = Room { let room = Room {
storage_id: stored_room.id, storage_id: stored_room.id,
@ -117,6 +128,7 @@ impl RoomRegistryInner {
members: HashSet::new(), // TODO load members from storage members: HashSet::new(), // TODO load members from storage
topic: stored_room.topic.into(), topic: stored_room.topic.into(),
message_count: stored_room.message_count, message_count: stored_room.message_count,
storage: self.storage.clone(),
}; };
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room))); let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
self.rooms.insert(room_id.clone(), room_handle.clone()); self.rooms.insert(room_id.clone(), room_handle.clone());
@ -140,13 +152,13 @@ impl RoomHandle {
lock.subscriptions.insert(player_id.clone(), player_handle); lock.subscriptions.insert(player_id.clone(), player_handle);
} }
#[tracing::instrument(skip(self, services), name = "RoomHandle::add_member")] #[tracing::instrument(skip(self), name = "RoomHandle::add_member")]
pub async fn add_member(&self, services: &Services, player_id: &PlayerId, player_storage_id: u32) { pub async fn add_member(&self, player_id: &PlayerId, player_storage_id: u32) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
tracing::info!("Adding a new member to a room"); tracing::info!("Adding a new member to a room");
let room_storage_id = lock.storage_id; let room_storage_id = lock.storage_id;
if !services.storage.is_room_member(room_storage_id, player_storage_id).await.unwrap() { if !lock.storage.is_room_member(room_storage_id, player_storage_id).await.unwrap() {
services.storage.add_room_member(room_storage_id, player_storage_id).await.unwrap(); lock.storage.add_room_member(room_storage_id, player_storage_id).await.unwrap();
} else { } else {
tracing::warn!("User {:#?} has already been added to the room.", player_id); tracing::warn!("User {:#?} has already been added to the room.", player_id);
} }
@ -164,12 +176,12 @@ impl RoomHandle {
lock.subscriptions.remove(player_id); lock.subscriptions.remove(player_id);
} }
#[tracing::instrument(skip(self, services), name = "RoomHandle::remove_member")] #[tracing::instrument(skip(self), name = "RoomHandle::remove_member")]
pub async fn remove_member(&self, services: &Services, player_id: &PlayerId, player_storage_id: u32) { pub async fn remove_member(&self, player_id: &PlayerId, player_storage_id: u32) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
tracing::info!("Removing a member from a room"); tracing::info!("Removing a member from a room");
let room_storage_id = lock.storage_id; let room_storage_id = lock.storage_id;
services.storage.remove_room_member(room_storage_id, player_storage_id).await.unwrap(); lock.storage.remove_room_member(room_storage_id, player_storage_id).await.unwrap();
lock.members.remove(player_id); lock.members.remove(player_id);
let update = Updates::RoomLeft { let update = Updates::RoomLeft {
room_id: lock.room_id.clone(), room_id: lock.room_id.clone(),
@ -178,10 +190,10 @@ impl RoomHandle {
lock.broadcast_update(update, player_id).await; lock.broadcast_update(update, player_id).await;
} }
#[tracing::instrument(skip(self, services, body, created_at), name = "RoomHandle::send_message")] #[tracing::instrument(skip(self, body, created_at), name = "RoomHandle::send_message")]
pub async fn send_message(&self, services: &Services, player_id: &PlayerId, body: Str, created_at: DateTime<Utc>) { pub async fn send_message(&self, player_id: &PlayerId, body: Str, created_at: DateTime<Utc>) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
let res = lock.send_message(services, player_id, body, created_at).await; let res = lock.send_message(player_id, body, created_at).await;
if let Err(err) = res { if let Err(err) = res {
log::warn!("Failed to send message: {err:?}"); log::warn!("Failed to send message: {err:?}");
} }
@ -197,12 +209,12 @@ impl RoomHandle {
} }
} }
#[tracing::instrument(skip(self, services, new_topic), name = "RoomHandle::set_topic")] #[tracing::instrument(skip(self, new_topic), name = "RoomHandle::set_topic")]
pub async fn set_topic(&self, services: &Services, changer_id: &PlayerId, new_topic: Str) { pub async fn set_topic(&self, changer_id: &PlayerId, new_topic: Str) {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
let storage_id = lock.storage_id; let storage_id = lock.storage_id;
lock.topic = new_topic.clone(); lock.topic = new_topic.clone();
services.storage.set_room_topic(storage_id, &new_topic).await.unwrap(); lock.storage.set_room_topic(storage_id, &new_topic).await.unwrap();
let update = Updates::RoomTopicChanged { let update = Updates::RoomTopicChanged {
room_id: lock.room_id.clone(), room_id: lock.room_id.clone(),
new_topic: new_topic.clone(), new_topic: new_topic.clone(),
@ -223,20 +235,14 @@ struct Room {
/// The total number of messages. Used to calculate the id of the new message. /// The total number of messages. Used to calculate the id of the new message.
message_count: u32, message_count: u32,
topic: Str, topic: Str,
storage: Storage,
} }
impl Room { impl Room {
#[tracing::instrument(skip(self, services, body, created_at), name = "Room::send_message")] #[tracing::instrument(skip(self, body, created_at), name = "Room::send_message")]
async fn send_message( async fn send_message(&mut self, author_id: &PlayerId, body: Str, created_at: DateTime<Utc>) -> Result<()> {
&mut self,
services: &Services,
author_id: &PlayerId,
body: Str,
created_at: DateTime<Utc>,
) -> Result<()> {
tracing::info!("Adding a message to room"); tracing::info!("Adding a message to room");
services self.storage
.storage
.insert_room_message( .insert_room_message(
self.storage_id, self.storage_id,
self.message_count, self.message_count,

39
crates/projection-irc/src/lib.rs
View File

@ -14,10 +14,10 @@ use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::channel; use tokio::sync::mpsc::channel;
use lavina_core::auth::Verdict; use lavina_core::auth::{Authenticator, Verdict};
use lavina_core::player::*; use lavina_core::player::*;
use lavina_core::prelude::*; use lavina_core::prelude::*;
use lavina_core::room::{RoomId, RoomInfo}; use lavina_core::room::{RoomId, RoomInfo, RoomRegistry};
use lavina_core::terminator::Terminator; use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore; use lavina_core::LavinaCore;
use proto_irc::client::CapabilitySubcommand; use proto_irc::client::CapabilitySubcommand;
@ -79,7 +79,7 @@ async fn handle_socket(
match registered_user { match registered_user {
Ok(user) => { Ok(user) => {
log::debug!("User registered"); log::debug!("User registered");
handle_registered_socket(config, &core, &mut reader, &mut writer, user).await?; handle_registered_socket(config, core.players, core.rooms, &mut reader, &mut writer, user).await?;
} }
Err(err) => { Err(err) => {
log::debug!("Registration failed: {err}"); log::debug!("Registration failed: {err}");
@ -215,7 +215,7 @@ impl RegistrationState {
realname, realname,
enabled_capabilities: self.enabled_capabilities, enabled_capabilities: self.enabled_capabilities,
}; };
self.finalize_auth(candidate_user, writer, core, config).await self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
} }
}, },
ClientMessage::Nick { nickname } => { ClientMessage::Nick { nickname } => {
@ -229,7 +229,7 @@ impl RegistrationState {
realname: realname.clone(), realname: realname.clone(),
enabled_capabilities: self.enabled_capabilities, enabled_capabilities: self.enabled_capabilities,
}; };
self.finalize_auth(candidate_user, writer, core, config).await self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
} else { } else {
self.future_nickname = Some(nickname); self.future_nickname = Some(nickname);
Ok(None) Ok(None)
@ -246,7 +246,7 @@ impl RegistrationState {
realname, realname,
enabled_capabilities: self.enabled_capabilities, enabled_capabilities: self.enabled_capabilities,
}; };
self.finalize_auth(candidate_user, writer, core, config).await self.finalize_auth(candidate_user, writer, &core.authenticator, config).await
} else { } else {
self.future_username = Some((username, realname)); self.future_username = Some((username, realname));
Ok(None) Ok(None)
@ -277,7 +277,7 @@ impl RegistrationState {
} }
} else { } else {
let body = AuthBody::from_str(body.as_bytes())?; let body = AuthBody::from_str(body.as_bytes())?;
if let Err(e) = auth_user(core, &body.login, &body.password).await { if let Err(e) = auth_user(&core.authenticator, &body.login, &body.password).await {
tracing::warn!("Authentication failed: {:?}", e); tracing::warn!("Authentication failed: {:?}", e);
let target = self.future_nickname.clone().unwrap_or_else(|| "*".into()); let target = self.future_nickname.clone().unwrap_or_else(|| "*".into());
sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into()) sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into())
@ -325,7 +325,7 @@ impl RegistrationState {
&mut self, &mut self,
candidate_user: RegisteredUser, candidate_user: RegisteredUser,
writer: &mut BufWriter<WriteHalf<'_>>, writer: &mut BufWriter<WriteHalf<'_>>,
core: &LavinaCore, authenticator: &Authenticator,
config: &ServerConfig, config: &ServerConfig,
) -> Result<Option<RegisteredUser>> { ) -> Result<Option<RegisteredUser>> {
if self.enabled_capabilities.contains(Capabilities::Sasl) if self.enabled_capabilities.contains(Capabilities::Sasl)
@ -344,7 +344,7 @@ impl RegistrationState {
writer.flush().await?; writer.flush().await?;
return Ok(None); return Ok(None);
}; };
auth_user(core, &*candidate_user.nickname, &*candidate_password).await?; auth_user(authenticator, &*candidate_user.nickname, &*candidate_password).await?;
Ok(Some(candidate_user)) Ok(Some(candidate_user))
} }
} }
@ -406,8 +406,8 @@ fn sasl_fail_message(sender: Str, nick: Str, text: Str) -> ServerMessage {
} }
} }
async fn auth_user(core: &LavinaCore, login: &str, plain_password: &str) -> Result<()> { async fn auth_user(authenticator: &Authenticator, login: &str, plain_password: &str) -> Result<()> {
let verdict = core.authenticate(login, plain_password).await?; let verdict = authenticator.authenticate(login, plain_password).await?;
// TODO properly map these onto protocol messages // TODO properly map these onto protocol messages
match verdict { match verdict {
Verdict::Authenticated => Ok(()), Verdict::Authenticated => Ok(()),
@ -418,7 +418,8 @@ async fn auth_user(core: &LavinaCore, login: &str, plain_password: &str) -> Resu
async fn handle_registered_socket<'a>( async fn handle_registered_socket<'a>(
config: ServerConfig, config: ServerConfig,
core: &LavinaCore, players: PlayerRegistry,
rooms: RoomRegistry,
reader: &mut BufReader<ReadHalf<'a>>, reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>, writer: &mut BufWriter<WriteHalf<'a>>,
user: RegisteredUser, user: RegisteredUser,
@ -427,7 +428,7 @@ async fn handle_registered_socket<'a>(
log::info!("Handling registered user: {user:?}"); log::info!("Handling registered user: {user:?}");
let player_id = PlayerId::from(user.nickname.clone())?; let player_id = PlayerId::from(user.nickname.clone())?;
let mut connection = core.connect_to_player(&player_id).await; let mut connection = players.connect_to_player(&player_id).await;
let text: Str = format!("Welcome to {} Server", &config.server_name).into(); let text: Str = format!("Welcome to {} Server", &config.server_name).into();
ServerMessage { ServerMessage {
@ -501,7 +502,7 @@ async fn handle_registered_socket<'a>(
len len
}; };
let incoming = std::str::from_utf8(&buffer[0..len-2])?; let incoming = std::str::from_utf8(&buffer[0..len-2])?;
if let HandleResult::Leave = handle_incoming_message(incoming, &config, &user, core, &mut connection, writer).await? { if let HandleResult::Leave = handle_incoming_message(incoming, &config, &user, &rooms, &mut connection, writer).await? {
break; break;
} }
buffer.clear(); buffer.clear();
@ -509,7 +510,7 @@ async fn handle_registered_socket<'a>(
update = connection.receiver.recv() => { update = connection.receiver.recv() => {
match update { match update {
Some(ConnectionMessage::Update(update)) => { Some(ConnectionMessage::Update(update)) => {
handle_update(&config, &user, &player_id, writer, core, update).await?; handle_update(&config, &user, &player_id, writer, &rooms, update).await?;
} }
Some(ConnectionMessage::Stop(_)) => { Some(ConnectionMessage::Stop(_)) => {
tracing::debug!("Connection is being terminated"); tracing::debug!("Connection is being terminated");
@ -560,14 +561,14 @@ async fn handle_update(
user: &RegisteredUser, user: &RegisteredUser,
player_id: &PlayerId, player_id: &PlayerId,
writer: &mut (impl AsyncWrite + Unpin), writer: &mut (impl AsyncWrite + Unpin),
core: &LavinaCore, rooms: &RoomRegistry,
update: Updates, update: Updates,
) -> Result<()> { ) -> Result<()> {
log::debug!("Sending irc message to player {player_id:?} on update {update:?}"); log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
match update { match update {
Updates::RoomJoined { new_member_id, room_id } => { Updates::RoomJoined { new_member_id, room_id } => {
if player_id == &new_member_id { if player_id == &new_member_id {
if let Some(room) = core.get_room(&room_id).await { if let Some(room) = rooms.get_room(&room_id).await {
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
let chan = Chan::Global(room_id.as_inner().clone()); let chan = Chan::Global(room_id.as_inner().clone());
produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?; produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?;
@ -690,7 +691,7 @@ async fn handle_incoming_message(
buffer: &str, buffer: &str,
config: &ServerConfig, config: &ServerConfig,
user: &RegisteredUser, user: &RegisteredUser,
core: &LavinaCore, rooms: &RoomRegistry,
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
writer: &mut (impl AsyncWrite + Unpin), writer: &mut (impl AsyncWrite + Unpin),
) -> Result<HandleResult> { ) -> Result<HandleResult> {
@ -774,7 +775,7 @@ async fn handle_incoming_message(
writer.flush().await?; writer.flush().await?;
} }
Recipient::Chan(Chan::Global(chan)) => { Recipient::Chan(Chan::Global(chan)) => {
let room = core.get_room(&RoomId::from(chan.clone())?).await; let room = rooms.get_room(&RoomId::from(chan.clone())?).await;
if let Some(room) = room { if let Some(room) = room {
let room_info = room.get_room_info().await; let room_info = room.get_room_info().await;
for member in room_info.members { for member in room_info.members {

121
crates/projection-irc/tests/lib.rs
View File

@ -8,6 +8,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{ReadHalf, WriteHalf}; use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::TcpStream; use tokio::net::TcpStream;
use lavina_core::auth::Authenticator;
use lavina_core::clustering::{ClusterConfig, ClusterMetadata}; use lavina_core::clustering::{ClusterConfig, ClusterMetadata};
use lavina_core::player::{JoinResult, PlayerId, SendMessageResult}; use lavina_core::player::{JoinResult, PlayerId, SendMessageResult};
use lavina_core::repo::{Storage, StorageConfig}; use lavina_core::repo::{Storage, StorageConfig};
@ -101,6 +102,8 @@ impl<'a> TestScope<'a> {
} }
struct TestServer { struct TestServer {
metrics: MetricsRegistry,
storage: Storage,
core: LavinaCore, core: LavinaCore,
server: RunningServer, server: RunningServer,
} }
@ -111,7 +114,7 @@ impl TestServer {
listen_on: "127.0.0.1:0".parse().unwrap(), listen_on: "127.0.0.1:0".parse().unwrap(),
server_name: "testserver".into(), server_name: "testserver".into(),
}; };
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig { let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(), db_path: ":memory:".into(),
}) })
@ -124,9 +127,14 @@ impl TestServer {
rooms: Default::default(), rooms: Default::default(),
}, },
}; };
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
Ok(TestServer { core, server }) Ok(TestServer {
metrics,
storage,
core,
server,
})
} }
async fn reboot(self) -> Result<TestServer> { async fn reboot(self) -> Result<TestServer> {
@ -142,19 +150,30 @@ impl TestServer {
rooms: Default::default(), rooms: Default::default(),
}, },
}; };
let TestServer { core, server } = self; let TestServer {
metrics: _,
storage,
core,
server,
} = self;
server.terminate().await?; server.terminate().await?;
let storage = core.shutdown().await; core.shutdown().await?;
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
Ok(TestServer { core, server }) Ok(TestServer {
metrics,
storage,
core,
server,
})
} }
async fn shutdown(self) { async fn shutdown(self) -> Result<()> {
let _ = self.server.terminate().await; self.server.terminate().await?;
let storage = self.core.shutdown().await; self.core.shutdown().await?;
let _ = storage.close().await; self.storage.close().await?;
Ok(())
} }
} }
@ -164,8 +183,8 @@ async fn scenario_basic() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -183,7 +202,7 @@ async fn scenario_basic() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -193,8 +212,8 @@ async fn scenario_join_and_reboot() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -253,7 +272,7 @@ async fn scenario_join_and_reboot() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -263,8 +282,8 @@ async fn scenario_force_join_msg() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1); let mut s1 = TestScope::new(&mut stream1);
@ -319,7 +338,7 @@ async fn scenario_force_join_msg() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -329,10 +348,10 @@ async fn scenario_two_users() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester1")?).await?; server.storage.create_user("tester1").await?;
server.core.set_password("tester1", "password").await?; Authenticator::new(server.storage.clone()).set_password("tester1", "password").await?;
server.core.create_player(&PlayerId::from("tester2")?).await?; server.storage.create_user("tester2").await?;
server.core.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 stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1); let mut s1 = TestScope::new(&mut stream1);
@ -391,7 +410,7 @@ async fn scenario_two_users() -> Result<()> {
stream2.shutdown().await?; stream2.shutdown().await?;
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -405,8 +424,8 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -435,7 +454,7 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -445,8 +464,8 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -474,7 +493,7 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -484,8 +503,8 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -512,7 +531,7 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -522,8 +541,8 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -556,7 +575,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -566,8 +585,8 @@ async fn terminate_socket_scenario() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -579,7 +598,7 @@ async fn terminate_socket_scenario() -> Result<()> {
s.send("AUTHENTICATE PLAIN").await?; s.send("AUTHENTICATE PLAIN").await?;
s.expect(":testserver AUTHENTICATE +").await?; s.expect(":testserver AUTHENTICATE +").await?;
server.shutdown().await; server.shutdown().await?;
assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof); assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);
Ok(()) Ok(())
@ -591,8 +610,8 @@ async fn server_time_capability() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -617,8 +636,8 @@ async fn server_time_capability() -> Result<()> {
s.expect(":testserver 353 tester = #test :tester").await?; s.expect(":testserver 353 tester = #test :tester").await?;
s.expect(":testserver 366 tester #test :End of /NAMES list").await?; s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
server.core.create_player(&PlayerId::from("some_guy")?).await?; server.storage.create_user("some_guy").await?;
let mut conn = server.core.connect_to_player(&PlayerId::from("some_guy").unwrap()).await; let mut conn = server.core.players.connect_to_player(&PlayerId::from("some_guy").unwrap()).await;
let res = conn.join_room(RoomId::from("test").unwrap()).await?; let res = conn.join_room(RoomId::from("test").unwrap()).await?;
let JoinResult::Success(_) = res else { let JoinResult::Success(_) = res else {
panic!("Failed to join room"); panic!("Failed to join room");
@ -650,7 +669,7 @@ async fn server_time_capability() -> Result<()> {
// wrap up // wrap up
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }
@ -660,10 +679,10 @@ async fn scenario_two_players_dialog() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester1")?).await?; server.storage.create_user("tester1").await?;
server.core.set_password("tester1", "password").await?; server.storage.set_password("tester1", "password").await?;
server.core.create_player(&PlayerId::from("tester2")?).await?; server.storage.create_user("tester2").await?;
server.core.set_password("tester2", "password").await?; server.storage.set_password("tester2", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?; let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1); let mut s1 = TestScope::new(&mut stream1);
@ -714,7 +733,7 @@ async fn scenario_two_players_dialog() -> Result<()> {
stream1.shutdown().await?; stream1.shutdown().await?;
stream2.shutdown().await?; stream2.shutdown().await?;
server.shutdown().await; server.shutdown().await?;
Ok(()) Ok(())
} }

15
crates/projection-xmpp/src/iq.rs
View File

@ -2,10 +2,9 @@
use quick_xml::events::Event; use quick_xml::events::Event;
use lavina_core::room::RoomId; use lavina_core::room::{RoomId, RoomRegistry};
use lavina_core::LavinaCore;
use proto_xmpp::bind::{BindRequest, BindResponse, Jid, Name, Server}; use proto_xmpp::bind::{BindRequest, BindResponse, Jid, Name, Server};
use proto_xmpp::client::{Iq, IqError, IqErrorCondition, IqErrorType, IqType}; use proto_xmpp::client::{Iq, IqError, IqErrorType, IqType};
use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery}; use proto_xmpp::disco::{Feature, Identity, InfoQuery, Item, ItemQuery};
use proto_xmpp::mam::{Fin, Set}; use proto_xmpp::mam::{Fin, Set};
use proto_xmpp::roster::RosterQuery; use proto_xmpp::roster::RosterQuery;
@ -74,7 +73,7 @@ impl<'a> XmppConnection<'a> {
} }
} }
IqClientBody::DiscoItem(item) => { IqClientBody::DiscoItem(item) => {
let response = self.disco_items(iq.to.as_ref(), &item, self.core).await; let response = self.disco_items(iq.to.as_ref(), &item, self.rooms).await;
let req = Iq { let req = Iq {
from: iq.to, from: iq.to,
id: iq.id, id: iq.id,
@ -104,7 +103,6 @@ impl<'a> XmppConnection<'a> {
r#type: IqType::Error, r#type: IqType::Error,
body: IqError { body: IqError {
r#type: IqErrorType::Cancel, r#type: IqErrorType::Cancel,
condition: None,
}, },
}; };
req.serialize(output); req.serialize(output);
@ -164,7 +162,7 @@ impl<'a> XmppConnection<'a> {
resource: None, resource: None,
}) if server.0 == self.hostname_rooms => { }) if server.0 == self.hostname_rooms => {
let room_id = RoomId::from(room_name.0.clone()).unwrap(); let room_id = RoomId::from(room_name.0.clone()).unwrap();
let Some(_) = self.core.get_room(&room_id).await else { let Some(_) = self.rooms.get_room(&room_id).await else {
// TODO should return item-not-found // TODO should return item-not-found
// example: // example:
// <error type="cancel"> // <error type="cancel">
@ -173,7 +171,6 @@ impl<'a> XmppConnection<'a> {
// </error> // </error>
return Err(IqError { return Err(IqError {
r#type: IqErrorType::Cancel, r#type: IqErrorType::Cancel,
condition: Some(IqErrorCondition::ItemNotFound),
}); });
}; };
identity = vec![Identity { identity = vec![Identity {
@ -199,7 +196,7 @@ impl<'a> XmppConnection<'a> {
}) })
} }
async fn disco_items(&self, to: Option<&Jid>, req: &ItemQuery, core: &LavinaCore) -> ItemQuery { async fn disco_items(&self, to: Option<&Jid>, req: &ItemQuery, rooms: &RoomRegistry) -> ItemQuery {
let item = match to { let item = match to {
Some(Jid { Some(Jid {
name: None, name: None,
@ -221,7 +218,7 @@ impl<'a> XmppConnection<'a> {
server, server,
resource: None, resource: None,
}) if server.0 == self.hostname_rooms => { }) if server.0 == self.hostname_rooms => {
let room_list = core.get_all_rooms().await; let room_list = rooms.get_all_rooms().await;
room_list room_list
.into_iter() .into_iter()
.map(|room_info| Item { .map(|room_info| Item {

13
crates/projection-xmpp/src/lib.rs
View File

@ -25,6 +25,7 @@ use tokio_rustls::TlsAcceptor;
use lavina_core::auth::Verdict; use lavina_core::auth::Verdict;
use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, StopReason}; use lavina_core::player::{ConnectionMessage, PlayerConnection, PlayerId, StopReason};
use lavina_core::prelude::*; use lavina_core::prelude::*;
use lavina_core::room::RoomRegistry;
use lavina_core::terminator::Terminator; use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore; use lavina_core::LavinaCore;
use proto_xmpp::bind::{Name, Resource}; use proto_xmpp::bind::{Name, Resource};
@ -202,14 +203,14 @@ async fn handle_socket(
authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &core, &hostname) => { authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf, &core, &hostname) => {
match authenticated { match authenticated {
Ok(authenticated) => { Ok(authenticated) => {
let mut connection = core.connect_to_player(&authenticated.player_id).await; let mut connection = core.players.connect_to_player(&authenticated.player_id).await;
socket_final( socket_final(
&mut xml_reader, &mut xml_reader,
&mut xml_writer, &mut xml_writer,
&mut reader_buf, &mut reader_buf,
&authenticated, &authenticated,
&mut connection, &mut connection,
&core, &core.rooms,
&hostname, &hostname,
) )
.await?; .await?;
@ -295,7 +296,7 @@ async fn socket_auth(
match AuthBody::from_str(&auth.body) { match AuthBody::from_str(&auth.body) {
Ok(logopass) => { Ok(logopass) => {
let name = &logopass.login; let name = &logopass.login;
let verdict = core.authenticate(name, &logopass.password).await?; let verdict = core.authenticator.authenticate(name, &logopass.password).await?;
match verdict { match verdict {
Verdict::Authenticated => { Verdict::Authenticated => {
proto_xmpp::sasl::Success.write_xml(xml_writer).await?; proto_xmpp::sasl::Success.write_xml(xml_writer).await?;
@ -325,7 +326,7 @@ async fn socket_final(
reader_buf: &mut Vec<u8>, reader_buf: &mut Vec<u8>,
authenticated: &Authenticated, authenticated: &Authenticated,
user_handle: &mut PlayerConnection, user_handle: &mut PlayerConnection,
core: &LavinaCore, rooms: &RoomRegistry,
hostname: &Str, hostname: &Str,
) -> Result<()> { ) -> Result<()> {
// TODO validate the server hostname received in the stream start // TODO validate the server hostname received in the stream start
@ -358,7 +359,7 @@ async fn socket_final(
let mut conn = XmppConnection { let mut conn = XmppConnection {
user: authenticated, user: authenticated,
user_handle, user_handle,
core, rooms,
hostname: hostname.clone(), hostname: hostname.clone(),
hostname_rooms: format!("rooms.{}", hostname).into(), hostname_rooms: format!("rooms.{}", hostname).into(),
}; };
@ -441,7 +442,7 @@ async fn socket_final(
struct XmppConnection<'a> { struct XmppConnection<'a> {
user: &'a Authenticated, user: &'a Authenticated,
user_handle: &'a mut PlayerConnection, user_handle: &'a mut PlayerConnection,
core: &'a LavinaCore, rooms: &'a RoomRegistry,
hostname: Str, hostname: Str,
hostname_rooms: Str, hostname_rooms: Str,
} }

29
crates/projection-xmpp/src/presence.rs
View File

@ -1,12 +1,11 @@
//! Handling of all client2server presence stanzas //! Handling of all client2server presence stanzas
use anyhow::Result;
use quick_xml::events::Event; use quick_xml::events::Event;
use lavina_core::prelude::*;
use lavina_core::room::RoomId; use lavina_core::room::RoomId;
use proto_xmpp::bind::{Jid, Name, Server}; use proto_xmpp::bind::{Jid, Name, Server};
use proto_xmpp::client::Presence; use proto_xmpp::client::Presence;
use proto_xmpp::muc::XUser;
use proto_xmpp::xml::{Ignore, ToXml}; use proto_xmpp::xml::{Ignore, ToXml};
use crate::XmppConnection; use crate::XmppConnection;
@ -61,10 +60,10 @@ impl<'a> XmppConnection<'a> {
} }
// todo: return Presence and serialize on the outside. // todo: return Presence and serialize on the outside.
async fn muc_presence(&mut self, name: &Name) -> Result<(Presence<XUser>)> { async fn muc_presence(&mut self, name: &Name) -> Result<(Presence<()>)> {
let a = self.user_handle.join_room(RoomId::from(name.0.clone())?).await?; let a = self.user_handle.join_room(RoomId::from(name.0.clone())?).await?;
// TODO handle bans // TODO handle bans
let response = Presence { let response = Presence::<()> {
to: Some(Jid { to: Some(Jid {
name: Some(self.user.xmpp_name.clone()), name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()), server: Server(self.hostname.clone()),
@ -75,7 +74,6 @@ impl<'a> XmppConnection<'a> {
server: Server(self.hostname_rooms.clone()), server: Server(self.hostname_rooms.clone()),
resource: Some(self.user.xmpp_muc_name.clone()), resource: Some(self.user.xmpp_muc_name.clone()),
}), }),
custom: vec![XUser],
..Default::default() ..Default::default()
}; };
Ok(response) Ok(response)
@ -89,17 +87,15 @@ impl<'a> XmppConnection<'a> {
mod tests { mod tests {
use crate::testkit::{expect_user_authenticated, TestServer}; use crate::testkit::{expect_user_authenticated, TestServer};
use crate::Authenticated; use crate::Authenticated;
use anyhow::Result;
use lavina_core::player::PlayerId; use lavina_core::player::PlayerId;
use proto_xmpp::bind::{Jid, Name, Resource, Server}; use proto_xmpp::bind::{Jid, Name, Resource, Server};
use proto_xmpp::client::Presence; use proto_xmpp::client::Presence;
use proto_xmpp::muc::XUser;
#[tokio::test] #[tokio::test]
async fn test_muc_joining() -> Result<()> { async fn test_muc_joining() {
let server = TestServer::start().await.unwrap(); let server = TestServer::start().await.unwrap();
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await.unwrap();
let player_id = PlayerId::from("tester").unwrap(); let player_id = PlayerId::from("tester").unwrap();
let user = Authenticated { let user = Authenticated {
@ -109,11 +105,11 @@ mod tests {
xmpp_muc_name: Resource("tester".into()), xmpp_muc_name: Resource("tester".into()),
}; };
let mut player_conn = server.core.connect_to_player(&user.player_id).await; let mut player_conn = server.core.players.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap(); let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let response = conn.muc_presence(&user.xmpp_name).await.unwrap(); let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence { let expected = Presence::<()> {
to: Some(Jid { to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()), name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()), server: Server(conn.hostname.clone()),
@ -124,22 +120,20 @@ mod tests {
server: Server(conn.hostname_rooms.clone()), server: Server(conn.hostname_rooms.clone()),
resource: Some(conn.user.xmpp_muc_name.clone()), resource: Some(conn.user.xmpp_muc_name.clone()),
}), }),
custom: vec![XUser],
..Default::default() ..Default::default()
}; };
assert_eq!(expected, response); assert_eq!(expected, response);
server.shutdown().await.unwrap(); server.shutdown().await.unwrap();
Ok(())
} }
// Test that joining a room second time after a server restart, // Test that joining a room second time after a server restart,
// i.e. in-memory cache of memberships is cleaned, does not cause any issues. // i.e. in-memory cache of memberships is cleaned, does not cause any issues.
#[tokio::test] #[tokio::test]
async fn test_muc_joining_twice() -> Result<()> { async fn test_muc_joining_twice() {
let server = TestServer::start().await.unwrap(); let server = TestServer::start().await.unwrap();
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await.unwrap();
let player_id = PlayerId::from("tester").unwrap(); let player_id = PlayerId::from("tester").unwrap();
let user = Authenticated { let user = Authenticated {
@ -149,7 +143,7 @@ mod tests {
xmpp_muc_name: Resource("tester".into()), xmpp_muc_name: Resource("tester".into()),
}; };
let mut player_conn = server.core.connect_to_player(&user.player_id).await; let mut player_conn = server.core.players.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap(); let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let response = conn.muc_presence(&user.xmpp_name).await.unwrap(); let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
@ -171,13 +165,12 @@ mod tests {
drop(conn); drop(conn);
let server = server.reboot().await.unwrap(); let server = server.reboot().await.unwrap();
let mut player_conn = server.core.connect_to_player(&user.player_id).await; let mut player_conn = server.core.players.connect_to_player(&user.player_id).await;
let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap(); let mut conn = expect_user_authenticated(&server, &user, &mut player_conn).await.unwrap();
let response = conn.muc_presence(&user.xmpp_name).await.unwrap(); let response = conn.muc_presence(&user.xmpp_name).await.unwrap();
assert_eq!(expected, response); assert_eq!(expected, response);
server.shutdown().await.unwrap(); server.shutdown().await.unwrap();
Ok(())
} }
} }

26
crates/projection-xmpp/src/testkit.rs
View File

@ -8,13 +8,15 @@ use lavina_core::LavinaCore;
use proto_xmpp::bind::{BindRequest, BindResponse, Jid, Name, Resource, Server}; use proto_xmpp::bind::{BindRequest, BindResponse, Jid, Name, Resource, Server};
pub(crate) struct TestServer { pub(crate) struct TestServer {
pub metrics: MetricsRegistry,
pub storage: Storage,
pub core: LavinaCore, pub core: LavinaCore,
} }
impl TestServer { impl TestServer {
pub async fn start() -> anyhow::Result<TestServer> { pub async fn start() -> anyhow::Result<TestServer> {
let _ = tracing_subscriber::fmt::try_init(); let _ = tracing_subscriber::fmt::try_init();
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig { let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(), db_path: ":memory:".into(),
}) })
@ -27,14 +29,14 @@ impl TestServer {
}, },
addresses: vec![], addresses: vec![],
}; };
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
Ok(TestServer { core }) Ok(TestServer { metrics, storage, core })
} }
pub async fn reboot(self) -> anyhow::Result<TestServer> { pub async fn reboot(self) -> anyhow::Result<TestServer> {
let storage = self.core.shutdown().await; self.core.shutdown().await?;
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let cluster_config = ClusterConfig { let cluster_config = ClusterConfig {
metadata: ClusterMetadata { metadata: ClusterMetadata {
node_id: 0, node_id: 0,
@ -43,14 +45,18 @@ impl TestServer {
}, },
addresses: vec![], addresses: vec![],
}; };
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, self.storage.clone()).await?;
Ok(TestServer { core }) Ok(TestServer {
metrics,
storage: self.storage.clone(),
core,
})
} }
pub async fn shutdown(self) -> anyhow::Result<()> { pub async fn shutdown(self) -> anyhow::Result<()> {
let storage = self.core.shutdown().await; self.core.shutdown().await?;
storage.close().await; self.storage.close().await?;
Ok(()) Ok(())
} }
} }
@ -63,7 +69,7 @@ pub async fn expect_user_authenticated<'a>(
let conn = XmppConnection { let conn = XmppConnection {
user: &user, user: &user,
user_handle: conn, user_handle: conn,
core: &server.core, rooms: &server.core.rooms,
hostname: "localhost".into(), hostname: "localhost".into(),
hostname_rooms: "rooms.localhost".into(), hostname_rooms: "rooms.localhost".into(),
}; };

39
crates/projection-xmpp/tests/lib.rs
View File

@ -18,8 +18,8 @@ use tokio_rustls::rustls::client::ServerCertVerifier;
use tokio_rustls::rustls::{ClientConfig, ServerName}; use tokio_rustls::rustls::{ClientConfig, ServerName};
use tokio_rustls::TlsConnector; use tokio_rustls::TlsConnector;
use lavina_core::auth::Authenticator;
use lavina_core::clustering::{ClusterConfig, ClusterMetadata}; use lavina_core::clustering::{ClusterConfig, ClusterMetadata};
use lavina_core::player::PlayerId;
use lavina_core::repo::{Storage, StorageConfig}; use lavina_core::repo::{Storage, StorageConfig};
use lavina_core::LavinaCore; use lavina_core::LavinaCore;
use projection_xmpp::{launch, RunningServer, ServerConfig}; use projection_xmpp::{launch, RunningServer, ServerConfig};
@ -141,6 +141,8 @@ impl ServerCertVerifier for IgnoreCertVerification {
} }
struct TestServer { struct TestServer {
metrics: MetricsRegistry,
storage: Storage,
core: LavinaCore, core: LavinaCore,
server: RunningServer, server: RunningServer,
} }
@ -154,7 +156,7 @@ impl TestServer {
key: "tests/certs/xmpp.key".parse().unwrap(), key: "tests/certs/xmpp.key".parse().unwrap(),
hostname: "localhost".into(), hostname: "localhost".into(),
}; };
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig { let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(), db_path: ":memory:".into(),
}) })
@ -167,15 +169,20 @@ impl TestServer {
rooms: Default::default(), rooms: Default::default(),
}, },
}; };
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let server = launch(config, core.clone(), metrics.clone()).await.unwrap(); let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
Ok(TestServer { core, server }) Ok(TestServer {
metrics,
storage,
core,
server,
})
} }
async fn shutdown(self) -> Result<()> { async fn shutdown(self) -> Result<()> {
self.server.terminate().await?; self.server.terminate().await?;
let storage = self.core.shutdown().await; self.core.shutdown().await?;
storage.close().await; self.storage.close().await?;
Ok(()) Ok(())
} }
} }
@ -186,8 +193,8 @@ async fn scenario_basic() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -254,8 +261,8 @@ async fn scenario_wrong_password() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -309,8 +316,8 @@ async fn scenario_basic_without_headers() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -355,8 +362,8 @@ async fn terminate_socket() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);
@ -395,8 +402,8 @@ async fn test_message_archive_request() -> Result<()> {
// test scenario // test scenario
server.core.create_player(&PlayerId::from("tester")?).await?; server.storage.create_user("tester").await?;
server.core.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 stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream); let mut s = TestScope::new(&mut stream);

28
crates/proto-xmpp/src/client.rs
View File

@ -260,18 +260,6 @@ impl MessageType {
/// https://xmpp.org/rfcs/rfc6120.html#stanzas-error /// https://xmpp.org/rfcs/rfc6120.html#stanzas-error
pub struct IqError { pub struct IqError {
pub r#type: IqErrorType, pub r#type: IqErrorType,
pub condition: Option<IqErrorCondition>,
}
pub enum IqErrorCondition {
ItemNotFound,
}
impl IqErrorCondition {
pub fn as_str(&self) -> &'static str {
match self {
IqErrorCondition::ItemNotFound => "item-not-found",
}
}
} }
pub enum IqErrorType { pub enum IqErrorType {
@ -301,18 +289,7 @@ impl IqErrorType {
impl ToXml for IqError { impl ToXml for IqError {
fn serialize(&self, events: &mut Vec<Event<'static>>) { fn serialize(&self, events: &mut Vec<Event<'static>>) {
let bytes = BytesStart::new(format!(r#"error xmlns="{}" type="{}""#, XMLNS, self.r#type.as_str())); let bytes = BytesStart::new(format!(r#"error xmlns="{}" type="{}""#, XMLNS, self.r#type.as_str()));
match self.condition { events.push(Event::Empty(bytes));
None => {
events.push(Event::Empty(bytes));
}
Some(IqErrorCondition::ItemNotFound) => {
events.push(Event::Start(bytes));
let bytes2 = BytesStart::new(r#"item-not-found xmlns="urn:ietf:params:xml:ns:xmpp-stanzas""#);
events.push(Event::Empty(bytes2));
let bytes = BytesEnd::new("error");
events.push(Event::End(bytes));
}
}
} }
} }
@ -669,9 +646,6 @@ impl<T: ToXml> ToXml for Presence<T> {
Event::End(BytesEnd::new("priority")), Event::End(BytesEnd::new("priority")),
]); ]);
} }
for c in &self.custom {
c.serialize(events);
}
events.push(Event::End(BytesEnd::new("presence"))); events.push(Event::End(BytesEnd::new("presence")));
} }
} }

13
crates/proto-xmpp/src/muc/mod.rs
View File

@ -1,13 +1,12 @@
#![allow(unused_variables)] #![allow(unused_variables)]
use quick_xml::events::{BytesStart, Event}; use quick_xml::events::Event;
use quick_xml::name::ResolveResult; use quick_xml::name::ResolveResult;
use crate::xml::*; use crate::xml::*;
use anyhow::{anyhow, Result}; use anyhow::{anyhow, Result};
pub const XMLNS: &'static str = "http://jabber.org/protocol/muc"; pub const XMLNS: &'static str = "http://jabber.org/protocol/muc";
pub const XMLNS_USER: &'static str = "http://jabber.org/protocol/muc#user";
#[derive(PartialEq, Eq, Debug, Default)] #[derive(PartialEq, Eq, Debug, Default)]
pub struct History { pub struct History {
@ -144,16 +143,6 @@ impl FromXml for X {
} }
} }
#[derive(Debug, PartialEq, Eq)]
pub struct XUser;
impl ToXml for XUser {
fn serialize(&self, output: &mut Vec<Event<'static>>) {
let mut tag = BytesStart::new("x");
tag.push_attribute(("xmlns", XMLNS_USER));
output.push(Event::Empty(tag));
}
}
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;

45
src/http.rs
View File

@ -13,9 +13,10 @@ use serde::{Deserialize, Serialize};
use tokio::net::TcpListener; use tokio::net::TcpListener;
use lavina_core::auth::UpdatePasswordResult; use lavina_core::auth::UpdatePasswordResult;
use lavina_core::player::{PlayerId, SendMessageResult}; use lavina_core::player::{PlayerId, PlayerRegistry, SendMessageResult};
use lavina_core::prelude::*; use lavina_core::prelude::*;
use lavina_core::room::RoomId; use lavina_core::repo::Storage;
use lavina_core::room::{RoomId, RoomRegistry};
use lavina_core::terminator::Terminator; use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore; use lavina_core::LavinaCore;
use mgmt_api::*; use mgmt_api::*;
@ -29,11 +30,16 @@ pub struct ServerConfig {
pub listen_on: SocketAddr, pub listen_on: SocketAddr,
} }
pub async fn launch(config: ServerConfig, metrics: MetricsRegistry, core: LavinaCore) -> Result<Terminator> { pub async fn launch(
config: ServerConfig,
metrics: MetricsRegistry,
core: LavinaCore,
storage: Storage,
) -> Result<Terminator> {
log::info!("Starting the http service"); log::info!("Starting the http service");
let listener = TcpListener::bind(config.listen_on).await?; let listener = TcpListener::bind(config.listen_on).await?;
log::debug!("Listener started"); log::debug!("Listener started");
let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, core, rx.map(|_| ()))); let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, core, storage, rx.map(|_| ())));
Ok(terminator) Ok(terminator)
} }
@ -41,6 +47,7 @@ async fn main_loop(
listener: TcpListener, listener: TcpListener,
metrics: MetricsRegistry, metrics: MetricsRegistry,
core: LavinaCore, core: LavinaCore,
storage: Storage,
termination: impl Future<Output = ()>, termination: impl Future<Output = ()>,
) -> Result<()> { ) -> Result<()> {
pin!(termination); pin!(termination);
@ -53,8 +60,9 @@ async fn main_loop(
let stream = TokioIo::new(stream); let stream = TokioIo::new(stream);
let metrics = metrics.clone(); let metrics = metrics.clone();
let core = core.clone(); let core = core.clone();
let storage = storage.clone();
tokio::task::spawn(async move { tokio::task::spawn(async move {
let svc_fn = service_fn(|r| route(&metrics, &core, r)); let svc_fn = service_fn(|r| route(&metrics, &core, &storage, r));
let server = http1::Builder::new().serve_connection(stream, svc_fn); let server = http1::Builder::new().serve_connection(stream, svc_fn);
if let Err(err) = server.await { if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err); tracing::error!("Error serving connection: {:?}", err);
@ -71,18 +79,19 @@ async fn main_loop(
async fn route( async fn route(
registry: &MetricsRegistry, registry: &MetricsRegistry,
core: &LavinaCore, core: &LavinaCore,
storage: &Storage,
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
) -> HttpResult<Response<Full<Bytes>>> { ) -> HttpResult<Response<Full<Bytes>>> {
propagade_span_from_headers(&request); propagade_span_from_headers(&request);
let res = match (request.method(), request.uri().path()) { let res = match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => endpoint_metrics(registry), (&Method::GET, "/metrics") => endpoint_metrics(registry),
(&Method::GET, "/rooms") => endpoint_rooms(core).await, (&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, core).await.or5xx(), (&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
(&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core).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, 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::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, rooms::paths::SET_TOPIC) => endpoint_set_room_topic(request, core).await.or5xx(),
_ => clustering::route(core, request).await.unwrap_or_else(endpoint_not_found), _ => clustering::route(core, storage, request).await.unwrap_or_else(endpoint_not_found),
}; };
Ok(res) Ok(res)
} }
@ -95,23 +104,23 @@ fn endpoint_metrics(registry: &MetricsRegistry) -> Response<Full<Bytes>> {
} }
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_rooms(core: &LavinaCore) -> Response<Full<Bytes>> { async fn endpoint_rooms(rooms: &RoomRegistry) -> Response<Full<Bytes>> {
// TODO introduce management API types independent from core-domain types // TODO introduce management API types independent from core-domain types
// TODO remove `Serialize` implementations from all core-domain types // TODO remove `Serialize` implementations from all core-domain types
let room_list = core.get_all_rooms().await.to_body(); let room_list = rooms.get_all_rooms().await.to_body();
Response::new(room_list) Response::new(room_list)
} }
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_create_player( async fn endpoint_create_player(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: &LavinaCore, storage: &Storage,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
return Ok(malformed_request()); return Ok(malformed_request());
}; };
core.create_player(&PlayerId::from(res.name)?).await?; storage.create_user(&res.name).await?;
log::info!("Player {} created", res.name); log::info!("Player {} created", res.name);
let mut response = Response::new(Full::<Bytes>::default()); let mut response = Response::new(Full::<Bytes>::default());
*response.status_mut() = StatusCode::CREATED; *response.status_mut() = StatusCode::CREATED;
@ -121,7 +130,7 @@ async fn endpoint_create_player(
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn endpoint_stop_player( async fn endpoint_stop_player(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: &LavinaCore, players: &PlayerRegistry,
) -> Result<Response<Full<Bytes>>> { ) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<StopPlayerRequest>(&str[..]) else {
@ -130,7 +139,7 @@ async fn endpoint_stop_player(
let Ok(player_id) = PlayerId::from(res.name) else { let Ok(player_id) = PlayerId::from(res.name) else {
return Ok(player_not_found()); return Ok(player_not_found());
}; };
let Some(()) = core.stop_player(&player_id).await? else { let Some(()) = players.stop_player(&player_id).await? else {
return Ok(player_not_found()); return Ok(player_not_found());
}; };
Ok(empty_204_request()) Ok(empty_204_request())
@ -145,7 +154,7 @@ async fn endpoint_set_password(
let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else { let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
return Ok(malformed_request()); return Ok(malformed_request());
}; };
let verdict = core.set_password(&res.player_name, &res.password).await?; let verdict = core.authenticator.set_password(&res.player_name, &res.password).await?;
match verdict { match verdict {
UpdatePasswordResult::PasswordUpdated => {} UpdatePasswordResult::PasswordUpdated => {}
UpdatePasswordResult::UserNotFound => { UpdatePasswordResult::UserNotFound => {
@ -170,7 +179,7 @@ async fn endpoint_send_room_message(
let Ok(player_id) = PlayerId::from(req.author_id) else { let Ok(player_id) = PlayerId::from(req.author_id) else {
return Ok(player_not_found()); return Ok(player_not_found());
}; };
let mut player = core.connect_to_player(&player_id).await; let mut player = core.players.connect_to_player(&player_id).await;
let res = player.send_message(room_id, req.message.into()).await?; let res = player.send_message(room_id, req.message.into()).await?;
match res { match res {
SendMessageResult::NoSuchRoom => Ok(room_not_found()), SendMessageResult::NoSuchRoom => Ok(room_not_found()),
@ -193,7 +202,7 @@ async fn endpoint_set_room_topic(
let Ok(player_id) = PlayerId::from(req.author_id) else { let Ok(player_id) = PlayerId::from(req.author_id) else {
return Ok(player_not_found()); return Ok(player_not_found());
}; };
let mut player = core.connect_to_player(&player_id).await; let mut player = core.players.connect_to_player(&player_id).await;
player.change_topic(room_id, req.topic.into()).await?; player.change_topic(room_id, req.topic.into()).await?;
Ok(empty_204_request()) Ok(empty_204_request())
} }

26
src/http/clustering.rs
View File

@ -6,14 +6,19 @@ use super::Or5xx;
use crate::http::{empty_204_request, malformed_request, player_not_found, room_not_found}; use crate::http::{empty_204_request, malformed_request, player_not_found, room_not_found};
use lavina_core::clustering::room::{paths, JoinRoomReq, SendMessageReq}; use lavina_core::clustering::room::{paths, JoinRoomReq, SendMessageReq};
use lavina_core::player::PlayerId; use lavina_core::player::PlayerId;
use lavina_core::repo::Storage;
use lavina_core::room::RoomId; use lavina_core::room::RoomId;
use lavina_core::LavinaCore; use lavina_core::LavinaCore;
// TODO move this into core // TODO move this into core
pub async fn route(core: &LavinaCore, request: Request<hyper::body::Incoming>) -> Option<Response<Full<Bytes>>> { pub async fn route(
core: &LavinaCore,
storage: &Storage,
request: Request<hyper::body::Incoming>,
) -> Option<Response<Full<Bytes>>> {
match (request.method(), request.uri().path()) { match (request.method(), request.uri().path()) {
(&Method::POST, paths::JOIN) => Some(endpoint_cluster_join_room(request, core).await.or5xx()), (&Method::POST, paths::JOIN) => Some(endpoint_cluster_join_room(request, core, storage).await.or5xx()),
(&Method::POST, paths::ADD_MESSAGE) => Some(endpoint_cluster_add_message(request, core).await.or5xx()), (&Method::POST, paths::ADD_MESSAGE) => Some(endpoint_cluster_add_message(request, core).await.or5xx()),
_ => None, _ => None,
} }
@ -23,6 +28,7 @@ pub async fn route(core: &LavinaCore, request: Request<hyper::body::Incoming>) -
async fn endpoint_cluster_join_room( async fn endpoint_cluster_join_room(
request: Request<hyper::body::Incoming>, request: Request<hyper::body::Incoming>,
core: &LavinaCore, core: &LavinaCore,
storage: &Storage,
) -> lavina_core::prelude::Result<Response<Full<Bytes>>> { ) -> lavina_core::prelude::Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes(); let str = request.collect().await?.to_bytes();
let Ok(req) = serde_json::from_slice::<JoinRoomReq>(&str[..]) else { let Ok(req) = serde_json::from_slice::<JoinRoomReq>(&str[..]) else {
@ -37,7 +43,9 @@ async fn endpoint_cluster_join_room(
dbg!(&req.player_id); dbg!(&req.player_id);
return Ok(player_not_found()); return Ok(player_not_found());
}; };
core.cluster_join_room(room_id, &player_id).await?; let room_handle = core.rooms.get_or_create_room(room_id).await.unwrap();
let storage_id = storage.create_or_retrieve_user_id_by_name(req.player_id).await?;
room_handle.add_member(&player_id, storage_id).await;
Ok(empty_204_request()) Ok(empty_204_request())
} }
@ -63,10 +71,10 @@ async fn endpoint_cluster_add_message(
dbg!(&req.player_id); dbg!(&req.player_id);
return Ok(player_not_found()); return Ok(player_not_found());
}; };
let res = core.cluster_send_room_message(room_id, &player_id, req.message.into(), created_at.to_utc()).await?; let Some(room_handle) = core.rooms.get_room(&room_id).await else {
if let Some(_) = res { dbg!(&room_id);
Ok(empty_204_request()) return Ok(room_not_found());
} else { };
Ok(room_not_found()) room_handle.send_message(&player_id, req.message.into(), created_at.to_utc()).await;
} Ok(empty_204_request())
} }

9
src/main.rs
View File

@ -69,10 +69,10 @@ async fn main() -> Result<()> {
cluster: cluster_config, cluster: cluster_config,
tracing: _, tracing: _,
} = config; } = config;
let mut metrics = MetricsRegistry::new(); let metrics = MetricsRegistry::new();
let storage = Storage::open(storage_config).await?; let storage = Storage::open(storage_config).await?;
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?; let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), core.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()).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?; let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone()).await?;
tracing::info!("Started"); tracing::info!("Started");
@ -83,8 +83,7 @@ async fn main() -> Result<()> {
xmpp.terminate().await?; xmpp.terminate().await?;
irc.terminate().await?; irc.terminate().await?;
telemetry_terminator.terminate().await?; telemetry_terminator.terminate().await?;
let storage = core.shutdown().await; core.shutdown().await?;
storage.close().await;
tracing::info!("Shutdown complete"); tracing::info!("Shutdown complete");
Ok(()) Ok(())
} }