252 lines
9.8 KiB
Rust
252 lines
9.8 KiB
Rust
//! 平台适配器(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<InboundEvent>) -> anyhow::Result<()>;
|
|
|
|
/// 出站动作分发(网关核心调用)
|
|
async fn dispatch(&self, action: OutboundAction) -> anyhow::Result<()>;
|
|
|
|
/// 清理平台的流式消息状态(网关在新建会话映射时调用,兜底上轮异常中止的残留)。
|
|
/// 默认空实现:无流式覆写语义的平台(Web 直接推帧)无需关心。
|
|
async fn reset_stream(&self, _chat_id: &str) {}
|
|
}
|
|
|
|
/// Web Chat 适配器(网关直接托管浏览器 WS,无三方依赖)
|
|
pub mod web {
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
|
|
use async_trait::async_trait;
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use tokio::sync::{mpsc, Mutex};
|
|
use tokio_tungstenite::tungstenite::Message;
|
|
|
|
use super::{InboundEvent, OutboundAction, PlatformAdapter};
|
|
|
|
/// 浏览器连接注册表(chat_id → ws 发送通道)
|
|
type WsRegistry = Arc<Mutex<HashMap<String, mpsc::Sender<Message>>>>;
|
|
|
|
/// Web 适配器:浏览器 WS 直连网关 0.0.0.0:9191(accept 协议见 handle_web_ws)
|
|
#[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<InboundEvent>) -> anyhow::Result<()> {
|
|
let listener = tokio::net::TcpListener::bind("0.0.0.0:9191").await?;
|
|
tracing::info!("[web] WS 服务监听 0.0.0.0:9191");
|
|
loop {
|
|
let (stream, _peer) = listener.accept().await?;
|
|
let registry = self.registry.clone();
|
|
let events = events.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = handle_web_ws(stream, registry, events).await {
|
|
tracing::warn!("[web] 浏览器连接异常退出: {e}");
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
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(Message::Text(text.to_string())).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// 单个浏览器 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(
|
|
stream: tokio::net::TcpStream,
|
|
registry: WsRegistry,
|
|
events: mpsc::Sender<InboundEvent>,
|
|
) -> anyhow::Result<()> {
|
|
let ws = tokio_tungstenite::accept_async(stream).await?;
|
|
let (mut sink, mut upstream) = ws.split();
|
|
|
|
// 首帧 hello 绑定 chat_id
|
|
let first = upstream.next().await.ok_or_else(|| anyhow::anyhow!("浏览器连接即断"))??;
|
|
let hello: serde_json::Value = match &first {
|
|
Message::Text(t) => serde_json::from_str(t)?,
|
|
_ => anyhow::bail!("首帧非文本"),
|
|
};
|
|
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::<Message>(64);
|
|
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 {
|
|
match msg? {
|
|
Message::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();
|
|
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" => {
|
|
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);
|
|
events
|
|
.send(InboundEvent::Approval {
|
|
platform: "web".into(),
|
|
chat_id: chat_id.clone(),
|
|
approval_id,
|
|
approved,
|
|
})
|
|
.await
|
|
.map_err(|_| anyhow::anyhow!("核心入站通道关闭"))?;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
Message::Close(_) => break,
|
|
_ => {}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
.await;
|
|
|
|
registry.lock().await.remove(&chat_id);
|
|
forward.abort();
|
|
tracing::info!("[web] 浏览器断开 chat_id={chat_id}");
|
|
result
|
|
}
|
|
}
|