diff --git a/Cargo.lock b/Cargo.lock index 597643f79e..d89a3ddff2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7217,7 +7217,6 @@ version = "0.15.0" dependencies = [ "atomicwrites", "authc", - "axum", "bincode", "censor", "chrono", @@ -7228,7 +7227,6 @@ dependencies = [ "futures-util", "hashbrown 0.13.2", "humantime", - "hyper", "itertools", "lazy_static", "noise", @@ -7283,12 +7281,16 @@ dependencies = [ name = "veloren-server-cli" version = "0.15.0" dependencies = [ + "axum", "cansi", + "chrono", "clap", "crossterm 0.26.1", + "hyper", "lazy_static", "mimalloc", "num_cpus", + "prometheus", "ron 0.8.1", "serde", "shell-words", diff --git a/server-cli/Cargo.toml b/server-cli/Cargo.toml index 5485670b61..c42ba33618 100644 --- a/server-cli/Cargo.toml +++ b/server-cli/Cargo.toml @@ -47,6 +47,12 @@ tracing = { workspace = true } ron = { workspace = true } serde = { workspace = true, features = [ "rc", "derive" ]} +#HTTP +axum = { version = "0.6.20" } +hyper = "0.14.26" +prometheus = { workspace = true } +chrono = { workspace = true } + [target.'cfg(windows)'.dependencies] mimalloc = "0.1.29" diff --git a/server-cli/src/main.rs b/server-cli/src/main.rs index c269adbb77..13f47e1e99 100644 --- a/server-cli/src/main.rs +++ b/server-cli/src/main.rs @@ -16,6 +16,7 @@ mod settings; mod shutdown_coordinator; mod tui_runner; mod tuilog; +mod web; use crate::{ cli::{Admin, ArgvApp, ArgvCommand, Message, SharedCommand, Shutdown}, shutdown_coordinator::ShutdownCoordinator, @@ -31,6 +32,7 @@ use std::{ sync::{atomic::AtomicBool, mpsc, Arc}, time::{Duration, Instant}, }; +use tokio::sync::Notify; use tracing::{info, trace}; lazy_static::lazy_static! { @@ -185,7 +187,7 @@ fn main() -> io::Result<()> { info!("Starting server..."); let protocols_and_addresses = server_settings.gameserver_protocols.clone(); - let metrics_port = &server_settings.metrics_address.port(); + let web_port = &settings.web_address.port(); // Create server let mut server = Server::new( server_settings, @@ -193,10 +195,27 @@ fn main() -> io::Result<()> { database_settings, &server_data_dir, &|_| {}, - runtime, + Arc::clone(&runtime), ) .expect("Failed to create server instance!"); + let registry = Arc::clone(server.metrics_registry()); + let chat = server.chat_cache().clone(); + let metrics_shutdown = Arc::new(Notify::new()); + let metrics_shutdown_clone = Arc::clone(&metrics_shutdown); + let web_chat_secret = settings.web_chat_secret.clone(); + + runtime.spawn(async move { + web::run( + registry, + chat, + web_chat_secret, + settings.web_address, + metrics_shutdown_clone.notified(), + ) + .await + }); + // Collect addresses that the server is listening to log. let gameserver_addresses = protocols_and_addresses .into_iter() @@ -210,7 +229,7 @@ fn main() -> io::Result<()> { }); info!( - ?metrics_port, + ?web_port, ?gameserver_addresses, "Server is ready to accept connections." ); @@ -243,6 +262,7 @@ fn main() -> io::Result<()> { tick_no += 1; // Terminate the server if instructed to do so by the shutdown coordinator if shutdown_coordinator.check(&mut server, &settings) { + metrics_shutdown.notify_one(); break; } diff --git a/server-cli/src/settings.rs b/server-cli/src/settings.rs index b771a49299..00e30f5354 100644 --- a/server-cli/src/settings.rs +++ b/server-cli/src/settings.rs @@ -1,5 +1,9 @@ use serde::{Deserialize, Serialize}; -use std::{fs, path::PathBuf}; +use std::{ + fs, + net::{Ipv4Addr, SocketAddr}, + path::PathBuf, +}; use tracing::warn; #[derive(Clone, Debug, Serialize, Deserialize)] @@ -7,6 +11,10 @@ use tracing::warn; pub struct Settings { pub update_shutdown_grace_period_secs: u32, pub update_shutdown_message: String, + pub web_address: SocketAddr, + /// SECRET API HEADER used to access the chat api, if disabled the API is + /// unreachable + pub web_chat_secret: Option, } impl Default for Settings { @@ -14,6 +22,8 @@ impl Default for Settings { Self { update_shutdown_grace_period_secs: 120, update_shutdown_message: "The server is restarting for an update".to_owned(), + web_address: SocketAddr::from((Ipv4Addr::LOCALHOST, 14005)), + web_chat_secret: None, } } } diff --git a/server/src/web/chat.rs b/server-cli/src/web/chat.rs similarity index 52% rename from server/src/web/chat.rs rename to server-cli/src/web/chat.rs index f935770a58..81efa3a5f7 100644 --- a/server/src/web/chat.rs +++ b/server-cli/src/web/chat.rs @@ -5,63 +5,16 @@ use axum::{ routing::get, Json, Router, }; -use chrono::{DateTime, Utc}; -use common::comp::ChatMsg; +use chrono::DateTime; use hyper::{Request, StatusCode}; -use serde::{Deserialize, Deserializer, Serialize}; -use std::{ - borrow::Cow, collections::VecDeque, mem::size_of, ops::Sub, str::FromStr, sync::Arc, - time::Duration, -}; -use tokio::sync::Mutex; - -/// The chat cache gets it data from the gameserver and will keep it for some -/// time It will be made available for its consumers, the REST Api -#[derive(Clone)] -pub struct ChatCache { - messages: Arc>>, -} +use serde::{Deserialize, Deserializer}; +use server::chat::ChatCache; +use std::{borrow::Cow, str::FromStr}; /// Keep Size small, so we dont have to Clone much for each request. #[derive(Clone)] struct ChatToken { - secret_token: Cow<'static, str>, -} - -pub struct ChatExporter { - messages: Arc>>, - keep_duration: chrono::Duration, -} - -impl ChatExporter { - pub fn send(&self, msg: ChatMsg) { - let time = Utc::now(); - let drop_older_than = time.sub(self.keep_duration); - let mut messages = self.messages.blocking_lock(); - while let Some(msg) = messages.front() && msg.time < drop_older_than { - messages.pop_front(); - } - messages.push_back(ChatMessage { time, msg }); - const MAX_CACHE_BYTES: usize = 10_000_000; // approx. because HashMap allocates on Heap - if messages.len() * size_of::() > MAX_CACHE_BYTES { - let msg_count = messages.len(); - tracing::debug!(?msg_count, "shrinking cache"); - messages.shrink_to_fit(); - } - } -} - -impl ChatCache { - pub fn new(keep_duration: Duration) -> (Self, ChatExporter) { - let messages: Arc>> = Default::default(); - let messages_clone = Arc::clone(&messages); - let keep_duration = chrono::Duration::from_std(keep_duration).unwrap(); - - (Self { messages }, ChatExporter { - messages: messages_clone, - keep_duration, - }) - } + secret_token: Option>, } async fn validate_secret( @@ -69,22 +22,25 @@ async fn validate_secret( req: Request, next: Next, ) -> Result { + // check if this endpoint is disabled + let secret_token = token.secret_token.ok_or(StatusCode::METHOD_NOT_ALLOWED)?; + pub const X_SECRET_TOKEN: &str = "X-Secret-Token"; let session_cookie = req .headers() .get(X_SECRET_TOKEN) .ok_or(StatusCode::UNAUTHORIZED)?; - if session_cookie.as_bytes() != token.secret_token.as_bytes() { + if session_cookie.as_bytes() != secret_token.as_bytes() { return Err(StatusCode::UNAUTHORIZED); } Ok(next.run(req).await) } -pub fn router(cache: ChatCache, secret_token: String) -> Router { +pub fn router(cache: ChatCache, secret_token: Option) -> Router { let token = ChatToken { - secret_token: Cow::Owned(secret_token), + secret_token: secret_token.map(Cow::Owned), }; Router::new() .route("/history", get(history)) @@ -92,12 +48,6 @@ pub fn router(cache: ChatCache, secret_token: String) -> Router { .with_state(cache) } -#[derive(Debug, Clone, Serialize, Deserialize)] -struct ChatMessage { - time: DateTime, - msg: ChatMsg, -} - #[derive(Debug, Deserialize)] struct Params { #[serde(default, deserialize_with = "empty_string_as_none")] diff --git a/server/src/web/mod.rs b/server-cli/src/web/mod.rs similarity index 93% rename from server/src/web/mod.rs rename to server-cli/src/web/mod.rs index bc3471c3ad..d5f5bfabd3 100644 --- a/server/src/web/mod.rs +++ b/server-cli/src/web/mod.rs @@ -2,15 +2,15 @@ use axum::{extract::State, response::IntoResponse, routing::get, Router}; use core::{future::Future, ops::Deref}; use hyper::{header, http, Body, StatusCode}; use prometheus::{Registry, TextEncoder}; +use server::chat::ChatCache; use std::net::SocketAddr; mod chat; -pub use chat::{ChatCache, ChatExporter}; pub async fn run( registry: R, cache: ChatCache, - chat_token: String, + chat_secret: Option, addr: S, shutdown: F, ) -> Result<(), hyper::Error> @@ -24,7 +24,7 @@ where .with_state(registry.deref().clone()); let app = Router::new() - .nest("/chat/v1", chat::router(cache, chat_token)) + .nest("/chat/v1", chat::router(cache, chat_secret)) .nest("/metrics", metrics) .route("/health", get(|| async {})); diff --git a/server/Cargo.toml b/server/Cargo.toml index 079faeb39a..ff50ae522d 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -64,10 +64,6 @@ enum-map = { workspace = true } noise = { version = "0.7", default-features = false } censor = "0.3" -#HTTP -axum = { version = "0.6.20" } -hyper = "0.14.26" - rusqlite = { version = "0.28.0", features = ["array", "vtab", "bundled", "trace"] } refinery = { version = "0.8.8", features = ["rusqlite"] } diff --git a/server/src/chat.rs b/server/src/chat.rs new file mode 100644 index 0000000000..16fecfbdc4 --- /dev/null +++ b/server/src/chat.rs @@ -0,0 +1,228 @@ +use chrono::{DateTime, Utc}; +use common::{ + comp, + comp::{chat::KillType, ChatType, Content, Group, Player, UnresolvedChatMsg}, + uid::IdMaps, + uuid::Uuid, +}; +use serde::{Deserialize, Serialize}; +use specs::{Join, World, WorldExt}; +use std::{collections::VecDeque, ops::Sub, sync::Arc, time::Duration}; +use tokio::sync::Mutex; + +#[derive(Clone, Serialize, Deserialize)] +pub struct PlayerInfo { + uuid: Uuid, + alias: String, +} + +#[derive(Clone, Serialize, Deserialize)] +pub enum KillSource { + Player(PlayerInfo, KillType), + NonPlayer(String, KillType), + NonExistent(KillType), + Environment(String), + FallDamage, + Suicide, + Other, +} + +#[derive(Clone, Serialize, Deserialize)] +/// partially mapped to common::comp::ChatMsg +pub enum ChatParties { + Online(PlayerInfo), + Offline(PlayerInfo), + CommandInfo(PlayerInfo), + CommandError(PlayerInfo), + Kill(KillSource, PlayerInfo), + GroupMeta(Vec), + Group(PlayerInfo, Vec), + Tell(PlayerInfo, PlayerInfo), + Say(PlayerInfo), + FactionMeta(String), + Faction(PlayerInfo, String), + Region(PlayerInfo), + World(PlayerInfo), +} + +#[derive(Clone, Serialize, Deserialize)] +pub struct ChatMessage { + pub time: DateTime, + pub parties: ChatParties, + pub content: Content, +} + +type MessagesStore = Arc>>; + +/// The chat cache gets it data from the gameserver and will keep it for some +/// time It will be made available for its consumers, the REST Api +#[derive(Clone)] +pub struct ChatCache { + pub messages: MessagesStore, +} + +pub struct ChatExporter { + messages: MessagesStore, + keep_duration: chrono::Duration, +} + +impl ChatMessage { + fn new(chatmsg: &UnresolvedChatMsg, parties: ChatParties) -> Self { + ChatMessage { + time: Utc::now(), + content: chatmsg.content().clone(), + parties, + } + } +} + +impl ChatExporter { + pub fn generate(chatmsg: &UnresolvedChatMsg, ecs: &World) -> Option { + let id_maps = ecs.read_resource::(); + let players = ecs.read_storage::(); + let player_info_from_uid = |uid| { + id_maps + .uid_entity(uid) + .and_then(|entry| players.get(entry)) + .map(|player| PlayerInfo { + alias: player.alias.clone(), + uuid: player.uuid(), + }) + }; + let group_members_from_group = |g| -> Vec<_> { + let groups = ecs.read_storage::(); + (&players, &groups) + .join() + .filter_map(|(player, group)| { + if g == group { + Some(PlayerInfo { + alias: player.alias.clone(), + uuid: player.uuid(), + }) + } else { + None + } + }) + .collect() + }; + + match &chatmsg.chat_type { + ChatType::Offline(from) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new(chatmsg, ChatParties::Offline(player_info))); + } + }, + ChatType::Online(from) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new(chatmsg, ChatParties::Online(player_info))); + } + }, + ChatType::Region(from) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new(chatmsg, ChatParties::Region(player_info))); + } + }, + ChatType::World(from) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new(chatmsg, ChatParties::World(player_info))); + } + }, + ChatType::Say(from) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new(chatmsg, ChatParties::Say(player_info))); + } + }, + ChatType::Tell(from, to) => { + if let (Some(from_player_info), Some(to_player_info)) = + (player_info_from_uid(*from), player_info_from_uid(*to)) + { + return Some(ChatMessage::new( + chatmsg, + ChatParties::Tell(from_player_info, to_player_info), + )); + } + }, + ChatType::Kill(kill_source, from) => { + let kill_source = match kill_source.clone() { + comp::chat::KillSource::Player(uid, t) => { + if let Some(player_info) = player_info_from_uid(uid) { + KillSource::Player(player_info, t) + } else { + return None; + } + }, + comp::chat::KillSource::NonPlayer(str, t) => KillSource::NonPlayer(str, t), + comp::chat::KillSource::NonExistent(t) => KillSource::NonExistent(t), + comp::chat::KillSource::Environment(str) => KillSource::Environment(str), + comp::chat::KillSource::FallDamage => KillSource::FallDamage, + comp::chat::KillSource::Suicide => KillSource::Suicide, + comp::chat::KillSource::Other => KillSource::Other, + }; + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new( + chatmsg, + ChatParties::Kill(kill_source, player_info), + )); + } + }, + ChatType::FactionMeta(s) => { + return Some(ChatMessage::new( + chatmsg, + ChatParties::FactionMeta(s.clone()), + )); + }, + ChatType::Faction(from, s) => { + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new( + chatmsg, + ChatParties::Faction(player_info, s.clone()), + )); + } + }, + ChatType::GroupMeta(g) => { + let members = group_members_from_group(g); + return Some(ChatMessage::new(chatmsg, ChatParties::GroupMeta(members))); + }, + ChatType::Group(from, g) => { + let members = group_members_from_group(g); + if let Some(player_info) = player_info_from_uid(*from) { + return Some(ChatMessage::new( + chatmsg, + ChatParties::Group(player_info, members), + )); + } + }, + _ => (), + }; + + None + } + + pub fn send(&self, msg: ChatMessage) { + let drop_older_than = msg.time.sub(self.keep_duration); + let mut messages = self.messages.blocking_lock(); + while let Some(msg) = messages.front() && msg.time < drop_older_than { + messages.pop_front(); + } + messages.push_back(msg); + const MAX_CACHE_MESSAGES: usize = 10_000; // in case we have a short spam of many many messages, we dont want to keep the capacity forever + if messages.capacity() > messages.len() + MAX_CACHE_MESSAGES { + let msg_count = messages.len(); + tracing::debug!(?msg_count, "shrinking cache"); + messages.shrink_to_fit(); + } + } +} + +impl ChatCache { + pub fn new(keep_duration: Duration) -> (Self, ChatExporter) { + let messages: Arc>> = Default::default(); + let messages_clone = Arc::clone(&messages); + let keep_duration = chrono::Duration::from_std(keep_duration).unwrap(); + + (Self { messages }, ChatExporter { + messages: messages_clone, + keep_duration, + }) + } +} diff --git a/server/src/lib.rs b/server/src/lib.rs index 961e869514..051abe6e08 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -8,6 +8,7 @@ pub mod automod; mod character_creator; +pub mod chat; pub mod chunk_generator; mod chunk_serialize; pub mod client; @@ -31,7 +32,6 @@ pub mod sys; #[cfg(feature = "persistent_world")] pub mod terrain_persistence; #[cfg(not(feature = "worldgen"))] mod test_world; -mod web; mod weather; @@ -104,7 +104,7 @@ use std::{ }; #[cfg(not(feature = "worldgen"))] use test_world::{IndexOwned, World}; -use tokio::{runtime::Runtime, sync::Notify}; +use tokio::runtime::Runtime; use tracing::{debug, error, info, trace, warn}; use vek::*; pub use world::{civ::WorldCivStage, sim::WorldSimStage, WorldGenerateStage}; @@ -124,7 +124,7 @@ use { common_state::plugin::{memory_manager::EcsWorld, PluginMgr}, }; -use crate::{persistence::character_loader::CharacterScreenResponseKind, web::ChatCache}; +use crate::{chat::ChatCache, persistence::character_loader::CharacterScreenResponseKind}; use common::comp::Anchor; #[cfg(feature = "worldgen")] pub use world::{ @@ -211,7 +211,8 @@ pub struct Server { runtime: Arc, - metrics_shutdown: Arc, + metrics_registry: Arc, + chat_cache: ChatCache, database_settings: Arc>, disconnect_all_clients_requested: bool, @@ -483,21 +484,8 @@ impl Server { state.ecs_mut().insert(DeletedEntities::default()); let network = Network::new_with_registry(Pid::new(), &runtime, ®istry); - let metrics_shutdown = Arc::new(Notify::new()); - let metrics_shutdown_clone = Arc::clone(&metrics_shutdown); - let addr = settings.metrics_address; let (chat_cache, chat_tracker) = ChatCache::new(Duration::from_secs(60)); state.ecs_mut().insert(chat_tracker); - runtime.spawn(async move { - web::run( - Arc::clone(®istry), - chat_cache, - "secretpassword".to_string(), - addr, - metrics_shutdown_clone.notified(), - ) - .await - }); let mut printed_quic_warning = false; for protocol in &settings.gameserver_protocols { @@ -603,7 +591,8 @@ impl Server { connection_handler, runtime, - metrics_shutdown, + metrics_registry: registry, + chat_cache, database_settings, disconnect_all_clients_requested: false, @@ -668,6 +657,12 @@ impl Server { /// Get a reference to the server's world. pub fn world(&self) -> &World { &self.world } + /// Get a reference to the Metrics Registry + pub fn metrics_registry(&self) -> &Arc { &self.metrics_registry } + + /// Get a reference to the Chat Cache + pub fn chat_cache(&self) -> &ChatCache { &self.chat_cache } + fn parse_locations(&self, character_list_data: &mut [CharacterItem]) { character_list_data.iter_mut().for_each(|c| { let name = c @@ -1497,8 +1492,6 @@ impl Server { impl Drop for Server { fn drop(&mut self) { - self.metrics_shutdown.notify_one(); - self.state .notify_players(ServerGeneral::Disconnect(DisconnectReason::Shutdown)); diff --git a/server/src/settings.rs b/server/src/settings.rs index caf7a968d9..4756011a66 100644 --- a/server/src/settings.rs +++ b/server/src/settings.rs @@ -160,7 +160,6 @@ impl CalendarMode { #[serde(default)] pub struct Settings { pub gameserver_protocols: Vec, - pub metrics_address: SocketAddr, pub auth_server_address: Option, pub max_players: u16, pub world_seed: u32, @@ -202,7 +201,6 @@ impl Default for Settings { address: SocketAddr::from((Ipv4Addr::UNSPECIFIED, 14004)), }, ], - metrics_address: SocketAddr::from((Ipv4Addr::LOCALHOST, 14005)), auth_server_address: Some("https://auth.veloren.net".into()), world_seed: DEFAULT_WORLD_SEED, server_name: "Veloren Server".into(), @@ -285,10 +283,6 @@ impl Settings { pick_unused_port().expect("Failed to find unused port!"), )), }], - metrics_address: SocketAddr::from(( - Ipv4Addr::LOCALHOST, - pick_unused_port().expect("Failed to find unused port!"), - )), auth_server_address: None, // If loading the default map file, make sure the seed is also default. world_seed: if load.map_file.is_some() { diff --git a/server/src/state_ext.rs b/server/src/state_ext.rs index 1330744a2a..716912fac1 100644 --- a/server/src/state_ext.rs +++ b/server/src/state_ext.rs @@ -1,5 +1,6 @@ use crate::{ automod::AutoMod, + chat::ChatExporter, client::Client, events::{self, update_map_markers}, persistence::PersistedComponents, @@ -8,7 +9,6 @@ use crate::{ rtsim::RtSim, settings::Settings, sys::sentinel::DeletedEntities, - web::ChatExporter, wiring, BattleModeBuffer, SpawnPoint, }; use common::{ @@ -905,6 +905,10 @@ impl StateExt for State { let group_info = msg.get_group().and_then(|g| group_manager.group_info(*g)); + if let Some(exported_message) = ChatExporter::generate(&msg, ecs) { + chat_exporter.send(exported_message); + } + let resolved_msg = msg .clone() .map_group(|_| group_info.map_or_else(|| "???".to_string(), |i| i.name.clone())); @@ -927,7 +931,6 @@ impl StateExt for State { | comp::ChatType::CommandError | comp::ChatType::Meta | comp::ChatType::World(_) => { - chat_exporter.send(resolved_msg.clone()); self.notify_players(ServerGeneral::ChatMsg(resolved_msg)) }, comp::ChatType::Online(u) => { @@ -938,7 +941,6 @@ impl StateExt for State { client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone())); } } - chat_exporter.send(resolved_msg); }, comp::ChatType::Tell(from, to) => { for (client, uid) in @@ -948,12 +950,10 @@ impl StateExt for State { client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone())); } } - chat_exporter.send(resolved_msg); }, comp::ChatType::Kill(kill_source, uid) => { let clients = ecs.read_storage::(); let clients_count = clients.count(); - chat_exporter.send(resolved_msg.clone()); // Avoid chat spam, send kill message only to group or nearby players if a // certain amount of clients are online if clients_count @@ -1008,7 +1008,6 @@ impl StateExt for State { } } } - chat_exporter.send(resolved_msg); }, comp::ChatType::Region(uid) => { let entity_opt = entity_from_uid(*uid); @@ -1021,7 +1020,6 @@ impl StateExt for State { } } } - chat_exporter.send(resolved_msg); }, comp::ChatType::Npc(uid) => { let entity_opt = entity_from_uid(*uid); @@ -1067,7 +1065,6 @@ impl StateExt for State { client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone())); } } - chat_exporter.send(resolved_msg); }, comp::ChatType::Group(from, g) => { if group_info.is_none() { @@ -1089,7 +1086,6 @@ impl StateExt for State { } else { send_to_group(g, ecs, &resolved_msg); } - chat_exporter.send(resolved_msg); }, comp::ChatType::GroupMeta(g) => { send_to_group(g, ecs, &resolved_msg);