3019 lines
105 KiB
Rust
3019 lines
105 KiB
Rust
use std::collections::HashMap;
|
||
use std::path::Path;
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, Instant};
|
||
|
||
use async_trait::async_trait;
|
||
use futures_util::{SinkExt, StreamExt};
|
||
use prost::{Message as ProstMessage, bytes::Bytes};
|
||
use serde::Deserialize;
|
||
use tokio::sync::{Mutex, RwLock};
|
||
use tokio::task::JoinHandle;
|
||
use tokio_util::sync::CancellationToken;
|
||
|
||
use crate::bus::{MediaItem, MessageBus, OutboundMessage};
|
||
use crate::channels::base::{Channel, ChannelError, LivePolicy, TurnSink, TurnTarget};
|
||
use crate::config::FeishuChannelConfig;
|
||
use crate::session::{ToolStatus, TurnBlock, TurnSnapshot, TurnStatus};
|
||
|
||
const FEISHU_API_BASE: &str = "https://open.feishu.cn/open-apis";
|
||
const FEISHU_WS_BASE: &str = "https://open.feishu.cn";
|
||
|
||
/// Heartbeat timeout for WS connection — must be larger than ping_interval (default 120 s).
|
||
/// If no binary frame (pong or event) is received within this window, reconnect.
|
||
const WS_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(300);
|
||
const CHANNEL_STOP_GRACE: Duration = if cfg!(test) {
|
||
Duration::from_millis(100)
|
||
} else {
|
||
Duration::from_secs(5)
|
||
};
|
||
/// Refresh tenant token this many seconds before the announced expiry.
|
||
const TOKEN_REFRESH_SKEW: Duration = Duration::from_secs(120);
|
||
/// Default tenant token TTL when `expire`/`expires_in` is absent.
|
||
const DEFAULT_TOKEN_TTL: Duration = Duration::from_secs(7200);
|
||
/// Dedup cache TTL (30 minutes).
|
||
const DEDUP_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
// Protobuf types for Feishu WebSocket protocol (pbbp2.proto)
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
#[derive(Clone, PartialEq, prost::Message)]
|
||
struct PbHeader {
|
||
#[prost(string, tag = "1")]
|
||
pub key: String,
|
||
#[prost(string, tag = "2")]
|
||
pub value: String,
|
||
}
|
||
|
||
/// Feishu WS frame.
|
||
/// method=0 → CONTROL (ping/pong) method=1 → DATA (events)
|
||
#[derive(Clone, PartialEq, prost::Message)]
|
||
struct PbFrame {
|
||
#[prost(uint64, tag = "1")]
|
||
pub seq_id: u64,
|
||
#[prost(uint64, tag = "2")]
|
||
pub log_id: u64,
|
||
#[prost(int32, tag = "3")]
|
||
pub service: i32,
|
||
#[prost(int32, tag = "4")]
|
||
pub method: i32,
|
||
#[prost(message, repeated, tag = "5")]
|
||
pub headers: Vec<PbHeader>,
|
||
#[prost(bytes = "vec", optional, tag = "8")]
|
||
pub payload: Option<Vec<u8>>,
|
||
}
|
||
|
||
/// POST /callback/ws/endpoint response
|
||
#[derive(Deserialize)]
|
||
struct WsEndpointResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
data: Option<WsEndpoint>,
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct WsEndpoint {
|
||
#[serde(rename = "URL")]
|
||
url: String,
|
||
#[serde(default)]
|
||
client_config: Option<WsClientConfig>,
|
||
}
|
||
|
||
#[derive(Deserialize, Default)]
|
||
struct WsClientConfig {
|
||
#[serde(rename = "PingInterval")]
|
||
ping_interval: Option<u64>,
|
||
}
|
||
|
||
/// Lark event envelope (method=1 / type=event payload)
|
||
#[derive(Deserialize)]
|
||
struct LarkEvent {
|
||
header: LarkEventHeader,
|
||
event: serde_json::Value,
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct LarkEventHeader {
|
||
event_type: String,
|
||
event_id: String,
|
||
}
|
||
|
||
impl std::fmt::Debug for LarkEventHeader {
|
||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||
f.debug_struct("LarkEventHeader")
|
||
.field("event_type", &self.event_type)
|
||
.field("event_id", &self.event_id)
|
||
.finish()
|
||
}
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct MsgReceivePayload {
|
||
sender: LarkSender,
|
||
message: LarkMessage,
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct LarkSender {
|
||
sender_id: LarkSenderId,
|
||
#[serde(default)]
|
||
sender_type: String,
|
||
}
|
||
|
||
#[derive(Deserialize, Default)]
|
||
struct LarkSenderId {
|
||
open_id: Option<String>,
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct LarkMessage {
|
||
message_id: String,
|
||
chat_id: String,
|
||
#[serde(default)]
|
||
chat_type: String,
|
||
message_type: String,
|
||
#[serde(default)]
|
||
content: String,
|
||
#[serde(default)]
|
||
parent_id: Option<String>,
|
||
#[serde(default)]
|
||
mentions: Vec<serde_json::Value>,
|
||
}
|
||
|
||
// ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
/// Cached tenant token with proactive refresh metadata.
|
||
#[derive(Clone)]
|
||
struct CachedTenantToken {
|
||
value: String,
|
||
refresh_after: Instant,
|
||
}
|
||
|
||
#[derive(Clone)]
|
||
pub struct FeishuChannel {
|
||
config: FeishuChannelConfig,
|
||
http_client: reqwest::Client,
|
||
running: Arc<RwLock<bool>>,
|
||
shutdown: Arc<RwLock<Option<CancellationToken>>>,
|
||
run_task: Arc<Mutex<Option<JoinHandle<()>>>>,
|
||
connected: Arc<RwLock<bool>>,
|
||
/// Cached tenant access token with proactive refresh.
|
||
tenant_token: Arc<RwLock<Option<CachedTenantToken>>>,
|
||
/// Dedup cache: WS message_ids seen in the last ~30 min.
|
||
seen_message_ids: Arc<RwLock<HashMap<String, Instant>>>,
|
||
/// Bot identity used to enforce group-chat @mention admission.
|
||
bot_open_id: Arc<RwLock<Option<String>>>,
|
||
}
|
||
|
||
/// Parsed message data from a Feishu frame
|
||
struct ParsedMessage {
|
||
message_id: String,
|
||
open_id: String,
|
||
chat_id: String,
|
||
content: String,
|
||
media: Vec<MediaItem>,
|
||
/// ID of the message this message is replying to (if any).
|
||
/// Used to fetch quoted message content for display.
|
||
parent_id: Option<String>,
|
||
}
|
||
|
||
impl FeishuChannel {
|
||
pub fn new(
|
||
mut config: FeishuChannelConfig,
|
||
workspace_dir: &Path,
|
||
) -> Result<Self, ChannelError> {
|
||
// Override media_dir to use workspace_dir/media/feishu
|
||
let media_dir = workspace_dir.join("media").join("feishu");
|
||
config.media_dir = media_dir.to_string_lossy().to_string();
|
||
|
||
Ok(Self {
|
||
config,
|
||
http_client: reqwest::Client::new(),
|
||
running: Arc::new(RwLock::new(false)),
|
||
shutdown: Arc::new(RwLock::new(None)),
|
||
run_task: Arc::new(Mutex::new(None)),
|
||
connected: Arc::new(RwLock::new(false)),
|
||
tenant_token: Arc::new(RwLock::new(None)),
|
||
seen_message_ids: Arc::new(RwLock::new(HashMap::new())),
|
||
bot_open_id: Arc::new(RwLock::new(None)),
|
||
})
|
||
}
|
||
|
||
async fn refresh_bot_open_id(&self) -> Result<String, ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
let response = self
|
||
.http_client
|
||
.get(format!("{}/bot/v3/info", FEISHU_API_BASE))
|
||
.bearer_auth(token)
|
||
.send()
|
||
.await
|
||
.map_err(|error| {
|
||
ChannelError::ConnectionError(format!("Bot info HTTP error: {error}"))
|
||
})?;
|
||
let status = response.status();
|
||
let body: serde_json::Value = response.json().await.map_err(|error| {
|
||
ChannelError::Other(format!("Failed to parse bot info response: {error}"))
|
||
})?;
|
||
if !status.is_success() || body.get("code").and_then(|value| value.as_i64()) != Some(0) {
|
||
return Err(ChannelError::Other(format!(
|
||
"Bot info request failed: status={status}, code={}",
|
||
body.get("code")
|
||
.and_then(|value| value.as_i64())
|
||
.unwrap_or(-1)
|
||
)));
|
||
}
|
||
let open_id = body
|
||
.pointer("/bot/open_id")
|
||
.or_else(|| body.pointer("/data/bot/open_id"))
|
||
.and_then(|value| value.as_str())
|
||
.map(str::trim)
|
||
.filter(|value| !value.is_empty())
|
||
.ok_or_else(|| ChannelError::Other("Bot info response has no open_id".to_string()))?
|
||
.to_string();
|
||
*self.bot_open_id.write().await = Some(open_id.clone());
|
||
Ok(open_id)
|
||
}
|
||
|
||
/// Get WebSocket endpoint URL from Feishu API
|
||
async fn get_ws_endpoint(
|
||
&self,
|
||
client: &reqwest::Client,
|
||
) -> Result<(String, WsClientConfig), ChannelError> {
|
||
let resp = client
|
||
.post(format!("{}/callback/ws/endpoint", FEISHU_WS_BASE))
|
||
.header("locale", "zh")
|
||
.json(&serde_json::json!({
|
||
"AppID": self.config.app_id,
|
||
"AppSecret": self.config.app_secret,
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| ChannelError::ConnectionError(format!("HTTP error: {}", e)))?;
|
||
|
||
let endpoint_resp: WsEndpointResp = resp.json().await.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Failed to parse endpoint response: {}", e))
|
||
})?;
|
||
|
||
if endpoint_resp.code != 0 {
|
||
return Err(ChannelError::ConnectionError(format!(
|
||
"WS endpoint failed: code={} msg={}",
|
||
endpoint_resp.code,
|
||
endpoint_resp.msg.as_deref().unwrap_or("unknown")
|
||
)));
|
||
}
|
||
|
||
let ep = endpoint_resp
|
||
.data
|
||
.ok_or_else(|| ChannelError::ConnectionError("Empty endpoint data".to_string()))?;
|
||
|
||
let client_config = ep.client_config.unwrap_or_default();
|
||
Ok((ep.url, client_config))
|
||
}
|
||
|
||
/// Get tenant access token (cached with proactive refresh).
|
||
async fn get_tenant_access_token(&self) -> Result<String, ChannelError> {
|
||
// 1. Check cache
|
||
{
|
||
let cached = self.tenant_token.read().await;
|
||
if let Some(ref token) = *cached
|
||
&& Instant::now() < token.refresh_after
|
||
{
|
||
return Ok(token.value.clone());
|
||
}
|
||
}
|
||
|
||
// 2. Fetch new token
|
||
let (token, ttl) = self.fetch_new_token().await?;
|
||
|
||
// 3. Cache with proactive refresh time (提前 120 秒)
|
||
let refresh_after = Instant::now() + ttl.saturating_sub(TOKEN_REFRESH_SKEW);
|
||
{
|
||
let mut cached = self.tenant_token.write().await;
|
||
*cached = Some(CachedTenantToken {
|
||
value: token.clone(),
|
||
refresh_after,
|
||
});
|
||
}
|
||
|
||
Ok(token)
|
||
}
|
||
|
||
/// Fetch a new tenant access token from Feishu.
|
||
async fn fetch_new_token(&self) -> Result<(String, Duration), ChannelError> {
|
||
let resp = self
|
||
.http_client
|
||
.post(format!(
|
||
"{}/auth/v3/tenant_access_token/internal",
|
||
FEISHU_API_BASE
|
||
))
|
||
.header("Content-Type", "application/json")
|
||
.json(&serde_json::json!({
|
||
"app_id": self.config.app_id,
|
||
"app_secret": self.config.app_secret,
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| ChannelError::ConnectionError(format!("HTTP error: {}", e)))?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct TokenResponse {
|
||
code: i32,
|
||
tenant_access_token: Option<String>,
|
||
expire: Option<i64>,
|
||
}
|
||
|
||
let token_resp: TokenResponse = resp
|
||
.json()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to parse token response: {}", e)))?;
|
||
|
||
if token_resp.code != 0 {
|
||
return Err(ChannelError::Other("Auth failed".to_string()));
|
||
}
|
||
|
||
let token = token_resp
|
||
.tenant_access_token
|
||
.ok_or_else(|| ChannelError::Other("No token in response".to_string()))?;
|
||
|
||
let ttl = token_resp
|
||
.expire
|
||
.and_then(|v| u64::try_from(v).ok())
|
||
.map(Duration::from_secs)
|
||
.unwrap_or(DEFAULT_TOKEN_TTL);
|
||
|
||
Ok((token, ttl))
|
||
}
|
||
|
||
/// Check if message_id has been seen (dedup), and mark it as seen if not.
|
||
/// Returns true if the message was already processed.
|
||
/// Note: GC of stale entries is handled in the heartbeat timeout_check loop.
|
||
async fn is_message_seen(&self, message_id: &str) -> bool {
|
||
let mut seen = self.seen_message_ids.write().await;
|
||
let now = Instant::now();
|
||
|
||
if seen.contains_key(message_id) {
|
||
true
|
||
} else {
|
||
seen.insert(message_id.to_string(), now);
|
||
false
|
||
}
|
||
}
|
||
|
||
/// Download media and save locally, return (description, media_item)
|
||
async fn download_media(
|
||
&self,
|
||
msg_type: &str,
|
||
content_json: &serde_json::Value,
|
||
message_id: &str,
|
||
) -> Result<(String, Option<MediaItem>), ChannelError> {
|
||
let media_dir = Path::new(&self.config.media_dir);
|
||
tokio::fs::create_dir_all(media_dir)
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to create media dir: {}", e)))?;
|
||
|
||
match msg_type {
|
||
"image" => {
|
||
self.download_image(content_json, message_id, media_dir)
|
||
.await
|
||
}
|
||
"audio" | "file" | "media" => {
|
||
self.download_file(content_json, message_id, media_dir, msg_type)
|
||
.await
|
||
}
|
||
_ => Ok((format!("[unsupported media type: {}]", msg_type), None)),
|
||
}
|
||
}
|
||
|
||
/// Download image from Feishu
|
||
async fn download_image(
|
||
&self,
|
||
content_json: &serde_json::Value,
|
||
message_id: &str,
|
||
media_dir: &Path,
|
||
) -> Result<(String, Option<MediaItem>), ChannelError> {
|
||
let image_key = content_json
|
||
.get("image_key")
|
||
.and_then(|v| v.as_str())
|
||
.ok_or_else(|| ChannelError::Other("No image_key in message".to_string()))?;
|
||
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
// Use message resource API for downloading message images
|
||
let url = format!(
|
||
"{}/im/v1/messages/{}/resources/{}?type=image",
|
||
FEISHU_API_BASE, message_id, image_key
|
||
);
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(url = %url, image_key = %image_key, message_id = %message_id, "Downloading image from Feishu via message resource API");
|
||
|
||
let resp = self
|
||
.http_client
|
||
.get(&url)
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Download image HTTP error: {}", e))
|
||
})?;
|
||
|
||
let status = resp.status();
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(status = %status, "Image download response status");
|
||
|
||
if !status.is_success() {
|
||
let error_text = resp.text().await.unwrap_or_default();
|
||
return Err(ChannelError::Other(format!(
|
||
"Image download failed {}: {}",
|
||
status, error_text
|
||
)));
|
||
}
|
||
|
||
let content_type = resp
|
||
.headers()
|
||
.get("content-type")
|
||
.and_then(|v| v.to_str().ok())
|
||
.unwrap_or("image/jpeg")
|
||
.to_string();
|
||
|
||
let ext = resolve_image_ext(&content_type);
|
||
|
||
let data = resp
|
||
.bytes()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read image data: {}", e)))?
|
||
.to_vec();
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(data_len = %data.len(), content_type = %content_type, "Downloaded image data");
|
||
|
||
let filename = format!(
|
||
"{}_{}.{}",
|
||
message_id,
|
||
&image_key[..8.min(image_key.len())],
|
||
ext
|
||
);
|
||
let file_path = resolve_unique_path(media_dir, &filename).await;
|
||
|
||
tokio::fs::write(&file_path, &data)
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to write image: {}", e)))?;
|
||
|
||
let media_item = MediaItem::new(file_path.to_string_lossy().to_string(), "image");
|
||
|
||
tracing::info!(message_id = %message_id, filename = %filename, "Downloaded image");
|
||
|
||
Ok((String::new(), Some(media_item)))
|
||
}
|
||
|
||
/// Download file/audio from Feishu
|
||
async fn download_file(
|
||
&self,
|
||
content_json: &serde_json::Value,
|
||
message_id: &str,
|
||
media_dir: &Path,
|
||
file_type: &str,
|
||
) -> Result<(String, Option<MediaItem>), ChannelError> {
|
||
let file_key = content_json
|
||
.get("file_key")
|
||
.and_then(|v| v.as_str())
|
||
.ok_or_else(|| ChannelError::Other("No file_key in message".to_string()))?;
|
||
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
// Use message resource API for downloading message files
|
||
let url = format!(
|
||
"{}/im/v1/messages/{}/resources/{}?type=file",
|
||
FEISHU_API_BASE, message_id, file_key
|
||
);
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(url = %url, file_key = %file_key, message_id = %message_id, "Downloading file from Feishu via message resource API");
|
||
|
||
let resp = self
|
||
.http_client
|
||
.get(&url)
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Download file HTTP error: {}", e))
|
||
})?;
|
||
|
||
let status = resp.status();
|
||
if !status.is_success() {
|
||
let error_text = resp.text().await.unwrap_or_default();
|
||
return Err(ChannelError::Other(format!(
|
||
"File download failed {}: {}",
|
||
status, error_text
|
||
)));
|
||
}
|
||
|
||
let data = resp
|
||
.bytes()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read file data: {}", e)))?
|
||
.to_vec();
|
||
|
||
let filename = content_json
|
||
.get("file_name")
|
||
.and_then(|v| v.as_str())
|
||
.map(|s| s.to_string())
|
||
.unwrap_or_else(|| {
|
||
let ext = resolve_file_ext(content_json);
|
||
if ext.is_empty() {
|
||
format!("{}_{}", message_id, &file_key[..8.min(file_key.len())])
|
||
} else {
|
||
format!(
|
||
"{}_{}.{}",
|
||
message_id,
|
||
&file_key[..8.min(file_key.len())],
|
||
ext
|
||
)
|
||
}
|
||
});
|
||
let file_path = resolve_unique_path(media_dir, &filename).await;
|
||
|
||
tokio::fs::write(&file_path, &data)
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to write file: {}", e)))?;
|
||
|
||
let media_item = MediaItem::new(file_path.to_string_lossy().to_string(), file_type);
|
||
|
||
tracing::info!(message_id = %message_id, filename = %filename, file_type = %file_type, "Downloaded file");
|
||
|
||
Ok((String::new(), Some(media_item)))
|
||
}
|
||
|
||
/// Upload image to Feishu and return the image_key
|
||
async fn upload_image(&self, file_path: &str) -> Result<String, ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let mime = mime_guess::from_path(file_path)
|
||
.first_or_octet_stream()
|
||
.to_string();
|
||
|
||
let file_name = std::path::Path::new(file_path)
|
||
.file_name()
|
||
.and_then(|n| n.to_str())
|
||
.unwrap_or("image.jpg");
|
||
|
||
let file_data = tokio::fs::read(file_path)
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read file: {}", e)))?;
|
||
|
||
let part = reqwest::multipart::Part::bytes(file_data)
|
||
.file_name(file_name.to_string())
|
||
.mime_str(&mime)
|
||
.map_err(|e| ChannelError::Other(format!("Invalid mime type: {}", e)))?;
|
||
|
||
let form = reqwest::multipart::Form::new()
|
||
.text("image_type", "message".to_string())
|
||
.part("image", part);
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!("{}/im/v1/images", FEISHU_API_BASE))
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.multipart(form)
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Upload image HTTP error: {}", e))
|
||
})?;
|
||
|
||
let status = resp.status();
|
||
let body_text = resp
|
||
.text()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read upload response: {}", e)))?;
|
||
tracing::debug!(status = %status, body = %body_text, "Feishu upload image");
|
||
|
||
#[derive(Deserialize)]
|
||
struct UploadResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
data: Option<UploadData>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct UploadData {
|
||
image_key: String,
|
||
}
|
||
|
||
let result: UploadResp = serde_json::from_str(&body_text).map_err(|e| {
|
||
ChannelError::Other(format!(
|
||
"Parse upload response error: {} | body: {}",
|
||
e, &body_text
|
||
))
|
||
})?;
|
||
|
||
if result.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Upload image failed: code={} msg={}",
|
||
result.code,
|
||
result.msg.as_deref().unwrap_or("unknown")
|
||
)));
|
||
}
|
||
|
||
result
|
||
.data
|
||
.map(|d| d.image_key)
|
||
.ok_or_else(|| ChannelError::Other("No image_key in response".to_string()))
|
||
}
|
||
|
||
/// Upload file to Feishu and return the file_key
|
||
async fn upload_file(&self, file_path: &str) -> Result<String, ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let file_name = std::path::Path::new(file_path)
|
||
.file_name()
|
||
.and_then(|n| n.to_str())
|
||
.unwrap_or("file.bin");
|
||
|
||
let extension = std::path::Path::new(file_path)
|
||
.extension()
|
||
.and_then(|e| e.to_str())
|
||
.unwrap_or("")
|
||
.to_lowercase();
|
||
|
||
let file_type = match extension.as_str() {
|
||
"opus" => "opus",
|
||
"mp4" | "mov" | "avi" | "mkv" => "mp4",
|
||
"pdf" => "pdf",
|
||
"doc" | "docx" => "doc",
|
||
"xls" | "xlsx" => "xls",
|
||
"ppt" | "pptx" => "ppt",
|
||
_ => "stream",
|
||
};
|
||
|
||
let file_data = tokio::fs::read(file_path)
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read file: {}", e)))?;
|
||
|
||
let part = reqwest::multipart::Part::bytes(file_data)
|
||
.file_name(file_name.to_string())
|
||
.mime_str("application/octet-stream")
|
||
.map_err(|e| ChannelError::Other(format!("Invalid mime type: {}", e)))?;
|
||
|
||
let form = reqwest::multipart::Form::new()
|
||
.text("file_type", file_type.to_string())
|
||
.text("file_name", file_name.to_string())
|
||
.part("file", part);
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!("{}/im/v1/files", FEISHU_API_BASE))
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.multipart(form)
|
||
.send()
|
||
.await
|
||
.map_err(|e| ChannelError::ConnectionError(format!("Upload file HTTP error: {}", e)))?;
|
||
|
||
let status = resp.status();
|
||
let body_text = resp
|
||
.text()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read upload response: {}", e)))?;
|
||
tracing::debug!(status = %status, body = %body_text, "Feishu upload file");
|
||
|
||
#[derive(Deserialize)]
|
||
struct UploadResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
data: Option<UploadData>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct UploadData {
|
||
file_key: String,
|
||
}
|
||
|
||
let result: UploadResp = serde_json::from_str(&body_text).map_err(|e| {
|
||
ChannelError::Other(format!(
|
||
"Parse upload response error: {} | body: {}",
|
||
e, &body_text
|
||
))
|
||
})?;
|
||
|
||
if result.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Upload file failed: code={} msg={}",
|
||
result.code,
|
||
result.msg.as_deref().unwrap_or("unknown")
|
||
)));
|
||
}
|
||
|
||
result
|
||
.data
|
||
.map(|d| d.file_key)
|
||
.ok_or_else(|| ChannelError::Other("No file_key in response".to_string()))
|
||
}
|
||
|
||
/// Add a reaction emoji to a message and store the reaction_id for later removal.
|
||
/// Returns the reaction_id if successful, None otherwise.
|
||
async fn add_reaction(&self, message_id: &str) -> Result<Option<String>, ChannelError> {
|
||
let emoji = self.config.reaction_emoji.as_str();
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!(
|
||
"{}/im/v1/messages/{}/reactions",
|
||
FEISHU_API_BASE, message_id
|
||
))
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.json(&serde_json::json!({
|
||
"reaction_type": { "emoji_type": emoji }
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Add reaction HTTP error: {}", e))
|
||
})?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct ReactionResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
data: Option<ReactionData>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct ReactionData {
|
||
reaction_id: Option<String>,
|
||
}
|
||
|
||
let result: ReactionResp = resp
|
||
.json()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Parse reaction response error: {}", e)))?;
|
||
|
||
if result.code != 0 {
|
||
tracing::warn!(
|
||
"Failed to add reaction to message {}: code={} msg={}",
|
||
message_id,
|
||
result.code,
|
||
result.msg.as_deref().unwrap_or("unknown")
|
||
);
|
||
return Ok(None);
|
||
}
|
||
|
||
let reaction_id = result.data.and_then(|d| d.reaction_id);
|
||
Ok(reaction_id)
|
||
}
|
||
|
||
/// Remove reaction using feishu metadata propagated through OutboundMessage.
|
||
/// Reads feishu.message_id and feishu.reaction_id from metadata.
|
||
async fn remove_reaction_from_metadata(
|
||
&self,
|
||
metadata: &std::collections::HashMap<String, String>,
|
||
) {
|
||
let (message_id, reaction_id) = match (
|
||
metadata.get("feishu.message_id"),
|
||
metadata.get("feishu.reaction_id"),
|
||
) {
|
||
(Some(msg_id), Some(rid)) => (msg_id.clone(), rid.clone()),
|
||
_ => return,
|
||
};
|
||
if let Err(e) = self.remove_reaction(&message_id, &reaction_id).await {
|
||
tracing::debug!(error = %e, message_id = %message_id, "Failed to remove reaction");
|
||
}
|
||
}
|
||
|
||
/// Remove a reaction emoji from a message.
|
||
async fn remove_reaction(
|
||
&self,
|
||
message_id: &str,
|
||
reaction_id: &str,
|
||
) -> Result<(), ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let resp = self
|
||
.http_client
|
||
.delete(format!(
|
||
"{}/im/v1/messages/{}/reactions/{}",
|
||
FEISHU_API_BASE, message_id, reaction_id
|
||
))
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Remove reaction HTTP error: {}", e))
|
||
})?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct ReactionResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
}
|
||
|
||
let result: ReactionResp = resp.json().await.map_err(|e| {
|
||
ChannelError::Other(format!("Parse remove reaction response error: {}", e))
|
||
})?;
|
||
|
||
if result.code != 0 {
|
||
tracing::debug!(
|
||
"Failed to remove reaction {} from message {}: code={} msg={}",
|
||
reaction_id,
|
||
message_id,
|
||
result.code,
|
||
result.msg.as_deref().unwrap_or("unknown")
|
||
);
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
const REPLY_CONTEXT_MAX_LEN: usize = 500;
|
||
|
||
/// Fetch the text content of a Feishu message by ID.
|
||
/// Returns a "[Reply to: ...]" context string, or None on failure.
|
||
async fn get_message_content(&self, message_id: &str) -> Option<String> {
|
||
let token = match self.get_tenant_access_token().await {
|
||
Ok(t) => t,
|
||
Err(e) => {
|
||
tracing::debug!(error = %e, message_id = %message_id, "Feishu: failed to get token for fetching parent message");
|
||
return None;
|
||
}
|
||
};
|
||
|
||
let resp = self
|
||
.http_client
|
||
.get(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id))
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.send()
|
||
.await
|
||
.ok()?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct MessageResp {
|
||
code: i32,
|
||
msg: Option<String>,
|
||
data: Option<MessageData>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct MessageData {
|
||
items: Option<Vec<MessageItem>>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct MessageItem {
|
||
msg_type: String,
|
||
body: Option<MessageBody>,
|
||
}
|
||
#[derive(Deserialize)]
|
||
struct MessageBody {
|
||
content: Option<String>,
|
||
}
|
||
|
||
let result: MessageResp = match resp.json().await {
|
||
Ok(r) => r,
|
||
Err(e) => {
|
||
tracing::debug!(error = %e, message_id = %message_id, "Feishu: failed to parse parent message response");
|
||
return None;
|
||
}
|
||
};
|
||
|
||
if result.code != 0 {
|
||
tracing::debug!(
|
||
message_id = %message_id,
|
||
code = %result.code,
|
||
msg = ?result.msg,
|
||
"Feishu: failed to fetch parent message"
|
||
);
|
||
return None;
|
||
}
|
||
|
||
let items = result.data?.items?;
|
||
let msg_obj = items.first()?;
|
||
|
||
let raw_content = msg_obj.body.as_ref()?.content.as_ref()?;
|
||
let msg_type = msg_obj.msg_type.as_str();
|
||
|
||
let text = match msg_type {
|
||
"text" => serde_json::from_str::<serde_json::Value>(raw_content)
|
||
.ok()?
|
||
.get("text")?
|
||
.as_str()?
|
||
.to_string(),
|
||
"post" => parse_post_content(raw_content),
|
||
_ => String::new(),
|
||
};
|
||
|
||
if text.is_empty() {
|
||
return None;
|
||
}
|
||
|
||
let text = if text.len() > Self::REPLY_CONTEXT_MAX_LEN {
|
||
format!("{}...", &text[..Self::REPLY_CONTEXT_MAX_LEN])
|
||
} else {
|
||
text
|
||
};
|
||
|
||
Some(format!("[Reply to: {}]", text))
|
||
}
|
||
|
||
/// Send a message to Feishu chat with specified message type and content.
|
||
/// Content is passed as-is (already a JSON string for file/media, or plain text for fallback).
|
||
async fn send_message_to_feishu(
|
||
&self,
|
||
receive_id: &str,
|
||
receive_id_type: &str,
|
||
msg_type: &str,
|
||
content: &str,
|
||
) -> Result<(), ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!(
|
||
"{}/im/v1/messages?receive_id_type={}",
|
||
FEISHU_API_BASE, receive_id_type
|
||
))
|
||
.header("Content-Type", "application/json")
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.json(&serde_json::json!({
|
||
"receive_id": receive_id,
|
||
"msg_type": msg_type,
|
||
"content": content
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Send message HTTP error: {}", e))
|
||
})?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct SendResp {
|
||
code: i32,
|
||
msg: String,
|
||
}
|
||
|
||
let send_resp: SendResp = resp
|
||
.json()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Parse send response error: {}", e)))?;
|
||
|
||
if send_resp.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Send message failed: code={} msg={}",
|
||
send_resp.code, send_resp.msg
|
||
)));
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Extract service_id from WebSocket URL query params
|
||
fn extract_service_id(url: &str) -> i32 {
|
||
url.split('?')
|
||
.nth(1)
|
||
.and_then(|qs| {
|
||
qs.split('&')
|
||
.find(|kv| kv.starts_with("service_id="))
|
||
.and_then(|kv| kv.split('=').nth(1))
|
||
.and_then(|v| v.parse::<i32>().ok())
|
||
})
|
||
.unwrap_or(0)
|
||
}
|
||
|
||
/// Handle incoming binary PbFrame - returns Some(ParsedMessage) if we need to ack
|
||
async fn handle_frame(&self, frame: &PbFrame) -> Result<Option<ParsedMessage>, ChannelError> {
|
||
// method 0 = CONTROL (ping/pong)
|
||
if frame.method == 0 {
|
||
return Ok(None);
|
||
}
|
||
|
||
// method 1 = DATA (events)
|
||
if frame.method != 1 {
|
||
return Ok(None);
|
||
}
|
||
|
||
let payload = frame
|
||
.payload
|
||
.as_deref()
|
||
.ok_or_else(|| ChannelError::Other("No payload in frame".to_string()))?;
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(payload_len = %payload.len(), "Received frame payload");
|
||
|
||
let event: LarkEvent = serde_json::from_slice(payload)
|
||
.map_err(|e| ChannelError::Other(format!("Parse event error: {}", e)))?;
|
||
|
||
let event_type = event.header.event_type.as_str();
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(event_type = %event_type, "Received event type");
|
||
if event_type != "im.message.receive_v1" {
|
||
return Ok(None);
|
||
}
|
||
|
||
let payload_data: MsgReceivePayload = serde_json::from_value(event.event.clone())
|
||
.map_err(|e| ChannelError::Other(format!("Parse payload error: {}", e)))?;
|
||
|
||
// Never let bot/app traffic trigger another model turn.
|
||
if matches!(payload_data.sender.sender_type.as_str(), "bot" | "app") {
|
||
return Ok(None);
|
||
}
|
||
|
||
let open_id = payload_data
|
||
.sender
|
||
.sender_id
|
||
.open_id
|
||
.ok_or_else(|| ChannelError::Other("No open_id".to_string()))?;
|
||
|
||
if !self.is_allowed(&open_id) {
|
||
tracing::warn!(sender = %open_id, "Rejected unauthorized Feishu sender");
|
||
return Ok(None);
|
||
}
|
||
|
||
let message_id = payload_data.message.message_id.clone();
|
||
let msg = payload_data.message;
|
||
if msg.chat_type == "group" && self.config.require_mention {
|
||
let bot_open_id = self.bot_open_id.read().await.clone();
|
||
if !message_mentions_bot(&msg, bot_open_id.as_deref()) {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(message_id = %message_id, "Ignoring group message without bot mention");
|
||
return Ok(None);
|
||
}
|
||
}
|
||
|
||
// Deduplicate only after admission so rejected traffic does not consume cache capacity.
|
||
if self.is_message_seen(&message_id).await {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(message_id = %message_id, "Duplicate message, skipping");
|
||
return Ok(None);
|
||
}
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(message_id = %message_id, "Received Feishu message");
|
||
|
||
let chat_id = msg.chat_id.clone();
|
||
let msg_type = msg.message_type.as_str();
|
||
let raw_content = msg.content.clone();
|
||
let parent_id = msg.parent_id.clone();
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(msg_type = %msg_type, chat_id = %chat_id, open_id = %open_id, "Parsing message content");
|
||
|
||
let (mut content, media) = self
|
||
.parse_and_download_message(msg_type, &raw_content, &message_id)
|
||
.await?;
|
||
content = normalize_mentions(
|
||
&content,
|
||
&msg.mentions,
|
||
self.bot_open_id.read().await.as_deref(),
|
||
);
|
||
|
||
// Fetch and prepend quoted message content if this is a reply
|
||
if let Some(ref pid) = parent_id
|
||
&& let Some(reply_ctx) = self.get_message_content(pid).await
|
||
{
|
||
content = format!("{}\n{}", reply_ctx, content);
|
||
}
|
||
|
||
#[cfg(debug_assertions)]
|
||
for m in &media {
|
||
tracing::debug!(media_type = %m.media_type, media_path = %m.path, "Media downloaded successfully");
|
||
}
|
||
|
||
Ok(Some(ParsedMessage {
|
||
message_id,
|
||
open_id,
|
||
chat_id,
|
||
content,
|
||
media,
|
||
parent_id,
|
||
}))
|
||
}
|
||
|
||
/// Parse message content and download media if needed
|
||
async fn parse_and_download_message(
|
||
&self,
|
||
msg_type: &str,
|
||
content: &str,
|
||
message_id: &str,
|
||
) -> Result<(String, Vec<MediaItem>), ChannelError> {
|
||
let (text, media) = match msg_type {
|
||
"text" => {
|
||
let text = if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
|
||
parsed
|
||
.get("text")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or(content)
|
||
.to_string()
|
||
} else {
|
||
content.to_string()
|
||
};
|
||
(text, Vec::new())
|
||
}
|
||
"post" => {
|
||
let text = parse_post_content(content);
|
||
let mut media = Vec::new();
|
||
|
||
for image_key in collect_post_image_keys(content) {
|
||
let content_json = serde_json::json!({ "image_key": image_key });
|
||
match self
|
||
.download_media("image", &content_json, message_id)
|
||
.await
|
||
{
|
||
Ok((_text, Some(item))) => media.push(item),
|
||
Ok((_text, None)) => {}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, "Failed to download image from Feishu post message");
|
||
}
|
||
}
|
||
}
|
||
|
||
(text, media)
|
||
}
|
||
"image" | "audio" | "file" | "media" => {
|
||
if let Ok(content_json) = serde_json::from_str::<serde_json::Value>(content) {
|
||
match self
|
||
.download_media(msg_type, &content_json, message_id)
|
||
.await
|
||
{
|
||
Ok((text, Some(media))) => (text, vec![media]),
|
||
Ok((text, None)) => (text, Vec::new()),
|
||
Err(_) => (format!("[{}: content unavailable]", msg_type), Vec::new()),
|
||
}
|
||
} else {
|
||
(format!("[{}: content unavailable]", msg_type), Vec::new())
|
||
}
|
||
}
|
||
"share_chat" => {
|
||
// Shared chat/cannel messages
|
||
if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
|
||
let chat_id = parsed
|
||
.get("chat_id")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("unknown");
|
||
(format!("[shared chat: {}]", chat_id), Vec::new())
|
||
} else {
|
||
("[shared chat]".to_string(), Vec::new())
|
||
}
|
||
}
|
||
"share_user" => {
|
||
// Shared user messages
|
||
if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
|
||
let user_id = parsed
|
||
.get("user_id")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("unknown");
|
||
(format!("[shared user: {}]", user_id), Vec::new())
|
||
} else {
|
||
("[shared user]".to_string(), Vec::new())
|
||
}
|
||
}
|
||
"interactive" => {
|
||
// Interactive card messages - extract text content
|
||
match extract_interactive_content(content) {
|
||
Ok((text, Some(media))) => (text, vec![media]),
|
||
Ok((text, None)) => (text, Vec::new()),
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, "Failed to extract interactive content");
|
||
(content.to_string(), Vec::new())
|
||
}
|
||
}
|
||
}
|
||
"list" => {
|
||
// List/bullet messages
|
||
match parse_list_content(content) {
|
||
Ok((text, Some(media))) => (text, vec![media]),
|
||
Ok((text, None)) => (text, Vec::new()),
|
||
Err(_) => (content.to_string(), Vec::new()),
|
||
}
|
||
}
|
||
"merge_forward" => ("[merged forward messages]".to_string(), Vec::new()),
|
||
"share_calendar_event" => {
|
||
if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) {
|
||
let event_key = parsed
|
||
.get("event_key")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("unknown");
|
||
(
|
||
format!("[shared calendar event: {}]", event_key),
|
||
Vec::new(),
|
||
)
|
||
} else {
|
||
("[shared calendar event]".to_string(), Vec::new())
|
||
}
|
||
}
|
||
"system" => ("[system message]".to_string(), Vec::new()),
|
||
_ => (content.to_string(), Vec::new()),
|
||
};
|
||
|
||
Ok((text, media))
|
||
}
|
||
|
||
/// Send acknowledgment for a message
|
||
async fn send_ack(
|
||
frame: &PbFrame,
|
||
write: &mut futures_util::stream::SplitSink<
|
||
tokio_tungstenite::WebSocketStream<
|
||
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
|
||
>,
|
||
tokio_tungstenite::tungstenite::Message,
|
||
>,
|
||
) -> Result<(), ChannelError> {
|
||
let mut ack = frame.clone();
|
||
ack.payload = Some(br#"{"code":200,"headers":{},"data":[]}"#.to_vec());
|
||
ack.headers.push(PbHeader {
|
||
key: "biz_rt".into(),
|
||
value: "0".into(),
|
||
});
|
||
write
|
||
.send(tokio_tungstenite::tungstenite::Message::Binary(
|
||
ack.encode_to_vec().into(),
|
||
))
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to send ack: {}", e)))?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn run_ws_loop(
|
||
&self,
|
||
bus: Arc<MessageBus>,
|
||
shutdown: CancellationToken,
|
||
) -> Result<(), ChannelError> {
|
||
let (wss_url, client_config) = tokio::select! {
|
||
result = self.get_ws_endpoint(&self.http_client) => result?,
|
||
_ = shutdown.cancelled() => return Ok(()),
|
||
};
|
||
|
||
let service_id = Self::extract_service_id(&wss_url);
|
||
tracing::info!(url = %wss_url, "Connecting to Feishu WebSocket");
|
||
|
||
let (ws_stream, _) = tokio::select! {
|
||
result = tokio_tungstenite::connect_async(&wss_url) => result.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("WebSocket connection failed: {}", e))
|
||
})?,
|
||
_ = shutdown.cancelled() => return Ok(()),
|
||
};
|
||
|
||
*self.connected.write().await = true;
|
||
tracing::info!("Feishu WebSocket connected");
|
||
|
||
let (mut write, mut read) = ws_stream.split();
|
||
|
||
// Send initial ping
|
||
let ping_frame = PbFrame {
|
||
seq_id: 1,
|
||
log_id: 0,
|
||
service: service_id,
|
||
method: 0,
|
||
headers: vec![PbHeader {
|
||
key: "type".into(),
|
||
value: "ping".into(),
|
||
}],
|
||
payload: None,
|
||
};
|
||
tokio::select! {
|
||
result = write.send(tokio_tungstenite::tungstenite::Message::Binary(
|
||
ping_frame.encode_to_vec().into(),
|
||
)) => result.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Failed to send initial ping: {}", e))
|
||
})?,
|
||
_ = shutdown.cancelled() => return Ok(()),
|
||
};
|
||
|
||
let ping_interval = client_config.ping_interval.unwrap_or(120).max(10);
|
||
let mut ping_interval_tok =
|
||
tokio::time::interval(tokio::time::Duration::from_secs(ping_interval));
|
||
let mut timeout_check = tokio::time::interval(tokio::time::Duration::from_secs(10));
|
||
let mut seq: u64 = 1;
|
||
let mut last_recv = Instant::now();
|
||
|
||
// Consume the immediate tick
|
||
ping_interval_tok.tick().await;
|
||
timeout_check.tick().await;
|
||
|
||
loop {
|
||
tokio::select! {
|
||
msg = read.next() => {
|
||
match msg {
|
||
Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(data))) => {
|
||
last_recv = Instant::now();
|
||
let bytes: Bytes = data;
|
||
if let Ok(frame) = PbFrame::decode(bytes.as_ref()) {
|
||
match self.handle_frame(&frame).await {
|
||
Ok(Some(parsed)) => {
|
||
// Send ACK immediately (Feishu requires within 3 s)
|
||
if let Err(e) = Self::send_ack(&frame, &mut write).await {
|
||
tracing::error!(error = %e, "Failed to send ACK to Feishu");
|
||
}
|
||
|
||
// Add reaction emoji (await so we get the reaction_id for later removal)
|
||
let message_id = parsed.message_id.clone();
|
||
let reaction_id = match self.add_reaction(&message_id).await {
|
||
Ok(Some(rid)) => Some(rid),
|
||
Ok(None) => None,
|
||
Err(e) => {
|
||
tracing::debug!(error = %e, message_id = %message_id, "Failed to add reaction");
|
||
None
|
||
}
|
||
};
|
||
|
||
let mut private_context = std::collections::HashMap::new();
|
||
private_context.insert("feishu.message_id".to_string(), message_id.clone());
|
||
if let Some(ref rid) = reaction_id {
|
||
private_context.insert("feishu.reaction_id".to_string(), rid.clone());
|
||
}
|
||
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, content_len = %parsed.content.len(), media_count = %parsed.media.len(), "Publishing message to bus");
|
||
let msg = crate::bus::InboundMessage {
|
||
channel: "feishu".to_string(),
|
||
sender_id: parsed.open_id.clone(),
|
||
chat_id: parsed.chat_id.clone(),
|
||
content: parsed.content.clone(),
|
||
received_at: crate::bus::message::current_timestamp(),
|
||
media: parsed.media.clone(),
|
||
channel_context: crate::bus::ChannelContext {
|
||
reply_to: parsed.parent_id.clone(),
|
||
private: private_context,
|
||
},
|
||
};
|
||
if let Err(e) = self.handle_and_publish(&bus, &msg).await {
|
||
tracing::error!(error = %e, open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Failed to publish Feishu message to bus");
|
||
} else {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!(open_id = %parsed.open_id, chat_id = %parsed.chat_id, "Message published to bus successfully");
|
||
}
|
||
}
|
||
Ok(None) => {}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, "Failed to parse Feishu frame");
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(data))) => {
|
||
last_recv = Instant::now();
|
||
let pong = PbFrame {
|
||
seq_id: seq.wrapping_add(1),
|
||
log_id: 0,
|
||
service: service_id,
|
||
method: 0,
|
||
headers: vec![PbHeader {
|
||
key: "type".into(),
|
||
value: "pong".into(),
|
||
}],
|
||
payload: Some(data.to_vec()),
|
||
};
|
||
let _ = write.send(tokio_tungstenite::tungstenite::Message::Binary(pong.encode_to_vec().into())).await;
|
||
}
|
||
Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(_))) => {
|
||
last_recv = Instant::now();
|
||
}
|
||
Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => {
|
||
#[cfg(debug_assertions)]
|
||
tracing::debug!("Feishu WebSocket closed");
|
||
break;
|
||
}
|
||
Some(Err(e)) => {
|
||
tracing::warn!(error = %e, "Feishu WebSocket error");
|
||
break;
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
_ = ping_interval_tok.tick() => {
|
||
seq = seq.wrapping_add(1);
|
||
let ping = PbFrame {
|
||
seq_id: seq,
|
||
log_id: 0,
|
||
service: service_id,
|
||
method: 0,
|
||
headers: vec![PbHeader {
|
||
key: "type".into(),
|
||
value: "ping".into(),
|
||
}],
|
||
payload: None,
|
||
};
|
||
if write.send(tokio_tungstenite::tungstenite::Message::Binary(ping.encode_to_vec().into())).await.is_err() {
|
||
tracing::warn!("Feishu ping failed, reconnecting");
|
||
break;
|
||
}
|
||
}
|
||
_ = timeout_check.tick() => {
|
||
if last_recv.elapsed() > WS_HEARTBEAT_TIMEOUT {
|
||
tracing::warn!("Feishu WebSocket heartbeat timeout, reconnecting");
|
||
break;
|
||
}
|
||
// GC dedup cache: remove entries older than TTL (matches zeroclaw pattern)
|
||
let now = Instant::now();
|
||
let mut seen = self.seen_message_ids.write().await;
|
||
seen.retain(|_, ts| now.duration_since(*ts) < DEDUP_CACHE_TTL);
|
||
}
|
||
_ = shutdown.cancelled() => {
|
||
tracing::info!("Feishu channel shutdown signal received");
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
*self.connected.write().await = false;
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
fn parse_post_content(content: &str) -> String {
|
||
/// Extract text from a single post element (text, link, at-mention).
|
||
fn extract_element(el: &serde_json::Value, out: &mut Vec<String>) {
|
||
match el.get("tag").and_then(|t| t.as_str()).unwrap_or("") {
|
||
"text" => {
|
||
if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
|
||
out.push(text.to_string());
|
||
}
|
||
}
|
||
"a" => {
|
||
let link_text = el
|
||
.get("text")
|
||
.and_then(|t| t.as_str())
|
||
.filter(|s| !s.is_empty())
|
||
.or_else(|| el.get("href").and_then(|h| h.as_str()))
|
||
.unwrap_or("");
|
||
out.push(link_text.to_string());
|
||
}
|
||
"at" => {
|
||
let name = el
|
||
.get("user_name")
|
||
.and_then(|n| n.as_str())
|
||
.or_else(|| el.get("user_id").and_then(|i| i.as_str()))
|
||
.unwrap_or("user");
|
||
out.push(format!("@{}", name));
|
||
}
|
||
"img" => {
|
||
out.push("[image]".to_string());
|
||
}
|
||
"code_block" => {
|
||
let lang = el.get("language").and_then(|l| l.as_str()).unwrap_or("");
|
||
let code_text = el.get("text").and_then(|t| t.as_str()).unwrap_or("");
|
||
out.push(format!("\n```{}\n{}\n```\n", lang, code_text));
|
||
}
|
||
_ => {
|
||
if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
|
||
out.push(text.to_string());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Parse a single block {title, content: [[...]]} and append text to out.
|
||
fn parse_block(block: &serde_json::Value, out: &mut Vec<String>) {
|
||
let title = block
|
||
.get("title")
|
||
.and_then(|t| t.as_str())
|
||
.filter(|s| !s.is_empty());
|
||
if let Some(t) = title {
|
||
out.push(t.to_string());
|
||
out.push("\n\n".to_string());
|
||
}
|
||
|
||
if let Some(content_arr) = block.get("content").and_then(|c| c.as_array()) {
|
||
for row in content_arr {
|
||
if let Some(row_arr) = row.as_array() {
|
||
for el in row_arr {
|
||
extract_element(el, out);
|
||
}
|
||
out.push("\n".to_string());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) else {
|
||
return content.to_string();
|
||
};
|
||
|
||
let mut texts = Vec::new();
|
||
|
||
// Unwrap optional {"post": ...} envelope (nanobot: root = root["post"])
|
||
let root = if parsed.get("post").and_then(|p| p.as_object()).is_some() {
|
||
parsed.get("post").unwrap()
|
||
} else {
|
||
&parsed
|
||
};
|
||
|
||
// Try direct format: {"title": ..., "content": [[...]]}
|
||
if root.get("content").and_then(|c| c.as_array()).is_some() {
|
||
parse_block(root, &mut texts);
|
||
let result = texts.join("");
|
||
if !result.trim().is_empty() {
|
||
return result.trim().to_string();
|
||
}
|
||
texts.clear();
|
||
}
|
||
|
||
// Try localized: {"zh_cn": {"title": ..., "content": [...]}}
|
||
for key in ["zh_cn", "en_us", "ja_jp"] {
|
||
if let Some(locale_data) = root.get(key).and_then(|l| l.as_object()) {
|
||
parse_block(&serde_json::json!(locale_data), &mut texts);
|
||
let result = texts.join("");
|
||
if !result.trim().is_empty() {
|
||
return result.trim().to_string();
|
||
}
|
||
texts.clear();
|
||
}
|
||
}
|
||
|
||
// Fall back: try any dict child
|
||
if let Some(root_obj) = root.as_object() {
|
||
for (_key, val) in root_obj {
|
||
if let Some(obj) = val.as_object()
|
||
&& obj.get("content").and_then(|c| c.as_array()).is_some()
|
||
{
|
||
parse_block(val, &mut texts);
|
||
let result = texts.join("");
|
||
if !result.trim().is_empty() {
|
||
return result.trim().to_string();
|
||
}
|
||
texts.clear();
|
||
}
|
||
}
|
||
}
|
||
|
||
content.to_string()
|
||
}
|
||
|
||
fn collect_post_image_keys(content: &str) -> Vec<String> {
|
||
fn visit(value: &serde_json::Value, keys: &mut Vec<String>) {
|
||
match value {
|
||
serde_json::Value::Object(map) => {
|
||
if let Some(image_key) = map.get("image_key").and_then(|v| v.as_str())
|
||
&& !keys.iter().any(|k| k == image_key)
|
||
{
|
||
keys.push(image_key.to_string());
|
||
}
|
||
|
||
for child in map.values() {
|
||
visit(child, keys);
|
||
}
|
||
}
|
||
serde_json::Value::Array(items) => {
|
||
for item in items {
|
||
visit(item, keys);
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
|
||
let Ok(parsed) = serde_json::from_str::<serde_json::Value>(content) else {
|
||
return Vec::new();
|
||
};
|
||
|
||
let mut keys = Vec::new();
|
||
visit(&parsed, &mut keys);
|
||
keys
|
||
}
|
||
|
||
/// Extract text content from interactive card messages
|
||
fn extract_interactive_content(content: &str) -> Result<(String, Option<MediaItem>), ChannelError> {
|
||
let parsed = match serde_json::from_str::<serde_json::Value>(content) {
|
||
Ok(p) => p,
|
||
Err(_) => return Ok((content.to_string(), None)),
|
||
};
|
||
|
||
let mut texts = Vec::new();
|
||
|
||
// Extract from elements array
|
||
if let Some(elements) = parsed.get("elements").and_then(|e| e.as_array()) {
|
||
for el in elements {
|
||
extract_element_content(el, &mut texts);
|
||
}
|
||
}
|
||
|
||
// Extract from card object
|
||
if let Some(card) = parsed.get("card").and_then(|c| c.as_object())
|
||
&& let Some(elements) = card.get("elements").and_then(|e| e.as_array())
|
||
{
|
||
for el in elements {
|
||
extract_element_content(el, &mut texts);
|
||
}
|
||
}
|
||
|
||
// Extract from header
|
||
if let Some(header) = parsed.get("header").and_then(|h| h.as_object())
|
||
&& let Some(title) = header.get("title").and_then(|t| t.as_object())
|
||
&& let Some(text) = title.get("content").and_then(|c| c.as_str())
|
||
{
|
||
texts.push(format!("title: {}\n", text));
|
||
}
|
||
|
||
let result = texts.join("").trim().to_string();
|
||
if result.is_empty() {
|
||
Ok((content.to_string(), None))
|
||
} else {
|
||
Ok((result, None))
|
||
}
|
||
}
|
||
|
||
/// Extract content from a single card element
|
||
fn extract_element_content(element: &serde_json::Value, texts: &mut Vec<String>) {
|
||
let tag = element.get("tag").and_then(|t| t.as_str()).unwrap_or("");
|
||
|
||
match tag {
|
||
"markdown" | "lark_md" => {
|
||
if let Some(content) = element.get("content").and_then(|c| c.as_str()) {
|
||
texts.push(content.to_string());
|
||
texts.push("\n".to_string());
|
||
}
|
||
}
|
||
"div" => {
|
||
if let Some(text_obj) = element.get("text").and_then(|t| t.as_object()) {
|
||
let content = text_obj
|
||
.get("content")
|
||
.and_then(|c| c.as_str())
|
||
.unwrap_or("");
|
||
texts.push(content.to_string());
|
||
} else if let Some(content) = element.get("text").and_then(|t| t.as_str()) {
|
||
texts.push(content.to_string());
|
||
}
|
||
texts.push("\n".to_string());
|
||
}
|
||
"a" => {
|
||
let href = element.get("href").and_then(|h| h.as_str()).unwrap_or("");
|
||
let text = element.get("text").and_then(|t| t.as_str()).unwrap_or("");
|
||
if !text.is_empty() {
|
||
texts.push(text.to_string());
|
||
} else if !href.is_empty() {
|
||
texts.push(format!("link: {}", href));
|
||
}
|
||
}
|
||
"img" => {
|
||
let alt = element.get("alt");
|
||
let alt_text = alt
|
||
.and_then(|a| a.as_str())
|
||
.or_else(|| {
|
||
alt.and_then(|a| a.as_object())
|
||
.and_then(|o| o.get("content"))
|
||
.and_then(|c| c.as_str())
|
||
})
|
||
.unwrap_or("[image]");
|
||
texts.push(format!("{}\n", alt_text));
|
||
}
|
||
"note" => {
|
||
if let Some(elements) = element.get("elements").and_then(|e| e.as_array()) {
|
||
for el in elements {
|
||
extract_element_content(el, texts);
|
||
}
|
||
}
|
||
}
|
||
"column_set" => {
|
||
if let Some(columns) = element.get("columns").and_then(|c| c.as_array()) {
|
||
for col in columns {
|
||
if let Some(elements) = col.get("elements").and_then(|e| e.as_array()) {
|
||
for el in elements {
|
||
extract_element_content(el, texts);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
"table" => {
|
||
// Tables are complex, just indicate presence
|
||
texts.push("[table]\n".to_string());
|
||
}
|
||
_ => {
|
||
// Recursively check for nested elements
|
||
if let Some(elements) = element.get("elements").and_then(|e| e.as_array()) {
|
||
for el in elements {
|
||
extract_element_content(el, texts);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Parse Feishu list/bullet message content into plain text
|
||
fn parse_list_content(content: &str) -> Result<(String, Option<MediaItem>), ChannelError> {
|
||
let parsed = match serde_json::from_str::<serde_json::Value>(content) {
|
||
Ok(p) => p,
|
||
Err(_) => return Ok((content.to_string(), None)),
|
||
};
|
||
|
||
let items = parsed
|
||
.get("items")
|
||
.and_then(|i| i.as_array())
|
||
.or_else(|| parsed.get("content").and_then(|c| c.as_array()));
|
||
|
||
let Some(items) = items else {
|
||
return Ok((content.to_string(), None));
|
||
};
|
||
|
||
let mut lines = Vec::new();
|
||
collect_list_items(items, &mut lines, 0);
|
||
|
||
let result = lines.join("\n").trim().to_string();
|
||
if result.is_empty() {
|
||
Ok((content.to_string(), None))
|
||
} else {
|
||
Ok((result, None))
|
||
}
|
||
}
|
||
|
||
/// Recursively collect list item text with indentation
|
||
fn collect_list_items(items: &[serde_json::Value], lines: &mut Vec<String>, depth: usize) {
|
||
let indent = " ".repeat(depth);
|
||
|
||
for item in items {
|
||
// Items can be arrays of inline elements or objects with content/children
|
||
let inline_elements = if let Some(arr) = item.as_array() {
|
||
arr.as_slice()
|
||
} else if let Some(obj) = item.as_object() {
|
||
obj.get("content")
|
||
.and_then(|c| c.as_array())
|
||
.map(|a| a.as_slice())
|
||
.unwrap_or(&[])
|
||
} else {
|
||
continue;
|
||
};
|
||
|
||
let mut text = String::new();
|
||
for el in inline_elements {
|
||
extract_inline_text(el, &mut text);
|
||
}
|
||
|
||
let trimmed = text.trim();
|
||
if !trimmed.is_empty() {
|
||
lines.push(format!("{}- {}", indent, trimmed));
|
||
}
|
||
|
||
// Handle nested children
|
||
if let Some(obj) = item.as_object() {
|
||
if let Some(children) = obj.get("children").and_then(|c| c.as_array()) {
|
||
collect_list_items(children, lines, depth + 1);
|
||
}
|
||
} else if let Some(children_arr) = item.as_array().and_then(|arr| {
|
||
arr.iter()
|
||
.find(|child| child.as_object().and_then(|o| o.get("children")).is_some())
|
||
}) && let Some(children) = children_arr
|
||
.as_object()
|
||
.and_then(|o| o.get("children"))
|
||
.and_then(|c| c.as_array())
|
||
{
|
||
collect_list_items(children, lines, depth + 1);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Extract text from inline elements (text, link, at-mention)
|
||
fn extract_inline_text(el: &serde_json::Value, out: &mut String) {
|
||
match el.get("tag").and_then(|t| t.as_str()).unwrap_or("") {
|
||
"text" => {
|
||
if let Some(text) = el.get("text").and_then(|t| t.as_str()) {
|
||
out.push_str(text);
|
||
}
|
||
}
|
||
"a" => {
|
||
let text = el
|
||
.get("text")
|
||
.and_then(|t| t.as_str())
|
||
.filter(|s| !s.is_empty())
|
||
.or_else(|| el.get("href").and_then(|h| h.as_str()))
|
||
.unwrap_or("");
|
||
out.push_str(text);
|
||
}
|
||
"at" => {
|
||
let name = el
|
||
.get("user_name")
|
||
.and_then(|n| n.as_str())
|
||
.or_else(|| el.get("user_id").and_then(|i| i.as_str()))
|
||
.unwrap_or("user");
|
||
out.push_str(&format!("@{}", name));
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
|
||
fn mention_open_id(mention: &serde_json::Value) -> Option<&str> {
|
||
mention
|
||
.pointer("/id/open_id")
|
||
.or_else(|| mention.get("open_id"))
|
||
.and_then(|value| value.as_str())
|
||
}
|
||
|
||
fn message_mentions_bot(message: &LarkMessage, bot_open_id: Option<&str>) -> bool {
|
||
let Some(bot_open_id) = bot_open_id.filter(|value| !value.is_empty()) else {
|
||
return false;
|
||
};
|
||
if message
|
||
.mentions
|
||
.iter()
|
||
.any(|mention| mention_open_id(mention) == Some(bot_open_id))
|
||
{
|
||
return true;
|
||
}
|
||
|
||
fn post_contains_bot(value: &serde_json::Value, bot_open_id: &str) -> bool {
|
||
match value {
|
||
serde_json::Value::Object(map) => {
|
||
let is_bot_mention = map.get("tag").and_then(|value| value.as_str()) == Some("at")
|
||
&& map
|
||
.get("user_id")
|
||
.or_else(|| map.get("open_id"))
|
||
.and_then(|value| value.as_str())
|
||
== Some(bot_open_id);
|
||
is_bot_mention
|
||
|| map
|
||
.values()
|
||
.any(|value| post_contains_bot(value, bot_open_id))
|
||
}
|
||
serde_json::Value::Array(values) => values
|
||
.iter()
|
||
.any(|value| post_contains_bot(value, bot_open_id)),
|
||
_ => false,
|
||
}
|
||
}
|
||
|
||
serde_json::from_str::<serde_json::Value>(&message.content)
|
||
.is_ok_and(|content| post_contains_bot(&content, bot_open_id))
|
||
}
|
||
|
||
/// Remove the bot's own placeholder while preserving human mentions as readable names.
|
||
fn normalize_mentions(
|
||
text: &str,
|
||
mentions: &[serde_json::Value],
|
||
bot_open_id: Option<&str>,
|
||
) -> String {
|
||
let mut normalized = text.to_string();
|
||
for mention in mentions {
|
||
let Some(key) = mention.get("key").and_then(|value| value.as_str()) else {
|
||
continue;
|
||
};
|
||
let replacement = if mention_open_id(mention) == bot_open_id {
|
||
String::new()
|
||
} else {
|
||
mention
|
||
.get("name")
|
||
.and_then(|value| value.as_str())
|
||
.filter(|value| !value.is_empty())
|
||
.map(|name| format!("@{name}"))
|
||
.unwrap_or_else(|| key.to_string())
|
||
};
|
||
normalized = normalized.replace(key, &replacement);
|
||
}
|
||
normalized.trim().to_string()
|
||
}
|
||
|
||
fn resolve_image_ext(content_type: &str) -> &str {
|
||
match content_type {
|
||
"image/png" => "png",
|
||
"image/gif" => "gif",
|
||
"image/webp" => "webp",
|
||
"image/bmp" => "bmp",
|
||
_ => "jpg",
|
||
}
|
||
}
|
||
|
||
fn resolve_file_ext(content_json: &serde_json::Value) -> String {
|
||
if let Some(name) = content_json.get("file_name").and_then(|v| v.as_str())
|
||
&& let Some(ext) = std::path::Path::new(name)
|
||
.extension()
|
||
.and_then(|e| e.to_str())
|
||
{
|
||
return ext.to_string();
|
||
}
|
||
String::new()
|
||
}
|
||
|
||
async fn resolve_unique_path(dir: &Path, filename: &str) -> std::path::PathBuf {
|
||
let candidate = dir.join(filename);
|
||
if !tokio::fs::try_exists(&candidate).await.unwrap_or(false) {
|
||
return candidate;
|
||
}
|
||
let stem = std::path::Path::new(filename)
|
||
.file_stem()
|
||
.and_then(|s| s.to_str())
|
||
.unwrap_or(filename);
|
||
let ext = std::path::Path::new(filename)
|
||
.extension()
|
||
.and_then(|s| s.to_str())
|
||
.unwrap_or("");
|
||
let mut n = 1;
|
||
loop {
|
||
let candidate = if ext.is_empty() {
|
||
dir.join(format!("{}({})", stem, n))
|
||
} else {
|
||
dir.join(format!("{}({}).{}", stem, n, ext))
|
||
};
|
||
if !tokio::fs::try_exists(&candidate).await.unwrap_or(false) {
|
||
return candidate;
|
||
}
|
||
n += 1;
|
||
}
|
||
}
|
||
|
||
impl FeishuChannel {
|
||
/// Build a Card JSON 2.0 interactive card with a single markdown element.
|
||
fn build_card_content(markdown: &str) -> String {
|
||
serde_json::json!({
|
||
"schema": "2.0",
|
||
"body": {
|
||
"elements": [{
|
||
"tag": "markdown",
|
||
"content": markdown
|
||
}]
|
||
}
|
||
})
|
||
.to_string()
|
||
}
|
||
|
||
/// Max byte-size for markdown content in a single card.
|
||
/// Card payloads have a ~30 KB limit; leave margin for JSON envelope.
|
||
const CARD_MARKDOWN_MAX_BYTES: usize = 28_000;
|
||
|
||
/// Split markdown content into chunks that fit within the card size limit.
|
||
/// Splits on line boundaries to avoid breaking markdown syntax.
|
||
fn split_markdown_chunks(text: &str) -> Vec<String> {
|
||
if text.len() <= Self::CARD_MARKDOWN_MAX_BYTES {
|
||
return vec![text.to_string()];
|
||
}
|
||
|
||
let mut chunks: Vec<String> = Vec::new();
|
||
let mut start = 0;
|
||
|
||
while start < text.len() {
|
||
if start + Self::CARD_MARKDOWN_MAX_BYTES >= text.len() {
|
||
chunks.push(text[start..].to_string());
|
||
break;
|
||
}
|
||
|
||
let end = text.floor_char_boundary(start + Self::CARD_MARKDOWN_MAX_BYTES);
|
||
let search_region = &text[start..end];
|
||
let split_at = search_region
|
||
.rfind('\n')
|
||
.map(|pos| start + pos + 1)
|
||
.unwrap_or(end);
|
||
|
||
let split_at = if text.is_char_boundary(split_at) {
|
||
split_at
|
||
} else {
|
||
(start..split_at)
|
||
.rev()
|
||
.find(|&i| text.is_char_boundary(i))
|
||
.unwrap_or(start)
|
||
};
|
||
|
||
if split_at <= start {
|
||
let forced = (end..=text.len())
|
||
.find(|&i| text.is_char_boundary(i))
|
||
.unwrap_or(text.len());
|
||
chunks.push(text[start..forced].to_string());
|
||
start = forced;
|
||
} else {
|
||
chunks.push(text[start..split_at].to_string());
|
||
start = split_at;
|
||
}
|
||
}
|
||
|
||
chunks
|
||
}
|
||
|
||
/// Send an interactive card message to Feishu.
|
||
async fn send_interactive_card(
|
||
&self,
|
||
receive_id: &str,
|
||
receive_id_type: &str,
|
||
card_content: &str,
|
||
) -> Result<String, ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!(
|
||
"{}/im/v1/messages?receive_id_type={}",
|
||
FEISHU_API_BASE, receive_id_type
|
||
))
|
||
.header("Content-Type", "application/json")
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.json(&serde_json::json!({
|
||
"receive_id": receive_id,
|
||
"msg_type": "interactive",
|
||
"content": card_content
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Send interactive card HTTP error: {}", e))
|
||
})?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct SendResp {
|
||
code: i32,
|
||
msg: String,
|
||
data: Option<SendData>,
|
||
}
|
||
|
||
#[derive(Deserialize)]
|
||
struct SendData {
|
||
message_id: String,
|
||
}
|
||
|
||
let send_resp: SendResp = resp.json().await.map_err(|e| {
|
||
ChannelError::Other(format!("Parse send interactive card response error: {}", e))
|
||
})?;
|
||
|
||
if send_resp.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Send interactive card failed: code={} msg={}",
|
||
send_resp.code, send_resp.msg
|
||
)));
|
||
}
|
||
|
||
send_resp
|
||
.data
|
||
.map(|data| data.message_id)
|
||
.filter(|message_id| !message_id.is_empty())
|
||
.ok_or_else(|| ChannelError::Other("Feishu send response has no message_id".into()))
|
||
}
|
||
|
||
async fn update_interactive_card(
|
||
&self,
|
||
message_id: &str,
|
||
card_content: &str,
|
||
) -> Result<(), ChannelError> {
|
||
let token = self.get_tenant_access_token().await?;
|
||
let card: serde_json::Value = serde_json::from_str(card_content)
|
||
.map_err(|error| ChannelError::Other(format!("Invalid card JSON: {error}")))?;
|
||
let response = self
|
||
.http_client
|
||
.patch(format!("{}/im/v1/messages/{}", FEISHU_API_BASE, message_id))
|
||
.header("Content-Type", "application/json")
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.json(&serde_json::json!({ "card": card }))
|
||
.send()
|
||
.await
|
||
.map_err(|error| {
|
||
ChannelError::ConnectionError(format!("Update card HTTP error: {error}"))
|
||
})?;
|
||
|
||
#[derive(Deserialize)]
|
||
struct UpdateResp {
|
||
code: i32,
|
||
msg: String,
|
||
}
|
||
let result: UpdateResp = response.json().await.map_err(|error| {
|
||
ChannelError::Other(format!("Parse update card response error: {error}"))
|
||
})?;
|
||
if result.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Update card failed: code={} msg={}",
|
||
result.code, result.msg
|
||
)));
|
||
}
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
trait FeishuTurnApi: Send {
|
||
async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError>;
|
||
async fn update_card(&mut self, message_id: &str, markdown: &str) -> Result<(), ChannelError>;
|
||
async fn cleanup(&mut self);
|
||
}
|
||
|
||
struct FeishuTurnBackend {
|
||
channel: FeishuChannel,
|
||
receive_id: String,
|
||
receive_id_type: &'static str,
|
||
metadata: HashMap<String, String>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl FeishuTurnApi for FeishuTurnBackend {
|
||
async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError> {
|
||
let card = FeishuChannel::build_card_content(markdown);
|
||
self.channel
|
||
.send_interactive_card(&self.receive_id, self.receive_id_type, &card)
|
||
.await
|
||
}
|
||
|
||
async fn update_card(&mut self, message_id: &str, markdown: &str) -> Result<(), ChannelError> {
|
||
let card = FeishuChannel::build_card_content(markdown);
|
||
self.channel
|
||
.update_interactive_card(message_id, &card)
|
||
.await
|
||
}
|
||
|
||
async fn cleanup(&mut self) {
|
||
self.channel
|
||
.remove_reaction_from_metadata(&self.metadata)
|
||
.await;
|
||
}
|
||
}
|
||
|
||
struct FeishuTurnSink {
|
||
api: Box<dyn FeishuTurnApi>,
|
||
message_id: Option<String>,
|
||
cleaned_up: bool,
|
||
}
|
||
|
||
impl FeishuTurnSink {
|
||
fn new(api: Box<dyn FeishuTurnApi>) -> Self {
|
||
Self {
|
||
api,
|
||
message_id: None,
|
||
cleaned_up: false,
|
||
}
|
||
}
|
||
|
||
async fn cleanup(&mut self) {
|
||
if !self.cleaned_up {
|
||
self.api.cleanup().await;
|
||
self.cleaned_up = true;
|
||
}
|
||
}
|
||
|
||
async fn send_chunks(&mut self, chunks: &[String]) -> Result<(), ChannelError> {
|
||
for chunk in chunks {
|
||
self.api.create_card(chunk).await?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
async fn finish_snapshot(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
|
||
let markdown = render_feishu_turn(snapshot);
|
||
let chunks = if markdown.is_empty() {
|
||
Vec::new()
|
||
} else {
|
||
FeishuChannel::split_markdown_chunks(&markdown)
|
||
};
|
||
|
||
let result = if chunks.is_empty() {
|
||
Ok(())
|
||
} else if let Some(message_id) = self.message_id.clone() {
|
||
match self.api.update_card(&message_id, &chunks[0]).await {
|
||
Ok(()) => self.send_chunks(&chunks[1..]).await,
|
||
Err(error) => {
|
||
tracing::warn!(error = %error, "Final Feishu card update failed; sending complete fallback");
|
||
self.send_chunks(&chunks).await
|
||
}
|
||
}
|
||
} else {
|
||
self.send_chunks(&chunks).await
|
||
};
|
||
self.cleanup().await;
|
||
result
|
||
}
|
||
}
|
||
|
||
#[async_trait]
|
||
impl TurnSink for FeishuTurnSink {
|
||
async fn update(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
|
||
let markdown = render_feishu_turn(snapshot);
|
||
if markdown.is_empty() {
|
||
return Ok(());
|
||
}
|
||
let live_markdown = truncate_feishu_live_markdown(&markdown);
|
||
if let Some(message_id) = self.message_id.clone() {
|
||
self.api.update_card(&message_id, &live_markdown).await
|
||
} else {
|
||
self.message_id = Some(self.api.create_card(&live_markdown).await?);
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
async fn finish(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
|
||
self.finish_snapshot(snapshot).await
|
||
}
|
||
|
||
async fn abort(&mut self, snapshot: &TurnSnapshot) -> Result<(), ChannelError> {
|
||
self.finish_snapshot(snapshot).await
|
||
}
|
||
}
|
||
|
||
fn render_feishu_turn(snapshot: &TurnSnapshot) -> String {
|
||
let mut sections = Vec::new();
|
||
for block in &snapshot.blocks {
|
||
match block {
|
||
TurnBlock::Reasoning { text, .. } if !text.trim().is_empty() => {
|
||
sections.push(format!(
|
||
"> **思考过程**\n> {}",
|
||
text.trim().replace('\n', "\n> ")
|
||
));
|
||
}
|
||
TurnBlock::Assistant { text, .. } if !text.trim().is_empty() => {
|
||
sections.push(text.trim().to_string());
|
||
}
|
||
TurnBlock::Tool {
|
||
name,
|
||
status,
|
||
preview,
|
||
..
|
||
} => {
|
||
let status = match status {
|
||
ToolStatus::Running => "执行中",
|
||
ToolStatus::Completed => "已完成",
|
||
ToolStatus::Failed => "失败",
|
||
};
|
||
let mut section = format!("> 🔧 **{name}** · {status}");
|
||
if let Some(preview) = preview.as_deref().filter(|value| !value.trim().is_empty()) {
|
||
section.push_str("\n> ");
|
||
section.push_str(&preview.trim().replace('\n', "\n> "));
|
||
}
|
||
sections.push(section);
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
|
||
if sections.is_empty() {
|
||
if snapshot.status == TurnStatus::Failed {
|
||
sections.push(format!(
|
||
"⚠️ 回复失败：{}",
|
||
snapshot.error.as_deref().unwrap_or("未知错误")
|
||
));
|
||
} else if snapshot.status == TurnStatus::Cancelled {
|
||
sections.push("已停止生成。".to_string());
|
||
} else {
|
||
return String::new();
|
||
}
|
||
}
|
||
|
||
let status = match snapshot.status {
|
||
TurnStatus::Running => Some(match snapshot.phase {
|
||
crate::session::TurnPhase::Queued => "排队中",
|
||
crate::session::TurnPhase::Reasoning => "思考中",
|
||
crate::session::TurnPhase::Responding => "生成中",
|
||
crate::session::TurnPhase::Acting => "调用工具中",
|
||
crate::session::TurnPhase::Finalizing => "收尾中",
|
||
}),
|
||
TurnStatus::Cancelled => Some("已停止"),
|
||
TurnStatus::Failed => Some("失败"),
|
||
TurnStatus::Completed => None,
|
||
};
|
||
if let Some(status) = status {
|
||
sections.push(format!("_{status}_"));
|
||
}
|
||
sections.join("\n\n")
|
||
}
|
||
|
||
fn truncate_feishu_live_markdown(markdown: &str) -> String {
|
||
if markdown.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES {
|
||
return markdown.to_string();
|
||
}
|
||
const SUFFIX: &str = "\n\n_内容仍在生成，已暂时截断…_";
|
||
let limit = FeishuChannel::CARD_MARKDOWN_MAX_BYTES.saturating_sub(SUFFIX.len());
|
||
let boundary = markdown.floor_char_boundary(limit);
|
||
format!("{}{SUFFIX}", &markdown[..boundary])
|
||
}
|
||
|
||
#[async_trait]
|
||
impl Channel for FeishuChannel {
|
||
fn name(&self) -> &str {
|
||
"feishu"
|
||
}
|
||
|
||
fn is_allowed(&self, sender_id: &str) -> bool {
|
||
self.config.allow_from.iter().any(|allowed| {
|
||
let allowed = allowed.trim();
|
||
allowed == "*" || allowed == sender_id
|
||
})
|
||
}
|
||
|
||
/// Handle an inbound message: check for slash commands first, then publish to bus
|
||
async fn handle_and_publish(
|
||
&self,
|
||
bus: &Arc<MessageBus>,
|
||
msg: &crate::bus::InboundMessage,
|
||
) -> Result<(), ChannelError> {
|
||
// All messages (including slash commands) go through the normal inbound flow
|
||
// SessionManager handles session creation/reuse internally
|
||
bus.publish_inbound(msg.clone()).await?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn start(&self, bus: Arc<MessageBus>) -> Result<(), ChannelError> {
|
||
if self.config.app_id.is_empty() || self.config.app_secret.is_empty() {
|
||
return Err(ChannelError::ConfigError(
|
||
"Feishu app_id or app_secret is not configured".to_string(),
|
||
));
|
||
}
|
||
|
||
if self.config.require_mention {
|
||
match self.refresh_bot_open_id().await {
|
||
Ok(open_id) => {
|
||
tracing::info!(bot_open_id = %open_id, "Resolved Feishu bot identity")
|
||
}
|
||
Err(error) => tracing::warn!(
|
||
error = %error,
|
||
"Failed to resolve Feishu bot identity; group messages will be ignored"
|
||
),
|
||
}
|
||
}
|
||
|
||
let mut run_task = self.run_task.lock().await;
|
||
if run_task.as_ref().is_some_and(|task| !task.is_finished()) {
|
||
return Ok(());
|
||
}
|
||
if let Some(finished) = run_task.take() {
|
||
let _ = finished.await;
|
||
}
|
||
|
||
*self.running.write().await = true;
|
||
|
||
let shutdown = CancellationToken::new();
|
||
*self.shutdown.write().await = Some(shutdown.clone());
|
||
|
||
let channel = self.clone();
|
||
let bus = bus.clone();
|
||
*run_task = Some(tokio::spawn(async move {
|
||
let mut consecutive_failures = 0;
|
||
let max_failures = 3;
|
||
|
||
loop {
|
||
if !*channel.running.read().await {
|
||
break;
|
||
}
|
||
|
||
match channel.run_ws_loop(bus.clone(), shutdown.clone()).await {
|
||
Ok(_) => {
|
||
tracing::info!("Feishu WebSocket disconnected");
|
||
}
|
||
Err(e) => {
|
||
consecutive_failures += 1;
|
||
tracing::error!(attempt = consecutive_failures, error = %e, "Feishu WebSocket error");
|
||
if consecutive_failures >= max_failures {
|
||
tracing::error!("Feishu channel: max failures reached, stopping");
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
if !*channel.running.read().await || shutdown.is_cancelled() {
|
||
break;
|
||
}
|
||
|
||
tracing::info!("Feishu channel retrying in 5s...");
|
||
tokio::select! {
|
||
_ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => {}
|
||
_ = shutdown.cancelled() => break,
|
||
}
|
||
}
|
||
|
||
*channel.running.write().await = false;
|
||
tracing::info!("Feishu channel stopped");
|
||
}));
|
||
|
||
tracing::info!("Feishu channel started");
|
||
Ok(())
|
||
}
|
||
|
||
async fn stop(&self) -> Result<(), ChannelError> {
|
||
*self.running.write().await = false;
|
||
*self.connected.write().await = false;
|
||
|
||
if let Some(shutdown) = self.shutdown.write().await.take() {
|
||
shutdown.cancel();
|
||
}
|
||
|
||
let task = { self.run_task.lock().await.take() };
|
||
if let Some(mut task) = task {
|
||
match tokio::time::timeout(CHANNEL_STOP_GRACE, &mut task).await {
|
||
Ok(result) => result.map_err(|error| {
|
||
ChannelError::Other(format!("Feishu channel task failed to join: {error}"))
|
||
})?,
|
||
Err(_) => {
|
||
tracing::warn!(
|
||
grace_ms = CHANNEL_STOP_GRACE.as_millis(),
|
||
"Feishu channel did not stop in time; aborting connection task"
|
||
);
|
||
task.abort();
|
||
let _ = task.await;
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
fn is_running(&self) -> bool {
|
||
self.running.try_read().map(|r| *r).unwrap_or(false)
|
||
}
|
||
|
||
fn live_policy(&self) -> LivePolicy {
|
||
if self.config.live_updates {
|
||
LivePolicy::Snapshot {
|
||
min_interval: Duration::from_millis(
|
||
self.config.live_update_interval_ms.clamp(250, 5_000),
|
||
),
|
||
}
|
||
} else {
|
||
LivePolicy::FinalOnly
|
||
}
|
||
}
|
||
|
||
fn presentation_policy(&self) -> crate::delivery::PresentationPolicy {
|
||
crate::delivery::PresentationPolicy::external(self.config.live_updates)
|
||
}
|
||
|
||
async fn open_turn(&self, target: TurnTarget) -> Result<Box<dyn TurnSink>, ChannelError> {
|
||
let (receive_id, receive_id_type) = if target.chat_id.starts_with("oc_") {
|
||
(target.chat_id, "chat_id")
|
||
} else {
|
||
(target.reply_to.unwrap_or(target.chat_id), "open_id")
|
||
};
|
||
Ok(Box::new(FeishuTurnSink::new(Box::new(FeishuTurnBackend {
|
||
channel: self.clone(),
|
||
receive_id,
|
||
receive_id_type,
|
||
metadata: target.metadata,
|
||
}))))
|
||
}
|
||
|
||
async fn send(&self, msg: OutboundMessage) -> Result<(), ChannelError> {
|
||
let receive_id = if msg.chat_id.starts_with("oc_") {
|
||
&msg.chat_id
|
||
} else {
|
||
msg.reply_to.as_ref().unwrap_or(&msg.chat_id)
|
||
};
|
||
let receive_id_type = if msg.chat_id.starts_with("oc_") {
|
||
"chat_id"
|
||
} else {
|
||
"open_id"
|
||
};
|
||
|
||
// If no media, send as interactive card with raw markdown
|
||
if msg.media.is_empty() {
|
||
let content = msg.content.trim();
|
||
|
||
// Empty content
|
||
if content.is_empty() {
|
||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||
return Ok(());
|
||
}
|
||
|
||
let chunks = Self::split_markdown_chunks(content);
|
||
for chunk in &chunks {
|
||
let card = Self::build_card_content(chunk);
|
||
if let Err(e) = self
|
||
.send_interactive_card(receive_id, receive_id_type, &card)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, "Failed to send interactive card, falling back to text");
|
||
let text_content = serde_json::json!({ "text": chunk }).to_string();
|
||
let result = self
|
||
.send_message_to_feishu(receive_id, receive_id_type, "text", &text_content)
|
||
.await;
|
||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||
return result;
|
||
}
|
||
}
|
||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||
return Ok(());
|
||
}
|
||
|
||
// Handle multimodal message - send with media
|
||
let token = self.get_tenant_access_token().await?;
|
||
|
||
// Separate images (can embed in post) from files (sent as separate messages)
|
||
let mut image_items = Vec::new();
|
||
let mut file_items = Vec::new();
|
||
for media_item in &msg.media {
|
||
match media_item.media_type.as_str() {
|
||
"image" => image_items.push(media_item),
|
||
"audio" | "video" | "file" => file_items.push(media_item),
|
||
_ => {
|
||
tracing::warn!(media_type = %media_item.media_type, "Unsupported media type for sending");
|
||
}
|
||
}
|
||
}
|
||
|
||
// Upload and send files as separate messages (one per file)
|
||
for item in &file_items {
|
||
match self.upload_file(&item.path).await {
|
||
Ok(file_key) => {
|
||
let file_msg_type = match item.media_type.as_str() {
|
||
"audio" => "audio",
|
||
"video" => "media",
|
||
_ => "file",
|
||
};
|
||
let file_content = serde_json::json!({"file_key": file_key}).to_string();
|
||
if let Err(e) = self
|
||
.send_message_to_feishu(
|
||
receive_id,
|
||
receive_id_type,
|
||
file_msg_type,
|
||
&file_content,
|
||
)
|
||
.await
|
||
{
|
||
tracing::warn!(error = %e, msg_type = file_msg_type, "Failed to send file message");
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, path = %item.path, "Failed to upload file");
|
||
}
|
||
}
|
||
}
|
||
|
||
// Build content parts for post (text + images)
|
||
let mut content_parts = Vec::new();
|
||
|
||
if !msg.content.is_empty() {
|
||
const MAX_TEXT_LENGTH: usize = 60_000;
|
||
let truncated_text = if msg.content.len() > MAX_TEXT_LENGTH {
|
||
format!(
|
||
"{}...\n\n[Content truncated due to length limit]",
|
||
&msg.content[..msg.content.ceil_char_boundary(MAX_TEXT_LENGTH)]
|
||
)
|
||
} else {
|
||
msg.content.clone()
|
||
};
|
||
content_parts.push(serde_json::json!({
|
||
"tag": "text",
|
||
"text": truncated_text
|
||
}));
|
||
}
|
||
|
||
for item in &image_items {
|
||
match self.upload_image(&item.path).await {
|
||
Ok(image_key) => {
|
||
content_parts.push(serde_json::json!({
|
||
"tag": "img",
|
||
"image_key": image_key
|
||
}));
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(error = %e, path = %item.path, "Failed to upload image");
|
||
}
|
||
}
|
||
}
|
||
|
||
// If no post content after processing (no text, no images), skip
|
||
if content_parts.is_empty() {
|
||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||
return Ok(());
|
||
}
|
||
|
||
// Determine message type and build content
|
||
let msg_type = if msg.content.is_empty() && image_items.len() == 1 {
|
||
"image"
|
||
} else {
|
||
"post"
|
||
};
|
||
|
||
let content = if msg_type == "image" {
|
||
// Image-only: content is just {"image_key": "..."}
|
||
let image_key = content_parts[0]["image_key"].as_str().unwrap_or("");
|
||
serde_json::json!({"image_key": image_key}).to_string()
|
||
} else {
|
||
// Post with media: zh_cn wrapped post structure
|
||
let post_content: Vec<Vec<serde_json::Value>> =
|
||
content_parts.into_iter().map(|part| vec![part]).collect();
|
||
serde_json::json!({
|
||
"zh_cn": {
|
||
"title": "",
|
||
"content": post_content
|
||
}
|
||
})
|
||
.to_string()
|
||
};
|
||
|
||
let resp = self
|
||
.http_client
|
||
.post(format!(
|
||
"{}/im/v1/messages?receive_id_type={}",
|
||
FEISHU_API_BASE, receive_id_type
|
||
))
|
||
.header("Content-Type", "application/json")
|
||
.header("Authorization", format!("Bearer {}", token))
|
||
.json(&serde_json::json!({
|
||
"receive_id": receive_id,
|
||
"msg_type": msg_type,
|
||
"content": content
|
||
}))
|
||
.send()
|
||
.await
|
||
.map_err(|e| {
|
||
ChannelError::ConnectionError(format!("Send multimodal message HTTP error: {}", e))
|
||
})?;
|
||
|
||
let send_status = resp.status();
|
||
let send_body = resp
|
||
.text()
|
||
.await
|
||
.map_err(|e| ChannelError::Other(format!("Failed to read send response: {}", e)))?;
|
||
tracing::debug!(status = %send_status, body = %send_body, msg_type = %msg_type, "Feishu send message");
|
||
|
||
#[derive(Deserialize)]
|
||
struct SendResp {
|
||
code: i32,
|
||
msg: String,
|
||
}
|
||
|
||
let send_resp: SendResp = serde_json::from_str(&send_body).map_err(|e| {
|
||
ChannelError::Other(format!(
|
||
"Parse send response error: {} | body: {}",
|
||
e, &send_body
|
||
))
|
||
})?;
|
||
|
||
if send_resp.code != 0 {
|
||
return Err(ChannelError::Other(format!(
|
||
"Send multimodal message failed: code={} msg={}",
|
||
send_resp.code, send_resp.msg
|
||
)));
|
||
}
|
||
|
||
// Remove pending reaction after successfully sending
|
||
self.remove_reaction_from_metadata(&msg.metadata).await;
|
||
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use crate::agent::TurnEvent;
|
||
use crate::delivery::{PresentationPolicy, project_snapshot};
|
||
|
||
#[derive(Default)]
|
||
struct MockTurnState {
|
||
created: Vec<String>,
|
||
updated: Vec<(String, String)>,
|
||
cleanups: usize,
|
||
fail_updates: usize,
|
||
}
|
||
|
||
struct MockTurnApi {
|
||
state: Arc<Mutex<MockTurnState>>,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl FeishuTurnApi for MockTurnApi {
|
||
async fn create_card(&mut self, markdown: &str) -> Result<String, ChannelError> {
|
||
let mut state = self.state.lock().await;
|
||
state.created.push(markdown.to_string());
|
||
Ok(format!("card-{}", state.created.len()))
|
||
}
|
||
|
||
async fn update_card(
|
||
&mut self,
|
||
message_id: &str,
|
||
markdown: &str,
|
||
) -> Result<(), ChannelError> {
|
||
let mut state = self.state.lock().await;
|
||
if state.fail_updates > 0 {
|
||
state.fail_updates -= 1;
|
||
return Err(ChannelError::Other("card can no longer be edited".into()));
|
||
}
|
||
state
|
||
.updated
|
||
.push((message_id.to_string(), markdown.to_string()));
|
||
Ok(())
|
||
}
|
||
|
||
async fn cleanup(&mut self) {
|
||
self.state.lock().await.cleanups += 1;
|
||
}
|
||
}
|
||
|
||
fn mock_sink(state: Arc<Mutex<MockTurnState>>) -> FeishuTurnSink {
|
||
FeishuTurnSink::new(Box::new(MockTurnApi { state }))
|
||
}
|
||
|
||
fn test_channel() -> FeishuChannel {
|
||
FeishuChannel::new(
|
||
FeishuChannelConfig {
|
||
enabled: true,
|
||
app_id: "test-app".to_string(),
|
||
app_secret: "test-secret".to_string(),
|
||
allow_from: vec!["*".to_string()],
|
||
require_mention: true,
|
||
agent: String::new(),
|
||
media_dir: String::new(),
|
||
reaction_emoji: "THUMBSUP".to_string(),
|
||
live_updates: false,
|
||
live_update_interval_ms: 500,
|
||
},
|
||
Path::new("/tmp"),
|
||
)
|
||
.expect("test channel should be valid")
|
||
}
|
||
|
||
fn inbound_frame(
|
||
message_id: &str,
|
||
sender_id: &str,
|
||
chat_type: &str,
|
||
text: &str,
|
||
mentions: serde_json::Value,
|
||
) -> PbFrame {
|
||
PbFrame {
|
||
seq_id: 1,
|
||
log_id: 1,
|
||
service: 1,
|
||
method: 1,
|
||
headers: vec![],
|
||
payload: Some(
|
||
serde_json::json!({
|
||
"header": {
|
||
"event_type": "im.message.receive_v1",
|
||
"event_id": format!("event-{message_id}")
|
||
},
|
||
"event": {
|
||
"sender": {
|
||
"sender_id": { "open_id": sender_id },
|
||
"sender_type": "user"
|
||
},
|
||
"message": {
|
||
"message_id": message_id,
|
||
"chat_id": "oc_test",
|
||
"chat_type": chat_type,
|
||
"message_type": "text",
|
||
"content": serde_json::json!({ "text": text }).to_string(),
|
||
"mentions": mentions
|
||
}
|
||
}
|
||
})
|
||
.to_string()
|
||
.into_bytes(),
|
||
),
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn inbound_admission_enforces_allowlist_before_parsing() {
|
||
let mut channel = test_channel();
|
||
channel.config.allow_from = vec!["ou_allowed".to_string()];
|
||
let frame = inbound_frame(
|
||
"om_denied",
|
||
"ou_denied",
|
||
"p2p",
|
||
"hello",
|
||
serde_json::json!([]),
|
||
);
|
||
|
||
assert!(channel.handle_frame(&frame).await.unwrap().is_none());
|
||
assert!(channel.seen_message_ids.read().await.is_empty());
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn group_message_requires_bot_mention_and_removes_only_self_mention() {
|
||
let channel = test_channel();
|
||
*channel.bot_open_id.write().await = Some("ou_bot".to_string());
|
||
|
||
let ignored = inbound_frame(
|
||
"om_ignored",
|
||
"ou_user",
|
||
"group",
|
||
"hello",
|
||
serde_json::json!([]),
|
||
);
|
||
assert!(channel.handle_frame(&ignored).await.unwrap().is_none());
|
||
|
||
let admitted = inbound_frame(
|
||
"om_admitted",
|
||
"ou_user",
|
||
"group",
|
||
"@_user_1 ask @_user_2",
|
||
serde_json::json!([
|
||
{"key": "@_user_1", "id": {"open_id": "ou_bot"}, "name": "PicoBot"},
|
||
{"key": "@_user_2", "id": {"open_id": "ou_peer"}, "name": "Alice"}
|
||
]),
|
||
);
|
||
let parsed = channel
|
||
.handle_frame(&admitted)
|
||
.await
|
||
.unwrap()
|
||
.expect("mentioned group message should be admitted");
|
||
assert_eq!(parsed.content, "ask @Alice");
|
||
}
|
||
|
||
#[test]
|
||
fn post_mentions_can_gate_group_messages_when_top_level_mentions_are_absent() {
|
||
let message = LarkMessage {
|
||
message_id: "om_post".to_string(),
|
||
chat_id: "oc_test".to_string(),
|
||
chat_type: "group".to_string(),
|
||
message_type: "post".to_string(),
|
||
content: serde_json::json!({
|
||
"zh_cn": {"content": [[{"tag": "at", "user_id": "ou_bot"}]]}
|
||
})
|
||
.to_string(),
|
||
parent_id: None,
|
||
mentions: vec![],
|
||
};
|
||
|
||
assert!(message_mentions_bot(&message, Some("ou_bot")));
|
||
assert!(!message_mentions_bot(&message, Some("ou_other")));
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn turn_sink_creates_once_updates_same_card_and_cleans_up_at_finish() {
|
||
let state = Arc::new(Mutex::new(MockTurnState::default()));
|
||
let mut sink = mock_sink(state.clone());
|
||
let (controller, emitter, _) =
|
||
crate::session::TurnController::start("feishu:chat:dialog", "message");
|
||
emitter
|
||
.emit(TurnEvent::TextDelta {
|
||
iteration: 0,
|
||
delta: "hello".into(),
|
||
})
|
||
.unwrap();
|
||
sink.update(&project_snapshot(
|
||
&controller.snapshot(),
|
||
PresentationPolicy::external(true),
|
||
))
|
||
.await
|
||
.unwrap();
|
||
emitter
|
||
.emit(TurnEvent::TextDelta {
|
||
iteration: 0,
|
||
delta: " world".into(),
|
||
})
|
||
.unwrap();
|
||
sink.update(&project_snapshot(
|
||
&controller.snapshot(),
|
||
PresentationPolicy::external(true),
|
||
))
|
||
.await
|
||
.unwrap();
|
||
controller.complete(None);
|
||
sink.finish(&project_snapshot(
|
||
&controller.snapshot(),
|
||
PresentationPolicy::external(true),
|
||
))
|
||
.await
|
||
.unwrap();
|
||
|
||
let state = state.lock().await;
|
||
assert_eq!(state.created.len(), 1);
|
||
assert_eq!(state.updated.len(), 2);
|
||
assert!(state.updated.iter().all(|(id, _)| id == "card-1"));
|
||
assert!(state.updated.last().unwrap().1.contains("hello world"));
|
||
assert_eq!(state.cleanups, 1);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn final_update_failure_sends_complete_fallback_and_cleanup_is_idempotent() {
|
||
let state = Arc::new(Mutex::new(MockTurnState::default()));
|
||
let mut sink = mock_sink(state.clone());
|
||
let (controller, emitter, _) =
|
||
crate::session::TurnController::start("feishu:chat:dialog", "message");
|
||
emitter
|
||
.emit(TurnEvent::TextDelta {
|
||
iteration: 0,
|
||
delta: "partial".into(),
|
||
})
|
||
.unwrap();
|
||
sink.update(&controller.snapshot()).await.unwrap();
|
||
state.lock().await.fail_updates = 1;
|
||
emitter
|
||
.emit(TurnEvent::TextDelta {
|
||
iteration: 0,
|
||
delta: " final".into(),
|
||
})
|
||
.unwrap();
|
||
controller.complete(None);
|
||
sink.finish(&controller.snapshot()).await.unwrap();
|
||
sink.finish(&controller.snapshot()).await.unwrap();
|
||
|
||
let state = state.lock().await;
|
||
assert_eq!(state.created.len(), 2);
|
||
assert!(state.created[1].contains("partial final"));
|
||
assert_eq!(state.cleanups, 1);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn final_only_sink_sends_no_fragments_and_abort_without_text_is_visible() {
|
||
let state = Arc::new(Mutex::new(MockTurnState::default()));
|
||
let mut sink = mock_sink(state.clone());
|
||
let (controller, _emitter, _) =
|
||
crate::session::TurnController::start("feishu:chat:dialog", "message");
|
||
controller.fail("provider unavailable");
|
||
|
||
sink.abort(&controller.snapshot()).await.unwrap();
|
||
|
||
let state = state.lock().await;
|
||
assert_eq!(state.created.len(), 1);
|
||
assert!(state.created[0].contains("provider unavailable"));
|
||
assert_eq!(state.updated.len(), 0);
|
||
assert_eq!(state.cleanups, 1);
|
||
}
|
||
|
||
#[test]
|
||
fn external_projection_removes_reasoning_before_feishu_rendering() {
|
||
let (controller, emitter, _) =
|
||
crate::session::TurnController::start("feishu:chat:dialog", "message");
|
||
emitter
|
||
.emit(TurnEvent::ReasoningDelta {
|
||
iteration: 0,
|
||
delta: "private".into(),
|
||
})
|
||
.unwrap();
|
||
emitter
|
||
.emit(TurnEvent::TextDelta {
|
||
iteration: 0,
|
||
delta: "public".into(),
|
||
})
|
||
.unwrap();
|
||
let projected =
|
||
project_snapshot(&controller.snapshot(), PresentationPolicy::external(true));
|
||
|
||
let markdown = render_feishu_turn(&projected);
|
||
assert!(markdown.contains("public"));
|
||
assert!(!markdown.contains("private"));
|
||
}
|
||
|
||
#[test]
|
||
fn live_card_truncation_preserves_utf8_and_payload_limit() {
|
||
let markdown = "你".repeat(FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
|
||
let truncated = truncate_feishu_live_markdown(&markdown);
|
||
|
||
assert!(truncated.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
|
||
assert!(truncated.ends_with("_内容仍在生成，已暂时截断…_"));
|
||
}
|
||
|
||
#[test]
|
||
fn final_card_chunking_preserves_long_utf8_content() {
|
||
let markdown = "你".repeat(FeishuChannel::CARD_MARKDOWN_MAX_BYTES);
|
||
let chunks = FeishuChannel::split_markdown_chunks(&markdown);
|
||
|
||
assert!(chunks.len() > 1);
|
||
assert!(
|
||
chunks
|
||
.iter()
|
||
.all(|chunk| chunk.len() <= FeishuChannel::CARD_MARKDOWN_MAX_BYTES)
|
||
);
|
||
assert_eq!(chunks.concat(), markdown);
|
||
}
|
||
|
||
#[test]
|
||
fn live_policy_uses_configured_bounded_interval() {
|
||
let mut channel = test_channel();
|
||
assert_eq!(channel.live_policy(), LivePolicy::FinalOnly);
|
||
channel.config.live_updates = true;
|
||
channel.config.live_update_interval_ms = 10;
|
||
|
||
assert_eq!(
|
||
channel.live_policy(),
|
||
LivePolicy::Snapshot {
|
||
min_interval: Duration::from_millis(250)
|
||
}
|
||
);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn stop_aborts_connection_task_that_ignores_cancellation() {
|
||
let channel = test_channel();
|
||
let shutdown = CancellationToken::new();
|
||
|
||
*channel.running.write().await = true;
|
||
*channel.connected.write().await = true;
|
||
*channel.shutdown.write().await = Some(shutdown.clone());
|
||
*channel.run_task.lock().await = Some(tokio::spawn(std::future::pending()));
|
||
|
||
tokio::time::timeout(Duration::from_secs(1), channel.stop())
|
||
.await
|
||
.expect("stop must have a hard deadline")
|
||
.expect("stop should succeed after aborting the stuck task");
|
||
|
||
assert!(shutdown.is_cancelled());
|
||
assert!(!channel.is_running());
|
||
assert!(!*channel.connected.read().await);
|
||
assert!(channel.run_task.lock().await.is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn collect_post_image_keys_finds_nested_images() {
|
||
let content = serde_json::json!({
|
||
"zh_cn": {
|
||
"title": "",
|
||
"content": [[
|
||
{"tag": "img", "image_key": "img_v3_001"},
|
||
{"tag": "text", "text": "这是哪里？"},
|
||
{"tag": "img", "image_key": "img_v3_002"},
|
||
{"tag": "img", "image_key": "img_v3_001"}
|
||
]]
|
||
}
|
||
})
|
||
.to_string();
|
||
|
||
assert_eq!(
|
||
collect_post_image_keys(&content),
|
||
vec!["img_v3_001".to_string(), "img_v3_002".to_string()]
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn parse_post_content_preserves_image_positions() {
|
||
let content = serde_json::json!({
|
||
"zh_cn": {
|
||
"title": "",
|
||
"content": [[
|
||
{"tag": "text", "text": "这是一张图："},
|
||
{"tag": "img", "image_key": "img_v3_001"},
|
||
{"tag": "text", "text": "看完继续说"}
|
||
]]
|
||
}
|
||
})
|
||
.to_string();
|
||
|
||
assert_eq!(
|
||
parse_post_content(&content),
|
||
"这是一张图：[image]看完继续说"
|
||
);
|
||
}
|
||
}
|