mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-17 04:15:05 +00:00
feat(channels): migrate STT to unified audio.Manager
Migrate Telegram, Discord, Feishu channels to audio.Manager.Transcribe for consistent voice message handling. WhatsApp STT opt-in (default off; Decision 6 — E2E encryption trade-off). Fix tenant-scoped context in Discord factory. Remove per-channel STT implementations.
This commit is contained in:
1 parent
4db33dc822
commit
5e0a60da36
20 files changed
+361
-404
No files matched your search
+6
-6
@@ -133,7 +133,7 @@ func runGateway() {
|
||||
tools.DetectServerIPs(context.Background())
|
||||
}
|
||||
|
||||
toolsReg, execApprovalMgr, mcpMgr, sandboxMgr, browserMgr, webFetchTool, ttsTool, permPE, toolPE, dataDir, agentCfg := setupToolRegistry(cfg, workspace, providerRegistry)
|
||||
toolsReg, execApprovalMgr, mcpMgr, sandboxMgr, browserMgr, webFetchTool, ttsTool, audioMgr, permPE, toolPE, dataDir, agentCfg := setupToolRegistry(cfg, workspace, providerRegistry)
|
||||
if browserMgr != nil {
|
||||
defer browserMgr.Close()
|
||||
}
|
||||
@@ -425,12 +425,12 @@ func runGateway() {
|
||||
instanceLoader = channels.NewInstanceLoader(pgStores.ChannelInstances, pgStores.Agents, channelMgr, msgBus, pgStores.Pairing)
|
||||
instanceLoader.SetProviderRegistry(providerRegistry)
|
||||
instanceLoader.SetPendingCompactionConfig(cfg.Channels.PendingCompaction)
|
||||
instanceLoader.RegisterFactory(channels.TypeTelegram, telegram.FactoryWithStores(pgStores.Agents, pgStores.ConfigPermissions, pgStores.Teams, pgStores.SubagentTasks, pgStores.PendingMessages))
|
||||
instanceLoader.RegisterFactory(channels.TypeDiscord, discord.FactoryWithStores(pgStores.Agents, pgStores.ConfigPermissions, pgStores.PendingMessages))
|
||||
instanceLoader.RegisterFactory(channels.TypeFeishu, feishu.FactoryWithPendingStore(pgStores.PendingMessages))
|
||||
instanceLoader.RegisterFactory(channels.TypeTelegram, telegram.FactoryWithStoresAndAudio(pgStores.Agents, pgStores.ConfigPermissions, pgStores.Teams, pgStores.SubagentTasks, pgStores.PendingMessages, audioMgr))
|
||||
instanceLoader.RegisterFactory(channels.TypeDiscord, discord.FactoryWithStoresAndAudio(pgStores.Agents, pgStores.ConfigPermissions, pgStores.PendingMessages, audioMgr))
|
||||
instanceLoader.RegisterFactory(channels.TypeFeishu, feishu.FactoryWithPendingStoreAndAudio(pgStores.PendingMessages, audioMgr))
|
||||
instanceLoader.RegisterFactory(channels.TypeZaloOA, zalo.Factory)
|
||||
instanceLoader.RegisterFactory(channels.TypeZaloPersonal, zalopersonal.FactoryWithPendingStore(pgStores.PendingMessages))
|
||||
instanceLoader.RegisterFactory(channels.TypeWhatsApp, whatsapp.FactoryWithDB(pgStores.DB, pgStores.PendingMessages, "pgx"))
|
||||
instanceLoader.RegisterFactory(channels.TypeWhatsApp, whatsapp.FactoryWithDBAudio(pgStores.DB, pgStores.PendingMessages, "pgx", audioMgr, pgStores.BuiltinTools))
|
||||
instanceLoader.RegisterFactory(channels.TypeSlack, slackchannel.FactoryWithPendingStore(pgStores.PendingMessages))
|
||||
instanceLoader.RegisterFactory(channels.TypeFacebook, facebook.Factory)
|
||||
instanceLoader.RegisterFactory(channels.TypePancake, pancake.Factory)
|
||||
@@ -440,7 +440,7 @@ func runGateway() {
|
||||
}
|
||||
|
||||
// Register config-based channels as fallback when no DB instances loaded.
|
||||
registerConfigChannels(cfg, channelMgr, msgBus, pgStores, instanceLoader)
|
||||
registerConfigChannels(cfg, channelMgr, msgBus, pgStores, instanceLoader, audioMgr)
|
||||
|
||||
// Register channels/instances/links/teams RPC methods
|
||||
wireChannelRPCMethods(server, pgStores, channelMgr, agentRouter, msgBus, workspace)
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/agent"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/discord"
|
||||
@@ -27,7 +28,8 @@ import (
|
||||
)
|
||||
|
||||
// registerConfigChannels registers config-based channels as fallback when no DB instances are loaded.
|
||||
func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, msgBus *bus.MessageBus, pgStores *store.Stores, instanceLoader *channels.InstanceLoader) {
|
||||
// audioMgr is optional (nil = STT disabled for channels).
|
||||
func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, msgBus *bus.MessageBus, pgStores *store.Stores, instanceLoader *channels.InstanceLoader, audioMgr *audio.Manager) {
|
||||
if instanceLoader != nil {
|
||||
return
|
||||
}
|
||||
@@ -46,7 +48,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
|
||||
if cfg.Channels.Telegram.Enabled {
|
||||
if cfg.Channels.Telegram.Token == "" {
|
||||
recordMissingConfig(channels.TypeTelegram, "Set channels.telegram.token in config.")
|
||||
} else if tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pgStores.Pairing); err != nil {
|
||||
} else if tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pgStores.Pairing, audioMgr); err != nil {
|
||||
channelMgr.RecordFailure(channels.TypeTelegram, "", err)
|
||||
slog.Error("failed to initialize telegram channel", "error", err)
|
||||
} else {
|
||||
@@ -58,7 +60,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
|
||||
if cfg.Channels.Discord.Enabled {
|
||||
if cfg.Channels.Discord.Token == "" {
|
||||
recordMissingConfig(channels.TypeDiscord, "Set channels.discord.token in config.")
|
||||
} else if dc, err := discord.New(cfg.Channels.Discord, msgBus, nil, nil, nil, nil); err != nil {
|
||||
} else if dc, err := discord.New(cfg.Channels.Discord, msgBus, nil, nil, nil, nil, audioMgr); err != nil {
|
||||
channelMgr.RecordFailure(channels.TypeDiscord, "", err)
|
||||
slog.Error("failed to initialize discord channel", "error", err)
|
||||
} else {
|
||||
@@ -72,7 +74,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
|
||||
if strings.Contains(fmt.Sprintf("%T", pgStores.DB.Driver()), "sqlite") {
|
||||
waDialect = "sqlite3"
|
||||
}
|
||||
wa, err := whatsapp.New(cfg.Channels.WhatsApp, msgBus, pgStores.Pairing, pgStores.DB, pgStores.PendingMessages, waDialect)
|
||||
wa, err := whatsapp.New(cfg.Channels.WhatsApp, msgBus, pgStores.Pairing, pgStores.DB, pgStores.PendingMessages, waDialect, audioMgr, pgStores.BuiltinTools)
|
||||
if err != nil {
|
||||
channelMgr.RecordFailure(channels.TypeWhatsApp, "", err)
|
||||
slog.Error("failed to initialize whatsapp channel", "error", err)
|
||||
@@ -131,7 +133,7 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
|
||||
feishu.WithAgentStore(pgStores.Agents),
|
||||
feishu.WithConfigPermStore(pgStores.ConfigPermissions),
|
||||
}
|
||||
if f, err := feishu.New(cfg.Channels.Feishu, msgBus, pgStores.Pairing, nil, feishuOpts...); err != nil {
|
||||
if f, err := feishu.New(cfg.Channels.Feishu, msgBus, pgStores.Pairing, nil, audioMgr, feishuOpts...); err != nil {
|
||||
channelMgr.RecordFailure(channels.TypeFeishu, "", err)
|
||||
slog.Error("failed to initialize feishu channel", "error", err)
|
||||
} else {
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"github.com/bwmarrin/discordgo"
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/typing"
|
||||
@@ -29,15 +30,17 @@ type Channel struct {
|
||||
typingCtrls sync.Map // channelID string → *typing.Controller
|
||||
agentStore store.AgentStore // for agent key lookup (nil = writer commands disabled)
|
||||
configPermStore store.ConfigPermissionStore // for group file writer management (nil = writer commands disabled)
|
||||
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
|
||||
// pairingService, pairingDebounce, approvedGroups, groupHistory, historyLimit, requireMention
|
||||
// are inherited from channels.BaseChannel.
|
||||
}
|
||||
|
||||
// New creates a new Discord channel from config.
|
||||
// agentStore and configPermStore are optional (nil = writer commands disabled).
|
||||
// audioMgr is optional (nil = STT disabled).
|
||||
func New(cfg config.DiscordConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore,
|
||||
agentStore store.AgentStore, configPermStore store.ConfigPermissionStore,
|
||||
pendingStore store.PendingMessageStore) (*Channel, error) {
|
||||
pendingStore store.PendingMessageStore, audioMgr *audio.Manager) (*Channel, error) {
|
||||
session, err := discordgo.New("Bot " + cfg.Token)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create discord session: %w", err)
|
||||
@@ -67,6 +70,7 @@ func New(cfg config.DiscordConfig, msgBus *bus.MessageBus, pairingSvc store.Pair
|
||||
config: cfg,
|
||||
agentStore: agentStore,
|
||||
configPermStore: configPermStore,
|
||||
audioMgr: audioMgr,
|
||||
}
|
||||
ch.SetRequireMention(requireMention)
|
||||
ch.SetPairingService(pairingSvc)
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -34,21 +35,26 @@ type discordInstanceConfig struct {
|
||||
// Factory creates a Discord channel from DB instance data (no extra stores).
|
||||
func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, nil, nil, nil)
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, nil, nil, nil, nil)
|
||||
}
|
||||
|
||||
// FactoryWithStores returns a ChannelFactory that includes agent, configPerm, and pending message stores.
|
||||
func FactoryWithStores(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, pendingStore store.PendingMessageStore) channels.ChannelFactory {
|
||||
return FactoryWithStoresAndAudio(agentStore, configPermStore, pendingStore, nil)
|
||||
}
|
||||
|
||||
// FactoryWithStoresAndAudio returns a ChannelFactory with all stores and STT support.
|
||||
func FactoryWithStoresAndAudio(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
|
||||
return func(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, agentStore, configPermStore, pendingStore)
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, agentStore, configPermStore, pendingStore, audioMgr)
|
||||
}
|
||||
}
|
||||
|
||||
func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore,
|
||||
agentStore store.AgentStore, configPermStore store.ConfigPermissionStore,
|
||||
pendingStore store.PendingMessageStore) (channels.Channel, error) {
|
||||
pendingStore store.PendingMessageStore, audioMgr *audio.Manager) (channels.Channel, error) {
|
||||
|
||||
var c discordCreds
|
||||
if len(creds) > 0 {
|
||||
@@ -89,7 +95,7 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
dcCfg.GroupPolicy = "pairing"
|
||||
}
|
||||
|
||||
ch, err := New(dcCfg, msgBus, pairingSvc, agentStore, configPermStore, pendingStore)
|
||||
ch, err := New(dcCfg, msgBus, pairingSvc, agentStore, configPermStore, pendingStore, audioMgr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"github.com/bwmarrin/discordgo"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
@@ -114,7 +115,18 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
|
||||
|
||||
switch mi.Type {
|
||||
case media.TypeAudio, media.TypeVoice:
|
||||
transcript, sttErr := c.transcribeAudio(context.Background(), mi.FilePath)
|
||||
var transcript string
|
||||
var sttErr error
|
||||
if c.audioMgr != nil {
|
||||
sttCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
res, err := c.audioMgr.Transcribe(sttCtx, audio.STTInput{FilePath: mi.FilePath, MimeType: "audio/ogg"}, audio.STTOptions{})
|
||||
cancel()
|
||||
if err == nil && res != nil {
|
||||
transcript = res.Text
|
||||
} else {
|
||||
sttErr = err
|
||||
}
|
||||
}
|
||||
if sttErr != nil {
|
||||
slog.Warn("discord: STT transcription failed",
|
||||
"type", mi.Type, "error", sttErr,
|
||||
|
||||
@@ -1,18 +0,0 @@
|
||||
package discord
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
)
|
||||
|
||||
// transcribeAudio calls the shared STT proxy service with the given audio file.
|
||||
// Returns ("", nil) silently when STT is not configured or filePath is empty.
|
||||
func (c *Channel) transcribeAudio(ctx context.Context, filePath string) (string, error) {
|
||||
return media.TranscribeAudio(ctx, media.STTConfig{
|
||||
ProxyURL: c.config.STTProxyURL,
|
||||
APIKey: c.config.STTAPIKey,
|
||||
TenantID: c.config.STTTenantID,
|
||||
TimeoutSeconds: c.config.STTTimeoutSeconds,
|
||||
}, filePath)
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
@@ -266,7 +267,18 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) {
|
||||
|
||||
switch m.Type {
|
||||
case media.TypeAudio, media.TypeVoice:
|
||||
transcript, sttErr := c.transcribeAudio(ctx, m.FilePath)
|
||||
var transcript string
|
||||
var sttErr error
|
||||
if c.audioMgr != nil {
|
||||
sttCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
res, err := c.audioMgr.Transcribe(sttCtx, audio.STTInput{FilePath: m.FilePath, MimeType: "audio/ogg"}, audio.STTOptions{})
|
||||
cancel()
|
||||
if err == nil && res != nil {
|
||||
transcript = res.Text
|
||||
} else {
|
||||
sttErr = err
|
||||
}
|
||||
}
|
||||
if sttErr != nil {
|
||||
slog.Warn("feishu: STT transcription failed",
|
||||
"type", m.Type, "error", sttErr,
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -100,7 +101,7 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
fsCfg.GroupPolicy = "pairing"
|
||||
}
|
||||
|
||||
ch, err := New(fsCfg, msgBus, pairingSvc, nil)
|
||||
ch, err := New(fsCfg, msgBus, pairingSvc, nil, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -111,6 +112,11 @@ func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
|
||||
// FactoryWithPendingStore returns a ChannelFactory with persistent history support.
|
||||
func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.ChannelFactory {
|
||||
return FactoryWithPendingStoreAndAudio(pendingStore, nil)
|
||||
}
|
||||
|
||||
// FactoryWithPendingStoreAndAudio returns a ChannelFactory with persistent history and STT support.
|
||||
func FactoryWithPendingStoreAndAudio(pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
|
||||
return func(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
|
||||
@@ -165,7 +171,7 @@ func FactoryWithPendingStore(pendingStore store.PendingMessageStore) channels.Ch
|
||||
fsCfg.GroupPolicy = "pairing"
|
||||
}
|
||||
|
||||
ch, err := New(fsCfg, msgBus, pairingSvc, pendingStore)
|
||||
ch, err := New(fsCfg, msgBus, pairingSvc, pendingStore, audioMgr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -49,6 +50,7 @@ type Channel struct {
|
||||
stopCh chan struct{}
|
||||
httpServer *http.Server
|
||||
wsClient *WSClient
|
||||
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
|
||||
// pairingService, pairingDebounce, approvedGroups, groupHistory, historyLimit
|
||||
// are inherited from channels.BaseChannel.
|
||||
}
|
||||
@@ -91,7 +93,8 @@ type senderCacheEntry struct {
|
||||
}
|
||||
|
||||
// New creates a new Feishu/Lark channel.
|
||||
func New(cfg config.FeishuConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, pendingStore store.PendingMessageStore, opts ...Option) (*Channel, error) {
|
||||
// audioMgr is optional (nil = STT disabled).
|
||||
func New(cfg config.FeishuConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, pendingStore store.PendingMessageStore, audioMgr *audio.Manager, opts ...Option) (*Channel, error) {
|
||||
if cfg.AppID == "" || cfg.AppSecret == "" {
|
||||
return nil, fmt.Errorf("feishu app_id and app_secret are required")
|
||||
}
|
||||
@@ -116,6 +119,7 @@ func New(cfg config.FeishuConfig, msgBus *bus.MessageBus, pairingSvc store.Pairi
|
||||
docCache: newDocCache(larkDocCacheSize, larkDocCacheTTL),
|
||||
groupAllowList: cfg.GroupAllowFrom,
|
||||
stopCh: make(chan struct{}),
|
||||
audioMgr: audioMgr,
|
||||
}
|
||||
ch.SetPairingService(pairingSvc)
|
||||
ch.SetGroupHistory(channels.MakeHistory(channels.TypeFeishu, pendingStore, base.TenantID()))
|
||||
|
||||
@@ -1,18 +0,0 @@
|
||||
package feishu
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
)
|
||||
|
||||
// transcribeAudio calls the shared STT proxy service with the given audio file.
|
||||
// Returns ("", nil) silently when STT is not configured or filePath is empty.
|
||||
func (c *Channel) transcribeAudio(ctx context.Context, filePath string) (string, error) {
|
||||
return media.TranscribeAudio(ctx, media.STTConfig{
|
||||
ProxyURL: c.cfg.STTProxyURL,
|
||||
APIKey: c.cfg.STTAPIKey,
|
||||
TenantID: c.cfg.STTTenantID,
|
||||
TimeoutSeconds: c.cfg.STTTimeoutSeconds,
|
||||
}, filePath)
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/google/uuid"
|
||||
"github.com/mymmrac/telego"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -43,6 +44,7 @@ type Channel struct {
|
||||
handlerWg sync.WaitGroup // tracks in-flight handler goroutines for graceful shutdown
|
||||
handlerSem chan struct{} // bounded semaphore for concurrent handler goroutines
|
||||
pendingDraftID sync.Map // localKey string → int (draftID)
|
||||
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
|
||||
// pairingService, approvedGroups, pairingDebounce, groupHistory, historyLimit, requireMention
|
||||
// are inherited from channels.BaseChannel.
|
||||
}
|
||||
@@ -85,8 +87,9 @@ func WithPendingMessageStore(s store.PendingMessageStore) Option {
|
||||
|
||||
// New creates a new Telegram channel from config.
|
||||
// pairingSvc is optional (nil = fall back to allowlist only).
|
||||
// audioMgr is optional (nil = STT disabled).
|
||||
// Optional stores are set via Option functions.
|
||||
func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, chanOpts ...Option) (*Channel, error) {
|
||||
func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.PairingStore, audioMgr *audio.Manager, chanOpts ...Option) (*Channel, error) {
|
||||
var botOpts []telego.BotOption
|
||||
|
||||
if cfg.APIServer != "" {
|
||||
@@ -152,6 +155,7 @@ func New(cfg config.TelegramConfig, msgBus *bus.MessageBus, pairingSvc store.Pai
|
||||
httpClient: httpClient,
|
||||
transport: transport,
|
||||
mentionMode: mentionMode,
|
||||
audioMgr: audioMgr,
|
||||
}
|
||||
ch.SetPairingService(pairingSvc)
|
||||
ch.SetGroupHistory(channels.MakeHistory(channels.TypeTelegram, nil, base.TenantID()))
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -42,14 +43,19 @@ type telegramInstanceConfig struct {
|
||||
// Factory creates a Telegram channel from DB instance data (no extra stores).
|
||||
func Factory(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc)
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, nil)
|
||||
}
|
||||
|
||||
// FactoryWithStores returns a ChannelFactory that includes optional stores via functional options.
|
||||
func FactoryWithStores(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, subagentTaskStore store.SubagentTaskStore, pendingStore store.PendingMessageStore) channels.ChannelFactory {
|
||||
return FactoryWithStoresAndAudio(agentStore, configPermStore, teamStore, subagentTaskStore, pendingStore, nil)
|
||||
}
|
||||
|
||||
// FactoryWithStoresAndAudio returns a ChannelFactory with all stores and STT support.
|
||||
func FactoryWithStoresAndAudio(agentStore store.AgentStore, configPermStore store.ConfigPermissionStore, teamStore store.TeamStore, subagentTaskStore store.SubagentTaskStore, pendingStore store.PendingMessageStore, audioMgr *audio.Manager) channels.ChannelFactory {
|
||||
return func(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc,
|
||||
return buildChannel(name, creds, cfg, msgBus, pairingSvc, audioMgr,
|
||||
WithAgentStore(agentStore),
|
||||
WithConfigPermStore(configPermStore),
|
||||
WithTeamStore(teamStore),
|
||||
@@ -60,7 +66,7 @@ func FactoryWithStores(agentStore store.AgentStore, configPermStore store.Config
|
||||
}
|
||||
|
||||
func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore, opts ...Option) (channels.Channel, error) {
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore, audioMgr *audio.Manager, opts ...Option) (channels.Channel, error) {
|
||||
|
||||
var c telegramCreds
|
||||
if len(creds) > 0 {
|
||||
@@ -117,7 +123,7 @@ func buildChannel(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
tgCfg.GroupPolicy = "pairing"
|
||||
}
|
||||
|
||||
ch, err := New(tgCfg, msgBus, pairingSvc, opts...)
|
||||
ch, err := New(tgCfg, msgBus, pairingSvc, audioMgr, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"github.com/mymmrac/telego"
|
||||
tu "github.com/mymmrac/telego/telegoutil"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/typing"
|
||||
@@ -407,7 +408,18 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
|
||||
m := &mediaList[i]
|
||||
switch m.Type {
|
||||
case "audio", "voice":
|
||||
transcript, sttErr := c.transcribeAudio(ctx, m.FilePath)
|
||||
var transcript string
|
||||
var sttErr error
|
||||
if c.audioMgr != nil {
|
||||
sttCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
res, err := c.audioMgr.Transcribe(sttCtx, audio.STTInput{FilePath: m.FilePath, MimeType: "audio/ogg"}, audio.STTOptions{})
|
||||
cancel()
|
||||
if err == nil && res != nil {
|
||||
transcript = res.Text
|
||||
} else {
|
||||
sttErr = err
|
||||
}
|
||||
}
|
||||
if sttErr != nil {
|
||||
slog.Warn("telegram: STT transcription failed",
|
||||
"type", m.Type, "error", sttErr)
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
package telegram
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels/media"
|
||||
)
|
||||
|
||||
// transcribeAudio calls the configured STT proxy service with the given audio file and returns
|
||||
// the transcribed text. Returns ("", nil) silently when STT is not configured or filePath is empty.
|
||||
// Delegates to the shared media.TranscribeAudio implementation.
|
||||
func (c *Channel) transcribeAudio(ctx context.Context, filePath string) (string, error) {
|
||||
return media.TranscribeAudio(ctx, media.STTConfig{
|
||||
ProxyURL: c.config.STTProxyURL,
|
||||
APIKey: c.config.STTAPIKey,
|
||||
TenantID: c.config.STTTenantID,
|
||||
TimeoutSeconds: c.config.STTTimeoutSeconds,
|
||||
}, filePath)
|
||||
}
|
||||
@@ -1,312 +0,0 @@
|
||||
package telegram
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
)
|
||||
|
||||
// sttTestResponse mirrors the STT proxy JSON response for test assertions.
|
||||
type sttTestResponse struct {
|
||||
Transcript string `json:"transcript"`
|
||||
}
|
||||
|
||||
// sttEndpoint is the expected STT endpoint path.
|
||||
const sttEndpoint = "/transcribe_audio"
|
||||
|
||||
// newChannelWithSTT is a minimal Channel stub for STT unit tests.
|
||||
// It skips bot initialisation (which requires a real Telegram token).
|
||||
func newChannelWithSTT(cfg config.TelegramConfig) *Channel {
|
||||
return &Channel{config: cfg}
|
||||
}
|
||||
|
||||
// writeTempAudio writes a fake audio file and returns its path.
|
||||
// The caller is responsible for removing the file after the test.
|
||||
func writeTempAudio(t *testing.T, content string) string {
|
||||
t.Helper()
|
||||
f, err := os.CreateTemp("", "stt_test_*.ogg")
|
||||
if err != nil {
|
||||
t.Fatalf("create temp audio file: %v", err)
|
||||
}
|
||||
if _, err := f.WriteString(content); err != nil {
|
||||
t.Fatalf("write temp audio file: %v", err)
|
||||
}
|
||||
f.Close()
|
||||
return f.Name()
|
||||
}
|
||||
|
||||
// --- transcribeAudio unit tests ---
|
||||
|
||||
// TestTranscribeAudio_NoProxy verifies that when STTProxyURL is empty, the
|
||||
// function returns ("", nil) without making any HTTP call.
|
||||
func TestTranscribeAudio_NoProxy(t *testing.T) {
|
||||
c := newChannelWithSTT(config.TelegramConfig{})
|
||||
transcript, err := c.transcribeAudio(context.Background(), "/any/file.ogg")
|
||||
if err != nil {
|
||||
t.Fatalf("expected nil error, got: %v", err)
|
||||
}
|
||||
if transcript != "" {
|
||||
t.Fatalf("expected empty transcript, got: %q", transcript)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_EmptyFilePath verifies that an empty filePath is a silent
|
||||
// no-op even when STT is configured.
|
||||
func TestTranscribeAudio_EmptyFilePath(t *testing.T) {
|
||||
c := newChannelWithSTT(config.TelegramConfig{
|
||||
STTProxyURL: "https://stt.example.com",
|
||||
})
|
||||
transcript, err := c.transcribeAudio(context.Background(), "")
|
||||
if err != nil {
|
||||
t.Fatalf("expected nil error, got: %v", err)
|
||||
}
|
||||
if transcript != "" {
|
||||
t.Fatalf("expected empty transcript, got: %q", transcript)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_MissingFile verifies that a non-existent file path returns
|
||||
// an error (not a silent empty result).
|
||||
func TestTranscribeAudio_MissingFile(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// Should never be reached.
|
||||
t.Error("unexpected HTTP call for missing file")
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
_, err := c.transcribeAudio(context.Background(), "/nonexistent/file.ogg")
|
||||
if err == nil {
|
||||
t.Fatal("expected an error for missing file, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_Success verifies the happy path: a real HTTP server returns
|
||||
// {"transcript": "hello world"} and the function returns that string.
|
||||
func TestTranscribeAudio_Success(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// Verify endpoint path.
|
||||
if r.URL.Path != sttEndpoint {
|
||||
t.Errorf("unexpected path: %s", r.URL.Path)
|
||||
}
|
||||
// Verify method.
|
||||
if r.Method != http.MethodPost {
|
||||
t.Errorf("expected POST, got %s", r.Method)
|
||||
}
|
||||
// Verify multipart body contains a "file" field.
|
||||
if err := r.ParseMultipartForm(1 << 20); err != nil {
|
||||
t.Errorf("parse multipart: %v", err)
|
||||
}
|
||||
if _, _, err := r.FormFile("file"); err != nil {
|
||||
t.Errorf("expected 'file' field in multipart form: %v", err)
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: "hello world"})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
transcript, err := c.transcribeAudio(context.Background(), audioFile)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if transcript != "hello world" {
|
||||
t.Errorf("expected %q, got %q", "hello world", transcript)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_BearerToken verifies that STTAPIKey is sent as an
|
||||
// Authorization: Bearer header.
|
||||
func TestTranscribeAudio_BearerToken(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
const wantKey = "super-secret-key"
|
||||
var gotAuth string
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotAuth = r.Header.Get("Authorization")
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: "ok"})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{
|
||||
STTProxyURL: srv.URL,
|
||||
STTAPIKey: wantKey,
|
||||
})
|
||||
if _, err := c.transcribeAudio(context.Background(), audioFile); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if gotAuth != "Bearer "+wantKey {
|
||||
t.Errorf("expected Authorization %q, got %q", "Bearer "+wantKey, gotAuth)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_NoAuthHeader verifies that no Authorization header is sent
|
||||
// when STTAPIKey is empty.
|
||||
func TestTranscribeAudio_NoAuthHeader(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if auth := r.Header.Get("Authorization"); auth != "" {
|
||||
t.Errorf("expected no Authorization header, got %q", auth)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: "ok"})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
if _, err := c.transcribeAudio(context.Background(), audioFile); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_TenantID verifies that STTTenantID is forwarded as a
|
||||
// multipart "tenant_id" field when set.
|
||||
func TestTranscribeAudio_TenantID(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
const wantTenant = "acme-corp"
|
||||
var gotTenant string
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if err := r.ParseMultipartForm(1 << 20); err == nil {
|
||||
gotTenant = r.FormValue("tenant_id")
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: "ok"})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{
|
||||
STTProxyURL: srv.URL,
|
||||
STTTenantID: wantTenant,
|
||||
})
|
||||
if _, err := c.transcribeAudio(context.Background(), audioFile); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if gotTenant != wantTenant {
|
||||
t.Errorf("expected tenant_id %q, got %q", wantTenant, gotTenant)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_NoTenantField verifies that when STTTenantID is empty, the
|
||||
// multipart form does NOT include a "tenant_id" field.
|
||||
func TestTranscribeAudio_NoTenantField(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if err := r.ParseMultipartForm(1 << 20); err == nil {
|
||||
if tid := r.FormValue("tenant_id"); tid != "" {
|
||||
t.Errorf("expected no tenant_id field, got %q", tid)
|
||||
}
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: "ok"})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
if _, err := c.transcribeAudio(context.Background(), audioFile); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_UpstreamError verifies that a non-200 response is surfaced
|
||||
// as an error (not silently swallowed).
|
||||
func TestTranscribeAudio_UpstreamError(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "service unavailable", http.StatusServiceUnavailable)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
_, err := c.transcribeAudio(context.Background(), audioFile)
|
||||
if err == nil {
|
||||
t.Fatal("expected error for non-200 response, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "503") {
|
||||
t.Errorf("expected error to mention status 503, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_InvalidJSON verifies that a 200 response with malformed
|
||||
// JSON is returned as an error.
|
||||
func TestTranscribeAudio_InvalidJSON(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Write([]byte(`not-json`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
_, err := c.transcribeAudio(context.Background(), audioFile)
|
||||
if err == nil {
|
||||
t.Fatal("expected error for invalid JSON, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_EmptyTranscript verifies that a 200 response with an empty
|
||||
// transcript field returns ("", nil) — not an error.
|
||||
func TestTranscribeAudio_EmptyTranscript(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sttTestResponse{Transcript: ""})
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
transcript, err := c.transcribeAudio(context.Background(), audioFile)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if transcript != "" {
|
||||
t.Errorf("expected empty transcript, got %q", transcript)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTranscribeAudio_ContextCancelled verifies that a cancelled context causes
|
||||
// the HTTP call to fail fast.
|
||||
func TestTranscribeAudio_ContextCancelled(t *testing.T) {
|
||||
audioFile := writeTempAudio(t, "fake-ogg-bytes")
|
||||
defer os.Remove(audioFile)
|
||||
|
||||
// Server that blocks until the test is done — ensures the cancel fires first.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
<-r.Context().Done()
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // cancel immediately
|
||||
|
||||
c := newChannelWithSTT(config.TelegramConfig{STTProxyURL: srv.URL})
|
||||
_, err := c.transcribeAudio(ctx, audioFile)
|
||||
if err == nil {
|
||||
t.Fatal("expected error for cancelled context, got nil")
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -24,6 +25,13 @@ type whatsappInstanceConfig struct {
|
||||
// FactoryWithDB returns a ChannelFactory with DB access for whatsmeow auth state.
|
||||
// dialect must be "pgx" (PostgreSQL) or "sqlite3" (SQLite/desktop).
|
||||
func FactoryWithDB(db *sql.DB, pendingStore store.PendingMessageStore, dialect string) channels.ChannelFactory {
|
||||
return FactoryWithDBAudio(db, pendingStore, dialect, nil, nil)
|
||||
}
|
||||
|
||||
// FactoryWithDBAudio returns a ChannelFactory with DB access, STT support, and builtin-tools store
|
||||
// for reading stt.whatsapp_enabled opt-in setting per message.
|
||||
func FactoryWithDBAudio(db *sql.DB, pendingStore store.PendingMessageStore, dialect string,
|
||||
audioMgr *audio.Manager, builtinToolStore store.BuiltinToolStore) channels.ChannelFactory {
|
||||
return func(name string, creds json.RawMessage, cfg json.RawMessage,
|
||||
msgBus *bus.MessageBus, pairingSvc store.PairingStore) (channels.Channel, error) {
|
||||
|
||||
@@ -64,7 +72,7 @@ func FactoryWithDB(db *sql.DB, pendingStore store.PendingMessageStore, dialect s
|
||||
waCfg.GroupPolicy = "pairing"
|
||||
}
|
||||
|
||||
ch, err := New(waCfg, msgBus, pairingSvc, db, pendingStore, dialect)
|
||||
ch, err := New(waCfg, msgBus, pairingSvc, db, pendingStore, dialect, audioMgr, builtinToolStore)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -111,6 +111,21 @@ func (c *Channel) handleIncomingMessage(evt *events.Message) {
|
||||
metadata["user_name"] = evt.Info.PushName
|
||||
}
|
||||
|
||||
// STT: transcribe audio items (opt-in via builtin_tools[stt].settings.whatsapp_enabled,
|
||||
// default false per Decision 6 — enabling breaks E2E encryption for voice messages).
|
||||
waSttSettings := c.loadSTTSettings(ctx)
|
||||
locale := "" // i18n.T falls back to English when locale is empty
|
||||
for i := range mediaList {
|
||||
m := &mediaList[i]
|
||||
if m.Type == media.TypeAudio || m.Type == media.TypeVoice {
|
||||
mimeType := m.ContentType
|
||||
if mimeType == "" {
|
||||
mimeType = "audio/ogg"
|
||||
}
|
||||
m.Transcript = c.transcribeVoice(ctx, m.FilePath, mimeType, locale, waSttSettings)
|
||||
}
|
||||
}
|
||||
|
||||
// Build media tags and bus.MediaFile list.
|
||||
var mediaFiles []bus.MediaFile
|
||||
if len(mediaList) > 0 {
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package whatsapp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
)
|
||||
|
||||
// sttSettings is the minimal contract Phase 5 needs from builtin_tools[stt].settings.
|
||||
// whatsapp_enabled defaults to false (opt-in) per Decision 6 — admin must explicitly
|
||||
// enable it, acknowledging that WhatsApp STT breaks E2E encryption for voice messages.
|
||||
type sttSettings struct {
|
||||
WhatsappEnabled bool `json:"whatsapp_enabled"`
|
||||
}
|
||||
|
||||
// loadSTTSettings fetches stt builtin-tool settings and parses whatsapp_enabled.
|
||||
// Returns opt-out (false) on any failure: nil store, DB error, parse error, or missing key.
|
||||
// Called per voice message — rare enough that per-message DB read is acceptable.
|
||||
func (c *Channel) loadSTTSettings(ctx context.Context) sttSettings {
|
||||
if c.builtinToolStore == nil {
|
||||
return sttSettings{WhatsappEnabled: false}
|
||||
}
|
||||
raw, err := c.builtinToolStore.GetSettings(ctx, "stt")
|
||||
if err != nil {
|
||||
slog.Debug("whatsapp: stt settings fetch failed, defaulting to opt-out", "error", err)
|
||||
return sttSettings{WhatsappEnabled: false}
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
return sttSettings{WhatsappEnabled: false}
|
||||
}
|
||||
var s sttSettings
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
slog.Debug("whatsapp: stt settings parse failed, defaulting to opt-out", "error", err)
|
||||
return sttSettings{WhatsappEnabled: false}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// transcribeVoice converts a downloaded voice file to text via audio.Manager.
|
||||
// Returns i18n fallback "[Voice message]" when:
|
||||
// - settings.WhatsappEnabled == false (default; opt-in per Decision 6)
|
||||
// - audioMgr is nil
|
||||
// - Transcribe fails / times out (10s wall clock)
|
||||
func (c *Channel) transcribeVoice(ctx context.Context, filePath, mimeType, locale string, settings sttSettings) string {
|
||||
fallback := i18n.T(locale, i18n.MsgVoiceMessageFallback)
|
||||
if !settings.WhatsappEnabled {
|
||||
return fallback
|
||||
}
|
||||
if c.audioMgr == nil {
|
||||
return fallback
|
||||
}
|
||||
sttCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
res, err := c.audioMgr.Transcribe(sttCtx, audio.STTInput{FilePath: filePath, MimeType: mimeType}, audio.STTOptions{})
|
||||
if err != nil || res == nil {
|
||||
slog.Warn("whatsapp: stt failed or timed out", "error", err, "file", filePath)
|
||||
return fallback
|
||||
}
|
||||
slog.Info("audio.stt.whatsapp_invoked", "duration_secs", res.Duration, "lang", res.Language)
|
||||
return res.Text
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package whatsapp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/i18n"
|
||||
)
|
||||
|
||||
// stubSTTProvider is a minimal in-process STTProvider for unit tests.
|
||||
type stubSTTProvider struct {
|
||||
name string
|
||||
result *audio.TranscriptResult
|
||||
err error
|
||||
delay time.Duration // simulate slow provider
|
||||
called int
|
||||
}
|
||||
|
||||
func (s *stubSTTProvider) Name() string { return s.name }
|
||||
|
||||
func (s *stubSTTProvider) Transcribe(ctx context.Context, in audio.STTInput, opts audio.STTOptions) (*audio.TranscriptResult, error) {
|
||||
s.called++
|
||||
if s.delay > 0 {
|
||||
select {
|
||||
case <-time.After(s.delay):
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
return s.result, s.err
|
||||
}
|
||||
|
||||
// newTestChannel builds a minimal Channel with audioMgr for unit tests.
|
||||
func newTestChannelWithAudio(audioMgr *audio.Manager) *Channel {
|
||||
return &Channel{audioMgr: audioMgr}
|
||||
}
|
||||
|
||||
// newAudioMgr creates an audio.Manager with one stub STT provider.
|
||||
func newAudioMgr(stub *stubSTTProvider) *audio.Manager {
|
||||
mgr := audio.NewManager(audio.ManagerConfig{})
|
||||
mgr.RegisterSTT(stub)
|
||||
mgr.SetSTTChain([]string{stub.Name()})
|
||||
return mgr
|
||||
}
|
||||
|
||||
// Case 1: opt-in enabled, success — transcribeVoice calls Transcribe and returns transcript.
|
||||
func TestTranscribeVoice_OptInSuccess_CallsManager(t *testing.T) {
|
||||
stub := &stubSTTProvider{name: "stub", result: &audio.TranscriptResult{Text: "hello world"}}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
got := ch.transcribeVoice(context.Background(), "/tmp/voice.ogg", "audio/ogg", "en", settings)
|
||||
|
||||
if stub.called != 1 {
|
||||
t.Errorf("expected Transcribe called once, got %d", stub.called)
|
||||
}
|
||||
if got != "hello world" {
|
||||
t.Errorf("expected %q, got %q", "hello world", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 2: opt-in + Transcribe succeeds → returns transcript text.
|
||||
func TestTranscribeVoice_Success_ReturnsTranscript(t *testing.T) {
|
||||
stub := &stubSTTProvider{name: "stub", result: &audio.TranscriptResult{Text: "test transcript"}}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
got := ch.transcribeVoice(context.Background(), "/tmp/audio.ogg", "audio/ogg", "", settings)
|
||||
|
||||
if got != "test transcript" {
|
||||
t.Errorf("expected %q, got %q", "test transcript", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 3: STT error → fallback "[Voice message]" returned.
|
||||
func TestTranscribeVoice_STTError_ReturnsFallback(t *testing.T) {
|
||||
stub := &stubSTTProvider{name: "stub", err: errors.New("provider unavailable")}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
got := ch.transcribeVoice(context.Background(), "/tmp/voice.ogg", "audio/ogg", "en", settings)
|
||||
|
||||
fallback := i18n.T("en", i18n.MsgVoiceMessageFallback)
|
||||
if got != fallback {
|
||||
t.Errorf("expected fallback %q, got %q", fallback, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 4: ctx DeadlineExceeded → fallback, no crash.
|
||||
func TestTranscribeVoice_ContextDeadlineExceeded_ReturnsFallback(t *testing.T) {
|
||||
// Stub that blocks until ctx done — simulates slow Scribe.
|
||||
stub := &stubSTTProvider{name: "stub", delay: 5 * time.Second}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
got := ch.transcribeVoice(ctx, "/tmp/voice.ogg", "audio/ogg", "en", settings)
|
||||
|
||||
fallback := i18n.T("en", i18n.MsgVoiceMessageFallback)
|
||||
if got != fallback {
|
||||
t.Errorf("expected fallback %q, got %q", fallback, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 5: Manager has zero STT providers → ErrAllSTTProvidersFailed → fallback.
|
||||
func TestTranscribeVoice_NoProviders_ReturnsFallback(t *testing.T) {
|
||||
mgr := audio.NewManager(audio.ManagerConfig{})
|
||||
// No providers registered — Transcribe returns ErrAllSTTProvidersFailed.
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
got := ch.transcribeVoice(context.Background(), "/tmp/voice.ogg", "audio/ogg", "", settings)
|
||||
|
||||
fallback := i18n.T("", i18n.MsgVoiceMessageFallback)
|
||||
if got != fallback {
|
||||
t.Errorf("expected fallback %q, got %q", fallback, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 6: ctx Done before 10s wall clock → Transcribe aborts, returns fallback.
|
||||
func TestTranscribeVoice_CtxDoneBeforeTimeout_AbortsCleanly(t *testing.T) {
|
||||
stub := &stubSTTProvider{name: "stub", delay: 30 * time.Second}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: true}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // immediately cancelled
|
||||
|
||||
got := ch.transcribeVoice(ctx, "/tmp/voice.ogg", "audio/ogg", "vi", settings)
|
||||
|
||||
fallback := i18n.T("vi", i18n.MsgVoiceMessageFallback)
|
||||
if got != fallback {
|
||||
t.Errorf("expected fallback %q, got %q", fallback, got)
|
||||
}
|
||||
}
|
||||
|
||||
// Case 7: whatsapp_enabled=false (default opt-out) → Transcribe never called, fallback only.
|
||||
func TestTranscribeVoice_OptOut_NoTranscribeCall_ReturnsFallback(t *testing.T) {
|
||||
stub := &stubSTTProvider{name: "stub", result: &audio.TranscriptResult{Text: "should not appear"}}
|
||||
mgr := newAudioMgr(stub)
|
||||
ch := newTestChannelWithAudio(mgr)
|
||||
settings := sttSettings{WhatsappEnabled: false} // opt-out (default)
|
||||
|
||||
got := ch.transcribeVoice(context.Background(), "/tmp/voice.ogg", "audio/ogg", "en", settings)
|
||||
|
||||
if stub.called != 0 {
|
||||
t.Errorf("expected Transcribe NOT called, got %d calls", stub.called)
|
||||
}
|
||||
fallback := i18n.T("en", i18n.MsgVoiceMessageFallback)
|
||||
if got != fallback {
|
||||
t.Errorf("expected fallback %q, got %q", fallback, got)
|
||||
}
|
||||
}
|
||||
@@ -13,8 +13,8 @@ import (
|
||||
"go.mau.fi/whatsmeow/store/sqlstore"
|
||||
"go.mau.fi/whatsmeow/types"
|
||||
"go.mau.fi/whatsmeow/types/events"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/nextlevelbuilder/goclaw/internal/audio"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/bus"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/channels"
|
||||
"github.com/nextlevelbuilder/goclaw/internal/config"
|
||||
@@ -28,7 +28,7 @@ const (
|
||||
|
||||
func init() {
|
||||
// Set device name shown in WhatsApp's "Linked Devices" screen (once at package init).
|
||||
wastore.DeviceProps.Os = proto.String("GoClaw")
|
||||
wastore.DeviceProps.Os = new("GoClaw")
|
||||
}
|
||||
|
||||
// Channel connects directly to WhatsApp via go.mau.fi/whatsmeow.
|
||||
@@ -41,7 +41,9 @@ type Channel struct {
|
||||
mu sync.Mutex
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
parentCtx context.Context // stored from Start() for Reauth() context chain
|
||||
parentCtx context.Context // stored from Start() for Reauth() context chain
|
||||
audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT)
|
||||
builtinToolStore store.BuiltinToolStore // reads stt settings (whatsapp_enabled) per voice message; nil = opt-out
|
||||
|
||||
// QR state
|
||||
lastQRMu sync.RWMutex
|
||||
@@ -58,7 +60,6 @@ type Channel struct {
|
||||
// pairingService, pairingDebounce, approvedGroups, groupHistory are inherited from channels.BaseChannel.
|
||||
}
|
||||
|
||||
|
||||
// GetLastQRB64 returns the most recent QR PNG (base64).
|
||||
func (c *Channel) GetLastQRB64() string {
|
||||
c.lastQRMu.RLock()
|
||||
@@ -82,9 +83,12 @@ func (c *Channel) cacheQR(pngB64 string) {
|
||||
|
||||
// New creates a new WhatsApp channel backed by whatsmeow.
|
||||
// dialect must be "pgx" (PostgreSQL) or "sqlite3" (SQLite/desktop).
|
||||
// audioMgr is optional (nil = STT disabled).
|
||||
// builtinToolStore is optional (nil = STT permanently opt-out regardless of admin toggle).
|
||||
func New(cfg config.WhatsAppConfig, msgBus *bus.MessageBus,
|
||||
pairingSvc store.PairingStore, db *sql.DB,
|
||||
pendingStore store.PendingMessageStore, dialect string) (*Channel, error) {
|
||||
pendingStore store.PendingMessageStore, dialect string, audioMgr *audio.Manager,
|
||||
builtinToolStore store.BuiltinToolStore) (*Channel, error) {
|
||||
|
||||
base := channels.NewBaseChannel(channels.TypeWhatsApp, msgBus, cfg.AllowFrom)
|
||||
base.ValidatePolicy(cfg.DMPolicy, cfg.GroupPolicy)
|
||||
@@ -95,9 +99,11 @@ func New(cfg config.WhatsAppConfig, msgBus *bus.MessageBus,
|
||||
}
|
||||
|
||||
ch := &Channel{
|
||||
BaseChannel: base,
|
||||
config: cfg,
|
||||
container: container,
|
||||
BaseChannel: base,
|
||||
config: cfg,
|
||||
container: container,
|
||||
audioMgr: audioMgr,
|
||||
builtinToolStore: builtinToolStore,
|
||||
}
|
||||
ch.SetPairingService(pairingSvc)
|
||||
ch.SetGroupHistory(channels.MakeHistory("whatsapp", pendingStore, base.TenantID()))
|
||||
@@ -117,7 +123,6 @@ func (c *Channel) Start(ctx context.Context) error {
|
||||
return fmt.Errorf("whatsapp get device: %w", err)
|
||||
}
|
||||
|
||||
|
||||
c.client = whatsmeow.NewClient(deviceStore, nil)
|
||||
c.client.AddEventHandler(c.handleEvent)
|
||||
|
||||
|
||||
Reference in new issue
Block a user