use std::sync::Arc; use matrix_sdk::config::SyncSettings; use matrix_sdk::room::Room; use matrix_sdk::Client; use ruma::events::room::member::StrippedRoomMemberEvent; use ruma::events::room::message::OriginalSyncRoomMessageEvent; use ruma::events::room::redaction::OriginalSyncRoomRedactionEvent; use tokio::sync::Mutex; use tracing::{error, info, warn}; use crate::archive::indexer::Indexer; use crate::archive::schema::ArchiveDocument; use crate::brain::conversation::{ContextMessage, ConversationManager}; use crate::brain::evaluator::{Engagement, Evaluator}; use crate::brain::responder::Responder; use crate::config::Config; use crate::matrix_utils; pub struct AppState { pub config: Arc, pub indexer: Arc, pub evaluator: Arc, pub responder: Arc, pub conversations: Arc>, pub mistral: Arc, } pub async fn start_sync(client: Client, state: Arc) -> anyhow::Result<()> { // Register event handlers let s = state.clone(); client.add_event_handler( move |event: OriginalSyncRoomMessageEvent, room: Room| { let state = s.clone(); async move { if let Err(e) = handle_message(event, room, state).await { error!("Error handling message: {e}"); } } }, ); let s = state.clone(); client.add_event_handler( move |event: OriginalSyncRoomRedactionEvent, _room: Room| { let state = s.clone(); async move { handle_redaction(event, &state).await; } }, ); client.add_event_handler( move |event: StrippedRoomMemberEvent, room: Room| async move { handle_invite(event, room).await; }, ); info!("Starting Matrix sync loop"); let settings = SyncSettings::default(); client.sync(settings).await?; Ok(()) } async fn handle_message( event: OriginalSyncRoomMessageEvent, room: Room, state: Arc, ) -> anyhow::Result<()> { let sender = event.sender.to_string(); let room_id = room.room_id().to_string(); let event_id = event.event_id.to_string(); let timestamp = event.origin_server_ts.0.into(); // Check if this is an edit if let Some((original_id, new_body)) = matrix_utils::extract_edit(&event) { state .indexer .update_edit(&original_id.to_string(), &new_body) .await; return Ok(()); } let Some(body) = matrix_utils::extract_body(&event) else { return Ok(()); }; let room_name = matrix_utils::room_display_name(&room); let sender_name = room .get_member_no_sync(&event.sender) .await .ok() .flatten() .and_then(|m| m.display_name().map(|s| s.to_string())); let reply_to = matrix_utils::extract_reply_to(&event).map(|id| id.to_string()); let thread_id = matrix_utils::extract_thread_id(&event).map(|id| id.to_string()); // Archive the message let doc = ArchiveDocument { event_id: event_id.clone(), room_id: room_id.clone(), room_name: Some(room_name.clone()), sender: sender.clone(), sender_name: sender_name.clone(), timestamp, content: body.clone(), reply_to, thread_id, media_urls: Vec::new(), event_type: "m.room.message".into(), edited: false, redacted: false, }; state.indexer.add(doc).await; // Update conversation context let is_dm = room.is_direct().await.unwrap_or(false); { let mut convs = state.conversations.lock().await; convs.add_message( &room_id, is_dm, ContextMessage { sender: sender_name.clone().unwrap_or_else(|| sender.clone()), content: body.clone(), timestamp, }, ); } // Evaluate whether to respond let recent: Vec = { let convs = state.conversations.lock().await; convs .get_context(&room_id) .iter() .map(|m| format!("{}: {}", m.sender, m.content)) .collect() }; let engagement = state .evaluator .evaluate(&sender, &body, is_dm, &recent, &state.mistral) .await; let (should_respond, is_spontaneous) = match engagement { Engagement::MustRespond { reason } => { info!(?reason, "Must respond"); (true, false) } Engagement::MaybeRespond { relevance, hook } => { info!(relevance, hook = hook.as_str(), "Maybe respond (spontaneous)"); (true, true) } Engagement::Ignore => (false, false), }; if !should_respond { return Ok(()); } // Show typing indicator let _ = room.typing_notice(true).await; let context = { let convs = state.conversations.lock().await; convs.get_context(&room_id) }; let members = matrix_utils::room_member_names(&room).await; let display_sender = sender_name.as_deref().unwrap_or(&sender); let response = state .responder .generate_response( &context, &body, display_sender, &room_name, &members, is_spontaneous, &state.mistral, ) .await; // Stop typing indicator let _ = room.typing_notice(false).await; if let Some(text) = response { let content = matrix_utils::make_reply_content(&text, event.event_id.to_owned()); if let Err(e) = room.send(content).await { error!("Failed to send response: {e}"); } } Ok(()) } async fn handle_redaction(event: OriginalSyncRoomRedactionEvent, state: &AppState) { if let Some(redacted_id) = &event.redacts { state.indexer.update_redaction(&redacted_id.to_string()).await; } } async fn handle_invite(event: StrippedRoomMemberEvent, room: Room) { // Only handle our own invites if event.state_key != room.own_user_id() { return; } info!(room_id = %room.room_id(), "Received invite, auto-joining"); tokio::spawn(async move { for attempt in 0..3u32 { match room.join().await { Ok(_) => { info!(room_id = %room.room_id(), "Joined room"); return; } Err(e) => { warn!(room_id = %room.room_id(), attempt, "Failed to join: {e}"); tokio::time::sleep(std::time::Duration::from_secs(2u64.pow(attempt))).await; } } } error!(room_id = %room.room_id(), "Failed to join after retries"); }); }