refactor to pdu.get_content() for serde_json::from_ elim

Signed-off-by: Jason Volk <jason@zemos.net>

src/api/client/sync.rs

@@ -635,17 +635,8 @@ async fn load_joined_room(
 				.await?
 				.ready_filter(|(_, pdu)| pdu.kind == RoomMember)
 				.filter_map(|(_, pdu)| async move {
-					let Ok(content) = serde_json::from_str::<RoomMemberEventContent>(pdu.content.get()) else {
+					let content: RoomMemberEventContent = pdu.get_content().ok()?;
-						return None;
+					let user_id: &UserId = pdu.state_key.as_deref().map(TryInto::try_into).flat_ok()?;
-					};
-
-					let Some(state_key) = &pdu.state_key else {
-						return None;
-					};
-
-					let Ok(user_id) = UserId::parse(state_key) else {
-						return None;
-					};
 
 					if user_id == sender_user {
 						return None;
@@ -656,22 +647,17 @@ async fn load_joined_room(
 						return None;
 					}
 
-					if !services
+					let is_invited = services.rooms.state_cache.is_invited(user_id, room_id);
-						.rooms
+
-						.state_cache
+					let is_joined = services.rooms.state_cache.is_joined(user_id, room_id);
-						.is_joined(&user_id, room_id)
+
-						.await && services
+					if !is_joined.await && is_invited.await {
-						.rooms
-						.state_cache
-						.is_invited(&user_id, room_id)
-						.await
-					{
 						return None;
 					}
 
-					Some(user_id)
+					Some(user_id.to_owned())
 				})
-				.collect::<HashSet<_>>()
+				.collect::<HashSet<OwnedUserId>>()
 				.await;
 
 			Ok::<_, Error>((
@@ -839,11 +825,9 @@ async fn load_joined_room(
 							continue;
 						}
 
-						let new_membership = serde_json::from_str::<RoomMemberEventContent>(state_event.content.get())
+						let content: RoomMemberEventContent = state_event.get_content()?;
-							.map_err(|_| Error::bad_database("Invalid PDU in database."))?
-							.membership;
 
-						match new_membership {
+						match content.membership {
 							MembershipState::Join => {
 								// A new user joined an encrypted room
 								if !share_encrypted_room(services, sender_user, &user_id, Some(room_id)).await {
@@ -1357,12 +1341,8 @@ pub(crate) async fn sync_events_v4_route(
 										continue;
 									}
 
-									let new_membership =
+									let content: RoomMemberEventContent = pdu.get_content()?;
-										serde_json::from_str::<RoomMemberEventContent>(pdu.content.get())
+									match content.membership {
-											.map_err(|_| Error::bad_database("Invalid PDU in database."))?
-											.membership;
-
-									match new_membership {
 										MembershipState::Join => {
 											// A new user joined an encrypted room
 											if !share_encrypted_room(&services, sender_user, &user_id, Some(room_id))

src/service/rooms/event_handler/mod.rs

@@ -614,9 +614,7 @@ impl Service {
 							}
 						},
 						_ => {
-							let content = serde_json::from_str::<RoomRedactionEventContent>(incoming_pdu.content.get())
+							let content: RoomRedactionEventContent = incoming_pdu.get_content()?;
-								.map_err(|_| Error::bad_database("Invalid content in redaction pdu."))?;
-
 							if let Some(redact_id) = &content.redacts {
 								!self
 									.services
@@ -1432,10 +1430,10 @@ impl Service {
 	}
 
 	fn get_room_version_id(create_event: &PduEvent) -> Result<RoomVersionId> {
-		let create_event_content: RoomCreateEventContent = serde_json::from_str(create_event.content.get())
+		let content: RoomCreateEventContent = create_event.get_content()?;
-			.map_err(|e| err!(Database("Invalid create event: {e}")))?;
+		let room_version = content.room_version;
 
-		Ok(create_event_content.room_version)
+		Ok(room_version)
 	}
 
 	#[inline]

src/service/rooms/pdu_metadata/mod.rs

@@ -102,7 +102,7 @@ impl Service {
 					return false;
 				}
 
-				let Ok(content) = serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get()) else {
+				let Ok(content) = pdu.get_content::<ExtractRelatesToEventId>() else {
 					return false;
 				};
 

src/service/rooms/spaces/mod.rs

@@ -596,13 +596,11 @@ impl Service {
 				.await
 				.map_err(|e| err!(Database("Event {id:?} in space state not found: {e:?}")))?;
 
-			if serde_json::from_str::<SpaceChildEventContent>(pdu.content.get())
+			if let Ok(content) = pdu.get_content::<SpaceChildEventContent>() {
-				.ok()
+				if content.via.is_empty() {
-				.map(|c| c.via)
-				.map_or(true, |v| v.is_empty())
-			{
 					continue;
 				}
+			}
 
 			if OwnedRoomId::try_from(state_key).is_ok() {
 				children_pdus.push(pdu.to_stripped_spacechild_state_event());

src/service/rooms/state/mod.rs

@@ -93,28 +93,17 @@ impl Service {
 
 		pin_mut!(event_ids);
 		while let Some(event_id) = event_ids.next().await {
-			let Ok(pdu) = self.services.timeline.get_pdu_json(&event_id).await else {
+			let Ok(pdu) = self.services.timeline.get_pdu(&event_id).await else {
 				continue;
 			};
 
-			let pdu: PduEvent = match serde_json::from_str(
-				&serde_json::to_string(&pdu).expect("CanonicalJsonObj can be serialized to JSON"),
-			) {
-				Ok(pdu) => pdu,
-				Err(_) => continue,
-			};
-
 			match pdu.kind {
 				TimelineEventType::RoomMember => {
-					let Ok(membership_event) = serde_json::from_str::<RoomMemberEventContent>(pdu.content.get()) else {
+					let Some(user_id) = pdu.state_key.as_ref().map(UserId::parse).flat_ok() else {
 						continue;
 					};
 
-					let Some(state_key) = pdu.state_key else {
+					let Ok(membership_event) = pdu.get_content::<RoomMemberEventContent>() else {
-						continue;
-					};
-
-					let Ok(user_id) = UserId::parse(state_key) else {
 						continue;
 					};
 

src/service/rooms/state_accessor/mod.rs

@@ -325,11 +325,9 @@ impl Service {
 	where
 		T: for<'de> Deserialize<'de> + Send,
 	{
-		use serde_json::from_str;
-
 		self.room_state_get(room_id, event_type, state_key)
 			.await
-			.and_then(|event| from_str::<T>(event.content.get()).map_err(Into::into))
+			.and_then(|event| event.get_content())
 	}
 
 	pub async fn get_name(&self, room_id: &RoomId) -> Result<String> {

src/service/rooms/timeline/mod.rs

@@ -471,12 +471,7 @@ impl Service {
 						}
 					},
 					_ => {
-						let content =
+						let content: RoomRedactionEventContent = pdu.get_content()?;
-							serde_json::from_str::<RoomRedactionEventContent>(pdu.content.get()).map_err(|e| {
-								warn!("Invalid content in redaction pdu: {e}");
-								Error::bad_database("Invalid content in redaction pdu")
-							})?;
-
 						if let Some(redact_id) = &content.redacts {
 							if self
 								.services
@@ -506,11 +501,7 @@ impl Service {
 					let target_user_id =
 						UserId::parse(state_key.clone()).expect("This state_key was previously validated");
 
-					let content = serde_json::from_str::<RoomMemberEventContent>(pdu.content.get()).map_err(|e| {
+					let content: RoomMemberEventContent = pdu.get_content()?;
-						error!("Invalid room member event content in pdu: {e}");
-						Error::bad_database("Invalid room member event content in pdu.")
-					})?;
-
 					let invite_state = match content.membership {
 						MembershipState::Invite => self.services.state.summary_stripped(pdu).await.into(),
 						_ => None,
@@ -533,9 +524,7 @@ impl Service {
 				}
 			},
 			TimelineEventType::RoomMessage => {
-				let content = serde_json::from_str::<ExtractBody>(pdu.content.get())
+				let content: ExtractBody = pdu.get_content()?;
-					.map_err(|_| Error::bad_database("Invalid content in pdu."))?;
-
 				if let Some(body) = content.body {
 					self.services.search.index_pdu(shortroomid, &pdu_id, &body);
 
@@ -549,7 +538,7 @@ impl Service {
 			_ => {},
 		}
 
-		if let Ok(content) = serde_json::from_str::<ExtractRelatesToEventId>(pdu.content.get()) {
+		if let Ok(content) = pdu.get_content::<ExtractRelatesToEventId>() {
 			if let Ok(related_pducount) = self.get_pdu_count(&content.relates_to.event_id).await {
 				self.services
 					.pdu_metadata
@@ -557,7 +546,7 @@ impl Service {
 			}
 		}
 
-		if let Ok(content) = serde_json::from_str::<ExtractRelatesTo>(pdu.content.get()) {
+		if let Ok(content) = pdu.get_content::<ExtractRelatesTo>() {
 			match content.relates_to {
 				Relation::Reply {
 					in_reply_to,
@@ -712,10 +701,7 @@ impl Service {
 				.room_state_get(room_id, &event_type.to_string().into(), state_key)
 				.await
 			{
-				unsigned.insert(
+				unsigned.insert("prev_content".to_owned(), prev_pdu.get_content_as_value());
-					"prev_content".to_owned(),
-					serde_json::from_str(prev_pdu.content.get()).expect("string is valid json"),
-				);
 				unsigned.insert(
 					"prev_sender".to_owned(),
 					serde_json::to_value(&prev_pdu.sender).expect("UserId::to_value always works"),
@@ -874,9 +860,7 @@ impl Service {
 					};
 				},
 				_ => {
-					let content = serde_json::from_str::<RoomRedactionEventContent>(pdu.content.get())
+					let content: RoomRedactionEventContent = pdu.get_content()?;
-						.map_err(|e| err!(Database("Invalid content in redaction pdu: {e:?}")))?;
-
 					if let Some(redact_id) = &content.redacts {
 						if !self
 							.services
@@ -1026,7 +1010,7 @@ impl Service {
 			.await
 			.map_err(|e| err!(Database(error!(?pdu_id, ?event_id, ?e, "PDU ID points to invalid PDU."))))?;
 
-		if let Ok(content) = serde_json::from_str::<ExtractBody>(pdu.content.get()) {
+		if let Ok(content) = pdu.get_content::<ExtractBody>() {
 			if let Some(body) = content.body {
 				self.services
 					.search
@@ -1200,9 +1184,7 @@ impl Service {
 		drop(insert_lock);
 
 		if pdu.kind == TimelineEventType::RoomMessage {
-			let content = serde_json::from_str::<ExtractBody>(pdu.content.get())
+			let content: ExtractBody = pdu.get_content()?;
-				.map_err(|e| err!(Database("Invalid content in pdu: {e:?}")))?;
-
 			if let Some(body) = content.body {
 				self.services.search.index_pdu(shortroomid, &pdu_id, &body);
 			}