新增: 飞书直连渠道接线(文本链路第一批)

- 桌面端出站WS长连接,复用spawn_agentic_session通用核心与完成通知桥
- 配置读feishu-config KV,白名单空=放行并自助提示open_id
- todo并入下午分析产出(存储架构/知识库/规范治理/基建走查)
This commit is contained in:
lxy
2026-10-05 14:59:55 +08:00
parent d5766c2142
commit 8233ba095e
+221
View File
@@ -0,0 +1,221 @@
//! 飞书直连渠道:桌面端出站 WS 长连接(免公网回调),消息进 ai-chat 会话链路
//!
//! 形态依据:docs/02-架构设计/专项设计/飞书直连接入-形态修正与问题记录-2026-10-05.md
//! (用户 10-05 推翻网关链路,定稿桌面本地直连;参照 u-claw feishu_channel.go 单进程闭环)
//! 内核自 crates/df-im-gateway/src/feishu.rs 提炼(用户令:并入现有,不新建 crate);
//! 会话复用 spawn_agentic_session 通用核心(pin_active=false,与编排节点同款调用方),
//! 完成通知走 AppState 通知桥(register/notify_agent_completion,编排节点先例)。
//!
//! 第一批:文本收发(整段回复);流式 edit_text/审批卡片放第二批(内核留存于网关 feishu.rs)。
use std::collections::HashMap;
use std::sync::Arc;
use tauri::Manager;
use larksuite_oapi_sdk_rs::channel::{Channel, ChannelPolicy, SendInput};
use larksuite_oapi_sdk_rs::{EventDispatcher, LarkClient, LarkError, RequestOption};
use serde::Deserialize;
use df_storage::crud::SettingsRepo;
use df_types::types::new_id;
/// app_settings 键:飞书渠道配置(与设置面板 FeishuPanel 同键契约)
const KV_FEISHU_CONFIG: &str = "feishu-config";
/// app_settings 键:飞书 chat_id → conversation_id 映射(持久化,跨重启续会话)
const KV_CHAT_MAP: &str = "feishu-chat-map";
#[derive(Debug, Clone, Deserialize)]
struct FeishuConfig {
#[serde(default)]
enabled: bool,
#[serde(default)]
app_id: String,
#[serde(default)]
app_secret: String,
/// 空 = 放行所有(首启抓 open_id 回填);open_id 按应用隔离,严禁跨应用搬运
#[serde(default)]
whitelist: Vec<String>,
}
/// settings 值归一:前端 JSON.stringify 落库带引号,去首尾引号(SettingsNorm 正典同款)
fn norm(raw: Option<String>) -> String {
raw.map(|s| s.trim().trim_matches('"').to_string())
.unwrap_or_default()
}
/// 渠道运行上下文(回调闭包共享)
struct Ctx {
app: tauri::AppHandle,
client: LarkClient,
whitelist: Vec<String>,
/// chat_id → conversation_id(启动时从 KV 恢复,变更回写)
chat_map: Arc<tokio::sync::Mutex<HashMap<String, String>>>,
}
/// 启动入口:lib.rs setup 调用。未启用/配置缺失/构造失败一律记日志返回,不阻塞应用启动。
pub fn start_if_enabled(app: tauri::AppHandle) {
tauri::async_runtime::spawn(async move {
let db = app.state::<crate::state::AppState>().db.clone();
let repo = SettingsRepo::new(&db);
let raw = norm(repo.get(KV_FEISHU_CONFIG).await.ok().flatten());
if raw.is_empty() {
tracing::info!("[feishu] feishu-config 未配置,渠道不启动");
return;
}
let Ok(cfg) = serde_json::from_str::<FeishuConfig>(&raw) else {
tracing::error!("[feishu] feishu-config 解析失败,渠道不启动");
return;
};
if !cfg.enabled || cfg.app_id.is_empty() || cfg.app_secret.is_empty() {
tracing::info!("[feishu] enabled=false 或凭证缺失,渠道不启动");
return;
}
let client = match LarkClient::builder(&cfg.app_id, &cfg.app_secret).build() {
Ok(c) => c,
Err(e) => {
tracing::error!("[feishu] LarkClient 构造失败: {e}");
return;
}
};
// 启动时从 KV 恢复 chat 映射
let restored: HashMap<String, String> = serde_json::from_str(
&norm(repo.get(KV_CHAT_MAP).await.ok().flatten()),
)
.unwrap_or_default();
let ctx = Arc::new(Ctx {
app,
client: client.clone(),
whitelist: cfg.whitelist,
chat_map: Arc::new(tokio::sync::Mutex::new(restored)),
});
if let Err(e) = run_loop(ctx, client).await {
tracing::error!("[feishu] 长连接退出: {e}");
}
});
}
async fn run_loop(ctx: Arc<Ctx>, client: LarkClient) -> anyhow::Result<()> {
let dispatcher = EventDispatcher::new("", "");
let ctx_msg = ctx.clone();
// Channel<'a> 借用 client;Box::leak 提升为 'static(渠道生命周期即进程生命周期)
let leaked: &'static LarkClient = Box::leak(Box::new(client));
let channel: Channel<'static> = Channel::builder(leaked, dispatcher)
.policy(ChannelPolicy::default().require_mention(false))
.on_message(move |msg| {
let ctx = ctx_msg.clone();
async move {
handle_message(ctx, msg).await;
Ok(())
}
})
.build();
tracing::info!("[feishu] 长连接启动(桌面端直连,免公网回调)");
channel
.start()
.await
.map_err(|e: LarkError| anyhow::anyhow!("飞书长连接退出: {e}"))?;
Ok(())
}
/// 入站文本:群聊需 @bot;白名单校验;/new 重置;其余走 ai-chat 会话链路
async fn handle_message(
ctx: Arc<Ctx>,
msg: larksuite_oapi_sdk_rs::channel::NormalizedMessage,
) {
if msg.message_type != "text" {
return;
}
if msg.chat_type == "group" && !msg.mentioned_bot {
return;
}
let text = msg.text.clone().unwrap_or_default().trim().to_string();
if text.is_empty() {
return;
}
let chat_id = msg.chat_id.clone();
let open_id = msg
.sender
.user_id
.as_ref()
.and_then(|u| u.open_id.clone())
.unwrap_or_default();
// 白名单:空=放行;非空且不在内 → 提示并附自己的 open_id(首启自助回填,免翻日志)
if !ctx.whitelist.is_empty() && !ctx.whitelist.iter().any(|w| w == &open_id) {
let hint = format!("⛔ 你不在白名单内。你的 open_id:`{open_id}`(发给管理员回填后重试)");
send_text(&ctx.client, &chat_id, &hint).await;
return;
}
// /new 重置该飞书会话映射(下次消息开新 conv)
if text == "/new" {
ctx.chat_map.lock().await.remove(&chat_id);
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let snapshot = ctx.chat_map.lock().await.clone();
save_chat_map(&db, &snapshot).await;
send_text(&ctx.client, &chat_id, "✅ 已重置会话").await;
return;
}
// chat_id → conv_id 映射(内存命中优先,未命中新建)
let conv_id = {
let mut map = ctx.chat_map.lock().await;
map.entry(chat_id.clone()).or_insert_with(new_id).clone()
};
let state = ctx.app.state::<crate::state::AppState>();
let (tx, rx) = tokio::sync::oneshot::channel::<String>();
state.register_agent_completion(conv_id.clone(), tx);
match crate::commands::ai::commands::chat::spawn_agentic_session(
&ctx.app,
&state,
text,
None,
None,
None,
Some(conv_id.clone()),
None,
None,
false, // 不劫持用户活跃面板(编排节点同款语义)
)
.await
{
Ok(_) => {
let db = ctx.app.state::<crate::state::AppState>().db.clone();
let snapshot = ctx.chat_map.lock().await.clone();
save_chat_map(&db, &snapshot).await;
}
Err(e) => {
// 启动失败:通知桥条目清掉防死条目(编排节点同款)
state.notify_agent_completion(&conv_id, String::new());
send_text(&ctx.client, &chat_id, &format!("❌ 会话启动失败: {e}")).await;
return;
}
}
// 等最终回复(通知桥 oneshot;超时兜底防悬挂)
let reply = match tokio::time::timeout(std::time::Duration::from_secs(600), rx).await {
Ok(Ok(text)) if !text.trim().is_empty() => text,
Ok(_) => "(会话已结束,无文本回复)".to_string(),
Err(_) => "⏱ 回复超时,请重试或 /new 重置".to_string(),
};
send_text(&ctx.client, &chat_id, &reply).await;
}
async fn send_text(client: &LarkClient, chat_id: &str, text: &str) {
let input = SendInput {
receive_id: Some(chat_id.to_string()),
text: Some(text.to_string()),
..Default::default()
};
if let Err(e) = client.channel_messaging().send(&input, &RequestOption::default()).await {
tracing::error!("[feishu] 发送失败: {e}");
}
}
async fn save_chat_map(db: &df_storage::db::Database, map: &HashMap<String, String>) {
let repo = SettingsRepo::new(db);
let json = serde_json::to_string(map).unwrap_or_default();
let _ = repo.set(KV_CHAT_MAP, &json).await;
}