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

Compare commits

base: nikita:8b05c852e743b328114e3ee1c175ee0e4071f618
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. "8b05c852e743b328114e3ee1c175ee0e4071f618" and "d1a72e7978f6dbe441b35619c90aab4c9885def8" have entirely different histories.
8b05c852e7 ... d1a72e7978

22 changed files with 509 additions and 515 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 rand_core::OsRng;
use crate::LavinaCore;
use crate::prelude::log;
use crate::repo::Storage;
pub enum Verdict {
Authenticated,
@ -16,10 +17,18 @@ pub enum UpdatePasswordResult {
UserNotFound,
}
impl LavinaCore {
#[tracing::instrument(skip(self, provided_password), name = "Services::authenticate")]
#[derive(Clone)]
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> {
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);
};
if let Some(argon2_hash) = stored_user.argon2_hash {
@ -39,9 +48,9 @@ impl LavinaCore {
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> {
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);
};
@ -51,8 +60,8 @@ impl LavinaCore {
.hash_password(provided_password.as_bytes(), &salt)
.map_err(|e| anyhow!("Failed to hash password: {e:?}"))?;
self.services.storage.set_argon2_challenge(u.id, password_hash.to_string().as_str()).await?;
tracing::info!("Password changed for player {login}");
self.storage.set_argon2_challenge(u.id, password_hash.to_string().as_str()).await?;
log::info!("Password changed for player {login}");
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 reqwest::Client;
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_tracing::{DefaultSpanBackend, TracingMiddleware};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
pub mod broadcast;
pub mod room;
@ -27,15 +27,19 @@ pub struct ClusterMetadata {
pub rooms: HashMap<String, u32>,
}
#[derive(Clone)]
pub struct LavinaClient {
addresses: Addresses,
addresses: Arc<Addresses>,
client: ClientWithMiddleware,
}
impl LavinaClient {
pub fn new(addresses: Addresses) -> Self {
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<()> {

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

@ -1,32 +1,41 @@
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use chrono::{DateTime, Utc};
use tokio::sync::Mutex;
use crate::player::{PlayerId, Updates};
use crate::player::{PlayerId, PlayerRegistry, Updates};
use crate::prelude::Str;
use crate::room::RoomId;
use crate::Services;
/// Receives updates from other nodes and broadcasts them to local player actors.
struct BroadcastingInner {
subscriptions: HashMap<RoomId, HashSet<PlayerId>>,
}
pub struct Broadcasting(Mutex<BroadcastingInner>);
impl Broadcasting {}
#[derive(Clone)]
pub struct Broadcasting(Arc<Mutex<BroadcastingInner>>);
impl Broadcasting {
pub fn new() -> Self {
let inner = BroadcastingInner {
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.
#[tracing::instrument(skip(self, message, created_at))]
pub async fn broadcast(&self, room_id: RoomId, author_id: PlayerId, message: Str, created_at: DateTime<Utc>) {
let inner = self.broadcasting.0.lock().await;
#[tracing::instrument(skip(self, players, message, created_at), name = "Broadcasting::broadcast")]
pub async fn broadcast(
&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 {
return;
};
@ -40,7 +49,7 @@ impl Services {
if i == &author_id {
continue;
}
let Some(player) = self.players.get_player(i).await else {
let Some(player) = players.get_player(i).await else {
continue;
};
player.update(update.clone()).await;
@ -48,6 +57,6 @@ impl Services {
}
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 crate::clustering::LavinaClient;
use crate::player::PlayerId;
use crate::prelude::Str;
use crate::room::RoomId;
use crate::LavinaCore;
pub mod paths {
pub const JOIN: &'static str = "/cluster/rooms/join";
@ -43,46 +38,22 @@ pub struct SetRoomTopicReq<'a> {
impl LavinaClient {
#[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
}
#[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
}
#[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
}
#[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
}
}
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.
use std::collections::HashMap;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use tokio::sync::RwLock as AsyncRwLock;
use crate::player::{PlayerId, Updates};
use crate::player::{PlayerId, PlayerRegistry, Updates};
use crate::prelude::*;
use crate::Services;
use crate::repo::Storage;
/// Id of a conversation between two players.
///
@ -44,50 +45,55 @@ struct Dialog {
struct DialogRegistryInner {
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 {
#[tracing::instrument(skip(self, body, created_at))]
pub async fn send_dialog_message(
impl DialogRegistry {
pub async fn send_message(
&self,
from: PlayerId,
to: PlayerId,
body: Str,
created_at: &DateTime<Utc>,
) -> Result<()> {
let guard = self.dialogs.0.read().await;
let mut guard = self.0.read().await;
let id = DialogId::new(from.clone(), to.clone());
let dialog = guard.dialogs.get(&id);
if let Some(d) = dialog {
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)
.await?;
d.message_count += 1;
} else {
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
if let Some(d) = guard2.dialogs.get(&id) {
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)
.await?;
d.message_count += 1;
} else {
let (p1, p2) = id.as_inner();
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,
None => {
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");
self.storage
guard2
.storage
.insert_dialog_message(
stored_dialog.id,
stored_dialog.message_count,
@ -104,10 +110,14 @@ impl Services {
};
guard2.dialogs.insert(id.clone(), AsyncRwLock::new(dialog));
}
drop(guard2);
guard = guard2.downgrade();
}
// 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");
return Ok(());
};
@ -123,13 +133,33 @@ impl Services {
}
impl DialogRegistry {
pub fn new() -> DialogRegistry {
DialogRegistry(AsyncRwLock::new(DialogRegistryInner {
pub fn new(storage: Storage) -> DialogRegistry {
DialogRegistry(Arc::new(AsyncRwLock::new(DialogRegistryInner {
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)]

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

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

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,
//! so that they don't overload the room actor.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use chrono::{DateTime, Utc};
use prometheus::{IntGauge, Registry as MetricsRegistry};
@ -16,11 +17,14 @@ use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::RwLock;
use tracing::{Instrument, Span};
use crate::clustering::broadcast::Broadcasting;
use crate::clustering::room::*;
use crate::clustering::{ClusterMetadata, LavinaClient};
use crate::dialog::DialogRegistry;
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::LavinaCore;
/// Opaque player identifier. Cannot contain spaces, must be shorter than 32.
#[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.
pub(crate) struct PlayerRegistry(RwLock<PlayerRegistryInner>);
#[derive(Clone)]
pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
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")?;
metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner {
room_registry,
dialogs,
storage,
cluster_metadata,
cluster_client,
broadcasting,
players: HashMap::new(),
metric_active_players,
};
Ok(PlayerRegistry(RwLock::new(inner)))
Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
}
pub fn shutdown(self) {
let res = self.0.into_inner();
pub fn shutdown(self) -> Result<()> {
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);
Ok(())
}
#[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")]
pub async fn get_or_launch_player(&self, core: &LavinaCore, id: &PlayerId) -> PlayerHandle {
#[tracing::instrument(skip(self), name = "PlayerRegistry::get_or_launch_player")]
pub async fn get_or_launch_player(&self, id: &PlayerId) -> PlayerHandle {
let inner = self.0.read().await;
if let Some((handle, _)) = inner.players.get(id) {
handle.clone()
@ -310,7 +334,17 @@ impl PlayerRegistry {
if let Some((handle, _)) = inner.players.get(id) {
handle.clone()
} 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.metric_active_players.inc();
handle
@ -318,13 +352,13 @@ impl PlayerRegistry {
}
}
#[tracing::instrument(skip(self, core), name = "PlayerRegistry::connect_to_player")]
pub async fn connect_to_player(&self, core: &LavinaCore, id: &PlayerId) -> PlayerConnection {
let player_handle = self.get_or_launch_player(core, id).await;
#[tracing::instrument(skip(self), name = "PlayerRegistry::connect_to_player")]
pub async fn connect_to_player(&self, id: &PlayerId) -> PlayerConnection {
let player_handle = self.get_or_launch_player(id).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;
for (i, (k, j)) in inner.players.drain() {
k.send(ActorCommand::Stop).await;
@ -339,6 +373,12 @@ impl PlayerRegistry {
/// The player registry state representation.
struct PlayerRegistryInner {
room_registry: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
/// Active player actors.
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
metric_active_players: IntGauge,
@ -358,14 +398,29 @@ struct Player {
banned_from: HashSet<RoomId>,
rx: Receiver<(ActorCommand, Span)>,
handle: PlayerHandle,
services: LavinaCore,
players: PlayerRegistry,
rooms: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
cluster_metadata: Arc<ClusterMetadata>,
cluster_client: LavinaClient,
broadcasting: Broadcasting,
}
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 handle = PlayerHandle { tx };
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 {
player_id,
storage_id,
@ -377,16 +432,22 @@ impl Player {
banned_from: HashSet::new(),
rx,
handle,
services: core,
players,
rooms,
dialogs,
storage,
cluster_metadata,
cluster_client,
broadcasting,
};
let fiber = tokio::task::spawn(player.main_loop());
(handle_clone, fiber)
}
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 node = res.unwrap_or(self.services.cluster_metadata.main_owner);
if node == self.services.cluster_metadata.node_id {
let res = self.cluster_metadata.rooms.get(room_id.as_inner().as_ref()).copied();
let node = res.unwrap_or(self.cluster_metadata.main_owner);
if node == self.cluster_metadata.node_id {
None
} else {
Some(node)
@ -394,13 +455,13 @@ impl Player {
}
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 {
if let Some(remote_node) = self.room_location(&room_id) {
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 {
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 {
self.my_rooms.insert(room_id, RoomRef::Local(room));
} else {
@ -526,10 +587,9 @@ impl Player {
room_id: room_id.as_inner(),
player_id: self.player_id.as_inner(),
};
self.services.client.join_room(remote_node, req).await.unwrap();
let room_storage_id =
self.services.storage.create_or_retrieve_room_id_by_name(room_id.as_inner()).await.unwrap();
self.services.storage.add_room_member(room_storage_id, self.storage_id).await.unwrap();
self.cluster_client.join_room(remote_node, req).await.unwrap();
let room_storage_id = self.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.my_rooms.insert(room_id.clone(), RoomRef::Remote { node_id: remote_node });
JoinResult::Success(RoomInfo {
id: room_id,
@ -537,14 +597,14 @@ impl Player {
members: vec![],
})
} 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,
Err(e) => {
log::error!("Failed to get or create room: {e}");
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;
self.my_rooms.insert(room_id.clone(), RoomRef::Local(room.clone()));
let room_info = room.get_room_info().await;
@ -564,17 +624,17 @@ impl Player {
match room {
RoomRef::Local(room) => {
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 } => {
let req = LeaveRoomReq {
room_id: room_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 =
self.services.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.create_or_retrieve_room_id_by_name(room_id.as_inner()).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");
return SendMessageResult::NoSuchRoom;
};
let created_at = Utc::now();
let created_at = chrono::Utc::now();
match 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 } => {
let req = SendMessageReq {
@ -608,9 +668,10 @@ impl Player {
message: &*body,
created_at: &*created_at.to_rfc3339(),
};
self.services.client.send_room_message(*node_id, req).await.unwrap();
self.services
self.cluster_client.send_room_message(*node_id, req).await.unwrap();
self.broadcasting
.broadcast(
&self.players,
room_id.clone(),
self.player_id.clone(),
body.clone(),
@ -637,7 +698,7 @@ impl Player {
};
match 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 } => {
let req = SetRoomTopicReq {
@ -645,7 +706,7 @@ impl Player {
player_id: self.player_id.as_inner(),
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 };
@ -675,11 +736,8 @@ impl Player {
#[tracing::instrument(skip(self, body), name = "Player::send_dialog_message")]
async fn send_dialog_message(&self, connection_id: ConnectionId, recipient: PlayerId, body: Str) {
let created_at = Utc::now();
self.services
.send_dialog_message(self.player_id.clone(), recipient.clone(), body.clone(), &created_at)
.await
.unwrap();
let created_at = chrono::Utc::now();
self.dialogs.send_message(self.player_id.clone(), recipient.clone(), body.clone(), &created_at).await.unwrap();
let update = Updates::NewDialogMessage {
sender: self.player_id.clone(),
receiver: recipient.clone(),
@ -691,7 +749,7 @@ impl Player {
#[tracing::instrument(skip(self), name = "Player::check_user_existence")]
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
} else {
GetInfoResult::UserDoesntExist

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

@ -1,6 +1,7 @@
//! Storage and persistence logic.
use std::str::FromStr;
use std::sync::Arc;
use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions;
@ -19,8 +20,9 @@ pub struct StorageConfig {
pub db_path: String,
}
#[derive(Clone)]
pub struct Storage {
conn: Mutex<SqliteConnection>,
conn: Arc<Mutex<SqliteConnection>>,
}
impl Storage {
pub async fn open(config: StorageConfig) -> Result<Storage> {
@ -32,17 +34,17 @@ impl Storage {
migrator.run(&mut conn).await?;
log::info!("Migrations passed");
let conn = Mutex::new(conn);
let conn = Arc::new(Mutex::new(conn));
Ok(Storage { conn })
}
pub async fn close(self) {
let res = self.conn.into_inner();
match res.close().await {
Ok(_) => {}
Err(e) => {
tracing::error!("Failed to close the DB connection: {e:?}");
}
}
pub async fn close(self) -> Result<()> {
let res = match Arc::try_unwrap(self.conn) {
Ok(e) => e,
Err(_) => return Err(fail("failed to acquire DB ownership on shutdown")),
};
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")]
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 (id,): (u32,) = sqlx::query_as(
"insert into rooms(name, topic)
@ -47,7 +47,7 @@ impl Storage {
#[tracing::instrument(skip(self, content, created_at), name = "Storage::insert_room_message")]
pub async fn insert_room_message(
&self,
&mut self,
room_id: u32,
id: u32,
content: &str,
@ -130,7 +130,7 @@ impl Storage {
}
#[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;
sqlx::query(
"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::prelude::*;
use crate::Services;
use crate::repo::Storage;
/// Opaque room id
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
@ -35,32 +35,41 @@ impl RoomId {
}
/// Shared data structure for storing metadata about rooms.
pub(crate) struct RoomRegistry(AsyncRwLock<RoomRegistryInner>);
#[derive(Clone)]
pub struct RoomRegistry(Arc<AsyncRwLock<RoomRegistryInner>>);
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")?;
metrics.register(Box::new(metric_active_rooms.clone()))?;
let inner = RoomRegistryInner {
rooms: HashMap::new(),
metric_active_rooms,
storage,
};
Ok(RoomRegistry(AsyncRwLock::new(inner)))
Ok(RoomRegistry(Arc::new(AsyncRwLock::new(inner))))
}
pub fn shutdown(self) {
// TODO iterate over rooms and stop them
pub fn shutdown(self) -> Result<()> {
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")]
pub async fn get_or_create_room(&self, services: &Services, room_id: RoomId) -> Result<RoomHandle> {
#[tracing::instrument(skip(self), name = "RoomRegistry::get_or_create_room")]
pub async fn get_or_create_room(&self, room_id: RoomId) -> Result<RoomHandle> {
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())
} else {
log::debug!("Creating room {}...", &room_id.0);
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 {
storage_id: id,
room_id: room_id.clone(),
@ -68,6 +77,7 @@ impl RoomRegistry {
members: HashSet::new(),
topic: topic.into(),
message_count: 0,
storage: inner.storage.clone(),
};
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
inner.rooms.insert(room_id, room_handle.clone());
@ -76,10 +86,10 @@ impl RoomRegistry {
}
}
#[tracing::instrument(skip(self, services), name = "RoomRegistry::get_room")]
pub async fn get_room(&self, services: &Services, room_id: &RoomId) -> Option<RoomHandle> {
#[tracing::instrument(skip(self), name = "RoomRegistry::get_room")]
pub async fn get_room(&self, room_id: &RoomId) -> Option<RoomHandle> {
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")]
@ -100,15 +110,16 @@ impl RoomRegistry {
struct RoomRegistryInner {
rooms: HashMap<RoomId, RoomHandle>,
metric_active_rooms: IntGauge,
storage: Storage,
}
impl RoomRegistryInner {
#[tracing::instrument(skip(self, services), name = "RoomRegistryInner::get_or_load_room")]
async fn get_or_load_room(&mut self, services: &Services, room_id: &RoomId) -> Result<Option<RoomHandle>> {
#[tracing::instrument(skip(self), name = "RoomRegistryInner::get_or_load_room")]
async fn get_or_load_room(&mut self, room_id: &RoomId) -> Result<Option<RoomHandle>> {
if let Some(room_handle) = self.rooms.get(room_id) {
log::debug!("Room {} was loaded already", &room_id.0);
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);
let room = Room {
storage_id: stored_room.id,
@ -117,6 +128,7 @@ impl RoomRegistryInner {
members: HashSet::new(), // TODO load members from storage
topic: stored_room.topic.into(),
message_count: stored_room.message_count,
storage: self.storage.clone(),
};
let room_handle = RoomHandle(Arc::new(AsyncRwLock::new(room)));
self.rooms.insert(room_id.clone(), room_handle.clone());
@ -140,13 +152,13 @@ impl RoomHandle {
lock.subscriptions.insert(player_id.clone(), player_handle);
}
#[tracing::instrument(skip(self, services), name = "RoomHandle::add_member")]
pub async fn add_member(&self, services: &Services, player_id: &PlayerId, player_storage_id: u32) {
#[tracing::instrument(skip(self), name = "RoomHandle::add_member")]
pub async fn add_member(&self, player_id: &PlayerId, player_storage_id: u32) {
let mut lock = self.0.write().await;
tracing::info!("Adding a new member to a room");
let room_storage_id = lock.storage_id;
if !services.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();
if !lock.storage.is_room_member(room_storage_id, player_storage_id).await.unwrap() {
lock.storage.add_room_member(room_storage_id, player_storage_id).await.unwrap();
} else {
tracing::warn!("User {:#?} has already been added to the room.", player_id);
}
@ -164,12 +176,12 @@ impl RoomHandle {
lock.subscriptions.remove(player_id);
}
#[tracing::instrument(skip(self, services), name = "RoomHandle::remove_member")]
pub async fn remove_member(&self, services: &Services, player_id: &PlayerId, player_storage_id: u32) {
#[tracing::instrument(skip(self), name = "RoomHandle::remove_member")]
pub async fn remove_member(&self, player_id: &PlayerId, player_storage_id: u32) {
let mut lock = self.0.write().await;
tracing::info!("Removing a member from a room");
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);
let update = Updates::RoomLeft {
room_id: lock.room_id.clone(),
@ -178,10 +190,10 @@ impl RoomHandle {
lock.broadcast_update(update, player_id).await;
}
#[tracing::instrument(skip(self, services, body, created_at), name = "RoomHandle::send_message")]
pub async fn send_message(&self, services: &Services, player_id: &PlayerId, body: Str, created_at: DateTime<Utc>) {
#[tracing::instrument(skip(self, body, created_at), name = "RoomHandle::send_message")]
pub async fn send_message(&self, player_id: &PlayerId, body: Str, created_at: DateTime<Utc>) {
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 {
log::warn!("Failed to send message: {err:?}");
}
@ -197,12 +209,12 @@ impl RoomHandle {
}
}
#[tracing::instrument(skip(self, services, new_topic), name = "RoomHandle::set_topic")]
pub async fn set_topic(&self, services: &Services, changer_id: &PlayerId, new_topic: Str) {
#[tracing::instrument(skip(self, new_topic), name = "RoomHandle::set_topic")]
pub async fn set_topic(&self, changer_id: &PlayerId, new_topic: Str) {
let mut lock = self.0.write().await;
let storage_id = lock.storage_id;
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 {
room_id: lock.room_id.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.
message_count: u32,
topic: Str,
storage: Storage,
}
impl Room {
#[tracing::instrument(skip(self, services, body, created_at), name = "Room::send_message")]
async fn send_message(
&mut self,
services: &Services,
author_id: &PlayerId,
body: Str,
created_at: DateTime<Utc>,
) -> Result<()> {
#[tracing::instrument(skip(self, body, created_at), name = "Room::send_message")]
async fn send_message(&mut self, author_id: &PlayerId, body: Str, created_at: DateTime<Utc>) -> Result<()> {
tracing::info!("Adding a message to room");
services
.storage
self.storage
.insert_room_message(
self.storage_id,
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::sync::mpsc::channel;
use lavina_core::auth::Verdict;
use lavina_core::auth::{Authenticator, Verdict};
use lavina_core::player::*;
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::LavinaCore;
use proto_irc::client::CapabilitySubcommand;
@ -79,7 +79,7 @@ async fn handle_socket(
match registered_user {
Ok(user) => {
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) => {
log::debug!("Registration failed: {err}");
@ -215,7 +215,7 @@ impl RegistrationState {
realname,
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 } => {
@ -229,7 +229,7 @@ impl RegistrationState {
realname: realname.clone(),
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 {
self.future_nickname = Some(nickname);
Ok(None)
@ -246,7 +246,7 @@ impl RegistrationState {
realname,
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 {
self.future_username = Some((username, realname));
Ok(None)
@ -277,7 +277,7 @@ impl RegistrationState {
}
} else {
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);
let target = self.future_nickname.clone().unwrap_or_else(|| "*".into());
sasl_fail_message(config.server_name.clone(), target, "Bad credentials".into())
@ -325,7 +325,7 @@ impl RegistrationState {
&mut self,
candidate_user: RegisteredUser,
writer: &mut BufWriter<WriteHalf<'_>>,
core: &LavinaCore,
authenticator: &Authenticator,
config: &ServerConfig,
) -> Result<Option<RegisteredUser>> {
if self.enabled_capabilities.contains(Capabilities::Sasl)
@ -344,7 +344,7 @@ impl RegistrationState {
writer.flush().await?;
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))
}
}
@ -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<()> {
let verdict = core.authenticate(login, plain_password).await?;
async fn auth_user(authenticator: &Authenticator, login: &str, plain_password: &str) -> Result<()> {
let verdict = authenticator.authenticate(login, plain_password).await?;
// TODO properly map these onto protocol messages
match verdict {
Verdict::Authenticated => Ok(()),
@ -418,7 +418,8 @@ async fn auth_user(core: &LavinaCore, login: &str, plain_password: &str) -> Resu
async fn handle_registered_socket<'a>(
config: ServerConfig,
core: &LavinaCore,
players: PlayerRegistry,
rooms: RoomRegistry,
reader: &mut BufReader<ReadHalf<'a>>,
writer: &mut BufWriter<WriteHalf<'a>>,
user: RegisteredUser,
@ -427,7 +428,7 @@ async fn handle_registered_socket<'a>(
log::info!("Handling registered user: {user:?}");
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();
ServerMessage {
@ -501,7 +502,7 @@ async fn handle_registered_socket<'a>(
len
};
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;
}
buffer.clear();
@ -509,7 +510,7 @@ async fn handle_registered_socket<'a>(
update = connection.receiver.recv() => {
match 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(_)) => {
tracing::debug!("Connection is being terminated");
@ -560,14 +561,14 @@ async fn handle_update(
user: &RegisteredUser,
player_id: &PlayerId,
writer: &mut (impl AsyncWrite + Unpin),
core: &LavinaCore,
rooms: &RoomRegistry,
update: Updates,
) -> Result<()> {
log::debug!("Sending irc message to player {player_id:?} on update {update:?}");
match update {
Updates::RoomJoined { new_member_id, room_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 chan = Chan::Global(room_id.as_inner().clone());
produce_on_join_cmd_messages(&config, &user, &chan, &room_info, writer).await?;
@ -690,7 +691,7 @@ async fn handle_incoming_message(
buffer: &str,
config: &ServerConfig,
user: &RegisteredUser,
core: &LavinaCore,
rooms: &RoomRegistry,
user_handle: &mut PlayerConnection,
writer: &mut (impl AsyncWrite + Unpin),
) -> Result<HandleResult> {
@ -774,7 +775,7 @@ async fn handle_incoming_message(
writer.flush().await?;
}
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 {
let room_info = room.get_room_info().await;
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::TcpStream;
use lavina_core::auth::Authenticator;
use lavina_core::clustering::{ClusterConfig, ClusterMetadata};
use lavina_core::player::{JoinResult, PlayerId, SendMessageResult};
use lavina_core::repo::{Storage, StorageConfig};
@ -101,6 +102,8 @@ impl<'a> TestScope<'a> {
}
struct TestServer {
metrics: MetricsRegistry,
storage: Storage,
core: LavinaCore,
server: RunningServer,
}
@ -111,7 +114,7 @@ impl TestServer {
listen_on: "127.0.0.1:0".parse().unwrap(),
server_name: "testserver".into(),
};
let mut metrics = MetricsRegistry::new();
let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(),
})
@ -124,9 +127,14 @@ impl TestServer {
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();
Ok(TestServer { core, server })
Ok(TestServer {
metrics,
storage,
core,
server,
})
}
async fn reboot(self) -> Result<TestServer> {
@ -142,19 +150,30 @@ impl TestServer {
rooms: Default::default(),
},
};
let TestServer { core, server } = self;
let TestServer {
metrics: _,
storage,
core,
server,
} = self;
server.terminate().await?;
let storage = core.shutdown().await;
let mut metrics = MetricsRegistry::new();
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?;
core.shutdown().await?;
let metrics = MetricsRegistry::new();
let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
Ok(TestServer { core, server })
Ok(TestServer {
metrics,
storage,
core,
server,
})
}
async fn shutdown(self) {
let _ = self.server.terminate().await;
let storage = self.core.shutdown().await;
let _ = storage.close().await;
async fn shutdown(self) -> Result<()> {
self.server.terminate().await?;
self.core.shutdown().await?;
self.storage.close().await?;
Ok(())
}
}
@ -164,8 +183,8 @@ async fn scenario_basic() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -183,7 +202,7 @@ async fn scenario_basic() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -193,8 +212,8 @@ async fn scenario_join_and_reboot() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -253,7 +272,7 @@ async fn scenario_join_and_reboot() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -263,8 +282,8 @@ async fn scenario_force_join_msg() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1);
@ -319,7 +338,7 @@ async fn scenario_force_join_msg() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -329,10 +348,10 @@ async fn scenario_two_users() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester1")?).await?;
server.core.set_password("tester1", "password").await?;
server.core.create_player(&PlayerId::from("tester2")?).await?;
server.core.set_password("tester2", "password").await?;
server.storage.create_user("tester1").await?;
Authenticator::new(server.storage.clone()).set_password("tester1", "password").await?;
server.storage.create_user("tester2").await?;
Authenticator::new(server.storage.clone()).set_password("tester2", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1);
@ -391,7 +410,7 @@ async fn scenario_two_users() -> Result<()> {
stream2.shutdown().await?;
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -405,8 +424,8 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -435,7 +454,7 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -445,8 +464,8 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -474,7 +493,7 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -484,8 +503,8 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -512,7 +531,7 @@ async fn scenario_cap_short_negotiation() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -522,8 +541,8 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -556,7 +575,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -566,8 +585,8 @@ async fn terminate_socket_scenario() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -579,7 +598,7 @@ async fn terminate_socket_scenario() -> Result<()> {
s.send("AUTHENTICATE PLAIN").await?;
s.expect(":testserver AUTHENTICATE +").await?;
server.shutdown().await;
server.shutdown().await?;
assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);
Ok(())
@ -591,8 +610,8 @@ async fn server_time_capability() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -617,8 +636,8 @@ async fn server_time_capability() -> Result<()> {
s.expect(":testserver 353 tester = #test :tester").await?;
s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
server.core.create_player(&PlayerId::from("some_guy")?).await?;
let mut conn = server.core.connect_to_player(&PlayerId::from("some_guy").unwrap()).await;
server.storage.create_user("some_guy").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 JoinResult::Success(_) = res else {
panic!("Failed to join room");
@ -650,7 +669,7 @@ async fn server_time_capability() -> Result<()> {
// wrap up
server.shutdown().await;
server.shutdown().await?;
Ok(())
}
@ -660,10 +679,10 @@ async fn scenario_two_players_dialog() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester1")?).await?;
server.core.set_password("tester1", "password").await?;
server.core.create_player(&PlayerId::from("tester2")?).await?;
server.core.set_password("tester2", "password").await?;
server.storage.create_user("tester1").await?;
server.storage.set_password("tester1", "password").await?;
server.storage.create_user("tester2").await?;
server.storage.set_password("tester2", "password").await?;
let mut stream1 = TcpStream::connect(server.server.addr).await?;
let mut s1 = TestScope::new(&mut stream1);
@ -714,7 +733,7 @@ async fn scenario_two_players_dialog() -> Result<()> {
stream1.shutdown().await?;
stream2.shutdown().await?;
server.shutdown().await;
server.shutdown().await?;
Ok(())
}

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

@ -2,10 +2,9 @@
use quick_xml::events::Event;
use lavina_core::room::RoomId;
use lavina_core::LavinaCore;
use lavina_core::room::{RoomId, RoomRegistry};
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::mam::{Fin, Set};
use proto_xmpp::roster::RosterQuery;
@ -74,7 +73,7 @@ impl<'a> XmppConnection<'a> {
}
}
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 {
from: iq.to,
id: iq.id,
@ -104,7 +103,6 @@ impl<'a> XmppConnection<'a> {
r#type: IqType::Error,
body: IqError {
r#type: IqErrorType::Cancel,
condition: None,
},
};
req.serialize(output);
@ -164,7 +162,7 @@ impl<'a> XmppConnection<'a> {
resource: None,
}) if server.0 == self.hostname_rooms => {
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
// example:
// <error type="cancel">
@ -173,7 +171,6 @@ impl<'a> XmppConnection<'a> {
// </error>
return Err(IqError {
r#type: IqErrorType::Cancel,
condition: Some(IqErrorCondition::ItemNotFound),
});
};
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 {
Some(Jid {
name: None,
@ -221,7 +218,7 @@ impl<'a> XmppConnection<'a> {
server,
resource: None,
}) if server.0 == self.hostname_rooms => {
let room_list = core.get_all_rooms().await;
let room_list = rooms.get_all_rooms().await;
room_list
.into_iter()
.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::player::{ConnectionMessage, PlayerConnection, PlayerId, StopReason};
use lavina_core::prelude::*;
use lavina_core::room::RoomRegistry;
use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore;
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) => {
match 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(
&mut xml_reader,
&mut xml_writer,
&mut reader_buf,
&authenticated,
&mut connection,
&core,
&core.rooms,
&hostname,
)
.await?;
@ -295,7 +296,7 @@ async fn socket_auth(
match AuthBody::from_str(&auth.body) {
Ok(logopass) => {
let name = &logopass.login;
let verdict = core.authenticate(name, &logopass.password).await?;
let verdict = core.authenticator.authenticate(name, &logopass.password).await?;
match verdict {
Verdict::Authenticated => {
proto_xmpp::sasl::Success.write_xml(xml_writer).await?;
@ -325,7 +326,7 @@ async fn socket_final(
reader_buf: &mut Vec<u8>,
authenticated: &Authenticated,
user_handle: &mut PlayerConnection,
core: &LavinaCore,
rooms: &RoomRegistry,
hostname: &Str,
) -> Result<()> {
// TODO validate the server hostname received in the stream start
@ -358,7 +359,7 @@ async fn socket_final(
let mut conn = XmppConnection {
user: authenticated,
user_handle,
core,
rooms,
hostname: hostname.clone(),
hostname_rooms: format!("rooms.{}", hostname).into(),
};
@ -441,7 +442,7 @@ async fn socket_final(
struct XmppConnection<'a> {
user: &'a Authenticated,
user_handle: &'a mut PlayerConnection,
core: &'a LavinaCore,
rooms: &'a RoomRegistry,
hostname: Str,
hostname_rooms: Str,
}

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

@ -1,12 +1,11 @@
//! Handling of all client2server presence stanzas
use anyhow::Result;
use quick_xml::events::Event;
use lavina_core::prelude::*;
use lavina_core::room::RoomId;
use proto_xmpp::bind::{Jid, Name, Server};
use proto_xmpp::client::{Message, MessageType, Presence};
use proto_xmpp::muc::XUser;
use proto_xmpp::client::Presence;
use proto_xmpp::xml::{Ignore, ToXml};
use crate::XmppConnection;
@ -23,30 +22,11 @@ impl<'a> XmppConnection<'a> {
// resources in MUCs are members' personas – not implemented (yet?)
resource: Some(_),
}) if server.0 == self.hostname_rooms => {
let response = self.muc_presence(p.id.as_ref(), &name).await?;
let message = Message::<()> {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
resource: Some(self.user.xmpp_resource.clone()),
}),
from: Some(Jid {
name: Some(name.clone()),
server: Server(self.hostname_rooms.clone()),
resource: None,
}),
r#type: MessageType::Groupchat,
subject: Some("".into()),
id: None,
lang: None,
body: "".into(),
custom: vec![],
};
let response = self.muc_presence(&name).await?;
response.serialize(output);
message.serialize(output);
}
_ => {
// TODO other presence casess
// TODO other presence cases
let response = Presence::<()>::default();
response.serialize(output);
}
@ -75,18 +55,15 @@ impl<'a> XmppConnection<'a> {
};
response.serialize(output);
}
_ => {
tracing::error!("TODO");
}
_ => todo!(),
}
}
// todo: return Presence and serialize on the outside.
async fn muc_presence(&mut self, id: Option<&String>, 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?;
// TODO handle bans
let response = Presence {
id: id.map(ToOwned::to_owned),
let response = Presence::<()> {
to: Some(Jid {
name: Some(self.user.xmpp_name.clone()),
server: Server(self.hostname.clone()),
@ -97,7 +74,6 @@ impl<'a> XmppConnection<'a> {
server: Server(self.hostname_rooms.clone()),
resource: Some(self.user.xmpp_muc_name.clone()),
}),
custom: vec![XUser],
..Default::default()
};
Ok(response)
@ -111,17 +87,15 @@ impl<'a> XmppConnection<'a> {
mod tests {
use crate::testkit::{expect_user_authenticated, TestServer};
use crate::Authenticated;
use anyhow::Result;
use lavina_core::player::PlayerId;
use proto_xmpp::bind::{Jid, Name, Resource, Server};
use proto_xmpp::client::Presence;
use proto_xmpp::muc::XUser;
#[tokio::test]
async fn test_muc_joining() -> Result<()> {
async fn test_muc_joining() {
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 user = Authenticated {
@ -131,11 +105,11 @@ mod tests {
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 response = conn.muc_presence(&user.xmpp_name).await.unwrap();
let expected = Presence {
let expected = Presence::<()> {
to: Some(Jid {
name: Some(conn.user.xmpp_name.clone()),
server: Server(conn.hostname.clone()),
@ -146,22 +120,20 @@ mod tests {
server: Server(conn.hostname_rooms.clone()),
resource: Some(conn.user.xmpp_muc_name.clone()),
}),
custom: vec![XUser],
..Default::default()
};
assert_eq!(expected, response);
server.shutdown().await.unwrap();
Ok(())
}
// 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.
#[tokio::test]
async fn test_muc_joining_twice() -> Result<()> {
async fn test_muc_joining_twice() {
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 user = Authenticated {
@ -171,7 +143,7 @@ mod tests {
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 response = conn.muc_presence(&user.xmpp_name).await.unwrap();
@ -193,13 +165,12 @@ mod tests {
drop(conn);
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 response = conn.muc_presence(&user.xmpp_name).await.unwrap();
assert_eq!(expected, response);
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};
pub(crate) struct TestServer {
pub metrics: MetricsRegistry,
pub storage: Storage,
pub core: LavinaCore,
}
impl TestServer {
pub async fn start() -> anyhow::Result<TestServer> {
let _ = tracing_subscriber::fmt::try_init();
let mut metrics = MetricsRegistry::new();
let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(),
})
@ -27,14 +29,14 @@ impl TestServer {
},
addresses: vec![],
};
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?;
Ok(TestServer { core })
let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
Ok(TestServer { metrics, storage, core })
}
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 {
metadata: ClusterMetadata {
node_id: 0,
@ -43,14 +45,18 @@ impl TestServer {
},
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<()> {
let storage = self.core.shutdown().await;
storage.close().await;
self.core.shutdown().await?;
self.storage.close().await?;
Ok(())
}
}
@ -63,7 +69,7 @@ pub async fn expect_user_authenticated<'a>(
let conn = XmppConnection {
user: &user,
user_handle: conn,
core: &server.core,
rooms: &server.core.rooms,
hostname: "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::TlsConnector;
use lavina_core::auth::Authenticator;
use lavina_core::clustering::{ClusterConfig, ClusterMetadata};
use lavina_core::player::PlayerId;
use lavina_core::repo::{Storage, StorageConfig};
use lavina_core::LavinaCore;
use projection_xmpp::{launch, RunningServer, ServerConfig};
@ -141,6 +141,8 @@ impl ServerCertVerifier for IgnoreCertVerification {
}
struct TestServer {
metrics: MetricsRegistry,
storage: Storage,
core: LavinaCore,
server: RunningServer,
}
@ -154,7 +156,7 @@ impl TestServer {
key: "tests/certs/xmpp.key".parse().unwrap(),
hostname: "localhost".into(),
};
let mut metrics = MetricsRegistry::new();
let metrics = MetricsRegistry::new();
let storage = Storage::open(StorageConfig {
db_path: ":memory:".into(),
})
@ -167,15 +169,20 @@ impl TestServer {
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();
Ok(TestServer { core, server })
Ok(TestServer {
metrics,
storage,
core,
server,
})
}
async fn shutdown(self) -> Result<()> {
self.server.terminate().await?;
let storage = self.core.shutdown().await;
storage.close().await;
self.core.shutdown().await?;
self.storage.close().await?;
Ok(())
}
}
@ -186,8 +193,8 @@ async fn scenario_basic() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -254,8 +261,8 @@ async fn scenario_wrong_password() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -309,8 +316,8 @@ async fn scenario_basic_without_headers() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -355,8 +362,8 @@ async fn terminate_socket() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
@ -395,8 +402,8 @@ async fn test_message_archive_request() -> Result<()> {
// test scenario
server.core.create_player(&PlayerId::from("tester")?).await?;
server.core.set_password("tester", "password").await?;
server.storage.create_user("tester").await?;
Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);

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

@ -20,7 +20,7 @@ pub struct Message<T> {
// default is Normal
pub r#type: MessageType,
pub lang: Option<Str>,
pub subject: Option<Str>, // TODO there are three states of <subject -- not present, empty and some string
pub subject: Option<Str>,
pub body: Str,
pub custom: Vec<T>,
}
@ -50,7 +50,6 @@ enum MessageParserInner<T: FromXml> {
InBody(MessageParserState<T>),
InCustom(MessageParserState<T>, T::P),
}
#[derive(Default)]
struct MessageParserState<T> {
from: Option<Jid>,
@ -62,7 +61,6 @@ struct MessageParserState<T> {
body: Option<Str>,
custom: Vec<T>,
}
impl<T: FromXml> Parser for MessageParser<T> {
type Output = Result<Message<T>>;
@ -210,18 +208,9 @@ impl<T: ToXml> ToXml for Message<T> {
value: self.r#type.as_str().as_bytes().into(),
});
events.push(Event::Start(bytes));
if self.subject.is_none() {
// TODO make body optional
events.push(Event::Start(BytesStart::new("body")));
events.push(Event::Text(BytesText::new(&self.body).into_owned()));
events.push(Event::End(BytesEnd::new("body")));
}
if let Some(subject) = &self.subject {
// TODO make subject tri-state
events.push(Event::Start(BytesStart::new("subject")));
events.push(Event::Text(BytesText::new(&subject).into_owned()));
events.push(Event::End(BytesEnd::new("subject")));
}
events.push(Event::Start(BytesStart::new("body")));
events.push(Event::Text(BytesText::new(&self.body).into_owned()));
events.push(Event::End(BytesEnd::new("body")));
events.push(Event::End(BytesEnd::new("message")));
}
}
@ -271,19 +260,6 @@ impl MessageType {
/// https://xmpp.org/rfcs/rfc6120.html#stanzas-error
pub struct IqError {
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 {
@ -298,7 +274,6 @@ pub enum IqErrorType {
/// Retry after waiting (the error is temporary)
Wait,
}
impl IqErrorType {
pub fn as_str(&self) -> &'static str {
match self {
@ -314,18 +289,7 @@ impl IqErrorType {
impl ToXml for IqError {
fn serialize(&self, events: &mut Vec<Event<'static>>) {
let bytes = BytesStart::new(format!(r#"error xmlns="{}" type="{}""#, XMLNS, self.r#type.as_str()));
match self.condition {
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));
}
}
events.push(Event::Empty(bytes));
}
}
@ -358,7 +322,6 @@ enum IqParserInner<T: FromXml> {
ParsingBody(IqParserState<T>, T::P),
Final(IqParserState<T>),
}
struct IqParserState<T> {
pub from: Option<Jid>,
pub id: Option<String>,
@ -501,7 +464,6 @@ impl<T: ToXml> ToXml for Iq<T> {
#[derive(PartialEq, Eq, Debug)]
pub struct Presence<T> {
pub id: Option<String>,
pub to: Option<Jid>,
pub from: Option<Jid>,
pub priority: Option<PresencePriority>,
@ -514,7 +476,6 @@ pub struct Presence<T> {
impl<T> Default for Presence<T> {
fn default() -> Self {
Self {
id: Default::default(),
to: Default::default(),
from: Default::default(),
priority: Default::default(),
@ -589,10 +550,6 @@ impl<T: FromXml> FromXml for Presence<T> {
let s = std::str::from_utf8(&attr.value)?;
p.r#type = Some(s.into());
}
b"id" => {
let s = std::str::from_utf8(&attr.value)?;
p.r#type = Option::from(s.to_string());
}
_ => {}
}
}
@ -680,12 +637,6 @@ impl<T: ToXml> ToXml for Presence<T> {
value: from.to_string().as_bytes().into(),
}]);
}
if let Some(ref id) = self.id {
start.extend_attributes([Attribute {
key: QName(b"id"),
value: id.to_string().as_bytes().into(),
}]);
}
events.push(Event::Start(start));
if let Some(ref priority) = self.priority {
let s = priority.0.to_string();
@ -695,9 +646,6 @@ impl<T: ToXml> ToXml for Presence<T> {
Event::End(BytesEnd::new("priority")),
]);
}
for c in &self.custom {
c.serialize(events);
}
events.push(Event::End(BytesEnd::new("presence")));
}
}
@ -720,7 +668,7 @@ mod tests {
to: Some(Jid {
name: Some(Name("chelik".into())),
server: Server("xmpp.ru".into()),
resource: None,
resource: None
}),
r#type: MessageType::Chat,
lang: None,
@ -743,7 +691,7 @@ mod tests {
to: Some(Jid {
name: Some(Name("chelik".into())),
server: Server("xmpp.ru".into()),
resource: None,
resource: None
}),
r#type: MessageType::Chat,
lang: None,
@ -765,7 +713,7 @@ mod tests {
id: "bind_1".to_string(),
to: None,
r#type: IqType::Set,
body: BindRequest(Resource("mobile".into())),
body: BindRequest(Resource("mobile".into()))
}
)
}

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

@ -1,13 +1,12 @@
#![allow(unused_variables)]
use quick_xml::events::{BytesEnd, BytesStart, Event};
use quick_xml::events::Event;
use quick_xml::name::ResolveResult;
use crate::xml::*;
use anyhow::{anyhow, Result};
pub const XMLNS: &'static str = "http://jabber.org/protocol/muc";
pub const XMLNS_USER: &'static str = "http://jabber.org/protocol/muc#user";
#[derive(PartialEq, Eq, Debug, Default)]
pub struct History {
@ -144,28 +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::Start(tag));
let mut meg = BytesStart::new("item");
meg.push_attribute(("affiliation", "owner"));
meg.push_attribute(("role", "moderator"));
meg.push_attribute(("jid", "user@localhost/user"));
output.push(Event::Empty(meg));
let mut veg = BytesStart::new("status");
veg.push_attribute(("code", "110"));
output.push(Event::Empty(veg));
output.push(Event::End(BytesEnd::new("x")));
}
}
#[cfg(test)]
mod test {
use super::*;

45
src/http.rs
View File

@ -13,9 +13,10 @@ use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
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::room::RoomId;
use lavina_core::repo::Storage;
use lavina_core::room::{RoomId, RoomRegistry};
use lavina_core::terminator::Terminator;
use lavina_core::LavinaCore;
use mgmt_api::*;
@ -29,11 +30,16 @@ pub struct ServerConfig {
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");
let listener = TcpListener::bind(config.listen_on).await?;
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)
}
@ -41,6 +47,7 @@ async fn main_loop(
listener: TcpListener,
metrics: MetricsRegistry,
core: LavinaCore,
storage: Storage,
termination: impl Future<Output = ()>,
) -> Result<()> {
pin!(termination);
@ -53,8 +60,9 @@ async fn main_loop(
let stream = TokioIo::new(stream);
let metrics = metrics.clone();
let core = core.clone();
let storage = storage.clone();
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);
if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err);
@ -71,18 +79,19 @@ async fn main_loop(
async fn route(
registry: &MetricsRegistry,
core: &LavinaCore,
storage: &Storage,
request: Request<hyper::body::Incoming>,
) -> HttpResult<Response<Full<Bytes>>> {
propagade_span_from_headers(&request);
let res = match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => endpoint_metrics(registry),
(&Method::GET, "/rooms") => endpoint_rooms(core).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, core).await.or5xx(),
(&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, core).await.or5xx(),
(&Method::GET, "/rooms") => endpoint_rooms(&core.rooms).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
(&Method::POST, paths::STOP_PLAYER) => endpoint_stop_player(request, &core.players).await.or5xx(),
(&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, 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(),
_ => 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)
}
@ -95,23 +104,23 @@ fn endpoint_metrics(registry: &MetricsRegistry) -> Response<Full<Bytes>> {
}
#[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 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)
}
#[tracing::instrument(skip_all)]
async fn endpoint_create_player(
request: Request<hyper::body::Incoming>,
core: &LavinaCore,
storage: &Storage,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
return Ok(malformed_request());
};
core.create_player(&PlayerId::from(res.name)?).await?;
storage.create_user(&res.name).await?;
log::info!("Player {} created", res.name);
let mut response = Response::new(Full::<Bytes>::default());
*response.status_mut() = StatusCode::CREATED;
@ -121,7 +130,7 @@ async fn endpoint_create_player(
#[tracing::instrument(skip_all)]
async fn endpoint_stop_player(
request: Request<hyper::body::Incoming>,
core: &LavinaCore,
players: &PlayerRegistry,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
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 {
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());
};
Ok(empty_204_request())
@ -145,7 +154,7 @@ async fn endpoint_set_password(
let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
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 {
UpdatePasswordResult::PasswordUpdated => {}
UpdatePasswordResult::UserNotFound => {
@ -170,7 +179,7 @@ async fn endpoint_send_room_message(
let Ok(player_id) = PlayerId::from(req.author_id) else {
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?;
match res {
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 {
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?;
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 lavina_core::clustering::room::{paths, JoinRoomReq, SendMessageReq};
use lavina_core::player::PlayerId;
use lavina_core::repo::Storage;
use lavina_core::room::RoomId;
use lavina_core::LavinaCore;
// 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()) {
(&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()),
_ => None,
}
@ -23,6 +28,7 @@ pub async fn route(core: &LavinaCore, request: Request<hyper::body::Incoming>) -
async fn endpoint_cluster_join_room(
request: Request<hyper::body::Incoming>,
core: &LavinaCore,
storage: &Storage,
) -> lavina_core::prelude::Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(req) = serde_json::from_slice::<JoinRoomReq>(&str[..]) else {
@ -37,7 +43,9 @@ async fn endpoint_cluster_join_room(
dbg!(&req.player_id);
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())
}
@ -63,10 +71,10 @@ async fn endpoint_cluster_add_message(
dbg!(&req.player_id);
return Ok(player_not_found());
};
let res = core.cluster_send_room_message(room_id, &player_id, req.message.into(), created_at.to_utc()).await?;
if let Some(_) = res {
Ok(empty_204_request())
} else {
Ok(room_not_found())
}
let Some(room_handle) = core.rooms.get_room(&room_id).await else {
dbg!(&room_id);
return 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,
tracing: _,
} = config;
let mut metrics = MetricsRegistry::new();
let metrics = MetricsRegistry::new();
let storage = Storage::open(storage_config).await?;
let core = LavinaCore::new(&mut metrics, cluster_config, storage).await?;
let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), core.clone()).await?;
let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
let telemetry_terminator = http::launch(telemetry_config, metrics.clone(), core.clone(), storage.clone()).await?;
let irc = projection_irc::launch(irc_config, core.clone(), metrics.clone()).await?;
let xmpp = projection_xmpp::launch(xmpp_config, core.clone(), metrics.clone()).await?;
tracing::info!("Started");
@ -83,8 +83,7 @@ async fn main() -> Result<()> {
xmpp.terminate().await?;
irc.terminate().await?;
telemetry_terminator.terminate().await?;
let storage = core.shutdown().await;
storage.close().await;
core.shutdown().await?;
tracing::info!("Shutdown complete");
Ok(())
}