From 23f1957c56a9cb8a06ab6efd6aef94f4d84b2cf8 Mon Sep 17 00:00:00 2001 From: viettranx Date: Tue, 10 Mar 2026 18:45:48 +0700 Subject: [PATCH] feat(channels): wire contact auto-collector across all channel handlers Pass contactCollector through channel manager to all channel handlers (Telegram, Discord, Feishu, Slack, Zalo) so contacts are automatically collected when users interact with the agent. --- cmd/gateway.go | 1 + internal/channels/channel.go | 20 +++++++++----- internal/channels/discord/handler.go | 5 ++++ internal/channels/feishu/bot.go | 5 ++++ internal/channels/manager.go | 30 +++++++++++++++++---- internal/channels/slack/handlers.go | 6 +++++ internal/channels/telegram/handlers.go | 5 ++++ internal/channels/zalo/personal/handlers.go | 6 +++++ 8 files changed, 67 insertions(+), 11 deletions(-) diff --git a/cmd/gateway.go b/cmd/gateway.go index b38e1bd1..4a6c0fd8 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -874,6 +874,7 @@ func runGateway() { var contactCollector *store.ContactCollector if pgStores.Contacts != nil { contactCollector = store.NewContactCollector(pgStores.Contacts, cache.NewInMemoryCache[bool]()) + channelMgr.SetContactCollector(contactCollector) // propagate to all channel handlers } go consumeInboundMessages(ctx, msgBus, agentRouter, cfg, sched, channelMgr, consumerTeamStore, quotaChecker, delegateMgr, pgStores.Sessions, pgStores.Agents, contactCollector) diff --git a/internal/channels/channel.go b/internal/channels/channel.go index a7df05b7..def6a188 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -15,6 +15,7 @@ import ( "strings" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/store" ) // InternalChannels are system channels excluded from outbound dispatch. @@ -134,12 +135,13 @@ type ReactionChannel interface { // BaseChannel provides shared functionality for all channel implementations. // Channel implementations should embed this struct. type BaseChannel struct { - name string - channelType string // platform type; defaults to name if unset - bus *bus.MessageBus - running bool - allowList []string - agentID string // for DB instances: routes to specific agent (empty = use resolveAgentRoute) + name string + channelType string // platform type; defaults to name if unset + bus *bus.MessageBus + running bool + allowList []string + agentID string // for DB instances: routes to specific agent (empty = use resolveAgentRoute) + contactCollector *store.ContactCollector // optional: auto-collect contacts from channel messages } // NewBaseChannel creates a new BaseChannel with the given parameters. @@ -174,6 +176,12 @@ func (c *BaseChannel) AgentID() string { return c.agentID } // SetAgentID sets the explicit agent ID for routing (used by InstanceLoader for DB instances). func (c *BaseChannel) SetAgentID(id string) { c.agentID = id } +// SetContactCollector sets the contact collector for auto-collecting contacts from messages. +func (c *BaseChannel) SetContactCollector(cc *store.ContactCollector) { c.contactCollector = cc } + +// ContactCollector returns the contact collector (may be nil). +func (c *BaseChannel) ContactCollector() *store.ContactCollector { return c.contactCollector } + // IsRunning returns whether the channel is running. func (c *BaseChannel) IsRunning() bool { return c.running } diff --git a/internal/channels/discord/handler.go b/internal/channels/discord/handler.go index 05e67e07..c0a60dbe 100644 --- a/internal/channels/discord/handler.go +++ b/internal/channels/discord/handler.go @@ -147,6 +147,11 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate MessageID: m.ID, }, c.historyLimit) + // Collect contact even when bot is not mentioned (cache prevents DB spam). + if cc := c.ContactCollector(); cc != nil { + cc.EnsureContact(context.Background(), c.Type(), c.Name(), senderID, senderID, senderName, m.Author.Username, "group") + } + slog.Debug("discord group message recorded (no mention)", "channel_id", channelID, "user_id", senderID, diff --git a/internal/channels/feishu/bot.go b/internal/channels/feishu/bot.go index d9ea1d50..ca9dc5ec 100644 --- a/internal/channels/feishu/bot.go +++ b/internal/channels/feishu/bot.go @@ -85,6 +85,11 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) { MessageID: messageID, }, c.historyLimit) + // Collect contact even when bot is not mentioned (cache prevents DB spam). + if cc := c.ContactCollector(); cc != nil { + cc.EnsureContact(ctx, c.Type(), c.Name(), mc.SenderID, mc.SenderID, senderName, "", "group") + } + slog.Debug("feishu group message recorded (no mention)", "chat_id", mc.ChatID, "sender", senderName, ) diff --git a/internal/channels/manager.go b/internal/channels/manager.go index d1a6bef6..2ed3a1ff 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -6,6 +6,7 @@ import ( "sync" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/store" ) // RunContext tracks an active agent run for streaming/reaction event forwarding. @@ -25,11 +26,12 @@ type RunContext struct { // Manager manages all registered channels, handling their lifecycle // and routing outbound messages to the correct channel. type Manager struct { - channels map[string]Channel - bus *bus.MessageBus - runs sync.Map // runID string → *RunContext - dispatchTask *asyncTask - mu sync.RWMutex + channels map[string]Channel + bus *bus.MessageBus + runs sync.Map // runID string → *RunContext + dispatchTask *asyncTask + mu sync.RWMutex + contactCollector *store.ContactCollector } type asyncTask struct { @@ -137,9 +139,27 @@ func (m *Manager) GetEnabledChannels() []string { func (m *Manager) RegisterChannel(name string, channel Channel) { m.mu.Lock() defer m.mu.Unlock() + // Propagate contact collector to channels that embed BaseChannel. + if m.contactCollector != nil { + if bc, ok := channel.(interface{ SetContactCollector(*store.ContactCollector) }); ok { + bc.SetContactCollector(m.contactCollector) + } + } m.channels[name] = channel } +// SetContactCollector sets the contact collector for all current and future channels. +func (m *Manager) SetContactCollector(cc *store.ContactCollector) { + m.mu.Lock() + defer m.mu.Unlock() + m.contactCollector = cc + for _, ch := range m.channels { + if bc, ok := ch.(interface{ SetContactCollector(*store.ContactCollector) }); ok { + bc.SetContactCollector(cc) + } + } +} + // ChannelTypeForName returns the platform type for a channel instance name. // Reads directly from the Channel.Type() method — no separate map needed. func (m *Manager) ChannelTypeForName(name string) string { diff --git a/internal/channels/slack/handlers.go b/internal/channels/slack/handlers.go index b104eec4..79827028 100644 --- a/internal/channels/slack/handlers.go +++ b/internal/channels/slack/handlers.go @@ -1,6 +1,7 @@ package slack import ( + "context" "fmt" "log/slog" "strings" @@ -172,6 +173,11 @@ func (c *Channel) handleMessage(ev *slackevents.MessageEvent) { MessageID: ev.TimeStamp, }, c.historyLimit) + // Collect contact even when bot is not mentioned (cache prevents DB spam). + if cc := c.ContactCollector(); cc != nil { + cc.EnsureContact(context.Background(), c.Type(), c.Name(), senderID, senderID, displayName, "", "group") + } + slog.Debug("slack group message recorded (no mention)", "channel_id", channelID, "user", displayName) return diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 3b55d2d9..8bd512f8 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -251,6 +251,11 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { MessageID: fmt.Sprintf("%d", message.MessageID), }, c.historyLimit) + // Collect contact even when bot is not mentioned (cache prevents DB spam). + if cc := c.ContactCollector(); cc != nil { + cc.EnsureContact(ctx, c.Type(), c.Name(), userID, userID, user.FirstName, user.Username, "group") + } + slog.Debug("telegram group message recorded (no mention)", "chat_id", chatID, "sender", senderLabel, ) diff --git a/internal/channels/zalo/personal/handlers.go b/internal/channels/zalo/personal/handlers.go index b11a9c41..e9425626 100644 --- a/internal/channels/zalo/personal/handlers.go +++ b/internal/channels/zalo/personal/handlers.go @@ -88,6 +88,12 @@ func (c *Channel) handleGroupMessage(msg protocol.GroupMessage) { Timestamp: time.Now(), MessageID: msg.Data.MsgID, }, c.historyLimit) + + // Collect contact even when bot is not mentioned (cache prevents DB spam). + if cc := c.ContactCollector(); cc != nil { + cc.EnsureContact(context.Background(), c.Type(), c.Name(), senderID, senderID, senderName, "", "group") + } + slog.Debug("zalo_personal group message recorded (no mention)", "group_id", threadID, "sender", senderName,