Chimera Settle service

This commit is contained in:
Stephan D
2026-03-06 15:42:32 +01:00
parent ea5ec79a6e
commit 10bcdb4fe2
43 changed files with 8070 additions and 0 deletions

View File

@@ -0,0 +1,27 @@
package appversion
import (
"github.com/tech/sendico/pkg/version"
vf "github.com/tech/sendico/pkg/version/factory"
)
// Build information. Populated at build-time.
var (
Version string
Revision string
Branch string
BuildUser string
BuildDate string
)
func Create() version.Printer {
info := version.Info{
Program: "Sendico Payment Gateway Service",
Revision: Revision,
Branch: Branch,
BuildUser: BuildUser,
BuildDate: BuildDate,
Version: Version,
}
return vf.Create(&info)
}

View File

@@ -0,0 +1,230 @@
package serverimp
import (
"context"
"os"
"strings"
"time"
"github.com/tech/sendico/gateway/chsettle/internal/service/gateway"
"github.com/tech/sendico/gateway/chsettle/storage"
gatewaymongo "github.com/tech/sendico/gateway/chsettle/storage/mongo"
"github.com/tech/sendico/pkg/api/routers"
"github.com/tech/sendico/pkg/db"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
msg "github.com/tech/sendico/pkg/messaging"
mb "github.com/tech/sendico/pkg/messaging/broker"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/server/grpcapp"
"go.uber.org/zap"
"gopkg.in/yaml.v3"
)
type Imp struct {
logger mlogger.Logger
file string
debug bool
config *config
app *grpcapp.App[storage.Repository]
service *gateway.Service
discoveryWatcher *discovery.RegistryWatcher
discoveryReg *discovery.Registry
}
type config struct {
*grpcapp.Config `yaml:",inline"`
Gateway gatewayConfig `yaml:"gateway"`
Treasury treasuryConfig `yaml:"treasury"`
}
type gatewayConfig struct {
Rail string `yaml:"rail"`
TargetChatIDEnv string `yaml:"target_chat_id_env"`
TimeoutSeconds int32 `yaml:"timeout_seconds"`
AcceptedUserIDs []string `yaml:"accepted_user_ids"`
SuccessReaction string `yaml:"success_reaction"`
}
type treasuryConfig struct {
ExecutionDelay time.Duration `yaml:"execution_delay"`
PollInterval time.Duration `yaml:"poll_interval"`
Ledger ledgerConfig `yaml:"ledger"`
Limits treasuryLimitsConfig `yaml:"limits"`
}
type treasuryLimitsConfig struct {
MaxAmountPerOperation string `yaml:"max_amount_per_operation"`
MaxDailyAmount string `yaml:"max_daily_amount"`
}
type ledgerConfig struct {
Timeout time.Duration `yaml:"timeout"`
}
func Create(logger mlogger.Logger, file string, debug bool) (*Imp, error) {
return &Imp{
logger: logger.Named("server"),
file: file,
debug: debug,
}, nil
}
func (i *Imp) Shutdown() {
if i.app == nil {
return
}
timeout := 15 * time.Second
if i.config != nil && i.config.Runtime != nil {
timeout = i.config.Runtime.ShutdownTimeout()
}
if i.service != nil {
i.service.Shutdown()
}
if i.discoveryWatcher != nil {
i.discoveryWatcher.Stop()
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
i.app.Shutdown(ctx)
}
func (i *Imp) Start() error {
cfg, err := i.loadConfig()
if err != nil {
i.logger.Error("Service startup aborted: invalid configuration", zap.Error(err))
return err
}
i.config = cfg
var broker mb.Broker
if cfg.Messaging != nil && cfg.Messaging.Driver != "" {
broker, err = msg.CreateMessagingBroker(i.logger, cfg.Messaging)
if err != nil {
i.logger.Warn("Failed to create messaging broker", zap.Error(err))
}
}
if broker != nil {
registry := discovery.NewRegistry()
watcher, watcherErr := discovery.NewRegistryWatcher(i.logger, broker, registry)
if watcherErr != nil {
i.logger.Warn("Failed to initialise discovery registry watcher", zap.Error(watcherErr))
} else if startErr := watcher.Start(); startErr != nil {
i.logger.Warn("Failed to start discovery registry watcher", zap.Error(startErr))
} else {
i.discoveryWatcher = watcher
i.discoveryReg = registry
i.logger.Info("Discovery registry watcher started")
}
}
repoFactory := func(logger mlogger.Logger, conn *db.MongoConnection) (storage.Repository, error) {
return gatewaymongo.New(logger, conn)
}
serviceFactory := func(logger mlogger.Logger, repo storage.Repository, producer msg.Producer) (grpcapp.Service, error) {
invokeURI := ""
if cfg.GRPC != nil {
invokeURI = cfg.GRPC.DiscoveryInvokeURI()
}
msgSettings := map[string]any(nil)
if cfg.Messaging != nil {
msgSettings = cfg.Messaging.Settings
}
gwCfg := gateway.Config{
Rail: cfg.Gateway.Rail,
TargetChatIDEnv: cfg.Gateway.TargetChatIDEnv,
TimeoutSeconds: cfg.Gateway.TimeoutSeconds,
AcceptedUserIDs: cfg.Gateway.AcceptedUserIDs,
SuccessReaction: cfg.Gateway.SuccessReaction,
InvokeURI: invokeURI,
MessagingSettings: msgSettings,
DiscoveryRegistry: i.discoveryReg,
Treasury: gateway.TreasuryConfig{
ExecutionDelay: cfg.Treasury.ExecutionDelay,
PollInterval: cfg.Treasury.PollInterval,
Ledger: gateway.LedgerConfig{
Timeout: cfg.Treasury.Ledger.Timeout,
},
Limits: gateway.TreasuryLimitsConfig{
MaxAmountPerOperation: cfg.Treasury.Limits.MaxAmountPerOperation,
MaxDailyAmount: cfg.Treasury.Limits.MaxDailyAmount,
},
},
}
svc := gateway.NewService(logger, repo, producer, broker, gwCfg)
i.service = svc
return svc, nil
}
app, err := grpcapp.NewApp(i.logger, "chsettle_gateway", cfg.Config, i.debug, repoFactory, serviceFactory)
if err != nil {
i.logger.Error("Service startup aborted: failed to construct app", zap.Error(err))
return err
}
i.app = app
if err := i.app.Start(); err != nil {
i.logger.Error("Service startup aborted: app start failed", zap.Error(err))
return err
}
grpcAddress := ""
metricsAddress := ""
if cfg.GRPC != nil {
grpcAddress = strings.TrimSpace(cfg.GRPC.Address)
}
if cfg.Metrics != nil {
metricsAddress = strings.TrimSpace(cfg.Metrics.Address)
}
i.logger.Info("ChimeraSettle gateway started",
zap.String("grpc_address", grpcAddress),
zap.String("metrics_address", metricsAddress))
return nil
}
func (i *Imp) loadConfig() (*config, error) {
data, err := os.ReadFile(i.file)
if err != nil {
i.logger.Error("Could not read configuration file", zap.String("config_file", i.file), zap.Error(err))
return nil, err
}
cfg := &config{Config: &grpcapp.Config{}}
if err := yaml.Unmarshal(data, cfg); err != nil {
i.logger.Error("Failed to parse configuration", zap.Error(err))
return nil, err
}
if cfg.Runtime == nil {
cfg.Runtime = &grpcapp.RuntimeConfig{ShutdownTimeoutSeconds: 15}
}
if cfg.GRPC == nil {
cfg.GRPC = &routers.GRPCConfig{
Network: "tcp",
Address: ":50080",
EnableReflection: true,
EnableHealth: true,
}
}
if cfg.Metrics == nil {
cfg.Metrics = &grpcapp.MetricsConfig{Address: ":9406"}
}
if cfg.Treasury.ExecutionDelay <= 0 {
cfg.Treasury.ExecutionDelay = 30 * time.Second
}
if cfg.Treasury.PollInterval <= 0 {
cfg.Treasury.PollInterval = 30 * time.Second
}
if cfg.Treasury.Ledger.Timeout <= 0 {
cfg.Treasury.Ledger.Timeout = 5 * time.Second
}
cfg.Gateway.Rail = discovery.NormalizeRail(cfg.Gateway.Rail)
if cfg.Gateway.Rail == "" {
i.logger.Error("Invalid configuration: gateway rail is required")
return nil, merrors.InvalidArgument("gateway rail is required", "gateway.rail")
}
if !discovery.IsKnownRail(cfg.Gateway.Rail) {
i.logger.Error("Invalid configuration: gateway rail is unknown", zap.String("gateway_rail", cfg.Gateway.Rail))
return nil, merrors.InvalidArgument("gateway rail must be a known token", "gateway.rail")
}
return cfg, nil
}

View File

@@ -0,0 +1,11 @@
package server
import (
serverimp "github.com/tech/sendico/gateway/chsettle/internal/server/internal"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/server"
)
func Create(logger mlogger.Logger, file string, debug bool) (server.Application, error) {
return serverimp.Create(logger, file, debug)
}

View File

@@ -0,0 +1,443 @@
package gateway
import (
"context"
"errors"
"regexp"
"strings"
"time"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
confirmations "github.com/tech/sendico/pkg/messaging/notifications/confirmations"
tnotifications "github.com/tech/sendico/pkg/messaging/notifications/telegram"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
"go.uber.org/zap"
)
var amountPattern = regexp.MustCompile(`^[0-9]+(\.[0-9]+)?$`)
var currencyPattern = regexp.MustCompile(`^[A-Za-z]{3,10}$`)
func (s *Service) startConfirmationTimeoutWatcher() {
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return
}
if s.timeoutCancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
s.timeoutCtx = ctx
s.timeoutCancel = cancel
s.timeoutWG.Add(1)
go func() {
defer s.timeoutWG.Done()
ticker := time.NewTicker(defaultConfirmationSweepInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.sweepExpiredConfirmations(ctx)
}
}
}()
}
func (s *Service) sweepExpiredConfirmations(ctx context.Context) {
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return
}
expired, err := s.repo.PendingConfirmations().ListExpired(ctx, time.Now(), 100)
if err != nil {
s.logger.Warn("Failed to list expired pending confirmations", zap.Error(err))
return
}
for i := range expired {
pending := &expired[i]
if strings.TrimSpace(pending.RequestID) == "" {
continue
}
result := &model.ConfirmationResult{
RequestID: pending.RequestID,
Status: model.ConfirmationStatusTimeout,
}
if err := s.publishPendingConfirmationResult(pending, result); err != nil {
s.logger.Warn("Failed to publish timeout confirmation result", zap.Error(err), zap.String("request_id", pending.RequestID))
continue
}
if err := s.clearPendingConfirmation(ctx, pending.RequestID); err != nil {
s.logger.Warn("Failed to remove expired pending confirmation", zap.Error(err), zap.String("request_id", pending.RequestID))
}
}
}
func (s *Service) persistPendingConfirmation(ctx context.Context, request *model.ConfirmationRequest) error {
if request == nil {
return merrors.InvalidArgument("confirmation request is nil", "request")
}
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return merrors.Internal("pending confirmations store unavailable")
}
timeout := request.TimeoutSeconds
if timeout <= 0 {
timeout = int32(defaultConfirmationTimeoutSeconds)
}
pending := &storagemodel.PendingConfirmation{
RequestID: strings.TrimSpace(request.RequestID),
TargetChatID: strings.TrimSpace(request.TargetChatID),
AcceptedUserIDs: normalizeStringList(request.AcceptedUserIDs),
RequestedMoney: request.RequestedMoney,
SourceService: strings.TrimSpace(request.SourceService),
Rail: strings.TrimSpace(request.Rail),
ExpiresAt: time.Now().Add(time.Duration(timeout) * time.Second),
}
return s.repo.PendingConfirmations().Upsert(ctx, pending)
}
func (s *Service) clearPendingConfirmation(ctx context.Context, requestID string) error {
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return nil
}
requestID = strings.TrimSpace(requestID)
if requestID == "" {
return nil
}
return s.repo.PendingConfirmations().DeleteByRequestID(ctx, requestID)
}
func (s *Service) onConfirmationDispatch(ctx context.Context, dispatch *model.ConfirmationRequestDispatch) error {
if dispatch == nil {
return merrors.InvalidArgument("confirmation dispatch is nil", "dispatch")
}
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return merrors.Internal("pending confirmations store unavailable")
}
requestID := strings.TrimSpace(dispatch.RequestID)
messageID := strings.TrimSpace(dispatch.MessageID)
if requestID == "" {
return merrors.InvalidArgument("confirmation request_id is required", "request_id")
}
if messageID == "" {
return merrors.InvalidArgument("confirmation message_id is required", "message_id")
}
if err := s.repo.PendingConfirmations().AttachMessage(ctx, requestID, messageID); err != nil {
if errors.Is(err, merrors.ErrNoData) {
s.logger.Info("Confirmation dispatch ignored: pending request not found",
zap.String("request_id", requestID),
zap.String("message_id", messageID))
return nil
}
s.logger.Warn("Failed to attach confirmation message id", zap.Error(err), zap.String("request_id", requestID), zap.String("message_id", messageID))
return err
}
s.logger.Info("Pending confirmation message attached", zap.String("request_id", requestID), zap.String("message_id", messageID))
return nil
}
func (s *Service) onTelegramUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) error {
if update == nil || update.Message == nil {
return nil
}
if s == nil || s.repo == nil || s.repo.PendingConfirmations() == nil {
return merrors.Internal("pending confirmations store unavailable")
}
message := update.Message
replyToID := strings.TrimSpace(message.ReplyToMessageID)
if replyToID == "" {
s.handleTreasuryTelegramUpdate(ctx, update)
return nil
}
replyFields := telegramReplyLogFields(update)
pending, err := s.repo.PendingConfirmations().FindByMessageID(ctx, replyToID)
if err != nil {
return err
}
if pending == nil {
if s.handleTreasuryTelegramUpdate(ctx, update) {
return nil
}
s.logger.Warn("Telegram confirmation reply dropped",
append(replyFields,
zap.String("outcome", "dropped"),
zap.String("reason", "no_pending_confirmation"),
)...)
return nil
}
replyFields = append(replyFields,
zap.String("request_id", strings.TrimSpace(pending.RequestID)),
zap.String("target_chat_id", strings.TrimSpace(pending.TargetChatID)),
)
if !pending.ExpiresAt.IsZero() && time.Now().After(pending.ExpiresAt) {
result := &model.ConfirmationResult{
RequestID: pending.RequestID,
Status: model.ConfirmationStatusTimeout,
}
if err := s.publishPendingConfirmationResult(pending, result); err != nil {
return err
}
if err := s.clearPendingConfirmation(ctx, pending.RequestID); err != nil {
return err
}
s.logger.Info("Telegram confirmation reply processed",
append(replyFields,
zap.String("outcome", "processed"),
zap.String("result_status", string(result.Status)),
zap.String("reason", "expired_confirmation"),
)...)
return nil
}
if strings.TrimSpace(message.ChatID) != strings.TrimSpace(pending.TargetChatID) {
s.logger.Warn("Telegram confirmation reply dropped",
append(replyFields,
zap.String("outcome", "dropped"),
zap.String("reason", "chat_mismatch"),
zap.String("expected_chat_id", strings.TrimSpace(pending.TargetChatID)),
)...)
return nil
}
if !isUserAllowed(message.FromUserID, pending.AcceptedUserIDs) {
result := &model.ConfirmationResult{
RequestID: pending.RequestID,
Status: model.ConfirmationStatusRejected,
ParseError: "unauthorized_user",
RawReply: message,
}
if err := s.publishPendingConfirmationResult(pending, result); err != nil {
return err
}
if e := s.sendTelegramText(ctx, &model.TelegramTextRequest{
RequestID: pending.RequestID,
ChatID: pending.TargetChatID,
ReplyToMessageID: message.MessageID,
Text: "Only approved users can confirm this payment.",
}); e != nil {
s.logger.Warn("Failed to create telegram text", append(replyFields, zap.Error(err))...)
}
if err := s.clearPendingConfirmation(ctx, pending.RequestID); err != nil {
return err
}
s.logger.Info("Telegram confirmation reply processed",
append(replyFields,
zap.String("outcome", "processed"),
zap.String("result_status", string(result.Status)),
zap.String("reason", "unauthorized_user"),
)...)
return nil
}
money, reason, err := parseConfirmationReply(message.Text)
if err != nil {
if markErr := s.repo.PendingConfirmations().MarkClarified(ctx, pending.RequestID); markErr != nil {
s.logger.Warn("Failed to mark confirmation as clarified", zap.Error(markErr), zap.String("request_id", pending.RequestID))
}
if e := s.sendTelegramText(ctx, &model.TelegramTextRequest{
RequestID: pending.RequestID,
ChatID: pending.TargetChatID,
ReplyToMessageID: message.MessageID,
Text: clarificationMessage(reason),
}); e != nil {
s.logger.Warn("Failed to create telegram text", append(replyFields, zap.Error(err))...)
}
s.logger.Warn("Telegram confirmation reply dropped",
append(replyFields,
zap.String("outcome", "dropped"),
zap.String("reason", "invalid_reply_format"),
zap.String("parse_reason", reason),
)...)
return nil
}
status := model.ConfirmationStatusConfirmed
if pending.Clarified {
status = model.ConfirmationStatusClarified
}
result := &model.ConfirmationResult{
RequestID: pending.RequestID,
Money: money,
RawReply: message,
Status: status,
}
if err := s.publishPendingConfirmationResult(pending, result); err != nil {
return err
}
if err := s.clearPendingConfirmation(ctx, pending.RequestID); err != nil {
return err
}
s.logger.Info("Telegram confirmation reply processed",
append(replyFields,
zap.String("outcome", "processed"),
zap.String("result_status", string(result.Status)),
)...)
return nil
}
func (s *Service) handleTreasuryTelegramUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) bool {
if s == nil || s.treasury == nil || update == nil || update.Message == nil {
return false
}
return s.treasury.HandleUpdate(ctx, update)
}
func telegramReplyLogFields(update *model.TelegramWebhookUpdate) []zap.Field {
if update == nil || update.Message == nil {
return nil
}
message := update.Message
return []zap.Field{
zap.Int64("update_id", update.UpdateID),
zap.String("message_id", strings.TrimSpace(message.MessageID)),
zap.String("reply_to_message_id", strings.TrimSpace(message.ReplyToMessageID)),
zap.String("chat_id", strings.TrimSpace(message.ChatID)),
zap.String("from_user_id", strings.TrimSpace(message.FromUserID)),
}
}
func (s *Service) publishPendingConfirmationResult(pending *storagemodel.PendingConfirmation, result *model.ConfirmationResult) error {
if pending == nil || result == nil {
return merrors.InvalidArgument("pending confirmation context is required")
}
if s == nil || s.producer == nil {
return merrors.Internal("messaging producer is not configured")
}
sourceService := strings.TrimSpace(pending.SourceService)
if sourceService == "" {
sourceService = string(mservice.PaymentGateway)
}
rail := strings.TrimSpace(pending.Rail)
if rail == "" {
rail = s.rail
}
env := confirmations.ConfirmationResult(string(mservice.PaymentGateway), result, sourceService, rail)
if err := s.producer.SendMessage(env); err != nil {
s.logger.Warn("Failed to publish confirmation result", zap.Error(err),
zap.String("request_id", strings.TrimSpace(result.RequestID)),
zap.String("status", string(result.Status)),
zap.String("source_service", sourceService),
zap.String("rail", rail))
return err
}
return nil
}
func (s *Service) sendTelegramText(_ context.Context, request *model.TelegramTextRequest) error {
if request == nil {
return merrors.InvalidArgument("telegram text request is nil", "request")
}
if s == nil || s.producer == nil {
return merrors.Internal("messaging producer is not configured")
}
request.ChatID = strings.TrimSpace(request.ChatID)
request.Text = strings.TrimSpace(request.Text)
request.ReplyToMessageID = strings.TrimSpace(request.ReplyToMessageID)
if request.ChatID == "" || request.Text == "" {
return merrors.InvalidArgument("telegram chat_id and text are required", "chat_id", "text")
}
env := tnotifications.TelegramText(string(mservice.PaymentGateway), request)
if err := s.producer.SendMessage(env); err != nil {
s.logger.Warn("Failed to publish telegram text request", zap.Error(err),
zap.String("request_id", request.RequestID),
zap.String("chat_id", request.ChatID),
zap.String("reply_to_message_id", request.ReplyToMessageID))
return err
}
return nil
}
func isFinalConfirmationStatus(status model.ConfirmationStatus) bool {
switch status {
case model.ConfirmationStatusConfirmed, model.ConfirmationStatusRejected, model.ConfirmationStatusTimeout, model.ConfirmationStatusClarified:
return true
default:
return false
}
}
func isUserAllowed(userID string, allowed []string) bool {
allowed = normalizeStringList(allowed)
if len(allowed) == 0 {
return true
}
userID = strings.TrimSpace(userID)
if userID == "" {
return false
}
for _, id := range allowed {
if id == userID {
return true
}
}
return false
}
func parseConfirmationReply(text string) (*paymenttypes.Money, string, error) {
text = strings.TrimSpace(text)
if text == "" {
return nil, "empty", merrors.InvalidArgument("empty reply")
}
parts := strings.Fields(text)
if len(parts) < 2 {
if len(parts) == 1 && amountPattern.MatchString(parts[0]) {
return nil, "missing_currency", merrors.InvalidArgument("currency is required")
}
return nil, "missing_amount", merrors.InvalidArgument("amount is required")
}
if len(parts) > 2 {
return nil, "format", merrors.InvalidArgument("reply format is invalid")
}
amount := parts[0]
currency := parts[1]
if !amountPattern.MatchString(amount) {
return nil, "invalid_amount", merrors.InvalidArgument("amount format is invalid")
}
if !currencyPattern.MatchString(currency) {
return nil, "invalid_currency", merrors.InvalidArgument("currency format is invalid")
}
return &paymenttypes.Money{
Amount: amount,
Currency: strings.ToUpper(currency),
}, "", nil
}
func clarificationMessage(reason string) string {
switch reason {
case "missing_currency":
return "Currency code is required. Reply with \"<amount> <currency>\" (e.g., 12.34 USD)."
case "missing_amount":
return "Amount is required. Reply with \"<amount> <currency>\" (e.g., 12.34 USD)."
case "invalid_amount":
return "Amount must be a decimal number. Reply with \"<amount> <currency>\" (e.g., 12.34 USD)."
case "invalid_currency":
return "Currency must be a code like USD or EUR. Reply with \"<amount> <currency>\"."
default:
return "Reply with \"<amount> <currency>\" (e.g., 12.34 USD)."
}
}
func normalizeStringList(values []string) []string {
if len(values) == 0 {
return nil
}
result := make([]string, 0, len(values))
seen := map[string]struct{}{}
for _, value := range values {
value = strings.TrimSpace(value)
if value == "" {
continue
}
if _, ok := seen[value]; ok {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
if len(result) == 0 {
return nil
}
return result
}

View File

@@ -0,0 +1,413 @@
package gateway
import (
"context"
"errors"
"strings"
"github.com/tech/sendico/pkg/connector/params"
"github.com/tech/sendico/pkg/merrors"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
"go.uber.org/zap"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/structpb"
)
const (
chsettleConnectorID = "chsettle"
connectorScenarioParam = "scenario"
connectorScenarioMetaKey = "chsettle_scenario"
)
func (s *Service) GetCapabilities(_ context.Context, _ *connectorv1.GetCapabilitiesRequest) (*connectorv1.GetCapabilitiesResponse, error) {
return &connectorv1.GetCapabilitiesResponse{
Capabilities: &connectorv1.ConnectorCapabilities{
ConnectorType: chsettleConnectorID,
Version: "",
SupportedAccountKinds: nil,
SupportedOperationTypes: []connectorv1.OperationType{connectorv1.OperationType_TRANSFER},
OperationParams: chsettleOperationParams(),
},
}, nil
}
func (s *Service) OpenAccount(_ context.Context, _ *connectorv1.OpenAccountRequest) (*connectorv1.OpenAccountResponse, error) {
return &connectorv1.OpenAccountResponse{Error: connectorError(connectorv1.ErrorCode_UNSUPPORTED_ACCOUNT_KIND, "open_account: unsupported", nil, "")}, nil
}
func (s *Service) GetAccount(_ context.Context, _ *connectorv1.GetAccountRequest) (*connectorv1.GetAccountResponse, error) {
return nil, merrors.NotImplemented("get_account: unsupported")
}
func (s *Service) ListAccounts(_ context.Context, _ *connectorv1.ListAccountsRequest) (*connectorv1.ListAccountsResponse, error) {
return nil, merrors.NotImplemented("list_accounts: unsupported")
}
func (s *Service) GetBalance(_ context.Context, _ *connectorv1.GetBalanceRequest) (*connectorv1.GetBalanceResponse, error) {
return nil, merrors.NotImplemented("get_balance: unsupported")
}
func (s *Service) SubmitOperation(ctx context.Context, req *connectorv1.SubmitOperationRequest) (*connectorv1.SubmitOperationResponse, error) {
if req == nil || req.GetOperation() == nil {
s.logger.Warn("Submit operation rejected", zap.String("reason", "operation is required"))
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, "submit_operation: operation is required", nil, "")}}, nil
}
op := req.GetOperation()
s.logger.Debug("Submit operation request received",
zap.String("idempotency_key", strings.TrimSpace(op.GetIdempotencyKey())),
zap.String("intent_ref", strings.TrimSpace(op.GetIntentRef())),
zap.String("operation_ref", strings.TrimSpace(op.GetOperationRef())))
if strings.TrimSpace(op.GetIdempotencyKey()) == "" {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.String("reason", "idempotency_key is required"))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, "submit_operation: idempotency_key is required", op, "")}}, nil
}
if op.GetType() != connectorv1.OperationType_TRANSFER {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.String("reason", "unsupported operation type"))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_UNSUPPORTED_OPERATION, "submit_operation: unsupported operation type", op, "")}}, nil
}
reader := params.New(op.GetParams())
metadata := reader.StringMap("metadata")
if metadata == nil {
metadata = map[string]string{}
}
paymentIntentID := strings.TrimSpace(reader.String("payment_intent_id"))
if paymentIntentID == "" {
paymentIntentID = strings.TrimSpace(reader.String("payment_ref"))
}
if paymentIntentID == "" {
paymentIntentID = strings.TrimSpace(metadata[metadataPaymentIntentID])
}
if paymentIntentID == "" {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.String("reason", "payment_intent_id is required"))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, "submit_operation: payment_intent_id is required", op, "")}}, nil
}
source := operationAccountID(op.GetFrom())
if source == "" {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.String("reason", "from.account is required"))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, "transfer: from.account is required", op, "")}}, nil
}
dest, err := transferDestinationFromOperation(op)
if err != nil {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.Error(err))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, err.Error(), op, "")}}, nil
}
amount := op.GetMoney()
if amount == nil {
s.logger.Warn("Submit operation rejected", append(operationLogFields(op), zap.String("reason", "money is required"))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(connectorv1.ErrorCode_INVALID_PARAMS, "transfer: money is required", op, "")}}, nil
}
metadata[metadataPaymentIntentID] = paymentIntentID
quoteRef := strings.TrimSpace(reader.String("quote_ref"))
if quoteRef != "" {
metadata[metadataQuoteRef] = quoteRef
}
targetChatID := strings.TrimSpace(reader.String("target_chat_id"))
if targetChatID != "" {
metadata[metadataTargetChatID] = targetChatID
}
outgoingLeg := normalizeRail(reader.String("outgoing_leg"))
if outgoingLeg != "" {
metadata[metadataOutgoingLeg] = outgoingLeg
}
if scenario := strings.TrimSpace(reader.String(connectorScenarioParam)); scenario != "" {
metadata[connectorScenarioMetaKey] = scenario
}
s.logger.Debug("Submit operation parsed transfer metadata",
zap.String("idempotency_key", strings.TrimSpace(op.GetIdempotencyKey())),
zap.String("payment_intent_id", paymentIntentID),
zap.String("quote_ref", quoteRef),
zap.String("target_chat_id", targetChatID),
zap.String("outgoing_leg", outgoingLeg),
zap.String("scenario_override", strings.TrimSpace(metadata[connectorScenarioMetaKey])))
normalizedAmount := normalizeMoneyForTransfer(amount)
logFields := append(operationLogFields(op),
zap.String("payment_intent_id", paymentIntentID),
zap.String("organization_ref", strings.TrimSpace(reader.String("organization_ref"))),
zap.String("source_wallet_ref", source),
zap.String("amount", strings.TrimSpace(normalizedAmount.GetAmount())),
zap.String("currency", strings.TrimSpace(normalizedAmount.GetCurrency())),
zap.String("quote_ref", quoteRef),
zap.String("operation_ref", req.Operation.GetOperationRef()),
zap.String("intent_ref", op.GetIntentRef()),
zap.String("outgoing_leg", outgoingLeg),
)
logFields = append(logFields, transferDestinationLogFields(dest)...)
s.logger.Debug("Submit operation forwarding to transfer handler", logFields...)
resp, err := s.SubmitTransfer(ctx, &chainv1.SubmitTransferRequest{
IdempotencyKey: strings.TrimSpace(op.GetIdempotencyKey()),
OrganizationRef: strings.TrimSpace(reader.String("organization_ref")),
SourceWalletRef: source,
Destination: dest,
Amount: normalizedAmount,
Metadata: metadata,
PaymentRef: paymentIntentID,
IntentRef: strings.TrimSpace(op.GetIntentRef()),
OperationRef: strings.TrimSpace(op.GetOperationRef()),
})
if err != nil {
s.logger.Warn("Submit operation transfer failed", append(logFields, zap.Error(err))...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{Error: connectorError(mapErrorCode(err), err.Error(), op, "")}}, nil
}
transfer := resp.GetTransfer()
operationID := strings.TrimSpace(transfer.GetOperationRef())
if operationID == "" {
s.logger.Warn("Submit operation transfer response missing operation_ref", append(logFields,
zap.String("transfer_ref", strings.TrimSpace(transfer.GetTransferRef())),
)...)
return &connectorv1.SubmitOperationResponse{Receipt: &connectorv1.OperationReceipt{
Error: connectorError(connectorv1.ErrorCode_TEMPORARY_UNAVAILABLE, "submit_operation: operation_ref is missing in transfer response", op, ""),
}}, nil
}
s.logger.Info("Submit operation transfer submitted", append(logFields,
zap.String("transfer_ref", strings.TrimSpace(transfer.GetTransferRef())),
zap.String("status", transfer.GetStatus().String()),
)...)
return &connectorv1.SubmitOperationResponse{
Receipt: &connectorv1.OperationReceipt{
OperationId: operationID,
Status: transferStatusToOperation(transfer.GetStatus()),
ProviderRef: strings.TrimSpace(transfer.GetTransferRef()),
},
}, nil
}
func (s *Service) GetOperation(ctx context.Context, req *connectorv1.GetOperationRequest) (*connectorv1.GetOperationResponse, error) {
if req == nil || strings.TrimSpace(req.GetOperationId()) == "" {
s.logger.Warn("Get operation rejected", zap.String("reason", "operation_id is required"))
return nil, merrors.InvalidArgument("get_operation: operation_id is required")
}
operationID := strings.TrimSpace(req.GetOperationId())
s.logger.Debug("Get operation request received", zap.String("operation_id", operationID))
if s.repo == nil || s.repo.Payments() == nil {
s.logger.Warn("Get operation storage unavailable", zap.String("operation_id", operationID))
return nil, merrors.Internal("get_operation: storage is not configured")
}
record, err := s.repo.Payments().FindByOperationRef(ctx, operationID)
if err != nil {
s.logger.Warn("Get operation lookup by operation_ref failed", zap.String("operation_id", operationID), zap.Error(err))
return nil, err
}
if record == nil {
s.logger.Info("Get operation not found", zap.String("operation_id", operationID))
return nil, status.Error(codes.NotFound, "operation not found")
}
return &connectorv1.GetOperationResponse{Operation: transferToOperation(transferFromPayment(record, nil))}, nil
}
func (s *Service) ListOperations(_ context.Context, _ *connectorv1.ListOperationsRequest) (*connectorv1.ListOperationsResponse, error) {
return nil, merrors.NotImplemented("list_operations: unsupported")
}
func chsettleOperationParams() []*connectorv1.OperationParamSpec {
return []*connectorv1.OperationParamSpec{
{OperationType: connectorv1.OperationType_TRANSFER, Params: []*connectorv1.ParamSpec{
{Key: "payment_intent_id", Type: connectorv1.ParamType_STRING, Required: true},
{Key: "organization_ref", Type: connectorv1.ParamType_STRING, Required: false},
{Key: "quote_ref", Type: connectorv1.ParamType_STRING, Required: false},
{Key: "target_chat_id", Type: connectorv1.ParamType_STRING, Required: false},
{Key: "outgoing_leg", Type: connectorv1.ParamType_STRING, Required: false},
{Key: connectorScenarioParam, Type: connectorv1.ParamType_STRING, Required: false},
{Key: "metadata", Type: connectorv1.ParamType_JSON, Required: false},
}},
}
}
func transferDestinationFromOperation(op *connectorv1.Operation) (*chainv1.TransferDestination, error) {
if op == nil {
return nil, merrors.InvalidArgument("transfer: operation is required")
}
if to := op.GetTo(); to != nil {
if account := to.GetAccount(); account != nil {
return &chainv1.TransferDestination{Destination: &chainv1.TransferDestination_ManagedWalletRef{ManagedWalletRef: strings.TrimSpace(account.GetAccountId())}}, nil
}
if ext := to.GetExternal(); ext != nil {
return &chainv1.TransferDestination{Destination: &chainv1.TransferDestination_ExternalAddress{ExternalAddress: strings.TrimSpace(ext.GetExternalRef())}}, nil
}
}
return nil, merrors.InvalidArgument("transfer: to.account or to.external is required")
}
func normalizeMoneyForTransfer(m *moneyv1.Money) *moneyv1.Money {
if m == nil {
return nil
}
currency := strings.TrimSpace(m.GetCurrency())
if idx := strings.Index(currency, "-"); idx > 0 {
currency = currency[:idx]
}
return &moneyv1.Money{
Amount: strings.TrimSpace(m.GetAmount()),
Currency: currency,
}
}
func transferToOperation(transfer *chainv1.Transfer) *connectorv1.Operation {
if transfer == nil {
return nil
}
op := &connectorv1.Operation{
OperationId: strings.TrimSpace(transfer.GetOperationRef()),
Type: connectorv1.OperationType_TRANSFER,
Status: transferStatusToOperation(transfer.GetStatus()),
Money: transfer.GetRequestedAmount(),
ProviderRef: strings.TrimSpace(transfer.GetTransferRef()),
IntentRef: strings.TrimSpace(transfer.GetIntentRef()),
OperationRef: strings.TrimSpace(transfer.GetOperationRef()),
CreatedAt: transfer.GetCreatedAt(),
UpdatedAt: transfer.GetUpdatedAt(),
}
params := map[string]interface{}{}
if paymentRef := strings.TrimSpace(transfer.GetPaymentRef()); paymentRef != "" {
params["payment_ref"] = paymentRef
}
if organizationRef := strings.TrimSpace(transfer.GetOrganizationRef()); organizationRef != "" {
params["organization_ref"] = organizationRef
}
if failureReason := strings.TrimSpace(transfer.GetFailureReason()); failureReason != "" {
params["failure_reason"] = failureReason
}
if len(params) > 0 {
op.Params = structFromMap(params)
}
if source := strings.TrimSpace(transfer.GetSourceWalletRef()); source != "" {
op.From = &connectorv1.OperationParty{Ref: &connectorv1.OperationParty_Account{Account: &connectorv1.AccountRef{
ConnectorId: chsettleConnectorID,
AccountId: source,
}}}
}
if dest := transfer.GetDestination(); dest != nil {
switch d := dest.GetDestination().(type) {
case *chainv1.TransferDestination_ManagedWalletRef:
op.To = &connectorv1.OperationParty{Ref: &connectorv1.OperationParty_Account{Account: &connectorv1.AccountRef{
ConnectorId: chsettleConnectorID,
AccountId: strings.TrimSpace(d.ManagedWalletRef),
}}}
case *chainv1.TransferDestination_ExternalAddress:
op.To = &connectorv1.OperationParty{Ref: &connectorv1.OperationParty_External{External: &connectorv1.ExternalRef{
ExternalRef: strings.TrimSpace(d.ExternalAddress),
}}}
}
}
return op
}
func transferStatusToOperation(status chainv1.TransferStatus) connectorv1.OperationStatus {
switch status {
case chainv1.TransferStatus_TRANSFER_CREATED:
return connectorv1.OperationStatus_OPERATION_CREATED
case chainv1.TransferStatus_TRANSFER_PROCESSING:
return connectorv1.OperationStatus_OPERATION_PROCESSING
case chainv1.TransferStatus_TRANSFER_WAITING:
return connectorv1.OperationStatus_OPERATION_WAITING
case chainv1.TransferStatus_TRANSFER_SUCCESS:
return connectorv1.OperationStatus_OPERATION_SUCCESS
case chainv1.TransferStatus_TRANSFER_FAILED:
return connectorv1.OperationStatus_OPERATION_FAILED
case chainv1.TransferStatus_TRANSFER_CANCELLED:
return connectorv1.OperationStatus_OPERATION_CANCELLED
case chainv1.TransferStatus_TRANSFER_STATUS_UNSPECIFIED:
fallthrough
default:
return connectorv1.OperationStatus_OPERATION_STATUS_UNSPECIFIED
}
}
func operationAccountID(party *connectorv1.OperationParty) string {
if party == nil {
return ""
}
if account := party.GetAccount(); account != nil {
return strings.TrimSpace(account.GetAccountId())
}
return ""
}
func structFromMap(values map[string]interface{}) *structpb.Struct {
if len(values) == 0 {
return nil
}
result, err := structpb.NewStruct(values)
if err != nil {
return nil
}
return result
}
func operationLogFields(op *connectorv1.Operation) []zap.Field {
if op == nil {
return nil
}
return []zap.Field{
zap.String("operation_id", strings.TrimSpace(op.GetOperationId())),
zap.String("idempotency_key", strings.TrimSpace(op.GetIdempotencyKey())),
zap.String("correlation_id", strings.TrimSpace(op.GetCorrelationId())),
zap.String("parent_intent_id", strings.TrimSpace(op.GetParentIntentId())),
zap.String("operation_type", op.GetType().String()),
zap.String("intent_ref", strings.TrimSpace(op.GetIntentRef())),
}
}
func transferDestinationLogFields(dest *chainv1.TransferDestination) []zap.Field {
if dest == nil {
return nil
}
switch d := dest.GetDestination().(type) {
case *chainv1.TransferDestination_ManagedWalletRef:
return []zap.Field{
zap.String("destination_type", "managed_wallet"),
zap.String("destination_ref", strings.TrimSpace(d.ManagedWalletRef)),
}
case *chainv1.TransferDestination_ExternalAddress:
return []zap.Field{
zap.String("destination_type", "external_address"),
zap.String("destination_ref", strings.TrimSpace(d.ExternalAddress)),
}
default:
return []zap.Field{zap.String("destination_type", "unknown")}
}
}
func connectorError(code connectorv1.ErrorCode, message string, op *connectorv1.Operation, accountID string) *connectorv1.ConnectorError {
err := &connectorv1.ConnectorError{
Code: code,
Message: strings.TrimSpace(message),
AccountId: strings.TrimSpace(accountID),
}
if op != nil {
err.CorrelationId = strings.TrimSpace(op.GetCorrelationId())
err.ParentIntentId = strings.TrimSpace(op.GetParentIntentId())
err.OperationId = strings.TrimSpace(op.GetOperationId())
}
return err
}
func mapErrorCode(err error) connectorv1.ErrorCode {
switch {
case errors.Is(err, merrors.ErrInvalidArg):
return connectorv1.ErrorCode_INVALID_PARAMS
case errors.Is(err, merrors.ErrNoData):
return connectorv1.ErrorCode_NOT_FOUND
case errors.Is(err, merrors.ErrNotImplemented):
return connectorv1.ErrorCode_UNSUPPORTED_OPERATION
case errors.Is(err, merrors.ErrInternal):
return connectorv1.ErrorCode_TEMPORARY_UNAVAILABLE
default:
return connectorv1.ErrorCode_PROVIDER_ERROR
}
}

View File

@@ -0,0 +1,119 @@
package gateway
import (
"context"
"testing"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func TestSubmitOperation_UsesOperationRefAsOperationID(t *testing.T) {
svc, _, _ := newTestService(t)
svc.chatID = "1"
req := &connectorv1.SubmitOperationRequest{
Operation: &connectorv1.Operation{
Type: connectorv1.OperationType_TRANSFER,
IdempotencyKey: "idem-settlement-1",
OperationRef: "payment-1:hop_2_settlement_fx_convert",
IntentRef: "intent-1",
Money: &moneyv1.Money{Amount: "1.00", Currency: "USDT"},
From: &connectorv1.OperationParty{
Ref: &connectorv1.OperationParty_Account{Account: &connectorv1.AccountRef{
ConnectorId: chsettleConnectorID,
AccountId: "wallet-src",
}},
},
To: &connectorv1.OperationParty{
Ref: &connectorv1.OperationParty_Account{Account: &connectorv1.AccountRef{
ConnectorId: chsettleConnectorID,
AccountId: "wallet-dst",
}},
},
Params: structFromMap(map[string]interface{}{
"payment_ref": "payment-1",
"organization_ref": "org-1",
}),
},
}
resp, err := svc.SubmitOperation(context.Background(), req)
if err != nil {
t.Fatalf("SubmitOperation returned error: %v", err)
}
if resp.GetReceipt() == nil {
t.Fatal("expected receipt")
}
if got := resp.GetReceipt().GetError(); got != nil {
t.Fatalf("expected no connector error, got: %v", got)
}
if got, want := resp.GetReceipt().GetOperationId(), "payment-1:hop_2_settlement_fx_convert"; got != want {
t.Fatalf("operation_id mismatch: got=%q want=%q", got, want)
}
if got, want := resp.GetReceipt().GetProviderRef(), "idem-settlement-1"; got != want {
t.Fatalf("provider_ref mismatch: got=%q want=%q", got, want)
}
}
func TestGetOperation_UsesOperationRefIdentity(t *testing.T) {
svc, repo, _ := newTestService(t)
record := &storagemodel.PaymentRecord{
IdempotencyKey: "idem-settlement-2",
OperationRef: "payment-2:hop_2_settlement_fx_convert",
PaymentIntentID: "pi-2",
PaymentRef: "payment-2",
RequestedMoney: &paymenttypes.Money{Amount: "5.00", Currency: "USDT"},
Status: storagemodel.PaymentStatusSuccess,
}
if err := repo.payments.Upsert(context.Background(), record); err != nil {
t.Fatalf("failed to seed payment record: %v", err)
}
resp, err := svc.GetOperation(context.Background(), &connectorv1.GetOperationRequest{
OperationId: "payment-2:hop_2_settlement_fx_convert",
})
if err != nil {
t.Fatalf("GetOperation returned error: %v", err)
}
if resp.GetOperation() == nil {
t.Fatal("expected operation")
}
if got, want := resp.GetOperation().GetOperationId(), "payment-2:hop_2_settlement_fx_convert"; got != want {
t.Fatalf("operation_id mismatch: got=%q want=%q", got, want)
}
if got, want := resp.GetOperation().GetProviderRef(), "idem-settlement-2"; got != want {
t.Fatalf("provider_ref mismatch: got=%q want=%q", got, want)
}
}
func TestGetOperation_DoesNotResolveByIdempotencyKey(t *testing.T) {
svc, repo, _ := newTestService(t)
record := &storagemodel.PaymentRecord{
IdempotencyKey: "idem-settlement-3",
OperationRef: "payment-3:hop_2_settlement_fx_convert",
PaymentIntentID: "pi-3",
PaymentRef: "payment-3",
RequestedMoney: &paymenttypes.Money{Amount: "5.00", Currency: "USDT"},
Status: storagemodel.PaymentStatusSuccess,
}
if err := repo.payments.Upsert(context.Background(), record); err != nil {
t.Fatalf("failed to seed payment record: %v", err)
}
_, err := svc.GetOperation(context.Background(), &connectorv1.GetOperationRequest{
OperationId: "idem-settlement-3",
})
if err == nil {
t.Fatal("expected not found error")
}
if status.Code(err) != codes.NotFound {
t.Fatalf("unexpected error code: got=%s want=%s", status.Code(err), codes.NotFound)
}
}

View File

@@ -0,0 +1,47 @@
package gateway
import (
"context"
gatewayoutbox "github.com/tech/sendico/gateway/common/outbox"
"github.com/tech/sendico/pkg/db/transaction"
me "github.com/tech/sendico/pkg/messaging/envelope"
)
type tgOutboxProvider interface {
Outbox() gatewayoutbox.Store
}
type tgTransactionProvider interface {
TransactionFactory() transaction.Factory
}
func (s *Service) outboxStore() gatewayoutbox.Store {
provider, ok := s.repo.(tgOutboxProvider)
if !ok || provider == nil {
return nil
}
return provider.Outbox()
}
func (s *Service) startOutboxReliableProducer() error {
if s == nil || s.repo == nil {
return nil
}
return s.outbox.Start(s.logger, s.producer, s.outboxStore(), s.msgCfg)
}
func (s *Service) sendWithOutbox(ctx context.Context, env me.Envelope) error {
if err := s.startOutboxReliableProducer(); err != nil {
return err
}
return s.outbox.Send(ctx, env)
}
func (s *Service) executeTransaction(ctx context.Context, cb transaction.Callback) (any, error) {
provider, ok := s.repo.(tgTransactionProvider)
if !ok || provider == nil || provider.TransactionFactory() == nil {
return cb(ctx)
}
return provider.TransactionFactory().CreateTransaction().Execute(ctx, cb)
}

View File

@@ -0,0 +1,302 @@
package gateway
import (
"hash/fnv"
"strconv"
"strings"
"time"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
)
const (
scenarioMetadataKey = "chsettle_scenario"
scenarioMetadataAliasKey = "scenario"
)
type settlementScenario struct {
Name string
InitialStatus storagemodel.PaymentStatus
FinalStatus storagemodel.PaymentStatus
FinalDelay time.Duration
FailureReason string
}
type settlementScenarioTrace struct {
Source string
OverrideRaw string
OverrideNormalized string
AmountRaw string
AmountCurrency string
BucketSlot int
}
var scenarioFastSuccess = settlementScenario{
Name: "fast_success",
InitialStatus: storagemodel.PaymentStatusSuccess,
}
var scenarioSlowSuccess = settlementScenario{
Name: "slow_success",
InitialStatus: storagemodel.PaymentStatusWaiting,
FinalStatus: storagemodel.PaymentStatusSuccess,
FinalDelay: 30 * time.Second,
}
var scenarioFailImmediate = settlementScenario{
Name: "fail_immediate",
InitialStatus: storagemodel.PaymentStatusFailed,
FailureReason: "simulated_fail_immediate",
}
var scenarioFailTimeout = settlementScenario{
Name: "fail_timeout",
InitialStatus: storagemodel.PaymentStatusWaiting,
FinalStatus: storagemodel.PaymentStatusFailed,
FinalDelay: 45 * time.Second,
FailureReason: "simulated_fail_timeout",
}
var scenarioStuckPending = settlementScenario{
Name: "stuck_pending",
InitialStatus: storagemodel.PaymentStatusWaiting,
}
var scenarioRetryThenSuccess = settlementScenario{
Name: "retry_then_success",
InitialStatus: storagemodel.PaymentStatusProcessing,
FinalStatus: storagemodel.PaymentStatusSuccess,
FinalDelay: 25 * time.Second,
}
var scenarioWebhookDelayedSuccess = settlementScenario{
Name: "webhook_delayed_success",
InitialStatus: storagemodel.PaymentStatusWaiting,
FinalStatus: storagemodel.PaymentStatusSuccess,
FinalDelay: 60 * time.Second,
}
var scenarioSlowThenFail = settlementScenario{
Name: "slow_then_fail",
InitialStatus: storagemodel.PaymentStatusProcessing,
FinalStatus: storagemodel.PaymentStatusFailed,
FinalDelay: 75 * time.Second,
FailureReason: "simulated_slow_then_fail",
}
var scenarioPartialProgressStuck = settlementScenario{
Name: "partial_progress_stuck",
InitialStatus: storagemodel.PaymentStatusProcessing,
}
func resolveSettlementScenario(idempotencyKey string, amount *paymenttypes.Money, metadata map[string]string) settlementScenario {
scenario, _ := resolveSettlementScenarioWithTrace(idempotencyKey, amount, metadata)
return scenario
}
func resolveSettlementScenarioWithTrace(idempotencyKey string, amount *paymenttypes.Money, metadata map[string]string) (settlementScenario, settlementScenarioTrace) {
trace := settlementScenarioTrace{
BucketSlot: -1,
}
if amount != nil {
trace.AmountRaw = strings.TrimSpace(amount.Amount)
trace.AmountCurrency = strings.TrimSpace(amount.Currency)
}
overrideScenario, overrideRaw, overrideNormalized, overrideApplied := parseScenarioOverride(metadata)
if overrideRaw != "" {
trace.OverrideRaw = overrideRaw
trace.OverrideNormalized = overrideNormalized
}
if overrideApplied {
trace.Source = "explicit_override"
return overrideScenario, trace
}
slot, ok := amountModuloSlot(amount)
if ok {
if trace.OverrideRaw != "" {
trace.Source = "invalid_override_amount_bucket"
} else {
trace.Source = "amount_bucket"
}
trace.BucketSlot = slot
return scenarioBySlot(slot, idempotencyKey), trace
}
slot = hashModulo(idempotencyKey, 1000)
if trace.OverrideRaw != "" {
trace.Source = "invalid_override_idempotency_hash_bucket"
} else {
trace.Source = "idempotency_hash_bucket"
}
trace.BucketSlot = slot
return scenarioBySlot(slot, idempotencyKey), trace
}
func parseScenarioOverride(metadata map[string]string) (settlementScenario, string, string, bool) {
if len(metadata) == 0 {
return settlementScenario{}, "", "", false
}
overrideRaw := strings.TrimSpace(metadata[scenarioMetadataKey])
if overrideRaw == "" {
overrideRaw = strings.TrimSpace(metadata[scenarioMetadataAliasKey])
}
if overrideRaw == "" {
return settlementScenario{}, "", "", false
}
scenario, normalized, ok := scenarioByName(overrideRaw)
return scenario, overrideRaw, normalized, ok
}
func scenarioByName(value string) (settlementScenario, string, bool) {
key := normalizeScenarioName(value)
switch key {
case "fast_success", "success_fast", "instant_success":
return scenarioFastSuccess, key, true
case "slow_success", "success_slow":
return scenarioSlowSuccess, key, true
case "fail_immediate", "immediate_fail", "failed":
return scenarioFailImmediate, key, true
case "fail_timeout", "timeout_fail":
return scenarioFailTimeout, key, true
case "stuck", "stuck_pending", "pending_stuck":
return scenarioStuckPending, key, true
case "retry_then_success":
return scenarioRetryThenSuccess, key, true
case "webhook_delayed_success":
return scenarioWebhookDelayedSuccess, key, true
case "slow_then_fail":
return scenarioSlowThenFail, key, true
case "partial_progress_stuck":
return scenarioPartialProgressStuck, key, true
case "chaos", "chaos_random_seeded":
return scenarioBySlot(950, ""), key, true
default:
return settlementScenario{}, key, false
}
}
func normalizeScenarioName(value string) string {
key := strings.ToLower(strings.TrimSpace(value))
key = strings.ReplaceAll(key, "-", "_")
return key
}
func scenarioBySlot(slot int, seed string) settlementScenario {
switch {
case slot < 100:
return scenarioFastSuccess
case slot < 200:
return scenarioSlowSuccess
case slot < 300:
return scenarioFailImmediate
case slot < 400:
return scenarioFailTimeout
case slot < 500:
return scenarioStuckPending
case slot < 600:
return scenarioRetryThenSuccess
case slot < 700:
return scenarioWebhookDelayedSuccess
case slot < 800:
return scenarioSlowThenFail
case slot < 900:
return scenarioPartialProgressStuck
default:
return chaosScenario(seed)
}
}
func chaosScenario(seed string) settlementScenario {
choices := []settlementScenario{
scenarioFastSuccess,
scenarioSlowSuccess,
scenarioFailImmediate,
scenarioFailTimeout,
scenarioStuckPending,
scenarioSlowThenFail,
}
idx := hashModulo(seed, len(choices))
return choices[idx]
}
func amountModuloSlot(amount *paymenttypes.Money) (int, bool) {
if amount == nil {
return 0, false
}
raw := strings.TrimSpace(amount.Amount)
if raw == "" {
return 0, false
}
sign := 1
if strings.HasPrefix(raw, "+") {
raw = strings.TrimPrefix(raw, "+")
}
if strings.HasPrefix(raw, "-") {
sign = -1
raw = strings.TrimPrefix(raw, "-")
}
parts := strings.SplitN(raw, ".", 3)
if len(parts) == 0 || len(parts) > 2 {
return 0, false
}
whole := parts[0]
if whole == "" || !digitsOnly(whole) {
return 0, false
}
frac := "00"
if len(parts) == 2 {
f := parts[1]
if f == "" || !digitsOnly(f) {
return 0, false
}
if len(f) >= 2 {
frac = f[:2]
} else {
frac = f + "0"
}
}
wholeMod := digitsMod(whole, 10)
fracVal, _ := strconv.Atoi(frac)
slot := (wholeMod*100 + fracVal) % 1000
if sign < 0 {
slot = (-slot + 1000) % 1000
}
return slot, true
}
func digitsOnly(value string) bool {
if value == "" {
return false
}
for i := 0; i < len(value); i++ {
if value[i] < '0' || value[i] > '9' {
return false
}
}
return true
}
func digitsMod(value string, mod int) int {
if mod <= 0 {
return 0
}
result := 0
for i := 0; i < len(value); i++ {
digit := int(value[i] - '0')
result = (result*10 + digit) % mod
}
return result
}
func hashModulo(input string, mod int) int {
if mod <= 0 {
return 0
}
h := fnv.New32a()
_, _ = h.Write([]byte(strings.TrimSpace(input)))
return int(h.Sum32() % uint32(mod))
}
func (s settlementScenario) delayedTransitionEnabled() bool {
return s.FinalStatus != "" && s.FinalDelay > 0
}

View File

@@ -0,0 +1,105 @@
package gateway
import (
"context"
"testing"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
)
func TestResolveSettlementScenario_AmountBuckets(t *testing.T) {
tests := []struct {
name string
amount string
want string
}{
{name: "bucket_000_fast_success", amount: "10.00", want: scenarioFastSuccess.Name},
{name: "bucket_100_slow_success", amount: "11.00", want: scenarioSlowSuccess.Name},
{name: "bucket_200_fail_immediate", amount: "12.00", want: scenarioFailImmediate.Name},
{name: "bucket_300_fail_timeout", amount: "13.00", want: scenarioFailTimeout.Name},
{name: "bucket_400_stuck_pending", amount: "14.00", want: scenarioStuckPending.Name},
{name: "bucket_500_retry_then_success", amount: "15.00", want: scenarioRetryThenSuccess.Name},
{name: "bucket_600_webhook_delayed_success", amount: "16.00", want: scenarioWebhookDelayedSuccess.Name},
{name: "bucket_700_slow_then_fail", amount: "17.00", want: scenarioSlowThenFail.Name},
{name: "bucket_800_partial_progress_stuck", amount: "18.00", want: scenarioPartialProgressStuck.Name},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got := resolveSettlementScenario("idem-"+tc.name, &paymenttypes.Money{Amount: tc.amount, Currency: "USD"}, nil)
if got.Name != tc.want {
t.Fatalf("scenario mismatch: got=%q want=%q", got.Name, tc.want)
}
})
}
}
func TestResolveSettlementScenario_ExplicitOverride(t *testing.T) {
got := resolveSettlementScenario("idem-override", &paymenttypes.Money{Amount: "10.00", Currency: "USD"}, map[string]string{
scenarioMetadataKey: "stuck",
})
if got.Name != scenarioStuckPending.Name {
t.Fatalf("scenario mismatch: got=%q want=%q", got.Name, scenarioStuckPending.Name)
}
}
func TestSubmitTransfer_UsesImmediateFailureScenario(t *testing.T) {
svc, repo, _ := newTestService(t)
resp, err := svc.SubmitTransfer(context.Background(), &chainv1.SubmitTransferRequest{
IdempotencyKey: "idem-immediate-fail",
IntentRef: "intent-immediate-fail",
OperationRef: "op-immediate-fail",
PaymentRef: "payment-immediate-fail",
Amount: &moneyv1.Money{Amount: "9.99", Currency: "USD"},
Metadata: map[string]string{
scenarioMetadataAliasKey: "fail_immediate",
},
})
if err != nil {
t.Fatalf("submit transfer failed: %v", err)
}
if got, want := resp.GetTransfer().GetStatus(), chainv1.TransferStatus_TRANSFER_FAILED; got != want {
t.Fatalf("transfer status mismatch: got=%v want=%v", got, want)
}
record := repo.payments.records["idem-immediate-fail"]
if record == nil {
t.Fatalf("expected payment record")
}
if got, want := record.Status, storagemodel.PaymentStatusFailed; got != want {
t.Fatalf("record status mismatch: got=%s want=%s", got, want)
}
if got, want := record.Scenario, scenarioFailImmediate.Name; got != want {
t.Fatalf("record scenario mismatch: got=%q want=%q", got, want)
}
}
func TestApplyScenarioTransition_SetsFinalSuccess(t *testing.T) {
svc, repo, _ := newTestService(t)
_ = repo.payments.Upsert(context.Background(), &storagemodel.PaymentRecord{
IdempotencyKey: "idem-transition-success",
IntentRef: "intent-transition-success",
OperationRef: "op-transition-success",
PaymentRef: "payment-transition-success",
RequestedMoney: &paymenttypes.Money{Amount: "5.00", Currency: "USD"},
Status: storagemodel.PaymentStatusWaiting,
Scenario: scenarioSlowSuccess.Name,
})
svc.applyScenarioTransition("idem-transition-success", scenarioSlowSuccess)
record := repo.payments.records["idem-transition-success"]
if record == nil {
t.Fatalf("expected payment record")
}
if got, want := record.Status, storagemodel.PaymentStatusSuccess; got != want {
t.Fatalf("record status mismatch: got=%s want=%s", got, want)
}
if record.ExecutedMoney == nil {
t.Fatalf("expected executed money")
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,417 @@
package gateway
import (
"context"
"sync"
"testing"
"time"
"github.com/tech/sendico/gateway/chsettle/storage"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/discovery"
envelope "github.com/tech/sendico/pkg/messaging/envelope"
tnotifications "github.com/tech/sendico/pkg/messaging/notifications/telegram"
mloggerfactory "github.com/tech/sendico/pkg/mlogger/factory"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
paymenttypes "github.com/tech/sendico/pkg/payments/types"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
)
//
// FAKE STORES
//
type fakePaymentsStore struct {
mu sync.Mutex
records map[string]*storagemodel.PaymentRecord
}
func (f *fakePaymentsStore) FindByIdempotencyKey(_ context.Context, key string) (*storagemodel.PaymentRecord, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
return nil, nil
}
return f.records[key], nil
}
func (f *fakePaymentsStore) FindByOperationRef(_ context.Context, key string) (*storagemodel.PaymentRecord, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
return nil, nil
}
for _, record := range f.records {
if record != nil && record.OperationRef == key {
return record, nil
}
}
return nil, nil
}
func (f *fakePaymentsStore) Upsert(_ context.Context, record *storagemodel.PaymentRecord) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
f.records = map[string]*storagemodel.PaymentRecord{}
}
f.records[record.IdempotencyKey] = record
return nil
}
type fakeTelegramStore struct {
mu sync.Mutex
records map[string]*storagemodel.TelegramConfirmation
}
func (f *fakeTelegramStore) Upsert(_ context.Context, record *storagemodel.TelegramConfirmation) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
f.records = map[string]*storagemodel.TelegramConfirmation{}
}
f.records[record.RequestID] = record
return nil
}
type fakeRepo struct {
payments *fakePaymentsStore
tg *fakeTelegramStore
pending *fakePendingStore
treasury storage.TreasuryRequestsStore
users storage.TreasuryTelegramUsersStore
}
func (f *fakeRepo) Payments() storage.PaymentsStore {
return f.payments
}
func (f *fakeRepo) TelegramConfirmations() storage.TelegramConfirmationsStore {
return f.tg
}
func (f *fakeRepo) PendingConfirmations() storage.PendingConfirmationsStore {
return f.pending
}
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
}
func (f *fakePendingStore) Upsert(_ context.Context, record *storagemodel.PendingConfirmation) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
f.records = map[string]*storagemodel.PendingConfirmation{}
}
cp := *record
f.records[record.RequestID] = &cp
return nil
}
func (f *fakePendingStore) FindByRequestID(_ context.Context, requestID string) (*storagemodel.PendingConfirmation, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.records == nil {
return nil, nil
}
return f.records[requestID], nil
}
func (f *fakePendingStore) FindByMessageID(_ context.Context, messageID string) (*storagemodel.PendingConfirmation, error) {
f.mu.Lock()
defer f.mu.Unlock()
for _, record := range f.records {
if record != nil && record.MessageID == messageID {
return record, nil
}
}
return nil, nil
}
func (f *fakePendingStore) MarkClarified(_ context.Context, requestID string) error {
f.mu.Lock()
defer f.mu.Unlock()
if record := f.records[requestID]; record != nil {
record.Clarified = true
}
return nil
}
func (f *fakePendingStore) AttachMessage(_ context.Context, requestID string, messageID string) error {
f.mu.Lock()
defer f.mu.Unlock()
if record := f.records[requestID]; record != nil {
if record.MessageID == "" {
record.MessageID = messageID
}
}
return nil
}
func (f *fakePendingStore) DeleteByRequestID(_ context.Context, requestID string) error {
f.mu.Lock()
defer f.mu.Unlock()
delete(f.records, requestID)
return nil
}
func (f *fakePendingStore) ListExpired(_ context.Context, now time.Time, limit int64) ([]storagemodel.PendingConfirmation, error) {
f.mu.Lock()
defer f.mu.Unlock()
if limit <= 0 {
limit = 100
}
result := make([]storagemodel.PendingConfirmation, 0)
for _, record := range f.records {
if record == nil || record.ExpiresAt.IsZero() || record.ExpiresAt.After(now) {
continue
}
result = append(result, *record)
if int64(len(result)) >= limit {
break
}
}
return result, nil
}
//
// FAKE BROKER (ОБЯЗАТЕЛЕН ДЛЯ СЕРВИСА)
//
type fakeBroker struct{}
func (f *fakeBroker) Publish(_ envelope.Envelope) error {
return nil
}
func (f *fakeBroker) Subscribe(event model.NotificationEvent) (<-chan envelope.Envelope, error) {
return nil, nil
}
func (f *fakeBroker) Unsubscribe(event model.NotificationEvent, subChan <-chan envelope.Envelope) error {
return nil
}
//
// CAPTURE ONLY TELEGRAM REACTIONS
//
type captureProducer struct {
mu sync.Mutex
reactions []envelope.Envelope
sig string
}
func (c *captureProducer) SendMessage(env envelope.Envelope) error {
if env.GetSignature().ToString() != c.sig {
return nil
}
c.mu.Lock()
c.reactions = append(c.reactions, env)
c.mu.Unlock()
return nil
}
//
// TESTS
//
func newTestService(_ *testing.T) (*Service, *fakeRepo, *captureProducer) {
logger := mloggerfactory.NewLogger(false)
repo := &fakeRepo{
payments: &fakePaymentsStore{},
tg: &fakeTelegramStore{},
pending: &fakePendingStore{},
}
sigEnv := tnotifications.TelegramReaction(string(mservice.PaymentGateway), &model.TelegramReactionRequest{
RequestID: "x",
ChatID: "1",
MessageID: "2",
Emoji: "ok",
})
prod := &captureProducer{
sig: sigEnv.GetSignature().ToString(),
}
svc := NewService(logger, repo, prod, &fakeBroker{}, Config{
Rail: "card",
SuccessReaction: "👍",
})
return svc, repo, prod
}
func TestConfirmed(t *testing.T) {
svc, repo, prod := newTestService(t)
_ = repo.payments.Upsert(context.Background(), &storagemodel.PaymentRecord{
IdempotencyKey: "idem-1",
PaymentIntentID: "pi-1",
QuoteRef: "quote-1",
OutgoingLeg: "card",
RequestedMoney: &paymenttypes.Money{Amount: "5", Currency: "EUR"},
Status: storagemodel.PaymentStatusWaiting,
})
result := &model.ConfirmationResult{
RequestID: "idem-1",
Money: &paymenttypes.Money{Amount: "5", Currency: "EUR"},
Status: model.ConfirmationStatusConfirmed,
RawReply: &model.TelegramMessage{ChatID: "1", MessageID: "2"},
}
_ = svc.onConfirmationResult(context.Background(), result)
rec := repo.payments.records["idem-1"]
if rec.Status != storagemodel.PaymentStatusSuccess {
t.Fatalf("expected success, got %s", rec.Status)
}
if rec.RequestedMoney == nil {
t.Fatalf("requested money not set")
}
if rec.ExecutedAt.IsZero() {
t.Fatalf("executedAt not set")
}
if repo.tg.records["idem-1"] == nil {
t.Fatalf("telegram confirmation not stored")
}
if len(prod.reactions) != 1 {
t.Fatalf("reaction must be published")
}
}
func TestClarified(t *testing.T) {
svc, repo, prod := newTestService(t)
_ = repo.payments.Upsert(context.Background(), &storagemodel.PaymentRecord{
IdempotencyKey: "idem-2",
Status: storagemodel.PaymentStatusWaiting,
})
result := &model.ConfirmationResult{
RequestID: "idem-2",
Status: model.ConfirmationStatusClarified,
RawReply: &model.TelegramMessage{ChatID: "1", MessageID: "2"},
}
_ = svc.onConfirmationResult(context.Background(), result)
rec := repo.payments.records["idem-2"]
if rec.Status != storagemodel.PaymentStatusWaiting {
t.Fatalf("clarified must not change status")
}
if repo.tg.records["idem-2"] == nil {
t.Fatalf("telegram confirmation must be stored")
}
if len(prod.reactions) != 0 {
t.Fatalf("clarified must not publish reaction")
}
}
func TestRejected(t *testing.T) {
svc, repo, prod := newTestService(t)
// ВАЖНО: чтобы текущий emitTransferStatusEvent не падал на nil,
// даем минимально ожидаемые поля + non-nil ExecutedMoney.
_ = repo.payments.Upsert(context.Background(), &storagemodel.PaymentRecord{
IdempotencyKey: "idem-3",
PaymentIntentID: "pi-3",
QuoteRef: "quote-3",
OutgoingLeg: "card",
RequestedMoney: &paymenttypes.Money{Amount: "5", Currency: "EUR"},
ExecutedMoney: &paymenttypes.Money{Amount: "0", Currency: "EUR"},
Status: storagemodel.PaymentStatusWaiting,
})
result := &model.ConfirmationResult{
RequestID: "idem-3",
Status: model.ConfirmationStatusRejected,
RawReply: &model.TelegramMessage{ChatID: "1", MessageID: "2"},
}
_ = svc.onConfirmationResult(context.Background(), result)
rec := repo.payments.records["idem-3"]
if rec.Status != storagemodel.PaymentStatusFailed {
t.Fatalf("expected failed")
}
if repo.tg.records["idem-3"] == nil {
t.Fatalf("telegram confirmation must be stored")
}
if len(prod.reactions) != 0 {
t.Fatalf("rejected must not publish reaction")
}
}
func TestTimeout(t *testing.T) {
svc, repo, prod := newTestService(t)
// ВАЖНО: чтобы текущий emitTransferStatusEvent не падал на nil,
// даем минимально ожидаемые поля + non-nil ExecutedMoney.
_ = repo.payments.Upsert(context.Background(), &storagemodel.PaymentRecord{
IdempotencyKey: "idem-4",
PaymentIntentID: "pi-4",
QuoteRef: "quote-4",
OutgoingLeg: "card",
RequestedMoney: &paymenttypes.Money{Amount: "5", Currency: "EUR"},
ExecutedMoney: &paymenttypes.Money{Amount: "0", Currency: "EUR"},
Status: storagemodel.PaymentStatusWaiting,
})
result := &model.ConfirmationResult{
RequestID: "idem-4",
Status: model.ConfirmationStatusTimeout,
RawReply: &model.TelegramMessage{ChatID: "1", MessageID: "2"},
}
_ = svc.onConfirmationResult(context.Background(), result)
rec := repo.payments.records["idem-4"]
if rec.Status != storagemodel.PaymentStatusFailed {
t.Fatalf("timeout must be failed")
}
if repo.tg.records["idem-4"] == nil {
t.Fatalf("telegram confirmation must be stored")
}
if len(prod.reactions) != 0 {
t.Fatalf("timeout must not publish reaction")
}
}
func TestIntentFromSubmitTransfer_NormalizesOutgoingLeg(t *testing.T) {
intent, err := intentFromSubmitTransfer(&chainv1.SubmitTransferRequest{
IdempotencyKey: "idem-5",
IntentRef: "pi-5",
OperationRef: "op-5",
PaymentRef: "pay-5",
Amount: &moneyv1.Money{Amount: "10", Currency: "USD"},
Metadata: map[string]string{
metadataOutgoingLeg: "card",
},
}, "provider_settlement", "")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got, want := intent.OutgoingLeg, discovery.RailCardPayout; got != want {
t.Fatalf("unexpected outgoing leg: got=%q want=%q", got, want)
}
}

View File

@@ -0,0 +1,108 @@
package gateway
import (
"context"
"github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
paymentgateway "github.com/tech/sendico/pkg/messaging/notifications/paymentgateway"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
"github.com/tech/sendico/pkg/mutil/mzap"
"github.com/tech/sendico/pkg/payments/rail"
"go.uber.org/zap"
)
func isFinalStatus(t *model.PaymentRecord) bool {
switch t.Status {
case model.PaymentStatusFailed, model.PaymentStatusSuccess, model.PaymentStatusCancelled:
return true
default:
return false
}
}
func toOpStatus(t *model.PaymentRecord) (rail.OperationResult, error) {
switch t.Status {
case model.PaymentStatusFailed:
return rail.OperationResultFailed, nil
case model.PaymentStatusSuccess:
return rail.OperationResultSuccess, nil
case model.PaymentStatusCancelled:
return rail.OperationResultCancelled, nil
default:
return rail.OperationResultFailed, merrors.InvalidArgument("unexpected transfer status", "payment.status")
}
}
func (s *Service) updateTransferStatus(ctx context.Context, record *model.PaymentRecord) error {
if record == nil {
return merrors.InvalidArgument("payment record is required", "record")
}
s.logger.Debug("Persisting transfer status",
zap.String("idempotency_key", record.IdempotencyKey),
zap.String("payment_ref", record.PaymentIntentID),
zap.String("status", string(record.Status)),
zap.Bool("is_final", isFinalStatus(record)))
if !isFinalStatus(record) {
if err := s.repo.Payments().Upsert(ctx, record); err != nil {
s.logger.Warn("Failed to update transfer status", zap.String("payment_ref", record.PaymentIntentID), zap.String("status", string(record.Status)), zap.Error(err))
return err
}
s.logger.Debug("Transfer status persisted (non-final)",
zap.String("idempotency_key", record.IdempotencyKey),
zap.String("status", string(record.Status)))
return nil
}
_, err := s.executeTransaction(ctx, func(txCtx context.Context) (any, error) {
if upsertErr := s.repo.Payments().Upsert(txCtx, record); upsertErr != nil {
return nil, upsertErr
}
if isFinalStatus(record) {
if emitErr := s.emitTransferStatusEvent(txCtx, record); emitErr != nil {
return nil, emitErr
}
}
return nil, nil
})
if err != nil {
s.logger.Warn("Failed to update transfer status", zap.String("payment_ref", record.PaymentIntentID), zap.String("status", string(record.Status)), zap.Error(err))
return err
}
s.logger.Info("Transfer status persisted (final)",
zap.String("idempotency_key", record.IdempotencyKey),
zap.String("status", string(record.Status)))
return nil
}
func (s *Service) emitTransferStatusEvent(ctx context.Context, record *model.PaymentRecord) error {
if s == nil || record == nil {
return nil
}
if s.producer == nil || s.outboxStore() == nil {
return nil
}
status, err := toOpStatus(record)
if err != nil {
s.logger.Warn("Failed to map transfer status for transfer status event", zap.Error(err), mzap.ObjRef("transfer_ref", record.ID))
return err
}
exec := pmodel.PaymentGatewayExecution{
PaymentIntentID: record.PaymentIntentID,
IdempotencyKey: record.IdempotencyKey,
ExecutedMoney: record.ExecutedMoney,
PaymentRef: record.PaymentRef,
Status: status,
OperationRef: record.OperationRef,
Error: record.FailureReason,
TransferRef: record.ID.Hex(),
}
env := paymentgateway.PaymentGatewayExecution(mservice.ChSettle, &exec)
if sendErr := s.sendWithOutbox(ctx, env); sendErr != nil {
s.logger.Warn("Failed to publish transfer status event", zap.Error(sendErr), mzap.ObjRef("transfer_ref", record.ID))
return sendErr
}
return nil
}

View File

@@ -0,0 +1,94 @@
package bot
import "strings"
type Command string
const (
CommandStart Command = "start"
CommandHelp Command = "help"
CommandFund Command = "fund"
CommandWithdraw Command = "withdraw"
CommandConfirm Command = "confirm"
CommandCancel Command = "cancel"
)
var supportedCommands = []Command{
CommandStart,
CommandHelp,
CommandFund,
CommandWithdraw,
CommandConfirm,
CommandCancel,
}
func (c Command) Slash() string {
name := strings.TrimSpace(string(c))
if name == "" {
return ""
}
return "/" + name
}
func parseCommand(text string) Command {
text = strings.TrimSpace(text)
if !strings.HasPrefix(text, "/") {
return ""
}
token := text
if idx := strings.IndexAny(token, " \t\n\r"); idx >= 0 {
token = token[:idx]
}
token = strings.TrimPrefix(token, "/")
if idx := strings.Index(token, "@"); idx >= 0 {
token = token[:idx]
}
return Command(strings.ToLower(strings.TrimSpace(token)))
}
func supportedCommandsMessage() string {
lines := make([]string, 0, len(supportedCommands)+2)
lines = append(lines, "*Supported Commands*")
lines = append(lines, "")
for _, cmd := range supportedCommands {
lines = append(lines, markdownCommand(cmd))
}
return strings.Join(lines, "\n")
}
func confirmationCommandsMessage() string {
return strings.Join([]string{
"*Confirm Operation*",
"",
"Use " + markdownCommand(CommandConfirm) + " to execute.",
"Use " + markdownCommand(CommandCancel) + " to abort.",
}, "\n")
}
func helpMessage(accountCode string, currency string) string {
accountCode = strings.TrimSpace(accountCode)
currency = strings.ToUpper(strings.TrimSpace(currency))
if accountCode == "" {
accountCode = "N/A"
}
if currency == "" {
currency = "N/A"
}
lines := []string{
"*Treasury Bot Help*",
"",
"*Attached account:* " + markdownCode(accountCode) + " (" + markdownCode(currency) + ")",
"",
"*How to use*",
"1. Start funding with " + markdownCommand(CommandFund) + " or withdrawal with " + markdownCommand(CommandWithdraw) + ".",
"2. Enter amount as decimal with dot separator and no currency.",
" Example: " + markdownCode("1250.75"),
"3. Confirm with " + markdownCommand(CommandConfirm) + " or abort with " + markdownCommand(CommandCancel) + ".",
"",
"*Cooldown*",
"After confirmation there is a cooldown window. You can cancel during it with " + markdownCommand(CommandCancel) + ".",
"You will receive a follow-up message with execution success or failure.",
}
return strings.Join(lines, "\n")
}

View File

@@ -0,0 +1,73 @@
package bot
import (
"strings"
"sync"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
)
type DialogState string
const (
DialogStateWaitingAmount DialogState = "waiting_amount"
DialogStateWaitingConfirmation DialogState = "waiting_confirmation"
)
type DialogSession struct {
State DialogState
OperationType storagemodel.TreasuryOperationType
LedgerAccountID string
RequestID string
}
type Dialogs struct {
mu sync.Mutex
sessions map[string]DialogSession
}
func NewDialogs() *Dialogs {
return &Dialogs{
sessions: map[string]DialogSession{},
}
}
func (d *Dialogs) Get(telegramUserID string) (DialogSession, bool) {
if d == nil {
return DialogSession{}, false
}
telegramUserID = strings.TrimSpace(telegramUserID)
if telegramUserID == "" {
return DialogSession{}, false
}
d.mu.Lock()
defer d.mu.Unlock()
session, ok := d.sessions[telegramUserID]
return session, ok
}
func (d *Dialogs) Set(telegramUserID string, session DialogSession) {
if d == nil {
return
}
telegramUserID = strings.TrimSpace(telegramUserID)
if telegramUserID == "" {
return
}
d.mu.Lock()
defer d.mu.Unlock()
d.sessions[telegramUserID] = session
}
func (d *Dialogs) Clear(telegramUserID string) {
if d == nil {
return
}
telegramUserID = strings.TrimSpace(telegramUserID)
if telegramUserID == "" {
return
}
d.mu.Lock()
defer d.mu.Unlock()
delete(d.sessions, telegramUserID)
}

View File

@@ -0,0 +1,18 @@
package bot
import (
"strings"
)
func markdownCode(value string) string {
value = strings.TrimSpace(value)
if value == "" {
value = "N/A"
}
value = strings.ReplaceAll(value, "`", "'")
return "`" + value + "`"
}
func markdownCommand(command Command) string {
return command.Slash()
}

View File

@@ -0,0 +1,527 @@
package bot
import (
"context"
"errors"
"strconv"
"strings"
"time"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/model"
"go.uber.org/zap"
)
const unauthorizedMessage = "*Unauthorized*\nYour Telegram account is not allowed to perform treasury operations."
const unauthorizedChatMessage = "*Unauthorized Chat*\nThis Telegram chat is not allowed to perform treasury operations."
const amountInputHint = "*Amount format*\nEnter amount as a decimal number using a dot separator and without currency.\nExample: `1250.75`"
type SendTextFunc func(ctx context.Context, chatID string, text string) error
type ScheduleTracker interface {
TrackScheduled(record *storagemodel.TreasuryRequest)
Untrack(requestID string)
}
type AccountProfile struct {
AccountID string
AccountCode string
Currency string
}
type CreateRequestInput struct {
OperationType storagemodel.TreasuryOperationType
TelegramUserID string
LedgerAccountID string
ChatID string
Amount string
}
type TreasuryService interface {
ExecutionDelay() time.Duration
MaxPerOperationLimit() string
GetActiveRequestForAccount(ctx context.Context, ledgerAccountID string) (*storagemodel.TreasuryRequest, error)
GetAccountProfile(ctx context.Context, ledgerAccountID string) (*AccountProfile, error)
CreateRequest(ctx context.Context, input CreateRequestInput) (*storagemodel.TreasuryRequest, error)
ConfirmRequest(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 {
error
LimitKind() string
LimitMax() string
}
type Router struct {
logger mlogger.Logger
service TreasuryService
dialogs *Dialogs
send SendTextFunc
tracker ScheduleTracker
users UserBindingResolver
}
func NewRouter(
logger mlogger.Logger,
service TreasuryService,
send SendTextFunc,
tracker ScheduleTracker,
users UserBindingResolver,
) *Router {
if logger != nil {
logger = logger.Named("treasury_router")
}
return &Router{
logger: logger,
service: service,
dialogs: NewDialogs(),
send: send,
tracker: tracker,
users: users,
}
}
func (r *Router) Enabled() bool {
return r != nil && r.service != nil && r.users != nil
}
func (r *Router) HandleUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) bool {
if !r.Enabled() || update == nil || update.Message == nil {
return false
}
message := update.Message
chatID := strings.TrimSpace(message.ChatID)
userID := strings.TrimSpace(message.FromUserID)
text := strings.TrimSpace(message.Text)
if chatID == "" || userID == "" {
return false
}
command := parseCommand(text)
if r.logger != nil {
r.logger.Debug("Telegram treasury update received",
zap.Int64("update_id", update.UpdateID),
zap.String("chat_id", chatID),
zap.String("telegram_user_id", userID),
zap.String("command", strings.TrimSpace(string(command))),
zap.String("message_text", text),
zap.String("reply_to_message_id", strings.TrimSpace(message.ReplyToMessageID)),
)
}
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
}
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:
profile := r.resolveAccountProfile(ctx, accountID)
_ = r.sendText(ctx, chatID, welcomeMessage(profile))
return true
case CommandHelp:
profile := r.resolveAccountProfile(ctx, accountID)
_ = r.sendText(ctx, chatID, helpMessage(displayAccountCode(profile, accountID), profile.Currency))
return true
case CommandFund:
if r.logger != nil {
r.logger.Info("Treasury funding dialog requested",
zap.String("chat_id", chatID),
zap.String("telegram_user_id", userID),
zap.String("ledger_account_id", accountID))
}
r.startAmountDialog(ctx, userID, accountID, chatID, storagemodel.TreasuryOperationFund)
return true
case CommandWithdraw:
if r.logger != nil {
r.logger.Info("Treasury withdrawal dialog requested",
zap.String("chat_id", chatID),
zap.String("telegram_user_id", userID),
zap.String("ledger_account_id", accountID))
}
r.startAmountDialog(ctx, userID, accountID, chatID, storagemodel.TreasuryOperationWithdraw)
return true
case CommandConfirm:
if r.logger != nil {
r.logger.Info("Treasury confirmation requested",
zap.String("chat_id", chatID),
zap.String("telegram_user_id", userID),
zap.String("ledger_account_id", accountID))
}
r.confirm(ctx, userID, accountID, chatID)
return true
case CommandCancel:
if r.logger != nil {
r.logger.Info("Treasury cancellation requested",
zap.String("chat_id", chatID),
zap.String("telegram_user_id", userID),
zap.String("ledger_account_id", accountID))
}
r.cancel(ctx, userID, accountID, chatID)
return true
}
session, hasSession := r.dialogs.Get(userID)
if hasSession {
switch session.State {
case DialogStateWaitingAmount:
r.captureAmount(ctx, userID, accountID, chatID, session.OperationType, text)
return true
case DialogStateWaitingConfirmation:
_ = r.sendText(ctx, chatID, confirmationCommandsMessage())
return true
}
}
if strings.HasPrefix(text, "/") {
_ = r.sendText(ctx, chatID, supportedCommandsMessage())
return true
}
if strings.TrimSpace(message.ReplyToMessageID) != "" {
return false
}
if text != "" {
_ = r.sendText(ctx, chatID, supportedCommandsMessage())
return true
}
return false
}
func (r *Router) startAmountDialog(ctx context.Context, userID, accountID, chatID string, operation storagemodel.TreasuryOperationType) {
active, err := r.service.GetActiveRequestForAccount(ctx, accountID)
if err != nil {
if r.logger != nil {
r.logger.Warn("Failed to check active treasury request", zap.Error(err), zap.String("telegram_user_id", userID), zap.String("ledger_account_id", accountID))
}
_ = r.sendText(ctx, chatID, "*Temporary issue*\nUnable to check pending treasury operations right now. Please try again.")
return
}
if active != nil {
_ = r.sendText(ctx, chatID, pendingRequestMessage(active))
r.dialogs.Set(userID, DialogSession{
State: DialogStateWaitingConfirmation,
LedgerAccountID: accountID,
RequestID: active.RequestID,
})
return
}
r.dialogs.Set(userID, DialogSession{
State: DialogStateWaitingAmount,
OperationType: operation,
LedgerAccountID: accountID,
})
profile := r.resolveAccountProfile(ctx, accountID)
_ = r.sendText(ctx, chatID, amountPromptMessage(operation, profile, accountID))
}
func (r *Router) captureAmount(ctx context.Context, userID, accountID, chatID string, operation storagemodel.TreasuryOperationType, amount string) {
record, err := r.service.CreateRequest(ctx, CreateRequestInput{
OperationType: operation,
TelegramUserID: userID,
LedgerAccountID: accountID,
ChatID: chatID,
Amount: amount,
})
if err != nil {
if record != nil {
_ = r.sendText(ctx, chatID, pendingRequestMessage(record))
r.dialogs.Set(userID, DialogSession{
State: DialogStateWaitingConfirmation,
LedgerAccountID: accountID,
RequestID: record.RequestID,
})
return
}
if typed, ok := err.(limitError); ok {
switch typed.LimitKind() {
case "per_operation":
_ = r.sendText(ctx, chatID, "*Amount exceeds allowed limit*\n\n*Max per operation:* "+markdownCode(typed.LimitMax())+"\n\nEnter another amount or "+markdownCommand(CommandCancel)+".")
return
case "daily":
_ = r.sendText(ctx, chatID, "*Daily amount limit exceeded*\n\n*Max per day:* "+markdownCode(typed.LimitMax())+"\n\nEnter another amount or "+markdownCommand(CommandCancel)+".")
return
}
}
if errors.Is(err, merrors.ErrInvalidArg) {
_ = r.sendText(ctx, chatID, "*Invalid amount*\n\n"+amountInputHint+"\n\nEnter another amount or "+markdownCommand(CommandCancel)+".")
return
}
_ = r.sendText(ctx, chatID, "*Failed to create treasury request*\n\nEnter another amount or "+markdownCommand(CommandCancel)+".")
return
}
if record == nil {
_ = r.sendText(ctx, chatID, "*Failed to create treasury request*\n\nEnter another amount or "+markdownCommand(CommandCancel)+".")
return
}
r.dialogs.Set(userID, DialogSession{
State: DialogStateWaitingConfirmation,
LedgerAccountID: accountID,
RequestID: record.RequestID,
})
_ = r.sendText(ctx, chatID, confirmationPrompt(record))
}
func (r *Router) confirm(ctx context.Context, userID string, accountID string, chatID string) {
requestID := ""
if session, ok := r.dialogs.Get(userID); ok && strings.TrimSpace(session.RequestID) != "" {
requestID = strings.TrimSpace(session.RequestID)
} else {
active, err := r.service.GetActiveRequestForAccount(ctx, accountID)
if err == nil && active != nil {
requestID = strings.TrimSpace(active.RequestID)
}
}
if requestID == "" {
_ = r.sendText(ctx, chatID, "*No pending treasury operation.*")
return
}
record, err := r.service.ConfirmRequest(ctx, requestID, userID)
if err != nil {
_ = r.sendText(ctx, chatID, "*Unable to confirm treasury request.*\n\nUse "+markdownCommand(CommandCancel)+" or create a new request with "+markdownCommand(CommandFund)+" or "+markdownCommand(CommandWithdraw)+".")
return
}
if r.tracker != nil {
r.tracker.TrackScheduled(record)
}
r.dialogs.Clear(userID)
delay := int64(r.service.ExecutionDelay().Seconds())
if delay < 0 {
delay = 0
}
_ = r.sendText(ctx, chatID,
"*Operation confirmed*\n\n"+
"*Execution:* scheduled in "+markdownCode(formatSeconds(delay))+".\n"+
"You can cancel during this cooldown with "+markdownCommand(CommandCancel)+".\n\n"+
"You will receive a follow-up message with execution success or failure.\n\n"+
"*Request ID:* "+markdownCode(strings.TrimSpace(record.RequestID)))
}
func (r *Router) cancel(ctx context.Context, userID string, accountID string, chatID string) {
requestID := ""
if session, ok := r.dialogs.Get(userID); ok && strings.TrimSpace(session.RequestID) != "" {
requestID = strings.TrimSpace(session.RequestID)
} else {
active, err := r.service.GetActiveRequestForAccount(ctx, accountID)
if err == nil && active != nil {
requestID = strings.TrimSpace(active.RequestID)
}
}
if requestID == "" {
r.dialogs.Clear(userID)
_ = r.sendText(ctx, chatID, "*No pending treasury operation.*")
return
}
record, err := r.service.CancelRequest(ctx, requestID, userID)
if err != nil {
_ = r.sendText(ctx, chatID, "*Unable to cancel treasury request.*")
return
}
if r.tracker != nil {
r.tracker.Untrack(record.RequestID)
}
r.dialogs.Clear(userID)
_ = r.sendText(ctx, chatID, "*Operation cancelled*\n\n*Request ID:* "+markdownCode(strings.TrimSpace(record.RequestID)))
}
func (r *Router) sendText(ctx context.Context, chatID string, text string) error {
if r == nil || r.send == nil {
return nil
}
chatID = strings.TrimSpace(chatID)
text = strings.TrimSpace(text)
if chatID == "" || text == "" {
return nil
}
if err := r.send(ctx, chatID, text); err != nil {
if r.logger != nil {
r.logger.Warn("Failed to send treasury bot response",
zap.Error(err),
zap.String("chat_id", chatID),
zap.String("message_text", text))
}
return err
}
return nil
}
func (r *Router) logUnauthorized(update *model.TelegramWebhookUpdate) {
if r == nil || r.logger == nil || update == nil || update.Message == nil {
return
}
message := update.Message
r.logger.Warn("unauthorized_access",
zap.String("event", "unauthorized_access"),
zap.String("telegram_user_id", strings.TrimSpace(message.FromUserID)),
zap.String("chat_id", strings.TrimSpace(message.ChatID)),
zap.String("message_text", strings.TrimSpace(message.Text)),
zap.Time("timestamp", time.Now()),
)
}
func pendingRequestMessage(record *storagemodel.TreasuryRequest) string {
if record == nil {
return "*Pending treasury operation already exists.*\n\nUse " + markdownCommand(CommandCancel) + "."
}
return "*Pending Treasury Operation*\n\n" +
"*Account:* " + markdownCode(requestAccountDisplay(record)) + "\n" +
"*Request ID:* " + markdownCode(strings.TrimSpace(record.RequestID)) + "\n" +
"*Status:* " + markdownCode(strings.TrimSpace(string(record.Status))) + "\n" +
"*Amount:* " + markdownCode(strings.TrimSpace(record.Amount)+" "+strings.TrimSpace(record.Currency)) + "\n\n" +
"Wait for execution or cancel with " + markdownCommand(CommandCancel) + "."
}
func confirmationPrompt(record *storagemodel.TreasuryRequest) string {
if record == nil {
return "*Request created.*\n\nUse " + markdownCommand(CommandConfirm) + " or " + markdownCommand(CommandCancel) + "."
}
title := "*Funding request created.*"
if record.OperationType == storagemodel.TreasuryOperationWithdraw {
title = "*Withdrawal request created.*"
}
return title + "\n\n" +
"*Account:* " + markdownCode(requestAccountDisplay(record)) + "\n" +
"*Amount:* " + markdownCode(strings.TrimSpace(record.Amount)+" "+strings.TrimSpace(record.Currency)) + "\n\n" +
confirmationCommandsMessage()
}
func welcomeMessage(profile *AccountProfile) string {
accountCode := displayAccountCode(profile, "")
currency := ""
if profile != nil {
currency = strings.ToUpper(strings.TrimSpace(profile.Currency))
}
if accountCode == "" {
accountCode = "N/A"
}
if currency == "" {
currency = "N/A"
}
return "*Sendico Treasury Bot*\n\n" +
"*Attached account:* " + markdownCode(accountCode) + " (" + markdownCode(currency) + ")\n" +
"Use " + markdownCommand(CommandFund) + " to credit your account and " + markdownCommand(CommandWithdraw) + " to debit it.\n" +
"After entering an amount, use " + markdownCommand(CommandConfirm) + " or " + markdownCommand(CommandCancel) + ".\n" +
"Use " + markdownCommand(CommandHelp) + " for detailed usage."
}
func amountPromptMessage(operation storagemodel.TreasuryOperationType, profile *AccountProfile, fallbackAccountID string) string {
title := "*Funding request*"
if operation == storagemodel.TreasuryOperationWithdraw {
title = "*Withdrawal request*"
}
accountCode := displayAccountCode(profile, fallbackAccountID)
currency := ""
if profile != nil {
currency = strings.ToUpper(strings.TrimSpace(profile.Currency))
}
if accountCode == "" {
accountCode = "N/A"
}
if currency == "" {
currency = "N/A"
}
return title + "\n\n" +
"*Account:* " + markdownCode(accountCode) + " (" + markdownCode(currency) + ")\n\n" +
amountInputHint
}
func requestAccountDisplay(record *storagemodel.TreasuryRequest) string {
if record == nil {
return ""
}
if code := strings.TrimSpace(record.LedgerAccountCode); code != "" {
return code
}
return strings.TrimSpace(record.LedgerAccountID)
}
func displayAccountCode(profile *AccountProfile, fallbackAccountID string) string {
if profile != nil {
if code := strings.TrimSpace(profile.AccountCode); code != "" {
return code
}
if id := strings.TrimSpace(profile.AccountID); id != "" {
return id
}
}
return strings.TrimSpace(fallbackAccountID)
}
func (r *Router) resolveAccountProfile(ctx context.Context, ledgerAccountID string) *AccountProfile {
if r == nil || r.service == nil {
return &AccountProfile{AccountID: strings.TrimSpace(ledgerAccountID)}
}
profile, err := r.service.GetAccountProfile(ctx, ledgerAccountID)
if err != nil {
if r.logger != nil {
r.logger.Warn("Failed to resolve treasury account profile",
zap.Error(err),
zap.String("ledger_account_id", strings.TrimSpace(ledgerAccountID)))
}
return &AccountProfile{AccountID: strings.TrimSpace(ledgerAccountID)}
}
if profile == nil {
return &AccountProfile{AccountID: strings.TrimSpace(ledgerAccountID)}
}
if strings.TrimSpace(profile.AccountID) == "" {
profile.AccountID = strings.TrimSpace(ledgerAccountID)
}
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"
}
return strconv.FormatInt(value, 10) + " seconds"
}

View File

@@ -0,0 +1,362 @@
package bot
import (
"context"
"testing"
"time"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
mloggerfactory "github.com/tech/sendico/pkg/mlogger/factory"
"github.com/tech/sendico/pkg/model"
)
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
}
func (fakeService) MaxPerOperationLimit() string {
return "1000000"
}
func (fakeService) GetActiveRequestForAccount(context.Context, string) (*storagemodel.TreasuryRequest, error) {
return nil, nil
}
func (fakeService) GetAccountProfile(_ context.Context, ledgerAccountID string) (*AccountProfile, error) {
return &AccountProfile{
AccountID: ledgerAccountID,
AccountCode: ledgerAccountID,
Currency: "USD",
}, nil
}
func (fakeService) CreateRequest(context.Context, CreateRequestInput) (*storagemodel.TreasuryRequest, error) {
return nil, nil
}
func (fakeService) ConfirmRequest(context.Context, string, string) (*storagemodel.TreasuryRequest, error) {
return nil, nil
}
func (fakeService) CancelRequest(context.Context, string, string) (*storagemodel.TreasuryRequest, error) {
return nil, nil
}
func TestRouterUnauthorizedInAllowedChatSendsAccessDenied(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "100",
FromUserID: "999",
Text: "/fund",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != unauthorizedMessage {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterUnknownChatGetsDenied(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
AllowedChatIDs: []string{"100"},
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "999",
FromUserID: "123",
Text: "/fund",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != unauthorizedChatMessage {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterEmptyAllowedChats_AllowsAnyChatForAuthorizedUser(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "999",
FromUserID: "123",
Text: "/fund",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != amountPromptMessage(
storagemodel.TreasuryOperationFund,
&AccountProfile{AccountID: "acct-1", AccountCode: "acct-1", Currency: "USD"},
"acct-1",
) {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterEmptyAllowedChats_UnauthorizedUserGetsDenied(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "777",
FromUserID: "999",
Text: "/fund",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != unauthorizedMessage {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterStartAuthorizedShowsWelcome(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "777",
FromUserID: "123",
Text: "/start",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != welcomeMessage(&AccountProfile{AccountID: "acct-1", AccountCode: "acct-1", Currency: "USD"}) {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterHelpAuthorizedShowsHelp(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "777",
FromUserID: "123",
Text: "/help",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != helpMessage("acct-1", "USD") {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterStartUnauthorizedGetsDenied(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "777",
FromUserID: "999",
Text: "/start",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != unauthorizedMessage {
t.Fatalf("unexpected message: %q", sent[0])
}
}
func TestRouterPlainTextWithoutSession_ShowsSupportedCommands(t *testing.T) {
var sent []string
router := NewRouter(
mloggerfactory.NewLogger(false),
fakeService{},
func(_ context.Context, _ string, text string) error {
sent = append(sent, text)
return nil
},
nil,
fakeUserBindingResolver{
bindings: map[string]*UserBinding{
"123": {
TelegramUserID: "123",
LedgerAccountID: "acct-1",
},
},
},
)
handled := router.HandleUpdate(context.Background(), &model.TelegramWebhookUpdate{
Message: &model.TelegramMessage{
ChatID: "777",
FromUserID: "123",
Text: "hello",
},
})
if !handled {
t.Fatalf("expected update to be handled")
}
if len(sent) != 1 {
t.Fatalf("expected one message, got %d", len(sent))
}
if sent[0] != supportedCommandsMessage() {
t.Fatalf("unexpected message: %q", sent[0])
}
}

View File

@@ -0,0 +1,11 @@
package treasury
import "time"
type Config struct {
ExecutionDelay time.Duration
PollInterval time.Duration
MaxAmountPerOperation string
MaxDailyAmount string
}

View File

@@ -0,0 +1,312 @@
package ledger
import (
"context"
"crypto/tls"
"fmt"
"net/url"
"strings"
"time"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/types/known/structpb"
)
const ledgerConnectorID = "ledger"
type Config struct {
Endpoint string
Timeout time.Duration
Insecure bool
}
type Account struct {
AccountID string
AccountCode string
Currency string
OrganizationRef string
}
type Balance struct {
AccountID string
Amount string
Currency string
}
type PostRequest struct {
AccountID string
OrganizationRef string
Amount string
Currency string
Reference string
IdempotencyKey string
}
type OperationResult struct {
Reference string
}
type Client interface {
GetAccount(ctx context.Context, accountID string) (*Account, error)
GetBalance(ctx context.Context, accountID string) (*Balance, error)
ExternalCredit(ctx context.Context, req PostRequest) (*OperationResult, error)
ExternalDebit(ctx context.Context, req PostRequest) (*OperationResult, error)
Close() error
}
type grpcConnectorClient interface {
GetAccount(ctx context.Context, in *connectorv1.GetAccountRequest, opts ...grpc.CallOption) (*connectorv1.GetAccountResponse, error)
GetBalance(ctx context.Context, in *connectorv1.GetBalanceRequest, opts ...grpc.CallOption) (*connectorv1.GetBalanceResponse, error)
SubmitOperation(ctx context.Context, in *connectorv1.SubmitOperationRequest, opts ...grpc.CallOption) (*connectorv1.SubmitOperationResponse, error)
}
type connectorClient struct {
cfg Config
conn *grpc.ClientConn
client grpcConnectorClient
}
func New(cfg Config) (Client, error) {
cfg.Endpoint = strings.TrimSpace(cfg.Endpoint)
if cfg.Endpoint == "" {
return nil, merrors.InvalidArgument("ledger endpoint is required", "ledger.endpoint")
}
if normalized, insecure := normalizeEndpoint(cfg.Endpoint); normalized != "" {
cfg.Endpoint = normalized
if insecure {
cfg.Insecure = true
}
}
if cfg.Timeout <= 0 {
cfg.Timeout = 5 * time.Second
}
dialOpts := []grpc.DialOption{}
if cfg.Insecure {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
} else {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})))
}
conn, err := grpc.NewClient(cfg.Endpoint, dialOpts...)
if err != nil {
return nil, merrors.InternalWrap(err, fmt.Sprintf("ledger: dial %s", cfg.Endpoint))
}
return &connectorClient{
cfg: cfg,
conn: conn,
client: connectorv1.NewConnectorServiceClient(conn),
}, nil
}
func (c *connectorClient) Close() error {
if c == nil || c.conn == nil {
return nil
}
return c.conn.Close()
}
func (c *connectorClient) GetAccount(ctx context.Context, accountID string) (*Account, error) {
accountID = strings.TrimSpace(accountID)
if accountID == "" {
return nil, merrors.InvalidArgument("ledger account_id is required", "account_id")
}
ctx, cancel := c.callContext(ctx)
defer cancel()
resp, err := c.client.GetAccount(ctx, &connectorv1.GetAccountRequest{
AccountRef: &connectorv1.AccountRef{
ConnectorId: ledgerConnectorID,
AccountId: accountID,
},
})
if err != nil {
return nil, err
}
account := resp.GetAccount()
if account == nil {
return nil, merrors.NoData("ledger account not found")
}
accountCode := strings.TrimSpace(account.GetLabel())
organizationRef := strings.TrimSpace(account.GetOwnerRef())
if organizationRef == "" && account.GetProviderDetails() != nil {
details := account.GetProviderDetails().AsMap()
if organizationRef == "" {
organizationRef = firstDetailValue(details, "organization_ref", "organizationRef", "org_ref")
}
if accountCode == "" {
accountCode = firstDetailValue(details, "account_code", "accountCode", "code", "ledger_account_code")
}
}
return &Account{
AccountID: accountID,
AccountCode: accountCode,
Currency: strings.ToUpper(strings.TrimSpace(account.GetAsset())),
OrganizationRef: organizationRef,
}, nil
}
func (c *connectorClient) GetBalance(ctx context.Context, accountID string) (*Balance, error) {
accountID = strings.TrimSpace(accountID)
if accountID == "" {
return nil, merrors.InvalidArgument("ledger account_id is required", "account_id")
}
ctx, cancel := c.callContext(ctx)
defer cancel()
resp, err := c.client.GetBalance(ctx, &connectorv1.GetBalanceRequest{
AccountRef: &connectorv1.AccountRef{
ConnectorId: ledgerConnectorID,
AccountId: accountID,
},
})
if err != nil {
return nil, err
}
balance := resp.GetBalance()
if balance == nil || balance.GetAvailable() == nil {
return nil, merrors.Internal("ledger balance is unavailable")
}
return &Balance{
AccountID: accountID,
Amount: strings.TrimSpace(balance.GetAvailable().GetAmount()),
Currency: strings.ToUpper(strings.TrimSpace(balance.GetAvailable().GetCurrency())),
}, nil
}
func (c *connectorClient) ExternalCredit(ctx context.Context, req PostRequest) (*OperationResult, error) {
return c.submitExternalOperation(ctx, connectorv1.OperationType_CREDIT, discovery.OperationExternalCredit, req)
}
func (c *connectorClient) ExternalDebit(ctx context.Context, req PostRequest) (*OperationResult, error) {
return c.submitExternalOperation(ctx, connectorv1.OperationType_DEBIT, discovery.OperationExternalDebit, req)
}
func (c *connectorClient) submitExternalOperation(ctx context.Context, opType connectorv1.OperationType, operation string, req PostRequest) (*OperationResult, error) {
req.AccountID = strings.TrimSpace(req.AccountID)
req.OrganizationRef = strings.TrimSpace(req.OrganizationRef)
req.Amount = strings.TrimSpace(req.Amount)
req.Currency = strings.ToUpper(strings.TrimSpace(req.Currency))
req.Reference = strings.TrimSpace(req.Reference)
req.IdempotencyKey = strings.TrimSpace(req.IdempotencyKey)
if req.AccountID == "" {
return nil, merrors.InvalidArgument("ledger account_id is required", "account_id")
}
if req.OrganizationRef == "" {
return nil, merrors.InvalidArgument("ledger organization_ref is required", "organization_ref")
}
if req.Amount == "" || req.Currency == "" {
return nil, merrors.InvalidArgument("ledger amount is required", "amount")
}
if req.IdempotencyKey == "" {
return nil, merrors.InvalidArgument("ledger idempotency_key is required", "idempotency_key")
}
params := map[string]any{
"organization_ref": req.OrganizationRef,
"operation": operation,
"description": "chsettle treasury operation",
"metadata": map[string]any{
"reference": req.Reference,
},
}
operationReq := &connectorv1.Operation{
Type: opType,
IdempotencyKey: req.IdempotencyKey,
Money: &moneyv1.Money{
Amount: req.Amount,
Currency: req.Currency,
},
Params: structFromMap(params),
}
account := &connectorv1.AccountRef{ConnectorId: ledgerConnectorID, AccountId: req.AccountID}
switch opType {
case connectorv1.OperationType_CREDIT:
operationReq.To = &connectorv1.OperationParty{Ref: &connectorv1.OperationParty_Account{Account: account}}
case connectorv1.OperationType_DEBIT:
operationReq.From = &connectorv1.OperationParty{Ref: &connectorv1.OperationParty_Account{Account: account}}
}
ctx, cancel := c.callContext(ctx)
defer cancel()
resp, err := c.client.SubmitOperation(ctx, &connectorv1.SubmitOperationRequest{Operation: operationReq})
if err != nil {
return nil, err
}
if resp.GetReceipt() == nil {
return nil, merrors.Internal("ledger receipt is unavailable")
}
if receiptErr := resp.GetReceipt().GetError(); receiptErr != nil {
message := strings.TrimSpace(receiptErr.GetMessage())
if message == "" {
message = "ledger operation failed"
}
return nil, merrors.InvalidArgument(message)
}
reference := strings.TrimSpace(resp.GetReceipt().GetOperationId())
if reference == "" {
reference = req.Reference
}
return &OperationResult{Reference: reference}, nil
}
func (c *connectorClient) callContext(ctx context.Context) (context.Context, context.CancelFunc) {
if ctx == nil {
ctx = context.Background()
}
return context.WithTimeout(ctx, c.cfg.Timeout)
}
func structFromMap(values map[string]any) *structpb.Struct {
if len(values) == 0 {
return nil
}
result, err := structpb.NewStruct(values)
if err != nil {
return nil
}
return result
}
func normalizeEndpoint(raw string) (string, bool) {
raw = strings.TrimSpace(raw)
if raw == "" {
return "", false
}
parsed, err := url.Parse(raw)
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return raw, false
}
switch strings.ToLower(strings.TrimSpace(parsed.Scheme)) {
case "http", "grpc":
return parsed.Host, true
case "https", "grpcs":
return parsed.Host, false
default:
return raw, false
}
}
func firstDetailValue(values map[string]any, keys ...string) string {
if len(values) == 0 || len(keys) == 0 {
return ""
}
for _, key := range keys {
key = strings.TrimSpace(key)
if key == "" {
continue
}
if value, ok := values[key]; ok {
if text := strings.TrimSpace(fmt.Sprint(value)); text != "" {
return text
}
}
}
return ""
}

View File

@@ -0,0 +1,235 @@
package ledger
import (
"context"
"fmt"
"net"
"net/url"
"sort"
"strings"
"sync"
"time"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mservice"
"go.uber.org/zap"
)
type DiscoveryConfig struct {
Logger mlogger.Logger
Registry *discovery.Registry
Timeout time.Duration
}
type discoveryEndpoint struct {
address string
insecure bool
raw string
}
func (e discoveryEndpoint) key() string {
return fmt.Sprintf("%s|%t", e.address, e.insecure)
}
type discoveryClient struct {
logger mlogger.Logger
registry *discovery.Registry
timeout time.Duration
mu sync.Mutex
client Client
endpointKey string
}
func NewDiscoveryClient(cfg DiscoveryConfig) (Client, error) {
if cfg.Registry == nil {
return nil, merrors.InvalidArgument("treasury ledger discovery registry is required", "registry")
}
if cfg.Timeout <= 0 {
cfg.Timeout = 5 * time.Second
}
logger := cfg.Logger
if logger != nil {
logger = logger.Named("treasury_ledger_discovery")
}
return &discoveryClient{
logger: logger,
registry: cfg.Registry,
timeout: cfg.Timeout,
}, nil
}
func (c *discoveryClient) Close() error {
if c == nil {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
if c.client != nil {
err := c.client.Close()
c.client = nil
c.endpointKey = ""
return err
}
return nil
}
func (c *discoveryClient) GetAccount(ctx context.Context, accountID string) (*Account, error) {
client, err := c.resolveClient(ctx)
if err != nil {
return nil, err
}
return client.GetAccount(ctx, accountID)
}
func (c *discoveryClient) GetBalance(ctx context.Context, accountID string) (*Balance, error) {
client, err := c.resolveClient(ctx)
if err != nil {
return nil, err
}
return client.GetBalance(ctx, accountID)
}
func (c *discoveryClient) ExternalCredit(ctx context.Context, req PostRequest) (*OperationResult, error) {
client, err := c.resolveClient(ctx)
if err != nil {
return nil, err
}
return client.ExternalCredit(ctx, req)
}
func (c *discoveryClient) ExternalDebit(ctx context.Context, req PostRequest) (*OperationResult, error) {
client, err := c.resolveClient(ctx)
if err != nil {
return nil, err
}
return client.ExternalDebit(ctx, req)
}
func (c *discoveryClient) resolveClient(_ context.Context) (Client, error) {
if c == nil || c.registry == nil {
return nil, merrors.Internal("treasury ledger discovery is unavailable")
}
endpoint, err := c.resolveEndpoint()
if err != nil {
return nil, err
}
key := endpoint.key()
c.mu.Lock()
defer c.mu.Unlock()
if c.client != nil && c.endpointKey == key {
return c.client, nil
}
if c.client != nil {
_ = c.client.Close()
c.client = nil
c.endpointKey = ""
}
next, err := New(Config{
Endpoint: endpoint.address,
Timeout: c.timeout,
Insecure: endpoint.insecure,
})
if err != nil {
return nil, err
}
c.client = next
c.endpointKey = key
if c.logger != nil {
c.logger.Info("Discovered ledger endpoint selected",
zap.String("service", string(mservice.Ledger)),
zap.String("invoke_uri", endpoint.raw),
zap.String("address", endpoint.address),
zap.Bool("insecure", endpoint.insecure))
}
return c.client, nil
}
func (c *discoveryClient) resolveEndpoint() (discoveryEndpoint, error) {
entries := c.registry.List(time.Now(), true)
type match struct {
entry discovery.RegistryEntry
opMatch bool
}
matches := make([]match, 0, len(entries))
requiredOps := discovery.LedgerServiceOperations()
for _, entry := range entries {
if !matchesService(entry.Service, mservice.Ledger) {
continue
}
matches = append(matches, match{
entry: entry,
opMatch: discovery.HasAnyOperation(entry.Operations, requiredOps),
})
}
if len(matches) == 0 {
return discoveryEndpoint{}, merrors.NoData("discovery: ledger service unavailable")
}
sort.Slice(matches, func(i, j int) bool {
if matches[i].opMatch != matches[j].opMatch {
return matches[i].opMatch
}
if matches[i].entry.RoutingPriority != matches[j].entry.RoutingPriority {
return matches[i].entry.RoutingPriority > matches[j].entry.RoutingPriority
}
if matches[i].entry.ID != matches[j].entry.ID {
return matches[i].entry.ID < matches[j].entry.ID
}
return matches[i].entry.InstanceID < matches[j].entry.InstanceID
})
return parseDiscoveryEndpoint(matches[0].entry.InvokeURI)
}
func matchesService(service string, candidate mservice.Type) bool {
service = strings.TrimSpace(service)
if service == "" || strings.TrimSpace(string(candidate)) == "" {
return false
}
return strings.EqualFold(service, strings.TrimSpace(string(candidate)))
}
func parseDiscoveryEndpoint(raw string) (discoveryEndpoint, error) {
raw = strings.TrimSpace(raw)
if raw == "" {
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: invoke uri is required")
}
if !strings.Contains(raw, "://") {
if _, _, splitErr := net.SplitHostPort(raw); splitErr != nil {
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: invoke uri must include host:port")
}
return discoveryEndpoint{address: raw, insecure: true, raw: raw}, nil
}
parsed, err := url.Parse(raw)
if err != nil || parsed.Scheme == "" {
if err != nil {
return discoveryEndpoint{}, err
}
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: invoke uri must include host:port")
}
scheme := strings.ToLower(strings.TrimSpace(parsed.Scheme))
switch scheme {
case "grpc":
address := strings.TrimSpace(parsed.Host)
if _, _, splitErr := net.SplitHostPort(address); splitErr != nil {
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: invoke uri must include host:port")
}
return discoveryEndpoint{address: address, insecure: true, raw: raw}, nil
case "grpcs":
address := strings.TrimSpace(parsed.Host)
if _, _, splitErr := net.SplitHostPort(address); splitErr != nil {
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: invoke uri must include host:port")
}
return discoveryEndpoint{address: address, insecure: false, raw: raw}, nil
case "dns", "passthrough":
return discoveryEndpoint{address: raw, insecure: true, raw: raw}, nil
default:
return discoveryEndpoint{}, merrors.InvalidArgument("discovery: unsupported invoke uri scheme")
}
}

View File

@@ -0,0 +1,205 @@
package treasury
import (
"context"
"strings"
"time"
"github.com/tech/sendico/gateway/chsettle/internal/service/treasury/bot"
"github.com/tech/sendico/gateway/chsettle/internal/service/treasury/ledger"
"github.com/tech/sendico/gateway/chsettle/storage"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/model"
)
type Module struct {
logger mlogger.Logger
service *Service
router *bot.Router
scheduler *Scheduler
ledger ledger.Client
}
func NewModule(
logger mlogger.Logger,
repo storage.TreasuryRequestsStore,
users storage.TreasuryTelegramUsersStore,
ledgerClient ledger.Client,
cfg Config,
send bot.SendTextFunc,
) (*Module, error) {
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,
ledgerClient,
cfg.ExecutionDelay,
cfg.MaxAmountPerOperation,
cfg.MaxDailyAmount,
)
if err != nil {
return nil, err
}
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, &botUsersAdapter{store: users})
return module, nil
}
func (m *Module) Enabled() bool {
return m != nil && m.router != nil && m.router.Enabled() && m.scheduler != nil
}
func (m *Module) Start() {
if m == nil || m.scheduler == nil {
return
}
m.scheduler.Start()
}
func (m *Module) Shutdown() {
if m == nil {
return
}
if m.scheduler != nil {
m.scheduler.Shutdown()
}
if m.ledger != nil {
_ = m.ledger.Close()
}
}
func (m *Module) HandleUpdate(ctx context.Context, update *model.TelegramWebhookUpdate) bool {
if m == nil || m.router == nil {
return false
}
return m.router.HandleUpdate(ctx, update)
}
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
}
return a.svc.ExecutionDelay()
}
func (a *botServiceAdapter) MaxPerOperationLimit() string {
if a == nil || a.svc == nil {
return ""
}
return a.svc.MaxPerOperationLimit()
}
func (a *botServiceAdapter) GetActiveRequestForAccount(ctx context.Context, ledgerAccountID string) (*storagemodel.TreasuryRequest, error) {
if a == nil || a.svc == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return a.svc.GetActiveRequestForAccount(ctx, ledgerAccountID)
}
func (a *botServiceAdapter) GetAccountProfile(ctx context.Context, ledgerAccountID string) (*bot.AccountProfile, error) {
if a == nil || a.svc == nil {
return nil, merrors.Internal("treasury service unavailable")
}
profile, err := a.svc.GetAccountProfile(ctx, ledgerAccountID)
if err != nil {
return nil, err
}
if profile == nil {
return nil, nil
}
return &bot.AccountProfile{
AccountID: strings.TrimSpace(profile.AccountID),
AccountCode: strings.TrimSpace(profile.AccountCode),
Currency: strings.TrimSpace(profile.Currency),
}, nil
}
func (a *botServiceAdapter) CreateRequest(ctx context.Context, input bot.CreateRequestInput) (*storagemodel.TreasuryRequest, error) {
if a == nil || a.svc == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return a.svc.CreateRequest(ctx, CreateRequestInput{
OperationType: input.OperationType,
TelegramUserID: input.TelegramUserID,
LedgerAccountID: input.LedgerAccountID,
ChatID: input.ChatID,
Amount: input.Amount,
})
}
func (a *botServiceAdapter) ConfirmRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error) {
if a == nil || a.svc == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return a.svc.ConfirmRequest(ctx, requestID, telegramUserID)
}
func (a *botServiceAdapter) CancelRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error) {
if a == nil || a.svc == nil {
return nil, merrors.Internal("treasury service unavailable")
}
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

@@ -0,0 +1,327 @@
package treasury
import (
"context"
"strings"
"sync"
"time"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/mlogger"
"go.uber.org/zap"
)
type NotifyFunc func(ctx context.Context, chatID string, text string) error
type Scheduler struct {
logger mlogger.Logger
service *Service
notify NotifyFunc
safetySweepInterval time.Duration
cancel context.CancelFunc
wg sync.WaitGroup
timersMu sync.Mutex
timers map[string]*time.Timer
}
func NewScheduler(logger mlogger.Logger, service *Service, notify NotifyFunc, safetySweepInterval time.Duration) *Scheduler {
if logger != nil {
logger = logger.Named("treasury_scheduler")
}
if safetySweepInterval <= 0 {
safetySweepInterval = 30 * time.Second
}
return &Scheduler{
logger: logger,
service: service,
notify: notify,
safetySweepInterval: safetySweepInterval,
timers: map[string]*time.Timer{},
}
}
func (s *Scheduler) Start() {
if s == nil || s.service == nil || s.cancel != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
// Rebuild in-memory timers from DB on startup.
s.hydrateTimers(ctx)
// Safety pass for overdue items at startup.
s.sweep(ctx)
s.wg.Add(1)
go func() {
defer s.wg.Done()
ticker := time.NewTicker(s.safetySweepInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.sweep(ctx)
}
}
}()
}
func (s *Scheduler) Shutdown() {
if s == nil || s.cancel == nil {
return
}
s.cancel()
s.wg.Wait()
s.timersMu.Lock()
for requestID, timer := range s.timers {
if timer != nil {
timer.Stop()
}
delete(s.timers, requestID)
}
s.timersMu.Unlock()
}
func (s *Scheduler) TrackScheduled(record *storagemodel.TreasuryRequest) {
if s == nil || s.service == nil || record == nil {
return
}
if strings.TrimSpace(record.RequestID) == "" {
return
}
if record.Status != storagemodel.TreasuryRequestStatusScheduled {
return
}
requestID := strings.TrimSpace(record.RequestID)
when := record.ScheduledAt
if when.IsZero() {
when = time.Now()
}
delay := time.Until(when)
if delay <= 0 {
s.Untrack(requestID)
go s.executeAndNotifyByID(context.Background(), requestID)
return
}
s.timersMu.Lock()
if existing := s.timers[requestID]; existing != nil {
existing.Stop()
}
s.timers[requestID] = time.AfterFunc(delay, func() {
s.Untrack(requestID)
s.executeAndNotifyByID(context.Background(), requestID)
})
s.timersMu.Unlock()
}
func (s *Scheduler) Untrack(requestID string) {
if s == nil {
return
}
requestID = strings.TrimSpace(requestID)
if requestID == "" {
return
}
s.timersMu.Lock()
if timer := s.timers[requestID]; timer != nil {
timer.Stop()
}
delete(s.timers, requestID)
s.timersMu.Unlock()
}
func (s *Scheduler) hydrateTimers(ctx context.Context) {
if s == nil || s.service == nil {
return
}
scheduled, err := s.service.ScheduledRequests(ctx, 1000)
if err != nil {
s.logger.Warn("Failed to hydrate scheduled treasury requests", zap.Error(err))
return
}
for _, record := range scheduled {
s.TrackScheduled(&record)
}
}
func (s *Scheduler) sweep(ctx context.Context) {
if s == nil || s.service == nil {
return
}
now := time.Now()
confirmed, err := s.service.DueRequests(ctx, []storagemodel.TreasuryRequestStatus{
storagemodel.TreasuryRequestStatusConfirmed,
}, now, 100)
if err != nil {
s.logger.Warn("Failed to list confirmed treasury requests", zap.Error(err))
return
}
for _, request := range confirmed {
s.executeAndNotifyByID(ctx, strings.TrimSpace(request.RequestID))
}
scheduled, err := s.service.DueRequests(ctx, []storagemodel.TreasuryRequestStatus{
storagemodel.TreasuryRequestStatusScheduled,
}, now, 100)
if err != nil {
s.logger.Warn("Failed to list scheduled treasury requests", zap.Error(err))
return
}
for _, request := range scheduled {
s.Untrack(strings.TrimSpace(request.RequestID))
s.executeAndNotifyByID(ctx, strings.TrimSpace(request.RequestID))
}
}
func (s *Scheduler) executeAndNotifyByID(ctx context.Context, requestID string) {
if s == nil || s.service == nil {
return
}
requestID = strings.TrimSpace(requestID)
if requestID == "" {
return
}
runCtx := ctx
if runCtx == nil {
runCtx = context.Background()
}
withTimeout, cancel := context.WithTimeout(runCtx, 30*time.Second)
defer cancel()
result, err := s.service.ExecuteRequest(withTimeout, requestID)
if err != nil {
s.logger.Warn("Failed to execute treasury request", zap.Error(err), zap.String("request_id", requestID))
return
}
if result == nil || result.Request == nil {
s.logger.Debug("Treasury execution produced no result", zap.String("request_id", requestID))
return
}
if s.notify == nil {
s.logger.Warn("Treasury execution notifier is unavailable", zap.String("request_id", requestID))
return
}
text := executionMessage(result)
if strings.TrimSpace(text) == "" {
s.logger.Debug("Treasury execution result has no notification text",
zap.String("request_id", strings.TrimSpace(result.Request.RequestID)),
zap.String("status", strings.TrimSpace(string(result.Request.Status))))
return
}
chatID := strings.TrimSpace(result.Request.ChatID)
if chatID == "" {
s.logger.Warn("Treasury execution notification skipped: empty chat_id",
zap.String("request_id", strings.TrimSpace(result.Request.RequestID)))
return
}
s.logger.Info("Sending treasury execution notification",
zap.String("request_id", strings.TrimSpace(result.Request.RequestID)),
zap.String("chat_id", chatID),
zap.String("status", strings.TrimSpace(string(result.Request.Status))))
notifyCtx := context.Background()
if ctx != nil {
notifyCtx = ctx
}
notifyCtx, notifyCancel := context.WithTimeout(notifyCtx, 15*time.Second)
defer notifyCancel()
if err := s.notify(notifyCtx, chatID, text); err != nil {
s.logger.Warn("Failed to notify treasury execution result",
zap.Error(err),
zap.String("request_id", strings.TrimSpace(result.Request.RequestID)),
zap.String("chat_id", chatID),
zap.String("status", strings.TrimSpace(string(result.Request.Status))))
return
}
s.logger.Info("Treasury execution notification sent",
zap.String("request_id", strings.TrimSpace(result.Request.RequestID)),
zap.String("chat_id", chatID),
zap.String("status", strings.TrimSpace(string(result.Request.Status))))
}
func executionMessage(result *ExecutionResult) string {
if result == nil || result.Request == nil {
return ""
}
request := result.Request
switch request.Status {
case storagemodel.TreasuryRequestStatusExecuted:
op := "Funding"
sign := "+"
if request.OperationType == storagemodel.TreasuryOperationWithdraw {
op = "Withdrawal"
sign = "-"
}
balanceAmount := "unavailable"
balanceCurrency := strings.TrimSpace(request.Currency)
if result.NewBalance != nil {
if strings.TrimSpace(result.NewBalance.Amount) != "" {
balanceAmount = strings.TrimSpace(result.NewBalance.Amount)
}
if strings.TrimSpace(result.NewBalance.Currency) != "" {
balanceCurrency = strings.TrimSpace(result.NewBalance.Currency)
}
}
return "*" + op + " completed*\n\n" +
"*Account:* " + markdownCode(requestAccountCode(request)) + "\n" +
"*Amount:* " + markdownCode(sign+strings.TrimSpace(request.Amount)+" "+strings.TrimSpace(request.Currency)) + "\n" +
"*New balance:* " + markdownCode(balanceAmount+" "+balanceCurrency) + "\n\n" +
"*Reference:* " + markdownCode(strings.TrimSpace(request.RequestID))
case storagemodel.TreasuryRequestStatusFailed:
reason := strings.TrimSpace(request.ErrorMessage)
if reason == "" && result.ExecutionError != nil {
reason = strings.TrimSpace(result.ExecutionError.Error())
}
if reason == "" {
reason = "Unknown error."
}
return "*Execution failed*\n\n" +
"*Account:* " + markdownCode(requestAccountCode(request)) + "\n" +
"*Amount:* " + markdownCode(strings.TrimSpace(request.Amount)+" "+strings.TrimSpace(request.Currency)) + "\n" +
"*Status:* " + markdownCode("FAILED") + "\n" +
"*Reason:* " + markdownCode(compactForMarkdown(reason)) + "\n\n" +
"*Request ID:* " + markdownCode(strings.TrimSpace(request.RequestID))
default:
return ""
}
}
func requestAccountCode(request *storagemodel.TreasuryRequest) string {
if request == nil {
return ""
}
if code := strings.TrimSpace(request.LedgerAccountCode); code != "" {
return code
}
return strings.TrimSpace(request.LedgerAccountID)
}
func markdownCode(value string) string {
value = strings.TrimSpace(value)
if value == "" {
value = "N/A"
}
value = strings.ReplaceAll(value, "`", "'")
return "`" + value + "`"
}
func compactForMarkdown(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "Unknown error."
}
value = strings.ReplaceAll(value, "\r\n", " ")
value = strings.ReplaceAll(value, "\n", " ")
value = strings.ReplaceAll(value, "\r", " ")
return strings.Join(strings.Fields(value), " ")
}

View File

@@ -0,0 +1,457 @@
package treasury
import (
"context"
"errors"
"fmt"
"math/big"
"strings"
"time"
"github.com/tech/sendico/gateway/chsettle/internal/service/treasury/ledger"
"github.com/tech/sendico/gateway/chsettle/storage"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"go.mongodb.org/mongo-driver/v2/bson"
"go.uber.org/zap"
)
var ErrActiveTreasuryRequest = errors.New("active treasury request exists")
type CreateRequestInput struct {
OperationType storagemodel.TreasuryOperationType
TelegramUserID string
LedgerAccountID string
ChatID string
Amount string
}
type AccountProfile struct {
AccountID string
AccountCode string
Currency string
}
type ExecutionResult struct {
Request *storagemodel.TreasuryRequest
NewBalance *ledger.Balance
ExecutionError error
}
type Service struct {
logger mlogger.Logger
repo storage.TreasuryRequestsStore
ledger ledger.Client
validator *Validator
executionDelay time.Duration
}
func NewService(
logger mlogger.Logger,
repo storage.TreasuryRequestsStore,
ledgerClient ledger.Client,
executionDelay time.Duration,
maxPerOperation string,
maxDaily string,
) (*Service, error) {
if logger == nil {
return nil, merrors.InvalidArgument("logger is required", "logger")
}
if repo == nil {
return nil, merrors.InvalidArgument("treasury repository is required", "repo")
}
if ledgerClient == nil {
return nil, merrors.InvalidArgument("ledger client is required", "ledger_client")
}
if executionDelay <= 0 {
executionDelay = 30 * time.Second
}
validator, err := NewValidator(repo, maxPerOperation, maxDaily)
if err != nil {
return nil, err
}
return &Service{
logger: logger.Named("treasury_service"),
repo: repo,
ledger: ledgerClient,
validator: validator,
executionDelay: executionDelay,
}, nil
}
func (s *Service) ExecutionDelay() time.Duration {
if s == nil {
return 0
}
return s.executionDelay
}
func (s *Service) MaxPerOperationLimit() string {
if s == nil || s.validator == nil {
return ""
}
return s.validator.MaxPerOperation()
}
func (s *Service) GetActiveRequestForAccount(ctx context.Context, ledgerAccountID string) (*storagemodel.TreasuryRequest, error) {
if s == nil || s.repo == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return s.repo.FindActiveByLedgerAccountID(ctx, ledgerAccountID)
}
func (s *Service) GetRequest(ctx context.Context, requestID string) (*storagemodel.TreasuryRequest, error) {
if s == nil || s.repo == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return s.repo.FindByRequestID(ctx, requestID)
}
func (s *Service) GetAccountProfile(ctx context.Context, ledgerAccountID string) (*AccountProfile, error) {
if s == nil || s.ledger == nil {
return nil, merrors.Internal("treasury service unavailable")
}
ledgerAccountID = strings.TrimSpace(ledgerAccountID)
if ledgerAccountID == "" {
return nil, merrors.InvalidArgument("ledger_account_id is required", "ledger_account_id")
}
account, err := s.ledger.GetAccount(ctx, ledgerAccountID)
if err != nil {
return nil, err
}
if account == nil {
return nil, merrors.NoData("ledger account not found")
}
return &AccountProfile{
AccountID: ledgerAccountID,
AccountCode: resolveAccountCode(account, ledgerAccountID),
Currency: strings.ToUpper(strings.TrimSpace(account.Currency)),
}, nil
}
func (s *Service) CreateRequest(ctx context.Context, input CreateRequestInput) (*storagemodel.TreasuryRequest, error) {
if s == nil || s.repo == nil || s.ledger == nil || s.validator == nil {
return nil, merrors.Internal("treasury service unavailable")
}
input.TelegramUserID = strings.TrimSpace(input.TelegramUserID)
input.LedgerAccountID = strings.TrimSpace(input.LedgerAccountID)
input.ChatID = strings.TrimSpace(input.ChatID)
input.Amount = strings.TrimSpace(input.Amount)
switch input.OperationType {
case storagemodel.TreasuryOperationFund, storagemodel.TreasuryOperationWithdraw:
default:
return nil, merrors.InvalidArgument("treasury operation is invalid", "operation_type")
}
if input.TelegramUserID == "" {
return nil, merrors.InvalidArgument("telegram_user_id is required", "telegram_user_id")
}
if input.LedgerAccountID == "" {
return nil, merrors.InvalidArgument("ledger_account_id is required", "ledger_account_id")
}
if input.ChatID == "" {
return nil, merrors.InvalidArgument("chat_id is required", "chat_id")
}
active, err := s.repo.FindActiveByLedgerAccountID(ctx, input.LedgerAccountID)
if err != nil {
return nil, err
}
if active != nil {
return active, ErrActiveTreasuryRequest
}
amountRat, normalizedAmount, err := s.validator.ValidateAmount(input.Amount)
if err != nil {
return nil, err
}
if err := s.validator.ValidateDailyLimit(ctx, input.LedgerAccountID, amountRat, time.Now()); err != nil {
return nil, err
}
account, err := s.ledger.GetAccount(ctx, input.LedgerAccountID)
if err != nil {
return nil, err
}
if account == nil || strings.TrimSpace(account.Currency) == "" {
return nil, merrors.Internal("ledger account currency is unavailable")
}
if strings.TrimSpace(account.OrganizationRef) == "" {
return nil, merrors.Internal("ledger account organization is unavailable")
}
requestID := newRequestID()
record := &storagemodel.TreasuryRequest{
RequestID: requestID,
OperationType: input.OperationType,
TelegramUserID: input.TelegramUserID,
LedgerAccountID: input.LedgerAccountID,
LedgerAccountCode: resolveAccountCode(account, input.LedgerAccountID),
OrganizationRef: account.OrganizationRef,
ChatID: input.ChatID,
Amount: normalizedAmount,
Currency: strings.ToUpper(strings.TrimSpace(account.Currency)),
Status: storagemodel.TreasuryRequestStatusCreated,
IdempotencyKey: fmt.Sprintf("chsettle:%s", requestID),
Active: true,
}
if err := s.repo.Create(ctx, record); err != nil {
if errors.Is(err, storage.ErrDuplicate) {
active, fetchErr := s.repo.FindActiveByLedgerAccountID(ctx, input.LedgerAccountID)
if fetchErr != nil {
return nil, fetchErr
}
if active != nil {
return active, ErrActiveTreasuryRequest
}
return nil, err
}
return nil, err
}
s.logRequest(record, "created", nil)
return record, nil
}
func (s *Service) ConfirmRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error) {
requestID = strings.TrimSpace(requestID)
telegramUserID = strings.TrimSpace(telegramUserID)
if requestID == "" {
return nil, merrors.InvalidArgument("request_id is required", "request_id")
}
record, err := s.repo.FindByRequestID(ctx, requestID)
if err != nil {
return nil, err
}
if record == nil {
return nil, merrors.NoData("treasury request not found")
}
if telegramUserID != "" && record.TelegramUserID != telegramUserID {
return nil, merrors.Unauthorized("treasury request ownership mismatch")
}
switch record.Status {
case storagemodel.TreasuryRequestStatusScheduled:
return record, nil
case storagemodel.TreasuryRequestStatusCreated, storagemodel.TreasuryRequestStatusConfirmed:
now := time.Now()
record.ConfirmedAt = now
record.ScheduledAt = now.Add(s.executionDelay)
record.Status = storagemodel.TreasuryRequestStatusScheduled
record.Active = true
record.ErrorMessage = ""
default:
return nil, merrors.InvalidArgument("treasury request cannot be confirmed in current status", "status")
}
if err := s.repo.Update(ctx, record); err != nil {
return nil, err
}
s.logRequest(record, "scheduled", nil)
return record, nil
}
func (s *Service) CancelRequest(ctx context.Context, requestID string, telegramUserID string) (*storagemodel.TreasuryRequest, error) {
requestID = strings.TrimSpace(requestID)
telegramUserID = strings.TrimSpace(telegramUserID)
if requestID == "" {
return nil, merrors.InvalidArgument("request_id is required", "request_id")
}
record, err := s.repo.FindByRequestID(ctx, requestID)
if err != nil {
return nil, err
}
if record == nil {
return nil, merrors.NoData("treasury request not found")
}
if telegramUserID != "" && record.TelegramUserID != telegramUserID {
return nil, merrors.Unauthorized("treasury request ownership mismatch")
}
switch record.Status {
case storagemodel.TreasuryRequestStatusCancelled:
return record, nil
case storagemodel.TreasuryRequestStatusCreated, storagemodel.TreasuryRequestStatusConfirmed, storagemodel.TreasuryRequestStatusScheduled:
record.Status = storagemodel.TreasuryRequestStatusCancelled
record.CancelledAt = time.Now()
record.Active = false
default:
return nil, merrors.InvalidArgument("treasury request cannot be cancelled in current status", "status")
}
if err := s.repo.Update(ctx, record); err != nil {
return nil, err
}
s.logRequest(record, "cancelled", nil)
return record, nil
}
func (s *Service) ExecuteRequest(ctx context.Context, requestID string) (*ExecutionResult, error) {
requestID = strings.TrimSpace(requestID)
if requestID == "" {
return nil, merrors.InvalidArgument("request_id is required", "request_id")
}
record, err := s.repo.FindByRequestID(ctx, requestID)
if err != nil {
return nil, err
}
if record == nil {
return nil, nil
}
switch record.Status {
case storagemodel.TreasuryRequestStatusExecuted,
storagemodel.TreasuryRequestStatusCancelled,
storagemodel.TreasuryRequestStatusFailed:
return nil, nil
case storagemodel.TreasuryRequestStatusScheduled:
claimed, err := s.repo.ClaimScheduled(ctx, requestID)
if err != nil {
return nil, err
}
if !claimed {
return nil, nil
}
record, err = s.repo.FindByRequestID(ctx, requestID)
if err != nil {
return nil, err
}
if record == nil {
return nil, nil
}
}
if record.Status != storagemodel.TreasuryRequestStatusConfirmed {
return nil, nil
}
return s.executeClaimed(ctx, record)
}
func (s *Service) executeClaimed(ctx context.Context, record *storagemodel.TreasuryRequest) (*ExecutionResult, error) {
if record == nil {
return nil, merrors.InvalidArgument("treasury request is required", "request")
}
postReq := ledger.PostRequest{
AccountID: record.LedgerAccountID,
OrganizationRef: record.OrganizationRef,
Amount: record.Amount,
Currency: record.Currency,
Reference: record.RequestID,
IdempotencyKey: record.IdempotencyKey,
}
var (
opResult *ledger.OperationResult
err error
)
switch record.OperationType {
case storagemodel.TreasuryOperationFund:
opResult, err = s.ledger.ExternalCredit(ctx, postReq)
case storagemodel.TreasuryOperationWithdraw:
opResult, err = s.ledger.ExternalDebit(ctx, postReq)
default:
err = merrors.InvalidArgument("treasury operation is invalid", "operation_type")
}
now := time.Now()
if err != nil {
record.Status = storagemodel.TreasuryRequestStatusFailed
record.Active = false
record.ExecutedAt = now
record.ErrorMessage = strings.TrimSpace(err.Error())
if saveErr := s.repo.Update(ctx, record); saveErr != nil {
return nil, saveErr
}
s.logRequest(record, "failed", err)
return &ExecutionResult{
Request: record,
ExecutionError: err,
}, nil
}
if opResult != nil {
record.LedgerReference = strings.TrimSpace(opResult.Reference)
}
record.Status = storagemodel.TreasuryRequestStatusExecuted
record.Active = false
record.ExecutedAt = now
record.ErrorMessage = ""
balance, balanceErr := s.ledger.GetBalance(ctx, record.LedgerAccountID)
if balanceErr != nil {
record.ErrorMessage = strings.TrimSpace(balanceErr.Error())
}
if saveErr := s.repo.Update(ctx, record); saveErr != nil {
return nil, saveErr
}
s.logRequest(record, "executed", nil)
return &ExecutionResult{
Request: record,
NewBalance: balance,
ExecutionError: balanceErr,
}, nil
}
func (s *Service) DueRequests(ctx context.Context, statuses []storagemodel.TreasuryRequestStatus, now time.Time, limit int64) ([]storagemodel.TreasuryRequest, error) {
if s == nil || s.repo == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return s.repo.FindDueByStatus(ctx, statuses, now, limit)
}
func (s *Service) ScheduledRequests(ctx context.Context, limit int64) ([]storagemodel.TreasuryRequest, error) {
if s == nil || s.repo == nil {
return nil, merrors.Internal("treasury service unavailable")
}
return s.repo.FindDueByStatus(
ctx,
[]storagemodel.TreasuryRequestStatus{storagemodel.TreasuryRequestStatusScheduled},
time.Now().Add(10*365*24*time.Hour),
limit,
)
}
func (s *Service) ParseAmount(value string) (*big.Rat, error) {
return parseAmountRat(value)
}
func (s *Service) logRequest(record *storagemodel.TreasuryRequest, status string, err error) {
if s == nil || s.logger == nil || record == nil {
return
}
fields := []zap.Field{
zap.String("request_id", strings.TrimSpace(record.RequestID)),
zap.String("telegram_user_id", strings.TrimSpace(record.TelegramUserID)),
zap.String("ledger_account_id", strings.TrimSpace(record.LedgerAccountID)),
zap.String("ledger_account_code", strings.TrimSpace(record.LedgerAccountCode)),
zap.String("chat_id", strings.TrimSpace(record.ChatID)),
zap.String("operation_type", strings.TrimSpace(string(record.OperationType))),
zap.String("amount", strings.TrimSpace(record.Amount)),
zap.String("currency", strings.TrimSpace(record.Currency)),
zap.String("status", status),
zap.String("ledger_reference", strings.TrimSpace(record.LedgerReference)),
zap.String("error_message", strings.TrimSpace(record.ErrorMessage)),
}
if err != nil {
fields = append(fields, zap.Error(err))
}
s.logger.Info("treasury_request", fields...)
}
func newRequestID() string {
return "TG-TREASURY-" + strings.ToUpper(bson.NewObjectID().Hex())
}
func resolveAccountCode(account *ledger.Account, fallbackAccountID string) string {
if account != nil {
if code := strings.TrimSpace(account.AccountCode); code != "" {
return code
}
if code := strings.TrimSpace(account.AccountID); code != "" {
return code
}
}
return strings.TrimSpace(fallbackAccountID)
}

View File

@@ -0,0 +1,178 @@
package treasury
import (
"context"
"math/big"
"regexp"
"strings"
"time"
"github.com/tech/sendico/gateway/chsettle/storage"
storagemodel "github.com/tech/sendico/gateway/chsettle/storage/model"
"github.com/tech/sendico/pkg/merrors"
)
var treasuryAmountPattern = regexp.MustCompile(`^[0-9]+(\.[0-9]+)?$`)
type LimitKind string
const (
LimitKindPerOperation LimitKind = "per_operation"
LimitKindDaily LimitKind = "daily"
)
type LimitError struct {
Kind LimitKind
Max string
}
func (e *LimitError) Error() string {
if e == nil {
return "limit exceeded"
}
switch e.Kind {
case LimitKindPerOperation:
return "max amount per operation exceeded"
case LimitKindDaily:
return "max daily amount exceeded"
default:
return "limit exceeded"
}
}
func (e *LimitError) LimitKind() string {
if e == nil {
return ""
}
return string(e.Kind)
}
func (e *LimitError) LimitMax() string {
if e == nil {
return ""
}
return e.Max
}
type Validator struct {
repo storage.TreasuryRequestsStore
maxPerOperation *big.Rat
maxDaily *big.Rat
maxPerOperationRaw string
maxDailyRaw string
}
func NewValidator(repo storage.TreasuryRequestsStore, maxPerOperation string, maxDaily string) (*Validator, error) {
validator := &Validator{
repo: repo,
maxPerOperationRaw: strings.TrimSpace(maxPerOperation),
maxDailyRaw: strings.TrimSpace(maxDaily),
}
if validator.maxPerOperationRaw != "" {
value, err := parseAmountRat(validator.maxPerOperationRaw)
if err != nil {
return nil, merrors.InvalidArgument("treasury max_amount_per_operation is invalid", "treasury.limits.max_amount_per_operation")
}
validator.maxPerOperation = value
}
if validator.maxDailyRaw != "" {
value, err := parseAmountRat(validator.maxDailyRaw)
if err != nil {
return nil, merrors.InvalidArgument("treasury max_daily_amount is invalid", "treasury.limits.max_daily_amount")
}
validator.maxDaily = value
}
return validator, nil
}
func (v *Validator) MaxPerOperation() string {
if v == nil {
return ""
}
return v.maxPerOperationRaw
}
func (v *Validator) MaxDaily() string {
if v == nil {
return ""
}
return v.maxDailyRaw
}
func (v *Validator) ValidateAmount(amount string) (*big.Rat, string, error) {
amount = strings.TrimSpace(amount)
value, err := parseAmountRat(amount)
if err != nil {
return nil, "", err
}
if v != nil && v.maxPerOperation != nil && value.Cmp(v.maxPerOperation) > 0 {
return nil, "", &LimitError{
Kind: LimitKindPerOperation,
Max: v.maxPerOperationRaw,
}
}
return value, amount, nil
}
func (v *Validator) ValidateDailyLimit(ctx context.Context, ledgerAccountID string, amount *big.Rat, now time.Time) error {
if v == nil || v.maxDaily == nil || v.repo == nil {
return nil
}
if amount == nil {
return merrors.InvalidArgument("amount is required", "amount")
}
dayStart := time.Date(now.UTC().Year(), now.UTC().Month(), now.UTC().Day(), 0, 0, 0, 0, time.UTC)
dayEnd := dayStart.Add(24 * time.Hour)
records, err := v.repo.ListByAccountAndStatuses(
ctx,
ledgerAccountID,
[]storagemodel.TreasuryRequestStatus{
storagemodel.TreasuryRequestStatusCreated,
storagemodel.TreasuryRequestStatusConfirmed,
storagemodel.TreasuryRequestStatusScheduled,
storagemodel.TreasuryRequestStatusExecuted,
},
dayStart,
dayEnd,
)
if err != nil {
return err
}
total := new(big.Rat)
for _, record := range records {
next, err := parseAmountRat(record.Amount)
if err != nil {
return merrors.Internal("treasury request amount is invalid")
}
total.Add(total, next)
}
total.Add(total, amount)
if total.Cmp(v.maxDaily) > 0 {
return &LimitError{
Kind: LimitKindDaily,
Max: v.maxDailyRaw,
}
}
return nil
}
func parseAmountRat(value string) (*big.Rat, error) {
value = strings.TrimSpace(value)
if value == "" {
return nil, merrors.InvalidArgument("amount is required", "amount")
}
if !treasuryAmountPattern.MatchString(value) {
return nil, merrors.InvalidArgument("amount format is invalid", "amount")
}
amount := new(big.Rat)
if _, ok := amount.SetString(value); !ok {
return nil, merrors.InvalidArgument("amount format is invalid", "amount")
}
if amount.Sign() <= 0 {
return nil, merrors.InvalidArgument("amount must be positive", "amount")
}
return amount, nil
}