add admin command to delete all remote media files

src/service/admin.rs

@@ -23,8 +23,8 @@ use ruma::{
         TimelineEventType,
     },
     signatures::verify_json,
-    EventId, MilliSecondsSinceUnixEpoch, OwnedMxcUri, OwnedRoomId, RoomId,
-    RoomVersionId, ServerName, UserId,
+    EventId, MilliSecondsSinceUnixEpoch, OwnedMxcUri, OwnedRoomId,
+    OwnedServerName, RoomId, RoomVersionId, ServerName, UserId,
 };
 use serde_json::value::to_raw_value;
 use tokio::sync::{mpsc, Mutex, RwLock};
@@ -185,6 +185,17 @@ enum AdminCommand {
         mxc: OwnedMxcUri,
     },
 
+    /// Delete cached remote media from the database.
+    ///
+    /// This media may still be fetched and cached again in the future.
+    DeleteRemoteMedia {
+        /// If specified, only delete remote media from this origin.
+        ///
+        /// If not specified, all remote media will be deleted.
+        #[clap(long)]
+        origin: Option<OwnedServerName>,
+    },
+
     /// Verify json signatures
     /// [commandbody]()
     /// # ```
@@ -803,6 +814,50 @@ impl Service {
                 services().media.delete(mxc).await?;
                 RoomMessageEventContent::text_plain("Media deleted.")
             }
+            AdminCommand::DeleteRemoteMedia {
+                origin,
+            } => {
+                if origin.as_deref() == Some(services().globals.server_name()) {
+                    return Ok(RoomMessageEventContent::text_plain(
+                        "Specified origin is this server. Will not delete \
+                         anything.",
+                    ));
+                }
+
+                let mut count = 0;
+
+                // The `media.iter_all()` iterator is not `Send`, so spawn it in
+                // a separate thread and send the results over a channel.
+                let (tx, mut rx) = mpsc::channel(1);
+                tokio::task::spawn_blocking(move || {
+                    for mxc in services().media.iter_all() {
+                        if tx.blocking_send(mxc).is_err() {
+                            break;
+                        }
+                    }
+                });
+
+                while let Some(mxc) = rx.recv().await {
+                    let mxc = mxc?;
+                    let server_name = mxc.server_name();
+
+                    if server_name == Ok(services().globals.server_name()) {
+                        continue;
+                    }
+                    if let Some(origin) = &origin {
+                        if server_name != Ok(origin) {
+                            continue;
+                        }
+                    }
+
+                    count += 1;
+                    services().media.delete(mxc).await?;
+                }
+
+                RoomMessageEventContent::text_plain(format!(
+                    "{count} media objects deleted."
+                ))
+            }
             AdminCommand::DeactivateUser {
                 leave_rooms,
                 user_id,