tech/sendico
2
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into SEND066

Browse Source 
merge main into SEND066
This commit is contained in:
Arseni
2026-03-05 21:12:43 +03:00
parent d6a3a0cc5b 15393765b9
commit 39c04beb21
22 changed files with 528 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
api/gateway/tgsettle/config.dev.yml
View File

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

7
api/gateway/tgsettle/config.yml
View File

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

65
api/gateway/tgsettle/internal/server/internal/serverimp.go
View File

@@ -3,7 +3,6 @@ package serverimp
import (
"context"
"os"
"strings"
"time"
"github.com/tech/sendico/gateway/tgsettle/internal/service/gateway"
@@ -38,8 +37,6 @@ type config struct {
*grpcapp.Config `yaml:",inline"`
Gateway gatewayConfig `yaml:"gateway"`
Treasury treasuryConfig `yaml:"treasury"`
Ledger ledgerConfig `yaml:"ledger"` // deprecated: use treasury.ledger
Telegram telegramConfig `yaml:"telegram"` // deprecated: use treasury.telegram
}
type gatewayConfig struct {
@@ -50,20 +47,9 @@ type gatewayConfig struct {
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 {
ExecutionDelay time.Duration `yaml:"execution_delay"`
PollInterval time.Duration `yaml:"poll_interval"`
Telegram telegramConfig `yaml:"telegram"`
Ledger ledgerConfig `yaml:"ledger"`
Limits treasuryLimitsConfig `yaml:"limits"`
}
@@ -145,8 +131,6 @@ func (i *Imp) Start() error {
if cfg.Messaging != nil {
msgSettings = cfg.Messaging.Settings
}
treasuryTelegram := treasuryTelegramConfig(cfg, i.logger)
treasuryLedger := treasuryLedgerConfig(cfg, i.logger)
gwCfg := gateway.Config{
Rail: cfg.Gateway.Rail,
TargetChatIDEnv: cfg.Gateway.TargetChatIDEnv,
@@ -159,12 +143,8 @@ func (i *Imp) Start() error {
Treasury: gateway.TreasuryConfig{
ExecutionDelay: cfg.Treasury.ExecutionDelay,
PollInterval: cfg.Treasury.PollInterval,
Telegram: gateway.TelegramConfig{
AllowedChats: treasuryTelegram.AllowedChats,
Users: telegramUsers(treasuryTelegram.Users),
},
Ledger: gateway.LedgerConfig{
Timeout: treasuryLedger.Timeout,
Timeout: cfg.Treasury.Ledger.Timeout,
},
Limits: gateway.TreasuryLimitsConfig{
MaxAmountPerOperation: cfg.Treasury.Limits.MaxAmountPerOperation,
@@ -228,46 +208,3 @@ func (i *Imp) loadConfig() (*config, error) {
}
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
}

42
api/gateway/tgsettle/internal/service/gateway/service.go
View File

@@ -68,20 +68,9 @@ type Config struct {
Treasury TreasuryConfig
}
type TelegramConfig struct {
AllowedChats []string
Users []TelegramUserBinding
}
type TelegramUserBinding struct {
TelegramUserID string
LedgerAccount string
}
type TreasuryConfig struct {
ExecutionDelay time.Duration
PollInterval time.Duration
Telegram TelegramConfig
Ledger LedgerConfig
Limits TreasuryLimitsConfig
}
@@ -181,39 +170,13 @@ func (s *Service) Shutdown() {
}
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
}
if s.cfg.DiscoveryRegistry == nil {
s.logger.Warn("Treasury module disabled: discovery registry is unavailable")
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
if ledgerTimeout <= 0 {
@@ -241,10 +204,9 @@ func (s *Service) startTreasuryModule() {
module, err := treasurysvc.NewModule(
s.logger,
s.repo.TreasuryRequests(),
s.repo.TreasuryTelegramUsers(),
ledgerClient,
treasurysvc.Config{
AllowedChats: s.cfg.Treasury.Telegram.AllowedChats,
Users: users,
ExecutionDelay: executionDelay,
PollInterval: pollInterval,
MaxAmountPerOperation: s.cfg.Treasury.Limits.MaxAmountPerOperation,

5
api/gateway/tgsettle/internal/service/gateway/service_test.go
View File

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

2
api/gateway/tgsettle/internal/service/treasury/bot/markup.go
View File

@@ -14,5 +14,5 @@ func markdownCode(value string) string {
}
func markdownCommand(command Command) string {
return markdownCode(command.Slash())
return command.Slash()
}

90
api/gateway/tgsettle/internal/service/treasury/bot/router.go
View File

@@ -51,6 +51,16 @@ type TreasuryService interface {
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 {
error
LimitKind() string
@@ -65,9 +75,7 @@ type Router struct {
send SendTextFunc
tracker ScheduleTracker
allowedChats map[string]struct{}
userAccounts map[string]string
allowAnyChat bool
users UserBindingResolver
}
func NewRouter(
@@ -75,43 +83,23 @@ func NewRouter(
service TreasuryService,
send SendTextFunc,
tracker ScheduleTracker,
allowedChats []string,
userAccounts map[string]string,
users UserBindingResolver,
) *Router {
if logger != nil {
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{
logger: logger,
service: service,
dialogs: NewDialogs(),
send: send,
tracker: tracker,
allowedChats: allowed,
userAccounts: users,
allowAnyChat: len(allowed) == 0,
logger: logger,
service: service,
dialogs: NewDialogs(),
send: send,
tracker: tracker,
users: users,
}
}
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 {
@@ -138,20 +126,28 @@ func (r *Router) HandleUpdate(ctx context.Context, update *model.TelegramWebhook
)
}
if !r.allowAnyChat {
if _, ok := r.allowedChats[chatID]; !ok {
r.logUnauthorized(update)
_ = r.sendText(ctx, chatID, unauthorizedChatMessage)
return true
binding, err := r.users.ResolveUserBinding(ctx, userID)
if err != nil {
if r.logger != nil {
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
}
accountID, ok := r.userAccounts[userID]
if !ok || strings.TrimSpace(accountID) == "" {
if binding == nil || strings.TrimSpace(binding.LedgerAccountID) == "" {
r.logUnauthorized(update)
_ = r.sendText(ctx, chatID, unauthorizedMessage)
return true
}
if !isChatAllowed(chatID, binding.AllowedChatIDs) {
r.logUnauthorized(update)
_ = r.sendText(ctx, chatID, unauthorizedChatMessage)
return true
}
accountID := strings.TrimSpace(binding.LedgerAccountID)
switch command {
case CommandStart:
@@ -507,6 +503,22 @@ func (r *Router) resolveAccountProfile(ctx context.Context, ledgerAccountID stri
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 {
if value == 1 {
return "1 second"

97
api/gateway/tgsettle/internal/service/treasury/bot/router_test.go
View File

@@ -12,6 +12,21 @@ import (
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 {
return 30 * time.Second
}
@@ -54,8 +69,15 @@ func TestRouterUnauthorizedInAllowedChatSendsAccessDenied(t *testing.T) {
return nil
},
nil,
[]string{"100"},
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -85,8 +107,15 @@ func TestRouterUnknownChatGetsDenied(t *testing.T) {
return nil
},
nil,
[]string{"100"},
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -116,8 +145,14 @@ func TestRouterEmptyAllowedChats_AllowsAnyChatForAuthorizedUser(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -151,8 +186,14 @@ func TestRouterEmptyAllowedChats_UnauthorizedUserGetsDenied(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -182,8 +223,14 @@ func TestRouterStartAuthorizedShowsWelcome(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -213,8 +260,14 @@ func TestRouterHelpAuthorizedShowsHelp(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -244,8 +297,14 @@ func TestRouterStartUnauthorizedGetsDenied(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
@@ -275,8 +334,14 @@ func TestRouterPlainTextWithoutSession_ShowsSupportedCommands(t *testing.T) {
return nil
},
nil,
nil,
map[string]string{"123": "acct-1"},
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{

8
api/gateway/tgsettle/internal/service/treasury/config.go
View File

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

61
api/gateway/tgsettle/internal/service/treasury/module.go
View File

@@ -26,6 +26,7 @@ type Module struct {
func NewModule(
logger mlogger.Logger,
repo storage.TreasuryRequestsStore,
users storage.TreasuryTelegramUsersStore,
ledgerClient ledger.Client,
cfg Config,
send bot.SendTextFunc,
@@ -33,6 +34,9 @@ func NewModule(
if logger != nil {
logger = logger.Named("treasury")
}
if users == nil {
return nil, merrors.InvalidArgument("treasury telegram users store is required", "users")
}
service, err := NewService(
logger,
repo,
@@ -45,23 +49,13 @@ func NewModule(
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{
logger: logger,
service: service,
ledger: ledgerClient,
}
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
}
@@ -99,6 +93,28 @@ type botServiceAdapter struct {
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) {
if a == nil || a.svc == nil {
return 0
@@ -164,3 +180,26 @@ func (a *botServiceAdapter) CancelRequest(ctx context.Context, requestID string,
}
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
}

2
api/gateway/tgsettle/internal/service/treasury/service.go
View File

@@ -441,7 +441,7 @@ func (s *Service) logRequest(record *storagemodel.TreasuryRequest, status string
}
func newRequestID() string {
return "TGSETTLE-" + strings.ToUpper(bson.NewObjectID().Hex()[:8])
return "TG-TREASURY-" + strings.ToUpper(bson.NewObjectID().Hex())
}
func resolveAccountCode(account *ledger.Account, fallbackAccountID string) string {

5
api/gateway/tgsettle/storage/model/storable.go
View File

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

8
api/gateway/tgsettle/storage/model/treasury.go
View File

@@ -49,3 +49,11 @@ type TreasuryRequest struct {
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"`
}

11
api/gateway/tgsettle/storage/mongo/repository.go
View File

@@ -25,6 +25,7 @@ type Repository struct {
tg storage.TelegramConfirmationsStore
pending storage.PendingConfirmationsStore
treasury storage.TreasuryRequestsStore
users storage.TreasuryTelegramUsersStore
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"))
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)
if err != nil {
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.pending = pendingStore
result.treasury = treasuryStore
result.users = treasuryUsersStore
result.outbox = outboxStore
result.logger.Info("Payment gateway MongoDB storage initialised")
return result, nil
@@ -110,6 +117,10 @@ func (r *Repository) TreasuryRequests() storage.TreasuryRequestsStore {
return r.treasury
}
func (r *Repository) TreasuryTelegramUsers() storage.TreasuryTelegramUsersStore {
return r.users
}
func (r *Repository) Outbox() gatewayoutbox.Store {
return r.outbox
}

42
api/gateway/tgsettle/storage/mongo/store/treasury_requests.go
View File

@@ -296,20 +296,48 @@ func (t *TreasuryRequests) Update(ctx context.Context, record *model.TreasuryReq
Set(repository.Field("operationType"), record.OperationType).
Set(repository.Field("telegramUserId"), record.TelegramUserID).
Set(repository.Field("ledgerAccountId"), record.LedgerAccountID).
Set(repository.Field("ledgerAccountCode"), record.LedgerAccountCode).
Set(repository.Field("organizationRef"), record.OrganizationRef).
Set(repository.Field("chatId"), record.ChatID).
Set(repository.Field("amount"), record.Amount).
Set(repository.Field("currency"), record.Currency).
Set(repository.Field(fieldTreasuryStatus), record.Status).
Set(repository.Field("confirmedAt"), record.ConfirmedAt).
Set(repository.Field("scheduledAt"), record.ScheduledAt).
Set(repository.Field("executedAt"), record.ExecutedAt).
Set(repository.Field("cancelledAt"), record.CancelledAt).
Set(repository.Field(fieldTreasuryIdempotencyKey), record.IdempotencyKey).
Set(repository.Field("ledgerReference"), record.LedgerReference).
Set(repository.Field("errorMessage"), record.ErrorMessage).
Set(repository.Field(fieldTreasuryActive), record.Active)
if record.LedgerAccountCode != "" {
patch = patch.Set(repository.Field("ledgerAccountCode"), record.LedgerAccountCode)
} else {
patch = patch.Unset(repository.Field("ledgerAccountCode"))
}
if !record.ConfirmedAt.IsZero() {
patch = patch.Set(repository.Field("confirmedAt"), record.ConfirmedAt)
} else {
patch = patch.Unset(repository.Field("confirmedAt"))
}
if !record.ScheduledAt.IsZero() {
patch = patch.Set(repository.Field("scheduledAt"), record.ScheduledAt)
} else {
patch = patch.Unset(repository.Field("scheduledAt"))
}
if !record.ExecutedAt.IsZero() {
patch = patch.Set(repository.Field("executedAt"), record.ExecutedAt)
} else {
patch = patch.Unset(repository.Field("executedAt"))
}
if !record.CancelledAt.IsZero() {
patch = patch.Set(repository.Field("cancelledAt"), record.CancelledAt)
} else {
patch = patch.Unset(repository.Field("cancelledAt"))
}
if record.LedgerReference != "" {
patch = patch.Set(repository.Field("ledgerReference"), record.LedgerReference)
} else {
patch = patch.Unset(repository.Field("ledgerReference"))
}
if record.ErrorMessage != "" {
patch = patch.Set(repository.Field("errorMessage"), record.ErrorMessage)
} else {
patch = patch.Unset(repository.Field("errorMessage"))
}
if _, err := t.repo.PatchMany(ctx, repository.Filter(fieldTreasuryRequestID, record.RequestID), patch); err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
t.logger.Warn("Failed to update treasury request", zap.Error(err), zap.String("request_id", record.RequestID))

87
api/gateway/tgsettle/storage/mongo/store/treasury_telegram_users.go Normal file
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)

5
api/gateway/tgsettle/storage/storage.go
View File

@@ -15,6 +15,7 @@ type Repository interface {
TelegramConfirmations() TelegramConfirmationsStore
PendingConfirmations() PendingConfirmationsStore
TreasuryRequests() TreasuryRequestsStore
TreasuryTelegramUsers() TreasuryTelegramUsersStore
}
type PaymentsStore interface {
@@ -46,3 +47,7 @@ type TreasuryRequestsStore interface {
Update(ctx context.Context, record *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)
}

102
frontend/pweb/lib/controllers/payouts/multiple_payouts.dart
View File

@@ -1,23 +1,29 @@
import 'dart:async';
import 'package:flutter/foundation.dart';
import 'package:pshared/controllers/payment/source.dart';
import 'package:pshared/models/money.dart';
import 'package:pshared/models/payment/asset.dart';
import 'package:pshared/models/payment/chain_network.dart';
import 'package:pshared/models/payment/methods/data.dart';
import 'package:pshared/models/payment/methods/ledger.dart';
import 'package:pshared/models/payment/methods/managed_wallet.dart';
import 'package:pshared/models/payment/payment.dart';
import 'package:pshared/models/payment/quote/status_type.dart';
import 'package:pshared/models/payment/wallet.dart';
import 'package:pweb/models/payment/multiple_payouts/csv_row.dart';
import 'package:pweb/models/payment/multiple_payouts/state.dart';
import 'package:pweb/providers/multiple_payouts.dart';
import 'package:pweb/services/payments/csv_input.dart';
class MultiplePayoutsController extends ChangeNotifier {
final CsvInputService _csvInput;
MultiplePayoutsProvider? _provider;
PaymentSourceController? _sourceController;
_PickState _pickState = _PickState.idle;
Exception? _uiError;
String? _lastSourceKey;
MultiplePayoutsController({required CsvInputService csvInput})
: _csvInput = csvInput;
@@ -37,6 +43,7 @@ class MultiplePayoutsController extends ChangeNotifier {
_sourceController?.removeListener(_onSourceChanged);
_sourceController = sourceController;
_sourceController?.addListener(_onSourceChanged);
_lastSourceKey = _currentSourceKey;
shouldNotify = true;
}
if (shouldNotify) {
@@ -60,16 +67,16 @@ class MultiplePayoutsController extends ChangeNotifier {
_provider?.quoteStatusType ?? QuoteStatusType.missing;
Duration? get quoteTimeLeft => _provider?.quoteTimeLeft;
bool get canSend => (_provider?.canSend ?? false) && _selectedWallet != null;
bool get canSend => (_provider?.canSend ?? false) && _selectedSource != null;
Money? get aggregateDebitAmount =>
_provider?.aggregateDebitAmountFor(_selectedWallet);
_provider?.aggregateDebitAmountForCurrency(_selectedSourceCurrencyCode);
Money? get requestedSentAmount => _provider?.requestedSentAmount;
Money? get aggregateSettlementAmount =>
_provider?.aggregateSettlementAmountFor(_selectedWallet);
Money? get aggregateSettlementAmount => _provider
?.aggregateSettlementAmountForCurrency(_selectedSourceCurrencyCode);
Money? get aggregateFeeAmount =>
_provider?.aggregateFeeAmountFor(_selectedWallet);
_provider?.aggregateFeeAmountForCurrency(_selectedSourceCurrencyCode);
double? get aggregateFeePercent =>
_provider?.aggregateFeePercentFor(_selectedWallet);
_provider?.aggregateFeePercentForCurrency(_selectedSourceCurrencyCode);
Future<void> pickAndQuote() async {
if (_pickState == _PickState.picking) return;
@@ -84,15 +91,16 @@ class MultiplePayoutsController extends ChangeNotifier {
try {
final picked = await _csvInput.pickCsv();
if (picked == null) return;
final wallet = _selectedWallet;
if (wallet == null) {
_setUiError(StateError('Select source wallet first'));
final source = _selectedSource;
if (source == null) {
_setUiError(StateError('Select source of funds first'));
return;
}
await provider.quoteFromCsv(
fileName: picked.name,
content: picked.content,
sourceWallet: wallet,
sourceMethod: source.method,
sourceCurrencyCode: source.currencyCode,
);
} catch (e) {
_setUiError(e);
@@ -131,10 +139,78 @@ class MultiplePayoutsController extends ChangeNotifier {
}
void _onSourceChanged() {
final currentSourceKey = _currentSourceKey;
final sourceChanged = currentSourceKey != _lastSourceKey;
_lastSourceKey = currentSourceKey;
if (sourceChanged) {
unawaited(_requoteWithUploadedRows());
}
notifyListeners();
}
Wallet? get _selectedWallet => _sourceController?.selectedWallet;
String? get _selectedSourceCurrencyCode =>
_sourceController?.selectedCurrencyCode;
String? get _currentSourceKey {
final source = _sourceController;
if (source == null ||
source.selectedType == null ||
source.selectedRef == null) {
return null;
}
return '${source.selectedType!.name}:${source.selectedRef!}';
}
({PaymentMethodData method, String currencyCode})? get _selectedSource {
final source = _sourceController;
if (source == null) return null;
final currencyCode = source.selectedCurrencyCode;
if (currencyCode == null || currencyCode.isEmpty) return null;
final wallet = source.selectedWallet;
if (wallet != null) {
final hasAsset = (wallet.tokenSymbol ?? '').isNotEmpty;
final asset = hasAsset
? PaymentAsset(
chain: wallet.network ?? ChainNetwork.unspecified,
tokenSymbol: wallet.tokenSymbol!,
contractAddress: wallet.contractAddress,
)
: null;
return (
method: ManagedWalletPaymentMethod(
managedWalletRef: wallet.id,
asset: asset,
),
currencyCode: currencyCode,
);
}
final ledger = source.selectedLedgerAccount;
if (ledger != null) {
return (
method: LedgerPaymentMethod(ledgerAccountRef: ledger.ledgerAccountRef),
currencyCode: currencyCode,
);
}
return null;
}
Future<void> _requoteWithUploadedRows() async {
final provider = _provider;
if (provider == null) return;
if (provider.selectedFileName == null || provider.rows.isEmpty) return;
final source = _selectedSource;
if (source == null) return;
_clearUiError(notify: false);
await provider.requoteUploadedRows(
sourceMethod: source.method,
sourceCurrencyCode: source.currencyCode,
);
}
void _setUiError(Object error) {
_uiError = error is Exception ? error : Exception(error.toString());

13
frontend/pweb/lib/pages/dashboard/payouts/multiple/panels/source_quote/helpers.dart
View File

@@ -5,6 +5,8 @@ import 'package:pshared/utils/money.dart';
import 'package:pweb/controllers/payouts/multiple_payouts.dart';
import 'package:pweb/generated/i18n/app_localizations.dart';
String moneyLabel(Money? money) {
if (money == null) return 'N/A';
@@ -12,10 +14,7 @@ String moneyLabel(Money? money) {
if (amount.isNaN) return '${money.amount} ${money.currency}';
try {
return assetToString(
Asset(
currency: currencyStringToCode(money.currency),
amount: amount,
),
Asset(currency: currencyStringToCode(money.currency), amount: amount),
);
} catch (_) {
return '${money.amount} ${money.currency}';
@@ -31,6 +30,8 @@ String sentAmountLabel(MultiplePayoutsController controller) {
return moneyLabel(requested);
}
String feeLabel(MultiplePayoutsController controller) {
return moneyLabel(controller.aggregateFeeAmount);
String feeLabel(MultiplePayoutsController controller, AppLocalizations l10n) {
final fee = controller.aggregateFeeAmount;
if (fee == null) return l10n.noFee;
return moneyLabel(fee);
}

4
frontend/pweb/lib/pages/dashboard/payouts/multiple/panels/source_quote/summary.dart
View File

@@ -4,7 +4,7 @@ import 'package:pweb/controllers/payouts/multiple_payouts.dart';
import 'package:pweb/models/dashboard/summary_values.dart';
import 'package:pweb/pages/dashboard/payouts/multiple/panels/source_quote/helpers.dart';
import 'package:pweb/pages/dashboard/payouts/summary/widget.dart';
import 'package:pweb/generated/i18n/app_localizations.dart';
class SourceQuoteSummary extends StatelessWidget {
const SourceQuoteSummary({
@@ -21,7 +21,7 @@ class SourceQuoteSummary extends StatelessWidget {
return PaymentSummary(
spacing: spacing,
values: PaymentSummaryValues(
fee: feeLabel(controller),
fee: feeLabel(controller, AppLocalizations.of(context)!),
recipientReceives: moneyLabel(controller.aggregateSettlementAmount),
total: moneyLabel(controller.aggregateDebitAmount),
),

105
frontend/pweb/lib/providers/multiple_payouts.dart
View File

@@ -4,7 +4,7 @@ import 'package:pshared/models/money.dart';
import 'package:pshared/models/payment/payment.dart';
import 'package:pshared/models/payment/quote/quote.dart';
import 'package:pshared/models/payment/quote/status_type.dart';
import 'package:pshared/models/payment/wallet.dart';
import 'package:pshared/models/payment/methods/data.dart';
import 'package:pshared/provider/payment/multiple/provider.dart';
import 'package:pshared/provider/payment/multiple/quotation.dart';
import 'package:pshared/utils/currency.dart';
@@ -76,12 +76,12 @@ class MultiplePayoutsProvider extends ChangeNotifier {
return quoteRef != null && quoteRef.isNotEmpty;
}
Money? aggregateDebitAmountFor(Wallet? sourceWallet) {
Money? aggregateDebitAmountForCurrency(String? sourceCurrencyCode) {
if (_rows.isEmpty) return null;
final totals = aggregateMoneyByCurrency(
_quoteItems().map((quote) => quote.amounts?.sourceDebitTotal),
);
return _moneyForSourceCurrency(totals, sourceWallet);
return _moneyForSourceCurrency(totals, sourceCurrencyCode);
}
Money? get requestedSentAmount {
@@ -97,23 +97,23 @@ class MultiplePayoutsProvider extends ChangeNotifier {
return Money(amount: amountToString(total), currency: currency);
}
Money? aggregateSettlementAmountFor(Wallet? sourceWallet) {
Money? aggregateSettlementAmountForCurrency(String? sourceCurrencyCode) {
if (_rows.isEmpty) return null;
final totals = aggregateMoneyByCurrency(
_quoteItems().map((quote) => quote.amounts?.destinationSettlement),
);
return _moneyForSourceCurrency(totals, sourceWallet);
return _moneyForSourceCurrency(totals, sourceCurrencyCode);
}
Money? aggregateFeeAmountFor(Wallet? sourceWallet) {
Money? aggregateFeeAmountForCurrency(String? sourceCurrencyCode) {
if (_rows.isEmpty) return null;
final totals = aggregateMoneyByCurrency(_quoteItems().map(quoteFeeTotal));
return _moneyForSourceCurrency(totals, sourceWallet);
return _moneyForSourceCurrency(totals, sourceCurrencyCode);
}
double? aggregateFeePercentFor(Wallet? sourceWallet) {
final debit = aggregateDebitAmountFor(sourceWallet);
final fee = aggregateFeeAmountFor(sourceWallet);
double? aggregateFeePercentForCurrency(String? sourceCurrencyCode) {
final debit = aggregateDebitAmountForCurrency(sourceCurrencyCode);
final fee = aggregateFeeAmountForCurrency(sourceCurrencyCode);
if (debit == null || fee == null) return null;
final debitValue = parseMoneyAmount(debit.amount, fallback: double.nan);
@@ -126,7 +126,8 @@ class MultiplePayoutsProvider extends ChangeNotifier {
Future<void> quoteFromCsv({
required String fileName,
required String content,
required Wallet sourceWallet,
required PaymentMethodData sourceMethod,
required String sourceCurrencyCode,
}) async {
if (isBusy) return;
@@ -144,18 +145,43 @@ class MultiplePayoutsProvider extends ChangeNotifier {
_sentCount = 0;
final rows = _csvParser.parseRows(content);
final intents = _intentBuilder.buildIntents(sourceWallet, rows);
await _quoteRows(
quotation: quotation,
fileName: fileName,
rows: rows,
sourceMethod: sourceMethod,
sourceCurrencyCode: sourceCurrencyCode,
);
_selectedFileName = fileName;
_rows = rows;
if (quotation.error != null) {
_setErrorObject(quotation.error!);
}
} catch (e) {
_setErrorObject(e);
} finally {
_setState(MultiplePayoutsState.idle);
}
}
await quotation.quotePayments(
intents,
metadata: <String, String>{
'upload_filename': fileName,
'upload_rows': rows.length.toString(),
...?_uploadAmountMetadata(),
},
Future<void> requoteUploadedRows({
required PaymentMethodData sourceMethod,
required String sourceCurrencyCode,
}) async {
if (isBusy || _rows.isEmpty || _selectedFileName == null) return;
final quotation = _quotation;
if (quotation == null) return;
try {
_setState(MultiplePayoutsState.quoting);
_error = null;
_sentCount = 0;
await _quoteRows(
quotation: quotation,
fileName: _selectedFileName!,
rows: _rows,
sourceMethod: sourceMethod,
sourceCurrencyCode: sourceCurrencyCode,
);
if (quotation.error != null) {
@@ -254,13 +280,16 @@ class MultiplePayoutsProvider extends ChangeNotifier {
};
}
Money? _moneyForSourceCurrency(List<Money>? values, Wallet? sourceWallet) {
Money? _moneyForSourceCurrency(
List<Money>? values,
String? sourceCurrencyCode,
) {
if (values == null || values.isEmpty) return null;
if (sourceWallet != null) {
final sourceCurrency = currencyCodeToString(sourceWallet.currency);
if (sourceCurrencyCode != null && sourceCurrencyCode.isNotEmpty) {
final sourceCurrency = sourceCurrencyCode.trim().toUpperCase();
for (final value in values) {
if (value.currency.toUpperCase() == sourceCurrency.toUpperCase()) {
if (value.currency.toUpperCase() == sourceCurrency) {
return value;
}
}
@@ -272,6 +301,32 @@ class MultiplePayoutsProvider extends ChangeNotifier {
List<PaymentQuote> _quoteItems() =>
_quotation?.quotation?.items ?? const <PaymentQuote>[];
Future<void> _quoteRows({
required MultiQuotationProvider quotation,
required String fileName,
required List<CsvPayoutRow> rows,
required PaymentMethodData sourceMethod,
required String sourceCurrencyCode,
}) async {
final intents = _intentBuilder.buildIntents(
sourceMethod: sourceMethod,
sourceCurrency: sourceCurrencyCode,
rows: rows,
);
_selectedFileName = fileName;
_rows = rows;
await quotation.quotePayments(
intents,
metadata: <String, String>{
'upload_filename': fileName,
'upload_rows': rows.length.toString(),
...?_uploadAmountMetadata(),
},
);
}
@override
void dispose() {
_quotation?.removeListener(_onQuotationChanged);

29
frontend/pweb/lib/utils/payment/multiple_intent_builder.dart
View File

@@ -1,14 +1,10 @@
import 'package:pshared/models/money.dart';
import 'package:pshared/models/payment/asset.dart';
import 'package:pshared/models/payment/chain_network.dart';
import 'package:pshared/models/payment/fees/treatment.dart';
import 'package:pshared/models/payment/intent.dart';
import 'package:pshared/models/payment/kind.dart';
import 'package:pshared/models/payment/methods/card.dart';
import 'package:pshared/models/payment/methods/managed_wallet.dart';
import 'package:pshared/models/payment/methods/data.dart';
import 'package:pshared/models/payment/settlement_mode.dart';
import 'package:pshared/models/payment/wallet.dart';
import 'package:pshared/utils/currency.dart';
import 'package:pshared/utils/payment/fx_helpers.dart';
import 'package:pweb/models/payment/multiple_payouts/csv_row.dart';
@@ -16,19 +12,11 @@ import 'package:pweb/models/payment/multiple_payouts/csv_row.dart';
class MultipleIntentBuilder {
static const String _currency = 'RUB';
List<PaymentIntent> buildIntents(
Wallet sourceWallet,
List<CsvPayoutRow> rows,
) {
final sourceCurrency = currencyCodeToString(sourceWallet.currency);
final hasAsset = (sourceWallet.tokenSymbol ?? '').isNotEmpty;
final sourceAsset = hasAsset
? PaymentAsset(
chain: sourceWallet.network ?? ChainNetwork.unspecified,
tokenSymbol: sourceWallet.tokenSymbol!,
contractAddress: sourceWallet.contractAddress,
)
: null;
List<PaymentIntent> buildIntents({
required PaymentMethodData sourceMethod,
required String sourceCurrency,
required List<CsvPayoutRow> rows,
}) {
final fxIntent = FxIntentHelper.buildSellBaseBuyQuote(
baseCurrency: sourceCurrency,
quoteCurrency: _currency,
@@ -39,10 +27,7 @@ class MultipleIntentBuilder {
final amount = Money(amount: row.amount, currency: _currency);
return PaymentIntent(
kind: PaymentKind.payout,
source: ManagedWalletPaymentMethod(
managedWalletRef: sourceWallet.id,
asset: sourceAsset,
),
source: sourceMethod,
destination: CardPaymentMethod(
pan: row.pan,
firstName: row.firstName,