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

Merge branch 'main' into proto-xmpp-crate

This commit is contained in:
Nikita Vilunov 2023-09-30 17:42:18 +02:00
parent 4c9ab1676e 563811cbca
commit ae2ce37e6b
15 changed files with 369 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
.gitea/workflows/test-pr.yml Normal file
View File

@ -0,0 +1,19 @@
name: check-and-test
on: [push, pull_request]
jobs:
check-and-test:
runs-on: ubuntu-latest
steps:
- name: git checkout repository
uses: actions/checkout@v4
- name: setup rust
uses: https://github.com/actions-rs/toolchain@v1
- name: cargo check
uses: https://github.com/actions-rs/cargo@v1
with:
command: check
- name: test
uses: https://github.com/actions-rs/cargo@v1
with:
command: test
args: --workspace -- --skip projections::irc

115
Cargo.lock generated
View File

@ -44,6 +44,54 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0942ffc6dcaadf03badf6e6a2d0228460359d5e34b57ccdc720b7382dfbd5ec5"
[[package]]
name = "anstream"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f58811cfac344940f1a400b6e6231ce35171f614f26439e80f8c1465c5cc0c"
dependencies = [
"anstyle",
"anstyle-parse",
"anstyle-query",
"anstyle-wincon",
"colorchoice",
"utf8parse",
]
[[package]]
name = "anstyle"
version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b84bf0a05bbb2a83e5eb6fa36bb6e87baa08193c35ff52bbf6b38d8af2890e46"
[[package]]
name = "anstyle-parse"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "938874ff5980b03a87c5524b3ae5b59cf99b1d6bc836848df7bc5ada9643c333"
dependencies = [
"utf8parse",
]
[[package]]
name = "anstyle-query"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ca11d4be1bab0c8bc8734a9aa7bf4ee8316d462a08c6ac5052f888fef5b494b"
dependencies = [
"windows-sys",
]
[[package]]
name = "anstyle-wincon"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "58f54d10c6dfa51283a066ceab3ec1ab78d13fae00aa49243a45e4571fb79dfd"
dependencies = [
"anstyle",
"windows-sys",
]
[[package]]
name = "anyhow"
version = "1.0.75"
@ -161,6 +209,52 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "clap"
version = "4.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1d7b8d5ec32af0fadc644bf1fd509a688c2103b185644bb1e29d164e0703136"
dependencies = [
"clap_builder",
"clap_derive",
]
[[package]]
name = "clap_builder"
version = "4.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5179bb514e4d7c2051749d8fcefa2ed6d06a9f4e6d69faf3805f5d80b8cf8d56"
dependencies = [
"anstream",
"anstyle",
"clap_lex",
"strsim",
]
[[package]]
name = "clap_derive"
version = "4.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0862016ff20d69b84ef8247369fabf5c008a7417002411897d40ee1f4532b873"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn 2.0.29",
]
[[package]]
name = "clap_lex"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cd7cc57abe963c6d3b9d8be5b06ba7c8957a930305ca90304f24ef040aa6f961"
[[package]]
name = "colorchoice"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
[[package]]
name = "const-oid"
version = "0.9.5"
@ -754,11 +848,13 @@ version = "0.0.1-dev"
dependencies = [
"anyhow",
"assert_matches",
"clap",
"derive_more",
"figment",
"futures-util",
"http-body-util",
"hyper 1.0.0-rc.3",
"mgmt-api",
"nonempty",
"prometheus",
"proto-irc",
@ -846,6 +942,13 @@ version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
[[package]]
name = "mgmt-api"
version = "0.0.1-dev"
dependencies = [
"serde",
]
[[package]]
name = "mime"
version = "0.3.17"
@ -1794,6 +1897,12 @@ dependencies = [
"unicode-normalization",
]
[[package]]
name = "strsim"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
[[package]]
name = "subtle"
version = "2.5.0"
@ -2104,6 +2213,12 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "utf8parse"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a"
[[package]]
name = "uuid"
version = "1.4.1"

7
Cargo.toml
View File

@ -3,6 +3,7 @@ members = [
".",
"crates/proto-irc",
"crates/proto-xmpp",
"crates/mgmt-api",
]
[workspace.package]
@ -19,6 +20,8 @@ quick-xml = { version = "0.30.0", features = ["async-tokio"] }
lazy_static = "1.4.0"
regex = "1.7.1"
derive_more = "0.99.17"
clap = { version = "4.4.4", features = ["derive"] }
serde = { version = "1.0.152", features = ["rc", "serde_derive"] }
[package]
name = "lavina"
@ -31,7 +34,7 @@ anyhow.workspace = true
figment = { version = "0.10.8", features = ["env", "toml"] } # configuration files
hyper = { version = "1.0.0-rc.3,<1.0.0-rc.4", features = ["server", "http1"] } # http server
http-body-util = "0.1.0-rc.3"
serde = { version = "1.0.152", features = ["rc", "serde_derive"] }
serde.workspace = true
serde_json = "1.0.93"
tokio.workspace = true
tracing = "0.1.37" # logging & tracing api
@ -47,6 +50,8 @@ uuid = { version = "1.3.0", features = ["v4"] }
sqlx = { version = "0.7.0-alpha.2", features = ["sqlite", "migrate"] }
proto-irc = { path = "crates/proto-irc" }
proto-xmpp = { path = "crates/proto-xmpp" }
mgmt-api = { path = "crates/mgmt-api" }
clap.workspace = true
[dev-dependencies]
assert_matches.workspace = true

8
crates/mgmt-api/Cargo.toml Normal file
View File

@ -0,0 +1,8 @@
[package]
name = "mgmt-api"
edition = "2021"
version.workspace = true
publish = false
[dependencies]
serde.workspace = true

29
crates/mgmt-api/src/lib.rs Normal file
View File

@ -0,0 +1,29 @@
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
pub struct ErrorResponse<'a> {
pub code: &'a str,
pub message: &'a str,
}
#[derive(Serialize, Deserialize)]
pub struct CreatePlayerRequest<'a> {
pub name: &'a str,
}
#[derive(Serialize, Deserialize)]
pub struct ChangePasswordRequest<'a> {
pub player_name: &'a str,
pub password: &'a str,
}
pub mod paths {
pub const CREATE_PLAYER: &'static str = "/mgmt/create_player";
pub const SET_PASSWORD: &'static str = "/mgmt/set_password";
}
pub mod errors {
pub const INVALID_PATH: &'static str = "invalid_path";
pub const MALFORMED_REQUEST: &'static str = "malformed_request";
pub const PLAYER_NOT_FOUND: &'static str = "player_not_found";
}

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

@ -2,6 +2,7 @@
name = "proto-irc"
edition = "2021"
version.workspace = true
publish = false
[dependencies]
nom.workspace = true

2
dist/alpine3.18.Dockerfile vendored
View File

@ -8,4 +8,4 @@ FROM alpine:3.18@sha256:7144f7bab3d4c2648d7e59409f15ec52a18006a128c733fcff20d3a4
COPY --from=bld target/release/lavina /usr/bin/lavina
VOLUME ["/etc/lavina/", "/var/lib/lavina/"]
ENTRYPOINT ["lavina"]
ENTRYPOINT ["lavina", "--config", "/etc/lavina/config.toml"]

1
rustfmt.toml Normal file
View File

@ -0,0 +1 @@
max_width = 120

46
src/core/repo/mod.rs
View File

@ -5,7 +5,7 @@ use std::sync::Arc;
use serde::Deserialize;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::{ConnectOptions, Connection, FromRow, SqliteConnection};
use sqlx::{ConnectOptions, Connection, FromRow, Sqlite, SqliteConnection, Transaction};
use tokio::sync::Mutex;
use crate::prelude::*;
@ -102,6 +102,50 @@ impl Storage {
res.close().await?;
Ok(())
}
pub async fn create_user(&mut self, name: &str) -> Result<()> {
let query = sqlx::query(
"insert into users(name)
values (?);",
)
.bind(name);
let mut executor = self.conn.lock().await;
query.execute(&mut *executor).await?;
Ok(())
}
pub async fn set_password<'a>(&'a mut self, name: &'a str, pwd: &'a str) -> Result<Option<()>> {
async fn inner(txn: &mut Transaction<'_, Sqlite>, name: &str, pwd: &str) -> Result<Option<()>> {
let id: Option<(u32,)> = sqlx::query_as("select * from users where name = ? limit 1;")
.bind(name)
.fetch_optional(&mut **txn)
.await?;
let Some((id,)) = id else {
return Ok(None);
};
sqlx::query("insert or replace into challenges_plain_password(user_id, password) values (?, ?);")
.bind(id)
.bind(pwd)
.execute(&mut **txn)
.await?;
Ok(Some(()))
}
let mut executor = self.conn.lock().await;
let mut tx = executor.begin().await?;
let res = inner(&mut tx, name, pwd).await;
match res {
Ok(e) => {
tx.commit().await?;
Ok(e)
}
Err(e) => {
tx.rollback().await?;
Err(e)
}
}
}
}
#[derive(FromRow)]

16
src/main.rs
View File

@ -11,7 +11,9 @@ mod projections;
mod util;
use std::future::Future;
use std::path::Path;
use clap::Parser;
use figment::providers::Format;
use figment::{providers::Toml, Figment};
use prometheus::Registry as MetricsRegistry;
@ -30,11 +32,15 @@ struct ServerConfig {
storage: core::repo::StorageConfig,
}
#[derive(Parser)]
struct CliArgs {
#[arg(long)]
config: Box<Path>,
}
fn load_config() -> Result<ServerConfig> {
// TODO get config path as a cmd line arg
let raw_config = Figment::new()
.merge(Toml::file("config.toml"))
.merge(Toml::file("/etc/lavina/config.toml"));
let args = CliArgs::parse();
let raw_config = Figment::from(Toml::file(args.config));
let config: ServerConfig = raw_config.extract()?;
Ok(config)
}
@ -58,7 +64,7 @@ async fn main() -> Result<()> {
let rooms = RoomRegistry::new(&mut metrics, storage.clone())?;
let mut players = PlayerRegistry::empty(rooms.clone(), &mut metrics)?;
let telemetry_terminator =
util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone()).await?;
util::telemetry::launch(telemetry_config, metrics.clone(), rooms.clone(), storage.clone()).await?;
let irc = projections::irc::launch(irc_config, players.clone(), rooms.clone(), metrics.clone(), storage.clone()).await?;
let xmpp = projections::xmpp::launch(xmpp_config, players.clone(), rooms.clone(), metrics.clone()).await?;
tracing::info!("Started");

6
src/projections/irc/mod.rs
View File

@ -52,6 +52,7 @@ async fn handle_socket(
termination: Deferred<()>, // TODO use it to stop the connection gracefully
mut storage: Storage,
) -> Result<()> {
log::info!("Received an IRC connection from {socket_addr}");
let (reader, writer) = stream.split();
let mut reader: BufReader<ReadHalf> = BufReader::new(reader);
let mut writer = BufWriter::new(writer);
@ -74,10 +75,13 @@ async fn handle_socket(
match registered_user {
Ok(user) => {
log::debug!("User registered");
handle_registered_socket(config, players, rooms, &mut reader, &mut writer, user)
.await?;
}
Err(_) => {}
Err(_) => {
log::debug!("Registration failed");
}
}
stream.shutdown().await?;

4
src/projections/xmpp/mod.rs
View File

@ -148,7 +148,7 @@ async fn handle_socket(
rooms: RoomRegistry,
termination: Deferred<()>, // TODO use it to stop the connection gracefully
) -> Result<()> {
log::debug!("Received an XMPP connection from {socket_addr}");
log::info!("Received an XMPP connection from {socket_addr}");
let mut reader_buf = vec![];
let (reader, writer) = stream.split();
let mut buf_reader = BufReader::new(reader);
@ -171,6 +171,7 @@ async fn handle_socket(
let mut xml_writer = Writer::new(b);
let authenticated = socket_auth(&mut xml_reader, &mut xml_writer, &mut reader_buf).await?;
log::debug!("User authenticated");
let mut connection = players
.connect_to_player(authenticated.player_id.clone())
.await;
@ -313,7 +314,6 @@ async fn socket_final(
match parser.consume(ns, &event) {
Continuation::Final(res) => {
let res = res?;
dbg!(&res);
let stop = handle_packet(&mut events, res, authenticated, user_handle, rooms).await?;
for i in &events {
xml_writer.write_event_async(i).await?;

10
src/util/http.rs
View File

@ -1,10 +0,0 @@
use std::convert::Infallible;
use http_body_util::Full;
use hyper::{body::Bytes, Response, StatusCode};
pub fn not_found() -> std::result::Result<Response<Full<Bytes>>, Infallible> {
let mut response = Response::new(Full::new(Bytes::from("404")));
*response.status_mut() = StatusCode::NOT_FOUND;
Ok(response)
}

1
src/util/mod.rs
View File

@ -1,6 +1,5 @@
use crate::prelude::*;
pub mod http;
pub mod table;
pub mod telemetry;
#[cfg(test)]

149
src/util/telemetry.rs
View File

@ -6,18 +6,18 @@ use http_body_util::{BodyExt, Full};
use hyper::body::Bytes;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response};
use hyper::{Method, Request, Response, StatusCode};
use prometheus::{Encoder, Registry as MetricsRegistry, TextEncoder};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use crate::core::repo::Storage;
use crate::core::room::RoomRegistry;
use crate::prelude::*;
use crate::util::http::*;
use crate::util::Terminator;
type BoxBody = http_body_util::combinators::BoxBody<Bytes, Infallible>;
use mgmt_api::*;
type HttpResult<T> = std::result::Result<T, Infallible>;
#[derive(Deserialize, Debug)]
@ -29,11 +29,12 @@ pub async fn launch(
config: ServerConfig,
metrics: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
) -> Result<Terminator> {
log::info!("Starting the telemetry service");
let listener = TcpListener::bind(config.listen_on).await?;
log::debug!("Listener started");
let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, rx.map(|_| ())));
let terminator = Terminator::spawn(|rx| main_loop(listener, metrics, rooms, storage, rx.map(|_| ())));
Ok(terminator)
}
@ -41,6 +42,7 @@ async fn main_loop(
listener: TcpListener,
metrics: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
termination: impl Future<Output = ()>,
) -> Result<()> {
pin!(termination);
@ -52,10 +54,12 @@ async fn main_loop(
let (stream, _) = result?;
let metrics = metrics.clone();
let rooms = rooms.clone();
let storage = storage.clone();
tokio::task::spawn(async move {
let registry = metrics.clone();
let rooms = rooms.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), r)));
let storage = storage.clone();
let server = http1::Builder::new().serve_connection(stream, service_fn(move |r| route(registry.clone(), rooms.clone(), storage.clone(), r)));
if let Err(err) = server.await {
tracing::error!("Error serving connection: {:?}", err);
}
@ -70,27 +74,124 @@ async fn main_loop(
async fn route(
registry: MetricsRegistry,
rooms: RoomRegistry,
storage: Storage,
request: Request<hyper::body::Incoming>,
) -> std::result::Result<Response<BoxBody>, Infallible> {
match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => Ok(endpoint_metrics(registry)?.map(BodyExt::boxed)),
(&Method::GET, "/rooms") => Ok(endpoint_rooms(rooms).await?.map(BodyExt::boxed)),
_ => Ok(not_found()?.map(BodyExt::boxed)),
) -> HttpResult<Response<Full<Bytes>>> {
let res = match (request.method(), request.uri().path()) {
(&Method::GET, "/metrics") => endpoint_metrics(registry),
(&Method::GET, "/rooms") => endpoint_rooms(rooms).await,
(&Method::POST, paths::CREATE_PLAYER) => endpoint_create_player(request, storage).await.or5xx(),
(&Method::POST, paths::SET_PASSWORD) => endpoint_set_password(request, storage).await.or5xx(),
_ => not_found(),
};
Ok(res)
}
fn endpoint_metrics(registry: MetricsRegistry) -> Response<Full<Bytes>> {
let mf = registry.gather();
let mut buffer = vec![];
TextEncoder.encode(&mf, &mut buffer).expect("write to vec cannot fail");
Response::new(Full::new(Bytes::from(buffer)))
}
async fn endpoint_rooms(rooms: RoomRegistry) -> Response<Full<Bytes>> {
// TODO introduce management API types independent from core-domain types
// TODO remove `Serialize` implementations from all core-domain types
let room_list = rooms.get_all_rooms().await.to_body();
Response::new(room_list)
}
async fn endpoint_create_player(
request: Request<hyper::body::Incoming>,
mut storage: Storage,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<CreatePlayerRequest>(&str[..]) else {
let payload = ErrorResponse {
code: errors::MALFORMED_REQUEST,
message: "The request payload contains incorrect JSON value",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::BAD_REQUEST;
return Ok(response);
};
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;
Ok(response)
}
async fn endpoint_set_password(
request: Request<hyper::body::Incoming>,
mut storage: Storage,
) -> Result<Response<Full<Bytes>>> {
let str = request.collect().await?.to_bytes();
let Ok(res) = serde_json::from_slice::<ChangePasswordRequest>(&str[..]) else {
let payload = ErrorResponse {
code: errors::MALFORMED_REQUEST,
message: "The request payload contains incorrect JSON value",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::BAD_REQUEST;
return Ok(response);
};
let Some(_) = storage.set_password(&res.player_name, &res.password).await? else {
let payload = ErrorResponse {
code: errors::PLAYER_NOT_FOUND,
message: "No such player exists",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::UNPROCESSABLE_ENTITY;
return Ok(response);
};
log::info!("Password changed for player {}", res.player_name);
let mut response = Response::new(Full::<Bytes>::default());
*response.status_mut() = StatusCode::NO_CONTENT;
Ok(response)
}
pub fn not_found() -> Response<Full<Bytes>> {
let payload = ErrorResponse {
code: errors::INVALID_PATH,
message: "The path does not exist",
}
.to_body();
let mut response = Response::new(payload);
*response.status_mut() = StatusCode::NOT_FOUND;
response
}
trait Or5xx {
fn or5xx(self) -> Response<Full<Bytes>>;
}
impl Or5xx for Result<Response<Full<Bytes>>> {
fn or5xx(self) -> Response<Full<Bytes>> {
match self {
Ok(e) => e,
Err(e) => {
let mut response = Response::new(Full::new(e.to_string().into()));
*response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
response
}
}
}
}
fn endpoint_metrics(registry: MetricsRegistry) -> HttpResult<Response<Full<Bytes>>> {
let mf = registry.gather();
let mut buffer = vec![];
TextEncoder
.encode(&mf, &mut buffer)
.expect("write to vec cannot fail");
Ok(Response::new(Full::new(Bytes::from(buffer))))
trait ToBody {
fn to_body(&self) -> Full<Bytes>;
}
async fn endpoint_rooms(rooms: RoomRegistry) -> HttpResult<Response<Full<Bytes>>> {
let room_list = rooms.get_all_rooms().await;
impl<T> ToBody for T
where
T: Serialize,
{
fn to_body(&self) -> Full<Bytes> {
let mut buffer = vec![];
serde_json::to_writer(&mut buffer, &room_list).expect("unexpected fail when writing to vec");
Ok(Response::new(Full::new(Bytes::from(buffer))))
serde_json::to_writer(&mut buffer, self).expect("unexpected fail when writing to vec");
Full::new(Bytes::from(buffer))
}
}