fix: Use a database migration to fix corrupted us.cloke.msc4175.tz fields

(cherry picked from commit 4a893ce4cc81487bcf324dccefd8184ddef5b215)

src/service/migrations.rs

@@ -9,7 +9,7 @@ use conduwuit::{
 	},
 	warn,
 };
-use futures::{FutureExt, StreamExt};
+use futures::{FutureExt, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use ruma::{
 	OwnedUserId, RoomId, UserId,
@@ -138,6 +138,14 @@ async fn migrate(services: &Services) -> Result<()> {
 		info!("Migration: Bumped database version to 17");
 	}
 
+	if db["global"]
+		.get(FIXED_CORRUPT_MSC4133_FIELDS_MARKER)
+		.await
+		.is_not_found()
+	{
+		fix_corrupt_msc4133_fields(services).await?;
+	}
+
 	if services.globals.db.database_version().await < 18 {
 		services.globals.db.bump_database_version(18);
 		info!("Migration: Bumped database version to 18");
@@ -564,3 +572,54 @@ async fn fix_readreceiptid_readreceipt_duplicates(services: &Services) -> Result
 	db["global"].insert(b"fix_readreceiptid_readreceipt_duplicates", []);
 	db.db.sort()
 }
+
+const FIXED_CORRUPT_MSC4133_FIELDS_MARKER: &'static [u8] = b"fix_corrupt_msc4133_fields";
+async fn fix_corrupt_msc4133_fields(services: &Services) -> Result {
+	use serde_json::{Value, from_slice};
+	type KeyVal<'a> = ((OwnedUserId, String), &'a [u8]);
+
+	warn!("Fixing corrupted `us.cloke.msc4175.tz` fields...");
+
+	let db = &services.db;
+	let cork = db.cork_and_sync();
+	let useridprofilekey_value = db["useridprofilekey_value"].clone();
+
+	let (total, fixed) = useridprofilekey_value
+		.stream()
+		.try_fold(
+			(0_usize, 0_usize),
+			async |(mut total, mut fixed),
+			       ((user, key), value): KeyVal<'_>|
+			       -> Result<(usize, usize)> {
+				if let Err(error) = from_slice::<Value>(value) {
+					// Due to an old bug, some conduwuit databases have `us.cloke.msc4175.tz` user
+					// profile fields with raw strings instead of quoted JSON ones.
+					// This migration fixes that.
+					let new_value = if key == "us.cloke.msc4175.tz" {
+						Value::String(String::from_utf8(value.to_vec())?)
+					} else {
+						return Err!(
+							"failed to deserialize msc4133 key {} of user {}: {}",
+							key,
+							user,
+							error
+						);
+					};
+
+					useridprofilekey_value.put((user, key), new_value);
+					fixed += 1;
+				}
+				total += 1;
+
+				Ok((total, fixed))
+			},
+		)
+		.await?;
+
+	drop(cork);
+	info!(?total, ?fixed, "Fixed corrupted `us.cloke.msc4175.tz` fields.");
+
+	db["global"].insert(FIXED_CORRUPT_MSC4133_FIELDS_MARKER, []);
+	db.db.sort()?;
+	Ok(())
+}

src/service/users/mod.rs

@@ -1102,34 +1102,6 @@ impl Service {
 		Ok(user_id)
 	}
 
-	#[inline]
-	fn parse_profile_kv(
-		&self,
-		user_id: &UserId,
-		key: &str,
-		value: Vec<u8>,
-	) -> Result<serde_json::Value> {
-		match serde_json::from_slice(&value) {
-			| Ok(value) => Ok(value),
-			| Err(error) => {
-				// Due to an old bug, some conduwuit databases have `us.cloke.msc4175.tz` user
-				// profile fields with raw strings instead of quoted JSON ones.
-				if key == "us.cloke.msc4175.tz" {
-					// TODO insert a hint about this being a cold path
-					debug_warn!(
-						"Fixing corrupt `us.cloke.msc4175.tz` field in the profile of {}",
-						user_id
-					);
-					let raw_tz = serde_json::Value::String(String::from_utf8(value)?);
-					self.set_profile_key(user_id, "us.cloke.msc4175.tz", Some(raw_tz.clone()));
-					Ok(raw_tz)
-				} else {
-					Err(error.into())
-				}
-			},
-		}
-	}
-
 	/// Gets a specific user profile key
 	pub async fn profile_key(
 		&self,
@@ -1141,7 +1113,7 @@ impl Service {
 			.useridprofilekey_value
 			.qry(&key)
 			.await
-			.and_then(|handle| self.parse_profile_kv(user_id, profile_key, handle.to_vec()))
+			.and_then(|handle| serde_json::from_slice(&handle).map_err(|err| err.into()))
 	}
 
 	/// Gets all the user's profile keys and values in an iterator
@@ -1156,10 +1128,7 @@ impl Service {
 			.useridprofilekey_value
 			.stream_prefix(&prefix)
 			.ignore_err()
-			.map(|((_, key), value): KeyVal<'_>| {
-				let value = self.parse_profile_kv(user_id, &key, value.to_vec())?;
-				Ok((key, value))
-			})
+			.map(|((_, key), value): KeyVal<'_>| Ok((key, serde_json::from_slice(value)?)))
 			.ignore_err()
 	}