From 2f8247581102fe63164ed391e5d75bf9fba6b98f Mon Sep 17 00:00:00 2001 From: timedout Date: Sat, 31 Oct 2026 23:09:37 +0000 Subject: [PATCH] feat: Move `get-remote-pdu` to graph command, simplify --- src/admin/debug/commands.rs | 144 +----------------- src/admin/debug/mod.rs | 24 --- src/admin/room/graph.rs | 34 ++++- .../rooms/event_handler/parse_incoming_pdu.rs | 2 +- 4 files changed, 36 insertions(+), 168 deletions(-) diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index 012764785..e1fe9afd3 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -8,14 +8,14 @@ use std::{ use conduwuit::{ Err, Result, debug_error, err, info, matrix::{Event, pdu::PduEvent}, - trace, utils, + utils, utils::{ stream::{IterStream, ReadyExt}, string::EMPTY, }, warn, }; -use futures::{FutureExt, StreamExt, TryStreamExt}; +use futures::{StreamExt, TryStreamExt}; use lettre::message::Mailbox; use resolvematrix::{ resolution::ResolvedDestination, @@ -294,146 +294,6 @@ impl crate::Context<'_> { .await } - pub(super) async fn get_remote_pdu_list( - &self, - server: OwnedServerName, - force: bool, - ) -> Result { - if !self.services.server.config.allow_federation { - return Err!("Federation is disabled on this homeserver.",); - } - - if server == self.services.globals.server_name() { - return Err!( - "Not allowed to send federation requests to ourselves. Please use `get-pdu` for \ - fetching local PDUs from the database.", - ); - } - - if self.body.len() < 2 - || !self.body[0].trim().starts_with("```") - || self.body.last().unwrap_or(&EMPTY).trim() != "```" - { - return Err!("Expected code block in command body. Add --help for details.",); - } - - let list = self - .body - .iter() - .collect::>() - .drain(1..self.body.len().saturating_sub(1)) - .filter_map(|pdu| EventId::parse(pdu).ok()) - .collect::>(); - - let mut failed_count: usize = 0; - let mut success_count: usize = 0; - - for event_id in list { - if force { - match self.get_remote_pdu(event_id.clone(), server.clone()).await { - | Err(e) => { - failed_count = failed_count.saturating_add(1); - self.services - .admin - .send_text(&format!("Failed to get remote PDU, ignoring error: {e}")) - .await; - - warn!("Failed to get remote PDU, ignoring error: {e}"); - }, - | _ => { - success_count = success_count.saturating_add(1); - }, - } - } else { - self.get_remote_pdu(event_id.clone(), server.clone()) - .await?; - success_count = success_count.saturating_add(1); - } - } - - let out = format!( - "Fetched {success_count} remote PDUs successfully with {failed_count} failures" - ); - - self.write_str(&out).await - } - - pub(super) async fn get_remote_pdu( - &self, - event_id: OwnedEventId, - server: OwnedServerName, - ) -> Result { - if !self.services.server.config.allow_federation { - return Err!("Federation is disabled on this homeserver."); - } - - if server == self.services.globals.server_name() { - return Err!( - "Not allowed to send federation requests to ourselves. Please use `get-pdu` for \ - fetching local PDUs.", - ); - } - - match self - .services - .sending - .send_federation_request( - &server, - ruma::api::federation::event::get_event::v1::Request::new(event_id.clone()), - ) - .await - { - | Err(e) => { - return 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| { - warn!( - "Requested event ID {event_id} from server but failed to convert \ - from RawValue to CanonicalJsonObject (malformed event/response?): \ - {e}" - ); - err!(Request(Unknown( - "Received response from server but failed to parse PDU" - ))) - })?; - - trace!("Attempting to parse PDU: {:?}", &response.pdu); - let _parsed_pdu = { - let parsed_result = self - .services - .rooms - .event_handler - .parse_incoming_pdu(&response.pdu, None) - .boxed() - .await; - - let (event_id, value, room_id) = match parsed_result { - | Ok(t) => t, - | Err(e) => { - warn!("Failed to parse PDU: {e}"); - info!("Full PDU: {:?}", &response.pdu); - return Err!( - "Failed to parse PDU remote server {server} sent us: {e}" - ); - }, - }; - - vec![(event_id, value, room_id)] - }; - - let text = serde_json::to_string_pretty(&json)?; - let msg = "Got PDU from specified server:"; - write!(self, "{msg}. Event body:\n```json\n{text}\n```") - }, - } - .await - } - pub(super) async fn force_device_list_updates(&self) -> Result { // Force E2EE device list updates for all users self.services diff --git a/src/admin/debug/mod.rs b/src/admin/debug/mod.rs index bc20d07aa..1f75e70c9 100644 --- a/src/admin/debug/mod.rs +++ b/src/admin/debug/mod.rs @@ -40,30 +40,6 @@ pub enum DebugCommand { /// the command. ParsePdu, - /// Attempts to retrieve a PDU from a remote server. **Does not** insert - /// it into the database - /// or persist it anywhere. - GetRemotePdu { - /// An event ID (a $ followed by the base64 reference hash) - event_id: OwnedEventId, - - /// Argument for us to attempt to fetch the event from the - /// specified remote server. - server: OwnedServerName, - }, - - /// Same as `get-remote-pdu` but accepts a codeblock newline delimited - /// list of PDUs and a single server to fetch from - GetRemotePduList { - /// Argument for us to attempt to fetch all the events from the - /// specified remote server. - server: OwnedServerName, - - /// If set, ignores errors, else stops at the first error/failure. - #[arg(short, long)] - force: bool, - }, - /// Forces device lists for all local and remote users to be updated (as /// having new keys available) ForceDeviceListUpdates, diff --git a/src/admin/room/graph.rs b/src/admin/room/graph.rs index cdf1587c4..052ed7cc0 100644 --- a/src/admin/room/graph.rs +++ b/src/admin/room/graph.rs @@ -2,7 +2,7 @@ use clap::Subcommand; use conduwuit::{Err, Event, Result, err, utils::stream::BroadbandExt}; use conduwuit_macros::admin_command_dispatch; use futures::StreamExt; -use ruma::{OwnedEventId, OwnedRoomId, OwnedRoomOrAliasId}; +use ruma::{OwnedEventId, OwnedRoomId, OwnedRoomOrAliasId, OwnedServerName}; #[admin_command_dispatch] #[derive(Debug, Subcommand)] @@ -18,6 +18,12 @@ pub enum GraphCommand { event_id: OwnedEventId, }, + /// Fetches a PDU over federation. Does not store the result. + FetchPDU { + event_id: OwnedEventId, + server_name: OwnedServerName, + }, + /// Yields the first PDU we became aware of in a room. /// This is always the room's create event for local rooms, or the first /// membership event that introduced us to the remote room. @@ -120,6 +126,32 @@ impl crate::Context<'_> { .await } + pub(super) async fn fetch_pdu( + &self, + event_id: OwnedEventId, + server: OwnedServerName, + ) -> Result { + let response = self + .services + .sending + .send_federation_request( + &server, + ruma::api::federation::event::get_event::v1::Request::new(event_id.clone()), + ) + .await?; + let (real_event_id, room_id, pdu_json) = self + .services + .rooms + .event_handler + .parse_incoming_pdu(&response.pdu, None) + .await?; + self.write_str(&format!( + "Event ID: {real_event_id}\nRoom ID: {room_id}\nPDU:\n```json\n{}\n```", + serde_json::to_string_pretty(&pdu_json)? + )) + .await + } + pub(super) async fn first_pdu_in_room(&self, room_id: OwnedRoomId) -> Result { self.bail_restricted()?; diff --git a/src/service/rooms/event_handler/parse_incoming_pdu.rs b/src/service/rooms/event_handler/parse_incoming_pdu.rs index 3d9803440..0512f5170 100644 --- a/src/service/rooms/event_handler/parse_incoming_pdu.rs +++ b/src/service/rooms/event_handler/parse_incoming_pdu.rs @@ -93,7 +93,7 @@ impl super::Service { room_version_rules: Option<&RoomVersionRules>, ) -> Result { let value = serde_json::from_str::(pdu.get()).map_err(|e| { - err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}"))) + err!(BadServerResponse(debug_warn!("Error parsing incoming event as JSON {e:?}"))) })?; let event_type = value .get("type")