feat(channels): add Facebook Messenger and Pancake channel integrations (#731)

Add two new channel implementations for Facebook Fanpage (comment + Messenger
auto-reply, first inbox DM) and Pancake/pages.fm (multi-platform inbox via
Facebook, Zalo, Instagram, TikTok, WhatsApp, LINE).

Key features:
- Facebook: comment auto-reply, Messenger auto-reply, first inbox DM,
  HMAC-SHA256 webhook verification, multi-page webhook routing
- Pancake: multi-platform inbox, outbound echo dedup with HTML normalization,
  race-condition-safe echo fingerprinting, platform-aware formatting
- Bootstrap skip: pre-fill USER.md from channel metadata (Pancake)
- SanitizeDisplayName across all channels (defense in depth)

Code audit fixes:
- Fix truncateForTikTok byte→rune slicing (UTF-8 corruption)
- Fix empty message.ID shared dedup slot (silent message loss)
- Fix DisplayName markdown injection in buildPrefilledUser
- Consolidate duplicate ChannelMeta type (agent→bootstrap)
- Compile-time interface assertions, alphabetical type constants
- Per-message logs demoted to slog.Debug, errors.As for wrapped errors
- UI: alphabetical channel ordering, complete config schemas
- Remove deprecated WhatsApp bridge_url, fix nested error parsing
This commit is contained in:
Plateau Nguyen authored and GitHub committed 2026-04-09 23:15:08 +07:00
1 parent 7205e6bb3b
commit fb8afd41bf
44 files changed
+4214 -39

No files matched your search

+4
View File
@@ -20,6 +20,8 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/eventbus"
kg "github.com/nextlevelbuilder/goclaw/internal/knowledgegraph"
"github.com/nextlevelbuilder/goclaw/internal/channels/discord"
"github.com/nextlevelbuilder/goclaw/internal/channels/facebook"
"github.com/nextlevelbuilder/goclaw/internal/channels/pancake"
"github.com/nextlevelbuilder/goclaw/internal/channels/feishu"
slackchannel "github.com/nextlevelbuilder/goclaw/internal/channels/slack"
"github.com/nextlevelbuilder/goclaw/internal/channels/telegram"
@@ -417,6 +419,8 @@ func runGateway() {
instanceLoader.RegisterFactory(channels.TypeZaloPersonal, zalopersonal.FactoryWithPendingStore(pgStores.PendingMessages))
instanceLoader.RegisterFactory(channels.TypeWhatsApp, whatsapp.FactoryWithDB(pgStores.DB, pgStores.PendingMessages, "pgx"))
instanceLoader.RegisterFactory(channels.TypeSlack, slackchannel.FactoryWithPendingStore(pgStores.PendingMessages))
instanceLoader.RegisterFactory(channels.TypeFacebook, facebook.Factory)
instanceLoader.RegisterFactory(channels.TypePancake, pancake.Factory)
if err := instanceLoader.LoadAll(context.Background()); err != nil {
slog.Error("failed to load channel instances from DB", "error", err)
}
+2 -2
View File
@@ -30,8 +30,8 @@ func buildEnsureUserProfile(as store.AgentStore) agent.EnsureUserProfileFunc {
// isNew=true seeds all files; isNew=false only seeds if user has zero files
// (avoids re-seeding BOOTSTRAP.md after auto-cleanup on server restart).
func buildSeedUserFiles(as store.AgentStore) agent.SeedUserFilesFunc {
return func(ctx context.Context, agentID uuid.UUID, userID, agentType string, isNew bool) error {
_, err := bootstrap.SeedUserFiles(ctx, as, agentID, userID, agentType, !isNew)
return func(ctx context.Context, agentID uuid.UUID, userID, agentType string, isNew bool, channelMeta *bootstrap.ChannelMeta) error {
_, err := bootstrap.SeedUserFiles(ctx, as, agentID, userID, agentType, !isNew, channelMeta)
return err
}
}
+12
View File
@@ -91,6 +91,18 @@ func extractSessionMetadata(msg bus.InboundMessage, peerKind string) map[string]
return meta
}
// buildPancakeSessionLabel returns "Pancake:{senderName}:{pageName}" with non-empty parts only.
func buildPancakeSessionLabel(senderName, pageName string) string {
label := "Pancake"
if senderName != "" {
label += ":" + senderName
}
if pageName != "" {
label += ":" + pageName
}
return label
}
// buildAnnounceOutMeta builds outbound metadata for announce messages so that
// Send() can route replies to the correct forum topic or DM thread.
func buildAnnounceOutMeta(localKey string) map[string]string {
+6
View File
@@ -108,6 +108,12 @@ func processNormalMessage(
}
}
// Set session label for Pancake channels: "Pancake:{SenderName}:{PageName}"
if msg.Metadata["pancake_mode"] != "" {
label := buildPancakeSessionLabel(msg.Metadata["display_name"], msg.Metadata["page_name"])
deps.SessStore.SetLabel(ctx, sessionKey, label)
}
// Auto-collect channel contacts for the contact selector.
// Skip internal senders (system:*, notification:*, teammate:*, ticker:*, session_send_tool).
if deps.ContactCollector != nil && msg.SenderID != "" && !bus.IsInternalSender(msg.SenderID) {
+1
View File
@@ -157,6 +157,7 @@ func wireExtras(
ContextFileLoader: contextFileLoader,
BootstrapCleanup: buildBootstrapCleanup(stores.Agents),
CacheInvalidate: buildCacheInvalidate(contextFileInterceptor),
DefaultTimezone: appCfg.Cron.DefaultTimezone,
InjectionAction: injectionAction,
MaxMessageChars: appCfg.Gateway.MaxMessageChars,
CompactionCfg: appCfg.Agents.Defaults.Compaction,
+2 -2
View File
@@ -88,9 +88,9 @@ Every channel must implement the base interface:
| Interface | Purpose | Implemented By |
|-----------|---------|----------------|
| `StreamingChannel` | Real-time streaming updates | Telegram, Slack |
| `WebhookChannel` | Webhook HTTP handler mounting | Feishu |
| `WebhookChannel` | Webhook HTTP handler mounting | Facebook, Feishu/Lark, Pancake |
| `ReactionChannel` | Status reactions on messages | Telegram, Slack, Feishu |
| `BlockReplyChannel` | Override gateway block_reply setting | Slack |
| `BlockReplyChannel` | Override gateway block_reply setting | Discord, Feishu/Lark, Pancake, Slack, Zalo OA, Zalo Personal |
`BaseChannel` provides a shared implementation that all channels embed: allowlist matching, `HandleMessage()`, `CheckPolicy()`, and user ID extraction.
+2
View File
@@ -233,9 +233,11 @@ All notable changes to GoClaw Gateway are documented here. Format follows [Keep
- **File viewer**: Improved workspace file view/download and storage depth control
- **Pairing DB errors**: Handle transient errors gracefully
- **Provider thinking**: Corrected DashScope per-model thinking logic
- **Pancake Page loop guard**: Narrowed webhook ingress to `messaging` + `INBOX` events and normalized HTML-formatted echoes before short-TTL outbound echo suppression, reducing Facebook Page self-reply loops in Pancake inbox conversations
### Documentation
- Updated `05-channels-messaging.md` — Refreshed `WebhookChannel` / `BlockReplyChannel` implementation tables for Facebook, Pancake, Discord, and Zalo-family channels
- Updated `18-http-api.md` — Added section 17 for Runtime & Packages Management endpoints
- Updated `09-security.md` — Added Docker entrypoint documentation, pkg-helper architecture, privilege separation
- Updated `17-changelog.md` — New entries for packages management, Docker security, and auth fix
+2 -1
View File
@@ -115,7 +115,8 @@ func (l *Loop) injectContext(ctx context.Context, req *RunRequest) (contextSetup
// Seeding must run before buildMessages→resolveContextFiles reads context files.
// Team sessions skip seeding: members process tasks from leader, not end-user onboarding.
isTeamSession := bootstrap.IsTeamSession(req.SessionKey)
setup := l.getOrCreateUserSetup(ctx, req.UserID, req.Channel, isTeamSession)
channelMeta := l.buildChannelMeta(req)
setup := l.getOrCreateUserSetup(ctx, req.UserID, req.Channel, isTeamSession, channelMeta)
// Workspace resolution (layered pipeline).
// Layer order: tenant → team → project (future) → user/chat
+5 -1
View File
@@ -45,7 +45,8 @@ type EnsureUserProfileFunc func(ctx context.Context, agentID uuid.UUID, userID,
// Called once per user per Loop instance, independent of workspace.
// isNew indicates whether the profile was just created (seed all) or already existed
// (only seed if user has zero files — avoids re-seeding after BOOTSTRAP.md cleanup).
type SeedUserFilesFunc func(ctx context.Context, agentID uuid.UUID, userID, agentType string, isNew bool) error
// channelMeta carries optional channel-provided contact info for bootstrap skip decisions.
type SeedUserFilesFunc func(ctx context.Context, agentID uuid.UUID, userID, agentType string, isNew bool, channelMeta *bootstrap.ChannelMeta) error
// EnsureUserFilesFunc is the legacy combined callback (profile + seed + workspace).
// Deprecated: use EnsureUserProfileFunc + SeedUserFilesFunc separately.
@@ -72,6 +73,7 @@ type Loop struct {
agentUUID uuid.UUID // set for context propagation
tenantID uuid.UUID // agent's owning tenant
agentType string // "open" or "predefined"
defaultTimezone string // system default timezone for bootstrap pre-fill
provider providers.Provider
model string
contextWindow int
@@ -299,6 +301,7 @@ type LoopConfig struct {
ContextFileLoader ContextFileLoaderFunc
BootstrapCleanup BootstrapCleanupFunc
CacheInvalidate CacheInvalidateFunc // invalidate context file cache after seeding
DefaultTimezone string // system default timezone for bootstrap pre-fill
// Tracing collector (nil = no tracing)
TraceCollector *tracing.Collector
@@ -433,6 +436,7 @@ func NewLoop(cfg LoopConfig) *Loop {
skillAllowList: cfg.SkillAllowList,
hasMemory: cfg.HasMemory,
contextFiles: cfg.ContextFiles,
defaultTimezone: cfg.DefaultTimezone,
ensureUserProfile: cfg.EnsureUserProfile,
seedUserFiles: cfg.SeedUserFiles,
ensureUserFiles: cfg.EnsureUserFiles,
+15 -2
View File
@@ -64,10 +64,23 @@ func (l *Loop) shouldShareKnowledgeGraph() bool {
return l.workspaceSharing != nil && l.workspaceSharing.ShareKnowledgeGraph
}
// buildChannelMeta extracts channel metadata from RunRequest for bootstrap decisions.
// Returns nil when channel type is unknown (preserves normal bootstrap flow).
func (l *Loop) buildChannelMeta(req *RunRequest) *bootstrap.ChannelMeta {
if req == nil || req.ChannelType == "" {
return nil
}
return &bootstrap.ChannelMeta{
ChannelType: req.ChannelType,
DisplayName: req.SenderName,
DefaultTimezone: l.defaultTimezone,
}
}
// getOrCreateUserSetup returns the cached userSetup for a user, creating it on first call.
// On first call: seeds context files (non-team) and resolves workspace from user profile.
// On subsequent calls: returns cached setup immediately (no DB calls).
func (l *Loop) getOrCreateUserSetup(ctx context.Context, userID, channel string, isTeamSession bool) *userSetup {
func (l *Loop) getOrCreateUserSetup(ctx context.Context, userID, channel string, isTeamSession bool, channelMeta *bootstrap.ChannelMeta) *userSetup {
if userID == "" {
return &userSetup{workspace: l.workspace}
}
@@ -92,7 +105,7 @@ func (l *Loop) getOrCreateUserSetup(ctx context.Context, userID, channel string,
}
// Step 2: Seed context files (must run before buildMessages reads them).
// Passes isNew so SeedUserFiles knows whether to skip existing files.
if err := l.seedUserFiles(ctx, l.agentUUID, userID, l.agentType, isNew); err != nil {
if err := l.seedUserFiles(ctx, l.agentUUID, userID, l.agentType, isNew, channelMeta); err != nil {
slog.Warn("failed to seed user context files", "error", err)
// Seeding failed (e.g. SQLITE_BUSY after retries). Inject
// embedded bootstrap templates in-memory so the first turn
+2
View File
@@ -45,6 +45,7 @@ type ResolverDeps struct {
ContextFileLoader ContextFileLoaderFunc
BootstrapCleanup BootstrapCleanupFunc
CacheInvalidate CacheInvalidateFunc
DefaultTimezone string // system default timezone for bootstrap pre-fill
// Security
InjectionAction string // "log", "warn", "block", "off"
@@ -435,6 +436,7 @@ func NewManagedResolver(deps ResolverDeps) ResolverFunc {
ContextFileLoader: deps.ContextFileLoader,
BootstrapCleanup: deps.BootstrapCleanup,
CacheInvalidate: deps.CacheInvalidate,
DefaultTimezone: deps.DefaultTimezone,
OnEvent: deps.OnEvent,
TraceCollector: deps.TraceCollector,
InjectionAction: deps.InjectionAction,
+78 -1
View File
@@ -2,6 +2,7 @@ package bootstrap
import (
"context"
"fmt"
"log/slog"
"path/filepath"
"strings"
@@ -9,9 +10,47 @@ import (
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// ChannelMeta carries channel-provided contact info for bootstrap decisions.
type ChannelMeta struct {
ChannelType string
DisplayName string
DefaultTimezone string
}
// shouldSkipBootstrap returns true when channel provides enough user info
// to pre-fill USER.md, making the interactive bootstrap unnecessary.
// Currently only Pancake channel qualifies (provides Facebook profile name).
func shouldSkipBootstrap(meta *ChannelMeta) bool {
return meta != nil &&
meta.ChannelType == "pancake" &&
meta.DisplayName != ""
}
// buildPrefilledUser generates USER.md content pre-filled with channel-provided contact info.
func buildPrefilledUser(meta *ChannelMeta) string {
tz := meta.DefaultTimezone
if tz == "" {
tz = "(unknown)"
}
name := channels.SanitizeDisplayName(meta.DisplayName)
return fmt.Sprintf(`# USER.md - About This User
- **Name:** %s
- **What to call them:** %s
- **Timezone:** %s
## Context
_(First contact via %s channel. Profile info auto-filled from channel data.)_
`, name, name, tz, meta.ChannelType)
}
// retryOnBusy retries fn up to 3 times on SQLITE_BUSY errors with 500ms delay.
func retryOnBusy(fn func() error) error {
var lastErr error
@@ -140,7 +179,7 @@ var userSeedFilesPredefined = []string{
// This ensures wizard-configured owner profiles are preserved on first chat.
//
// Returns the list of file names that were seeded.
func SeedUserFiles(ctx context.Context, agentStore store.AgentStore, agentID uuid.UUID, userID, agentType string, skipIfAnyExist bool) ([]string, error) {
func SeedUserFiles(ctx context.Context, agentStore store.AgentStore, agentID uuid.UUID, userID, agentType string, skipIfAnyExist bool, channelMeta *ChannelMeta) ([]string, error) {
files := userSeedFilesOpen
if agentType == store.AgentTypePredefined {
files = userSeedFilesPredefined
@@ -160,6 +199,44 @@ func SeedUserFiles(ctx context.Context, agentStore store.AgentStore, agentID uui
return nil, nil
}
// Channel-provided contact info: skip bootstrap, pre-fill USER.md directly.
// Currently only Pancake channel (Facebook Messenger) provides enough user info
// (display_name from Facebook profile) to skip the interactive onboarding flow.
if shouldSkipBootstrap(channelMeta) {
userContent := buildPrefilledUser(channelMeta)
if err := retryOnBusy(func() error {
return agentStore.SetUserContextFile(ctx, agentID, userID, UserFile, userContent)
}); err != nil {
return nil, err
}
seeded := []string{UserFile}
// For open agents: still seed SOUL.md, IDENTITY.md, AGENTS.md (default templates)
// so the agent has its personality/identity files — just skip BOOTSTRAP.md.
if agentType == store.AgentTypeOpen {
for _, name := range userSeedFilesOpen {
if name == BootstrapFile || name == UserFile {
continue // skip bootstrap (the whole point) and user (already filled)
}
content, err := templateFS.ReadFile(filepath.Join("templates", name))
if err != nil {
continue
}
if err := retryOnBusy(func() error {
return agentStore.SetUserContextFile(ctx, agentID, userID, name, string(content))
}); err != nil {
return seeded, err
}
seeded = append(seeded, name)
}
}
slog.Info("bootstrap skipped (channel contact info)",
"agent", agentID, "user", userID, "channel", channelMeta.ChannelType,
"display_name", channelMeta.DisplayName, "files", seeded)
return seeded, nil
}
hasFile := make(map[string]bool, len(existing))
for _, f := range existing {
if f.Content != "" {
+28 -10
View File
@@ -3,6 +3,7 @@ package bootstrap
import (
"context"
"maps"
"strings"
"testing"
"github.com/google/uuid"
@@ -116,6 +117,23 @@ func (s *seedStubStore) PropagateContextFile(_ context.Context, _ uuid.UUID, _ s
}
// ---- Tests ----
// TestBuildPrefilledUser_SanitizesMarkdownInjection verifies that DisplayName with
// newlines or markdown syntax does not inject into USER.md structure.
func TestBuildPrefilledUser_SanitizesMarkdownInjection(t *testing.T) {
meta := &ChannelMeta{
ChannelType: "pancake",
DisplayName: "Evil\n- **Admin:** true\n## Override",
DefaultTimezone: "Asia/Ho_Chi_Minh",
}
content := buildPrefilledUser(meta)
if strings.Contains(content, "## Override") {
t.Error("DisplayName markdown injection not sanitized")
}
if strings.Contains(content, "\n- **Admin:**") {
t.Error("DisplayName newline injection not sanitized")
}
}
// TestSeedUserFiles_PredefinedAgent_UsesAgentLevelUserMD is the primary regression test.
// When a predefined agent has wizard-written USER.md in agent_context_files, SeedUserFiles
// must seed that content into user_context_files — NOT the blank embedded template.
@@ -127,7 +145,7 @@ func TestSeedUserFiles_PredefinedAgent_UsesAgentLevelUserMD(t *testing.T) {
// Simulate wizard writing USER.md at agent level via agents.files.set
as.agentFiles[UserFile] = wizardContent
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-alice", store.AgentTypePredefined, false)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-alice", store.AgentTypePredefined, false, nil)
if err != nil {
t.Fatalf("SeedUserFiles returned error: %v", err)
}
@@ -160,7 +178,7 @@ func TestSeedUserFiles_PredefinedAgent_FallsBackToTemplateWhenNoAgentLevelUserMD
agentID := uuid.New()
// No agent-level USER.md — wizard did not write one
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-bob", store.AgentTypePredefined, false)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-bob", store.AgentTypePredefined, false, nil)
if err != nil {
t.Fatalf("SeedUserFiles returned error: %v", err)
}
@@ -198,7 +216,7 @@ func TestSeedUserFiles_PredefinedAgent_DoesNotOverwriteExistingPerUserContent(t
// Also set wizard content at agent level
as.agentFiles[UserFile] = "wizard content that should NOT override personal content"
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-charlie", store.AgentTypePredefined, false)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-charlie", store.AgentTypePredefined, false, nil)
if err != nil {
t.Fatalf("SeedUserFiles returned error: %v", err)
}
@@ -223,7 +241,7 @@ func TestSeedUserFiles_OpenAgent_UsesEmbeddedTemplate(t *testing.T) {
agentID := uuid.New()
// Open agents should never check agent_context_files for USER.md
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-dave", store.AgentTypeOpen, false)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-dave", store.AgentTypeOpen, false, nil)
if err != nil {
t.Fatalf("SeedUserFiles returned error: %v", err)
}
@@ -256,14 +274,14 @@ func TestSeedUserFiles_IdempotentOnSecondCall(t *testing.T) {
agentID := uuid.New()
// First call — seeds files
SeedUserFiles(context.Background(), as, agentID, "user-frank", store.AgentTypePredefined, false)
SeedUserFiles(context.Background(), as, agentID, "user-frank", store.AgentTypePredefined, false, nil)
// Simulate what the first call wrote (move seededUserFiles → userFiles)
maps.Copy(as.userFiles, as.seededUserFiles)
as.seededUserFiles = make(map[string]string)
// Second call — must seed nothing (all files already exist)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-frank", store.AgentTypePredefined, false)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-frank", store.AgentTypePredefined, false, nil)
if err != nil {
t.Fatalf("second SeedUserFiles returned error: %v", err)
}
@@ -284,7 +302,7 @@ func TestSeedUserFiles_SkipIfAnyExist(t *testing.T) {
agentID := uuid.New()
// Step 1: Seed initial files (new user, skipIfAnyExist=false)
SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, false)
SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, false, nil)
maps.Copy(as.userFiles, as.seededUserFiles)
as.seededUserFiles = make(map[string]string)
@@ -292,7 +310,7 @@ func TestSeedUserFiles_SkipIfAnyExist(t *testing.T) {
delete(as.userFiles, BootstrapFile)
// Step 3: skipIfAnyExist=true (existing profile) → must NOT re-seed
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, true)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, true, nil)
if err != nil {
t.Fatalf("SeedUserFiles(skipIfAnyExist=true) returned error: %v", err)
}
@@ -301,7 +319,7 @@ func TestSeedUserFiles_SkipIfAnyExist(t *testing.T) {
}
// Step 4: skipIfAnyExist=false (force) → SHOULD re-seed missing BOOTSTRAP.md
seeded, err = SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, false)
seeded, err = SeedUserFiles(context.Background(), as, agentID, "user-eve", store.AgentTypePredefined, false, nil)
if err != nil {
t.Fatalf("SeedUserFiles(skipIfAnyExist=false) returned error: %v", err)
}
@@ -323,7 +341,7 @@ func TestSeedUserFiles_SkipIfAnyExist_EmptyUser(t *testing.T) {
agentID := uuid.New()
// No existing user files → skipIfAnyExist=true should still seed
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-ghost", store.AgentTypePredefined, true)
seeded, err := SeedUserFiles(context.Background(), as, agentID, "user-ghost", store.AgentTypePredefined, true, nil)
if err != nil {
t.Fatalf("SeedUserFiles returned error: %v", err)
}
+4 -2
View File
@@ -70,10 +70,12 @@ const (
// Channel type constants used across channel packages and gateway wiring.
const (
TypeTelegram = "telegram"
TypeDiscord = "discord"
TypeSlack = "slack"
TypeFacebook = "facebook"
TypeFeishu = "feishu"
TypePancake = "pancake"
TypeSlack = "slack"
TypeTelegram = "telegram"
TypeWhatsApp = "whatsapp"
TypeZaloOA = "zalo_oa"
TypeZaloPersonal = "zalo_personal"
+1 -1
View File
@@ -264,7 +264,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
"message_id": m.ID,
"user_id": senderID,
"username": m.Author.Username,
"display_name": senderName,
"display_name": channels.SanitizeDisplayName(senderName),
"guild_id": m.GuildID,
"channel_id": channelID,
"is_dm": fmt.Sprintf("%t", isDM),
+3
View File
@@ -71,6 +71,9 @@ func (m *Manager) dispatchOutbound(ctx context.Context) {
if err := channel.Send(ctx, msg); err != nil {
slog.Error("error sending message to channel",
"channel", msg.Channel,
"chat_id", msg.ChatID,
"content_len", len(msg.Content),
"content_preview", Truncate(msg.Content, 160),
"error", err,
)
// Try to send a text-only error notification back to the chat.
@@ -0,0 +1,99 @@
package facebook
import (
"fmt"
"log/slog"
"strings"
)
// handleCommentEvent processes a feed webhook change where item == "comment".
func (ch *Channel) handleCommentEvent(entry WebhookEntry, change ChangeValue) {
// Feature gate.
if !ch.config.Features.CommentReply {
return
}
// Only process new comments (not edits or deletions).
if change.Verb != "add" {
return
}
// Page routing: ensure this event belongs to our page instance (before dedup write).
if entry.ID != ch.pageID {
return
}
// Self-reply prevention: skip comments posted by the page itself.
if change.From.ID == ch.pageID {
return
}
// Dedup: Facebook may deliver the same event more than once.
if ch.isDup("comment:" + change.CommentID) {
slog.Debug("facebook: duplicate comment event skipped", "comment_id", change.CommentID)
return
}
// Build message content — optionally enriched with post + thread context.
// Use stopCtx so inflight Graph API calls are cancelled when the channel stops.
content := change.Message
if ch.config.CommentReplyOptions.IncludePostContext && change.PostID != "" {
content = ch.buildEnrichedContent(change)
}
senderID := change.From.ID
// Session key groups all comments by the same user on the same post.
chatID := fmt.Sprintf("%s:%s", change.PostID, senderID)
metadata := map[string]string{
"comment_id": change.CommentID,
"post_id": change.PostID,
"parent_id": change.ParentID,
"sender_name": change.From.Name,
"sender_id": senderID,
"fb_mode": "comment",
"reply_to_comment_id": change.CommentID,
}
ch.HandleMessage(senderID, chatID, content, nil, metadata, "direct")
}
// buildEnrichedContent fetches post content and comment thread, assembles a rich
// context string for the agent. Uses stopCtx so it is cancelled on channel Stop().
func (ch *Channel) buildEnrichedContent(change ChangeValue) string {
ctx := ch.stopCtx
var sb strings.Builder
// Post content.
post, err := ch.postFetcher.GetPost(ctx, change.PostID)
if err == nil && post != nil && post.Message != "" {
sb.WriteString("[Bài đăng] ")
sb.WriteString(post.Message)
sb.WriteString("\n\n")
}
// Comment thread (when this is a nested reply).
if change.ParentID != "" {
depth := ch.config.CommentReplyOptions.MaxThreadDepth
if depth <= 0 {
depth = 10
}
thread, err := ch.postFetcher.GetCommentThread(ctx, change.ParentID, depth)
if err == nil && len(thread) > 0 {
sb.WriteString("[Thread]\n")
for _, c := range thread {
sb.WriteString(fmt.Sprintf("- %s: %s\n", c.From.Name, c.Message))
}
sb.WriteString("\n")
}
}
// Current comment.
sb.WriteString("[Comment mới] ")
sb.WriteString(change.From.Name)
sb.WriteString(": ")
sb.WriteString(change.Message)
return sb.String()
}
+260
View File
@@ -0,0 +1,260 @@
package facebook
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"net/http"
"sync"
"time"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// Compile-time interface assertions.
var (
_ channels.Channel = (*Channel)(nil)
_ channels.WebhookChannel = (*Channel)(nil)
)
const (
webhookPath = "/channels/facebook/webhook"
dedupTTL = 24 * time.Hour // matches Facebook's max retry window
dedupCleanEvery = 5 * time.Minute // how often to evict stale dedup entries
)
// Channel implements channels.Channel and channels.WebhookChannel for Facebook Fanpage.
// Supports comment auto-reply, Messenger auto-reply, and first inbox DM.
type Channel struct {
*channels.BaseChannel
config facebookInstanceConfig
graphClient *GraphClient
webhookH *WebhookHandler
pageID string
// dedup prevents processing duplicate webhook deliveries.
// Value is the time.Time the event was first seen; entries are evicted after dedupTTL.
dedup sync.Map // eventKey(string) → time.Time
// firstInboxSent tracks which senders have already received a first-inbox DM (in-memory).
firstInboxSent sync.Map // senderID(string) → struct{}
// postFetcher caches post content to enrich comment context.
postFetcher *PostFetcher
// stopCh + stopCtx serve complementary roles:
// stopCh for select{} in goroutines (runDedupCleaner), stopCtx for HTTP client cancellation.
stopCh chan struct{}
stopCtx context.Context
stopFn context.CancelFunc
}
// New creates a Facebook channel from parsed credentials and config.
func New(cfg facebookInstanceConfig, creds facebookCreds,
msgBus *bus.MessageBus, _ store.PairingStore) (*Channel, error) {
if creds.PageAccessToken == "" {
return nil, fmt.Errorf("facebook: page_access_token is required")
}
if cfg.PageID == "" {
return nil, fmt.Errorf("facebook: page_id is required")
}
if creds.AppSecret == "" {
return nil, fmt.Errorf("facebook: app_secret is required")
}
if creds.VerifyToken == "" {
return nil, fmt.Errorf("facebook: verify_token is required")
}
base := channels.NewBaseChannel(channels.TypeFacebook, msgBus, cfg.AllowFrom)
graphClient := NewGraphClient(creds.PageAccessToken, cfg.PageID)
postFetcher := NewPostFetcher(graphClient, cfg.PostContextCacheTTL)
stopCtx, stopFn := context.WithCancel(context.Background())
ch := &Channel{
BaseChannel: base,
config: cfg,
graphClient: graphClient,
pageID: cfg.PageID,
postFetcher: postFetcher,
stopCh: make(chan struct{}),
stopCtx: stopCtx,
stopFn: stopFn,
}
wh := NewWebhookHandler(creds.AppSecret, creds.VerifyToken)
wh.onComment = ch.handleCommentEvent
wh.onMessage = ch.handleMessagingEvent
ch.webhookH = wh
return ch, nil
}
// Factory creates a Facebook Channel from DB instance data.
// Implements channels.ChannelFactory.
func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
var c facebookCreds
if err := json.Unmarshal(creds, &c); err != nil {
return nil, fmt.Errorf("facebook: decode credentials: %w", err)
}
var ic facebookInstanceConfig
if len(cfg) > 0 {
if err := json.Unmarshal(cfg, &ic); err != nil {
return nil, fmt.Errorf("facebook: decode config: %w", err)
}
}
ch, err := New(ic, c, msgBus, pairingSvc)
if err != nil {
return nil, err
}
ch.SetName(name)
return ch, nil
}
// Start connects the channel: verifies the page token, subscribes webhooks, marks healthy.
func (ch *Channel) Start(ctx context.Context) error {
ch.MarkStarting("connecting to Facebook page")
if err := ch.graphClient.VerifyToken(ctx); err != nil {
ch.MarkFailed("token invalid", err.Error(), channels.ChannelFailureKindAuth, false)
return err
}
// Best-effort: subscribe app to webhooks.
if err := ch.graphClient.SubscribeApp(ctx); err != nil {
slog.Warn("facebook: webhook subscription failed (check app install on page)", "err", err)
}
globalRouter.register(ch)
ch.MarkHealthy("connected to page " + ch.pageID)
ch.SetRunning(true)
// Background goroutine: evict stale dedup entries to prevent memory growth.
go ch.runDedupCleaner()
slog.Info("facebook channel started", "page_id", ch.pageID, "name", ch.Name())
return nil
}
// Stop gracefully shuts down the channel.
func (ch *Channel) Stop(_ context.Context) error {
globalRouter.unregister(ch.pageID)
ch.stopFn() // cancel stopCtx → cancels inflight Graph API calls
close(ch.stopCh) // stop background goroutines
ch.SetRunning(false)
ch.MarkStopped("stopped")
slog.Info("facebook channel stopped", "page_id", ch.pageID, "name", ch.Name())
return nil
}
// Send delivers an outbound message. Dispatches to comment reply or Messenger based on fb_mode metadata.
func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error {
mode := msg.Metadata["fb_mode"]
switch mode {
case "messenger":
text := FormatForMessenger(msg.Content)
parts := splitMessage(text, messengerMaxChars)
for _, part := range parts {
if _, err := ch.graphClient.SendMessage(ctx, msg.ChatID, part); err != nil {
ch.handleAPIError(err)
return err
}
}
default: // "comment"
commentID := msg.Metadata["reply_to_comment_id"]
if commentID == "" {
return fmt.Errorf("facebook: reply_to_comment_id missing in outbound metadata")
}
text := FormatForComment(msg.Content)
if _, err := ch.graphClient.ReplyToComment(ctx, commentID, text); err != nil {
ch.handleAPIError(err)
return err
}
// First inbox: send a private DM after comment reply (best-effort).
if ch.config.Features.FirstInbox {
senderID := msg.Metadata["sender_id"]
if senderID != "" {
ch.sendFirstInbox(ctx, senderID)
}
}
}
return nil
}
// WebhookHandler returns the shared webhook path and the global router as handler.
// Only the first facebook instance mounts the route; others return ("", nil).
func (ch *Channel) WebhookHandler() (string, http.Handler) {
return globalRouter.webhookRoute()
}
// handleAPIError maps Graph API errors to channel health states.
func (ch *Channel) handleAPIError(err error) {
if err == nil {
return
}
switch {
case IsAuthError(err):
ch.MarkFailed("token expired", err.Error(), channels.ChannelFailureKindAuth, false)
case IsPermissionError(err):
ch.MarkFailed("permission denied", err.Error(), channels.ChannelFailureKindAuth, false)
case IsRateLimitError(err):
ch.MarkDegraded("rate limited", err.Error(), channels.ChannelFailureKindNetwork, true)
default:
ch.MarkDegraded("api error", err.Error(), channels.ChannelFailureKindUnknown, true)
}
}
// sendFirstInbox sends a one-time private DM after a comment reply (best-effort).
func (ch *Channel) sendFirstInbox(ctx context.Context, senderID string) {
if _, alreadySent := ch.firstInboxSent.LoadOrStore(senderID, struct{}{}); alreadySent {
return
}
message := ch.config.FirstInboxMessage
if message == "" {
message = "Cảm ơn bạn đã comment! Mình có thể hỗ trợ thêm qua tin nhắn riêng."
}
if _, err := ch.graphClient.SendMessage(ctx, senderID, message); err != nil {
slog.Warn("facebook: first inbox send failed", "sender_id", senderID, "err", err)
ch.firstInboxSent.Delete(senderID) // allow retry on next comment
}
}
// runDedupCleaner evicts dedup entries older than dedupTTL every dedupCleanEvery.
func (ch *Channel) runDedupCleaner() {
ticker := time.NewTicker(dedupCleanEvery)
defer ticker.Stop()
for {
select {
case <-ch.stopCh:
return
case <-ticker.C:
now := time.Now()
ch.dedup.Range(func(k, v any) bool {
if t, ok := v.(time.Time); ok && now.Sub(t) > dedupTTL {
ch.dedup.Delete(k)
}
return true
})
}
}
}
// isDup checks and records a dedup key. Returns true if the key was already seen.
func (ch *Channel) isDup(key string) bool {
_, loaded := ch.dedup.LoadOrStore(key, time.Now())
return loaded
}
+395
View File
@@ -0,0 +1,395 @@
package facebook
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
// --- verifySignature ---
func TestVerifySignature(t *testing.T) {
secret := "test_secret"
body := []byte(`{"object":"page","entry":[]}`)
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(body)
validSig := "sha256=" + hex.EncodeToString(mac.Sum(nil))
tests := []struct {
name string
sig string
wantValid bool
}{
{"valid signature", validSig, true},
{"wrong prefix", "sha1=" + hex.EncodeToString(mac.Sum(nil)), false},
{"empty signature", "", false},
{"bad hex", "sha256=notvalidhex", false},
{"tampered body", "sha256=0000000000000000000000000000000000000000000000000000000000000000", false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := verifySignature(body, tt.sig, secret)
if got != tt.wantValid {
t.Errorf("verifySignature() = %v, want %v", got, tt.wantValid)
}
})
}
}
// --- WebhookHandler GET verification ---
func TestWebhookHandlerVerification(t *testing.T) {
wh := NewWebhookHandler("secret", "my_verify_token")
t.Run("valid challenge", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet,
"/webhook?hub.mode=subscribe&hub.verify_token=my_verify_token&hub.challenge=abc123", nil)
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", w.Code)
}
if w.Body.String() != "abc123" {
t.Errorf("body = %q, want %q", w.Body.String(), "abc123")
}
})
t.Run("wrong verify token", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet,
"/webhook?hub.mode=subscribe&hub.verify_token=wrong&hub.challenge=abc123", nil)
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusForbidden {
t.Errorf("status = %d, want 403", w.Code)
}
})
t.Run("wrong hub.mode", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet,
"/webhook?hub.mode=unsubscribe&hub.verify_token=my_verify_token&hub.challenge=abc123", nil)
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusForbidden {
t.Errorf("status = %d, want 403", w.Code)
}
})
t.Run("unsafe challenge rejected", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet,
"/webhook?hub.mode=subscribe&hub.verify_token=my_verify_token&hub.challenge=<script>alert(1)</script>", nil)
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Errorf("status = %d, want 400", w.Code)
}
})
}
// --- WebhookHandler POST event delivery ---
func signBody(t *testing.T, body []byte, secret string) string {
t.Helper()
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(body)
return "sha256=" + hex.EncodeToString(mac.Sum(nil))
}
func TestWebhookHandlerPostEvents(t *testing.T) {
appSecret := "app_secret_123"
t.Run("invalid signature returns 200 (no retry)", func(t *testing.T) {
wh := NewWebhookHandler(appSecret, "token")
body := []byte(`{"object":"page","entry":[]}`)
req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(string(body)))
req.Header.Set("X-Hub-Signature-256", "sha256=badbadbadbad")
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("status = %d, want 200 (Facebook must stop retrying)", w.Code)
}
})
t.Run("valid comment event dispatches callback", func(t *testing.T) {
var gotChange ChangeValue
wh := NewWebhookHandler(appSecret, "token")
wh.onComment = func(entry WebhookEntry, change ChangeValue) {
gotChange = change
}
payload := WebhookPayload{
Object: "page",
Entry: []WebhookEntry{{
ID: "111",
Time: 1234567890,
Changes: []WebhookChange{{
Field: "feed",
Value: ChangeValue{
From: FBUser{ID: "456", Name: "Alice"},
Item: "comment",
CommentID: "789",
PostID: "post1",
Message: "Hello page!",
Verb: "add",
},
}},
}},
}
body, _ := json.Marshal(payload)
req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(string(body)))
req.Header.Set("X-Hub-Signature-256", signBody(t, body, appSecret))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", w.Code)
}
if gotChange.CommentID != "789" {
t.Errorf("comment callback not called or wrong comment_id: %+v", gotChange)
}
})
t.Run("valid messenger event dispatches callback", func(t *testing.T) {
var gotEvent MessagingEvent
wh := NewWebhookHandler(appSecret, "token")
wh.onMessage = func(_ WebhookEntry, event MessagingEvent) {
gotEvent = event
}
mid := "m_abc123"
payload := WebhookPayload{
Object: "page",
Entry: []WebhookEntry{{
ID: "111",
Time: 1234567890,
Messaging: []MessagingEvent{{
Sender: FBUser{ID: "user1"},
Recipient: FBUser{ID: "111"},
Timestamp: 1234567890,
Message: &IncomingMessage{MID: mid, Text: "hi"},
}},
}},
}
body, _ := json.Marshal(payload)
req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(string(body)))
req.Header.Set("X-Hub-Signature-256", signBody(t, body, appSecret))
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", w.Code)
}
if gotEvent.Message == nil || gotEvent.Message.MID != mid {
t.Errorf("messenger callback not called: %+v", gotEvent)
}
})
t.Run("non-page object ignored", func(t *testing.T) {
called := false
wh := NewWebhookHandler(appSecret, "token")
wh.onComment = func(_ WebhookEntry, _ ChangeValue) { called = true }
payload := map[string]any{"object": "user", "entry": []any{}}
body, _ := json.Marshal(payload)
req := httptest.NewRequest(http.MethodPost, "/webhook", strings.NewReader(string(body)))
req.Header.Set("X-Hub-Signature-256", signBody(t, body, appSecret))
w := httptest.NewRecorder()
wh.ServeHTTP(w, req)
if called {
t.Error("callback should not be called for non-page object")
}
})
}
// --- Formatter ---
func TestFormatForComment(t *testing.T) {
tests := []struct {
input string
want string
}{
{"**bold** text", "bold text"},
{"*italic* text", "italic text"},
{"**bold** and *italic*", "bold and italic"},
{"[link](https://example.com)", "link (https://example.com)"},
{"# Heading\nbody", "Heading\nbody"},
{"<b>html</b>", "html"},
{"`code`", "code"},
// Long text truncated at 8000 chars.
{strings.Repeat("a", 9000), strings.Repeat("a", 8000)},
}
for _, tt := range tests {
got := FormatForComment(tt.input)
if got != tt.want {
t.Errorf("FormatForComment(%q) = %q, want %q", tt.input, got, tt.want)
}
}
}
func TestFormatForMessenger(t *testing.T) {
tests := []struct {
input string
want string
}{
{"**bold**", "bold"},
{"*italic*", "italic"},
{"[link](https://x.com)", "link (https://x.com)"},
{" leading space ", "leading space"},
}
for _, tt := range tests {
got := FormatForMessenger(tt.input)
if got != tt.want {
t.Errorf("FormatForMessenger(%q) = %q, want %q", tt.input, got, tt.want)
}
}
}
func TestSplitMessage(t *testing.T) {
t.Run("short message not split", func(t *testing.T) {
parts := splitMessage("hello", 2000)
if len(parts) != 1 || parts[0] != "hello" {
t.Errorf("unexpected parts: %v", parts)
}
})
t.Run("splits at sentence boundary", func(t *testing.T) {
long := strings.Repeat("word ", 200) + ". " + strings.Repeat("word ", 200)
parts := splitMessage(long, 500)
if len(parts) < 2 {
t.Errorf("expected multiple parts, got %d", len(parts))
}
for _, p := range parts {
if len([]rune(p)) > 500 {
t.Errorf("part exceeds maxChars: len=%d", len([]rune(p)))
}
}
})
t.Run("hard cut when no boundary found", func(t *testing.T) {
noSpace := strings.Repeat("a", 3000)
parts := splitMessage(noSpace, 2000)
if len(parts) != 2 {
t.Errorf("expected 2 parts, got %d", len(parts))
}
})
}
// --- validateFBID ---
func TestValidateFBID(t *testing.T) {
valid := []string{"123456789", "123_456", "1"}
invalid := []string{"", "../me", "123abc", "123 456", "../../accounts"}
for _, id := range valid {
if err := validateFBID(id); err != nil {
t.Errorf("validateFBID(%q) unexpectedly returned error: %v", id, err)
}
}
for _, id := range invalid {
if err := validateFBID(id); err == nil {
t.Errorf("validateFBID(%q) should have returned error", id)
}
}
}
// --- parseRetryAfter ---
func TestParseRetryAfter(t *testing.T) {
tests := []struct {
header string
want time.Duration
}{
{"", 5 * time.Second},
{"0", 5 * time.Second},
{"notanumber", 5 * time.Second},
{"10", 10 * time.Second},
{"60", 60 * time.Second},
// Capped at maxRetryAfterSec (60).
{"3600", 60 * time.Second},
{"999999", 60 * time.Second},
}
for _, tt := range tests {
resp := &http.Response{Header: make(http.Header)}
if tt.header != "" {
resp.Header.Set("Retry-After", tt.header)
}
got := parseRetryAfter(resp)
if got != tt.want {
t.Errorf("parseRetryAfter(%q) = %v, want %v", tt.header, got, tt.want)
}
}
}
// --- Dedup eviction ---
func TestDedupEviction(t *testing.T) {
// Override TTL to a tiny value so we can test eviction synchronously.
const shortTTL = 10 * time.Millisecond
ch := &Channel{}
key := "comment:test_event"
// First call: not a dup.
ch.dedup.Store(key, time.Now().Add(-shortTTL-time.Millisecond))
// Simulate cleaner: evict entries older than shortTTL.
now := time.Now()
ch.dedup.Range(func(k, v any) bool {
if t2, ok := v.(time.Time); ok && now.Sub(t2) > shortTTL {
ch.dedup.Delete(k)
}
return true
})
// After eviction, isDup should return false (entry gone).
if ch.isDup(key) {
// isDup stores the key now, so second call would be dup — just verify the first returns false.
// Since isDup uses LoadOrStore, a successful store means it wasn't there.
// The test above stored, deleted, so isDup should do a fresh store → return false.
}
// Verify second call IS a dup.
if !ch.isDup(key) {
t.Error("second isDup call should return true (key already stored)")
}
}
// --- graphAPIError classification ---
func TestErrorClassifiers(t *testing.T) {
authErr := &graphAPIError{code: 190, msg: "invalid token"}
permErr := &graphAPIError{code: 10, msg: "permission denied"}
rateErr := &graphAPIError{code: 4, msg: "rate limit"}
otherErr := &graphAPIError{code: 999, msg: "unknown"}
wrappedAuth := fmt.Errorf("send failed: %w", authErr)
if !IsAuthError(authErr) {
t.Error("IsAuthError should be true for code 190")
}
if !IsAuthError(wrappedAuth) {
t.Error("IsAuthError should work with wrapped errors")
}
if !IsPermissionError(permErr) {
t.Error("IsPermissionError should be true for code 10")
}
if !IsRateLimitError(rateErr) {
t.Error("IsRateLimitError should be true for code 4")
}
if IsAuthError(otherErr) || IsPermissionError(otherErr) || IsRateLimitError(otherErr) {
t.Error("unknown error should not match any classifier")
}
if IsAuthError(nil) {
t.Error("nil should not be auth error")
}
}
+104
View File
@@ -0,0 +1,104 @@
package facebook
import (
"regexp"
"strings"
)
const (
commentMaxChars = 8000
messengerMaxChars = 2000
)
var (
// Two-pass bold: **text** first, then *text*, to avoid stray asterisk artifacts.
reBoldDouble = regexp.MustCompile(`\*\*([^*]+)\*\*`)
reBoldSingle = regexp.MustCompile(`\*([^*]+)\*`)
// markdownLink converts [text](url) → "text (url)".
reLink = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`)
// HTML tags.
reHTML = regexp.MustCompile(`<[^>]+>`)
// Heading markers (# … at line start).
reHeader = regexp.MustCompile(`(?m)^#{1,6}\s+`)
// Inline `code` backticks.
reCode = regexp.MustCompile("`([^`]+)`")
)
// FormatForComment sanitizes agent output for posting as a Facebook comment.
// Facebook comments do not support markdown or HTML.
func FormatForComment(text string) string {
text = reHeader.ReplaceAllString(text, "")
text = reBoldDouble.ReplaceAllString(text, "$1") // ** first
text = reBoldSingle.ReplaceAllString(text, "$1") // * second
text = reLink.ReplaceAllString(text, "$1 ($2)")
text = reCode.ReplaceAllString(text, "$1")
text = reHTML.ReplaceAllString(text, "")
text = strings.TrimSpace(text)
return truncateRunes(text, commentMaxChars)
}
// FormatForMessenger sanitizes agent output for Messenger messages.
func FormatForMessenger(text string) string {
text = reHeader.ReplaceAllString(text, "")
text = reBoldDouble.ReplaceAllString(text, "$1")
text = reBoldSingle.ReplaceAllString(text, "$1")
text = reLink.ReplaceAllString(text, "$1 ($2)")
text = reCode.ReplaceAllString(text, "$1")
text = reHTML.ReplaceAllString(text, "")
return strings.TrimSpace(text)
}
// splitMessage splits text into chunks of at most maxChars runes,
// preferring paragraph or sentence boundaries to avoid mid-word cuts.
// Uses plain-text splitting (not channels.ChunkMarkdown) because Facebook
// strips all markdown — fenced code block repair is unnecessary.
func splitMessage(text string, maxChars int) []string {
runes := []rune(text) // convert once
if len(runes) <= maxChars {
return []string{text}
}
var parts []string
for len(runes) > maxChars {
chunk := string(runes[:maxChars])
// Try paragraph boundary.
if idx := strings.LastIndex(chunk, "\n\n"); idx > maxChars/2 {
parts = append(parts, strings.TrimSpace(chunk[:idx]))
runes = []rune(strings.TrimSpace(string(runes[idx+2:])))
continue
}
// Try sentence boundary.
split := false
for _, sep := range []string{". ", "! ", "? ", "\n"} {
if idx := strings.LastIndex(chunk, sep); idx > maxChars/2 {
parts = append(parts, strings.TrimSpace(chunk[:idx+1]))
runes = []rune(strings.TrimSpace(string(runes[idx+len(sep):])))
split = true
break
}
}
if split {
continue
}
// Hard cut at maxChars.
parts = append(parts, strings.TrimSpace(chunk))
runes = []rune(strings.TrimSpace(string(runes[maxChars:])))
}
if remaining := strings.TrimSpace(string(runes)); remaining != "" {
parts = append(parts, remaining)
}
return parts
}
// truncateRunes truncates s to at most n Unicode code points.
func truncateRunes(s string, n int) string {
runes := []rune(s)
if len(runes) <= n {
return s
}
return string(runes[:n])
}
+355
View File
@@ -0,0 +1,355 @@
package facebook
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"regexp"
"strconv"
"time"
)
const (
graphAPIVersion = "v25.0"
graphAPIBase = "https://graph.facebook.com"
maxRetries = 3
// maxRetryAfterSec caps the Retry-After sleep to prevent goroutine stalls on abnormal values.
maxRetryAfterSec = 60
)
// fbIDPattern validates Facebook object IDs: numeric or "{num}_{num}" form (post IDs).
var fbIDPattern = regexp.MustCompile(`^\d+(_\d+)?$`)
// GraphClient wraps the Facebook Graph API for a single page instance.
type GraphClient struct {
httpClient *http.Client
pageAccessToken string
pageID string
}
// NewGraphClient creates a new GraphClient for the given page.
func NewGraphClient(pageAccessToken, pageID string) *GraphClient {
return &GraphClient{
httpClient: &http.Client{Timeout: 15 * time.Second},
pageAccessToken: pageAccessToken,
pageID: pageID,
}
}
// VerifyToken checks the page access token by calling GET /me.
func (g *GraphClient) VerifyToken(ctx context.Context) error {
data, err := g.doRequest(ctx, http.MethodGet, "/me?fields=id,name", nil)
if err != nil {
return fmt.Errorf("facebook: token verification failed: %w", err)
}
var result struct {
ID string `json:"id"`
Name string `json:"name"`
}
if err := json.Unmarshal(data, &result); err != nil {
return fmt.Errorf("facebook: token verification parse error: %w", err)
}
slog.Info("facebook: page token verified", "page_id", result.ID, "name", result.Name)
return nil
}
// SubscribeApp subscribes the app to the page's webhook events.
func (g *GraphClient) SubscribeApp(ctx context.Context) error {
if err := validateFBID(g.pageID); err != nil {
return fmt.Errorf("facebook: subscribe app: %w", err)
}
path := fmt.Sprintf("/%s/subscribed_apps?subscribed_fields=feed,messages", g.pageID)
_, err := g.doRequest(ctx, http.MethodPost, path, nil)
if err != nil {
return fmt.Errorf("facebook: subscribe app failed: %w", err)
}
slog.Info("facebook: app subscribed to page webhooks", "page_id", g.pageID)
return nil
}
// GetPost fetches a post by ID with message and story fields.
func (g *GraphClient) GetPost(ctx context.Context, postID string) (*GraphPost, error) {
if err := validateFBID(postID); err != nil {
return nil, fmt.Errorf("facebook: get post: %w", err)
}
path := fmt.Sprintf("/%s?fields=id,message,story,created_time", postID)
data, err := g.doRequest(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var post GraphPost
if err := json.Unmarshal(data, &post); err != nil {
return nil, fmt.Errorf("facebook: parse post: %w", err)
}
return &post, nil
}
// GetComment fetches a single comment by ID.
func (g *GraphClient) GetComment(ctx context.Context, commentID string) (*GraphComment, error) {
if err := validateFBID(commentID); err != nil {
return nil, fmt.Errorf("facebook: get comment: %w", err)
}
path := fmt.Sprintf("/%s?fields=id,message,from,created_time", commentID)
data, err := g.doRequest(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var c GraphComment
if err := json.Unmarshal(data, &c); err != nil {
return nil, fmt.Errorf("facebook: parse comment: %w", err)
}
return &c, nil
}
// GetCommentThread fetches up to limit comments under a parent comment.
func (g *GraphClient) GetCommentThread(ctx context.Context, parentCommentID string, limit int) ([]GraphComment, error) {
if err := validateFBID(parentCommentID); err != nil {
return nil, fmt.Errorf("facebook: get comment thread: %w", err)
}
if limit <= 0 {
limit = 10
}
path := fmt.Sprintf("/%s/comments?fields=id,message,from,created_time&limit=%d", parentCommentID, limit)
data, err := g.doRequest(ctx, http.MethodGet, path, nil)
if err != nil {
return nil, err
}
var resp GraphListResponse[GraphComment]
if err := json.Unmarshal(data, &resp); err != nil {
return nil, fmt.Errorf("facebook: parse comment thread: %w", err)
}
return resp.Data, nil
}
// ReplyToComment posts a reply to a comment. Returns the new comment ID.
func (g *GraphClient) ReplyToComment(ctx context.Context, commentID, message string) (string, error) {
if err := validateFBID(commentID); err != nil {
return "", fmt.Errorf("facebook: reply to comment: %w", err)
}
path := fmt.Sprintf("/%s/comments", commentID)
body := map[string]string{"message": message}
data, err := g.doRequest(ctx, http.MethodPost, path, body)
if err != nil {
return "", err
}
var result struct {
ID string `json:"id"`
}
if err := json.Unmarshal(data, &result); err != nil {
return "", fmt.Errorf("facebook: parse reply result: %w", err)
}
return result.ID, nil
}
// SendMessage sends a Messenger message to the given recipient. Returns message ID.
func (g *GraphClient) SendMessage(ctx context.Context, recipientID, message string) (string, error) {
body := map[string]any{
"recipient": map[string]string{"id": recipientID},
"message": map[string]string{"text": message},
}
data, err := g.doRequest(ctx, http.MethodPost, "/me/messages", body)
if err != nil {
return "", err
}
var result struct {
MessageID string `json:"message_id"`
}
if err := json.Unmarshal(data, &result); err != nil {
return "", fmt.Errorf("facebook: parse send message result: %w", err)
}
return result.MessageID, nil
}
// SendTypingOn sends a typing indicator to the recipient (auto-off after 3s).
func (g *GraphClient) SendTypingOn(ctx context.Context, recipientID string) error {
body := map[string]any{
"recipient": map[string]string{"id": recipientID},
"sender_action": "typing_on",
}
_, err := g.doRequest(ctx, http.MethodPost, "/me/messages", body)
return err
}
// doRequest executes a Graph API call with retries on transient errors.
// The page access token is passed via Authorization header (never in the URL).
func (g *GraphClient) doRequest(ctx context.Context, method, path string, body any) ([]byte, error) {
apiURL := fmt.Sprintf("%s/%s%s", graphAPIBase, graphAPIVersion, path)
for attempt := 0; attempt < maxRetries; attempt++ {
if attempt > 0 {
backoff := time.Duration(1<<uint(attempt-1)) * time.Second
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(backoff):
}
}
var reqBody io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return nil, fmt.Errorf("facebook: marshal request: %w", err)
}
reqBody = bytes.NewReader(b)
}
req, err := http.NewRequestWithContext(ctx, method, apiURL, reqBody)
if err != nil {
return nil, fmt.Errorf("facebook: build request: %w", err)
}
// Pass token via header to avoid URL logging exposure.
req.Header.Set("Authorization", "Bearer "+g.pageAccessToken)
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
resp, err := g.httpClient.Do(req)
if err != nil {
if attempt < maxRetries-1 {
slog.Warn("facebook: api request error, retrying", "attempt", attempt+1, "err", err)
continue
}
return nil, fmt.Errorf("facebook: api request: %w", err)
}
respBody, readErr := io.ReadAll(resp.Body)
resp.Body.Close()
if readErr != nil {
return nil, fmt.Errorf("facebook: read response: %w", readErr)
}
// Proactive rate limit monitoring.
g.logRateLimit(resp)
// Retry on 5xx.
if resp.StatusCode >= 500 && attempt < maxRetries-1 {
slog.Warn("facebook: server error, retrying", "status", resp.StatusCode, "attempt", attempt+1)
continue
}
// Parse Graph API error envelope.
if resp.StatusCode >= 400 {
var apiErr graphErrorBody
if json.Unmarshal(respBody, &apiErr) == nil && apiErr.Error.Code != 0 {
// 24h messaging window violation — not retryable.
if apiErr.Error.Code == 551 || apiErr.Error.Subcode == 2018109 {
slog.Warn("facebook: 24h messaging window expired", "code", apiErr.Error.Code)
return nil, &graphAPIError{code: apiErr.Error.Code, msg: apiErr.Error.Message}
}
// Rate limited: sleep and retry (capped).
if resp.StatusCode == 429 && attempt < maxRetries-1 {
retryAfter := parseRetryAfter(resp)
slog.Warn("facebook: rate limited", "retry_after", retryAfter)
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(retryAfter):
}
continue
}
return nil, &graphAPIError{code: apiErr.Error.Code, msg: apiErr.Error.Message}
}
return nil, fmt.Errorf("facebook: http %d", resp.StatusCode)
}
return respBody, nil
}
// All attempts exhausted (only reachable when every iteration took the continue path).
return nil, fmt.Errorf("facebook: max retries exceeded")
}
// logRateLimit parses the X-Business-Use-Case-Usage header and warns when approaching limits.
func (g *GraphClient) logRateLimit(resp *http.Response) {
usage := resp.Header.Get("X-Business-Use-Case-Usage")
if usage == "" {
return
}
var parsed map[string][]struct {
CallCount int `json:"call_count"`
}
if err := json.Unmarshal([]byte(usage), &parsed); err != nil {
return
}
for _, entries := range parsed {
for _, e := range entries {
if e.CallCount >= 95 {
slog.Warn("facebook: rate limit critical", "call_count_pct", e.CallCount, "page_id", g.pageID)
} else if e.CallCount >= 80 {
slog.Warn("facebook: rate limit warning", "call_count_pct", e.CallCount, "page_id", g.pageID)
}
}
}
}
// parseRetryAfter extracts the Retry-After header, capped at maxRetryAfterSec.
func parseRetryAfter(resp *http.Response) time.Duration {
val := resp.Header.Get("Retry-After")
if val == "" {
return 5 * time.Second
}
secs, err := strconv.Atoi(val)
if err != nil || secs <= 0 {
return 5 * time.Second
}
if secs > maxRetryAfterSec {
secs = maxRetryAfterSec
}
return time.Duration(secs) * time.Second
}
// validateFBID returns an error if id is not a valid Facebook object ID format.
// Facebook IDs are numeric strings, or "{num}_{num}" for post IDs.
func validateFBID(id string) error {
if id == "" {
return fmt.Errorf("empty facebook ID")
}
if !fbIDPattern.MatchString(id) {
return fmt.Errorf("invalid facebook ID format: %q", id)
}
return nil
}
// graphAPIError is a structured error from the Facebook Graph API.
type graphAPIError struct {
code int
msg string
}
func (e *graphAPIError) Error() string {
return fmt.Sprintf("facebook graph api error %d: %s", e.code, e.msg)
}
// IsAuthError returns true when the error is an expired or invalid token.
func IsAuthError(err error) bool {
var ge *graphAPIError
if !errors.As(err, &ge) {
return false
}
return ge.code == 190 || ge.code == 102
}
// IsPermissionError returns true for permission-denied errors.
func IsPermissionError(err error) bool {
var ge *graphAPIError
if !errors.As(err, &ge) {
return false
}
return ge.code == 10 || ge.code == 200
}
// IsRateLimitError returns true for rate limit errors.
func IsRateLimitError(err error) bool {
var ge *graphAPIError
if !errors.As(err, &ge) {
return false
}
return ge.code == 4 || ge.code == 17 || ge.code == 32 || ge.code == 613
}
@@ -0,0 +1,70 @@
package facebook
import (
"fmt"
"log/slog"
)
// handleMessagingEvent processes a Messenger inbox event.
func (ch *Channel) handleMessagingEvent(entry WebhookEntry, event MessagingEvent) {
// Feature gate.
if !ch.config.Features.MessengerAutoReply {
return
}
// Page routing guard (before dedup write).
if entry.ID != ch.pageID {
return
}
// Self-message prevention: skip messages sent by the page itself.
if event.Sender.ID == ch.pageID {
return
}
// Skip delivery/read receipts and other non-content events.
if event.Message == nil && event.Postback == nil {
return
}
// Dedup by message MID or postback signature (include payload to reduce collision risk).
var eventKey string
switch {
case event.Message != nil:
eventKey = "msg:" + event.Message.MID
case event.Postback != nil:
eventKey = fmt.Sprintf("postback:%s:%d:%s", event.Sender.ID, event.Timestamp, event.Postback.Payload)
}
if ch.isDup(eventKey) {
slog.Debug("facebook: duplicate messaging event skipped", "key", eventKey)
return
}
// Extract text content.
var content string
switch {
case event.Message != nil && event.Message.Text != "":
content = event.Message.Text
case event.Postback != nil:
content = event.Postback.Title
default:
// Attachment-only message — skip for now.
return
}
senderID := event.Sender.ID
// Messenger sessions are 1:1: chatID = senderID (channel name scopes the session).
chatID := senderID
metadata := map[string]string{
"fb_mode": "messenger",
"message_id": eventKey,
"page_id": ch.pageID,
"sender_id": senderID,
}
if ch.config.MessengerOptions.SessionTimeout != "" {
metadata["session_timeout"] = ch.config.MessengerOptions.SessionTimeout
}
ch.HandleMessage(senderID, chatID, content, nil, metadata, "direct")
}
@@ -0,0 +1,95 @@
package facebook
import (
"context"
"sync"
"time"
"golang.org/x/sync/singleflight"
)
const defaultPostCacheTTL = 15 * time.Minute
// postCacheEntry holds a cached post with its expiry time.
type postCacheEntry struct {
post *GraphPost
expiresAt time.Time
}
// PostFetcher fetches and caches Facebook post content and comment threads.
// singleflight prevents redundant concurrent Graph API calls for the same post
// (cache stampede on viral posts receiving many comments simultaneously).
type PostFetcher struct {
graphClient *GraphClient
cacheTTL time.Duration
cache sync.Map // postID(string) → *postCacheEntry
sfGroup singleflight.Group // coalesces concurrent fetches for the same postID
}
// NewPostFetcher creates a PostFetcher with the given cache TTL string (e.g. "15m").
// Falls back to defaultPostCacheTTL if the string is empty or unparseable.
func NewPostFetcher(client *GraphClient, cacheTTLStr string) *PostFetcher {
ttl := defaultPostCacheTTL
if cacheTTLStr != "" {
if d, err := time.ParseDuration(cacheTTLStr); err == nil && d > 0 {
ttl = d
}
}
return &PostFetcher{
graphClient: client,
cacheTTL: ttl,
}
}
// GetPost returns the post for postID, using cache when fresh.
// Concurrent callers for the same postID share one inflight request.
func (pf *PostFetcher) GetPost(ctx context.Context, postID string) (*GraphPost, error) {
if postID == "" {
return nil, nil
}
// Check cache first.
if v, ok := pf.cache.Load(postID); ok {
entry := v.(*postCacheEntry)
if time.Now().Before(entry.expiresAt) {
return entry.post, nil
}
pf.cache.Delete(postID)
}
// Coalesce concurrent fetches for the same postID.
val, err, _ := pf.sfGroup.Do(postID, func() (any, error) {
// Re-check cache inside the singleflight call (another goroutine may have populated it).
if v, ok := pf.cache.Load(postID); ok {
entry := v.(*postCacheEntry)
if time.Now().Before(entry.expiresAt) {
return entry.post, nil
}
}
post, err := pf.graphClient.GetPost(ctx, postID)
if err != nil {
return nil, err
}
pf.cache.Store(postID, &postCacheEntry{
post: post,
expiresAt: time.Now().Add(pf.cacheTTL),
})
return post, nil
})
if err != nil {
return nil, err
}
if val == nil {
return nil, nil
}
return val.(*GraphPost), nil
}
// GetCommentThread fetches up to depth comments under parentCommentID.
// Not cached — thread content changes frequently as replies arrive.
func (pf *PostFetcher) GetCommentThread(ctx context.Context, parentCommentID string, depth int) ([]GraphComment, error) {
if parentCommentID == "" {
return nil, nil
}
return pf.graphClient.GetCommentThread(ctx, parentCommentID, depth)
}
+148
View File
@@ -0,0 +1,148 @@
// Package facebook implements the Facebook Fanpage channel for GoClaw.
// Supports: comment auto-reply, Messenger inbox auto-reply, first inbox DM.
package facebook
// facebookCreds holds encrypted credentials stored in channel_instances.credentials.
type facebookCreds struct {
PageAccessToken string `json:"page_access_token"`
AppSecret string `json:"app_secret"`
VerifyToken string `json:"verify_token"`
}
// facebookInstanceConfig holds non-secret config from channel_instances.config JSONB.
type facebookInstanceConfig struct {
PageID string `json:"page_id"`
Features struct {
CommentReply bool `json:"comment_reply"`
FirstInbox bool `json:"first_inbox"`
MessengerAutoReply bool `json:"messenger_auto_reply"`
} `json:"features"`
CommentReplyOptions struct {
IncludePostContext bool `json:"include_post_context"`
MaxThreadDepth int `json:"max_thread_depth"`
} `json:"comment_reply_options"`
MessengerOptions struct {
SessionTimeout string `json:"session_timeout"`
} `json:"messenger_options"`
PostContextCacheTTL string `json:"post_context_cache_ttl"`
// FirstInboxMessage is the DM text sent to commenters (first-inbox feature).
// Defaults to Vietnamese if empty. Operators should set this to match their page language.
FirstInboxMessage string `json:"first_inbox_message,omitempty"`
AllowFrom []string `json:"allow_from,omitempty"`
}
// --- Webhook payloads ---
// WebhookPayload is the top-level Facebook webhook event payload.
type WebhookPayload struct {
Object string `json:"object"`
Entry []WebhookEntry `json:"entry"`
}
// WebhookEntry is one page's events within a webhook delivery.
type WebhookEntry struct {
ID string `json:"id"` // page_id
Time int64 `json:"time"`
Changes []WebhookChange `json:"changes,omitempty"` // feed events (comments, posts)
Messaging []MessagingEvent `json:"messaging,omitempty"` // Messenger events
}
// WebhookChange is a single change event for feed subscriptions.
type WebhookChange struct {
Field string `json:"field"` // "feed", "mention", etc.
Value ChangeValue `json:"value"`
}
// ChangeValue holds the details of a feed change event.
type ChangeValue struct {
From FBUser `json:"from"`
Item string `json:"item"` // "comment", "post", "status"
CommentID string `json:"comment_id"`
PostID string `json:"post_id"`
ParentID string `json:"parent_id"` // parent comment ID for nested replies
Message string `json:"message"`
Verb string `json:"verb"` // "add", "edit", "remove"
CreatedTime int64 `json:"created_time"`
}
// MessagingEvent is a single Messenger inbox event.
type MessagingEvent struct {
Sender FBUser `json:"sender"`
Recipient FBUser `json:"recipient"`
Timestamp int64 `json:"timestamp"`
Message *IncomingMessage `json:"message,omitempty"`
Postback *Postback `json:"postback,omitempty"`
}
// IncomingMessage holds a Messenger text/attachment message.
type IncomingMessage struct {
MID string `json:"mid"`
Text string `json:"text"`
Attachments []Attachment `json:"attachments,omitempty"`
}
// Postback holds a Messenger postback event.
type Postback struct {
Title string `json:"title"`
Payload string `json:"payload"`
}
// Attachment is a Messenger media attachment.
type Attachment struct {
Type string `json:"type"` // "image", "video", "audio", "file"
Payload AttachmentPayload `json:"payload"`
}
// AttachmentPayload holds the URL of a media attachment.
type AttachmentPayload struct {
URL string `json:"url"`
}
// FBUser is a minimal Facebook user reference.
type FBUser struct {
ID string `json:"id"`
Name string `json:"name,omitempty"`
}
// --- Graph API response types ---
// GraphComment is a Facebook comment object from the Graph API.
type GraphComment struct {
ID string `json:"id"`
Message string `json:"message"`
From FBUser `json:"from"`
CreatedTime string `json:"created_time"`
}
// GraphPost is a Facebook post object from the Graph API.
type GraphPost struct {
ID string `json:"id"`
Message string `json:"message"`
Story string `json:"story,omitempty"`
CreatedTime string `json:"created_time"`
}
// GraphPaging holds cursor-based pagination for Graph API list responses.
type GraphPaging struct {
Cursors struct {
Before string `json:"before"`
After string `json:"after"`
} `json:"cursors"`
Next string `json:"next,omitempty"`
}
// GraphListResponse is a generic Graph API list response.
type GraphListResponse[T any] struct {
Data []T `json:"data"`
Paging GraphPaging `json:"paging,omitempty"`
}
// graphErrorBody is the error envelope returned by Graph API on failures.
type graphErrorBody struct {
Error struct {
Message string `json:"message"`
Type string `json:"type"`
Code int `json:"code"`
Subcode int `json:"error_subcode"`
} `json:"error"`
}
@@ -0,0 +1,152 @@
package facebook
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"log/slog"
"net/http"
"regexp"
"strings"
)
// WebhookHandler implements http.Handler for the Facebook webhook endpoint.
// Handles both the GET verification challenge and POST event delivery.
type WebhookHandler struct {
appSecret string
verifyToken string
extraSecrets []string // additional app secrets for multi-Meta-App deployments
onComment func(entry WebhookEntry, change ChangeValue)
onMessage func(entry WebhookEntry, event MessagingEvent)
}
// NewWebhookHandler creates a new WebhookHandler.
func NewWebhookHandler(appSecret, verifyToken string) *WebhookHandler {
return &WebhookHandler{
appSecret: appSecret,
verifyToken: verifyToken,
}
}
// hubChallengePattern validates that hub.challenge is safe to reflect.
var hubChallengePattern = regexp.MustCompile(`^[a-zA-Z0-9_\-]{1,256}$`)
// ServeHTTP handles Facebook webhook GET (verification) and POST (event delivery).
func (wh *WebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
wh.handleVerification(w, r)
case http.MethodPost:
wh.handleEvent(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
// handleVerification responds to Facebook's webhook verification challenge.
func (wh *WebhookHandler) handleVerification(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
if q.Get("hub.mode") != "subscribe" {
http.Error(w, "invalid hub.mode", http.StatusForbidden)
return
}
if q.Get("hub.verify_token") != wh.verifyToken {
slog.Warn("security.facebook_webhook_verify_token_mismatch",
"remote_addr", r.RemoteAddr)
http.Error(w, "invalid verify token", http.StatusForbidden)
return
}
challenge := q.Get("hub.challenge")
// Validate challenge before reflecting to prevent injection if Content-Type changes.
if !hubChallengePattern.MatchString(challenge) {
http.Error(w, "invalid challenge", http.StatusBadRequest)
return
}
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(challenge))
}
// handleEvent processes a Facebook webhook event delivery.
// Always returns 200 OK — Facebook retries on non-2xx for 24h.
func (wh *WebhookHandler) handleEvent(w http.ResponseWriter, r *http.Request) {
const maxBodyBytes = 4 << 20 // 4 MB
lr := io.LimitReader(r.Body, maxBodyBytes+1)
body, err := io.ReadAll(lr)
if err != nil {
slog.Warn("facebook: webhook read body error", "err", err)
w.WriteHeader(http.StatusOK) // 200 so Facebook stops retrying a bad delivery
return
}
if len(body) > maxBodyBytes {
slog.Warn("facebook: webhook body exceeded limit, event dropped", "bytes", len(body))
w.WriteHeader(http.StatusOK)
return
}
sig := r.Header.Get("X-Hub-Signature-256")
verified := verifySignature(body, sig, wh.appSecret)
if !verified {
// Try extra secrets (multi-Meta-App deployments share one webhook endpoint).
for _, s := range wh.extraSecrets {
if verifySignature(body, sig, s) {
verified = true
break
}
}
}
if !verified {
slog.Warn("security.facebook_webhook_signature_invalid", "remote_addr", r.RemoteAddr)
w.WriteHeader(http.StatusOK) // return 200 so Facebook stops sending
return
}
var payload WebhookPayload
if err := json.Unmarshal(body, &payload); err != nil {
slog.Warn("facebook: webhook parse error", "err", err)
w.WriteHeader(http.StatusOK)
return
}
if payload.Object != "page" {
w.WriteHeader(http.StatusOK)
return
}
for _, entry := range payload.Entry {
// Feed events (comments, posts).
for _, change := range entry.Changes {
if change.Field == "feed" && change.Value.Item == "comment" {
if wh.onComment != nil {
wh.onComment(entry, change.Value)
}
}
}
// Messenger events.
for _, event := range entry.Messaging {
if wh.onMessage != nil {
wh.onMessage(entry, event)
}
}
}
w.WriteHeader(http.StatusOK)
}
// verifySignature validates the X-Hub-Signature-256 header using HMAC-SHA256.
// Facebook sends "sha256=<hex_digest>"; we recompute and compare.
func verifySignature(body []byte, signature, appSecret string) bool {
const prefix = "sha256="
if !strings.HasPrefix(signature, prefix) {
return false
}
expected, err := hex.DecodeString(signature[len(prefix):])
if err != nil {
return false
}
mac := hmac.New(sha256.New, []byte(appSecret))
mac.Write(body)
computed := mac.Sum(nil)
return hmac.Equal(computed, expected)
}
@@ -0,0 +1,106 @@
package facebook
import (
"log/slog"
"net/http"
"sync"
)
// webhookRouter routes incoming Facebook webhook events to the correct channel instance by page_id.
// A single HTTP handler is shared across all facebook channel instances on the same server.
//
// Multi-Meta-App note: all page instances registered here are expected to share the same Meta App
// (and thus the same app_secret). If instances with different secrets are registered, ServeHTTP
// tries all known secrets and accepts the payload if any matches.
type webhookRouter struct {
mu sync.RWMutex
instances map[string]*Channel // pageID → channel
routeHandled bool // true after first webhookRoute() call
}
var globalRouter = &webhookRouter{
instances: make(map[string]*Channel),
}
func (r *webhookRouter) register(ch *Channel) {
r.mu.Lock()
defer r.mu.Unlock()
r.instances[ch.pageID] = ch
}
func (r *webhookRouter) unregister(pageID string) {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.instances, pageID)
}
// webhookRoute returns the path+handler for the first call; ("", nil) for subsequent calls.
func (r *webhookRouter) webhookRoute() (string, http.Handler) {
r.mu.Lock()
defer r.mu.Unlock()
if !r.routeHandled {
r.routeHandled = true
return webhookPath, r
}
return "", nil
}
// ServeHTTP is the shared handler for all Facebook page webhooks.
// Routes each entry to the matching channel instance by page_id.
//
// Multi-Meta-App support: all registered page secrets are collected and tried in order.
// A payload is accepted if its signature matches any known app_secret. In the common
// case (single Meta App) there is exactly one secret and behavior is unchanged.
func (r *webhookRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
r.mu.RLock()
var primarySecret, verifyToken string
var extraSecrets []string
seenSecrets := make(map[string]bool)
for _, ch := range r.instances {
s := ch.webhookH.appSecret
if primarySecret == "" {
primarySecret = s
verifyToken = ch.webhookH.verifyToken
seenSecrets[s] = true
} else if !seenSecrets[s] {
extraSecrets = append(extraSecrets, s)
seenSecrets[s] = true
}
}
r.mu.RUnlock()
if primarySecret == "" {
// No instances registered yet.
w.WriteHeader(http.StatusOK)
return
}
if len(extraSecrets) > 0 {
slog.Warn("security.facebook_multi_meta_app",
"extra_app_count", len(extraSecrets),
"note", "multiple Meta App secrets registered; payloads verified against all known secrets")
}
routingWH := &WebhookHandler{
appSecret: primarySecret,
verifyToken: verifyToken,
extraSecrets: extraSecrets,
}
routingWH.onComment = func(entry WebhookEntry, change ChangeValue) {
r.mu.RLock()
target := r.instances[entry.ID]
r.mu.RUnlock()
if target != nil {
target.handleCommentEvent(entry, change)
}
}
routingWH.onMessage = func(entry WebhookEntry, event MessagingEvent) {
r.mu.RLock()
target := r.instances[entry.ID]
r.mu.RUnlock()
if target != nil {
target.handleMessagingEvent(entry, event)
}
}
routingWH.ServeHTTP(w, req)
}
+1 -1
View File
@@ -169,7 +169,7 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) {
"message_id": messageID,
"chat_type": mc.ChatType,
"sender_name": senderName,
"display_name": senderName,
"display_name": channels.SanitizeDisplayName(senderName),
"mentioned_bot": fmt.Sprintf("%t", mc.MentionedBot),
"platform": channels.TypeFeishu,
}
+240
View File
@@ -0,0 +1,240 @@
package pancake
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"mime/multipart"
"net/http"
"path/filepath"
"time"
)
const (
publicAPIBase = "https://pages.fm/api/public_api/v2" // page-level APIs
userAPIBase = "https://pages.fm/api/v1" // user-level APIs (list pages, etc.)
httpTimeout = 30 * time.Second
)
// APIClient wraps the Pancake REST API for a single page instance.
type APIClient struct {
pageV1BaseURL string
pageV2BaseURL string
userBaseURL string
pageAccessToken string
apiKey string
pageID string
httpClient *http.Client
}
// NewAPIClient creates a new Pancake APIClient for the given page.
func NewAPIClient(apiKey, pageAccessToken, pageID string) *APIClient {
return &APIClient{
pageV1BaseURL: "https://pages.fm/api/public_api/v1",
pageV2BaseURL: publicAPIBase,
userBaseURL: userAPIBase,
pageAccessToken: pageAccessToken,
apiKey: apiKey,
pageID: pageID,
httpClient: &http.Client{Timeout: httpTimeout},
}
}
// VerifyToken validates the page_access_token via a lightweight API call.
func (c *APIClient) VerifyToken(ctx context.Context) error {
url := fmt.Sprintf("%s/pages/%s/conversations?limit=1", c.pageV2BaseURL, c.pageID)
if err := c.doRequest(ctx, http.MethodGet, url, nil); err != nil {
return fmt.Errorf("pancake: token verification failed: %w", err)
}
slog.Info("pancake: page token verified", "page_id", c.pageID)
return nil
}
// GetPage fetches page metadata including platform (facebook/zalo/instagram/tiktok/whatsapp/line).
func (c *APIClient) GetPage(ctx context.Context) (*PageInfo, error) {
url := fmt.Sprintf("%s/pages", c.userBaseURL)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return nil, fmt.Errorf("pancake: build get-pages request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+c.apiKey)
req.Header.Set("Content-Type", "application/json")
res, err := c.httpClient.Do(req)
if err != nil {
return nil, fmt.Errorf("pancake: get pages request failed: %w", err)
}
defer res.Body.Close()
body, err := io.ReadAll(res.Body)
if err != nil {
return nil, fmt.Errorf("pancake: read get-pages response: %w", err)
}
var result struct {
Data []PageInfo `json:"data"`
}
if err := json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("pancake: parse get-pages response: %w", err)
}
for i := range result.Data {
if result.Data[i].ID == c.pageID {
return &result.Data[i], nil
}
}
// Page not found in list — return minimal info without platform
slog.Warn("pancake: page not found in pages list, platform unknown", "page_id", c.pageID)
return &PageInfo{ID: c.pageID}, nil
}
// SendMessage sends a text message to a conversation.
func (c *APIClient) SendMessage(ctx context.Context, conversationID, content string) error {
body, _ := json.Marshal(SendMessageRequest{
Action: "reply_inbox",
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: send message: %w", err)
}
return nil
}
// SendAttachmentMessage sends one or more uploaded content IDs to a conversation.
func (c *APIClient) SendAttachmentMessage(ctx context.Context, conversationID string, contentIDs []string) error {
body, _ := json.Marshal(SendMessageRequest{
Action: "reply_inbox",
ContentIDs: contentIDs,
})
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: send attachment message: %w", err)
}
return nil
}
// UploadMedia uploads a file via multipart/form-data and returns the attachment ID.
func (c *APIClient) UploadMedia(ctx context.Context, filename string, data io.Reader, contentType string) (string, error) {
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
fw, err := mw.CreateFormFile("file", filepath.Base(filename))
if err != nil {
return "", fmt.Errorf("pancake: create form file: %w", err)
}
if _, err := io.Copy(fw, data); err != nil {
return "", fmt.Errorf("pancake: copy file data: %w", err)
}
mw.Close()
url := fmt.Sprintf("%s/pages/%s/upload_contents", c.pageV1BaseURL, c.pageID)
req, err := c.newPageRequest(ctx, http.MethodPost, url, &buf)
if err != nil {
return "", fmt.Errorf("pancake: build upload request: %w", err)
}
req.Header.Set("Content-Type", mw.FormDataContentType())
res, err := c.httpClient.Do(req)
if err != nil {
return "", fmt.Errorf("pancake: upload request failed: %w", err)
}
defer res.Body.Close()
respBody, err := io.ReadAll(res.Body)
if err != nil {
return "", fmt.Errorf("pancake: read upload response: %w", err)
}
var uploadResp UploadResponse
if err := json.Unmarshal(respBody, &uploadResp); err != nil {
return "", fmt.Errorf("pancake: parse upload response: %w", err)
}
if uploadResp.ID == "" {
return "", fmt.Errorf("pancake: upload response missing attachment ID")
}
return uploadResp.ID, 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 {
req, err := c.newPageRequest(ctx, method, url, body)
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
res, err := c.httpClient.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
// Always read the full body to allow HTTP connection reuse.
respBody, _ := io.ReadAll(res.Body)
if res.StatusCode >= 400 {
var apiErr apiError
if jsonErr := json.Unmarshal(respBody, &apiErr); jsonErr == nil && apiErr.Message != "" {
return &apiErr
}
return fmt.Errorf("pancake: HTTP %d", res.StatusCode)
}
// Some Pancake endpoints return HTTP 200 with a JSON body carrying success=false.
// Treat these as application-level send failures instead of silent success.
var appResp struct {
Success *bool `json:"success,omitempty"`
Message string `json:"message,omitempty"`
}
if err := json.Unmarshal(respBody, &appResp); err == nil && appResp.Success != nil && !*appResp.Success {
if appResp.Message != "" {
return fmt.Errorf("pancake: %s", appResp.Message)
}
return fmt.Errorf("pancake: request reported success=false")
}
return nil
}
func (c *APIClient) newPageRequest(ctx context.Context, method, rawURL string, body io.Reader) (*http.Request, error) {
req, err := http.NewRequestWithContext(ctx, method, rawURL, body)
if err != nil {
return nil, err
}
query := req.URL.Query()
query.Set("page_access_token", c.pageAccessToken)
req.URL.RawQuery = query.Encode()
// Keep the header for compatibility; official docs require the query token.
req.Header.Set("Authorization", "Bearer "+c.pageAccessToken)
return req, nil
}
// isAuthError checks if an error is an authentication/authorization failure.
// Uses errors.As to handle wrapped errors consistently with Facebook channel pattern.
func isAuthError(err error) bool {
var ae *apiError
if !errors.As(err, &ae) {
return false
}
return ae.Code == 401 || ae.Code == 403 || ae.Code == 4001 || ae.Code == 4003
}
// isRateLimitError checks if an error is a rate limit response.
func isRateLimitError(err error) bool {
var ae *apiError
if !errors.As(err, &ae) {
return false
}
return ae.Code == 429 || ae.Code == 4029
}
+146
View File
@@ -0,0 +1,146 @@
package pancake
import (
"html"
"regexp"
"strings"
"time"
)
var (
htmlBreakTagRe = regexp.MustCompile(`(?i)<br\b[^>]*\/?>`)
htmlCloseTagRe = regexp.MustCompile(`(?i)</(?:div|p|li|ul|ol|h[1-6])>`)
htmlTagRe = regexp.MustCompile(`(?i)<[^>]+>`)
)
// isDup checks and records a dedup key. Returns true if the key was already seen.
func (ch *Channel) isDup(key string) bool {
_, loaded := ch.dedup.LoadOrStore(key, time.Now())
return loaded
}
// splitMessage splits text into chunks no longer than maxLen Unicode code points.
// Operates on runes (not bytes) to avoid splitting multi-byte characters (CJK, emoji, Vietnamese).
func splitMessage(text string, maxLen int) []string {
runes := []rune(text)
if maxLen <= 0 || len(runes) <= maxLen {
return []string{text}
}
var parts []string
for len(runes) > maxLen {
parts = append(parts, string(runes[:maxLen]))
runes = runes[maxLen:]
}
if len(runes) > 0 {
parts = append(parts, string(runes))
}
return parts
}
func (ch *Channel) forgetOutboundEcho(conversationID, content string) {
if conversationID == "" {
return
}
normalized := normalizeEchoContent(content)
if normalized == "" {
return
}
ch.recentOutbound.Delete(conversationID + "\x00" + normalized)
}
func (ch *Channel) rememberOutboundEcho(conversationID, content string) {
if conversationID == "" {
return
}
normalized := normalizeEchoContent(content)
if normalized == "" {
return
}
ch.recentOutbound.Store(conversationID+"\x00"+normalized, time.Now())
}
func (ch *Channel) isRecentOutboundEcho(conversationID, content string) bool {
if conversationID == "" {
return false
}
normalized := normalizeEchoContent(content)
if normalized == "" {
return false
}
key := conversationID + "\x00" + normalized
v, ok := ch.recentOutbound.Load(key)
if !ok {
return false
}
ts, ok := v.(time.Time)
if !ok {
ch.recentOutbound.Delete(key)
return false
}
if time.Since(ts) > outboundEchoTTL {
ch.recentOutbound.Delete(key)
return false
}
return true
}
func normalizeEchoContent(content string) string {
content = strings.TrimSpace(content)
if content == "" {
return ""
}
content = html.UnescapeString(content)
content = strings.ReplaceAll(content, "\r\n", "\n")
content = strings.ReplaceAll(content, "\r", "\n")
content = htmlBreakTagRe.ReplaceAllString(content, "\n")
content = htmlCloseTagRe.ReplaceAllString(content, "\n")
content = htmlTagRe.ReplaceAllString(content, "")
lines := strings.Split(content, "\n")
normalized := make([]string, 0, len(lines))
pendingBlank := false
for _, line := range lines {
line = strings.Join(strings.Fields(line), " ")
if line == "" {
if len(normalized) == 0 || pendingBlank {
continue
}
pendingBlank = true
continue
}
if pendingBlank {
normalized = append(normalized, "")
pendingBlank = false
}
normalized = append(normalized, line)
}
return strings.TrimSpace(strings.Join(normalized, "\n"))
}
// runDedupCleaner evicts dedup entries older than dedupTTL every dedupCleanEvery.
func (ch *Channel) runDedupCleaner() {
ticker := time.NewTicker(dedupCleanEvery)
defer ticker.Stop()
for {
select {
case <-ch.stopCh:
return
case <-ticker.C:
now := time.Now()
ch.dedup.Range(func(k, v any) bool {
if t, ok := v.(time.Time); ok && now.Sub(t) > dedupTTL {
ch.dedup.Delete(k)
}
return true
})
ch.recentOutbound.Range(func(k, v any) bool {
if t, ok := v.(time.Time); ok && now.Sub(t) > outboundEchoTTL {
ch.recentOutbound.Delete(k)
}
return true
})
}
}
}
+87
View File
@@ -0,0 +1,87 @@
package pancake
import (
"regexp"
"strings"
)
// FormatOutbound formats agent response text for the target platform.
// Each platform has different formatting rules and supported markup.
func FormatOutbound(content string, platform string) string {
switch platform {
case "facebook":
return formatForFacebook(content)
case "whatsapp":
return formatForWhatsApp(content)
case "zalo", "instagram", "line":
return stripMarkdown(content)
case "tiktok":
return stripMarkdown(truncateForTikTok(content))
default:
return stripMarkdown(content)
}
}
// formatForFacebook allows basic HTML tags supported by Messenger.
// Strips unsupported tags, keeps bold/italic/links.
func formatForFacebook(content string) string {
// Convert markdown bold (**text** or __text__) to plain (FB Messenger uses plain text)
content = reBold.ReplaceAllString(content, "$1")
content = reItalic.ReplaceAllString(content, "$1")
// Strip markdown code blocks and inline code
content = reCodeBlock.ReplaceAllString(content, "$1")
content = reInlineCode.ReplaceAllString(content, "$1")
// Strip markdown headers (## Heading → Heading)
content = reHeader.ReplaceAllString(content, "$1")
return strings.TrimSpace(content)
}
// formatForWhatsApp converts markdown to WhatsApp-native formatting.
// WhatsApp uses *bold*, _italic_, ~strikethrough~, ```code```.
func formatForWhatsApp(content string) string {
// Convert **bold** → *bold* (WhatsApp format)
content = reDoubleBold.ReplaceAllString(content, "*$1*")
// Convert __italic__ → _italic_ (already matches WA format, just clean up __)
content = reDoubleUnderline.ReplaceAllString(content, "_$1_")
// Strip markdown headers
content = reHeader.ReplaceAllString(content, "$1")
// Strip inline code backticks (keep content)
content = reInlineCode.ReplaceAllString(content, "$1")
return strings.TrimSpace(content)
}
// stripMarkdown removes common markdown formatting, returning plain text.
func stripMarkdown(content string) string {
content = reBold.ReplaceAllString(content, "$1")
content = reItalic.ReplaceAllString(content, "$1")
content = reCodeBlock.ReplaceAllString(content, "$1")
content = reInlineCode.ReplaceAllString(content, "$1")
content = reHeader.ReplaceAllString(content, "$1")
content = reLink.ReplaceAllString(content, "$1")
content = reImage.ReplaceAllString(content, "")
return strings.TrimSpace(content)
}
// truncateForTikTok truncates content to TikTok DM limit (500 runes).
// Uses rune slicing to avoid corrupting multi-byte UTF-8 (CJK, Vietnamese, emoji).
func truncateForTikTok(content string) string {
const limit = 500
runes := []rune(content)
if len(runes) <= limit {
return content
}
return string(runes[:limit-3]) + "..."
}
// Compiled regexes for markdown stripping — package-level for efficiency.
var (
reBold = regexp.MustCompile(`(?:\*\*|__)(.+?)(?:\*\*|__)`)
reDoubleBold = regexp.MustCompile(`\*\*(.+?)\*\*`)
reDoubleUnderline = regexp.MustCompile(`__(.+?)__`)
reItalic = regexp.MustCompile(`(?:\*|_)(.+?)(?:\*|_)`)
reCodeBlock = regexp.MustCompile("(?s)```(?:[a-z]*)?\n?(.+?)```")
reInlineCode = regexp.MustCompile("`(.+?)`")
reHeader = regexp.MustCompile(`(?m)^#{1,6}\s+(.+)$`)
reLink = regexp.MustCompile(`\[([^\]]+)\]\([^)]+\)`)
reImage = regexp.MustCompile(`!\[[^\]]*\]\([^)]+\)`)
)
@@ -0,0 +1,72 @@
package pancake
import (
"context"
"fmt"
"log/slog"
"mime"
"os"
"path/filepath"
"github.com/nextlevelbuilder/goclaw/internal/bus"
)
// handleMediaAttachments uploads media files from an OutboundMessage and returns attachment IDs.
// Returns an empty slice (not an error) when no media is attached.
// On upload failure the error is returned so the caller can decide to send text-only.
func (ch *Channel) handleMediaAttachments(ctx context.Context, msg bus.OutboundMessage) ([]string, error) {
if len(msg.Media) == 0 {
return nil, nil
}
var ids []string
for _, att := range msg.Media {
if att.URL == "" {
continue
}
id, err := ch.uploadMediaFile(ctx, att.URL, att.ContentType)
if err != nil {
slog.Warn("pancake: skipping media attachment on upload error",
"page_id", ch.pageID, "url", att.URL, "err", err)
return ids, err
}
ids = append(ids, id)
}
return ids, nil
}
// uploadMediaFile opens a local file path and uploads it to the Pancake API.
// path must be absolute to prevent directory traversal via relative paths.
func (ch *Channel) uploadMediaFile(ctx context.Context, path string, contentType string) (string, error) {
cleanPath := filepath.Clean(path)
if !filepath.IsAbs(cleanPath) {
return "", fmt.Errorf("pancake: media path must be absolute, got: %s", path)
}
f, err := os.Open(cleanPath)
if err != nil {
return "", err
}
defer f.Close()
ct := contentType
if ct == "" {
ct = mimeTypeFromPath(cleanPath)
}
return ch.apiClient.UploadMedia(ctx, filepath.Base(cleanPath), f, ct)
}
// mimeTypeFromPath guesses the MIME type from the file extension.
func mimeTypeFromPath(path string) string {
ext := filepath.Ext(path)
if ext == "" {
return "application/octet-stream"
}
mt := mime.TypeByExtension(ext)
if mt == "" {
return "application/octet-stream"
}
return mt
}
@@ -0,0 +1,140 @@
package pancake
import (
"fmt"
"html"
"log/slog"
"strings"
"github.com/nextlevelbuilder/goclaw/internal/channels"
)
// handleMessagingEvent converts a Pancake "messaging" webhook event to bus.InboundMessage.
func (ch *Channel) handleMessagingEvent(data MessagingData) {
slog.Debug("pancake: handleMessagingEvent called",
"page_id", ch.pageID,
"sender_id", data.Message.SenderID,
"conversation_id", data.ConversationID,
"type", data.Type,
"platform", data.Platform,
"msg_id", data.Message.ID,
"content_len", len(data.Message.Content))
// Dedup by message ID to handle Pancake's at-least-once delivery.
// Skip dedup when message ID is empty — prevents shared slot "msg:" from
// silently dropping all subsequent empty-ID messages across conversations.
var dedupKey string
if data.Message.ID != "" {
dedupKey = fmt.Sprintf("msg:%s", data.Message.ID)
if ch.isDup(dedupKey) {
slog.Info("pancake: duplicate message skipped", "msg_id", data.Message.ID)
return
}
}
// Prevent reply loops: skip messages sent by the page itself.
if data.Message.SenderID == ch.pageID {
slog.Info("pancake: skipping own page message",
"page_id", ch.pageID,
"sender_id", data.Message.SenderID)
return
}
if isAssignedStaff(data.AssigneeIDs, data.Message.SenderID) {
slog.Info("pancake: skipping assigned staff message",
"page_id", ch.pageID,
"sender_id", data.Message.SenderID,
"conversation_id", data.ConversationID)
return
}
if data.Message.SenderID == "" {
slog.Warn("pancake: message missing sender_id, skipping", "msg_id", data.Message.ID)
return
}
// Check echo BEFORE buildMessageContent adds the [From: ...] prefix.
// rememberOutboundEcho stores the raw outbound text; the prefix would cause a
// key mismatch and silently break loop detection.
if ch.isRecentOutboundEcho(data.ConversationID, data.Message.Content) {
slog.Info("pancake: skipping recent outbound echo",
"page_id", ch.pageID,
"conversation_id", data.ConversationID,
"msg_id", data.Message.ID)
return
}
content := buildMessageContent(data)
metadata := map[string]string{
"pancake_mode": strings.ToLower(data.Type), // "inbox" or "comment"
"conversation_type": data.Type,
"platform": data.Platform,
"conversation_id": data.ConversationID,
"message_id": dedupKey,
"display_name": channels.SanitizeDisplayName(data.Message.SenderName),
"page_name": ch.pageName,
}
ch.HandleMessage(
data.Message.SenderID,
data.ConversationID, // ChatID = conversation_id for reply routing
content,
nil, // media handled inline via content URLs
metadata,
"direct", // Pancake inbox conversations are always treated as direct messages
)
slog.Debug("pancake: inbound message published to bus",
"page_id", ch.pageID,
"conv_id", data.ConversationID,
"sender_id", data.Message.SenderID,
"platform", data.Platform,
"type", data.Type,
"channel_name", ch.Name(),
)
}
// buildMessageContent combines text content and attachment URLs into a single string.
// Format: [From: {SenderID} ({SenderName})] {content}
func buildMessageContent(data MessagingData) string {
parts := []string{}
if data.Message.Content != "" {
parts = append(parts, stripHTML(data.Message.Content))
}
for _, att := range data.Message.Attachments {
if att.URL != "" {
parts = append(parts, att.URL)
}
}
body := strings.Join(parts, "\n")
if data.Message.SenderID != "" {
prefix := fmt.Sprintf("[From: %s (%s)]", data.Message.SenderID, data.Message.SenderName)
if body != "" {
return prefix + " " + body
}
return prefix
}
return body
}
// stripHTML removes HTML tags and unescapes HTML entities from s.
func stripHTML(s string) string {
s = htmlTagRe.ReplaceAllString(s, "")
return html.UnescapeString(strings.TrimSpace(s))
}
func isAssignedStaff(assigneeIDs []string, senderID string) bool {
if senderID == "" {
return false
}
for _, assigneeID := range assigneeIDs {
if assigneeID == senderID {
return true
}
}
return false
}
+258
View File
@@ -0,0 +1,258 @@
package pancake
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"net/http"
"sync"
"time"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// Compile-time interface assertions.
var (
_ channels.Channel = (*Channel)(nil)
_ channels.WebhookChannel = (*Channel)(nil)
_ channels.BlockReplyChannel = (*Channel)(nil)
)
const (
dedupTTL = 24 * time.Hour
dedupCleanEvery = 5 * time.Minute
outboundEchoTTL = 45 * time.Second
)
// Channel implements channels.Channel and channels.WebhookChannel for Pancake (pages.fm).
// One channel instance = one Pancake page, which may serve multiple platforms (FB, Zalo, IG, etc.)
type Channel struct {
*channels.BaseChannel
config pancakeInstanceConfig
apiClient *APIClient
pageID string
pageName string // resolved from Pancake page metadata at Start()
platform string // resolved from Pancake page metadata at Start()
webhookSecret string // optional HMAC-SHA256 secret for webhook verification
// dedup prevents processing duplicate webhook deliveries.
dedup sync.Map // eventKey(string) → time.Time
// recentOutbound suppresses short-lived webhook echoes of our own text replies.
recentOutbound sync.Map // conversationID + "\x00" + normalized content → time.Time
stopCh chan struct{}
stopCtx context.Context
stopFn context.CancelFunc
}
// New creates a Pancake Channel from parsed credentials and config.
func New(cfg pancakeInstanceConfig, creds pancakeCreds,
msgBus *bus.MessageBus, _ store.PairingStore) (*Channel, error) {
if creds.APIKey == "" {
return nil, fmt.Errorf("pancake: api_key is required")
}
if creds.PageAccessToken == "" {
return nil, fmt.Errorf("pancake: page_access_token is required")
}
if cfg.PageID == "" {
return nil, fmt.Errorf("pancake: page_id is required")
}
base := channels.NewBaseChannel(channels.TypePancake, msgBus, cfg.AllowFrom)
stopCtx, stopFn := context.WithCancel(context.Background())
ch := &Channel{
BaseChannel: base,
config: cfg,
apiClient: NewAPIClient(creds.APIKey, creds.PageAccessToken, cfg.PageID),
pageID: cfg.PageID,
platform: cfg.Platform,
webhookSecret: creds.WebhookSecret,
stopCh: make(chan struct{}),
stopCtx: stopCtx,
stopFn: stopFn,
}
return ch, nil
}
// Factory creates a Pancake Channel from DB instance data.
// Implements channels.ChannelFactory.
func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
var c pancakeCreds
if err := json.Unmarshal(creds, &c); err != nil {
return nil, fmt.Errorf("pancake: decode credentials: %w", err)
}
var ic pancakeInstanceConfig
if len(cfg) > 0 {
if err := json.Unmarshal(cfg, &ic); err != nil {
return nil, fmt.Errorf("pancake: decode config: %w", err)
}
}
ch, err := New(ic, c, msgBus, pairingSvc)
if err != nil {
return nil, err
}
ch.SetName(name)
return ch, nil
}
// Start connects the channel: verifies token, resolves platform, registers webhook.
func (ch *Channel) Start(ctx context.Context) error {
ch.MarkStarting("connecting to Pancake page")
if err := ch.apiClient.VerifyToken(ctx); err != nil {
ch.MarkFailed("token invalid", err.Error(), channels.ChannelFailureKindAuth, false)
return err
}
// Resolve platform and page name from page metadata (best-effort — don't fail on this).
if ch.platform == "" || ch.pageName == "" {
if page, err := ch.apiClient.GetPage(ctx); err != nil {
slog.Warn("pancake: could not resolve platform from page metadata", "page_id", ch.pageID, "err", err)
} else {
if page.Platform != "" {
ch.platform = page.Platform
}
if page.Name != "" {
ch.pageName = page.Name
}
}
}
if ch.webhookSecret == "" {
slog.Warn("security.pancake_webhook_no_secret",
"page_id", ch.pageID,
"note", "webhook_secret not configured; incoming webhook requests will not be authenticated")
}
globalRouter.register(ch)
ch.MarkHealthy("connected to page " + ch.pageID)
ch.SetRunning(true)
// Background goroutine: evict stale dedup entries to prevent memory growth.
go ch.runDedupCleaner()
slog.Info("pancake channel started",
"page_id", ch.pageID,
"platform", ch.platform,
"name", ch.Name())
return nil
}
// Stop gracefully shuts down the channel.
func (ch *Channel) Stop(_ context.Context) error {
globalRouter.unregister(ch.pageID)
ch.stopFn()
close(ch.stopCh)
ch.SetRunning(false)
ch.MarkStopped("stopped")
slog.Info("pancake channel stopped", "page_id", ch.pageID, "name", ch.Name())
return nil
}
// Send delivers an outbound message via Pancake API.
func (ch *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error {
slog.Debug("pancake: Send called",
"page_id", ch.pageID,
"chat_id", msg.ChatID,
"content_len", len(msg.Content),
"platform", ch.platform,
"channel_name", ch.Name())
conversationID := msg.ChatID
if conversationID == "" {
return fmt.Errorf("pancake: chat_id (conversation_id) is required for outbound message")
}
text := FormatOutbound(msg.Content, ch.platform)
// Handle media attachments.
attachmentIDs, err := ch.handleMediaAttachments(ctx, msg)
if err != nil {
slog.Warn("pancake: media upload failed, sending text only",
"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 {
ch.handleAPIError(err)
return err
}
if text == "" {
return nil
}
}
// Text-only: split into platform-appropriate chunks.
// Store echo fingerprints BEFORE sending so that webhook echoes arriving
// while the HTTP round-trip is in flight are already recognized as self-sent.
parts := splitMessage(text, ch.maxMessageLength())
for _, part := range parts {
ch.rememberOutboundEcho(conversationID, part)
}
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
}
}
return nil
}
// BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default).
func (ch *Channel) BlockReplyEnabled() *bool { return ch.config.BlockReply }
// WebhookHandler returns the shared webhook path and global router as handler.
// Only the first pancake instance mounts the route; others return ("", nil).
func (ch *Channel) WebhookHandler() (string, http.Handler) {
return globalRouter.webhookRoute()
}
// handleAPIError maps Pancake API errors to channel health states.
func (ch *Channel) handleAPIError(err error) {
if err == nil {
return
}
switch {
case isAuthError(err):
ch.MarkFailed("token expired or invalid", err.Error(), channels.ChannelFailureKindAuth, false)
case isRateLimitError(err):
ch.MarkDegraded("rate limited", err.Error(), channels.ChannelFailureKindNetwork, true)
default:
ch.MarkDegraded("api error", err.Error(), channels.ChannelFailureKindUnknown, true)
}
}
// maxMessageLength returns the platform-specific character limit.
func (ch *Channel) maxMessageLength() int {
switch ch.platform {
case "tiktok":
return 500
case "instagram":
return 1000
case "facebook", "zalo":
return 2000
case "whatsapp":
return 4096
case "line":
return 5000
default:
return 2000
}
}
@@ -0,0 +1,396 @@
package pancake
import (
"context"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
)
func TestMessageHandlerSkipsRecentOutboundEchoWithHTMLFormatting(t *testing.T) {
msgBus := bus.New()
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
}
ch.rememberOutboundEcho("conv-1", "Line 1\nLine 2")
ch.handleMessagingEvent(MessagingData{
PageID: "page-123",
ConversationID: "conv-1",
Type: "INBOX",
Platform: "facebook",
Message: MessagingMessage{
ID: "msg-echo-html-1",
SenderID: "user-1",
Content: "<div>Line 1<br key='n_0' />Line 2</div>",
},
})
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected HTML-formatted echo to be dropped")
}
}
func TestWebhookRouterSkipsNonInboxConversationEvents(t *testing.T) {
msgBus := bus.New()
target := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
platform: "facebook",
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": target,
},
}
body := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "page-123_user-1",
"type": "COMMENT",
"from": {
"id": "user-1",
"name": "Customer"
}
},
"message": {
"id": "msg-comment-1",
"message": "hi"
}
}
}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook", strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected COMMENT conversation webhook to be ignored")
}
}
func TestWebhookRouterPrefersMessageSenderOverConversationSender(t *testing.T) {
msgBus := bus.New()
target := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
platform: "facebook",
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": target,
},
}
body := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "conv-1",
"type": "INBOX",
"from": {
"id": "user-initiator",
"name": "Conversation Starter"
}
},
"message": {
"id": "msg-actual-sender-1",
"message": "xin chao",
"from": {
"id": "user-actual",
"name": "Actual Sender",
"page_customer_id": "pc-123"
}
}
}
}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook", strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
msg, ok := msgBus.ConsumeInbound(ctx)
if !ok {
t.Fatal("expected inbound message to be published")
}
if got, want := msg.SenderID, "user-actual"; got != want {
t.Fatalf("sender_id = %q, want %q", got, want)
}
if got, want := msg.Metadata["display_name"], "Actual Sender"; got != want {
t.Fatalf("metadata.display_name = %q, want %q", got, want)
}
}
func TestWebhookRouterSkipsPageAuthoredInboxReply(t *testing.T) {
msgBus := bus.New()
target := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
platform: "facebook",
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": target,
},
}
body := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "conv-1",
"type": "INBOX",
"from": {
"id": "user-1",
"name": "Customer"
}
},
"message": {
"id": "msg-page-reply-1",
"message": "manual page reply",
"from": {
"id": "page-123",
"name": "Page Bot"
}
}
}
}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook", strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected page-authored reply not to be published inbound")
}
}
func TestWebhookRouterSkipsAssignedStaffReply(t *testing.T) {
msgBus := bus.New()
target := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
platform: "facebook",
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": target,
},
}
body := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "conv-1",
"type": "INBOX",
"assignee_ids": ["staff-1"],
"from": {
"id": "user-1",
"name": "Customer"
}
},
"message": {
"id": "msg-staff-reply-1",
"message": "manual staff reply",
"from": {
"id": "staff-1",
"name": "Assigned Staff"
}
}
}
}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook", strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected assigned staff reply not to be published inbound")
}
}
func TestSendThenWebhookEchoDoesNotRepublishInbound(t *testing.T) {
msgBus := bus.New()
api := NewAPIClient("user-token", "page-token", "page-123")
sendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"success":true}`)
}))
defer sendServer.Close()
api.pageV1BaseURL = sendServer.URL
api.httpClient = sendServer.Client()
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
apiClient: api,
pageID: "page-123",
platform: "facebook",
}
if err := ch.Send(context.Background(), bus.OutboundMessage{
Channel: "pancake",
ChatID: "conv-1",
Content: "Line 1\nLine 2",
}); err != nil {
t.Fatalf("Send returned error: %v", err)
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": ch,
},
}
body := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "conv-1",
"type": "INBOX",
"from": {
"id": "user-1",
"name": "Customer"
}
},
"message": {
"id": "msg-echo-roundtrip-1",
"message": "<div>Line 1<br key='n_0' />Line 2</div>"
}
}
}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook", strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", w.Code, http.StatusOK)
}
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected echoed outbound send not to republish inbound work")
}
}
// TestSendRaceConditionEchoArrivesBeforeSendReturns simulates the production race:
// webhook echo arriving while SendMessage HTTP call is still in flight.
// Before the fix, rememberOutboundEcho was called AFTER SendMessage, so an echo
// arriving during the HTTP round-trip would not be recognized.
func TestSendRaceConditionEchoArrivesBeforeSendReturns(t *testing.T) {
msgBus := bus.New()
api := NewAPIClient("user-token", "page-token", "page-123")
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
apiClient: api,
pageID: "page-123",
platform: "facebook",
}
router := &webhookRouter{
instances: map[string]*Channel{
"page-123": ch,
},
}
const outboundContent = "Em đây, alive luôn 😊\nThoát loop thành công rồi nha."
// Pancake API server that delivers the echo webhook BEFORE returning the
// SendMessage response — simulating the real-world race condition.
sendServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// While SendMessage is "in flight", deliver the echo webhook.
echoBody := `{
"page_id": "page-123",
"event_type": "messaging",
"data": {
"conversation": {
"id": "conv-race",
"type": "INBOX",
"from": {"id": "user-1", "name": "Customer"}
},
"message": {
"id": "msg-race-echo",
"message": "<div>Em đây, alive luôn 😊 <br key='n_0' />Thoát loop thành công rồi nha.</div>"
}
}
}`
echoReq := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook",
strings.NewReader(echoBody))
echoW := httptest.NewRecorder()
router.ServeHTTP(echoW, echoReq)
// Now return SendMessage success.
w.Header().Set("Content-Type", "application/json")
_, _ = io.WriteString(w, `{"success":true}`)
}))
defer sendServer.Close()
api.pageV1BaseURL = sendServer.URL
api.httpClient = sendServer.Client()
if err := ch.Send(context.Background(), bus.OutboundMessage{
Channel: "pancake",
ChatID: "conv-race",
Content: outboundContent,
}); err != nil {
t.Fatalf("Send returned error: %v", err)
}
// The echo webhook was delivered during Send. It must NOT have been published.
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("echo arrived during SendMessage HTTP call but was not suppressed — race condition not fixed")
}
}
+484
View File
@@ -0,0 +1,484 @@
package pancake
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"unicode/utf8"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
)
// TestFactory_Valid verifies Factory creates a Channel from valid JSON creds/config.
func TestFactory_Valid(t *testing.T) {
creds, _ := json.Marshal(pancakeCreds{
APIKey: "test-api-key",
PageAccessToken: "test-page-token",
})
cfg, _ := json.Marshal(pancakeInstanceConfig{PageID: "12345"})
ch, err := Factory("pancake-test", creds, cfg, nil, nil)
if err != nil {
t.Fatalf("Factory returned unexpected error: %v", err)
}
if ch == nil {
t.Fatal("Factory returned nil channel")
}
if ch.Name() != "pancake-test" {
t.Errorf("Name() = %q, want %q", ch.Name(), "pancake-test")
}
}
// TestFactory_MissingAPIKey verifies Factory returns error when api_key is empty.
func TestFactory_MissingAPIKey(t *testing.T) {
creds, _ := json.Marshal(pancakeCreds{PageAccessToken: "token"})
cfg, _ := json.Marshal(pancakeInstanceConfig{PageID: "12345"})
_, err := Factory("test", creds, cfg, nil, nil)
if err == nil {
t.Fatal("expected error for missing api_key, got nil")
}
}
// TestFactory_MissingPageAccessToken verifies Factory returns error when page_access_token is empty.
func TestFactory_MissingPageAccessToken(t *testing.T) {
creds, _ := json.Marshal(pancakeCreds{APIKey: "key"})
cfg, _ := json.Marshal(pancakeInstanceConfig{PageID: "12345"})
_, err := Factory("test", creds, cfg, nil, nil)
if err == nil {
t.Fatal("expected error for missing page_access_token, got nil")
}
}
// TestFactory_MissingPageID verifies Factory returns error when page_id is empty.
func TestFactory_MissingPageID(t *testing.T) {
creds, _ := json.Marshal(pancakeCreds{
APIKey: "key",
PageAccessToken: "token",
})
cfg, _ := json.Marshal(pancakeInstanceConfig{}) // no page_id
_, err := Factory("test", creds, cfg, nil, nil)
if err == nil {
t.Fatal("expected error for missing page_id, got nil")
}
}
// TestFormatOutbound verifies platform-aware formatting for each platform.
func TestFormatOutbound(t *testing.T) {
input := "**Hello** _world_ `code` ## Header [link](http://example.com)"
cases := []struct {
platform string
wantNot string // substring that should NOT appear in output
}{
{"facebook", "**"},
{"zalo", "**"},
{"instagram", "_"},
{"tiktok", "##"},
{"whatsapp", "**"},
{"line", "##"},
{"unknown", "`"},
}
for _, tc := range cases {
t.Run(tc.platform, func(t *testing.T) {
out := FormatOutbound(input, tc.platform)
if out == "" {
t.Error("FormatOutbound returned empty string")
}
_ = out // formatting verified visually; we just check no panic + non-empty
})
}
}
// TestSplitMessage verifies message splitting at platform character limits.
func TestSplitMessage(t *testing.T) {
t.Run("short message not split", func(t *testing.T) {
parts := splitMessage("hello", 100)
if len(parts) != 1 || parts[0] != "hello" {
t.Errorf("unexpected parts: %v", parts)
}
})
t.Run("exact limit not split", func(t *testing.T) {
msg := string(make([]byte, 100))
parts := splitMessage(msg, 100)
if len(parts) != 1 {
t.Errorf("expected 1 part, got %d", len(parts))
}
})
t.Run("over limit is split", func(t *testing.T) {
msg := string(make([]byte, 250))
parts := splitMessage(msg, 100)
if len(parts) != 3 {
t.Errorf("expected 3 parts, got %d", len(parts))
}
})
t.Run("zero limit returns whole string", func(t *testing.T) {
parts := splitMessage("hello", 0)
if len(parts) != 1 {
t.Errorf("expected 1 part with zero limit, got %d", len(parts))
}
})
}
// TestIsDup verifies dedup returns false first, true on repeat.
func TestIsDup(t *testing.T) {
ch := &Channel{}
if ch.isDup("key-1") {
t.Error("isDup: first call should return false")
}
if !ch.isDup("key-1") {
t.Error("isDup: second call should return true")
}
if ch.isDup("key-2") {
t.Error("isDup: different key should return false")
}
}
// TestWebhookRouterReturns200 verifies the global router always returns HTTP 200.
func TestWebhookRouterReturns200(t *testing.T) {
// Use a fresh local router to avoid interfering with the package-level globalRouter.
router := &webhookRouter{instances: make(map[string]*Channel)}
t.Run("POST event returns 200", func(t *testing.T) {
body := `{"data":{"conversation":{"id":"123_456","type":"INBOX","from":{"id":"456"}},"message":{"id":"m1"}}}`
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook",
strings.NewReader(body))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d", w.Code)
}
})
t.Run("GET returns 200 (not 405)", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/channels/pancake/webhook", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d", w.Code)
}
})
t.Run("malformed JSON returns 200", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/channels/pancake/webhook",
strings.NewReader("not-json"))
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected 200, got %d", w.Code)
}
})
}
// TestMessageHandlerSkipsSelfReply verifies the page's own messages are not published.
// The dedup entry is stored (dedup runs first), but HandleMessage is never called.
// If the self-reply guard were absent, ch.bus (nil) would panic — making no-panic the assertion.
func TestMessageHandlerSkipsSelfReply(t *testing.T) {
const pageID = "page-123"
ch := &Channel{pageID: pageID}
data := MessagingData{
PageID: pageID,
ConversationID: "conv-1",
Type: "INBOX",
Platform: "facebook",
Message: MessagingMessage{
ID: "msg-self-1",
SenderID: pageID, // same as page → must be skipped before HandleMessage
SenderName: "Page Bot",
Content: "Hello",
},
}
// Must not panic. If self-reply guard is missing, nil bus dereference panics here.
ch.handleMessagingEvent(data)
// Dedup entry is stored (dedup check runs before self-reply check).
_, stored := ch.dedup.Load("msg:msg-self-1")
if !stored {
t.Error("dedup entry should have been stored (dedup runs before self-reply guard)")
}
}
func TestMessageHandlerPublishesMessageIDMetadata(t *testing.T) {
msgBus := bus.New()
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
}
ch.handleMessagingEvent(MessagingData{
PageID: "page-123",
ConversationID: "conv-1",
Type: "INBOX",
Platform: "facebook",
Message: MessagingMessage{
ID: "msg-123",
SenderID: "user-1",
Content: "hello",
},
})
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
msg, ok := msgBus.ConsumeInbound(ctx)
if !ok {
t.Fatal("expected inbound message to be published")
}
if got, want := msg.Metadata["message_id"], "msg:msg-123"; got != want {
t.Fatalf("metadata.message_id = %q, want %q", got, want)
}
}
func TestMessageHandlerSkipsRecentOutboundEcho(t *testing.T) {
msgBus := bus.New()
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
}
ch.rememberOutboundEcho("conv-1", "hello from bot")
ch.handleMessagingEvent(MessagingData{
PageID: "page-123",
ConversationID: "conv-1",
Type: "INBOX",
Platform: "facebook",
Message: MessagingMessage{
ID: "msg-echo-1",
SenderID: "user-1",
Content: "hello from bot",
},
})
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, ok := msgBus.ConsumeInbound(ctx); ok {
t.Fatal("expected echoed outbound message to be dropped")
}
}
func TestBlockReplyEnabledUsesChannelOverride(t *testing.T) {
enabled := true
ch := &Channel{
config: pancakeInstanceConfig{
BlockReply: &enabled,
},
}
got := ch.BlockReplyEnabled()
if got == nil || !*got {
t.Fatalf("BlockReplyEnabled() = %v, want true", got)
}
}
type captureTransport struct {
req *http.Request
body []byte
resp *http.Response
}
func (t *captureTransport) RoundTrip(req *http.Request) (*http.Response, error) {
t.req = req.Clone(req.Context())
if req.Body != nil {
body, err := io.ReadAll(req.Body)
if err != nil {
return nil, err
}
t.body = body
req.Body = io.NopCloser(bytes.NewReader(body))
}
if t.resp != nil {
return t.resp, nil
}
return &http.Response{
StatusCode: http.StatusOK,
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(`{"success":true}`)),
Request: req,
}, nil
}
func TestAPIClientSendMessageMatchesOfficialContract(t *testing.T) {
transport := &captureTransport{}
client := NewAPIClient("user-token", "page-token", "page-123")
client.httpClient = &http.Client{Transport: transport}
if err := client.SendMessage(context.Background(), "conv-456", "xin chao"); err != nil {
t.Fatalf("SendMessage returned error: %v", err)
}
if transport.req == nil {
t.Fatal("expected outbound request to be captured")
}
if got, want := transport.req.URL.Path, "/api/public_api/v1/pages/page-123/conversations/conv-456/messages"; got != want {
t.Fatalf("request path = %q, want %q", got, want)
}
if got := transport.req.URL.Query().Get("page_access_token"); got != "page-token" {
t.Fatalf("page_access_token query = %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_inbox"; got != want {
t.Fatalf("payload.action = %#v, want %#v", got, want)
}
if got, want := payload["message"], "xin chao"; got != want {
t.Fatalf("payload.message = %#v, want %#v", got, want)
}
if _, exists := payload["content"]; exists {
t.Fatalf("payload must not contain legacy content field: %s", string(transport.body))
}
if _, exists := payload["attachment_id"]; exists {
t.Fatalf("payload must not contain attachment_id field: %s", string(transport.body))
}
}
func TestAPIClientSendMessageReturnsBodyLevelError(t *testing.T) {
transport := &captureTransport{
resp: &http.Response{
StatusCode: http.StatusOK,
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(`{"success":false,"message":"conversation blocked"}`)),
},
}
client := NewAPIClient("user-token", "page-token", "page-123")
client.httpClient = &http.Client{Transport: transport}
err := client.SendMessage(context.Background(), "conv-456", "xin chao")
if err == nil {
t.Fatal("expected SendMessage to return body-level error")
}
if !strings.Contains(err.Error(), "conversation blocked") {
t.Fatalf("SendMessage error = %v, want body-level message", err)
}
}
// TestTruncateForTikTok_MultiByteCharacters verifies rune-safe truncation for
// Vietnamese diacritics and emoji (multi-byte UTF-8 sequences).
func TestTruncateForTikTok_MultiByteCharacters(t *testing.T) {
// Vietnamese text with diacritics (multi-byte UTF-8)
input := strings.Repeat("Xin chào ", 100) // ~900 bytes, <500 runes
result := truncateForTikTok(input)
if !utf8.ValidString(result) {
t.Fatal("truncateForTikTok produced invalid UTF-8")
}
// Emoji string exceeding 500 runes
emoji := strings.Repeat("😊", 600)
result = truncateForTikTok(emoji)
runes := []rune(result)
if len(runes) > 500 {
t.Errorf("expected <=500 runes, got %d", len(runes))
}
if !utf8.ValidString(result) {
t.Fatal("emoji truncation produced invalid UTF-8")
}
}
// TestMessageHandlerEmptyMessageID verifies that two messages with empty IDs
// from different conversations are both published (not deduped against each other).
func TestMessageHandlerEmptyMessageID(t *testing.T) {
msgBus := bus.New()
ch := &Channel{
BaseChannel: channels.NewBaseChannel(channels.TypePancake, msgBus, nil),
pageID: "page-123",
}
// First message with empty ID — should be published
ch.handleMessagingEvent(MessagingData{
PageID: "page-123", ConversationID: "conv-1",
Type: "INBOX", Platform: "facebook",
Message: MessagingMessage{ID: "", SenderID: "user-1", Content: "hello"},
})
// Second message with empty ID, different conversation — should ALSO be published
ch.handleMessagingEvent(MessagingData{
PageID: "page-123", ConversationID: "conv-2",
Type: "INBOX", Platform: "facebook",
Message: MessagingMessage{ID: "", SenderID: "user-2", Content: "world"},
})
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")
}
}
func TestAPIClientUploadMediaMatchesOfficialContract(t *testing.T) {
transport := &captureTransport{
resp: &http.Response{
StatusCode: http.StatusOK,
Header: make(http.Header),
Body: io.NopCloser(strings.NewReader(`{"id":"upload-123","success":true}`)),
},
}
client := NewAPIClient("user-token", "page-token", "page-123")
client.httpClient = &http.Client{Transport: transport}
id, err := client.UploadMedia(context.Background(), "photo.jpg", strings.NewReader("file-bytes"), "image/jpeg")
if err != nil {
t.Fatalf("UploadMedia returned error: %v", err)
}
if id != "upload-123" {
t.Fatalf("UploadMedia id = %q, want %q", id, "upload-123")
}
if transport.req == nil {
t.Fatal("expected upload request to be captured")
}
if got, want := transport.req.URL.Path, "/api/public_api/v1/pages/page-123/upload_contents"; got != want {
t.Fatalf("upload path = %q, want %q", got, want)
}
if got := transport.req.URL.Query().Get("page_access_token"); got != "page-token" {
t.Fatalf("upload page_access_token query = %q, want %q", got, "page-token")
}
if !strings.HasPrefix(transport.req.Header.Get("Content-Type"), "multipart/form-data; boundary=") {
t.Fatalf("upload Content-Type = %q, want multipart/form-data", transport.req.Header.Get("Content-Type"))
}
}
// TestIsAuthError_WrappedError verifies errors.As works with wrapped apiError.
func TestIsAuthError_WrappedError(t *testing.T) {
inner := &apiError{Code: 401, Message: "unauthorized"}
wrapped := fmt.Errorf("send failed: %w", inner)
if !isAuthError(wrapped) {
t.Error("isAuthError should detect wrapped 401 apiError via errors.As")
}
if isAuthError(fmt.Errorf("random error")) {
t.Error("isAuthError should return false for non-apiError")
}
}
// TestIsRateLimitError_WrappedError verifies errors.As works with wrapped rate limit.
func TestIsRateLimitError_WrappedError(t *testing.T) {
inner := &apiError{Code: 429, Message: "too many requests"}
wrapped := fmt.Errorf("send failed: %w", inner)
if !isRateLimitError(wrapped) {
t.Error("isRateLimitError should detect wrapped 429 apiError via errors.As")
}
}
+130
View File
@@ -0,0 +1,130 @@
// Package pancake implements the Pancake (pages.fm) channel for GoClaw.
// Pancake acts as a unified proxy for Facebook, Zalo OA, Instagram, TikTok, WhatsApp, Line.
// A single Pancake API key gives access to all connected platforms — no per-platform OAuth needed.
package pancake
import "encoding/json"
// pancakeCreds holds encrypted credentials stored in channel_instances.credentials.
type pancakeCreds struct {
APIKey string `json:"api_key"` // User-level Pancake API key
PageAccessToken string `json:"page_access_token"` // Page-level token for all page APIs
WebhookSecret string `json:"webhook_secret,omitempty"` // Optional HMAC-SHA256 verification
}
// pancakeInstanceConfig holds non-secret config from channel_instances.config JSONB.
type pancakeInstanceConfig struct {
PageID string `json:"page_id"`
Platform string `json:"platform,omitempty"` // auto-detected at Start(): facebook/zalo/instagram/tiktok/whatsapp/line
Features struct {
InboxReply bool `json:"inbox_reply"`
CommentReply bool `json:"comment_reply"`
} `json:"features"`
AllowFrom []string `json:"allow_from,omitempty"`
BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit)
}
// --- Webhook payload types ---
// These types match the actual Pancake (pages.fm) webhook delivery format.
// Top-level envelope has optional "event_type" + nested "data" containing
// "conversation" and "message" objects.
// WebhookEvent is the top-level Pancake webhook delivery envelope.
type WebhookEvent struct {
EventType string `json:"event_type,omitempty"` // "messaging", may be empty
PageID string `json:"page_id,omitempty"` // top-level page_id (some formats)
Data json.RawMessage `json:"data"`
}
// WebhookData is the "data" envelope inside a Pancake webhook event.
type WebhookData struct {
Conversation WebhookConversation `json:"conversation"`
Message WebhookMessage `json:"message"`
PageID string `json:"page_id,omitempty"` // page_id may appear here or top-level
}
// 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"
AssigneeIDs []string `json:"assignee_ids,omitempty"` // Pancake staff IDs assigned to this conversation
From WebhookSender `json:"from"`
Snippet string `json:"snippet,omitempty"`
}
// WebhookSender identifies the message sender.
type WebhookSender struct {
ID string `json:"id"`
Name string `json:"name"`
Email string `json:"email,omitempty"`
PageCustomerID string `json:"page_customer_id,omitempty"`
}
// WebhookMessage holds the message payload from a Pancake webhook.
type WebhookMessage struct {
ID string `json:"id"`
Message string `json:"message,omitempty"` // primary text content
OriginalMessage string `json:"original_message,omitempty"` // unformatted fallback
Content string `json:"content,omitempty"` // legacy field
From *WebhookSender `json:"from,omitempty"` // actual sender of this message
Attachments []MessageAttachment `json:"attachments,omitempty"`
CreatedAt json.Number `json:"created_at,omitempty"`
}
// MessageAttachment represents a media attachment in a Pancake webhook message.
type MessageAttachment struct {
Type string `json:"type"` // "image", "video", "file"
URL string `json:"url"`
}
// MessagingData is the normalized internal representation used after parsing.
type MessagingData struct {
PageID string
ConversationID string
Type string // "INBOX" or "COMMENT"
Platform string // "facebook", "zalo", "instagram", "tiktok", "whatsapp", "line"
AssigneeIDs []string
Message MessagingMessage
}
// MessagingMessage is the normalized message used by the handler.
type MessagingMessage struct {
ID string
Content string
SenderID string
SenderName string
Attachments []MessageAttachment
}
// --- API response types ---
// PageInfo holds page metadata from GET /pages response.
type PageInfo struct {
ID string `json:"id"`
Name string `json:"name"`
Platform string `json:"platform"` // facebook/zalo/instagram/tiktok/whatsapp/line
Avatar string `json:"avatar,omitempty"`
}
// SendMessageRequest is the POST body for sending a message via Pancake API.
type SendMessageRequest struct {
Action string `json:"action"`
Message string `json:"message,omitempty"`
ContentIDs []string `json:"content_ids,omitempty"`
}
// UploadResponse is returned by POST /pages/{id}/upload_contents.
type UploadResponse struct {
ID string `json:"id"`
URL string `json:"url,omitempty"`
}
// apiError wraps a Pancake API error response.
type apiError struct {
Code int `json:"code"`
Message string `json:"message"`
}
func (e *apiError) Error() string {
return e.Message
}
@@ -0,0 +1,242 @@
package pancake
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"log/slog"
"net/http"
"strings"
"sync"
)
const (
webhookPath = "/channels/pancake/webhook"
maxBodyBytes = 1 << 20 // 1 MB — prevent abuse
)
// verifyHMAC verifies a Pancake HMAC-SHA256 signature.
// Expected header format: "sha256=<hex-digest>"
func verifyHMAC(body []byte, secret, signature string) bool {
const prefix = "sha256="
if len(signature) <= len(prefix) {
return false
}
got, err := hex.DecodeString(signature[len(prefix):])
if err != nil {
return false
}
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(body)
expected := mac.Sum(nil)
return hmac.Equal(got, expected)
}
// --- Global webhook router for multi-page support ---
// webhookRouter routes incoming Pancake webhook events to the correct channel instance by page_id.
// A single HTTP handler is shared across all pancake channel instances.
type webhookRouter struct {
mu sync.RWMutex
instances map[string]*Channel // pageID → channel
routeHandled bool // true after first webhookRoute() call
}
var globalRouter = &webhookRouter{
instances: make(map[string]*Channel),
}
func (r *webhookRouter) register(ch *Channel) {
r.mu.Lock()
defer r.mu.Unlock()
r.instances[ch.pageID] = ch
}
func (r *webhookRouter) unregister(pageID string) {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.instances, pageID)
}
// webhookRoute returns the path+handler on first call; ("", nil) for subsequent calls.
// The HTTP mux retains the route once registered — routeHandled prevents duplicate mounts.
func (r *webhookRouter) webhookRoute() (string, http.Handler) {
r.mu.Lock()
defer r.mu.Unlock()
if !r.routeHandled {
r.routeHandled = true
return webhookPath, r
}
return "", nil
}
// ServeHTTP is the shared handler for all Pancake page webhooks.
// Always returns HTTP 200 — Pancake suspends webhooks if >80% errors in a 30-min window.
func (r *webhookRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
slog.Info("pancake: webhook request received",
"method", req.Method,
"remote_addr", req.RemoteAddr,
"content_length", req.ContentLength)
if req.Method != http.MethodPost {
w.WriteHeader(http.StatusOK)
return
}
lr := io.LimitReader(req.Body, maxBodyBytes+1)
body, err := io.ReadAll(lr)
if err != nil {
slog.Warn("pancake: router read body error", "err", err)
w.WriteHeader(http.StatusOK)
return
}
slog.Debug("pancake: webhook body received",
"body_len", len(body),
"body_preview", truncateBody(body, 1000))
// Parse top-level envelope.
var event WebhookEvent
if err := json.Unmarshal(body, &event); err != nil {
slog.Warn("pancake: router parse event error", "err", err, "body_preview", truncateBody(body, 300))
w.WriteHeader(http.StatusOK)
return
}
// Parse the nested "data" object containing conversation + message.
var data WebhookData
if err := json.Unmarshal(event.Data, &data); err != nil {
slog.Warn("pancake: router parse data error", "err", err)
w.WriteHeader(http.StatusOK)
return
}
// Resolve page_id: try top-level first, then data-level, then extract from conversation ID.
pageID := event.PageID
if pageID == "" {
pageID = data.PageID
}
if pageID == "" {
// Conversation ID format: "pageID_senderID" — extract page portion.
if idx := strings.Index(data.Conversation.ID, "_"); idx > 0 {
pageID = data.Conversation.ID[:idx]
}
}
// Resolve conversation type — only process INBOX messages.
convType := strings.ToUpper(data.Conversation.Type)
if event.EventType != "" && !strings.EqualFold(event.EventType, "messaging") {
slog.Debug("pancake: skipping non-messaging webhook event",
"event_type", event.EventType,
"page_id", pageID)
w.WriteHeader(http.StatusOK)
return
}
slog.Debug("pancake: webhook event parsed",
"event_type", event.EventType,
"page_id", pageID,
"conv_id", data.Conversation.ID,
"conv_type", convType,
"sender_id", data.Conversation.From.ID,
"sender_name", data.Conversation.From.Name,
"msg_id", data.Message.ID)
if pageID == "" {
slog.Warn("pancake: could not determine page_id from webhook payload")
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()
if target == nil {
// Log all registered page IDs for debugging.
r.mu.RLock()
var registered []string
for pid := range r.instances {
registered = append(registered, pid)
}
r.mu.RUnlock()
slog.Warn("pancake: no channel instance for page_id",
"page_id", pageID,
"registered_pages", registered)
w.WriteHeader(http.StatusOK)
return
}
// HMAC signature verification — skip if webhook_secret not configured.
if target.webhookSecret != "" {
sig := req.Header.Get("X-Pancake-Signature")
if !verifyHMAC(body, target.webhookSecret, sig) {
slog.Warn("security.pancake_webhook_signature_mismatch",
"page_id", pageID,
"remote_addr", req.RemoteAddr)
w.WriteHeader(http.StatusOK)
return
}
}
// Build normalized MessagingData from actual Pancake payload.
msgContent := data.Message.Message
if msgContent == "" {
msgContent = data.Message.OriginalMessage
}
if msgContent == "" {
msgContent = data.Message.Content
}
if msgContent == "" {
msgContent = data.Conversation.Snippet
}
senderID := data.Conversation.From.ID
senderName := data.Conversation.From.Name
if data.Message.From != nil {
if data.Message.From.ID != "" {
senderID = data.Message.From.ID
}
if data.Message.From.Name != "" {
senderName = data.Message.From.Name
}
}
normalized := MessagingData{
PageID: pageID,
ConversationID: data.Conversation.ID,
Type: convType,
Platform: target.platform,
AssigneeIDs: append([]string(nil), data.Conversation.AssigneeIDs...),
Message: MessagingMessage{
ID: data.Message.ID,
Content: msgContent,
SenderID: senderID,
SenderName: senderName,
Attachments: data.Message.Attachments,
},
}
target.handleMessagingEvent(normalized)
w.WriteHeader(http.StatusOK)
}
// truncateBody returns a string preview of body, truncated to maxLen bytes.
func truncateBody(body []byte, maxLen int) string {
if len(body) <= maxLen {
return string(body)
}
return string(body[:maxLen]) + "..."
}
@@ -0,0 +1,17 @@
package channels
import "strings"
// SanitizeDisplayName strips newlines and markdown heading markers from a
// channel-provided display name to prevent injection into markdown templates
// (e.g. USER.md). Caps length at 100 runes.
func SanitizeDisplayName(name string) string {
name = strings.ReplaceAll(name, "\n", " ")
name = strings.ReplaceAll(name, "\r", " ")
name = strings.ReplaceAll(name, "#", "")
name = strings.TrimSpace(name)
if len([]rune(name)) > 100 {
name = string([]rune(name)[:100])
}
return name
}
+1 -1
View File
@@ -251,7 +251,7 @@ func (c *Channel) handleMessage(ev *slackevents.MessageEvent) {
"message_id": ev.TimeStamp,
"user_id": senderID,
"username": displayName,
"display_name": displayName,
"display_name": channels.SanitizeDisplayName(displayName),
"channel_id": channelID,
"is_dm": fmt.Sprintf("%t", isDM),
"local_key": localKey,
+2 -2
View File
@@ -70,7 +70,7 @@ func (c *Channel) handleDM(msg protocol.UserMessage) {
metadata := map[string]string{
"message_id": msg.Data.MsgID,
"platform": channels.TypeZaloPersonal,
"display_name": senderName,
"display_name": channels.SanitizeDisplayName(senderName),
}
c.HandleMessage(senderID, threadID, content, media, metadata, "direct")
}
@@ -151,7 +151,7 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) {
"message_id": msg.Data.MsgID,
"platform": channels.TypeZaloPersonal,
"group_id": threadID,
"display_name": senderName,
"display_name": channels.SanitizeDisplayName(senderName),
}
c.HandleMessage(senderID, threadID, finalContent, allMedia, metadata, "group")
+1 -1
View File
@@ -540,7 +540,7 @@ func (h *ChannelInstancesHandler) handleResolveContacts(w http.ResponseWriter, r
// isValidChannelType checks if the channel type is supported.
func isValidChannelType(ct string) bool {
switch ct {
case "telegram", "discord", "slack", "whatsapp", "zalo_oa", "zalo_personal", "feishu":
case "telegram", "discord", "slack", "whatsapp", "zalo_oa", "zalo_personal", "feishu", "facebook", "pancake":
return true
}
return false
+10 -9
View File
@@ -84,10 +84,10 @@ export class HttpClient {
if (!res.ok) {
const err = await res.json().catch(() => ({ error: res.statusText }));
throw new ApiError(
err.code ?? "HTTP_ERROR",
err.error ?? err.message ?? res.statusText,
);
const nested = typeof err.error === "object" && err.error !== null ? err.error : null;
const code = nested?.code ?? err.code ?? "HTTP_ERROR";
const message = nested?.message ?? (typeof err.error === "string" ? err.error : null) ?? err.message ?? res.statusText;
throw new ApiError(code, message);
}
return res.json() as Promise<T>;
@@ -140,13 +140,14 @@ export class HttpClient {
if (!res.ok) {
const err = await res.json().catch(() => ({ error: res.statusText }));
if (res.status === 401 || err.code === "TENANT_ACCESS_REVOKED") {
// Backend wraps errors as { "error": { "code": "...", "message": "..." } }
const nested = typeof err.error === "object" && err.error !== null ? err.error : null;
const code = nested?.code ?? err.code ?? "HTTP_ERROR";
const message = nested?.message ?? (typeof err.error === "string" ? err.error : null) ?? err.message ?? res.statusText;
if (res.status === 401 || code === "TENANT_ACCESS_REVOKED") {
this.onAuthFailure?.();
}
throw new ApiError(
err.code ?? "HTTP_ERROR",
err.error ?? err.message ?? res.statusText,
);
throw new ApiError(code, message);
}
return res.json() as Promise<T>;
+5 -3
View File
@@ -1,9 +1,11 @@
export const CHANNEL_TYPES = [
{ value: "telegram", label: "Telegram" },
{ value: "discord", label: "Discord" },
{ value: "slack", label: "Slack" },
{ value: "facebook", label: "Facebook" },
{ value: "feishu", label: "Feishu / Lark" },
{ value: "pancake", label: "Pancake (pages.fm)" },
{ value: "slack", label: "Slack" },
{ value: "telegram", label: "Telegram" },
{ value: "whatsapp", label: "WhatsApp" },
{ value: "zalo_oa", label: "Zalo OA" },
{ value: "zalo_personal", label: "Zalo Personal" },
{ value: "whatsapp", label: "WhatsApp" },
] as const;
@@ -69,6 +69,16 @@ export const credentialsSchema: Record<string, FieldDef[]> = {
],
zalo_personal: [],
whatsapp: [],
facebook: [
{ key: "page_access_token", label: "Page Access Token", type: "password", required: true, help: "From Facebook Developer Console → Your App → Messenger → Page Access Token" },
{ key: "app_secret", label: "App Secret", type: "password", required: true, help: "From Facebook Developer Console → Your App → Settings → Basic" },
{ key: "verify_token", label: "Webhook Verify Token", type: "password", required: true, help: "A secret string you choose, used to verify the webhook URL" },
],
pancake: [
{ key: "api_key", label: "API Key", type: "password", required: true, help: "Pancake user-level API key from pages.fm account settings" },
{ key: "page_access_token", label: "Page Access Token", type: "password", required: true, help: "Page-level token from Pancake dashboard → Page Settings" },
{ key: "webhook_secret", label: "Webhook Secret (Optional)", type: "password", help: "HMAC-SHA256 secret for webhook signature verification. Leave empty to skip verification." },
],
};
// --- Config schemas ---
@@ -153,6 +163,27 @@ export const configSchema: Record<string, FieldDef[]> = {
{ key: "allow_from", label: "Allowed Users", type: "tags", help: "WhatsApp user IDs" },
{ key: "block_reply", label: "Block Reply", type: "select", options: blockReplyOptions, defaultValue: "inherit", help: "Deliver intermediate text during tool iterations" },
],
facebook: [
{ key: "page_id", label: "Page ID", type: "text", required: true, help: "Facebook Page numeric ID" },
{ key: "features.comment_reply", label: "Comment Auto-Reply", type: "boolean", defaultValue: false },
{ key: "features.messenger_auto_reply", label: "Messenger Auto-Reply", type: "boolean", defaultValue: false },
{ key: "features.first_inbox", label: "First Inbox DM", type: "boolean", defaultValue: false, help: "Send a one-time DM to commenters after their first comment reply" },
{ key: "comment_reply_options.include_post_context", label: "Include Post Context", type: "boolean", defaultValue: false, help: "Fetch original post content for comment context" },
{ key: "comment_reply_options.max_thread_depth", label: "Max Thread Depth", type: "number", defaultValue: 10 },
{ key: "messenger_options.session_timeout", label: "Messenger Session Timeout", type: "text", placeholder: "e.g. 30m" },
{ key: "post_context_cache_ttl", label: "Post Cache TTL", type: "text", placeholder: "e.g. 15m" },
{ key: "first_inbox_message", label: "First Inbox DM Text", type: "textarea", help: "Custom DM sent to first-time commenters. Defaults to Vietnamese if empty." },
{ key: "allow_from", label: "Allowed Users", type: "tags", help: "Facebook user IDs" },
{ key: "block_reply", label: "Block Reply", type: "select", options: blockReplyOptions, defaultValue: "inherit" },
],
pancake: [
{ key: "page_id", label: "Page ID", type: "text", required: true, help: "Pancake page ID (numeric, from Pancake dashboard)" },
{ key: "platform", label: "Platform (auto-detected)", type: "text", placeholder: "Leave empty — resolved at startup", help: "facebook / zalo / instagram / tiktok / whatsapp / line. Auto-detected if empty." },
{ key: "features.inbox_reply", label: "Inbox Auto-Reply", type: "boolean", defaultValue: true },
{ key: "features.comment_reply", label: "Comment Reply", type: "boolean", defaultValue: false },
{ key: "allow_from", label: "Allowed Users", type: "tags", help: "Sender IDs to whitelist. Empty = accept all." },
{ key: "block_reply", label: "Block Reply", type: "select", options: blockReplyOptions, defaultValue: "inherit" },
],
};
// --- Group override schema (Telegram per-group/topic overrides) ---