diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index df430e4d3..8badb991d 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -12,6 +12,7 @@ import ( "fmt" "math" "net/http" + "strings" "sync" "time" @@ -192,6 +193,22 @@ func NewManager(cfg *config.Config, messageBus *bus.MessageBus, store media.Medi return m, nil } +// injectChannelDependencies injects optional dependencies (MediaStore, PlaceholderRecorder, +// Owner) into a channel if it implements the corresponding setter interfaces. +func (m *Manager) injectChannelDependencies(ch Channel) { + if m.mediaStore != nil { + if setter, ok := ch.(interface{ SetMediaStore(s media.MediaStore) }); ok { + setter.SetMediaStore(m.mediaStore) + } + } + if setter, ok := ch.(interface{ SetPlaceholderRecorder(r PlaceholderRecorder) }); ok { + setter.SetPlaceholderRecorder(m) + } + if setter, ok := ch.(interface{ SetOwner(ch Channel) }); ok { + setter.SetOwner(ch) + } +} + // initChannel is a helper that looks up a factory by name and creates the channel. func (m *Manager) initChannel(name, displayName string) { f, ok := getFactory(name) @@ -211,20 +228,7 @@ func (m *Manager) initChannel(name, displayName string) { "error": err.Error(), }) } else { - // Inject MediaStore if channel supports it - if m.mediaStore != nil { - if setter, ok := ch.(interface{ SetMediaStore(s media.MediaStore) }); ok { - setter.SetMediaStore(m.mediaStore) - } - } - // Inject PlaceholderRecorder if channel supports it - if setter, ok := ch.(interface{ SetPlaceholderRecorder(r PlaceholderRecorder) }); ok { - setter.SetPlaceholderRecorder(m) - } - // Inject owner reference so BaseChannel.HandleMessage can auto-trigger typing/reaction - if setter, ok := ch.(interface{ SetOwner(ch Channel) }); ok { - setter.SetOwner(ch) - } + m.injectChannelDependencies(ch) m.channels[name] = ch logger.InfoCF("channels", "Channel enabled successfully", map[string]any{ "channel": displayName, @@ -232,11 +236,57 @@ func (m *Manager) initChannel(name, displayName string) { } } +// initTelegramBot initializes a single named Telegram bot and registers it as a channel. +func (m *Manager) initTelegramBot(bot config.TelegramBotConfig) { + channelName := bot.ChannelName() + displayName := "Telegram" + if bot.ID != "" && bot.ID != "default" { + displayName = "Telegram (" + bot.ID + ")" + } + + logger.DebugCF("channels", "Attempting to initialize channel", map[string]any{ + "channel": displayName, + }) + + f, ok := getTelegramBotFactory() + if !ok { + logger.WarnCF("channels", "Telegram bot factory not registered", map[string]any{ + "channel": displayName, + }) + return + } + + ch, err := f(bot, channelName, m.bus) + if err != nil { + logger.ErrorCF("channels", "Failed to initialize channel", map[string]any{ + "channel": displayName, + "error": err.Error(), + }) + return + } + + m.injectChannelDependencies(ch) + + if _, exists := m.channels[channelName]; exists { + logger.ErrorCF("channels", "Duplicate channel name — skipping bot", map[string]any{ + "channel": displayName, + "name": channelName, + }) + return + } + m.channels[channelName] = ch + logger.InfoCF("channels", "Channel enabled successfully", map[string]any{ + "channel": displayName, + }) +} + func (m *Manager) initChannels() error { logger.InfoC("channels", "Initializing channel manager") - if m.config.Channels.Telegram.Enabled && m.config.Channels.Telegram.Token != "" { - m.initChannel("telegram", "Telegram") + for _, bot := range m.config.Channels.TelegramBots { + if bot.Enabled && bot.Token != "" { + m.initTelegramBot(bot) + } } if m.config.Channels.WhatsApp.Enabled { @@ -478,6 +528,14 @@ func newChannelWorker(name string, ch Channel) *channelWorker { rateVal := float64(defaultRateLimit) if r, ok := channelRateConfig[name]; ok { rateVal = r + } else { + // Named channel variants (e.g. "telegram-amber") inherit the base channel's rate. + for prefix, r := range channelRateConfig { + if strings.HasPrefix(name, prefix+"-") { + rateVal = r + break + } + } } burst := int(math.Max(1, math.Ceil(rateVal/2))) diff --git a/pkg/channels/registry.go b/pkg/channels/registry.go index 36a05bf3e..2cdc513c8 100644 --- a/pkg/channels/registry.go +++ b/pkg/channels/registry.go @@ -11,9 +11,17 @@ import ( // Each channel subpackage registers one or more factories via init(). type ChannelFactory func(cfg *config.Config, bus *bus.MessageBus) (Channel, error) +// TelegramBotFactory is a constructor function that creates a TelegramChannel from a +// TelegramBotConfig and an explicit channel name. Registered by the telegram subpackage +// to allow the manager to initialize named bots without a direct import (avoiding cycles). +type TelegramBotFactory func(botCfg config.TelegramBotConfig, channelName string, bus *bus.MessageBus) (Channel, error) + var ( factoriesMu sync.RWMutex factories = map[string]ChannelFactory{} + + telegramBotFactoryMu sync.RWMutex + telegramBotFactory TelegramBotFactory ) // RegisterFactory registers a named channel factory. Called from subpackage init() functions. @@ -30,3 +38,18 @@ func getFactory(name string) (ChannelFactory, bool) { f, ok := factories[name] return f, ok } + +// RegisterTelegramBotFactory registers the factory used to create named Telegram bots. +// Called from the telegram subpackage init() to avoid a direct import cycle. +func RegisterTelegramBotFactory(f TelegramBotFactory) { + telegramBotFactoryMu.Lock() + defer telegramBotFactoryMu.Unlock() + telegramBotFactory = f +} + +// getTelegramBotFactory returns the registered TelegramBotFactory, if any. +func getTelegramBotFactory() (TelegramBotFactory, bool) { + telegramBotFactoryMu.RLock() + defer telegramBotFactoryMu.RUnlock() + return telegramBotFactory, telegramBotFactory != nil +} diff --git a/pkg/channels/telegram/init.go b/pkg/channels/telegram/init.go index ac87bb805..621233ca2 100644 --- a/pkg/channels/telegram/init.go +++ b/pkg/channels/telegram/init.go @@ -10,4 +10,7 @@ func init() { channels.RegisterFactory("telegram", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { return NewTelegramChannel(cfg, b) }) + channels.RegisterTelegramBotFactory(func(botCfg config.TelegramBotConfig, channelName string, b *bus.MessageBus) (channels.Channel, error) { + return NewTelegramChannelFromConfig(botCfg.AsTelegramConfig(), channelName, b) + }) } diff --git a/pkg/channels/telegram/telegram.go b/pkg/channels/telegram/telegram.go index 34ee46b7b..c7fd81283 100644 --- a/pkg/channels/telegram/telegram.go +++ b/pkg/channels/telegram/telegram.go @@ -40,12 +40,12 @@ var ( type TelegramChannel struct { *channels.BaseChannel - bot *telego.Bot - bh *th.BotHandler - config *config.Config - chatIDs map[string]int64 - ctx context.Context - cancel context.CancelFunc + bot *telego.Bot + bh *th.BotHandler + placeholderCfg config.PlaceholderConfig + chatIDs map[string]int64 + ctx context.Context + cancel context.CancelFunc registerFunc func(context.Context, []commands.Definition) error commandRegCancel context.CancelFunc @@ -95,10 +95,64 @@ func NewTelegramChannel(cfg *config.Config, bus *bus.MessageBus) (*TelegramChann ) return &TelegramChannel{ - BaseChannel: base, - bot: bot, - config: cfg, - chatIDs: make(map[string]int64), + BaseChannel: base, + bot: bot, + placeholderCfg: telegramCfg.Placeholder, + chatIDs: make(map[string]int64), + }, nil +} + +// NewTelegramChannelFromConfig creates a TelegramChannel from a TelegramConfig and +// an explicit channel name. Used when initializing named bots from telegram_bots config. +func NewTelegramChannelFromConfig(telegramCfg config.TelegramConfig, channelName string, b *bus.MessageBus) (*TelegramChannel, error) { + if telegramCfg.Token == "" { + return nil, fmt.Errorf("telegram bot token is required") + } + var opts []telego.BotOption + + if telegramCfg.Proxy != "" { + proxyURL, parseErr := url.Parse(telegramCfg.Proxy) + if parseErr != nil { + return nil, fmt.Errorf("invalid proxy URL %q: %w", telegramCfg.Proxy, parseErr) + } + opts = append(opts, telego.WithHTTPClient(&http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyURL(proxyURL), + }, + })) + } else if os.Getenv("HTTP_PROXY") != "" || os.Getenv("HTTPS_PROXY") != "" { + opts = append(opts, telego.WithHTTPClient(&http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + }, + })) + } + + if baseURL := strings.TrimRight(strings.TrimSpace(telegramCfg.BaseURL), "/"); baseURL != "" { + opts = append(opts, telego.WithAPIServer(baseURL)) + } + opts = append(opts, telego.WithLogger(logger.NewLogger("telego"))) + + bot, err := telego.NewBot(telegramCfg.Token, opts...) + if err != nil { + return nil, fmt.Errorf("failed to create telegram bot: %w", err) + } + + base := channels.NewBaseChannel( + channelName, + telegramCfg, + b, + telegramCfg.AllowFrom, + channels.WithMaxMessageLength(4000), + channels.WithGroupTrigger(telegramCfg.GroupTrigger), + channels.WithReasoningChannelID(telegramCfg.ReasoningChannelID), + ) + + return &TelegramChannel{ + BaseChannel: base, + bot: bot, + placeholderCfg: telegramCfg.Placeholder, + chatIDs: make(map[string]int64), }, nil } @@ -298,7 +352,7 @@ func (c *TelegramChannel) EditMessage(ctx context.Context, chatID string, messag // It sends a placeholder message (e.g. "Thinking... 💭") that will later be // edited to the actual response via EditMessage (channels.MessageEditor). func (c *TelegramChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { - phCfg := c.config.Channels.Telegram.Placeholder + phCfg := c.placeholderCfg if !phCfg.Enabled { return "", nil } diff --git a/pkg/config/config.go b/pkg/config/config.go index 71c7fec8f..d4c43a563 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -255,9 +255,10 @@ func (d *AgentDefaults) GetModelName() string { } type ChannelsConfig struct { - WhatsApp WhatsAppConfig `json:"whatsapp"` - Telegram TelegramConfig `json:"telegram"` - Feishu FeishuConfig `json:"feishu"` + WhatsApp WhatsAppConfig `json:"whatsapp"` + Telegram TelegramConfig `json:"telegram"` + TelegramBots []TelegramBotConfig `json:"telegram_bots,omitempty"` + Feishu FeishuConfig `json:"feishu"` Discord DiscordConfig `json:"discord"` MaixCam MaixCamConfig `json:"maixcam"` QQ QQConfig `json:"qq"` @@ -311,6 +312,47 @@ type TelegramConfig struct { ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_TELEGRAM_REASONING_CHANNEL_ID"` } +// TelegramBotConfig defines a single named Telegram bot for use in telegram_bots. +// Each entry creates a separate channel named "telegram-", except when id is +// empty or "default" which creates the standard "telegram" channel. +type TelegramBotConfig struct { + ID string `json:"id"` + Enabled bool `json:"enabled"` + Token string `json:"token"` + BaseURL string `json:"base_url,omitempty"` + Proxy string `json:"proxy,omitempty"` + AllowFrom FlexibleStringSlice `json:"allow_from,omitempty"` + GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` + Typing TypingConfig `json:"typing,omitempty"` + Placeholder PlaceholderConfig `json:"placeholder,omitempty"` + ReasoningChannelID string `json:"reasoning_channel_id,omitempty"` +} + +// ChannelName returns the channel identifier for this bot. +// Bots with an empty or "default" ID use "telegram" for backward compatibility. +// All other IDs produce "telegram-". +func (b TelegramBotConfig) ChannelName() string { + if b.ID == "" || b.ID == "default" { + return "telegram" + } + return "telegram-" + b.ID +} + +// AsTelegramConfig converts a TelegramBotConfig to the equivalent TelegramConfig. +func (b TelegramBotConfig) AsTelegramConfig() TelegramConfig { + return TelegramConfig{ + Enabled: b.Enabled, + Token: b.Token, + BaseURL: b.BaseURL, + Proxy: b.Proxy, + AllowFrom: b.AllowFrom, + GroupTrigger: b.GroupTrigger, + Typing: b.Typing, + Placeholder: b.Placeholder, + ReasoningChannelID: b.ReasoningChannelID, + } +} + type FeishuConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_FEISHU_ENABLED"` AppID string `json:"app_id" env:"PICOCLAW_CHANNELS_FEISHU_APP_ID"` @@ -844,6 +886,7 @@ func LoadConfig(path string) (*Config, error) { // Migrate legacy channel config fields to new unified structures cfg.migrateChannelConfigs() + cfg.normalizeTelegramBots() // Auto-migrate: if only legacy providers config exists, convert to model_list if len(cfg.ModelList) == 0 && cfg.HasProvidersConfig() { @@ -858,6 +901,35 @@ func LoadConfig(path string) (*Config, error) { return cfg, nil } +// normalizeTelegramBots folds the legacy single channels.telegram entry into +// channels.telegram_bots so the rest of the codebase only needs to range over +// TelegramBots. The legacy entry is prepended with id "default", which produces +// the channel name "telegram" — preserving all existing bindings unchanged. +func (c *Config) normalizeTelegramBots() { + if c.Channels.Telegram.Token == "" { + return + } + // Already represented in the list (same channel name) — avoid duplicates. + for _, b := range c.Channels.TelegramBots { + if b.ChannelName() == "telegram" { + return + } + } + legacy := TelegramBotConfig{ + ID: "default", + Enabled: c.Channels.Telegram.Enabled, + Token: c.Channels.Telegram.Token, + BaseURL: c.Channels.Telegram.BaseURL, + Proxy: c.Channels.Telegram.Proxy, + AllowFrom: c.Channels.Telegram.AllowFrom, + GroupTrigger: c.Channels.Telegram.GroupTrigger, + Typing: c.Channels.Telegram.Typing, + Placeholder: c.Channels.Telegram.Placeholder, + ReasoningChannelID: c.Channels.Telegram.ReasoningChannelID, + } + c.Channels.TelegramBots = append([]TelegramBotConfig{legacy}, c.Channels.TelegramBots...) +} + func (c *Config) migrateChannelConfigs() { // Discord: mention_only -> group_trigger.mention_only if c.Channels.Discord.MentionOnly && !c.Channels.Discord.GroupTrigger.MentionOnly {