mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(channel): claim each inbound message once so redeliveries stop answering twice
This commit is contained in:
matevip
parent
9bc4741aa7
commit
aaae3bf122
@ -0,0 +1,79 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
/**
|
||||
* Tunables for inbound channel-message deduplication.
|
||||
*
|
||||
* <p>IM platforms redeliver the same message when an acknowledgement is late,
|
||||
* lost, or answered with a non-200 — DingTalk, WeCom and Feishu all do this.
|
||||
* Every redelivery that reaches the router starts a full, independent agent
|
||||
* turn, so the user sees the same answer twice (and the conversation gains a
|
||||
* duplicate user/assistant pair). {@link InboundMessageDeduplicator} keeps a
|
||||
* short-lived record of the message identities already claimed so a
|
||||
* redelivery is dropped instead of answered again.
|
||||
*
|
||||
* <p>入站渠道消息去重配置。平台重投同一条消息时，若不去重则每次重投都会跑一轮完整
|
||||
* 的 Agent 回合，用户看到重复答复。
|
||||
*
|
||||
* <pre>
|
||||
* mate:
|
||||
* channel:
|
||||
* dedup:
|
||||
* enabled: true
|
||||
* ttl: 5m
|
||||
* max-size: 2000
|
||||
* </pre>
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "mate.channel.dedup")
|
||||
public class ChannelDedupProperties {
|
||||
|
||||
/**
|
||||
* Master switch. When false every message is treated as new — only useful
|
||||
* when debugging a suspected false-positive drop.
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
/**
|
||||
* How long a claimed message identity keeps suppressing redeliveries.
|
||||
*
|
||||
* <p>Must comfortably exceed the platforms' redelivery windows (seconds to
|
||||
* low minutes) while staying short enough that a user who genuinely resends
|
||||
* the identical payload later is not silenced. Note that a resend carries a
|
||||
* fresh platform message id in every channel we support, so the TTL only
|
||||
* matters for the id-less fallback identity.
|
||||
*/
|
||||
private Duration ttl = Duration.ofMinutes(5);
|
||||
|
||||
/**
|
||||
* Hard cap on tracked identities. Reached only under sustained traffic
|
||||
* within one TTL window; the oldest claims are dropped first.
|
||||
*/
|
||||
private int maxSize = 2000;
|
||||
|
||||
public boolean isEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
public Duration getTtl() {
|
||||
return ttl;
|
||||
}
|
||||
|
||||
public void setTtl(Duration ttl) {
|
||||
this.ttl = ttl;
|
||||
}
|
||||
|
||||
public int getMaxSize() {
|
||||
return maxSize;
|
||||
}
|
||||
|
||||
public void setMaxSize(int maxSize) {
|
||||
this.maxSize = maxSize;
|
||||
}
|
||||
}
|
||||
@ -70,6 +70,7 @@ public class ChannelMessageRouter {
|
||||
private final ChatStreamTracker streamTracker;
|
||||
private final ChannelChatOriginFactory chatOriginFactory;
|
||||
private final ChannelErrorClassifier errorClassifier;
|
||||
private final InboundMessageDeduplicator inboundDedup;
|
||||
/** Field-injected (rather than constructor) to avoid a signature
|
||||
* change that would ripple through every test that constructs the
|
||||
* router directly. Spring's stock publisher is always available. */
|
||||
@ -183,7 +184,8 @@ public class ChannelMessageRouter {
|
||||
ObjectMapper objectMapper,
|
||||
ChatStreamTracker streamTracker,
|
||||
ChannelChatOriginFactory chatOriginFactory,
|
||||
ChannelErrorClassifier errorClassifier) {
|
||||
ChannelErrorClassifier errorClassifier,
|
||||
InboundMessageDeduplicator inboundDedup) {
|
||||
this.agentService = agentService;
|
||||
this.conversationService = conversationService;
|
||||
this.channelService = channelService;
|
||||
@ -196,6 +198,7 @@ public class ChannelMessageRouter {
|
||||
this.streamTracker = streamTracker;
|
||||
this.chatOriginFactory = chatOriginFactory;
|
||||
this.errorClassifier = errorClassifier;
|
||||
this.inboundDedup = inboundDedup;
|
||||
}
|
||||
|
||||
// ==================== 防抖辅助类 ====================
|
||||
@ -268,6 +271,18 @@ public class ChannelMessageRouter {
|
||||
}
|
||||
channelEntity = fresh;
|
||||
|
||||
// Inbound idempotency, before ANY side effect (magic commands, trigger
|
||||
// fan-out, agent turn). IM platforms redeliver a message whose ack was
|
||||
// late, lost, or non-200; without this claim every redelivery runs its
|
||||
// own agent turn and the user gets the same answer again. Claiming here
|
||||
// rather than in each adapter means every channel — including the four
|
||||
// that never had dedup — is covered by one code path.
|
||||
if (!inboundDedup.claim(channelEntity.getId(), inboundIdentity(message))) {
|
||||
log.info("[{}] Duplicate inbound message (id={}) on channel {}; dropping",
|
||||
adapter.getChannelType(), inboundIdentity(message), channelEntity.getId());
|
||||
return;
|
||||
}
|
||||
|
||||
String conversationId = buildConversationId(message, channelEntity.getId());
|
||||
if (handleMagicCommand(message, adapter, channelEntity, conversationId)) {
|
||||
return;
|
||||
@ -373,15 +388,18 @@ public class ChannelMessageRouter {
|
||||
try {
|
||||
long ws = channelEntity.getWorkspaceId() == null ? 0L : channelEntity.getWorkspaceId();
|
||||
String channelType = adapter.getChannelType();
|
||||
// messageId may be null for adapters that don't surface one;
|
||||
// fall back to a sender+timestamp composite so the dedup key
|
||||
// is at least deterministic-ish per webhook delivery.
|
||||
String messageId = message.getMessageId();
|
||||
if (messageId == null || messageId.isBlank()) {
|
||||
messageId = channelType + ":" + message.getSenderId() + ":"
|
||||
+ (message.getTimestamp() == null ? System.currentTimeMillis()
|
||||
: message.getTimestamp());
|
||||
}
|
||||
// Reuse the identity the inbound claim uses so both agree on what
|
||||
// "the same message" is. Unlike the claim — which the deduplicator
|
||||
// scopes by channel id — this key travels to the trigger pipeline
|
||||
// unscoped, so anything we derive ourselves keeps the channelType
|
||||
// prefix: two channels can otherwise produce the same
|
||||
// sender+timestamp pair inside one workspace.
|
||||
String platformId = message.getMessageId();
|
||||
String messageId = (platformId != null && !platformId.isBlank())
|
||||
? platformId
|
||||
: channelType + ":" + (inboundIdentity(message) != null
|
||||
? inboundIdentity(message)
|
||||
: message.getSenderId() + "@" + System.currentTimeMillis());
|
||||
events.publishEvent(new ChannelMessageReceivedEvent(
|
||||
ws,
|
||||
channelType,
|
||||
@ -396,6 +414,53 @@ public class ChannelMessageRouter {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The stable identity of an inbound message, used both for the inbound
|
||||
* dedup claim and as the trigger pipeline's dedup key.
|
||||
*
|
||||
* <p>Prefers the platform message id — every adapter that has one puts it
|
||||
* on {@link ChannelMessage#getMessageId()}, and a redelivery carries the
|
||||
* same value. Adapters whose stable token is not the raw message id (WeCom
|
||||
* uses its {@code context_token}) put that token there instead.
|
||||
*
|
||||
* <p>Falls back to {@code sender@timestamp} when there is no id but the
|
||||
* platform stamped the message — still stable across redeliveries of the
|
||||
* same payload. Returns {@code null} when neither exists: there is nothing
|
||||
* to tell a redelivery apart from a fresh message, so the caller must fail
|
||||
* open rather than guess.
|
||||
*
|
||||
* <p>Package-private for unit-test access.
|
||||
*/
|
||||
static String inboundIdentity(ChannelMessage message) {
|
||||
if (message == null) {
|
||||
return null;
|
||||
}
|
||||
String messageId = message.getMessageId();
|
||||
if (messageId != null && !messageId.isBlank()) {
|
||||
return messageId;
|
||||
}
|
||||
if (message.getTimestamp() == null) {
|
||||
return null;
|
||||
}
|
||||
return message.getSenderId() + "@" + message.getTimestamp();
|
||||
}
|
||||
|
||||
/**
|
||||
* Has this inbound message already been claimed? A peek, not a claim —
|
||||
* the authoritative claim happens once, in {@link #enqueue}.
|
||||
*
|
||||
* <p>For adapters to call before expensive inbound work (media download,
|
||||
* payload decryption) so a known redelivery costs nothing. Adapters reach
|
||||
* it through the router they already hold, which keeps the deduplicator
|
||||
* out of every adapter constructor.
|
||||
*
|
||||
* @param identity the same value the adapter will put on
|
||||
* {@link ChannelMessage#getMessageId()}
|
||||
*/
|
||||
public boolean isDuplicateInbound(Long channelId, String identity) {
|
||||
return inboundDedup.contains(channelId, identity);
|
||||
}
|
||||
|
||||
/**
|
||||
* 防抖到期：将合并后的消息真正放入渠道队列
|
||||
*/
|
||||
@ -416,6 +481,12 @@ public class ChannelMessageRouter {
|
||||
if (!offered) {
|
||||
log.error("[{}] Message queue full (capacity={}), dropping message from {}",
|
||||
channelType, QUEUE_CAPACITY, pending.firstMessage.getSenderId());
|
||||
// Never handed off — give the claim back so the platform's own
|
||||
// retry can still get an answer. A turn that ran and *failed*
|
||||
// keeps its claim: the user already got the error reply, and a
|
||||
// retry would only produce a second one.
|
||||
inboundDedup.release(pending.channelEntity != null ? pending.channelEntity.getId() : null,
|
||||
inboundIdentity(pending.firstMessage));
|
||||
try {
|
||||
String replyTarget = resolveReplyTarget(pending.firstMessage);
|
||||
pending.adapter.sendMessage(replyTarget, "系统繁忙，请稍后再试");
|
||||
|
||||
@ -8,6 +8,7 @@ import org.springframework.context.event.EventListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
import vip.mate.channel.model.ChannelSessionEntity;
|
||||
import vip.mate.channel.repository.ChannelSessionMapper;
|
||||
import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Comparator;
|
||||
@ -56,6 +57,30 @@ public class ChannelSessionStore {
|
||||
log.info("ChannelSessionStore initialized: loaded {} sessions from DB", sessions.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop the cached session for a conversation the user just deleted.
|
||||
*
|
||||
* <p>{@code deleteConversation} removes the {@code mate_channel_session}
|
||||
* row inside its DB cascade, but the cache is this class's private state
|
||||
* and no DB delete can reach it. Without this listener the entry survives
|
||||
* as a phantom: the next inbound message takes the "update existing" branch
|
||||
* and calls {@code updateById} against a primary key that no longer exists,
|
||||
* which affects 0 rows and never re-inserts — so the channel session stays
|
||||
* missing and proactive push / cron channel resolution silently degrade
|
||||
* after the next restart.
|
||||
*
|
||||
* <p>Runs after the DB cascade commits — see {@link ConversationDeletedEvent}.
|
||||
*
|
||||
* <p>会话被删除后清理内存缓存，避免留下指向已删除行的幽灵条目。
|
||||
*/
|
||||
@EventListener
|
||||
public void onConversationDeleted(ConversationDeletedEvent event) {
|
||||
if (cache.remove(event.conversationId()) != null) {
|
||||
log.info("[ChannelSession] Evicted cached session for deleted conversation {}",
|
||||
event.conversationId());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存或更新会话标识（收到用户消息时调用）
|
||||
*
|
||||
@ -78,40 +103,51 @@ public class ChannelSessionStore {
|
||||
existing.setSenderName(senderName);
|
||||
existing.setChannelId(channelId);
|
||||
existing.setLastActiveTime(now);
|
||||
sessionMapper.updateById(existing);
|
||||
log.debug("Updated channel session: conversationId={}, targetId={}", conversationId, targetId);
|
||||
} else {
|
||||
// 先查 DB（可能是上次启动后的新记录）
|
||||
ChannelSessionEntity dbEntity = sessionMapper.selectOne(
|
||||
new LambdaQueryWrapper<ChannelSessionEntity>()
|
||||
.eq(ChannelSessionEntity::getConversationId, conversationId));
|
||||
|
||||
if (dbEntity != null) {
|
||||
dbEntity.setTargetId(targetId);
|
||||
dbEntity.setSenderId(senderId);
|
||||
dbEntity.setSenderName(senderName);
|
||||
dbEntity.setChannelId(channelId);
|
||||
dbEntity.setLastActiveTime(now);
|
||||
sessionMapper.updateById(dbEntity);
|
||||
cache.put(conversationId, dbEntity);
|
||||
log.debug("Updated channel session from DB: conversationId={}", conversationId);
|
||||
} else {
|
||||
// 新建
|
||||
ChannelSessionEntity entity = new ChannelSessionEntity();
|
||||
entity.setConversationId(conversationId);
|
||||
entity.setChannelType(channelType);
|
||||
entity.setTargetId(targetId);
|
||||
entity.setSenderId(senderId);
|
||||
entity.setSenderName(senderName);
|
||||
entity.setChannelId(channelId);
|
||||
entity.setLastActiveTime(now);
|
||||
sessionMapper.insert(entity);
|
||||
cache.put(conversationId, entity);
|
||||
log.debug("Created channel session: conversationId={}, targetId={}", conversationId, targetId);
|
||||
|
||||
// 容量保护：超过上限时淘汰最久未活跃的会话
|
||||
evictIfNeeded();
|
||||
int updated = sessionMapper.updateById(existing);
|
||||
if (updated > 0) {
|
||||
log.debug("Updated channel session: conversationId={}, targetId={}", conversationId, targetId);
|
||||
return;
|
||||
}
|
||||
// The cached entity points at a row that no longer exists — the
|
||||
// conversation was deleted out from under us (deletes are physical;
|
||||
// no logical-delete column is honoured project-wide). Without this
|
||||
// self-heal the update silently affects 0 rows on every subsequent
|
||||
// message and the session is never re-created, so proactive push
|
||||
// and cron channel resolution break after the next restart.
|
||||
log.info("Channel session row for {} vanished; re-creating from cache miss", conversationId);
|
||||
cache.remove(conversationId);
|
||||
}
|
||||
|
||||
// 先查 DB（可能是上次启动后的新记录）
|
||||
ChannelSessionEntity dbEntity = sessionMapper.selectOne(
|
||||
new LambdaQueryWrapper<ChannelSessionEntity>()
|
||||
.eq(ChannelSessionEntity::getConversationId, conversationId));
|
||||
|
||||
if (dbEntity != null) {
|
||||
dbEntity.setTargetId(targetId);
|
||||
dbEntity.setSenderId(senderId);
|
||||
dbEntity.setSenderName(senderName);
|
||||
dbEntity.setChannelId(channelId);
|
||||
dbEntity.setLastActiveTime(now);
|
||||
sessionMapper.updateById(dbEntity);
|
||||
cache.put(conversationId, dbEntity);
|
||||
log.debug("Updated channel session from DB: conversationId={}", conversationId);
|
||||
} else {
|
||||
// 新建
|
||||
ChannelSessionEntity entity = new ChannelSessionEntity();
|
||||
entity.setConversationId(conversationId);
|
||||
entity.setChannelType(channelType);
|
||||
entity.setTargetId(targetId);
|
||||
entity.setSenderId(senderId);
|
||||
entity.setSenderName(senderName);
|
||||
entity.setChannelId(channelId);
|
||||
entity.setLastActiveTime(now);
|
||||
sessionMapper.insert(entity);
|
||||
cache.put(conversationId, entity);
|
||||
log.debug("Created channel session: conversationId={}, targetId={}", conversationId, targetId);
|
||||
|
||||
// 容量保护：超过上限时淘汰最久未活跃的会话
|
||||
evictIfNeeded();
|
||||
}
|
||||
}
|
||||
|
||||
@ -189,13 +225,29 @@ public class ChannelSessionStore {
|
||||
}
|
||||
|
||||
/**
|
||||
* 删除会话
|
||||
* Drop a channel session from both layers.
|
||||
*
|
||||
* <p>Use this rather than the mapper: this class owns the cache, so a
|
||||
* caller that deletes the row directly leaves a phantom entry behind —
|
||||
* every later {@code saveOrUpdate} then updates a primary key that no
|
||||
* longer exists and the session is never re-created.
|
||||
*
|
||||
* <p>The conversation-delete cascade does not come through here: it removes
|
||||
* the row inside its own transaction and lets
|
||||
* {@link #onConversationDeleted} drop the cache after commit, so the cache
|
||||
* is never cleared for a delete that later rolls back.
|
||||
*
|
||||
* <p>删除会话（内存 + DB 双层）。
|
||||
*
|
||||
* @return number of DB rows removed
|
||||
*/
|
||||
public void remove(String conversationId) {
|
||||
ChannelSessionEntity removed = cache.remove(conversationId);
|
||||
if (removed != null) {
|
||||
sessionMapper.deleteById(removed.getId());
|
||||
public int remove(String conversationId) {
|
||||
cache.remove(conversationId);
|
||||
int deleted = sessionMapper.delete(new LambdaQueryWrapper<ChannelSessionEntity>()
|
||||
.eq(ChannelSessionEntity::getConversationId, conversationId));
|
||||
if (deleted > 0) {
|
||||
log.debug("Removed channel session: conversationId={}", conversationId);
|
||||
}
|
||||
return deleted;
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,187 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* TTL- and capacity-bounded claim register for inbound channel messages.
|
||||
*
|
||||
* <p>One shared implementation for every channel. Before this existed, four
|
||||
* adapters carried four hand-rolled variants (a 500-entry LRU, an unbounded
|
||||
* set halved on overflow, an access-ordered map) and four adapters carried
|
||||
* none at all — DingTalk among them, which is why a redelivered DingTalk
|
||||
* message produced a second full answer.
|
||||
*
|
||||
* <p>A message is identified by {@code channelId + identity}, where identity is
|
||||
* the platform message id (see
|
||||
* {@link ChannelMessageRouter#inboundIdentity(ChannelMessage)}). Scoping by
|
||||
* channel keeps two channels of the same type from colliding on a platform id
|
||||
* that is only unique per app.
|
||||
*
|
||||
* <p>Three operations, matching the three things a caller needs:
|
||||
* <ul>
|
||||
* <li>{@link #claim} — take ownership of a message. The first caller gets
|
||||
* {@code true} and proceeds; a redelivery inside the TTL gets
|
||||
* {@code false} and must drop the message.</li>
|
||||
* <li>{@link #contains} — peek without claiming, so an adapter can drop a
|
||||
* known redelivery <em>before</em> expensive inbound work (media
|
||||
* download, payload decryption) and still leave the authoritative claim
|
||||
* to the router.</li>
|
||||
* <li>{@link #release} — give a claim back when the message was never handed
|
||||
* off for processing (e.g. the channel queue was full), so the
|
||||
* platform's own retry can still get through.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>Fail-open by design: a blank identity means "this platform gave us
|
||||
* nothing stable to dedup on", and the message is always let through. Dropping
|
||||
* a real message is worse than answering a redelivery twice.
|
||||
*
|
||||
* <p>入站消息去重登记表（TTL + 容量双约束），全渠道共用一份实现。
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@EnableConfigurationProperties(ChannelDedupProperties.class)
|
||||
public class InboundMessageDeduplicator {
|
||||
|
||||
private final ChannelDedupProperties props;
|
||||
|
||||
/**
|
||||
* Claimed identity -> claim timestamp (epoch millis). Insertion-ordered so
|
||||
* the eldest entries sit at the head and overflow trimming is a head scan.
|
||||
* Guarded by its own monitor — claims are short, contended only by the
|
||||
* channel intake threads.
|
||||
*/
|
||||
private final LinkedHashMap<String, Long> claims = new LinkedHashMap<>();
|
||||
|
||||
public InboundMessageDeduplicator(ChannelDedupProperties props) {
|
||||
this.props = props;
|
||||
}
|
||||
|
||||
/**
|
||||
* Take ownership of an inbound message.
|
||||
*
|
||||
* @return {@code true} when the caller owns this message and should process
|
||||
* it; {@code false} when it is a redelivery already claimed inside
|
||||
* the TTL window and must be dropped
|
||||
*/
|
||||
public boolean claim(Long channelId, String identity) {
|
||||
String key = key(channelId, identity);
|
||||
if (key == null || !props.isEnabled()) {
|
||||
return true;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
long ttlMs = ttlMillis();
|
||||
synchronized (claims) {
|
||||
Long claimedAt = claims.get(key);
|
||||
if (claimedAt != null && now - claimedAt < ttlMs) {
|
||||
return false;
|
||||
}
|
||||
// Either new, or an expired claim being retaken. Remove first so
|
||||
// the re-insert moves the entry to the tail — insertion order is
|
||||
// what the overflow trim relies on to find the eldest claims.
|
||||
claims.remove(key);
|
||||
claims.put(key, now);
|
||||
if (claims.size() > props.getMaxSize()) {
|
||||
trim(now, ttlMs);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Peek at a claim without taking one. Lets an adapter short-circuit a
|
||||
* redelivery before doing expensive inbound work while leaving the single
|
||||
* authoritative claim to the router.
|
||||
*/
|
||||
public boolean contains(Long channelId, String identity) {
|
||||
String key = key(channelId, identity);
|
||||
if (key == null || !props.isEnabled()) {
|
||||
return false;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
long ttlMs = ttlMillis();
|
||||
synchronized (claims) {
|
||||
Long claimedAt = claims.get(key);
|
||||
if (claimedAt == null) {
|
||||
return false;
|
||||
}
|
||||
if (now - claimedAt < ttlMs) {
|
||||
return true;
|
||||
}
|
||||
claims.remove(key);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand a claim back. Call this only when the message was never handed off
|
||||
* for processing — a turn that ran and failed keeps its claim, because the
|
||||
* user already received the error and a platform retry would just send a
|
||||
* second one.
|
||||
*/
|
||||
public void release(Long channelId, String identity) {
|
||||
String key = key(channelId, identity);
|
||||
if (key == null) {
|
||||
return;
|
||||
}
|
||||
synchronized (claims) {
|
||||
claims.remove(key);
|
||||
}
|
||||
}
|
||||
|
||||
/** Drop every claim. Called when a channel restarts. */
|
||||
public void clear() {
|
||||
synchronized (claims) {
|
||||
claims.clear();
|
||||
}
|
||||
}
|
||||
|
||||
/** Live claim count. Package-private for tests. */
|
||||
int size() {
|
||||
synchronized (claims) {
|
||||
return claims.size();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Evict expired claims first; if the map is still over capacity (every
|
||||
* entry fresh under sustained traffic), drop the eldest until it fits.
|
||||
* Caller holds the monitor.
|
||||
*/
|
||||
private void trim(long now, long ttlMs) {
|
||||
claims.entrySet().removeIf(e -> now - e.getValue() >= ttlMs);
|
||||
int overflow = claims.size() - props.getMaxSize();
|
||||
if (overflow <= 0) {
|
||||
return;
|
||||
}
|
||||
Iterator<Map.Entry<String, Long>> it = claims.entrySet().iterator();
|
||||
for (int i = 0; i < overflow && it.hasNext(); i++) {
|
||||
it.next();
|
||||
it.remove();
|
||||
}
|
||||
log.debug("[dedup] Trimmed {} eldest claims (cap={})", overflow, props.getMaxSize());
|
||||
}
|
||||
|
||||
private long ttlMillis() {
|
||||
Duration ttl = props.getTtl();
|
||||
return ttl != null ? Math.max(1L, ttl.toMillis()) : Duration.ofMinutes(5).toMillis();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compose the tracking key, or {@code null} when there is nothing stable to
|
||||
* track. Scoped by channel id so two channels of the same type can't
|
||||
* collide on a per-app platform id.
|
||||
*/
|
||||
private static String key(Long channelId, String identity) {
|
||||
if (identity == null || identity.isBlank()) {
|
||||
return null;
|
||||
}
|
||||
return (channelId == null ? "-" : channelId.toString()) + ":" + identity;
|
||||
}
|
||||
}
|
||||
@ -27,6 +27,8 @@ import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* 钉钉渠道适配器
|
||||
@ -62,6 +64,12 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
/** AI Card 管理器（message_type=card 时初始化） */
|
||||
private DingTalkAICardManager aiCardManager;
|
||||
|
||||
/**
|
||||
* Off-callback worker for inbound parsing, so the Stream frame is acked
|
||||
* immediately. See {@link #dispatchInbound}.
|
||||
*/
|
||||
private volatile ExecutorService inboundExecutor;
|
||||
|
||||
/** 钉钉媒体上传器（doStart 时初始化） */
|
||||
private DingTalkMediaUploader mediaUploader;
|
||||
|
||||
@ -118,6 +126,11 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
|
||||
// 启动 Stream 模式或 Webhook 模式
|
||||
if (isStreamMode()) {
|
||||
this.inboundExecutor = Executors.newSingleThreadExecutor(r -> {
|
||||
Thread t = new Thread(r, "dingtalk-inbound-" + channelEntity.getId());
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
startStreamMode(clientId, clientSecret);
|
||||
} else {
|
||||
log.info("[dingtalk] Webhook mode: waiting for callbacks at /api/v1/channels/webhook/dingtalk");
|
||||
@ -234,12 +247,44 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
return;
|
||||
}
|
||||
|
||||
handleWebhook(payload);
|
||||
dispatchInbound(payload);
|
||||
} catch (Exception e) {
|
||||
log.error("[dingtalk-stream] Failed to parse stream message: {}", e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand the parsed payload to a worker and return, so the SDK can ack the
|
||||
* Stream frame immediately.
|
||||
*
|
||||
* <p>{@link #handleWebhook} resolves media inline — each attachment costs a
|
||||
* download-URL call plus a byte fetch against DingTalk. Running that on the
|
||||
* callback thread delays the ack by however long the downloads take, and a
|
||||
* late ack makes DingTalk redeliver the message: the user gets the same
|
||||
* answer once per redelivery. Acking first removes the cause; the router's
|
||||
* inbound claim is the second line of defence for redeliveries we can't
|
||||
* prevent.
|
||||
*
|
||||
* <p>Single-threaded on purpose — the agent turn itself already runs on the
|
||||
* router's queue, so this thread only parses, and keeping it serial
|
||||
* preserves the arrival order of a sender's messages.
|
||||
*/
|
||||
private void dispatchInbound(Map<String, Object> payload) {
|
||||
ExecutorService executor = inboundExecutor;
|
||||
if (executor == null || executor.isShutdown()) {
|
||||
// Channel stopped mid-flight — process inline rather than drop.
|
||||
handleWebhook(payload);
|
||||
return;
|
||||
}
|
||||
executor.execute(() -> {
|
||||
try {
|
||||
handleWebhook(payload);
|
||||
} catch (Exception e) {
|
||||
log.error("[dingtalk-stream] Inbound dispatch failed: {}", e.getMessage(), e);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
// 关闭 Stream 客户端
|
||||
@ -252,6 +297,10 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
|
||||
}
|
||||
streamClient = null;
|
||||
}
|
||||
if (inboundExecutor != null) {
|
||||
inboundExecutor.shutdownNow();
|
||||
inboundExecutor = null;
|
||||
}
|
||||
if (aiCardManager != null) {
|
||||
aiCardManager.cleanup();
|
||||
aiCardManager = null;
|
||||
|
||||
@ -60,14 +60,6 @@ public class DiscordChannelAdapter extends AbstractChannelAdapter {
|
||||
/** 媒体下载用 HttpClient（复用 http_proxy 配置） */
|
||||
private volatile HttpClient mediaHttpClient;
|
||||
|
||||
/** 已处理消息去重（LRU，最多保留 500 条） */
|
||||
private final Set<String> processedMessageIds = Collections.newSetFromMap(new LinkedHashMap<>() {
|
||||
@Override
|
||||
protected boolean removeEldestEntry(Map.Entry<String, Boolean> eldest) {
|
||||
return size() > 500;
|
||||
}
|
||||
});
|
||||
|
||||
public DiscordChannelAdapter(ChannelEntity channelEntity,
|
||||
ChannelMessageRouter messageRouter,
|
||||
ObjectMapper objectMapper) {
|
||||
@ -146,7 +138,6 @@ public class DiscordChannelAdapter extends AbstractChannelAdapter {
|
||||
}
|
||||
selfId = null;
|
||||
mediaHttpClient = null;
|
||||
processedMessageIds.clear();
|
||||
log.info("[discord] Discord channel stopped");
|
||||
}
|
||||
|
||||
@ -447,14 +438,10 @@ public class DiscordChannelAdapter extends AbstractChannelAdapter {
|
||||
return;
|
||||
}
|
||||
|
||||
// 去重
|
||||
// Inbound dedup lives in ChannelMessageRouter.enqueue now — msgId is
|
||||
// carried on the ChannelMessage below and claimed there, once, for
|
||||
// every channel.
|
||||
String msgId = message.getId();
|
||||
synchronized (processedMessageIds) {
|
||||
if (processedMessageIds.contains(msgId)) {
|
||||
return;
|
||||
}
|
||||
processedMessageIds.add(msgId);
|
||||
}
|
||||
|
||||
String channelId = message.getChannel().getId();
|
||||
String guildId = message.isFromGuild() ? message.getGuild().getId() : null;
|
||||
|
||||
@ -84,9 +84,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
/** 定时 Token 刷新任务 */
|
||||
private ScheduledFuture<?> tokenRefreshFuture;
|
||||
|
||||
/** 消息去重：最近处理过的 message_id */
|
||||
private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();
|
||||
|
||||
/**
|
||||
* 群内 bot 别名缓存：chatId → 学到的别名集合（openId / unionId / userId / name）。
|
||||
* <p>飞书 SDK 投递的 mention 里，bot 的标识可能是群内自定义别名（{@code ou_357e...} / 自定义名称），
|
||||
@ -460,7 +457,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
this.botName = null;
|
||||
this.botOpenIdLastFailureMs = 0L;
|
||||
}
|
||||
this.processedMessageIds.clear();
|
||||
this.chatBotAliases.clear();
|
||||
this.mentionTracker.clear();
|
||||
this.nicknameCache.clear();
|
||||
@ -1233,12 +1229,14 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
log.warn("[feishu] require_mention=true but bot open_id unavailable; allowing messageId={}", messageId);
|
||||
}
|
||||
|
||||
// 消息去重
|
||||
if (messageId != null && !processedMessageIds.add(messageId)) {
|
||||
// Early duplicate gate. The authoritative claim happens once, in
|
||||
// ChannelMessageRouter.enqueue; this peek only spares a redelivery the
|
||||
// side effects below (the "received" reaction, media downloads) that
|
||||
// would otherwise fire again before the router ever sees the message.
|
||||
if (messageRouter.isDuplicateInbound(channelEntity.getId(), messageId)) {
|
||||
log.debug("[feishu] Duplicate message_id: {}, skipping", messageId);
|
||||
return;
|
||||
}
|
||||
cleanupProcessedIds();
|
||||
|
||||
// 添加消息反应（非阻塞，表示"已收到"）
|
||||
if (messageId != null && getConfigBoolean("enable_reaction", true)) {
|
||||
@ -1299,21 +1297,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
|
||||
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--;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== 消息反应 ====================
|
||||
|
||||
/**
|
||||
|
||||
@ -60,9 +60,6 @@ public class WeixinChannelAdapter extends AbstractChannelAdapter {
|
||||
|
||||
public static final String CHANNEL_TYPE = "weixin";
|
||||
|
||||
/** 消息去重最大记录数 */
|
||||
private static final int PROCESSED_IDS_MAX = 2000;
|
||||
|
||||
// ==================== 运行时状态 ====================
|
||||
|
||||
private ILinkClient client;
|
||||
@ -99,14 +96,6 @@ public class WeixinChannelAdapter extends AbstractChannelAdapter {
|
||||
private static final long POLL_STUCK_THRESHOLD_MS = 90_000;
|
||||
private static final long WATCHDOG_INTERVAL_MS = 30_000;
|
||||
|
||||
/** 消息去重集合（LRU） */
|
||||
private final LinkedHashMap<String, Boolean> processedIds = new LinkedHashMap<>(256, 0.75f, true) {
|
||||
@Override
|
||||
protected boolean removeEldestEntry(Map.Entry<String, Boolean> eldest) {
|
||||
return size() > PROCESSED_IDS_MAX;
|
||||
}
|
||||
};
|
||||
|
||||
/** 用户最新 context_token 缓存（用于主动推送） */
|
||||
private final ConcurrentHashMap<String, String> userContextTokens = new ConcurrentHashMap<>();
|
||||
|
||||
@ -422,15 +411,18 @@ public class WeixinChannelAdapter extends AbstractChannelAdapter {
|
||||
return;
|
||||
}
|
||||
|
||||
// 去重
|
||||
// Stable inbound identity for this channel: context_token when the
|
||||
// platform supplies one (it survives redelivery where msg_id does not),
|
||||
// else sender + msg_id. Carried on the ChannelMessage as messageId so
|
||||
// the router claims exactly this key — see inboundIdentity().
|
||||
String dedupKey = !contextToken.isBlank() ? contextToken
|
||||
: fromUserId + "_" + getStr(msg, "msg_id");
|
||||
synchronized (processedIds) {
|
||||
if (processedIds.containsKey(dedupKey)) {
|
||||
log.debug("[weixin] Duplicate message skipped: {}", dedupKey.substring(0, Math.min(40, dedupKey.length())));
|
||||
return;
|
||||
}
|
||||
processedIds.put(dedupKey, Boolean.TRUE);
|
||||
// Early duplicate gate: the authoritative claim happens once in
|
||||
// ChannelMessageRouter.enqueue, but parsing below downloads media, so
|
||||
// a known redelivery is dropped before paying for that.
|
||||
if (messageRouter.isDuplicateInbound(channelEntity.getId(), dedupKey)) {
|
||||
log.debug("[weixin] Duplicate message skipped: {}", dedupKey.substring(0, Math.min(40, dedupKey.length())));
|
||||
return;
|
||||
}
|
||||
|
||||
// 解析消息内容
|
||||
@ -644,7 +636,9 @@ public class WeixinChannelAdapter extends AbstractChannelAdapter {
|
||||
String replyToken = contextToken + "|" + fromUserId;
|
||||
|
||||
ChannelMessage channelMessage = ChannelMessage.builder()
|
||||
.messageId(getStr(msg, "msg_id"))
|
||||
// dedupKey, not the raw msg_id: it is this channel's stable
|
||||
// inbound identity and the router claims exactly this value.
|
||||
.messageId(dedupKey)
|
||||
.channelType(CHANNEL_TYPE)
|
||||
.senderId(fromUserId)
|
||||
.senderName(fromUserId) // iLink API 不提供昵称
|
||||
|
||||
@ -1173,7 +1173,7 @@ public class WikiProcessingService {
|
||||
int totalPlanned = createMetas.size() + updateSlugs.size();
|
||||
|
||||
// Chunk fallback: if route returned nothing for a non-trivial chunk, inject an overview page
|
||||
// so no content is silently dropped (mirrors llm_wiki source-summary guarantee).
|
||||
// so no content is silently dropped: every source chunk is represented by at least one page.
|
||||
if (totalPlanned == 0 && textContent.length() >= properties.getChunkFallbackMinChars()) {
|
||||
String overviewSlug = WikiPageService.toSlug(rawTitle) + "-overview";
|
||||
ObjectNode fallbackMeta =
|
||||
|
||||
@ -290,6 +290,14 @@ mateclaw:
|
||||
|
||||
# MateClaw Agent 配置
|
||||
mate:
|
||||
channel:
|
||||
# 入站消息去重：IM 平台在 ack 迟到 / 丢失 / 非 200 时会重投同一条消息，
|
||||
# 不去重则每次重投都会跑一轮完整 Agent 回合，用户看到重复答复。
|
||||
dedup:
|
||||
enabled: true
|
||||
# 需明显长于各平台重投窗口（秒级到分钟级）
|
||||
ttl: 5m
|
||||
max-size: 2000
|
||||
agent:
|
||||
# Deterministic Markdown cleanup of the final answer (heading spaces, glued
|
||||
# ---, table pipe alignment) before persistence / channel delivery. Set to
|
||||
|
||||
@ -362,7 +362,8 @@ class ChannelMagicCommandTest {
|
||||
router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier);
|
||||
chatOriginFactory, errorClassifier,
|
||||
new InboundMessageDeduplicator(new ChannelDedupProperties()));
|
||||
when(adapter.getChannelType()).thenReturn("wecom");
|
||||
channel.setAgentId(100L);
|
||||
// modelConfigService is field-injected on the real router (optional
|
||||
|
||||
@ -37,7 +37,8 @@ class ChannelMessageRouterApprovalDenyTest {
|
||||
ChannelMessageRouter router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier);
|
||||
chatOriginFactory, errorClassifier,
|
||||
new InboundMessageDeduplicator(new ChannelDedupProperties()));
|
||||
|
||||
PendingApproval pending = new PendingApproval("abcdef123", "conv-1", "alice",
|
||||
"dangerous_tool", "{}", "needs approval");
|
||||
|
||||
@ -149,7 +149,8 @@ class ChannelMessageRouterExecutionMetadataTest {
|
||||
router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier);
|
||||
chatOriginFactory, errorClassifier,
|
||||
new InboundMessageDeduplicator(new ChannelDedupProperties()));
|
||||
when(adapter.getChannelType()).thenReturn("telegram");
|
||||
when(chatOriginFactory.from(any(), any(), any(), any())).thenReturn(ChatOrigin.EMPTY);
|
||||
channel.setAgentId(100L);
|
||||
|
||||
@ -0,0 +1,196 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.approval.ApprovalWorkflowService;
|
||||
import vip.mate.channel.model.ChannelEntity;
|
||||
import vip.mate.channel.notification.ApprovalNotificationService;
|
||||
import vip.mate.channel.service.ChannelService;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.memory.event.ConversationCompletionPublisher;
|
||||
import vip.mate.tts.TtsService;
|
||||
import vip.mate.workspace.conversation.ConversationService;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* Issue #526: a redelivered IM message must not start a second agent turn.
|
||||
*
|
||||
* <p>Platforms redeliver when an ack is late, lost, or non-200. Before the
|
||||
* router claimed inbound identities, every redelivery ran a full independent
|
||||
* turn — the user got the same answer twice and the conversation gained a
|
||||
* duplicate user/assistant pair. These tests pin the claim at {@code enqueue},
|
||||
* which is the one path every channel goes through.
|
||||
*/
|
||||
class ChannelMessageRouterInboundDedupTest {
|
||||
|
||||
@Test
|
||||
@DisplayName("a redelivered message with the same id runs the turn only once")
|
||||
void redeliveryIsDropped() {
|
||||
Fixture f = new Fixture();
|
||||
|
||||
// Spaced past the debounce window so a NON-deduped redelivery would
|
||||
// start its own turn — otherwise the merger alone would produce one
|
||||
// turn and the assertion would prove nothing.
|
||||
f.enqueueSpaced("dt-msg-1", "你好");
|
||||
f.enqueueSpaced("dt-msg-1", "你好");
|
||||
f.enqueueSpaced("dt-msg-1", "你好");
|
||||
|
||||
f.awaitTurns(1);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("distinct message ids each get their own turn")
|
||||
void distinctMessagesEachRun() {
|
||||
Fixture f = new Fixture();
|
||||
|
||||
f.enqueueSpaced("dt-msg-1", "你好");
|
||||
f.enqueueSpaced("dt-msg-2", "再问一次");
|
||||
|
||||
f.awaitTurns(2);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("no platform id and no timestamp fails open rather than dropping a real message")
|
||||
void missingIdentityFailsOpen() {
|
||||
Fixture f = new Fixture();
|
||||
|
||||
f.enqueueWithoutIdentity("你好");
|
||||
Fixture.sleep(ChannelMessageRouter.DEBOUNCE_MS * 3);
|
||||
f.enqueueWithoutIdentity("你好");
|
||||
|
||||
// Both must get through: with nothing stable to key on, silently
|
||||
// swallowing the second message would lose real user input.
|
||||
f.awaitTurns(2);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("the same platform id on a different channel is not a duplicate")
|
||||
void identitiesAreScopedPerChannel() {
|
||||
Fixture f = new Fixture();
|
||||
|
||||
f.enqueue(1L, "shared-id", "你好");
|
||||
f.enqueue(2L, "shared-id", "你好");
|
||||
|
||||
f.awaitTurns(2);
|
||||
}
|
||||
|
||||
// ==================== helpers ====================
|
||||
|
||||
private static final class Fixture {
|
||||
final ConversationService conversationService = mock(ConversationService.class);
|
||||
final ChannelService channelService = mock(ChannelService.class);
|
||||
final ChannelAdapter adapter = mock(ChannelAdapter.class);
|
||||
final ChannelMessageRouter router;
|
||||
|
||||
Fixture() {
|
||||
AgentService agentService = mock(AgentService.class);
|
||||
ChannelSessionStore channelSessionStore = mock(ChannelSessionStore.class);
|
||||
ApprovalWorkflowService approvalService = mock(ApprovalWorkflowService.class);
|
||||
ApprovalNotificationService approvalNotificationService = mock(ApprovalNotificationService.class);
|
||||
ConversationCompletionPublisher completionPublisher = mock(ConversationCompletionPublisher.class);
|
||||
TtsService ttsService = mock(TtsService.class);
|
||||
ChatStreamTracker streamTracker = mock(ChatStreamTracker.class);
|
||||
ChannelChatOriginFactory chatOriginFactory = mock(ChannelChatOriginFactory.class);
|
||||
ChannelErrorClassifier errorClassifier = mock(ChannelErrorClassifier.class);
|
||||
|
||||
router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier,
|
||||
new InboundMessageDeduplicator(new ChannelDedupProperties()));
|
||||
|
||||
when(adapter.getChannelType()).thenReturn("dingtalk");
|
||||
// Channel lookups return a live, agent-bound row for any id.
|
||||
when(channelService.getChannel(anyLong())).thenAnswer(inv -> channel(inv.getArgument(0)));
|
||||
}
|
||||
|
||||
private static ChannelEntity channel(Long id) {
|
||||
ChannelEntity e = new ChannelEntity();
|
||||
e.setId(id);
|
||||
e.setName("dingtalk-" + id);
|
||||
e.setChannelType("dingtalk");
|
||||
e.setEnabled(true);
|
||||
e.setAgentId(100L);
|
||||
e.setWorkspaceId(1L);
|
||||
return e;
|
||||
}
|
||||
|
||||
void enqueue(String messageId, String content) {
|
||||
enqueue(1L, messageId, content);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue, then wait out the debounce window so the next send cannot be
|
||||
* merged into this one. Dedup and the merger both collapse traffic into
|
||||
* fewer turns; spacing them apart is what makes the assertion attribute
|
||||
* the collapse to dedup.
|
||||
*/
|
||||
void enqueueSpaced(String messageId, String content) {
|
||||
enqueue(1L, messageId, content);
|
||||
sleep(ChannelMessageRouter.DEBOUNCE_MS * 3);
|
||||
}
|
||||
|
||||
void enqueue(Long channelId, String messageId, String content) {
|
||||
router.enqueue(ChannelMessage.builder()
|
||||
.messageId(messageId)
|
||||
.channelType("dingtalk")
|
||||
.senderId("alice")
|
||||
.content(content)
|
||||
.replyToken("reply-1")
|
||||
.timestamp(LocalDateTime.of(2026, 1, 1, 0, 0))
|
||||
.build(), adapter, channel(channelId));
|
||||
}
|
||||
|
||||
void enqueueWithoutIdentity(String content) {
|
||||
router.enqueue(ChannelMessage.builder()
|
||||
.channelType("dingtalk")
|
||||
.senderId("alice")
|
||||
.content(content)
|
||||
.replyToken("reply-1")
|
||||
.build(), adapter, channel(1L));
|
||||
}
|
||||
|
||||
/**
|
||||
* Count turns by the get-or-create every processed message performs
|
||||
* before the agent runs. Polls past the debounce + queue handoff, then
|
||||
* holds steady long enough that a late extra turn would still fail the
|
||||
* count.
|
||||
*/
|
||||
void awaitTurns(int expected) {
|
||||
long deadline = System.currentTimeMillis() + 15_000;
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
if (turnCount() >= expected) {
|
||||
break;
|
||||
}
|
||||
sleep(50);
|
||||
}
|
||||
// A duplicate would arrive one debounce window behind the original;
|
||||
// wait it out before asserting the exact count.
|
||||
sleep(ChannelMessageRouter.DEBOUNCE_MS * 2);
|
||||
verify(conversationService, times(expected)).getOrCreateSharedConversation(
|
||||
anyString(), eq(100L), anyLong(), isNull(), isNull());
|
||||
}
|
||||
|
||||
private int turnCount() {
|
||||
return mockingDetails(conversationService).getInvocations().stream()
|
||||
.filter(i -> "getOrCreateSharedConversation".equals(i.getMethod().getName()))
|
||||
.filter(i -> i.getArguments().length == 5)
|
||||
.toList()
|
||||
.size();
|
||||
}
|
||||
|
||||
static void sleep(long ms) {
|
||||
try {
|
||||
Thread.sleep(ms);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -137,7 +137,8 @@ class ChannelMessageRouterNarrationTest {
|
||||
router = new ChannelMessageRouter(agentService, conversationService,
|
||||
channelService, channelSessionStore, approvalService, approvalNotificationService,
|
||||
completionPublisher, ttsService, new ObjectMapper(), streamTracker,
|
||||
chatOriginFactory, errorClassifier);
|
||||
chatOriginFactory, errorClassifier,
|
||||
new InboundMessageDeduplicator(new ChannelDedupProperties()));
|
||||
when(adapter.getChannelType()).thenReturn("telegram");
|
||||
when(chatOriginFactory.from(any(), any(), any(), any())).thenReturn(ChatOrigin.EMPTY);
|
||||
channel.setAgentId(100L);
|
||||
|
||||
@ -0,0 +1,87 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import vip.mate.channel.model.ChannelSessionEntity;
|
||||
import vip.mate.channel.repository.ChannelSessionMapper;
|
||||
import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* Issue #526: deleting a conversation removes the {@code mate_channel_session}
|
||||
* row inside the DB cascade, but the cache is this class's private state. If
|
||||
* the entry survives, the next inbound message updates a primary key that no
|
||||
* longer exists — 0 rows, no re-insert — and the channel session stays missing.
|
||||
*/
|
||||
class ChannelSessionStoreTest {
|
||||
|
||||
private ChannelSessionMapper mapper;
|
||||
private ChannelSessionStore store;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
mapper = mock(ChannelSessionMapper.class);
|
||||
store = new ChannelSessionStore(mapper);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("deleting the conversation evicts the cached session")
|
||||
void conversationDeleteEvictsCache() {
|
||||
when(mapper.selectOne(any())).thenReturn(null);
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-1", "alice", "Alice", 1L);
|
||||
assertNotNull(store.getSession("dingtalk:1:alice"));
|
||||
|
||||
store.onConversationDeleted(new ConversationDeletedEvent("dingtalk:1:alice"));
|
||||
|
||||
assertNull(store.getSession("dingtalk:1:alice"),
|
||||
"a phantom entry would keep updating a row that no longer exists");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a cached entry pointing at a deleted row self-heals into a fresh insert")
|
||||
void staleCacheEntryIsRecreated() {
|
||||
when(mapper.selectOne(any())).thenReturn(null);
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-1", "alice", "Alice", 1L);
|
||||
verify(mapper, times(1)).insert(any(ChannelSessionEntity.class));
|
||||
|
||||
// The row is deleted behind our back (console delete on another node,
|
||||
// or an event listener that never ran) — updateById now affects 0 rows.
|
||||
when(mapper.updateById(any(ChannelSessionEntity.class))).thenReturn(0);
|
||||
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-2", "alice", "Alice", 1L);
|
||||
|
||||
// Must not stop at the failed update: re-insert so proactive push and
|
||||
// cron channel resolution keep working.
|
||||
verify(mapper, times(2)).insert(any(ChannelSessionEntity.class));
|
||||
assertEquals("hook-2", store.getTargetId("dingtalk:1:alice"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("a successful update does not fall through to an insert")
|
||||
void liveRowIsUpdatedInPlace() {
|
||||
when(mapper.selectOne(any())).thenReturn(null);
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-1", "alice", "Alice", 1L);
|
||||
when(mapper.updateById(any(ChannelSessionEntity.class))).thenReturn(1);
|
||||
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-2", "alice", "Alice", 1L);
|
||||
|
||||
verify(mapper, times(1)).insert(any(ChannelSessionEntity.class));
|
||||
assertEquals("hook-2", store.getTargetId("dingtalk:1:alice"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("remove() clears both layers")
|
||||
void removeClearsCacheAndRow() {
|
||||
when(mapper.selectOne(any())).thenReturn(null);
|
||||
when(mapper.delete(any())).thenReturn(1);
|
||||
store.saveOrUpdate("dingtalk:1:alice", "dingtalk", "hook-1", "alice", "Alice", 1L);
|
||||
|
||||
assertEquals(1, store.remove("dingtalk:1:alice"));
|
||||
assertNull(store.getSession("dingtalk:1:alice"));
|
||||
verify(mapper).delete(any());
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,166 @@
|
||||
package vip.mate.channel;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/**
|
||||
* Inbound dedup contract: a redelivered message is dropped, a genuinely new
|
||||
* one is not, and nothing is dropped when the platform gave us no stable
|
||||
* identity to key on.
|
||||
*/
|
||||
class InboundMessageDeduplicatorTest {
|
||||
|
||||
private static InboundMessageDeduplicator dedup(ChannelDedupProperties props) {
|
||||
return new InboundMessageDeduplicator(props);
|
||||
}
|
||||
|
||||
private static ChannelDedupProperties props() {
|
||||
return new ChannelDedupProperties();
|
||||
}
|
||||
|
||||
@Test
|
||||
void firstClaimWinsAndRedeliveryIsDropped() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"), "first delivery must be processed");
|
||||
assertFalse(d.claim(1L, "msg-1"), "redelivery must be dropped");
|
||||
assertFalse(d.claim(1L, "msg-1"), "and stay dropped");
|
||||
assertTrue(d.claim(1L, "msg-2"), "a different message is unaffected");
|
||||
}
|
||||
|
||||
@Test
|
||||
void claimsAreScopedPerChannel() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
// Same platform message id on a different channel row: platform ids are
|
||||
// only unique per app, so these must not collide.
|
||||
assertTrue(d.claim(2L, "msg-1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void blankIdentityFailsOpen() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
// No stable identity => cannot tell a redelivery from a new message.
|
||||
// Dropping a real message is worse than answering a redelivery twice.
|
||||
assertTrue(d.claim(1L, null));
|
||||
assertTrue(d.claim(1L, null));
|
||||
assertTrue(d.claim(1L, " "));
|
||||
assertTrue(d.claim(1L, " "));
|
||||
assertEquals(0, d.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void expiredClaimIsRetakeable() throws Exception {
|
||||
ChannelDedupProperties p = props();
|
||||
p.setTtl(Duration.ofMillis(30));
|
||||
InboundMessageDeduplicator d = dedup(p);
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
assertFalse(d.claim(1L, "msg-1"));
|
||||
Thread.sleep(60);
|
||||
assertTrue(d.claim(1L, "msg-1"), "past the TTL the identity is free again");
|
||||
}
|
||||
|
||||
@Test
|
||||
void releaseLetsThePlatformRetryThrough() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
assertFalse(d.claim(1L, "msg-1"));
|
||||
// Never handed off for processing (queue full) — hand the claim back.
|
||||
d.release(1L, "msg-1");
|
||||
assertTrue(d.claim(1L, "msg-1"), "retry must be able to get through");
|
||||
}
|
||||
|
||||
@Test
|
||||
void containsPeeksWithoutClaiming() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
assertFalse(d.contains(1L, "msg-1"), "peek must not claim");
|
||||
assertFalse(d.contains(1L, "msg-1"));
|
||||
assertEquals(0, d.size(), "peeking must leave the register untouched");
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
assertTrue(d.contains(1L, "msg-1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void capacityIsEnforcedAndOldestClaimsGoFirst() {
|
||||
ChannelDedupProperties p = props();
|
||||
p.setMaxSize(10);
|
||||
InboundMessageDeduplicator d = dedup(p);
|
||||
|
||||
for (int i = 0; i < 50; i++) {
|
||||
assertTrue(d.claim(1L, "msg-" + i));
|
||||
}
|
||||
assertTrue(d.size() <= 10, "register must stay bounded, was " + d.size());
|
||||
assertTrue(d.contains(1L, "msg-49"), "the newest claim must survive the trim");
|
||||
assertFalse(d.contains(1L, "msg-0"), "the eldest claim is the one dropped");
|
||||
}
|
||||
|
||||
@Test
|
||||
void disabledSwitchLetsEverythingThrough() {
|
||||
ChannelDedupProperties p = props();
|
||||
p.setEnabled(false);
|
||||
InboundMessageDeduplicator d = dedup(p);
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
assertFalse(d.contains(1L, "msg-1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void clearDropsEveryClaim() {
|
||||
InboundMessageDeduplicator d = dedup(props());
|
||||
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
d.clear();
|
||||
assertEquals(0, d.size());
|
||||
assertTrue(d.claim(1L, "msg-1"));
|
||||
}
|
||||
|
||||
// ---- identity derivation (ChannelMessageRouter.inboundIdentity) ----
|
||||
|
||||
@Test
|
||||
void identityPrefersThePlatformMessageId() {
|
||||
ChannelMessage m = ChannelMessage.builder()
|
||||
.messageId("dt-123")
|
||||
.channelType("dingtalk")
|
||||
.senderId("alice")
|
||||
.timestamp(LocalDateTime.of(2026, 1, 1, 0, 0))
|
||||
.build();
|
||||
|
||||
assertEquals("dt-123", ChannelMessageRouter.inboundIdentity(m));
|
||||
}
|
||||
|
||||
@Test
|
||||
void identityFallsBackToSenderAndTimestamp() {
|
||||
LocalDateTime ts = LocalDateTime.of(2026, 1, 1, 0, 0);
|
||||
ChannelMessage m = ChannelMessage.builder()
|
||||
.channelType("telegram")
|
||||
.senderId("alice")
|
||||
.timestamp(ts)
|
||||
.build();
|
||||
|
||||
// Stable across redeliveries of the same payload.
|
||||
assertEquals("alice@" + ts, ChannelMessageRouter.inboundIdentity(m));
|
||||
}
|
||||
|
||||
@Test
|
||||
void identityIsNullWhenNothingStableExists() {
|
||||
ChannelMessage m = ChannelMessage.builder()
|
||||
.channelType("telegram")
|
||||
.senderId("alice")
|
||||
.build();
|
||||
|
||||
assertNull(ChannelMessageRouter.inboundIdentity(m),
|
||||
"no id and no timestamp => must fail open, not invent a key");
|
||||
assertNull(ChannelMessageRouter.inboundIdentity(null));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user