use std::collections::{HashSet, VecDeque}; use axum::extract::State; use conduwuit::{Err, Event, Result, debug, info, trace, utils::to_canonical_object, warn}; use ruma::{OwnedEventId, api::federation::event::get_missing_events}; use serde_json::value::RawValue; use service::rooms::event_handler::GET_MISSING_EVENTS_MAX_BATCH_SIZE; use super::AccessCheck; use crate::Ruma; /// # `POST /_matrix/federation/v1/get_missing_events/{roomId}` /// /// Retrieves events that the sender is missing. pub(crate) async fn get_missing_events_route( State(services): State, body: Ruma, ) -> Result { AccessCheck { services: &services, origin: &body.identity, room_id: &body.room_id, event_id: None, } .assert() .await?; if !services .rooms .state_cache .server_in_room(services.globals.server_name(), &body.room_id) .await { info!( origin = body.identity.as_str(), "Refusing to serve state for room we aren't participating in" ); return Err!(Request(NotFound("This server is not participating in that room."))); } let limit = body .limit .try_into() .unwrap_or(10) .min(GET_MISSING_EVENTS_MAX_BATCH_SIZE); let room_version = services.rooms.state.get_room_version(&body.room_id).await?; let mut queue: VecDeque = VecDeque::from(body.latest_events.clone()); let mut results: Vec> = Vec::with_capacity(limit); let mut seen: HashSet = HashSet::from_iter(body.earliest_events.clone()); while let Some(next_event_id) = queue.pop_front() { if seen.contains(&next_event_id) { trace!(%next_event_id, "already seen event, skipping"); continue; } if results.len() >= limit { debug!(%next_event_id, "reached limit of events to return, breaking"); break; } let mut pdu = match services.rooms.timeline.get_pdu(&next_event_id).await { | Ok(pdu) => pdu, | Err(e) => { warn!("could not find event {next_event_id} while walking missing events: {e}"); continue; }, }; if pdu.room_id_or_hash() != body.room_id { return Err!(Request(Unknown( "Event {next_event_id} is not in room {}", body.room_id ))); } if services .rooms .pdu_metadata .is_event_rejected(pdu.event_id()) .await { debug!(%next_event_id, "event rejected, not traversing"); continue; } if !services .rooms .state_accessor .server_can_see_event(&body.identity, &body.room_id, pdu.event_id()) .await { debug!(%next_event_id, origin = %body.identity, "redacting event origin cannot see"); pdu.redact(&room_version, None)?; } trace!( %next_event_id, prev_events = ?pdu.prev_events().collect::>(), "adding event to results and queueing prev events" ); queue.extend(pdu.prev_events.clone()); seen.insert(next_event_id.clone()); if body.latest_events.contains(&next_event_id) { continue; // Don't include latest_events in results, // but do include their prev_events in the queue } results.push( services .sending .convert_to_outgoing_federation_event(to_canonical_object(pdu)?) .await, ); trace!( %next_event_id, queue_len = queue.len(), seen_len = seen.len(), results_len = results.len(), "event added to results" ); } if !queue.is_empty() { debug!("limit reached before queue was empty"); } results.reverse(); // return oldest first Ok(get_missing_events::v1::Response::new(results)) }