TudbuT/conduit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

de-global services from admin

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2024-07-27 00:11:41 +00:00
parent 7a3cc3941e
commit 7e50db4193
37 changed files with 1131 additions and 1127 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
src/admin/admin.rs
View file

@ -3,14 +3,14 @@ use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use crate::{
appservice, appservice::AppserviceCommand, check, check::CheckCommand, debug, debug::DebugCommand, federation,
federation::FederationCommand, media, media::MediaCommand, query, query::QueryCommand, room, room::RoomCommand,
server, server::ServerCommand, user, user::UserCommand,
appservice, appservice::AppserviceCommand, check, check::CheckCommand, command::Command, debug,
debug::DebugCommand, federation, federation::FederationCommand, media, media::MediaCommand, query,
query::QueryCommand, room, room::RoomCommand, server, server::ServerCommand, user, user::UserCommand,
};
#[derive(Debug, Parser)]
#[command(name = "admin", version = env!("CARGO_PKG_VERSION"))]
pub(crate) enum AdminCommand {
pub(super) enum AdminCommand {
#[command(subcommand)]
/// - Commands for managing appservices
Appservices(AppserviceCommand),
@ -49,18 +49,18 @@ pub(crate) enum AdminCommand {
}
#[tracing::instrument(skip_all, name = "command")]
pub(crate) async fn process(command: AdminCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
let reply_message_content = match command {
AdminCommand::Appservices(command) => appservice::process(command, body).await?,
AdminCommand::Media(command) => media::process(command, body).await?,
AdminCommand::Users(command) => user::process(command, body).await?,
AdminCommand::Rooms(command) => room::process(command, body).await?,
AdminCommand::Federation(command) => federation::process(command, body).await?,
AdminCommand::Server(command) => server::process(command, body).await?,
AdminCommand::Debug(command) => debug::process(command, body).await?,
AdminCommand::Query(command) => query::process(command, body).await?,
AdminCommand::Check(command) => check::process(command, body).await?,
};
pub(super) async fn process(command: AdminCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
use AdminCommand::*;
Ok(reply_message_content)
Ok(match command {
Appservices(command) => appservice::process(command, context).await?,
Media(command) => media::process(command, context).await?,
Users(command) => user::process(command, context).await?,
Rooms(command) => room::process(command, context).await?,
Federation(command) => federation::process(command, context).await?,
Server(command) => server::process(command, context).await?,
Debug(command) => debug::process(command, context).await?,
Query(command) => query::process(command, context).await?,
Check(command) => check::process(command, context).await?,
})
}

29
src/admin/appservice/commands.rs
View file

@ -1,18 +1,20 @@
use ruma::{api::appservice::Registration, events::room::message::RoomMessageEventContent};
use crate::{services, Result};
use crate::{admin_command, Result};
pub(super) async fn register(body: Vec<&str>) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn register(&self) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let appservice_config = body[1..body.len().checked_sub(1).unwrap()].join("\n");
let appservice_config = self.body[1..self.body.len().checked_sub(1).unwrap()].join("\n");
let parsed_config = serde_yaml::from_str::<Registration>(&appservice_config);
match parsed_config {
Ok(yaml) => match services().appservice.register_appservice(yaml).await {
Ok(yaml) => match self.services.appservice.register_appservice(yaml).await {
Ok(id) => Ok(RoomMessageEventContent::text_plain(format!(
"Appservice registered with ID: {id}."
))),
@ -26,8 +28,10 @@ pub(super) async fn register(body: Vec<&str>) -> Result<RoomMessageEventContent>
}
}
pub(super) async fn unregister(_body: Vec<&str>, appservice_identifier: String) -> Result<RoomMessageEventContent> {
match services()
#[admin_command]
pub(super) async fn unregister(&self, appservice_identifier: String) -> Result<RoomMessageEventContent> {
match self
.services
.appservice
.unregister_appservice(&appservice_identifier)
.await
@ -39,8 +43,10 @@ pub(super) async fn unregister(_body: Vec<&str>, appservice_identifier: String)
}
}
pub(super) async fn show(_body: Vec<&str>, appservice_identifier: String) -> Result<RoomMessageEventContent> {
match services()
#[admin_command]
pub(super) async fn show_appservice_config(&self, appservice_identifier: String) -> Result<RoomMessageEventContent> {
match self
.services
.appservice
.get_registration(&appservice_identifier)
.await
@ -54,8 +60,9 @@ pub(super) async fn show(_body: Vec<&str>, appservice_identifier: String) -> Res
}
}
pub(super) async fn list(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let appservices = services().appservice.iter_ids().await;
#[admin_command]
pub(super) async fn list_registered(&self) -> Result<RoomMessageEventContent> {
let appservices = self.services.appservice.iter_ids().await;
let output = format!("Appservices ({}): {}", appservices.len(), appservices.join(", "));
Ok(RoomMessageEventContent::text_plain(output))
}

23
src/admin/appservice/mod.rs
View file

@ -2,11 +2,11 @@ mod commands;
use clap::Subcommand;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use self::commands::*;
use crate::admin_command_dispatch;
#[derive(Debug, Subcommand)]
#[admin_command_dispatch]
pub(super) enum AppserviceCommand {
/// - Register an appservice using its registration YAML
///
@ -28,24 +28,13 @@ pub(super) enum AppserviceCommand {
/// - Show an appservice's config using its ID
///
/// You can find the ID using the `list-appservices` command.
Show {
#[clap(alias("show"))]
ShowAppserviceConfig {
/// The appservice to show
appservice_identifier: String,
},
/// - List all the currently registered appservices
List,
}
pub(super) async fn process(command: AppserviceCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
AppserviceCommand::Register => register(body).await?,
AppserviceCommand::Unregister {
appservice_identifier,
} => unregister(body, appservice_identifier).await?,
AppserviceCommand::Show {
appservice_identifier,
} => show(body, appservice_identifier).await?,
AppserviceCommand::List => list(body).await?,
})
#[clap(alias("list"))]
ListRegistered,
}

6
src/admin/check/commands.rs
View file

@ -1,12 +1,14 @@
use conduit::Result;
use conduit_macros::implement;
use ruma::events::room::message::RoomMessageEventContent;
use crate::services;
use crate::{services, Command};
/// Uses the iterator in `src/database/key_value/users.rs` to iterator over
/// every user in our database (remote and local). Reports total count, any
/// errors if there were any, etc
pub(super) async fn check_all_users(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[implement(Command, params = "<'_>")]
pub(super) async fn check_all_users(&self) -> Result<RoomMessageEventContent> {
let timer = tokio::time::Instant::now();
let results = services().users.db.iter();
let query_time = timer.elapsed();

6
src/admin/check/mod.rs
View file

@ -4,15 +4,15 @@ use clap::Subcommand;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use self::commands::*;
use crate::Command;
#[derive(Debug, Subcommand)]
pub(super) enum CheckCommand {
AllUsers,
}
pub(super) async fn process(command: CheckCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
pub(super) async fn process(command: CheckCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
Ok(match command {
CheckCommand::AllUsers => check_all_users(body).await?,
CheckCommand::AllUsers => context.check_all_users().await?,
})
}

6
src/admin/command.rs Normal file
View file

@ -0,0 +1,6 @@
use service::Services;
pub(crate) struct Command<'a> {
pub(crate) services: &'a Services,
pub(crate) body: &'a [&'a str],
}

258
src/admin/debug/commands.rs
View file

@ -16,19 +16,22 @@ use ruma::{
events::room::message::RoomMessageEventContent,
CanonicalJsonObject, EventId, OwnedRoomOrAliasId, RoomId, RoomVersionId, ServerName,
};
use service::services;
use tokio::sync::RwLock;
use tracing_subscriber::EnvFilter;
pub(super) async fn echo(_body: &[&str], message: Vec<String>) -> Result<RoomMessageEventContent> {
use crate::admin_command;
#[admin_command]
pub(super) async fn echo(&self, message: Vec<String>) -> Result<RoomMessageEventContent> {
let message = message.join(" ");
Ok(RoomMessageEventContent::notice_plain(message))
}
pub(super) async fn get_auth_chain(_body: &[&str], event_id: Box<EventId>) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn get_auth_chain(&self, event_id: Box<EventId>) -> Result<RoomMessageEventContent> {
let event_id = Arc::<EventId>::from(event_id);
if let Some(event) = services().rooms.timeline.get_pdu_json(&event_id)? {
if let Some(event) = self.services.rooms.timeline.get_pdu_json(&event_id)? {
let room_id_str = event
.get("room_id")
.and_then(|val| val.as_str())
@ -36,13 +39,16 @@ pub(super) async fn get_auth_chain(_body: &[&str], event_id: Box<EventId>) -> Re
let room_id = <&RoomId>::try_from(room_id_str)
.map_err(|_| Error::bad_database("Invalid room id field in event in database"))?;
let start = Instant::now();
let count = services()
let count = self
.services
.rooms
.auth_chain
.event_ids_iter(room_id, vec![event_id])
.await?
.count();
let elapsed = start.elapsed();
Ok(RoomMessageEventContent::text_plain(format!(
"Loaded auth chain with length {count} in {elapsed:?}"
@ -52,14 +58,16 @@ pub(super) async fn get_auth_chain(_body: &[&str], event_id: Box<EventId>) -> Re
}
}
pub(super) async fn parse_pdu(body: &[&str]) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn parse_pdu(&self) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let string = body[1..body.len().saturating_sub(1)].join("\n");
let string = self.body[1..self.body.len().saturating_sub(1)].join("\n");
match serde_json::from_str(&string) {
Ok(value) => match ruma::signatures::reference_hash(&value, &RoomVersionId::V6) {
Ok(hash) => {
@ -80,15 +88,17 @@ pub(super) async fn parse_pdu(body: &[&str]) -> Result<RoomMessageEventContent>
}
}
pub(super) async fn get_pdu(_body: &[&str], event_id: Box<EventId>) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn get_pdu(&self, event_id: Box<EventId>) -> Result<RoomMessageEventContent> {
let mut outlier = false;
let mut pdu_json = services()
let mut pdu_json = self
.services
.rooms
.timeline
.get_non_outlier_pdu_json(&event_id)?;
if pdu_json.is_none() {
outlier = true;
pdu_json = services().rooms.timeline.get_pdu_json(&event_id)?;
pdu_json = self.services.rooms.timeline.get_pdu_json(&event_id)?;
}
match pdu_json {
Some(json) => {
@ -107,39 +117,42 @@ pub(super) async fn get_pdu(_body: &[&str], event_id: Box<EventId>) -> Result<Ro
}
}
#[admin_command]
pub(super) async fn get_remote_pdu_list(
body: &[&str], server: Box<ServerName>, force: bool,
&self, server: Box<ServerName>, force: bool,
) -> Result<RoomMessageEventContent> {
if !services().globals.config.allow_federation {
if !self.services.globals.config.allow_federation {
return Ok(RoomMessageEventContent::text_plain(
"Federation is disabled on this homeserver.",
));
}
if server == services().globals.server_name() {
if server == self.services.globals.server_name() {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to send federation requests to ourselves. Please use `get-pdu` for fetching local PDUs from \
the database.",
));
}
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let list = body
let list = self
.body
.iter()
.collect::<Vec<_>>()
.drain(1..body.len().saturating_sub(1))
.drain(1..self.body.len().saturating_sub(1))
.filter_map(|pdu| EventId::parse(pdu).ok())
.collect::<Vec<_>>();
for pdu in list {
if force {
if let Err(e) = get_remote_pdu(&[], Box::from(pdu), server.clone()).await {
services()
if let Err(e) = self.get_remote_pdu(Box::from(pdu), server.clone()).await {
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"Failed to get remote PDU, ignoring error: {e}"
@ -148,29 +161,31 @@ pub(super) async fn get_remote_pdu_list(
warn!(%e, "Failed to get remote PDU, ignoring error");
}
} else {
get_remote_pdu(&[], Box::from(pdu), server.clone()).await?;
self.get_remote_pdu(Box::from(pdu), server.clone()).await?;
}
}
Ok(RoomMessageEventContent::text_plain("Fetched list of remote PDUs."))
}
#[admin_command]
pub(super) async fn get_remote_pdu(
_body: &[&str], event_id: Box<EventId>, server: Box<ServerName>,
&self, event_id: Box<EventId>, server: Box<ServerName>,
) -> Result<RoomMessageEventContent> {
if !services().globals.config.allow_federation {
if !self.services.globals.config.allow_federation {
return Ok(RoomMessageEventContent::text_plain(
"Federation is disabled on this homeserver.",
));
}
if server == services().globals.server_name() {
if server == self.services.globals.server_name() {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to send federation requests to ourselves. Please use `get-pdu` for fetching local PDUs.",
));
}
match services()
match self
.services
.sending
.send_federation_request(
&server,
@ -191,7 +206,8 @@ pub(super) async fn get_remote_pdu(
debug!("Attempting to parse PDU: {:?}", &response.pdu);
let parsed_pdu = {
let parsed_result = services()
let parsed_result = self
.services
.rooms
.event_handler
.parse_incoming_pdu(&response.pdu);
@ -212,7 +228,7 @@ pub(super) async fn get_remote_pdu(
let pub_key_map = RwLock::new(BTreeMap::new());
debug!("Attempting to fetch homeserver signing keys for {server}");
services()
self.services
.rooms
.event_handler
.fetch_required_signing_keys(parsed_pdu.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
@ -222,7 +238,7 @@ pub(super) async fn get_remote_pdu(
});
info!("Attempting to handle event ID {event_id} as backfilled PDU");
services()
self.services
.rooms
.timeline
.backfill_pdu(&server, response.pdu, &pub_key_map)
@ -241,9 +257,11 @@ pub(super) async fn get_remote_pdu(
}
}
pub(super) async fn get_room_state(_body: &[&str], room: OwnedRoomOrAliasId) -> Result<RoomMessageEventContent> {
let room_id = services().rooms.alias.resolve(&room).await?;
let room_state = services()
#[admin_command]
pub(super) async fn get_room_state(&self, room: OwnedRoomOrAliasId) -> Result<RoomMessageEventContent> {
let room_id = self.services.rooms.alias.resolve(&room).await?;
let room_state = self
.services
.rooms
.state_accessor
.room_state_full(&room_id)
@ -268,8 +286,9 @@ pub(super) async fn get_room_state(_body: &[&str], room: OwnedRoomOrAliasId) ->
Ok(RoomMessageEventContent::notice_markdown(format!("```json\n{json}\n```")))
}
pub(super) async fn ping(_body: &[&str], server: Box<ServerName>) -> Result<RoomMessageEventContent> {
if server == services().globals.server_name() {
#[admin_command]
pub(super) async fn ping(&self, server: Box<ServerName>) -> Result<RoomMessageEventContent> {
if server == self.services.globals.server_name() {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to send federation requests to ourselves.",
));
@ -277,7 +296,8 @@ pub(super) async fn ping(_body: &[&str], server: Box<ServerName>) -> Result<Room
let timer = tokio::time::Instant::now();
match services()
match self
.services
.sending
.send_federation_request(&server, ruma::api::federation::discovery::get_server_version::v1::Request {})
.await
@ -306,23 +326,23 @@ pub(super) async fn ping(_body: &[&str], server: Box<ServerName>) -> Result<Room
}
}
pub(super) async fn force_device_list_updates(_body: &[&str]) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn force_device_list_updates(&self) -> Result<RoomMessageEventContent> {
// Force E2EE device list updates for all users
for user_id in services().users.iter().filter_map(Result::ok) {
services().users.mark_device_key_update(&user_id)?;
for user_id in self.services.users.iter().filter_map(Result::ok) {
self.services.users.mark_device_key_update(&user_id)?;
}
Ok(RoomMessageEventContent::text_plain(
"Marked all devices for all users as having new keys to update",
))
}
pub(super) async fn change_log_level(
_body: &[&str], filter: Option<String>, reset: bool,
) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn change_log_level(&self, filter: Option<String>, reset: bool) -> Result<RoomMessageEventContent> {
let handles = &["console"];
if reset {
let old_filter_layer = match EnvFilter::try_new(&services().globals.config.log) {
let old_filter_layer = match EnvFilter::try_new(&self.services.globals.config.log) {
Ok(s) => s,
Err(e) => {
return Ok(RoomMessageEventContent::text_plain(format!(
@ -331,7 +351,8 @@ pub(super) async fn change_log_level(
},
};
match services()
match self
.services
.server
.log
.reload
@ -340,7 +361,7 @@ pub(super) async fn change_log_level(
Ok(()) => {
return Ok(RoomMessageEventContent::text_plain(format!(
"Successfully changed log level back to config value {}",
services().globals.config.log
self.services.globals.config.log
)));
},
Err(e) => {
@ -361,7 +382,8 @@ pub(super) async fn change_log_level(
},
};
match services()
match self
.services
.server
.log
.reload
@ -381,19 +403,21 @@ pub(super) async fn change_log_level(
Ok(RoomMessageEventContent::text_plain("No log level was specified."))
}
pub(super) async fn sign_json(body: &[&str]) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn sign_json(&self) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let string = body[1..body.len().checked_sub(1).unwrap()].join("\n");
let string = self.body[1..self.body.len().checked_sub(1).unwrap()].join("\n");
match serde_json::from_str(&string) {
Ok(mut value) => {
ruma::signatures::sign_json(
services().globals.server_name().as_str(),
services().globals.keypair(),
self.services.globals.server_name().as_str(),
self.services.globals.keypair(),
&mut value,
)
.expect("our request json is what ruma expects");
@ -404,19 +428,21 @@ pub(super) async fn sign_json(body: &[&str]) -> Result<RoomMessageEventContent>
}
}
pub(super) async fn verify_json(body: &[&str]) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn verify_json(&self) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let string = body[1..body.len().checked_sub(1).unwrap()].join("\n");
let string = self.body[1..self.body.len().checked_sub(1).unwrap()].join("\n");
match serde_json::from_str(&string) {
Ok(value) => {
let pub_key_map = RwLock::new(BTreeMap::new());
services()
self.services
.rooms
.event_handler
.fetch_required_signing_keys([&value], &pub_key_map)
@ -434,19 +460,22 @@ pub(super) async fn verify_json(body: &[&str]) -> Result<RoomMessageEventContent
}
}
#[tracing::instrument(skip(_body))]
pub(super) async fn first_pdu_in_room(_body: &[&str], room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
if !services()
#[admin_command]
#[tracing::instrument(skip(self))]
pub(super) async fn first_pdu_in_room(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
if !self
.services
.rooms
.state_cache
.server_in_room(&services().globals.config.server_name, &room_id)?
.server_in_room(&self.services.globals.config.server_name, &room_id)?
{
return Ok(RoomMessageEventContent::text_plain(
"We are not participating in the room / we don't know about the room ID.",
));
}
let first_pdu = services()
let first_pdu = self
.services
.rooms
.timeline
.first_pdu_in_room(&room_id)?
@ -455,19 +484,22 @@ pub(super) async fn first_pdu_in_room(_body: &[&str], room_id: Box<RoomId>) -> R
Ok(RoomMessageEventContent::text_plain(format!("{first_pdu:?}")))
}
#[tracing::instrument(skip(_body))]
pub(super) async fn latest_pdu_in_room(_body: &[&str], room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
if !services()
#[admin_command]
#[tracing::instrument(skip(self))]
pub(super) async fn latest_pdu_in_room(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
if !self
.services
.rooms
.state_cache
.server_in_room(&services().globals.config.server_name, &room_id)?
.server_in_room(&self.services.globals.config.server_name, &room_id)?
{
return Ok(RoomMessageEventContent::text_plain(
"We are not participating in the room / we don't know about the room ID.",
));
}
let latest_pdu = services()
let latest_pdu = self
.services
.rooms
.timeline
.latest_pdu_in_room(&room_id)?
@ -476,32 +508,36 @@ pub(super) async fn latest_pdu_in_room(_body: &[&str], room_id: Box<RoomId>) ->
Ok(RoomMessageEventContent::text_plain(format!("{latest_pdu:?}")))
}
#[tracing::instrument(skip(_body))]
#[admin_command]
#[tracing::instrument(skip(self))]
pub(super) async fn force_set_room_state_from_server(
_body: &[&str], room_id: Box<RoomId>, server_name: Box<ServerName>,
&self, room_id: Box<RoomId>, server_name: Box<ServerName>,
) -> Result<RoomMessageEventContent> {
if !services()
if !self
.services
.rooms
.state_cache
.server_in_room(&services().globals.config.server_name, &room_id)?
.server_in_room(&self.services.globals.config.server_name, &room_id)?
{
return Ok(RoomMessageEventContent::text_plain(
"We are not participating in the room / we don't know about the room ID.",
));
}
let first_pdu = services()
let first_pdu = self
.services
.rooms
.timeline
.latest_pdu_in_room(&room_id)?
.ok_or_else(|| Error::bad_database("Failed to find the latest PDU in database"))?;
let room_version = services().rooms.state.get_room_version(&room_id)?;
let room_version = self.services.rooms.state.get_room_version(&room_id)?;
let mut state: HashMap<u64, Arc<EventId>> = HashMap::new();
let pub_key_map = RwLock::new(BTreeMap::new());
let remote_state_response = services()
let remote_state_response = self
.services
.sending
.send_federation_request(
&server_name,
@ -515,7 +551,7 @@ pub(super) async fn force_set_room_state_from_server(
let mut events = Vec::with_capacity(remote_state_response.pdus.len());
for pdu in remote_state_response.pdus.clone() {
events.push(match services().rooms.event_handler.parse_incoming_pdu(&pdu) {
events.push(match self.services.rooms.event_handler.parse_incoming_pdu(&pdu) {
Ok(t) => t,
Err(e) => {
warn!("Could not parse PDU, ignoring: {e}");
@ -525,7 +561,7 @@ pub(super) async fn force_set_room_state_from_server(
}
info!("Fetching required signing keys for all the state events we got");
services()
self.services
.rooms
.event_handler
.fetch_required_signing_keys(events.iter().map(|(_event_id, event, _room_id)| event), &pub_key_map)
@ -535,7 +571,7 @@ pub(super) async fn force_set_room_state_from_server(
for result in remote_state_response
.pdus
.iter()
.map(|pdu| validate_and_add_event_id(services(), pdu, &room_version, &pub_key_map))
.map(|pdu| validate_and_add_event_id(self.services, pdu, &room_version, &pub_key_map))
{
let Ok((event_id, value)) = result.await else {
continue;
@ -546,12 +582,13 @@ pub(super) async fn force_set_room_state_from_server(
Error::BadServerResponse("Invalid PDU in send_join response.")
})?;
services()
self.services
.rooms
.outlier
.add_pdu_outlier(&event_id, &value)?;
if let Some(state_key) = &pdu.state_key {
let shortstatekey = services()
let shortstatekey = self
.services
.rooms
.short
.get_or_create_shortstatekey(&pdu.kind.to_string().into(), state_key)?;
@ -563,32 +600,34 @@ pub(super) async fn force_set_room_state_from_server(
for result in remote_state_response
.auth_chain
.iter()
.map(|pdu| validate_and_add_event_id(services(), pdu, &room_version, &pub_key_map))
.map(|pdu| validate_and_add_event_id(self.services, pdu, &room_version, &pub_key_map))
{
let Ok((event_id, value)) = result.await else {
continue;
};
services()
self.services
.rooms
.outlier
.add_pdu_outlier(&event_id, &value)?;
}
let new_room_state = services()
let new_room_state = self
.services
.rooms
.event_handler
.resolve_state(room_id.clone().as_ref(), &room_version, state)
.await?;
info!("Forcing new room state");
let (short_state_hash, new, removed) = services()
let (short_state_hash, new, removed) = self
.services
.rooms
.state_compressor
.save_state(room_id.clone().as_ref(), new_room_state)?;
let state_lock = services().rooms.state.mutex.lock(&room_id).await;
services()
let state_lock = self.services.rooms.state.mutex.lock(&room_id).await;
self.services
.rooms
.state
.force_state(room_id.clone().as_ref(), short_state_hash, new, removed, &state_lock)
@ -598,7 +637,10 @@ pub(super) async fn force_set_room_state_from_server(
"Updating joined counts for room just in case (e.g. we may have found a difference in the room's \
m.room.member state"
);
services().rooms.state_cache.update_joined_count(&room_id)?;
self.services
.rooms
.state_cache
.update_joined_count(&room_id)?;
drop(state_lock);
@ -607,28 +649,30 @@ pub(super) async fn force_set_room_state_from_server(
))
}
#[admin_command]
pub(super) async fn get_signing_keys(
_body: &[&str], server_name: Option<Box<ServerName>>, _cached: bool,
&self, server_name: Option<Box<ServerName>>, _cached: bool,
) -> Result<RoomMessageEventContent> {
let server_name = server_name.unwrap_or_else(|| services().server.config.server_name.clone().into());
let signing_keys = services().globals.signing_keys_for(&server_name)?;
let server_name = server_name.unwrap_or_else(|| self.services.server.config.server_name.clone().into());
let signing_keys = self.services.globals.signing_keys_for(&server_name)?;
Ok(RoomMessageEventContent::notice_markdown(format!(
"```rs\n{signing_keys:#?}\n```"
)))
}
#[admin_command]
#[allow(dead_code)]
pub(super) async fn get_verify_keys(
_body: &[&str], server_name: Option<Box<ServerName>>, cached: bool,
&self, server_name: Option<Box<ServerName>>, cached: bool,
) -> Result<RoomMessageEventContent> {
let server_name = server_name.unwrap_or_else(|| services().server.config.server_name.clone().into());
let server_name = server_name.unwrap_or_else(|| self.services.server.config.server_name.clone().into());
let mut out = String::new();
if cached {
writeln!(out, "| Key ID | VerifyKey |")?;
writeln!(out, "| --- | --- |")?;
for (key_id, verify_key) in services().globals.verify_keys_for(&server_name)? {
for (key_id, verify_key) in self.services.globals.verify_keys_for(&server_name)? {
writeln!(out, "| {key_id} | {verify_key:?} |")?;
}
@ -636,7 +680,8 @@ pub(super) async fn get_verify_keys(
}
let signature_ids: Vec<String> = Vec::new();
let keys = services()
let keys = self
.services
.rooms
.event_handler
.fetch_signing_keys_for_server(&server_name, signature_ids)
@ -651,16 +696,17 @@ pub(super) async fn get_verify_keys(
Ok(RoomMessageEventContent::notice_markdown(out))
}
#[admin_command]
pub(super) async fn resolve_true_destination(
_body: &[&str], server_name: Box<ServerName>, no_cache: bool,
&self, server_name: Box<ServerName>, no_cache: bool,
) -> Result<RoomMessageEventContent> {
if !services().globals.config.allow_federation {
if !self.services.globals.config.allow_federation {
return Ok(RoomMessageEventContent::text_plain(
"Federation is disabled on this homeserver.",
));
}
if server_name == services().globals.config.server_name {
if server_name == self.services.globals.config.server_name {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to send federation requests to ourselves. Please use `get-pdu` for fetching local PDUs.",
));
@ -672,12 +718,13 @@ pub(super) async fn resolve_true_destination(
&& matches!(data.span_name(), "actual" | "well-known" | "srv")
};
let state = &services().server.log.capture;
let state = &self.services.server.log.capture;
let logs = Arc::new(Mutex::new(String::new()));
let capture = Capture::new(state, Some(filter), capture::fmt_markdown(logs.clone()));
let capture_scope = capture.start();
let actual = services()
let actual = self
.services
.resolver
.resolve_actual_dest(&server_name, !no_cache)
.await?;
@ -692,7 +739,8 @@ pub(super) async fn resolve_true_destination(
Ok(RoomMessageEventContent::text_markdown(msg))
}
pub(super) async fn memory_stats(_body: &[&str]) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn memory_stats(&self) -> Result<RoomMessageEventContent> {
let html_body = conduit::alloc::memory_stats();
if html_body.is_none() {
@ -708,8 +756,9 @@ pub(super) async fn memory_stats(_body: &[&str]) -> Result<RoomMessageEventConte
}
#[cfg(tokio_unstable)]
pub(super) async fn runtime_metrics(_body: &[&str]) -> Result<RoomMessageEventContent> {
let out = services().server.metrics.runtime_metrics().map_or_else(
#[admin_command]
pub(super) async fn runtime_metrics(&self) -> Result<RoomMessageEventContent> {
let out = self.services.server.metrics.runtime_metrics().map_or_else(
|| "Runtime metrics are not available.".to_owned(),
|metrics| format!("```rs\n{metrics:#?}\n```"),
);
@ -718,15 +767,17 @@ pub(super) async fn runtime_metrics(_body: &[&str]) -> Result<RoomMessageEventCo
}
#[cfg(not(tokio_unstable))]
pub(super) async fn runtime_metrics(_body: &[&str]) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn runtime_metrics(&self) -> Result<RoomMessageEventContent> {
Ok(RoomMessageEventContent::text_markdown(
"Runtime metrics require building with `tokio_unstable`.",
))
}
#[cfg(tokio_unstable)]
pub(super) async fn runtime_interval(_body: &[&str]) -> Result<RoomMessageEventContent> {
let out = services().server.metrics.runtime_interval().map_or_else(
#[admin_command]
pub(super) async fn runtime_interval(&self) -> Result<RoomMessageEventContent> {
let out = self.services.server.metrics.runtime_interval().map_or_else(
|| "Runtime metrics are not available.".to_owned(),
|metrics| format!("```rs\n{metrics:#?}\n```"),
);
@ -735,18 +786,21 @@ pub(super) async fn runtime_interval(_body: &[&str]) -> Result<RoomMessageEventC
}
#[cfg(not(tokio_unstable))]
pub(super) async fn runtime_interval(_body: &[&str]) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn runtime_interval(&self) -> Result<RoomMessageEventContent> {
Ok(RoomMessageEventContent::text_markdown(
"Runtime metrics require building with `tokio_unstable`.",
))
}
pub(super) async fn time(_body: &[&str]) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn time(&self) -> Result<RoomMessageEventContent> {
let now = SystemTime::now();
Ok(RoomMessageEventContent::text_markdown(utils::time::format(now, "%+")))
}
pub(super) async fn list_dependencies(_body: &[&str], names: bool) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn list_dependencies(&self, names: bool) -> Result<RoomMessageEventContent> {
if names {
let out = info::cargo::dependencies_names().join(" ");
return Ok(RoomMessageEventContent::notice_markdown(out));

6
src/admin/debug/mod.rs
View file

@ -3,10 +3,10 @@ pub(crate) mod tester;
use clap::Subcommand;
use conduit::Result;
use conduit_macros::admin_command_dispatch;
use ruma::{events::room::message::RoomMessageEventContent, EventId, OwnedRoomOrAliasId, RoomId, ServerName};
use ruma::{EventId, OwnedRoomOrAliasId, RoomId, ServerName};
use self::{commands::*, tester::TesterCommand};
use self::tester::TesterCommand;
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]

18
src/admin/debug/tester.rs
View file

@ -1,33 +1,29 @@
use ruma::events::room::message::RoomMessageEventContent;
use crate::Result;
use crate::{admin_command, admin_command_dispatch, Result};
#[admin_command_dispatch]
#[derive(Debug, clap::Subcommand)]
pub(crate) enum TesterCommand {
Tester,
Timer,
}
pub(super) async fn process(command: TesterCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
match command {
TesterCommand::Tester => tester(body).await,
TesterCommand::Timer => timer(body).await,
}
}
#[inline(never)]
#[rustfmt::skip]
#[allow(unused_variables)]
async fn tester(body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[admin_command]
async fn tester(&self) -> Result<RoomMessageEventContent> {
Ok(RoomMessageEventContent::notice_plain("completed"))
}
#[inline(never)]
#[rustfmt::skip]
async fn timer(body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[admin_command]
async fn timer(&self) -> Result<RoomMessageEventContent> {
let started = std::time::Instant::now();
timed(&body);
timed(self.body);
let elapsed = started.elapsed();
Ok(RoomMessageEventContent::notice_plain(format!("completed in {elapsed:#?}")))

39
src/admin/federation/commands.rs
View file

@ -1,21 +1,26 @@
use std::fmt::Write;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, OwnedRoomId, RoomId, ServerName, UserId};
use crate::{escape_html, get_room_info, services, Result};
use crate::{admin_command, escape_html, get_room_info};
pub(super) async fn disable_room(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
services().rooms.metadata.disable_room(&room_id, true)?;
#[admin_command]
pub(super) async fn disable_room(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
self.services.rooms.metadata.disable_room(&room_id, true)?;
Ok(RoomMessageEventContent::text_plain("Room disabled."))
}
pub(super) async fn enable_room(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
services().rooms.metadata.disable_room(&room_id, false)?;
#[admin_command]
pub(super) async fn enable_room(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
self.services.rooms.metadata.disable_room(&room_id, false)?;
Ok(RoomMessageEventContent::text_plain("Room enabled."))
}
pub(super) async fn incoming_federation(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let map = services()
#[admin_command]
pub(super) async fn incoming_federation(&self) -> Result<RoomMessageEventContent> {
let map = self
.services
.rooms
.event_handler
.federation_handletime
@ -31,10 +36,10 @@ pub(super) async fn incoming_federation(_body: Vec<&str>) -> Result<RoomMessageE
Ok(RoomMessageEventContent::text_plain(&msg))
}
pub(super) async fn fetch_support_well_known(
_body: Vec<&str>, server_name: Box<ServerName>,
) -> Result<RoomMessageEventContent> {
let response = services()
#[admin_command]
pub(super) async fn fetch_support_well_known(&self, server_name: Box<ServerName>) -> Result<RoomMessageEventContent> {
let response = self
.services
.client
.default
.get(format!("https://{server_name}/.well-known/matrix/support"))
@ -72,25 +77,27 @@ pub(super) async fn fetch_support_well_known(
)))
}
pub(super) async fn remote_user_in_rooms(_body: Vec<&str>, user_id: Box<UserId>) -> Result<RoomMessageEventContent> {
if user_id.server_name() == services().globals.config.server_name {
#[admin_command]
pub(super) async fn remote_user_in_rooms(&self, user_id: Box<UserId>) -> Result<RoomMessageEventContent> {
if user_id.server_name() == self.services.globals.config.server_name {
return Ok(RoomMessageEventContent::text_plain(
"User belongs to our server, please use `list-joined-rooms` user admin command instead.",
));
}
if !services().users.exists(&user_id)? {
if !self.services.users.exists(&user_id)? {
return Ok(RoomMessageEventContent::text_plain(
"Remote user does not exist in our database.",
));
}
let mut rooms: Vec<(OwnedRoomId, u64, String)> = services()
let mut rooms: Vec<(OwnedRoomId, u64, String)> = self
.services
.rooms
.state_cache
.rooms_joined(&user_id)
.filter_map(Result::ok)
.map(|room_id| get_room_info(services(), &room_id))
.map(|room_id| get_room_info(self.services, &room_id))
.collect();
if rooms.is_empty() {

23
src/admin/federation/mod.rs
View file

@ -2,10 +2,11 @@ mod commands;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomId, ServerName, UserId};
use ruma::{RoomId, ServerName, UserId};
use self::commands::*;
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(super) enum FederationCommand {
/// - List all rooms we are currently handling an incoming pdu from
@ -39,21 +40,3 @@ pub(super) enum FederationCommand {
user_id: Box<UserId>,
},
}
pub(super) async fn process(command: FederationCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
FederationCommand::DisableRoom {
room_id,
} => disable_room(body, room_id).await?,
FederationCommand::EnableRoom {
room_id,
} => enable_room(body, room_id).await?,
FederationCommand::IncomingFederation => incoming_federation(body).await?,
FederationCommand::FetchSupportWellKnown {
server_name,
} => fetch_support_well_known(body, server_name).await?,
FederationCommand::RemoteUserInRooms {
user_id,
} => remote_user_in_rooms(body, user_id).await?,
})
}

23
src/admin/handler.rs
View file

@ -15,10 +15,10 @@ use service::{
Services,
};
use crate::{admin, admin::AdminCommand};
use crate::{admin, admin::AdminCommand, Command};
struct Handler<'a> {
services: &'a Services,
struct Handler {
services: &'static Services,
}
#[must_use]
@ -68,13 +68,12 @@ fn reply(mut content: RoomMessageEventContent, reply_id: Option<OwnedEventId>) -
Some(content)
}
impl Handler<'_> {
impl Handler {
// Parse and process a message from the admin room
async fn process(&self, msg: &str) -> CommandOutput {
let mut lines = msg.lines().filter(|l| !l.trim().is_empty());
let command = lines.next().expect("each string has at least one line");
let body = lines.collect::<Vec<_>>();
let parsed = match self.parse_command(command) {
let (parsed, body) = match self.parse_command(command) {
Ok(parsed) => parsed,
Err(error) => {
let server_name = self.services.globals.server_name();
@ -84,7 +83,12 @@ impl Handler<'_> {
};
let timer = Instant::now();
let result = Box::pin(admin::process(parsed, body)).await;
let body: Vec<&str> = body.iter().map(String::as_str).collect();
let context = Command {
services: self.services,
body: &body,
};
let result = Box::pin(admin::process(parsed, &context)).await;
let elapsed = timer.elapsed();
conduit::debug!(?command, ok = result.is_ok(), "command processed in {elapsed:?}");
match result {
@ -96,9 +100,10 @@ impl Handler<'_> {
}
// Parse chat messages from the admin room into an AdminCommand object
fn parse_command(&self, command_line: &str) -> Result<AdminCommand, String> {
fn parse_command(&self, command_line: &str) -> Result<(AdminCommand, Vec<String>), String> {
let argv = self.parse_line(command_line);
AdminCommand::try_parse_from(argv).map_err(|error| error.to_string())
let com = AdminCommand::try_parse_from(&argv).map_err(|error| error.to_string())?;
Ok((com, argv))
}
fn complete_command(&self, mut cmd: clap::Command, line: &str) -> String {

37
src/admin/media/commands.rs
View file

@ -1,11 +1,11 @@
use conduit::Result;
use conduit::{debug, info, Result};
use ruma::{events::room::message::RoomMessageEventContent, EventId, MxcUri};
use tracing::{debug, info};
use crate::services;
use crate::admin_command;
#[admin_command]
pub(super) async fn delete(
_body: Vec<&str>, mxc: Option<Box<MxcUri>>, event_id: Option<Box<EventId>>,
&self, mxc: Option<Box<MxcUri>>, event_id: Option<Box<EventId>>,
) -> Result<RoomMessageEventContent> {
if event_id.is_some() && mxc.is_some() {
return Ok(RoomMessageEventContent::text_plain(
@ -15,7 +15,7 @@ pub(super) async fn delete(
if let Some(mxc) = mxc {
debug!("Got MXC URL: {mxc}");
services().media.delete(mxc.as_ref()).await?;
self.services.media.delete(mxc.as_ref()).await?;
return Ok(RoomMessageEventContent::text_plain(
"Deleted the MXC from our database and on our filesystem.",
@ -27,7 +27,7 @@ pub(super) async fn delete(
let mut mxc_deletion_count: usize = 0;
// parsing the PDU for any MXC URLs begins here
if let Some(event_json) = services().rooms.timeline.get_pdu_json(&event_id)? {
if let Some(event_json) = self.services.rooms.timeline.get_pdu_json(&event_id)? {
if let Some(content_key) = event_json.get("content") {
debug!("Event ID has \"content\".");
let content_obj = content_key.as_object();
@ -123,7 +123,7 @@ pub(super) async fn delete(
}
for mxc_url in mxc_urls {
services().media.delete(&mxc_url).await?;
self.services.media.delete(&mxc_url).await?;
mxc_deletion_count = mxc_deletion_count.saturating_add(1);
}
@ -138,23 +138,26 @@ pub(super) async fn delete(
))
}
pub(super) async fn delete_list(body: Vec<&str>) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn delete_list(&self) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let mxc_list = body
.clone()
.drain(1..body.len().checked_sub(1).unwrap())
let mxc_list = self
.body
.to_vec()
.drain(1..self.body.len().checked_sub(1).unwrap())
.collect::<Vec<_>>();
let mut mxc_deletion_count: usize = 0;
for mxc in mxc_list {
debug!("Deleting MXC {mxc} in bulk");
services().media.delete(mxc).await?;
self.services.media.delete(mxc).await?;
mxc_deletion_count = mxc_deletion_count
.checked_add(1)
.expect("mxc_deletion_count should not get this high");
@ -165,10 +168,10 @@ pub(super) async fn delete_list(body: Vec<&str>) -> Result<RoomMessageEventConte
)))
}
pub(super) async fn delete_past_remote_media(
_body: Vec<&str>, duration: String, force: bool,
) -> Result<RoomMessageEventContent> {
let deleted_count = services()
#[admin_command]
pub(super) async fn delete_past_remote_media(&self, duration: String, force: bool) -> Result<RoomMessageEventContent> {
let deleted_count = self
.services
.media
.delete_all_remote_media_at_after_time(duration, force)
.await?;

19
src/admin/media/mod.rs
View file

@ -2,10 +2,11 @@ mod commands;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, EventId, MxcUri};
use ruma::{EventId, MxcUri};
use self::commands::*;
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(super) enum MediaCommand {
/// - Deletes a single media file from our database and on the filesystem
@ -36,17 +37,3 @@ pub(super) enum MediaCommand {
force: bool,
},
}
pub(super) async fn process(command: MediaCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
MediaCommand::Delete {
mxc,
event_id,
} => delete(body, mxc, event_id).await?,
MediaCommand::DeleteList => delete_list(body).await?,
MediaCommand::DeletePastRemoteMedia {
duration,
force,
} => delete_past_remote_media(body, duration, force).await?,
})
}

7
src/admin/mod.rs
View file

@ -3,6 +3,7 @@
#![allow(clippy::enum_glob_use)]
pub(crate) mod admin;
pub(crate) mod command;
pub(crate) mod handler;
mod tests;
pub(crate) mod utils;
@ -22,9 +23,13 @@ extern crate conduit_core as conduit;
extern crate conduit_service as service;
pub(crate) use conduit::Result;
pub(crate) use conduit_macros::{admin_command, admin_command_dispatch};
pub(crate) use service::services;
pub(crate) use crate::utils::{escape_html, get_room_info};
pub(crate) use crate::{
command::Command,
utils::{escape_html, get_room_info},
};
pub(crate) const PAGE_SIZE: usize = 100;

46
src/admin/query/account_data.rs
View file

@ -1,18 +1,48 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{
events::{room::message::RoomMessageEventContent, RoomAccountDataEventType},
RoomId, UserId,
};
use super::AccountData;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/account_data.rs
pub(crate) enum AccountDataCommand {
/// - Returns all changes to the account data that happened after `since`.
ChangesSince {
/// Full user ID
user_id: Box<UserId>,
/// UNIX timestamp since (u64)
since: u64,
/// Optional room ID of the account data
room_id: Option<Box<RoomId>>,
},
/// - Searches the account data for a specific kind.
Get {
/// Full user ID
user_id: Box<UserId>,
/// Account data event type
kind: RoomAccountDataEventType,
/// Optional room ID of the account data
room_id: Option<Box<RoomId>>,
},
}
/// All the getters and iterators from src/database/key_value/account_data.rs
pub(super) async fn account_data(subcommand: AccountData) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: AccountDataCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
AccountData::ChangesSince {
AccountDataCommand::ChangesSince {
user_id,
since,
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
let results = services
.account_data
.changes_since(room_id.as_deref(), &user_id, since)?;
let query_time = timer.elapsed();
@ -21,13 +51,13 @@ pub(super) async fn account_data(subcommand: AccountData) -> Result<RoomMessageE
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
AccountData::Get {
AccountDataCommand::Get {
user_id,
kind,
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
let results = services
.account_data
.get(room_id.as_deref(), &user_id, kind)?;
let query_time = timer.elapsed();

30
src/admin/query/appservice.rs
View file

@ -1,16 +1,32 @@
use clap::Subcommand;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use super::Appservice;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/appservice.rs
pub(crate) enum AppserviceCommand {
/// - Gets the appservice registration info/details from the ID as a string
GetRegistration {
/// Appservice registration ID
appservice_id: Box<str>,
},
/// - Gets all appservice registrations with their ID and registration info
All,
}
/// All the getters and iterators from src/database/key_value/appservice.rs
pub(super) async fn appservice(subcommand: Appservice) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: AppserviceCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
Appservice::GetRegistration {
AppserviceCommand::GetRegistration {
appservice_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
let results = services
.appservice
.db
.get_registration(appservice_id.as_ref());
@ -20,9 +36,9 @@ pub(super) async fn appservice(subcommand: Appservice) -> Result<RoomMessageEven
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Appservice::All => {
AppserviceCommand::All => {
let timer = tokio::time::Instant::now();
let results = services().appservice.all();
let results = services.appservice.all();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(

49
src/admin/query/globals.rs
View file

@ -1,52 +1,73 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, ServerName};
use super::Globals;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/globals.rs
pub(crate) enum GlobalsCommand {
DatabaseVersion,
CurrentCount,
LastCheckForUpdatesId,
LoadKeypair,
/// - This returns an empty `Ok(BTreeMap<..>)` when there are no keys found
/// for the server.
SigningKeysFor {
origin: Box<ServerName>,
},
}
/// All the getters and iterators from src/database/key_value/globals.rs
pub(super) async fn globals(subcommand: Globals) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: GlobalsCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
Globals::DatabaseVersion => {
GlobalsCommand::DatabaseVersion => {
let timer = tokio::time::Instant::now();
let results = services().globals.db.database_version();
let results = services.globals.db.database_version();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Globals::CurrentCount => {
GlobalsCommand::CurrentCount => {
let timer = tokio::time::Instant::now();
let results = services().globals.db.current_count();
let results = services.globals.db.current_count();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Globals::LastCheckForUpdatesId => {
GlobalsCommand::LastCheckForUpdatesId => {
let timer = tokio::time::Instant::now();
let results = services().updates.last_check_for_updates_id();
let results = services.updates.last_check_for_updates_id();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Globals::LoadKeypair => {
GlobalsCommand::LoadKeypair => {
let timer = tokio::time::Instant::now();
let results = services().globals.db.load_keypair();
let results = services.globals.db.load_keypair();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Globals::SigningKeysFor {
GlobalsCommand::SigningKeysFor {
origin,
} => {
let timer = tokio::time::Instant::now();
let results = services().globals.db.verify_keys_for(&origin);
let results = services.globals.db.verify_keys_for(&origin);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(

281
src/admin/query/mod.rs
View file

@ -10,304 +10,51 @@ mod users;
use clap::Subcommand;
use conduit::Result;
use room_state_cache::room_state_cache;
use ruma::{
events::{room::message::RoomMessageEventContent, RoomAccountDataEventType},
OwnedServerName, RoomAliasId, RoomId, ServerName, UserId,
};
use self::{
account_data::account_data, appservice::appservice, globals::globals, presence::presence, resolver::resolver,
room_alias::room_alias, sending::sending, users::users,
account_data::AccountDataCommand, appservice::AppserviceCommand, globals::GlobalsCommand,
presence::PresenceCommand, resolver::ResolverCommand, room_alias::RoomAliasCommand,
room_state_cache::RoomStateCacheCommand, sending::SendingCommand, users::UsersCommand,
};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
/// Query tables from database
pub(super) enum QueryCommand {
/// - account_data.rs iterators and getters
#[command(subcommand)]
AccountData(AccountData),
AccountData(AccountDataCommand),
/// - appservice.rs iterators and getters
#[command(subcommand)]
Appservice(Appservice),
Appservice(AppserviceCommand),
/// - presence.rs iterators and getters
#[command(subcommand)]
Presence(Presence),
Presence(PresenceCommand),
/// - rooms/alias.rs iterators and getters
#[command(subcommand)]
RoomAlias(RoomAlias),
RoomAlias(RoomAliasCommand),
/// - rooms/state_cache iterators and getters
#[command(subcommand)]
RoomStateCache(RoomStateCache),
RoomStateCache(RoomStateCacheCommand),
/// - globals.rs iterators and getters
#[command(subcommand)]
Globals(Globals),
Globals(GlobalsCommand),
/// - sending.rs iterators and getters
#[command(subcommand)]
Sending(Sending),
Sending(SendingCommand),
/// - users.rs iterators and getters
#[command(subcommand)]
Users(Users),
Users(UsersCommand),
/// - resolver service
#[command(subcommand)]
Resolver(Resolver),
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/account_data.rs
pub(super) enum AccountData {
/// - Returns all changes to the account data that happened after `since`.
ChangesSince {
/// Full user ID
user_id: Box<UserId>,
/// UNIX timestamp since (u64)
since: u64,
/// Optional room ID of the account data
room_id: Option<Box<RoomId>>,
},
/// - Searches the account data for a specific kind.
Get {
/// Full user ID
user_id: Box<UserId>,
/// Account data event type
kind: RoomAccountDataEventType,
/// Optional room ID of the account data
room_id: Option<Box<RoomId>>,
},
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/appservice.rs
pub(super) enum Appservice {
/// - Gets the appservice registration info/details from the ID as a string
GetRegistration {
/// Appservice registration ID
appservice_id: Box<str>,
},
/// - Gets all appservice registrations with their ID and registration info
All,
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/presence.rs
pub(super) enum Presence {
/// - Returns the latest presence event for the given user.
GetPresence {
/// Full user ID
user_id: Box<UserId>,
},
/// - Iterator of the most recent presence updates that happened after the
/// event with id `since`.
PresenceSince {
/// UNIX timestamp since (u64)
since: u64,
},
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/rooms/alias.rs
pub(super) enum RoomAlias {
ResolveLocalAlias {
/// Full room alias
alias: Box<RoomAliasId>,
},
/// - Iterator of all our local room aliases for the room ID
LocalAliasesForRoom {
/// Full room ID
room_id: Box<RoomId>,
},
/// - Iterator of all our local aliases in our database with their room IDs
AllLocalAliases,
}
#[derive(Debug, Subcommand)]
pub(super) enum RoomStateCache {
ServerInRoom {
server: Box<ServerName>,
room_id: Box<RoomId>,
},
RoomServers {
room_id: Box<RoomId>,
},
ServerRooms {
server: Box<ServerName>,
},
RoomMembers {
room_id: Box<RoomId>,
},
LocalUsersInRoom {
room_id: Box<RoomId>,
},
ActiveLocalUsersInRoom {
room_id: Box<RoomId>,
},
RoomJoinedCount {
room_id: Box<RoomId>,
},
RoomInvitedCount {
room_id: Box<RoomId>,
},
RoomUserOnceJoined {
room_id: Box<RoomId>,
},
RoomMembersInvited {
room_id: Box<RoomId>,
},
GetInviteCount {
room_id: Box<RoomId>,
user_id: Box<UserId>,
},
GetLeftCount {
room_id: Box<RoomId>,
user_id: Box<UserId>,
},
RoomsJoined {
user_id: Box<UserId>,
},
RoomsLeft {
user_id: Box<UserId>,
},
RoomsInvited {
user_id: Box<UserId>,
},
InviteState {
user_id: Box<UserId>,
room_id: Box<RoomId>,
},
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/globals.rs
pub(super) enum Globals {
DatabaseVersion,
CurrentCount,
LastCheckForUpdatesId,
LoadKeypair,
/// - This returns an empty `Ok(BTreeMap<..>)` when there are no keys found
/// for the server.
SigningKeysFor {
origin: Box<ServerName>,
},
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/sending.rs
pub(super) enum Sending {
/// - Queries database for all `servercurrentevent_data`
ActiveRequests,
/// - Queries database for `servercurrentevent_data` but for a specific
/// destination
///
/// This command takes only *one* format of these arguments:
///
/// appservice_id
/// server_name
/// user_id AND push_key
///
/// See src/service/sending/mod.rs for the definition of the `Destination`
/// enum
ActiveRequestsFor {
#[arg(short, long)]
appservice_id: Option<String>,
#[arg(short, long)]
server_name: Option<Box<ServerName>>,
#[arg(short, long)]
user_id: Option<Box<UserId>>,
#[arg(short, long)]
push_key: Option<String>,
},
/// - Queries database for `servernameevent_data` which are the queued up
/// requests that will eventually be sent
///
/// This command takes only *one* format of these arguments:
///
/// appservice_id
/// server_name
/// user_id AND push_key
///
/// See src/service/sending/mod.rs for the definition of the `Destination`
/// enum
QueuedRequests {
#[arg(short, long)]
appservice_id: Option<String>,
#[arg(short, long)]
server_name: Option<Box<ServerName>>,
#[arg(short, long)]
user_id: Option<Box<UserId>>,
#[arg(short, long)]
push_key: Option<String>,
},
GetLatestEduCount {
server_name: Box<ServerName>,
},
}
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/users.rs
pub(super) enum Users {
Iter,
}
#[derive(Debug, Subcommand)]
/// Resolver service and caches
pub(super) enum Resolver {
/// Query the destinations cache
DestinationsCache {
server_name: Option<OwnedServerName>,
},
/// Query the overrides cache
OverridesCache {
name: Option<String>,
},
}
/// Processes admin query commands
pub(super) async fn process(command: QueryCommand, _body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
QueryCommand::AccountData(command) => account_data(command).await?,
QueryCommand::Appservice(command) => appservice(command).await?,
QueryCommand::Presence(command) => presence(command).await?,
QueryCommand::RoomAlias(command) => room_alias(command).await?,
QueryCommand::RoomStateCache(command) => room_state_cache(command).await?,
QueryCommand::Globals(command) => globals(command).await?,
QueryCommand::Sending(command) => sending(command).await?,
QueryCommand::Users(command) => users(command).await?,
QueryCommand::Resolver(command) => resolver(command).await?,
})
Resolver(ResolverCommand),
}

36
src/admin/query/presence.rs
View file

@ -1,27 +1,47 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, UserId};
use super::Presence;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/presence.rs
pub(crate) enum PresenceCommand {
/// - Returns the latest presence event for the given user.
GetPresence {
/// Full user ID
user_id: Box<UserId>,
},
/// - Iterator of the most recent presence updates that happened after the
/// event with id `since`.
PresenceSince {
/// UNIX timestamp since (u64)
since: u64,
},
}
/// All the getters and iterators in key_value/presence.rs
pub(super) async fn presence(subcommand: Presence) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: PresenceCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
Presence::GetPresence {
PresenceCommand::GetPresence {
user_id,
} => {
let timer = tokio::time::Instant::now();
let results = services().presence.db.get_presence(&user_id)?;
let results = services.presence.db.get_presence(&user_id)?;
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
Presence::PresenceSince {
PresenceCommand::PresenceSince {
since,
} => {
let timer = tokio::time::Instant::now();
let results = services().presence.db.presence_since(since);
let results = services.presence.db.presence_since(since);
let presence_since: Vec<(_, _, _)> = results.collect();
let query_time = timer.elapsed();

44
src/admin/query/resolver.rs
View file

@ -1,24 +1,28 @@
use std::fmt::Write;
use clap::Subcommand;
use conduit::{utils::time, Result};
use ruma::{events::room::message::RoomMessageEventContent, OwnedServerName};
use super::Resolver;
use crate::services;
use crate::{admin_command, admin_command_dispatch};
/// All the getters and iterators in key_value/users.rs
pub(super) async fn resolver(subcommand: Resolver) -> Result<RoomMessageEventContent> {
match subcommand {
Resolver::DestinationsCache {
server_name,
} => destinations_cache(server_name).await,
Resolver::OverridesCache {
name,
} => overrides_cache(name).await,
}
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
/// Resolver service and caches
pub(crate) enum ResolverCommand {
/// Query the destinations cache
DestinationsCache {
server_name: Option<OwnedServerName>,
},
/// Query the overrides cache
OverridesCache {
name: Option<String>,
},
}
async fn destinations_cache(server_name: Option<OwnedServerName>) -> Result<RoomMessageEventContent> {
#[admin_command]
async fn destinations_cache(&self, server_name: Option<OwnedServerName>) -> Result<RoomMessageEventContent> {
use service::resolver::cache::CachedDest;
let mut out = String::new();
@ -36,7 +40,8 @@ async fn destinations_cache(server_name: Option<OwnedServerName>) -> Result<Room
writeln!(out, "| {name} | {dest} | {host} | {expire} |").expect("wrote line");
};
let map = services()
let map = self
.services
.resolver
.cache
.destinations
@ -52,7 +57,8 @@ async fn destinations_cache(server_name: Option<OwnedServerName>) -> Result<Room
Ok(RoomMessageEventContent::notice_markdown(out))
}
async fn overrides_cache(server_name: Option<String>) -> Result<RoomMessageEventContent> {
#[admin_command]
async fn overrides_cache(&self, server_name: Option<String>) -> Result<RoomMessageEventContent> {
use service::resolver::cache::CachedOverride;
let mut out = String::new();
@ -70,7 +76,13 @@ async fn overrides_cache(server_name: Option<String>) -> Result<RoomMessageEvent
writeln!(out, "| {name} | {ips:?} | {port} | {expire} |").expect("wrote line");
};
let map = services().resolver.cache.overrides.read().expect("locked");
let map = self
.services
.resolver
.cache
.overrides
.read()
.expect("locked");
if let Some(server_name) = server_name.as_ref() {
map.get_key_value(server_name).map(row);

41
src/admin/query/room_alias.rs
View file

@ -1,27 +1,48 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomAliasId, RoomId};
use super::RoomAlias;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/rooms/alias.rs
pub(crate) enum RoomAliasCommand {
ResolveLocalAlias {
/// Full room alias
alias: Box<RoomAliasId>,
},
/// - Iterator of all our local room aliases for the room ID
LocalAliasesForRoom {
/// Full room ID
room_id: Box<RoomId>,
},
/// - Iterator of all our local aliases in our database with their room IDs
AllLocalAliases,
}
/// All the getters and iterators in src/database/key_value/rooms/alias.rs
pub(super) async fn room_alias(subcommand: RoomAlias) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: RoomAliasCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
RoomAlias::ResolveLocalAlias {
RoomAliasCommand::ResolveLocalAlias {
alias,
} => {
let timer = tokio::time::Instant::now();
let results = services().rooms.alias.resolve_local_alias(&alias);
let results = services.rooms.alias.resolve_local_alias(&alias);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomAlias::LocalAliasesForRoom {
RoomAliasCommand::LocalAliasesForRoom {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services().rooms.alias.local_aliases_for_room(&room_id);
let results = services.rooms.alias.local_aliases_for_room(&room_id);
let aliases: Vec<_> = results.collect();
let query_time = timer.elapsed();
@ -29,9 +50,9 @@ pub(super) async fn room_alias(subcommand: RoomAlias) -> Result<RoomMessageEvent
"Query completed in {query_time:?}:\n\n```rs\n{aliases:#?}\n```"
)))
},
RoomAlias::AllLocalAliases => {
RoomAliasCommand::AllLocalAliases => {
let timer = tokio::time::Instant::now();
let results = services().rooms.alias.all_local_aliases();
let results = services.rooms.alias.all_local_aliases();
let aliases: Vec<_> = results.collect();
let query_time = timer.elapsed();

170
src/admin/query/room_state_cache.rs
View file

@ -1,71 +1,136 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomId, ServerName, UserId};
use super::RoomStateCache;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
pub(crate) enum RoomStateCacheCommand {
ServerInRoom {
server: Box<ServerName>,
room_id: Box<RoomId>,
},
RoomServers {
room_id: Box<RoomId>,
},
ServerRooms {
server: Box<ServerName>,
},
RoomMembers {
room_id: Box<RoomId>,
},
LocalUsersInRoom {
room_id: Box<RoomId>,
},
ActiveLocalUsersInRoom {
room_id: Box<RoomId>,
},
RoomJoinedCount {
room_id: Box<RoomId>,
},
RoomInvitedCount {
room_id: Box<RoomId>,
},
RoomUserOnceJoined {
room_id: Box<RoomId>,
},
RoomMembersInvited {
room_id: Box<RoomId>,
},
GetInviteCount {
room_id: Box<RoomId>,
user_id: Box<UserId>,
},
GetLeftCount {
room_id: Box<RoomId>,
user_id: Box<UserId>,
},
RoomsJoined {
user_id: Box<UserId>,
},
RoomsLeft {
user_id: Box<UserId>,
},
RoomsInvited {
user_id: Box<UserId>,
},
InviteState {
user_id: Box<UserId>,
room_id: Box<RoomId>,
},
}
pub(super) async fn process(
subcommand: RoomStateCacheCommand, context: &Command<'_>,
) -> Result<RoomMessageEventContent> {
let services = context.services;
pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomMessageEventContent> {
match subcommand {
RoomStateCache::ServerInRoom {
RoomStateCacheCommand::ServerInRoom {
server,
room_id,
} => {
let timer = tokio::time::Instant::now();
let result = services()
.rooms
.state_cache
.server_in_room(&server, &room_id);
let result = services.rooms.state_cache.server_in_room(&server, &room_id);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"
)))
},
RoomStateCache::RoomServers {
RoomStateCacheCommand::RoomServers {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
.rooms
.state_cache
.room_servers(&room_id)
.collect();
let results: Result<Vec<_>> = services.rooms.state_cache.room_servers(&room_id).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::ServerRooms {
RoomStateCacheCommand::ServerRooms {
server,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services().rooms.state_cache.server_rooms(&server).collect();
let results: Result<Vec<_>> = services.rooms.state_cache.server_rooms(&server).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomMembers {
RoomStateCacheCommand::RoomMembers {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
.rooms
.state_cache
.room_members(&room_id)
.collect();
let results: Result<Vec<_>> = services.rooms.state_cache.room_members(&room_id).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::LocalUsersInRoom {
RoomStateCacheCommand::LocalUsersInRoom {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Vec<_> = services()
let results: Vec<_> = services
.rooms
.state_cache
.local_users_in_room(&room_id)
@ -76,11 +141,11 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::ActiveLocalUsersInRoom {
RoomStateCacheCommand::ActiveLocalUsersInRoom {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Vec<_> = services()
let results: Vec<_> = services
.rooms
.state_cache
.active_local_users_in_room(&room_id)
@ -91,33 +156,33 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomJoinedCount {
RoomStateCacheCommand::RoomJoinedCount {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services().rooms.state_cache.room_joined_count(&room_id);
let results = services.rooms.state_cache.room_joined_count(&room_id);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomInvitedCount {
RoomStateCacheCommand::RoomInvitedCount {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services().rooms.state_cache.room_invited_count(&room_id);
let results = services.rooms.state_cache.room_invited_count(&room_id);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomUserOnceJoined {
RoomStateCacheCommand::RoomUserOnceJoined {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
let results: Result<Vec<_>> = services
.rooms
.state_cache
.room_useroncejoined(&room_id)
@ -128,11 +193,11 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomMembersInvited {
RoomStateCacheCommand::RoomMembersInvited {
room_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
let results: Result<Vec<_>> = services
.rooms
.state_cache
.room_members_invited(&room_id)
@ -143,12 +208,12 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::GetInviteCount {
RoomStateCacheCommand::GetInviteCount {
room_id,
user_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
let results = services
.rooms
.state_cache
.get_invite_count(&room_id, &user_id);
@ -158,12 +223,12 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::GetLeftCount {
RoomStateCacheCommand::GetLeftCount {
room_id,
user_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
let results = services
.rooms
.state_cache
.get_left_count(&room_id, &user_id);
@ -173,56 +238,45 @@ pub(super) async fn room_state_cache(subcommand: RoomStateCache) -> Result<RoomM
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomsJoined {
RoomStateCacheCommand::RoomsJoined {
user_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
.rooms
.state_cache
.rooms_joined(&user_id)
.collect();
let results: Result<Vec<_>> = services.rooms.state_cache.rooms_joined(&user_id).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomsInvited {
RoomStateCacheCommand::RoomsInvited {
user_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services()
.rooms
.state_cache
.rooms_invited(&user_id)
.collect();
let results: Result<Vec<_>> = services.rooms.state_cache.rooms_invited(&user_id).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::RoomsLeft {
RoomStateCacheCommand::RoomsLeft {
user_id,
} => {
let timer = tokio::time::Instant::now();
let results: Result<Vec<_>> = services().rooms.state_cache.rooms_left(&user_id).collect();
let results: Result<Vec<_>> = services.rooms.state_cache.rooms_left(&user_id).collect();
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(
"Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"
)))
},
RoomStateCache::InviteState {
RoomStateCacheCommand::InviteState {
user_id,
room_id,
} => {
let timer = tokio::time::Instant::now();
let results = services()
.rooms
.state_cache
.invite_state(&user_id, &room_id);
let results = services.rooms.state_cache.invite_state(&user_id, &room_id);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(

91
src/admin/query/sending.rs
View file

@ -1,14 +1,73 @@
use ruma::events::room::message::RoomMessageEventContent;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, ServerName, UserId};
use service::sending::Destination;
use super::Sending;
use crate::{service::sending::Destination, services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/sending.rs
pub(crate) enum SendingCommand {
/// - Queries database for all `servercurrentevent_data`
ActiveRequests,
/// - Queries database for `servercurrentevent_data` but for a specific
/// destination
///
/// This command takes only *one* format of these arguments:
///
/// appservice_id
/// server_name
/// user_id AND push_key
///
/// See src/service/sending/mod.rs for the definition of the `Destination`
/// enum
ActiveRequestsFor {
#[arg(short, long)]
appservice_id: Option<String>,
#[arg(short, long)]
server_name: Option<Box<ServerName>>,
#[arg(short, long)]
user_id: Option<Box<UserId>>,
#[arg(short, long)]
push_key: Option<String>,
},
/// - Queries database for `servernameevent_data` which are the queued up
/// requests that will eventually be sent
///
/// This command takes only *one* format of these arguments:
///
/// appservice_id
/// server_name
/// user_id AND push_key
///
/// See src/service/sending/mod.rs for the definition of the `Destination`
/// enum
QueuedRequests {
#[arg(short, long)]
appservice_id: Option<String>,
#[arg(short, long)]
server_name: Option<Box<ServerName>>,
#[arg(short, long)]
user_id: Option<Box<UserId>>,
#[arg(short, long)]
push_key: Option<String>,
},
GetLatestEduCount {
server_name: Box<ServerName>,
},
}
/// All the getters and iterators in key_value/sending.rs
pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: SendingCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
Sending::ActiveRequests => {
SendingCommand::ActiveRequests => {
let timer = tokio::time::Instant::now();
let results = services().sending.db.active_requests();
let results = services.sending.db.active_requests();
let active_requests: Result<Vec<(_, _, _)>> = results.collect();
let query_time = timer.elapsed();
@ -16,7 +75,7 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
"Query completed in {query_time:?}:\n\n```rs\n{active_requests:#?}\n```"
)))
},
Sending::QueuedRequests {
SendingCommand::QueuedRequests {
appservice_id,
server_name,
user_id,
@ -38,12 +97,12 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
));
}
services()
services
.sending
.db
.queued_requests(&Destination::Appservice(appservice_id))
},
(None, Some(server_name), None, None) => services()
(None, Some(server_name), None, None) => services
.sending
.db
.queued_requests(&Destination::Normal(server_name.into())),
@ -55,7 +114,7 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
));
}
services()
services
.sending
.db
.queued_requests(&Destination::Push(user_id.into(), push_key))
@ -81,7 +140,7 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
"Query completed in {query_time:?}:\n\n```rs\n{queued_requests:#?}\n```"
)))
},
Sending::ActiveRequestsFor {
SendingCommand::ActiveRequestsFor {
appservice_id,
server_name,
user_id,
@ -104,12 +163,12 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
));
}
services()
services
.sending
.db
.active_requests_for(&Destination::Appservice(appservice_id))
},
(None, Some(server_name), None, None) => services()
(None, Some(server_name), None, None) => services
.sending
.db
.active_requests_for(&Destination::Normal(server_name.into())),
@ -121,7 +180,7 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
));
}
services()
services
.sending
.db
.active_requests_for(&Destination::Push(user_id.into(), push_key))
@ -147,11 +206,11 @@ pub(super) async fn sending(subcommand: Sending) -> Result<RoomMessageEventConte
"Query completed in {query_time:?}:\n\n```rs\n{active_requests:#?}\n```"
)))
},
Sending::GetLatestEduCount {
SendingCommand::GetLatestEduCount {
server_name,
} => {
let timer = tokio::time::Instant::now();
let results = services().sending.db.get_latest_educount(&server_name);
let results = services.sending.db.get_latest_educount(&server_name);
let query_time = timer.elapsed();
Ok(RoomMessageEventContent::notice_markdown(format!(

19
src/admin/query/users.rs
View file

@ -1,14 +1,23 @@
use clap::Subcommand;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use super::Users;
use crate::{services, Result};
use crate::Command;
#[derive(Debug, Subcommand)]
/// All the getters and iterators from src/database/key_value/users.rs
pub(crate) enum UsersCommand {
Iter,
}
/// All the getters and iterators in key_value/users.rs
pub(super) async fn users(subcommand: Users) -> Result<RoomMessageEventContent> {
pub(super) async fn process(subcommand: UsersCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match subcommand {
Users::Iter => {
UsersCommand::Iter => {
let timer = tokio::time::Instant::now();
let results = services().users.db.iter();
let results = services.users.db.iter();
let users = results.collect::<Vec<_>>();
let query_time = timer.elapsed();

67
src/admin/room/room_alias_commands.rs → src/admin/room/alias.rs
View file

@ -1,12 +1,49 @@
use std::fmt::Write;
use ruma::{events::room::message::RoomMessageEventContent, RoomAliasId};
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomAliasId, RoomId};
use super::RoomAliasCommand;
use crate::{escape_html, services, Result};
use crate::{escape_html, Command};
pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Result<RoomMessageEventContent> {
let server_user = &services().globals.server_user;
#[derive(Debug, Subcommand)]
pub(crate) enum RoomAliasCommand {
/// - Make an alias point to a room.
Set {
#[arg(short, long)]
/// Set the alias even if a room is already using it
force: bool,
/// The room id to set the alias on
room_id: Box<RoomId>,
/// The alias localpart to use (`alias`, not `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - Remove a local alias
Remove {
/// The alias localpart to remove (`alias`, not `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - Show which room is using an alias
Which {
/// The alias localpart to look up (`alias`, not
/// `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - List aliases currently being used
List {
/// If set, only list the aliases for this room
room_id: Option<Box<RoomId>>,
},
}
pub(super) async fn process(command: RoomAliasCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
let server_user = &services.globals.server_user;
match command {
RoomAliasCommand::Set {
@ -19,7 +56,7 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
| RoomAliasCommand::Which {
ref room_alias_localpart,
} => {
let room_alias_str = format!("#{}:{}", room_alias_localpart, services().globals.server_name());
let room_alias_str = format!("#{}:{}", room_alias_localpart, services.globals.server_name());
let room_alias = match RoomAliasId::parse_box(room_alias_str) {
Ok(alias) => alias,
Err(err) => return Ok(RoomMessageEventContent::text_plain(format!("Failed to parse alias: {err}"))),
@ -29,8 +66,8 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
force,
room_id,
..
} => match (force, services().rooms.alias.resolve_local_alias(&room_alias)) {
(true, Ok(Some(id))) => match services()
} => match (force, services.rooms.alias.resolve_local_alias(&room_alias)) {
(true, Ok(Some(id))) => match services
.rooms
.alias
.set_alias(&room_alias, &room_id, server_user)
@ -43,7 +80,7 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
(false, Ok(Some(id))) => Ok(RoomMessageEventContent::text_plain(format!(
"Refusing to overwrite in use alias for {id}, use -f or --force to overwrite"
))),
(_, Ok(None)) => match services()
(_, Ok(None)) => match services
.rooms
.alias
.set_alias(&room_alias, &room_id, server_user)
@ -55,8 +92,8 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
},
RoomAliasCommand::Remove {
..
} => match services().rooms.alias.resolve_local_alias(&room_alias) {
Ok(Some(id)) => match services()
} => match services.rooms.alias.resolve_local_alias(&room_alias) {
Ok(Some(id)) => match services
.rooms
.alias
.remove_alias(&room_alias, server_user)
@ -70,7 +107,7 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
},
RoomAliasCommand::Which {
..
} => match services().rooms.alias.resolve_local_alias(&room_alias) {
} => match services.rooms.alias.resolve_local_alias(&room_alias) {
Ok(Some(id)) => Ok(RoomMessageEventContent::text_plain(format!("Alias resolves to {id}"))),
Ok(None) => Ok(RoomMessageEventContent::text_plain("Alias isn't in use.")),
Err(err) => Ok(RoomMessageEventContent::text_plain(format!("Unable to lookup alias: {err}"))),
@ -84,7 +121,7 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
room_id,
} => {
if let Some(room_id) = room_id {
let aliases = services()
let aliases = services
.rooms
.alias
.local_aliases_for_room(&room_id)
@ -109,14 +146,14 @@ pub(super) async fn process(command: RoomAliasCommand, _body: Vec<&str>) -> Resu
Err(err) => Ok(RoomMessageEventContent::text_plain(format!("Unable to list aliases: {err}"))),
}
} else {
let aliases = services()
let aliases = services
.rooms
.alias
.all_local_aliases()
.collect::<Result<Vec<_>, _>>();
match aliases {
Ok(aliases) => {
let server_name = services().globals.server_name();
let server_name = services.globals.server_name();
let plain_list = aliases
.iter()
.fold(String::new(), |mut output, (alias, id)| {

19
src/admin/room/room_commands.rs → src/admin/room/commands.rs
View file

@ -1,15 +1,18 @@
use std::fmt::Write;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use crate::{escape_html, get_room_info, services, Result, PAGE_SIZE};
use crate::{admin_command, escape_html, get_room_info, PAGE_SIZE};
pub(super) async fn list(
_body: Vec<&str>, page: Option<usize>, exclude_disabled: bool, exclude_banned: bool,
#[admin_command]
pub(super) async fn list_rooms(
&self, page: Option<usize>, exclude_disabled: bool, exclude_banned: bool,
) -> Result<RoomMessageEventContent> {
// TODO: i know there's a way to do this with clap, but i can't seem to find it
let page = page.unwrap_or(1);
let mut rooms = services()
let mut rooms = self
.services
.rooms
.metadata
.iter_ids()
@ -18,7 +21,8 @@ pub(super) async fn list(
.ok()
.filter(|room_id| {
if exclude_disabled
&& services()
&& self
.services
.rooms
.metadata
.is_disabled(room_id)
@ -28,7 +32,8 @@ pub(super) async fn list(
}
if exclude_banned
&& services()
&& self
.services
.rooms
.metadata
.is_banned(room_id)
@ -39,7 +44,7 @@ pub(super) async fn list(
true
})
.map(|room_id| get_room_info(services(), &room_id))
.map(|room_id| get_room_info(self.services, &room_id))
})
.collect::<Vec<_>>();
rooms.sort_by_key(|r| r.1);

38
src/admin/room/room_directory_commands.rs → src/admin/room/directory.rs
View file

@ -1,21 +1,43 @@
use std::fmt::Write;
use ruma::{events::room::message::RoomMessageEventContent, OwnedRoomId};
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, OwnedRoomId, RoomId};
use super::RoomDirectoryCommand;
use crate::{escape_html, get_room_info, services, Result, PAGE_SIZE};
use crate::{escape_html, get_room_info, Command, PAGE_SIZE};
pub(super) async fn process(command: RoomDirectoryCommand, _body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[derive(Debug, Subcommand)]
pub(crate) enum RoomDirectoryCommand {
/// - Publish a room to the room directory
Publish {
/// The room id of the room to publish
room_id: Box<RoomId>,
},
/// - Unpublish a room to the room directory
Unpublish {
/// The room id of the room to unpublish
room_id: Box<RoomId>,
},
/// - List rooms that are published
List {
page: Option<usize>,
},
}
pub(super) async fn process(command: RoomDirectoryCommand, context: &Command<'_>) -> Result<RoomMessageEventContent> {
let services = context.services;
match command {
RoomDirectoryCommand::Publish {
room_id,
} => match services().rooms.directory.set_public(&room_id) {
} => match services.rooms.directory.set_public(&room_id) {
Ok(()) => Ok(RoomMessageEventContent::text_plain("Room published")),
Err(err) => Ok(RoomMessageEventContent::text_plain(format!("Unable to update room: {err}"))),
},
RoomDirectoryCommand::Unpublish {
room_id,
} => match services().rooms.directory.set_not_public(&room_id) {
} => match services.rooms.directory.set_not_public(&room_id) {
Ok(()) => Ok(RoomMessageEventContent::text_plain("Room unpublished")),
Err(err) => Ok(RoomMessageEventContent::text_plain(format!("Unable to update room: {err}"))),
},
@ -24,12 +46,12 @@ pub(super) async fn process(command: RoomDirectoryCommand, _body: Vec<&str>) ->
} => {
// TODO: i know there's a way to do this with clap, but i can't seem to find it
let page = page.unwrap_or(1);
let mut rooms = services()
let mut rooms = services
.rooms
.directory
.public_rooms()
.filter_map(Result::ok)
.map(|id: OwnedRoomId| get_room_info(services(), &id))
.map(|id: OwnedRoomId| get_room_info(services, &id))
.collect::<Vec<_>>();
rooms.sort_by_key(|r| r.1);
rooms.reverse();

51
src/admin/room/room_info_commands.rs → src/admin/room/info.rs
View file

@ -1,22 +1,30 @@
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomId};
use service::services;
use super::RoomInfoCommand;
use crate::Result;
use crate::{admin_command, admin_command_dispatch};
pub(super) async fn process(command: RoomInfoCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
match command {
RoomInfoCommand::ListJoinedMembers {
room_id,
} => list_joined_members(body, room_id).await,
RoomInfoCommand::ViewRoomTopic {
room_id,
} => view_room_topic(body, room_id).await,
}
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(crate) enum RoomInfoCommand {
/// - List joined members in a room
ListJoinedMembers {
room_id: Box<RoomId>,
},
/// - Displays room topic
///
/// Room topics can be huge, so this is in its
/// own separate command
ViewRoomTopic {
room_id: Box<RoomId>,
},
}
async fn list_joined_members(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let room_name = services()
#[admin_command]
async fn list_joined_members(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let room_name = self
.services
.rooms
.state_accessor
.get_name(&room_id)
@ -24,7 +32,8 @@ async fn list_joined_members(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<R
.flatten()
.unwrap_or_else(|| room_id.to_string());
let members = services()
let members = self
.services
.rooms
.state_cache
.room_members(&room_id)
@ -35,7 +44,7 @@ async fn list_joined_members(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<R
.map(|user_id| {
(
user_id.clone(),
services()
self.services
.users
.displayname(&user_id)
.unwrap_or(None)
@ -58,8 +67,14 @@ async fn list_joined_members(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<R
Ok(RoomMessageEventContent::notice_markdown(output_plain))
}
async fn view_room_topic(_body: Vec<&str>, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let Some(room_topic) = services().rooms.state_accessor.get_room_topic(&room_id)? else {
#[admin_command]
async fn view_room_topic(&self, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let Some(room_topic) = self
.services
.rooms
.state_accessor
.get_room_topic(&room_id)?
else {
return Ok(RoomMessageEventContent::text_plain("Room does not have a room topic set."));
};

166
src/admin/room/mod.rs
View file

@ -1,19 +1,23 @@
mod room_alias_commands;
mod room_commands;
mod room_directory_commands;
mod room_info_commands;
mod room_moderation_commands;
mod alias;
mod commands;
mod directory;
mod info;
mod moderation;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, RoomId, RoomOrAliasId};
use self::room_commands::list;
use self::{
alias::RoomAliasCommand, directory::RoomDirectoryCommand, info::RoomInfoCommand, moderation::RoomModerationCommand,
};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(super) enum RoomCommand {
/// - List all rooms the server knows about
List {
#[clap(alias = "list")]
ListRooms {
page: Option<usize>,
/// Excludes rooms that we have federation disabled with
@ -41,149 +45,3 @@ pub(super) enum RoomCommand {
/// - Manage the room directory
Directory(RoomDirectoryCommand),
}
#[derive(Debug, Subcommand)]
pub(super) enum RoomInfoCommand {
/// - List joined members in a room
ListJoinedMembers {
room_id: Box<RoomId>,
},
/// - Displays room topic
///
/// Room topics can be huge, so this is in its
/// own separate command
ViewRoomTopic {
room_id: Box<RoomId>,
},
}
#[derive(Debug, Subcommand)]
pub(super) enum RoomAliasCommand {
/// - Make an alias point to a room.
Set {
#[arg(short, long)]
/// Set the alias even if a room is already using it
force: bool,
/// The room id to set the alias on
room_id: Box<RoomId>,
/// The alias localpart to use (`alias`, not `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - Remove a local alias
Remove {
/// The alias localpart to remove (`alias`, not `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - Show which room is using an alias
Which {
/// The alias localpart to look up (`alias`, not
/// `#alias:servername.tld`)
room_alias_localpart: String,
},
/// - List aliases currently being used
List {
/// If set, only list the aliases for this room
room_id: Option<Box<RoomId>>,
},
}
#[derive(Debug, Subcommand)]
pub(super) enum RoomDirectoryCommand {
/// - Publish a room to the room directory
Publish {
/// The room id of the room to publish
room_id: Box<RoomId>,
},
/// - Unpublish a room to the room directory
Unpublish {
/// The room id of the room to unpublish
room_id: Box<RoomId>,
},
/// - List rooms that are published
List {
page: Option<usize>,
},
}
#[derive(Debug, Subcommand)]
pub(super) enum RoomModerationCommand {
/// - Bans a room from local users joining and evicts all our local users
/// from the room. Also blocks any invites (local and remote) for the
/// banned room.
///
/// Server admins (users in the conduwuit admin room) will not be evicted
/// and server admins can still join the room. To evict admins too, use
/// --force (also ignores errors) To disable incoming federation of the
/// room, use --disable-federation
BanRoom {
#[arg(short, long)]
/// Evicts admins out of the room and ignores any potential errors when
/// making our local users leave the room
force: bool,
#[arg(long)]
/// Disables incoming federation of the room after banning and evicting
/// users
disable_federation: bool,
/// The room in the format of `!roomid:example.com` or a room alias in
/// the format of `#roomalias:example.com`
room: Box<RoomOrAliasId>,
},
/// - Bans a list of rooms (room IDs and room aliases) from a newline
/// delimited codeblock similar to `user deactivate-all`
BanListOfRooms {
#[arg(short, long)]
/// Evicts admins out of the room and ignores any potential errors when
/// making our local users leave the room
force: bool,
#[arg(long)]
/// Disables incoming federation of the room after banning and evicting
/// users
disable_federation: bool,
},
/// - Unbans a room to allow local users to join again
///
/// To re-enable incoming federation of the room, use --enable-federation
UnbanRoom {
#[arg(long)]
/// Enables incoming federation of the room after unbanning
enable_federation: bool,
/// The room in the format of `!roomid:example.com` or a room alias in
/// the format of `#roomalias:example.com`
room: Box<RoomOrAliasId>,
},
/// - List of all rooms we have banned
ListBannedRooms,
}
pub(super) async fn process(command: RoomCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
RoomCommand::Info(command) => room_info_commands::process(command, body).await?,
RoomCommand::Alias(command) => room_alias_commands::process(command, body).await?,
RoomCommand::Directory(command) => room_directory_commands::process(command, body).await?,
RoomCommand::Moderation(command) => room_moderation_commands::process(command, body).await?,
RoomCommand::List {
page,
exclude_disabled,
exclude_banned,
} => list(body, page, exclude_disabled, exclude_banned).await?,
})
}

195
src/admin/room/room_moderation_commands.rs → src/admin/room/moderation.rs
View file

@ -1,37 +1,77 @@
use api::client::leave_room;
use clap::Subcommand;
use conduit::{debug, error, info, warn, Result};
use ruma::{events::room::message::RoomMessageEventContent, OwnedRoomId, RoomAliasId, RoomId, RoomOrAliasId};
use super::RoomModerationCommand;
use crate::{get_room_info, services};
use crate::{admin_command, admin_command_dispatch, get_room_info};
pub(super) async fn process(command: RoomModerationCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
match command {
RoomModerationCommand::BanRoom {
force,
room,
disable_federation,
} => ban_room(body, force, room, disable_federation).await,
RoomModerationCommand::BanListOfRooms {
force,
disable_federation,
} => ban_list_of_rooms(body, force, disable_federation).await,
RoomModerationCommand::UnbanRoom {
room,
enable_federation,
} => unban_room(body, room, enable_federation).await,
RoomModerationCommand::ListBannedRooms => list_banned_rooms(body).await,
}
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(crate) enum RoomModerationCommand {
/// - Bans a room from local users joining and evicts all our local users
/// from the room. Also blocks any invites (local and remote) for the
/// banned room.
///
/// Server admins (users in the conduwuit admin room) will not be evicted
/// and server admins can still join the room. To evict admins too, use
/// --force (also ignores errors) To disable incoming federation of the
/// room, use --disable-federation
BanRoom {
#[arg(short, long)]
/// Evicts admins out of the room and ignores any potential errors when
/// making our local users leave the room
force: bool,
#[arg(long)]
/// Disables incoming federation of the room after banning and evicting
/// users
disable_federation: bool,
/// The room in the format of `!roomid:example.com` or a room alias in
/// the format of `#roomalias:example.com`
room: Box<RoomOrAliasId>,
},
/// - Bans a list of rooms (room IDs and room aliases) from a newline
/// delimited codeblock similar to `user deactivate-all`
BanListOfRooms {
#[arg(short, long)]
/// Evicts admins out of the room and ignores any potential errors when
/// making our local users leave the room
force: bool,
#[arg(long)]
/// Disables incoming federation of the room after banning and evicting
/// users
disable_federation: bool,
},
/// - Unbans a room to allow local users to join again
///
/// To re-enable incoming federation of the room, use --enable-federation
UnbanRoom {
#[arg(long)]
/// Enables incoming federation of the room after unbanning
enable_federation: bool,
/// The room in the format of `!roomid:example.com` or a room alias in
/// the format of `#roomalias:example.com`
room: Box<RoomOrAliasId>,
},
/// - List of all rooms we have banned
ListBannedRooms,
}
#[admin_command]
async fn ban_room(
_body: Vec<&str>, force: bool, room: Box<RoomOrAliasId>, disable_federation: bool,
&self, force: bool, disable_federation: bool, room: Box<RoomOrAliasId>,
) -> Result<RoomMessageEventContent> {
debug!("Got room alias or ID: {}", room);
let admin_room_alias = &services().globals.admin_alias;
let admin_room_alias = &self.services.globals.admin_alias;
if let Some(admin_room_id) = services().admin.get_admin_room()? {
if let Some(admin_room_id) = self.services.admin.get_admin_room()? {
if room.to_string().eq(&admin_room_id) || room.to_string().eq(admin_room_alias) {
return Ok(RoomMessageEventContent::text_plain("Not allowed to ban the admin room."));
}
@ -50,7 +90,7 @@ async fn ban_room(
debug!("Room specified is a room ID, banning room ID");
services().rooms.metadata.ban_room(&room_id, true)?;
self.services.rooms.metadata.ban_room(&room_id, true)?;
room_id
} else if room.is_room_alias_id() {
@ -69,12 +109,13 @@ async fn ban_room(
get_alias_helper to fetch room ID remotely"
);
let room_id = if let Some(room_id) = services().rooms.alias.resolve_local_alias(&room_alias)? {
let room_id = if let Some(room_id) = self.services.rooms.alias.resolve_local_alias(&room_alias)? {
room_id
} else {
debug!("We don't have this room alias to a room ID locally, attempting to fetch room ID over federation");
match services()
match self
.services
.rooms
.alias
.resolve_alias(&room_alias, None)
@ -92,7 +133,7 @@ async fn ban_room(
}
};
services().rooms.metadata.ban_room(&room_id, true)?;
self.services.rooms.metadata.ban_room(&room_id, true)?;
room_id
} else {
@ -104,20 +145,21 @@ async fn ban_room(
debug!("Making all users leave the room {}", &room);
if force {
for local_user in services()
for local_user in self
.services
.rooms
.state_cache
.room_members(&room_id)
.filter_map(|user| {
user.ok().filter(|local_user| {
services().globals.user_is_local(local_user)
self.services.globals.user_is_local(local_user)
// additional wrapped check here is to avoid adding remote users
// who are in the admin room to the list of local users (would
// fail auth check)
&& (services().globals.user_is_local(local_user)
&& (self.services.globals.user_is_local(local_user)
// since this is a force operation, assume user is an admin
// if somehow this fails
&& services()
&& self.services
.users
.is_admin(local_user)
.unwrap_or(true))
@ -128,30 +170,31 @@ async fn ban_room(
&local_user, &room_id
);
if let Err(e) = leave_room(services(), &local_user, &room_id, None).await {
if let Err(e) = leave_room(self.services, &local_user, &room_id, None).await {
warn!(%e, "Failed to leave room");
}
}
} else {
for local_user in services()
for local_user in self
.services
.rooms
.state_cache
.room_members(&room_id)
.filter_map(|user| {
user.ok().filter(|local_user| {
local_user.server_name() == services().globals.server_name()
local_user.server_name() == self.services.globals.server_name()
// additional wrapped check here is to avoid adding remote users
// who are in the admin room to the list of local users (would fail auth check)
&& (local_user.server_name()
== services().globals.server_name()
&& !services()
== self.services.globals.server_name()
&& !self.services
.users
.is_admin(local_user)
.unwrap_or(false))
})
}) {
debug!("Attempting leave for user {} in room {}", &local_user, &room_id);
if let Err(e) = leave_room(services(), &local_user, &room_id, None).await {
if let Err(e) = leave_room(self.services, &local_user, &room_id, None).await {
error!(
"Error attempting to make local user {} leave room {} during room banning: {}",
&local_user, &room_id, e
@ -166,7 +209,7 @@ async fn ban_room(
}
if disable_federation {
services().rooms.metadata.disable_room(&room_id, true)?;
self.services.rooms.metadata.disable_room(&room_id, true)?;
return Ok(RoomMessageEventContent::text_plain(
"Room banned, removed all our local users, and disabled incoming federation with room.",
));
@ -178,19 +221,22 @@ async fn ban_room(
))
}
async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: bool) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
async fn ban_list_of_rooms(&self, force: bool, disable_federation: bool) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let rooms_s = body
.clone()
.drain(1..body.len().saturating_sub(1))
let rooms_s = self
.body
.to_vec()
.drain(1..self.body.len().saturating_sub(1))
.collect::<Vec<_>>();
let admin_room_alias = &services().globals.admin_alias;
let admin_room_alias = &self.services.globals.admin_alias;
let mut room_ban_count: usize = 0;
let mut room_ids: Vec<OwnedRoomId> = Vec::new();
@ -198,7 +244,7 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
for &room in &rooms_s {
match <&RoomOrAliasId>::try_from(room) {
Ok(room_alias_or_id) => {
if let Some(admin_room_id) = services().admin.get_admin_room()? {
if let Some(admin_room_id) = self.services.admin.get_admin_room()? {
if room.to_owned().eq(&admin_room_id) || room.to_owned().eq(admin_room_alias) {
info!("User specified admin room in bulk ban list, ignoring");
continue;
@ -231,7 +277,7 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
match RoomAliasId::parse(room_alias_or_id) {
Ok(room_alias) => {
let room_id =
if let Some(room_id) = services().rooms.alias.resolve_local_alias(&room_alias)? {
if let Some(room_id) = self.services.rooms.alias.resolve_local_alias(&room_alias)? {
room_id
} else {
debug!(
@ -239,7 +285,8 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
ID over federation"
);
match services()
match self
.services
.rooms
.alias
.resolve_alias(&room_alias, None)
@ -303,28 +350,35 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
}
for room_id in room_ids {
if services().rooms.metadata.ban_room(&room_id, true).is_ok() {
if self
.services
.rooms
.metadata
.ban_room(&room_id, true)
.is_ok()
{
debug!("Banned {room_id} successfully");
room_ban_count = room_ban_count.saturating_add(1);
}
debug!("Making all users leave the room {}", &room_id);
if force {
for local_user in services()
for local_user in self
.services
.rooms
.state_cache
.room_members(&room_id)
.filter_map(|user| {
user.ok().filter(|local_user| {
local_user.server_name() == services().globals.server_name()
local_user.server_name() == self.services.globals.server_name()
// additional wrapped check here is to avoid adding remote
// users who are in the admin room to the list of local
// users (would fail auth check)
&& (local_user.server_name()
== services().globals.server_name()
== self.services.globals.server_name()
// since this is a force operation, assume user is an
// admin if somehow this fails
&& services()
&& self.services
.users
.is_admin(local_user)
.unwrap_or(true))
@ -334,31 +388,32 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
"Attempting leave for user {} in room {} (forced, ignoring all errors, evicting admins too)",
&local_user, room_id
);
if let Err(e) = leave_room(services(), &local_user, &room_id, None).await {
if let Err(e) = leave_room(self.services, &local_user, &room_id, None).await {
warn!(%e, "Failed to leave room");
}
}
} else {
for local_user in services()
for local_user in self
.services
.rooms
.state_cache
.room_members(&room_id)
.filter_map(|user| {
user.ok().filter(|local_user| {
local_user.server_name() == services().globals.server_name()
local_user.server_name() == self.services.globals.server_name()
// additional wrapped check here is to avoid adding remote
// users who are in the admin room to the list of local
// users (would fail auth check)
&& (local_user.server_name()
== services().globals.server_name()
&& !services()
== self.services.globals.server_name()
&& !self.services
.users
.is_admin(local_user)
.unwrap_or(false))
})
}) {
debug!("Attempting leave for user {} in room {}", &local_user, &room_id);
if let Err(e) = leave_room(services(), &local_user, &room_id, None).await {
if let Err(e) = leave_room(self.services, &local_user, &room_id, None).await {
error!(
"Error attempting to make local user {} leave room {} during bulk room banning: {}",
&local_user, &room_id, e
@ -374,7 +429,7 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
}
if disable_federation {
services().rooms.metadata.disable_room(&room_id, true)?;
self.services.rooms.metadata.disable_room(&room_id, true)?;
}
}
@ -390,9 +445,8 @@ async fn ban_list_of_rooms(body: Vec<&str>, force: bool, disable_federation: boo
}
}
async fn unban_room(
_body: Vec<&str>, room: Box<RoomOrAliasId>, enable_federation: bool,
) -> Result<RoomMessageEventContent> {
#[admin_command]
async fn unban_room(&self, enable_federation: bool, room: Box<RoomOrAliasId>) -> Result<RoomMessageEventContent> {
let room_id = if room.is_room_id() {
let room_id = match RoomId::parse(&room) {
Ok(room_id) => room_id,
@ -406,7 +460,7 @@ async fn unban_room(
debug!("Room specified is a room ID, unbanning room ID");
services().rooms.metadata.ban_room(&room_id, false)?;
self.services.rooms.metadata.ban_room(&room_id, false)?;
room_id
} else if room.is_room_alias_id() {
@ -425,12 +479,13 @@ async fn unban_room(
get_alias_helper to fetch room ID remotely"
);
let room_id = if let Some(room_id) = services().rooms.alias.resolve_local_alias(&room_alias)? {
let room_id = if let Some(room_id) = self.services.rooms.alias.resolve_local_alias(&room_alias)? {
room_id
} else {
debug!("We don't have this room alias to a room ID locally, attempting to fetch room ID over federation");
match services()
match self
.services
.rooms
.alias
.resolve_alias(&room_alias, None)
@ -448,7 +503,7 @@ async fn unban_room(
}
};
services().rooms.metadata.ban_room(&room_id, false)?;
self.services.rooms.metadata.ban_room(&room_id, false)?;
room_id
} else {
@ -459,7 +514,7 @@ async fn unban_room(
};
if enable_federation {
services().rooms.metadata.disable_room(&room_id, false)?;
self.services.rooms.metadata.disable_room(&room_id, false)?;
return Ok(RoomMessageEventContent::text_plain("Room unbanned."));
}
@ -469,8 +524,10 @@ async fn unban_room(
))
}
async fn list_banned_rooms(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let rooms = services()
#[admin_command]
async fn list_banned_rooms(&self) -> Result<RoomMessageEventContent> {
let rooms = self
.services
.rooms
.metadata
.list_banned_rooms()
@ -484,7 +541,7 @@ async fn list_banned_rooms(_body: Vec<&str>) -> Result<RoomMessageEventContent>
let mut rooms = room_ids
.into_iter()
.map(|room_id| get_room_info(services(), &room_id))
.map(|room_id| get_room_info(self.services, &room_id))
.collect::<Vec<_>>();
rooms.sort_by_key(|r| r.1);
rooms.reverse();

83
src/admin/server/commands.rs
View file

@ -1,12 +1,14 @@
use std::fmt::Write;
use std::{fmt::Write, sync::Arc};
use conduit::{info, utils::time, warn, Err, Result};
use ruma::events::room::message::RoomMessageEventContent;
use crate::services;
use crate::admin_command;
pub(super) async fn uptime(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let elapsed = services()
#[admin_command]
pub(super) async fn uptime(&self) -> Result<RoomMessageEventContent> {
let elapsed = self
.services
.server
.started
.elapsed()
@ -16,13 +18,15 @@ pub(super) async fn uptime(_body: Vec<&str>) -> Result<RoomMessageEventContent>
Ok(RoomMessageEventContent::notice_plain(format!("{result}.")))
}
pub(super) async fn show_config(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn show_config(&self) -> Result<RoomMessageEventContent> {
// Construct and send the response
Ok(RoomMessageEventContent::text_plain(format!("{}", services().globals.config)))
Ok(RoomMessageEventContent::text_plain(format!("{}", self.services.globals.config)))
}
#[admin_command]
pub(super) async fn list_features(
_body: Vec<&str>, available: bool, enabled: bool, comma: bool,
&self, available: bool, enabled: bool, comma: bool,
) -> Result<RoomMessageEventContent> {
let delim = if comma {
","
@ -62,9 +66,10 @@ pub(super) async fn list_features(
Ok(RoomMessageEventContent::text_markdown(features))
}
pub(super) async fn memory_usage(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let services_usage = services().memory_usage().await?;
let database_usage = services().db.db.memory_usage()?;
#[admin_command]
pub(super) async fn memory_usage(&self) -> Result<RoomMessageEventContent> {
let services_usage = self.services.memory_usage().await?;
let database_usage = self.services.db.db.memory_usage()?;
let allocator_usage = conduit::alloc::memory_usage().map_or(String::new(), |s| format!("\nAllocator:\n{s}"));
Ok(RoomMessageEventContent::text_plain(format!(
@ -72,14 +77,16 @@ pub(super) async fn memory_usage(_body: Vec<&str>) -> Result<RoomMessageEventCon
)))
}
pub(super) async fn clear_caches(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
services().clear_cache().await;
#[admin_command]
pub(super) async fn clear_caches(&self) -> Result<RoomMessageEventContent> {
self.services.clear_cache().await;
Ok(RoomMessageEventContent::text_plain("Done."))
}
pub(super) async fn list_backups(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let result = services().globals.db.backup_list()?;
#[admin_command]
pub(super) async fn list_backups(&self) -> Result<RoomMessageEventContent> {
let result = self.services.globals.db.backup_list()?;
if result.is_empty() {
Ok(RoomMessageEventContent::text_plain("No backups found."))
@ -88,46 +95,51 @@ pub(super) async fn list_backups(_body: Vec<&str>) -> Result<RoomMessageEventCon
}
}
pub(super) async fn backup_database(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let mut result = services()
#[admin_command]
pub(super) async fn backup_database(&self) -> Result<RoomMessageEventContent> {
let globals = Arc::clone(&self.services.globals);
let mut result = self
.services
.server
.runtime()
.spawn_blocking(move || match services().globals.db.backup() {
.spawn_blocking(move || match globals.db.backup() {
Ok(()) => String::new(),
Err(e) => (*e).to_string(),
})
.await
.unwrap();
.await?;
if result.is_empty() {
result = services().globals.db.backup_list()?;
result = self.services.globals.db.backup_list()?;
}
Ok(RoomMessageEventContent::text_plain(&result))
}
pub(super) async fn list_database_files(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
let result = services().globals.db.file_list()?;
Ok(RoomMessageEventContent::notice_markdown(result))
}
pub(super) async fn admin_notice(_body: Vec<&str>, message: Vec<String>) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn list_database_files(&self) -> Result<RoomMessageEventContent> {
let result = self.services.globals.db.file_list()?;
Ok(RoomMessageEventContent::notice_markdown(result))
}
#[admin_command]
pub(super) async fn admin_notice(&self, message: Vec<String>) -> Result<RoomMessageEventContent> {
let message = message.join(" ");
services().admin.send_text(&message).await;
self.services.admin.send_text(&message).await;
Ok(RoomMessageEventContent::notice_plain("Notice was sent to #admins"))
}
#[cfg(conduit_mods)]
pub(super) async fn reload(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
services().server.reload()?;
#[admin_command]
pub(super) async fn reload_mods(&self) -> Result<RoomMessageEventContent> {
self.services.server.reload()?;
Ok(RoomMessageEventContent::notice_plain("Reloading server..."))
}
#[admin_command]
#[cfg(unix)]
pub(super) async fn restart(_body: Vec<&str>, force: bool) -> Result<RoomMessageEventContent> {
pub(super) async fn restart(&self, force: bool) -> Result<RoomMessageEventContent> {
use conduit::utils::sys::current_exe_deleted;
if !force && current_exe_deleted() {
@ -137,14 +149,15 @@ pub(super) async fn restart(_body: Vec<&str>, force: bool) -> Result<RoomMessage
);
}
services().server.restart()?;
self.services.server.restart()?;
Ok(RoomMessageEventContent::notice_plain("Restarting server..."))
}
pub(super) async fn shutdown(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn shutdown(&self) -> Result<RoomMessageEventContent> {
warn!("shutdown command");
services().server.shutdown()?;
self.services.server.shutdown()?;
Ok(RoomMessageEventContent::notice_plain("Shutting down server..."))
}

35
src/admin/server/mod.rs
View file

@ -2,10 +2,10 @@ mod commands;
use clap::Subcommand;
use conduit::Result;
use ruma::events::room::message::RoomMessageEventContent;
use self::commands::*;
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(super) enum ServerCommand {
/// - Time elapsed since startup
@ -47,9 +47,9 @@ pub(super) enum ServerCommand {
message: Vec<String>,
},
#[cfg(conduit_mods)]
/// - Hot-reload the server
Reload,
#[clap(alias = "reload")]
ReloadMods,
#[cfg(unix)]
/// - Restart the server
@ -61,30 +61,3 @@ pub(super) enum ServerCommand {
/// - Shutdown the server
Shutdown,
}
pub(super) async fn process(command: ServerCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
ServerCommand::Uptime => uptime(body).await?,
ServerCommand::ShowConfig => show_config(body).await?,
ServerCommand::ListFeatures {
available,
enabled,
comma,
} => list_features(body, available, enabled, comma).await?,
ServerCommand::MemoryUsage => memory_usage(body).await?,
ServerCommand::ClearCaches => clear_caches(body).await?,
ServerCommand::ListBackups => list_backups(body).await?,
ServerCommand::BackupDatabase => backup_database(body).await?,
ServerCommand::ListDatabaseFiles => list_database_files(body).await?,
ServerCommand::AdminNotice {
message,
} => admin_notice(body, message).await?,
#[cfg(conduit_mods)]
ServerCommand::Reload => reload(body).await?,
#[cfg(unix)]
ServerCommand::Restart {
force,
} => restart(body, force).await?,
ServerCommand::Shutdown => shutdown(body).await?,
})
}

180
src/admin/user/commands.rs
View file

@ -1,7 +1,7 @@
use std::{collections::BTreeMap, fmt::Write as _};
use api::client::{join_room_by_id_helper, leave_all_rooms, update_avatar_url, update_displayname};
use conduit::{utils, Result};
use conduit::{error, info, utils, warn, Result};
use ruma::{
events::{
room::message::RoomMessageEventContent,
@ -10,17 +10,17 @@ use ruma::{
},
OwnedRoomId, OwnedRoomOrAliasId, OwnedUserId, RoomId,
};
use tracing::{error, info, warn};
use crate::{
escape_html, get_room_info, services,
admin_command, escape_html, get_room_info,
utils::{parse_active_local_user_id, parse_local_user_id},
};
const AUTO_GEN_PASSWORD_LENGTH: usize = 25;
pub(super) async fn list(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
match services().users.list_local_users() {
#[admin_command]
pub(super) async fn list_users(&self) -> Result<RoomMessageEventContent> {
match self.services.users.list_local_users() {
Ok(users) => {
let mut plain_msg = format!("Found {} local user account(s):\n```\n", users.len());
plain_msg += users.join("\n").as_str();
@ -32,13 +32,12 @@ pub(super) async fn list(_body: Vec<&str>) -> Result<RoomMessageEventContent> {
}
}
pub(super) async fn create(
_body: Vec<&str>, username: String, password: Option<String>,
) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn create_user(&self, username: String, password: Option<String>) -> Result<RoomMessageEventContent> {
// Validate user id
let user_id = parse_local_user_id(services(), &username)?;
let user_id = parse_local_user_id(self.services, &username)?;
if services().users.exists(&user_id)? {
if self.services.users.exists(&user_id)? {
return Ok(RoomMessageEventContent::text_plain(format!("Userid {user_id} already exists")));
}
@ -51,30 +50,33 @@ pub(super) async fn create(
let password = password.unwrap_or_else(|| utils::random_string(AUTO_GEN_PASSWORD_LENGTH));
// Create user
services().users.create(&user_id, Some(password.as_str()))?;
self.services
.users
.create(&user_id, Some(password.as_str()))?;
// Default to pretty displayname
let mut displayname = user_id.localpart().to_owned();
// If `new_user_displayname_suffix` is set, registration will push whatever
// content is set to the user's display name with a space before it
if !services()
if !self
.services
.globals
.config
.new_user_displayname_suffix
.is_empty()
{
write!(displayname, " {}", services().globals.config.new_user_displayname_suffix)
write!(displayname, " {}", self.services.globals.config.new_user_displayname_suffix)
.expect("should be able to write to string buffer");
}
services()
self.services
.users
.set_displayname(&user_id, Some(displayname))
.await?;
// Initial account data
services().account_data.update(
self.services.account_data.update(
None,
&user_id,
ruma::events::GlobalAccountDataEventType::PushRules
@ -88,12 +90,13 @@ pub(super) async fn create(
.expect("to json value always works"),
)?;
if !services().globals.config.auto_join_rooms.is_empty() {
for room in &services().globals.config.auto_join_rooms {
if !services()
if !self.services.globals.config.auto_join_rooms.is_empty() {
for room in &self.services.globals.config.auto_join_rooms {
if !self
.services
.rooms
.state_cache
.server_in_room(services().globals.server_name(), room)?
.server_in_room(self.services.globals.server_name(), room)?
{
warn!("Skipping room {room} to automatically join as we have never joined before.");
continue;
@ -101,11 +104,11 @@ pub(super) async fn create(
if let Some(room_id_server_name) = room.server_name() {
match join_room_by_id_helper(
services(),
self.services,
&user_id,
room,
Some("Automatically joining this room upon registration".to_owned()),
&[room_id_server_name.to_owned(), services().globals.server_name().to_owned()],
&[room_id_server_name.to_owned(), self.services.globals.server_name().to_owned()],
None,
)
.await
@ -130,38 +133,38 @@ pub(super) async fn create(
)))
}
pub(super) async fn deactivate(
_body: Vec<&str>, no_leave_rooms: bool, user_id: String,
) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn deactivate(&self, no_leave_rooms: bool, user_id: String) -> Result<RoomMessageEventContent> {
// Validate user id
let user_id = parse_local_user_id(services(), &user_id)?;
let user_id = parse_local_user_id(self.services, &user_id)?;
// don't deactivate the server service account
if user_id == services().globals.server_user {
if user_id == self.services.globals.server_user {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to deactivate the server service account.",
));
}
services().users.deactivate_account(&user_id)?;
self.services.users.deactivate_account(&user_id)?;
if !no_leave_rooms {
services()
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"Making {user_id} leave all rooms after deactivation..."
)))
.await;
let all_joined_rooms: Vec<OwnedRoomId> = services()
let all_joined_rooms: Vec<OwnedRoomId> = self
.services
.rooms
.state_cache
.rooms_joined(&user_id)
.filter_map(Result::ok)
.collect();
update_displayname(services(), user_id.clone(), None, all_joined_rooms.clone()).await?;
update_avatar_url(services(), user_id.clone(), None, None, all_joined_rooms).await?;
leave_all_rooms(services(), &user_id).await;
update_displayname(self.services, user_id.clone(), None, all_joined_rooms.clone()).await?;
update_avatar_url(self.services, user_id.clone(), None, None, all_joined_rooms).await?;
leave_all_rooms(self.services, &user_id).await;
}
Ok(RoomMessageEventContent::text_plain(format!(
@ -169,10 +172,11 @@ pub(super) async fn deactivate(
)))
}
pub(super) async fn reset_password(_body: Vec<&str>, username: String) -> Result<RoomMessageEventContent> {
let user_id = parse_local_user_id(services(), &username)?;
#[admin_command]
pub(super) async fn reset_password(&self, username: String) -> Result<RoomMessageEventContent> {
let user_id = parse_local_user_id(self.services, &username)?;
if user_id == services().globals.server_user {
if user_id == self.services.globals.server_user {
return Ok(RoomMessageEventContent::text_plain(
"Not allowed to set the password for the server account. Please use the emergency password config option.",
));
@ -180,7 +184,8 @@ pub(super) async fn reset_password(_body: Vec<&str>, username: String) -> Result
let new_password = utils::random_string(AUTO_GEN_PASSWORD_LENGTH);
match services()
match self
.services
.users
.set_password(&user_id, Some(new_password.as_str()))
{
@ -193,28 +198,29 @@ pub(super) async fn reset_password(_body: Vec<&str>, username: String) -> Result
}
}
pub(super) async fn deactivate_all(
body: Vec<&str>, no_leave_rooms: bool, force: bool,
) -> Result<RoomMessageEventContent> {
if body.len() < 2 || !body[0].trim().starts_with("```") || body.last().unwrap_or(&"").trim() != "```" {
#[admin_command]
pub(super) async fn deactivate_all(&self, no_leave_rooms: bool, force: bool) -> Result<RoomMessageEventContent> {
if self.body.len() < 2 || !self.body[0].trim().starts_with("```") || self.body.last().unwrap_or(&"").trim() != "```"
{
return Ok(RoomMessageEventContent::text_plain(
"Expected code block in command body. Add --help for details.",
));
}
let usernames = body
.clone()
.drain(1..body.len().saturating_sub(1))
let usernames = self
.body
.to_vec()
.drain(1..self.body.len().saturating_sub(1))
.collect::<Vec<_>>();
let mut user_ids: Vec<OwnedUserId> = Vec::with_capacity(usernames.len());
let mut admins = Vec::new();
for username in usernames {
match parse_active_local_user_id(services(), username) {
match parse_active_local_user_id(self.services, username) {
Ok(user_id) => {
if services().users.is_admin(&user_id)? && !force {
services()
if self.services.users.is_admin(&user_id)? && !force {
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"{username} is an admin and --force is not set, skipping over"
@ -225,8 +231,8 @@ pub(super) async fn deactivate_all(
}
// don't deactivate the server service account
if user_id == services().globals.server_user {
services()
if user_id == self.services.globals.server_user {
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"{username} is the server service account, skipping over"
@ -238,7 +244,7 @@ pub(super) async fn deactivate_all(
user_ids.push(user_id);
},
Err(e) => {
services()
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!(
"{username} is not a valid username, skipping over: {e}"
@ -252,24 +258,25 @@ pub(super) async fn deactivate_all(
let mut deactivation_count: usize = 0;
for user_id in user_ids {
match services().users.deactivate_account(&user_id) {
match self.services.users.deactivate_account(&user_id) {
Ok(()) => {
deactivation_count = deactivation_count.saturating_add(1);
if !no_leave_rooms {
info!("Forcing user {user_id} to leave all rooms apart of deactivate-all");
let all_joined_rooms: Vec<OwnedRoomId> = services()
let all_joined_rooms: Vec<OwnedRoomId> = self
.services
.rooms
.state_cache
.rooms_joined(&user_id)
.filter_map(Result::ok)
.collect();
update_displayname(services(), user_id.clone(), None, all_joined_rooms.clone()).await?;
update_avatar_url(services(), user_id.clone(), None, None, all_joined_rooms).await?;
leave_all_rooms(services(), &user_id).await;
update_displayname(self.services, user_id.clone(), None, all_joined_rooms.clone()).await?;
update_avatar_url(self.services, user_id.clone(), None, None, all_joined_rooms).await?;
leave_all_rooms(self.services, &user_id).await;
}
},
Err(e) => {
services()
self.services
.admin
.send_message(RoomMessageEventContent::text_plain(format!("Failed deactivating user: {e}")))
.await;
@ -290,16 +297,18 @@ pub(super) async fn deactivate_all(
}
}
pub(super) async fn list_joined_rooms(_body: Vec<&str>, user_id: String) -> Result<RoomMessageEventContent> {
#[admin_command]
pub(super) async fn list_joined_rooms(&self, user_id: String) -> Result<RoomMessageEventContent> {
// Validate user id
let user_id = parse_local_user_id(services(), &user_id)?;
let user_id = parse_local_user_id(self.services, &user_id)?;
let mut rooms: Vec<(OwnedRoomId, u64, String)> = services()
let mut rooms: Vec<(OwnedRoomId, u64, String)> = self
.services
.rooms
.state_cache
.rooms_joined(&user_id)
.filter_map(Result::ok)
.map(|room_id| get_room_info(services(), &room_id))
.map(|room_id| get_room_info(self.services, &room_id))
.collect();
if rooms.is_empty() {
@ -341,35 +350,38 @@ pub(super) async fn list_joined_rooms(_body: Vec<&str>, user_id: String) -> Resu
Ok(RoomMessageEventContent::text_html(output_plain, output_html))
}
#[admin_command]
pub(super) async fn force_join_room(
_body: Vec<&str>, user_id: String, room_id: OwnedRoomOrAliasId,
&self, user_id: String, room_id: OwnedRoomOrAliasId,
) -> Result<RoomMessageEventContent> {
let user_id = parse_local_user_id(services(), &user_id)?;
let room_id = services().rooms.alias.resolve(&room_id).await?;
let user_id = parse_local_user_id(self.services, &user_id)?;
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
assert!(
services().globals.user_is_local(&user_id),
self.services.globals.user_is_local(&user_id),
"Parsed user_id must be a local user"
);
join_room_by_id_helper(services(), &user_id, &room_id, None, &[], None).await?;
join_room_by_id_helper(self.services, &user_id, &room_id, None, &[], None).await?;
Ok(RoomMessageEventContent::notice_markdown(format!(
"{user_id} has been joined to {room_id}.",
)))
}
pub(super) async fn make_user_admin(_body: Vec<&str>, user_id: String) -> Result<RoomMessageEventContent> {
let user_id = parse_local_user_id(services(), &user_id)?;
let displayname = services()
#[admin_command]
pub(super) async fn make_user_admin(&self, user_id: String) -> Result<RoomMessageEventContent> {
let user_id = parse_local_user_id(self.services, &user_id)?;
let displayname = self
.services
.users
.displayname(&user_id)?
.unwrap_or_else(|| user_id.to_string());
assert!(
services().globals.user_is_local(&user_id),
self.services.globals.user_is_local(&user_id),
"Parsed user_id must be a local user"
);
services()
self.services
.admin
.make_user_admin(&user_id, displayname)
.await?;
@ -379,12 +391,14 @@ pub(super) async fn make_user_admin(_body: Vec<&str>, user_id: String) -> Result
)))
}
#[admin_command]
pub(super) async fn put_room_tag(
_body: Vec<&str>, user_id: String, room_id: Box<RoomId>, tag: String,
&self, user_id: String, room_id: Box<RoomId>, tag: String,
) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(services(), &user_id)?;
let user_id = parse_active_local_user_id(self.services, &user_id)?;
let event = services()
let event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)?;
@ -402,7 +416,7 @@ pub(super) async fn put_room_tag(
.tags
.insert(tag.clone().into(), TagInfo::new());
services().account_data.update(
self.services.account_data.update(
Some(&room_id),
&user_id,
RoomAccountDataEventType::Tag,
@ -414,12 +428,14 @@ pub(super) async fn put_room_tag(
)))
}
#[admin_command]
pub(super) async fn delete_room_tag(
_body: Vec<&str>, user_id: String, room_id: Box<RoomId>, tag: String,
&self, user_id: String, room_id: Box<RoomId>, tag: String,
) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(services(), &user_id)?;
let user_id = parse_active_local_user_id(self.services, &user_id)?;
let event = services()
let event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)?;
@ -434,7 +450,7 @@ pub(super) async fn delete_room_tag(
tags_event.content.tags.remove(&tag.clone().into());
services().account_data.update(
self.services.account_data.update(
Some(&room_id),
&user_id,
RoomAccountDataEventType::Tag,
@ -446,12 +462,12 @@ pub(super) async fn delete_room_tag(
)))
}
pub(super) async fn get_room_tags(
_body: Vec<&str>, user_id: String, room_id: Box<RoomId>,
) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(services(), &user_id)?;
#[admin_command]
pub(super) async fn get_room_tags(&self, user_id: String, room_id: Box<RoomId>) -> Result<RoomMessageEventContent> {
let user_id = parse_active_local_user_id(self.services, &user_id)?;
let event = services()
let event = self
.services
.account_data
.get(Some(&room_id), &user_id, RoomAccountDataEventType::Tag)?;

56
src/admin/user/mod.rs
View file

@ -2,14 +2,16 @@ mod commands;
use clap::Subcommand;
use conduit::Result;
use ruma::{events::room::message::RoomMessageEventContent, OwnedRoomOrAliasId, RoomId};
use ruma::{OwnedRoomOrAliasId, RoomId};
use self::commands::*;
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
pub(super) enum UserCommand {
/// - Create a new user
Create {
#[clap(alias = "create")]
CreateUser {
/// Username of the new user
username: String,
/// Password of the new user, if unspecified one is generated
@ -56,7 +58,8 @@ pub(super) enum UserCommand {
},
/// - List local users in the database
List,
#[clap(alias = "list")]
ListUsers,
/// - Lists all the rooms (local and remote) that the specified user is
/// joined in
@ -101,48 +104,3 @@ pub(super) enum UserCommand {
room_id: Box<RoomId>,
},
}
pub(super) async fn process(command: UserCommand, body: Vec<&str>) -> Result<RoomMessageEventContent> {
Ok(match command {
UserCommand::List => list(body).await?,
UserCommand::Create {
username,
password,
} => create(body, username, password).await?,
UserCommand::Deactivate {
no_leave_rooms,
user_id,
} => deactivate(body, no_leave_rooms, user_id).await?,
UserCommand::ResetPassword {
username,
} => reset_password(body, username).await?,
UserCommand::DeactivateAll {
no_leave_rooms,
force,
} => deactivate_all(body, no_leave_rooms, force).await?,
UserCommand::ListJoinedRooms {
user_id,
} => list_joined_rooms(body, user_id).await?,
UserCommand::ForceJoinRoom {
user_id,
room_id,
} => force_join_room(body, user_id, room_id).await?,
UserCommand::MakeUserAdmin {
user_id,
} => make_user_admin(body, user_id).await?,
UserCommand::PutRoomTag {
user_id,
room_id,
tag,
} => put_room_tag(body, user_id, room_id, tag).await?,
UserCommand::DeleteRoomTag {
user_id,
room_id,
tag,
} => delete_room_tag(body, user_id, room_id, tag).await?,
UserCommand::GetRoomTags {
user_id,
room_id,
} => get_room_tags(body, user_id, room_id).await?,
})
}

22
src/macros/admin.rs
View file

@ -2,15 +2,27 @@ use itertools::Itertools;
use proc_macro::{Span, TokenStream};
use proc_macro2::TokenStream as TokenStream2;
use quote::{quote, ToTokens};
use syn::{Error, Fields, Ident, ItemEnum, Meta, Variant};
use syn::{parse_quote, Attribute, Error, Fields, Ident, ItemEnum, ItemFn, Meta, Variant};
use crate::{utils::camel_to_snake_string, Result};
pub(super) fn command(mut item: ItemFn, _args: &[Meta]) -> Result<TokenStream> {
let attr: Attribute = parse_quote! {
#[conduit_macros::implement(crate::Command, params = "<'_>")]
};
item.attrs.push(attr);
Ok(item.into_token_stream().into())
}
pub(super) fn command_dispatch(item: ItemEnum, _args: &[Meta]) -> Result<TokenStream> {
let name = &item.ident;
let arm: Vec<TokenStream2> = item.variants.iter().map(dispatch_arm).try_collect()?;
let switch = quote! {
pub(super) async fn process(command: #name, body: Vec<&str>) -> Result<RoomMessageEventContent> {
pub(super) async fn process(
command: #name,
context: &crate::Command<'_>
) -> Result<ruma::events::room::message::RoomMessageEventContent> {
use #name::*;
#[allow(non_snake_case)]
Ok(match command {
@ -34,7 +46,7 @@ fn dispatch_arm(v: &Variant) -> Result<TokenStream2> {
let field = fields.named.iter().filter_map(|f| f.ident.as_ref());
let arg = field.clone();
quote! {
#name { #( #field ),* } => Box::pin(#handler(&body, #( #arg ),*)).await?,
#name { #( #field ),* } => Box::pin(context.#handler(#( #arg ),*)).await?,
}
},
Fields::Unnamed(fields) => {
@ -42,12 +54,12 @@ fn dispatch_arm(v: &Variant) -> Result<TokenStream2> {
return Err(Error::new(Span::call_site().into(), "One unnamed field required"));
};
quote! {
#name ( #field ) => Box::pin(#handler::process(#field, body)).await?,
#name ( #field ) => Box::pin(#handler::process(#field, context)).await?,
}
},
Fields::Unit => {
quote! {
#name => Box::pin(#handler(&body)).await?,
#name => Box::pin(context.#handler()).await?,
}
},
};

5
src/macros/mod.rs
View file

@ -14,6 +14,11 @@ use syn::{
pub(crate) type Result<T> = std::result::Result<T, Error>;
#[proc_macro_attribute]
pub fn admin_command(args: TokenStream, input: TokenStream) -> TokenStream {
attribute_macro::<ItemFn, _>(args, input, admin::command)
}
#[proc_macro_attribute]
pub fn admin_command_dispatch(args: TokenStream, input: TokenStream) -> TokenStream {
attribute_macro::<ItemEnum, _>(args, input, admin::command_dispatch)