fix

Author: stevessr

src/admin/query/sync.rs

@@ -49,7 +49,7 @@ pub(super) async fn show_connection(
 	// Admin CLI/backend calls don't have an X-Forwarded-For header context,
 	// so pass None for the forwarded-for field.
 	let key = into_connection_key(user_id, device_id, conn_id, None::<String>);
-	let cache = self.services.sync.find_connection(&key)?;
+	let cache = self.services.sync.find_connection(&key).await?;
 
 	let out;
 	{

src/api/client/read_marker.rs

@@ -29,7 +29,8 @@ pub(crate) async fn set_read_marker_route(
 	if body.private_read_receipt.is_some() || body.read_receipt.is_some() {
 		services
 			.pusher
-			.reset_notification_counts(sender_user, &body.room_id);
+			.reset_notification_counts(sender_user, &body.room_id)
+			.await;
 	}
 
 	if let Some(event) = &body.fully_read {
@@ -120,7 +121,8 @@ pub(crate) async fn create_receipt_route(
 	) {
 		services
 			.pusher
-			.reset_notification_counts(sender_user, &body.room_id);
+			.reset_notification_counts(sender_user, &body.room_id)
+			.await;
 	}
 
 	match body.receipt_type {

src/api/client/session/oauth.rs

@@ -54,7 +54,7 @@ fn consume_redirect(state_token: &str) -> Option<String> {
 
 /// OAuth 2.0 token response
 #[derive(Clone, Debug, Serialize, Deserialize)]
-pub(super) struct OAuthTokenResponse {
+pub(crate) struct OAuthTokenResponse {
 	pub access_token: String,
 	pub token_type: String,
 	#[serde(skip_serializing_if = "Option::is_none")]
@@ -69,7 +69,7 @@ pub(super) struct OAuthTokenResponse {
 
 /// OAuth 2.0 userinfo response
 #[derive(Clone, Debug, Serialize, Deserialize)]
-pub(super) struct OAuthUserInfo {
+pub(crate) struct OAuthUserInfo {
 	pub sub: String,
 	#[serde(skip_serializing_if = "Option::is_none")]
 	pub name: Option<String>,
@@ -326,7 +326,7 @@ pub(crate) async fn oauth_callback_route(
 		use ruma::UserId;
 		let user_id_ref: &UserId = user_id.as_ref();
 		let token = utils::random_string(TOKEN_LENGTH);
-		services
+		let _ = services
 			.users
 			.create_login_token(user_id_ref, &token);
 		token

src/api/client/sync/v5.rs

@@ -80,7 +80,7 @@ pub(crate) async fn sync_events_v5_route(
 	InsecureClientIp(client): InsecureClientIp,
 	body: Ruma<Request>,
 ) -> Result<Response> {
-	let (sender_user, sender_device) = body.sender();
+	let (_sender_user, _sender_device) = body.sender();
 	let request = &body.body;
 	let since = request
 		.pos
@@ -113,11 +113,12 @@ pub(crate) async fn sync_events_v5_route(
 		request.conn_id.as_deref(),
 		x_forwarded_for,
 	);
-	let conn_val = since
-		.ne(&0)
-		.then(|| services.sync.find_connection(&conn_key))
-		.unwrap_or_else(|| Ok(services.sync.init_connection(&conn_key)))
-		.map_err(|_| err!(Request(UnknownPos("Connection lost; restarting sync stream."))))?;
+	let conn_val = if since.ne(&0) {
+		services.sync.find_connection(&conn_key).boxed()
+	} else {
+		async { Ok(services.sync.init_connection(&conn_key).await) }.boxed()
+	}
+	.await?;
 
 	let conn = conn_val.lock();
 	let ping_presence = services

src/service/pusher/mod.rs

@@ -61,6 +61,25 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
+#[implement(Service)]
+#[tracing::instrument(level = "debug", skip(self))]
+pub async fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) {
+	let count = self.services.globals.next_count();
+
+	let userroom_id = (user_id, room_id);
+	self.db
+		.userroomid_highlightcount
+		.put(userroom_id, 0_u64);
+	self.db
+		.userroomid_notificationcount
+		.put(userroom_id, 0_u64);
+
+	let roomuser_id = (room_id, user_id);
+	self.db
+		.roomuserid_lastnotificationread
+		.put(roomuser_id, *count);
+}
+
 #[implement(Service)]
 pub async fn set_pusher(
 	&self,

src/service/pusher/notification.rs

@@ -5,51 +5,9 @@ use tuwunel_core::{
 };
 use tuwunel_database::{Deserialized, Interfix};
 
-pub struct Service {
-	db: Data,
-	services: Arc<crate::services::OnceServices>,
-}
-
-struct Data {
-	userroomid_notificationcount: Arc<Map>,
-	userroomid_highlightcount: Arc<Map>,
-	roomuserid_lastnotificationread: Arc<Map>,
-}
 
-impl crate::Service for Service {
-	fn build(args: &crate::Args<'_>) -> Result<Arc<Self>> {
-		Ok(Arc::new(Self {
-			db: Data {
-				userroomid_notificationcount: args.db["userroomid_notificationcount"].clone(),
-				userroomid_highlightcount: args.db["userroomid_highlightcount"].clone(),
-				roomuserid_lastnotificationread: args.db["roomuserid_lastnotificationread"]
-					.clone(),
-			},
-			services: args.services.clone(),
-		}))
-	}
 
-	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
-}
 
-#[implement(Service)]
-#[tracing::instrument(level = "debug", skip(self))]
-pub fn reset_notification_counts(&self, user_id: &UserId, room_id: &RoomId) {
-	let count = self.services.globals.next_count();
-
-	let userroom_id = (user_id, room_id);
-	self.db
-		.userroomid_highlightcount
-		.put(userroom_id, 0_u64);
-	self.db
-		.userroomid_notificationcount
-		.put(userroom_id, 0_u64);
-
-	let roomuser_id = (room_id, user_id);
-	self.db
-		.roomuserid_lastnotificationread
-		.put(roomuser_id, *count);
-}
 
 #[implement(super::Service)]
 #[tracing::instrument(level = "debug", skip(self), ret(level = "trace"))]

src/service/rooms/timeline/append.rs

@@ -169,7 +169,8 @@ where
 
 	self.services
 		.pusher
-		.reset_notification_counts(pdu.sender(), pdu.room_id());
+		.reset_notification_counts(pdu.sender(), pdu.room_id())
+		.await;
 
 	let count = PduCount::Normal(*next_count1);
 	let pdu_id: RawPduId = PduId { shortroomid, count }.into();

src/service/sync/mod.rs

@@ -111,15 +111,15 @@ pub async fn clear_connections(
 	self.connections
 		.lock()
 		.await
-		.retain(|(conn_user_id, conn_device_id, conn_conn_id), _| {
+		.retain(|(conn_user_id, conn_device_id, conn_conn_id, conn_xff), _| {
 			let retain = user_id.is_none_or(is_equal_to!(conn_user_id))
 				&& device_id.is_none_or(is_equal_to!(conn_device_id))
 				&& (conn_id.is_none() || conn_id == conn_conn_id.as_ref());
 
 			if !retain {
 				self.db
 					.userdeviceconnid_conn
-					.del((conn_user_id, conn_device_id, conn_conn_id));
+					.del((conn_user_id, conn_device_id, conn_conn_id, conn_xff));
 			}
 
 			retain
@@ -129,10 +129,10 @@ pub async fn clear_connections(
 #[implement(Service)]
 #[tracing::instrument(level = "debug", skip(self))]
 pub async fn drop_connection(&self, key: &ConnectionKey) {
-	let mut cache = self.connections.lock().await;
-
-	self.db.userdeviceconnid_conn.del(key);
-	cache.remove(key);
+	self.connections
+		.lock()
+		.await
+		.remove(key);
 }
 
 #[implement(Service)]
@@ -220,15 +220,7 @@ pub async fn is_connection_stored(&self, key: &ConnectionKey) -> bool {
 	self.db.userdeviceconnid_conn.contains(key).await
 }
 
-#[inline]
-pub fn into_connection_key<U, D, C>(user_id: U, device_id: D, conn_id: Option<C>) -> ConnectionKey
-where
-	U: Into<OwnedUserId>,
-	D: Into<OwnedDeviceId>,
-	C: Into<ConnectionId>,
-{
-	(user_id.into(), device_id.into(), conn_id.map(Into::into))
-}
+
 
 #[implement(Connection)]
 #[tracing::instrument(level = "debug", skip(self, service))]
@@ -373,66 +365,48 @@ fn some_or_sticky<T: Clone>(target: Option<&T>, cached: &mut Option<T>) {
 	}
 }
 
-#[implement(Service)]
-pub fn clear_connections(
-	&self,
-	user_id: Option<&UserId>,
-	device_id: Option<&DeviceId>,
-	conn_id: Option<&ConnectionId>,
-) {
-	self.connections.lock().expect("locked").retain(
-		|(conn_user_id, conn_device_id, conn_conn_id, _conn_xff), _| {
-			!(user_id.is_none_or(is_equal_to!(conn_user_id))
-				&& device_id.is_none_or(is_equal_to!(conn_device_id))
-				&& (conn_id.is_none() || conn_id == conn_conn_id.as_ref()))
-		},
-	);
-}
 
-#[implement(Service)]
-pub fn drop_connection(&self, key: &ConnectionKey) {
-	self.connections
-		.lock()
-		.expect("locked")
-		.remove(key);
-}
 
 #[implement(Service)]
-pub fn list_connections(&self) -> Vec<ConnectionKey> {
+#[tracing::instrument(level = "trace", skip(self))]
+pub async fn list_connections(&self) -> Vec<ConnectionKey> {
 	self.connections
 		.lock()
-		.expect("locked")
+		.await
 		.keys()
 		.cloned()
 		.collect()
 }
 
 #[implement(Service)]
-pub fn init_connection(&self, key: &ConnectionKey) -> ConnectionVal {
+#[tracing::instrument(level = "debug", skip(self))]
+pub async fn init_connection(&self, key: &ConnectionKey) -> ConnectionVal {
 	self.connections
 		.lock()
-		.expect("locked")
+		.await
 		.entry(key.clone())
 		.and_modify(|existing| *existing = ConnectionVal::default())
 		.or_default()
 		.clone()
 }
 
 #[implement(Service)]
-pub fn find_connection(&self, key: &ConnectionKey) -> Result<ConnectionVal> {
+#[tracing::instrument(level = "debug", skip(self))]
+pub async fn find_connection(&self, key: &ConnectionKey) -> Result<ConnectionVal> {
 	self.connections
 		.lock()
-		.expect("locked")
+		.await
 		.get(key)
 		.cloned()
 		.ok_or_else(|| err!(Request(NotFound("Connection not found."))))
 }
 
 #[implement(Service)]
-pub fn contains_connection(&self, key: &ConnectionKey) -> bool {
+#[tracing::instrument(level = "trace", skip(self))]
+pub async fn contains_connection(&self, key: &ConnectionKey) -> bool {
 	self.connections
 		.lock()
-		.expect("locked")
+		.await
 		.contains_key(key)
 }