//! 平台适配器(P0 预留项 ④:trait 分层,统一 InboundEvent/OutboundAction) //! //! 网关核心只认识这两个类型,四端差异(飞书卡片/企微 Markdown/钉钉卡片/Web JSON)收敛在各 adapter。 //! P0 只实现 Web,飞书/企微/钉钉留 trait 位(P1/P2/P4 各自落地,不动核心)。 use async_trait::async_trait; use tokio::sync::mpsc; /// 平台入站事件(IM 平台 → 网关,已归一化) #[derive(Debug, Clone)] pub enum InboundEvent { /// 用户发消息 Message { platform: String, chat_id: String, user_id: String, text: String, }, /// 会话切换(/new 或菜单) NewSession { platform: String, chat_id: String }, /// 审批按钮回调 Approval { platform: String, chat_id: String, approval_id: String, approved: bool, }, } /// 平台出站动作(网关 → IM 平台,已归一化) #[derive(Debug, Clone)] pub enum OutboundAction { /// 发/更新 AI 流式文本(全量内容,适配器自选卡片或文本) StreamText { chat_id: String, text: String, done: bool }, /// 发送审批卡片(同意/拒绝按钮) ApprovalCard { chat_id: String, approval_id: String, title: String, detail: String }, /// 发系统提示 Notice { chat_id: String, text: String }, } /// 平台适配器 #[async_trait] pub trait PlatformAdapter: Send + Sync { /// 平台名(feishu/wecom/dingtalk/web,与 session_map.platform 对应)。 /// 核心循环经 platform_index 注册表间接使用。 #[allow(dead_code)] fn name(&self) -> &str; /// 启动适配器(长连接/回调服务器/WS),入站事件从 events 通道流出。阻塞运行。 async fn run(&self, events: mpsc::Sender) -> anyhow::Result<()>; /// 出站动作分发(网关核心调用) async fn dispatch(&self, action: OutboundAction) -> anyhow::Result<()>; /// 清理平台的流式消息状态(网关在新建会话映射时调用,兜底上轮异常中止的残留)。 /// 默认空实现:无流式覆写语义的平台(Web 直接推帧)无需关心。 async fn reset_stream(&self, _chat_id: &str) {} } /// Web Chat 适配器(P1:axum 同端口托管静态页 + WS,浏览器直连) /// /// - `GET /` → 内嵌单文件聊天页(include_str!,零运行时文件依赖) /// - `GET /ws` → WS 升级,协议见 handle_web_ws(与 P0 纯 WS 版完全一致) pub mod web { use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::State; use axum::response::{Html, Response}; use axum::routing::get; use axum::Router; use futures_util::{SinkExt, StreamExt}; use tokio::sync::{mpsc, Mutex}; use super::{InboundEvent, OutboundAction, PlatformAdapter}; /// 内嵌聊天页(单文件,无外部依赖,构建期打进二进制) const INDEX_HTML: &str = include_str!("../web/index.html"); /// 浏览器连接注册表(chat_id → ws 发送通道) type WsRegistry = Arc>>>; /// axum 路由共享状态(注册表 + 核心入站通道) #[derive(Clone)] struct WsState { registry: WsRegistry, events: mpsc::Sender, } /// Web 适配器:HTTP+WS 服务监听 0.0.0.0:9191 #[derive(Clone)] pub struct WebAdapter { registry: WsRegistry, } impl WebAdapter { pub fn new() -> Self { Self { registry: Arc::new(Mutex::new(HashMap::new())) } } } impl Default for WebAdapter { fn default() -> Self { Self::new() } } #[async_trait] impl PlatformAdapter for WebAdapter { fn name(&self) -> &str { "web" } async fn run(&self, events: mpsc::Sender) -> anyhow::Result<()> { let state = WsState { registry: self.registry.clone(), events }; let app = Router::new() .route("/", get(|| async { Html(INDEX_HTML) })) .route("/ws", get(ws_upgrade)) .with_state(state); let listener = tokio::net::TcpListener::bind("0.0.0.0:9191").await?; tracing::info!("[web] HTTP+WS 服务监听 http://0.0.0.0:9191 (页面 / , WS /ws)"); axum::serve(listener, app).await?; Ok(()) } async fn dispatch(&self, action: OutboundAction) -> anyhow::Result<()> { match action { OutboundAction::StreamText { chat_id, text, done } => { let frame = serde_json::json!({ "type": if done { "stream_done" } else { "stream_delta" }, "text": text, }); self.send_ws(&chat_id, &frame.to_string()).await } OutboundAction::ApprovalCard { chat_id, approval_id, title, detail } => { let frame = serde_json::json!({ "type": "approval", "approval_id": approval_id, "title": title, "detail": detail, }); self.send_ws(&chat_id, &frame.to_string()).await } OutboundAction::Notice { chat_id, text } => { let frame = serde_json::json!({ "type": "notice", "text": text }); self.send_ws(&chat_id, &frame.to_string()).await } } } } impl WebAdapter { async fn send_ws(&self, chat_id: &str, text: &str) -> anyhow::Result<()> { let registry = self.registry.lock().await; if let Some(tx) = registry.get(chat_id) { tx.send(WsMessage::Text(text.to_string())).await?; } Ok(()) } } /// WS 升级入口 async fn ws_upgrade(State(state): State, ws: WebSocketUpgrade) -> Response { ws.on_upgrade(move |socket| async move { if let Err(e) = handle_web_ws(socket, state).await { tracing::warn!("[web] WS 会话异常结束: {e:#}"); } }) } /// 单个浏览器 WS 会话。 /// /// 协议(浏览器 → 网关): /// ```json /// 首帧绑定: {"type":"hello","chat_id":"...","user_id":"..."} /// 消息: {"type":"message","text":"..."} /// 重置会话: {"type":"new_session"} /// 审批: {"type":"approval","approval_id":"...","approved":true|false} /// ``` /// 网关 → 浏览器:stream_delta / stream_done / approval / notice(见 dispatch)。 async fn handle_web_ws(socket: WebSocket, state: WsState) -> anyhow::Result<()> { let (mut sink, mut upstream) = socket.split(); // 首帧 hello 绑定 chat_id let first = match upstream.next().await { Some(Ok(m)) => m, _ => { tracing::warn!("[web] 浏览器连接即断(无首帧)"); return Ok(()); } }; let hello: serde_json::Value = match &first { WsMessage::Text(t) => serde_json::from_str(t).unwrap_or_default(), _ => { tracing::warn!("[web] 首帧非文本,拒绝绑定"); return Ok(()); } }; let chat_id = hello .get("chat_id") .and_then(|v| v.as_str()) .unwrap_or("web-anon") .to_string(); let user_id = hello.get("user_id").and_then(|v| v.as_str()).unwrap_or("anon").to_string(); // 注册发送通道:出站帧经 ws_tx → forward 任务写 socket let (ws_tx, mut ws_rx) = mpsc::channel::(64); state.registry.lock().await.insert(chat_id.clone(), ws_tx); tracing::info!("[web] 浏览器接入 chat_id={chat_id} user={user_id}"); let forward = tokio::spawn(async move { while let Some(frame) = ws_rx.recv().await { if sink.send(frame).await.is_err() { break; } } }); let result: anyhow::Result<()> = async { while let Some(msg) = upstream.next().await { let msg = msg?; match msg { WsMessage::Text(t) => { let v: serde_json::Value = serde_json::from_str(&t).unwrap_or_default(); match v.get("type").and_then(|x| x.as_str()).unwrap_or("") { "message" => { let text = v.get("text").and_then(|x| x.as_str()).unwrap_or("").to_string(); state .events .send(InboundEvent::Message { platform: "web".into(), chat_id: chat_id.clone(), user_id: user_id.clone(), text, }) .await .map_err(|_| anyhow::anyhow!("核心入站通道关闭"))?; } "new_session" => { state .events .send(InboundEvent::NewSession { platform: "web".into(), chat_id: chat_id.clone(), }) .await .map_err(|_| anyhow::anyhow!("核心入站通道关闭"))?; } "approval" => { let approval_id = v.get("approval_id").and_then(|x| x.as_str()).unwrap_or("").to_string(); let approved = v.get("approved").and_then(|x| x.as_bool()).unwrap_or(false); state .events .send(InboundEvent::Approval { platform: "web".into(), chat_id: chat_id.clone(), approval_id, approved, }) .await .map_err(|_| anyhow::anyhow!("核心入站通道关闭"))?; } _ => {} } } WsMessage::Close(_) => break, _ => {} } } Ok(()) } .await; state.registry.lock().await.remove(&chat_id); forward.abort(); tracing::info!("[web] 浏览器断开 chat_id={chat_id}"); result } }