From 137a986d4fc58bc42278dabd56ac2ac9f8705331 Mon Sep 17 00:00:00 2001 From: Duc Nguyen Date: Mon, 9 Mar 2026 07:02:37 +0700 Subject: [PATCH] feat(channels): add Slack channel (#83) * feat(channels): add Slack channel via Socket Mode (#37) Implement Slack integration using Socket Mode (xapp-/xoxb- tokens): - Event-driven messaging via app_mention + message events - Policy checks: open, pairing, allowlist, disabled (DM + group) - Thread participation with configurable TTL - Markdown-to-mrkdwn formatting pipeline - Streaming support (edit-in-place + native ChatStreamer) - SSRF-protected file downloads - Debounce, dedup, reactions, group history context - 170 unit tests (format, helpers, stream, SSRF) Fix BaseChannel.HandleMessage allowlist to also check chatID, enabling group allowlist with channel IDs across all channels. Closes #37 * feat(slack): add file/media support and edit-to-mention handling - Wire inbound file download into handleMessage (images, audio, documents) - Add media.go with resolveMedia, classifyMime, buildMediaTags - Extract shared ExtractDocumentContent to channels/media_utils.go (DRY with Telegram) - Support file_share and message_changed subtypes - Handle edit-to-mention: respond when user edits old message to add @bot - Add MediaMaxBytes config field (default 20MB) - Fix debounce media accumulation (was silently dropping files) - Add 60s HTTP client timeout on file downloads - Refactor downloadFile signature for slack.File compatibility --- Makefile | 4 +- cmd/gateway.go | 12 + docs/05-channels-messaging.md | 95 ++- go.mod | 1 + go.sum | 2 + internal/channels/media_utils.go | 76 ++ internal/channels/slack/channel.go | 521 ++++++++++++++ internal/channels/slack/channel_test.go | 203 ++++++ internal/channels/slack/factory.go | 90 +++ internal/channels/slack/format.go | 198 ++++++ internal/channels/slack/format_test.go | 536 ++++++++++++++ internal/channels/slack/handlers.go | 667 ++++++++++++++++++ internal/channels/slack/handlers_test.go | 118 ++++ internal/channels/slack/media.go | 99 +++ internal/channels/slack/reactions.go | 99 +++ internal/channels/slack/stream.go | 133 ++++ internal/channels/slack/stream_test.go | 121 ++++ internal/config/config_channels.go | 20 +- internal/config/config_load.go | 6 + internal/gateway/methods/channel_instances.go | 2 +- internal/http/channel_instances.go | 2 +- ui/web/src/constants/channels.ts | 1 + ui/web/src/pages/channels/channel-schemas.ts | 19 + .../pages/channels/channels-status-view.tsx | 1 + 24 files changed, 2995 insertions(+), 31 deletions(-) create mode 100644 internal/channels/media_utils.go create mode 100644 internal/channels/slack/channel.go create mode 100644 internal/channels/slack/channel_test.go create mode 100644 internal/channels/slack/factory.go create mode 100644 internal/channels/slack/format.go create mode 100644 internal/channels/slack/format_test.go create mode 100644 internal/channels/slack/handlers.go create mode 100644 internal/channels/slack/handlers_test.go create mode 100644 internal/channels/slack/media.go create mode 100644 internal/channels/slack/reactions.go create mode 100644 internal/channels/slack/stream.go create mode 100644 internal/channels/slack/stream_test.go diff --git a/Makefile b/Makefile index 2fb73a7f..ec0c602a 100644 --- a/Makefile +++ b/Makefile @@ -16,10 +16,12 @@ clean: version: @echo $(VERSION) -COMPOSE = docker compose -f docker-compose.yml -f docker-compose.managed.yml -f docker-compose.selfservice.yml +COMPOSE = docker compose -f docker-compose.yml -f docker-compose.postgres.yml -f docker-compose.selfservice.yml +UPGRADE = docker compose -f docker-compose.yml -f docker-compose.postgres.yml -f docker-compose.upgrade.yml up: $(COMPOSE) up -d --build + $(UPGRADE) run --rm upgrade down: $(COMPOSE) down diff --git a/cmd/gateway.go b/cmd/gateway.go index 7b57e93f..42e6841a 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -17,6 +17,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/channels/discord" "github.com/nextlevelbuilder/goclaw/internal/channels/feishu" + slackchannel "github.com/nextlevelbuilder/goclaw/internal/channels/slack" "github.com/nextlevelbuilder/goclaw/internal/channels/telegram" "github.com/nextlevelbuilder/goclaw/internal/channels/whatsapp" "github.com/nextlevelbuilder/goclaw/internal/channels/zalo" @@ -700,6 +701,7 @@ func runGateway() { instanceLoader.RegisterFactory("zalo_oa", zalo.Factory) instanceLoader.RegisterFactory("zalo_personal", zalopersonal.Factory) instanceLoader.RegisterFactory("whatsapp", whatsapp.Factory) + instanceLoader.RegisterFactory("slack", slackchannel.Factory) if err := instanceLoader.LoadAll(context.Background()); err != nil { slog.Error("failed to load channel instances from DB", "error", err) } @@ -756,6 +758,16 @@ func runGateway() { } } + if cfg.Channels.Slack.Enabled && cfg.Channels.Slack.BotToken != "" && cfg.Channels.Slack.AppToken != "" && instanceLoader == nil { + sl, err := slackchannel.New(cfg.Channels.Slack, msgBus, nil) + if err != nil { + slog.Error("failed to initialize slack channel", "error", err) + } else { + channelMgr.RegisterChannel("slack", sl) + slog.Info("slack channel enabled (config)") + } + } + if cfg.Channels.Feishu.Enabled && cfg.Channels.Feishu.AppID != "" && instanceLoader == nil { f, err := feishu.New(cfg.Channels.Feishu, msgBus, pgStores.Pairing) if err != nil { diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index 3166d756..d4981f81 100644 --- a/docs/05-channels-messaging.md +++ b/docs/05-channels-messaging.md @@ -11,6 +11,7 @@ flowchart LR subgraph Platforms TG["Telegram"] DC["Discord"] + SL["Slack"] FS["Feishu/Lark"] ZL["Zalo OA"] ZLP["Zalo Personal"] @@ -34,6 +35,7 @@ flowchart LR TG --> CH DC --> CH + SL --> CH FS --> CH ZL --> CH ZLP --> CH @@ -46,6 +48,7 @@ flowchart LR DISPATCH --> SEND SEND --> TG SEND --> DC + SEND --> SL SEND --> FS SEND --> ZL SEND --> ZLP @@ -154,22 +157,22 @@ flowchart TD ## 4. Channel Comparison -| Feature | Telegram | Feishu/Lark | Discord | WhatsApp | Zalo OA | Zalo Personal | -|---------|----------|-------------|---------|----------|---------|---------------| -| Connection | Long polling | WS (default) / Webhook | Gateway events | External WS bridge | Long polling | Internal protocol | -| DM support | Yes | Yes | Yes | Yes | Yes (DM only) | Yes | -| Group support | Yes (mention gating) | Yes | Yes | Yes | No | Yes | -| Forum/Topics | Yes (per-topic config) | Yes (topic session mode) | -- | -- | -- | -- | -| Message limit | 4,096 chars | Configurable (default 4,000) | 2,000 chars | N/A (bridge) | 2,000 chars | 2,000 chars | -| Streaming | Typing indicator | Streaming message cards | Edit "Thinking..." | No | No | No | -| Media | Photos, voice, files | Images, files (30 MB) | Files, embeds | JSON messages | Images (5 MB) | -- | -| Speech-to-text | Yes (STT proxy) | -- | -- | -- | -- | -- | -| Voice routing | Yes (VoiceAgentID) | -- | -- | -- | -- | -- | -| Rich formatting | Markdown → HTML | Card messages | Markdown | Plain text | Plain text | Plain text | -| Bot commands | 10+ commands | -- | -- | -- | -- | -- | -| Tool allow list | Per-topic | -- | -- | -- | -- | -- | -| Pairing support | Yes | Yes | Yes | Yes | Yes | Yes | -| Status reactions | Yes | Yes | -- | -- | -- | -- | +| Feature | Telegram | Feishu/Lark | Discord | Slack | WhatsApp | Zalo OA | Zalo Personal | +|---------|----------|-------------|---------|-------|----------|---------|---------------| +| Connection | Long polling | WS (default) / Webhook | Gateway events | Socket Mode | External WS bridge | Long polling | Internal protocol | +| DM support | Yes | Yes | Yes | Yes | Yes | Yes (DM only) | Yes | +| Group support | Yes (mention gating) | Yes | Yes | Yes (mention gating + thread cache) | Yes | No | Yes | +| Forum/Topics | Yes (per-topic config) | Yes (topic session mode) | -- | -- | -- | -- | -- | +| Message limit | 4,096 chars | Configurable (default 4,000) | 2,000 chars | 4,000 chars | N/A (bridge) | 2,000 chars | 2,000 chars | +| Streaming | Typing indicator | Streaming message cards | Edit "Thinking..." | Edit "Thinking..." (throttled 1s) | No | No | No | +| Media | Photos, voice, files | Images, files (30 MB) | Files, embeds | Files (download w/ SSRF protection) | JSON messages | Images (5 MB) | -- | +| Speech-to-text | Yes (STT proxy) | -- | -- | -- | -- | -- | -- | +| Voice routing | Yes (VoiceAgentID) | -- | -- | -- | -- | -- | -- | +| Rich formatting | Markdown → HTML | Card messages | Markdown | Markdown → mrkdwn | Plain text | Plain text | Plain text | +| Bot commands | 10+ commands | -- | -- | -- | -- | -- | -- | +| Tool allow list | Per-topic | -- | -- | -- | -- | -- | -- | +| Pairing support | Yes | Yes | Yes | Yes | Yes | Yes | Yes | +| Status reactions | Yes | Yes | -- | Yes | -- | -- | -- | --- @@ -385,7 +388,50 @@ The Discord channel uses the `discordgo` library to connect via the Discord Gate --- -## 8. WhatsApp +## 8. Slack + +The Slack channel uses the `slack-go/slack` library to connect via Socket Mode (WebSocket). + +### Key Behaviors + +- **Socket Mode**: Uses `xapp-` App-Level Token for WebSocket connection (no public URL needed) +- **Three token types**: `xoxb-` (Bot Token, required), `xapp-` (App-Level Token, required), `xoxp-` (User Token, optional for custom identity) +- **Token prefix validation**: Tokens validated at startup (`xoxb-`, `xapp-`, `xoxp-` prefixes) +- **Message limit**: 4,000-character limit with automatic splitting at newline boundaries +- **Placeholder editing**: Sends "Thinking..." → edits with actual response (same as Discord) +- **Mention gating**: `requireMention` default true; `<@botUserID>` stripped from content +- **Thread participation cache**: After bot replies in a thread, subsequent messages in that thread auto-trigger response without @mention (24h TTL) +- **Message dedup**: `channel+ts` key prevents duplicate processing on Socket Mode reconnect +- **Message debounce**: Per-thread batching of rapid messages (300ms default, configurable) +- **Dead socket classification**: Non-retryable auth errors (invalid_auth, token_revoked) fail fast instead of infinite reconnect +- **Streaming**: Edit-in-place via `chat.update` with 1000ms throttle (Slack Tier 3 rate limit) +- **Reactions**: Status emoji on user messages (thinking_face, hammer_and_wrench, white_check_mark, x, hourglass) +- **SSRF protection**: File download hostname allowlist (*.slack.com, *.slack-edge.com, *.slack-files.com), auth token stripped on redirect +- **Health probe**: `auth.test()` with 2.5s timeout for monitoring integration + +### Formatting Pipeline + +``` +LLM markdown → htmlTagsToMarkdown() → extractSlackTokens() → escapeHTMLEntities() +→ extractCodeBlocks() → convertTablesToCodeBlocks() → bold/strike/header/link conversion +→ restore tokens/code blocks → Slack mrkdwn +``` + +Key conversions: `**bold**` → `*bold*`, `~~strike~~` → `~strike~`, `[text](url)` → ``, `# Header` → `*Header*`, tables → code blocks. + +### Environment Variables + +``` +GOCLAW_SLACK_BOT_TOKEN → channels.slack.bot_token +GOCLAW_SLACK_APP_TOKEN → channels.slack.app_token +GOCLAW_SLACK_USER_TOKEN → channels.slack.user_token (optional) +``` + +Auto-enables when both bot_token and app_token are set. + +--- + +## 9. WhatsApp The WhatsApp channel communicates through an external WebSocket bridge (e.g., whatsapp-web.js based). GoClaw does not implement the WhatsApp protocol directly. @@ -399,7 +445,7 @@ The WhatsApp channel communicates through an external WebSocket bridge (e.g., wh --- -## 9. Zalo OA +## 10. Zalo OA The Zalo OA (Official Account) channel connects to the Zalo OA Bot API. @@ -414,7 +460,7 @@ The Zalo OA (Official Account) channel connects to the Zalo OA Bot API. --- -## 10. Zalo Personal +## 11. Zalo Personal The Zalo Personal channel provides access to personal Zalo accounts using a reverse-engineered protocol. This is an unofficial integration. @@ -443,7 +489,7 @@ Zalo Personal uses an unofficial, reverse-engineered protocol. The account used --- -## 11. Channel-Isolated Workspaces +## 12. Channel-Isolated Workspaces Each channel instance can target a specific agent, providing workspace isolation across channels. @@ -462,7 +508,7 @@ In managed mode, channel instances are loaded from the database with their assig --- -## 12. Local Key Propagation +## 13. Local Key Propagation Thread/topic context is preserved through the entire message pipeline using a `local_key` in message metadata. This ensures subagent, delegation, and team message results land in the correct thread — not the root chat. @@ -478,7 +524,7 @@ All channel state — placeholders, streams, reactions, typing controllers, thre --- -## 13. Per-User Isolation +## 14. Per-User Isolation Channels provide per-user isolation through compound sender IDs and context propagation: @@ -489,7 +535,7 @@ Channels provide per-user isolation through compound sender IDs and context prop --- -## 14. Pairing System +## 15. Pairing System The pairing system provides a DM authentication flow for channels using the `pairing` DM policy. @@ -540,6 +586,9 @@ flowchart TD | `internal/channels/feishu/bot.go` | Bot message handlers | | `internal/channels/feishu/bot_policy.go` | Policy evaluation | | `internal/channels/discord/discord.go` | Discord: gateway events, placeholder editing | +| `internal/channels/slack/slack.go` | Slack: Socket Mode, mention gating, thread caching | +| `internal/channels/slack/format.go` | Markdown → Slack mrkdwn pipeline | +| `internal/channels/slack/reactions.go` | Status emoji reactions on messages | | `internal/channels/whatsapp/whatsapp.go` | WhatsApp: external WS bridge | | `internal/channels/zalo/zalo.go` | Zalo OA: Bot API, long polling | | `internal/channels/zalo/personal/channel.go` | Zalo Personal: reverse-engineered protocol | diff --git a/go.mod b/go.mod index 7f66c123..38c5e7b9 100644 --- a/go.mod +++ b/go.mod @@ -86,6 +86,7 @@ require ( github.com/pires/go-proxyproto v0.8.1 // indirect github.com/prometheus-community/pro-bing v0.4.0 // indirect github.com/safchain/ethtool v0.3.0 // indirect + github.com/slack-go/slack v0.19.0 // indirect github.com/spf13/cast v1.7.1 // indirect github.com/tailscale/certstore v0.1.1-0.20231202035212-d3fa0460f47e // indirect github.com/tailscale/go-winio v0.0.0-20231025203758-c4f33415bf55 // indirect diff --git a/go.sum b/go.sum index 05a4ae90..2f9634ae 100644 --- a/go.sum +++ b/go.sum @@ -329,6 +329,8 @@ github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/safchain/ethtool v0.3.0 h1:gimQJpsI6sc1yIqP/y8GYgiXn/NjgvpM0RNoWLVVmP0= github.com/safchain/ethtool v0.3.0/go.mod h1:SA9BwrgyAqNo7M+uaL6IYbxpm5wk3L7Mm6ocLW+CJUs= +github.com/slack-go/slack v0.19.0 h1:J8lL/nGTsIUX53HU8YxZeI3PDkA+sxZsFrI2Dew7h44= +github.com/slack-go/slack v0.19.0/go.mod h1:K81UmCivcYd/5Jmz8vLBfuyoZ3B4rQC2GHVXHteXiAE= github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= diff --git a/internal/channels/media_utils.go b/internal/channels/media_utils.go new file mode 100644 index 00000000..b17dd1ac --- /dev/null +++ b/internal/channels/media_utils.go @@ -0,0 +1,76 @@ +package channels + +import ( + "fmt" + "html" + "os" + "path/filepath" + "strings" +) + +const ( + // DocMaxChars is the max characters to extract from text documents (matching TS: 200K). + DocMaxChars = 200_000 +) + +// TextExtensions maps file extensions to MIME types for text files we can extract. +var TextExtensions = map[string]string{ + ".txt": "text/plain", + ".md": "text/markdown", + ".csv": "text/csv", + ".tsv": "text/tab-separated-values", + ".json": "application/json", + ".yaml": "text/yaml", + ".yml": "text/yaml", + ".xml": "text/xml", + ".log": "text/plain", + ".ini": "text/plain", + ".cfg": "text/plain", + ".env": "text/plain", + ".sh": "text/x-shellscript", + ".py": "text/x-python", + ".go": "text/x-go", + ".js": "text/javascript", + ".ts": "text/typescript", + ".html": "text/html", + ".css": "text/css", + ".sql": "text/x-sql", + ".rs": "text/x-rust", + ".java": "text/x-java", + ".c": "text/x-c", + ".cpp": "text/x-c++", + ".h": "text/x-c", + ".rb": "text/x-ruby", + ".php": "text/x-php", + ".toml": "text/x-toml", +} + +// ExtractDocumentContent reads a document file and returns its content wrapped in XML tags. +// For text files: extracts content, truncates at DocMaxChars, wraps in block. +// For binary files: returns a placeholder message. +func ExtractDocumentContent(filePath, fileName string) (string, error) { + if filePath == "" { + return fmt.Sprintf("[File: %s — download failed]", fileName), nil + } + + ext := strings.ToLower(filepath.Ext(fileName)) + mime, isText := TextExtensions[ext] + if !isText { + return fmt.Sprintf("[File: %s — binary format not supported, only text files can be processed]", fileName), nil + } + + data, err := os.ReadFile(filePath) + if err != nil { + return "", fmt.Errorf("read file %s: %w", fileName, err) + } + + content := string(data) + + if len(content) > DocMaxChars { + content = content[:DocMaxChars] + "\n... [truncated]" + } + + escaped := html.EscapeString(content) + + return fmt.Sprintf("\n%s\n", fileName, mime, escaped), nil +} diff --git a/internal/channels/slack/channel.go b/internal/channels/slack/channel.go new file mode 100644 index 00000000..4ab94bc2 --- /dev/null +++ b/internal/channels/slack/channel.go @@ -0,0 +1,521 @@ +package slack + +import ( + "context" + "fmt" + "log/slog" + "regexp" + "strings" + "sync" + "time" + + slackapi "github.com/slack-go/slack" + "github.com/slack-go/slack/socketmode" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +const ( + pairingDebounceTime = 60 * time.Second + maxMessageLen = 4000 // Slack mrkdwn text limit + userCacheTTL = 1 * time.Hour + healthProbeTimeout = 2500 * time.Millisecond +) + +// Channel connects to Slack via Socket Mode for event-driven messaging. +type Channel struct { + *channels.BaseChannel + api *slackapi.Client // Bot Token API client (xoxb-) + userAPI *slackapi.Client // User Token API client (xoxp-, optional) + sm *socketmode.Client // Socket Mode client (xapp-) + config config.SlackConfig + botUserID string // populated on Start() via auth.test + teamID string // populated on Start() via auth.test + requireMention bool // require @bot in channels (default true) + + placeholders sync.Map // localKey -> placeholderTS + dedup sync.Map // channel+ts -> time.Time + threadParticip sync.Map // channelID+threadTS -> time.Time (auto-reply without @mention) + streams sync.Map // localKey -> *streamState + reactions sync.Map // chatID:messageID -> *reactionState + pairingDebounce sync.Map // senderID -> time.Time + approvedGroups sync.Map // channelID -> true + + // High-churn map: sync.Mutex + regular map for debounce timers + debounceMu sync.Mutex + debounceTimers map[string]*debounceEntry + + // Read-heavy map: sync.RWMutex + regular map for user display name cache + userCacheMu sync.RWMutex + userCache map[string]cachedUser + + pairingService store.PairingStore + groupHistory *channels.PendingHistory + historyLimit int + debounceDelay time.Duration + threadTTL time.Duration // thread participation expiry (0 = disabled) + wg sync.WaitGroup // tracks goroutines for clean shutdown + cancelFn context.CancelFunc +} + +type cachedUser struct { + displayName string + fetchedAt time.Time +} + +// Compile-time interface assertions. +var _ channels.Channel = (*Channel)(nil) +var _ channels.StreamingChannel = (*Channel)(nil) +var _ channels.ReactionChannel = (*Channel)(nil) +var _ channels.BlockReplyChannel = (*Channel)(nil) + +// New creates a new Slack channel from config. +func New(cfg config.SlackConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore) (*Channel, error) { + if cfg.BotToken == "" { + return nil, fmt.Errorf("slack bot_token is required") + } + if cfg.AppToken == "" { + return nil, fmt.Errorf("slack app_token is required for Socket Mode") + } + + // Token prefix validation: catch misconfigured tokens early. + if !strings.HasPrefix(cfg.BotToken, "xoxb-") { + return nil, fmt.Errorf("slack bot_token must start with xoxb-") + } + if !strings.HasPrefix(cfg.AppToken, "xapp-") { + return nil, fmt.Errorf("slack app_token must start with xapp-") + } + if cfg.UserToken != "" && !strings.HasPrefix(cfg.UserToken, "xoxp-") { + return nil, fmt.Errorf("slack user_token must start with xoxp-") + } + + base := channels.NewBaseChannel("slack", msgBus, cfg.AllowFrom) + base.ValidatePolicy(cfg.DMPolicy, cfg.GroupPolicy) + + requireMention := true + if cfg.RequireMention != nil { + requireMention = *cfg.RequireMention + } + + historyLimit := cfg.HistoryLimit + if historyLimit == 0 { + historyLimit = channels.DefaultGroupHistoryLimit + } + + debounceDelay := time.Duration(cfg.DebounceDelay) * time.Millisecond + if cfg.DebounceDelay == 0 { + debounceDelay = 300 * time.Millisecond + } + + threadTTL := 24 * time.Hour // default: 24h + if cfg.ThreadTTL != nil { + if *cfg.ThreadTTL <= 0 { + threadTTL = 0 // explicitly disabled + } else { + threadTTL = time.Duration(*cfg.ThreadTTL) * time.Hour + } + } + + return &Channel{ + BaseChannel: base, + config: cfg, + requireMention: requireMention, + pairingService: pairingSvc, + groupHistory: channels.NewPendingHistory(), + historyLimit: historyLimit, + debounceDelay: debounceDelay, + threadTTL: threadTTL, + debounceTimers: make(map[string]*debounceEntry), + userCache: make(map[string]cachedUser), + }, nil +} + +// Start opens the Socket Mode connection and begins receiving events. +func (c *Channel) Start(ctx context.Context) error { + slog.Info("starting slack bot (socket mode)") + + c.api = slackapi.New( + c.config.BotToken, + slackapi.OptionAppLevelToken(c.config.AppToken), + ) + + if c.config.UserToken != "" { + c.userAPI = slackapi.New(c.config.UserToken) + slog.Info("slack user token configured (custom identity enabled)") + } + + authResp, err := c.api.AuthTest() + if err != nil { + return fmt.Errorf("slack auth.test failed: %w", err) + } + c.botUserID = authResp.UserID + c.teamID = authResp.TeamID + + c.sm = socketmode.New( + c.api, + socketmode.OptionDebug(false), + ) + + smCtx, cancel := context.WithCancel(ctx) + c.cancelFn = cancel + + c.wg.Add(3) // event loop + RunContext loop + periodic sweep + + // Goroutine 1: Event loop + go func() { + defer c.wg.Done() + c.eventLoop(smCtx) + }() + + // Goroutine 2: Socket Mode connection with dead socket error classification + go func() { + defer c.wg.Done() + for { + if err := c.sm.RunContext(smCtx); err != nil { + if smCtx.Err() != nil { + return + } + if isNonRetryableAuthError(err.Error()) { + slog.Error("slack: non-retryable auth error, stopping channel", "error", err) + c.SetRunning(false) + return + } + slog.Warn("slack socket mode error, reconnecting in 5s", "error", err) + time.Sleep(5 * time.Second) + } + } + }() + + // Goroutine 3: Periodic sweep (every 2 minutes) for TTL-based map eviction + go func() { + defer c.wg.Done() + ticker := time.NewTicker(2 * time.Minute) + defer ticker.Stop() + for { + select { + case <-smCtx.Done(): + return + case <-ticker.C: + c.sweepMaps() + } + } + }() + + c.SetRunning(true) + slog.Info("slack bot connected", "user_id", c.botUserID, "team", authResp.Team) + return nil +} + +// sweepMaps performs age-based eviction across all TTL-controlled maps. +func (c *Channel) sweepMaps() { + now := time.Now() + + c.dedup.Range(func(k, v any) bool { + if now.Sub(v.(time.Time)) > 5*time.Minute { + c.dedup.Delete(k) + } + return true + }) + + if c.threadTTL > 0 { + c.threadParticip.Range(func(k, v any) bool { + if now.Sub(v.(time.Time)) > c.threadTTL { + c.threadParticip.Delete(k) + } + return true + }) + } + + c.userCacheMu.Lock() + for k, v := range c.userCache { + if now.Sub(v.fetchedAt) > userCacheTTL { + delete(c.userCache, k) + } + } + c.userCacheMu.Unlock() + + c.pairingDebounce.Range(func(k, v any) bool { + if now.Sub(v.(time.Time)) > pairingDebounceTime*10 { + c.pairingDebounce.Delete(k) + } + return true + }) + + c.streams.Range(func(k, v any) bool { + st := v.(*streamState) + st.mu.Lock() + stale := now.Sub(st.lastUpdate) > 10*time.Minute + st.mu.Unlock() + if stale { + c.streams.Delete(k) + } + return true + }) +} + +// eventLoop processes Socket Mode events. +func (c *Channel) eventLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case evt, ok := <-c.sm.Events: + if !ok { + return + } + c.handleEvent(evt) + } + } +} + +func (c *Channel) handleEvent(evt socketmode.Event) { + switch evt.Type { + case socketmode.EventTypeEventsAPI: + c.handleEventsAPI(evt) + case socketmode.EventTypeDisconnect: + slog.Info("slack socket mode disconnecting (will auto-reconnect)") + } +} + +// Stop gracefully shuts down the Slack channel. +func (c *Channel) Stop(_ context.Context) error { + slog.Info("stopping slack bot") + c.SetRunning(false) + + if c.cancelFn != nil { + c.cancelFn() + } + + // Flush all pending debounce entries before shutdown + c.debounceMu.Lock() + pendingKeys := make([]string, 0, len(c.debounceTimers)) + for k, entry := range c.debounceTimers { + entry.mu.Lock() + if entry.timer != nil { + entry.timer.Stop() + } + entry.mu.Unlock() + pendingKeys = append(pendingKeys, k) + } + c.debounceMu.Unlock() + + for _, k := range pendingKeys { + c.flushDebounce(k) + } + + // Wait for all goroutines with timeout + doneCh := make(chan struct{}) + go func() { + c.wg.Wait() + close(doneCh) + }() + + select { + case <-doneCh: + case <-time.After(10 * time.Second): + slog.Warn("slack bot stop timed out after 10s") + } + + return nil +} + +// HandleMessage overrides BaseChannel to allow messages when the chatID (Slack channel) +// is in the allowlist, enabling group-level allowlisting without requiring individual user IDs. +// This is Slack-specific: other channels only check senderID in BaseChannel.HandleMessage. +func (c *Channel) HandleMessage(senderID, chatID, content string, mediaPaths []string, metadata map[string]string, peerKind string) { + // Allow if either the sender or the Slack channel ID is in the allowlist. + if !c.IsAllowed(senderID) && !c.IsAllowed(chatID) { + return + } + + userID := senderID + if idx := strings.IndexByte(senderID, '|'); idx > 0 { + userID = senderID[:idx] + } + + var mediaFiles []bus.MediaFile + for _, p := range mediaPaths { + mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) + } + + c.Bus().PublishInbound(bus.InboundMessage{ + Channel: c.Name(), + SenderID: senderID, + ChatID: chatID, + Content: content, + Media: mediaFiles, + PeerKind: peerKind, + UserID: userID, + Metadata: metadata, + AgentID: c.AgentID(), + }) +} + +// BlockReplyEnabled returns the per-channel block_reply override. +func (c *Channel) BlockReplyEnabled() *bool { return c.config.BlockReply } + +// Send delivers an outbound message to Slack. +func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return fmt.Errorf("slack bot not running") + } + + channelID := msg.ChatID + if channelID == "" { + return fmt.Errorf("empty chat ID for slack send") + } + + placeholderKey := channelID + if pk := msg.Metadata["placeholder_key"]; pk != "" { + placeholderKey = pk + } + threadTS := msg.Metadata["message_thread_id"] + + // Placeholder update (LLM retry notification) + if msg.Metadata["placeholder_update"] == "true" { + if pTS, ok := c.placeholders.Load(placeholderKey); ok { + ts := pTS.(string) + _, _, _, _ = c.api.UpdateMessage(channelID, ts, + slackapi.MsgOptionText(msg.Content, false)) + } + return nil + } + + content := msg.Content + + // NO_REPLY: delete placeholder, return + if content == "" { + if pTS, ok := c.placeholders.Load(placeholderKey); ok { + c.placeholders.Delete(placeholderKey) + ts := pTS.(string) + _, _, _ = c.api.DeleteMessage(channelID, ts) + } + return nil + } + + content = markdownToSlackMrkdwn(content) + + // Edit placeholder with first chunk, send rest as follow-ups + if pTS, ok := c.placeholders.Load(placeholderKey); ok { + c.placeholders.Delete(placeholderKey) + ts := pTS.(string) + + editContent, remaining := splitAtLimit(content, maxMessageLen) + + opts := []slackapi.MsgOption{slackapi.MsgOptionText(editContent, false)} + if threadTS != "" { + opts = append(opts, slackapi.MsgOptionTS(threadTS)) + } + + if _, _, _, editErr := c.api.UpdateMessage(channelID, ts, opts...); editErr == nil { + if remaining != "" { + return c.sendChunked(channelID, remaining, threadTS) + } + return nil + } else { + slog.Warn("slack placeholder edit failed, sending new message", + "channel_id", channelID, "error", editErr) + } + } + + // Handle media attachments + for _, media := range msg.Media { + if err := c.uploadFile(channelID, threadTS, media); err != nil { + slog.Warn("slack: file upload failed", + "file", media.URL, "error", err) + c.sendChunked(channelID, fmt.Sprintf("[File upload failed: %s]", media.URL), threadTS) + } + } + + return c.sendChunked(channelID, content, threadTS) +} + +func (c *Channel) sendChunked(channelID, content, threadTS string) error { + for len(content) > 0 { + chunk, rest := splitAtLimit(content, maxMessageLen) + content = rest + + opts := []slackapi.MsgOption{slackapi.MsgOptionText(chunk, false)} + if threadTS != "" { + opts = append(opts, slackapi.MsgOptionTS(threadTS)) + } + + if _, _, err := c.api.PostMessage(channelID, opts...); err != nil { + return fmt.Errorf("send slack message: %w", err) + } + } + return nil +} + +// splitAtLimit splits content at maxLen runes, preferring newline boundaries. +func splitAtLimit(content string, maxLen int) (chunk, remaining string) { + runes := []rune(content) + if len(runes) <= maxLen { + return content, "" + } + cutAt := maxLen + // Try to break at a newline in the second half + candidate := string(runes[:maxLen]) + if idx := strings.LastIndex(candidate, "\n"); idx > len(candidate)/2 { + return content[:idx+1], content[idx+1:] + } + return string(runes[:cutAt]), string(runes[cutAt:]) +} + +// resolveDisplayName fetches and caches the Slack display name for a user ID. +func (c *Channel) resolveDisplayName(userID string) string { + c.userCacheMu.RLock() + cu, found := c.userCache[userID] + c.userCacheMu.RUnlock() + + if found && time.Since(cu.fetchedAt) < userCacheTTL { + return cu.displayName + } + + user, err := c.api.GetUserInfo(userID) + if err != nil { + slog.Debug("slack: failed to resolve user", "user_id", userID, "error", err) + return userID + } + + name := user.Profile.DisplayName + if name == "" { + name = user.RealName + } + if name == "" { + name = user.Name + } + + c.userCacheMu.Lock() + c.userCache[userID] = cachedUser{displayName: name, fetchedAt: time.Now()} + c.userCacheMu.Unlock() + + return name +} + +// nonRetryableAuthErrors matches Slack errors that indicate permanent auth failure. +var nonRetryableAuthErrors = regexp.MustCompile( + `(?i)(invalid_auth|token_revoked|account_inactive|not_authed|team_not_found|missing_scope)`, +) + +func isNonRetryableAuthError(errMsg string) bool { + return nonRetryableAuthErrors.MatchString(errMsg) +} + +// HealthProbe performs an auth.test call to verify the Slack connection is alive. +func (c *Channel) HealthProbe(ctx context.Context) (ok bool, elapsed time.Duration, err error) { + if c.api == nil { + return false, 0, fmt.Errorf("slack client not initialized (Start() not called)") + } + + start := time.Now() + probeCtx, cancel := context.WithTimeout(ctx, healthProbeTimeout) + defer cancel() + + _, err = c.api.AuthTestContext(probeCtx) + elapsed = time.Since(start) + return err == nil, elapsed, err +} diff --git a/internal/channels/slack/channel_test.go b/internal/channels/slack/channel_test.go new file mode 100644 index 00000000..11824690 --- /dev/null +++ b/internal/channels/slack/channel_test.go @@ -0,0 +1,203 @@ +package slack + +import ( + "testing" +) + +func TestSplitAtLimit(t *testing.T) { + tests := []struct { + name string + content string + maxLen int + expectedChunk string + expectedRem string + }{ + { + name: "under limit", + content: "hello world", + maxLen: 20, + expectedChunk: "hello world", + expectedRem: "", + }, + { + name: "exact limit", + content: "hello world", + maxLen: 11, + expectedChunk: "hello world", + expectedRem: "", + }, + { + name: "over limit with newline at half", + content: "line1\nline2\nline3", + maxLen: 10, + expectedChunk: "line1\nline", + expectedRem: "2\nline3", + }, + { + name: "over limit no newline", + content: "a very long string with no newlines at all", + maxLen: 10, + expectedChunk: "a very lon", + expectedRem: "g string with no newlines at all", + }, + { + name: "newline at boundary but not in second half", + content: "short\nvery long remaining text here", + maxLen: 10, + expectedChunk: "short\nvery", + expectedRem: " long remaining text here", + }, + { + name: "empty string", + content: "", + maxLen: 10, + expectedChunk: "", + expectedRem: "", + }, + { + name: "single character", + content: "a", + maxLen: 10, + expectedChunk: "a", + expectedRem: "", + }, + { + name: "cjk characters", + content: "中文test日本語", + maxLen: 5, + expectedChunk: "中文tes", + expectedRem: "t日本語", + }, + { + name: "emoji characters", + content: "hello 👋 world 🌍 test", + maxLen: 10, + expectedChunk: "hello 👋 wo", + expectedRem: "rld 🌍 test", + }, + { + name: "newline in second half with long first half", + content: "aaaaaaaaaa\nbbbbbbbbbb", + maxLen: 15, + expectedChunk: "aaaaaaaaaa\n", + expectedRem: "bbbbbbbbbb", + }, + { + name: "multiple newlines", + content: "line1\nline2\nline3\nline4\nline5", + maxLen: 20, + expectedChunk: "line1\nline2\nline3\n", + expectedRem: "line4\nline5", + }, + { + name: "newline at exact boundary", + content: "12345\n67890", + maxLen: 10, + expectedChunk: "12345\n6789", + expectedRem: "0", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + chunk, rem := splitAtLimit(tt.content, tt.maxLen) + if chunk != tt.expectedChunk || rem != tt.expectedRem { + t.Errorf("splitAtLimit(%q, %d) = (%q, %q), want (%q, %q)", + tt.content, tt.maxLen, chunk, rem, tt.expectedChunk, tt.expectedRem) + } + }) + } +} + +func TestIsNonRetryableAuthError(t *testing.T) { + tests := []struct { + name string + errMsg string + expected bool + }{ + { + name: "invalid_auth error", + errMsg: "invalid_auth", + expected: true, + }, + { + name: "token_revoked error", + errMsg: "token_revoked", + expected: true, + }, + { + name: "account_inactive error", + errMsg: "account_inactive", + expected: true, + }, + { + name: "not_authed error", + errMsg: "not_authed", + expected: true, + }, + { + name: "team_not_found error", + errMsg: "team_not_found", + expected: true, + }, + { + name: "missing_scope error", + errMsg: "missing_scope", + expected: true, + }, + { + name: "retryable error", + errMsg: "rate_limited", + expected: false, + }, + { + name: "random error", + errMsg: "some random error", + expected: false, + }, + { + name: "empty string", + errMsg: "", + expected: false, + }, + { + name: "case insensitive invalid_auth", + errMsg: "Invalid_Auth", + expected: true, + }, + { + name: "case insensitive token_revoked", + errMsg: "TOKEN_REVOKED", + expected: true, + }, + { + name: "error message contains non_retryable", + errMsg: "error: invalid_auth while connecting", + expected: true, + }, + { + name: "similar but different error", + errMsg: "invalid_token_auth", + expected: false, + }, + { + name: "socket error", + errMsg: "i/o timeout", + expected: false, + }, + { + name: "connection refused", + errMsg: "connection refused", + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isNonRetryableAuthError(tt.errMsg) + if got != tt.expected { + t.Errorf("isNonRetryableAuthError(%q) = %v, want %v", tt.errMsg, got, tt.expected) + } + }) + } +} diff --git a/internal/channels/slack/factory.go b/internal/channels/slack/factory.go new file mode 100644 index 00000000..53c5167b --- /dev/null +++ b/internal/channels/slack/factory.go @@ -0,0 +1,90 @@ +package slack + +import ( + "encoding/json" + "fmt" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// slackCreds maps the credentials JSON from the channel_instances table. +type slackCreds struct { + BotToken string `json:"bot_token"` // xoxb-... + AppToken string `json:"app_token"` // xapp-... (Socket Mode) + UserToken string `json:"user_token,omitempty"` // xoxp-... (optional: custom identity) +} + +// slackInstanceConfig maps the non-secret config JSONB from the channel_instances table. +type slackInstanceConfig struct { + DMPolicy string `json:"dm_policy,omitempty"` + GroupPolicy string `json:"group_policy,omitempty"` + AllowFrom []string `json:"allow_from,omitempty"` + RequireMention *bool `json:"require_mention,omitempty"` + HistoryLimit int `json:"history_limit,omitempty"` + DMStream *bool `json:"dm_stream,omitempty"` + GroupStream *bool `json:"group_stream,omitempty"` + NativeStream *bool `json:"native_stream,omitempty"` + ReactionLevel string `json:"reaction_level,omitempty"` + BlockReply *bool `json:"block_reply,omitempty"` + DebounceDelay int `json:"debounce_delay,omitempty"` + ThreadTTL *int `json:"thread_ttl,omitempty"` +} + +// Factory creates a Slack channel from DB instance data. +func Factory(name string, creds json.RawMessage, cfg json.RawMessage, + msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) { + + var c slackCreds + if len(creds) > 0 { + if err := json.Unmarshal(creds, &c); err != nil { + return nil, fmt.Errorf("decode slack credentials: %w", err) + } + } + if c.BotToken == "" { + return nil, fmt.Errorf("slack bot_token is required") + } + if c.AppToken == "" { + return nil, fmt.Errorf("slack app_token is required for Socket Mode") + } + + var ic slackInstanceConfig + if len(cfg) > 0 { + if err := json.Unmarshal(cfg, &ic); err != nil { + return nil, fmt.Errorf("decode slack config: %w", err) + } + } + + slackCfg := config.SlackConfig{ + Enabled: true, + BotToken: c.BotToken, + AppToken: c.AppToken, + UserToken: c.UserToken, + AllowFrom: ic.AllowFrom, + DMPolicy: ic.DMPolicy, + GroupPolicy: ic.GroupPolicy, + RequireMention: ic.RequireMention, + HistoryLimit: ic.HistoryLimit, + DMStream: ic.DMStream, + GroupStream: ic.GroupStream, + NativeStream: ic.NativeStream, + ReactionLevel: ic.ReactionLevel, + BlockReply: ic.BlockReply, + DebounceDelay: ic.DebounceDelay, + ThreadTTL: ic.ThreadTTL, + } + + // Secure default: DB instances default to "pairing" for groups. + if slackCfg.GroupPolicy == "" { + slackCfg.GroupPolicy = "pairing" + } + + ch, err := New(slackCfg, msgBus, pairingSvc) + if err != nil { + return nil, err + } + ch.SetName(name) + return ch, nil +} diff --git a/internal/channels/slack/format.go b/internal/channels/slack/format.go new file mode 100644 index 00000000..41591ffe --- /dev/null +++ b/internal/channels/slack/format.go @@ -0,0 +1,198 @@ +package slack + +import ( + "fmt" + "regexp" + "strings" +) + +// markdownToSlackMrkdwn converts standard markdown (LLM output) to Slack mrkdwn syntax. +// Pipeline order (CRITICAL): +// 1. htmlTagsToMarkdown -- convert HTML tags to markdown +// 2. extractSlackTokens -- protect <@U123>, <#C456>, BEFORE escaping +// 3. escapeHTMLEntities -- escape &, <, > (won't touch protected tokens) +// 4. extractCodeBlocks -- protect code fences from formatting +// 5. extractInlineCodes -- protect inline code +// 6. convertTablesToCodeBlocks -- render tables as code blocks +func markdownToSlackMrkdwn(text string) string { + if text == "" { + return "" + } + + text = htmlTagsToMarkdown(text) + + // Preserve Slack-native tokens BEFORE HTML entity escaping + slackTokens, text := extractSlackTokens(text) + + text = escapeHTMLEntities(text) + + codeBlocks, text := extractCodeBlocks(text) + inlineCodes, text := extractInlineCodes(text) + + text = convertTablesToCodeBlocks(text) + + // Convert markdown links: [text](url) -> + text = reLink.ReplaceAllString(text, "<$2|$1>") + + // Convert bold: **text** or __text__ -> *text* + text = reBoldDouble.ReplaceAllString(text, "*$1*") + text = reBoldUnderscore.ReplaceAllString(text, "*$1*") + + // Convert strikethrough: ~~text~~ -> ~text~ + text = reStrike.ReplaceAllString(text, "~$1~") + + // Convert headers: # Header -> *Header* (no native header in mrkdwn) + text = reHeader.ReplaceAllString(text, "*$1*") + + // Restore Slack tokens + for i, token := range slackTokens { + text = strings.Replace(text, fmt.Sprintf("\x00ST%d\x00", i), token, 1) + } + + // Restore inline code + for i, code := range inlineCodes { + text = strings.Replace(text, fmt.Sprintf("\x00IC%d\x00", i), "`"+code+"`", 1) + } + + // Restore code blocks + for i, block := range codeBlocks { + text = strings.Replace(text, fmt.Sprintf("\x00CB%d\x00", i), "```"+block+"```", 1) + } + + return text +} + +// Compiled regex patterns. +var ( + reLink = regexp.MustCompile(`\[([^\]]+)\]\(([^)]+)\)`) + reBoldDouble = regexp.MustCompile(`\*\*(.+?)\*\*`) + reBoldUnderscore = regexp.MustCompile(`__(.+?)__`) + reStrike = regexp.MustCompile(`~~(.+?)~~`) + reHeader = regexp.MustCompile(`(?m)^#{1,6}\s+(.+)$`) + reHTMLBold = regexp.MustCompile(`(?i)<(?:b|strong)>([\s\S]*?)`) + reHTMLItalic = regexp.MustCompile(`(?i)<(?:i|em)>([\s\S]*?)`) + reHTMLStrike = regexp.MustCompile(`(?i)<(?:s|strike|del)>([\s\S]*?)`) + reHTMLCode = regexp.MustCompile(`(?i)([\s\S]*?)`) + reHTMLLink = regexp.MustCompile(`(?i)]*>([\s\S]*?)`) + reHTMLBreak = regexp.MustCompile(`(?i)`) + reHTMLPara = regexp.MustCompile(`(?i)`) + reTableRow = regexp.MustCompile(`(?m)^\|(.+)\|$`) + reTableSep = regexp.MustCompile(`(?m)^\|[\s:]*-+[\s:|-]*\|$`) + reSlackToken = regexp.MustCompile(`<[@#!][^>]+>|<(?:mailto|tel|https?|ftp):[^>]+>`) +) + +func escapeHTMLEntities(text string) string { + text = strings.ReplaceAll(text, "&", "&") + text = strings.ReplaceAll(text, "<", "<") + text = strings.ReplaceAll(text, ">", ">") + return text +} + +func extractSlackTokens(text string) (tokens []string, result string) { + result = reSlackToken.ReplaceAllStringFunc(text, func(match string) string { + tokens = append(tokens, match) + return fmt.Sprintf("\x00ST%d\x00", len(tokens)-1) + }) + return tokens, result +} + +func htmlTagsToMarkdown(text string) string { + text = reHTMLBreak.ReplaceAllString(text, "\n") + text = reHTMLPara.ReplaceAllString(text, "\n") + text = reHTMLBold.ReplaceAllString(text, "**$1**") + text = reHTMLItalic.ReplaceAllString(text, "_${1}_") + text = reHTMLStrike.ReplaceAllString(text, "~~$1~~") + text = reHTMLCode.ReplaceAllString(text, "`$1`") + text = reHTMLLink.ReplaceAllString(text, "[$2]($1)") + return text +} + +func extractCodeBlocks(text string) (blocks []string, result string) { + parts := strings.Split(text, "```") + if len(parts) < 3 { + return nil, text + } + + var sb strings.Builder + for i, part := range parts { + if i%2 == 1 { + blocks = append(blocks, part) + sb.WriteString(fmt.Sprintf("\x00CB%d\x00", len(blocks)-1)) + } else { + sb.WriteString(part) + } + } + // If odd number of ```, the last unpaired one is literal + if len(parts)%2 == 0 { + sb.WriteString("```") + sb.WriteString(parts[len(parts)-1]) + } + return blocks, sb.String() +} + +func extractInlineCodes(text string) (codes []string, result string) { + var sb strings.Builder + inCode := false + codeStart := 0 + + for i := 0; i < len(text); i++ { + if text[i] == '`' { + if inCode { + codes = append(codes, text[codeStart:i]) + sb.WriteString(fmt.Sprintf("\x00IC%d\x00", len(codes)-1)) + inCode = false + } else { + inCode = true + codeStart = i + 1 + } + } else if !inCode { + sb.WriteByte(text[i]) + } + } + + if inCode { + sb.WriteByte('`') + sb.WriteString(text[codeStart:]) + } + + return codes, sb.String() +} + +func convertTablesToCodeBlocks(text string) string { + lines := strings.Split(text, "\n") + var result []string + var tableLines []string + inTable := false + + for _, line := range lines { + isTableRow := reTableRow.MatchString(line) + + if isTableRow { + if !inTable { + inTable = true + tableLines = nil + } + if reTableSep.MatchString(line) { + continue + } + tableLines = append(tableLines, line) + } else { + if inTable { + result = append(result, "```") + result = append(result, tableLines...) + result = append(result, "```") + inTable = false + tableLines = nil + } + result = append(result, line) + } + } + + if inTable && len(tableLines) > 0 { + result = append(result, "```") + result = append(result, tableLines...) + result = append(result, "```") + } + + return strings.Join(result, "\n") +} diff --git a/internal/channels/slack/format_test.go b/internal/channels/slack/format_test.go new file mode 100644 index 00000000..273b4bdc --- /dev/null +++ b/internal/channels/slack/format_test.go @@ -0,0 +1,536 @@ +package slack + +import ( + "testing" +) + +func TestMarkdownToSlackMrkdwn(t *testing.T) { + tests := []struct { + name string + input string + expected string + }{ + { + name: "empty string", + input: "", + expected: "", + }, + { + name: "plain text", + input: "hello world", + expected: "hello world", + }, + { + name: "bold double asterisk", + input: "this is **bold** text", + expected: "this is *bold* text", + }, + { + name: "bold underscore", + input: "this is __bold__ text", + expected: "this is *bold* text", + }, + { + name: "strikethrough", + input: "this is ~~striked~~ text", + expected: "this is ~striked~ text", + }, + { + name: "header h1", + input: "# Header Title", + expected: "*Header Title*", + }, + { + name: "header h2", + input: "## Sub Header", + expected: "*Sub Header*", + }, + { + name: "header h6", + input: "###### Small Header", + expected: "*Small Header*", + }, + { + name: "markdown link", + input: "[click here](https://example.com)", + expected: "", + }, + { + name: "inline code", + input: "use `variable` in code", + expected: "use `variable` in code", + }, + { + name: "code block", + input: "```\nfunction test() {}\n```", + expected: "```\nfunction test() {}\n```", + }, + { + name: "preserve slack tokens user mention", + input: "hey <@U123456> check this", + expected: "hey <@U123456> check this", + }, + { + name: "preserve slack tokens channel mention", + input: "discuss in <#C123456>", + expected: "discuss in <#C123456>", + }, + { + name: "preserve slack tokens url", + input: "see ", + expected: "see ", + }, + { + name: "preserve slack tokens mailto", + input: "email ", + expected: "email ", + }, + { + name: "mixed formatting", + input: "**bold** and ~~strike~~ and `code`", + expected: "*bold* and ~strike~ and `code`", + }, + { + name: "html bold tag", + input: "this is bold text", + expected: "this is *bold* text", + }, + { + name: "html italic tag", + input: "this is italic text", + expected: "this is _italic_ text", + }, + { + name: "html strike tag", + input: "this is struck text", + expected: "this is ~struck~ text", + }, + { + name: "html code tag", + input: "use var here", + expected: "use `var` here", + }, + { + name: "html link tag", + input: "click here", + expected: "click ", + }, + { + name: "html br tag", + input: "line1
line2", + expected: "line1\nline2", + }, + { + name: "html p tags", + input: "

paragraph

", + expected: "\nparagraph\n", + }, + { + name: "complex mixed", + input: "# Title\n\nSee **bold** at and `code`", + expected: "*Title*\n\nSee *bold* at and `code`", + }, + { + name: "special chars escaped", + input: "a & b < c > d", + expected: "a & b < c > d", + }, + { + name: "slack tokens with special chars preserved", + input: "email ", + expected: "email ", + }, + { + name: "multiple slack tokens", + input: "<@U123> and <#C456> with ", + expected: "<@U123> and <#C456> with ", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := markdownToSlackMrkdwn(tt.input) + if got != tt.expected { + t.Errorf("markdownToSlackMrkdwn(%q) = %q, want %q", tt.input, got, tt.expected) + } + }) + } +} + +func TestEscapeHTMLEntities(t *testing.T) { + tests := []struct { + name string + input string + expected string + }{ + { + name: "no special chars", + input: "hello world", + expected: "hello world", + }, + { + name: "ampersand", + input: "a & b", + expected: "a & b", + }, + { + name: "less than", + input: "a < b", + expected: "a < b", + }, + { + name: "greater than", + input: "a > b", + expected: "a > b", + }, + { + name: "all special chars", + input: "a & b < c > d", + expected: "a & b < c > d", + }, + { + name: "multiple ampersands", + input: "a & b & c & d", + expected: "a & b & c & d", + }, + { + name: "empty string", + input: "", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := escapeHTMLEntities(tt.input) + if got != tt.expected { + t.Errorf("escapeHTMLEntities(%q) = %q, want %q", tt.input, got, tt.expected) + } + }) + } +} + +func TestExtractSlackTokens(t *testing.T) { + tests := []struct { + name string + input string + expectedCount int + expectedContent string // should have placeholders + }{ + { + name: "no tokens", + input: "hello world", + expectedCount: 0, + }, + { + name: "user mention", + input: "hey <@U123456>", + expectedCount: 1, + }, + { + name: "channel mention", + input: "in <#C789012>", + expectedCount: 1, + }, + { + name: "https url", + input: "see ", + expectedCount: 1, + }, + { + name: "http url", + input: "see ", + expectedCount: 1, + }, + { + name: "ftp url", + input: "download ", + expectedCount: 1, + }, + { + name: "mailto", + input: "email ", + expectedCount: 1, + }, + { + name: "tel", + input: "call ", + expectedCount: 1, + }, + { + name: "multiple tokens", + input: "<@U123> and <#C456> with ", + expectedCount: 3, + }, + { + name: "empty string", + input: "", + expectedCount: 0, + }, + { + name: "user mention with special chars", + input: "<@U123456|user.name>", + expectedCount: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tokens, result := extractSlackTokens(tt.input) + if len(tokens) != tt.expectedCount { + t.Errorf("extractSlackTokens(%q) returned %d tokens, want %d", tt.input, len(tokens), tt.expectedCount) + } + if tt.expectedCount > 0 && result == tt.input { + t.Errorf("extractSlackTokens(%q) should replace tokens with placeholders, but result = %q", tt.input, result) + } + }) + } +} + +func TestExtractCodeBlocks(t *testing.T) { + tests := []struct { + name string + input string + expectedCount int + }{ + { + name: "no code blocks", + input: "hello world", + expectedCount: 0, + }, + { + name: "single code block", + input: "text `code` more", + expectedCount: 0, // inline, not block + }, + { + name: "paired code block fences", + input: "before\n```\ncode\n```\nafter", + expectedCount: 1, + }, + { + name: "multiple code blocks", + input: "```\nblock1\n```\nmiddle\n```\nblock2\n```", + expectedCount: 2, + }, + { + name: "unpaired fence odd", + input: "text\n```\ncode (no closing)", + expectedCount: 0, + }, + { + name: "empty code block", + input: "```\n\n```", + expectedCount: 1, + }, + { + name: "code with language marker", + input: "```python\ncode\n```", + expectedCount: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + blocks, result := extractCodeBlocks(tt.input) + if len(blocks) != tt.expectedCount { + t.Errorf("extractCodeBlocks(%q) returned %d blocks, want %d", tt.input, len(blocks), tt.expectedCount) + } + if tt.expectedCount > 0 && result == tt.input { + t.Errorf("extractCodeBlocks(%q) should replace blocks with placeholders", tt.input) + } + }) + } +} + +func TestExtractInlineCodes(t *testing.T) { + tests := []struct { + name string + input string + expectedCount int + }{ + { + name: "no inline code", + input: "hello world", + expectedCount: 0, + }, + { + name: "single inline code", + input: "use `variable` here", + expectedCount: 1, + }, + { + name: "multiple inline codes", + input: "`var1` and `var2` and `var3`", + expectedCount: 3, + }, + { + name: "unpaired backtick", + input: "use `code", + expectedCount: 0, + }, + { + name: "empty inline code", + input: "use `` here", + expectedCount: 1, + }, + { + name: "with special chars inside", + input: "use `my-var_name` here", + expectedCount: 1, + }, + { + name: "backticks in code block are also extracted", + input: "```\n`inside`\n```", + expectedCount: 4, // extractInlineCodes doesn't know about code blocks + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + codes, result := extractInlineCodes(tt.input) + if len(codes) != tt.expectedCount { + t.Errorf("extractInlineCodes(%q) returned %d codes, want %d", tt.input, len(codes), tt.expectedCount) + } + if tt.expectedCount > 0 && result == tt.input { + t.Errorf("extractInlineCodes(%q) should replace codes with placeholders", tt.input) + } + }) + } +} + +func TestConvertTablesToCodeBlocks(t *testing.T) { + tests := []struct { + name string + input string + expected string + }{ + { + name: "no table", + input: "just text", + expected: "just text", + }, + { + name: "simple table", + input: "| Col1 | Col2 |\n|------|------|\n| A | B |", + expected: "```\n| Col1 | Col2 |\n| A | B |\n```", + }, + { + name: "table with separator stripped", + input: "| Header |\n|--------|", + expected: "```\n| Header |\n```", + }, + { + name: "table at end", + input: "text\n| Col |\n|-----|\n| A |", + expected: "text\n```\n| Col |\n| A |\n```", + }, + { + name: "text after table", + input: "| A |\n|---|\nmore text", + expected: "```\n| A |\n```\nmore text", + }, + { + name: "multiple tables", + input: "| A |\n|---|\ntext\n| B |\n|---|", + expected: "```\n| A |\n```\ntext\n```\n| B |\n```", + }, + { + name: "empty string", + input: "", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := convertTablesToCodeBlocks(tt.input) + if got != tt.expected { + t.Errorf("convertTablesToCodeBlocks(%q) = %q, want %q", tt.input, got, tt.expected) + } + }) + } +} + +func TestHTMLTagsToMarkdown(t *testing.T) { + tests := []struct { + name string + input string + expected string + }{ + { + name: "bold tag", + input: "bold", + expected: "**bold**", + }, + { + name: "strong tag", + input: "strong", + expected: "**strong**", + }, + { + name: "italic tag", + input: "italic", + expected: "_italic_", + }, + { + name: "em tag", + input: "emphasis", + expected: "_emphasis_", + }, + { + name: "strike tag", + input: "struck", + expected: "~~struck~~", + }, + { + name: "del tag", + input: "deleted", + expected: "~~deleted~~", + }, + { + name: "code tag", + input: "code", + expected: "`code`", + }, + { + name: "link tag", + input: "link", + expected: "[link](https://example.com)", + }, + { + name: "br tag", + input: "line1
line2", + expected: "line1\nline2", + }, + { + name: "br self-closing tag", + input: "line1
line2", + expected: "line1\nline2", + }, + { + name: "p tag", + input: "

paragraph

", + expected: "\nparagraph\n", + }, + { + name: "case insensitive", + input: "bold", + expected: "**bold**", + }, + { + name: "multiple tags", + input: "bold and italic", + expected: "**bold** and _italic_", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := htmlTagsToMarkdown(tt.input) + if got != tt.expected { + t.Errorf("htmlTagsToMarkdown(%q) = %q, want %q", tt.input, got, tt.expected) + } + }) + } +} diff --git a/internal/channels/slack/handlers.go b/internal/channels/slack/handlers.go new file mode 100644 index 00000000..180d912b --- /dev/null +++ b/internal/channels/slack/handlers.go @@ -0,0 +1,667 @@ +package slack + +import ( + "bytes" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "time" + + slackapi "github.com/slack-go/slack" + "github.com/slack-go/slack/slackevents" + "github.com/slack-go/slack/socketmode" + + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" +) + +func (c *Channel) handleEventsAPI(evt socketmode.Event) { + eventsAPI, ok := evt.Data.(slackevents.EventsAPIEvent) + if !ok { + return + } + + // Ack immediately (Slack requires ack within ~3s) + c.sm.Ack(*evt.Request) + + switch ev := eventsAPI.InnerEvent.Data.(type) { + case *slackevents.MessageEvent: + c.handleMessage(ev) + case *slackevents.AppMentionEvent: + c.handleAppMention(ev) + } +} + +func (c *Channel) handleMessage(ev *slackevents.MessageEvent) { + // For message_changed: extract user/text from the nested Message field. + // Only process if the edit introduces a new @bot mention. + if ev.SubType == "message_changed" { + if ev.Message == nil { + return + } + // Skip bot's own edits or messages without a user + if ev.Message.User == c.botUserID || ev.Message.User == "" { + return + } + // Only process if the edited message mentions the bot + if !c.isBotMentioned(ev.Message.Text) { + return + } + // Check that the previous version did NOT mention the bot (newly added mention) + if ev.PreviousMessage != nil && c.isBotMentioned(ev.PreviousMessage.Text) { + return + } + // Promote nested fields to top-level for unified processing below + ev.User = ev.Message.User + ev.Text = ev.Message.Text + ev.TimeStamp = ev.Message.Timestamp + ev.ThreadTimeStamp = ev.Message.ThreadTimestamp + } + + if ev.User == c.botUserID || ev.User == "" { + return + } + + // Skip message subtypes (edits, deletes, bot_message, joins, etc.) + // Allow "file_share" and "message_changed" subtypes. + if ev.SubType != "" && ev.SubType != "file_share" && ev.SubType != "message_changed" { + return + } + + // Explicit dedup: prevent duplicate processing on Socket Mode reconnect + dedupKey := ev.Channel + ":" + ev.TimeStamp + if _, loaded := c.dedup.LoadOrStore(dedupKey, time.Now()); loaded { + return + } + + senderID := ev.User + channelID := ev.Channel + content := ev.Text + + isDM := ev.ChannelType == "im" + peerKind := "group" + if isDM { + peerKind = "direct" + } + + // Resolve display name; strip "|" to prevent compound senderID corruption + displayName := strings.ReplaceAll(c.resolveDisplayName(senderID), "|", "_") + compoundSenderID := fmt.Sprintf("%s|%s", senderID, displayName) + + // Policy check + if isDM { + if !c.checkDMPolicy(senderID, channelID) { + return + } + } else { + if !c.checkGroupPolicy(senderID, channelID) { + return + } + } + + // For DMs, apply global allowlist filter (allow_from contains user IDs). + // For groups, skip — group policy already handles channel/user filtering. + if isDM && !c.IsAllowed(compoundSenderID) { + slog.Debug("slack message rejected by allowlist", + "user_id", senderID, "display_name", displayName) + return + } + + // Process file attachments from Slack message + var mediaPaths []string + if ev.Message != nil && len(ev.Message.Files) > 0 { + items, docContent := c.resolveMedia(ev.Message.Files) + + for _, item := range items { + if item.FilePath != "" { + mediaPaths = append(mediaPaths, item.FilePath) + } + } + + // Prepend media tags and document content to message text + mediaTags := buildMediaTags(items) + if mediaTags != "" { + if content != "" { + content = mediaTags + "\n\n" + content + } else { + content = mediaTags + } + } + if docContent != "" { + if content != "" { + content = content + "\n\n" + docContent + } else { + content = docContent + } + } + } + + if content == "" { + return + } + + // Determine local_key and thread context + localKey := channelID + threadTS := ev.ThreadTimeStamp + if !isDM && threadTS != "" { + localKey = fmt.Sprintf("%s:thread:%s", channelID, threadTS) + } + + // Mention gating in groups (with thread participation cache) + if !isDM && c.requireMention { + mentioned := c.isBotMentioned(content) + + // Thread participation cache: auto-reply in threads where bot previously participated + if !mentioned && threadTS != "" && c.threadTTL > 0 { + participKey := channelID + ":particip:" + threadTS + if lastReply, ok := c.threadParticip.Load(participKey); ok { + if time.Since(lastReply.(time.Time)) < c.threadTTL { + mentioned = true + slog.Debug("slack: auto-reply in participated thread", + "channel_id", channelID, "thread_ts", threadTS) + } else { + c.threadParticip.Delete(participKey) + } + } + } + + if !mentioned { + c.groupHistory.Record(localKey, channels.HistoryEntry{ + Sender: displayName, + Body: content, + Timestamp: time.Now(), + MessageID: ev.TimeStamp, + }, c.historyLimit) + + slog.Debug("slack group message recorded (no mention)", + "channel_id", channelID, "user", displayName) + return + } + } + + content = c.stripBotMention(content) + content = strings.TrimSpace(content) + + slog.Debug("slack message received", + "sender_id", senderID, "channel_id", channelID, + "is_dm", isDM, "preview", channels.Truncate(content, 50)) + + // Send "Thinking..." placeholder + replyThreadTS := threadTS + if !isDM && replyThreadTS == "" { + replyThreadTS = ev.TimeStamp // start thread from the triggering message + } + + placeholderOpts := []slackapi.MsgOption{ + slackapi.MsgOptionText("Thinking...", false), + } + if replyThreadTS != "" { + placeholderOpts = append(placeholderOpts, slackapi.MsgOptionTS(replyThreadTS)) + } + + _, placeholderTS, err := c.api.PostMessage(channelID, placeholderOpts...) + if err == nil { + c.placeholders.Store(localKey, placeholderTS) + } + + // Build final content with group history context + finalContent := content + if peerKind == "group" { + annotated := fmt.Sprintf("[From: %s]\n%s", displayName, content) + if c.historyLimit > 0 { + finalContent = c.groupHistory.BuildContext(localKey, annotated, c.historyLimit) + } else { + finalContent = annotated + } + } + + metadata := map[string]string{ + "message_id": ev.TimeStamp, + "user_id": senderID, + "username": displayName, + "channel_id": channelID, + "is_dm": fmt.Sprintf("%t", isDM), + "local_key": localKey, + "placeholder_key": localKey, + } + if replyThreadTS != "" { + metadata["message_thread_id"] = replyThreadTS + } + + // Message debounce: batch rapid messages per-thread + if c.debounceDelay > 0 { + if c.debounceMessage(localKey, compoundSenderID, channelID, finalContent, mediaPaths, metadata, peerKind) { + // Record thread participation even when debounced + if peerKind == "group" && replyThreadTS != "" { + participKey := channelID + ":particip:" + replyThreadTS + c.threadParticip.Store(participKey, time.Now()) + } + return + } + } + + c.HandleMessage(compoundSenderID, channelID, finalContent, mediaPaths, metadata, peerKind) + + // Record thread participation for auto-reply cache + if peerKind == "group" { + if replyThreadTS != "" { + participKey := channelID + ":particip:" + replyThreadTS + c.threadParticip.Store(participKey, time.Now()) + } + c.groupHistory.Clear(localKey) + } +} + +func (c *Channel) handleAppMention(ev *slackevents.AppMentionEvent) { + if ev.User == c.botUserID || ev.User == "" { + return + } + + // Dedup: app_mention may arrive alongside a message event + dedupKey := ev.Channel + ":" + ev.TimeStamp + if _, loaded := c.dedup.LoadOrStore(dedupKey, time.Now()); loaded { + return + } + + // If requireMention is false, message handler already processes all channel messages + if !c.requireMention { + return + } + + senderID := ev.User + channelID := ev.Channel + content := ev.Text + + displayName := strings.ReplaceAll(c.resolveDisplayName(senderID), "|", "_") + compoundSenderID := fmt.Sprintf("%s|%s", senderID, displayName) + + if !c.checkGroupPolicy(senderID, channelID) { + return + } + + content = c.stripBotMention(content) + content = strings.TrimSpace(content) + + if content == "" { + return + } + + localKey := channelID + threadTS := ev.ThreadTimeStamp + if threadTS != "" { + localKey = fmt.Sprintf("%s:thread:%s", channelID, threadTS) + } + + slog.Debug("slack app_mention received", + "sender_id", senderID, "channel_id", channelID, + "preview", channels.Truncate(content, 50)) + + replyThreadTS := threadTS + if replyThreadTS == "" { + replyThreadTS = ev.TimeStamp + } + + placeholderOpts := []slackapi.MsgOption{ + slackapi.MsgOptionText("Thinking...", false), + } + if replyThreadTS != "" { + placeholderOpts = append(placeholderOpts, slackapi.MsgOptionTS(replyThreadTS)) + } + + _, placeholderTS, err := c.api.PostMessage(channelID, placeholderOpts...) + if err == nil { + c.placeholders.Store(localKey, placeholderTS) + } + + annotated := fmt.Sprintf("[From: %s]\n%s", displayName, content) + finalContent := annotated + if c.historyLimit > 0 { + finalContent = c.groupHistory.BuildContext(localKey, annotated, c.historyLimit) + } + + metadata := map[string]string{ + "message_id": ev.TimeStamp, + "user_id": senderID, + "username": displayName, + "channel_id": channelID, + "is_dm": "false", + "local_key": localKey, + "placeholder_key": localKey, + } + if replyThreadTS != "" { + metadata["message_thread_id"] = replyThreadTS + } + + c.HandleMessage(compoundSenderID, channelID, finalContent, nil, metadata, "group") + + // Record thread participation + if replyThreadTS != "" { + participKey := channelID + ":particip:" + replyThreadTS + c.threadParticip.Store(participKey, time.Now()) + } + + c.groupHistory.Clear(localKey) +} + +// isBotMentioned checks if the message text contains <@botUserID>. +func (c *Channel) isBotMentioned(text string) bool { + return strings.Contains(text, "<@"+c.botUserID+">") +} + +// stripBotMention removes <@botUserID> from message text. +func (c *Channel) stripBotMention(text string) string { + return strings.ReplaceAll(text, "<@"+c.botUserID+">", "") +} + +// --- Message debounce/batching --- + +type debounceEntry struct { + timer *time.Timer + messages []string + mu sync.Mutex + senderID string + channelID string + media []string + metadata map[string]string + peerKind string +} + +// debounceMessage batches rapid messages. Returns true if message was debounced. +func (c *Channel) debounceMessage(localKey, senderID, channelID, content string, media []string, metadata map[string]string, peerKind string) bool { + c.debounceMu.Lock() + entry, loaded := c.debounceTimers[localKey] + if !loaded { + entry = &debounceEntry{ + senderID: senderID, + channelID: channelID, + media: media, + metadata: metadata, + peerKind: peerKind, + } + c.debounceTimers[localKey] = entry + } + c.debounceMu.Unlock() + + entry.mu.Lock() + defer entry.mu.Unlock() + + entry.messages = append(entry.messages, content) + if loaded { + // Only append media for subsequent messages; first message's media is set in constructor. + entry.media = append(entry.media, media...) + } + entry.metadata = metadata // use latest message's metadata + + if !loaded { + entry.timer = time.AfterFunc(c.debounceDelay, func() { + c.flushDebounce(localKey) + }) + return true + } + + if entry.timer != nil { + entry.timer.Reset(c.debounceDelay) + } + return true +} + +func (c *Channel) flushDebounce(localKey string) { + c.debounceMu.Lock() + entry, ok := c.debounceTimers[localKey] + if ok { + delete(c.debounceTimers, localKey) + } + c.debounceMu.Unlock() + + if !ok { + return + } + + entry.mu.Lock() + combined := strings.Join(entry.messages, "\n") + entry.mu.Unlock() + + c.HandleMessage(entry.senderID, entry.channelID, combined, entry.media, entry.metadata, entry.peerKind) + + if entry.peerKind == "group" { + c.groupHistory.Clear(localKey) + } +} + +// --- Policy checks --- + +func (c *Channel) checkDMPolicy(senderID, channelID string) bool { + dmPolicy := c.config.DMPolicy + if dmPolicy == "" { + dmPolicy = "pairing" + } + + switch dmPolicy { + case "disabled": + return false + case "open": + return true + case "allowlist": + return c.HasAllowList() && c.IsAllowed(senderID) + default: // "pairing" + if c.pairingService != nil && c.pairingService.IsPaired(senderID, c.Name()) { + return true + } + if c.HasAllowList() && c.IsAllowed(senderID) { + return true + } + c.sendPairingReply(senderID, channelID) + return false + } +} + +func (c *Channel) checkGroupPolicy(senderID, channelID string) bool { + groupPolicy := c.config.GroupPolicy + if groupPolicy == "" { + groupPolicy = "open" + } + + switch groupPolicy { + case "disabled": + return false + case "allowlist": + if !c.HasAllowList() { + return false + } + // Allow if user ID or channel ID is in the allowlist + return c.IsAllowed(senderID) || c.IsAllowed(channelID) + case "pairing": + if c.HasAllowList() && c.IsAllowed(senderID) { + return true + } + if _, cached := c.approvedGroups.Load(channelID); cached { + return true + } + groupSenderID := fmt.Sprintf("group:%s", channelID) + if c.pairingService != nil && c.pairingService.IsPaired(groupSenderID, c.Name()) { + c.approvedGroups.Store(channelID, true) + return true + } + c.sendPairingReply(groupSenderID, channelID) + return false + default: // "open" + return true + } +} + +func (c *Channel) sendPairingReply(senderID, channelID string) { + if c.pairingService == nil { + return + } + + if lastSent, ok := c.pairingDebounce.Load(senderID); ok { + if time.Since(lastSent.(time.Time)) < pairingDebounceTime { + return + } + } + + code, err := c.pairingService.RequestPairing(senderID, c.Name(), channelID, "default", nil) + if err != nil { + slog.Warn("slack: failed to request pairing code", "error", err) + return + } + + // Security: do not expose pairing code in group channels (visible to all members). + // Instead, direct admin to CLI or web UI where pending codes are listed. + var msg string + if strings.HasPrefix(senderID, "group:") { + msg = fmt.Sprintf("This channel is not authorized to use this bot.\n\n"+ + "An admin can approve via CLI:\n goclaw pairing approve %s\n\n"+ + "Or approve via the GoClaw web UI (Pairing section).", code) + } else { + msg = fmt.Sprintf("GoClaw: access not configured.\n\nYour Slack user ID: %s\n\nPairing code: %s\n\nAsk the bot owner to approve with:\n goclaw pairing approve %s", + senderID, code, code) + } + if _, _, err := c.api.PostMessage(channelID, slackapi.MsgOptionText(msg, false)); err != nil { + slog.Warn("slack: failed to send pairing reply", + "channel_id", channelID, "error", err) + } + c.pairingDebounce.Store(senderID, time.Now()) +} + +// --- File download (SSRF-protected) --- + +var slackDownloadAllowlist = []string{ + ".slack.com", + ".slack-edge.com", + ".slack-files.com", +} + +func isAllowedDownloadHost(rawURL string) bool { + u, err := url.Parse(rawURL) + if err != nil || u.Scheme != "https" { + return false + } + host := strings.ToLower(u.Hostname()) + for _, suffix := range slackDownloadAllowlist { + if strings.HasSuffix(host, suffix) { + return true + } + } + return false +} + +func (c *Channel) downloadFile(name, urlPrivate, urlPrivateDownload string, maxBytes int64) (string, error) { + downloadURL := urlPrivateDownload + if downloadURL == "" { + downloadURL = urlPrivate + } + if downloadURL == "" { + return "", fmt.Errorf("no download URL for file %s", name) + } + + if !isAllowedDownloadHost(downloadURL) { + return "", fmt.Errorf("security: download URL hostname not in Slack allowlist: %s", downloadURL) + } + + ext := filepath.Ext(name) + if ext == "" { + ext = ".dat" + } + tmpFile, err := os.CreateTemp("", "slack-file-*"+ext) + if err != nil { + return "", fmt.Errorf("create temp file: %w", err) + } + defer tmpFile.Close() + + client := &http.Client{ + Timeout: 60 * time.Second, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + req.Header.Del("Authorization") // strip auth on redirect (CDN presigned URL) + if req.URL.Scheme != "https" { + return fmt.Errorf("security: redirect to non-HTTPS URL blocked: %s", req.URL) + } + // Only allow redirects to known Slack CDN domains to prevent SSRF. + host := req.URL.Hostname() + if !isAllowedSlackHost(host) { + return fmt.Errorf("security: redirect to untrusted host blocked: %s", host) + } + if len(via) >= 3 { + return fmt.Errorf("too many redirects") + } + return nil + }, + } + + req, err := http.NewRequest("GET", downloadURL, nil) + if err != nil { + os.Remove(tmpFile.Name()) + return "", err + } + req.Header.Set("Authorization", "Bearer "+c.config.BotToken) + + resp, err := client.Do(req) + if err != nil { + os.Remove(tmpFile.Name()) + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + os.Remove(tmpFile.Name()) + return "", fmt.Errorf("download failed: HTTP %d", resp.StatusCode) + } + + if _, err := io.Copy(tmpFile, io.LimitReader(resp.Body, maxBytes)); err != nil { + os.Remove(tmpFile.Name()) + return "", err + } + + return tmpFile.Name(), nil +} + +// allowedSlackHosts contains trusted Slack CDN domains for redirect validation. +var allowedSlackHosts = []string{ + ".slack-edge.com", + ".slack.com", + "files.slack.com", +} + +// isAllowedSlackHost checks if a hostname belongs to a known Slack CDN domain. +func isAllowedSlackHost(host string) bool { + for _, suffix := range allowedSlackHosts { + if host == strings.TrimPrefix(suffix, ".") || strings.HasSuffix(host, suffix) { + return true + } + } + return false +} + +// --- File upload (v2 3-step API) --- + +func (c *Channel) uploadFile(channelID, threadTS string, media bus.MediaAttachment) error { + filePath := media.URL + fileName := filepath.Base(filePath) + + data, err := os.ReadFile(filePath) + if err != nil { + return fmt.Errorf("read file %s: %w", filePath, err) + } + + params := slackapi.UploadFileParameters{ + Filename: fileName, + FileSize: len(data), + Reader: bytes.NewReader(data), + Title: fileName, + Channel: channelID, + ThreadTimestamp: threadTS, + } + + _, err = c.api.UploadFile(params) + if err != nil { + return fmt.Errorf("upload file: %w", err) + } + + return nil +} diff --git a/internal/channels/slack/handlers_test.go b/internal/channels/slack/handlers_test.go new file mode 100644 index 00000000..b7648c48 --- /dev/null +++ b/internal/channels/slack/handlers_test.go @@ -0,0 +1,118 @@ +package slack + +import ( + "testing" +) + +func TestIsAllowedDownloadHost(t *testing.T) { + tests := []struct { + name string + rawURL string + expected bool + }{ + { + name: "valid slack.com domain", + rawURL: "https://files.slack.com/path/to/file", + expected: true, + }, + { + name: "valid slack-edge.com domain", + rawURL: "https://a.b.slack-edge.com/x/y/z", + expected: true, + }, + { + name: "valid slack-files.com domain", + rawURL: "https://files.slack-files.com/path/to/file", + expected: true, + }, + { + name: "http instead of https", + rawURL: "http://files.slack.com/path", + expected: false, + }, + { + name: "non-slack domain", + rawURL: "https://evil.com/malicious", + expected: false, + }, + { + name: "slack in subdomain but different tld", + rawURL: "https://files.slack.org/path", + expected: false, + }, + { + name: "slack.com but with subdomain not in allowlist", + rawURL: "https://notfiles.slack.com/path", + expected: true, // matches .slack.com suffix + }, + { + name: "slack-edge.com variations", + rawURL: "https://cdn.slack-edge.com/file", + expected: true, + }, + { + name: "empty string", + rawURL: "", + expected: false, + }, + { + name: "invalid url", + rawURL: "not a url at all", + expected: false, + }, + { + name: "hostname with special chars", + rawURL: "https://a]b.slack-edge.com/x", + expected: true, // Go's URL parser handles this, suffix matches + }, + { + name: "domain with slack.com.evil.com", + rawURL: "https://slack.com.evil.com/path", + expected: false, + }, + { + name: "ftp scheme", + rawURL: "ftp://files.slack.com/path", + expected: false, + }, + { + name: "case insensitive hostname", + rawURL: "https://FILES.SLACK.COM/path", + expected: true, + }, + { + name: "multiple subdomains", + rawURL: "https://a.b.c.slack-files.com/file", + expected: true, + }, + { + name: "hostname with port", + rawURL: "https://files.slack.com:443/path", + expected: true, + }, + { + name: "url with query params", + rawURL: "https://files.slack.com/file?token=abc&download=true", + expected: true, + }, + { + name: "url with fragment", + rawURL: "https://files.slack.com/file#section", + expected: true, + }, + { + name: "very long subdomain chain", + rawURL: "https://very.long.subdomain.chain.slack.com/file", + expected: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isAllowedDownloadHost(tt.rawURL) + if got != tt.expected { + t.Errorf("isAllowedDownloadHost(%q) = %v, want %v", tt.rawURL, got, tt.expected) + } + }) + } +} diff --git a/internal/channels/slack/media.go b/internal/channels/slack/media.go new file mode 100644 index 00000000..47756635 --- /dev/null +++ b/internal/channels/slack/media.go @@ -0,0 +1,99 @@ +package slack + +import ( + "fmt" + "log/slog" + "strings" + + slackapi "github.com/slack-go/slack" + + "github.com/nextlevelbuilder/goclaw/internal/channels/media" +) + +const defaultMediaMaxBytes int64 = 20 * 1024 * 1024 // 20MB + +// mediaItem represents a downloaded file from a Slack message. +type mediaItem struct { + Type string // "image", "audio", "document" + FilePath string // local temp file path + FileName string // original filename + ContentType string // MIME type +} + +// resolveMedia downloads and classifies files attached to a Slack message. +// Returns media items and any extra content (extracted document text). +func (c *Channel) resolveMedia(files []slackapi.File) (items []mediaItem, extraContent string) { + maxBytes := c.config.MediaMaxBytes + if maxBytes == 0 { + maxBytes = defaultMediaMaxBytes + } + + for _, f := range files { + // Skip files exceeding size limit before download + if int64(f.Size) > maxBytes { + slog.Warn("slack: file too large, skipping", + "file", f.Name, "size", f.Size, "max", maxBytes) + continue + } + + mtype := classifyMime(f.Mimetype) + + filePath, err := c.downloadFile(f.Name, f.URLPrivate, f.URLPrivateDownload, maxBytes) + if err != nil { + slog.Warn("slack: file download failed", + "file", f.Name, "error", err) + continue + } + + items = append(items, mediaItem{ + Type: mtype, + FilePath: filePath, + FileName: f.Name, + ContentType: f.Mimetype, + }) + + // Extract text from document files + if mtype == "document" { + docContent, err := media.ExtractDocumentContent(filePath, f.Name) + if err != nil { + slog.Warn("slack: document extraction failed", + "file", f.Name, "error", err) + continue + } + if extraContent != "" { + extraContent += "\n" + } + extraContent += docContent + } + } + + return items, extraContent +} + +// classifyMime maps a MIME type to a media category. +func classifyMime(mime string) string { + switch { + case strings.HasPrefix(mime, "image/"): + return "image" + case strings.HasPrefix(mime, "audio/"): + return "audio" + default: + return "document" + } +} + +// buildMediaTags generates content tags for media items. +func buildMediaTags(items []mediaItem) string { + var tags []string + for _, m := range items { + switch m.Type { + case "image": + tags = append(tags, "") + case "audio": + tags = append(tags, "") + case "document": + tags = append(tags, fmt.Sprintf("", m.FileName)) + } + } + return strings.Join(tags, "\n") +} diff --git a/internal/channels/slack/reactions.go b/internal/channels/slack/reactions.go new file mode 100644 index 00000000..4f6294cb --- /dev/null +++ b/internal/channels/slack/reactions.go @@ -0,0 +1,99 @@ +package slack + +import ( + "context" + "log/slog" + "sync" + "time" + + slackapi "github.com/slack-go/slack" +) + +const reactionDebounceInterval = 700 * time.Millisecond + +// statusEmoji maps GoClaw agent status to Slack emoji names (without colons). +var statusEmoji = map[string]string{ + "thinking": "thinking_face", + "tool": "hammer_and_wrench", + "done": "white_check_mark", + "error": "x", + "stall": "hourglass_flowing_sand", +} + +// reactionState tracks per-message reaction state. +type reactionState struct { + currentEmoji string + lastUpdate time.Time + mu sync.Mutex +} + +// OnReactionEvent adds a status emoji reaction to the user's message. +func (c *Channel) OnReactionEvent(_ context.Context, chatID string, messageID string, status string) error { + if c.config.ReactionLevel == "" || c.config.ReactionLevel == "off" { + return nil + } + + emoji, ok := statusEmoji[status] + if !ok { + return nil + } + + // For "minimal" level, only show thinking and done + if c.config.ReactionLevel == "minimal" && status != "thinking" && status != "done" { + return nil + } + + channelID := extractChannelID(chatID) + + stateKey := chatID + ":" + messageID + stateVal, _ := c.reactions.LoadOrStore(stateKey, &reactionState{}) + st := stateVal.(*reactionState) + + st.mu.Lock() + defer st.mu.Unlock() + + if time.Since(st.lastUpdate) < reactionDebounceInterval { + return nil + } + + // Remove previous reaction (if different) + if st.currentEmoji != "" && st.currentEmoji != emoji { + if err := c.api.RemoveReaction(st.currentEmoji, + slackapi.ItemRef{Channel: channelID, Timestamp: messageID}); err != nil { + slog.Debug("slack: remove reaction failed", "emoji", st.currentEmoji, "error", err) + } + } + + if err := c.api.AddReaction(emoji, + slackapi.ItemRef{Channel: channelID, Timestamp: messageID}); err != nil { + slog.Debug("slack: add reaction failed", "emoji", emoji, "error", err) + return nil + } + + st.currentEmoji = emoji + st.lastUpdate = time.Now() + return nil +} + +// ClearReaction removes the current status emoji from a message. +func (c *Channel) ClearReaction(_ context.Context, chatID string, messageID string) error { + stateKey := chatID + ":" + messageID + stateVal, ok := c.reactions.LoadAndDelete(stateKey) + if !ok { + return nil + } + + st := stateVal.(*reactionState) + st.mu.Lock() + defer st.mu.Unlock() + + if st.currentEmoji != "" { + channelID := extractChannelID(chatID) + if err := c.api.RemoveReaction(st.currentEmoji, + slackapi.ItemRef{Channel: channelID, Timestamp: messageID}); err != nil { + slog.Debug("slack: clear reaction failed", "emoji", st.currentEmoji, "error", err) + } + } + + return nil +} diff --git a/internal/channels/slack/stream.go b/internal/channels/slack/stream.go new file mode 100644 index 00000000..f8b8836a --- /dev/null +++ b/internal/channels/slack/stream.go @@ -0,0 +1,133 @@ +package slack + +import ( + "context" + "log/slog" + "strings" + "sync" + "time" + + slackapi "github.com/slack-go/slack" +) + +const streamThrottleInterval = 1000 * time.Millisecond + +// streamState tracks per-chat streaming state. +type streamState struct { + channelID string + threadTS string + msgTS string // placeholder message timestamp + lastUpdate time.Time // last chat.update call + mu sync.Mutex +} + +// StreamEnabled reports whether streaming is active for DMs or groups. +func (c *Channel) StreamEnabled(isGroup bool) bool { + if isGroup { + return c.config.GroupStream != nil && *c.config.GroupStream + } + return c.config.DMStream != nil && *c.config.DMStream +} + +// OnStreamStart is called when a new agent run starts. +// The placeholder "Thinking..." was already sent in handleMessage. +func (c *Channel) OnStreamStart(_ context.Context, _ string) error { + return nil +} + +// OnChunkEvent edits the placeholder with accumulated text, throttled to avoid rate limits. +func (c *Channel) OnChunkEvent(_ context.Context, chatID string, fullText string) error { + stateVal, ok := c.streams.Load(chatID) + if !ok { + pTS, pOK := c.placeholders.Load(chatID) + if !pOK { + return nil + } + + st := &streamState{ + channelID: extractChannelID(chatID), + msgTS: pTS.(string), + } + if threadTS := extractThreadTS(chatID); threadTS != "" { + st.threadTS = threadTS + } + c.streams.Store(chatID, st) + stateVal = st + } + + st := stateVal.(*streamState) + st.mu.Lock() + defer st.mu.Unlock() + + if time.Since(st.lastUpdate) < streamThrottleInterval { + return nil + } + + formatted := markdownToSlackMrkdwn(fullText) + if len(formatted) > maxMessageLen { + formatted = formatted[:maxMessageLen] + "..." + } + + opts := []slackapi.MsgOption{slackapi.MsgOptionText(formatted, false)} + _, _, _, err := c.api.UpdateMessage(st.channelID, st.msgTS, opts...) + if err != nil { + slog.Debug("slack stream chunk update failed", "error", err) + return nil + } + + st.lastUpdate = time.Now() + return nil +} + +// OnStreamEnd sends final formatted text. +func (c *Channel) OnStreamEnd(_ context.Context, chatID string, finalText string) error { + if finalText == "" { + c.streams.Delete(chatID) + return nil + } + + channelID := extractChannelID(chatID) + var msgTS, threadTS string + + if stateVal, ok := c.streams.Load(chatID); ok { + st := stateVal.(*streamState) + msgTS = st.msgTS + threadTS = st.threadTS + } else if pTS, ok := c.placeholders.Load(chatID); ok { + msgTS = pTS.(string) + } + + defer c.streams.Delete(chatID) + + if msgTS == "" { + return nil + } + + formatted := markdownToSlackMrkdwn(finalText) + if len(formatted) <= maxMessageLen { + _, _, _, err := c.api.UpdateMessage(channelID, msgTS, slackapi.MsgOptionText(formatted, false)) + return err + } + + // Final text exceeds limit -- update placeholder with first chunk, send rest as follow-ups + first, rest := splitAtLimit(formatted, maxMessageLen) + _, _, _, _ = c.api.UpdateMessage(channelID, msgTS, slackapi.MsgOptionText(first, false)) + return c.sendChunked(channelID, rest, threadTS) +} + +// extractChannelID gets the channel ID from a local_key. +func extractChannelID(localKey string) string { + if idx := strings.Index(localKey, ":thread:"); idx > 0 { + return localKey[:idx] + } + return localKey +} + +// extractThreadTS gets the thread_ts from a local_key, or "" if not threaded. +func extractThreadTS(localKey string) string { + const prefix = ":thread:" + if idx := strings.Index(localKey, prefix); idx > 0 { + return localKey[idx+len(prefix):] + } + return "" +} diff --git a/internal/channels/slack/stream_test.go b/internal/channels/slack/stream_test.go new file mode 100644 index 00000000..362e2a61 --- /dev/null +++ b/internal/channels/slack/stream_test.go @@ -0,0 +1,121 @@ +package slack + +import ( + "testing" +) + +func TestExtractChannelID(t *testing.T) { + tests := []struct { + name string + localKey string + expected string + }{ + { + name: "plain channel id", + localKey: "C123456", + expected: "C123456", + }, + { + name: "threaded message", + localKey: "C123456:thread:1234.5678", + expected: "C123456", + }, + { + name: "threaded with different ts format", + localKey: "C999:thread:999999.999999", + expected: "C999", + }, + { + name: "no thread marker", + localKey: "C123456789", + expected: "C123456789", + }, + { + name: "empty string", + localKey: "", + expected: "", + }, + { + name: "only thread marker", + localKey: ":thread:1234.5678", + expected: ":thread:1234.5678", + }, + { + name: "colon but no thread text", + localKey: "C123:thread:", + expected: "C123", + }, + { + name: "multiple colons", + localKey: "C123:thread:1234:5678", + expected: "C123", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := extractChannelID(tt.localKey) + if got != tt.expected { + t.Errorf("extractChannelID(%q) = %q, want %q", tt.localKey, got, tt.expected) + } + }) + } +} + +func TestExtractThreadTS(t *testing.T) { + tests := []struct { + name string + localKey string + expected string + }{ + { + name: "threaded message", + localKey: "C123456:thread:1234.5678", + expected: "1234.5678", + }, + { + name: "plain channel id", + localKey: "C123456", + expected: "", + }, + { + name: "empty string", + localKey: "", + expected: "", + }, + { + name: "only thread marker (no channel id)", + localKey: ":thread:1234.5678", + expected: "", // idx must be > 0, so this returns "" + }, + { + name: "thread marker but empty ts", + localKey: "C123:thread:", + expected: "", + }, + { + name: "multiple colons after thread", + localKey: "C123:thread:1234:5678:extra", + expected: "1234:5678:extra", + }, + { + name: "thread marker not found", + localKey: "C123:other:1234.5678", + expected: "", + }, + { + name: "thread marker case sensitive", + localKey: "C123:THREAD:1234.5678", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := extractThreadTS(tt.localKey) + if got != tt.expected { + t.Errorf("extractThreadTS(%q) = %q, want %q", tt.localKey, got, tt.expected) + } + }) + } +} diff --git a/internal/config/config_channels.go b/internal/config/config_channels.go index 4650f7b4..7b64a939 100644 --- a/internal/config/config_channels.go +++ b/internal/config/config_channels.go @@ -88,12 +88,22 @@ type DiscordConfig struct { type SlackConfig struct { Enabled bool `json:"enabled"` - BotToken string `json:"bot_token"` - AppToken string `json:"app_token"` + BotToken string `json:"bot_token"` // xoxb-... (Bot User OAuth Token) + AppToken string `json:"app_token"` // xapp-... (App-Level Token for Socket Mode) + UserToken string `json:"user_token,omitempty"` // xoxp-... (Optional: custom bot identity) AllowFrom FlexibleStringSlice `json:"allow_from"` - DMPolicy string `json:"dm_policy,omitempty"` // "open" (default), "allowlist", "disabled" - GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "allowlist", "disabled" - RequireMention bool `json:"require_mention,omitempty"` // only respond to @bot in channels (default true) + DMPolicy string `json:"dm_policy,omitempty"` // "pairing" (default), "allowlist", "open", "disabled" + GroupPolicy string `json:"group_policy,omitempty"` // "open" (default), "pairing", "allowlist", "disabled" + RequireMention *bool `json:"require_mention,omitempty"` // require @bot mention in channels (default true) + HistoryLimit int `json:"history_limit,omitempty"` // max pending group messages for context (default 50, 0=disabled) + DMStream *bool `json:"dm_stream,omitempty"` // enable streaming for DMs (default false) + GroupStream *bool `json:"group_stream,omitempty"` // enable streaming for groups (default false) + NativeStream *bool `json:"native_stream,omitempty"` // use Slack ChatStreamer API if available (default false) + ReactionLevel string `json:"reaction_level,omitempty"` // "off" (default), "minimal", "full" + BlockReply *bool `json:"block_reply,omitempty"` // override gateway block_reply (nil = inherit) + DebounceDelay int `json:"debounce_delay,omitempty"` // ms delay before dispatching rapid messages (default 300, 0=disabled) + ThreadTTL *int `json:"thread_ttl,omitempty"` // hours before thread participation expires (default 24, 0=disabled — always require @mention) + MediaMaxBytes int64 `json:"media_max_bytes,omitempty"` // max file download size in bytes (default 20MB) } type WhatsAppConfig struct { diff --git a/internal/config/config_load.go b/internal/config/config_load.go index 233e8d4d..32b17236 100644 --- a/internal/config/config_load.go +++ b/internal/config/config_load.go @@ -119,6 +119,9 @@ func (c *Config) applyEnvOverrides() { envStr("GOCLAW_LARK_ENCRYPT_KEY", &c.Channels.Feishu.EncryptKey) envStr("GOCLAW_LARK_VERIFICATION_TOKEN", &c.Channels.Feishu.VerificationToken) envStr("GOCLAW_WHATSAPP_BRIDGE_URL", &c.Channels.WhatsApp.BridgeURL) + envStr("GOCLAW_SLACK_BOT_TOKEN", &c.Channels.Slack.BotToken) + envStr("GOCLAW_SLACK_APP_TOKEN", &c.Channels.Slack.AppToken) + envStr("GOCLAW_SLACK_USER_TOKEN", &c.Channels.Slack.UserToken) // TTS secrets envStr("GOCLAW_TTS_OPENAI_API_KEY", &c.Tts.OpenAI.APIKey) @@ -142,6 +145,9 @@ func (c *Config) applyEnvOverrides() { if c.Channels.WhatsApp.BridgeURL != "" { c.Channels.WhatsApp.Enabled = true } + if c.Channels.Slack.BotToken != "" && c.Channels.Slack.AppToken != "" { + c.Channels.Slack.Enabled = true + } // Claude CLI provider envStr("GOCLAW_CLAUDE_CLI_PATH", &c.Providers.ClaudeCLI.CLIPath) diff --git a/internal/gateway/methods/channel_instances.go b/internal/gateway/methods/channel_instances.go index 2d139028..6b64d2aa 100644 --- a/internal/gateway/methods/channel_instances.go +++ b/internal/gateway/methods/channel_instances.go @@ -245,7 +245,7 @@ func maskInstance(inst store.ChannelInstanceData) map[string]interface{} { // isValidChannelType checks if the channel type is supported. func isValidChannelType(ct string) bool { switch ct { - case "telegram", "discord", "whatsapp", "zalo_oa", "zalo_personal", "feishu": + case "telegram", "discord", "slack", "whatsapp", "zalo_oa", "zalo_personal", "feishu": return true } return false diff --git a/internal/http/channel_instances.go b/internal/http/channel_instances.go index f85d763d..c532bc0b 100644 --- a/internal/http/channel_instances.go +++ b/internal/http/channel_instances.go @@ -378,7 +378,7 @@ func (h *ChannelInstancesHandler) handleRemoveWriter(w http.ResponseWriter, r *h // isValidChannelType checks if the channel type is supported. func isValidChannelType(ct string) bool { switch ct { - case "telegram", "discord", "whatsapp", "zalo_oa", "zalo_personal", "feishu": + case "telegram", "discord", "slack", "whatsapp", "zalo_oa", "zalo_personal", "feishu": return true } return false diff --git a/ui/web/src/constants/channels.ts b/ui/web/src/constants/channels.ts index 50a8bff8..ba504285 100644 --- a/ui/web/src/constants/channels.ts +++ b/ui/web/src/constants/channels.ts @@ -1,6 +1,7 @@ export const CHANNEL_TYPES = [ { value: "telegram", label: "Telegram" }, { value: "discord", label: "Discord" }, + { value: "slack", label: "Slack" }, { value: "feishu", label: "Feishu / Lark" }, { value: "zalo_oa", label: "Zalo OA" }, { value: "zalo_personal", label: "Zalo Personal" }, diff --git a/ui/web/src/pages/channels/channel-schemas.ts b/ui/web/src/pages/channels/channel-schemas.ts index e6eda621..4d62914c 100644 --- a/ui/web/src/pages/channels/channel-schemas.ts +++ b/ui/web/src/pages/channels/channel-schemas.ts @@ -44,6 +44,11 @@ export const credentialsSchema: Record = { discord: [ { key: "token", label: "Bot Token", type: "password", required: true, placeholder: "Discord bot token" }, ], + slack: [ + { key: "bot_token", label: "Bot Token", type: "password", required: true, placeholder: "xoxb-...", help: "Bot User OAuth Token from your Slack app's OAuth & Permissions page" }, + { key: "app_token", label: "App-Level Token", type: "password", required: true, placeholder: "xapp-...", help: "App-Level Token with connections:write scope (required for Socket Mode)" }, + { key: "user_token", label: "User Token (Optional)", type: "password", required: false, placeholder: "xoxp-...", help: "Optional: User OAuth Token for custom bot identity. Leave empty to use default bot identity." }, + ], feishu: [ { key: "app_id", label: "App ID", type: "text", required: true, placeholder: "cli_xxxxx" }, { key: "app_secret", label: "App Secret", type: "password", required: true }, @@ -84,6 +89,20 @@ export const configSchema: Record = { { key: "allow_from", label: "Allowed Users", type: "tags", help: "Discord user IDs" }, { key: "block_reply", label: "Block Reply", type: "select", options: blockReplyOptions, defaultValue: "inherit", help: "Deliver intermediate text during tool iterations" }, ], + slack: [ + { key: "dm_policy", label: "DM Policy", type: "select", options: dmPolicyOptions, defaultValue: "pairing", help: "How to handle direct messages from unknown users" }, + { key: "group_policy", label: "Group Policy", type: "select", options: groupPolicyOptions, defaultValue: "pairing", help: "How to handle messages from channels/groups" }, + { key: "require_mention", label: "Require @mention in channels", type: "boolean", defaultValue: true, help: "Bot only responds when explicitly @mentioned in channels (recommended)" }, + { key: "history_limit", label: "Group History Limit", type: "number", defaultValue: 50, help: "Max pending group messages for context (0 = disabled)" }, + { key: "dm_stream", label: "DM Streaming", type: "boolean", defaultValue: false, help: "Progressively edit placeholder message as LLM generates (DMs)" }, + { key: "group_stream", label: "Group Streaming", type: "boolean", defaultValue: false, help: "Progressively edit placeholder message as LLM generates (channels)" }, + { key: "native_stream", label: "Native Streaming (Agents & AI Apps)", type: "boolean", defaultValue: false, help: "Use Slack's ChatStreamer API for native streaming. Falls back to edit-in-place if unavailable." }, + { key: "debounce_delay", label: "Debounce Delay (ms)", type: "number", defaultValue: 300, help: "Milliseconds to wait before dispatching rapid messages. Set 0 to disable." }, + { key: "thread_ttl", label: "Thread Participation TTL (hours)", type: "number", defaultValue: 24, help: "Hours before bot stops auto-replying in threads it participated in. 0 = always require @mention." }, + { key: "reaction_level", label: "Reaction Level", type: "select", options: [{ value: "off", label: "Off" }, { value: "minimal", label: "Minimal (thinking + done)" }, { value: "full", label: "Full (all status emoji)" }], defaultValue: "off", help: "Show emoji reactions on user messages during agent processing" }, + { key: "allow_from", label: "Allowed Users", type: "tags", help: "Slack user IDs (U...) allowed to interact; empty = no allowlist filter" }, + { key: "block_reply", label: "Block Reply", type: "select", options: blockReplyOptions, defaultValue: "inherit", help: "Deliver intermediate text during tool iterations" }, + ], feishu: [ { key: "domain", label: "Domain", type: "select", options: [{ value: "lark", label: "Lark (global) — webhook only" }, { value: "feishu", label: "Feishu (China)" }], defaultValue: "lark" }, { key: "connection_mode", label: "Connection Mode", type: "select", options: [{ value: "webhook", label: "Webhook" }, { value: "websocket", label: "WebSocket (Feishu only)" }], defaultValue: "webhook", help: "Lark Global only supports Webhook" }, diff --git a/ui/web/src/pages/channels/channels-status-view.tsx b/ui/web/src/pages/channels/channels-status-view.tsx index e55246f0..baffa3aa 100644 --- a/ui/web/src/pages/channels/channels-status-view.tsx +++ b/ui/web/src/pages/channels/channels-status-view.tsx @@ -10,6 +10,7 @@ import type { ChannelStatus } from "./hooks/use-channels"; const channelTypeLabels: Record = { telegram: "Telegram", discord: "Discord", + slack: "Slack", feishu: "Feishu / Lark", zalo_oa: "Zalo OA", zalo_personal: "Zalo Personal",