tuwunel_service/rooms/pdu_metadata/
references.rs

1use futures::{Stream, StreamExt, TryFutureExt};
2use ruma::{EventId, OwnedEventId, RoomId};
3use tuwunel_core::{
4	PduId, Result, implement,
5	matrix::{Event, Pdu},
6	trace,
7	utils::{
8		stream::{ReadyExt, TryIgnore, WidebandExt},
9		u64_from_u8,
10	},
11};
12use tuwunel_database::Interfix;
13
14use super::{
15	Service,
16	typed_relations::{Tag, prefix},
17};
18
19/// Cap on the `m.reference` bundle chunk; /relations is the paginated fallback.
20const BUNDLE_MAX: usize = 100;
21
22/// MSC2675/MSC3267: the event ids of `parent`'s `m.reference` children, oldest
23/// first, from the typed index, capped at `BUNDLE_MAX`. Empty when
24/// `parent` is redacted or unreferenced. The ids come from the index value (the
25/// child shorteventid) without loading the children, so the chunk is filtered
26/// for neither ignored users nor history visibility. The ignored-user posture
27/// matches the /relations endpoint, which also does not filter relation
28/// children by ignored sender; the history-visibility posture matches the
29/// thread and edit bundles and is less strict than /relations, which does
30/// filter children by visibility.
31#[implement(Service)]
32#[tracing::instrument(skip_all, level = "trace")]
33pub(super) async fn references(&self, parent: &Pdu) -> Vec<OwnedEventId> {
34	if parent.is_redacted() {
35		return Vec::new();
36	}
37
38	let Ok(parent_id) = self
39		.services
40		.timeline
41		.get_pdu_id(parent.event_id())
42		.map_ok(PduId::from)
43		.await
44	else {
45		return Vec::new();
46	};
47
48	self.referenced_children(parent_id)
49		.take(BUNDLE_MAX)
50		.collect()
51		.await
52}
53
54#[implement(Service)]
55fn referenced_children(&self, parent_id: PduId) -> impl Stream<Item = OwnedEventId> + Send + '_ {
56	let prefix = prefix(parent_id.shortroomid, parent_id.count, Tag::Reference);
57	let seek = prefix.clone();
58
59	self.db
60		.relatesto_typed
61		.raw_stream_from(seek.as_slice())
62		.ignore_err()
63		.ready_take_while(move |(key, _)| key.starts_with(&prefix))
64		.map(|(_, val)| u64_from_u8(val))
65		.wide_filter_map(async |short| {
66			self.services
67				.short
68				.get_eventid_from_short(short)
69				.await
70				.ok()
71		})
72}
73
74#[implement(Service)]
75#[tracing::instrument(skip_all, level = "debug")]
76pub fn mark_as_referenced<'a, I>(&self, room_id: &RoomId, event_ids: I)
77where
78	I: Iterator<Item = &'a EventId>,
79{
80	for event_id in event_ids {
81		let key = (room_id, event_id);
82
83		self.db.referencedevents.put_raw(key, []);
84	}
85}
86
87#[implement(Service)]
88#[tracing::instrument(skip(self), level = "debug", ret)]
89pub async fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> bool {
90	let key = (room_id, event_id);
91
92	self.db.referencedevents.qry(&key).await.is_ok()
93}
94
95#[implement(Service)]
96#[tracing::instrument(skip(self), level = "debug")]
97pub fn mark_event_soft_failed(&self, event_id: &EventId) {
98	self.db.softfailedeventids.insert(event_id, []);
99}
100
101#[implement(Service)]
102#[tracing::instrument(skip(self), level = "debug", ret)]
103pub async fn is_event_soft_failed(&self, event_id: &EventId) -> bool {
104	self.db
105		.softfailedeventids
106		.get(event_id)
107		.await
108		.is_ok()
109}
110
111/// Streams owned event IDs with soft-fail markers.
112///
113/// Each ID is copied before the database cursor advances.
114#[implement(Service)]
115pub fn soft_failed_event_ids(&self) -> impl Stream<Item = OwnedEventId> + Send + '_ {
116	self.db
117		.softfailedeventids
118		.keys()
119		.ignore_err()
120		.map(|event_id: &EventId| event_id.to_owned())
121}
122
123/// Clears one event's soft-fail marker.
124///
125/// A later processing attempt can evaluate the event again.
126#[implement(Service)]
127pub fn clear_event_soft_failed(&self, event_id: &EventId) {
128	self.db.softfailedeventids.remove(event_id);
129}
130
131#[implement(Service)]
132#[tracing::instrument(skip(self), level = "debug")]
133pub async fn delete_all_referenced_for_room(&self, room_id: &RoomId) -> Result {
134	let prefix = (room_id, Interfix);
135
136	self.db
137		.referencedevents
138		.keys_prefix_raw(&prefix)
139		.ignore_err()
140		.ready_for_each(|key| {
141			trace!(?key, "Removing key");
142			self.db.referencedevents.remove(key);
143		})
144		.await;
145
146	Ok(())
147}
tuwunel_service/rooms/pdu_metadata/references.rs

tuwunel_service/rooms/pdu_metadata/
references.rs
