mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
54e3f7f3fa
stopWebSocket() only nullified the wsClient reference without calling disconnect() on the SDK client. This left the old WebSocket connection's pingLoop thread and ExecutorService running, leaking file descriptors and threads on each reconnect. Over time, accumulated leaks prevented new connections from being established, causing the Feishu channel to silently stop receiving messages. Fix: use reflection to access the SDK's protected `conn` field and call close(1000) on the OkHttp WebSocket, triggering the SDK's onClosed → disconnect() cleanup chain. Note: oapi-sdk 2.7.1 adds a public close() method that would make this reflection unnecessary. Consider upgrading as a follow-up. Closes #220
2592 lines
122 KiB
Java
2592 lines
122 KiB
Java
package vip.mate.channel.feishu;
|
||
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.lark.oapi.event.EventDispatcher;
|
||
import com.lark.oapi.service.im.ImService;
|
||
import com.lark.oapi.service.im.v1.model.P2MessageReceiveV1;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import reactor.core.publisher.Flux;
|
||
import vip.mate.agent.AgentService.StreamDelta;
|
||
import vip.mate.channel.AbstractChannelAdapter;
|
||
import vip.mate.channel.ChannelMessage;
|
||
import vip.mate.channel.ChannelMessageRouter;
|
||
import vip.mate.channel.ExponentialBackoff;
|
||
import vip.mate.channel.StreamingChannelAdapter;
|
||
import vip.mate.channel.media.GeneratedFileScrubber;
|
||
import vip.mate.channel.media.MediaSource;
|
||
import vip.mate.channel.media.MediaUploadException;
|
||
import vip.mate.channel.media.MediaUploadRequest;
|
||
import vip.mate.channel.media.MediaUploadResult;
|
||
import vip.mate.channel.model.ChannelEntity;
|
||
import vip.mate.workspace.conversation.model.MessageContentPart;
|
||
|
||
import java.io.InputStream;
|
||
import java.net.URI;
|
||
import java.net.http.HttpClient;
|
||
import java.net.http.HttpRequest;
|
||
import java.net.http.HttpResponse;
|
||
import java.nio.file.Files;
|
||
import java.nio.file.Path;
|
||
import java.nio.file.StandardCopyOption;
|
||
import java.time.Duration;
|
||
import java.time.LocalDateTime;
|
||
import java.util.ArrayList;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
import java.util.Set;
|
||
import java.util.concurrent.ConcurrentHashMap;
|
||
import java.util.concurrent.ScheduledFuture;
|
||
import java.util.concurrent.TimeUnit;
|
||
|
||
/**
|
||
* 飞书渠道适配器
|
||
* <p>
|
||
* 飞书渠道实现：
|
||
* - 接入模式：WebSocket 长连接（默认，无需公网 IP）或 Event Subscription（HTTP 回调）
|
||
* - 发送方式：通过 Open API 发送消息
|
||
* - 消息去重：基于 message_id 防止重复处理
|
||
* - 引用消息：自动拉取 parent_id 对应的父消息内容并注入上下文
|
||
* <p>
|
||
* 配置项（configJson）：
|
||
* - app_id: 飞书应用 App ID
|
||
* - app_secret: 飞书应用 App Secret
|
||
* - connection_mode: 接入模式 "websocket"（默认）或 "webhook"
|
||
* - domain: "feishu"（默认）或 "lark"（国际版）
|
||
* - encrypt_key: 事件加密密钥（webhook 模式必填）
|
||
* - verification_token: 事件验证 Token（webhook 模式可选）
|
||
* - enable_reaction: 是否在收到消息后添加表情反应（默认 true）
|
||
* - enable_nickname_cache: 是否通过 Contact API 获取用户昵称（默认 true）
|
||
* - media_download_enabled: 是否下载消息中的媒体文件（默认 false）
|
||
* - enable_quoted_context: 是否拉取被引用消息内容注入到 prompt（默认 true）
|
||
* - silent_disconnect_threshold_seconds: WebSocket 静默断连阈值（默认 1800，0 禁用）
|
||
* - stale_event_threshold_seconds: 过滤旧事件阈值（默认 30，0 禁用）
|
||
* - card_format: 卡片格式化模式 "auto"（默认）| "always" | "never"
|
||
* auto: 根据内容自动检测；always: 全部包卡片；never: 全部纯文本（降级/调试用）
|
||
* - card_header: Markdown 卡片 header 文案，默认 "AI 助手"；设为空串可隐藏 header
|
||
* - require_mention: 群聊中是否需要 @机器人 才响应（默认 false）
|
||
* true: 仅当消息中 @了机器人才处理；通过飞书 mentions 字段精确判断，无需配置 botPrefix
|
||
*
|
||
* @author MateClaw Team
|
||
*/
|
||
@Slf4j
|
||
public class FeishuChannelAdapter extends AbstractChannelAdapter implements StreamingChannelAdapter {
|
||
|
||
public static final String CHANNEL_TYPE = "feishu";
|
||
|
||
private HttpClient httpClient;
|
||
private String tenantAccessToken;
|
||
private long tokenExpireTime;
|
||
|
||
/** 定时 Token 刷新任务 */
|
||
private ScheduledFuture<?> tokenRefreshFuture;
|
||
|
||
/** 消息去重：最近处理过的 message_id */
|
||
private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();
|
||
|
||
/** 昵称缓存：open_id → 显示名称 */
|
||
private final ConcurrentHashMap<String, String> nicknameCache = new ConcurrentHashMap<>();
|
||
private static final int NICKNAME_CACHE_MAX = 500;
|
||
|
||
/** WebSocket 客户端（websocket 模式） */
|
||
private volatile com.lark.oapi.ws.Client wsClient;
|
||
|
||
/** WebSocket 连接线程 */
|
||
private volatile Thread wsThread;
|
||
|
||
/** WebSocket 静默断连看门狗：定期检查最近事件时间，超过阈值就强制重连 */
|
||
private ScheduledFuture<?> silentDisconnectWatchdog;
|
||
|
||
/** 是否已收到至少一个事件（用于避免新连接立即触发静默超时） */
|
||
private volatile boolean hasReceivedFirstEvent = false;
|
||
|
||
/** 看门狗检查间隔（秒） */
|
||
private static final long WATCHDOG_INTERVAL_SECONDS = 60L;
|
||
|
||
/** 静默断连默认阈值（秒）：30 分钟无事件就视为可疑 */
|
||
private static final long DEFAULT_SILENT_THRESHOLD_SECONDS = 1800L;
|
||
|
||
/** 旧事件过滤默认阈值（秒）：超过 30 秒的事件视为重连后回放 */
|
||
private static final long DEFAULT_STALE_THRESHOLD_SECONDS = 30L;
|
||
|
||
/** Bot's own open_id, fetched once from /open-apis/bot/v3/info and cached. */
|
||
private volatile String botOpenId;
|
||
|
||
/** Serializes lazy bot-open-id fetches so concurrent group messages share one API roundtrip. */
|
||
private final Object botOpenIdLock = new Object();
|
||
|
||
/**
|
||
* Last failure timestamp for {@code /open-apis/bot/v3/info}. While the call is in the
|
||
* {@link #BOT_OPENID_FAILURE_BACKOFF_MS} negative-cache window, {@link #getBotOpenId()}
|
||
* returns {@code null} fast so a Feishu outage doesn't trigger one synchronous retry
|
||
* per inbound group message.
|
||
*/
|
||
private volatile long botOpenIdLastFailureMs = 0L;
|
||
|
||
/** Negative-cache window for {@link #getBotOpenId()} failures (60 s). */
|
||
private static final long BOT_OPENID_FAILURE_BACKOFF_MS = 60_000L;
|
||
|
||
/** SDK-backed uploader for image / file / audio / video parts. Nullable for legacy callers. */
|
||
private final FeishuMediaUploader mediaUploader;
|
||
|
||
/** Scrubs {@code /api/v1/files/generated/{id}} URLs into native attachments. Nullable for legacy callers. */
|
||
private final GeneratedFileScrubber generatedFileScrubber;
|
||
|
||
/** CardKit streaming-card manager. Nullable for legacy callers / tests. */
|
||
private final FeishuStreamingCardManager streamingCardManager;
|
||
|
||
/** Interactive-card dispatcher (approval cards etc.). Nullable for legacy callers / tests. */
|
||
private final vip.mate.channel.feishu.cards.FeishuCardDispatcher cardDispatcher;
|
||
|
||
/**
|
||
* Per-channel SDK {@code Client} factory — drives inbound resource
|
||
* downloads via {@code client.im().v1().messageResource().get(...)}
|
||
* instead of the legacy hand-rolled {@code HttpClient} path. Nullable
|
||
* for legacy callers / tests, in which case the adapter falls back
|
||
* to the raw HTTP path (still works, just bypasses retries / token
|
||
* refresh / domain switching that the SDK handles).
|
||
*/
|
||
private final FeishuClientFactory clientFactory;
|
||
|
||
/**
|
||
* Process-local cache that exposes downloaded inbound media via
|
||
* {@code /api/v1/files/generated/{id}} so the conversation memory
|
||
* and admin UI can render the file without holding a tenant token.
|
||
* Nullable for legacy callers / tests.
|
||
*/
|
||
private final vip.mate.tool.document.GeneratedFileCache generatedFileCache;
|
||
|
||
/**
|
||
* STT service for transcribing inbound voice messages. Feishu, unlike
|
||
* WeCom / DingTalk, does NOT include ASR text in the webhook payload —
|
||
* its inbound audio carries only a {@code file_key}. So we download the
|
||
* bytes (via {@link #downloadResource}) and run them through
|
||
* {@link vip.mate.stt.SttService} here, prepending the transcript as a
|
||
* text {@link MessageContentPart} so the agent reasons over actual
|
||
* content instead of the bare {@code "[音频]"} placeholder.
|
||
*
|
||
* <p>Nullable for legacy callers / tests — STT is skipped entirely when
|
||
* absent, and the message still goes through as audio-only (degraded
|
||
* but not broken).
|
||
*/
|
||
private final vip.mate.stt.SttService sttService;
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper) {
|
||
this(channelEntity, messageRouter, objectMapper, null, null, null, null, null, null, null);
|
||
}
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper,
|
||
FeishuMediaUploader mediaUploader,
|
||
GeneratedFileScrubber generatedFileScrubber) {
|
||
this(channelEntity, messageRouter, objectMapper, mediaUploader, generatedFileScrubber,
|
||
null, null, null, null, null);
|
||
}
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper,
|
||
FeishuMediaUploader mediaUploader,
|
||
GeneratedFileScrubber generatedFileScrubber,
|
||
FeishuStreamingCardManager streamingCardManager) {
|
||
this(channelEntity, messageRouter, objectMapper, mediaUploader,
|
||
generatedFileScrubber, streamingCardManager, null, null, null, null);
|
||
}
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper,
|
||
FeishuMediaUploader mediaUploader,
|
||
GeneratedFileScrubber generatedFileScrubber,
|
||
FeishuStreamingCardManager streamingCardManager,
|
||
vip.mate.channel.feishu.cards.FeishuCardDispatcher cardDispatcher) {
|
||
this(channelEntity, messageRouter, objectMapper, mediaUploader,
|
||
generatedFileScrubber, streamingCardManager, cardDispatcher, null, null, null);
|
||
}
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper,
|
||
FeishuMediaUploader mediaUploader,
|
||
GeneratedFileScrubber generatedFileScrubber,
|
||
FeishuStreamingCardManager streamingCardManager,
|
||
vip.mate.channel.feishu.cards.FeishuCardDispatcher cardDispatcher,
|
||
FeishuClientFactory clientFactory,
|
||
vip.mate.tool.document.GeneratedFileCache generatedFileCache) {
|
||
this(channelEntity, messageRouter, objectMapper, mediaUploader,
|
||
generatedFileScrubber, streamingCardManager, cardDispatcher,
|
||
clientFactory, generatedFileCache, null);
|
||
}
|
||
|
||
public FeishuChannelAdapter(ChannelEntity channelEntity,
|
||
ChannelMessageRouter messageRouter,
|
||
ObjectMapper objectMapper,
|
||
FeishuMediaUploader mediaUploader,
|
||
GeneratedFileScrubber generatedFileScrubber,
|
||
FeishuStreamingCardManager streamingCardManager,
|
||
vip.mate.channel.feishu.cards.FeishuCardDispatcher cardDispatcher,
|
||
FeishuClientFactory clientFactory,
|
||
vip.mate.tool.document.GeneratedFileCache generatedFileCache,
|
||
vip.mate.stt.SttService sttService) {
|
||
super(channelEntity, messageRouter, objectMapper);
|
||
this.mediaUploader = mediaUploader;
|
||
this.generatedFileScrubber = generatedFileScrubber;
|
||
this.streamingCardManager = streamingCardManager;
|
||
this.cardDispatcher = cardDispatcher;
|
||
this.clientFactory = clientFactory;
|
||
this.generatedFileCache = generatedFileCache;
|
||
this.sttService = sttService;
|
||
// Feishu WebSocket reconnect: 2s→4s→8s→16s→30s, infinite retry
|
||
this.backoff = new ExponentialBackoff(2000, 30000, 2.0, -1);
|
||
}
|
||
|
||
// ==================== 生命周期 ====================
|
||
|
||
@Override
|
||
protected void doStart() {
|
||
String appId = getConfigString("app_id");
|
||
String appSecret = getConfigString("app_secret");
|
||
|
||
if (appId == null || appSecret == null) {
|
||
throw new IllegalStateException("Feishu channel requires app_id and app_secret in configJson");
|
||
}
|
||
|
||
// HttpClient 两种模式都需要（发送消息、下载媒体、联系人 API）
|
||
this.httpClient = HttpClient.newBuilder()
|
||
.connectTimeout(Duration.ofSeconds(10))
|
||
.build();
|
||
|
||
// 获取初始 tenant_access_token
|
||
refreshTenantAccessToken();
|
||
|
||
// 定时刷新 Token：过期前 5 分钟自动刷新
|
||
scheduleTokenRefresh();
|
||
|
||
// Prefetch the bot's own open_id once the token is valid so the first
|
||
// require_mention check on the WebSocket dispatch thread doesn't pay
|
||
// the 5 s API latency. Failure is non-fatal — getBotOpenId() handles
|
||
// it and the negative cache keeps subsequent retries cheap.
|
||
getBotOpenId();
|
||
|
||
String connectionMode = getConfigString("connection_mode", "websocket");
|
||
if ("websocket".equals(connectionMode)) {
|
||
startWebSocket(appId, appSecret);
|
||
} else {
|
||
// webhook 模式下 encrypt_key 必须配置，否则 fail-fast 拒绝启动；
|
||
// 没有加密密钥 + 无签名校验 = 任何人可伪造 webhook 请求触发 agent 消息
|
||
String encryptKey = getConfigString("encrypt_key", null);
|
||
if (encryptKey == null || encryptKey.isBlank()) {
|
||
throw new IllegalStateException(
|
||
"Feishu channel in webhook mode requires encrypt_key in configJson " +
|
||
"(fail-closed to prevent unauthenticated webhook abuse). " +
|
||
"Configure encrypt_key on the Feishu Event Subscriptions page and mirror " +
|
||
"it in this channel's configJson, or switch connection_mode to websocket.");
|
||
}
|
||
log.info("[feishu] Webhook mode (encrypt_key configured), waiting for callbacks at /api/v1/channels/webhook/feishu");
|
||
}
|
||
|
||
log.info("[feishu] Feishu channel initialized: appId={}, mode={}, domain={}",
|
||
appId, connectionMode, getConfigString("domain", "feishu"));
|
||
}
|
||
|
||
@Override
|
||
protected void doStop() {
|
||
// 取消定时 Token 刷新
|
||
if (tokenRefreshFuture != null) {
|
||
tokenRefreshFuture.cancel(false);
|
||
tokenRefreshFuture = null;
|
||
}
|
||
|
||
// 关闭 WebSocket
|
||
stopWebSocket();
|
||
|
||
this.httpClient = null;
|
||
this.tenantAccessToken = null;
|
||
// Reset bot-open-id state under the same lock getBotOpenId uses, so a
|
||
// dispatch thread mid-fetch sees a consistent (cleared) view rather than
|
||
// a torn write that could re-cache a stale id.
|
||
synchronized (botOpenIdLock) {
|
||
this.botOpenId = null;
|
||
this.botOpenIdLastFailureMs = 0L;
|
||
}
|
||
this.processedMessageIds.clear();
|
||
this.nicknameCache.clear();
|
||
this.quotedMessageCache.clear();
|
||
log.info("[feishu] Feishu channel stopped");
|
||
}
|
||
|
||
@Override
|
||
protected void doReconnect() {
|
||
String appId = getConfigString("app_id");
|
||
String appSecret = getConfigString("app_secret");
|
||
String connectionMode = getConfigString("connection_mode", "websocket");
|
||
|
||
// 重新建立 HTTP 客户端
|
||
this.httpClient = HttpClient.newBuilder()
|
||
.connectTimeout(Duration.ofSeconds(10))
|
||
.build();
|
||
|
||
try {
|
||
refreshTenantAccessToken();
|
||
} catch (Exception e) {
|
||
log.warn("[feishu] Token refresh during reconnect failed: {}", e.getMessage());
|
||
}
|
||
|
||
// Re-prefetch bot open_id after reconnect: same dispatch-thread latency
|
||
// concern as doStart, plus picks up a rotated app identity if any.
|
||
getBotOpenId();
|
||
|
||
if ("websocket".equals(connectionMode)) {
|
||
log.info("[feishu] Reconnecting WebSocket...");
|
||
stopWebSocket();
|
||
// 同步连接：在当前重连线程中直接阻塞调用 start()
|
||
// 连接成功 start() 会一直阻塞（不会返回到这里）
|
||
// 连接失败 start() 抛异常，由 AbstractChannelAdapter.scheduleReconnect 捕获并触发 onReconnectFailed
|
||
startWebSocketSync(appId, appSecret);
|
||
}
|
||
|
||
log.info("[feishu] Reconnect completed for: {}", channelEntity.getName());
|
||
}
|
||
|
||
// ==================== WebSocket 长连接 ====================
|
||
|
||
/**
|
||
* 创建 WebSocket 客户端实例（不启动连接）
|
||
*/
|
||
private com.lark.oapi.ws.Client createWsClient(String appId, String appSecret) {
|
||
EventDispatcher eventDispatcher = EventDispatcher.newBuilder("", "")
|
||
.onP2MessageReceiveV1(new ImService.P2MessageReceiveV1Handler() {
|
||
@Override
|
||
public void handle(P2MessageReceiveV1 event) throws Exception {
|
||
if (!running.get()) return;
|
||
|
||
// 检查 app_id 匹配（防止多实例事件错路由）
|
||
if (event.getHeader() != null && event.getHeader().getAppId() != null
|
||
&& !event.getHeader().getAppId().equals(appId)) {
|
||
log.debug("[feishu] Dropping misrouted event, app_id={} (expected {})",
|
||
event.getHeader().getAppId(), appId);
|
||
return;
|
||
}
|
||
|
||
try {
|
||
handleWebSocketEvent(event);
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to handle WebSocket event: {}", e.getMessage(), e);
|
||
}
|
||
}
|
||
})
|
||
// Silently ignore all other IM events to avoid HandlerNotFoundException.
|
||
// When the SDK receives an event without a registered handler, it throws
|
||
// HandlerNotFoundException → SDK catches it and sends a 500 response →
|
||
// the Feishu server may close the WebSocket connection. The exception is
|
||
// swallowed inside the SDK so it never appears in application logs.
|
||
.onP2MessageReactionCreatedV1(new ImService.P2MessageReactionCreatedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2MessageReactionCreatedV1 event) {}
|
||
})
|
||
.onP2MessageReactionDeletedV1(new ImService.P2MessageReactionDeletedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2MessageReactionDeletedV1 event) {}
|
||
})
|
||
.onP2MessageReadV1(new ImService.P2MessageReadV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2MessageReadV1 event) {}
|
||
})
|
||
.onP2MessageRecalledV1(new ImService.P2MessageRecalledV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2MessageRecalledV1 event) {}
|
||
})
|
||
.onP2ChatMemberBotAddedV1(new ImService.P2ChatMemberBotAddedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatMemberBotAddedV1 event) {}
|
||
})
|
||
.onP2ChatMemberBotDeletedV1(new ImService.P2ChatMemberBotDeletedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatMemberBotDeletedV1 event) {}
|
||
})
|
||
.onP2ChatMemberUserAddedV1(new ImService.P2ChatMemberUserAddedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatMemberUserAddedV1 event) {}
|
||
})
|
||
.onP2ChatMemberUserDeletedV1(new ImService.P2ChatMemberUserDeletedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatMemberUserDeletedV1 event) {}
|
||
})
|
||
.onP2ChatMemberUserWithdrawnV1(new ImService.P2ChatMemberUserWithdrawnV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatMemberUserWithdrawnV1 event) {}
|
||
})
|
||
.onP2ChatUpdatedV1(new ImService.P2ChatUpdatedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatUpdatedV1 event) {}
|
||
})
|
||
.onP2ChatDisbandedV1(new ImService.P2ChatDisbandedV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatDisbandedV1 event) {}
|
||
})
|
||
.onP2ChatAccessEventBotP2pChatEnteredV1(new ImService.P2ChatAccessEventBotP2pChatEnteredV1Handler() {
|
||
@Override
|
||
public void handle(com.lark.oapi.service.im.v1.model.P2ChatAccessEventBotP2pChatEnteredV1 event) {}
|
||
})
|
||
// Interactive card button clicks (Schema 2.0) — routed through cardDispatcher.
|
||
// The returned P2CardActionTriggerResponse is how Schema-2.0
|
||
// cards update in-place; PATCH /im/v1/messages/{id} is a
|
||
// silent no-op for V2 cards and must not be used.
|
||
.onP2CardActionTrigger(new com.lark.oapi.event.cardcallback.P2CardActionTriggerHandler() {
|
||
@Override
|
||
public com.lark.oapi.event.cardcallback.model.P2CardActionTriggerResponse handle(
|
||
com.lark.oapi.event.cardcallback.model.P2CardActionTrigger event) {
|
||
if (!running.get()) return null;
|
||
try {
|
||
return handleCardActionTrigger(event);
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to handle card action: {}", e.getMessage(), e);
|
||
return null;
|
||
}
|
||
}
|
||
})
|
||
.build();
|
||
|
||
return new com.lark.oapi.ws.Client.Builder(appId, appSecret)
|
||
.eventHandler(eventDispatcher)
|
||
.autoReconnect(false) // 由我们的 ExponentialBackoff 控制重连，不用 SDK 内置重连
|
||
.domain("lark".equals(getConfigString("domain", "feishu"))
|
||
? "https://open.larksuite.com"
|
||
: "https://open.feishu.cn")
|
||
.build();
|
||
}
|
||
|
||
/**
|
||
* 启动 WebSocket 长连接（异步，用于 doStart 首次启动）
|
||
* 在守护线程中运行，避免阻塞主线程。连接失败时触发 onDisconnected → 退避重连
|
||
*/
|
||
private void startWebSocket(String appId, String appSecret) {
|
||
wsClient = createWsClient(appId, appSecret);
|
||
|
||
wsThread = new Thread(() -> {
|
||
try {
|
||
log.info("[feishu] WebSocket connecting (long connection)...");
|
||
wsClient.start();
|
||
// start() blocks until disconnect; if it returns normally, it means disconnected
|
||
if (running.get()) {
|
||
onDisconnected("WebSocket connection ended");
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("[feishu] WebSocket error: {}", e.getMessage(), e);
|
||
if (running.get()) {
|
||
onDisconnected("WebSocket error: " + e.getMessage());
|
||
}
|
||
}
|
||
}, "feishu-ws-" + channelEntity.getId());
|
||
wsThread.setDaemon(true);
|
||
wsThread.start();
|
||
|
||
startSilentDisconnectWatchdog();
|
||
}
|
||
|
||
/**
|
||
* 启动静默断连看门狗。
|
||
* <p>
|
||
* SDK 内部有 ping/pong 心跳，正常情况下连接断开会触发 start() 返回 → onDisconnected。
|
||
* 但少数场景 TCP 层认为连接还在但事件不再流入（NAT 超时、半开连接、对端 hang 死），
|
||
* SDK 也察觉不到。看门狗每 60s 检查一次「最近事件时间」，超过阈值就强制重连。
|
||
* <p>
|
||
* 静默判定有两个前置条件：
|
||
* 1. 已经收到过至少一个事件（避免新连接立即触发误报）
|
||
* 2. silent_disconnect_threshold_seconds > 0（设为 0 可禁用看门狗）
|
||
* <p>
|
||
* 默认阈值 30 分钟。安静的渠道（一天没几条消息）建议调大到 1-2 小时；
|
||
* 高频渠道可以调小到 5-10 分钟以更快发现问题。
|
||
*/
|
||
private void startSilentDisconnectWatchdog() {
|
||
long thresholdSec = getConfigLong("silent_disconnect_threshold_seconds",
|
||
DEFAULT_SILENT_THRESHOLD_SECONDS);
|
||
if (thresholdSec <= 0) {
|
||
log.debug("[feishu] Silent disconnect watchdog disabled (threshold=0)");
|
||
return;
|
||
}
|
||
long thresholdMs = thresholdSec * 1000L;
|
||
|
||
cancelSilentDisconnectWatchdog();
|
||
silentDisconnectWatchdog = ensureReconnectScheduler().scheduleAtFixedRate(() -> {
|
||
if (!running.get() || wsClient == null) return;
|
||
if (!hasReceivedFirstEvent) return; // 没收到首个事件前不算静默
|
||
|
||
long silentMs = System.currentTimeMillis() - lastEventTimeMs.get();
|
||
if (silentMs > thresholdMs) {
|
||
log.warn("[feishu] Silent WebSocket detected: no events for {}s (threshold {}s), forcing reconnect",
|
||
silentMs / 1000, thresholdSec);
|
||
onDisconnected("silent disconnect: no events for " + (silentMs / 1000) + "s");
|
||
}
|
||
}, WATCHDOG_INTERVAL_SECONDS, WATCHDOG_INTERVAL_SECONDS, TimeUnit.SECONDS);
|
||
log.info("[feishu] Silent disconnect watchdog started (threshold {}s, check every {}s)",
|
||
thresholdSec, WATCHDOG_INTERVAL_SECONDS);
|
||
}
|
||
|
||
private void cancelSilentDisconnectWatchdog() {
|
||
if (silentDisconnectWatchdog != null) {
|
||
silentDisconnectWatchdog.cancel(false);
|
||
silentDisconnectWatchdog = null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 启动 WebSocket 长连接（同步，用于 doReconnect 重连线程）
|
||
* 在当前线程中阻塞调用 start()：
|
||
* - 连接成功后 start() 会一直阻塞（收消息），不会返回
|
||
* - 连接失败 start() 抛异常，由 scheduleReconnect 的 catch 捕获 → onReconnectFailed → 退避递增
|
||
*/
|
||
private void startWebSocketSync(String appId, String appSecret) {
|
||
wsClient = createWsClient(appId, appSecret);
|
||
log.info("[feishu] WebSocket connecting (long connection)...");
|
||
// 看门狗必须在 start() 之前启动，否则 start() 阻塞后这一行永远到不了，
|
||
// 一旦断连后重连这条路径，watchdog 就再也不会被恢复
|
||
startSilentDisconnectWatchdog();
|
||
wsClient.start(); // 阻塞：成功则永驻，失败则抛异常
|
||
}
|
||
|
||
/**
|
||
* 关闭 WebSocket 连接
|
||
* <p>
|
||
* 必须主动关闭底层 WebSocket 连接并触发 SDK 的内部清理（停止 pingLoop、
|
||
* 释放 ExecutorService）。仅置空引用会导致旧连接的 pingLoop 和线程池
|
||
* 持续运行，造成文件描述符和线程泄漏，最终使新连接无法建立。
|
||
* <p>
|
||
* SDK 的 {@code disconnect()} 是 protected 的，无法直接调用。通过反射
|
||
* 获取内部 {@code conn} 字段并发送 close(1000) 触发 SDK 的
|
||
* {@code onClosed → disconnect()} 清理链路。
|
||
*/
|
||
private void stopWebSocket() {
|
||
cancelSilentDisconnectWatchdog();
|
||
hasReceivedFirstEvent = false;
|
||
if (wsThread != null) {
|
||
wsThread.interrupt();
|
||
wsThread = null;
|
||
}
|
||
if (wsClient != null) {
|
||
try {
|
||
// 反射获取 protected conn 字段，发送 close(1000) 触发 SDK 清理
|
||
var connField = wsClient.getClass().getDeclaredField("conn");
|
||
connField.setAccessible(true);
|
||
Object conn = connField.get(wsClient);
|
||
if (conn != null) {
|
||
// conn 是 OkHttp WebSocket，close(1000) 会触发 onClosed → disconnect()
|
||
var closeMethod = conn.getClass().getMethod("close", int.class, String.class);
|
||
closeMethod.invoke(conn, 1000, "client closed");
|
||
}
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] Error during WebSocket disconnect: {}", e.getMessage());
|
||
}
|
||
}
|
||
wsClient = null;
|
||
}
|
||
|
||
/**
|
||
* 处理 WebSocket 事件：从 P2MessageReceiveV1 提取字段，调用统一入口
|
||
*/
|
||
private void handleWebSocketEvent(P2MessageReceiveV1 event) {
|
||
var eventBody = event.getEvent();
|
||
if (eventBody == null || eventBody.getMessage() == null) {
|
||
return;
|
||
}
|
||
|
||
var message = eventBody.getMessage();
|
||
var sender = eventBody.getSender();
|
||
|
||
// 任何事件到达都更新活跃时间戳，看门狗用它判断是否静默断连
|
||
touchActivity();
|
||
hasReceivedFirstEvent = true;
|
||
|
||
// 旧事件过滤：SDK 在重连后可能回放历史事件，按 message.create_time 过滤
|
||
// 远超 stale 阈值的消息（典型场景：连接断了 5 分钟后恢复，5 分钟前的消息再处理一次没意义）
|
||
long staleThresholdMs = getConfigLong("stale_event_threshold_seconds",
|
||
DEFAULT_STALE_THRESHOLD_SECONDS) * 1000L;
|
||
if (staleThresholdMs > 0 && message.getCreateTime() != null) {
|
||
try {
|
||
long msgCreateTimeMs = Long.parseLong(message.getCreateTime());
|
||
long ageMs = System.currentTimeMillis() - msgCreateTimeMs;
|
||
if (ageMs > staleThresholdMs) {
|
||
log.info("[feishu] Dropping stale event: messageId={}, age={}s (threshold {}s)",
|
||
message.getMessageId(), ageMs / 1000, staleThresholdMs / 1000);
|
||
return;
|
||
}
|
||
} catch (NumberFormatException ignored) {
|
||
// create_time 不是合法的毫秒数，跳过过滤而不是误丢
|
||
}
|
||
}
|
||
|
||
String messageId = message.getMessageId();
|
||
String messageType = message.getMessageType();
|
||
String contentStr = message.getContent();
|
||
String chatId = message.getChatId();
|
||
String chatType = message.getChatType();
|
||
String parentId = message.getParentId();
|
||
|
||
String senderOpenId = null;
|
||
if (sender != null && sender.getSenderId() != null) {
|
||
senderOpenId = sender.getSenderId().getOpenId();
|
||
}
|
||
|
||
boolean isBotMentioned = isBotMentionedInEvent(message.getMentions());
|
||
handleFeishuMessage(messageId, messageType, contentStr, chatId, chatType, senderOpenId, parentId, isBotMentioned, event);
|
||
}
|
||
|
||
// ==================== @提及检测 ====================
|
||
|
||
private boolean isBotMentionedInEvent(com.lark.oapi.service.im.v1.model.MentionEvent[] mentions) {
|
||
return eventMentionsContainBot(mentions, getBotOpenId());
|
||
}
|
||
|
||
private boolean isBotMentionedInWebhookMessage(Map<String, Object> message) {
|
||
Object mentionsObj = message.get("mentions");
|
||
if (!(mentionsObj instanceof List<?> list)) return false;
|
||
return webhookMentionsContainBot(list, getBotOpenId());
|
||
}
|
||
|
||
/** Package-private for testing: 判断 SDK mentions 数组中是否包含指定 open_id */
|
||
static boolean eventMentionsContainBot(com.lark.oapi.service.im.v1.model.MentionEvent[] mentions,
|
||
String botOpenId) {
|
||
if (mentions == null || mentions.length == 0 || botOpenId == null) return false;
|
||
for (var mention : mentions) {
|
||
if (mention.getId() != null && botOpenId.equals(mention.getId().getOpenId())) return true;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
/** Package-private for testing: 判断 Webhook mentions 列表中是否包含指定 open_id */
|
||
static boolean webhookMentionsContainBot(List<?> mentions, String botOpenId) {
|
||
if (mentions == null || botOpenId == null) return false;
|
||
for (Object item : mentions) {
|
||
if (!(item instanceof Map<?, ?> mention)) continue;
|
||
Object idObj = mention.get("id");
|
||
if (!(idObj instanceof Map<?, ?> id)) continue;
|
||
if (botOpenId.equals(id.get("open_id"))) return true;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
/**
|
||
* Package-private for testing: returns true when a group message should be dropped
|
||
* because {@code require_mention} is on, the bot was not mentioned, AND we know our
|
||
* own open_id (so we trust the negative answer).
|
||
*
|
||
* <p>When {@code botOpenId} is {@code null} the bot identity is unavailable — either
|
||
* the {@code /open-apis/bot/v3/info} fetch hasn't succeeded yet, or it failed and is
|
||
* in the negative-cache window. In that case the gate falls open so a transient
|
||
* Feishu API outage doesn't silence the bot in every group it's in. The accompanying
|
||
* warn-level log makes the degraded mode visible.
|
||
*/
|
||
static boolean isGroupNonMentionDrop(boolean isGroup,
|
||
boolean requireMention,
|
||
boolean isBotMentioned,
|
||
String botOpenId) {
|
||
return isGroup && requireMention && !isBotMentioned && botOpenId != null;
|
||
}
|
||
|
||
/**
|
||
* Fetches the bot's own open_id once and caches it. Returns {@code null}
|
||
* when the identity is unavailable; callers (currently the
|
||
* {@code require_mention} gate) treat {@code null} as "identity unknown"
|
||
* and fall open so a transient API outage doesn't silence the bot.
|
||
*
|
||
* <p>Concurrent callers share a single API roundtrip via
|
||
* {@link #botOpenIdLock}. On failure, {@link #botOpenIdLastFailureMs} is
|
||
* stamped so callers within the next {@link #BOT_OPENID_FAILURE_BACKOFF_MS}
|
||
* ms return {@code null} immediately instead of triggering a fresh 5 s
|
||
* synchronous fetch per inbound message.
|
||
*/
|
||
private String getBotOpenId() {
|
||
String cached = botOpenId;
|
||
if (cached != null) return cached;
|
||
if (withinFailureBackoff()) return null;
|
||
synchronized (botOpenIdLock) {
|
||
// Re-check under the lock — another thread may have populated the
|
||
// cache or stamped a fresh failure while we were waiting.
|
||
if (botOpenId != null) return botOpenId;
|
||
if (withinFailureBackoff()) return null;
|
||
try {
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/bot/v3/info"))
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.GET()
|
||
.timeout(Duration.ofSeconds(5))
|
||
.build();
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
Map<?, ?> body = objectMapper.readValue(response.body(), Map.class);
|
||
Map<?, ?> bot = (Map<?, ?>) body.get("bot");
|
||
if (bot != null && bot.get("open_id") instanceof String openId && !openId.isBlank()) {
|
||
botOpenId = openId;
|
||
log.info("[feishu] Bot open_id fetched and cached: {}", openId);
|
||
return openId;
|
||
}
|
||
// 2xx with no bot.open_id field → treat as transient failure.
|
||
botOpenIdLastFailureMs = System.currentTimeMillis();
|
||
log.warn("[feishu] /open-apis/bot/v3/info returned no bot.open_id; require_mention gate falls open for {}s",
|
||
BOT_OPENID_FAILURE_BACKOFF_MS / 1000);
|
||
} catch (Exception e) {
|
||
botOpenIdLastFailureMs = System.currentTimeMillis();
|
||
log.warn("[feishu] Failed to fetch bot open_id (require_mention gate falls open for {}s): {}",
|
||
BOT_OPENID_FAILURE_BACKOFF_MS / 1000, e.getMessage());
|
||
}
|
||
return null;
|
||
}
|
||
}
|
||
|
||
private boolean withinFailureBackoff() {
|
||
long last = botOpenIdLastFailureMs;
|
||
return last != 0L && System.currentTimeMillis() - last < BOT_OPENID_FAILURE_BACKOFF_MS;
|
||
}
|
||
|
||
// ==================== Token 管理 ====================
|
||
|
||
/**
|
||
* 定时刷新 Token：每隔 (expireSeconds - 300) 秒刷新一次
|
||
*/
|
||
private void scheduleTokenRefresh() {
|
||
// Token 默认有效期 7200s，提前 5 分钟刷新 => 周期 6900s
|
||
long refreshIntervalSeconds = Math.max(300, 7200 - 300);
|
||
tokenRefreshFuture = ensureReconnectScheduler().scheduleAtFixedRate(() -> {
|
||
if (!running.get()) return;
|
||
try {
|
||
refreshTenantAccessToken();
|
||
log.debug("[feishu] Scheduled token refresh succeeded");
|
||
} catch (Exception e) {
|
||
log.warn("[feishu] Scheduled token refresh failed: {}, will retry on next interval",
|
||
e.getMessage());
|
||
lastError = "Token refresh failed: " + e.getMessage();
|
||
}
|
||
}, refreshIntervalSeconds, refreshIntervalSeconds, TimeUnit.SECONDS);
|
||
log.info("[feishu] Token auto-refresh scheduled every {}s", refreshIntervalSeconds);
|
||
}
|
||
|
||
/**
|
||
* 获取/刷新 tenant_access_token
|
||
*/
|
||
private void refreshTenantAccessToken() {
|
||
String appId = getConfigString("app_id");
|
||
String appSecret = getConfigString("app_secret");
|
||
String apiBase = getApiBaseUrl();
|
||
|
||
try {
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"app_id", appId,
|
||
"app_secret", appSecret
|
||
));
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/auth/v3/tenant_access_token/internal"))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
@SuppressWarnings("unchecked")
|
||
Map<String, Object> result = objectMapper.readValue(response.body(), Map.class);
|
||
|
||
Integer code = result.get("code") instanceof Number n ? n.intValue() : null;
|
||
if (code != null && code != 0) {
|
||
throw new RuntimeException("Feishu API error: code=" + code + ", msg=" + result.get("msg"));
|
||
}
|
||
|
||
this.tenantAccessToken = (String) result.get("tenant_access_token");
|
||
Object expire = result.get("expire");
|
||
int expireSeconds = expire instanceof Number n ? n.intValue() : 7200;
|
||
this.tokenExpireTime = System.currentTimeMillis() + (expireSeconds - 300) * 1000L;
|
||
|
||
log.info("[feishu] tenant_access_token refreshed, expires in {}s", expireSeconds);
|
||
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to refresh tenant_access_token: {}", e.getMessage(), e);
|
||
throw new RuntimeException("Token refresh failed: " + e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
private void ensureTokenValid() {
|
||
if (tenantAccessToken == null || System.currentTimeMillis() >= tokenExpireTime) {
|
||
try {
|
||
refreshTenantAccessToken();
|
||
} catch (Exception e) {
|
||
log.warn("[feishu] On-demand token refresh failed: {}", e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
|
||
// ==================== Domain 国际化 ====================
|
||
|
||
/**
|
||
* 获取 API 基础 URL
|
||
* domain=feishu → https://open.feishu.cn
|
||
* domain=lark → https://open.larksuite.com
|
||
*/
|
||
private String getApiBaseUrl() {
|
||
String domain = getConfigString("domain", "feishu");
|
||
return "lark".equals(domain)
|
||
? "https://open.larksuite.com"
|
||
: "https://open.feishu.cn";
|
||
}
|
||
|
||
// ==================== Webhook 处理 ====================
|
||
|
||
/**
|
||
* 处理飞书 Event Subscription 回调
|
||
* 由 ChannelWebhookController 调用
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
public Map<String, Object> handleWebhook(Map<String, Object> payload) {
|
||
// 处理 URL 验证请求
|
||
String type = (String) payload.get("type");
|
||
if ("url_verification".equals(type)) {
|
||
String challenge = (String) payload.get("challenge");
|
||
log.info("[feishu] URL verification challenge received");
|
||
return Map.of("challenge", challenge != null ? challenge : "");
|
||
}
|
||
|
||
try {
|
||
// 解析 v2 事件格式
|
||
Map<String, Object> header = (Map<String, Object>) payload.get("header");
|
||
Map<String, Object> event = (Map<String, Object>) payload.get("event");
|
||
|
||
if (header == null || event == null) {
|
||
log.warn("[feishu] Invalid event payload: missing header or event");
|
||
return Map.of("code", 0);
|
||
}
|
||
|
||
String eventType = (String) header.get("event_type");
|
||
if (!"im.message.receive_v1".equals(eventType)) {
|
||
log.debug("[feishu] Ignoring event type: {}", eventType);
|
||
return Map.of("code", 0);
|
||
}
|
||
|
||
// 解析消息
|
||
Map<String, Object> message = (Map<String, Object>) event.get("message");
|
||
if (message == null) {
|
||
return Map.of("code", 0);
|
||
}
|
||
|
||
String messageId = (String) message.get("message_id");
|
||
String messageType = (String) message.get("message_type");
|
||
String contentStr = (String) message.get("content");
|
||
String chatId = (String) message.get("chat_id");
|
||
String chatType = (String) message.get("chat_type");
|
||
String parentId = (String) message.get("parent_id");
|
||
|
||
// 提取发送者 open_id
|
||
Map<String, Object> sender = (Map<String, Object>) event.get("sender");
|
||
String senderOpenId = null;
|
||
if (sender != null) {
|
||
Map<String, Object> senderIdObj = (Map<String, Object>) sender.get("sender_id");
|
||
if (senderIdObj != null) {
|
||
senderOpenId = (String) senderIdObj.get("open_id");
|
||
}
|
||
}
|
||
|
||
boolean isBotMentioned = isBotMentionedInWebhookMessage(message);
|
||
handleFeishuMessage(messageId, messageType, contentStr, chatId, chatType, senderOpenId, parentId, isBotMentioned, payload);
|
||
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to handle webhook: {}", e.getMessage(), e);
|
||
}
|
||
|
||
return Map.of("code", 0);
|
||
}
|
||
|
||
// ==================== 统一消息处理入口 ====================
|
||
|
||
/**
|
||
* 统一消息处理入口（Webhook 和 WebSocket 共用）
|
||
*
|
||
* @param messageId 消息 ID
|
||
* @param messageType 消息类型（text/image/post/file/audio/media）
|
||
* @param contentStr 消息内容 JSON 字符串
|
||
* @param chatId 群组 ID（私聊为 null）
|
||
* @param chatType "p2p" 或 "group"
|
||
* @param senderOpenId 发送者 open_id
|
||
* @param parentId 被引用消息的 message_id（无引用时为 null）
|
||
* @param rawPayload 原始负载（用于调试）
|
||
*/
|
||
private void handleFeishuMessage(String messageId, String messageType, String contentStr,
|
||
String chatId, String chatType, String senderOpenId,
|
||
String parentId, boolean isBotMentioned, Object rawPayload) {
|
||
// require_mention 群聊过滤：群聊中必须 @机器人才响应。
|
||
// 当 botOpenId 为 null 时（API 抖动 / 尚未拉取成功），失败回退到放行 —
|
||
// 避免飞书 /open-apis/bot/v3/info 短暂不可用时整个群机器人变哑巴。
|
||
boolean isGroup = "group".equals(chatType);
|
||
boolean requireMention = getConfigBoolean("require_mention", false);
|
||
if (isGroupNonMentionDrop(isGroup, requireMention, isBotMentioned, botOpenId)) {
|
||
log.debug("[feishu] require_mention=true but bot not mentioned, dropping messageId={}", messageId);
|
||
return;
|
||
}
|
||
if (isGroup && requireMention && !isBotMentioned) {
|
||
// botOpenId is null here — identity unknown, gate falls open.
|
||
log.warn("[feishu] require_mention=true but bot open_id unavailable; allowing messageId={}", messageId);
|
||
}
|
||
|
||
// 消息去重
|
||
if (messageId != null && !processedMessageIds.add(messageId)) {
|
||
log.debug("[feishu] Duplicate message_id: {}, skipping", messageId);
|
||
return;
|
||
}
|
||
cleanupProcessedIds();
|
||
|
||
// 添加消息反应（非阻塞，表示"已收到"）
|
||
if (messageId != null && getConfigBoolean("enable_reaction", true)) {
|
||
addReactionAsync(messageId, "THUMBSUP");
|
||
}
|
||
|
||
// 获取用户昵称
|
||
String senderName = senderOpenId;
|
||
if (senderOpenId != null && getConfigBoolean("enable_nickname_cache", true)) {
|
||
senderName = getUserName(senderOpenId);
|
||
}
|
||
|
||
// 解析消息内容
|
||
List<MessageContentPart> contentParts = new ArrayList<>();
|
||
String textContent = extractContentParts(messageId, messageType, contentStr, contentParts);
|
||
|
||
if (contentParts.isEmpty() && (textContent == null || textContent.isBlank())) {
|
||
log.debug("[feishu] Empty message content, ignoring");
|
||
return;
|
||
}
|
||
|
||
// 引用消息（用户在飞书里"引用"了之前的某条消息回复）：拉取被引用消息内容并注入上下文，
|
||
// 让 agent 能理解 "解释一下" 这种缺主语的引用回复 —— 不然就只看到"解释一下"三个字。
|
||
if (parentId != null && !parentId.isBlank() && getConfigBoolean("enable_quoted_context", true)) {
|
||
String quotedText = fetchQuotedMessageText(parentId);
|
||
if (quotedText != null && !quotedText.isBlank()) {
|
||
String prefix = "[引用消息: " + quotedText + "]\n";
|
||
textContent = prefix + (textContent != null ? textContent : "");
|
||
// 同步加一个 text part 到最前面，让多模态消息也能看到引用上下文
|
||
contentParts.add(0, MessageContentPart.text(prefix));
|
||
}
|
||
}
|
||
|
||
// 生成短会话后缀
|
||
String shortSuffix = generateShortSessionSuffix(chatId, senderOpenId, isGroup);
|
||
|
||
ChannelMessage channelMessage = ChannelMessage.builder()
|
||
.messageId(messageId)
|
||
.channelType(CHANNEL_TYPE)
|
||
.senderId(senderOpenId)
|
||
.senderName(senderName)
|
||
.chatId(isGroup ? shortSuffix : null)
|
||
.content(textContent != null ? textContent : "")
|
||
.contentType(messageType)
|
||
.contentParts(contentParts)
|
||
.inputMode("audio".equals(messageType) ? "voice" : "text")
|
||
.timestamp(LocalDateTime.now())
|
||
.rawPayload(rawPayload)
|
||
.build();
|
||
|
||
// replyToken <20><>留完整 chatId（<64><EFBC88>送消息需要完整 ID）
|
||
channelMessage.setReplyToken(chatId);
|
||
onMessage(channelMessage);
|
||
}
|
||
|
||
/**
|
||
* 清理旧的去重记录：超过 1000 条时保留最近添加的（移除最早的一半）
|
||
*/
|
||
private void cleanupProcessedIds() {
|
||
if (processedMessageIds.size() > 1000) {
|
||
int toRemove = processedMessageIds.size() / 2;
|
||
var iterator = processedMessageIds.iterator();
|
||
while (iterator.hasNext() && toRemove > 0) {
|
||
iterator.next();
|
||
iterator.remove();
|
||
toRemove--;
|
||
}
|
||
}
|
||
}
|
||
|
||
// ==================== 消息反应 ====================
|
||
|
||
/**
|
||
* Acknowledge a successful agent reply by reacting to the inbound
|
||
* message with a "DONE" (✅) emoji. Gated by {@code enable_done_reaction}
|
||
* (default true) — operators that prefer a quiet UI can disable it
|
||
* without losing the existing inbound "THUMBSUP" ack.
|
||
*/
|
||
@Override
|
||
public void onAgentCompleted(ChannelMessage inboundMessage) {
|
||
if (inboundMessage == null) return;
|
||
String messageId = inboundMessage.getMessageId();
|
||
if (messageId == null || messageId.isBlank()) return;
|
||
if (!getConfigBoolean("enable_done_reaction", true)) return;
|
||
addReactionAsync(messageId, "DONE");
|
||
}
|
||
|
||
// ==================== Approval card ====================
|
||
|
||
/**
|
||
* Card-button approval is the canonical path for Feishu when the
|
||
* dispatcher is wired. Tells {@code ChannelMessageRouter} not to
|
||
* auto-cancel pending approvals on subsequent user messages —
|
||
* users decide via clicking, not by typing {@code /approve}.
|
||
* Returns false only in legacy / test contexts where the
|
||
* dispatcher isn't injected (the inherited text-flow path applies).
|
||
*/
|
||
@Override
|
||
public boolean usesInteractiveApprovalCards() {
|
||
return cardDispatcher != null;
|
||
}
|
||
|
||
/**
|
||
* Render the approval notice as a Schema-2.0 interactive button card
|
||
* so the user can approve / deny in-channel without bouncing to the
|
||
* web UI. Falls back to the inherited markdown-text path when the
|
||
* dispatcher isn't wired (legacy constructors / test rigs), the
|
||
* card oversizes, or the platform refuses to deliver.
|
||
*/
|
||
@Override
|
||
public void sendApprovalNotice(String targetId,
|
||
vip.mate.channel.notification.ApprovalNotice notice) {
|
||
if (cardDispatcher == null || notice == null || targetId == null) {
|
||
super.sendApprovalNotice(targetId, notice);
|
||
return;
|
||
}
|
||
var kindOpt = cardDispatcher.lookupByName(
|
||
vip.mate.channel.feishu.cards.tool_guard.ToolGuardCardKindFactory.KIND_NAME);
|
||
if (kindOpt.isEmpty()) {
|
||
super.sendApprovalNotice(targetId, notice);
|
||
return;
|
||
}
|
||
try {
|
||
Map<String, Object> cardJson = kindOpt.get().renderer().render(notice);
|
||
boolean sent = sendCard(targetId, cardJson);
|
||
if (!sent) {
|
||
log.warn("[feishu-toolguard] sendCard returned false; falling back to text");
|
||
super.sendApprovalNotice(targetId, notice);
|
||
}
|
||
} catch (vip.mate.channel.cards.CardOversizedException e) {
|
||
log.warn("[feishu-toolguard] approval card oversized ({}); falling back to text", e.getMessage());
|
||
super.sendApprovalNotice(targetId, notice);
|
||
} catch (Exception e) {
|
||
log.error("[feishu-toolguard] render/send approval card failed; falling back to text: {}",
|
||
e.getMessage(), e);
|
||
super.sendApprovalNotice(targetId, notice);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Dispatch a {@code P2CardActionTrigger} (button click on an
|
||
* interactive card) to the matching {@code FeishuCardKind} via the
|
||
* dispatcher and propagate the kind's response back to Feishu so
|
||
* the card can update in-place. Returns null when no dispatcher is
|
||
* wired or no kind matches the action prefix — Feishu leaves the
|
||
* original card unchanged in that case.
|
||
*/
|
||
private com.lark.oapi.event.cardcallback.model.P2CardActionTriggerResponse handleCardActionTrigger(
|
||
com.lark.oapi.event.cardcallback.model.P2CardActionTrigger event) {
|
||
if (cardDispatcher == null || event == null || event.getEvent() == null) {
|
||
return null;
|
||
}
|
||
com.lark.oapi.event.cardcallback.model.P2CardActionTriggerData data = event.getEvent();
|
||
com.lark.oapi.event.cardcallback.model.CallBackAction action = data.getAction();
|
||
if (action == null || action.getValue() == null) {
|
||
return null;
|
||
}
|
||
Object actionField = action.getValue().get("action");
|
||
String actionStr = actionField != null ? actionField.toString() : null;
|
||
var kindOpt = cardDispatcher.lookupByAction(actionStr);
|
||
if (kindOpt.isEmpty()) {
|
||
log.debug("[feishu] No card kind registered for action={}", actionStr);
|
||
return null;
|
||
}
|
||
return kindOpt.get().handler().handle(this, data);
|
||
}
|
||
|
||
/**
|
||
* Re-enter the router as if the given message arrived from this
|
||
* channel. Used by the tool-guard card handler to re-emit the
|
||
* button click as a synthetic {@code /approve <pendingId>} or
|
||
* {@code /deny <pendingId>} so the router runs its canonical
|
||
* text-approve path ({@code resolveAndConsume + replay}).
|
||
*
|
||
* <p>External callers must go through {@link #onMessage}; this is
|
||
* the in-package fast lane that skips the WS / webhook decode.
|
||
*/
|
||
public void injectSyntheticMessage(ChannelMessage message) {
|
||
messageRouter.enqueue(message, this, channelEntity);
|
||
}
|
||
|
||
/**
|
||
* 非阻塞地给消息添加表情反应
|
||
* 在新线程中执行，失败只 log.debug 不影响主流程
|
||
*/
|
||
private void addReactionAsync(String messageId, String emojiType) {
|
||
Thread reactionThread = new Thread(() -> {
|
||
try {
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"reaction_type", Map.of("emoji_type", emojiType)
|
||
));
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages/" + messageId + "/reactions"))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.timeout(Duration.ofSeconds(5))
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.debug("[feishu] Add reaction failed: status={}, body={}", response.statusCode(), response.body());
|
||
} else {
|
||
log.debug("[feishu] Reaction {} added to message {}", emojiType, messageId);
|
||
}
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] Add reaction error: {}", e.getMessage());
|
||
}
|
||
}, "feishu-reaction");
|
||
reactionThread.setDaemon(true);
|
||
reactionThread.start();
|
||
}
|
||
|
||
// ==================== 联系人昵称 ====================
|
||
|
||
/**
|
||
* 通过 open_id 获取用户昵称
|
||
* 优先查缓存 → 调用 Contact API → 降级返回 open_id 后缀
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
private String getUserName(String openId) {
|
||
if (openId == null || openId.isBlank()) return openId;
|
||
|
||
// 1. 查缓存
|
||
String cached = nicknameCache.get(openId);
|
||
if (cached != null) return cached;
|
||
|
||
// 2. 调用 Contact API
|
||
try {
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/contact/v3/users/" + openId + "?user_id_type=open_id"))
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.timeout(Duration.ofSeconds(2))
|
||
.GET()
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() == 200) {
|
||
Map<String, Object> result = objectMapper.readValue(response.body(), Map.class);
|
||
Integer code = result.get("code") instanceof Number n ? n.intValue() : null;
|
||
if (code != null && code == 0) {
|
||
Map<String, Object> data = (Map<String, Object>) result.get("data");
|
||
if (data != null) {
|
||
Map<String, Object> user = (Map<String, Object>) data.get("user");
|
||
if (user != null) {
|
||
String name = firstNonBlank(
|
||
(String) user.get("name"),
|
||
(String) user.get("en_name")
|
||
);
|
||
if (name != null) {
|
||
// 缓存超限时清理最早的一半
|
||
if (nicknameCache.size() >= NICKNAME_CACHE_MAX) {
|
||
int toRemove = nicknameCache.size() / 2;
|
||
var iterator = nicknameCache.keySet().iterator();
|
||
while (iterator.hasNext() && toRemove > 0) {
|
||
iterator.next();
|
||
iterator.remove();
|
||
toRemove--;
|
||
}
|
||
}
|
||
nicknameCache.put(openId, name);
|
||
return name;
|
||
}
|
||
}
|
||
}
|
||
} else {
|
||
log.debug("[feishu] Contact API error for {}: code={}", openId, code);
|
||
}
|
||
}
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] getUserName failed for {}: {}", openId, e.getMessage());
|
||
}
|
||
|
||
// 3. 降级：返回 open_id 后 6 位
|
||
String fallback = openId.length() > 6 ? openId.substring(openId.length() - 6) : openId;
|
||
return fallback;
|
||
}
|
||
|
||
private static String firstNonBlank(String... values) {
|
||
for (String v : values) {
|
||
if (v != null && !v.isBlank()) return v.trim();
|
||
}
|
||
return null;
|
||
}
|
||
|
||
// ==================== 引用消息上下文 ====================
|
||
|
||
/** 引用消息内容缓存：parent_message_id → 文本摘要（避免同一条引用反复拉 API） */
|
||
private final ConcurrentHashMap<String, String> quotedMessageCache = new ConcurrentHashMap<>();
|
||
private static final int QUOTED_CACHE_MAX = 200;
|
||
|
||
/**
|
||
* 拉取被引用消息的文本摘要。
|
||
* <p>
|
||
* GET /open-apis/im/v1/messages/{message_id} 返回 items[0]，body.content 是一段 JSON 字符串，
|
||
* 形如 {"text": "..."} 或 post 富文本。我们只取一个简短文本表示，给 agent 当上下文用，
|
||
* 不还原完整富文本/媒体（成本高且 prompt 容易冗余）。
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
private String fetchQuotedMessageText(String parentMessageId) {
|
||
String cached = quotedMessageCache.get(parentMessageId);
|
||
if (cached != null) return cached;
|
||
|
||
try {
|
||
ensureTokenValid();
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(getApiBaseUrl() + "/open-apis/im/v1/messages/" + parentMessageId))
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.timeout(Duration.ofSeconds(3))
|
||
.GET()
|
||
.build();
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.debug("[feishu] Fetch quoted message failed: status={}", response.statusCode());
|
||
return null;
|
||
}
|
||
Map<String, Object> result = objectMapper.readValue(response.body(), Map.class);
|
||
Integer code = result.get("code") instanceof Number n ? n.intValue() : null;
|
||
if (code == null || code != 0) {
|
||
log.debug("[feishu] Fetch quoted message API error: code={}, msg={}", code, result.get("msg"));
|
||
return null;
|
||
}
|
||
Map<String, Object> data = (Map<String, Object>) result.get("data");
|
||
if (data == null) return null;
|
||
List<Map<String, Object>> items = (List<Map<String, Object>>) data.get("items");
|
||
if (items == null || items.isEmpty()) return null;
|
||
Map<String, Object> item = items.get(0);
|
||
String msgType = (String) item.get("msg_type");
|
||
Map<String, Object> body = (Map<String, Object>) item.get("body");
|
||
if (body == null) return null;
|
||
String contentJson = (String) body.get("content");
|
||
String summary = summarizeQuotedContent(msgType, contentJson);
|
||
|
||
if (summary != null && !summary.isBlank()) {
|
||
if (quotedMessageCache.size() >= QUOTED_CACHE_MAX) {
|
||
int toRemove = quotedMessageCache.size() / 2;
|
||
var iter = quotedMessageCache.keySet().iterator();
|
||
while (iter.hasNext() && toRemove > 0) {
|
||
iter.next(); iter.remove(); toRemove--;
|
||
}
|
||
}
|
||
quotedMessageCache.put(parentMessageId, summary);
|
||
}
|
||
return summary;
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] fetchQuotedMessageText failed: {}", e.getMessage());
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 把引用消息的 body.content 折叠成一个简短文本表示，控制 prompt 注入大小。
|
||
* 长文本截断到 200 字符以内，post 取段落首文本，图片/文件给类型占位。
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
private String summarizeQuotedContent(String msgType, String contentJson) {
|
||
if (contentJson == null) return null;
|
||
try {
|
||
Map<String, Object> obj = objectMapper.readValue(contentJson, Map.class);
|
||
String text = switch (msgType != null ? msgType : "") {
|
||
case "text" -> (String) obj.get("text");
|
||
case "image" -> "[图片]";
|
||
case "file" -> "[文件: " + obj.getOrDefault("file_name", "") + "]";
|
||
case "audio" -> "[音频]";
|
||
case "media" -> "[视频]";
|
||
case "post" -> extractPostFirstText(obj);
|
||
default -> "[" + msgType + "]";
|
||
};
|
||
if (text == null) return null;
|
||
text = text.trim();
|
||
if (text.length() > 200) text = text.substring(0, 200) + "…";
|
||
return text;
|
||
} catch (Exception e) {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
@SuppressWarnings("unchecked")
|
||
private String extractPostFirstText(Map<String, Object> postObj) {
|
||
// post 结构：{"zh_cn": {"title": "...", "content": [[{tag:text, text:"..."}], ...]}}
|
||
// 我们只取第一个语种的 title + 第一段的首个 text 元素，作为引用摘要
|
||
for (Object langValue : postObj.values()) {
|
||
if (!(langValue instanceof Map<?, ?> lang)) continue;
|
||
String title = (String) ((Map<String, Object>) lang).get("title");
|
||
List<List<Map<String, Object>>> content = (List<List<Map<String, Object>>>) ((Map<String, Object>) lang).get("content");
|
||
StringBuilder sb = new StringBuilder();
|
||
if (title != null && !title.isBlank()) sb.append(title).append("：");
|
||
if (content != null) {
|
||
outer:
|
||
for (var paragraph : content) {
|
||
for (var inline : paragraph) {
|
||
if ("text".equals(inline.get("tag"))) {
|
||
sb.append(inline.get("text"));
|
||
break outer;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return sb.toString();
|
||
}
|
||
return null;
|
||
}
|
||
|
||
// ==================== 会话 ID 优化 ====================
|
||
|
||
/**
|
||
* 生成更短的会话标识后缀
|
||
* - 群聊：app_id 后 4 位 + "_" + chat_id 后 8 位
|
||
* - 私聊：open_id 后 12 位
|
||
*/
|
||
private String generateShortSessionSuffix(String chatId, String openId, boolean isGroup) {
|
||
if (isGroup && chatId != null) {
|
||
String appId = getConfigString("app_id", "");
|
||
String appSuffix = appId.length() >= 4 ? appId.substring(appId.length() - 4) : appId;
|
||
String chatSuffix = chatId.length() >= 8 ? chatId.substring(chatId.length() - 8) : chatId;
|
||
return appSuffix + "_" + chatSuffix;
|
||
}
|
||
if (openId != null) {
|
||
return openId.length() >= 12 ? openId.substring(openId.length() - 12) : openId;
|
||
}
|
||
if (chatId != null) {
|
||
return chatId.length() >= 12 ? chatId.substring(chatId.length() - 12) : chatId;
|
||
}
|
||
return null;
|
||
}
|
||
|
||
// ==================== 消息内容解析 ====================
|
||
|
||
/**
|
||
* 解析飞书消息内容为 contentParts
|
||
*
|
||
* @param messageId 消息 ID（用于媒体下载）
|
||
* @param messageType 消息类型
|
||
* @param contentStr 消息内容 JSON 字符串
|
||
* @param parts 输出的 content parts
|
||
* @return 纯文本摘要
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
private String extractContentParts(String messageId, String messageType, String contentStr,
|
||
List<MessageContentPart> parts) {
|
||
if (contentStr == null) return null;
|
||
|
||
try {
|
||
Map<String, Object> contentObj = objectMapper.readValue(contentStr, Map.class);
|
||
|
||
return switch (messageType) {
|
||
case "text" -> {
|
||
String text = (String) contentObj.get("text");
|
||
if (text != null && !text.isBlank()) {
|
||
parts.add(MessageContentPart.text(text));
|
||
}
|
||
yield text;
|
||
}
|
||
case "post" -> {
|
||
yield parsePostContent(messageId, contentObj, parts);
|
||
}
|
||
case "image" -> {
|
||
String imageKey = (String) contentObj.get("image_key");
|
||
if (imageKey != null) {
|
||
// Images need bytes for vision: the Feishu CDN URL
|
||
// requires tenant_access_token, so a downstream vision
|
||
// tool that only sees image_key cannot fetch the
|
||
// image. Default-on for images (separate from the
|
||
// file/audio/video gate) so vision works out of the
|
||
// box; admins can opt out with feishu_image_download_enabled=false.
|
||
DownloadedResource dl = maybeDownloadImage(messageId, imageKey);
|
||
MessageContentPart part = MessageContentPart.image(imageKey, null);
|
||
applyDownload(part, dl);
|
||
parts.add(part);
|
||
}
|
||
yield "[图片]";
|
||
}
|
||
case "file" -> {
|
||
String fileKey = (String) contentObj.get("file_key");
|
||
String fileName = (String) contentObj.get("file_name");
|
||
if (fileKey != null) {
|
||
DownloadedResource dl = maybeDownloadResource(messageId, fileKey, "file", fileName);
|
||
MessageContentPart part = MessageContentPart.file(fileKey, fileName, null);
|
||
applyDownload(part, dl);
|
||
parts.add(part);
|
||
}
|
||
yield "[文件: " + (fileName != null ? fileName : "") + "]";
|
||
}
|
||
case "audio" -> {
|
||
String fileKey = (String) contentObj.get("file_key");
|
||
if (fileKey != null) {
|
||
// Feishu voice messages are always opus (no extension on the
|
||
// wire) — give downloadResource the hint so the on-disk file
|
||
// ends in .opus and SttService gets a usable MIME for routing.
|
||
DownloadedResource dl = maybeDownloadResource(messageId, fileKey, "file", "voice.opus");
|
||
MessageContentPart part = MessageContentPart.audio(fileKey, null);
|
||
applyDownload(part, dl);
|
||
// STT hop: inject the transcript as a sibling text part BEFORE
|
||
// the audio part so ChannelMessageRouter.buildPromptFromParts
|
||
// sees real content instead of just "[音频]". WeCom / DingTalk
|
||
// skip this step because their webhooks already include ASR
|
||
// text; Feishu does not.
|
||
String transcript = transcribeInboundAudio(dl);
|
||
if (transcript != null && !transcript.isBlank()) {
|
||
parts.add(MessageContentPart.text(transcript));
|
||
}
|
||
parts.add(part);
|
||
}
|
||
yield "[音频]";
|
||
}
|
||
case "media" -> {
|
||
String fileKey = (String) contentObj.get("file_key");
|
||
String fileName = (String) contentObj.get("file_name");
|
||
if (fileKey != null) {
|
||
DownloadedResource dl = maybeDownloadResource(messageId, fileKey, "file", fileName);
|
||
MessageContentPart part = MessageContentPart.video(fileKey, fileName);
|
||
applyDownload(part, dl);
|
||
parts.add(part);
|
||
}
|
||
yield "[视频]";
|
||
}
|
||
default -> {
|
||
parts.add(MessageContentPart.text("[" + messageType + " 消息]"));
|
||
yield "[" + messageType + " 消息暂不支持处理]";
|
||
}
|
||
};
|
||
} catch (Exception e) {
|
||
log.warn("[feishu] Failed to parse message content: {}", e.getMessage());
|
||
parts.add(MessageContentPart.text(contentStr));
|
||
return contentStr;
|
||
}
|
||
}
|
||
|
||
// ==================== Post 富文本解析 ====================
|
||
|
||
/**
|
||
* 解析飞书 post 富文本消息
|
||
* <p>
|
||
* 飞书 post 结构：
|
||
* <pre>
|
||
* {
|
||
* "zh_cn": {
|
||
* "title": "标题",
|
||
* "content": [ ← 段落数组
|
||
* [ ← 每个段落是行内元素数组
|
||
* {"tag": "text", "text": "内容"},
|
||
* {"tag": "a", "text": "链接", "href": "url"},
|
||
* {"tag": "at", "user_name": "名字", "user_id": "id"},
|
||
* {"tag": "img", "image_key": "key"},
|
||
* {"tag": "media", "file_key": "key"},
|
||
* {"tag": "code_block", "text": "code"},
|
||
* {"tag": "md", "text": "markdown"}
|
||
* ]
|
||
* ]
|
||
* }
|
||
* }
|
||
* </pre>
|
||
*/
|
||
@SuppressWarnings("unchecked")
|
||
private String parsePostContent(String messageId, Map<String, Object> contentObj,
|
||
List<MessageContentPart> parts) {
|
||
// 取 zh_cn 或第一个可用的 locale 分支
|
||
Map<String, Object> localeBranch = (Map<String, Object>) contentObj.get("zh_cn");
|
||
if (localeBranch == null) {
|
||
localeBranch = (Map<String, Object>) contentObj.get("en_us");
|
||
}
|
||
if (localeBranch == null && !contentObj.isEmpty()) {
|
||
// 取第一个 locale
|
||
for (Object val : contentObj.values()) {
|
||
if (val instanceof Map) {
|
||
localeBranch = (Map<String, Object>) val;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
if (localeBranch == null) {
|
||
return null;
|
||
}
|
||
|
||
StringBuilder text = new StringBuilder();
|
||
|
||
// 标题
|
||
String title = (String) localeBranch.get("title");
|
||
if (title != null && !title.isBlank()) {
|
||
text.append(title).append("\n");
|
||
}
|
||
|
||
// 段落内容
|
||
List<List<Map<String, Object>>> paragraphs =
|
||
(List<List<Map<String, Object>>>) localeBranch.get("content");
|
||
if (paragraphs == null) return text.toString().trim();
|
||
|
||
// Same default-on behaviour as the standalone file/audio/video branch — see maybeDownloadResource.
|
||
boolean mediaDownload = getConfigBoolean("media_download_enabled", true);
|
||
|
||
for (int i = 0; i < paragraphs.size(); i++) {
|
||
List<Map<String, Object>> paragraph = paragraphs.get(i);
|
||
if (paragraph == null) continue;
|
||
|
||
for (Map<String, Object> element : paragraph) {
|
||
String tag = (String) element.get("tag");
|
||
if (tag == null) continue;
|
||
|
||
switch (tag) {
|
||
case "text" -> {
|
||
String t = (String) element.get("text");
|
||
if (t != null) text.append(t);
|
||
}
|
||
case "code_block", "md" -> {
|
||
String t = (String) element.get("text");
|
||
if (t != null) text.append(t);
|
||
}
|
||
case "a" -> {
|
||
String linkText = (String) element.get("text");
|
||
String href = (String) element.get("href");
|
||
if (linkText != null && href != null) {
|
||
text.append("[").append(linkText).append("](").append(href).append(")");
|
||
} else if (linkText != null) {
|
||
text.append(linkText);
|
||
} else if (href != null) {
|
||
text.append(href);
|
||
}
|
||
}
|
||
case "at" -> {
|
||
String userName = (String) element.get("user_name");
|
||
String userId = (String) element.get("user_id");
|
||
if (userName != null && !userName.isBlank()) {
|
||
text.append("@").append(userName);
|
||
} else if (userId != null) {
|
||
text.append("@").append(userId);
|
||
}
|
||
}
|
||
case "img" -> {
|
||
String imageKey = (String) element.get("image_key");
|
||
if (imageKey != null) {
|
||
// Same reasoning as the standalone image case: vision
|
||
// pipelines need bytes; image_key alone is opaque.
|
||
DownloadedResource dl = maybeDownloadImage(messageId, imageKey);
|
||
MessageContentPart imgPart = MessageContentPart.image(imageKey, null);
|
||
applyDownload(imgPart, dl);
|
||
parts.add(imgPart);
|
||
text.append("[图片]");
|
||
}
|
||
}
|
||
case "media" -> {
|
||
String fileKey = (String) element.get("file_key");
|
||
if (fileKey != null) {
|
||
DownloadedResource dl = mediaDownload
|
||
? maybeDownloadResource(messageId, fileKey, "file", null)
|
||
: null;
|
||
MessageContentPart mediaPart = MessageContentPart.file(fileKey, null, null);
|
||
applyDownload(mediaPart, dl);
|
||
parts.add(mediaPart);
|
||
text.append("[媒体]");
|
||
}
|
||
}
|
||
default -> {
|
||
String t = (String) element.get("text");
|
||
if (t != null) text.append(t);
|
||
}
|
||
}
|
||
}
|
||
|
||
// 段落间用换行分隔
|
||
if (i < paragraphs.size() - 1) {
|
||
text.append("\n");
|
||
}
|
||
}
|
||
|
||
String result = text.toString().trim();
|
||
if (!result.isEmpty()) {
|
||
parts.add(0, MessageContentPart.text(result));
|
||
}
|
||
return result.isEmpty() ? null : result;
|
||
}
|
||
|
||
// ==================== Inbound media download ====================
|
||
|
||
/**
|
||
* Result of a successful inbound-resource download.
|
||
*
|
||
* @param path absolute path on disk under
|
||
* {@code ~/.mateclaw/media/feishu/} — fed to
|
||
* {@link MessageContentPart#setPath(String)} so
|
||
* vision / file-reading tools can consume bytes
|
||
* @param fileUrl {@code /api/v1/files/generated/{id}} URL backed
|
||
* by {@link vip.mate.tool.document.GeneratedFileCache}
|
||
* (10-min TTL) so the admin UI and conversation
|
||
* memory can render the file without a tenant
|
||
* token; {@code null} when the cache wasn't wired
|
||
* in (e.g. legacy tests)
|
||
* @param fileName server-side filename (extension inferred from
|
||
* content-type, falls back to {@code fileNameHint}
|
||
* or the SDK-returned name)
|
||
* @param contentType MIME type from the download response headers
|
||
*/
|
||
// Package-private for unit tests — pin the field copy contract.
|
||
record DownloadedResource(String path, String fileUrl,
|
||
String fileName, String contentType) {}
|
||
|
||
/**
|
||
* Download a {@code file} / {@code audio} / {@code video} inbound
|
||
* resource, but only when the per-channel
|
||
* {@code media_download_enabled} flag is on. Default is now <b>true</b>
|
||
* so a user uploading a file to the bot lands as a real file the
|
||
* agent can read — set it to {@code false} on the channel config to
|
||
* opt out (e.g. tighter privacy, no disk usage).
|
||
*
|
||
* <p>Mirrors the WeCom / DingTalk pattern: text-only adapters dropping
|
||
* file bytes was the "did you receive my doc?" gap reported in
|
||
* production.
|
||
*/
|
||
private DownloadedResource maybeDownloadResource(String messageId, String fileKey, String type, String fileNameHint) {
|
||
if (!getConfigBoolean("media_download_enabled", true)) {
|
||
return null;
|
||
}
|
||
return downloadResource(messageId, fileKey, type, fileNameHint);
|
||
}
|
||
|
||
/**
|
||
* Image-specific download: default ON so the vision/STT pipeline
|
||
* downstream actually has bytes to analyze. Without the local file,
|
||
* vision providers see only an opaque {@code image_key} and the
|
||
* Feishu CDN URL needs tenant_access_token to fetch — neither of
|
||
* which the model can resolve. Admins who want to suppress image
|
||
* downloads (e.g. tighter privacy, no disk usage) can set
|
||
* {@code feishu_image_download_enabled=false} on the channel config.
|
||
*/
|
||
private DownloadedResource maybeDownloadImage(String messageId, String imageKey) {
|
||
if (!getConfigBoolean("feishu_image_download_enabled", true)) {
|
||
return null;
|
||
}
|
||
return downloadResource(messageId, imageKey, "image", null);
|
||
}
|
||
|
||
/**
|
||
* Copy a {@link DownloadedResource}'s fields onto a
|
||
* {@link MessageContentPart}. {@code null} input is a no-op so the
|
||
* caller can keep the part as a bare {@code file_key} placeholder
|
||
* when download was disabled / failed.
|
||
*/
|
||
/**
|
||
* Read the downloaded audio bytes from disk and run them through
|
||
* {@link vip.mate.stt.SttService}. Returns the transcribed text, or
|
||
* {@code null} when STT is not wired, disabled, the download didn't
|
||
* produce a usable path, or every provider failed.
|
||
*
|
||
* <p>Best-effort by design — STT failure must not block the agent
|
||
* from seeing the audio part. The user gets the "[音频]" placeholder
|
||
* and any agent that's tooled-up can still investigate.
|
||
*/
|
||
// Package-private for unit tests.
|
||
String transcribeInboundAudio(DownloadedResource dl) {
|
||
if (sttService == null || dl == null || dl.path() == null) {
|
||
return null;
|
||
}
|
||
try {
|
||
byte[] audioBytes = Files.readAllBytes(Path.of(dl.path()));
|
||
if (audioBytes.length == 0) {
|
||
log.debug("[feishu-stt] empty audio file at {}, skipping STT", dl.path());
|
||
return null;
|
||
}
|
||
String fileName = dl.fileName() != null ? dl.fileName() : "voice.opus";
|
||
String contentType = dl.contentType() != null ? dl.contentType() : "audio/opus";
|
||
// language=null → SttService falls back to the system-settings
|
||
// UI language hint; works for both Chinese and English without
|
||
// needing per-channel config.
|
||
Map<String, Object> result = sttService.transcribe(
|
||
audioBytes, fileName, contentType, null);
|
||
if (!Boolean.TRUE.equals(result.get("success"))) {
|
||
log.warn("[feishu-stt] transcription failed: {}", result.get("error"));
|
||
return null;
|
||
}
|
||
String text = (String) result.get("text");
|
||
if (text == null || text.isBlank()) {
|
||
log.debug("[feishu-stt] transcription returned empty text");
|
||
return null;
|
||
}
|
||
log.info("[feishu-stt] transcribed {} bytes → {} chars",
|
||
audioBytes.length, text.length());
|
||
return text;
|
||
} catch (Exception e) {
|
||
log.warn("[feishu-stt] transcription threw: {}", e.getMessage());
|
||
return null;
|
||
}
|
||
}
|
||
|
||
// Package-private for unit tests.
|
||
static void applyDownload(MessageContentPart part, DownloadedResource dl) {
|
||
if (dl == null || part == null) return;
|
||
if (dl.path() != null) part.setPath(dl.path());
|
||
if (dl.fileUrl() != null) part.setFileUrl(dl.fileUrl());
|
||
if (dl.fileName() != null && (part.getFileName() == null || part.getFileName().isBlank())) {
|
||
part.setFileName(dl.fileName());
|
||
}
|
||
if (dl.contentType() != null) part.setContentType(dl.contentType());
|
||
}
|
||
|
||
/**
|
||
* Pull an inbound resource (image / file / audio / video) from
|
||
* Feishu and persist it both to disk and (best-effort) to the
|
||
* {@link vip.mate.tool.document.GeneratedFileCache}.
|
||
*
|
||
* <p>Implementation prefers the official {@code oapi-sdk}
|
||
* ({@code client.im().v1().messageResource().get(...)}) when a
|
||
* {@link FeishuClientFactory} was wired in — gets token refresh,
|
||
* domain switching (feishu / lark), retries, and the same
|
||
* {@code Authorization} contract the rest of the SDK uses for free.
|
||
* Falls back to the legacy hand-rolled {@code HttpClient} path
|
||
* when no factory is available (3-arg legacy ctor / unit tests).
|
||
*
|
||
* @param messageId Feishu message id ({@code om_…}) — required
|
||
* by the resource endpoint to scope the file
|
||
* @param fileKey resource key ({@code img_…} for images,
|
||
* {@code file_…} for everything else)
|
||
* @param type SDK type discriminator: {@code "image"} for
|
||
* images, {@code "file"} for file/audio/video
|
||
* @param fileNameHint optional original filename (used for
|
||
* extension inference when the response has no
|
||
* useful Content-Type)
|
||
* @return a {@link DownloadedResource} on success, {@code null} on
|
||
* any failure (logged at debug) — caller treats {@code null}
|
||
* as "skip, just pass the opaque key through"
|
||
*/
|
||
private DownloadedResource downloadResource(String messageId, String fileKey, String type, String fileNameHint) {
|
||
// ---- Prefer SDK path: token refresh, retries, domain handled by the SDK
|
||
if (clientFactory != null && channelEntity != null && channelEntity.getId() != null) {
|
||
try {
|
||
com.lark.oapi.Client client = clientFactory.client(channelEntity.getId());
|
||
com.lark.oapi.service.im.v1.model.GetMessageResourceReq req =
|
||
com.lark.oapi.service.im.v1.model.GetMessageResourceReq.newBuilder()
|
||
.messageId(messageId)
|
||
.fileKey(fileKey)
|
||
.type(type)
|
||
.build();
|
||
com.lark.oapi.service.im.v1.model.GetMessageResourceResp resp =
|
||
client.im().v1().messageResource().get(req);
|
||
if (!resp.success() || resp.getData() == null) {
|
||
log.debug("[feishu] SDK resource.get failed: code={}, msg={}",
|
||
resp.getCode(), resp.getMsg());
|
||
return null;
|
||
}
|
||
byte[] bytes = resp.getData().toByteArray();
|
||
// SDK exposes Content-Disposition's filename via getFileName();
|
||
// prefer it over the hint when present.
|
||
String sdkFileName = resp.getFileName();
|
||
String preferredHint = (sdkFileName != null && !sdkFileName.isBlank())
|
||
? sdkFileName : fileNameHint;
|
||
// No Content-Type accessor on GetMessageResourceResp — infer
|
||
// from filename extension (good enough for cache MIME and
|
||
// for vision pipelines that key on extension).
|
||
String contentType = inferMimeFromName(preferredHint);
|
||
return persistAndCache(messageId, fileKey, bytes, preferredHint, contentType);
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] SDK resource.get threw, falling back to raw HTTP: {}", e.getMessage());
|
||
// Fall through to raw HTTP — keeps the inbound path resilient
|
||
// when the SDK is temporarily unhappy (e.g. invalid req shape
|
||
// on a new event type the SDK doesn't yet model).
|
||
}
|
||
}
|
||
|
||
// ---- Legacy fallback: hand-rolled HTTP. Kept so the 3-arg ctor
|
||
// and unit tests that don't wire a factory still get bytes.
|
||
try {
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages/" + messageId
|
||
+ "/resources/" + fileKey + "?type=" + type))
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.timeout(Duration.ofSeconds(30))
|
||
.GET()
|
||
.build();
|
||
|
||
HttpResponse<InputStream> response = httpClient.send(request,
|
||
HttpResponse.BodyHandlers.ofInputStream());
|
||
|
||
// BodyHandlers.ofInputStream does NOT auto-close the response body —
|
||
// callers must consume or close it on every path, including non-2xx
|
||
// and early-return-after-content-type. Use try-with-resources around
|
||
// the entire post-send block so failed downloads don't leak file
|
||
// descriptors during sustained traffic.
|
||
try (InputStream is = response.body()) {
|
||
if (response.statusCode() != 200) {
|
||
log.debug("[feishu] Download resource failed: status={}", response.statusCode());
|
||
return null;
|
||
}
|
||
String contentType = response.headers().firstValue("Content-Type")
|
||
.orElse(inferMimeFromName(fileNameHint));
|
||
byte[] bytes = is.readAllBytes();
|
||
return persistAndCache(messageId, fileKey, bytes, fileNameHint, contentType);
|
||
}
|
||
|
||
} catch (Exception e) {
|
||
log.debug("[feishu] Download resource failed: {}", e.getMessage());
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Write {@code bytes} to {@code ~/.mateclaw/media/feishu/} and,
|
||
* if {@link #generatedFileCache} is wired, also push them into the
|
||
* cache so the UI gets a {@code /api/v1/files/generated/{id}} URL.
|
||
* Disk persist is always attempted; cache push is best-effort.
|
||
*/
|
||
private DownloadedResource persistAndCache(String messageId, String fileKey,
|
||
byte[] bytes, String fileNameHint,
|
||
String contentType) throws java.io.IOException {
|
||
Path mediaDir = Path.of(System.getProperty("user.home"), ".mateclaw", "media", "feishu");
|
||
Files.createDirectories(mediaDir);
|
||
|
||
String safeKey = fileKey.replaceAll("[^a-zA-Z0-9_]", "");
|
||
if (safeKey.isEmpty()) safeKey = "file";
|
||
|
||
String ext = extensionFor(contentType, fileNameHint);
|
||
String resolvedFileName = (fileNameHint != null && !fileNameHint.isBlank())
|
||
? fileNameHint
|
||
: (safeKey + "." + ext);
|
||
|
||
Path filePath = mediaDir.resolve(messageId + "_" + safeKey + "." + ext);
|
||
Files.write(filePath, bytes);
|
||
log.debug("[feishu] Downloaded resource to: {} ({} bytes, {})",
|
||
filePath, bytes.length, contentType);
|
||
|
||
// Best-effort cache push so the UI / agent can reference the file
|
||
// via a tokenless URL. Failure here doesn't kill the inbound path —
|
||
// the local path on its own is still useful for vision tools.
|
||
String fileUrl = null;
|
||
if (generatedFileCache != null) {
|
||
try {
|
||
String id = generatedFileCache.put(bytes, resolvedFileName, contentType);
|
||
fileUrl = "/api/v1/files/generated/" + id;
|
||
} catch (Exception cacheEx) {
|
||
log.debug("[feishu] cache put failed: {}", cacheEx.getMessage());
|
||
}
|
||
}
|
||
return new DownloadedResource(filePath.toAbsolutePath().toString(),
|
||
fileUrl, resolvedFileName, contentType);
|
||
}
|
||
|
||
/** Map content-type / file-name hint to a short extension for the on-disk filename. */
|
||
// Package-private for unit tests.
|
||
static String extensionFor(String contentType, String fileNameHint) {
|
||
if (contentType != null) {
|
||
String ct = contentType.toLowerCase(java.util.Locale.ROOT);
|
||
if (ct.contains("jpeg") || ct.contains("jpg")) return "jpg";
|
||
if (ct.contains("png")) return "png";
|
||
if (ct.contains("gif")) return "gif";
|
||
if (ct.contains("webp")) return "webp";
|
||
if (ct.contains("pdf")) return "pdf";
|
||
if (ct.contains("opus")) return "opus";
|
||
if (ct.contains("mpeg") || ct.contains("mp3")) return "mp3";
|
||
if (ct.contains("mp4")) return "mp4";
|
||
if (ct.contains("wordprocessing") || ct.contains("msword")) return "docx";
|
||
if (ct.contains("spreadsheet") || ct.contains("excel")) return "xlsx";
|
||
if (ct.contains("presentation") || ct.contains("powerpoint")) return "pptx";
|
||
}
|
||
if (fileNameHint != null && fileNameHint.contains(".")) {
|
||
String hint = fileNameHint.substring(fileNameHint.lastIndexOf('.') + 1)
|
||
.toLowerCase(java.util.Locale.ROOT);
|
||
if (!hint.isBlank()) return hint;
|
||
}
|
||
return "bin";
|
||
}
|
||
|
||
/**
|
||
* Best-effort MIME guess from a filename — the SDK download response
|
||
* doesn't expose Content-Type, so we fall back to the extension. Good
|
||
* enough for the cache (used for the {@code Content-Type} header the
|
||
* UI sets when re-serving) and for vision providers (most key on
|
||
* extension anyway).
|
||
*/
|
||
// Package-private for unit tests.
|
||
static String inferMimeFromName(String fileName) {
|
||
if (fileName == null) return "application/octet-stream";
|
||
String lower = fileName.toLowerCase(java.util.Locale.ROOT);
|
||
if (lower.endsWith(".jpg") || lower.endsWith(".jpeg")) return "image/jpeg";
|
||
if (lower.endsWith(".png")) return "image/png";
|
||
if (lower.endsWith(".gif")) return "image/gif";
|
||
if (lower.endsWith(".webp")) return "image/webp";
|
||
if (lower.endsWith(".pdf")) return "application/pdf";
|
||
if (lower.endsWith(".opus")) return "audio/opus";
|
||
if (lower.endsWith(".mp3")) return "audio/mpeg";
|
||
if (lower.endsWith(".mp4")) return "video/mp4";
|
||
if (lower.endsWith(".docx")) return "application/vnd.openxmlformats-officedocument.wordprocessingml.document";
|
||
if (lower.endsWith(".xlsx")) return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
|
||
if (lower.endsWith(".pptx")) return "application/vnd.openxmlformats-officedocument.presentationml.presentation";
|
||
if (lower.endsWith(".txt") || lower.endsWith(".md")) return "text/plain";
|
||
return "application/octet-stream";
|
||
}
|
||
|
||
// ==================== 消息发送 ====================
|
||
|
||
/**
|
||
* Maps a Feishu target identifier to the corresponding
|
||
* {@code receive_id_type} query parameter expected by the Open API.
|
||
* <p>
|
||
* Targets prefixed with {@code "ou_"} are user open_ids; everything else
|
||
* (group chat_ids start with {@code "oc_"}, but any non-{@code ou_} value
|
||
* is treated as a chat_id by default) is sent with
|
||
* {@code receive_id_type=chat_id}. Keeping this routing in one place
|
||
* keeps {@link #sendOneTextChunk}, {@link #sendCard},
|
||
* {@link #sendFeishuMedia} and {@link #proactiveSend} from drifting —
|
||
* silently misrouting the text-fallback after a failed card send to an
|
||
* individual was the previous regression.
|
||
*/
|
||
private static String resolveReceiveIdType(String targetId) {
|
||
return targetId != null && targetId.startsWith("ou_") ? "open_id" : "chat_id";
|
||
}
|
||
|
||
/**
|
||
* Conservative per-message char ceiling. Feishu's documented limit is on
|
||
* the encoded JSON body (~150KB), but UTF-8 Chinese is 3 bytes/char and
|
||
* JSON escape adds further overhead, so a 4000-char chunk is comfortably
|
||
* under any realistic limit and also gives readable IM chunking instead
|
||
* of one wall of text. Split at paragraph / line boundaries when possible.
|
||
*/
|
||
static final int MAX_TEXT_MESSAGE_CHARS = 4000;
|
||
|
||
// ==================== StreamingChannelAdapter ====================
|
||
|
||
/**
|
||
* Stream consumption strategy:
|
||
* <ul>
|
||
* <li>{@code card_streaming_enabled=true} (default) + manager wired →
|
||
* create a {@code cardkit/v1} streaming card, throttle-update its
|
||
* markdown element on every delta, finalise on completion. The
|
||
* receiver sees text appearing character-by-character like a
|
||
* chat-app's typing animation.</li>
|
||
* <li>Manager missing OR card creation fails OR config opts out →
|
||
* fall back to {@link #processStreamAsText}: accumulate every
|
||
* chunk and send one final regular message.</li>
|
||
* </ul>
|
||
*/
|
||
@Override
|
||
public String processStream(Flux<StreamDelta> stream, ChannelMessage message, String conversationId) {
|
||
if (!isCardStreamingEnabled() || streamingCardManager == null
|
||
|| channelEntity == null) {
|
||
return processStreamAsText(stream, message);
|
||
}
|
||
|
||
String receiveId = pickReceiveId(message);
|
||
if (receiveId == null) {
|
||
log.warn("[feishu-stream] No usable receive id on message; falling back to text mode");
|
||
return processStreamAsText(stream, message);
|
||
}
|
||
String receiveIdType = resolveReceiveIdType(receiveId);
|
||
|
||
String sessionKey = streamingCardManager.createAndDeliver(
|
||
channelEntity.getId(), receiveIdType, receiveId, null);
|
||
if (sessionKey == null) {
|
||
log.info("[feishu-stream] Card create/deliver failed; falling back to text mode");
|
||
return processStreamAsText(stream, message);
|
||
}
|
||
|
||
StringBuilder accumulator = new StringBuilder();
|
||
try {
|
||
stream.doOnNext(delta -> {
|
||
if (delta.content() != null) {
|
||
accumulator.append(delta.content());
|
||
streamingCardManager.appendContent(sessionKey, delta.content(), false);
|
||
}
|
||
})
|
||
.doOnError(err -> {
|
||
log.error("[feishu-stream] stream error: sessionKey={}, err={}",
|
||
sessionKey, err.getMessage());
|
||
streamingCardManager.failCard(sessionKey, err.getMessage());
|
||
})
|
||
.blockLast(Duration.ofMinutes(5));
|
||
|
||
String finalContent = accumulator.toString();
|
||
if (finalContent.isBlank()) {
|
||
finalContent = "（无回复内容）";
|
||
}
|
||
// Strip any /api/v1/files/generated/{id} URLs out of the card
|
||
// text (replacing each with a "📎 filename" marker) AND send
|
||
// each cache-resolved file as a native Feishu attachment. The
|
||
// streaming card can only render markdown — without this hop
|
||
// the user sees a broken-looking download link instead of the
|
||
// actual file. Cache-miss URLs fall back to the user-facing
|
||
// retry hint that GeneratedFileScrubber emits.
|
||
String renderedContent = scrubAndSendAttachments(receiveId, finalContent);
|
||
streamingCardManager.finishCard(sessionKey, renderedContent);
|
||
log.info("[feishu-stream] Card streaming completed: sessionKey={}, contentLen={}",
|
||
sessionKey, renderedContent.length());
|
||
return finalContent;
|
||
|
||
} catch (Exception e) {
|
||
log.error("[feishu-stream] Card streaming failed: sessionKey={}, err={}",
|
||
sessionKey, e.getMessage(), e);
|
||
streamingCardManager.failCard(sessionKey, e.getMessage());
|
||
|
||
// Tag returned content with the "[错误] " prefix so
|
||
// ChannelMessageRouter.isErrorReply flips status='error' on the
|
||
// persisted row and BaseAgent.sanitizeForLlm filters it out of
|
||
// the next turn's history. Without this, partial streaming
|
||
// output (e.g. LLM 400'd mid-stream) would re-enter the prompt
|
||
// as a valid assistant turn and re-trigger the same 400.
|
||
String partial = accumulator.toString();
|
||
String errorPrefix = "[错误] Feishu CardKit streaming failed: " + e.getMessage();
|
||
if (!partial.isBlank()) {
|
||
return errorPrefix + "\n\n（已生成的部分内容，已忽略）\n" + partial;
|
||
}
|
||
throw new RuntimeException(errorPrefix, e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Streaming fallback — accumulate all deltas, then send through the
|
||
* existing {@link #sendMessage} path so the message goes out as a
|
||
* regular text bubble (auto-upgraded to a non-streaming card by the
|
||
* existing {@code card_format} logic when content looks card-worthy).
|
||
*/
|
||
private String processStreamAsText(Flux<StreamDelta> stream, ChannelMessage message) {
|
||
StringBuilder accumulator = new StringBuilder();
|
||
stream.doOnNext(delta -> {
|
||
if (delta.content() != null) {
|
||
accumulator.append(delta.content());
|
||
}
|
||
})
|
||
.blockLast(Duration.ofMinutes(5));
|
||
String finalContent = accumulator.toString();
|
||
if (!finalContent.isBlank()) {
|
||
String replyTarget = message.getReplyToken() != null
|
||
? message.getReplyToken()
|
||
: (message.getChatId() != null ? message.getChatId() : message.getSenderId());
|
||
if (replyTarget != null) {
|
||
// Same scrub-and-upload hop as the streaming card finish path —
|
||
// a generated-file URL in plain text would otherwise reach the
|
||
// user as a markdown link that opens to nothing useful in IM.
|
||
String renderedContent = scrubAndSendAttachments(replyTarget, finalContent);
|
||
sendMessage(replyTarget, renderedContent);
|
||
}
|
||
}
|
||
return finalContent;
|
||
}
|
||
|
||
/**
|
||
* Run {@link GeneratedFileScrubber} over outbound text. For every
|
||
* {@code /api/v1/files/generated/{id}} URL the cache resolves, send
|
||
* the bytes as a native Feishu attachment to {@code targetId} via
|
||
* the existing upload path. Returns the rewritten text (URLs replaced
|
||
* with {@code "📎 filename"} markers for hits, retry hints for
|
||
* cache-misses) so the caller can use it for the text bubble / card.
|
||
*
|
||
* <p>No-op when scrubber / uploader / channel entity is missing —
|
||
* returns input unchanged so legacy callers (3-arg ctor, tests)
|
||
* continue to behave like the pre-scrubber adapter. The cost of the
|
||
* regex scan on a normal reply with no generated URLs is one
|
||
* Matcher.find() that returns false, negligible.
|
||
*/
|
||
// Package-private so tests can call directly without driving the full
|
||
// streaming pipeline. Production callers are processStream and
|
||
// processStreamAsText, both internal.
|
||
String scrubAndSendAttachments(String targetId, String content) {
|
||
if (content == null || content.isEmpty()
|
||
|| generatedFileScrubber == null
|
||
|| mediaUploader == null
|
||
|| channelEntity == null
|
||
|| channelEntity.getId() == null) {
|
||
return content;
|
||
}
|
||
GeneratedFileScrubber.ScrubResult scrubbed = generatedFileScrubber.scrub(content);
|
||
Long channelId = channelEntity.getId();
|
||
for (GeneratedFileScrubber.AttachmentHit hit : scrubbed.attachments()) {
|
||
uploadAndSendAttachment(targetId, channelId,
|
||
new MediaSource.Bytes(hit.bytes()),
|
||
hit.fileName(),
|
||
hit.mediaType(),
|
||
hit.mimeType(),
|
||
null);
|
||
}
|
||
return scrubbed.rewrittenText();
|
||
}
|
||
|
||
/** Resolve the best id to receive a streaming card — prefer reply token, then chat, then sender. */
|
||
private static String pickReceiveId(ChannelMessage message) {
|
||
if (message == null) return null;
|
||
if (message.getReplyToken() != null && !message.getReplyToken().isBlank()) {
|
||
return message.getReplyToken();
|
||
}
|
||
if (message.getChatId() != null && !message.getChatId().isBlank()) {
|
||
return message.getChatId();
|
||
}
|
||
if (message.getSenderId() != null && !message.getSenderId().isBlank()) {
|
||
return message.getSenderId();
|
||
}
|
||
return null;
|
||
}
|
||
|
||
private boolean isCardStreamingEnabled() {
|
||
// Default true — streaming cards are the better UX when CardKit is
|
||
// available. Operators can flip card_streaming_enabled=false in
|
||
// configJson to fall back to the text path (useful for debugging
|
||
// or when targeting an old Feishu tenant that hasn't rolled out
|
||
// CardKit v1 universally).
|
||
return getConfigBoolean("card_streaming_enabled", true);
|
||
}
|
||
|
||
@Override
|
||
public void sendMessage(String targetId, String content) {
|
||
if (httpClient == null) {
|
||
log.warn("[feishu] Channel not started, cannot send message");
|
||
return;
|
||
}
|
||
if (content == null) {
|
||
return;
|
||
}
|
||
|
||
ensureTokenValid();
|
||
|
||
String cardFormat = getConfigString("card_format", "auto");
|
||
|
||
if ("never".equals(cardFormat)) {
|
||
splitTextForFeishu(content, MAX_TEXT_MESSAGE_CHARS)
|
||
.forEach(c -> sendOneTextChunk(targetId, c));
|
||
return;
|
||
}
|
||
|
||
FeishuCardFormatter.ContentFormat fmt = FeishuCardFormatter.detect(content);
|
||
|
||
boolean preferCard = "always".equals(cardFormat) || fmt != FeishuCardFormatter.ContentFormat.PLAIN_TEXT;
|
||
if (preferCard) {
|
||
String cardHeader = getConfigString("card_header", FeishuCardFormatter.DEFAULT_MARKDOWN_HEADER);
|
||
boolean sent = sendCard(targetId, FeishuCardFormatter.render(content, fmt, cardHeader));
|
||
if (sent) return;
|
||
// Card path bailed (oversized payload, null guard, or network error) —
|
||
// fall through to text so the user doesn't get nothing.
|
||
}
|
||
splitTextForFeishu(content, MAX_TEXT_MESSAGE_CHARS)
|
||
.forEach(c -> sendOneTextChunk(targetId, c));
|
||
}
|
||
|
||
private void sendOneTextChunk(String targetId, String content) {
|
||
String apiBase = getApiBaseUrl();
|
||
String receiveIdType = resolveReceiveIdType(targetId);
|
||
try {
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"receive_id", targetId,
|
||
"msg_type", "text",
|
||
"content", objectMapper.writeValueAsString(Map.of("text", content))
|
||
));
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages?receive_id_type=" + receiveIdType))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.warn("[feishu] Send message failed: status={}, body={}", response.statusCode(), response.body());
|
||
} else {
|
||
log.debug("[feishu] Message sent to {}={} ({} chars)", receiveIdType, targetId, content.length());
|
||
}
|
||
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to send message: {}", e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Feishu rejects interactive messages whose {@code content} payload exceeds
|
||
* roughly 30 KB once stringified. We compare the serialized card against a
|
||
* slightly tighter ceiling and fall back to plain text rather than letting
|
||
* the request error out at the API.
|
||
*/
|
||
static final int MAX_CARD_CONTENT_BYTES = 30_000;
|
||
|
||
/**
|
||
* Posts an Interactive Card. Returns {@code true} when the card was sent
|
||
* (HTTP 2xx), {@code false} when a pre-flight check rejected the call
|
||
* (null inputs, channel stopped, payload oversized) or the HTTP request
|
||
* itself failed — letting {@link #sendMessage(String, String)} fall back
|
||
* to plain text instead of going silent.
|
||
*
|
||
* <p>{@code targetId} format follows {@link #resolveReceiveIdType}:
|
||
* {@code "ou_"} prefix → open_id (1:1), anything else → chat_id (group).
|
||
*/
|
||
public boolean sendCard(String targetId, Map<String, Object> cardJson) {
|
||
if (httpClient == null) {
|
||
log.warn("[feishu] Channel not started, cannot send card");
|
||
return false;
|
||
}
|
||
if (targetId == null || cardJson == null) {
|
||
log.warn("[feishu] sendCard called with null target or card");
|
||
return false;
|
||
}
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
String receiveIdType = resolveReceiveIdType(targetId);
|
||
try {
|
||
String cardContent = objectMapper.writeValueAsString(cardJson);
|
||
int cardBytes = cardContent.getBytes(java.nio.charset.StandardCharsets.UTF_8).length;
|
||
if (cardBytes > MAX_CARD_CONTENT_BYTES) {
|
||
log.warn("[feishu] Card content {} bytes exceeds {} byte limit, falling back to text",
|
||
cardBytes, MAX_CARD_CONTENT_BYTES);
|
||
return false;
|
||
}
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"receive_id", targetId,
|
||
"msg_type", "interactive",
|
||
"content", cardContent
|
||
));
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages?receive_id_type=" + receiveIdType))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.warn("[feishu] Send card failed: status={}, body={}", response.statusCode(), response.body());
|
||
return false;
|
||
}
|
||
log.debug("[feishu] Card sent to {} (type={})", targetId, receiveIdType);
|
||
return true;
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to send card: {}", e.getMessage(), e);
|
||
return false;
|
||
}
|
||
}
|
||
|
||
public void updateCard(String messageId, Map<String, Object> cardJson) {
|
||
if (httpClient == null) {
|
||
log.warn("[feishu] Channel not started, cannot update card");
|
||
return;
|
||
}
|
||
if (messageId == null || cardJson == null) {
|
||
log.warn("[feishu] updateCard called with null messageId or card");
|
||
return;
|
||
}
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
try {
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"msg_type", "interactive",
|
||
"content", objectMapper.writeValueAsString(cardJson)
|
||
));
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages/" + messageId))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.method("PATCH", HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.warn("[feishu] Update card failed: status={}, body={}", response.statusCode(), response.body());
|
||
} else {
|
||
log.debug("[feishu] Card updated: messageId={}", messageId);
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to update card: {}", e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Split a possibly-oversized message into chunks no larger than
|
||
* {@code maxChars}, preferring paragraph (\n\n) then line (\n) then
|
||
* whitespace boundaries. Hard-cuts as a last resort so we never silently
|
||
* drop content.
|
||
*/
|
||
static List<String> splitTextForFeishu(String content, int maxChars) {
|
||
if (content == null || content.isEmpty()) {
|
||
return List.of();
|
||
}
|
||
if (content.length() <= maxChars) {
|
||
return List.of(content);
|
||
}
|
||
List<String> chunks = new ArrayList<>();
|
||
int idx = 0;
|
||
int n = content.length();
|
||
while (idx < n) {
|
||
int end = Math.min(idx + maxChars, n);
|
||
if (end < n) {
|
||
int boundary = -1;
|
||
int paragraphCut = content.lastIndexOf("\n\n", end);
|
||
if (paragraphCut > idx + maxChars / 2) {
|
||
boundary = paragraphCut + 2;
|
||
}
|
||
if (boundary < 0) {
|
||
int lineCut = content.lastIndexOf('\n', end);
|
||
if (lineCut > idx + maxChars / 2) {
|
||
boundary = lineCut + 1;
|
||
}
|
||
}
|
||
if (boundary < 0) {
|
||
int spaceCut = content.lastIndexOf(' ', end);
|
||
if (spaceCut > idx + maxChars / 2) {
|
||
boundary = spaceCut + 1;
|
||
}
|
||
}
|
||
if (boundary > idx) {
|
||
end = boundary;
|
||
}
|
||
}
|
||
chunks.add(content.substring(idx, end));
|
||
idx = end;
|
||
}
|
||
return chunks;
|
||
}
|
||
|
||
@Override
|
||
public void sendContentParts(String targetId, List<MessageContentPart> parts) {
|
||
if (httpClient == null) {
|
||
log.warn("[feishu] Channel not started, cannot send message");
|
||
return;
|
||
}
|
||
if (parts == null || parts.isEmpty()) return;
|
||
|
||
ensureTokenValid();
|
||
|
||
// Carry an explicit channelId for media uploads. Available on
|
||
// the entity since CRUD; only null in degenerate test stubs.
|
||
Long channelId = channelEntity != null ? channelEntity.getId() : null;
|
||
|
||
for (MessageContentPart part : parts) {
|
||
if (part == null) continue;
|
||
try {
|
||
switch (part.getType()) {
|
||
case "text" -> handleTextPart(targetId, channelId, part);
|
||
case "refusal" -> {
|
||
String refusal = part.getText();
|
||
if (refusal != null && !refusal.isBlank()) {
|
||
sendMessage(targetId, "⚠️ " + refusal);
|
||
}
|
||
}
|
||
case "image" -> handleMediaPart(targetId, channelId, part, "image", "image.jpg");
|
||
case "file" -> handleMediaPart(targetId, channelId, part, "file", "file.bin");
|
||
case "audio" -> handleMediaPart(targetId, channelId, part, "audio", "voice_reply.opus");
|
||
case "video" -> handleMediaPart(targetId, channelId, part, "video", "video.mp4");
|
||
default -> {
|
||
if (part.getText() != null) sendMessage(targetId, part.getText());
|
||
}
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to send content part ({}): {}", part.getType(), e.getMessage());
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Handle one text part: if the scrubber is wired, rewrite any
|
||
* {@code /api/v1/files/generated/{id}} URL to a {@code "📎 filename"}
|
||
* marker and upload the cached bytes as a native attachment; if not,
|
||
* fall back to sending the text as-is (legacy behavior).
|
||
*/
|
||
private void handleTextPart(String targetId, Long channelId, MessageContentPart part) {
|
||
String text = part.getText() != null ? part.getText() : "";
|
||
if (generatedFileScrubber == null || mediaUploader == null || channelId == null) {
|
||
sendMessage(targetId, text);
|
||
return;
|
||
}
|
||
GeneratedFileScrubber.ScrubResult scrubbed = generatedFileScrubber.scrub(text);
|
||
sendMessage(targetId, scrubbed.rewrittenText());
|
||
for (GeneratedFileScrubber.AttachmentHit hit : scrubbed.attachments()) {
|
||
uploadAndSendAttachment(targetId, channelId,
|
||
new MediaSource.Bytes(hit.bytes()),
|
||
hit.fileName(),
|
||
hit.mediaType(),
|
||
hit.mimeType(),
|
||
null);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Handle one image / file / audio / video part. If {@code mediaId} is
|
||
* already a Feishu key (image_key / file_key) — e.g. echoed back
|
||
* from an earlier inbound message — send directly. Otherwise build a
|
||
* {@link MediaSource} from {@code path} / {@code fileUrl} /
|
||
* {@code text}(base64 not yet supported here) and upload via the SDK
|
||
* uploader first.
|
||
*/
|
||
private void handleMediaPart(String targetId, Long channelId, MessageContentPart part,
|
||
String mediaType, String defaultFileName) {
|
||
// 1. mediaId is already a Feishu key → send straight away
|
||
String existingKey = part.getMediaId();
|
||
if (existingKey != null && (existingKey.startsWith("img_") || existingKey.startsWith("file_"))) {
|
||
String keyField = "image".equals(mediaType) ? "image_key" : "file_key";
|
||
sendFeishuMedia(targetId, mediaType, Map.of(keyField, existingKey));
|
||
return;
|
||
}
|
||
if (mediaUploader == null || channelId == null) {
|
||
sendFallbackText(targetId, part, mediaType);
|
||
return;
|
||
}
|
||
MediaSource source = resolveSource(part);
|
||
if (source == null) {
|
||
sendFallbackText(targetId, part, mediaType);
|
||
return;
|
||
}
|
||
String fileName = part.getFileName() != null ? part.getFileName() : defaultFileName;
|
||
uploadAndSendAttachment(targetId, channelId, source, fileName,
|
||
mediaType, part.getContentType(), null);
|
||
}
|
||
|
||
/**
|
||
* Resolve a {@link MessageContentPart}'s payload source — prefer
|
||
* already-on-disk path, fall back to fileUrl. Returns {@code null}
|
||
* when neither is usable (e.g. the part only carries a placeholder).
|
||
*/
|
||
private static MediaSource resolveSource(MessageContentPart part) {
|
||
if (part.getPath() != null && !part.getPath().isBlank()) {
|
||
java.nio.file.Path p = java.nio.file.Path.of(part.getPath());
|
||
if (java.nio.file.Files.exists(p)) {
|
||
return new MediaSource.LocalPath(p);
|
||
}
|
||
}
|
||
String url = part.getFileUrl();
|
||
if (url != null && !url.isBlank()) {
|
||
return new MediaSource.RemoteUrl(url);
|
||
}
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* Upload via the SDK-backed uploader and dispatch the resulting
|
||
* key as a Feishu media message. Surfaces any rejection /
|
||
* downgrade note as a follow-up text bubble so users understand
|
||
* why a bubble isn't native.
|
||
*/
|
||
// Package-private + non-final so tests can override and capture upload
|
||
// attempts without standing up a real Feishu HTTP client.
|
||
void uploadAndSendAttachment(String targetId, Long channelId,
|
||
MediaSource source, String fileName,
|
||
String mediaType, String contentType,
|
||
Integer durationMillis) {
|
||
try {
|
||
MediaUploadResult result = mediaUploader.upload(new MediaUploadRequest(
|
||
channelId, source, fileName, mediaType, contentType, durationMillis));
|
||
String finalType = result.finalMediaType();
|
||
String keyField = "image".equals(finalType) ? "image_key" : "file_key";
|
||
sendFeishuMedia(targetId, finalType, Map.of(keyField, result.mediaId()));
|
||
if (result.downgradeNote() != null) {
|
||
sendMessage(targetId, "ℹ️ " + result.downgradeNote());
|
||
}
|
||
} catch (MediaUploadException e) {
|
||
log.warn("[feishu] Upload rejected for {} ({}): {}", fileName, mediaType, e.getMessage());
|
||
sendMessage(targetId, "⚠️ " + e.getMessage());
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Upload failed for {} ({}): {}", fileName, mediaType, e.getMessage(), e);
|
||
sendMessage(targetId, "⚠️ 附件 " + fileName + " 发送失败：" + e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Last-ditch text representation when neither a Feishu key nor a
|
||
* usable source could be derived from the part. Keeps the bubble
|
||
* informative instead of swallowing the message.
|
||
*/
|
||
private void sendFallbackText(String targetId, MessageContentPart part, String mediaType) {
|
||
StringBuilder sb = new StringBuilder("⚠️ 无法发送 ");
|
||
sb.append(mediaType);
|
||
if (part.getFileName() != null) sb.append("：").append(part.getFileName());
|
||
if (part.getFileUrl() != null) sb.append(" (").append(part.getFileUrl()).append(")");
|
||
sendMessage(targetId, sb.toString());
|
||
}
|
||
|
||
private void sendFeishuMedia(String targetId, String msgType, Map<String, Object> content) {
|
||
String apiBase = getApiBaseUrl();
|
||
String receiveIdType = resolveReceiveIdType(targetId);
|
||
try {
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"receive_id", targetId,
|
||
"msg_type", msgType,
|
||
"content", objectMapper.writeValueAsString(content)
|
||
));
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages?receive_id_type=" + receiveIdType))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.warn("[feishu] Send {} failed: status={}, body={}", msgType, response.statusCode(), response.body());
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to send {}: {}", msgType, e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
// ==================== 主动推送 ====================
|
||
|
||
@Override
|
||
public boolean supportsProactiveSend() {
|
||
return true;
|
||
}
|
||
|
||
/**
|
||
* 主动推送消息
|
||
* <p>
|
||
* targetId 可以是：
|
||
* - chat_id（以 oc_ 开头）：发送到群聊
|
||
* - open_id（以 ou_ 开头）：发送到个人
|
||
* - 其他：默认按 chat_id 处理
|
||
*/
|
||
@Override
|
||
public void proactiveSend(String targetId, String content) {
|
||
if (httpClient == null) {
|
||
log.warn("[feishu] Channel not started, cannot proactive send");
|
||
return;
|
||
}
|
||
|
||
ensureTokenValid();
|
||
String apiBase = getApiBaseUrl();
|
||
String receiveIdType = resolveReceiveIdType(targetId);
|
||
|
||
try {
|
||
String jsonBody = objectMapper.writeValueAsString(Map.of(
|
||
"receive_id", targetId,
|
||
"msg_type", "text",
|
||
"content", objectMapper.writeValueAsString(Map.of("text", content))
|
||
));
|
||
|
||
HttpRequest request = HttpRequest.newBuilder()
|
||
.uri(URI.create(apiBase + "/open-apis/im/v1/messages?receive_id_type=" + receiveIdType))
|
||
.header("Content-Type", "application/json; charset=utf-8")
|
||
.header("Authorization", "Bearer " + tenantAccessToken)
|
||
.POST(HttpRequest.BodyPublishers.ofString(jsonBody))
|
||
.build();
|
||
|
||
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
|
||
if (response.statusCode() != 200) {
|
||
log.warn("[feishu] Proactive send failed: status={}, body={}", response.statusCode(), response.body());
|
||
} else {
|
||
log.debug("[feishu] Proactive message sent to {} (type={})", targetId, receiveIdType);
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("[feishu] Failed to proactive send: {}", e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
@Override
|
||
public String getChannelType() {
|
||
return CHANNEL_TYPE;
|
||
}
|
||
|
||
/**
|
||
* WebSocket mode opens a long-lived connection to Lark's gateway, which
|
||
* caps concurrent connections per bot app (~2). In a multi-instance
|
||
* deployment every node would race for that quota and reconnect-loop on
|
||
* {@code 1000040350: the number of connections exceeded the limit}.
|
||
* The leader gate ensures only one node holds the connection at a time.
|
||
*
|
||
* <p>Webhook mode is exempt: callbacks are HTTP-fanned by the load
|
||
* balancer, so all nodes can safely subscribe.
|
||
*/
|
||
@Override
|
||
public boolean requiresSingleLeader() {
|
||
return "websocket".equals(getConfigString("connection_mode", "websocket"));
|
||
}
|
||
}
|