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