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. + * + *
入站渠道消息去重配置。平台重投同一条消息时，若不去重则每次重投都会跑一轮完整 + * 的 Agent 回合，用户看到重复答复。 + * + *
+ * mate: + * channel: + * dedup: + * enabled: true + * ttl: 5m + * max-size: 2000 + *+ */ +@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. + * + *
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; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java index 7aedfcd7..23db1dd8 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -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. + * + *
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. + * + *
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. + * + *
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}. + * + *
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, "系统繁忙，请稍后再试"); diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java index d6a85032..aed2c5c7 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java @@ -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. + * + *
{@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. + * + *
Runs after the DB cascade commits — see {@link ConversationDeletedEvent}. + * + *
会话被删除后清理内存缓存，避免留下指向已删除行的幽灵条目。
+ */
+ @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 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.
+ *
+ * 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.
+ *
+ * 删除会话（内存 + 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 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.
+ *
+ * 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.
+ *
+ * Three operations, matching the three things a caller needs:
+ * 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.
+ *
+ * 入站消息去重登记表（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 {@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.
+ *
+ * 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 飞书 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--;
- }
- }
- }
-
// ==================== 消息反应 ====================
/**
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/weixin/WeixinChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/weixin/WeixinChannelAdapter.java
index 1f53b2d3..e415d4d3 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/weixin/WeixinChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/weixin/WeixinChannelAdapter.java
@@ -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 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();
+ }
+ }
+ }
+}
diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java
index fb89bc84..200d547f 100644
--- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java
+++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterNarrationTest.java
@@ -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);
diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelSessionStoreTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelSessionStoreTest.java
new file mode 100644
index 00000000..dfac76db
--- /dev/null
+++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelSessionStoreTest.java
@@ -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());
+ }
+}
diff --git a/mateclaw-server/src/test/java/vip/mate/channel/InboundMessageDeduplicatorTest.java b/mateclaw-server/src/test/java/vip/mate/channel/InboundMessageDeduplicatorTest.java
new file mode 100644
index 00000000..e5ab4d07
--- /dev/null
+++ b/mateclaw-server/src/test/java/vip/mate/channel/InboundMessageDeduplicatorTest.java
@@ -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));
+ }
+}
+ *
+ *
+ *