添加 MQTT 消息去重队列：Hook 层基于 payload+topic hash 去重，TTL 15 秒，定时清理

internal/mqttforward/dedup_queue.go

@@ -0,0 +1,72 @@
+package mqttforward
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sync"
+	"time"
+)
+
+const dedupTTL = 15 * time.Second
+
+type DedupQueue struct {
+	mu      sync.Mutex
+	entries map[string]time.Time
+	stopCh  chan struct{}
+}
+
+func NewDedupQueue() *DedupQueue {
+	return &DedupQueue{
+		entries: make(map[string]time.Time),
+		stopCh:  make(chan struct{}),
+	}
+}
+
+func (dq *DedupQueue) TryForward(topic string, payload []byte) bool {
+	hash := dedupHash(topic, payload)
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	if expiry, ok := dq.entries[hash]; ok && now.Before(expiry) {
+		return false
+	}
+	dq.entries[hash] = now.Add(dedupTTL)
+	return true
+}
+
+func (dq *DedupQueue) Start() {
+	go func() {
+		ticker := time.NewTicker(dedupTTL)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				dq.cleanup()
+			case <-dq.stopCh:
+				return
+			}
+		}
+	}()
+}
+
+func (dq *DedupQueue) Stop() {
+	close(dq.stopCh)
+}
+
+func (dq *DedupQueue) cleanup() {
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	for hash, expiry := range dq.entries {
+		if now.After(expiry) {
+			delete(dq.entries, hash)
+		}
+	}
+}
+
+func dedupHash(topic string, payload []byte) string {
+	h := sha256.New()
+	h.Write([]byte(topic))
+	h.Write(payload)
+	return hex.EncodeToString(h.Sum(nil))
+}

main.go

@@ -57,8 +57,9 @@ type meshtasticFilterHook struct {
 	settings        *rspkg.Cache
 	pkiResolver     func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
 	autoAcker       func(record map[string]any)
-	consoleLog      bool // 控制台是否打印 MQTT 连接/订阅事件
+	consoleLog       bool // 控制台是否打印 MQTT 连接/订阅事件
 	packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
+	dedupQueue       *mqttforwardpkg.DedupQueue
 }
 
 // ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
@@ -225,6 +226,10 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
 		h.rejectPublish(cl, pk, record)
 		return pk, packets.ErrRejectPacket
 	}
+	if h.dedupQueue != nil && !h.dedupQueue.TryForward(pk.TopicName, pk.Payload) {
+		h.stats.IncDropped()
+		return pk, packets.ErrRejectPacket
+	}
 	h.stats.IncForwarded()
 
 	h.dbQueue.EnqueueRecord(record, mqttClientInfoFromClient(cl))
@@ -521,6 +526,9 @@ func run(cfg *configpkg.Config) error {
 	if err := server.Close(); err != nil && runErr == nil {
 		runErr = err
 	}
+	if mqttHook.dedupQueue != nil {
+		mqttHook.dedupQueue.Stop()
+	}
 	return runErr
 }
 
@@ -529,6 +537,8 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 	if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
 		return nil, nil, "", err
 	}
+	dedupQueue := mqttforwardpkg.NewDedupQueue()
+	dedupQueue.Start()
 	hook := &meshtasticFilterHook{
 		server:           server,
 		key:              cfg.Key,
@@ -540,6 +550,7 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 		pkiResolver:      botpkg.NewPKIKeyResolver(store),
 		consoleLog:       cfg.ConsoleLog.MQTT,
 		packetConsoleLog: cfg.ConsoleLog.Meshtastic,
+		dedupQueue:       dedupQueue,
 	}
 	if err := server.AddHook(hook, nil); err != nil {
 		return nil, nil, "", err