tuwunel_api_federation/
backfill.rs

1use std::cmp;
2
3use axum::extract::State;
4use futures::{FutureExt, StreamExt, TryStreamExt};
5use ruma::{MilliSecondsSinceUnixEpoch, api::federation::backfill::get_backfill};
6use tuwunel_core::{
7	Result,
8	utils::{IterStream, ReadyExt, math::usize_from_ruma_bounded},
9};
10use tuwunel_matrix::PduCount;
11
12use super::AccessCheck;
13use crate::Ruma;
14
15/// arbitrary number but synapse's is 100 and we can handle lots of these
16/// anyways
17const LIMIT_MAX: usize = 150;
18/// no spec defined number but we can handle a lot of these
19const LIMIT_DEFAULT: usize = 50;
20
21/// # `GET /_matrix/federation/v1/backfill/<room_id>`
22///
23/// Retrieves events from before the sender joined the room, if the room's
24/// history visibility allows.
25pub(crate) async fn get_backfill_route(
26	State(services): State<crate::State>,
27	ref body: Ruma<get_backfill::v1::Request>,
28) -> Result<get_backfill::v1::Response> {
29	AccessCheck {
30		services: &services,
31		origin: body.origin(),
32		room_id: &body.room_id,
33		event_id: None,
34	}
35	.check()
36	.await?;
37
38	let limit = usize_from_ruma_bounded(body.limit, LIMIT_DEFAULT, LIMIT_MAX);
39
40	let from = body
41		.v
42		.iter()
43		.stream()
44		.filter_map(|event_id| {
45			services
46				.timeline
47				.get_pdu_count(event_id)
48				.map(Result::ok)
49		})
50		.ready_fold(PduCount::min(), cmp::max)
51		.await;
52
53	let room_version = services
54		.state
55		.get_room_version(&body.room_id)
56		.await
57		.ok();
58
59	Ok(get_backfill::v1::Response {
60		origin_server_ts: MilliSecondsSinceUnixEpoch::now(),
61
62		origin: services.globals.server_name().to_owned(),
63
64		pdus: services
65			.timeline
66			.pdus_rev(None, &body.room_id, Some(from.saturating_add(1)))
67			.try_filter_map(async |(_, pdu)| {
68				Ok(services
69					.state_accessor
70					.server_can_see_event(body.origin(), &pdu.room_id, &pdu.event_id)
71					.await
72					.then_some(pdu))
73			})
74			.try_filter_map(async |pdu| {
75				Ok(services
76					.timeline
77					.get_pdu_json(&pdu.event_id)
78					.await
79					.ok())
80			})
81			.take(limit)
82			.and_then(|pdu| {
83				services
84					.state_accessor
85					.erased_for_server(body.origin(), pdu)
86					.map(Ok)
87			})
88			.and_then(|pdu| {
89				services
90					.federation
91					.format_pdu_into(pdu, room_version.as_ref())
92					.map(Ok)
93			})
94			.try_collect()
95			.boxed() // size firewall
96			.await?,
97	})
98}
tuwunel_api_federation/backfill.rs

tuwunel_api_federation/
backfill.rs
