tuwunel_api/client/
threads.rs

1use std::collections::BTreeSet;
2
3use axum::extract::State;
4use futures::{StreamExt, TryStreamExt};
5use ruma::{
6	OwnedUserId,
7	api::client::threads::get_threads,
8	events::{GlobalAccountDataEventType, ignored_user_list::IgnoredUserListEvent},
9};
10use tuwunel_core::{
11	Err, Result, at,
12	matrix::{
13		Event,
14		pdu::{PduCount, PduEvent},
15	},
16	result::{FlatOk, LogErr},
17	utils::stream::TryWidebandExt,
18};
19use tuwunel_service::rooms::pdu_metadata::IgnoredThreadView;
20
21use crate::Ruma;
22
23/// # `GET /_matrix/client/r0/rooms/{roomId}/threads`
24pub(crate) async fn get_threads_route(
25	State(services): State<crate::State>,
26	ref body: Ruma<get_threads::v1::Request>,
27) -> Result<get_threads::v1::Response> {
28	let sender_user = body.sender_user();
29	let room_id = &body.room_id;
30
31	if !services.metadata.exists(room_id).await {
32		return Err!(Request(Forbidden("Room does not exist to this server")));
33	}
34
35	if !services
36		.state_accessor
37		.user_can_see_room(sender_user, room_id)
38		.await
39	{
40		return Err!(Request(Forbidden("You don't have permission to view this room.")));
41	}
42
43	// Use limit or else 10, with maximum 100
44	let limit = body
45		.limit
46		.map(usize::try_from)
47		.flat_ok()
48		.unwrap_or(10)
49		.min(100);
50
51	let from: PduCount = body
52		.from
53		.as_deref()
54		.map(str::parse)
55		.transpose()?
56		.unwrap_or_else(PduCount::max);
57
58	// MSC3856: the requester's ignore list adjusts the served threads.
59	let ignored: BTreeSet<OwnedUserId> = services
60		.account_data
61		.get_global(sender_user, GlobalAccountDataEventType::IgnoredUserList)
62		.await
63		.map(|event: IgnoredUserListEvent| event.content.ignored_users.into_keys().collect())
64		.unwrap_or_default();
65
66	// One extra row probes whether the list continues past this page.
67	let mut threads: Vec<(PduCount, PduEvent)> = services
68		.threads
69		.threads_until(sender_user, room_id, from, &body.include)
70		.try_filter_map(async |(count, pdu)| {
71			Ok(services
72				.state_accessor
73				.user_can_see_event(sender_user, room_id, &pdu.event_id)
74				.await
75				.then_some((count, pdu)))
76		})
77		.try_filter_map(async |(count, pdu)| {
78			let view = match ignored.is_empty() {
79				| true => IgnoredThreadView::Unchanged,
80				| false =>
81					services
82						.pdu_metadata
83						.ignored_thread_view(sender_user, &ignored, &pdu)
84						.await,
85			};
86
87			Ok(match view {
88				| IgnoredThreadView::Omitted => None,
89				| view => Some((count, pdu, view)),
90			})
91		})
92		.take(limit.saturating_add(1))
93		.wide_and_then(async |(count, pdu, view)| {
94			let pdu = services
95				.pdu_metadata
96				.bundle_aggregations(sender_user, pdu)
97				.await;
98
99			Ok((count, apply_ignored_view(pdu, view)))
100		})
101		.try_collect()
102		.await?;
103
104	let more = threads.len() > limit;
105
106	threads.truncate(limit);
107
108	Ok(get_threads::v1::Response {
109		next_batch: threads
110			.last()
111			.filter(|_| more)
112			.map(at!(0))
113			.as_ref()
114			.map(ToString::to_string),
115
116		chunk: threads
117			.into_iter()
118			.map(at!(1))
119			.map(Event::into_format)
120			.collect(),
121	})
122}
123
124/// MSC3856 ignored-user adjustments, applied after the bundle pass corrects
125/// the served `unsigned`: the redacted root replaces content only and keeps
126/// that `unsigned`, minus any `m.replace` bundle (a folded edit shares the
127/// root's sender, so it would re-serve the ignored content).
128fn apply_ignored_view(mut pdu: PduEvent, view: IgnoredThreadView) -> PduEvent {
129	let IgnoredThreadView::Adjusted { root, count, latest } = view else {
130		return pdu;
131	};
132
133	if let Some(count) = count {
134		pdu.set_thread_count(count).log_err().ok();
135	}
136
137	if let Some(latest) = latest {
138		pdu.set_thread_latest_event(&latest)
139			.log_err()
140			.ok();
141	}
142
143	match root {
144		| None => pdu,
145		| Some(mut root) => {
146			root.unsigned = pdu.unsigned;
147			root.remove_replacement_bundle().log_err().ok();
148
149			*root
150		},
151	}
152}
tuwunel_api/client/threads.rs

tuwunel_api/client/
threads.rs
