fix: Use server runtime instead of tokio

Cargo.lock

@@ -1311,7 +1311,7 @@ dependencies = [
 [[package]]
 name = "continuwuity-admin-api"
 version = "0.1.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "ruma-common",
  "serde",
@@ -1691,7 +1691,7 @@ dependencies = [
 [[package]]
 name = "draupnir-antispam"
 version = "0.1.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "ruma-common",
  "serde",
@@ -3039,7 +3039,7 @@ checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
 [[package]]
 name = "meowlnir-antispam"
 version = "0.1.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "ruma-common",
  "serde",
@@ -4254,7 +4254,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.10.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "assign",
  "continuwuity-admin-api",
@@ -4277,7 +4277,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.10.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -4289,7 +4289,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.18.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "as_variant",
  "assign",
@@ -4312,7 +4312,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -4344,7 +4344,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.28.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "as_variant",
  "indexmap",
@@ -4369,7 +4369,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "bytes",
  "headers",
@@ -4391,7 +4391,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.9.5"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "js_int",
  "thiserror 2.0.17",
@@ -4400,7 +4400,7 @@ dependencies = [
 [[package]]
 name = "ruma-identity-service-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -4410,7 +4410,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "cfg-if",
  "proc-macro-crate",
@@ -4425,7 +4425,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -4437,7 +4437,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.15.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=85d00fb5746cba23904234b4fd3c838dcf141541#85d00fb5746cba23904234b4fd3c838dcf141541"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=db86e37d602cc4935853bf14d5aace47d22358ed#db86e37d602cc4935853bf14d5aace47d22358ed"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",

Cargo.toml

@@ -342,7 +342,7 @@ version = "0.1.2"
 # Used for matrix spec type definitions and helpers
 [workspace.dependencies.ruma]
 git = "https://forgejo.ellis.link/continuwuation/ruwuma"
-rev = "85d00fb5746cba23904234b4fd3c838dcf141541"
+rev = "db86e37d602cc4935853bf14d5aace47d22358ed"
 features = [
     "compat",
     "rand",

src/api/client/media.rs

@@ -3,8 +3,11 @@ use std::time::Duration;
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduwuit::{
-	Err, Result, err,
+	Err, Result,
+	debug::DebugInspect,
+	debug_info, err,
 	utils::{self, content_disposition::make_content_disposition, math::ruma_from_usize},
+	warn,
 };
 use conduwuit_service::{
 	Services,
@@ -12,8 +15,9 @@ use conduwuit_service::{
 };
 use reqwest::Url;
 use ruma::{
-	Mxc, UserId,
+	Mxc, OwnedServerName, UserId,
 	api::client::{
+		authenticated_media,
 		authenticated_media::{
 			get_content, get_content_as_filename, get_content_thumbnail, get_media_config,
 			get_media_preview,
@@ -244,6 +248,73 @@ pub(crate) async fn get_media_preview_route(
 		})
 }
 
+async fn dispatch_redaction(
+	server_name: OwnedServerName,
+	media_id: String,
+	servers: Vec<OwnedServerName>,
+	services: crate::State,
+) {
+	for server in servers {
+		if services.globals.server_is_ours(&server) {
+			continue;
+		}
+
+		debug_info!("Asking {server} to redact media mxc://{server_name}/{media_id}");
+		services
+			.federation
+			.execute(&server, authenticated_media::redact::unstable::Request {
+				server_name: server_name.clone(),
+				media_id: media_id.clone(),
+			})
+			.await
+			.debug_inspect(|_| {
+				debug_info!("Asked {server} to redact media mxc://{server_name}/{media_id}");
+			})
+			.inspect_err(|e| {
+				warn!(
+					"Failed to ask {server} to redact media mxc://{server_name}/{media_id}: {e}"
+				);
+			})
+			.ok();
+	}
+}
+
+#[tracing::instrument(
+	name = "media_redact",
+	level = "debug",
+	skip_all,
+	fields(%_client),
+)]
+pub(crate) async fn redact_media_route(
+	State(services): State<crate::State>,
+	InsecureClientIp(_client): InsecureClientIp,
+	body: Ruma<authenticated_media::redact::unstable::Request>,
+) -> Result<authenticated_media::redact::unstable::Response> {
+	let user = body.sender_user();
+
+	let mxc = Mxc {
+		server_name: &body.server_name,
+		media_id: &body.media_id,
+	};
+
+	if !services.media.user_owns(user, &mxc).await {
+		return Err!(Request(Forbidden("You do not have permission to redact this attachment.")));
+	}
+
+	services.media.redact(&mxc).await?;
+
+	// TODO: This should be a persistent background task
+	let servers = services.media.get_interested_servers(&mxc).await;
+	services.server.runtime().spawn(dispatch_redaction(
+		mxc.server_name.to_owned(),
+		mxc.media_id.to_owned(),
+		servers,
+		services,
+	));
+
+	Ok(authenticated_media::redact::unstable::Response {})
+}
+
 async fn fetch_thumbnail(
 	services: &Services,
 	mxc: &Mxc<'_>,

src/api/router.rs

@@ -154,6 +154,7 @@ pub fn build(router: Router<State>, server: &Server) -> Router<State> {
 		.ruma_route(&client::get_content_route)
 		.ruma_route(&client::get_content_as_filename_route)
 		.ruma_route(&client::get_media_preview_route)
+		.ruma_route(&client::redact_media_route)
 		.ruma_route(&client::get_media_config_route)
 		.ruma_route(&client::get_devices_route)
 		.ruma_route(&client::get_device_route)

src/api/server/media.rs

@@ -46,6 +46,7 @@ pub(crate) async fn get_content_route(
 		content_type: content_type.map(Into::into),
 		content_disposition: Some(content_disposition),
 	};
+	services.media.mark_server_interested(&mxc, body.origin());
 
 	Ok(get_content::v1::Response {
 		content: FileOrLocation::File(content),
@@ -89,6 +90,7 @@ pub(crate) async fn get_content_thumbnail_route(
 		content_type: content_type.map(Into::into),
 		content_disposition: Some(content_disposition),
 	};
+	services.media.mark_server_interested(&mxc, body.origin());
 
 	Ok(get_content_thumbnail::v1::Response {
 		content: FileOrLocation::File(content),

src/database/maps.rs

@@ -108,6 +108,14 @@ pub(super) static MAPS: &[Descriptor] = &[
 		name: "mediaid_user",
 		..descriptor::RANDOM_SMALL
 	},
+	Descriptor {
+		name: "mediaid_redacted",
+		..descriptor::RANDOM_SMALL
+	},
+	Descriptor {
+		name: "mediaid_interestedservername",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "onetimekeyid_onetimekeys",
 		..descriptor::RANDOM_SMALL

src/service/media/data.rs

@@ -1,4 +1,4 @@
-use std::{sync::Arc, time::Duration};
+use std::{str::FromStr, sync::Arc, time::Duration};
 
 use conduwuit::{
 	Err, Result, debug, debug_info, err,
@@ -6,13 +6,15 @@ use conduwuit::{
 };
 use database::{Database, Interfix, Map};
 use futures::StreamExt;
-use ruma::{Mxc, OwnedMxcUri, UserId, http_headers::ContentDisposition};
+use ruma::{Mxc, OwnedMxcUri, OwnedServerName, UserId, http_headers::ContentDisposition};
 
 use super::{preview::UrlPreviewData, thumbnail::Dim};
 
 pub(crate) struct Data {
 	mediaid_file: Arc<Map>,
 	mediaid_user: Arc<Map>,
+	mediaid_redacted: Arc<Map>,
+	mediaid_interestedservername: Arc<Map>,
 	url_previews: Arc<Map>,
 }
 
@@ -28,6 +30,8 @@ impl Data {
 		Self {
 			mediaid_file: db["mediaid_file"].clone(),
 			mediaid_user: db["mediaid_user"].clone(),
+			mediaid_redacted: db["mediaid_redacted"].clone(),
+			mediaid_interestedservername: db["mediaid_interestedservername"].clone(),
 			url_previews: db["url_previews"].clone(),
 		}
 	}
@@ -77,6 +81,22 @@ impl Data {
 				self.mediaid_user.remove(key);
 			})
 			.await;
+
+		self.mediaid_interestedservername
+			.stream_prefix_raw(&prefix)
+			.ignore_err()
+			.ready_for_each(|(key, _)| {
+				debug_assert!(
+					key.starts_with(mxc.to_string().as_bytes()),
+					"key should start with the mxc"
+				);
+
+				debug_info!("Deleting interested server name key {key:?}");
+
+				self.mediaid_interestedservername.remove(key);
+			})
+			.await;
+		// NOTE: Redaction status is kept even after deletion
 	}
 
 	/// Searches for all files with the given MXC
@@ -275,4 +295,35 @@ impl Data {
 			image_height,
 		})
 	}
+
+	/// Marks a media item as redacted, preventing it from being served or
+	/// re-used.
+	pub(super) fn mark_redacted(&self, media_id: &str) {
+		self.mediaid_redacted.insert(media_id, []);
+	}
+
+	/// Checks if a media item is redacted.
+	pub(super) async fn is_redacted(&self, media_id: &str) -> bool {
+		self.mediaid_redacted.contains(media_id).await
+	}
+
+	pub(super) fn add_interested_server_name(&self, media_id: &str, server_name: &str) {
+		let key = (media_id, server_name);
+		self.mediaid_interestedservername
+			.insert(&database::serialize_key(key).expect("key must be serializable"), []);
+	}
+
+	pub(super) async fn interested_server_names(&self, media_id: &str) -> Vec<OwnedServerName> {
+		let prefix = (media_id, Interfix);
+		self.mediaid_interestedservername
+			.stream_prefix_raw(&prefix)
+			.ignore_err()
+			.map(|(key, _)| {
+				let parts: Vec<&[u8]> = key.rsplit(|&b| b == 0xFF).collect();
+				OwnedServerName::parse(string_from_bytes(parts[0]).unwrap_or_default())
+					.unwrap_or_else(|_| OwnedServerName::from_str("invalid.server").unwrap())
+			})
+			.collect()
+			.await
+	}
 }

src/service/media/mod.rs

@@ -17,7 +17,9 @@ use conduwuit::{
 	},
 	warn,
 };
-use ruma::{Mxc, OwnedMxcUri, UserId, http_headers::ContentDisposition};
+use ruma::{
+	Mxc, OwnedMxcUri, OwnedServerName, ServerName, UserId, http_headers::ContentDisposition,
+};
 use tokio::{
 	fs,
 	io::{AsyncReadExt, AsyncWriteExt, BufReader},
@@ -139,6 +141,28 @@ impl Service {
 		}
 	}
 
+	/// Marks a media ID as redacted, and deletes the associated file.
+	pub async fn redact(&self, mxc: &Mxc<'_>) -> Result<()> {
+		self.db.mark_redacted(mxc.media_id);
+		self.delete(mxc).await
+	}
+
+	/// Checks if a media ID is redacted.
+	pub async fn is_redacted(&self, mxc: &Mxc<'_>) -> bool {
+		self.db.is_redacted(mxc.media_id).await
+	}
+
+	/// Marks a server as "interested" (i.e. has downloaded this media from us).
+	pub fn mark_server_interested(&self, mxc: &Mxc<'_>, server_name: &ServerName) {
+		self.db
+			.add_interested_server_name(mxc.media_id, server_name.as_str());
+	}
+
+	/// Gets all servers interested in this media ID.
+	pub async fn get_interested_servers(&self, mxc: &Mxc<'_>) -> Vec<OwnedServerName> {
+		self.db.interested_server_names(mxc.media_id).await
+	}
+
 	/// Deletes all media by the specified user
 	///
 	/// currently, this is only practical for local users
@@ -229,6 +253,18 @@ impl Service {
 		Ok(mxcs)
 	}
 
+	/// Checks if a user owns a given MXC URI
+	pub async fn user_owns(&self, user: &UserId, mxc: &Mxc<'_>) -> bool {
+		self
+			.db
+			.get_all_user_mxcs(user)  // TODO: this can be more efficient.
+			.await
+			.iter()
+			.any(|v| {
+				v.parts().map(|x|(x.server_name, x.media_id)) == Ok((mxc.server_name, mxc.media_id))
+			})
+	}
+
 	/// Deletes all media files in the given time frame.
 	/// Returns a usize with the amount of media files deleted.
 	pub async fn delete_all_media_within_timeframe(

src/service/users/mod.rs

@@ -304,7 +304,7 @@ impl Service {
 	pub fn enable_login(&self, user_id: &UserId) { self.db.userid_logindisabled.remove(user_id); }
 
 	pub async fn is_login_disabled(&self, user_id: &UserId) -> bool {
-		self.db.userid_logindisabled.contains(user_id).await
+		self.db.userid_logindisabled.exists(user_id).await.is_ok()
 	}
 
 	/// Check if account is active, infallible