WIP

.typos.toml

@@ -1,5 +1,5 @@
 [files]
-extend-exclude = ["*.csr", "*.lock"]
+extend-exclude = ["*.csr"]
 
 [default.extend-words]
 "allocatedp" = "allocatedp"

Cargo.lock

@@ -3695,7 +3695,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.10.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "assign",
  "js_int",
@@ -3715,7 +3715,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.10.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3727,7 +3727,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.18.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "as_variant",
  "assign",
@@ -3750,7 +3750,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -3782,7 +3782,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.28.1"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "as_variant",
  "indexmap 2.9.0",
@@ -3807,7 +3807,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "bytes",
  "headers",
@@ -3829,7 +3829,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.9.5"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "js_int",
  "thiserror 2.0.12",
@@ -3838,7 +3838,7 @@ dependencies = [
 [[package]]
 name = "ruma-identity-service-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3848,7 +3848,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.13.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "cfg-if",
  "proc-macro-crate",
@@ -3863,7 +3863,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.9.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3875,7 +3875,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.15.0"
-source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=a48665b682be1016cea53ea5e7787442dfe7c1de#a48665b682be1016cea53ea5e7787442dfe7c1de"
+source = "git+https://forgejo.ellis.link/continuwuation/ruwuma?rev=d6870a7fb7f6cccff63f7fd0ff6c581bad80e983#d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",

Cargo.toml

@@ -350,7 +350,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://forgejo.ellis.link/continuwuation/ruwuma"
 #branch = "conduwuit-changes"
-rev = "a48665b682be1016cea53ea5e7787442dfe7c1de"
+rev = "d6870a7fb7f6cccff63f7fd0ff6c581bad80e983"
 features = [
     "compat",
     "rand",
@@ -381,7 +381,7 @@ features = [
     "unstable-msc4121",
     "unstable-msc4125",
     "unstable-msc4186",
-    "unstable-msc4203", # sending to-device events to appservices
+    "unstable-msc4203", # sending to-device events to appservices 
     "unstable-msc4210", # remove legacy mentions
     "unstable-extensible-events",
     "unstable-pdu",

src/api/client/sync/v3.rs

@@ -808,7 +808,7 @@ async fn load_joined_room(
 			let typings = services
 				.rooms
 				.typing
-				.typings_event_for_user(room_id, sender_user)
+				.typings_all(room_id, sender_user)
 				.await?;
 
 			Ok(vec![serde_json::from_str(&serde_json::to_string(&typings)?)?])

src/api/client/sync/v5.rs

@@ -33,7 +33,6 @@ use ruma::{
 	events::{
 		AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType, TimelineEventType,
 		room::member::{MembershipState, RoomMemberEventContent},
-		typing::TypingEventContent,
 	},
 	serde::Raw,
 	uint,
@@ -206,9 +205,6 @@ pub(crate) async fn sync_events_v5_route(
 		_ = tokio::time::timeout(duration, watcher).await;
 	}
 
-	let typing = collect_typing_events(services, sender_user, &body, &todo_rooms).await?;
-	response.extensions.typing = typing;
-
 	trace!(
 		rooms = ?response.rooms.len(),
 		account_data = ?response.extensions.account_data.rooms.len(),
@@ -292,8 +288,6 @@ where
 	Rooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
 	AllRooms: Iterator<Item = &'a RoomId> + Clone + Send + 'a,
 {
-	// TODO MSC4186: Implement remaining list filters: is_dm, is_encrypted,
-	// room_types.
 	for (list_id, list) in &body.lists {
 		let active_rooms: Vec<_> = match list.filters.as_ref().and_then(|f| f.is_invite) {
 			| None => all_rooms.clone().collect(),
@@ -671,62 +665,6 @@ where
 	}
 	Ok(rooms)
 }
-
-async fn collect_typing_events(
-	services: &Services,
-	sender_user: &UserId,
-	body: &sync_events::v5::Request,
-	todo_rooms: &TodoRooms,
-) -> Result<sync_events::v5::response::Typing> {
-	if !body.extensions.typing.enabled.unwrap_or(false) {
-		return Ok(sync_events::v5::response::Typing::default());
-	}
-	let rooms: Vec<_> = body.extensions.typing.rooms.clone().unwrap_or_else(|| {
-		body.room_subscriptions
-			.keys()
-			.map(ToOwned::to_owned)
-			.collect()
-	});
-	let lists: Vec<_> = body
-		.extensions
-		.typing
-		.lists
-		.clone()
-		.unwrap_or_else(|| body.lists.keys().map(ToOwned::to_owned).collect::<Vec<_>>());
-
-	if rooms.is_empty() && lists.is_empty() {
-		return Ok(sync_events::v5::response::Typing::default());
-	}
-
-	let mut typing_response = sync_events::v5::response::Typing::default();
-	for (room_id, (required_state_request, timeline_limit, roomsince)) in todo_rooms {
-		if services.rooms.typing.last_typing_update(room_id).await? <= *roomsince {
-			continue;
-		}
-
-		match services
-			.rooms
-			.typing
-			.typing_users_for_user(room_id, sender_user)
-			.await
-		{
-			| Ok(typing_users) => {
-				typing_response.rooms.insert(
-					room_id.to_owned(), // Already OwnedRoomId
-					Raw::new(&sync_events::v5::response::SyncTypingEvent {
-						content: TypingEventContent::new(typing_users),
-					})?,
-				);
-			},
-			| Err(e) => {
-				warn!(%room_id, "Failed to get typing events for room: {}", e);
-			},
-		}
-	}
-
-	Ok(typing_response)
-}
-
 async fn collect_account_data(
 	services: &Services,
 	(sender_user, _, globalsince, body): (&UserId, &DeviceId, u64, &sync_events::v5::Request),

src/service/rooms/read_receipt/mod.rs

@@ -1,6 +1,6 @@
 mod data;
 
-use std::{collections::BTreeMap, sync::Arc};
+use std::{collections::{BTreeMap, HashMap}, sync::Arc};
 
 use conduwuit::{
 	Result, debug, err,
@@ -9,12 +9,9 @@ use conduwuit::{
 };
 use futures::{Stream, TryFutureExt, try_join};
 use ruma::{
-	OwnedEventId, OwnedUserId, RoomId, UserId,
 	events::{
-		AnySyncEphemeralRoomEvent, SyncEphemeralRoomEvent,
-		receipt::{ReceiptEvent, ReceiptEventContent, Receipts},
-	},
-	serde::Raw,
+		receipt::{ReceiptEvent, ReceiptEventContent, ReceiptType, Receipts}, AnySyncEphemeralRoomEvent, SyncEphemeralRoomEvent
+	}, serde::Raw, OwnedEventId, OwnedUserId, RoomId, UserId
 };
 
 use self::data::{Data, ReceiptItem};
@@ -47,19 +44,48 @@ impl crate::Service for Service {
 }
 
 impl Service {
-	/// Replaces the previous read receipt.
+	/// Updates the public read receipt (`m.read`) based on the incoming event.
+	/// If the event referenced by the new public receipt is newer than the current
+	/// private read marker (`m.read.private`), the private marker is also updated
+	/// to match the public receipt's position.
 	pub async fn readreceipt_update(
 		&self,
 		user_id: &UserId,
 		room_id: &RoomId,
 		event: &ReceiptEvent,
 	) {
-		self.db.readreceipt_update(user_id, room_id, event).await;
-		self.services
-			.sending
-			.flush_room(room_id)
-			.await
-			.expect("room flush failed");
+		debug!(target: "readreceipt", %room_id, %user_id, "Updating read receipt in database.");
+
+		// 2. Find the maximum PDU count for the m.read event(s) referenced in the new receipt
+		let mut max_new_public_pdu_count: Option<PduCount> = None;
+		for (event_id, receipts) in event.content.0.iter() {
+			// Check if this event_id has an m.read receipt for the target user
+			if let Some(user_receipts) = receipts.get(&ReceiptType::Read) {
+				if user_receipts.contains_key(user_id) {
+					// Try to get the PDU count (timeline position) for this event_id
+					match self.services.timeline.get_pdu_count(event_id).await {
+						Ok(count) => {
+							// Update the maximum count found so far
+							let current_max = max_new_public_pdu_count.unwrap_or(PduCount::Normal(0));
+							max_new_public_pdu_count = Some(current_max.max(count));
+							debug!(target: "readreceipt", %room_id, %user_id, %event_id, count, "Found PDU count for new public receipt event.");
+						}
+						Err(e) => {
+							warn!(
+								target: "readreceipt", %room_id, %user_id, %event_id,
+								"Failed to get PDU count for event ID from new public read receipt: {}",
+								e
+							);
+						}
+					}
+				}
+			}
+		}
+
+		// Flush the sending queue for the room to notify clients
+		if let Err(e) = self.services.sending.flush_room(room_id).await {
+			warn!(target: "readreceipt", %room_id, %user_id, "Failed to flush room after read receipt update: {}", e);
+		}
 	}
 
 	/// Gets the latest private read receipt from the user in the room

src/service/rooms/typing/mod.rs

@@ -179,15 +179,18 @@ impl Service {
 			.unwrap_or(0))
 	}
 
-	pub async fn typing_users_for_user(
+	/// Returns a new typing EDU.
+	pub async fn typings_all(
 		&self,
 		room_id: &RoomId,
 		sender_user: &UserId,
-	) -> Result<Vec<OwnedUserId>> {
+	) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
 		let room_typing_indicators = self.typing.read().await.get(room_id).cloned();
 
 		let Some(typing_indicators) = room_typing_indicators else {
-			return Ok(Vec::new());
+			return Ok(SyncEphemeralRoomEvent {
+				content: ruma::events::typing::TypingEventContent { user_ids: Vec::new() },
+			});
 		};
 
 		let user_ids: Vec<_> = typing_indicators
@@ -204,19 +207,8 @@ impl Service {
 			.collect()
 			.await;
 
-		Ok(user_ids)
-	}
-
-	/// Returns a new typing EDU.
-	pub async fn typings_event_for_user(
-		&self,
-		room_id: &RoomId,
-		sender_user: &UserId,
-	) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
 		Ok(SyncEphemeralRoomEvent {
-			content: ruma::events::typing::TypingEventContent {
-				user_ids: self.typing_users_for_user(room_id, sender_user).await?,
-			},
+			content: ruma::events::typing::TypingEventContent { user_ids },
 		})
 	}