fix(weixin): make inbound message dedup configurable (#1667) (#1733)

ilink can retransmit the same message_id on ACK delay or network jitter,
which caused the bot to process the same user message twice and reply with
two different answers. This was already mitigated on main by an inline map
in dispatchInbound (5-minute window, keyed on from|msg_id|seq|create_time_ms|client_id),
but the cache was hardcoded: no per-deployment override, no opt-out, no
shared helper.

Changes:
- core/dedup: MessageDedup now accepts a configurable TTL via
  NewMessageDedup(ttl); zero-value still uses DefaultDedupTTL (60s). Adds
  nil-receiver safety, so callers that embed an uninitialised cache do not
  panic.
- platform/weixin: replace the inline map with core.MessageDedup, parse
  the new dedup_enabled (default true) and dedup_window_seconds (default
  300) options. DEBUG-level log line on drop.
- Tests: extend core/dedup_test.go with nil-receiver, configurable-TTL and
  race-detector coverage; add platform/weixin/dedup_test.go covering
  default-on, opt-out, custom-window, dispatch drop on duplicate and
  accept-all when disabled.
- config.example.toml: document the new keys (zh+en) for the weixin
  platform block.

No behaviour change for existing deployments that do not touch the new
options: weixin continues to dedup with the same 5-minute window it had
before.

config.example.toml

@@ -1484,6 +1484,18 @@ app_secret = "your-feishu-app-secret"
 # 使其低于触发阈值。设为 0 关闭配额。
 # burst_limit = 4           # max separate messages per window / 每窗口最多独立消息数
 # burst_window_secs = 86400 # window length (default 24h) / 窗口时长（秒，默认24小时）
+#
+# Inbound message dedup (issue #1667):
+# ilink can retransmit the same message_id on ACK delay or jitter, which
+# otherwise causes the bot to reply twice with different answers. The dedup
+# cache is keyed on from + message_id + seq + create_time_ms + client_id.
+# Default: enabled with a 5-minute window. Set dedup_enabled=false to disable.
+# 入站消息去重（issue #1667）：ilink 在 ACK 延迟或网络抖动时会重传相同
+# message_id，导致 bot 回复两次不同的答案。缓存键由 from + message_id + seq
+# + create_time_ms + client_id 组合。默认开启，窗口 5 分钟。设为
+# dedup_enabled=false 可关闭。
+# dedup_enabled = true
+# dedup_window_seconds = 300
 
 # -----------------------------------------------------------------------------
 # Tencent Yuanbao (yuanbao.tencent.com bot platform)

core/dedup.go

@@ -5,7 +5,11 @@ import (
 	"time"
 )
 
-const dedupTTL = 60 * time.Second
+// DefaultDedupTTL is the default dedup window used when a platform does not
+// configure one explicitly. It is intentionally generous — long enough to
+// absorb a network jitter / ACK-delay retry, short enough to keep the map
+// bounded under steady traffic.
+const DefaultDedupTTL = 60 * time.Second
 
 // StartTime is set once at process startup.
 // Platforms use it to discard messages created before the current process started,
@@ -14,15 +18,40 @@ var StartTime = time.Now()
 
 // MessageDedup tracks recently seen message IDs to prevent duplicate processing.
 // Safe for concurrent use.
+//
+// A zero-value MessageDedup uses DefaultDedupTTL; platforms can override via
+// NewMessageDedup(ttl). The struct intentionally stays a value type so existing
+// platform code that embeds `dedup core.MessageDedup` keeps working unchanged.
 type MessageDedup struct {
 	mu   sync.Mutex
 	seen map[string]time.Time
+	ttl  time.Duration // 0 means "use DefaultDedupTTL"
+}
+
+// NewMessageDedup returns a MessageDedup with a custom TTL. ttl <= 0 falls back
+// to DefaultDedupTTL.
+func NewMessageDedup(ttl time.Duration) *MessageDedup {
+	if ttl <= 0 {
+		ttl = DefaultDedupTTL
+	}
+	return &MessageDedup{ttl: ttl}
+}
+
+// TTL returns the dedup window in effect (handy for diagnostics and tests).
+func (d *MessageDedup) TTL() time.Duration {
+	if d == nil {
+		return 0
+	}
+	if d.ttl <= 0 {
+		return DefaultDedupTTL
+	}
+	return d.ttl
 }
 
 // IsDuplicate returns true if msgID was already seen within the TTL window.
 // Empty msgID is never considered a duplicate.
 func (d *MessageDedup) IsDuplicate(msgID string) bool {
-	if msgID == "" {
+	if d == nil || msgID == "" {
 		return false
 	}
 	d.mu.Lock()
@@ -31,8 +60,9 @@ func (d *MessageDedup) IsDuplicate(msgID string) bool {
 		d.seen = make(map[string]time.Time)
 	}
 	now := time.Now()
+	ttl := d.TTL()
 	for k, t := range d.seen {
-		if now.Sub(t) > dedupTTL {
+		if now.Sub(t) > ttl {
 			delete(d.seen, k)
 		}
 	}

core/dedup_test.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"sync"
 	"testing"
 	"time"
 )
@@ -42,6 +43,69 @@ func TestMessageDedup_Concurrent(t *testing.T) {
 	}
 }
 
+// TestMessageDedup_NilReceiver covers the nil-safety path: a nil *MessageDedup
+// must treat every ID as new (the dispatcher may carry an uninitialised cache
+// before factory init runs).
+func TestMessageDedup_NilReceiver(t *testing.T) {
+	var d *MessageDedup
+	if d.IsDuplicate("anything") {
+		t.Error("nil receiver should never report duplicate")
+	}
+	if d.TTL() != 0 {
+		t.Errorf("nil TTL() = %v, want 0", d.TTL())
+	}
+}
+
+// TestMessageDedup_ConfigurableTTL exercises the new configurable window
+// (issue #1667). A fresh cache with TTL=20ms should let an ID re-appear
+// after the window elapses.
+func TestMessageDedup_ConfigurableTTL(t *testing.T) {
+	d := NewMessageDedup(20 * time.Millisecond)
+	if got := d.TTL(); got != 20*time.Millisecond {
+		t.Fatalf("TTL() = %v, want 20ms", got)
+	}
+	if d.IsDuplicate("k") {
+		t.Fatal("first call should not be duplicate")
+	}
+	if !d.IsDuplicate("k") {
+		t.Fatal("second call within window should be duplicate")
+	}
+	time.Sleep(40 * time.Millisecond)
+	if d.IsDuplicate("k") {
+		t.Fatal("after window elapsed, ID should be accepted again")
+	}
+}
+
+// TestMessageDedup_DefaultTTLForZero ensures callers that pass ttl<=0 fall
+// back to DefaultDedupTTL instead of dividing by zero or panicking.
+func TestMessageDedup_DefaultTTLForZero(t *testing.T) {
+	d := NewMessageDedup(0)
+	if got := d.TTL(); got != DefaultDedupTTL {
+		t.Fatalf("TTL() = %v, want DefaultDedupTTL %v", got, DefaultDedupTTL)
+	}
+	d2 := NewMessageDedup(-1 * time.Second)
+	if got := d2.TTL(); got != DefaultDedupTTL {
+		t.Fatalf("negative TTL fallback = %v, want DefaultDedupTTL", got)
+	}
+}
+
+// TestMessageDedup_ConcurrentConfigurable stress-tests the configurable-TTL
+// path so race detector catches any future regression in the cleanup loop.
+func TestMessageDedup_ConcurrentConfigurable(t *testing.T) {
+	d := NewMessageDedup(50 * time.Millisecond)
+	var wg sync.WaitGroup
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func(n int) {
+			defer wg.Done()
+			for j := 0; j < 20; j++ {
+				d.IsDuplicate("k")
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
 func TestIsOldMessage(t *testing.T) {
 	if IsOldMessage(time.Now()) {
 		t.Error("current time should not be considered old")

platform/weixin/dedup_test.go

@@ -0,0 +1,156 @@
+package weixin
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/chenhg5/cc-connect/core"
+)
+
+// TestPickBool covers the boolean parsing helper used for the new
+// dedup_enabled / dedup_window_seconds config keys (issue #1667).
+func TestPickBool(t *testing.T) {
+	cases := []struct {
+		in  any
+		want bool
+	}{
+		{true, true},
+		{false, false},
+		{1, true},
+		{0, false},
+		{int64(2), true},
+		{float64(0.5), true},
+		{float64(0), false},
+		{"true", true},
+		{"TRUE", true},
+		{"yes", true},
+		{"on", true},
+		{"1", true},
+		{"false", false},
+		{"", false},
+		{nil, false},
+	}
+	for _, c := range cases {
+		if got := pickBool(c.in); got != c.want {
+			t.Errorf("pickBool(%v) = %v, want %v", c.in, got, c.want)
+		}
+	}
+}
+
+// TestNew_DedupDefaultsToEnabled guards the contract that weixin ships with
+// dedup on by default (ilink can retransmit on ACK delay, see issue #1667).
+func TestNew_DedupDefaultsToEnabled(t *testing.T) {
+	p, err := New(map[string]any{"token": "t"})
+	if err != nil {
+		t.Fatalf("New error = %v", err)
+	}
+	plat := p.(*Platform)
+	if !plat.dedupEnabled {
+		t.Fatal("dedup should default to true for weixin")
+	}
+	if plat.dedup == nil {
+		t.Fatal("dedup cache should be initialised by factory")
+	}
+	if plat.dedup.TTL() != 5*time.Minute {
+		t.Fatalf("default TTL = %v, want 5m", plat.dedup.TTL())
+	}
+}
+
+// TestNew_DedupCanBeDisabled verifies operators can opt out, e.g. for
+// debugging or when they run a downstream component that already dedups.
+func TestNew_DedupCanBeDisabled(t *testing.T) {
+	p, err := New(map[string]any{
+		"token":         "t",
+		"dedup_enabled": false,
+	})
+	if err != nil {
+		t.Fatalf("New error = %v", err)
+	}
+	plat := p.(*Platform)
+	if plat.dedupEnabled {
+		t.Fatal("dedup_enabled=false should be respected")
+	}
+}
+
+// TestNew_DedupWindowRespected covers the per-deployment TTL override.
+func TestNew_DedupWindowRespected(t *testing.T) {
+	p, err := New(map[string]any{
+		"token":                 "t",
+		"dedup_window_seconds":  15,
+	})
+	if err != nil {
+		t.Fatalf("New error = %v", err)
+	}
+	plat := p.(*Platform)
+	if plat.dedup.TTL() != 15*time.Second {
+		t.Fatalf("TTL = %v, want 15s", plat.dedup.TTL())
+	}
+}
+
+// TestDispatchInbound_DropsDuplicateMessageID exercises the regression path
+// from issue #1667: ilink sends the same MessageID twice; only the first
+// must reach the handler.
+func TestDispatchInbound_DropsDuplicateMessageID(t *testing.T) {
+	p, err := New(map[string]any{
+		"token":                 "t",
+		"dedup_window_seconds":  60,
+	})
+	if err != nil {
+		t.Fatalf("New error = %v", err)
+	}
+	plat := p.(*Platform)
+
+	var calls int
+	h := func(_ core.Platform, _ *core.Message) { calls++ }
+
+	msg := &weixinMessage{
+		MessageType:  messageTypeUser,
+		FromUserID:   "u-1",
+		MessageID:    7492913259736648968,
+		Seq:          42,
+		CreateTimeMs: time.Now().UnixMilli(),
+		ClientID:     "client-a",
+		ItemList:     []messageItem{{Type: messageItemText, TextItem: &textItem{Text: "hi"}}},
+	}
+
+	plat.dispatchInbound(context.Background(), msg, h)
+	if calls != 1 {
+		t.Fatalf("first call: handler calls = %d, want 1", calls)
+	}
+	// Same key — must be dropped silently.
+	plat.dispatchInbound(context.Background(), msg, h)
+	if calls != 1 {
+		t.Fatalf("duplicate should be dropped, handler calls = %d, want 1", calls)
+	}
+}
+
+// TestDispatchInbound_DedupDisabledAcceptsDuplicates confirms the toggle
+// actually disables the cache (so QA can verify the regression guard).
+func TestDispatchInbound_DedupDisabledAcceptsDuplicates(t *testing.T) {
+	p, err := New(map[string]any{
+		"token":         "t",
+		"dedup_enabled": false,
+	})
+	if err != nil {
+		t.Fatalf("New error = %v", err)
+	}
+	plat := p.(*Platform)
+
+	var calls int
+	h := func(_ core.Platform, _ *core.Message) { calls++ }
+	msg := &weixinMessage{
+		MessageType:  messageTypeUser,
+		FromUserID:   "u-1",
+		MessageID:    100,
+		Seq:          1,
+		CreateTimeMs: time.Now().UnixMilli(),
+		ClientID:     "client-a",
+		ItemList:     []messageItem{{Type: messageItemText, TextItem: &textItem{Text: "hi"}}},
+	}
+	plat.dispatchInbound(context.Background(), msg, h)
+	plat.dispatchInbound(context.Background(), msg, h)
+	if calls != 2 {
+		t.Fatalf("dedup_enabled=false: handler calls = %d, want 2 (both should pass)", calls)
+	}
+}

platform/weixin/weixin.go

@@ -81,8 +81,12 @@ type Platform struct {
 	syncBuf     string
 	syncBufPath string
 
-	dedupMu sync.Mutex
-	dedup   map[string]time.Time
+	// dedupEnabled mirrors the optional dedup_enabled config key (default true for
+	// weixin because ilink can retransmit the same message_id on ACK delay —
+	// see issue #1667). When false, the dispatch path skips the cache entirely.
+	dedupEnabled bool
+	// dedup tracks recently seen composite keys to absorb retransmissions.
+	dedup *core.MessageDedup
 
 	pauseMu    sync.Mutex
 	pauseUntil time.Time
@@ -196,6 +200,18 @@ func New(opts map[string]any) (core.Platform, error) {
 		burstWindow = defaultBurstWindowSecs
 	}
 
+	// dedup_enabled (default true) and dedup_window_seconds (default 300s, the
+	// legacy value) — see issue #1667. ilink can retransmit the same message_id
+	// on ACK delay; without dedup the bot replies twice.
+	dedupEnabled := pickBool(opts["dedup_enabled"])
+	if _, ok := opts["dedup_enabled"]; !ok {
+		dedupEnabled = true
+	}
+	dedupWindow := pickInt(opts["dedup_window_seconds"])
+	if dedupWindow <= 0 {
+		dedupWindow = 300
+	}
+
 	p := &Platform{
 		token:           token,
 		baseURL:         baseURL,
@@ -208,7 +224,8 @@ func New(opts map[string]any) (core.Platform, error) {
 		httpClient:      httpClient,
 		cdnHttpClient:   cdnHttpClient,
 		tokens:          make(map[string]string),
-		dedup:           make(map[string]time.Time),
+		dedupEnabled:    dedupEnabled,
+		dedup:           core.NewMessageDedup(time.Duration(dedupWindow) * time.Second),
 		typingTickets:   make(map[string]typingTicketEntry),
 		sendQuotaLimit:  burstLimit,
 		sendQuotaWindow: time.Duration(burstWindow) * time.Second,
@@ -241,6 +258,26 @@ func pickInt(v any) int {
 	}
 }
 
+// pickBool interprets an any value as a bool. Numeric values are treated as
+// "non-zero => true" so the same key works for both booleans and 0/1 ints.
+func pickBool(v any) bool {
+	switch x := v.(type) {
+	case bool:
+		return x
+	case int:
+		return x != 0
+	case int64:
+		return x != 0
+	case float64:
+		return x != 0
+	case string:
+		s := strings.ToLower(strings.TrimSpace(x))
+		return s == "true" || s == "1" || s == "yes" || s == "on"
+	default:
+		return false
+	}
+}
+
 func (p *Platform) Name() string { return "weixin" }
 
 func (p *Platform) loadSyncBuf() {
@@ -485,24 +522,15 @@ func (p *Platform) dispatchInbound(ctx context.Context, m *weixinMessage, h core
 		}
 	}
 
-	// Include create_time_ms and client_id so (seq,message_id)=(0,0) or duplicates are less likely to collide.
+	// Include create_time_ms and client_id so (seq,message_id)=(0,0) or duplicates
+	// are less likely to collide. Dedup window is configurable via
+	// dedup_window_seconds; the cache is a thin wrapper over core.MessageDedup
+	// (see issue #1667).
 	dedupKey := fmt.Sprintf("%s|%d|%d|%d|%s", from, m.MessageID, m.Seq, m.CreateTimeMs, strings.TrimSpace(m.ClientID))
-	p.dedupMu.Lock()
-	if p.dedup == nil {
-		p.dedup = make(map[string]time.Time)
-	}
-	now := time.Now()
-	for k, ts := range p.dedup {
-		if now.Sub(ts) > 5*time.Minute {
-			delete(p.dedup, k)
-		}
-	}
-	if _, ok := p.dedup[dedupKey]; ok {
-		p.dedupMu.Unlock()
+	if p.dedupEnabled && p.dedup != nil && p.dedup.IsDuplicate(dedupKey) {
+		slog.Debug("weixin: dropping duplicate message", "from", from, "message_id", m.MessageID, "seq", m.Seq)
 		return
 	}
-	p.dedup[dedupKey] = now
-	p.dedupMu.Unlock()
 
 	if tok := strings.TrimSpace(m.ContextToken); tok != "" {
 		p.setContextToken(from, tok)

platform/weixin/weixin_test.go

@@ -288,7 +288,7 @@ func TestPollLoop_NotifiesReadyForPollAfterFirstSuccessfulGetUpdates(t *testing.
 		longPollMS:    100,
 		accountLabel:  "default",
 		httpClient:    &http.Client{},
-		dedup:         make(map[string]time.Time),
+		dedup:         core.NewMessageDedup(5 * time.Minute),
 		typingTickets: make(map[string]typingTicketEntry),
 	}
 	p.api = newAPIClient(srv.URL(), "tok", "", p.httpClient)
@@ -330,7 +330,7 @@ func TestPollLoop_DoesNotNotifyReadyForPollWhileGetUpdatesFails(t *testing.T) {
 		longPollMS:    100,
 		accountLabel:  "default",
 		httpClient:    &http.Client{},
-		dedup:         make(map[string]time.Time),
+		dedup:         core.NewMessageDedup(5 * time.Minute),
 		typingTickets: make(map[string]typingTicketEntry),
 	}
 	p.api = newAPIClient(srv.URL(), "tok", "", p.httpClient)