diff --git a/.codegraph/.gitignore b/.codegraph/.gitignore new file mode 100644 index 0000000..1a3da33 --- /dev/null +++ b/.codegraph/.gitignore @@ -0,0 +1,17 @@ +# CodeGraph data files +# These are local to each machine and should not be committed + +# Database +*.db +*.db-wal +*.db-shm + +# Cache +cache/ + +# Logs +*.log +*.pid + +# Hook markers +.dirty diff --git a/README.md b/README.md index 2f39b1f..ee87d34 100644 --- a/README.md +++ b/README.md @@ -48,8 +48,7 @@ The bot reacts to a single greeting or to flooding within a defined interval win - Go 1.26.3 - [Golang Telegram Bot](https://github.com/go-telegram/bot) - [DeepSeek API](https://platform.deepseek.com/) - -Redis or some tiny DB like SQLite is on the TODO list, in case the bot needs to grow. +- [Redis](https://redis.io/) ## How to deploy on your own VDS diff --git a/README_RU.md b/README_RU.md index 8ac5afb..f33110f 100644 --- a/README_RU.md +++ b/README_RU.md @@ -47,8 +47,7 @@ Telegram открыл возможность установить бота-се - Go 1.26.3 - [Golang Telegram Bot](https://github.com/go-telegram/bot) - [DeepSeek API](https://platform.deepseek.com/) - -В TODO занесен Redis или какая-то простенькая базка типа SQLite, если потребуется расширение бота. +- [Redis](https://redis.io/) ## Как поднять у себя на VDS diff --git a/cmd/service/main.go b/cmd/service/main.go index 78db123..3106b69 100644 --- a/cmd/service/main.go +++ b/cmd/service/main.go @@ -1,12 +1,15 @@ package main import ( + "context" "fmt" "log/slog" "noirbot/internal/domain/repository" "noirbot/internal/domain/service" "noirbot/internal/gateways/deepseek" + httpgw "noirbot/internal/gateways/http" "noirbot/internal/gateways/memory" + redisstore "noirbot/internal/gateways/redis" "noirbot/internal/gateways/telegram/inbound" "noirbot/internal/gateways/telegram/outbound" "noirbot/internal/usecase/handle_business_connection" @@ -14,9 +17,8 @@ import ( "noirbot/pkg/config" "os" - httpgw "noirbot/internal/gateways/http" - "github.com/go-telegram/bot" + "github.com/redis/go-redis/v9" "go.uber.org/fx" ) @@ -29,6 +31,7 @@ func main() { newGreetingDetector, newFloodDetector, + newShortVoiceDetector, newOwnerWhitelist, newBusinessConnectionStore, @@ -40,6 +43,8 @@ func main() { newDeepseekConfig, newLLMClient, + newRedisClient, + newHandleBusinessMessageConfig, handle_business_connection.New, handle_business_message.New, @@ -55,6 +60,7 @@ func main() { fx.Invoke( wireLazyHandler, httpgw.RegisterRoutes, + bindRedisLifecycle, bindHTTPServerLifecycle, ), ) @@ -73,6 +79,24 @@ func bindHTTPServerLifecycle(lc fx.Lifecycle, s *httpgw.Server) { }) } +func bindRedisLifecycle(cfg *config.Config, lc fx.Lifecycle, r *redis.Client) { + lc.Append(fx.Hook{ + OnStart: func(ctx context.Context) error { + pingCtx, cancel := context.WithTimeout(ctx, cfg.Redis.DialTimeout) + defer cancel() + + if err := r.Ping(pingCtx).Err(); err != nil { + return fmt.Errorf("ping redis: %w", err) + } + + return nil + }, + OnStop: func(_ context.Context) error { + return r.Close() + }, + }) +} + func newLogger() *slog.Logger { return slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{ Level: slog.LevelInfo, @@ -100,16 +124,22 @@ func newFloodDetector(cfg *config.Config, store repository.MessageWindowStore) * }, store) } +func newShortVoiceDetector(cfg *config.Config) *service.ShortVoiceDetector { + return service.NewShortVoiceDetector(service.ShortVoiceDetectorConfig{ + MaxDuration: cfg.ShortVoice.MaxDuration, + }) +} + func newOwnerWhitelist(cfg *config.Config) repository.OwnerWhitelist { return memory.NewOwnerWhitelist(cfg.AllowedOwners) } -func newBusinessConnectionStore() repository.BusinessConnectionStore { - return memory.NewBusinessConnectionStore() +func newBusinessConnectionStore(r *redis.Client, cfg *config.Config) repository.BusinessConnectionStore { + return redisstore.NewBusinessConnectionStore(r, cfg.Redis.BusinessConnectionTTL) } -func newMessageWindowStore() repository.MessageWindowStore { - return memory.NewMessageWindowStore() +func newMessageWindowStore(r *redis.Client, cfg *config.Config) repository.MessageWindowStore { + return redisstore.NewMessageWindowStore(r, cfg.Flood.WindowDuration, cfg.Flood.RedisTTL) } func newBusinessSender(b *bot.Bot) repository.BusinessSender { @@ -135,6 +165,21 @@ func newLLMClient(c deepseek.Config) repository.LLMClient { func newHandleBusinessMessageConfig(cfg *config.Config) handle_business_message.Config { return handle_business_message.Config{ - SystemPrompt: cfg.Bot.SystemPrompt, + SystemPrompt: cfg.Bot.SystemPrompt, + ShortVoicePrompt: cfg.Bot.ShortVoicePrompt, } } + +func newRedisClient(cfg *config.Config) *redis.Client { + rdb := redis.NewClient(&redis.Options{ + Addr: cfg.Redis.Addr, + Password: cfg.Redis.Password, + DB: cfg.Redis.DB, + DialTimeout: cfg.Redis.DialTimeout, + ReadTimeout: cfg.Redis.ReadTimeout, + WriteTimeout: cfg.Redis.WriteTimeout, + PoolSize: cfg.Redis.PoolSize, + }) + + return rdb +} diff --git a/docker/docker-compose.prod.yaml b/docker/docker-compose.prod.yaml index bf7ad63..371fbe2 100644 --- a/docker/docker-compose.prod.yaml +++ b/docker/docker-compose.prod.yaml @@ -3,12 +3,17 @@ services: bot: image: ${BOT_IMAGE:?BOT_IMAGE is required} container_name: noirBot + depends_on: + redis: + condition: service_healthy restart: unless-stopped env_file: - .env networks: - traefik-network + - bot-internal + labels: - "traefik.enable=true" - "traefik.docker.network=telegram-n8n-bot_traefik-network" @@ -33,7 +38,34 @@ services: - "traefik.http.routers.noirbot-health.priority=90" - "traefik.http.routers.noirbot-health.service=noirbot" + redis: + image: redis:7-alpine + container_name: noirBot-redis + restart: unless-stopped + command: redis-server --appendonly yes + --maxmemory 256mb + --maxmemory-policy noeviction + --requirepass ${REDIS_PASSWORD:?REDIS_PASSWORD is required} + volumes: + - redis-data:/data + + networks: + - bot-internal + + healthcheck: + test: [ "CMD", "redis-cli", "ping" ] + interval: 5s + timeout: 3s + retries: 5 + start_period: 5s + networks: traefik-network: external: true - name: telegram-n8n-bot_traefik-network \ No newline at end of file + name: telegram-n8n-bot_traefik-network + + bot-internal: + driver: bridge + +volumes: + redis-data: diff --git a/env.example b/env.example index 94a827f..b8b52f0 100644 --- a/env.example +++ b/env.example @@ -2,10 +2,20 @@ BOT_TOKEN=wowthisisbottokenamaizing! BOT_DOMAIN=bot.example.com WEBHOOK_SECRET=andthisiswebhooksecret +# Production: container name from docker-compose.prod.yaml +REDIS_ADDR=bot-redis:6379 +# Local dev (when polling/local-compose is added): +# REDIS_ADDR=localhost:6379 +REDIS_PASSWORD= +REDIS_DB=0 +REDIS_BUSINESS_TTL=604800s +FLOOD_REDIS_TTL=120s + DEEPSEEK_API_KEY=guesswhatisit! DEEPSEEK_MODEL=deepseek-v4-pro BOT_SYSTEM_PROMPT="You are breathtaking!" +BOT_SHORT_VOICE_PROMPT="Short voice messages is corruption of mankind!" ALLOWED_OWNERS=2281489 GREETINGS=hi,hello,sup,yo diff --git a/go.mod b/go.mod index c3e70d4..cfb80ba 100644 --- a/go.mod +++ b/go.mod @@ -3,9 +3,11 @@ module noirbot go 1.26.3 require ( + github.com/alicebob/miniredis/v2 v2.38.0 github.com/gin-gonic/gin v1.12.0 github.com/go-telegram/bot v1.21.0 github.com/kelseyhightower/envconfig v1.4.0 + github.com/redis/go-redis/v9 v9.20.0 github.com/stretchr/testify v1.11.1 go.uber.org/fx v1.24.0 go.uber.org/mock v0.6.0 @@ -16,6 +18,7 @@ require ( github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/gabriel-vasile/mimetype v1.4.12 // indirect @@ -37,7 +40,9 @@ require ( github.com/quic-go/quic-go v0.59.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect + github.com/yuin/gopher-lua v1.1.1 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.uber.org/dig v1.19.0 // indirect go.uber.org/multierr v1.10.0 // indirect go.uber.org/zap v1.26.0 // indirect diff --git a/go.sum b/go.sum index 82f76f5..9aa750c 100644 --- a/go.sum +++ b/go.sum @@ -1,9 +1,17 @@ +github.com/alicebob/miniredis/v2 v2.38.0 h1:nZAzCR+Lj+Vxk4ZXzm2NuKq2O33RXj1XxJ2e2uP9jiw= +github.com/alicebob/miniredis/v2 v2.38.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k= github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -59,6 +67,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= +github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -76,8 +86,14 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY= github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= +github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= +github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/dig v1.19.0 h1:BACLhebsYdpQ7IROQ1AGPjrXcP5dF80U3gKoFzbaq/4= go.uber.org/dig v1.19.0/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE= go.uber.org/fx v1.24.0 h1:wE8mruvpg2kiiL1Vqd0CC+tr0/24XIB10Iwp2lLWzkg= diff --git a/internal/domain/model/message.go b/internal/domain/model/message.go index 21bc9bc..ee6f460 100644 --- a/internal/domain/model/message.go +++ b/internal/domain/model/message.go @@ -2,11 +2,20 @@ package model import "time" +type MessageKind string + +const ( + MessageKindText MessageKind = "text" + MessageKindVoice MessageKind = "voice" +) + type IncomingMessage struct { BusinessConnectionID string OwnerID int64 GuestID int64 + Kind MessageKind Text string + VoiceDuration time.Duration ReceivedAt time.Time } diff --git a/internal/domain/model/trigger.go b/internal/domain/model/trigger.go index e5faaf0..4746aab 100644 --- a/internal/domain/model/trigger.go +++ b/internal/domain/model/trigger.go @@ -3,9 +3,10 @@ package model type TriggerKind string const ( - TriggerKindNone TriggerKind = "none" - TriggerKindGreeting TriggerKind = "greeting" - TriggerKindFlood TriggerKind = "flood" + TriggerKindNone TriggerKind = "none" + TriggerKindGreeting TriggerKind = "greeting" + TriggerKindFlood TriggerKind = "flood" + TriggerKindShortVoice TriggerKind = "short_voice" ) type TriggerDecision struct { diff --git a/internal/domain/service/short_voice_detector.go b/internal/domain/service/short_voice_detector.go new file mode 100644 index 0000000..b1cad27 --- /dev/null +++ b/internal/domain/service/short_voice_detector.go @@ -0,0 +1,35 @@ +package service + +import ( + "noirbot/internal/domain/model" + "time" +) + +type ShortVoiceDetectorConfig struct { + MaxDuration time.Duration +} + +type ShortVoiceDetector struct { + cfg ShortVoiceDetectorConfig +} + +func NewShortVoiceDetector(cfg ShortVoiceDetectorConfig) *ShortVoiceDetector { + return &ShortVoiceDetector{ + cfg: cfg, + } +} + +func (d *ShortVoiceDetector) Detect(msg model.IncomingMessage) model.TriggerDecision { + if msg.Kind != model.MessageKindVoice { + return model.TriggerDecision{Kind: model.TriggerKindNone} + } + + if msg.VoiceDuration > d.cfg.MaxDuration { + return model.TriggerDecision{Kind: model.TriggerKindNone} + } + + return model.TriggerDecision{ + Kind: model.TriggerKindShortVoice, + Reason: "voice " + msg.VoiceDuration.String() + " <= " + d.cfg.MaxDuration.String(), + } +} diff --git a/internal/domain/service/short_voice_detector_test.go b/internal/domain/service/short_voice_detector_test.go new file mode 100644 index 0000000..561fa96 --- /dev/null +++ b/internal/domain/service/short_voice_detector_test.go @@ -0,0 +1,69 @@ +package service_test + +import ( + "noirbot/internal/domain/model" + "noirbot/internal/domain/service" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestShortVoiceDetector_Detect(t *testing.T) { + const maxDuration = 10 * time.Second + + tests := []struct { + name string + msg model.IncomingMessage + wantKind model.TriggerKind + }{ + { + name: "voice короче порога — триггер short_voice", + msg: model.IncomingMessage{ + Kind: model.MessageKindVoice, + VoiceDuration: 5 * time.Second, + }, + wantKind: model.TriggerKindShortVoice, + }, + { + name: "voice ровно на пороге — триггер (граничный случай)", + msg: model.IncomingMessage{ + Kind: model.MessageKindVoice, + VoiceDuration: maxDuration, + }, + wantKind: model.TriggerKindShortVoice, + }, + { + name: "voice длиннее порога — пропускаем", + msg: model.IncomingMessage{ + Kind: model.MessageKindVoice, + VoiceDuration: 30 * time.Second, + }, + wantKind: model.TriggerKindNone, + }, + { + name: "текстовое сообщение — детектор не реагирует", + msg: model.IncomingMessage{ + Kind: model.MessageKindText, + Text: "привет", + }, + wantKind: model.TriggerKindNone, + }, + { + name: "пустой Kind — детектор не реагирует", + msg: model.IncomingMessage{}, + wantKind: model.TriggerKindNone, + }, + } + + detector := service.NewShortVoiceDetector(service.ShortVoiceDetectorConfig{ + MaxDuration: maxDuration, + }) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := detector.Detect(tt.msg) + require.Equal(t, tt.wantKind, got.Kind) + }) + } +} diff --git a/internal/gateways/memory/business_connection_store.go b/internal/gateways/memory/business_connection_store.go index b7904ce..4272ee8 100644 --- a/internal/gateways/memory/business_connection_store.go +++ b/internal/gateways/memory/business_connection_store.go @@ -14,6 +14,8 @@ type BusinessConnectionStore struct { conns map[string]model.BusinessConnection } +// Deprecated: business connection live in Redis now. Use +// redis.NewBusinessConnectionStore instead. Kept as a local-dev fallback. func NewBusinessConnectionStore() *BusinessConnectionStore { return &BusinessConnectionStore{ conns: make(map[string]model.BusinessConnection), diff --git a/internal/gateways/memory/message_window_store.go b/internal/gateways/memory/message_window_store.go index d590fd2..0e91cb8 100644 --- a/internal/gateways/memory/message_window_store.go +++ b/internal/gateways/memory/message_window_store.go @@ -20,6 +20,8 @@ type MessageWindowStore struct { windows map[windowKey][]time.Time } +// Deprecated: message windows live in Redis now. Use +// redis.NewMessageWindowStore instead. Kept as a local-dev fallback. func NewMessageWindowStore() *MessageWindowStore { return &MessageWindowStore{ windows: make(map[windowKey][]time.Time), diff --git a/internal/gateways/redis/business_connection_store.go b/internal/gateways/redis/business_connection_store.go new file mode 100644 index 0000000..3e3031f --- /dev/null +++ b/internal/gateways/redis/business_connection_store.go @@ -0,0 +1,81 @@ +package redis + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "noirbot/internal/domain/model" + "noirbot/internal/domain/repository" + "time" + + "github.com/redis/go-redis/v9" +) + +const bcKeyPrefix = "bc:" + +var _ repository.BusinessConnectionStore = (*BusinessConnectionStore)(nil) + +type BusinessConnectionStore struct { + client *redis.Client + ttl time.Duration +} + +func NewBusinessConnectionStore(client *redis.Client, ttl time.Duration) *BusinessConnectionStore { + return &BusinessConnectionStore{ + client: client, + ttl: ttl, + } +} + +func (s *BusinessConnectionStore) Get( + ctx context.Context, + connectionID string, +) (model.BusinessConnection, bool, error) { + key := bcKeyPrefix + connectionID + + data, err := s.client.Get(ctx, key).Result() + if errors.Is(err, redis.Nil) { + return model.BusinessConnection{}, false, nil + } + + if err != nil { + return model.BusinessConnection{}, false, fmt.Errorf("redis get business_connection %s: %w", connectionID, err) + } + + var dto businessConnectionDTO + if unMshErr := json.Unmarshal([]byte(data), &dto); unMshErr != nil { + return model.BusinessConnection{}, false, fmt.Errorf("redis unmarshal business_connection %s: %w", connectionID, unMshErr) + } + + return fromBusinessConnectionDTO(dto), true, nil +} + +func (s *BusinessConnectionStore) Put(ctx context.Context, conn model.BusinessConnection) error { + if conn.ID == "" { + return ErrEmptyConnectionID + } + + key := bcKeyPrefix + conn.ID + + data, err := json.Marshal(toBusinessConnectionDTO(conn)) + if err != nil { + return fmt.Errorf("redis marshal business_connection %s: %w", conn.ID, err) + } + + if setErr := s.client.Set(ctx, key, data, s.ttl).Err(); setErr != nil { + return fmt.Errorf("redis set business_connection %s: %w", conn.ID, setErr) + } + + return nil +} + +func (s *BusinessConnectionStore) Delete(ctx context.Context, connectionID string) error { + key := bcKeyPrefix + connectionID + + if err := s.client.Del(ctx, key).Err(); err != nil { + return fmt.Errorf("redis del business_connection %s: %w", connectionID, err) + } + + return nil +} diff --git a/internal/gateways/redis/business_connection_store_test.go b/internal/gateways/redis/business_connection_store_test.go new file mode 100644 index 0000000..5ebf051 --- /dev/null +++ b/internal/gateways/redis/business_connection_store_test.go @@ -0,0 +1,124 @@ +// internal/gateways/redis/business_connection_store_test.go +package redis_test + +import ( + "context" + "noirbot/internal/domain/model" + "testing" + "time" + + redisstore "noirbot/internal/gateways/redis" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const bcTestTTL = time.Hour + +// newBCStore wires a miniredis-backed BusinessConnectionStore for tests. +// The underlying client is closed automatically when the test finishes. +func newBCStore(t *testing.T) (*redisstore.BusinessConnectionStore, *miniredis.Miniredis) { + t.Helper() + + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + + t.Cleanup(func() { _ = client.Close() }) + + return redisstore.NewBusinessConnectionStore(client, bcTestTTL), mr +} + +// sampleConnection returns a fully-populated business connection. +// UserChatID is included on purpose to guard against the regression where +// the field was silently dropped in the DTO mapper. +func sampleConnection() model.BusinessConnection { + return model.BusinessConnection{ + ID: "conn-42", + Owner: model.Owner{UserID: 111}, + UserChatID: 222, + IsEnabled: true, + CanReply: true, + ConnectedAt: time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC), + } +} + +func TestBusinessConnectionStore_Get_Missing(t *testing.T) { + store, _ := newBCStore(t) + + conn, ok, err := store.Get(context.Background(), "nope") + + require.NoError(t, err) + assert.False(t, ok) + assert.Equal(t, model.BusinessConnection{}, conn) +} + +func TestBusinessConnectionStore_Get_InvalidJSON(t *testing.T) { + store, mr := newBCStore(t) + require.NoError(t, mr.Set("bc:bad", "{not json")) + + _, _, err := store.Get(context.Background(), "bad") + + require.Error(t, err) + assert.Contains(t, err.Error(), "redis unmarshal business_connection bad") +} + +func TestBusinessConnectionStore_Put_EmptyID(t *testing.T) { + store, _ := newBCStore(t) + + conn := sampleConnection() + conn.ID = "" + + err := store.Put(context.Background(), conn) + + require.ErrorIs(t, err, redisstore.ErrEmptyConnectionID) +} + +func TestBusinessConnectionStore_Put_AppliesTTL(t *testing.T) { + store, mr := newBCStore(t) + conn := sampleConnection() + + require.NoError(t, store.Put(context.Background(), conn)) + + key := "bc:" + conn.ID + assert.True(t, mr.Exists(key)) + assert.Equal(t, bcTestTTL, mr.TTL(key)) +} + +func TestBusinessConnectionStore_Delete_Missing(t *testing.T) { + store, _ := newBCStore(t) + + err := store.Delete(context.Background(), "ghost") + + assert.NoError(t, err) +} + +func TestBusinessConnectionStore_Delete_Existing(t *testing.T) { + store, mr := newBCStore(t) + conn := sampleConnection() + require.NoError(t, store.Put(context.Background(), conn)) + + key := "bc:" + conn.ID + require.True(t, mr.Exists(key)) + + err := store.Delete(context.Background(), conn.ID) + + require.NoError(t, err) + assert.False(t, mr.Exists(key)) +} + +// TestBusinessConnectionStore_RoundTrip verifies that every domain field +// survives JSON marshal/unmarshal through the DTO mapper. This is the +// regression guard for the UserChatID-drop bug. +func TestBusinessConnectionStore_RoundTrip(t *testing.T) { + store, _ := newBCStore(t) + original := sampleConnection() + require.NoError(t, store.Put(context.Background(), original)) + + got, ok, err := store.Get(context.Background(), original.ID) + + require.NoError(t, err) + require.True(t, ok) + assert.Equal(t, original, got) +} diff --git a/internal/gateways/redis/errors.go b/internal/gateways/redis/errors.go new file mode 100644 index 0000000..198bda3 --- /dev/null +++ b/internal/gateways/redis/errors.go @@ -0,0 +1,5 @@ +package redis + +import "errors" + +var ErrEmptyConnectionID = errors.New("redis bc: empty connection ID") diff --git a/internal/gateways/redis/mapper.go b/internal/gateways/redis/mapper.go new file mode 100644 index 0000000..e56e0c1 --- /dev/null +++ b/internal/gateways/redis/mapper.go @@ -0,0 +1,40 @@ +package redis + +import ( + "noirbot/internal/domain/model" + "time" +) + +type businessConnectionDTO struct { + ID string `json:"id"` + Owner ownerDTO `json:"owner"` + UserChatID int64 `json:"userChatId"` + IsEnabled bool `json:"isEnabled"` + CanReply bool `json:"canReply"` + ConnectedAt time.Time `json:"connectedAt"` +} +type ownerDTO struct { + UserID int64 `json:"userId"` +} + +func toBusinessConnectionDTO(c model.BusinessConnection) businessConnectionDTO { + return businessConnectionDTO{ + ID: c.ID, + Owner: ownerDTO{UserID: c.Owner.UserID}, + UserChatID: c.UserChatID, + IsEnabled: c.IsEnabled, + CanReply: c.CanReply, + ConnectedAt: c.ConnectedAt, + } +} + +func fromBusinessConnectionDTO(dto businessConnectionDTO) model.BusinessConnection { + return model.BusinessConnection{ + ID: dto.ID, + Owner: model.Owner{UserID: dto.Owner.UserID}, + UserChatID: dto.UserChatID, + IsEnabled: dto.IsEnabled, + CanReply: dto.CanReply, + ConnectedAt: dto.ConnectedAt, + } +} diff --git a/internal/gateways/redis/message_window_store.go b/internal/gateways/redis/message_window_store.go new file mode 100644 index 0000000..7e0d99e --- /dev/null +++ b/internal/gateways/redis/message_window_store.go @@ -0,0 +1,72 @@ +package redis + +import ( + "context" + "fmt" + "noirbot/internal/domain/model" + "noirbot/internal/domain/repository" + "strconv" + "sync/atomic" + "time" + + "github.com/redis/go-redis/v9" +) + +const floodKeyPrefix = "flood:" + +var _ repository.MessageWindowStore = (*MessageWindowStore)(nil) + +type MessageWindowStore struct { + client *redis.Client + window time.Duration + ttl time.Duration + memberSeq atomic.Uint64 +} + +func NewMessageWindowStore(client *redis.Client, window, ttl time.Duration) *MessageWindowStore { + return &MessageWindowStore{ + client: client, + window: window, + ttl: ttl, + } +} + +func (s *MessageWindowStore) Append(ctx context.Context, ownerID, guestID int64, _ model.IncomingMessage) error { + key := floodKey(ownerID, guestID) + now := time.Now() + score := float64(now.UnixNano()) + cutoff := strconv.FormatInt(now.Add(-s.window).UnixNano(), 10) + + pipe := s.client.TxPipeline() + pipe.ZAdd(ctx, key, redis.Z{Score: score, Member: s.newMember(now)}) + pipe.ZRemRangeByScore(ctx, key, "-inf", cutoff) + pipe.Expire(ctx, key, s.ttl) + + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("redis flood append owner=%d guest=%d: %w", ownerID, guestID, err) + } + + return nil +} + +func (s *MessageWindowStore) CountSince(ctx context.Context, ownerID, guestID int64, since time.Time) (int, error) { + key := floodKey(ownerID, guestID) + minScore := "(" + strconv.FormatInt(since.UnixNano(), 10) + + count, err := s.client.ZCount(ctx, key, minScore, "+inf").Result() + if err != nil { + return 0, fmt.Errorf("redis flood count owner=%d guest=%d: %w", ownerID, guestID, err) + } + + return int(count), nil +} + +func (s *MessageWindowStore) newMember(now time.Time) string { + seq := s.memberSeq.Add(1) + + return strconv.FormatInt(now.UnixNano(), 10) + ":" + strconv.FormatUint(seq, 10) +} + +func floodKey(ownerID, guestID int64) string { + return floodKeyPrefix + strconv.FormatInt(ownerID, 10) + ":" + strconv.FormatInt(guestID, 10) +} diff --git a/internal/gateways/redis/message_window_store_test.go b/internal/gateways/redis/message_window_store_test.go new file mode 100644 index 0000000..6b76e21 --- /dev/null +++ b/internal/gateways/redis/message_window_store_test.go @@ -0,0 +1,193 @@ +// internal/gateways/redis/message_window_store_test.go +package redis_test + +import ( + "context" + "noirbot/internal/domain/model" + "sync" + "testing" + "time" + + redisstore "noirbot/internal/gateways/redis" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Redis EXPIRE has 1-second granularity; using sub-second TTL would be +// silently rounded up by go-redis and break exact-match assertions. +// floodTestWindow stays sub-second because it is only used as a domain +// duration (compared client-side as nanoseconds) and never reaches EXPIRE. +const ( + floodTestWindow = 100 * time.Millisecond + floodTestTTL = 2 * time.Second + floodTestPoolSize = 50 + concurrentAppends = 100 + + testOwnerA int64 = 1 + testGuestA int64 = 2 + testOwnerB int64 = 3 + testGuestB int64 = 4 +) + +// since0 is a "definitely in the past" bound used as the lower edge for +// CountSince when we just want to count everything that ever was appended. +var since0 = time.Unix(0, 0) + +// newWindowStore wires a miniredis-backed MessageWindowStore for tests. +// The underlying client is closed automatically when the test finishes. +func newWindowStore(t *testing.T) (*redisstore.MessageWindowStore, *miniredis.Miniredis) { + t.Helper() + + mr := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{ + Addr: mr.Addr(), + PoolSize: floodTestPoolSize, + }) + + t.Cleanup(func() { _ = client.Close() }) + + return redisstore.NewMessageWindowStore(client, floodTestWindow, floodTestTTL), mr +} + +func TestMessageWindowStore_Append_Single(t *testing.T) { + store, _ := newWindowStore(t) + ctx := context.Background() + + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + + count, err := store.CountSince(ctx, testOwnerA, testGuestA, since0) + require.NoError(t, err) + assert.Equal(t, 1, count) +} + +func TestMessageWindowStore_Append_Multiple(t *testing.T) { + const writes = 3 + + store, _ := newWindowStore(t) + ctx := context.Background() + + for range writes { + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + } + + count, err := store.CountSince(ctx, testOwnerA, testGuestA, since0) + require.NoError(t, err) + assert.Equal(t, writes, count) +} + +func TestMessageWindowStore_CountSince_Empty(t *testing.T) { + store, _ := newWindowStore(t) + + count, err := store.CountSince(context.Background(), testOwnerA, testGuestA, since0) + + require.NoError(t, err) + assert.Equal(t, 0, count) +} + +func TestMessageWindowStore_CountSince_FutureBound(t *testing.T) { + store, _ := newWindowStore(t) + ctx := context.Background() + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + + future := time.Now().Add(time.Hour) + count, err := store.CountSince(ctx, testOwnerA, testGuestA, future) + + require.NoError(t, err) + assert.Equal(t, 0, count) +} + +// TestMessageWindowStore_AppendPrunesStale verifies that Append removes +// entries that have aged out of the sliding window. We use a very short +// window, sleep past it, then Append again and assert the previous entry +// is gone. +func TestMessageWindowStore_AppendPrunesStale(t *testing.T) { + store, mr := newWindowStore(t) + ctx := context.Background() + + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + + time.Sleep(floodTestWindow + 50*time.Millisecond) + + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + + keys := mr.Keys() + require.Len(t, keys, 1) + + members, err := mr.ZMembers(keys[0]) + require.NoError(t, err) + assert.Len(t, members, 1, "stale entry must be pruned by Append") +} + +func TestMessageWindowStore_AppliesTTL(t *testing.T) { + store, mr := newWindowStore(t) + + require.NoError(t, store.Append(context.Background(), testOwnerA, testGuestA, model.IncomingMessage{})) + + keys := mr.Keys() + require.Len(t, keys, 1) + assert.Equal(t, floodTestTTL, mr.TTL(keys[0])) +} + +func TestMessageWindowStore_KeyExpires(t *testing.T) { + store, mr := newWindowStore(t) + require.NoError(t, store.Append(context.Background(), testOwnerA, testGuestA, model.IncomingMessage{})) + + keys := mr.Keys() + require.Len(t, keys, 1) + + mr.FastForward(floodTestTTL + time.Second) + + assert.False(t, mr.Exists(keys[0]), "key must expire after TTL") +} + +func TestMessageWindowStore_IsolatesPairs(t *testing.T) { + store, mr := newWindowStore(t) + ctx := context.Background() + + require.NoError(t, store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{})) + require.NoError(t, store.Append(ctx, testOwnerB, testGuestB, model.IncomingMessage{})) + + countA, err := store.CountSince(ctx, testOwnerA, testGuestA, since0) + require.NoError(t, err) + assert.Equal(t, 1, countA) + + countB, err := store.CountSince(ctx, testOwnerB, testGuestB, since0) + require.NoError(t, err) + assert.Equal(t, 1, countB) + + cross, err := store.CountSince(ctx, testOwnerA, testGuestB, since0) + require.NoError(t, err) + assert.Equal(t, 0, cross, "different (owner, guest) pairs must not share a window") + + assert.Len(t, mr.Keys(), 2) +} + +// TestMessageWindowStore_Concurrent verifies that the atomic per-process +// counter produces unique ZSET members under load. If members collided, +// the second ZADD with the same member would silently overwrite the first +// and the count would drop below N. +func TestMessageWindowStore_Concurrent(t *testing.T) { + store, _ := newWindowStore(t) + ctx := context.Background() + + var wg sync.WaitGroup + + wg.Add(concurrentAppends) + + for range concurrentAppends { + go func() { + defer wg.Done() + + _ = store.Append(ctx, testOwnerA, testGuestA, model.IncomingMessage{}) + }() + } + + wg.Wait() + + count, err := store.CountSince(ctx, testOwnerA, testGuestA, since0) + require.NoError(t, err) + assert.Equal(t, concurrentAppends, count) +} diff --git a/internal/gateways/telegram/inbound/mapper.go b/internal/gateways/telegram/inbound/mapper.go index 5445d3f..20c6ec0 100644 --- a/internal/gateways/telegram/inbound/mapper.go +++ b/internal/gateways/telegram/inbound/mapper.go @@ -14,16 +14,30 @@ func NewUpdateMapper() *UpdateMapper { } func (m *UpdateMapper) ToIncomingMessage(src *tgmodels.Message) (model.IncomingMessage, bool) { - if src == nil || src.From == nil || src.Text == "" { + if src == nil || src.From == nil { return model.IncomingMessage{}, false } - return model.IncomingMessage{ + base := model.IncomingMessage{ BusinessConnectionID: src.BusinessConnectionID, GuestID: src.From.ID, - Text: src.Text, - ReceivedAt: time.Unix(int64(src.Date), 0), - }, true + ReceivedAt: time.Now().UTC(), + } + + switch { + case src.Text != "": + base.Kind = model.MessageKindText + base.Text = src.Text + + return base, true + case src.Voice != nil: + base.Kind = model.MessageKindVoice + base.VoiceDuration = time.Duration(src.Voice.Duration) * time.Second + + return base, true + default: + return model.IncomingMessage{}, false + } } func (m *UpdateMapper) ToBusinessConnection(src *tgmodels.BusinessConnection) model.BusinessConnection { diff --git a/internal/gateways/telegram/inbound/mapper_test.go b/internal/gateways/telegram/inbound/mapper_test.go new file mode 100644 index 0000000..c6476c1 --- /dev/null +++ b/internal/gateways/telegram/inbound/mapper_test.go @@ -0,0 +1,88 @@ +package inbound_test + +import ( + "noirbot/internal/domain/model" + "noirbot/internal/gateways/telegram/inbound" + "testing" + "time" + + tgmodels "github.com/go-telegram/bot/models" + "github.com/stretchr/testify/require" +) + +func TestUpdateMapper_ToIncomingMessage(t *testing.T) { + mapper := inbound.NewUpdateMapper() + from := &tgmodels.User{ID: 999} + + tests := []struct { + name string + src *tgmodels.Message + wantOK bool + wantKind model.MessageKind + wantText string + wantDur time.Duration + }{ + { + name: "text сообщение → MessageKindText", + src: &tgmodels.Message{ + BusinessConnectionID: "conn-1", + From: from, + Text: "привет", + }, + wantOK: true, + wantKind: model.MessageKindText, + wantText: "привет", + }, + { + name: "voice сообщение → MessageKindVoice + длительность", + src: &tgmodels.Message{ + BusinessConnectionID: "conn-1", + From: from, + Voice: &tgmodels.Voice{Duration: 7}, + }, + wantOK: true, + wantKind: model.MessageKindVoice, + wantDur: 7 * time.Second, + }, + { + name: "audio file (не voice) → пропускаем как неподдерживаемый тип", + src: &tgmodels.Message{ + BusinessConnectionID: "conn-1", + From: from, + Audio: &tgmodels.Audio{Duration: 5}, + }, + wantOK: false, + }, + { + name: "ни text, ни voice → пропускаем", + src: &tgmodels.Message{From: from}, + wantOK: false, + }, + { + name: "From == nil → пропускаем", + src: &tgmodels.Message{Text: "x"}, + wantOK: false, + }, + { + name: "src == nil → пропускаем", + src: nil, + wantOK: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := mapper.ToIncomingMessage(tt.src) + + require.Equal(t, tt.wantOK, ok) + + if !tt.wantOK { + return + } + + require.Equal(t, tt.wantKind, got.Kind) + require.Equal(t, tt.wantText, got.Text) + require.Equal(t, tt.wantDur, got.VoiceDuration) + }) + } +} diff --git a/internal/usecase/handle_business_message/usecase.go b/internal/usecase/handle_business_message/usecase.go index a0f55da..b269dce 100644 --- a/internal/usecase/handle_business_message/usecase.go +++ b/internal/usecase/handle_business_message/usecase.go @@ -10,19 +10,26 @@ import ( ) type Config struct { - SystemPrompt string + SystemPrompt string + ShortVoicePrompt string } type Usecase struct { - cfg Config - whitelist repository.OwnerWhitelist - connStore repository.BusinessConnectionStore - accountReader repository.BusinessAccountReader - greetingDetector *service.GreetingDetector - floodDetector *service.FloodDetector - llm repository.LLMClient - sender repository.BusinessSender - log *slog.Logger + cfg Config + whitelist repository.OwnerWhitelist + connStore repository.BusinessConnectionStore + accountReader repository.BusinessAccountReader + greetingDetector *service.GreetingDetector + floodDetector *service.FloodDetector + shortVoiceDetector *service.ShortVoiceDetector + llm repository.LLMClient + sender repository.BusinessSender + log *slog.Logger +} + +type llmInput struct { + SystemPrompt string + UserText string } func New( @@ -32,20 +39,22 @@ func New( accountReader repository.BusinessAccountReader, greetingDetector *service.GreetingDetector, floodDetector *service.FloodDetector, + shortVoiceDetector *service.ShortVoiceDetector, llm repository.LLMClient, sender repository.BusinessSender, log *slog.Logger, ) *Usecase { return &Usecase{ - cfg: cfg, - whitelist: whitelist, - connStore: connStore, - accountReader: accountReader, - greetingDetector: greetingDetector, - floodDetector: floodDetector, - llm: llm, - sender: sender, - log: log.With("usecase", "handle_business_message"), + cfg: cfg, + whitelist: whitelist, + connStore: connStore, + accountReader: accountReader, + greetingDetector: greetingDetector, + floodDetector: floodDetector, + shortVoiceDetector: shortVoiceDetector, + llm: llm, + sender: sender, + log: log.With("usecase", "handle_business_message"), } } @@ -95,7 +104,9 @@ func (uc *Usecase) Execute(ctx context.Context, msg model.IncomingMessage) error ) } - reply, err := uc.llm.Generate(ctx, uc.cfg.SystemPrompt, msg.Text) + in := uc.llmInputs(msg) + + reply, err := uc.llm.Generate(ctx, in.SystemPrompt, in.UserText) if err != nil { return fmt.Errorf("%w: %w", ErrLLMGenerate, err) } @@ -135,9 +146,24 @@ func (uc *Usecase) resolveOwner(ctx context.Context, connectionID string) (model } func (uc *Usecase) classify(ctx context.Context, msg model.IncomingMessage) (model.TriggerDecision, error) { - if decision := uc.greetingDetector.Detect(msg); decision.ShouldReply() { - return decision, nil + switch msg.Kind { + case model.MessageKindVoice: + return uc.shortVoiceDetector.Detect(msg), nil + case model.MessageKindText: + if decision := uc.greetingDetector.Detect(msg); decision.ShouldReply() { + return decision, nil + } + + return uc.floodDetector.Detect(ctx, msg) + default: + return model.TriggerDecision{Kind: model.TriggerKindNone}, nil + } +} + +func (uc *Usecase) llmInputs(msg model.IncomingMessage) llmInput { + if msg.Kind == model.MessageKindVoice { + return llmInput{SystemPrompt: uc.cfg.ShortVoicePrompt, UserText: ""} } - return uc.floodDetector.Detect(ctx, msg) + return llmInput{SystemPrompt: uc.cfg.SystemPrompt, UserText: msg.Text} } diff --git a/internal/usecase/handle_business_message/usecase_test.go b/internal/usecase/handle_business_message/usecase_test.go index 803e865..38bf1e6 100644 --- a/internal/usecase/handle_business_message/usecase_test.go +++ b/internal/usecase/handle_business_message/usecase_test.go @@ -30,11 +30,20 @@ var ( testMsg = model.IncomingMessage{ BusinessConnectionID: "conn-1", GuestID: 999, + Kind: model.MessageKindText, Text: "привет", ReceivedAt: time.Now(), } - testReply = "Ну какой привет, пиши сразу, что тебе надо!" - systemPrompt = "Отвечай как нуарный детектив, повидавший некоторое дерьмо" + testVoiceMsg = model.IncomingMessage{ + BusinessConnectionID: "conn-1", + GuestID: 999, + Kind: model.MessageKindVoice, + VoiceDuration: 5 * time.Second, + ReceivedAt: time.Now(), + } + testReply = "Ну какой привет, пиши сразу, что тебе надо!" + systemPrompt = "Отвечай как нуарный детектив, повидавший некоторое дерьмо" + shortVoicePrompt = "Тебе пришло голосовое — отреагируй нуарно" ) func expectShowThinking(ctx context.Context, sender *mock.MockBusinessSender, msg model.IncomingMessage) { @@ -69,13 +78,21 @@ func newUsecase( Threshold: 5, }, windowStore) + shortVoice := service.NewShortVoiceDetector(service.ShortVoiceDetectorConfig{ + MaxDuration: 10 * time.Second, + }) + return New( - Config{SystemPrompt: systemPrompt}, + Config{ + SystemPrompt: systemPrompt, + ShortVoicePrompt: shortVoicePrompt, + }, whitelist, connStore, accountReader, greeting, flood, + shortVoice, llm, sender, slog.Default(), @@ -149,6 +166,7 @@ func TestUsecase_Execute(t *testing.T) { msg: model.IncomingMessage{ BusinessConnectionID: "conn-1", GuestID: 999, + Kind: model.MessageKindText, Text: "это очень длинное сообщение которое точно больше двадцати символов", ReceivedAt: time.Now(), }, @@ -259,6 +277,53 @@ func TestUsecase_Execute(t *testing.T) { msg: testMsg, wantErr: nil, }, + { + name: "short voice ≤ порога — LLM вызван с short voice prompt и пустым userText", + setup: func(ctrl *gomock.Controller) *Usecase { + whitelist := mock.NewMockOwnerWhitelist(ctrl) + connStore := mock.NewMockBusinessConnectionStore(ctrl) + accountReader := mock.NewMockBusinessAccountReader(ctrl) + llm := mock.NewMockLLMClient(ctrl) + sender := mock.NewMockBusinessSender(ctrl) + + connStore.EXPECT().Get(ctx, testConn.ID).Return(testConn, true, nil) + whitelist.EXPECT().IsAllowed(ctx, testConn.Owner.UserID).Return(true, nil) + expectShowThinking(ctx, sender, testVoiceMsg) + llm.EXPECT().Generate(ctx, shortVoicePrompt, "").Return(testReply, nil) + sender.EXPECT().Send(ctx, model.ReplyDraft{ + BusinessConnectionID: testVoiceMsg.BusinessConnectionID, + GuestID: testVoiceMsg.GuestID, + Text: testReply, + }).Return(nil) + + return newUsecase(t, whitelist, connStore, accountReader, llm, sender) + }, + msg: testVoiceMsg, + wantErr: nil, + }, + { + name: "long voice > порога — бот молчит, LLM не вызывается", + setup: func(ctrl *gomock.Controller) *Usecase { + whitelist := mock.NewMockOwnerWhitelist(ctrl) + connStore := mock.NewMockBusinessConnectionStore(ctrl) + accountReader := mock.NewMockBusinessAccountReader(ctrl) + llm := mock.NewMockLLMClient(ctrl) + sender := mock.NewMockBusinessSender(ctrl) + + connStore.EXPECT().Get(ctx, testConn.ID).Return(testConn, true, nil) + whitelist.EXPECT().IsAllowed(ctx, testConn.Owner.UserID).Return(true, nil) + + return newUsecase(t, whitelist, connStore, accountReader, llm, sender) + }, + msg: model.IncomingMessage{ + BusinessConnectionID: "conn-1", + GuestID: 999, + Kind: model.MessageKindVoice, + VoiceDuration: 30 * time.Second, + ReceivedAt: time.Now(), + }, + wantErr: nil, + }, } for _, tt := range tests { diff --git a/pkg/config/config.go b/pkg/config/config.go index feaef54..e321fae 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -10,11 +10,13 @@ import ( type Config struct { Telegram TelegramConfig HTTP HTTPConfig + Redis RedisConfig DeepSeek DeepSeekConfig Bot BotConfig Flood FloodConfig - Greetings []string `default:"привет,прив,здоров,хай,ку" envconfig:"GREETINGS"` - AllowedOwners []int64 `envconfig:"ALLOWED_OWNERS"` + Greetings []string `default:"привет,прив,здоров,хай,ку" envconfig:"GREETINGS"` + ShortVoice ShortVoiceConfig + AllowedOwners []int64 `envconfig:"ALLOWED_OWNERS"` } type TelegramConfig struct { @@ -29,6 +31,17 @@ type HTTPConfig struct { ShutdownTimeout time.Duration `default:"5s" envconfig:"HTTP_SHUTDOWN_TIMEOUT"` } +type RedisConfig struct { + Addr string `default:"localhost:6379" envconfig:"REDIS_ADDR"` + Password string `default:"" envconfig:"REDIS_PASSWORD"` + DB int `default:"0" envconfig:"REDIS_DB"` + DialTimeout time.Duration `default:"5s" envconfig:"REDIS_DIAL_TIMEOUT"` + ReadTimeout time.Duration `default:"3s" envconfig:"REDIS_READ_TIMEOUT"` + WriteTimeout time.Duration `default:"3s" envconfig:"REDIS_WRITE_TIMEOUT"` + PoolSize int `default:"20" envconfig:"REDIS_POOL_SIZE"` + BusinessConnectionTTL time.Duration `default:"604800s" envconfig:"REDIS_BUSINESS_TTL"` +} + type DeepSeekConfig struct { BaseURL string `default:"https://api.deepseek.com/v1" envconfig:"DEEPSEEK_BASE_URL"` APIKey string `envconfig:"DEEPSEEK_API_KEY" required:"true"` @@ -37,13 +50,19 @@ type DeepSeekConfig struct { } type BotConfig struct { - SystemPrompt string `envconfig:"BOT_SYSTEM_PROMPT" required:"true"` + SystemPrompt string `envconfig:"BOT_SYSTEM_PROMPT" required:"true"` + ShortVoicePrompt string `envconfig:"BOT_SHORT_VOICE_PROMPT" required:"true"` } type FloodConfig struct { - WindowDuration time.Duration `default:"60s" envconfig:"FLOOD_WINDOW"` - MaxLen int `default:"20" envconfig:"FLOOD_MAX_LEN"` - Threshold int `default:"5" envconfig:"FLOOD_THRESHOLD"` + WindowDuration time.Duration `default:"60s" envconfig:"FLOOD_WINDOW"` + MaxLen int `default:"20" envconfig:"FLOOD_MAX_LEN"` + Threshold int `default:"5" envconfig:"FLOOD_THRESHOLD"` + RedisTTL time.Duration `default:"120s" envconfig:"FLOOD_REDIS_TTL"` +} + +type ShortVoiceConfig struct { + MaxDuration time.Duration `default:"10s" envconfig:"SHORT_VOICE_MAX_DURATION"` } func Load() (*Config, error) { @@ -52,5 +71,21 @@ func Load() (*Config, error) { return nil, fmt.Errorf("load config: %w", err) } + if err := cfg.validate(); err != nil { + return nil, fmt.Errorf("validate config: %w", err) + } + return cfg, nil } + +func (c *Config) validate() error { + if c.Flood.RedisTTL < c.Flood.WindowDuration { + return fmt.Errorf("%w: ttl=%s, window=%s", + ErrInvalidFloodTTL, + c.Flood.RedisTTL, + c.Flood.WindowDuration, + ) + } + + return nil +} diff --git a/pkg/config/errors.go b/pkg/config/errors.go new file mode 100644 index 0000000..ba22870 --- /dev/null +++ b/pkg/config/errors.go @@ -0,0 +1,5 @@ +package config + +import "errors" + +var ErrInvalidFloodTTL = errors.New("config: FLOOD_REDIS_TTL must be >= FLOOD_WINDOW")