Media deletion Feature

clippy.toml

@@ -1,7 +1,7 @@
 array-size-threshold = 4096
 cognitive-complexity-threshold = 100        # TODO reduce me ALARA
 excessive-nesting-threshold = 8
-future-size-threshold = 8192
+future-size-threshold = 12288
 stack-size-threshold = 196608               # TODO reduce me ALARA
 too-many-lines-threshold = 780              # TODO reduce me to <= 100
 type-complexity-threshold = 250             # reduce me to ~200

docs/deploying/generic.md

@@ -166,7 +166,7 @@ sudo systemctl enable --now caddy
 
 ### Other Reverse Proxies
 
-As we would prefer our users to use Caddy, we will not provide configuration files for other proxys.
+As we would prefer our users to use Caddy, we will not provide configuration files for other proxies.
 
 You will need to reverse proxy everything under following routes:
 - `/_matrix/` - core Matrix C-S and S-S APIs

src/admin/admin.rs

@@ -10,7 +10,8 @@ use crate::{
 };
 
 #[derive(Debug, Parser)]
-#[command(name = "tuwunel", version = tuwunel_core::version())]
+#[command(name = "admin", version = tuwunel_core::version())]
+#[command(arg_required_else_help = true)]
 #[command_dispatch]
 pub(super) enum AdminCommand {
 	#[command(subcommand)]

src/admin/debug/commands.rs

@@ -231,8 +231,9 @@ pub(super) async fn get_remote_pdu(
 		})
 		.await
 	{
-		| Err(e) =>
-			Err!("Remote server did not have PDU or failed sending request to remote server: {e}"),
+		| Err(e) => {
+			Err!("Remote server did not have PDU or failed sending request to remote server: {e}")
+		},
 		| Ok(response) => {
 			let json: CanonicalJsonObject =
 				serde_json::from_str(response.pdu.get()).map_err(|e| {
@@ -374,8 +375,9 @@ pub(super) async fn change_log_level(
 			.reload
 			.reload(&old_filter_layer, Some(handles))
 		{
-			| Err(e) =>
-				return Err!("Failed to modify and reload the global tracing log level: {e}"),
+			| Err(e) => {
+				return Err!("Failed to modify and reload the global tracing log level: {e}");
+			},
 			| Ok(()) => {
 				let value = &self.services.server.config.log;
 				return Ok(format!(
@@ -401,8 +403,9 @@ pub(super) async fn change_log_level(
 			| Ok(()) => {
 				return Ok("Successfully changed log level".to_owned());
 			},
-			| Err(e) =>
-				return Err!("Failed to modify and reload the global tracing log level: {e}"),
+			| Err(e) => {
+				return Err!("Failed to modify and reload the global tracing log level: {e}");
+			},
 		}
 	}
 

src/admin/mod.rs

@@ -1,5 +1,6 @@
 #![allow(clippy::wildcard_imports)]
 #![allow(clippy::enum_glob_use)]
+#![allow(clippy::large_futures)]
 
 pub(crate) mod admin;
 mod tests;

src/api/client/media.rs

@@ -66,6 +66,11 @@ pub(crate) async fn create_content_route(
 		.create(mxc, Some(user), Some(&content_disposition), content_type, &body.file)
 		.await?;
 
+	// Track this upload as potentially being used in an encrypted message soon
+	services
+		.media
+		.retention_track_pending_upload(user.as_str(), &mxc.to_string());
+
 	let blurhash = body.generate_blurhash.then(|| {
 		services
 			.media

src/api/client/message.rs

@@ -99,6 +99,7 @@ pub(crate) async fn get_message_events_route(
 		services
 			.timeline
 			.backfill_if_required(room_id, from)
+			.boxed()
 			.await
 			.log_err()
 			.ok();

src/api/client/space.rs

@@ -44,7 +44,7 @@ pub(crate) async fn get_hierarchy_route(
 		.as_ref()
 		.and_then(|s| PaginationToken::from_str(s).ok());
 
-	// Should prevent unexpeded behaviour in (bad) clients
+	// Should prevent unexpected behaviour in (bad) clients
 	if let Some(ref token) = key {
 		if token.suggested_only != body.suggested_only || token.max_depth != max_depth {
 			return Err!(Request(InvalidParam(

src/api/client/state.rs

@@ -349,7 +349,7 @@ async fn allowed_to_send_state_event(
 				},
 			}
 		},
-		| StateEventType::RoomMember =>
+		| StateEventType::RoomMember => {
 			match json.deserialize_as_unchecked::<RoomMemberEventContent>() {
 				| Ok(membership_content) => {
 					let Ok(_state_key) = UserId::parse(state_key) else {
@@ -394,7 +394,8 @@ async fn allowed_to_send_state_event(
 						 membership state: {e}"
 					)));
 				},
-			},
+			}
+		},
 		| _ => (),
 	}
 

src/api/router/auth.rs

@@ -122,11 +122,13 @@ pub(super) async fn auth(
 			Err(BadRequest(UnknownToken { soft_logout: true }, "Expired access token."))
 		},
 
-		| (AppserviceToken, User(_)) =>
-			Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint."))),
+		| (AppserviceToken, User(_)) => {
+			Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint.")))
+		},
 
-		| (ServerSignatures, Appservice(_) | User(_)) =>
-			Err!(Request(Unauthorized("Server signatures must be used on this endpoint."))),
+		| (ServerSignatures, Appservice(_) | User(_)) => {
+			Err!(Request(Unauthorized("Server signatures must be used on this endpoint.")))
+		},
 
 		| (ServerSignatures, Token::None) => Ok(auth_server(services, request, json_body).await?),
 
@@ -182,8 +184,9 @@ fn check_auth_still_required(services: &Services, metadata: &Metadata, token: &T
 				.require_auth_for_profile_requests =>
 			match token {
 				| Token::Appservice(_) | Token::User(_) => Ok(()),
-				| Token::None | Token::Expired(_) | Token::Invalid =>
-					Err!(Request(MissingToken("Missing or invalid access token."))),
+				| Token::None | Token::Expired(_) | Token::Invalid => {
+					Err!(Request(MissingToken("Missing or invalid access token.")))
+				},
 			},
 		| &get_public_rooms::v3::Request::METADATA
 			if !services
@@ -192,8 +195,9 @@ fn check_auth_still_required(services: &Services, metadata: &Metadata, token: &T
 				.allow_public_room_directory_without_auth =>
 			match token {
 				| Token::Appservice(_) | Token::User(_) => Ok(()),
-				| Token::None | Token::Expired(_) | Token::Invalid =>
-					Err!(Request(MissingToken("Missing or invalid access token."))),
+				| Token::None | Token::Expired(_) | Token::Invalid => {
+					Err!(Request(MissingToken("Missing or invalid access token.")))
+				},
 			},
 		| _ => Ok(()),
 	}

src/core/config/mod.rs

@@ -1530,6 +1530,10 @@ pub struct Config {
 	#[serde(default, with = "serde_regex")]
 	pub forbidden_remote_server_names: RegexSet,
 
+	/// Media retention configuration (flattened; formerly media.retention.*)
+	#[serde(default, alias = "media", alias = "media.retention")]
+	pub media: MediaRetentionConfig,
+
 	/// List of forbidden server names via regex patterns that we will block all
 	/// outgoing federated room directory requests for. Useful for preventing
 	/// our users from wandering into bad servers or spaces.
@@ -2132,6 +2136,28 @@ pub struct WellKnownConfig {
 	pub support_mxid: Option<OwnedUserId>,
 }
 
+#[derive(Clone, Debug, Deserialize, Default)]
+#[config_example_generator(filename = "tuwunel-example.toml", section = "global.media")]
+pub struct MediaRetentionConfig {
+	/// What to do with local media when an event referencing it is redacted.
+	///
+	/// Options:
+	///   "keep" - Never delete media (feature disabled)
+	///   "ask_sender" - Ask the user who sent the message via DM (shows
+	///   ✅/❌/♻️ reactions)   
+	///   "delete_always" - Always delete unreferenced media immediately
+	///
+	/// Default: "keep"
+	///
+	/// Note: Deletion is event-driven and immediate. Users can set
+	/// per-room-type auto-delete preferences using `!user retention` commands
+	/// or the ♻️ reaction when `ask_sender` is enabled.
+	#[serde(default = "default_media_retention_on_redaction")]
+	pub on_redaction: String,
+}
+
+fn default_media_retention_on_redaction() -> String { "keep".to_owned() }
+
 #[derive(Clone, Copy, Debug, Deserialize, Default)]
 #[allow(rustdoc::broken_intra_doc_links, rustdoc::bare_urls)]
 #[config_example_generator(
@@ -2573,6 +2599,10 @@ impl Config {
 	}
 
 	pub fn check(&self) -> Result<(), Error> { check(self) }
+
+	// Media retention helpers
+	#[must_use]
+	pub fn media_retention_on_redaction(&self) -> &str { self.media.on_redaction.as_str() }
 }
 
 fn true_fn() -> bool { true }

src/core/log/capture.rs

@@ -38,7 +38,12 @@ impl EventData {
 	}
 }
 
+impl Default for CaptureManager {
+	fn default() -> Self { Self::new() }
+}
+
 impl CaptureManager {
+	#[must_use]
 	pub fn new() -> Self { Self { captures: Mutex::new(Vec::new()) } }
 
 	pub fn start_capture(&self, span_id: &Id) {

src/database/map/insert.rs

@@ -59,3 +59,33 @@ where
 		self.engine.flush().expect("database flush error");
 	}
 }
+
+/// Atomically write a batch of raw put and delete operations.
+#[implement(super::Map)]
+#[tracing::instrument(skip(self, puts, dels), fields(%self), level = "trace")]
+pub fn write_batch_raw<Ip, Ik>(&self, puts: Ip, dels: Ik)
+where
+	Ip: IntoIterator<Item = (Vec<u8>, Vec<u8>)>,
+	Ik: IntoIterator<Item = Vec<u8>>,
+{
+	let mut batch = WriteBatchWithTransaction::<false>::default();
+	let cf = self.cf();
+	for (k, v) in puts {
+		batch.put_cf(&cf, &k, &v);
+	}
+	for k in dels {
+		batch.delete_cf(&cf, &k);
+	}
+
+	let write_options = &self.write_options;
+	use crate::util::or_else as db_or_else;
+	self.engine
+		.db
+		.write_opt(batch, write_options)
+		.or_else(db_or_else)
+		.expect("database write batch error");
+
+	if !self.engine.corked() {
+		self.engine.flush().expect("database flush error");
+	}
+}

src/database/maps.rs

@@ -108,6 +108,10 @@ pub(super) static MAPS: &[Descriptor] = &[
 		name: "mediaid_file",
 		..descriptor::RANDOM_SMALL
 	},
+	Descriptor {
+		name: "media_retention",
+		..descriptor::RANDOM_SMALL
+	},
 	Descriptor {
 		name: "mediaid_user",
 		..descriptor::RANDOM_SMALL

src/service/admin/create.rs

@@ -30,7 +30,10 @@ pub async fn create_admin_room(&self) -> Result {
 	let server_user = self.services.globals.server_user.as_ref();
 
 	let name = format!("{} Admin Room", self.services.config.server_name);
-	let topic = format!("Manage {} | Run commands prefixed with `!admin` | Run `!admin -h` for help | Documentation: https://github.com/matrix-construct/tuwunel/", self.services.config.server_name);
+	let topic = format!(
+		"Manage {} | Run commands prefixed with `!admin` | Run `!admin -h` for help | Documentation: https://github.com/matrix-construct/tuwunel/",
+		self.services.config.server_name
+	);
 
 	self.services
 		.create

src/service/admin/grant.rs

@@ -143,7 +143,7 @@ pub async fn make_user_admin(&self, user_id: &UserId) -> Result {
 
 	if self.services.server.config.admin_room_notices {
 		let welcome_message = String::from(
-			"## Thank you for trying out tuwunel!\n\nTuwunel is a continuation of conduwuit which was technically a hard fork of Conduit.\n\nHelpful links:\n> GitHub Repo: https://github.com/matrix-construct/tuwunel\n> Documentation: https://github.com/matrix-construct/tuwunel\n> Report issues: https://github.com/matri-construct/tuwunel/issues\n\nFor a list of available commands, send the following message in this room: `!admin --help`"
+			"## Thank you for trying out tuwunel!\n\nTuwunel is a continuation of conduwuit which was technically a hard fork of Conduit.\n\nHelpful links:\n> GitHub Repo: https://github.com/matrix-construct/tuwunel\n> Documentation: https://github.com/matrix-construct/tuwunel\n> Report issues: https://github.com/matri-construct/tuwunel/issues\n\nFor a list of available commands, send the following message in this room: `!admin --help`",
 		);
 
 		// Send welcome message
@@ -209,8 +209,9 @@ pub async fn revoke_admin(&self, user_id: &UserId) -> Result {
 
 		| Err(e) => return Err!(error!(?e, "Failure occurred while attempting revoke.")),
 
-		| Ok(event) if !matches!(event.membership, Invite | Knock | Join) =>
-			return Err!("Cannot revoke {user_id} in membership state {:?}.", event.membership),
+		| Ok(event) if !matches!(event.membership, Invite | Knock | Join) => {
+			return Err!("Cannot revoke {user_id} in membership state {:?}.", event.membership);
+		},
 
 		| Ok(event) => {
 			assert!(

src/service/admin/mod.rs

@@ -46,6 +46,7 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
+#[allow(clippy::enum_variant_names)]
 enum AdminCommandCheckVerdict {
 	NotAdminCommand,
 	AdminEscapeCommand,

src/service/command/mod.rs

@@ -16,7 +16,7 @@ pub struct CommandResult {
 
 pub struct CompletionTree {
 	pub name: String,
-	pub nodes: Vec<CompletionTree>,
+	pub nodes: Vec<Self>,
 }
 
 #[async_trait]

src/service/command/run.rs

@@ -47,12 +47,12 @@ impl Service {
 			let error = Error::from_panic(panic);
 			error!("Panic while processing command: {error:?}");
 			Err(format!(
-						"Panic occurred while processing command:\n\
+				"Panic occurred while processing command:\n\
 						```\n\
 						{error:#?}\n\
 						```\n\
 						Please submit a [bug report](https://github.com/matrix-construct/tuwunel/issues/new).🥺"
-					))
+			))
 		});
 
 		let (output, err) = match result {

src/service/command/run_matrix.rs

@@ -100,13 +100,14 @@ impl Service {
 		if let Some(capture_level) = capture_level {
 			output.push_str(&format_logs(&result.logs, capture_level));
 		}
-		if result.err {
-			output.push_str("Command completed with error:\n");
-		} else {
-			output.push_str("Command completed:\n");
+		if !result.output.starts_with("Usage:") {
+			if result.err {
+				output.push_str("Command completed with error:\n");
+			} else {
+				output.push_str("Command completed:\n");
+			}
 		}
 		output.push_str(&result.output);
-
 		let mut content = RoomMessageEventContent::notice_markdown(output);
 		content.relates_to = Some(Relation::Reply {
 			in_reply_to: InReplyTo { event_id: reply_id.to_owned() },

src/service/media/data.rs

@@ -1,7 +1,7 @@
 use std::{sync::Arc, time::Duration};
 
 use futures::StreamExt;
-use ruma::{Mxc, OwnedMxcUri, UserId, http_headers::ContentDisposition};
+use ruma::{Mxc, OwnedMxcUri, OwnedUserId, UserId, http_headers::ContentDisposition};
 use tuwunel_core::{
 	Err, Result, debug, debug_info, err,
 	utils::{ReadyExt, str_from_bytes, stream::TryIgnore, string_from_bytes},
@@ -153,6 +153,24 @@ impl Data {
 			.await
 	}
 
+	pub(super) async fn get_media_owner(&self, mxc: &str) -> Option<OwnedUserId> {
+		let prefix = (mxc, Interfix);
+		let mut stream = self
+			.mediaid_user
+			.stream_prefix_raw(&prefix)
+			.ignore_err();
+
+		while let Some((_, raw_user)) = stream.next().await {
+			if let Ok(user) = string_from_bytes(raw_user) {
+				if let Ok(user_id) = OwnedUserId::try_from(user) {
+					return Some(user_id);
+				}
+			}
+		}
+
+		None
+	}
+
 	/// Gets all the media keys in our database (this includes all the metadata
 	/// associated with it such as width, height, content-type, etc)
 	pub(crate) async fn get_all_media_keys(&self) -> Vec<Vec<u8>> {