tuwunel_api_client/read_marker/
mod.rs

1mod read_markers;
2mod receipt;
3
4use futures::future::try_join;
5use ruma::{
6	EventId, MilliSecondsSinceUnixEpoch, RoomId, UserId,
7	events::{
8		RoomAccountDataEventType,
9		fully_read::{FullyReadEvent, FullyReadEventContent},
10		receipt::ReceiptThread,
11	},
12};
13use serde_json::to_value;
14use tuwunel_core::{
15	Err, Result, debug, debug_warn, err,
16	utils::result::{LogErr, NotFound},
17};
18use tuwunel_matrix::{Event, PduCount, PduId};
19use tuwunel_service::{Services, rooms::read_receipt::PrivateRead};
20
21pub(crate) use self::{read_markers::set_read_marker_route, receipt::create_receipt_route};
22
23#[tracing::instrument(level = "trace", skip_all)]
24async fn ensure_receipt_visible(
25	services: &Services,
26	room_id: &RoomId,
27	user_id: &UserId,
28	event_id: &EventId,
29) -> Result {
30	let event = services
31		.timeline
32		.get_pdu(event_id)
33		.await
34		.map_err(|_| err!(Request(NotFound("Event not found."))))?;
35
36	if event.room_id() != room_id
37		|| !services
38			.state_accessor
39			.user_can_see_event(user_id, &event)
40			.await
41	{
42		return Err!(Request(NotFound("Event not found.")));
43	}
44
45	Ok(())
46}
47
48/// Stores `event` as the user's fully-read marker unless that would move the
49/// marker backwards.
50///
51/// The marker moves only when `event` sits strictly after the stored one or
52/// either position does not resolve, so writes that arrive one after another
53/// never move it back; two in flight at once are not ordered against each
54/// other. MSC4446's `allow_backward` skips the comparison. Positions compare on
55/// the global counter, whichever room the event belongs to.
56#[tracing::instrument(level = "trace", skip_all)]
57async fn set_fully_read(
58	services: &Services,
59	room_id: &RoomId,
60	user_id: &UserId,
61	event: &EventId,
62	allow_backward: bool,
63) -> Result {
64	if !allow_backward {
65		let stored = services
66			.account_data
67			.get_room(room_id, user_id, RoomAccountDataEventType::FullyRead)
68			.await
69			.optional()
70			.inspect_err(
71				|error| debug_warn!(%user_id, %room_id, %error, "Unreadable fully-read marker, replacing it"),
72			)
73			.ok()
74			.flatten()
75			.map(|stored: FullyReadEvent| stored.content.event_id);
76
77		if !services
78			.read_receipt
79			.receipt_advanced(stored.as_deref(), event)
80			.await
81		{
82			return Ok(());
83		}
84	}
85
86	let fully_read = FullyReadEvent {
87		content: FullyReadEventContent { event_id: event.to_owned() },
88	};
89
90	services
91		.account_data
92		.update(
93			Some(room_id),
94			user_id,
95			RoomAccountDataEventType::FullyRead,
96			&to_value(fully_read)?,
97		)
98		.await
99}
100
101/// Resolves `event` to its timeline position and stores the private read
102/// marker for `thread` there.
103///
104/// Returns whether the marker advanced. A backfilled event carries no forward
105/// position, so it is skipped like a non-advancing write rather than failing
106/// the request.
107async fn set_private_marker(
108	services: &Services,
109	room_id: &RoomId,
110	user_id: &UserId,
111	event: &EventId,
112	thread: &ReceiptThread,
113) -> Result<bool> {
114	let (pdu_id, shortroomid) =
115		try_join(services.timeline.get_pdu_id(event), services.short.get_shortroomid(room_id))
116			.await
117			.map_err(|_| err!(Request(NotFound("Event not found."))))?;
118
119	let pdu_id = PduId::from(pdu_id);
120
121	if pdu_id.shortroomid != shortroomid {
122		return Err!(Request(NotFound("Event not found.")));
123	}
124
125	let PduCount::Normal(count) = pdu_id.count else {
126		debug!(%user_id, %room_id, %event, "Skipping private read marker at a backfilled event");
127		return Ok(false);
128	};
129
130	let advanced = services
131		.read_receipt
132		.private_read_set(PrivateRead {
133			room_id,
134			user_id,
135			count,
136			ts: MilliSecondsSinceUnixEpoch::now(),
137			thread,
138			announce: true,
139		})
140		.await;
141
142	Ok(advanced)
143}
144
145/// Clears the receipt's notification counts and refreshes the push badge.
146///
147/// The refresh follows every advance because the gateway can hold a stale
148/// badge while the stored count is already zero; only a delivery reconciles
149/// it.
150async fn reset_and_refresh_badge(
151	services: &Services,
152	user_id: &UserId,
153	room_id: &RoomId,
154	acknowledged: Option<&EventId>,
155	thread: &ReceiptThread,
156) {
157	services
158		.pusher
159		.reset_notification_counts_for_thread(user_id, room_id, acknowledged, thread)
160		.await;
161
162	services
163		.sending
164		.refresh_push_badge(user_id)
165		.await
166		.log_err()
167		.ok();
168}
tuwunel_api_client/read_marker/mod.rs

tuwunel_api_client/read_marker/
mod.rs
