Initial commit

src/bot/implementation.rs

@@ -0,0 +1,419 @@
+use std::sync::Arc;
+use std::{future::Future, pin::Pin};
+
+use mxlink::matrix_sdk::media::{MediaFormat, MediaRequest};
+use mxlink::matrix_sdk::ruma::{
+    events::room::MediaSource, MilliSecondsSinceUnixEpoch, OwnedUserId,
+};
+use mxlink::matrix_sdk::Room;
+
+use mxlink::{
+    InitConfig, LoginConfig, LoginCredentials, LoginEncryption, MatrixLink, PersistenceConfig,
+};
+
+use mxlink::helpers::account_data_config::{
+    ConfigError, GlobalConfigManager as AccountDataGlobalConfigManager,
+    RoomConfigManager as AccountDataRoomConfigManager,
+};
+use mxlink::helpers::encryption::Manager as EncryptionManager;
+
+use crate::agent::Manager as AgentManager;
+use crate::entity::catch_up_marker::{
+    CatchUpMarker, CatchUpMarkerManager, DelayedCatchUpMarkerManager,
+};
+use crate::entity::cfg::Config;
+use crate::entity::globalconfig::{GlobalConfig, GlobalConfigurationManager};
+use crate::entity::roomconfig::{RoomConfig, RoomConfigurationManager};
+
+use crate::agent::Manager;
+
+use crate::conversation::matrix::{RoomDisplayNameFetcher, RoomEventFetcher};
+
+const ROOM_EVENT_FETCHER_LRU_CACHE_SIZE: usize = 1000;
+const ROOM_DISPLAY_NAME_FETCHER_LRU_CACHE_SIZE: usize = 1000;
+const ROOM_CONFIG_MANAGER_LRU_CACHE_SIZE: usize = 1000;
+
+const LOGO_BYTES: &[u8] = include_bytes!("../../etc/assets/baibot-torso-768.png");
+const LOGO_MIME_TYPE: &str = "image/png";
+
+/// Controls how often we persist the catch-up marker to Account Data.
+/// Consult the `DelayedCatchUpMarkerManager` documentation for more information.
+const DELAYED_CATCH_UP_MARKER_MANAGER_PERSIST_INTERVAL_DURATION: std::time::Duration =
+    std::time::Duration::from_secs(10);
+
+/// Controls what federation delay we will tolerate. The timestamp that gets persisted
+/// will be based on the last seen event's `origin_server_ts` minus this duration.
+/// Consult the `DelayedCatchUpMarkerManager` documentation for more information.
+const DELAYED_CATCH_UP_MARKER_MANAGER_FEDERATION_DELAY_TOLERANCE_DURATION: std::time::Duration =
+    std::time::Duration::from_secs(90);
+
+struct BotInner {
+    config: Config,
+
+    matrix_link: MatrixLink,
+    delayed_catch_up_marker_manager: DelayedCatchUpMarkerManager,
+    global_config_manager: tokio::sync::Mutex<GlobalConfigurationManager>,
+    room_config_manager: tokio::sync::Mutex<RoomConfigurationManager>,
+    room_event_fetcher: Arc<RoomEventFetcher>,
+    room_display_name_fetcher: Arc<RoomDisplayNameFetcher>,
+    agent_manager: Manager,
+    admin_pattern_regexes: Vec<regex::Regex>,
+}
+
+/// Bot represents a bot instance.
+///
+/// All of the state is held in an `Arc` so the `Bot` can be cloned freely.
+#[derive(Clone)]
+pub struct Bot {
+    inner: Arc<BotInner>,
+}
+
+impl Bot {
+    pub async fn new(config: Config) -> anyhow::Result<Self> {
+        // Take some potentially problematic configuration values out of the config early on.
+        // If we'd be failing, we'd like it to happen early, before we log in, etc.
+
+        let initial_global_config: GlobalConfig =
+            config.initial_global_config.clone().try_into()?;
+
+        let admin_pattern_regexes = config.access.admin_pattern_regexes()?;
+        let persistence_config_encryption_key = config.persistence.config_encryption_key()?;
+
+        let agent_manager = AgentManager::new(config.agents.static_definitions.clone())?;
+
+        let encryption_manager = EncryptionManager::new(persistence_config_encryption_key);
+
+        let matrix_link = create_matrix_link(&config).await?;
+
+        let catch_up_marker_manager = create_catch_up_marker_manager(matrix_link.clone());
+
+        let delayed_catch_up_marker_manager = DelayedCatchUpMarkerManager::new(
+            catch_up_marker_manager,
+            DELAYED_CATCH_UP_MARKER_MANAGER_PERSIST_INTERVAL_DURATION,
+            DELAYED_CATCH_UP_MARKER_MANAGER_FEDERATION_DELAY_TOLERANCE_DURATION,
+        );
+
+        let global_config_manager = tokio::sync::Mutex::new(create_global_configuration_manager(
+            matrix_link.clone(),
+            encryption_manager.clone(),
+            initial_global_config,
+        ));
+
+        let room_config_manager = tokio::sync::Mutex::new(create_room_configuration_manager(
+            matrix_link.clone(),
+            encryption_manager.clone(),
+        ));
+
+        let room_event_fetcher = RoomEventFetcher::new(Some(ROOM_EVENT_FETCHER_LRU_CACHE_SIZE));
+
+        let room_display_name_fetcher = RoomDisplayNameFetcher::new(
+            matrix_link.clone(),
+            Some(ROOM_DISPLAY_NAME_FETCHER_LRU_CACHE_SIZE),
+        );
+
+        Ok(Self {
+            inner: Arc::new(BotInner {
+                config,
+
+                matrix_link,
+                delayed_catch_up_marker_manager,
+                global_config_manager,
+                room_config_manager,
+                room_event_fetcher: Arc::new(room_event_fetcher),
+                room_display_name_fetcher: Arc::new(room_display_name_fetcher),
+                agent_manager,
+                admin_pattern_regexes,
+            }),
+        })
+    }
+
+    pub(crate) fn admin_patterns(&self) -> &Vec<String> {
+        &self.inner.config.access.admin_patterns
+    }
+
+    pub(crate) fn name(&self) -> &str {
+        &self.inner.config.user.name
+    }
+
+    pub(crate) fn command_prefix(&self) -> &str {
+        &self.inner.config.command_prefix
+    }
+
+    pub(crate) fn homeserver_name(&self) -> &str {
+        &self.inner.config.homeserver.server_name
+    }
+
+    pub(crate) fn global_config_manager(&self) -> &tokio::sync::Mutex<GlobalConfigurationManager> {
+        &self.inner.global_config_manager
+    }
+
+    pub(crate) fn room_config_manager(&self) -> &tokio::sync::Mutex<RoomConfigurationManager> {
+        &self.inner.room_config_manager
+    }
+
+    pub(crate) fn room_event_fetcher(&self) -> Arc<RoomEventFetcher> {
+        self.inner.room_event_fetcher.clone()
+    }
+
+    pub(crate) fn room_display_name_fetcher(&self) -> Arc<RoomDisplayNameFetcher> {
+        self.inner.room_display_name_fetcher.clone()
+    }
+
+    pub(crate) fn agent_manager(&self) -> &Manager {
+        &self.inner.agent_manager
+    }
+
+    pub(crate) fn matrix_link(&self) -> &MatrixLink {
+        &self.inner.matrix_link
+    }
+
+    pub(crate) fn user_id(&self) -> &OwnedUserId {
+        self.matrix_link().user_id()
+    }
+
+    pub(crate) fn reacting(&self) -> super::reacting::Reacting {
+        super::reacting::Reacting::new(self.clone())
+    }
+
+    pub(crate) fn rooms(&self) -> super::rooms::Rooms {
+        super::rooms::Rooms::new(self.clone())
+    }
+
+    pub(crate) fn messaging(&self) -> super::messaging::Messaging {
+        super::messaging::Messaging::new(self.clone())
+    }
+
+    pub(crate) fn admin_pattern_regexes(&self) -> &Vec<regex::Regex> {
+        &self.inner.admin_pattern_regexes
+    }
+
+    pub(crate) async fn global_config(&self) -> Result<GlobalConfig, ConfigError> {
+        let mut global_config_manager_guard = self.inner.global_config_manager.lock().await;
+
+        global_config_manager_guard.get_or_create().await
+    }
+
+    pub(crate) async fn is_caught_up(
+        &self,
+        event_origin_server_ts: MilliSecondsSinceUnixEpoch,
+    ) -> Result<bool, ConfigError> {
+        self.inner
+            .delayed_catch_up_marker_manager
+            .is_caught_up(event_origin_server_ts.0.into())
+            .await
+    }
+
+    pub(crate) async fn catch_up(&self, event_origin_server_ts: MilliSecondsSinceUnixEpoch) {
+        self.inner
+            .delayed_catch_up_marker_manager
+            .catch_up(event_origin_server_ts.0.into())
+            .await
+    }
+
+    pub async fn start(&self) -> anyhow::Result<()> {
+        self.rooms().attach_event_handlers().await;
+        self.messaging().attach_event_handlers().await;
+        self.reacting().attach_event_handlers().await;
+
+        self.inner.delayed_catch_up_marker_manager.start().await;
+
+        self.prepare_profile().await?;
+
+        self.inner
+            .matrix_link
+            .start()
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to sync: {:?}", e))
+    }
+
+    async fn prepare_profile(&self) -> anyhow::Result<()> {
+        use std::time::Duration;
+        use tokio::time::sleep;
+
+        let mut delay = Duration::from_secs(3);
+        let max_delay = Duration::from_secs(30);
+
+        loop {
+            match self.do_prepare_profile().await {
+                Ok(_) => return Ok(()),
+                Err(err) => {
+                    tracing::warn!(
+                        ?err,
+                        ?delay,
+                        "Failed to prepare profile.. Will retry after delay..."
+                    );
+
+                    sleep(delay).await;
+
+                    delay = std::cmp::min(delay * 2, max_delay);
+                }
+            }
+        }
+    }
+
+    async fn do_prepare_profile(&self) -> anyhow::Result<()> {
+        tracing::debug!("Preparing profile..");
+
+        let account = self.inner.matrix_link.client().account();
+        let media = self.inner.matrix_link.client().media();
+
+        let desired_display_name = self.inner.config.user.name.clone();
+
+        let profile = account
+            .get_profile()
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed fetching profile: {:?}", e))?;
+
+        let should_update_display_name = match &profile.displayname {
+            Some(displayname) => displayname != &desired_display_name,
+            None => true,
+        };
+
+        if should_update_display_name {
+            tracing::info!(
+                ?profile.displayname,
+                ?desired_display_name,
+                "Updating display name.."
+            );
+
+            if let Err(err) = account.set_display_name(Some(&desired_display_name)).await {
+                return Err(anyhow::anyhow!("Failed setting display name: {:?}", err));
+            }
+        }
+
+        let should_update_avatar = match &profile.avatar_url {
+            Some(avatar_url) => {
+                let request = MediaRequest {
+                    source: MediaSource::Plain(avatar_url.to_owned()),
+                    format: MediaFormat::File,
+                };
+
+                let content = media
+                    .get_media_content(&request, true)
+                    .await
+                    .map_err(|e| anyhow::anyhow!("Failed fetching existing avatar: {:?}", e))?;
+
+                content.as_slice() != LOGO_BYTES
+            }
+            None => true,
+        };
+
+        if should_update_avatar {
+            tracing::info!("Updating avatar..");
+
+            let mime_type = LOGO_MIME_TYPE
+                .parse()
+                .expect("Failed parsing mime type for logo");
+
+            account
+                .upload_avatar(&mime_type, LOGO_BYTES.to_vec())
+                .await
+                .map_err(|e| anyhow::anyhow!("Failed uploading avatar: {:?}", e))?;
+        }
+
+        Ok(())
+    }
+}
+
+async fn create_matrix_link(config: &Config) -> anyhow::Result<MatrixLink> {
+    let session_file_path = config.persistence.session_file_path()?;
+    let session_encryption_key = config.persistence.session_encryption_key()?;
+    let db_dir_path: std::path::PathBuf = config.persistence.db_dir_path()?;
+
+    let login_creds = LoginCredentials::UserPassword(
+        config.user.mxid_localpart.to_owned(),
+        config.user.password.to_owned(),
+    );
+
+    let login_encryption = LoginEncryption::new(
+        config.user.encryption.recovery_passphrase.clone(),
+        config.user.encryption.recovery_reset_allowed,
+    );
+
+    let login_config = LoginConfig::new(
+        config.homeserver.url.to_owned(),
+        login_creds,
+        Some(login_encryption),
+        config.user.name.to_owned(),
+    );
+
+    let persistence_config =
+        PersistenceConfig::new(session_file_path, session_encryption_key, db_dir_path);
+
+    let init_config = InitConfig::new(login_config, persistence_config);
+
+    mxlink::init(&init_config).await.map_err(|e| e.into())
+}
+
+pub fn create_global_configuration_manager(
+    matrix_link: MatrixLink,
+    encryption_manager: EncryptionManager,
+    initial_global_config: GlobalConfig,
+) -> GlobalConfigurationManager {
+    let initial_global_config_callback = move || {
+        let initial_global_config = initial_global_config.clone();
+
+        let future = create_initial_global_config(initial_global_config);
+
+        // Explicitly box the future to match the expected type
+        Box::pin(future) as Pin<Box<dyn Future<Output = GlobalConfig> + Send>>
+    };
+
+    AccountDataGlobalConfigManager::new(
+        matrix_link,
+        encryption_manager,
+        initial_global_config_callback,
+    )
+}
+
+async fn create_initial_global_config(initial_global_config: GlobalConfig) -> GlobalConfig {
+    initial_global_config
+}
+
+pub fn create_room_configuration_manager(
+    matrix_link: MatrixLink,
+    encryption_manager: EncryptionManager,
+) -> RoomConfigurationManager {
+    let initial_room_config_callback = |room: Room| {
+        let future = create_initial_room_config(room);
+
+        // Explicitly box the future to match the expected type
+        Box::pin(future) as Pin<Box<dyn Future<Output = RoomConfig> + Send>>
+    };
+
+    AccountDataRoomConfigManager::new(
+        matrix_link.user_id().clone(),
+        encryption_manager,
+        initial_room_config_callback,
+        Some(ROOM_CONFIG_MANAGER_LRU_CACHE_SIZE),
+    )
+}
+
+async fn create_initial_room_config(room: Room) -> RoomConfig {
+    RoomConfig::default().with_room(room).await
+}
+
+pub fn create_catch_up_marker_manager(matrix_link: MatrixLink) -> CatchUpMarkerManager {
+    let initial_global_config_callback = || {
+        let future = create_initial_catch_up_marker();
+
+        // Explicitly box the future to match the expected type
+        Box::pin(future) as Pin<Box<dyn Future<Output = CatchUpMarker> + Send>>
+    };
+
+    // Intentionally not using encryption, to make this resilient even if we lose our encryption key.
+    // We're not worried about the catch-up marker being read or tampered with, as it's not sensitive data.
+    let encryption_manager = EncryptionManager::new(None);
+
+    let catch_up_marker_manager: CatchUpMarkerManager = AccountDataGlobalConfigManager::new(
+        matrix_link.clone(),
+        encryption_manager,
+        initial_global_config_callback,
+    );
+
+    catch_up_marker_manager
+}
+
+async fn create_initial_catch_up_marker() -> CatchUpMarker {
+    CatchUpMarker::new(0)
+}

src/bot/load_config.rs

@@ -0,0 +1,110 @@
+use std::env;
+use std::path::PathBuf;
+
+use anyhow::anyhow;
+
+use crate::agent::AgentPurpose;
+
+pub use crate::entity::cfg::{defaults as cfg_defaults, env as cfg_env, Config};
+
+pub fn load() -> anyhow::Result<Config> {
+    let config_file_path = env::var(cfg_env::BAIBOT_CONFIG_FILE_PATH)
+        .unwrap_or_else(|_| cfg_defaults::config_file_path().to_owned());
+    let config_file_path = PathBuf::from(config_file_path);
+
+    if !config_file_path.exists() {
+        return Err(anyhow!(
+            "Config file ({}) not found. Adjust the {} environment variable to use another config file.",
+            config_file_path.display(),
+            cfg_env::BAIBOT_CONFIG_FILE_PATH,
+        ));
+    }
+
+    let config_str = std::fs::read_to_string(config_file_path)?;
+    let mut config: Config = serde_yaml::from_str(&config_str)?;
+
+    // Allow environment variables to override some configuration keys
+    for (key, value) in env::vars() {
+        match key.as_str() {
+            cfg_env::BAIBOT_HOMESERVER_SERVER_NAME => config.homeserver.server_name = value,
+            cfg_env::BAIBOT_HOMESERVER_URL => config.homeserver.url = value,
+            cfg_env::BAIBOT_USER_MXID_LOCALPART => config.user.mxid_localpart = value,
+            cfg_env::BAIBOT_USER_PASSWORD => config.user.password = value,
+            cfg_env::BAIBOT_USER_ENCRYPTION_RECOVERY_PASSPHRASE => {
+                config.user.encryption.recovery_passphrase = Some(value);
+            }
+            cfg_env::BAIBOT_USER_NAME => config.user.name = value,
+            cfg_env::BAIBOT_COMMAND_PREFIX => config.command_prefix = value,
+            cfg_env::BAIBOT_LOGGING => {
+                config.logging = value;
+            }
+            cfg_env::BAIBOT_ACCESS_ADMIN_PATTERNS => {
+                config.access.admin_patterns = value
+                    .split(' ')
+                    .map(|s| s.trim().to_string())
+                    .filter(|s| !s.is_empty())
+                    .collect();
+            }
+            cfg_env::BAIBOT_PERSISTENCE_DATA_DIR_PATH => {
+                config.persistence.data_dir_path = Some(value);
+            }
+            cfg_env::BAIBOT_PERSISTENCE_CONFIG_ENCRYPTION_KEY => {
+                config.persistence.config_encryption_key = Some(value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_HANDLER_CATCH_ALL => {
+                let value = if value.is_empty() { None } else { Some(value) };
+
+                config
+                    .initial_global_config
+                    .handler
+                    .set_by_purpose(AgentPurpose::CatchAll, value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_HANDLER_TEXT_GENERATION => {
+                let value = if value.is_empty() { None } else { Some(value) };
+
+                config
+                    .initial_global_config
+                    .handler
+                    .set_by_purpose(AgentPurpose::TextGeneration, value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_HANDLER_TEXT_TO_SPEECH => {
+                let value = if value.is_empty() { None } else { Some(value) };
+
+                config
+                    .initial_global_config
+                    .handler
+                    .set_by_purpose(AgentPurpose::TextToSpeech, value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_HANDLER_SPEECH_TO_TEXT => {
+                let value = if value.is_empty() { None } else { Some(value) };
+
+                config
+                    .initial_global_config
+                    .handler
+                    .set_by_purpose(AgentPurpose::SpeechToText, value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_HANDLER_IMAGE_GENERATION => {
+                let value = if value.is_empty() { None } else { Some(value) };
+
+                config
+                    .initial_global_config
+                    .handler
+                    .set_by_purpose(AgentPurpose::ImageGeneration, value);
+            }
+            cfg_env::BAIBOT_INITIAL_GLOBAL_CONFIG_USER_PATTERNS => {
+                config.initial_global_config.user_patterns = Some(
+                    value
+                        .split(' ')
+                        .map(|s| s.trim().to_string())
+                        .filter(|s| !s.is_empty())
+                        .collect(),
+                );
+            }
+            _ => {}
+        }
+    }
+
+    config.validate().map_err(|s| anyhow!(s))?;
+
+    Ok(config)
+}

src/bot/messaging.rs

@@ -0,0 +1,334 @@
+use mxlink::matrix_sdk::{
+    ruma::{
+        api::client::receipt::create_receipt::v3::ReceiptType,
+        events::room::message::OriginalSyncRoomMessageEvent, OwnedEventId,
+    },
+    Room,
+};
+
+use mxlink::{CallbackError, MessageResponseType};
+
+use tracing::Instrument;
+
+use crate::{
+    conversation::matrix::determine_thread_context_for_room_event,
+    entity::{MessageContext, MessagePayload, RoomConfigContext, TriggerEventInfo},
+};
+
+#[derive(Clone)]
+pub struct Messaging {
+    bot: super::Bot,
+}
+
+impl Messaging {
+    pub fn new(bot: super::Bot) -> Self {
+        Self { bot }
+    }
+
+    pub async fn send_text_markdown_no_fail(
+        &self,
+        room: &Room,
+        message: String,
+        response_type: MessageResponseType,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        let result = self
+            .bot
+            .matrix_link()
+            .messaging()
+            .send_text_markdown(room, message, response_type)
+            .await;
+
+        match result {
+            Ok(result) => Some(result),
+            Err(err) => {
+                tracing::error!(
+                    room_id = format!("{:?}", room.room_id()),
+                    ?err,
+                    "Failed to send text message to room",
+                );
+                None
+            }
+        }
+    }
+
+    pub async fn send_notice_markdown_no_fail(
+        &self,
+        room: &Room,
+        message: String,
+        response_type: MessageResponseType,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        let result = self
+            .bot
+            .matrix_link()
+            .messaging()
+            .send_notice_markdown(room, message, response_type)
+            .await;
+
+        match result {
+            Ok(result) => Some(result),
+            Err(err) => {
+                tracing::error!(
+                    room_id = format!("{:?}", room.room_id()),
+                    ?err,
+                    "Failed to send notice message to room",
+                );
+                None
+            }
+        }
+    }
+
+    pub async fn send_tooltip_markdown_no_fail(
+        &self,
+        room: &Room,
+        message: &str,
+        response_type: MessageResponseType,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        self.send_notice_markdown_no_fail(
+            room,
+            crate::utils::status::create_tooltip_message_text(message),
+            response_type,
+        )
+        .await
+    }
+
+    pub async fn send_success_markdown_no_fail(
+        &self,
+        room: &Room,
+        message: &str,
+        response_type: MessageResponseType,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        self.send_notice_markdown_no_fail(
+            room,
+            crate::utils::status::create_success_message_text(message),
+            response_type,
+        )
+        .await
+    }
+
+    pub async fn send_error_markdown_no_fail(
+        &self,
+        room: &Room,
+        err: &str,
+        response_type: MessageResponseType,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        self.send_notice_markdown_no_fail(
+            room,
+            crate::utils::status::create_error_message_text(err),
+            response_type,
+        )
+        .await
+    }
+
+    pub async fn redact_event_no_fail(
+        &self,
+        room: &Room,
+        target_event_id: OwnedEventId,
+        reason: Option<String>,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::redact::redact_event::v3::Response> {
+        let result = self
+            .bot
+            .matrix_link()
+            .messaging()
+            .redact_event(room, target_event_id.clone(), reason)
+            .await;
+
+        match result {
+            Ok(result) => Some(result),
+            Err(err) => {
+                tracing::error!(
+                    room_id = format!("{:?}", room.room_id()),
+                    ?target_event_id,
+                    ?err,
+                    "Failed to send redaction to room",
+                );
+
+                None
+            }
+        }
+    }
+
+    pub(super) async fn attach_event_handlers(&self) {
+        let matrix_link_messaging = self.bot.matrix_link().messaging();
+
+        let this = self.clone();
+        matrix_link_messaging.on_actionable_room_message(|event, room| async move {
+            this.on_actionable_message(event, room).await
+        });
+    }
+
+    #[tracing::instrument(name = "bot_on_actionable_message", skip_all, fields(room_id = room.room_id().as_str(), event_id = event.event_id.as_str()))]
+    async fn on_actionable_message(
+        &self,
+        event: OriginalSyncRoomMessageEvent,
+        room: Room,
+    ) -> Result<(), CallbackError> {
+        if self
+            .bot
+            .is_caught_up(event.origin_server_ts)
+            .await
+            .map_err(|e| {
+                CallbackError::Unknown(
+                    format!("Failed to determine catch-up state: {:?}", e).into(),
+                )
+            })?
+        {
+            tracing::debug!(
+                event_origin_server_ts = format!("{:?}", event.origin_server_ts),
+                "Ignoring old message event",
+            );
+
+            return Ok(());
+        }
+
+        tracing::info!("Processing message");
+
+        let global_config = self
+            .bot
+            .global_config()
+            .await
+            .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+        tracing::trace!(?global_config, "Global config");
+
+        let room_config = self
+            .bot
+            .room_config_manager()
+            .lock()
+            .await
+            .get_or_create_for_room(&room)
+            .await
+            .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+        tracing::trace!(?room_config, "Room config");
+
+        let trigger_event_sender_is_admin = mxidwc::match_user_id(
+            event.sender.clone().as_str(),
+            self.bot.admin_pattern_regexes(),
+        );
+
+        let trigger_event_sender_is_allowed_user = match &global_config.access.user_patterns {
+            Some(user_patterns) => {
+                let allowed_user_regexes = mxidwc::parse_patterns_vector(user_patterns)
+                    .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+                mxidwc::match_user_id(event.sender.clone().as_str(), &allowed_user_regexes)
+            }
+            None => false,
+        };
+
+        if !trigger_event_sender_is_admin && !trigger_event_sender_is_allowed_user {
+            tracing::debug!("Ignoring message from non-admin/non-allowed user");
+            return Ok(());
+        }
+
+        let payload: Result<MessagePayload, String> = event.content.msgtype.clone().try_into();
+        let payload = match payload {
+            Ok(payload) => payload,
+            Err(err) => {
+                tracing::debug!(
+                    msg_type = event.content.msgtype(),
+                    ?err,
+                    "Ignoring message not supported by us",
+                );
+                return Ok(());
+            }
+        };
+
+        let thread_context = determine_thread_context_for_room_event(
+            self.bot.user_id(),
+            &room,
+            &event,
+            &payload,
+            &self.bot.room_event_fetcher(),
+        )
+        .await;
+
+        let thread_context = match thread_context {
+            Ok(value) => value,
+            Err(err) => {
+                tracing::error!(?err, "Failed to determine thread context for event");
+                return Ok(());
+            }
+        };
+
+        let Some(thread_context) = thread_context else {
+            tracing::debug!("Ignoring message with unknown thread context (likely not a threaded message or a top-level message)");
+            return Ok(());
+        };
+
+        let room_config_context =
+            RoomConfigContext::new(global_config.clone(), room_config.clone());
+
+        let trigger_event_info = TriggerEventInfo::new(
+            event.event_id.clone(),
+            event.sender.clone(),
+            payload,
+            trigger_event_sender_is_admin,
+        );
+
+        let message_context = MessageContext::new(
+            room.clone(),
+            room_config_context,
+            self.bot.admin_pattern_regexes().clone(),
+            trigger_event_info,
+            thread_context.info.clone(),
+        );
+
+        let bot_display_name = self
+            .bot
+            .room_display_name_fetcher()
+            .own_display_name_in_room(message_context.room())
+            .await;
+
+        let bot_display_name = match bot_display_name {
+            Ok(value) => value,
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    "Failed to fetch bot display name. Proceeding without it"
+                );
+                None
+            }
+        };
+
+        // The first event in the thread determines which handler processes the current event.
+        let controller_type = crate::controller::determine_controller(
+            self.bot.command_prefix(),
+            &thread_context.first_message,
+            &message_context,
+            self.bot.user_id(),
+            &bot_display_name,
+        );
+
+        tracing::info!(?controller_type, "Determined controller");
+
+        let _ = room
+            .send_single_receipt(
+                ReceiptType::Read,
+                thread_context.info.clone().into(),
+                event.event_id.clone(),
+            )
+            .await;
+
+        let start_time = std::time::Instant::now();
+
+        let event_span = tracing::error_span!("message_controller", ?controller_type);
+
+        crate::controller::dispatch_controller(&controller_type, &message_context, &self.bot)
+            .instrument(event_span)
+            .await;
+
+        let duration = std::time::Instant::now().duration_since(start_time);
+
+        tracing::debug!(?duration, "Controller finished");
+
+        self.bot.catch_up(event.origin_server_ts).await;
+
+        return Ok(());
+    }
+}

src/bot/mod.rs

@@ -0,0 +1,8 @@
+mod implementation;
+mod load_config;
+mod messaging;
+mod reacting;
+mod rooms;
+
+pub use implementation::Bot;
+pub use load_config::load as load_config;

src/bot/reacting.rs

@@ -0,0 +1,262 @@
+use mxlink::matrix_sdk::{
+    ruma::{
+        events::{
+            room::message::Relation, AnyMessageLikeEvent, AnySyncTimelineEvent, AnyTimelineEvent,
+            MessageLikeEvent,
+        },
+        OwnedEventId, OwnedUserId,
+    },
+    Room,
+};
+
+use mxlink::CallbackError;
+use mxlink::ThreadInfo;
+
+use tracing::Instrument;
+
+use crate::entity::{MessageContext, MessagePayload, RoomConfigContext, TriggerEventInfo};
+
+#[derive(Clone)]
+pub struct Reacting {
+    bot: super::Bot,
+}
+
+impl Reacting {
+    pub fn new(bot: super::Bot) -> Self {
+        Self { bot }
+    }
+
+    pub async fn react_no_fail(
+        &self,
+        room: &Room,
+        target_event_id: OwnedEventId,
+        reaction_key: String,
+    ) -> Option<mxlink::matrix_sdk::ruma::api::client::message::send_message_event::v3::Response>
+    {
+        let result = self
+            .bot
+            .matrix_link()
+            .reacting()
+            .react(room, target_event_id.clone(), reaction_key)
+            .await;
+
+        match result {
+            Ok(result) => Some(result),
+            Err(err) => {
+                tracing::error!(
+                    "Failed to send reaction to {} in room {:?}: {:?}",
+                    target_event_id,
+                    room.room_id(),
+                    err
+                );
+                None
+            }
+        }
+    }
+
+    pub(super) async fn attach_event_handlers(&self) {
+        let matrix_link_reacting = self.bot.matrix_link().reacting();
+
+        let this = self.clone();
+        matrix_link_reacting.on_actionable_reaction(
+            |event, room, reaction_event_content| async move {
+                this.on_actionable_reaction(event, room, reaction_event_content)
+                    .await
+            },
+        );
+    }
+
+    #[tracing::instrument(name = "bot_on_actionable_reaction", skip_all, fields(room_id = room.room_id().as_str(), event_id = event.event_id().as_str()))]
+    async fn on_actionable_reaction(
+        &self,
+        event: AnySyncTimelineEvent,
+        room: Room,
+        reaction_event_content: mxlink::matrix_sdk::ruma::events::reaction::ReactionEventContent,
+    ) -> Result<(), CallbackError> {
+        if self
+            .bot
+            .is_caught_up(event.origin_server_ts())
+            .await
+            .map_err(|e| {
+                CallbackError::Unknown(
+                    format!("Failed to determine catch-up state: {:?}", e).into(),
+                )
+            })?
+        {
+            tracing::debug!(
+                event_origin_server_ts = format!("{:?}", event.origin_server_ts()),
+                "Ignoring old reaction event",
+            );
+
+            return Ok(());
+        }
+
+        tracing::info!("Handling reaction");
+
+        let global_config = self
+            .bot
+            .global_config()
+            .await
+            .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+        tracing::trace!(?global_config, "Global config");
+
+        let trigger_event_sender_is_admin =
+            mxidwc::match_user_id(event.sender().as_str(), self.bot.admin_pattern_regexes());
+
+        let trigger_event_sender_is_allowed_user = match &global_config.access.user_patterns {
+            Some(user_patterns) => {
+                let allowed_user_regexes = mxidwc::parse_patterns_vector(user_patterns)
+                    .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+                mxidwc::match_user_id(event.sender().as_str(), &allowed_user_regexes)
+            }
+            None => false,
+        };
+
+        if !trigger_event_sender_is_admin && !trigger_event_sender_is_allowed_user {
+            tracing::debug!("Ignoring reaction from non-admin/non-allowed user");
+            return Ok(());
+        }
+
+        let reacted_to_event_id = &reaction_event_content.relates_to.event_id;
+
+        let reacted_to_event = self
+            .bot
+            .room_event_fetcher()
+            .fetch_event_in_room(reacted_to_event_id, &room)
+            .await;
+
+        let reacted_to_event = match reacted_to_event {
+            Ok(value) => value,
+            Err(err) => {
+                tracing::error!(
+                    ?reacted_to_event_id,
+                    ?err,
+                    "Failed to fetch reacted-to event",
+                );
+                return Ok(());
+            }
+        };
+
+        let reacted_to_event_any_timeline_event = match reacted_to_event.event.deserialize() {
+            Ok(value) => value,
+            Err(err) => {
+                tracing::error!(
+                    ?reacted_to_event_id,
+                    ?err,
+                    "Failed to deserialize reacted-to event event",
+                );
+                return Ok(());
+            }
+        };
+
+        let reacted_to_event_sender_id: OwnedUserId =
+            reacted_to_event_any_timeline_event.sender().to_owned();
+
+        let AnyTimelineEvent::MessageLike(reacted_to_event_message_like) =
+            reacted_to_event_any_timeline_event
+        else {
+            tracing::debug!(
+                ?reacted_to_event_id,
+                "Ignoring non-MessageLike reacted-to event",
+            );
+            return Ok(());
+        };
+
+        let AnyMessageLikeEvent::RoomMessage(reacted_to_event_room_message) =
+            reacted_to_event_message_like
+        else {
+            tracing::debug!(
+                ?reacted_to_event_id,
+                "Ignoring non-RoomMessage reacted-to event",
+            );
+            return Ok(());
+        };
+
+        let MessageLikeEvent::Original(reacted_to_event_room_message_original) =
+            reacted_to_event_room_message
+        else {
+            tracing::debug!(?reacted_to_event_id, "Ignoring redacted reacted-to event",);
+            return Ok(());
+        };
+
+        let reacted_to_event_payload: Result<MessagePayload, String> =
+            reacted_to_event_room_message_original
+                .content
+                .msgtype
+                .clone()
+                .try_into();
+        let Ok(reacted_to_event_payload) = reacted_to_event_payload else {
+            tracing::debug!(
+                msg_type = reacted_to_event_room_message_original.content.msgtype(),
+                "Ignoring reaction to message of unknown type",
+            );
+            return Ok(());
+        };
+
+        let thread_root_event_id = match reacted_to_event_room_message_original.content.relates_to {
+            Some(relation) => {
+                if let Relation::Thread(thread_id) = relation {
+                    thread_id.event_id.clone()
+                } else {
+                    reacted_to_event_id.clone()
+                }
+            }
+            None => reacted_to_event_id.clone(),
+        };
+
+        let thread_info = ThreadInfo::new(thread_root_event_id, reacted_to_event_id.clone());
+
+        let room_config = self
+            .bot
+            .room_config_manager()
+            .lock()
+            .await
+            .get_or_create_for_room(&room)
+            .await
+            .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+        tracing::trace!(?room_config, "Room config");
+
+        let room_config_context =
+            RoomConfigContext::new(global_config.clone(), room_config.clone());
+
+        let trigger_event_info = TriggerEventInfo::new(
+            event.event_id().to_owned(),
+            event.sender().to_owned(),
+            MessagePayload::Reaction {
+                key: reaction_event_content.relates_to.key,
+                reacted_to_event_payload: Box::new(reacted_to_event_payload),
+                reacted_to_event_id: reaction_event_content.relates_to.event_id.clone(),
+                reacted_to_event_sender_id,
+            },
+            trigger_event_sender_is_admin,
+        );
+
+        let message_context = MessageContext::new(
+            room,
+            room_config_context,
+            self.bot.admin_pattern_regexes().clone(),
+            trigger_event_info,
+            thread_info,
+        );
+
+        tracing::info!("Handling reaction via reaction controller");
+
+        let event_span = tracing::error_span!("reaction_controller");
+
+        crate::controller::reaction::handle(
+            &self.bot,
+            self.bot.matrix_link().clone(),
+            &message_context,
+        )
+        .instrument(event_span)
+        .await
+        .map_err(|err| CallbackError::Unknown(err.into()))?;
+
+        self.bot.catch_up(event.origin_server_ts()).await;
+
+        Ok(())
+    }
+}

src/bot/rooms.rs

@@ -0,0 +1,145 @@
+use mxlink::{
+    matrix_sdk::{
+        ruma::events::{room::member::StrippedRoomMemberEvent, AnySyncTimelineEvent},
+        Room,
+    },
+    InvitationDecision,
+};
+
+use mxlink::CallbackError;
+
+use tracing::Instrument;
+
+use crate::entity::RoomConfigContext;
+
+#[derive(Clone)]
+pub struct Rooms {
+    bot: super::Bot,
+}
+
+impl Rooms {
+    pub fn new(bot: super::Bot) -> Self {
+        Self { bot }
+    }
+
+    pub(super) async fn attach_event_handlers(&self) {
+        let matrix_link_rooms = self.bot.matrix_link().rooms();
+
+        let this = self.clone();
+        matrix_link_rooms.on_being_last_member(|event, room| async move {
+            this.on_being_last_member(event, room).await
+        });
+
+        let this = self.clone();
+        matrix_link_rooms
+            .on_invitation(|event, room| async move { this.on_invitation(event, room).await });
+
+        let this = self.clone();
+        matrix_link_rooms.on_joined(|event, room| async move { this.on_joined(event, room).await });
+    }
+
+    async fn on_invitation(
+        &self,
+        room_member: StrippedRoomMemberEvent,
+        _room: Room,
+    ) -> Result<InvitationDecision, CallbackError> {
+        tracing::debug!("Deciding on room invitation");
+
+        let global_config = self
+            .bot
+            .global_config()
+            .await
+            .map_err(|e| CallbackError::Unknown(e.into()))?;
+
+        let sender_is_admin = mxidwc::match_user_id(
+            room_member.sender.clone().as_str(),
+            self.bot.admin_pattern_regexes(),
+        );
+
+        let sender_is_allowed_user = match &global_config.access.user_patterns {
+            Some(user_patterns) => {
+                let allowed_user_regexes = mxidwc::parse_patterns_vector(user_patterns)
+                    .map_err(|e| CallbackError::Unknown(e.into()))?;
+
+                mxidwc::match_user_id(room_member.sender.clone().as_str(), &allowed_user_regexes)
+            }
+            None => false,
+        };
+
+        if !(sender_is_admin || sender_is_allowed_user) {
+            return Ok(InvitationDecision::Reject);
+        }
+
+        Ok(InvitationDecision::Join)
+    }
+
+    #[tracing::instrument(name = "bot_on_joined", skip_all, fields(room_id = room.room_id().as_str(), event_id = event.event_id().as_str()))]
+    async fn on_joined(
+        &self,
+        event: AnySyncTimelineEvent,
+        room: Room,
+    ) -> Result<(), CallbackError> {
+        if self
+            .bot
+            .is_caught_up(event.origin_server_ts())
+            .await
+            .map_err(|e| {
+                CallbackError::Unknown(
+                    format!("Failed to determine catch-up state: {:?}", e).into(),
+                )
+            })?
+        {
+            tracing::debug!(
+                event_origin_server_ts = format!("{:?}", event.origin_server_ts()),
+                "Ignoring old room join event",
+            );
+
+            return Ok(());
+        }
+
+        tracing::info!("Handling room join");
+
+        let global_config = self
+            .bot
+            .global_config()
+            .await
+            .map_err(|e| CallbackError::Unknown(e.into()))?;
+
+        let room_config_manager = self.bot.room_config_manager().lock().await;
+
+        // We force-create a new config when we join anew to ensure we:
+        // - always start from a known clean state
+        // - record the last join timestamp, so we can accurately service the room (ignoring past messages, etc.)
+        let room_config = room_config_manager
+            .create_new_for_room(&room)
+            .await
+            .map_err(|e| CallbackError::Unknown(e.into()))?;
+
+        let room_config_context = RoomConfigContext::new(global_config, room_config);
+
+        let event_span = tracing::error_span!("join_controller");
+
+        let result = crate::controller::join::handle(&self.bot, &room, &room_config_context)
+            .instrument(event_span)
+            .await
+            .map_err(|e| CallbackError::Unknown(e.into()));
+
+        self.bot.catch_up(event.origin_server_ts()).await;
+
+        result
+    }
+
+    async fn on_being_last_member(
+        &self,
+        _event: AnySyncTimelineEvent,
+        room: mxlink::matrix_sdk::Room,
+    ) -> Result<(), CallbackError> {
+        tracing::info!(
+            "Leaving room {} because we are the last member",
+            room.room_id()
+        );
+
+        // We are last in this room. Let's just leave
+        room.leave().await.map_err(|e| e.into())
+    }
+}