tg-670 #671

Merged
tech merged 2 commits from tg-670 into main 2026-03-05 13:47:38 +00:00
14 changed files with 306 additions and 188 deletions

View File

@@ -45,9 +45,6 @@ gateway:
treasury: treasury:
execution_delay: 60s execution_delay: 60s
poll_interval: 60s poll_interval: 60s
telegram:
allowed_chats: []
users: []
ledger: ledger:
timeout: 5s timeout: 5s
limits: limits:

View File

@@ -50,10 +50,3 @@ treasury:
limits: limits:
max_amount_per_operation: "" max_amount_per_operation: ""
max_daily_amount: "" max_daily_amount: ""
telegram:
allowed_chats: []
users:
- telegram_user_id: "8273799472"
ledger_account: "6972c738949b91ea0395e5fb"
- telegram_user_id: "8273507566"
ledger_account: "6995d6c118bca1d8baa5f2be"

View File

@@ -3,7 +3,6 @@ package serverimp
import ( import (
"context" "context"
"os" "os"
"strings"
"time" "time"
"github.com/tech/sendico/gateway/tgsettle/internal/service/gateway" "github.com/tech/sendico/gateway/tgsettle/internal/service/gateway"
@@ -38,8 +37,6 @@ type config struct {
*grpcapp.Config `yaml:",inline"` *grpcapp.Config `yaml:",inline"`
Gateway gatewayConfig `yaml:"gateway"` Gateway gatewayConfig `yaml:"gateway"`
Treasury treasuryConfig `yaml:"treasury"` Treasury treasuryConfig `yaml:"treasury"`
Ledger ledgerConfig `yaml:"ledger"` // deprecated: use treasury.ledger
Telegram telegramConfig `yaml:"telegram"` // deprecated: use treasury.telegram
} }
type gatewayConfig struct { type gatewayConfig struct {
@@ -50,20 +47,9 @@ type gatewayConfig struct {
SuccessReaction string `yaml:"success_reaction"` SuccessReaction string `yaml:"success_reaction"`
} }
type telegramConfig struct {
AllowedChats []string `yaml:"allowed_chats"`
Users []telegramUserConfig `yaml:"users"`
}
type telegramUserConfig struct {
TelegramUserID string `yaml:"telegram_user_id"`
LedgerAccount string `yaml:"ledger_account"`
}
type treasuryConfig struct { type treasuryConfig struct {
ExecutionDelay time.Duration `yaml:"execution_delay"` ExecutionDelay time.Duration `yaml:"execution_delay"`
PollInterval time.Duration `yaml:"poll_interval"` PollInterval time.Duration `yaml:"poll_interval"`
Telegram telegramConfig `yaml:"telegram"`
Ledger ledgerConfig `yaml:"ledger"` Ledger ledgerConfig `yaml:"ledger"`
Limits treasuryLimitsConfig `yaml:"limits"` Limits treasuryLimitsConfig `yaml:"limits"`
} }
@@ -145,8 +131,6 @@ func (i *Imp) Start() error {
if cfg.Messaging != nil { if cfg.Messaging != nil {
msgSettings = cfg.Messaging.Settings msgSettings = cfg.Messaging.Settings
} }
treasuryTelegram := treasuryTelegramConfig(cfg, i.logger)
treasuryLedger := treasuryLedgerConfig(cfg, i.logger)
gwCfg := gateway.Config{ gwCfg := gateway.Config{
Rail: cfg.Gateway.Rail, Rail: cfg.Gateway.Rail,
TargetChatIDEnv: cfg.Gateway.TargetChatIDEnv, TargetChatIDEnv: cfg.Gateway.TargetChatIDEnv,
@@ -159,12 +143,8 @@ func (i *Imp) Start() error {
Treasury: gateway.TreasuryConfig{ Treasury: gateway.TreasuryConfig{
ExecutionDelay: cfg.Treasury.ExecutionDelay, ExecutionDelay: cfg.Treasury.ExecutionDelay,
PollInterval: cfg.Treasury.PollInterval, PollInterval: cfg.Treasury.PollInterval,
Telegram: gateway.TelegramConfig{
AllowedChats: treasuryTelegram.AllowedChats,
Users: telegramUsers(treasuryTelegram.Users),
},
Ledger: gateway.LedgerConfig{ Ledger: gateway.LedgerConfig{
Timeout: treasuryLedger.Timeout, Timeout: cfg.Treasury.Ledger.Timeout,
}, },
Limits: gateway.TreasuryLimitsConfig{ Limits: gateway.TreasuryLimitsConfig{
MaxAmountPerOperation: cfg.Treasury.Limits.MaxAmountPerOperation, MaxAmountPerOperation: cfg.Treasury.Limits.MaxAmountPerOperation,
@@ -228,46 +208,3 @@ func (i *Imp) loadConfig() (*config, error) {
} }
return cfg, nil return cfg, nil
} }
func telegramUsers(input []telegramUserConfig) []gateway.TelegramUserBinding {
result := make([]gateway.TelegramUserBinding, 0, len(input))
for _, next := range input {
result = append(result, gateway.TelegramUserBinding{
TelegramUserID: strings.TrimSpace(next.TelegramUserID),
LedgerAccount: strings.TrimSpace(next.LedgerAccount),
})
}
return result
}
func treasuryTelegramConfig(cfg *config, logger mlogger.Logger) telegramConfig {
if cfg == nil {
return telegramConfig{}
}
if len(cfg.Treasury.Telegram.Users) > 0 || len(cfg.Treasury.Telegram.AllowedChats) > 0 {
return cfg.Treasury.Telegram
}
if len(cfg.Telegram.Users) > 0 || len(cfg.Telegram.AllowedChats) > 0 {
if logger != nil {
logger.Warn("Deprecated config path used: telegram.*; move these settings to treasury.telegram.*")
}
return cfg.Telegram
}
return cfg.Treasury.Telegram
}
func treasuryLedgerConfig(cfg *config, logger mlogger.Logger) ledgerConfig {
if cfg == nil {
return ledgerConfig{}
}
if cfg.Treasury.Ledger.Timeout > 0 {
return cfg.Treasury.Ledger
}
if cfg.Ledger.Timeout > 0 {
if logger != nil {
logger.Warn("Deprecated config path used: ledger.*; move these settings to treasury.ledger.*")
}
return cfg.Ledger
}
return cfg.Treasury.Ledger
}

View File

@@ -68,20 +68,9 @@ type Config struct {
Treasury TreasuryConfig Treasury TreasuryConfig
} }
type TelegramConfig struct {
AllowedChats []string
Users []TelegramUserBinding
}
type TelegramUserBinding struct {
TelegramUserID string
LedgerAccount string
}
type TreasuryConfig struct { type TreasuryConfig struct {
ExecutionDelay time.Duration ExecutionDelay time.Duration
PollInterval time.Duration PollInterval time.Duration
Telegram TelegramConfig
Ledger LedgerConfig Ledger LedgerConfig
Limits TreasuryLimitsConfig Limits TreasuryLimitsConfig
} }
@@ -181,39 +170,13 @@ func (s *Service) Shutdown() {
} }
func (s *Service) startTreasuryModule() { func (s *Service) startTreasuryModule() {
if s == nil || s.repo == nil || s.repo.TreasuryRequests() == nil { if s == nil || s.repo == nil || s.repo.TreasuryRequests() == nil || s.repo.TreasuryTelegramUsers() == nil {
return return
} }
if s.cfg.DiscoveryRegistry == nil { if s.cfg.DiscoveryRegistry == nil {
s.logger.Warn("Treasury module disabled: discovery registry is unavailable") s.logger.Warn("Treasury module disabled: discovery registry is unavailable")
return return
} }
configuredUsers := s.cfg.Treasury.Telegram.Users
if len(configuredUsers) == 0 {
return
}
users := make([]treasurysvc.UserBinding, 0, len(configuredUsers))
configuredUserIDs := make([]string, 0, len(configuredUsers))
for _, binding := range configuredUsers {
userID := strings.TrimSpace(binding.TelegramUserID)
accountID := strings.TrimSpace(binding.LedgerAccount)
if userID != "" {
configuredUserIDs = append(configuredUserIDs, userID)
}
if userID == "" || accountID == "" {
continue
}
users = append(users, treasurysvc.UserBinding{
TelegramUserID: userID,
LedgerAccount: accountID,
})
}
if len(users) == 0 {
s.logger.Warn("Treasury module disabled: no valid treasury.telegram.users bindings",
zap.Int("configured_bindings", len(configuredUsers)),
zap.Strings("configured_user_ids", configuredUserIDs))
return
}
ledgerTimeout := s.cfg.Treasury.Ledger.Timeout ledgerTimeout := s.cfg.Treasury.Ledger.Timeout
if ledgerTimeout <= 0 { if ledgerTimeout <= 0 {
@@ -241,10 +204,9 @@ func (s *Service) startTreasuryModule() {
module, err := treasurysvc.NewModule( module, err := treasurysvc.NewModule(
s.logger, s.logger,
s.repo.TreasuryRequests(), s.repo.TreasuryRequests(),
s.repo.TreasuryTelegramUsers(),
ledgerClient, ledgerClient,
treasurysvc.Config{ treasurysvc.Config{
AllowedChats: s.cfg.Treasury.Telegram.AllowedChats,
Users: users,
ExecutionDelay: executionDelay, ExecutionDelay: executionDelay,
PollInterval: pollInterval, PollInterval: pollInterval,
MaxAmountPerOperation: s.cfg.Treasury.Limits.MaxAmountPerOperation, MaxAmountPerOperation: s.cfg.Treasury.Limits.MaxAmountPerOperation,

View File

@@ -81,6 +81,7 @@ type fakeRepo struct {
tg *fakeTelegramStore tg *fakeTelegramStore
pending *fakePendingStore pending *fakePendingStore
treasury storage.TreasuryRequestsStore treasury storage.TreasuryRequestsStore
users storage.TreasuryTelegramUsersStore
} }
func (f *fakeRepo) Payments() storage.PaymentsStore { func (f *fakeRepo) Payments() storage.PaymentsStore {
@@ -99,6 +100,10 @@ func (f *fakeRepo) TreasuryRequests() storage.TreasuryRequestsStore {
return f.treasury return f.treasury
} }
func (f *fakeRepo) TreasuryTelegramUsers() storage.TreasuryTelegramUsersStore {
return f.users
}
type fakePendingStore struct { type fakePendingStore struct {
mu sync.Mutex mu sync.Mutex
records map[string]*storagemodel.PendingConfirmation records map[string]*storagemodel.PendingConfirmation

View File

@@ -51,6 +51,16 @@ type TreasuryService interface {
CancelRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error) CancelRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error)
} }
type UserBinding struct {
TelegramUserID string
LedgerAccountID string
AllowedChatIDs []string
}
type UserBindingResolver interface {
ResolveUserBinding(ctx context.Context, telegramUserID string) (*UserBinding, error)
}
type limitError interface { type limitError interface {
error error
LimitKind() string LimitKind() string
@@ -65,9 +75,7 @@ type Router struct {
send SendTextFunc send SendTextFunc
tracker ScheduleTracker tracker ScheduleTracker
allowedChats map[string]struct{} users UserBindingResolver
userAccounts map[string]string
allowAnyChat bool
} }
func NewRouter( func NewRouter(
@@ -75,43 +83,23 @@ func NewRouter(
service TreasuryService, service TreasuryService,
send SendTextFunc, send SendTextFunc,
tracker ScheduleTracker, tracker ScheduleTracker,
allowedChats []string, users UserBindingResolver,
userAccounts map[string]string,
) *Router { ) *Router {
if logger != nil { if logger != nil {
logger = logger.Named("treasury_router") logger = logger.Named("treasury_router")
} }
allowed := map[string]struct{}{}
for _, chatID := range allowedChats {
chatID = strings.TrimSpace(chatID)
if chatID == "" {
continue
}
allowed[chatID] = struct{}{}
}
users := map[string]string{}
for userID, accountID := range userAccounts {
userID = strings.TrimSpace(userID)
accountID = strings.TrimSpace(accountID)
if userID == "" || accountID == "" {
continue
}
users[userID] = accountID
}
return &Router{ return &Router{
logger: logger, logger: logger,
service: service, service: service,
dialogs: NewDialogs(), dialogs: NewDialogs(),
send: send, send: send,
tracker: tracker, tracker: tracker,
allowedChats: allowed, users: users,
userAccounts: users,
allowAnyChat: len(allowed) == 0,
} }
} }
func (r *Router) Enabled() bool { func (r *Router) Enabled() bool {
return r != nil && r.service != nil && len(r.userAccounts) > 0 return r != nil && r.service != nil && r.users != nil
} }
func (r *Router) HandleUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) bool { func (r *Router) HandleUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) bool {
@@ -138,20 +126,28 @@ func (r *Router) HandleUpdate(ctx context.Context, update *model.TelegramWebhook
) )
} }
if !r.allowAnyChat { binding, err := r.users.ResolveUserBinding(ctx, userID)
if _, ok := r.allowedChats[chatID]; !ok { if err != nil {
r.logUnauthorized(update) if r.logger != nil {
_ = r.sendText(ctx, chatID, unauthorizedChatMessage) r.logger.Warn("Failed to resolve treasury user binding",
zap.Error(err),
zap.String("telegram_user_id", userID),
zap.String("chat_id", chatID))
}
_ = r.sendText(ctx, chatID, "*Temporary issue*\nUnable to check treasury authorization right now. Please try again.")
return true return true
} }
} if binding == nil || strings.TrimSpace(binding.LedgerAccountID) == "" {
accountID, ok := r.userAccounts[userID]
if !ok || strings.TrimSpace(accountID) == "" {
r.logUnauthorized(update) r.logUnauthorized(update)
_ = r.sendText(ctx, chatID, unauthorizedMessage) _ = r.sendText(ctx, chatID, unauthorizedMessage)
return true return true
} }
if !isChatAllowed(chatID, binding.AllowedChatIDs) {
r.logUnauthorized(update)
_ = r.sendText(ctx, chatID, unauthorizedChatMessage)
return true
}
accountID := strings.TrimSpace(binding.LedgerAccountID)
switch command { switch command {
case CommandStart: case CommandStart:
@@ -507,6 +503,22 @@ func (r *Router) resolveAccountProfile(ctx context.Context, ledgerAccountID stri
return profile return profile
} }
func isChatAllowed(chatID string, allowedChatIDs []string) bool {
chatID = strings.TrimSpace(chatID)
if chatID == "" {
return false
}
if len(allowedChatIDs) == 0 {
return true
}
for _, allowed := range allowedChatIDs {
if strings.TrimSpace(allowed) == chatID {
return true
}
}
return false
}
func formatSeconds(value int64) string { func formatSeconds(value int64) string {
if value == 1 { if value == 1 {
return "1 second" return "1 second"

View File

@@ -12,6 +12,21 @@ import (
type fakeService struct{} type fakeService struct{}
type fakeUserBindingResolver struct {
bindings map[string]*UserBinding
err error
}
func (f fakeUserBindingResolver) ResolveUserBinding(_ context.Context, telegramUserID string) (*UserBinding, error) {
if f.err != nil {
return nil, f.err
}
if f.bindings == nil {
return nil, nil
}
return f.bindings[telegramUserID], nil
}
func (fakeService) ExecutionDelay() time.Duration { func (fakeService) ExecutionDelay() time.Duration {
return 30 * time.Second return 30 * time.Second
} }
@@ -54,8 +69,15 @@ func TestRouterUnauthorizedInAllowedChatSendsAccessDenied(t *testing.T) {
return nil return nil
}, },
nil, nil,
[]string{"100"}, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -85,8 +107,15 @@ func TestRouterUnknownChatGetsDenied(t *testing.T) {
return nil return nil
}, },
nil, nil,
[]string{"100"}, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -116,8 +145,14 @@ func TestRouterEmptyAllowedChats_AllowsAnyChatForAuthorizedUser(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -151,8 +186,14 @@ func TestRouterEmptyAllowedChats_UnauthorizedUserGetsDenied(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -182,8 +223,14 @@ func TestRouterStartAuthorizedShowsWelcome(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -213,8 +260,14 @@ func TestRouterHelpAuthorizedShowsHelp(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -244,8 +297,14 @@ func TestRouterStartUnauthorizedGetsDenied(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{
@@ -275,8 +334,14 @@ func TestRouterPlainTextWithoutSession_ShowsSupportedCommands(t *testing.T) {
return nil return nil
}, },
nil, nil,
nil, fakeUserBindingResolver{
map[string]string{"123": "acct-1"}, bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
) )
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{ handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{ Message: &model.TelegramMessage{

View File

@@ -2,15 +2,7 @@ package treasury
import "time" import "time"
type UserBinding struct {
TelegramUserID string
LedgerAccount string
}
type Config struct { type Config struct {
AllowedChats []string
Users []UserBinding
ExecutionDelay time.Duration ExecutionDelay time.Duration
PollInterval time.Duration PollInterval time.Duration

View File

@@ -26,6 +26,7 @@ type Module struct {
func NewModule( func NewModule(
logger mlogger.Logger, logger mlogger.Logger,
repo storage.TreasuryRequestsStore, repo storage.TreasuryRequestsStore,
users storage.TreasuryTelegramUsersStore,
ledgerClient ledger.Client, ledgerClient ledger.Client,
cfg Config, cfg Config,
send bot.SendTextFunc, send bot.SendTextFunc,
@@ -33,6 +34,9 @@ func NewModule(
if logger != nil { if logger != nil {
logger = logger.Named("treasury") logger = logger.Named("treasury")
} }
if users == nil {
return nil, merrors.InvalidArgument("treasury telegram users store is required", "users")
}
service, err := NewService( service, err := NewService(
logger, logger,
repo, repo,
@@ -45,23 +49,13 @@ func NewModule(
return nil, err return nil, err
} }
users := map[string]string{}
for _, binding := range cfg.Users {
userID := strings.TrimSpace(binding.TelegramUserID)
accountID := strings.TrimSpace(binding.LedgerAccount)
if userID == "" || accountID == "" {
continue
}
users[userID] = accountID
}
module := &Module{ module := &Module{
logger: logger, logger: logger,
service: service, service: service,
ledger: ledgerClient, ledger: ledgerClient,
} }
module.scheduler = NewScheduler(logger, service, NotifyFunc(send), cfg.PollInterval) module.scheduler = NewScheduler(logger, service, NotifyFunc(send), cfg.PollInterval)
module.router = bot.NewRouter(logger, &botServiceAdapter{svc: service}, send, module.scheduler, cfg.AllowedChats, users) module.router = bot.NewRouter(logger, &botServiceAdapter{svc: service}, send, module.scheduler, &botUsersAdapter{store: users})
return module, nil return module, nil
} }
@@ -99,6 +93,28 @@ type botServiceAdapter struct {
svc *Service svc *Service
} }
type botUsersAdapter struct {
store storage.TreasuryTelegramUsersStore
}
func (a *botUsersAdapter) ResolveUserBinding(ctx context.Context, telegramUserID string) (*bot.UserBinding, error) {
if a == nil || a.store == nil {
return nil, merrors.Internal("treasury users store unavailable")
}
record, err := a.store.FindByTelegramUserID(ctx, telegramUserID)
if err != nil {
return nil, err
}
if record == nil {
return nil, nil
}
return &bot.UserBinding{
TelegramUserID: strings.TrimSpace(record.TelegramUserID),
LedgerAccountID: strings.TrimSpace(record.LedgerAccountID),
AllowedChatIDs: normalizeChatIDs(record.AllowedChatIDs),
}, nil
}
func (a *botServiceAdapter) ExecutionDelay() (delay time.Duration) { func (a *botServiceAdapter) ExecutionDelay() (delay time.Duration) {
if a == nil || a.svc == nil { if a == nil || a.svc == nil {
return 0 return 0
@@ -164,3 +180,26 @@ func (a *botServiceAdapter) CancelRequest(ctx context.Context, requestID string,
} }
return a.svc.CancelRequest(ctx, requestID, telegramUserID) return a.svc.CancelRequest(ctx, requestID, telegramUserID)
} }
func normalizeChatIDs(values []string) []string {
if len(values) == 0 {
return nil
}
out := make([]string, 0, len(values))
seen := map[string]struct{}{}
for _, next := range values {
next = strings.TrimSpace(next)
if next == "" {
continue
}
if _, ok := seen[next]; ok {
continue
}
seen[next] = struct{}{}
out = append(out, next)
}
if len(out) == 0 {
return nil
}
return out
}

View File

@@ -5,6 +5,7 @@ const (
telegramConfirmationsCollection = "telegram_confirmations" telegramConfirmationsCollection = "telegram_confirmations"
pendingConfirmationsCollection = "pending_confirmations" pendingConfirmationsCollection = "pending_confirmations"
treasuryRequestsCollection = "treasury_requests" treasuryRequestsCollection = "treasury_requests"
treasuryTelegramUsersCollection = "treasury_telegram_users"
) )
func (*PaymentRecord) Collection() string { func (*PaymentRecord) Collection() string {
@@ -22,3 +23,7 @@ func (*PendingConfirmation) Collection() string {
func (*TreasuryRequest) Collection() string { func (*TreasuryRequest) Collection() string {
return treasuryRequestsCollection return treasuryRequestsCollection
} }
func (*TreasuryTelegramUser) Collection() string {
return treasuryTelegramUsersCollection
}

View File

@@ -49,3 +49,11 @@ type TreasuryRequest struct {
Active bool `bson:"active,omitempty" json:"active,omitempty"` Active bool `bson:"active,omitempty" json:"active,omitempty"`
} }
type TreasuryTelegramUser struct {
storable.Base `bson:",inline" json:",inline"`
TelegramUserID string `bson:"telegramUserId,omitempty" json:"telegram_user_id,omitempty"`
LedgerAccountID string `bson:"ledgerAccountId,omitempty" json:"ledger_account_id,omitempty"`
AllowedChatIDs []string `bson:"allowedChatIds,omitempty" json:"allowed_chat_ids,omitempty"`
}

View File

@@ -25,6 +25,7 @@ type Repository struct {
tg storage.TelegramConfirmationsStore tg storage.TelegramConfirmationsStore
pending storage.PendingConfirmationsStore pending storage.PendingConfirmationsStore
treasury storage.TreasuryRequestsStore treasury storage.TreasuryRequestsStore
users storage.TreasuryTelegramUsersStore
outbox gatewayoutbox.Store outbox gatewayoutbox.Store
} }
@@ -80,6 +81,11 @@ func New(logger mlogger.Logger, conn *db.MongoConnection) (*Repository, error) {
result.logger.Error("Failed to initialise treasury requests store", zap.Error(err), zap.String("store", "treasury_requests")) result.logger.Error("Failed to initialise treasury requests store", zap.Error(err), zap.String("store", "treasury_requests"))
return nil, err return nil, err
} }
treasuryUsersStore, err := store.NewTreasuryTelegramUsers(result.logger, result.db)
if err != nil {
result.logger.Error("Failed to initialise treasury telegram users store", zap.Error(err), zap.String("store", "treasury_telegram_users"))
return nil, err
}
outboxStore, err := gatewayoutbox.NewMongoStore(result.logger, result.db) outboxStore, err := gatewayoutbox.NewMongoStore(result.logger, result.db)
if err != nil { if err != nil {
result.logger.Error("Failed to initialise outbox store", zap.Error(err), zap.String("store", "outbox")) result.logger.Error("Failed to initialise outbox store", zap.Error(err), zap.String("store", "outbox"))
@@ -89,6 +95,7 @@ func New(logger mlogger.Logger, conn *db.MongoConnection) (*Repository, error) {
result.tg = tgStore result.tg = tgStore
result.pending = pendingStore result.pending = pendingStore
result.treasury = treasuryStore result.treasury = treasuryStore
result.users = treasuryUsersStore
result.outbox = outboxStore result.outbox = outboxStore
result.logger.Info("Payment gateway MongoDB storage initialised") result.logger.Info("Payment gateway MongoDB storage initialised")
return result, nil return result, nil
@@ -110,6 +117,10 @@ func (r *Repository) TreasuryRequests() storage.TreasuryRequestsStore {
return r.treasury return r.treasury
} }
func (r *Repository) TreasuryTelegramUsers() storage.TreasuryTelegramUsersStore {
return r.users
}
func (r *Repository) Outbox() gatewayoutbox.Store { func (r *Repository) Outbox() gatewayoutbox.Store {
return r.outbox return r.outbox
} }

View File

@@ -0,0 +1,87 @@
package store
import (
"context"
"errors"
"strings"
"github.com/tech/sendico/gateway/tgsettle/storage"
"github.com/tech/sendico/gateway/tgsettle/storage/model"
"github.com/tech/sendico/pkg/db/repository"
ri "github.com/tech/sendico/pkg/db/repository/index"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.uber.org/zap"
)
const (
treasuryTelegramUsersCollection = "treasury_telegram_users"
fieldTreasuryTelegramUserID = "telegramUserId"
)
type TreasuryTelegramUsers struct {
logger mlogger.Logger
repo repository.Repository
}
func NewTreasuryTelegramUsers(logger mlogger.Logger, db *mongo.Database) (*TreasuryTelegramUsers, error) {
if db == nil {
return nil, merrors.InvalidArgument("mongo database is nil")
}
if logger == nil {
logger = zap.NewNop()
}
logger = logger.Named("treasury_telegram_users").With(zap.String("collection", treasuryTelegramUsersCollection))
repo := repository.CreateMongoRepository(db, treasuryTelegramUsersCollection)
if err := repo.CreateIndex(&ri.Definition{
Keys: []ri.Key{{Field: fieldTreasuryTelegramUserID, Sort: ri.Asc}},
Unique: true,
}); err != nil {
logger.Error("Failed to create treasury telegram users user_id index", zap.Error(err), zap.String("index_field", fieldTreasuryTelegramUserID))
return nil, err
}
return &TreasuryTelegramUsers{
logger: logger,
repo: repo,
}, nil
}
func (t *TreasuryTelegramUsers) FindByTelegramUserID(ctx context.Context, telegramUserID string) (*model.TreasuryTelegramUser, error) {
telegramUserID = strings.TrimSpace(telegramUserID)
if telegramUserID == "" {
return nil, merrors.InvalidArgument("telegram_user_id is required", "telegram_user_id")
}
var result model.TreasuryTelegramUser
err := t.repo.FindOneByFilter(ctx, repository.Filter(fieldTreasuryTelegramUserID, telegramUserID), &result)
if errors.Is(err, merrors.ErrNoData) {
return nil, nil
}
if err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
t.logger.Warn("Failed to load treasury telegram user", zap.Error(err), zap.String("telegram_user_id", telegramUserID))
}
return nil, err
}
result.TelegramUserID = strings.TrimSpace(result.TelegramUserID)
result.LedgerAccountID = strings.TrimSpace(result.LedgerAccountID)
if len(result.AllowedChatIDs) > 0 {
normalized := make([]string, 0, len(result.AllowedChatIDs))
for _, next := range result.AllowedChatIDs {
next = strings.TrimSpace(next)
if next == "" {
continue
}
normalized = append(normalized, next)
}
result.AllowedChatIDs = normalized
}
if result.TelegramUserID == "" || result.LedgerAccountID == "" {
return nil, nil
}
return &result, nil
}
var _ storage.TreasuryTelegramUsersStore = (*TreasuryTelegramUsers)(nil)

View File

@@ -15,6 +15,7 @@ type Repository interface {
TelegramConfirmations() TelegramConfirmationsStore TelegramConfirmations() TelegramConfirmationsStore
PendingConfirmations() PendingConfirmationsStore PendingConfirmations() PendingConfirmationsStore
TreasuryRequests() TreasuryRequestsStore TreasuryRequests() TreasuryRequestsStore
TreasuryTelegramUsers() TreasuryTelegramUsersStore
} }
type PaymentsStore interface { type PaymentsStore interface {
@@ -46,3 +47,7 @@ type TreasuryRequestsStore interface {
Update(ctx context.Context, record *model.TreasuryRequest) error Update(ctx context.Context, record *model.TreasuryRequest) error
ListByAccountAndStatuses(ctx context.Context, ledgerAccountID string, statuses []model.TreasuryRequestStatus, dayStart, dayEnd time.Time) ([]model.TreasuryRequest, error) ListByAccountAndStatuses(ctx context.Context, ledgerAccountID string, statuses []model.TreasuryRequestStatus, dayStart, dayEnd time.Time) ([]model.TreasuryRequest, error)
} }
type TreasuryTelegramUsersStore interface {
FindByTelegramUserID(ctx context.Context, telegramUserID string) (*model.TreasuryTelegramUser, error)
}