tuwunel_api_client/

context.rs

use axum::extract::State;
use futures::{
	FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt,
	future::{OptionFuture, join, join3, try_join},
};
use ruma::{
	DeviceId, EventId, OwnedEventId, RoomId, UInt, UserId,
	api::client::{context::get_context, filter::RoomEventFilter},
	events::{AnyStateEvent, StateEventType},
	serde::Raw,
};
use tuwunel_core::{
	Err, Result, at, debug_warn, err, ref_at,
	utils::{
		BoolExt, IterStream,
		future::TryExtExt,
		math::usize_from_ruma_bounded,
		stream::{BroadbandExt, ReadyExt, TryIgnore, WidebandExt},
	},
};
use tuwunel_matrix::{
	Event,
	pdu::{PduEvent, RawPduId},
};
use tuwunel_service::{
	Services,
	rooms::{
		lazy_loading,
		lazy_loading::{Options, Witness},
		short::{ShortRoomId, ShortStateHash, ShortStateKey},
		timeline::PdusIterItem,
	},
};

use crate::{
	Ruma, is_ignored_pdu,
	message::{
		add_membership_unsigned, event_filter, event_filters, ignored_filter,
		lazy_loading_witness, related_by_filter, with_membership,
	},
};

const LIMIT_MAX: usize = 100;
const LIMIT_DEFAULT: usize = 10;

/// # `GET /_matrix/client/r0/rooms/{roomId}/context/{eventId}`
///
/// Allows loading room history around an event.
///
/// - Only works if the user is joined (TODO: always allow, but only show events
///   if the user was joined, depending on history_visibility)
pub(crate) async fn get_context_route(
	State(services): State<crate::State>,
	body: Ruma<get_context::v3::Request>,
) -> Result<get_context::v3::Response> {
	event_context(&services, ContextArgs {
		room_id: &body.room_id,
		event_id: &body.event_id,
		sender_user: body.sender_user(),
		sender_device: body.sender_device.as_deref(),
		filter: &body.filter,
		limit: Some(body.limit),
		bypass_visibility: false,
	})
	.await
}

/// Shared inputs for [`event_context`], the core behind both the client-server
/// `/context` route and the admin room-context endpoint.
pub struct ContextArgs<'a> {
	/// Room containing the requested event.
	pub room_id: &'a RoomId,

	/// Event at the center of the timeline window.
	pub event_id: &'a EventId,

	/// User requesting the event context.
	pub sender_user: &'a UserId,

	/// Requesting device used for lazy-loading membership state.
	pub sender_device: Option<&'a DeviceId>,

	/// Filter for timeline events and membership state.
	pub filter: &'a RoomEventFilter,

	/// Requested maximum number of surrounding timeline events.
	pub limit: Option<UInt>,

	/// Skip the visibility and ignore checks on the base event and both halves,
	/// and the room-state check on an anchor without state, for admin callers.
	pub bypass_visibility: bool,
}

/// Loads the timeline window around an event with its state and aggregations,
/// applying (unless `bypass_visibility`) the per-user visibility and ignore
/// checks.
///
/// Powers the client-server `/context` route and its admin bypass twin.
pub async fn event_context(
	services: &Services,
	args: ContextArgs<'_>,
) -> Result<get_context::v3::Response> {
	let ContextArgs {
		room_id,
		event_id,
		sender_user,
		sender_device,
		filter,
		limit,
		bypass_visibility,
	} = args;

	if !services.metadata.exists(room_id).await {
		return Err!(Request(Forbidden("Room does not exist to this server")));
	}

	let limit = limit
		.map_or(LIMIT_DEFAULT, |limit| usize_from_ruma_bounded(limit, LIMIT_DEFAULT, LIMIT_MAX));

	let (base_id, base_pdu) =
		resolve_base_event(services, room_id, event_id, sender_user, bypass_visibility).await?;

	let base_count = base_id.pdu_count();

	let encrypted = services
		.state_accessor
		.is_encrypted_room(room_id)
		.await;

	let shortroomid = services.short.get_shortroomid(room_id).await?;

	let base_event = async {
		let item = if bypass_visibility {
			(base_count, base_pdu)
		} else {
			ignored_filter(services, (base_count, base_pdu), sender_user).await?
		};

		Some(add_membership_unsigned(services, item, sender_user, encrypted).await)
	};

	let half = TimelineHalf {
		services,
		filter,
		shortroomid,
		sender_user,
		encrypted,
		bypass_visibility,
	};

	let events_before = collect_timeline_half(
		half,
		services
			.timeline
			.pdus_rev(Some(sender_user), room_id, Some(base_count)),
		limit / 2,
	);

	let events_after = collect_timeline_half(
		half,
		services
			.timeline
			.pdus(Some(sender_user), room_id, Some(base_count)),
		limit.div_ceil(2),
	);

	let (base_event, events_before, events_after): (_, Vec<_>, Vec<_>) =
		join3(base_event, events_before, events_after)
			.boxed() // size firewall
			.await;

	let lazy_loading_context = lazy_loading::Context {
		user_id: sender_user,
		device_id: sender_device,
		room_id,
		token: Some(base_count.into_unsigned()),
		options: Some(&filter.lazy_load_options),
		mode: lazy_loading::Mode::Update,
	};

	let lazy_loading_witnessed = filter
		.lazy_load_options
		.is_enabled()
		.then_async(|| {
			let witnessed = base_event
				.iter()
				.chain(events_before.iter())
				.chain(events_after.iter());

			lazy_loading_witness(services, &lazy_loading_context, witnessed)
		});

	let state_at = events_after
		.last()
		.map(ref_at!(1))
		.map_or(event_id, Event::event_id);

	let state_ids = load_state_ids(services, room_id, state_at, sender_user, bypass_visibility);

	let (lazy_loading_witnessed, state_ids) = join(lazy_loading_witnessed, state_ids).await;

	let state = build_state_response(
		services,
		state_ids?,
		lazy_loading_witnessed.unwrap_or_default(),
		filter,
		sender_user,
		encrypted,
	)
	.await;

	let event = OptionFuture::from(base_event.map(at!(1)).map(|pdu| {
		services
			.pdu_metadata
			.bundle_aggregations(sender_user, pdu)
	}))
	.await
	.map(Event::into_format);

	Ok(get_context::v3::Response {
		event,

		start: events_before
			.last()
			.map(at!(0))
			.or(Some(base_count))
			.as_ref()
			.map(ToString::to_string),

		// `end` is one past the base so a backward page from it still yields the base;
		// `start` stays at `base_count` (a bare count can't suit both directions).
		end: events_after
			.last()
			.map(at!(0))
			.or_else(|| Some(base_count.saturating_add(1)))
			.as_ref()
			.map(ToString::to_string),

		events_before: events_before
			.into_iter()
			.map(at!(1))
			.map(Event::into_format)
			.collect(),

		events_after: events_after
			.into_iter()
			.map(at!(1))
			.map(Event::into_format)
			.collect(),

		state,
	})
}

async fn resolve_base_event(
	services: &Services,
	room_id: &RoomId,
	event_id: &EventId,
	sender_user: &UserId,
	bypass_visibility: bool,
) -> Result<(RawPduId, PduEvent)> {
	let lookup = || {
		let base_id = services
			.timeline
			.get_pdu_id(event_id)
			.map_err(|_| err!(Request(NotFound("Event not found."))));

		let base_pdu = services
			.timeline
			.get_pdu(event_id)
			.map_err(|_| err!(Request(NotFound("Base event not found."))));

		try_join(base_id, base_pdu)
	};

	let resolve_remote = services
		.config
		.federation
		.history
		.fetch_unreceived_contexts
		&& services.config.federation.enable;

	let (base_id, base_pdu) = match lookup().await {
		| Ok(found) => found,
		| Err(e) if !resolve_remote => return Err(e),
		| Err(_) => {
			services
				.timeline
				.fetch_remote_event(room_id, event_id)
				.await
				.ok();

			lookup().await?
		},
	};

	if base_pdu.room_id != *room_id || base_pdu.event_id != *event_id {
		return Err!(Request(NotFound("Base event not found.")));
	}

	if !bypass_visibility
		&& !services
			.state_accessor
			.user_can_see_event(sender_user, &base_pdu)
			.await
	{
		debug_warn!(
			req_evt = ?event_id, ?base_id, ?room_id, %sender_user,
			"Event requested but the requester is not allowed to see it."
		);

		return Err!(Request(NotFound("Event not found.")));
	}

	if !bypass_visibility && is_ignored_pdu(services, &base_pdu, sender_user).await {
		return Err!(HttpJson(NOT_FOUND, {
			"errcode": "M_SENDER_IGNORED",
			"error": "You have ignored the user that sent this event",
			"sender": base_pdu.sender().as_str(),
		}));
	}

	Ok((base_id, base_pdu.without_private_unless_sender(Some(sender_user))))
}

/// Shared inputs for the two [`collect_timeline_half`] calls assembling the
/// before and after windows; only the stream and take count differ per call.
#[derive(Clone, Copy)]
struct TimelineHalf<'a> {
	services: &'a Services,
	filter: &'a RoomEventFilter,
	shortroomid: ShortRoomId,
	sender_user: &'a UserId,
	encrypted: bool,
	bypass_visibility: bool,
}

async fn collect_timeline_half<'a, S>(
	half: TimelineHalf<'a>,
	pdus: S,
	take: usize,
) -> Vec<PdusIterItem>
where
	S: Stream<Item = Result<PdusIterItem>> + Send + 'a,
{
	let TimelineHalf {
		services,
		filter,
		shortroomid,
		sender_user,
		encrypted,
		bypass_visibility,
	} = half;

	pdus.ignore_err()
		.ready_filter_map(|item| event_filter(item, filter))
		.wide_filter_map(|item| related_by_filter(services, shortroomid, filter, item))
		.wide_filter_map(|item| event_filters(services, sender_user, item, bypass_visibility))
		.take(take)
		.wide_then(|item| add_membership_unsigned(services, item, sender_user, encrypted))
		.wide_then(async |(count, pdu)| {
			let pdu = services
				.pdu_metadata
				.bundle_aggregations(sender_user, pdu)
				.await;

			(count, pdu)
		})
		.collect()
		.await
}

/// Loads the room state at the event that anchors the response's state.
///
/// An anchor without recorded state, such as a room's create event or a
/// backfilled event, falls back to the state `fallback_shortstatehash` selects
/// for the requester, which may be none.
async fn load_state_ids(
	services: &Services,
	room_id: &RoomId,
	state_at: &EventId,
	sender_user: &UserId,
	bypass_visibility: bool,
) -> Result<Vec<(ShortStateKey, OwnedEventId)>> {
	let state_full_ids = |shortstatehash| {
		services
			.state_accessor
			.state_full_ids(shortstatehash)
	};

	services
		.state
		.pdu_shortstatehash(state_at)
		.map_ok(Some)
		.or_else(|_| fallback_shortstatehash(services, room_id, sender_user, bypass_visibility))
		.map_ok(|shortstatehash| {
			shortstatehash
				.into_iter()
				.stream()
				.flat_map(state_full_ids)
				.map(Ok)
		})
		.map_err(|e| err!(Database("State not found: {e}")))
		.try_flatten_stream()
		.try_collect()
		.boxed() // Send generality
		.await
}

/// Selects the state a requester reads in place of an event's missing state.
///
/// Admitted requesters reach the selection, which checks access again; departed
/// members read their departure snapshot, admins included. Only an admin may
/// read current state when the selection admits none.
async fn fallback_shortstatehash(
	services: &Services,
	room_id: &RoomId,
	sender_user: &UserId,
	bypass_visibility: bool,
) -> Result<Option<ShortStateHash>> {
	let admitted = bypass_visibility
		|| services
			.state_accessor
			.user_can_see_state_events(sender_user, room_id)
			.await;

	match admitted
		.then_async(|| {
			services
				.state_accessor
				.user_visible_shortstatehash(sender_user, room_id, None)
		})
		.await
		.transpose()?
		.flatten()
	{
		| None if bypass_visibility =>
			services
				.state
				.get_room_shortstatehash(room_id)
				.map_ok(Some)
				.await,
		| shortstatehash => Ok(shortstatehash),
	}
}

async fn build_state_response(
	services: &Services,
	state_ids: Vec<(ShortStateKey, OwnedEventId)>,
	lazy_loading_witnessed: Witness,
	filter: &RoomEventFilter,
	sender_user: &UserId,
	encrypted: bool,
) -> Vec<Raw<AnyStateEvent>> {
	let shortstatekeys = state_ids.iter().map(at!(0)).stream();
	let shorteventids = state_ids.iter().map(ref_at!(1)).stream();

	services
		.short
		.multi_get_statekey_from_short(shortstatekeys)
		.zip(shorteventids)
		.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
		.ready_filter_map(|((event_type, state_key), event_id)| {
			if filter.lazy_load_options.is_enabled()
				&& event_type == StateEventType::RoomMember
				&& state_key
					.as_str()
					.try_into()
					.is_ok_and(|user_id: &UserId| !lazy_loading_witnessed.contains(user_id))
			{
				return None;
			}

			Some(event_id)
		})
		.broad_filter_map(|event_id: &OwnedEventId| {
			services.timeline.get_pdu(event_id.as_ref()).ok()
		})
		.map(|pdu| pdu.without_private_unless_sender(Some(sender_user)))
		.broad_then(|pdu| with_membership(services, pdu, sender_user, encrypted))
		.map(Event::into_format)
		.collect()
		.await
}