diff --git a/internal/channels/pancake/api_client.go b/internal/channels/pancake/api_client.go index 507077b4..77f369ba 100644 --- a/internal/channels/pancake/api_client.go +++ b/internal/channels/pancake/api_client.go @@ -163,6 +163,69 @@ func (c *APIClient) UploadMedia(ctx context.Context, filename string, data io.Re return uploadResp.ID, nil } +// ReplyComment sends a reply to a comment on a post (action: "reply_comment"). +func (c *APIClient) ReplyComment(ctx context.Context, conversationID, content string) error { + body, _ := json.Marshal(SendMessageRequest{ + Action: "reply_comment", + Message: content, + }) + url := fmt.Sprintf("%s/pages/%s/conversations/%s/messages", c.pageV1BaseURL, c.pageID, conversationID) + if err := c.doRequest(ctx, http.MethodPost, url, bytes.NewReader(body)); err != nil { + return fmt.Errorf("pancake: reply comment: %w", err) + } + return nil +} + +// PrivateReply sends a one-time DM to a commenter (first-inbox, action: "private_reply"). +func (c *APIClient) PrivateReply(ctx context.Context, conversationID, content string) error { + body, _ := json.Marshal(SendMessageRequest{ + Action: "private_reply", + Message: content, + }) + url := fmt.Sprintf("%s/pages/%s/conversations/%s/messages", c.pageV1BaseURL, c.pageID, conversationID) + if err := c.doRequest(ctx, http.MethodPost, url, bytes.NewReader(body)); err != nil { + return fmt.Errorf("pancake: private reply: %w", err) + } + return nil +} + +// GetPosts fetches recent posts for the page (used by PostFetcher for comment context enrichment). +// Bypasses doRequest because it needs to parse a JSON response body (doRequest discards the body). +// Connection reuse is preserved: defer res.Body.Close() + io.ReadAll drain the body fully. +func (c *APIClient) GetPosts(ctx context.Context, limit int) ([]PancakePost, error) { + if limit <= 0 { + limit = 20 + } + url := fmt.Sprintf("%s/pages/%s/posts?limit=%d", c.pageV2BaseURL, c.pageID, limit) + req, err := c.newPageRequest(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("pancake: build get-posts request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + res, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("pancake: get posts request failed: %w", err) + } + defer res.Body.Close() + + respBody, err := io.ReadAll(res.Body) + if err != nil { + return nil, fmt.Errorf("pancake: read get-posts response: %w", err) + } + if res.StatusCode >= 400 { + return nil, fmt.Errorf("pancake: get posts HTTP %d", res.StatusCode) + } + + var result struct { + Data []PancakePost `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return nil, fmt.Errorf("pancake: parse get-posts response: %w", err) + } + return result.Data, nil +} + // doRequest executes an authenticated HTTP request using the page_access_token. // Always drains and closes the response body to enable connection reuse. func (c *APIClient) doRequest(ctx context.Context, method, url string, body io.Reader) error { diff --git a/internal/channels/pancake/api_client_test.go b/internal/channels/pancake/api_client_test.go new file mode 100644 index 00000000..db6a974c --- /dev/null +++ b/internal/channels/pancake/api_client_test.go @@ -0,0 +1,226 @@ +package pancake + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// --- ReplyComment --- + +func TestReplyComment_MatchesOfficialContract(t *testing.T) { + transport := &captureTransport{} + client := NewAPIClient("user-token", "page-token", "page-123") + client.httpClient = &http.Client{Transport: transport} + + if err := client.ReplyComment(context.Background(), "conv-123", "thank you"); err != nil { + t.Fatalf("ReplyComment returned error: %v", err) + } + + if transport.req == nil { + t.Fatal("expected outbound request to be captured") + } + wantPath := "/api/public_api/v1/pages/page-123/conversations/conv-123/messages" + if got := transport.req.URL.Path; got != wantPath { + t.Fatalf("request path = %q, want %q", got, wantPath) + } + if got := transport.req.URL.Query().Get("page_access_token"); got != "page-token" { + t.Fatalf("page_access_token = %q, want %q", got, "page-token") + } + + var payload map[string]any + if err := json.Unmarshal(transport.body, &payload); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if got, want := payload["action"], "reply_comment"; got != want { + t.Fatalf("payload.action = %#v, want %#v", got, want) + } + if got, want := payload["message"], "thank you"; got != want { + t.Fatalf("payload.message = %#v, want %#v", got, want) + } +} + +func TestReplyComment_ReturnsError(t *testing.T) { + transport := &captureTransport{ + resp: &http.Response{ + StatusCode: http.StatusBadRequest, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"code":400,"message":"bad request"}`)), + }, + } + client := NewAPIClient("user-token", "page-token", "page-123") + client.httpClient = &http.Client{Transport: transport} + + if err := client.ReplyComment(context.Background(), "conv-123", "thank you"); err == nil { + t.Fatal("expected ReplyComment to return error on HTTP 400") + } +} + +// --- PrivateReply --- + +func TestPrivateReply_MatchesOfficialContract(t *testing.T) { + transport := &captureTransport{} + client := NewAPIClient("user-token", "page-token", "page-123") + client.httpClient = &http.Client{Transport: transport} + + if err := client.PrivateReply(context.Background(), "conv-123", "thanks for commenting!"); err != nil { + t.Fatalf("PrivateReply returned error: %v", err) + } + + if transport.req == nil { + t.Fatal("expected outbound request to be captured") + } + wantPath := "/api/public_api/v1/pages/page-123/conversations/conv-123/messages" + if got := transport.req.URL.Path; got != wantPath { + t.Fatalf("request path = %q, want %q", got, wantPath) + } + + var payload map[string]any + if err := json.Unmarshal(transport.body, &payload); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if got, want := payload["action"], "private_reply"; got != want { + t.Fatalf("payload.action = %#v, want %#v", got, want) + } + if got, want := payload["message"], "thanks for commenting!"; got != want { + t.Fatalf("payload.message = %#v, want %#v", got, want) + } +} + +func TestPrivateReply_ReturnsError(t *testing.T) { + transport := &captureTransport{ + resp: &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"success":false,"message":"conversation not found"}`)), + }, + } + client := NewAPIClient("user-token", "page-token", "page-123") + client.httpClient = &http.Client{Transport: transport} + + if err := client.PrivateReply(context.Background(), "conv-123", "hi"); err == nil { + t.Fatal("expected PrivateReply to return error when success=false") + } +} + +// --- GetPosts --- + +func TestGetPosts_ParsesResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"success":true,"data":[{"id":"post-1","message":"Hello world","created_at":"2024-01-01"}]}`)) + })) + defer srv.Close() + + client := NewAPIClient("user-token", "page-token", "page-123") + client.pageV2BaseURL = srv.URL + client.httpClient = srv.Client() + + posts, err := client.GetPosts(context.Background(), 10) + if err != nil { + t.Fatalf("GetPosts returned error: %v", err) + } + if len(posts) != 1 { + t.Fatalf("expected 1 post, got %d", len(posts)) + } + if posts[0].ID != "post-1" { + t.Errorf("posts[0].ID = %q, want %q", posts[0].ID, "post-1") + } + if posts[0].Message != "Hello world" { + t.Errorf("posts[0].Message = %q, want %q", posts[0].Message, "Hello world") + } +} + +func TestGetPosts_EmptyResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"success":true,"data":[]}`)) + })) + defer srv.Close() + + client := NewAPIClient("user-token", "page-token", "page-123") + client.pageV2BaseURL = srv.URL + client.httpClient = srv.Client() + + posts, err := client.GetPosts(context.Background(), 10) + if err != nil { + t.Fatalf("GetPosts returned error: %v", err) + } + if len(posts) != 0 { + t.Fatalf("expected 0 posts, got %d", len(posts)) + } +} + +func TestGetPosts_ErrorResponse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(`internal error`)) + })) + defer srv.Close() + + client := NewAPIClient("user-token", "page-token", "page-123") + client.pageV2BaseURL = srv.URL + client.httpClient = srv.Client() + + _, err := client.GetPosts(context.Background(), 10) + if err == nil { + t.Fatal("expected error on HTTP 500") + } +} + +// --- Config Parsing --- + +func TestConfigParsing_CommentReplyOptions(t *testing.T) { + raw := `{ + "page_id": "123", + "features": {"comment_reply": true, "first_inbox": true}, + "comment_reply_options": {"filter": "keyword", "keywords": ["price", "buy"]}, + "first_inbox_message": "Thanks!", + "post_context_cache_ttl": "30m" + }` + + var cfg pancakeInstanceConfig + if err := json.Unmarshal([]byte(raw), &cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + if !cfg.Features.FirstInbox { + t.Error("Features.FirstInbox should be true") + } + if cfg.CommentReplyOptions.Filter != "keyword" { + t.Errorf("Filter = %q, want %q", cfg.CommentReplyOptions.Filter, "keyword") + } + if len(cfg.CommentReplyOptions.Keywords) != 2 || + cfg.CommentReplyOptions.Keywords[0] != "price" || + cfg.CommentReplyOptions.Keywords[1] != "buy" { + t.Errorf("Keywords = %v, want [price buy]", cfg.CommentReplyOptions.Keywords) + } + if cfg.FirstInboxMessage != "Thanks!" { + t.Errorf("FirstInboxMessage = %q, want %q", cfg.FirstInboxMessage, "Thanks!") + } + if cfg.PostContextCacheTTL != "30m" { + t.Errorf("PostContextCacheTTL = %q, want %q", cfg.PostContextCacheTTL, "30m") + } +} + +func TestConfigParsing_Defaults(t *testing.T) { + raw := `{"page_id":"123","features":{"inbox_reply":true}}` + + var cfg pancakeInstanceConfig + if err := json.Unmarshal([]byte(raw), &cfg); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + if cfg.Features.FirstInbox { + t.Error("Features.FirstInbox should default to false") + } + if cfg.CommentReplyOptions.Filter != "" { + t.Errorf("CommentReplyOptions.Filter should default to empty, got %q", cfg.CommentReplyOptions.Filter) + } + if cfg.FirstInboxMessage != "" { + t.Errorf("FirstInboxMessage should default to empty, got %q", cfg.FirstInboxMessage) + } +} diff --git a/internal/channels/pancake/comment_handler.go b/internal/channels/pancake/comment_handler.go new file mode 100644 index 00000000..fd851a88 --- /dev/null +++ b/internal/channels/pancake/comment_handler.go @@ -0,0 +1,164 @@ +package pancake + +import ( + "fmt" + "log/slog" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/channels" +) + +// handleCommentEvent processes a Pancake COMMENT webhook event. +// Mirrors the inbox handler pattern with additional comment-specific guards. +func (ch *Channel) handleCommentEvent(data MessagingData) { + // Feature gate. + if !ch.config.Features.CommentReply { + return + } + + // Self-reply prevention: skip messages from the page itself. + if data.Message.SenderID == ch.pageID { + slog.Debug("pancake: skipping own page comment", + "page_id", ch.pageID, "sender_id", data.Message.SenderID) + return + } + + // Skip assigned staff comments. + if isAssignedStaff(data.AssigneeIDs, data.Message.SenderID) { + slog.Debug("pancake: skipping assigned staff comment", + "page_id", ch.pageID, "sender_id", data.Message.SenderID) + return + } + + if data.Message.SenderID == "" { + slog.Warn("pancake: comment missing sender_id", "msg_id", data.Message.ID) + return + } + + if data.ConversationID == "" { + slog.Warn("pancake: comment missing conversation_id", "msg_id", data.Message.ID) + return + } + + // Dedup by message ID (skip when empty to avoid shared slot). + var dedupKey string + if data.Message.ID != "" { + dedupKey = fmt.Sprintf("comment:%s", data.Message.ID) + if ch.isDup(dedupKey) { + slog.Debug("pancake: duplicate comment skipped", "msg_id", data.Message.ID) + return + } + } + + // Comment filter. + if !ch.filterComment(data.Message.Content) { + slog.Debug("pancake: comment filtered out", + "page_id", ch.pageID, "msg_id", data.Message.ID) + return + } + + // Echo check before content enrichment. + if ch.isRecentOutboundEcho(data.ConversationID, data.Message.Content) { + slog.Debug("pancake: skipping comment outbound echo", + "page_id", ch.pageID, "msg_id", data.Message.ID) + return + } + + // Build content — optionally enriched with post context. + content := ch.buildCommentContent(data) + + metadata := map[string]string{ + "pancake_mode": "comment", + "conversation_type": data.Type, + "reply_to_comment_id": data.Message.ID, + "sender_id": data.Message.SenderID, + "platform": data.Platform, + "conversation_id": data.ConversationID, + "message_id": dedupKey, + "display_name": channels.SanitizeDisplayName(data.Message.SenderName), + "page_name": ch.pageName, + } + if data.PostID != "" { + metadata["post_id"] = data.PostID + } + + // ChatID = ConversationID: Pancake groups COMMENT conversations per sender per post. + ch.HandleMessage( + data.Message.SenderID, + data.ConversationID, + content, + nil, + metadata, + "direct", + ) + + slog.Debug("pancake: comment event published to bus", + "page_id", ch.pageID, + "conv_id", data.ConversationID, + "sender_id", data.Message.SenderID, + "platform", data.Platform, + ) +} + +// buildCommentContent assembles the comment content, optionally enriched with post context. +// Uses display name only (no senderID in content — senderID stays in metadata). +func (ch *Channel) buildCommentContent(data MessagingData) string { + commentText := stripHTML(data.Message.Content) + senderName := channels.SanitizeDisplayName(data.Message.SenderName) + senderPrefix := fmt.Sprintf("[From: %s]", senderName) + + if !ch.config.CommentReplyOptions.IncludePostContext || ch.postFetcher == nil { + if commentText != "" { + return senderPrefix + " " + commentText + } + return senderPrefix + } + + var sb strings.Builder + + // Fetch post context best-effort — on failure, fall back to comment text only. + if data.PostID != "" { + post, err := ch.postFetcher.GetPost(ch.stopCtx, data.PostID) + if err != nil { + slog.Debug("pancake: post context fetch failed, using comment only", + "page_id", ch.pageID, "post_id", data.PostID, "err", err) + } + if err == nil && post != nil && post.Message != "" { + sb.WriteString("[Bai dang] ") + sb.WriteString(post.Message) + sb.WriteString("\n\n") + } + } + + sb.WriteString("[Comment moi] ") + sb.WriteString(senderPrefix) + if commentText != "" { + sb.WriteString(" ") + sb.WriteString(commentText) + } + + return sb.String() +} + +// filterComment checks if the comment matches the configured filter. +// Returns true if the comment should be processed. +func (ch *Channel) filterComment(content string) bool { + switch ch.config.CommentReplyOptions.Filter { + case "keyword": + if len(ch.config.CommentReplyOptions.Keywords) == 0 { + // No keywords configured = block all (safe default). + slog.Warn("pancake: keyword filter active but no keywords configured, blocking all comments", + "page_id", ch.pageID) + return false + } + lower := strings.ToLower(content) + for _, kw := range ch.config.CommentReplyOptions.Keywords { + if strings.Contains(lower, strings.ToLower(kw)) { + return true + } + } + return false + default: // "all" or empty — process all comments + return true + } +} diff --git a/internal/channels/pancake/comment_handler_test.go b/internal/channels/pancake/comment_handler_test.go new file mode 100644 index 00000000..8c2c9c9b --- /dev/null +++ b/internal/channels/pancake/comment_handler_test.go @@ -0,0 +1,364 @@ +package pancake + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/bus" +) + +// newTestChannel builds a minimal Channel for handler tests. +// apiSrv may be nil when API calls are not expected. +func newTestChannel(t *testing.T, pageID string, cfg pancakeInstanceConfig) (*Channel, *bus.MessageBus) { + t.Helper() + msgBus := bus.New() + cfg.PageID = pageID + creds := pancakeCreds{ + APIKey: "test-key", + PageAccessToken: "test-token", + } + ch, err := New(cfg, creds, msgBus, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + ch.pageName = "Test Page" + ch.platform = "facebook" + ch.stopCtx = context.Background() + return ch, msgBus +} + +func newTestChannelWithSrv(t *testing.T, pageID string, cfg pancakeInstanceConfig, srv *httptest.Server) (*Channel, *bus.MessageBus) { + t.Helper() + ch, msgBus := newTestChannel(t, pageID, cfg) + if srv != nil { + ch.apiClient.pageV2BaseURL = srv.URL + ch.apiClient.httpClient = srv.Client() + ch.postFetcher = NewPostFetcher(ch.apiClient, "") + ch.postFetcher.stopCtx = context.Background() + } + return ch, msgBus +} + +func commentEvent(pageID, convID, senderID, msgID, content string) MessagingData { + return MessagingData{ + PageID: pageID, + ConversationID: convID, + Type: "COMMENT", + Platform: "facebook", + Message: MessagingMessage{ + ID: msgID, + SenderID: senderID, + SenderName: "Test User", + Content: content, + }, + } +} + +// consumeInbound drains one message from the bus with a short timeout. +func consumeInbound(t *testing.T, msgBus *bus.MessageBus, timeout time.Duration) (bus.InboundMessage, bool) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + return msgBus.ConsumeInbound(ctx) +} + +// --- Feature Gate --- + +func TestHandleCommentEvent_FeatureGated(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = false + ch, msgBus := newTestChannel(t, "page-1", cfg) + + ch.handleCommentEvent(commentEvent("page-1", "conv-1", "user-1", "msg-1", "hello")) + + _, ok := consumeInbound(t, msgBus, 50*time.Millisecond) + if ok { + t.Error("expected no message published when CommentReply is disabled") + } +} + +func TestHandleCommentEvent_FeatureEnabled(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + ch.handleCommentEvent(commentEvent("page-1", "conv-1", "user-1", "msg-1", "hello")) + + msg, ok := consumeInbound(t, msgBus, 100*time.Millisecond) + if !ok { + t.Fatal("expected message published when CommentReply is enabled") + } + if msg.Metadata["pancake_mode"] != "comment" { + t.Errorf("pancake_mode = %q, want %q", msg.Metadata["pancake_mode"], "comment") + } +} + +// --- Self-Reply Prevention --- + +func TestHandleCommentEvent_SkipsSelfReply(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + // senderID == pageID: must be skipped without panic + ch.handleCommentEvent(commentEvent("page-1", "conv-1", "page-1", "msg-self", "own reply")) + + _, ok := consumeInbound(t, msgBus, 50*time.Millisecond) + if ok { + t.Error("expected self-reply to be dropped") + } +} + +func TestHandleCommentEvent_SkipsAssignedStaff(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + data := commentEvent("page-1", "conv-1", "staff-1", "msg-staff", "staff comment") + data.AssigneeIDs = []string{"staff-1"} + ch.handleCommentEvent(data) + + _, ok := consumeInbound(t, msgBus, 50*time.Millisecond) + if ok { + t.Error("expected assigned staff message to be dropped") + } +} + +// --- Dedup --- + +func TestHandleCommentEvent_DedupDropsSecondCall(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + evt := commentEvent("page-1", "conv-1", "user-1", "msg-dup", "hello") + ch.handleCommentEvent(evt) + ch.handleCommentEvent(evt) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, ok1 := msgBus.ConsumeInbound(ctx) + if !ok1 { + t.Fatal("expected first message to be published") + } + _, ok2 := consumeInbound(t, msgBus, 30*time.Millisecond) + if ok2 { + t.Error("expected second duplicate to be dropped") + } +} + +func TestHandleCommentEvent_EmptyMsgIDNotDeduped(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + ch.handleCommentEvent(commentEvent("page-1", "conv-1", "user-1", "", "msg A")) + ch.handleCommentEvent(commentEvent("page-1", "conv-1", "user-1", "", "msg B")) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, ok1 := msgBus.ConsumeInbound(ctx) + if !ok1 { + t.Fatal("first empty-ID message should be published") + } + _, ok2 := msgBus.ConsumeInbound(ctx) + if !ok2 { + t.Fatal("second empty-ID message should NOT be deduped against first") + } +} + +// --- Comment Filter --- + +func TestFilterComment_AllMode(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.Filter = "all" + ch, _ := newTestChannel(t, "page-1", cfg) + + if !ch.filterComment("random text") { + t.Error("filter=all should pass all comments") + } +} + +func TestFilterComment_KeywordMatch(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.Filter = "keyword" + cfg.CommentReplyOptions.Keywords = []string{"price", "buy"} + ch, _ := newTestChannel(t, "page-1", cfg) + + if !ch.filterComment("what is the price?") { + t.Error("expected 'price' keyword to match") + } + if !ch.filterComment("I want to buy this") { + t.Error("expected 'buy' keyword to match") + } + if ch.filterComment("nice photo") { + t.Error("expected 'nice photo' to be filtered out") + } +} + +func TestFilterComment_KeywordCaseInsensitive(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.Filter = "keyword" + cfg.CommentReplyOptions.Keywords = []string{"Price"} + ch, _ := newTestChannel(t, "page-1", cfg) + + if !ch.filterComment("what is the PRICE?") { + t.Error("keyword match should be case-insensitive") + } +} + +func TestFilterComment_EmptyFilter(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.Filter = "" + ch, _ := newTestChannel(t, "page-1", cfg) + + if !ch.filterComment("anything") { + t.Error("empty filter should default to all (pass everything)") + } +} + +// --- Content Enrichment --- + +func TestBuildEnrichedContent_WithPostContext(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + cfg.CommentReplyOptions.IncludePostContext = true + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]any{ + "data": []PancakePost{{ID: "post-abc", Message: "Hello world post"}}, + }) + })) + defer srv.Close() + + ch, _ := newTestChannelWithSrv(t, "page-1", cfg, srv) + + data := commentEvent("page-1", "conv-1", "user-1", "msg-1", "question about this") + data.PostID = "post-abc" + content := ch.buildCommentContent(data) + + if !contains(content, "[Bai dang]") { + t.Errorf("expected [Bai dang] prefix in enriched content, got: %s", content) + } + if !contains(content, "Hello world post") { + t.Errorf("expected post message in enriched content, got: %s", content) + } + if !contains(content, "[Comment moi]") { + t.Errorf("expected [Comment moi] in enriched content, got: %s", content) + } + if !contains(content, "question about this") { + t.Errorf("expected comment text in enriched content, got: %s", content) + } +} + +func TestBuildEnrichedContent_WithoutPostContext(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.IncludePostContext = false + ch, _ := newTestChannel(t, "page-1", cfg) + + data := commentEvent("page-1", "conv-1", "user-1", "msg-1", "just a comment") + data.PostID = "post-abc" + content := ch.buildCommentContent(data) + + if contains(content, "[Bai dang]") { + t.Errorf("should not include post context when IncludePostContext=false, got: %s", content) + } + if !contains(content, "just a comment") { + t.Errorf("expected comment text in output, got: %s", content) + } +} + +func TestBuildEnrichedContent_PostFetchFails(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.CommentReplyOptions.IncludePostContext = true + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + ch, _ := newTestChannelWithSrv(t, "page-1", cfg, srv) + + data := commentEvent("page-1", "conv-1", "user-1", "msg-1", "comment text") + data.PostID = "post-fail" + content := ch.buildCommentContent(data) + + // Should fall back to comment text only — no panic, no error propagated. + if content == "" { + t.Error("expected non-empty content on post fetch failure") + } + if contains(content, "[Bai dang]") { + t.Errorf("should not include post prefix when fetch fails, got: %s", content) + } +} + +// --- Metadata --- + +func TestHandleCommentEvent_MetadataFields(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + ch.platform = "facebook" + + data := MessagingData{ + PageID: "page-1", + ConversationID: "conv-abc", + Type: "COMMENT", + Platform: "facebook", + Message: MessagingMessage{ + ID: "msg-xyz", + SenderID: "user-1", + SenderName: "John Doe", + Content: "test comment", + }, + } + ch.handleCommentEvent(data) + + msg, ok := consumeInbound(t, msgBus, 100*time.Millisecond) + if !ok { + t.Fatal("expected message published") + } + + checks := map[string]string{ + "pancake_mode": "comment", + "conversation_type": "COMMENT", + "reply_to_comment_id": "msg-xyz", + "sender_id": "user-1", + "platform": "facebook", + "page_name": "Test Page", + "conversation_id": "conv-abc", + } + for key, want := range checks { + if got := msg.Metadata[key]; got != want { + t.Errorf("metadata[%q] = %q, want %q", key, got, want) + } + } +} + +// --- Session Key --- + +func TestHandleCommentEvent_ChatIDIsConversationID(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + ch, msgBus := newTestChannel(t, "page-1", cfg) + + ch.handleCommentEvent(commentEvent("page-1", "conv-distinct", "user-1", "msg-1", "hello")) + + msg, ok := consumeInbound(t, msgBus, 100*time.Millisecond) + if !ok { + t.Fatal("expected message published") + } + if msg.ChatID != "conv-distinct" { + t.Errorf("ChatID = %q, want %q", msg.ChatID, "conv-distinct") + } +} + +// contains is a helper to check if a string contains a substring. +func contains(s, substr string) bool { + return strings.Contains(s, substr) +} diff --git a/internal/channels/pancake/echo_dedup.go b/internal/channels/pancake/echo_dedup.go index 2695cddd..fde97a67 100644 --- a/internal/channels/pancake/echo_dedup.go +++ b/internal/channels/pancake/echo_dedup.go @@ -119,7 +119,12 @@ func normalizeEchoContent(content string) string { return strings.TrimSpace(strings.Join(normalized, "\n")) } +// firstInboxSentTTL controls how long a senderID is retained in firstInboxSent. +// After this period, the sender can receive the first-inbox DM again (e.g. new session after a long gap). +const firstInboxSentTTL = 72 * time.Hour + // runDedupCleaner evicts dedup entries older than dedupTTL every dedupCleanEvery. +// Also evicts firstInboxSent entries to bound memory growth on high-traffic pages. func (ch *Channel) runDedupCleaner() { ticker := time.NewTicker(dedupCleanEvery) defer ticker.Stop() @@ -141,6 +146,12 @@ func (ch *Channel) runDedupCleaner() { } return true }) + ch.firstInboxSent.Range(func(k, v any) bool { + if t, ok := v.(time.Time); ok && now.Sub(t) > firstInboxSentTTL { + ch.firstInboxSent.Delete(k) + } + return true + }) } } } diff --git a/internal/channels/pancake/pancake.go b/internal/channels/pancake/pancake.go index 5ca9e796..0ab40bff 100644 --- a/internal/channels/pancake/pancake.go +++ b/internal/channels/pancake/pancake.go @@ -44,6 +44,13 @@ type Channel struct { // recentOutbound suppresses short-lived webhook echoes of our own text replies. recentOutbound sync.Map // conversationID + "\x00" + normalized content → time.Time + // firstInboxSent tracks which senders have already received the one-time first-inbox DM. + // In-memory only: resets on restart (acceptable — re-sending once is benign). + firstInboxSent sync.Map // senderID(string) → time.Time + + // postFetcher fetches and caches page post content for comment context enrichment. + postFetcher *PostFetcher + stopCh chan struct{} stopCtx context.Context stopFn context.CancelFunc @@ -66,17 +73,20 @@ func New(cfg pancakeInstanceConfig, creds pancakeCreds, base := channels.NewBaseChannel(channels.TypePancake, msgBus, cfg.AllowFrom) stopCtx, stopFn := context.WithCancel(context.Background()) + apiClient := NewAPIClient(creds.APIKey, creds.PageAccessToken, cfg.PageID) ch := &Channel{ BaseChannel: base, config: cfg, - apiClient: NewAPIClient(creds.APIKey, creds.PageAccessToken, cfg.PageID), + apiClient: apiClient, pageID: cfg.PageID, platform: cfg.Platform, webhookSecret: creds.WebhookSecret, + postFetcher: NewPostFetcher(apiClient, cfg.PostContextCacheTTL), stopCh: make(chan struct{}), stopCtx: stopCtx, stopFn: stopFn, } + ch.postFetcher.stopCtx = stopCtx return ch, nil } @@ -161,6 +171,7 @@ func (ch *Channel) Stop(_ context.Context) error { } // Send delivers an outbound message via Pancake API. +// Routes to sendCommentReply or sendInboxReply based on metadata["pancake_mode"]. func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { slog.Debug("pancake: Send called", "page_id", ch.pageID, @@ -169,11 +180,21 @@ func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { "platform", ch.platform, "channel_name", ch.Name()) - conversationID := msg.ChatID - if conversationID == "" { + if msg.ChatID == "" { return fmt.Errorf("pancake: chat_id (conversation_id) is required for outbound message") } + switch msg.Metadata["pancake_mode"] { + case "comment": + return ch.sendCommentReply(ctx, msg) + default: // "inbox" or unset — existing behavior + return ch.sendInboxReply(ctx, msg) + } +} + +// sendInboxReply handles outbound inbox messages (existing logic extracted from Send). +func (ch *Channel) sendInboxReply(ctx context.Context, msg bus.OutboundMessage) error { + conversationID := msg.ChatID text := FormatOutbound(msg.Content, ch.platform) // Handle media attachments. @@ -183,7 +204,6 @@ func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { "page_id", ch.pageID, "err", err) } - // Pancake's official contract makes `message` and `content_ids` mutually exclusive. // Deliver uploaded media first, then follow with text chunks if needed. if len(attachmentIDs) > 0 { if err := ch.apiClient.SendAttachmentMessage(ctx, conversationID, attachmentIDs); err != nil { @@ -205,8 +225,6 @@ func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { for _, part := range parts { if err := ch.apiClient.SendMessage(ctx, conversationID, part); err != nil { ch.handleAPIError(err) - // Remove pre-stored echo fingerprints for unsent parts so they - // don't suppress genuine customer messages that happen to match. ch.forgetOutboundEcho(conversationID, part) return err } @@ -214,6 +232,57 @@ func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return nil } +// sendCommentReply replies to a comment and optionally sends a one-time first-inbox DM. +func (ch *Channel) sendCommentReply(ctx context.Context, msg bus.OutboundMessage) error { + // Bound API calls: ReplyComment + PrivateReply can hang if Pancake is slow. + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + conversationID := msg.ChatID + text := FormatOutbound(msg.Content, ch.platform) + + // Remember echo before sending (same pattern as inbox). + parts := splitMessage(text, ch.maxMessageLength()) + for _, part := range parts { + ch.rememberOutboundEcho(conversationID, part) + } + + for _, part := range parts { + if err := ch.apiClient.ReplyComment(ctx, conversationID, part); err != nil { + ch.handleAPIError(err) + ch.forgetOutboundEcho(conversationID, part) + return err + } + } + + // First inbox: one-time DM after comment reply (best-effort). + if ch.config.Features.FirstInbox { + senderID := msg.Metadata["sender_id"] + if senderID != "" { + ch.sendFirstInbox(ctx, senderID, conversationID) + } + } + + return nil +} + +// sendFirstInbox sends a one-time DM to a commenter (best-effort, fire-and-forget). +// If the send fails, the firstInboxSent entry is deleted to allow retry on the next comment. +func (ch *Channel) sendFirstInbox(ctx context.Context, senderID, conversationID string) { + if _, loaded := ch.firstInboxSent.LoadOrStore(senderID, time.Now()); loaded { + return // already sent to this sender + } + message := ch.config.FirstInboxMessage + if message == "" { + message = "Thanks for your comment! We can assist you further via private message." + } + if err := ch.apiClient.PrivateReply(ctx, conversationID, message); err != nil { + slog.Warn("pancake: first inbox send failed", + "sender_id", senderID, "err", err) + ch.firstInboxSent.Delete(senderID) // allow retry on next comment + } +} + // BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default). func (ch *Channel) BlockReplyEnabled() *bool { return ch.config.BlockReply } diff --git a/internal/channels/pancake/pancake_test.go b/internal/channels/pancake/pancake_test.go index c1506c36..e7f49bfb 100644 --- a/internal/channels/pancake/pancake_test.go +++ b/internal/channels/pancake/pancake_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" "unicode/utf8" @@ -482,3 +483,521 @@ func TestIsRateLimitError_WrappedError(t *testing.T) { t.Error("isRateLimitError should detect wrapped 429 apiError via errors.As") } } + +// buildWebhookBody builds a Pancake webhook JSON body for test use. +func buildWebhookBody(pageID, convID, convType, senderID, msgID, content, postID string) string { + conv := fmt.Sprintf(`{"id":%q,"type":%q,"from":{"id":%q}}`, convID, convType, senderID) + if postID != "" { + conv = fmt.Sprintf(`{"id":%q,"type":%q,"post_id":%q,"from":{"id":%q}}`, convID, convType, postID, senderID) + } + return fmt.Sprintf(`{"page_id":%q,"data":{"conversation":%s,"message":{"id":%q,"message":%q}}}`, + pageID, conv, msgID, content) +} + +// newTestRouter creates an isolated webhookRouter with a registered channel. +func newTestRouter(t *testing.T, cfg pancakeInstanceConfig) (*webhookRouter, *Channel, *bus.MessageBus) { + t.Helper() + msgBus := bus.New() + cfg.PageID = "page-test" + creds := pancakeCreds{APIKey: "k", PageAccessToken: "t"} + ch, err := New(cfg, creds, msgBus, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + ch.platform = "facebook" + router := &webhookRouter{instances: map[string]*Channel{"page-test": ch}} + return router, ch, msgBus +} + +// --- Webhook Router --- + +func TestWebhookRouterRoutesCommentEvent(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + router, _, msgBus := newTestRouter(t, cfg) + + body := buildWebhookBody("page-test", "conv-1", "COMMENT", "user-1", "msg-1", "hello", "") + req := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body)) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected message published for COMMENT event") + } + if msg.Metadata["pancake_mode"] != "comment" { + t.Errorf("pancake_mode = %q, want %q", msg.Metadata["pancake_mode"], "comment") + } +} + +func TestWebhookRouterRoutesInboxEvent(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.InboxReply = true + router, _, msgBus := newTestRouter(t, cfg) + + body := buildWebhookBody("page-test", "conv-1", "INBOX", "user-1", "msg-2", "inbox msg", "") + req := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body)) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected message published for INBOX event") + } + // inbox handler sets pancake_mode = "inbox" + if msg.Metadata["pancake_mode"] != "inbox" { + t.Errorf("pancake_mode = %q, want %q", msg.Metadata["pancake_mode"], "inbox") + } +} + +func TestWebhookRouterSkipsUnknownType(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + router, _, msgBus := newTestRouter(t, cfg) + + body := buildWebhookBody("page-test", "conv-1", "UNKNOWN", "user-1", "msg-3", "ignored", "") + req := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body)) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, ok := msgBus.ConsumeInbound(ctx) + if ok { + t.Error("expected no message for unknown conversation type") + } +} + +func TestWebhookRouterCommentNormalizesPostID(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + router, _, msgBus := newTestRouter(t, cfg) + + body := buildWebhookBody("page-test", "conv-1", "COMMENT", "user-1", "msg-4", "hello", "post-123") + req := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body)) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + msg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected message published") + } + if msg.Metadata["post_id"] != "post-123" { + t.Errorf("metadata.post_id = %q, want %q", msg.Metadata["post_id"], "post-123") + } +} + +// --- Send Path --- + +// multiCaptureTransport records multiple requests (for first-inbox tests). +type multiCaptureTransport struct { + reqs []*http.Request + bodies [][]byte + mu sync.Mutex +} + +func (t *multiCaptureTransport) RoundTrip(req *http.Request) (*http.Response, error) { + t.mu.Lock() + defer t.mu.Unlock() + cloned := req.Clone(req.Context()) + var body []byte + if req.Body != nil { + body, _ = io.ReadAll(req.Body) + req.Body = io.NopCloser(bytes.NewReader(body)) + } + t.reqs = append(t.reqs, cloned) + t.bodies = append(t.bodies, body) + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"success":true}`)), + Request: req, + }, nil +} + +func newChannelWithMultiCapture(t *testing.T, cfg pancakeInstanceConfig) (*Channel, *multiCaptureTransport) { + t.Helper() + transport := &multiCaptureTransport{} + msgBus := bus.New() + cfg.PageID = "page-123" + creds := pancakeCreds{APIKey: "k", PageAccessToken: "t"} + ch, err := New(cfg, creds, msgBus, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + ch.apiClient.httpClient = &http.Client{Transport: transport} + ch.platform = "facebook" + return ch, transport +} + +func TestSend_CommentMode(t *testing.T) { + cfg := pancakeInstanceConfig{} + ch, transport := newChannelWithMultiCapture(t, cfg) + + err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "conv-123", + Content: "reply text", + Metadata: map[string]string{ + "pancake_mode": "comment", + "reply_to_comment_id": "msg-1", + "sender_id": "user-1", + }, + }) + if err != nil { + t.Fatalf("Send returned error: %v", err) + } + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) == 0 { + t.Fatal("expected at least one request") + } + var payload map[string]any + json.Unmarshal(transport.bodies[0], &payload) + if payload["action"] != "reply_comment" { + t.Errorf("action = %v, want reply_comment", payload["action"]) + } + if payload["message"] != "reply text" { + t.Errorf("message = %v, want 'reply text'", payload["message"]) + } +} + +func TestSend_CommentMode_WithFirstInbox(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.FirstInbox = true + cfg.FirstInboxMessage = "Thanks!" + ch, transport := newChannelWithMultiCapture(t, cfg) + + err := ch.Send(context.Background(), bus.OutboundMessage{ + ChatID: "conv-123", + Content: "reply text", + Metadata: map[string]string{ + "pancake_mode": "comment", + "sender_id": "user-1", + }, + }) + if err != nil { + t.Fatalf("Send returned error: %v", err) + } + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) != 2 { + t.Fatalf("expected 2 requests (reply_comment + private_reply), got %d", len(transport.reqs)) + } + var p1, p2 map[string]any + json.Unmarshal(transport.bodies[0], &p1) + json.Unmarshal(transport.bodies[1], &p2) + if p1["action"] != "reply_comment" { + t.Errorf("first request action = %v, want reply_comment", p1["action"]) + } + if p2["action"] != "private_reply" { + t.Errorf("second request action = %v, want private_reply", p2["action"]) + } + if p2["message"] != "Thanks!" { + t.Errorf("private_reply message = %v, want 'Thanks!'", p2["message"]) + } +} + +func TestSend_CommentMode_FirstInboxDedup(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.FirstInbox = true + cfg.FirstInboxMessage = "DM!" + ch, transport := newChannelWithMultiCapture(t, cfg) + + outMsg := bus.OutboundMessage{ + ChatID: "conv-123", + Content: "hi", + Metadata: map[string]string{ + "pancake_mode": "comment", + "sender_id": "user-1", + }, + } + ch.Send(context.Background(), outMsg) //nolint:errcheck + outMsg.ChatID = "conv-456" // second comment, different conv, same sender + ch.Send(context.Background(), outMsg) //nolint:errcheck + + transport.mu.Lock() + defer transport.mu.Unlock() + // Expected: reply_comment x2, private_reply x1 (deduped on sender) + if len(transport.reqs) != 3 { + t.Fatalf("expected 3 requests (2x reply_comment + 1x private_reply), got %d", len(transport.reqs)) + } + var actions []string + for _, body := range transport.bodies { + var p map[string]any + json.Unmarshal(body, &p) + if a, ok := p["action"].(string); ok { + actions = append(actions, a) + } + } + privateCount := 0 + for _, a := range actions { + if a == "private_reply" { + privateCount++ + } + } + if privateCount != 1 { + t.Errorf("expected exactly 1 private_reply, got %d (actions: %v)", privateCount, actions) + } +} + +func TestSend_CommentMode_FirstInboxDisabled(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.FirstInbox = false + ch, transport := newChannelWithMultiCapture(t, cfg) + + ch.Send(context.Background(), bus.OutboundMessage{ //nolint:errcheck + ChatID: "conv-123", + Content: "reply", + Metadata: map[string]string{ + "pancake_mode": "comment", + "sender_id": "user-1", + }, + }) + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) != 1 { + t.Fatalf("expected 1 request (reply_comment only), got %d", len(transport.reqs)) + } + var p map[string]any + json.Unmarshal(transport.bodies[0], &p) + if p["action"] == "private_reply" { + t.Error("should not send private_reply when FirstInbox is disabled") + } +} + +func TestSend_InboxMode_Unchanged(t *testing.T) { + cfg := pancakeInstanceConfig{} + ch, transport := newChannelWithMultiCapture(t, cfg) + + ch.Send(context.Background(), bus.OutboundMessage{ //nolint:errcheck + ChatID: "conv-123", + Content: "inbox reply", + Metadata: map[string]string{ + "pancake_mode": "inbox", + }, + }) + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) == 0 { + t.Fatal("expected a request for inbox mode") + } + var p map[string]any + json.Unmarshal(transport.bodies[0], &p) + if p["action"] != "reply_inbox" { + t.Errorf("action = %v, want reply_inbox", p["action"]) + } +} + +func TestSend_CommentMode_EchoRemembered(t *testing.T) { + cfg := pancakeInstanceConfig{} + ch, _ := newChannelWithMultiCapture(t, cfg) + + ch.Send(context.Background(), bus.OutboundMessage{ //nolint:errcheck + ChatID: "conv-echo", + Content: "some reply", + Metadata: map[string]string{ + "pancake_mode": "comment", + "sender_id": "user-1", + }, + }) + + if !ch.isRecentOutboundEcho("conv-echo", "some reply") { + t.Error("expected outbound echo to be remembered after Send") + } +} + +// --- First Inbox --- + +func TestSendFirstInbox_DefaultMessage(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.FirstInboxMessage = "" // empty = use default + ch, transport := newChannelWithMultiCapture(t, cfg) + + ch.sendFirstInbox(context.Background(), "user-1", "conv-123") + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) != 1 { + t.Fatalf("expected 1 request, got %d", len(transport.reqs)) + } + var p map[string]any + json.Unmarshal(transport.bodies[0], &p) + if p["action"] != "private_reply" { + t.Errorf("action = %v, want private_reply", p["action"]) + } + msg, _ := p["message"].(string) + if msg == "" { + t.Error("expected non-empty default first inbox message") + } +} + +func TestSendFirstInbox_CustomMessage(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.FirstInboxMessage = "Thanks for your comment!" + ch, transport := newChannelWithMultiCapture(t, cfg) + + ch.sendFirstInbox(context.Background(), "user-1", "conv-123") + + transport.mu.Lock() + defer transport.mu.Unlock() + if len(transport.reqs) != 1 { + t.Fatalf("expected 1 request, got %d", len(transport.reqs)) + } + var p map[string]any + json.Unmarshal(transport.bodies[0], &p) + if p["message"] != "Thanks for your comment!" { + t.Errorf("message = %v, want custom message", p["message"]) + } +} + +func TestSendFirstInbox_ErrorRetryAllowed(t *testing.T) { + errorTransport := &captureTransport{ + resp: &http.Response{ + StatusCode: http.StatusInternalServerError, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("error")), + }, + } + cfg := pancakeInstanceConfig{} + cfg.FirstInboxMessage = "DM" + msgBus := bus.New() + cfg.PageID = "page-123" + creds := pancakeCreds{APIKey: "k", PageAccessToken: "t"} + ch, _ := New(cfg, creds, msgBus, nil) + ch.apiClient.httpClient = &http.Client{Transport: errorTransport} + + // First call: API error → firstInboxSent entry should be deleted (allows retry). + ch.sendFirstInbox(context.Background(), "user-1", "conv-123") + _, alreadyStored := ch.firstInboxSent.Load("user-1") + if alreadyStored { + t.Error("firstInboxSent should be deleted on error (allow retry)") + } + + // Second call: should attempt again (retry allowed). + secondTransport := &captureTransport{} + ch.apiClient.httpClient = &http.Client{Transport: secondTransport} + ch.sendFirstInbox(context.Background(), "user-1", "conv-123") + if secondTransport.req == nil { + t.Error("expected retry request after error-deletion") + } +} + +// TestCommentFlowEndToEnd is the Phase 5 integration scenario wired inline. +func TestCommentFlowEndToEnd(t *testing.T) { + cfg := pancakeInstanceConfig{} + cfg.Features.CommentReply = true + cfg.Features.FirstInbox = true + cfg.FirstInboxMessage = "Welcome!" + transport := &multiCaptureTransport{} + msgBus := bus.New() + cfg.PageID = "page-e2e" + creds := pancakeCreds{APIKey: "k", PageAccessToken: "t"} + ch, err := New(cfg, creds, msgBus, nil) + if err != nil { + t.Fatalf("New: %v", err) + } + ch.apiClient.httpClient = &http.Client{Transport: transport} + ch.platform = "facebook" + + router := &webhookRouter{instances: map[string]*Channel{"page-e2e": ch}} + + // Step 1: POST comment webhook. + body := buildWebhookBody("page-e2e", "conv-e2e", "COMMENT", "user-e2e", "msg-e2e", "great product!", "") + req := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body)) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + + // Step 2: Consume inbound message from bus. + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + inMsg, ok := msgBus.ConsumeInbound(ctx) + if !ok { + t.Fatal("expected inbound message after comment webhook") + } + + // Step 3: Verify metadata. + if inMsg.Metadata["pancake_mode"] != "comment" { + t.Errorf("pancake_mode = %q, want comment", inMsg.Metadata["pancake_mode"]) + } + if inMsg.Metadata["sender_id"] != "user-e2e" { + t.Errorf("sender_id = %q, want user-e2e", inMsg.Metadata["sender_id"]) + } + + // Step 4: Send outbound reply. + outMsg := bus.OutboundMessage{ + ChatID: inMsg.ChatID, + Content: "thank you!", + Metadata: inMsg.Metadata, + } + if err := ch.Send(context.Background(), outMsg); err != nil { + t.Fatalf("Send error: %v", err) + } + + // Step 5: Verify reply_comment + private_reply. + transport.mu.Lock() + reqCount := len(transport.reqs) + var actions []string + for _, b := range transport.bodies { + var p map[string]any + json.Unmarshal(b, &p) + if a, ok := p["action"].(string); ok { + actions = append(actions, a) + } + } + transport.mu.Unlock() + + if reqCount != 2 { + t.Fatalf("expected 2 requests (reply_comment + private_reply), got %d (actions: %v)", reqCount, actions) + } + if actions[0] != "reply_comment" { + t.Errorf("first action = %q, want reply_comment", actions[0]) + } + if actions[1] != "private_reply" { + t.Errorf("second action = %q, want private_reply", actions[1]) + } + + // Step 6: Second comment from same sender — no second DM. + body2 := buildWebhookBody("page-e2e", "conv-e2e", "COMMENT", "user-e2e", "msg-e2e-2", "another comment", "") + req2 := httptest.NewRequest(http.MethodPost, webhookPath, strings.NewReader(body2)) + w2 := httptest.NewRecorder() + router.ServeHTTP(w2, req2) + + ctx2, cancel2 := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel2() + inMsg2, ok2 := msgBus.ConsumeInbound(ctx2) + if !ok2 { + t.Fatal("expected second inbound message") + } + outMsg2 := bus.OutboundMessage{ + ChatID: inMsg2.ChatID, + Content: "thanks again", + Metadata: inMsg2.Metadata, + } + ch.Send(context.Background(), outMsg2) //nolint:errcheck + + transport.mu.Lock() + finalCount := len(transport.reqs) + transport.mu.Unlock() + + // 2 (first round) + 1 (second reply_comment only, no second private_reply) + if finalCount != 3 { + t.Errorf("expected 3 total requests after dedup, got %d", finalCount) + } +} diff --git a/internal/channels/pancake/post_fetcher.go b/internal/channels/pancake/post_fetcher.go new file mode 100644 index 00000000..2052bfd6 --- /dev/null +++ b/internal/channels/pancake/post_fetcher.go @@ -0,0 +1,94 @@ +package pancake + +import ( + "context" + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +const defaultPostCacheTTL = 15 * time.Minute + +type postCacheEntry struct { + post *PancakePost // nil = negative cache (post not found) + expiresAt time.Time +} + +// PostFetcher caches page post content for comment context enrichment. +// singleflight prevents redundant concurrent Pancake API calls for the same page. +// Unlike Facebook (per-post Graph API), Pancake requires listing all posts then finding by ID, +// so the singleflight key is "posts" (one call fetches all, results cached individually). +type PostFetcher struct { + apiClient *APIClient + cacheTTL time.Duration + cache sync.Map // postID(string) -> *postCacheEntry + sfGroup singleflight.Group + stopCtx context.Context // channel-lifetime context — used inside singleflight to avoid per-request cancellation +} + +// NewPostFetcher creates a PostFetcher. cacheTTLStr is parsed as time.Duration; +// empty or invalid values fall back to defaultPostCacheTTL. +func NewPostFetcher(client *APIClient, cacheTTLStr string) *PostFetcher { + ttl := defaultPostCacheTTL + if cacheTTLStr != "" { + if d, err := time.ParseDuration(cacheTTLStr); err == nil && d > 0 { + ttl = d + } + } + return &PostFetcher{ + apiClient: client, + cacheTTL: ttl, + } +} + +// GetPost returns a cached post by ID, or fetches all recent posts and caches them. +// Returns (nil, nil) if postID is empty or post not found — graceful degradation. +// Uses stopCtx inside singleflight to avoid per-request cancellation killing shared fetches. +// Stores negative cache entries for missing posts to prevent repeated API stampedes. +func (pf *PostFetcher) GetPost(ctx context.Context, postID string) (*PancakePost, error) { + if postID == "" { + return nil, nil + } + + // Check cache. + if v, ok := pf.cache.Load(postID); ok { + entry := v.(*postCacheEntry) + if time.Now().Before(entry.expiresAt) { + return entry.post, nil // may be nil (negative cache) + } + pf.cache.Delete(postID) + } + + // Coalesce concurrent fetches via singleflight keyed on "posts". + // All posts are fetched in one API call and cached individually. + fetchCtx := pf.stopCtx + if fetchCtx == nil { + fetchCtx = ctx + } + _, err, _ := pf.sfGroup.Do("posts", func() (any, error) { + posts, err := pf.apiClient.GetPosts(fetchCtx, 50) + if err != nil { + return nil, err + } + expires := time.Now().Add(pf.cacheTTL) + for i := range posts { + pf.cache.Store(posts[i].ID, &postCacheEntry{ + post: &posts[i], + expiresAt: expires, + }) + } + return nil, nil + }) + if err != nil { + return nil, err + } + + // Re-check cache after fetch. + if v, ok := pf.cache.Load(postID); ok { + return v.(*postCacheEntry).post, nil + } + // Store negative cache entry — prevents re-fetching for old/missing posts. + pf.cache.Store(postID, &postCacheEntry{post: nil, expiresAt: time.Now().Add(pf.cacheTTL)}) + return nil, nil +} diff --git a/internal/channels/pancake/post_fetcher_test.go b/internal/channels/pancake/post_fetcher_test.go new file mode 100644 index 00000000..c611adcb --- /dev/null +++ b/internal/channels/pancake/post_fetcher_test.go @@ -0,0 +1,224 @@ +package pancake + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" +) + +func makePostFetcherClient(t *testing.T, srv *httptest.Server) *APIClient { + t.Helper() + client := NewAPIClient("user-token", "page-token", "page-123") + client.pageV2BaseURL = srv.URL + client.httpClient = srv.Client() + return client +} + +func TestNewPostFetcher_DefaultTTL(t *testing.T) { + client := NewAPIClient("u", "p", "page-1") + + pf1 := NewPostFetcher(client, "") + if pf1.cacheTTL != defaultPostCacheTTL { + t.Errorf("empty string: cacheTTL = %v, want %v", pf1.cacheTTL, defaultPostCacheTTL) + } + + pf2 := NewPostFetcher(client, "bogus") + if pf2.cacheTTL != defaultPostCacheTTL { + t.Errorf("bogus: cacheTTL = %v, want %v", pf2.cacheTTL, defaultPostCacheTTL) + } + + pf3 := NewPostFetcher(client, "30m") + if pf3.cacheTTL != 30*time.Minute { + t.Errorf("30m: cacheTTL = %v, want 30m", pf3.cacheTTL) + } +} + +func TestPostFetcher_GetPost_EmptyPageIDReturnsNil(t *testing.T) { + client := NewAPIClient("u", "p", "page-1") + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + post, err := pf.GetPost(context.Background(), "") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if post != nil { + t.Errorf("expected nil post for empty postID, got %+v", post) + } +} + +func TestPostFetcher_GetPost_CacheHit(t *testing.T) { + var callCount atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount.Add(1) + json.NewEncoder(w).Encode(map[string]any{"data": []PancakePost{}}) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + // Pre-populate cache. + pf.cache.Store("post-1", &postCacheEntry{ + post: &PancakePost{ID: "post-1", Message: "cached"}, + expiresAt: time.Now().Add(time.Hour), + }) + + post, err := pf.GetPost(context.Background(), "post-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if post == nil || post.Message != "cached" { + t.Errorf("expected cached post, got %+v", post) + } + if callCount.Load() != 0 { + t.Errorf("expected 0 API calls, got %d", callCount.Load()) + } +} + +func TestPostFetcher_GetPost_CacheMiss(t *testing.T) { + var callCount atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount.Add(1) + json.NewEncoder(w).Encode(map[string]any{ + "data": []PancakePost{{ID: "post-1", Message: "fresh"}}, + }) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + post, err := pf.GetPost(context.Background(), "post-1") + if err != nil { + t.Fatalf("unexpected error on first call: %v", err) + } + if post == nil || post.Message != "fresh" { + t.Errorf("expected fresh post, got %+v", post) + } + if callCount.Load() != 1 { + t.Errorf("expected 1 API call, got %d", callCount.Load()) + } + + // Second call should be a cache hit — no additional API calls. + _, err = pf.GetPost(context.Background(), "post-1") + if err != nil { + t.Fatalf("unexpected error on second call: %v", err) + } + if callCount.Load() != 1 { + t.Errorf("expected still 1 API call after cache hit, got %d", callCount.Load()) + } +} + +func TestPostFetcher_GetPost_CacheExpiry(t *testing.T) { + var callCount atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount.Add(1) + json.NewEncoder(w).Encode(map[string]any{ + "data": []PancakePost{{ID: "post-1", Message: "refetched"}}, + }) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + // Pre-populate with expired entry. + pf.cache.Store("post-1", &postCacheEntry{ + post: &PancakePost{ID: "post-1", Message: "stale"}, + expiresAt: time.Now().Add(-time.Minute), // already expired + }) + + post, err := pf.GetPost(context.Background(), "post-1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if post == nil || post.Message != "refetched" { + t.Errorf("expected refetched post, got %+v", post) + } + if callCount.Load() != 1 { + t.Errorf("expected 1 API call after cache expiry, got %d", callCount.Load()) + } +} + +func TestPostFetcher_GetPost_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + post, err := pf.GetPost(context.Background(), "post-1") + if err == nil { + t.Fatal("expected error from API 500, got nil") + } + if post != nil { + t.Errorf("expected nil post on error, got %+v", post) + } +} + +func TestPostFetcher_GetPost_PostNotFound(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"data": []PancakePost{}}) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + post, err := pf.GetPost(context.Background(), "missing-post") + if err != nil { + t.Fatalf("expected nil error for missing post (graceful degradation), got %v", err) + } + if post != nil { + t.Errorf("expected nil post for missing post, got %+v", post) + } +} + +func TestPostFetcher_GetPost_SingleflightCoalescing(t *testing.T) { + var callCount atomic.Int32 + ready := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-ready // block until all goroutines have started + callCount.Add(1) + json.NewEncoder(w).Encode(map[string]any{ + "data": []PancakePost{{ID: "post-1", Message: "coalesced"}}, + }) + })) + defer srv.Close() + + client := makePostFetcherClient(t, srv) + pf := NewPostFetcher(client, "") + pf.stopCtx = context.Background() + + const goroutines = 10 + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + go func() { + defer wg.Done() + pf.GetPost(context.Background(), "post-1") //nolint:errcheck + }() + } + + // Give goroutines time to pile up in singleflight, then unblock. + time.Sleep(20 * time.Millisecond) + close(ready) + wg.Wait() + + if callCount.Load() != 1 { + t.Errorf("singleflight: expected 1 API call, got %d", callCount.Load()) + } +} diff --git a/internal/channels/pancake/types.go b/internal/channels/pancake/types.go index a0a239df..9ef25699 100644 --- a/internal/channels/pancake/types.go +++ b/internal/channels/pancake/types.go @@ -19,9 +19,18 @@ type pancakeInstanceConfig struct { Features struct { InboxReply bool `json:"inbox_reply"` CommentReply bool `json:"comment_reply"` + FirstInbox bool `json:"first_inbox"` // send one-time DM to commenter after comment reply } `json:"features"` - AllowFrom []string `json:"allow_from,omitempty"` - BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) + CommentReplyOptions struct { + IncludePostContext bool `json:"include_post_context"` // prepend post text to comment content + MaxThreadDepth int `json:"max_thread_depth"` + Filter string `json:"filter"` // "all" | "keyword" (default: all) + Keywords []string `json:"keywords"` // required when filter = "keyword" + } `json:"comment_reply_options"` + FirstInboxMessage string `json:"first_inbox_message,omitempty"` // custom DM text; defaults to built-in message + PostContextCacheTTL string `json:"post_context_cache_ttl,omitempty"` // e.g. "30m"; defaults to 15m + AllowFrom []string `json:"allow_from,omitempty"` + BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) } // --- Webhook payload types --- @@ -45,8 +54,9 @@ type WebhookData struct { // WebhookConversation holds the conversation metadata from a Pancake webhook. type WebhookConversation struct { - ID string `json:"id"` // format: "pageID_senderID" - Type string `json:"type"` // "INBOX" or "COMMENT" + ID string `json:"id"` // format: "pageID_senderID" + Type string `json:"type"` // "INBOX" or "COMMENT" + PostID string `json:"post_id,omitempty"` // present for COMMENT events AssigneeIDs []string `json:"assignee_ids,omitempty"` // Pancake staff IDs assigned to this conversation From WebhookSender `json:"from"` Snippet string `json:"snippet,omitempty"` @@ -81,6 +91,7 @@ type MessageAttachment struct { type MessagingData struct { PageID string ConversationID string + PostID string // present for COMMENT events; empty for INBOX Type string // "INBOX" or "COMMENT" Platform string // "facebook", "zalo", "instagram", "tiktok", "whatsapp", "line" AssigneeIDs []string @@ -113,6 +124,13 @@ type SendMessageRequest struct { ContentIDs []string `json:"content_ids,omitempty"` } +// PancakePost represents a page post from Pancake GET /pages/{id}/posts. +type PancakePost struct { + ID string `json:"id"` + Message string `json:"message"` + CreatedAt string `json:"created_at,omitempty"` +} + // UploadResponse is returned by POST /pages/{id}/upload_contents. type UploadResponse struct { ID string `json:"id"` diff --git a/internal/channels/pancake/webhook_handler.go b/internal/channels/pancake/webhook_handler.go index 981f0cc5..deb4d533 100644 --- a/internal/channels/pancake/webhook_handler.go +++ b/internal/channels/pancake/webhook_handler.go @@ -150,16 +150,6 @@ func (r *webhookRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) { w.WriteHeader(http.StatusOK) return } - if convType != "INBOX" { - slog.Info("pancake: skipping non-inbox conversation event", - "page_id", pageID, - "conv_id", data.Conversation.ID, - "conv_type", convType, - "msg_id", data.Message.ID) - w.WriteHeader(http.StatusOK) - return - } - r.mu.RLock() target := r.instances[pageID] r.mu.RUnlock() @@ -217,6 +207,7 @@ func (r *webhookRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) { normalized := MessagingData{ PageID: pageID, ConversationID: data.Conversation.ID, + PostID: data.Conversation.PostID, // present for COMMENT events Type: convType, Platform: target.platform, AssigneeIDs: append([]string(nil), data.Conversation.AssigneeIDs...), @@ -229,7 +220,16 @@ func (r *webhookRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) { }, } - target.handleMessagingEvent(normalized) + // Route by conversation type. + switch convType { + case "INBOX": + target.handleMessagingEvent(normalized) + case "COMMENT": + target.handleCommentEvent(normalized) + default: + slog.Debug("pancake: skipping unknown conversation type", + "page_id", pageID, "conv_type", convType) + } w.WriteHeader(http.StatusOK) }