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.
This commit is contained in:
viettranx committed 2026-03-10 18:46:44 +07:00
1 parent 2495ff253c
commit 23f1957c56
8 files changed
+67 -11

No files matched your search

+1
View File
@@ -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)
+14 -6
View File
@@ -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 }
+5
View File
@@ -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,
+5
View File
@@ -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,
)
+25 -5
View File
@@ -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 {
+6
View File
@@ -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
+5
View File
@@ -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,
)
@@ -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,