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

feat: Move get-remote-pdu to graph command, simplify

Browse Source
This commit is contained in:
timedout committed 2026-10-31 23:09:37 +00:00
1 parent b651fbd088
commit 2f82475811
4 files changed
+36 -168
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/admin/debug/commands.rs
+2 -142
View File
@@ -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::<Vec<_>>()
.drain(1..self.body.len().saturating_sub(1))
.filter_map(|pdu| EventId::parse(pdu).ok())
.collect::<Vec<_>>();
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
src/admin/debug/mod.rs
-24
View File
@@ -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,
src/admin/room/graph.rs
+33 -1
View File
@@ -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()?;
src/service/rooms/event_handler/parse_incoming_pdu.rs
+1 -1
View File
@@ -93,7 +93,7 @@ impl super::Service {
room_version_rules: Option<&RoomVersionRules>,
) -> Result<Parsed> {
let value = serde_json::from_str::<CanonicalJsonObject>(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")