From aaae3bf1229b4e110eb56dbfac430de4653a0c7d Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 29 Jul 2026 05:43:01 -0400 Subject: [PATCH] fix(channel): claim each inbound message once so redeliveries stop answering twice --- .../mate/channel/ChannelDedupProperties.java | 79 +++++++ .../mate/channel/ChannelMessageRouter.java | 91 +++++++- .../vip/mate/channel/ChannelSessionStore.java | 128 ++++++++---- .../channel/InboundMessageDeduplicator.java | 187 +++++++++++++++++ .../dingtalk/DingTalkChannelAdapter.java | 51 ++++- .../discord/DiscordChannelAdapter.java | 19 +- .../channel/feishu/FeishuChannelAdapter.java | 27 +-- .../channel/weixin/WeixinChannelAdapter.java | 32 ++- .../wiki/service/WikiProcessingService.java | 2 +- .../src/main/resources/application.yml | 8 + .../mate/channel/ChannelMagicCommandTest.java | 3 +- .../ChannelMessageRouterApprovalDenyTest.java | 3 +- ...nelMessageRouterExecutionMetadataTest.java | 3 +- .../ChannelMessageRouterInboundDedupTest.java | 196 ++++++++++++++++++ .../ChannelMessageRouterNarrationTest.java | 3 +- .../mate/channel/ChannelSessionStoreTest.java | 87 ++++++++ .../InboundMessageDeduplicatorTest.java | 166 +++++++++++++++ 17 files changed, 974 insertions(+), 111 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/InboundMessageDeduplicator.java create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterInboundDedupTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/ChannelSessionStoreTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/channel/InboundMessageDeduplicatorTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java new file mode 100644 index 00000000..08227874 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java @@ -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. + * + *

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() - .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() + .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. + * + *

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() + .eq(ChannelSessionEntity::getConversationId, conversationId)); + if (deleted > 0) { log.debug("Removed channel session: conversationId={}", conversationId); } + return deleted; } } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/InboundMessageDeduplicator.java b/mateclaw-server/src/main/java/vip/mate/channel/InboundMessageDeduplicator.java new file mode 100644 index 00000000..9d02461e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/InboundMessageDeduplicator.java @@ -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. + * + *

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 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> 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; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java index 86453c85..3199a6fd 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java @@ -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. + * + *

{@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 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; diff --git a/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java index 86658eea..803ee926 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/discord/DiscordChannelAdapter.java @@ -60,14 +60,6 @@ public class DiscordChannelAdapter extends AbstractChannelAdapter { /** 媒体下载用 HttpClient（复用 http_proxy 配置） */ private volatile HttpClient mediaHttpClient; - /** 已处理消息去重（LRU，最多保留 500 条） */ - private final Set processedMessageIds = Collections.newSetFromMap(new LinkedHashMap<>() { - @Override - protected boolean removeEldestEntry(Map.Entry 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; diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java index 8847a187..e8cc4304 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java @@ -84,9 +84,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre /** 定时 Token 刷新任务 */ private ScheduledFuture tokenRefreshFuture; - /** 消息去重：最近处理过的 message_id */ - private final Set processedMessageIds = ConcurrentHashMap.newKeySet(); - /** * 群内 bot 别名缓存：chatId → 学到的别名集合（openId / unionId / userId / name）。 *

飞书 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 processedIds = new LinkedHashMap<>(256, 0.75f, true) { - @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - return size() > PROCESSED_IDS_MAX; - } - }; - /** 用户最新 context_token 缓存（用于主动推送） */ private final ConcurrentHashMap 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 不提供昵称 diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java index 5fca7f82..1d8cc1e5 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java @@ -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 = diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index 0443b848..cd3d9082 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -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 diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMagicCommandTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMagicCommandTest.java index a79b1c5c..1a18221c 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMagicCommandTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMagicCommandTest.java @@ -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 diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterApprovalDenyTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterApprovalDenyTest.java index ccc7d50b..a5be58aa 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterApprovalDenyTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterApprovalDenyTest.java @@ -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"); diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterExecutionMetadataTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterExecutionMetadataTest.java index 2725aa55..a734a567 100644 --- a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterExecutionMetadataTest.java +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterExecutionMetadataTest.java @@ -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); diff --git a/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterInboundDedupTest.java b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterInboundDedupTest.java new file mode 100644 index 00000000..ee330219 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/channel/ChannelMessageRouterInboundDedupTest.java @@ -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. + * + *

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)); + } +}