arminius_lib_roomservice_client/
roomserver_command_subject.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Type for creating subjects for roomserver commands
6
7mod tests;
8
9use arminius_lib_message_bus::{EscapingSubject, TokenEscapeError};
10use async_nats::{Subject, subject::ToSubject};
11use ruma::{OwnedRoomId, RoomId};
12use serde::{Deserialize, Serialize};
13
14/// Errors that can occur when parsing a roomserver subject.
15#[expect(
16	missing_docs,
17	reason = "TODO(#2003): This exception predates us enabling `clippy::allow_attributes_without_reason`, so we don't know why it was enabled"
18)]
19#[derive(Debug, thiserror::Error, Serialize, Deserialize, PartialEq)]
20pub enum NatsSubjectParseError {
21	#[error("First element of subject was {found}, should be {expected}")]
22	InvalidSubjectError { expected: String, found: String },
23	#[error("Subject is missing {0} token")]
24	MissingTokenError(String),
25	#[error("Expected partition number to be an integer, got {0}")]
26	PartitionNumberNotInt(String),
27	#[error("Could not parse room id {0}")]
28	InvalidRoomId(String),
29	#[error("Trailing tokens in subject")]
30	TrailingTokens(String),
31	#[error("Token escape error: {0}")]
32	TokenEscapeError(#[from] TokenEscapeError),
33}
34
35/// A command sent to the roomserver
36#[derive(Clone, Debug, PartialEq)]
37pub struct RoomserverCommand {
38	/// The ID of the room the command applies to. Some commands, like the room
39	/// creation command, do not contain a room ID
40	pub room_id: Option<OwnedRoomId>,
41	/// The command that should be sent
42	pub command: String,
43}
44
45impl ToSubject for RoomserverCommand {
46	fn to_subject(&self) -> Subject {
47		EscapingSubject::new(vec![
48			"rooms".to_owned(),
49			match &self.room_id {
50				None => "new".to_owned(),
51				Some(room_id) => room_id.to_string(),
52			},
53			self.command.clone(),
54		])
55		.to_subject_string()
56		.into()
57	}
58}
59
60impl TryFrom<&Subject> for RoomserverCommand {
61	type Error = NatsSubjectParseError;
62
63	/// Parses a RoomserverCommand. Guaranteed to work if the Subject has been
64	/// generated with [RoomserverCommand::to_subject]
65	fn try_from(value: &Subject) -> Result<Self, Self::Error> {
66		let EscapingSubject(tokens) = value.parse()?;
67		let mut tokens = tokens.into_iter();
68
69		let destination = tokens
70			.next()
71			.ok_or_else(|| NatsSubjectParseError::MissingTokenError("destination".to_owned()))?;
72		if destination != "rooms" {
73			tracing::error!(subject = %value, "Invalid subject in message to roomserver");
74			return Err(NatsSubjectParseError::InvalidSubjectError {
75				expected: "rooms".to_owned(),
76				found: destination,
77			});
78		}
79
80		let partition_id = tokens
81			.next()
82			.ok_or_else(|| NatsSubjectParseError::MissingTokenError("partition_id".to_owned()))?;
83		if let Err(err) = partition_id.parse::<i32>() {
84			tracing::error!(
85				subject = %value,
86				error = %err,
87				"Error parsing partition id in subject of message to roomserver",
88			);
89			return Err(NatsSubjectParseError::PartitionNumberNotInt(partition_id));
90		}
91		let room_id = tokens
92			.next()
93			.ok_or_else(|| NatsSubjectParseError::MissingTokenError("room_id".to_owned()))?;
94
95		let room_id: Option<OwnedRoomId> = if room_id == "new" {
96			None
97		} else {
98			Some(
99				RoomId::parse(room_id.clone())
100					.map_err(|_| NatsSubjectParseError::InvalidRoomId(room_id))?,
101			)
102		};
103
104		let command = tokens
105			.next()
106			.ok_or_else(|| NatsSubjectParseError::MissingTokenError("command".to_owned()))?;
107
108		let remainder =
109			itertools::Itertools::intersperse(tokens, ".".to_owned()).collect::<String>();
110		if !remainder.is_empty() {
111			return Err(NatsSubjectParseError::TrailingTokens(remainder));
112		}
113
114		Ok(RoomserverCommand { room_id, command })
115	}
116}
arminius_lib_roomservice_client/roomserver_command_subject.rs

arminius_lib_roomservice_client/
roomserver_command_subject.rs
