From 53704b1c6963e9df5dcddbaac27755af0ee4da38 Mon Sep 17 00:00:00 2001 From: khanhtranhd Date: Mon, 13 Apr 2026 21:13:28 +0700 Subject: [PATCH] fix(facebook): preserve fb_mode in outbound routing + admin reply detection (#860) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(facebook): preserve fb_mode metadata in outbound routing + admin reply detection Two issues fixed: 1. Messenger auto-reply never delivered because fb_mode metadata was stripped during outbound message construction. The routing whitelist in gateway_consumer_normal.go and channels/events.go only copied thread_id/local_key/group_id — facebook-specific keys (fb_mode, sender_id, page_id, reply_to_comment_id) were dropped, causing facebook.Send() to fall into the comment path and fail with "reply_to_comment_id missing". 2. Added admin reply detection: before sending a bot reply, check via Graph API if the last page message in the conversation was sent by an admin (human) rather than the bot itself. Skips bot reply when admin already responded, preventing duplicate messages. Uses timestamp comparison with bot's own send history to distinguish bot-sent vs admin-sent page messages (both have from.id = page_id). * chore: exclude compiled binary from git * fix(facebook): ignore bot echoes in messenger cooldown * fix(facebook): add memory cleanup for admin reply maps and reduce echo window - Add adminReplied and botSentAt eviction to runDedupCleaner to prevent unbounded memory growth on high-traffic pages - Reduce botEchoWindow from 60s to 15s to avoid misclassifying real admin replies as bot echoes - Restore doc comments on routingMetaKeys and copyRoutingMeta - Add cross-reference comment between consumer and events routing key lists - Simplify admin-reply skip log (remove redundant map re-read) --------- Co-authored-by: khanhtran <> Co-authored-by: Plateau Nguyen Co-authored-by: viettranx --- .gitignore | 1 + cmd/gateway_consumer_normal.go | 7 +- internal/channels/events.go | 6 +- internal/channels/facebook/facebook.go | 80 ++++++++++++++-- internal/channels/facebook/graph_api.go | 1 - internal/channels/facebook/handlers_test.go | 92 +++++++++++++++++++ .../channels/facebook/messenger_handler.go | 33 ++++++- 7 files changed, 208 insertions(+), 12 deletions(-) diff --git a/.gitignore b/.gitignore index 62bbd755..e12ef95f 100644 --- a/.gitignore +++ b/.gitignore @@ -88,3 +88,4 @@ compose.d/* /mise.*.toml /mise /.mise +goclaw-patched-linux-amd64 diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index 0180f080..85313c5f 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -220,7 +220,12 @@ func processNormalMessage( outMeta["reply_to_message_id"] = mid } } - for _, k := range []string{tools.MetaMessageThreadID, "local_key", "placeholder_key", "group_id", "feishu_reply_target_id"} { + // Channel routing keys — keep in sync with routingMetaKeys in channels/events.go. + for _, k := range []string{ + tools.MetaMessageThreadID, "local_key", "placeholder_key", "group_id", + "feishu_reply_target_id", + "fb_mode", "sender_id", "page_id", "reply_to_comment_id", + } { if v := msg.Metadata[k]; v != "" { outMeta[k] = v } diff --git a/internal/channels/events.go b/internal/channels/events.go index 127714fd..21faffbd 100644 --- a/internal/channels/events.go +++ b/internal/channels/events.go @@ -354,7 +354,11 @@ var routingMetaKeys = []string{ "message_thread_id", // telegram forum topics "local_key", // composite chat-id suffix "group_id", // legacy group identifier - "feishu_reply_target_id", // feishu/lark thread reply routing (issue #818) + "feishu_reply_target_id", // feishu/lark thread reply routing + "fb_mode", // facebook messenger vs comment routing + "sender_id", // facebook sender for first-inbox + "page_id", // facebook page routing + "reply_to_comment_id", // facebook comment reply target } // copyRoutingMeta copies channel routing metadata from RunContext.Metadata diff --git a/internal/channels/facebook/facebook.go b/internal/channels/facebook/facebook.go index 71d4d4f7..5b0ba32a 100644 --- a/internal/channels/facebook/facebook.go +++ b/internal/channels/facebook/facebook.go @@ -21,9 +21,11 @@ var ( ) const ( - webhookPath = "/channels/facebook/webhook" - dedupTTL = 24 * time.Hour // matches Facebook's max retry window - dedupCleanEvery = 5 * time.Minute // how often to evict stale dedup entries + webhookPath = "/channels/facebook/webhook" + dedupTTL = 24 * time.Hour // matches Facebook's max retry window + dedupCleanEvery = 5 * time.Minute // how often to evict stale dedup entries + adminReplyCooldown = 5 * time.Minute + botEchoWindow = 15 * time.Second ) // Channel implements channels.Channel and channels.WebhookChannel for Facebook Fanpage. @@ -42,6 +44,14 @@ type Channel struct { // firstInboxSent tracks which senders have already received a first-inbox DM (in-memory). firstInboxSent sync.Map // senderID(string) → struct{} + // adminReplied tracks conversations where admin (page) sent a message recently. + // Bot skips auto-reply for these conversations to avoid duplicate responses. + adminReplied sync.Map // chatID(string) → time.Time + + // botSentAt tracks when bot last sent a reply to each conversation. + // Used to distinguish bot replies from admin replies in Graph API checks. + botSentAt sync.Map // chatID(string) → time.Time + // postFetcher caches post content to enrich comment context. postFetcher *PostFetcher @@ -148,27 +158,71 @@ func (ch *Channel) Start(ctx context.Context) error { // Stop gracefully shuts down the channel. func (ch *Channel) Stop(_ context.Context) error { globalRouter.unregister(ch.pageID) - ch.stopFn() // cancel stopCtx → cancels inflight Graph API calls - close(ch.stopCh) // stop background goroutines + ch.stopFn() // cancel stopCtx → cancels inflight Graph API calls + close(ch.stopCh) // stop background goroutines ch.SetRunning(false) ch.MarkStopped("stopped") slog.Info("facebook channel stopped", "page_id", ch.pageID, "name", ch.Name()) return nil } +func (ch *Channel) adminRepliedRecently(chatID string, now time.Time) bool { + val, ok := ch.adminReplied.Load(chatID) + if !ok { + return false + } + repliedAt, ok := val.(time.Time) + if !ok { + ch.adminReplied.Delete(chatID) + return false + } + if now.Sub(repliedAt) < adminReplyCooldown { + return true + } + ch.adminReplied.Delete(chatID) + return false +} + +func (ch *Channel) isBotEcho(chatID string, eventAt time.Time) bool { + val, ok := ch.botSentAt.Load(chatID) + if !ok { + return false + } + sentAt, ok := val.(time.Time) + if !ok { + ch.botSentAt.Delete(chatID) + return false + } + return eventAt.Sub(sentAt).Abs() < botEchoWindow +} + // Send delivers an outbound message. Dispatches to comment reply or Messenger based on fb_mode metadata. func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { mode := msg.Metadata["fb_mode"] switch mode { case "messenger": + if ch.adminRepliedRecently(msg.ChatID, time.Now()) { + slog.Info("facebook: skipping bot reply (admin already responded)", + "chat_id", msg.ChatID) + return nil + } + text := FormatForMessenger(msg.Content) parts := splitMessage(text, messengerMaxChars) + sentAt := time.Now() + ch.botSentAt.Store(msg.ChatID, sentAt) + sentAny := false for _, part := range parts { if _, err := ch.graphClient.SendMessage(ctx, msg.ChatID, part); err != nil { + if !sentAny { + ch.botSentAt.Delete(msg.ChatID) + } ch.handleAPIError(err) return err } + sentAny = true + ch.botSentAt.Store(msg.ChatID, time.Now()) } default: // "comment" @@ -232,7 +286,8 @@ func (ch *Channel) sendFirstInbox(ctx context.Context, senderID string) { } } -// runDedupCleaner evicts dedup entries older than dedupTTL every dedupCleanEvery. +// runDedupCleaner evicts stale entries from dedup, adminReplied, and botSentAt +// maps every dedupCleanEvery to prevent unbounded memory growth. func (ch *Channel) runDedupCleaner() { ticker := time.NewTicker(dedupCleanEvery) defer ticker.Stop() @@ -248,6 +303,18 @@ func (ch *Channel) runDedupCleaner() { } return true }) + ch.adminReplied.Range(func(k, v any) bool { + if t, ok := v.(time.Time); ok && now.Sub(t) > adminReplyCooldown { + ch.adminReplied.Delete(k) + } + return true + }) + ch.botSentAt.Range(func(k, v any) bool { + if t, ok := v.(time.Time); ok && now.Sub(t) > botEchoWindow { + ch.botSentAt.Delete(k) + } + return true + }) } } } @@ -257,4 +324,3 @@ func (ch *Channel) isDup(key string) bool { _, loaded := ch.dedup.LoadOrStore(key, time.Now()) return loaded } - diff --git a/internal/channels/facebook/graph_api.go b/internal/channels/facebook/graph_api.go index b9976eec..10ab11ff 100644 --- a/internal/channels/facebook/graph_api.go +++ b/internal/channels/facebook/graph_api.go @@ -361,4 +361,3 @@ func IsRateLimitError(err error) bool { } return ge.code == 4 || ge.code == 17 || ge.code == 32 || ge.code == 613 } - diff --git a/internal/channels/facebook/handlers_test.go b/internal/channels/facebook/handlers_test.go index c90f8183..cf0bab5b 100644 --- a/internal/channels/facebook/handlers_test.go +++ b/internal/channels/facebook/handlers_test.go @@ -337,6 +337,98 @@ func TestSend_MessengerMode(t *testing.T) { } } +// TestHandleMessagingEvent_PageEchoDoesNotTrackAdminReply verifies webhook +// echoes for bot-sent messages do not arm the admin-reply cooldown. +func TestHandleMessagingEvent_PageEchoDoesNotTrackAdminReply(t *testing.T) { + cfg := facebookInstanceConfig{} + cfg.Features.MessengerAutoReply = true + ch := newTestChannel(t, "111", cfg) + + now := time.Now() + ch.botSentAt.Store("user-1", now) + + ch.handleMessagingEvent(WebhookEntry{ID: "111"}, MessagingEvent{ + Sender: FBUser{ID: "111"}, + Recipient: FBUser{ID: "user-1"}, + Timestamp: now.UnixMilli(), + Message: &IncomingMessage{MID: "echo-1", Text: "bot reply"}, + }) + + if _, ok := ch.adminReplied.Load("user-1"); ok { + t.Fatal("bot echo should not be tracked as admin reply") + } +} + +// TestSend_MessengerModeSkipsWhenAdminReplyTracked verifies bot replies are +// suppressed when an admin reply was observed recently via webhook. +func TestSend_MessengerModeSkipsWhenAdminReplyTracked(t *testing.T) { + var sends int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sends++ + _, _ = w.Write([]byte(`{"message_id":"mid"}`)) + })) + t.Cleanup(srv.Close) + swapGraphBase(t, srv.URL) + + mb := bus.New() + ch, _ := New(facebookInstanceConfig{PageID: "111"}, facebookCreds{ + PageAccessToken: "t", AppSecret: "s", VerifyToken: "v", + }, mb, nil) + ch.adminReplied.Store("user-1", time.Now()) + + err := ch.Send(t.Context(), bus.OutboundMessage{ + ChatID: "user-1", + Content: "hello", + Metadata: map[string]string{"fb_mode": "messenger"}, + }) + if err != nil { + t.Fatalf("Send: %v", err) + } + if sends != 0 { + t.Fatalf("sends = %d, want 0 when admin already replied", sends) + } +} + +// TestSend_MessengerModeDoesNotConsultGraphHistory verifies historical page +// messages do not suppress replies after process restart. +func TestSend_MessengerModeDoesNotConsultGraphHistory(t *testing.T) { + var gets, posts int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodGet: + gets++ + _, _ = w.Write([]byte(`{"data":[{"messages":{"data":[{"from":{"id":"111"},"created_time":"2026-04-13T00:00:00+0000"}]}}]}`)) + case r.Method == http.MethodPost: + posts++ + _, _ = w.Write([]byte(`{"message_id":"mid"}`)) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } + })) + t.Cleanup(srv.Close) + swapGraphBase(t, srv.URL) + + mb := bus.New() + ch, _ := New(facebookInstanceConfig{PageID: "111"}, facebookCreds{ + PageAccessToken: "t", AppSecret: "s", VerifyToken: "v", + }, mb, nil) + + err := ch.Send(t.Context(), bus.OutboundMessage{ + ChatID: "user-1", + Content: "hello", + Metadata: map[string]string{"fb_mode": "messenger"}, + }) + if err != nil { + t.Fatalf("Send: %v", err) + } + if gets != 0 { + t.Fatalf("unexpected graph history check: gets = %d, want 0", gets) + } + if posts != 1 { + t.Fatalf("posts = %d, want 1", posts) + } +} + // TestSend_CommentModeMissingMetadata verifies the comment path errors when // reply_to_comment_id metadata is absent. func TestSend_CommentModeMissingMetadata(t *testing.T) { diff --git a/internal/channels/facebook/messenger_handler.go b/internal/channels/facebook/messenger_handler.go index 6fcbd450..524f6317 100644 --- a/internal/channels/facebook/messenger_handler.go +++ b/internal/channels/facebook/messenger_handler.go @@ -3,6 +3,7 @@ package facebook import ( "fmt" "log/slog" + "time" ) // handleMessagingEvent processes a Messenger inbox event. @@ -17,8 +18,19 @@ func (ch *Channel) handleMessagingEvent(entry WebhookEntry, event MessagingEvent return } - // Self-message prevention: skip messages sent by the page itself. + // Track admin (page) replies: when the page itself sends a message, + // record the recipient's chat ID so the bot skips auto-reply for that + // conversation during the cooldown window. if event.Sender.ID == ch.pageID { + if event.Recipient.ID != "" { + eventAt := messagingEventTime(event.Timestamp) + if ch.isBotEcho(event.Recipient.ID, eventAt) { + slog.Debug("facebook: bot echo ignored", "chat_id", event.Recipient.ID) + return + } + ch.adminReplied.Store(event.Recipient.ID, eventAt) + slog.Debug("facebook: admin reply tracked", "chat_id", event.Recipient.ID) + } return } @@ -40,6 +52,13 @@ func (ch *Channel) handleMessagingEvent(entry WebhookEntry, event MessagingEvent return } + // Check if admin already replied to this conversation recently. + senderID := event.Sender.ID + if ch.adminRepliedRecently(senderID, time.Now()) { + slog.Info("facebook: skipping auto-reply (admin replied recently)", "chat_id", senderID) + return + } + // Extract text content. var content string switch { @@ -52,7 +71,6 @@ func (ch *Channel) handleMessagingEvent(entry WebhookEntry, event MessagingEvent return } - senderID := event.Sender.ID // Messenger sessions are 1:1: chatID = senderID (channel name scopes the session). chatID := senderID @@ -68,3 +86,14 @@ func (ch *Channel) handleMessagingEvent(entry WebhookEntry, event MessagingEvent ch.HandleMessage(senderID, chatID, content, nil, metadata, "direct") } + +func messagingEventTime(ts int64) time.Time { + switch { + case ts > 1_000_000_000_000: + return time.UnixMilli(ts) + case ts > 0: + return time.Unix(ts, 0) + default: + return time.Now() + } +}