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

Compare commits

base: nikita:c983b59ddcbea2494b19a93a5cb8bcebefcb4572
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:ddb348bee929a3aa7b8d6545a96f5448280b4ce0
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. "c983b59ddcbea2494b19a93a5cb8bcebefcb4572" and "ddb348bee929a3aa7b8d6545a96f5448280b4ce0" have entirely different histories.
c983b59ddc ... ddb348bee9

18 changed files with 32 additions and 564 deletions
Show Stats Expand all files Collapse all files

6
Cargo.lock generated
View File

@ -1263,7 +1263,6 @@ version = "0.0.2-dev"
dependencies = [
"anyhow",
"bitflags 2.5.0",
"chrono",
"futures-util",
"lavina-core",
"nonempty",
@ -1775,7 +1774,6 @@ dependencies = [
"atoi",
"byteorder",
"bytes",
"chrono",
"crc",
"crossbeam-queue",
"either",
@ -1834,7 +1832,6 @@ dependencies = [
"sha2",
"sqlx-core",
"sqlx-mysql",
"sqlx-postgres",
"sqlx-sqlite",
"syn 1.0.109",
"tempfile",
@ -1852,7 +1849,6 @@ dependencies = [
"bitflags 2.5.0",
"byteorder",
"bytes",
"chrono",
"crc",
"digest",
"dotenvy",
@ -1894,7 +1890,6 @@ dependencies = [
"base64 0.21.7",
"bitflags 2.5.0",
"byteorder",
"chrono",
"crc",
"dotenvy",
"etcetera",
@ -1930,7 +1925,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b244ef0a8414da0bed4bb1910426e890b19e5e9bccc27ada6b797d05c55ae0aa"
dependencies = [
"atoi",
"chrono",
"flume",
"futures-channel",
"futures-core",

1
Cargo.toml
View File

@ -31,7 +31,6 @@ base64 = "0.22.0"
lavina-core = { path = "crates/lavina-core" }
tracing-subscriber = "0.3.16"
sasl = { path = "crates/sasl" }
chrono = "0.4.37"
[package]
name = "lavina"

4
crates/lavina-core/Cargo.toml
View File

@ -5,9 +5,9 @@ version.workspace = true
[dependencies]
anyhow.workspace = true
sqlx = { version = "0.7.4", features = ["sqlite", "migrate", "chrono"] }
sqlx = { version = "0.7.4", features = ["sqlite", "migrate"] }
serde.workspace = true
tokio.workspace = true
tracing.workspace = true
prometheus.workspace = true
chrono.workspace = true
chrono = "0.4.37"

17
crates/lavina-core/migrations/3_dialogs.sql
View File

@ -1,17 +0,0 @@
create table dialogs(
id integer primary key autoincrement not null,
participant_1 integer not null,
participant_2 integer not null,
created_at timestamp not null,
message_count integer not null default 0,
unique (participant_1, participant_2)
);
create table dialog_messages(
dialog_id integer not null,
id integer not null, -- unique per dialog, sequential in one dialog
author_id integer not null,
content string not null,
created_at timestamp not null,
primary key (dialog_id, id)
);

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

@ -1,150 +0,0 @@
//! Domain of dialogs – conversations between two participants.
//!
//! Dialogs are different from rooms in that they are always between two participants.
//! There are no
use std::collections::HashMap;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use tokio::sync::RwLock as AsyncRwLock;
use crate::player::{PlayerId, PlayerRegistry, Updates};
use crate::prelude::*;
use crate::repo::Storage;
/// Id of a conversation between two players.
///
/// Dialogs are identified by the pair of participants' ids. The order of ids does not matter.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DialogId(PlayerId, PlayerId);
impl DialogId {
pub fn new(a: PlayerId, b: PlayerId) -> DialogId {
if a.as_inner() < b.as_inner() {
DialogId(a, b)
} else {
DialogId(b, a)
}
}
pub fn as_inner(&self) -> (&PlayerId, &PlayerId) {
(&self.0, &self.1)
}
pub fn into_inner(self) -> (PlayerId, PlayerId) {
(self.0, self.1)
}
}
struct Dialog {
storage_id: u32,
player_storage_id_1: u32,
player_storage_id_2: u32,
message_count: u32,
}
struct DialogRegistryInner {
dialogs: HashMap<DialogId, AsyncRwLock<Dialog>>,
players: Option<PlayerRegistry>,
storage: Storage,
}
#[derive(Clone)]
pub struct DialogRegistry(Arc<AsyncRwLock<DialogRegistryInner>>);
impl DialogRegistry {
pub async fn send_message(
&self,
from: PlayerId,
to: PlayerId,
body: Str,
created_at: &DateTime<Utc>,
) -> Result<()> {
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;
guard.storage.increment_dialog_message_count(d.storage_id).await?;
d.message_count += 1;
} else {
drop(guard);
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;
guard2.storage.increment_dialog_message_count(d.storage_id).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 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");
guard2.storage.initialize_dialog(p1.as_inner(), p2.as_inner(), created_at).await?
}
};
tracing::info!("Dialog {id:?} loaded");
guard2.storage.increment_dialog_message_count(stored_dialog.id).await?;
let dialog = Dialog {
storage_id: stored_dialog.id,
player_storage_id_1: stored_dialog.participant_1,
player_storage_id_2: stored_dialog.participant_2,
message_count: stored_dialog.message_count + 1,
};
guard2.dialogs.insert(id.clone(), AsyncRwLock::new(dialog));
}
guard = guard2.downgrade();
}
// TODO send message to the other player and persist it
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(());
};
let update = Updates::NewDialogMessage {
sender: from.clone(),
receiver: to.clone(),
body: body.clone(),
created_at: chrono::Utc::now(), // todo
};
player.update(update).await;
return Ok(());
}
}
impl DialogRegistry {
pub fn new(storage: Storage) -> DialogRegistry {
DialogRegistry(Arc::new(AsyncRwLock::new(DialogRegistryInner {
dialogs: HashMap::new(),
players: None,
storage,
})))
}
pub async fn set_players(&self, players: PlayerRegistry) {
let mut guard = self.0.write().await;
guard.players = Some(players);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_dialog_id_new() {
let a = PlayerId::from("a").unwrap();
let b = PlayerId::from("b").unwrap();
let id1 = DialogId::new(a.clone(), b.clone());
let id2 = DialogId::new(a.clone(), b.clone());
// Dialog ids are invariant with respect to the order of participants
assert_eq!(id1, id2);
assert_eq!(id1.as_inner(), (&a, &b));
assert_eq!(id2.as_inner(), (&a, &b));
}
}

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

@ -2,12 +2,10 @@
use anyhow::Result;
use prometheus::Registry as MetricsRegistry;
use crate::dialog::DialogRegistry;
use crate::player::PlayerRegistry;
use crate::repo::Storage;
use crate::room::RoomRegistry;
pub mod dialog;
pub mod player;
pub mod prelude;
pub mod repo;
@ -20,21 +18,14 @@ mod table;
pub struct LavinaCore {
pub players: PlayerRegistry,
pub rooms: RoomRegistry,
pub dialogs: DialogRegistry,
}
impl LavinaCore {
pub async fn new(mut metrics: MetricsRegistry, storage: Storage) -> Result<LavinaCore> {
// TODO shutdown all services in reverse order on error
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)?;
dialogs.set_players(players.clone()).await;
Ok(LavinaCore {
players,
rooms,
dialogs,
})
let players = PlayerRegistry::empty(rooms.clone(), storage.clone(), &mut metrics)?;
Ok(LavinaCore { players, rooms })
}
pub async fn shutdown(mut self) -> Result<()> {

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

@ -10,13 +10,11 @@
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use chrono::{DateTime, Utc};
use prometheus::{IntGauge, Registry as MetricsRegistry};
use serde::Serialize;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::RwLock;
use crate::dialog::DialogRegistry;
use crate::prelude::*;
use crate::repo::Storage;
use crate::room::{RoomHandle, RoomId, RoomInfo, RoomRegistry};
@ -59,7 +57,7 @@ pub struct PlayerConnection {
}
impl PlayerConnection {
/// Handled in [Player::send_message].
pub async fn send_message(&mut self, room_id: RoomId, body: Str) -> Result<SendMessageResult> {
pub async fn send_message(&mut self, room_id: RoomId, body: Str) -> Result<()> {
let (promise, deferred) = oneshot();
let cmd = ClientCommand::SendMessage { room_id, body, promise };
self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
@ -105,17 +103,6 @@ impl PlayerConnection {
self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
Ok(deferred.await?)
}
pub async fn send_dialog_message(&self, recipient: PlayerId, body: Str) -> Result<()> {
let (promise, deferred) = oneshot();
let cmd = ClientCommand::SendDialogMessage {
recipient,
body,
promise,
};
self.player_handle.send(ActorCommand::ClientCommand(cmd, self.connection_id.clone())).await;
Ok(deferred.await?)
}
}
/// Handle to a player actor.
@ -176,7 +163,7 @@ pub enum ClientCommand {
SendMessage {
room_id: RoomId,
body: Str,
promise: Promise<SendMessageResult>,
promise: Promise<()>,
},
ChangeTopic {
room_id: RoomId,
@ -186,11 +173,6 @@ pub enum ClientCommand {
GetRooms {
promise: Promise<Vec<RoomInfo>>,
},
SendDialogMessage {
recipient: PlayerId,
body: Str,
promise: Promise<()>,
},
}
pub enum JoinResult {
@ -199,11 +181,6 @@ pub enum JoinResult {
Banned,
}
pub enum SendMessageResult {
Success(DateTime<Utc>),
NoSuchRoom,
}
/// Player update event type which is sent to a player actor and from there to a connection handler.
#[derive(Clone, Debug)]
pub enum Updates {
@ -215,7 +192,6 @@ pub enum Updates {
room_id: RoomId,
author_id: PlayerId,
body: Str,
created_at: DateTime<Utc>,
},
RoomJoined {
room_id: RoomId,
@ -227,12 +203,6 @@ pub enum Updates {
},
/// The player was banned from the room and left it immediately.
BannedFrom(RoomId),
NewDialogMessage {
sender: PlayerId,
receiver: PlayerId,
body: Str,
created_at: DateTime<Utc>,
},
}
/// Handle to a player registry — a shared data structure containing information about players.
@ -241,7 +211,6 @@ pub struct PlayerRegistry(Arc<RwLock<PlayerRegistryInner>>);
impl PlayerRegistry {
pub fn empty(
room_registry: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
metrics: &mut MetricsRegistry,
) -> Result<PlayerRegistry> {
@ -249,7 +218,6 @@ impl PlayerRegistry {
metrics.register(Box::new(metric_active_players.clone()))?;
let inner = PlayerRegistryInner {
room_registry,
dialogs,
storage,
players: HashMap::new(),
metric_active_players,
@ -257,23 +225,12 @@ impl PlayerRegistry {
Ok(PlayerRegistry(Arc::new(RwLock::new(inner))))
}
pub async fn get_player(&self, id: &PlayerId) -> Option<PlayerHandle> {
let inner = self.0.read().await;
inner.players.get(id).map(|(handle, _)| handle.clone())
}
pub async fn get_or_launch_player(&mut self, id: &PlayerId) -> PlayerHandle {
let mut inner = self.0.write().await;
if let Some((handle, _)) = inner.players.get(id) {
handle.clone()
} else {
let (handle, fiber) = Player::launch(
id.clone(),
inner.room_registry.clone(),
inner.dialogs.clone(),
inner.storage.clone(),
)
.await;
let (handle, fiber) = Player::launch(id.clone(), inner.room_registry.clone(), inner.storage.clone()).await;
inner.players.insert(id.clone(), (handle.clone(), fiber));
inner.metric_active_players.inc();
handle
@ -301,7 +258,6 @@ impl PlayerRegistry {
/// The player registry state representation.
struct PlayerRegistryInner {
room_registry: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
/// Active player actors.
players: HashMap<PlayerId, (PlayerHandle, JoinHandle<Player>)>,
@ -318,16 +274,10 @@ struct Player {
rx: Receiver<ActorCommand>,
handle: PlayerHandle,
rooms: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
}
impl Player {
async fn launch(
player_id: PlayerId,
rooms: RoomRegistry,
dialogs: DialogRegistry,
storage: Storage,
) -> (PlayerHandle, JoinHandle<Player>) {
async fn launch(player_id: PlayerId, rooms: RoomRegistry, storage: Storage) -> (PlayerHandle, JoinHandle<Player>) {
let (tx, rx) = channel(32);
let handle = PlayerHandle { tx };
let handle_clone = handle.clone();
@ -344,7 +294,6 @@ impl Player {
rx,
handle,
rooms,
dialogs,
storage,
};
let fiber = tokio::task::spawn(player.main_loop());
@ -418,8 +367,8 @@ impl Player {
let _ = promise.send(());
}
ClientCommand::SendMessage { room_id, body, promise } => {
let result = self.send_message(connection_id, room_id, body).await;
let _ = promise.send(result);
self.send_message(connection_id, room_id, body).await;
let _ = promise.send(());
}
ClientCommand::ChangeTopic {
room_id,
@ -433,14 +382,6 @@ impl Player {
let result = self.get_rooms().await;
let _ = promise.send(result);
}
ClientCommand::SendDialogMessage {
recipient,
body,
promise,
} => {
self.send_dialog_message(connection_id, recipient, body).await;
let _ = promise.send(());
}
}
}
@ -484,21 +425,18 @@ impl Player {
self.broadcast_update(update, connection_id).await;
}
async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) -> SendMessageResult {
async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) {
let Some(room) = self.my_rooms.get(&room_id) else {
tracing::info!("no room found");
return SendMessageResult::NoSuchRoom;
return;
};
let created_at = chrono::Utc::now();
room.send_message(&self.player_id, body.clone(), created_at.clone()).await;
room.send_message(&self.player_id, body.clone()).await;
let update = Updates::NewMessage {
room_id,
author_id: self.player_id.clone(),
body,
created_at,
};
self.broadcast_update(update, connection_id).await;
SendMessageResult::Success(created_at)
}
async fn change_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) {
@ -519,18 +457,6 @@ impl Player {
response
}
async fn send_dialog_message(&self, connection_id: ConnectionId, recipient: PlayerId, body: Str) {
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(),
body,
created_at,
};
self.broadcast_update(update, connection_id).await;
}
/// Broadcasts an update to all connections except the one with the given id.
///
/// This is called after handling a client command.

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

@ -1,68 +0,0 @@
use anyhow::Result;
use chrono::{DateTime, Utc};
use sqlx::FromRow;
use crate::repo::Storage;
impl Storage {
pub async fn retrieve_dialog(&self, participant_1: &str, participant_2: &str) -> Result<Option<StoredDialog>> {
let mut executor = self.conn.lock().await;
let res = sqlx::query_as(
"select r.id, r.participant_1, r.participant_2, r.message_count
from dialogs r join users u1 on r.participant_1 = u1.id join users u2 on r.participant_2 = u2.id
where u1.name = ? and u2.name = ?;",
)
.bind(participant_1)
.bind(participant_2)
.fetch_optional(&mut *executor)
.await?;
Ok(res)
}
pub async fn increment_dialog_message_count(&self, storage_id: u32) -> Result<()> {
let mut executor = self.conn.lock().await;
sqlx::query(
"update rooms set message_count = message_count + 1
where id = ?;",
)
.bind(storage_id)
.execute(&mut *executor)
.await?;
Ok(())
}
pub async fn initialize_dialog(
&self,
participant_1: &str,
participant_2: &str,
created_at: &DateTime<Utc>,
) -> Result<StoredDialog> {
let mut executor = self.conn.lock().await;
let res: StoredDialog = sqlx::query_as(
"insert into dialogs(participant_1, participant_2, created_at)
values (
(select id from users where name = ?),
(select id from users where name = ?),
?
)
returning id, participant_1, participant_2, message_count;",
)
.bind(participant_1)
.bind(participant_2)
.bind(&created_at)
.fetch_one(&mut *executor)
.await?;
Ok(res)
}
}
#[derive(FromRow)]
pub struct StoredDialog {
pub id: u32,
pub participant_1: u32,
pub participant_2: u32,
pub message_count: u32,
}

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

@ -4,7 +4,6 @@ use std::str::FromStr;
use std::sync::Arc;
use anyhow::anyhow;
use chrono::{DateTime, Utc};
use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{ConnectOptions, Connection, FromRow, Sqlite, SqliteConnection, Transaction};
@ -12,7 +11,6 @@ use tokio::sync::Mutex;
use crate::prelude::*;
mod dialog;
mod room;
mod user;
@ -82,14 +80,7 @@ impl Storage {
Ok(id)
}
pub async fn insert_message(
&mut self,
room_id: u32,
id: u32,
content: &str,
author_id: &str,
created_at: &DateTime<Utc>,
) -> Result<()> {
pub async fn insert_message(&mut self, room_id: u32, id: u32, content: &str, author_id: &str) -> Result<()> {
let mut executor = self.conn.lock().await;
let res: Option<(u32,)> = sqlx::query_as("select id from users where name = ?;")
.bind(author_id)
@ -107,7 +98,7 @@ impl Storage {
.bind(id)
.bind(content)
.bind(author_id)
.bind(created_at.to_string())
.bind(chrono::Utc::now().to_string())
.bind(room_id)
.execute(&mut *executor)
.await?;

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

@ -2,7 +2,6 @@
use std::collections::HashSet;
use std::{collections::HashMap, hash::Hash, sync::Arc};
use chrono::{DateTime, Utc};
use prometheus::{IntGauge, Registry as MetricRegistry};
use serde::Serialize;
use tokio::sync::RwLock as AsyncRwLock;
@ -164,9 +163,9 @@ impl RoomHandle {
lock.broadcast_update(update, player_id).await;
}
pub async fn send_message(&self, player_id: &PlayerId, body: Str, created_at: DateTime<Utc>) {
pub async fn send_message(&self, player_id: &PlayerId, body: Str) {
let mut lock = self.0.write().await;
let res = lock.send_message(player_id, body, created_at).await;
let res = lock.send_message(player_id, body).await;
if let Err(err) = res {
log::warn!("Failed to send message: {err:?}");
}
@ -209,23 +208,14 @@ struct Room {
storage: Storage,
}
impl Room {
async fn send_message(&mut self, author_id: &PlayerId, body: Str, created_at: DateTime<Utc>) -> Result<()> {
async fn send_message(&mut self, author_id: &PlayerId, body: Str) -> Result<()> {
tracing::info!("Adding a message to room");
self.storage
.insert_message(
self.storage_id,
self.message_count,
&body,
&*author_id.as_inner(),
&created_at,
)
.await?;
self.storage.insert_message(self.storage_id, self.message_count, &body, &*author_id.as_inner()).await?;
self.message_count += 1;
let update = Updates::NewMessage {
room_id: self.room_id.clone(),
author_id: author_id.clone(),
body,
created_at,
};
self.broadcast_update(update, author_id).await;
Ok(())

1
crates/projection-irc/Cargo.toml
View File

@ -12,7 +12,6 @@ tokio.workspace = true
prometheus.workspace = true
futures-util.workspace = true
nonempty.workspace = true
chrono.workspace = true
bitflags = "2.4.1"
proto-irc = { path = "../proto-irc" }
sasl = { path = "../sasl" }

3
crates/projection-irc/src/cap.rs
View File

@ -1,10 +1,9 @@
use bitflags::bitflags;
bitflags! {
#[derive(Debug, Clone, Copy)]
#[derive(Debug)]
pub struct Capabilities: u32 {
const None = 0;
const Sasl = 1 << 0;
const ServerTime = 1 << 1;
}
}

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

@ -2,7 +2,6 @@ use std::collections::HashMap;
use std::net::SocketAddr;
use anyhow::{anyhow, Result};
use chrono::SecondsFormat;
use futures_util::future::join_all;
use nonempty::nonempty;
use nonempty::NonEmpty;
@ -25,7 +24,7 @@ use proto_irc::client::{client_message, ClientMessage};
use proto_irc::server::CapSubBody;
use proto_irc::server::{AwayStatus, ServerMessage, ServerMessageBody};
use proto_irc::user::PrefixedNick;
use proto_irc::{Chan, Recipient, Tag};
use proto_irc::{Chan, Recipient};
use sasl::AuthBody;
mod cap;
@ -50,7 +49,6 @@ struct RegisteredUser {
*/
username: Str,
realname: Str,
enabled_capabilities: Capabilities,
}
async fn handle_socket(
@ -138,7 +136,7 @@ impl RegistrationState {
sender: Some(config.server_name.clone().into()),
body: ServerMessageBody::Cap {
target: self.future_nickname.clone().unwrap_or_else(|| "*".into()),
subcmd: CapSubBody::Ls("sasl=PLAIN server-time".into()),
subcmd: CapSubBody::Ls("sasl=PLAIN".into()),
},
}
.write_async(writer)
@ -158,30 +156,16 @@ impl RegistrationState {
self.enabled_capabilities |= Capabilities::Sasl;
}
acked.push(cap);
} else if &*cap.name == "server-time" {
if cap.to_disable {
self.enabled_capabilities &= !Capabilities::ServerTime;
} else {
self.enabled_capabilities |= Capabilities::ServerTime;
}
acked.push(cap);
} else {
naked.push(cap);
}
}
let mut ack_body = String::new();
if let Some((first, tail)) = acked.split_first() {
if first.to_disable {
for cap in acked {
if cap.to_disable {
ack_body.push('-');
}
ack_body += &*first.name;
for cap in tail {
ack_body.push(' ');
if cap.to_disable {
ack_body.push('-');
}
ack_body += &*cap.name;
}
ack_body += &*cap.name;
}
ServerMessage {
tags: vec![],
@ -211,7 +195,6 @@ impl RegistrationState {
nickname: nickname.clone(),
username,
realname,
enabled_capabilities: self.enabled_capabilities,
};
self.finalize_auth(candidate_user, writer, storage, config).await
}
@ -225,7 +208,6 @@ impl RegistrationState {
nickname: nickname.clone(),
username: username.clone(),
realname: realname.clone(),
enabled_capabilities: self.enabled_capabilities,
};
self.finalize_auth(candidate_user, writer, storage, config).await
} else {
@ -242,7 +224,6 @@ impl RegistrationState {
nickname: nickname.clone(),
username,
realname,
enabled_capabilities: self.enabled_capabilities,
};
self.finalize_auth(candidate_user, writer, storage, config).await
} else {
@ -606,18 +587,9 @@ async fn handle_update(
author_id,
room_id,
body,
created_at,
} => {
let mut tags = vec![];
if user.enabled_capabilities.contains(Capabilities::ServerTime) {
let tag = Tag {
key: "time".into(),
value: Some(created_at.to_rfc3339_opts(SecondsFormat::Millis, true).into()),
};
tags.push(tag);
}
ServerMessage {
tags,
tags: vec![],
sender: Some(author_id.as_inner().clone()),
body: ServerMessageBody::PrivateMessage {
target: Recipient::Chan(Chan::Global(room_id.as_inner().clone())),
@ -653,32 +625,6 @@ async fn handle_update(
.await?;
writer.flush().await?
}
Updates::NewDialogMessage {
sender,
receiver,
body,
created_at,
} => {
let mut tags = vec![];
if user.enabled_capabilities.contains(Capabilities::ServerTime) {
let tag = Tag {
key: "time".into(),
value: Some(created_at.to_rfc3339_opts(SecondsFormat::Millis, true).into()),
};
tags.push(tag);
}
ServerMessage {
tags,
sender: Some(sender.as_inner().clone()),
body: ServerMessageBody::PrivateMessage {
target: Recipient::Nick(receiver.as_inner().clone()),
body: body.clone(),
},
}
.write_async(writer)
.await?;
writer.flush().await?
}
}
Ok(())
}
@ -725,10 +671,6 @@ async fn handle_incoming_message(
let room_id = RoomId::from(chan)?;
user_handle.send_message(room_id, body).await?;
}
Recipient::Nick(nick) => {
let receiver = PlayerId::from(nick)?;
user_handle.send_dialog_message(receiver, body).await?;
}
_ => log::warn!("Unsupported target type"),
},
ClientMessage::Topic { chan, topic } => {

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

@ -1,20 +1,17 @@
use std::io::ErrorKind;
use std::net::SocketAddr;
use std::time::Duration;
use anyhow::{anyhow, Result};
use chrono::{DateTime, SecondsFormat};
use prometheus::Registry as MetricsRegistry;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::TcpStream;
use lavina_core::player::{JoinResult, PlayerId, SendMessageResult};
use lavina_core::repo::{Storage, StorageConfig};
use lavina_core::room::RoomId;
use lavina_core::LavinaCore;
use projection_irc::APP_VERSION;
use projection_irc::{launch, read_irc_message, RunningServer, ServerConfig};
struct TestScope<'a> {
reader: BufReader<ReadHalf<'a>>,
writer: WriteHalf<'a>,
@ -92,11 +89,6 @@ impl<'a> TestScope<'a> {
Err(_) => Ok(()),
}
}
async fn expect_cap_ls(&mut self) -> Result<()> {
self.expect(":testserver CAP * LS :sasl=PLAIN server-time").await?;
Ok(())
}
}
struct TestServer {
@ -396,7 +388,7 @@ async fn scenario_cap_full_negotiation() -> Result<()> {
s.send("CAP LS 302").await?;
s.send("NICK tester").await?;
s.send("USER UserName 0 * :Real Name").await?;
s.expect_cap_ls().await?;
s.expect(":testserver CAP * LS :sasl=PLAIN").await?;
s.send("CAP REQ :sasl").await?;
s.expect(":testserver CAP tester ACK :sasl").await?;
s.send("AUTHENTICATE PLAIN").await?;
@ -434,7 +426,7 @@ async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
let mut s = TestScope::new(&mut stream);
s.send("CAP LS 302").await?;
s.expect_cap_ls().await?;
s.expect(":testserver CAP * LS :sasl=PLAIN").await?;
s.send("CAP REQ :sasl").await?;
s.expect(":testserver CAP * ACK :sasl").await?;
s.send("AUTHENTICATE PLAIN").await?;
@ -513,7 +505,7 @@ async fn scenario_cap_sasl_fail() -> Result<()> {
s.send("CAP LS 302").await?;
s.send("NICK tester").await?;
s.send("USER UserName 0 * :Real Name").await?;
s.expect_cap_ls().await?;
s.expect(":testserver CAP * LS :sasl=PLAIN").await?;
s.send("CAP REQ :sasl").await?;
s.expect(":testserver CAP tester ACK :sasl").await?;
s.send("AUTHENTICATE SHA256").await?;
@ -566,72 +558,3 @@ async fn terminate_socket_scenario() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn server_time_capability() -> Result<()> {
let mut server = TestServer::start().await?;
// test scenario
server.storage.create_user("tester").await?;
server.storage.set_password("tester", "password").await?;
let mut stream = TcpStream::connect(server.server.addr).await?;
let mut s = TestScope::new(&mut stream);
s.send("CAP LS 302").await?;
s.send("NICK tester").await?;
s.send("USER UserName 0 * :Real Name").await?;
s.expect_cap_ls().await?;
s.send("CAP REQ :sasl server-time").await?;
s.expect(":testserver CAP tester ACK :sasl server-time").await?;
s.send("AUTHENTICATE PLAIN").await?;
s.expect(":testserver AUTHENTICATE +").await?;
s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
s.expect(":testserver 903 tester :SASL authentication successful").await?;
s.send("CAP END").await?;
s.expect_server_introduction("tester").await?;
s.expect_nothing().await?;
s.send("JOIN #test").await?;
s.expect(":tester JOIN #test").await?;
s.expect(":testserver 332 tester #test :New room").await?;
s.expect(":testserver 353 tester = #test :tester").await?;
s.expect(":testserver 366 tester #test :End of /NAMES list").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");
};
s.expect(":some_guy JOIN #test").await?;
let SendMessageResult::Success(res) = conn.send_message(RoomId::from("test").unwrap(), "Hello".into()).await?
else {
panic!("Failed to send message");
};
s.expect(&format!(
"@time={} :some_guy PRIVMSG #test :Hello",
res.to_rfc3339_opts(SecondsFormat::Millis, true)
))
.await?;
// formatting check
assert_eq!(
DateTime::parse_from_rfc3339(&"2024-01-01T10:00:32.123Z").unwrap().to_rfc3339_opts(SecondsFormat::Millis, true),
"2024-01-01T10:00:32.123Z"
);
s.send("QUIT :Leaving").await?;
s.expect(":testserver ERROR :Leaving the server").await?;
s.expect_eof().await?;
stream.shutdown().await?;
// wrap up
server.server.terminate().await?;
Ok(())
}

4
crates/projection-xmpp/src/message.rs
View File

@ -1,6 +1,5 @@
//! Handling of all client2server message stanzas
use lavina_core::player::PlayerId;
use quick_xml::events::Event;
use lavina_core::prelude::*;
@ -41,9 +40,6 @@ impl<'a> XmppConnection<'a> {
}
.serialize(output);
Ok(())
} else if server.0.as_ref() == &*self.hostname && m.r#type == MessageType::Chat {
self.user_handle.send_dialog_message(PlayerId::from(name.0.clone())?, m.body.clone()).await?;
Ok(())
} else {
todo!()
}

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

@ -17,7 +17,6 @@ impl<'a> XmppConnection<'a> {
room_id,
author_id,
body,
created_at: _,
} => {
Message::<()> {
to: Some(Jid {
@ -39,34 +38,6 @@ impl<'a> XmppConnection<'a> {
}
.serialize(output);
}
Updates::NewDialogMessage {
sender,
receiver,
body,
created_at: _,
} => {
if receiver == self.user.player_id {
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(sender.as_inner().clone())),
server: Server(self.hostname.clone()),
resource: Some(Resource(sender.into_inner())),
}),
id: None,
r#type: MessageType::Chat,
lang: None,
subject: None,
body: body.into(),
custom: vec![],
}
.serialize(output);
}
}
_ => {}
}
Ok(())

15
crates/proto-irc/src/lib.rs
View File

@ -18,19 +18,8 @@ use tokio::io::{AsyncWrite, AsyncWriteExt};
/// Single message tag value.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Tag {
pub key: Str,
pub value: Option<Str>,
}
impl Tag {
pub async fn write_async(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
writer.write_all(self.key.as_bytes()).await?;
if let Some(value) = &self.value {
writer.write_all(b"=").await?;
writer.write_all(value.as_bytes()).await?;
}
Ok(())
}
key: Str,
value: Option<u8>,
}
fn receiver(input: &str) -> IResult<&str, &str> {

7
crates/proto-irc/src/server.rs
View File

@ -19,13 +19,6 @@ pub struct ServerMessage {
impl ServerMessage {
pub async fn write_async(&self, writer: &mut (impl AsyncWrite + Unpin)) -> std::io::Result<()> {
if !self.tags.is_empty() {
for tag in &self.tags {
writer.write_all(b"@").await?;
tag.write_async(writer).await?;
writer.write_all(b" ").await?;
}
}
match &self.sender {
Some(ref sender) => {
writer.write_all(b":").await?;