Mirrors/continuwuity
0
0 Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2026-11-13 12:16:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

style: Refactor admin crate, drop admin_command macro

Browse Source 
`admin_command` depended on `implement`, which is now gone.
This commit is contained in:
timedout committed 2026-06-27 18:37:26 +01:00
1 parent 478560daae
commit 6fdeadc356
22 files changed
+4068 -4106
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/admin/appservice/commands.rs
+59 -63
View File
@@ -1,80 +1,76 @@
use conduwuit::{Err, Result, checked};
use futures::{FutureExt, StreamExt, TryFutureExt};
use crate::admin_command;
impl crate::Context<'_> {
pub(super) async fn register(&self) -> Result {
let body = &self.body;
let body_len = self.body.len();
if body_len < 2
|| !body[0].trim().starts_with("```")
|| body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.");
}
#[admin_command]
pub(super) async fn register(&self) -> Result {
let body = &self.body;
let body_len = self.body.len();
if body_len < 2
|| !body[0].trim().starts_with("```")
|| body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.");
let range = 1..checked!(body_len - 1)?;
let appservice_config_body = body[range].join("\n");
let parsed_config = serde_saphyr::from_str(&appservice_config_body);
match parsed_config {
| Err(e) => return Err!("Could not parse appservice config as YAML: {e}"),
| Ok(registration) => match self
.services
.appservice
.register_appservice(&registration, &appservice_config_body)
.await
.map(|()| registration.id)
{
| Err(e) => return Err!("Failed to register appservice: {e}"),
| Ok(id) => write!(self, "Appservice registered with ID: {id}"),
},
}
.await
}
let range = 1..checked!(body_len - 1)?;
let appservice_config_body = body[range].join("\n");
let parsed_config = serde_saphyr::from_str(&appservice_config_body);
match parsed_config {
| Err(e) => return Err!("Could not parse appservice config as YAML: {e}"),
| Ok(registration) => match self
pub(super) async fn unregister(&self, appservice_identifier: String) -> Result {
match self
.services
.appservice
.register_appservice(&registration, &appservice_config_body)
.unregister_appservice(&appservice_identifier)
.await
.map(|()| registration.id)
{
| Err(e) => return Err!("Failed to register appservice: {e}"),
| Ok(id) => write!(self, "Appservice registered with ID: {id}"),
},
}
.await
}
#[admin_command]
pub(super) async fn unregister(&self, appservice_identifier: String) -> Result {
match self
.services
.appservice
.unregister_appservice(&appservice_identifier)
| Err(e) => return Err!("Failed to unregister appservice: {e}"),
| Ok(()) => write!(self, "Appservice unregistered."),
}
.await
{
| Err(e) => return Err!("Failed to unregister appservice: {e}"),
| Ok(()) => write!(self, "Appservice unregistered."),
}
.await
}
#[admin_command]
pub(super) async fn show_appservice_config(&self, appservice_identifier: String) -> Result {
match self
.services
.appservice
.get_registration(&appservice_identifier)
pub(super) async fn show_appservice_config(&self, appservice_identifier: String) -> Result {
match self
.services
.appservice
.get_registration(&appservice_identifier)
.await
{
| None => return Err!("Appservice does not exist."),
| Some(config) => {
let config_str = serde_saphyr::to_string(&config)?;
write!(self, "Config for {appservice_identifier}:\n\n```yaml\n{config_str}\n```")
},
}
.await
{
| None => return Err!("Appservice does not exist."),
| Some(config) => {
let config_str = serde_saphyr::to_string(&config)?;
write!(self, "Config for {appservice_identifier}:\n\n```yaml\n{config_str}\n```")
},
}
.await
}
#[admin_command]
pub(super) async fn list_registered(&self) -> Result {
self.services
.appservice
.iter_ids()
.collect()
.map(Ok)
.and_then(|appservices: Vec<_>| {
let len = appservices.len();
let list = appservices.join(", ");
write!(self, "Appservices ({len}): {list}")
})
.await
pub(super) async fn list_registered(&self) -> Result {
self.services
.appservice
.iter_ids()
.collect()
.map(Ok)
.and_then(|appservices: Vec<_>| {
let len = appservices.len();
let list = appservices.join(", ");
write!(self, "Appservices ({len}): {list}")
})
.await
}
}
src/admin/check/commands.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use futures::StreamExt;
use crate::Context;
impl Context {
impl Context<'_> {
pub(super) async fn check_all_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let users = self.services.users.stream().collect::<Vec<_>>().await;
src/admin/debug/commands.rs
+996 -1005
View File
File diff suppressed because it is too large. Load diff
src/admin/debug/tester.rs
+23 -28
View File
@@ -1,6 +1,6 @@
use conduwuit::{Err, Result};
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, clap::Subcommand)]
@@ -11,37 +11,32 @@ pub enum TesterCommand {
Timer,
}
#[rustfmt::skip]
#[admin_command]
async fn panic(&self) -> Result {
impl crate::Context<'_> {
#[rustfmt::skip]
async fn panic(&self) -> Result {
panic!("panicked")
}
panic!("panicked")
}
#[rustfmt::skip]
async fn failure(&self) -> Result {
Err!("failed")
}
#[rustfmt::skip]
#[admin_command]
async fn failure(&self) -> Result {
#[inline(never)]
#[rustfmt::skip]
async fn tester(&self) -> Result {
self.write_str("Ok").await
}
Err!("failed")
}
#[inline(never)]
#[rustfmt::skip]
async fn timer(&self) -> Result {
let started = std::time::Instant::now();
timed(self.body);
#[inline(never)]
#[rustfmt::skip]
#[admin_command]
async fn tester(&self) -> Result {
self.write_str("Ok").await
}
#[inline(never)]
#[rustfmt::skip]
#[admin_command]
async fn timer(&self) -> Result {
let started = std::time::Instant::now();
timed(self.body);
let elapsed = started.elapsed();
self.write_str(&format!("completed in {elapsed:#?}")).await
let elapsed = started.elapsed();
self.write_str(&format!("completed in {elapsed:#?}")).await
}
}
#[inline(never)]
src/admin/federation/commands.rs
+116 -112
View File
@@ -4,127 +4,131 @@ use conduwuit::{Err, Result, utils::response::LimitReadExt};
use futures::StreamExt;
use ruma::{OwnedRoomId, OwnedServerName, OwnedUserId};
use crate::{admin_command, get_room_info};
use crate::get_room_info;
#[admin_command]
pub(super) async fn disable_room(&self, room_id: OwnedRoomId) -> Result {
self.bail_restricted()?;
self.services.rooms.metadata.disable_room(&room_id, true);
self.write_str("Room disabled.").await
}
impl crate::Context<'_> {
pub(super) async fn disable_room(&self, room_id: OwnedRoomId) -> Result {
self.bail_restricted()?;
self.services.rooms.metadata.disable_room(&room_id, true);
self.write_str("Room disabled.").await
}
#[admin_command]
pub(super) async fn enable_room(&self, room_id: OwnedRoomId) -> Result {
self.bail_restricted()?;
self.services.rooms.metadata.disable_room(&room_id, false);
self.write_str("Room enabled.").await
}
pub(super) async fn enable_room(&self, room_id: OwnedRoomId) -> Result {
self.bail_restricted()?;
self.services.rooms.metadata.disable_room(&room_id, false);
self.write_str("Room enabled.").await
}
#[admin_command]
pub(super) async fn incoming_federation(&self) -> Result {
let msg = {
let map = self
pub(super) async fn incoming_federation(&self) -> Result {
let msg = {
let map = self
.services
.rooms
.event_handler
.federation_handletime
.read();
let mut msg = format!(
"Handling {} incoming PDUs across {} active transactions:\n",
map.len(),
self.services.transactions.txn_active_handle_count()
);
for (r, (e, i)) in map.iter() {
let elapsed = i.elapsed();
writeln!(
msg,
"{} {}: {}m{}s",
r,
e,
elapsed.as_secs() / 60,
elapsed.as_secs() % 60
)?;
}
msg
};
self.write_str(&msg).await
}
pub(super) async fn fetch_support_well_known(&self, server_name: OwnedServerName) -> Result {
let response = self
.services
.client
.default
.get(format!("https://{server_name}/.well-known/matrix/support"))
.send()
.await?;
let text = response
.limit_read_text(
self.services
.config
.max_request_size
.try_into()
.expect("u64 fits into usize"),
)
.await?;
if text.is_empty() {
return Err!("Response text/body is empty.");
}
if text.len() > 1500 {
return Err!(
"Response text/body is over 1500 characters, assuming no support well-known.",
);
}
let json: serde_json::Value = match serde_json::from_str(&text) {
| Ok(json) => json,
| Err(_) => {
return Err!("Response text/body is not valid JSON.",);
},
};
let pretty_json: String = match serde_json::to_string_pretty(&json) {
| Ok(json) => json,
| Err(_) => {
return Err!("Response text/body is not valid JSON.",);
},
};
self.write_str(&format!("Got JSON response:\n\n```json\n{pretty_json}\n```"))
.await
}
pub(super) async fn remote_user_in_rooms(&self, user_id: OwnedUserId) -> Result {
if user_id.server_name() == self.services.server.name {
return Err!(
"User belongs to our server, please use `list-joined-rooms` user admin command \
instead.",
);
}
let mut rooms: Vec<(OwnedRoomId, u64, String)> = self
.services
.rooms
.event_handler
.federation_handletime
.read();
.state_cache
.rooms_joined(&user_id)
.then(async |room_id| get_room_info(self.services, &room_id).await)
.collect()
.await;
let mut msg = format!(
"Handling {} incoming PDUs across {} active transactions:\n",
map.len(),
self.services.transactions.txn_active_handle_count()
);
for (r, (e, i)) in map.iter() {
let elapsed = i.elapsed();
writeln!(msg, "{} {}: {}m{}s", r, e, elapsed.as_secs() / 60, elapsed.as_secs() % 60)?;
if rooms.is_empty() {
return Err!("User is not in any rooms.");
}
msg
};
self.write_str(&msg).await
}
rooms.sort_by_key(|r| r.1);
rooms.reverse();
#[admin_command]
pub(super) async fn fetch_support_well_known(&self, server_name: OwnedServerName) -> Result {
let response = self
.services
.client
.default
.get(format!("https://{server_name}/.well-known/matrix/support"))
.send()
.await?;
let num = rooms.len();
let body = rooms
.iter()
.map(|(id, members, name)| format!("{id} | Members: {members} | Name: {name}"))
.collect::<Vec<_>>()
.join("\n");
let text = response
.limit_read_text(
self.services
.config
.max_request_size
.try_into()
.expect("u64 fits into usize"),
)
.await?;
if text.is_empty() {
return Err!("Response text/body is empty.");
}
if text.len() > 1500 {
return Err!(
"Response text/body is over 1500 characters, assuming no support well-known.",
);
}
let json: serde_json::Value = match serde_json::from_str(&text) {
| Ok(json) => json,
| Err(_) => {
return Err!("Response text/body is not valid JSON.",);
},
};
let pretty_json: String = match serde_json::to_string_pretty(&json) {
| Ok(json) => json,
| Err(_) => {
return Err!("Response text/body is not valid JSON.",);
},
};
self.write_str(&format!("Got JSON response:\n\n```json\n{pretty_json}\n```"))
.await
}
#[admin_command]
pub(super) async fn remote_user_in_rooms(&self, user_id: OwnedUserId) -> Result {
if user_id.server_name() == self.services.server.name {
return Err!(
"User belongs to our server, please use `list-joined-rooms` user admin command \
instead.",
);
}
let mut rooms: Vec<(OwnedRoomId, u64, String)> = self
.services
.rooms
.state_cache
.rooms_joined(&user_id)
.then(async |room_id| get_room_info(self.services, &room_id).await)
.collect()
.await;
if rooms.is_empty() {
return Err!("User is not in any rooms.");
self.write_str(&format!("Rooms {user_id} shares with us ({num}):\n```\n{body}\n```"))
.await
}
rooms.sort_by_key(|r| r.1);
rooms.reverse();
let num = rooms.len();
let body = rooms
.iter()
.map(|(id, members, name)| format!("{id} | Members: {members} | Name: {name}"))
.collect::<Vec<_>>()
.join("\n");
self.write_str(&format!("Rooms {user_id} shares with us ({num}):\n```\n{body}\n```"))
.await
}
src/admin/media/commands.rs
+336 -334
View File
@@ -9,401 +9,403 @@ use conduwuit_service::media::Dim;
use ruma::{OwnedEventId, OwnedMxcUri, OwnedServerName};
use service::media::mxc::Mxc;
use crate::{admin_command, utils::parse_local_user_id};
use crate::utils::parse_local_user_id;
#[admin_command]
pub(super) async fn delete(
&self,
mxc: Option<OwnedMxcUri>,
event_id: Option<OwnedEventId>,
) -> Result {
self.bail_restricted()?;
impl crate::Context<'_> {
pub(super) async fn delete(
&self,
mxc: Option<OwnedMxcUri>,
event_id: Option<OwnedEventId>,
) -> Result {
self.bail_restricted()?;
if event_id.is_some() && mxc.is_some() {
return Err!("Please specify either an MXC or an event ID, not both.",);
}
if event_id.is_some() && mxc.is_some() {
return Err!("Please specify either an MXC or an event ID, not both.",);
}
if let Some(mxc) = mxc {
trace!("Got MXC URL: {mxc}");
self.services
.media
.delete(&mxc.as_str().try_into()?)
.await?;
if let Some(mxc) = mxc {
trace!("Got MXC URL: {mxc}");
self.services
.media
.delete(&mxc.as_str().try_into()?)
.await?;
return self
.write_str("Deleted the MXC from our database and on our filesystem.")
.await;
}
return self
.write_str("Deleted the MXC from our database and on our filesystem.")
.await;
}
if let Some(event_id) = event_id {
trace!("Got event ID to delete media from: {event_id}");
if let Some(event_id) = event_id {
trace!("Got event ID to delete media from: {event_id}");
let mut mxc_urls = Vec::with_capacity(4);
let mut mxc_urls = Vec::with_capacity(4);
// parsing the PDU for any MXC URLs begins here
match self.services.rooms.timeline.get_pdu_json(&event_id).await {
| Ok(event_json) => {
if let Some(content_key) = event_json.get("content") {
debug!("Event ID has \"content\".");
let content_obj = content_key.as_object();
// parsing the PDU for any MXC URLs begins here
match self.services.rooms.timeline.get_pdu_json(&event_id).await {
| Ok(event_json) => {
if let Some(content_key) = event_json.get("content") {
debug!("Event ID has \"content\".");
let content_obj = content_key.as_object();
if let Some(content) = content_obj {
// 1. attempts to parse the "url" key
debug!("Attempting to go into \"url\" key for main media file");
if let Some(url) = content.get("url") {
debug!("Got a URL in the event ID {event_id}: {url}");
if let Some(content) = content_obj {
// 1. attempts to parse the "url" key
debug!("Attempting to go into \"url\" key for main media file");
if let Some(url) = content.get("url") {
debug!("Got a URL in the event ID {event_id}: {url}");
if url.to_string().starts_with("\"mxc://") {
debug!("Pushing URL {url} to list of MXCs to delete");
let final_url = url.to_string().replace('"', "");
mxc_urls.push(final_url);
} else {
info!(
"Found a URL in the event ID {event_id} but did not start \
with mxc://, ignoring"
);
}
}
// 2. attempts to parse the "info" key
debug!("Attempting to go into \"info\" key for thumbnails");
if let Some(info_key) = content.get("info") {
debug!("Event ID has \"info\".");
let info_obj = info_key.as_object();
if let Some(info) = info_obj {
if let Some(thumbnail_url) = info.get("thumbnail_url") {
debug!("Found a thumbnail_url in info key: {thumbnail_url}");
if thumbnail_url.to_string().starts_with("\"mxc://") {
debug!(
"Pushing thumbnail URL {thumbnail_url} to list of \
MXCs to delete"
);
let final_thumbnail_url =
thumbnail_url.to_string().replace('"', "");
mxc_urls.push(final_thumbnail_url);
} else {
info!(
"Found a thumbnail URL in the event ID {event_id} \
but did not start with mxc://, ignoring"
);
}
if url.to_string().starts_with("\"mxc://") {
debug!("Pushing URL {url} to list of MXCs to delete");
let final_url = url.to_string().replace('"', "");
mxc_urls.push(final_url);
} else {
info!(
"No \"thumbnail_url\" key in \"info\" key, assuming no \
thumbnails."
"Found a URL in the event ID {event_id} but did not \
start with mxc://, ignoring"
);
}
}
}
// 3. attempts to parse the "file" key
debug!("Attempting to go into \"file\" key");
if let Some(file_key) = content.get("file") {
debug!("Event ID has \"file\".");
let file_obj = file_key.as_object();
// 2. attempts to parse the "info" key
debug!("Attempting to go into \"info\" key for thumbnails");
if let Some(info_key) = content.get("info") {
debug!("Event ID has \"info\".");
let info_obj = info_key.as_object();
if let Some(file) = file_obj {
if let Some(url) = file.get("url") {
debug!("Found url in file key: {url}");
if let Some(info) = info_obj {
if let Some(thumbnail_url) = info.get("thumbnail_url") {
debug!(
"Found a thumbnail_url in info key: {thumbnail_url}"
);
if url.to_string().starts_with("\"mxc://") {
debug!("Pushing URL {url} to list of MXCs to delete");
let final_url = url.to_string().replace('"', "");
mxc_urls.push(final_url);
if thumbnail_url.to_string().starts_with("\"mxc://") {
debug!(
"Pushing thumbnail URL {thumbnail_url} to list \
of MXCs to delete"
);
let final_thumbnail_url =
thumbnail_url.to_string().replace('"', "");
mxc_urls.push(final_thumbnail_url);
} else {
info!(
"Found a thumbnail URL in the event ID \
{event_id} but did not start with mxc://, \
ignoring"
);
}
} else {
warn!(
"Found a URL in the event ID {event_id} but did not \
start with mxc://, ignoring"
info!(
"No \"thumbnail_url\" key in \"info\" key, assuming \
no thumbnails."
);
}
} else {
error!("No \"url\" key in \"file\" key.");
}
}
// 3. attempts to parse the "file" key
debug!("Attempting to go into \"file\" key");
if let Some(file_key) = content.get("file") {
debug!("Event ID has \"file\".");
let file_obj = file_key.as_object();
if let Some(file) = file_obj {
if let Some(url) = file.get("url") {
debug!("Found url in file key: {url}");
if url.to_string().starts_with("\"mxc://") {
debug!("Pushing URL {url} to list of MXCs to delete");
let final_url = url.to_string().replace('"', "");
mxc_urls.push(final_url);
} else {
warn!(
"Found a URL in the event ID {event_id} but did \
not start with mxc://, ignoring"
);
}
} else {
error!("No \"url\" key in \"file\" key.");
}
}
}
} else {
return Err!(
"Event ID does not have a \"content\" key or failed parsing the \
event ID JSON.",
);
}
} else {
return Err!(
"Event ID does not have a \"content\" key or failed parsing the \
event ID JSON.",
"Event ID does not have a \"content\" key, this is not a message or \
an event type that contains media.",
);
}
} else {
return Err!(
"Event ID does not have a \"content\" key, this is not a message or an \
event type that contains media.",
);
},
| _ => {
return Err!("Event ID does not exist or is not known to us.",);
},
}
if mxc_urls.is_empty() {
return Err!("Parsed event ID but found no MXC URLs.",);
}
let mut mxc_deletion_count: usize = 0;
for mxc_url in mxc_urls {
match self
.services
.media
.delete(&mxc_url.as_str().try_into()?)
.await
{
| Ok(()) => {
debug_info!(
"Successfully deleted {mxc_url} from filesystem and database"
);
mxc_deletion_count = mxc_deletion_count.saturating_add(1);
},
| Err(e) => {
debug_warn!(
"Failed to delete {mxc_url}, ignoring error and skipping: {e}"
);
continue;
},
}
},
| _ => {
return Err!("Event ID does not exist or is not known to us.",);
},
}
return self
.write_str(&format!(
"Deleted {mxc_deletion_count} total MXCs from our database and the \
filesystem from event ID {event_id}."
))
.await;
}
if mxc_urls.is_empty() {
return Err!("Parsed event ID but found no MXC URLs.",);
Err!(
"Please specify either an MXC using --mxc or an event ID using --event-id of the \
message containing an image. See --help for details."
)
}
pub(super) async fn delete_list(&self) -> Result {
self.bail_restricted()?;
if self.body.len() < 2
|| !self.body[0].trim().starts_with("```")
|| self.body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.",);
}
let mut failed_parsed_mxcs: usize = 0;
let mxc_list = self
.body
.to_vec()
.drain(1..self.body.len().checked_sub(1).unwrap())
.filter_map(|mxc_s| {
mxc_s
.try_into()
.inspect_err(|e| {
debug_warn!("Failed to parse user-provided MXC URI: {e}");
failed_parsed_mxcs = failed_parsed_mxcs.saturating_add(1);
})
.ok()
})
.collect::<Vec<Mxc<'_>>>();
let mut mxc_deletion_count: usize = 0;
for mxc_url in mxc_urls {
match self
.services
.media
.delete(&mxc_url.as_str().try_into()?)
.await
{
for mxc in &mxc_list {
trace!(%failed_parsed_mxcs, %mxc_deletion_count, "Deleting MXC {mxc} in bulk");
match self.services.media.delete(mxc).await {
| Ok(()) => {
debug_info!("Successfully deleted {mxc_url} from filesystem and database");
debug_info!("Successfully deleted {mxc} from filesystem and database");
mxc_deletion_count = mxc_deletion_count.saturating_add(1);
},
| Err(e) => {
debug_warn!("Failed to delete {mxc_url}, ignoring error and skipping: {e}");
debug_warn!("Failed to delete {mxc}, ignoring error and skipping: {e}");
continue;
},
}
}
return self
.write_str(&format!(
"Deleted {mxc_deletion_count} total MXCs from our database and the filesystem \
from event ID {event_id}."
))
.await;
self.write_str(&format!(
"Finished bulk MXC deletion, deleted {mxc_deletion_count} total MXCs from our \
database and the filesystem. {failed_parsed_mxcs} MXCs failed to be parsed from \
the database.",
))
.await
}
Err!(
"Please specify either an MXC using --mxc or an event ID using --event-id of the \
message containing an image. See --help for details."
)
}
pub(super) async fn delete_past_remote_media(
&self,
duration: String,
before: bool,
after: bool,
yes_i_want_to_delete_local_media: bool,
) -> Result {
self.bail_restricted()?;
#[admin_command]
pub(super) async fn delete_list(&self) -> Result {
self.bail_restricted()?;
if self.body.len() < 2
|| !self.body[0].trim().starts_with("```")
|| self.body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.",);
}
let mut failed_parsed_mxcs: usize = 0;
let mxc_list = self
.body
.to_vec()
.drain(1..self.body.len().checked_sub(1).unwrap())
.filter_map(|mxc_s| {
mxc_s
.try_into()
.inspect_err(|e| {
debug_warn!("Failed to parse user-provided MXC URI: {e}");
failed_parsed_mxcs = failed_parsed_mxcs.saturating_add(1);
})
.ok()
})
.collect::<Vec<Mxc<'_>>>();
let mut mxc_deletion_count: usize = 0;
for mxc in &mxc_list {
trace!(%failed_parsed_mxcs, %mxc_deletion_count, "Deleting MXC {mxc} in bulk");
match self.services.media.delete(mxc).await {
| Ok(()) => {
debug_info!("Successfully deleted {mxc} from filesystem and database");
mxc_deletion_count = mxc_deletion_count.saturating_add(1);
},
| Err(e) => {
debug_warn!("Failed to delete {mxc}, ignoring error and skipping: {e}");
continue;
},
if before && after {
return Err!("Please only pick one argument, --before or --after.",);
}
}
assert!(!(before && after), "--before and --after should not be specified together");
self.write_str(&format!(
"Finished bulk MXC deletion, deleted {mxc_deletion_count} total MXCs from our database \
and the filesystem. {failed_parsed_mxcs} MXCs failed to be parsed from the database.",
))
.await
}
#[admin_command]
pub(super) async fn delete_past_remote_media(
&self,
duration: String,
before: bool,
after: bool,
yes_i_want_to_delete_local_media: bool,
) -> Result {
self.bail_restricted()?;
if before && after {
return Err!("Please only pick one argument, --before or --after.",);
}
assert!(!(before && after), "--before and --after should not be specified together");
let direction = if after {
TimeDirection::After
} else {
TimeDirection::Before
};
let time_boundary = parse_timepoint_ago(&duration)?;
let deleted_count = self
.services
.media
.delete_all_media_within_timeframe(
time_boundary,
direction,
yes_i_want_to_delete_local_media,
)
.await?;
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
#[admin_command]
pub(super) async fn delete_all_from_user(&self, username: String) -> Result {
let user_id = parse_local_user_id(self.services, &username)?;
let deleted_count = self.services.media.delete_from_user(&user_id).await?;
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
#[admin_command]
pub(super) async fn delete_all_from_server(
&self,
server_name: OwnedServerName,
yes_i_want_to_delete_local_media: bool,
) -> Result {
self.bail_restricted()?;
if server_name == self.services.globals.server_name() && !yes_i_want_to_delete_local_media {
return Err!("This command only works for remote media by default.",);
}
let Ok(all_mxcs) = self
.services
.media
.get_all_mxcs()
.await
.inspect_err(|e| error!("Failed to get MXC URIs from our database: {e}"))
else {
return Err!("Failed to get MXC URIs from our database",);
};
let mut deleted_count: usize = 0;
for mxc in all_mxcs {
let Ok(mxc_server_name) = mxc.server_name().inspect_err(|e| {
debug_warn!(
"Failed to parse MXC {mxc} server name from database, ignoring error and \
skipping: {e}"
);
}) else {
continue;
let direction = if after {
TimeDirection::After
} else {
TimeDirection::Before
};
if mxc_server_name != server_name
|| (self.services.globals.server_is_ours(mxc_server_name)
&& !yes_i_want_to_delete_local_media)
let time_boundary = parse_timepoint_ago(&duration)?;
let deleted_count = self
.services
.media
.delete_all_media_within_timeframe(
time_boundary,
direction,
yes_i_want_to_delete_local_media,
)
.await?;
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
pub(super) async fn delete_all_from_user(&self, username: String) -> Result {
let user_id = parse_local_user_id(self.services, &username)?;
let deleted_count = self.services.media.delete_from_user(&user_id).await?;
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
pub(super) async fn delete_all_from_server(
&self,
server_name: OwnedServerName,
yes_i_want_to_delete_local_media: bool,
) -> Result {
self.bail_restricted()?;
if server_name == self.services.globals.server_name() && !yes_i_want_to_delete_local_media
{
trace!("skipping MXC URI {mxc}");
continue;
return Err!("This command only works for remote media by default.",);
}
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let Ok(all_mxcs) = self
.services
.media
.get_all_mxcs()
.await
.inspect_err(|e| error!("Failed to get MXC URIs from our database: {e}"))
else {
return Err!("Failed to get MXC URIs from our database",);
};
match self.services.media.delete(&mxc).await {
| Ok(()) => {
deleted_count = deleted_count.saturating_add(1);
},
| Err(e) => {
debug_warn!("Failed to delete {mxc}, ignoring error and skipping: {e}");
let mut deleted_count: usize = 0;
for mxc in all_mxcs {
let Ok(mxc_server_name) = mxc.server_name().inspect_err(|e| {
debug_warn!(
"Failed to parse MXC {mxc} server name from database, ignoring error and \
skipping: {e}"
);
}) else {
continue;
},
};
if mxc_server_name != server_name
|| (self.services.globals.server_is_ours(mxc_server_name)
&& !yes_i_want_to_delete_local_media)
{
trace!("skipping MXC URI {mxc}");
continue;
}
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
match self.services.media.delete(&mxc).await {
| Ok(()) => {
deleted_count = deleted_count.saturating_add(1);
},
| Err(e) => {
debug_warn!("Failed to delete {mxc}, ignoring error and skipping: {e}");
continue;
},
}
}
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
self.write_str(&format!("Deleted {deleted_count} total files."))
.await
}
pub(super) async fn get_file_info(&self, mxc: OwnedMxcUri) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let metadata = self.services.media.get_metadata(&mxc).await;
#[admin_command]
pub(super) async fn get_file_info(&self, mxc: OwnedMxcUri) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let metadata = self.services.media.get_metadata(&mxc).await;
self.write_str(&format!("```\n{metadata:#?}\n```")).await
}
#[admin_command]
pub(super) async fn get_remote_file(
&self,
mxc: OwnedMxcUri,
server: Option<OwnedServerName>,
timeout: u32,
) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let timeout = Duration::from_millis(timeout.into());
let mut result = self
.services
.media
.fetch_remote_content(&mxc, None, server.as_deref(), timeout)
.await?;
// Grab the length of the content before clearing it to not flood the output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
self.write_str(&format!("```\n{result:#?}\nreceived {len} bytes for file content.\n```"))
.await
}
#[admin_command]
pub(super) async fn get_remote_thumbnail(
&self,
mxc: OwnedMxcUri,
server: Option<OwnedServerName>,
timeout: u32,
width: u32,
height: u32,
) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let timeout = Duration::from_millis(timeout.into());
let dim = Dim::new(width, height, None);
let mut result = self
.services
.media
.fetch_remote_thumbnail(&mxc, None, server.as_deref(), timeout, &dim)
.await?;
// Grab the length of the content before clearing it to not flood the output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
self.write_str(&format!("```\n{result:#?}\nreceived {len} bytes for file content.\n```"))
.await
}
#[admin_command]
pub(super) async fn delete_url_preview(&self, url: Option<String>, all: bool) -> Result {
if all {
self.services.media.clear_url_previews().await;
return self.write_str("Deleted all cached URL previews.").await;
self.write_str(&format!("```\n{metadata:#?}\n```")).await
}
let url = url.expect("clap enforces url is required unless --all");
pub(super) async fn get_remote_file(
&self,
mxc: OwnedMxcUri,
server: Option<OwnedServerName>,
timeout: u32,
) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let timeout = Duration::from_millis(timeout.into());
let mut result = self
.services
.media
.fetch_remote_content(&mxc, None, server.as_deref(), timeout)
.await?;
self.services.media.remove_url_preview(&url).await?;
// Grab the length of the content before clearing it to not flood the output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
self.write_str(&format!("Deleted cached URL preview for: {url}"))
.await
self.write_str(&format!("```\n{result:#?}\nreceived {len} bytes for file content.\n```"))
.await
}
pub(super) async fn get_remote_thumbnail(
&self,
mxc: OwnedMxcUri,
server: Option<OwnedServerName>,
timeout: u32,
width: u32,
height: u32,
) -> Result {
let mxc: Mxc<'_> = mxc.as_str().try_into()?;
let timeout = Duration::from_millis(timeout.into());
let dim = Dim::new(width, height, None);
let mut result = self
.services
.media
.fetch_remote_thumbnail(&mxc, None, server.as_deref(), timeout, &dim)
.await?;
// Grab the length of the content before clearing it to not flood the output
let len = result.content.as_ref().expect("content").len();
result.content.as_mut().expect("content").clear();
self.write_str(&format!("```\n{result:#?}\nreceived {len} bytes for file content.\n```"))
.await
}
pub(super) async fn delete_url_preview(&self, url: Option<String>, all: bool) -> Result {
if all {
self.services.media.clear_url_previews().await;
return self.write_str("Deleted all cached URL previews.").await;
}
let url = url.expect("clap enforces url is required unless --all");
self.services.media.remove_url_preview(&url).await?;
self.write_str(&format!("Deleted cached URL preview for: {url}"))
.await
}
}
src/admin/mod.rs
+1 -1
View File
@@ -26,7 +26,7 @@ extern crate conduwuit_api as api;
extern crate conduwuit_core as conduwuit;
extern crate conduwuit_service as service;
pub(crate) use conduwuit_macros::{admin_command, admin_command_dispatch};
pub(crate) use conduwuit_macros::admin_command_dispatch;
pub(crate) use crate::{context::Context, utils::get_room_info};
src/admin/query/account_data.rs
+43 -43
View File
@@ -4,7 +4,7 @@ use conduwuit_database::Deserialized as _;
use futures::StreamExt;
use ruma::{OwnedRoomId, OwnedUserId, exports::serde::Serialize};
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -31,50 +31,50 @@ pub enum AccountDataCommand {
},
}
#[admin_command]
async fn changes_since(
&self,
user_id: OwnedUserId,
since: u64,
room_id: Option<OwnedRoomId>,
) -> Result {
let timer = tokio::time::Instant::now();
let results: Vec<_> = self
.services
.account_data
.changes_since(room_id.as_deref(), &user_id, Some(since), None)
.collect()
.await;
let query_time = timer.elapsed();
impl crate::Context<'_> {
async fn changes_since(
&self,
user_id: OwnedUserId,
since: u64,
room_id: Option<OwnedRoomId>,
) -> Result {
let timer = tokio::time::Instant::now();
let results: Vec<_> = self
.services
.account_data
.changes_since(room_id.as_deref(), &user_id, Some(since), None)
.collect()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"))
.await
}
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"))
.await
}
#[admin_command]
async fn account_data_get(
&self,
user_id: OwnedUserId,
kind: String,
room_id: Option<OwnedRoomId>,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.account_data
.get_raw(room_id.as_deref(), &user_id, &kind)
.await;
let query_time = timer.elapsed();
async fn account_data_get(
&self,
user_id: OwnedUserId,
kind: String,
room_id: Option<OwnedRoomId>,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.account_data
.get_raw(room_id.as_deref(), &user_id, &kind)
.await;
let query_time = timer.elapsed();
let json = serde_json::to_string_pretty(&match room_id {
| None => result
.deserialized::<ruma::serde::Raw<ruma::events::AnyGlobalAccountDataEvent>>()?
.serialize(serde_json::value::Serializer)?,
| Some(_) => result
.deserialized::<ruma::serde::Raw<ruma::events::AnyRoomAccountDataEvent>>()?
.serialize(serde_json::value::Serializer)?,
})?;
let json = serde_json::to_string_pretty(&match room_id {
| None => result
.deserialized::<ruma::serde::Raw<ruma::events::AnyGlobalAccountDataEvent>>()?
.serialize(serde_json::value::Serializer)?,
| Some(_) => result
.deserialized::<ruma::serde::Raw<ruma::events::AnyRoomAccountDataEvent>>()?
.serialize(serde_json::value::Serializer)?,
})?;
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{json}\n```"))
.await
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{json}\n```"))
.await
}
}
src/admin/query/raw.rs
+246 -238
View File
@@ -13,7 +13,7 @@ use conduwuit_service::Services;
use futures::{FutureExt, Stream, StreamExt, TryStreamExt};
use tokio::time::Instant;
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -159,279 +159,287 @@ pub enum RawCommand {
},
}
#[admin_command]
pub(super) async fn compact(
&self,
map: Option<Vec<String>>,
start: Option<String>,
stop: Option<String>,
from: Option<usize>,
into: Option<usize>,
parallelism: Option<usize>,
exhaustive: bool,
) -> Result {
use conduwuit_database::compact::Options;
impl crate::Context<'_> {
pub(super) async fn compact(
&self,
map: Option<Vec<String>>,
start: Option<String>,
stop: Option<String>,
from: Option<usize>,
into: Option<usize>,
parallelism: Option<usize>,
exhaustive: bool,
) -> Result {
use conduwuit_database::compact::Options;
let default_all_maps: Option<_> = map.is_none().then(|| {
self.services
.db
.keys()
.map(Deref::deref)
.map(ToOwned::to_owned)
});
let default_all_maps: Option<_> = map.is_none().then(|| {
self.services
.db
.keys()
.map(Deref::deref)
.map(ToOwned::to_owned)
});
let maps: Vec<_> = map
.unwrap_or_default()
.into_iter()
.chain(default_all_maps.into_iter().flatten())
.map(|map| self.services.db.get(&map))
.filter_map(Result::ok)
.cloned()
.collect();
let maps: Vec<_> = map
.unwrap_or_default()
.into_iter()
.chain(default_all_maps.into_iter().flatten())
.map(|map| self.services.db.get(&map))
.filter_map(Result::ok)
.cloned()
.collect();
if maps.is_empty() {
return Err!("--map argument invalid. not found in database");
if maps.is_empty() {
return Err!("--map argument invalid. not found in database");
}
let range = (
start.as_ref().map(String::as_bytes).map(Into::into),
stop.as_ref().map(String::as_bytes).map(Into::into),
);
let options = Options {
range,
level: (from, into),
exclusive: parallelism.is_some_and(is_zero!()),
exhaustive,
};
let runtime = self.services.server.runtime().clone();
let parallelism = parallelism.unwrap_or(1);
let results = maps
.into_iter()
.try_stream::<conduwuit::Error>()
.paralleln_and_then(runtime, parallelism, move |map| {
map.compact_blocking(options.clone())?;
Ok(map.name().to_owned())
})
.collect::<Vec<_>>();
let timer = Instant::now();
let results = results.await;
let query_time = timer.elapsed();
self.write_str(&format!("Jobs completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"))
.await
}
let range = (
start.as_ref().map(String::as_bytes).map(Into::into),
stop.as_ref().map(String::as_bytes).map(Into::into),
);
pub(super) async fn raw_count(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let options = Options {
range,
level: (from, into),
exclusive: parallelism.is_some_and(is_zero!()),
exhaustive,
};
let timer = Instant::now();
let count = with_maps_or(map.as_deref(), self.services)
.then(|map| map.raw_count_prefix(&prefix))
.ready_fold(0_usize, usize::saturating_add)
.await;
let runtime = self.services.server.runtime().clone();
let parallelism = parallelism.unwrap_or(1);
let results = maps
.into_iter()
.try_stream::<conduwuit::Error>()
.paralleln_and_then(runtime, parallelism, move |map| {
map.compact_blocking(options.clone())?;
Ok(map.name().to_owned())
})
.collect::<Vec<_>>();
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{count:#?}\n```"))
.await
}
let timer = Instant::now();
let results = results.await;
let query_time = timer.elapsed();
self.write_str(&format!("Jobs completed in {query_time:?}:\n\n```rs\n{results:#?}\n```"))
.await
}
pub(super) async fn raw_keys(&self, map: String, prefix: Option<String>) -> Result {
writeln!(self, "```").boxed().await?;
#[admin_command]
pub(super) async fn raw_count(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let map = self.services.db.get(map.as_str())?;
let timer = Instant::now();
prefix
.as_deref()
.map_or_else(|| map.raw_keys().boxed(), |prefix| map.raw_keys_prefix(prefix).boxed())
.map_ok(String::from_utf8_lossy)
.try_for_each(|str| writeln!(self, "{str:?}"))
.boxed()
.await?;
let timer = Instant::now();
let count = with_maps_or(map.as_deref(), self.services)
.then(|map| map.raw_count_prefix(&prefix))
.ready_fold(0_usize, usize::saturating_add)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{count:#?}\n```"))
.await
}
pub(super) async fn raw_keys_sizes(
&self,
map: Option<String>,
prefix: Option<String>,
) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
#[admin_command]
pub(super) async fn raw_keys(&self, map: String, prefix: Option<String>) -> Result {
writeln!(self, "```").boxed().await?;
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_keys_prefix(&prefix))
.flatten()
.ignore_err()
.map(<[u8]>::len)
.ready_fold_default(|mut map: BTreeMap<_, usize>, len| {
let entry = map.entry(len).or_default();
*entry = entry.saturating_add(1);
map
})
.await;
let map = self.services.db.get(map.as_str())?;
let timer = Instant::now();
prefix
.as_deref()
.map_or_else(|| map.raw_keys().boxed(), |prefix| map.raw_keys_prefix(prefix).boxed())
.map_ok(String::from_utf8_lossy)
.try_for_each(|str| writeln!(self, "{str:?}"))
.boxed()
.await?;
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_keys_total(
&self,
map: Option<String>,
prefix: Option<String>,
) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
#[admin_command]
pub(super) async fn raw_keys_sizes(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_keys_prefix(&prefix))
.flatten()
.ignore_err()
.map(<[u8]>::len)
.ready_fold_default(|acc: usize, len| acc.saturating_add(len))
.await;
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_keys_prefix(&prefix))
.flatten()
.ignore_err()
.map(<[u8]>::len)
.ready_fold_default(|mut map: BTreeMap<_, usize>, len| {
let entry = map.entry(len).or_default();
*entry = entry.saturating_add(1);
map
})
.await;
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_vals_sizes(
&self,
map: Option<String>,
prefix: Option<String>,
) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
#[admin_command]
pub(super) async fn raw_keys_total(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_stream_prefix(&prefix))
.flatten()
.ignore_err()
.map(at!(1))
.map(<[u8]>::len)
.ready_fold_default(|mut map: BTreeMap<_, usize>, len| {
let entry = map.entry(len).or_default();
*entry = entry.saturating_add(1);
map
})
.await;
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_keys_prefix(&prefix))
.flatten()
.ignore_err()
.map(<[u8]>::len)
.ready_fold_default(|acc: usize, len| acc.saturating_add(len))
.await;
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_vals_total(
&self,
map: Option<String>,
prefix: Option<String>,
) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
#[admin_command]
pub(super) async fn raw_vals_sizes(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_stream_prefix(&prefix))
.flatten()
.ignore_err()
.map(at!(1))
.map(<[u8]>::len)
.ready_fold_default(|acc: usize, len| acc.saturating_add(len))
.await;
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_stream_prefix(&prefix))
.flatten()
.ignore_err()
.map(at!(1))
.map(<[u8]>::len)
.ready_fold_default(|mut map: BTreeMap<_, usize>, len| {
let entry = map.entry(len).or_default();
*entry = entry.saturating_add(1);
map
})
.await;
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_iter(&self, map: String, prefix: Option<String>) -> Result {
writeln!(self, "```").await?;
#[admin_command]
pub(super) async fn raw_vals_total(&self, map: Option<String>, prefix: Option<String>) -> Result {
let prefix = prefix.as_deref().unwrap_or(EMPTY);
let map = self.services.db.get(&map)?;
let timer = Instant::now();
prefix
.as_deref()
.map_or_else(
|| map.raw_stream().boxed(),
|prefix| map.raw_stream_prefix(prefix).boxed(),
)
.map_ok(apply!(2, String::from_utf8_lossy))
.map_ok(apply!(2, Cow::into_owned))
.try_for_each(|keyval| writeln!(self, "{keyval:?}"))
.boxed()
.await?;
let timer = Instant::now();
let result = with_maps_or(map.as_deref(), self.services)
.map(|map| map.raw_stream_prefix(&prefix))
.flatten()
.ignore_err()
.map(at!(1))
.map(<[u8]>::len)
.ready_fold_default(|acc: usize, len| acc.saturating_add(len))
.await;
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("```\n{result:#?}\n\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_keys_from(
&self,
map: String,
start: String,
limit: Option<usize>,
) -> Result {
writeln!(self, "```").await?;
#[admin_command]
pub(super) async fn raw_iter(&self, map: String, prefix: Option<String>) -> Result {
writeln!(self, "```").await?;
let map = self.services.db.get(&map)?;
let timer = Instant::now();
map.raw_keys_from(&start)
.map_ok(String::from_utf8_lossy)
.take(limit.unwrap_or(usize::MAX))
.try_for_each(|str| writeln!(self, "{str:?}"))
.boxed()
.await?;
let map = self.services.db.get(&map)?;
let timer = Instant::now();
prefix
.as_deref()
.map_or_else(|| map.raw_stream().boxed(), |prefix| map.raw_stream_prefix(prefix).boxed())
.map_ok(apply!(2, String::from_utf8_lossy))
.map_ok(apply!(2, Cow::into_owned))
.try_for_each(|keyval| writeln!(self, "{keyval:?}"))
.boxed()
.await?;
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
pub(super) async fn raw_iter_from(
&self,
map: String,
start: String,
limit: Option<usize>,
) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
let result = map
.raw_stream_from(&start)
.map_ok(apply!(2, String::from_utf8_lossy))
.map_ok(apply!(2, Cow::into_owned))
.take(limit.unwrap_or(usize::MAX))
.try_collect::<Vec<(String, String)>>()
.await?;
#[admin_command]
pub(super) async fn raw_keys_from(
&self,
map: String,
start: String,
limit: Option<usize>,
) -> Result {
writeln!(self, "```").await?;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
let map = self.services.db.get(&map)?;
let timer = Instant::now();
map.raw_keys_from(&start)
.map_ok(String::from_utf8_lossy)
.take(limit.unwrap_or(usize::MAX))
.try_for_each(|str| writeln!(self, "{str:?}"))
.boxed()
.await?;
pub(super) async fn raw_del(&self, map: String, key: String) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
map.remove(&key);
let query_time = timer.elapsed();
self.write_str(&format!("\n```\n\nQuery completed in {query_time:?}"))
.await
}
let query_time = timer.elapsed();
self.write_str(&format!("Operation completed in {query_time:?}"))
.await
}
#[admin_command]
pub(super) async fn raw_iter_from(
&self,
map: String,
start: String,
limit: Option<usize>,
) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
let result = map
.raw_stream_from(&start)
.map_ok(apply!(2, String::from_utf8_lossy))
.map_ok(apply!(2, Cow::into_owned))
.take(limit.unwrap_or(usize::MAX))
.try_collect::<Vec<(String, String)>>()
.await?;
pub(super) async fn raw_get(&self, map: String, key: String) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
let handle = map.get(&key).await?;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
let query_time = timer.elapsed();
let result = String::from_utf8_lossy(&handle);
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:?}\n```"))
.await
}
#[admin_command]
pub(super) async fn raw_del(&self, map: String, key: String) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
map.remove(&key);
pub(super) async fn raw_maps(&self) -> Result {
let list: Vec<_> = self.services.db.iter().map(at!(0)).copied().collect();
let query_time = timer.elapsed();
self.write_str(&format!("Operation completed in {query_time:?}"))
.await
}
#[admin_command]
pub(super) async fn raw_get(&self, map: String, key: String) -> Result {
let map = self.services.db.get(&map)?;
let timer = Instant::now();
let handle = map.get(&key).await?;
let query_time = timer.elapsed();
let result = String::from_utf8_lossy(&handle);
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:?}\n```"))
.await
}
#[admin_command]
pub(super) async fn raw_maps(&self) -> Result {
let list: Vec<_> = self.services.db.iter().map(at!(0)).copied().collect();
self.write_str(&format!("{list:#?}")).await
self.write_str(&format!("{list:#?}")).await
}
}
fn with_maps_or<'a>(
src/admin/query/resolver.rs
+48 -47
View File
@@ -3,7 +3,7 @@ use conduwuit::{Err, Result, utils::time};
use futures::StreamExt;
use ruma::OwnedServerName;
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -39,67 +39,68 @@ pub enum ResolverCommand {
},
}
#[admin_command]
async fn destinations_cache(&self, server_name: Option<OwnedServerName>) -> Result {
use service::resolver::cache::CachedDest;
impl crate::Context<'_> {
async fn destinations_cache(&self, server_name: Option<OwnedServerName>) -> Result {
use service::resolver::cache::CachedDest;
writeln!(self, "| Server Name | Destination | Hostname | Expires |").await?;
writeln!(self, "| ----------- | ----------- | -------- | ------- |").await?;
writeln!(self, "| Server Name | Destination | Hostname | Expires |").await?;
writeln!(self, "| ----------- | ----------- | -------- | ------- |").await?;
let mut destinations = self.services.resolver.cache.destinations().boxed();
let mut destinations = self.services.resolver.cache.destinations().boxed();
while let Some((name, CachedDest { dest, host, expire })) = destinations.next().await {
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
while let Some((name, CachedDest { dest, host, expire })) = destinations.next().await {
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
}
}
let expire = time::format(expire, "%+");
self.write_str(&format!("| {name} | {dest} | {host} | {expire} |\n"))
.await?;
}
let expire = time::format(expire, "%+");
self.write_str(&format!("| {name} | {dest} | {host} | {expire} |\n"))
.await?;
Ok(())
}
Ok(())
}
async fn overrides_cache(&self, server_name: Option<String>) -> Result {
use service::resolver::cache::CachedOverride;
#[admin_command]
async fn overrides_cache(&self, server_name: Option<String>) -> Result {
use service::resolver::cache::CachedOverride;
writeln!(self, "| Server Name | IP | Port | Expires | Overriding |").await?;
writeln!(self, "| ----------- | --- | ----:| ------- | ---------- |").await?;
writeln!(self, "| Server Name | IP | Port | Expires | Overriding |").await?;
writeln!(self, "| ----------- | --- | ----:| ------- | ---------- |").await?;
let mut overrides = self.services.resolver.cache.overrides().boxed();
let mut overrides = self.services.resolver.cache.overrides().boxed();
while let Some((name, CachedOverride { ips, port, expire, overriding })) =
overrides.next().await
{
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
while let Some((name, CachedOverride { ips, port, expire, overriding })) =
overrides.next().await
{
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
}
}
let expire = time::format(expire, "%+");
self.write_str(&format!(
"| {name} | {ips:?} | {port} | {expire} | {overriding:?} |\n"
))
.await?;
}
let expire = time::format(expire, "%+");
self.write_str(&format!("| {name} | {ips:?} | {port} | {expire} | {overriding:?} |\n"))
.await?;
Ok(())
}
Ok(())
}
#[admin_command]
async fn flush_cache(&self, name: Option<OwnedServerName>, all: bool) -> Result {
if all {
self.services.resolver.cache.clear().await;
writeln!(self, "Resolver caches cleared!").await
} else if let Some(name) = name {
self.services.resolver.cache.del_destination(&name);
self.services.resolver.cache.del_override(&name);
self.write_str(&format!("Cleared {name} from resolver caches!"))
.await
} else {
Err!("Missing name. Supply a name or use --all to flush the whole cache.")
async fn flush_cache(&self, name: Option<OwnedServerName>, all: bool) -> Result {
if all {
self.services.resolver.cache.clear().await;
writeln!(self, "Resolver caches cleared!").await
} else if let Some(name) = name {
self.services.resolver.cache.del_destination(&name);
self.services.resolver.cache.del_override(&name);
self.write_str(&format!("Cleared {name} from resolver caches!"))
.await
} else {
Err!("Missing name. Supply a name or use --all to flush the whole cache.")
}
}
}
src/admin/query/room_timeline.rs
+30 -30
View File
@@ -3,7 +3,7 @@ use conduwuit::{PduCount, Result, utils::stream::TryTools};
use futures::TryStreamExt;
use ruma::OwnedRoomOrAliasId;
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -23,39 +23,39 @@ pub enum RoomTimelineCommand {
},
}
#[admin_command]
pub(super) async fn last(&self, room_id: OwnedRoomOrAliasId) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
impl crate::Context<'_> {
pub(super) async fn last(&self, room_id: OwnedRoomOrAliasId) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
let result = self
.services
.rooms
.timeline
.last_timeline_count(&room_id)
.await?;
let result = self
.services
.rooms
.timeline
.last_timeline_count(&room_id)
.await?;
self.write_str(&format!("{result:#?}")).await
}
self.write_str(&format!("{result:#?}")).await
}
#[admin_command]
pub(super) async fn pdus(
&self,
room_id: OwnedRoomOrAliasId,
from: Option<String>,
limit: Option<usize>,
) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
pub(super) async fn pdus(
&self,
room_id: OwnedRoomOrAliasId,
from: Option<String>,
limit: Option<usize>,
) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
let from: Option<PduCount> = from.as_deref().map(str::parse).transpose()?;
let from: Option<PduCount> = from.as_deref().map(str::parse).transpose()?;
let result: Vec<_> = self
.services
.rooms
.timeline
.pdus_rev(&room_id, from)
.try_take(limit.unwrap_or(3))
.try_collect()
.await?;
let result: Vec<_> = self
.services
.rooms
.timeline
.pdus_rev(&room_id, from)
.try_take(limit.unwrap_or(3))
.try_collect()
.await?;
self.write_str(&format!("```\n{result:#?}\n```")).await
self.write_str(&format!("```\n{result:#?}\n```")).await
}
}
src/admin/query/short.rs
+16 -16
View File
@@ -2,7 +2,7 @@ use clap::Subcommand;
use conduwuit::Result;
use ruma::{OwnedEventId, OwnedRoomOrAliasId};
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -17,23 +17,23 @@ pub enum ShortCommand {
},
}
#[admin_command]
pub(super) async fn short_event_id(&self, event_id: OwnedEventId) -> Result {
let shortid = self
.services
.rooms
.short
.get_shorteventid(&event_id)
.await?;
impl crate::Context<'_> {
pub(super) async fn short_event_id(&self, event_id: OwnedEventId) -> Result {
let shortid = self
.services
.rooms
.short
.get_shorteventid(&event_id)
.await?;
self.write_str(&format!("{shortid:#?}")).await
}
self.write_str(&format!("{shortid:#?}")).await
}
#[admin_command]
pub(super) async fn short_room_id(&self, room_id: OwnedRoomOrAliasId) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
pub(super) async fn short_room_id(&self, room_id: OwnedRoomOrAliasId) -> Result {
let room_id = self.services.rooms.alias.resolve(&room_id).await?;
let shortid = self.services.rooms.short.get_shortroomid(&room_id).await?;
let shortid = self.services.rooms.short.get_shortroomid(&room_id).await?;
self.write_str(&format!("{shortid:#?}")).await
self.write_str(&format!("{shortid:#?}")).await
}
}
src/admin/query/users.rs
+255 -259
View File
@@ -3,7 +3,7 @@ use conduwuit::Result;
use futures::stream::StreamExt;
use ruma::{OwnedDeviceId, OwnedRoomId, OwnedUserId};
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -92,263 +92,259 @@ pub enum UsersCommand {
},
}
#[admin_command]
async fn get_shared_rooms(&self, user_a: OwnedUserId, user_b: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<_> = self
.services
.rooms
.state_cache
.get_shared_rooms(&user_a, &user_b)
.collect()
.await;
let query_time = timer.elapsed();
impl crate::Context<'_> {
async fn get_shared_rooms(&self, user_a: OwnedUserId, user_b: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<_> = self
.services
.rooms
.state_cache
.get_shared_rooms(&user_a, &user_b)
.collect()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_backup_session(
&self,
user_id: OwnedUserId,
version: String,
room_id: OwnedRoomId,
session_id: String,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_session(&user_id, &version, &room_id, &session_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_room_backups(
&self,
user_id: OwnedUserId,
version: String,
room_id: OwnedRoomId,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_room(&user_id, &version, &room_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_all_backups(&self, user_id: OwnedUserId, version: String) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.key_backups.get_all(&user_id, &version).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_backup_algorithm(&self, user_id: OwnedUserId, version: String) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_backup(&user_id, &version)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_latest_backup_version(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_latest_backup_version(&user_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_latest_backup(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.key_backups.get_latest_backup(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn iter_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<OwnedUserId> = self.services.users.stream().map(Into::into).collect().await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn iter_users2(&self) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<_> = self.services.users.stream().collect().await;
let result: Vec<_> = result
.into_iter()
.map(|user_id| String::from_utf8_lossy(user_id.as_bytes()).into_owned())
.collect();
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:?}\n```"))
.await
}
#[admin_command]
async fn count_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.users.count().await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn list_devices(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let devices = self
.services
.users
.all_device_ids(&user_id)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{devices:#?}\n```"))
.await
}
#[admin_command]
async fn list_devices_metadata(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let devices = self
.services
.users
.all_devices_metadata(&user_id)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{devices:#?}\n```"))
.await
}
#[admin_command]
async fn get_device_metadata(&self, user_id: OwnedUserId, device_id: OwnedDeviceId) -> Result {
let timer = tokio::time::Instant::now();
let device = self
.services
.users
.get_device_metadata(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{device:#?}\n```"))
.await
}
#[admin_command]
async fn get_devices_version(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let device = self.services.users.get_devicelist_version(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{device:#?}\n```"))
.await
}
#[admin_command]
async fn count_one_time_keys(&self, user_id: OwnedUserId, device_id: OwnedDeviceId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.count_one_time_keys(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_device_keys(&self, user_id: OwnedUserId, device_id: OwnedDeviceId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_device_keys(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_user_signing_key(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.users.get_user_signing_key(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_master_key(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_master_key(None, &user_id, &|_| true)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
#[admin_command]
async fn get_to_device_events(&self, user_id: OwnedUserId, device_id: OwnedDeviceId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_to_device_events(&user_id, &device_id, None, None)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_backup_session(
&self,
user_id: OwnedUserId,
version: String,
room_id: OwnedRoomId,
session_id: String,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_session(&user_id, &version, &room_id, &session_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_room_backups(
&self,
user_id: OwnedUserId,
version: String,
room_id: OwnedRoomId,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_room(&user_id, &version, &room_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_all_backups(&self, user_id: OwnedUserId, version: String) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.key_backups.get_all(&user_id, &version).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_backup_algorithm(&self, user_id: OwnedUserId, version: String) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_backup(&user_id, &version)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_latest_backup_version(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.key_backups
.get_latest_backup_version(&user_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_latest_backup(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.key_backups.get_latest_backup(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn iter_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<OwnedUserId> =
self.services.users.stream().map(Into::into).collect().await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn iter_users2(&self) -> Result {
let timer = tokio::time::Instant::now();
let result: Vec<_> = self.services.users.stream().collect().await;
let result: Vec<_> = result
.into_iter()
.map(|user_id| String::from_utf8_lossy(user_id.as_bytes()).into_owned())
.collect();
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:?}\n```"))
.await
}
async fn count_users(&self) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.users.count().await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn list_devices(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let devices = self
.services
.users
.all_device_ids(&user_id)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{devices:#?}\n```"))
.await
}
async fn list_devices_metadata(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let devices = self
.services
.users
.all_devices_metadata(&user_id)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{devices:#?}\n```"))
.await
}
async fn get_device_metadata(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
) -> Result {
let timer = tokio::time::Instant::now();
let device = self
.services
.users
.get_device_metadata(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{device:#?}\n```"))
.await
}
async fn get_devices_version(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let device = self.services.users.get_devicelist_version(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{device:#?}\n```"))
.await
}
async fn count_one_time_keys(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.count_one_time_keys(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_device_keys(&self, user_id: OwnedUserId, device_id: OwnedDeviceId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_device_keys(&user_id, &device_id)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_user_signing_key(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self.services.users.get_user_signing_key(&user_id).await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_master_key(&self, user_id: OwnedUserId) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_master_key(None, &user_id, &|_| true)
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
async fn get_to_device_events(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
) -> Result {
let timer = tokio::time::Instant::now();
let result = self
.services
.users
.get_to_device_events(&user_id, &device_id, None, None)
.collect::<Vec<_>>()
.await;
let query_time = timer.elapsed();
self.write_str(&format!("Query completed in {query_time:?}:\n\n```rs\n{result:#?}\n```"))
.await
}
}
src/admin/room/commands.rs
+72 -72
View File
@@ -2,83 +2,83 @@ use conduwuit::{Err, Result};
use futures::StreamExt;
use ruma::OwnedRoomId;
use crate::{PAGE_SIZE, admin_command, get_room_info};
use crate::{PAGE_SIZE, get_room_info};
#[allow(clippy::fn_params_excessive_bools)]
#[admin_command]
pub(super) async fn list_rooms(
&self,
page: Option<usize>,
exclude_disabled: bool,
exclude_banned: bool,
include_empty: bool,
no_details: bool,
) -> Result {
// 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 = self
.services
.rooms
.metadata
.iter_ids()
.filter_map(|room_id| async move {
(!exclude_disabled || !self.services.rooms.metadata.is_disabled(&room_id).await)
.then_some(room_id)
})
.filter_map(|room_id| async move {
(!exclude_banned || !self.services.rooms.metadata.is_banned(&room_id).await)
.then_some(room_id)
})
.then(async |room_id| get_room_info(self.services, &room_id).await)
.then(|(room_id, total_members, name)| async move {
let local_members: Vec<_> = self
.services
.rooms
.state_cache
.active_local_users_in_room(&room_id)
.collect()
.await;
let local_members = local_members.len();
(room_id, total_members, local_members, name)
})
.filter_map(|(room_id, total_members, local_members, name)| async move {
(include_empty || local_members > 0).then_some((room_id, total_members, name))
})
.collect::<Vec<_>>()
.await;
impl crate::Context<'_> {
#[allow(clippy::fn_params_excessive_bools)]
pub(super) async fn list_rooms(
&self,
page: Option<usize>,
exclude_disabled: bool,
exclude_banned: bool,
include_empty: bool,
no_details: bool,
) -> Result {
// 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 = self
.services
.rooms
.metadata
.iter_ids()
.filter_map(|room_id| async move {
(!exclude_disabled || !self.services.rooms.metadata.is_disabled(&room_id).await)
.then_some(room_id)
})
.filter_map(|room_id| async move {
(!exclude_banned || !self.services.rooms.metadata.is_banned(&room_id).await)
.then_some(room_id)
})
.then(async |room_id| get_room_info(self.services, &room_id).await)
.then(|(room_id, total_members, name)| async move {
let local_members: Vec<_> = self
.services
.rooms
.state_cache
.active_local_users_in_room(&room_id)
.collect()
.await;
let local_members = local_members.len();
(room_id, total_members, local_members, name)
})
.filter_map(|(room_id, total_members, local_members, name)| async move {
(include_empty || local_members > 0).then_some((room_id, total_members, name))
})
.collect::<Vec<_>>()
.await;
rooms.sort_by_key(|r| r.1);
rooms.reverse();
rooms.sort_by_key(|r| r.1);
rooms.reverse();
let rooms = rooms
.into_iter()
.skip(page.saturating_sub(1).saturating_mul(PAGE_SIZE))
.take(PAGE_SIZE)
.collect::<Vec<_>>();
let rooms = rooms
.into_iter()
.skip(page.saturating_sub(1).saturating_mul(PAGE_SIZE))
.take(PAGE_SIZE)
.collect::<Vec<_>>();
if rooms.is_empty() {
return Err!("No more rooms.");
if rooms.is_empty() {
return Err!("No more rooms.");
}
let body = rooms
.iter()
.map(|(id, members, name)| {
if no_details {
format!("{id}")
} else {
format!("{id}\tMembers: {members}\tName: {name}")
}
})
.collect::<Vec<_>>()
.join("\n");
self.write_str(&format!("Rooms ({}):\n```\n{body}\n```", rooms.len()))
.await
}
let body = rooms
.iter()
.map(|(id, members, name)| {
if no_details {
format!("{id}")
} else {
format!("{id}\tMembers: {members}\tName: {name}")
}
})
.collect::<Vec<_>>()
.join("\n");
pub(super) async fn exists(&self, room_id: OwnedRoomId) -> Result {
let result = self.services.rooms.metadata.exists(&room_id).await;
self.write_str(&format!("Rooms ({}):\n```\n{body}\n```", rooms.len()))
.await
}
#[admin_command]
pub(super) async fn exists(&self, room_id: OwnedRoomId) -> Result {
let result = self.services.rooms.metadata.exists(&room_id).await;
self.write_str(&format!("{result}")).await
self.write_str(&format!("{result}")).await
}
}
src/admin/room/info.rs
+54 -54
View File
@@ -3,7 +3,7 @@ use conduwuit::{Err, Result, utils::ReadyExt};
use futures::StreamExt;
use ruma::OwnedRoomId;
use crate::{admin_command, admin_command_dispatch};
use crate::admin_command_dispatch;
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -26,62 +26,62 @@ pub enum RoomInfoCommand {
},
}
#[admin_command]
async fn list_joined_members(&self, room_id: OwnedRoomId, local_only: bool) -> Result {
let room_name = self
.services
.rooms
.state_accessor
.get_name(&room_id)
.await
.unwrap_or_else(|_| room_id.to_string());
impl crate::Context<'_> {
async fn list_joined_members(&self, room_id: OwnedRoomId, local_only: bool) -> Result {
let room_name = self
.services
.rooms
.state_accessor
.get_name(&room_id)
.await
.unwrap_or_else(|_| room_id.to_string());
let member_info: Vec<_> = self
.services
.rooms
.state_cache
.room_members(&room_id)
.ready_filter(|user_id| {
local_only
.then(|| self.services.globals.user_is_local(user_id))
.unwrap_or(true)
})
.filter_map(|user_id| async move {
Some((
self.services
.users
.displayname(&user_id)
.await
.unwrap_or_else(|_| user_id.to_string()),
user_id,
))
})
.collect()
.await;
let member_info: Vec<_> = self
.services
.rooms
.state_cache
.room_members(&room_id)
.ready_filter(|user_id| {
local_only
.then(|| self.services.globals.user_is_local(user_id))
.unwrap_or(true)
})
.filter_map(|user_id| async move {
Some((
self.services
.users
.displayname(&user_id)
.await
.unwrap_or_else(|_| user_id.to_string()),
user_id,
))
})
.collect()
.await;
let num = member_info.len();
let body = member_info
.into_iter()
.map(|(displayname, mxid)| format!("{mxid} | {displayname}"))
.collect::<Vec<_>>()
.join("\n");
let num = member_info.len();
let body = member_info
.into_iter()
.map(|(displayname, mxid)| format!("{mxid} | {displayname}"))
.collect::<Vec<_>>()
.join("\n");
self.write_str(&format!("{num} Members in Room \"{room_name}\":\n```\n{body}\n```"))
.await
}
self.write_str(&format!("{num} Members in Room \"{room_name}\":\n```\n{body}\n```"))
.await
}
#[admin_command]
async fn view_room_topic(&self, room_id: OwnedRoomId) -> Result {
let Ok(room_topic) = self
.services
.rooms
.state_accessor
.get_room_topic(&room_id)
.await
else {
return Err!("Room does not have a room topic set.");
};
async fn view_room_topic(&self, room_id: OwnedRoomId) -> Result {
let Ok(room_topic) = self
.services
.rooms
.state_accessor
.get_room_topic(&room_id)
.await
else {
return Err!("Room does not have a room topic set.");
};
self.write_str(&format!("Room topic:\n```\n{room_topic}\n```"))
.await
self.write_str(&format!("Room topic:\n```\n{room_topic}\n```"))
.await
}
}
src/admin/room/moderation.rs
+352 -351
View File
@@ -8,7 +8,7 @@ use conduwuit::{
use futures::{FutureExt, StreamExt};
use ruma::{OwnedRoomId, OwnedRoomOrAliasId, RoomAliasId, RoomId, RoomOrAliasId};
use crate::{admin_command, admin_command_dispatch, get_room_info};
use crate::{admin_command_dispatch, get_room_info};
#[admin_command_dispatch]
#[derive(Debug, Subcommand)]
@@ -45,238 +45,72 @@ pub enum RoomModerationCommand {
},
}
#[admin_command]
async fn ban_room(&self, room: OwnedRoomOrAliasId) -> Result {
debug!("Got room alias or ID: {}", room);
impl crate::Context<'_> {
async fn ban_room(&self, room: OwnedRoomOrAliasId) -> Result {
debug!("Got room alias or ID: {}", room);
let admin_room_alias = &self.services.globals.admin_alias;
let admin_room_alias = &self.services.globals.admin_alias;
if let Ok(admin_room_id) = self.services.admin.get_admin_room().await {
if room.to_string().eq(&admin_room_id) || room.to_string().eq(admin_room_alias) {
return Err!("Not allowed to ban the admin room.");
if let Ok(admin_room_id) = self.services.admin.get_admin_room().await {
if room.to_string().eq(&admin_room_id) || room.to_string().eq(admin_room_alias) {
return Err!("Not allowed to ban the admin room.");
}
}
}
let room_id = if room.is_room_id() {
let room_id = match RoomId::parse(&room) {
| Ok(room_id) => room_id,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full room \
ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
let room_id = if room.is_room_id() {
let room_id = match RoomId::parse(&room) {
| Ok(room_id) => room_id,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full \
room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!("Room specified is a room ID, banning room ID");
room_id.clone()
} else if room.is_room_alias_id() {
let room_alias = match RoomAliasId::parse(&room) {
| Ok(room_alias) => room_alias,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full \
room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!(
"Room specified is not a room ID, attempting to resolve room alias to a room ID \
locally, if not using get_alias_helper to fetch room ID remotely"
);
match self.services.rooms.alias.resolve_alias(&room_alias).await {
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for room {room}"
);
room_id
},
| Err(e) => {
return Err!("Failed to resolve room alias {room} to a room ID: {e}");
},
}
} else {
return Err!(
"Room specified is not a room ID or room alias. Please note that this requires \
a full room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`)",
);
};
debug!("Room specified is a room ID, banning room ID");
room_id.clone()
} else if room.is_room_alias_id() {
let room_alias = match RoomAliasId::parse(&room) {
| Ok(room_alias) => room_alias,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full room \
ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!(
"Room specified is not a room ID, attempting to resolve room alias to a room ID \
locally, if not using get_alias_helper to fetch room ID remotely"
);
match self.services.rooms.alias.resolve_alias(&room_alias).await {
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for room {room}"
);
room_id
},
| Err(e) => {
return Err!("Failed to resolve room alias {room} to a room ID: {e}");
},
}
} else {
return Err!(
"Room specified is not a room ID or room alias. Please note that this requires a \
full room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`)",
);
};
info!("Making all users leave the room {room_id} and forgetting it");
let mut users = self
.services
.rooms
.state_cache
.room_members(&room_id)
.ready_filter(|user| self.services.globals.user_is_local(user))
.boxed();
while let Some(ref user_id) = users.next().await {
info!(
"Attempting leave for user {user_id} in room {room_id} (ignoring all errors, \
evicting admins too)",
);
if let Err(e) = leave_room(self.services, user_id, &room_id, None)
.boxed()
.await
{
warn!("Failed to leave room: {e}");
}
self.services.rooms.state_cache.forget(&room_id, user_id);
}
self.services
.rooms
.alias
.local_aliases_for_room(&room_id)
.for_each(|local_alias| async move {
self.services
.rooms
.alias
.remove_alias(&local_alias, &self.services.globals.server_user)
.await
.ok();
})
.await;
self.services.rooms.directory.set_not_public(&room_id); // remove from the room directory
self.services.rooms.metadata.ban_room(&room_id, true); // prevent further joins
self.services.rooms.metadata.disable_room(&room_id, true); // disable federation
self.write_str(
"Room banned, removed all our local users, and disabled incoming federation with room.",
)
.await
}
#[admin_command]
async fn ban_list_of_rooms(&self) -> Result {
if self.body.len() < 2
|| !self.body[0].trim().starts_with("```")
|| self.body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.",);
}
let rooms_s = self
.body
.to_vec()
.drain(1..self.body.len().saturating_sub(1))
.collect::<Vec<_>>();
let admin_room_alias = &self.services.globals.admin_alias;
let mut room_ban_count: usize = 0;
let mut room_ids: Vec<OwnedRoomId> = Vec::new();
for &room in &rooms_s {
match <&RoomOrAliasId>::try_from(room) {
| Ok(room_alias_or_id) => {
if let Ok(admin_room_id) = self.services.admin.get_admin_room().await {
if room.to_owned().eq(&admin_room_id) || room.to_owned().eq(admin_room_alias)
{
warn!("User specified admin room in bulk ban list, ignoring");
continue;
}
}
if room_alias_or_id.is_room_id() {
let room_id = match RoomId::parse(room_alias_or_id) {
| Ok(room_id) => room_id,
| Err(e) => {
// ignore rooms we failed to parse
warn!(
"Error parsing room \"{room}\" during bulk room banning, \
ignoring error and logging here: {e}"
);
continue;
},
};
room_ids.push(room_id.clone());
}
if room_alias_or_id.is_room_alias_id() {
match RoomAliasId::parse(room_alias_or_id) {
| Ok(room_alias) => {
let room_id = match self
.services
.rooms
.alias
.resolve_local_alias(&room_alias)
.await
{
| Ok(room_id) => room_id,
| _ => {
debug!(
"We don't have this room alias to a room ID locally, \
attempting to fetch room ID over federation"
);
match self
.services
.rooms
.alias
.resolve_alias(&room_alias)
.await
{
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for \
{room}",
);
room_id
},
| Err(e) => {
warn!(
"Failed to resolve room alias {room} to a room \
ID: {e}"
);
continue;
},
}
},
};
room_ids.push(room_id);
},
| Err(e) => {
warn!(
"Error parsing room \"{room}\" during bulk room banning, \
ignoring error and logging here: {e}"
);
continue;
},
}
}
},
| Err(e) => {
warn!(
"Error parsing room \"{room}\" during bulk room banning, ignoring error and \
logging here: {e}"
);
continue;
},
}
}
for room_id in room_ids {
debug!("Banned {room_id} successfully");
room_ban_count = room_ban_count.saturating_add(1);
debug!("Making all users leave the room {room_id} and forgetting it");
info!("Making all users leave the room {room_id} and forgetting it");
let mut users = self
.services
.rooms
@@ -286,7 +120,7 @@ async fn ban_list_of_rooms(&self) -> Result {
.boxed();
while let Some(ref user_id) = users.next().await {
debug!(
info!(
"Attempting leave for user {user_id} in room {room_id} (ignoring all errors, \
evicting admins too)",
);
@@ -301,7 +135,6 @@ async fn ban_list_of_rooms(&self) -> Result {
self.services.rooms.state_cache.forget(&room_id, user_id);
}
// remove any local aliases, ignore errors
self.services
.rooms
.alias
@@ -316,139 +149,307 @@ async fn ban_list_of_rooms(&self) -> Result {
})
.await;
self.services.rooms.metadata.ban_room(&room_id, true);
// unpublish from room directory, ignore errors
self.services.rooms.directory.set_not_public(&room_id);
self.services.rooms.metadata.disable_room(&room_id, true);
self.services.rooms.directory.set_not_public(&room_id); // remove from the room directory
self.services.rooms.metadata.ban_room(&room_id, true); // prevent further joins
self.services.rooms.metadata.disable_room(&room_id, true); // disable federation
self.write_str(
"Room banned, removed all our local users, and disabled incoming federation with \
room.",
)
.await
}
self.write_str(&format!(
"Finished bulk room ban, banned {room_ban_count} total rooms, evicted all users, and \
disabled incoming federation with the room."
))
.await
}
async fn ban_list_of_rooms(&self) -> Result {
if self.body.len() < 2
|| !self.body[0].trim().starts_with("```")
|| self.body.last().unwrap_or(&"").trim() != "```"
{
return Err!("Expected code block in command body. Add --help for details.",);
}
#[admin_command]
async fn unban_room(&self, room: OwnedRoomOrAliasId) -> Result {
let room_id = if room.is_room_id() {
let room_id = match RoomId::parse(&room) {
| Ok(room_id) => room_id,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full room \
ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
let rooms_s = self
.body
.to_vec()
.drain(1..self.body.len().saturating_sub(1))
.collect::<Vec<_>>();
let admin_room_alias = &self.services.globals.admin_alias;
let mut room_ban_count: usize = 0;
let mut room_ids: Vec<OwnedRoomId> = Vec::new();
for &room in &rooms_s {
match <&RoomOrAliasId>::try_from(room) {
| Ok(room_alias_or_id) => {
if let Ok(admin_room_id) = self.services.admin.get_admin_room().await {
if room.to_owned().eq(&admin_room_id)
|| room.to_owned().eq(admin_room_alias)
{
warn!("User specified admin room in bulk ban list, ignoring");
continue;
}
}
if room_alias_or_id.is_room_id() {
let room_id = match RoomId::parse(room_alias_or_id) {
| Ok(room_id) => room_id,
| Err(e) => {
// ignore rooms we failed to parse
warn!(
"Error parsing room \"{room}\" during bulk room banning, \
ignoring error and logging here: {e}"
);
continue;
},
};
room_ids.push(room_id.clone());
}
if room_alias_or_id.is_room_alias_id() {
match RoomAliasId::parse(room_alias_or_id) {
| Ok(room_alias) => {
let room_id = match self
.services
.rooms
.alias
.resolve_local_alias(&room_alias)
.await
{
| Ok(room_id) => room_id,
| _ => {
debug!(
"We don't have this room alias to a room ID \
locally, attempting to fetch room ID over \
federation"
);
match self
.services
.rooms
.alias
.resolve_alias(&room_alias)
.await
{
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for \
{room}",
);
room_id
},
| Err(e) => {
warn!(
"Failed to resolve room alias {room} to a \
room ID: {e}"
);
continue;
},
}
},
};
room_ids.push(room_id);
},
| Err(e) => {
warn!(
"Error parsing room \"{room}\" during bulk room banning, \
ignoring error and logging here: {e}"
);
continue;
},
}
}
},
| Err(e) => {
warn!(
"Error parsing room \"{room}\" during bulk room banning, ignoring error \
and logging here: {e}"
);
continue;
},
}
}
for room_id in room_ids {
debug!("Banned {room_id} successfully");
room_ban_count = room_ban_count.saturating_add(1);
debug!("Making all users leave the room {room_id} and forgetting it");
let mut users = self
.services
.rooms
.state_cache
.room_members(&room_id)
.ready_filter(|user| self.services.globals.user_is_local(user))
.boxed();
while let Some(ref user_id) = users.next().await {
debug!(
"Attempting leave for user {user_id} in room {room_id} (ignoring all \
errors, evicting admins too)",
);
},
if let Err(e) = leave_room(self.services, user_id, &room_id, None)
.boxed()
.await
{
warn!("Failed to leave room: {e}");
}
self.services.rooms.state_cache.forget(&room_id, user_id);
}
// remove any local aliases, ignore errors
self.services
.rooms
.alias
.local_aliases_for_room(&room_id)
.for_each(|local_alias| async move {
self.services
.rooms
.alias
.remove_alias(&local_alias, &self.services.globals.server_user)
.await
.ok();
})
.await;
self.services.rooms.metadata.ban_room(&room_id, true);
// unpublish from room directory, ignore errors
self.services.rooms.directory.set_not_public(&room_id);
self.services.rooms.metadata.disable_room(&room_id, true);
}
self.write_str(&format!(
"Finished bulk room ban, banned {room_ban_count} total rooms, evicted all users, \
and disabled incoming federation with the room."
))
.await
}
async fn unban_room(&self, room: OwnedRoomOrAliasId) -> Result {
let room_id = if room.is_room_id() {
let room_id = match RoomId::parse(&room) {
| Ok(room_id) => room_id,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full \
room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!("Room specified is a room ID, unbanning room ID");
self.services.rooms.metadata.ban_room(&room_id, false);
room_id.clone()
} else if room.is_room_alias_id() {
let room_alias = match RoomAliasId::parse(&room) {
| Ok(room_alias) => room_alias,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full \
room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!(
"Room specified is not a room ID, attempting to resolve room alias to a room ID \
locally, if not using get_alias_helper to fetch room ID remotely"
);
let room_id = match self
.services
.rooms
.alias
.resolve_local_alias(&room_alias)
.await
{
| Ok(room_id) => room_id,
| _ => {
debug!(
"We don't have this room alias to a room ID locally, attempting to \
fetch room ID over federation"
);
match self.services.rooms.alias.resolve_alias(&room_alias).await {
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for room {room}"
);
room_id
},
| Err(e) => {
return Err!("Failed to resolve room alias {room} to a room ID: {e}");
},
}
},
};
self.services.rooms.metadata.ban_room(&room_id, false);
room_id
} else {
return Err!(
"Room specified is not a room ID or room alias. Please note that this requires \
a full room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`)",
);
};
debug!("Room specified is a room ID, unbanning room ID");
self.services.rooms.metadata.ban_room(&room_id, false);
self.services.rooms.metadata.disable_room(&room_id, false);
self.write_str("Room unbanned and federation re-enabled.")
.await
}
room_id.clone()
} else if room.is_room_alias_id() {
let room_alias = match RoomAliasId::parse(&room) {
| Ok(room_alias) => room_alias,
| Err(e) => {
return Err!(
"Failed to parse room ID {room}. Please note that this requires a full room \
ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`): {e}"
);
},
};
debug!(
"Room specified is not a room ID, attempting to resolve room alias to a room ID \
locally, if not using get_alias_helper to fetch room ID remotely"
);
let room_id = match self
async fn list_banned_rooms(&self, no_details: bool) -> Result {
let room_ids: Vec<OwnedRoomId> = self
.services
.rooms
.alias
.resolve_local_alias(&room_alias)
.await
{
| Ok(room_id) => room_id,
| _ => {
debug!(
"We don't have this room alias to a room ID locally, attempting to fetch \
room ID over federation"
);
.metadata
.list_banned_rooms()
.map(Into::into)
.collect()
.await;
match self.services.rooms.alias.resolve_alias(&room_alias).await {
| Ok((room_id, servers)) => {
debug!(
%room_id,
?servers,
"Got federation response fetching room ID for room {room}"
);
room_id
},
| Err(e) => {
return Err!("Failed to resolve room alias {room} to a room ID: {e}");
},
if room_ids.is_empty() {
return Err!("No rooms are banned.");
}
let mut rooms = room_ids
.iter()
.stream()
.then(|room_id| get_room_info(self.services, room_id))
.collect::<Vec<_>>()
.await;
rooms.sort_by_key(|r| r.1);
rooms.reverse();
let num = rooms.len();
let body = rooms
.iter()
.map(|(id, members, name)| {
if no_details {
format!("{id}")
} else {
format!("{id}\tMembers: {members}\tName: {name}")
}
},
};
})
.collect::<Vec<_>>()
.join("\n");
self.services.rooms.metadata.ban_room(&room_id, false);
room_id
} else {
return Err!(
"Room specified is not a room ID or room alias. Please note that this requires a \
full room ID (`!awIh6gGInaS5wLQJwa:example.com`) or a room alias \
(`#roomalias:example.com`)",
);
};
self.services.rooms.metadata.disable_room(&room_id, false);
self.write_str("Room unbanned and federation re-enabled.")
.await
}
#[admin_command]
async fn list_banned_rooms(&self, no_details: bool) -> Result {
let room_ids: Vec<OwnedRoomId> = self
.services
.rooms
.metadata
.list_banned_rooms()
.map(Into::into)
.collect()
.await;
if room_ids.is_empty() {
return Err!("No rooms are banned.");
self.write_str(&format!("Rooms Banned ({num}):\n```\n{body}\n```"))
.await
}
let mut rooms = room_ids
.iter()
.stream()
.then(|room_id| get_room_info(self.services, room_id))
.collect::<Vec<_>>()
.await;
rooms.sort_by_key(|r| r.1);
rooms.reverse();
let num = rooms.len();
let body = rooms
.iter()
.map(|(id, members, name)| {
if no_details {
format!("{id}")
} else {
format!("{id}\tMembers: {members}\tName: {name}")
}
})
.collect::<Vec<_>>()
.join("\n");
self.write_str(&format!("Rooms Banned ({num}):\n```\n{body}\n```"))
.await
}
src/admin/server/commands.rs
+197 -210
View File
@@ -7,243 +7,230 @@ use conduwuit::{
};
use futures::TryStreamExt;
use crate::admin_command;
impl crate::Context<'_> {
pub(super) async fn uptime(&self) -> Result {
let elapsed = self
.services
.server
.started
.elapsed()
.expect("standard duration");
#[admin_command]
pub(super) async fn uptime(&self) -> Result {
let elapsed = self
.services
.server
.started
.elapsed()
.expect("standard duration");
let result = time::pretty(elapsed);
self.write_str(&format!("{result}.")).await
}
let result = time::pretty(elapsed);
self.write_str(&format!("{result}.")).await
}
pub(super) async fn show_config(&self) -> Result {
self.bail_restricted()?;
#[admin_command]
pub(super) async fn show_config(&self) -> Result {
self.bail_restricted()?;
self.write_str(&format!("{}", *self.services.server.config))
.await
}
self.write_str(&format!("{}", *self.services.server.config))
.await
}
pub(super) async fn reload_config(&self, path: Option<PathBuf>) -> Result {
// The path argument is only what's optionally passed via the admin command,
// so we need to merge it with the existing paths if any were given at startup.
let mut paths = Vec::new();
#[admin_command]
pub(super) async fn reload_config(&self, path: Option<PathBuf>) -> Result {
// The path argument is only what's optionally passed via the admin command,
// so we need to merge it with the existing paths if any were given at startup.
let mut paths = Vec::new();
// Add previously saved paths to the argument list
self.services
.config
.config_paths
.clone()
.unwrap_or_default()
.iter()
.for_each(|p| paths.push(p.to_owned()));
// Add previously saved paths to the argument list
self.services
.config
.config_paths
.clone()
.unwrap_or_default()
.iter()
.for_each(|p| paths.push(p.to_owned()));
// If a path is given, and it's not already in the list,
// add it last, so that it overrides earlier files
if let Some(p) = path {
if !paths.contains(&p) {
paths.push(p);
// If a path is given, and it's not already in the list,
// add it last, so that it overrides earlier files
if let Some(p) = path {
if !paths.contains(&p) {
paths.push(p);
}
}
self.services.config.reload(&paths)?;
self.write_str(&format!("Successfully reconfigured from paths: {paths:?}"))
.await
}
self.services.config.reload(&paths)?;
pub(super) async fn memory_usage(&self) -> Result {
let services_usage = self.services.memory_usage().await?;
let database_usage = self.services.db.db.memory_usage()?;
let allocator_usage = conduwuit::alloc::memory_usage()
.map_or(String::new(), |s| format!("\nAllocator:\n{s}"));
self.write_str(&format!("Successfully reconfigured from paths: {paths:?}"))
self.write_str(&format!(
"Services:\n{services_usage}\nDatabase:\n{database_usage}{allocator_usage}",
))
.await
}
#[admin_command]
pub(super) async fn memory_usage(&self) -> Result {
let services_usage = self.services.memory_usage().await?;
let database_usage = self.services.db.db.memory_usage()?;
let allocator_usage =
conduwuit::alloc::memory_usage().map_or(String::new(), |s| format!("\nAllocator:\n{s}"));
self.write_str(&format!(
"Services:\n{services_usage}\nDatabase:\n{database_usage}{allocator_usage}",
))
.await
}
#[admin_command]
pub(super) async fn clear_caches(&self) -> Result {
self.services.clear_cache().await;
self.write_str("Done.").await
}
#[admin_command]
pub(super) async fn list_backups(&self) -> Result {
self.services
.db
.db
.backup_list()?
.try_stream()
.try_for_each(|result| writeln!(self, "{result}"))
.await
}
#[admin_command]
pub(super) async fn backup_database(&self) -> Result {
self.bail_restricted()?;
let db = Arc::clone(&self.services.db);
let result = self
.services
.server
.runtime()
.spawn_blocking(move || match db.db.backup() {
| Ok(()) => "Done".to_owned(),
| Err(e) => format!("Failed: {e}"),
})
.await?;
let count = self.services.db.db.backup_count()?;
self.write_str(&format!("{result}. Currently have {count} backups."))
.await
}
#[admin_command]
pub(super) async fn admin_notice(&self, message: Vec<String>) -> Result {
let message = message.join(" ");
self.services.admin.send_text(&message).await;
self.write_str("Notice was sent to #admins").await
}
#[admin_command]
pub(super) async fn reload_mods(&self) -> Result {
self.bail_restricted()?;
self.services.server.reload()?;
self.write_str("Reloading server...").await
}
#[admin_command]
#[cfg(unix)]
pub(super) async fn restart(&self, force: bool) -> Result {
use conduwuit::utils::sys::current_exe_deleted;
if !force && current_exe_deleted() {
return Err!(
"The server cannot be restarted because the executable changed. If this is expected \
use --force to override."
);
}
self.services.server.restart()?;
pub(super) async fn clear_caches(&self) -> Result {
self.services.clear_cache().await;
self.write_str("Restarting server...").await
}
#[admin_command]
pub(super) async fn shutdown(&self) -> Result {
self.bail_restricted()?;
warn!("shutdown command");
self.services.server.shutdown()?;
self.write_str("Shutting down server...").await
}
#[admin_command]
pub(super) async fn list_features(&self) -> Result {
let mut enabled_features = conduwuit::info::introspection::ENABLED_FEATURES
.lock()
.expect("locked")
.values()
.flat_map(|f| f.iter())
.collect::<Vec<_>>();
enabled_features.sort_unstable();
enabled_features.dedup();
let mut available_features = conduwuit::build_metadata::WORKSPACE_FEATURES
.iter()
.flat_map(|(_, f)| f.iter())
.collect::<Vec<_>>();
available_features.sort_unstable();
available_features.dedup();
let mut features = String::new();
for feature in available_features {
let active = enabled_features.contains(&feature);
let emoji = if active { "✅" } else { "❌" };
let remark = if active { "[enabled]" } else { "" };
writeln!(features, "{emoji} {feature} {remark}")?;
self.write_str("Done.").await
}
self.write_str(&features).await
}
#[admin_command]
pub(super) async fn build_info(&self) -> Result {
use conduwuit::build_metadata::built;
pub(super) async fn list_backups(&self) -> Result {
self.services
.db
.db
.backup_list()?
.try_stream()
.try_for_each(|result| writeln!(self, "{result}"))
.await
}
let mut info = String::new();
pub(super) async fn backup_database(&self) -> Result {
self.bail_restricted()?;
let db = Arc::clone(&self.services.db);
let result = self
.services
.server
.runtime()
.spawn_blocking(move || match db.db.backup() {
| Ok(()) => "Done".to_owned(),
| Err(e) => format!("Failed: {e}"),
})
.await?;
let count = self.services.db.db.backup_count()?;
self.write_str(&format!("{result}. Currently have {count} backups."))
.await
}
// Version information
writeln!(info, "# Build Information\n")?;
writeln!(info, "**Version:** {}", built::PKG_VERSION)?;
writeln!(info, "**Package:** {}", built::PKG_NAME)?;
writeln!(info, "**Description:** {}", built::PKG_DESCRIPTION)?;
pub(super) async fn admin_notice(&self, message: Vec<String>) -> Result {
let message = message.join(" ");
self.services.admin.send_text(&message).await;
// Git information
writeln!(info, "\n## Git Information\n")?;
if let Some(hash) = conduwuit::build_metadata::GIT_COMMIT_HASH {
writeln!(info, "**Commit Hash:** {hash}")?;
}
if let Some(hash) = conduwuit::build_metadata::GIT_COMMIT_HASH_SHORT {
writeln!(info, "**Commit Hash (short):** {hash}")?;
}
if let Some(url) = conduwuit::build_metadata::GIT_REMOTE_WEB_URL {
writeln!(info, "**Repository:** {url}")?;
self.write_str("Notice was sent to #admins").await
}
if let Some(url) = conduwuit::build_metadata::GIT_REMOTE_COMMIT_URL {
writeln!(info, "**Commit URL:** {url}")?;
pub(super) async fn reload_mods(&self) -> Result {
self.bail_restricted()?;
self.services.server.reload()?;
self.write_str("Reloading server...").await
}
// Build environment
writeln!(info, "\n## Build Environment\n")?;
writeln!(info, "**Profile:** {}", built::PROFILE)?;
writeln!(info, "**Optimization Level:** {}", built::OPT_LEVEL)?;
writeln!(info, "**Debug:** {}", built::DEBUG)?;
writeln!(info, "**Target:** {}", built::TARGET)?;
writeln!(info, "**Host:** {}", built::HOST)?;
#[cfg(unix)]
pub(super) async fn restart(&self, force: bool) -> Result {
use conduwuit::utils::sys::current_exe_deleted;
// Rust compiler information
writeln!(info, "\n## Compiler Information\n")?;
writeln!(info, "**Rustc Version:** {}", built::RUSTC_VERSION)?;
if !built::RUSTDOC_VERSION.is_empty() {
writeln!(info, "**Rustdoc Version:** {}", built::RUSTDOC_VERSION)?;
if !force && current_exe_deleted() {
return Err!(
"The server cannot be restarted because the executable changed. If this is \
expected use --force to override."
);
}
self.services.server.restart()?;
self.write_str("Restarting server...").await
}
// Target configuration
writeln!(info, "\n## Target Configuration\n")?;
writeln!(info, "**Architecture:** {}", built::CFG_TARGET_ARCH)?;
writeln!(info, "**OS:** {}", built::CFG_OS)?;
writeln!(info, "**Family:** {}", built::CFG_FAMILY)?;
writeln!(info, "**Endianness:** {}", built::CFG_ENDIAN)?;
writeln!(info, "**Pointer Width:** {} bits", built::CFG_POINTER_WIDTH)?;
if !built::CFG_ENV.is_empty() {
writeln!(info, "**Environment:** {}", built::CFG_ENV)?;
pub(super) async fn shutdown(&self) -> Result {
self.bail_restricted()?;
warn!("shutdown command");
self.services.server.shutdown()?;
self.write_str("Shutting down server...").await
}
// CI information
if let Some(ci) = built::CI_PLATFORM {
writeln!(info, "\n## CI Platform\n")?;
writeln!(info, "**Platform:** {ci}")?;
pub(super) async fn list_features(&self) -> Result {
let mut enabled_features = conduwuit::info::introspection::ENABLED_FEATURES
.lock()
.expect("locked")
.values()
.flat_map(|f| f.iter())
.collect::<Vec<_>>();
enabled_features.sort_unstable();
enabled_features.dedup();
let mut available_features = conduwuit::build_metadata::WORKSPACE_FEATURES
.iter()
.flat_map(|(_, f)| f.iter())
.collect::<Vec<_>>();
available_features.sort_unstable();
available_features.dedup();
let mut features = String::new();
for feature in available_features {
let active = enabled_features.contains(&feature);
let emoji = if active { "✅" } else { "❌" };
let remark = if active { "[enabled]" } else { "" };
writeln!(features, "{emoji} {feature} {remark}")?;
}
self.write_str(&features).await
}
self.write_str(&info).await
pub(super) async fn build_info(&self) -> Result {
use conduwuit::build_metadata::built;
let mut info = String::new();
// Version information
writeln!(info, "# Build Information\n")?;
writeln!(info, "**Version:** {}", built::PKG_VERSION)?;
writeln!(info, "**Package:** {}", built::PKG_NAME)?;
writeln!(info, "**Description:** {}", built::PKG_DESCRIPTION)?;
// Git information
writeln!(info, "\n## Git Information\n")?;
if let Some(hash) = conduwuit::build_metadata::GIT_COMMIT_HASH {
writeln!(info, "**Commit Hash:** {hash}")?;
}
if let Some(hash) = conduwuit::build_metadata::GIT_COMMIT_HASH_SHORT {
writeln!(info, "**Commit Hash (short):** {hash}")?;
}
if let Some(url) = conduwuit::build_metadata::GIT_REMOTE_WEB_URL {
writeln!(info, "**Repository:** {url}")?;
}
if let Some(url) = conduwuit::build_metadata::GIT_REMOTE_COMMIT_URL {
writeln!(info, "**Commit URL:** {url}")?;
}
// Build environment
writeln!(info, "\n## Build Environment\n")?;
writeln!(info, "**Profile:** {}", built::PROFILE)?;
writeln!(info, "**Optimization Level:** {}", built::OPT_LEVEL)?;
writeln!(info, "**Debug:** {}", built::DEBUG)?;
writeln!(info, "**Target:** {}", built::TARGET)?;
writeln!(info, "**Host:** {}", built::HOST)?;
// Rust compiler information
writeln!(info, "\n## Compiler Information\n")?;
writeln!(info, "**Rustc Version:** {}", built::RUSTC_VERSION)?;
if !built::RUSTDOC_VERSION.is_empty() {
writeln!(info, "**Rustdoc Version:** {}", built::RUSTDOC_VERSION)?;
}
// Target configuration
writeln!(info, "\n## Target Configuration\n")?;
writeln!(info, "**Architecture:** {}", built::CFG_TARGET_ARCH)?;
writeln!(info, "**OS:** {}", built::CFG_OS)?;
writeln!(info, "**Family:** {}", built::CFG_FAMILY)?;
writeln!(info, "**Endianness:** {}", built::CFG_ENDIAN)?;
writeln!(info, "**Pointer Width:** {} bits", built::CFG_POINTER_WIDTH)?;
if !built::CFG_ENV.is_empty() {
writeln!(info, "**Environment:** {}", built::CFG_ENV)?;
}
// CI information
if let Some(ci) = built::CI_PLATFORM {
writeln!(info, "\n## CI Platform\n")?;
writeln!(info, "**Platform:** {ci}")?;
}
self.write_str(&info).await
}
}
src/admin/token/commands.rs
+66 -66
View File
@@ -1,76 +1,76 @@
use conduwuit::{Err, Result, utils};
use conduwuit_macros::admin_command;
use futures::StreamExt;
use service::registration_tokens::TokenExpires;
#[admin_command]
pub(super) async fn issue_token(&self, expires: super::TokenExpires) -> Result {
let expires = {
if expires.immortal {
None
} else if let Some(max_uses) = expires.max_uses {
Some(TokenExpires::AfterUses(max_uses))
} else if expires.once {
Some(TokenExpires::AfterUses(1))
} else if let Some(max_age) = expires
.max_age
.as_deref()
.map(|max_age| utils::time::timepoint_from_now(utils::time::parse_duration(max_age)?))
.transpose()?
{
Some(TokenExpires::AfterTime(max_age))
} else {
unreachable!();
}
};
impl crate::Context<'_> {
pub(super) async fn issue_token(&self, expires: super::TokenExpires) -> Result {
let expires = {
if expires.immortal {
None
} else if let Some(max_uses) = expires.max_uses {
Some(TokenExpires::AfterUses(max_uses))
} else if expires.once {
Some(TokenExpires::AfterUses(1))
} else if let Some(max_age) = expires
.max_age
.as_deref()
.map(|max_age| {
utils::time::timepoint_from_now(utils::time::parse_duration(max_age)?)
})
.transpose()?
{
Some(TokenExpires::AfterTime(max_age))
} else {
unreachable!();
}
};
let (token, info) = self
.services
.registration_tokens
.issue_token(self.sender_or_service_user().into(), expires);
let (token, info) = self
.services
.registration_tokens
.issue_token(self.sender_or_service_user().into(), expires);
self.write_str(&format!(
"New registration token issued: `{token}`. {}.",
if let Some(expires) = info.expires {
format!("{expires}")
} else {
"Never expires".to_owned()
}
))
.await
}
#[admin_command]
pub(super) async fn revoke_token(&self, token: String) -> Result {
let Some(token) = self
.services
.registration_tokens
.validate_token(token)
self.write_str(&format!(
"New registration token issued: `{token}`. {}.",
if let Some(expires) = info.expires {
format!("{expires}")
} else {
"Never expires".to_owned()
}
))
.await
else {
return Err!("This token does not exist or has already expired.");
};
self.services.registration_tokens.revoke_token(token)?;
self.write_str("Token revoked successfully.").await
}
#[admin_command]
pub(super) async fn list_tokens(&self) -> Result {
let tokens: Vec<_> = self
.services
.registration_tokens
.iterate_tokens()
.collect()
.await;
self.write_str(&format!("Found {} registration tokens:\n", tokens.len()))
.await?;
for token in tokens {
self.write_str(&format!("- {token}\n")).await?;
}
Ok(())
pub(super) async fn revoke_token(&self, token: String) -> Result {
let Some(token) = self
.services
.registration_tokens
.validate_token(token)
.await
else {
return Err!("This token does not exist or has already expired.");
};
self.services.registration_tokens.revoke_token(token)?;
self.write_str("Token revoked successfully.").await
}
pub(super) async fn list_tokens(&self) -> Result {
let tokens: Vec<_> = self
.services
.registration_tokens
.iterate_tokens()
.collect()
.await;
self.write_str(&format!("Found {} registration tokens:\n", tokens.len()))
.await?;
for token in tokens {
self.write_str(&format!("- {token}\n")).await?;
}
Ok(())
}
}
src/admin/user/commands.rs
+1155 -1158
View File
File diff suppressed because it is too large. Load diff
src/macros/admin.rs
+1 -10
View File
@@ -2,19 +2,10 @@ use itertools::Itertools;
use proc_macro::{Span, TokenStream};
use proc_macro2::TokenStream as TokenStream2;
use quote::{ToTokens, quote};
use syn::{Attribute, Error, Fields, Ident, ItemEnum, ItemFn, Meta, Variant, parse_quote};
use syn::{Error, Fields, Ident, ItemEnum, Meta, Variant};
use crate::{Result, utils::camel_to_snake_string};
pub(super) fn command(mut item: ItemFn, _args: &[Meta]) -> Result<TokenStream> {
let attr: Attribute = parse_quote! {
#[conduwuit_macros::implement(crate::Context, 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()?;
src/macros/mod.rs
-5
View File
@@ -14,11 +14,6 @@ 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)
src/macros/utils.rs
+1 -3
View File
@@ -1,8 +1,6 @@
use std::collections::HashMap;
use syn::{Expr, ExprLit, Generics, Lit, Meta, MetaNameValue, parse_str};
use crate::Result;
use syn::{Expr, ExprLit, Lit, Meta, MetaNameValue};
pub(crate) fn get_simple_settings(args: &[Meta]) -> HashMap<String, String> {
args.iter().fold(HashMap::new(), |mut map, arg| {