feat(pancake): comment auto-reply + first-inbox DM (#841)

* feat(pancake): add comment auto-reply + first-inbox inbox implementation

- Route COMMENT webhook events with feature gate and self-reply prevention
- Add PostFetcher with sync.Map cache + singleflight for comment context
- Implement ReplyComment() and PrivateReply() API client methods
- Split Send() into sendCommentReply() / sendInboxReply() flows
- Add sendFirstInbox() for one-time inbox DM with configurable greeting
- New config: Features.FirstInbox, CommentReplyOptions, FirstInboxMessage, PostContextCacheTTL
- 72 tests passing with race detection; both PG and SQLite builds clean

* fix(pancake): address PR 841 review issues

- Add slog.Debug when post context fetch fails in buildCommentContent
- Add comment to GetPosts explaining why it bypasses doRequest
- Add 72h TTL eviction for firstInboxSent in runDedupCleaner
- Add 30s timeout to sendCommentReply to bound API hang risk
- Replace custom containsStr helper with strings.Contains in tests
This commit is contained in:
Plateau Nguyen authored and GitHub committed 2026-04-12 17:12:57 +07:00
1 parent 6608e3dafe
commit 18c09be21b
11 files changed
+1773 -21

No files matched your search

+63
View File
@@ -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 {
@@ -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)
}
}
@@ -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
}
}
@@ -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)
}
+11
View File
@@ -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
})
}
}
}
+75 -6
View File
@@ -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 }
+519
View File
@@ -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)
}
}
+94
View File
@@ -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
}
@@ -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())
}
}
+22 -4
View File
@@ -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"`
+11 -11
View File
@@ -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)
}