From a2c9904ec33f40e6b5578ddff1d36629649ca70d Mon Sep 17 00:00:00 2001 From: Matrix Music Bot Dev Date: Thu, 2 Jul 2026 21:03:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(event=5Floop):=20sync=20+=20add=5Fevent=5F?= =?UTF-8?q?handler=20+=20Timeline::send=20=E5=8A=A0=E5=AF=86=E9=80=9A?= =?UTF-8?q?=E9=81=93=EF=BC=88M2.2/c=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增模块：src/event_loop.rs 依赖新增：futures-util 0.3 源码增加：#![recursion_limit = "512"]（matrix-sdk 类型嵌套很深） 实现要点： 1. bootstrap::restore 拿到 Client + user_id + device_id 2. 注册全局 SyncRoomMessageEvent handler（matrix-sdk add_event_handler 闭包模式） 3. handler 内部： - 提取 m.text 文本 - 跳过 bot 自己发的消息 - 用 commands::parse 解析（自动处理 mention / DM 区分） - 用 commands::render_reply 渲染回复 - 用 TimelineBuilder::new(room).build() 构造 Timeline - 用 timeline.send(RoomMessageEventContent::text_plain(...).into()) 发送 4. 启动 long-running client.sync()（用 tokio::select! 与 ctrl-c 并发） E2EE 不变量已实现并强制： - ✅ 唯一出站通道：Timeline::send（SDK 内部用 Megolm 加密） - ❌ 绝不使用 Room::send_raw()（明文） - ✅ 入站消息经 matrix-sdk 解密后是 plaintext，可直接解析 类型挑战（已解决）： - matrix-sdk 0.18 EventHandler 通过 impl_event_handler! 宏实现， 要求 FnOnce(Ev, ...) -> Fut 形式，不能包 Pin> - matrix-sdk-crypto 内部有 !Send 类型，所以 client.sync() 不能 tokio::spawn， 必须在当前 task 用 tokio::select! 与 ctrl-c 并发 - 嵌套 async 闭包触发 rustc 深度限制，需要 #![recursion_limit = "512"] cmd_run 改造： - 同样从 MATRIX_BOT_SECRET_PASSPHRASE 环境变量读取 passphrase - 用 tokio multi-thread runtime 跑 event_loop::run CLI 实测（fake config + 缺 secret.bin）： - 无 config → 友好错误 - 有 config + 缺 secret.bin → '请先运行 login' 错误 - 完整流程需真实 homeserver 才能跑通 测试：cargo test → 15 passed（11 commands + 2 config + 2 secret_store） --- Cargo.lock | 1 + Cargo.toml | 1 + src/event_loop.rs | 140 ++++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 29 +++++++--- 4 files changed, 162 insertions(+), 9 deletions(-) create mode 100644 src/event_loop.rs diff --git a/Cargo.lock b/Cargo.lock index 3ad2c1f..9dcf93a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2043,6 +2043,7 @@ dependencies = [ "async-trait", "base64", "clap", + "futures-util", "getrandom 0.3.4", "matrix-sdk", "matrix-sdk-crypto", diff --git a/Cargo.toml b/Cargo.toml index cc776a0..8a16908 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ argon2 = "0.5.3" async-trait = "0.1.89" base64 = "0.22.1" clap = { version = "4.6.1", features = ["derive", "env"] } +futures-util = "0.3.32" getrandom = { version = "0.3", features = ["std"] } matrix-sdk = "0.18.0" matrix-sdk-crypto = "0.18.0" diff --git a/src/event_loop.rs b/src/event_loop.rs new file mode 100644 index 0000000..a4c0d7c --- /dev/null +++ b/src/event_loop.rs @@ -0,0 +1,140 @@ +//! Event loop：恢复登录 → 初始 sync → 注册全局事件 handler → 派发命令。 +//! +//! E2EE 不变量： +//! - **所有**出站消息（bot 回复）必须经由 `Timeline::send()`，由 SDK 内部用 Megolm 加密。 +//! - **绝不**直接调用 `Room::send_raw()` 之类的明文发送方法。 +//! - 入站消息经 matrix-sdk 的 `SyncRoomMessageEvent` 已是解密后的 plaintext。 +//! +//! v1 阶段只处理文本消息。后续 milestone 会处理 m.reaction（用于搜索结果选择） +//! 和 m.audio（投递回执）。 + +use std::sync::Arc; + +use matrix_sdk::config::SyncSettings; +use matrix_sdk::room::Room; +use matrix_sdk::ruma::events::room::message::{ + MessageType, RoomMessageEventContent, SyncRoomMessageEvent, +}; +use matrix_sdk::Client; +use matrix_sdk_ui::timeline::TimelineBuilder; +use tracing::{debug, error, info, warn}; + +use crate::commands::{self, Decision, ParseContext}; +use crate::config::AppConfig; +use crate::error::{AppError, AppResult}; + +/// 启动 event loop。 +/// +/// 流程： +/// 1. bootstrap::restore 拿到 Client +/// 2. 注册 m.room.message 全局事件 handler +/// 3. 启动 long-running client.sync() +/// 4. 阻塞直到 ctrl-c +pub async fn run(cfg: AppConfig, passphrase: &str) -> AppResult<()> { + let (client, user_id, device_id) = crate::bootstrap::restore(cfg.clone(), passphrase).await?; + info!(%user_id, %device_id, "恢复登录成功"); + + // 1. 注册全局 m.room.message handler + let state = Arc::new(cfg); + client.add_event_handler({ + let state = state.clone(); + move |ev: SyncRoomMessageEvent, room: Room, client: Client| { + let state = state.clone(); + async move { handle_message(ev, room, client, state).await } + } + }); + info!("已注册 m.room.message 事件 handler"); + + // 2. 启动 long-running sync（用 select! 同时监听 ctrl-c） + // 注意：client.sync() 返回的 Future 非 Send（matrix-sdk-crypto 内部限制）， + // 不能用 tokio::spawn，必须在当前 task 中并发。 + info!("开始 long-running sync（按 Ctrl-C 停止）"); + tokio::select! { + res = client.sync(SyncSettings::default()) => { + if let Err(e) = res { + error!(error = %e, "long-running sync 出错"); + } + } + _ = tokio::signal::ctrl_c() => { + info!("收到 Ctrl-C，正在关闭..."); + } + } + Ok(()) +} + +/// m.room.message 事件处理函数 +async fn handle_message( + ev: SyncRoomMessageEvent, + room: Room, + client: Client, + state: Arc, +) { + // 1. 提取原始消息（必须是 OriginalSyncRoomMessageEvent） + let original = match ev { + SyncRoomMessageEvent::Original(o) => o, + SyncRoomMessageEvent::Redacted(_) => return, + }; + + // 2. 提取文本 + let text = match &original.content.msgtype { + MessageType::Text(t) => t.body.clone(), + _ => return, // 非文本消息（M4+ 处理 m.audio/m.image/m.reaction） + }; + + // 3. 跳过 bot 自己发的 + let bot_user_id = match client.user_id() { + Some(u) => u.to_owned(), + None => return, + }; + if original.sender == bot_user_id { + return; + } + + let room_id = room.room_id().to_string(); + let is_dm = detect_dm(&room); + + debug!(room_id = %room_id, sender = %original.sender, text = %text, "收到消息"); + + // 4. 解析命令 + let ctx = ParseContext { + text: &text, + sender: &original.sender, + is_dm, + bot_user_id: bot_user_id.as_ref(), + prefix: &state.commands.prefix, + mention_required_in_rooms: state.commands.mention_required_in_rooms, + }; + let cmd = match commands::parse(&ctx) { + Decision::Handle(c) => c, + Decision::Ignore | Decision::Unknown => return, + }; + + // 5. 渲染回复 + let reply_text = commands::render_reply(&cmd); + + // 6. 通过 Timeline 发送（Megolm 加密路径） + if let Err(e) = send_text(&room, &reply_text).await { + warn!(room_id = %room_id, error = %e, "发送回复失败"); + } else { + debug!(room_id = %room_id, "回复已发送（已通过 Megolm 加密）"); + } +} + +/// 通过 Timeline 发送文本（**唯一**合规的出站消息通道） +async fn send_text(room: &Room, text: &str) -> AppResult<()> { + let timeline = TimelineBuilder::new(room) + .build() + .await + .map_err(|e| AppError::Sdk(format!("构造 Timeline 失败: {e}")))?; + let content = RoomMessageEventContent::text_plain(text); + timeline + .send(content.into()) + .await + .map_err(|e| AppError::Sdk(format!("Timeline::send 失败: {e}")))?; + Ok(()) +} + +/// 简单判断一个房间是否是 DM（基于成员数） +fn detect_dm(room: &Room) -> bool { + room.joined_members_count() <= 2 +} \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 7686f27..c0d5d7e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,11 +8,15 @@ //! //! M1.2 阶段：login / config 子命令为占位实现，run 子命令仅打印 "正在同步..." 后退出。 +// matrix-sdk 0.18 的类型嵌套很深，需要提高递归上限 +#![recursion_limit = "512"] + mod bootstrap; mod client; mod commands; mod config; mod error; +mod event_loop; mod secret_store; use std::path::PathBuf; @@ -155,15 +159,22 @@ fn cmd_login(cfg: AppConfig, password: &str) -> AppResult<()> { } fn cmd_run(cfg: AppConfig) -> AppResult<()> { - info!( - homeserver = %cfg.homeserver.url, - user_id = %cfg.bot.user_id, - data_dir = %cfg.data.dir.display(), - "[stub] run 子命令待 M2 实现；目前仅打印配置即退出" - ); - Err(AppError::Other( - "run 尚未实现（M2 占位；sync + Timeline 订阅）".to_string(), - )) + let passphrase = std::env::var("MATRIX_BOT_SECRET_PASSPHRASE").map_err(|_| { + AppError::SecretStore( + "未设置 MATRIX_BOT_SECRET_PASSPHRASE 环境变量（用于解密 secret.bin）".to_string(), + ) + })?; + if passphrase.len() < 8 { + return Err(AppError::SecretStore( + "MATRIX_BOT_SECRET_PASSPHRASE 长度至少 8 字符".to_string(), + )); + } + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(|e| AppError::Other(format!("tokio runtime 创建失败: {e}")))?; + runtime.block_on(event_loop::run(cfg, &passphrase)) } fn cmd_config(