Set up default currency accounts automatically (#72)

This commit is contained in:
Stephan D committed 2026-10-23 12:43:55 +02:00
1 parent 62df9eac5f
commit 5c255296d3
21 files changed
+675 -62

No files matched your search

+1
View File
@@ -75,6 +75,7 @@ func OpenDependencies(_ context.Context, logger mlogger.Logger, config *api.Conf
return nil, nil, err
}
deps.Payments = &payments.Service{Store: paymentStore, Provider: provider, Identity: deps.Identity, Logger: logger.Named("payments")}
deps.Auth.Wallet = deps.Payments
logger.Info("Sendico sign-up facts integration enabled")
}
records, err := factorstore.Open(logger, db)
+15
View File
@@ -31,8 +31,13 @@ type SignupFacts interface {
Ready(context.Context, bson.ObjectID) error
}
type CurrencyAccounts interface {
Ensure(context.Context, *customer.Account) error
}
type Service struct {
Facts SignupFacts
Wallet CurrencyAccounts
Accounts account.DB
Customers accountservice.AccountService
Tokens verification.DB
@@ -110,6 +115,11 @@ func (s *Service) Signup(ctx context.Context, a *customer.Account, ip, key strin
return nil, err
}
}
if s.Wallet != nil {
if err := s.Wallet.Ensure(ctx, a); err != nil {
return nil, err
}
}
if raw != "" {
u, _ := url.Parse(s.ActivationURL)
q := u.Query()
@@ -148,6 +158,11 @@ func (s *Service) ResendActivation(ctx context.Context, email, ip, key string) e
return err
}
}
if s.Wallet != nil {
if err := s.Wallet.Ensure(ctx, a); err != nil {
return err
}
}
_, err = s.sendActivation(ctx, a, key)
return err
}
@@ -75,6 +75,17 @@ func (m *capturedMail) Send(_ context.Context, r notifications.Message) error {
}
return nil
}
type signupWallet struct {
customers []bson.ObjectID
err error
}
func (s *signupWallet) Ensure(_ context.Context, a *customer.Account) error {
s.customers = append(s.customers, a.ID)
return s.err
}
func TestSignupActivationAndLoginEmailFlow(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
@@ -102,6 +113,8 @@ func TestSignupActivationAndLoginEmailFlow(t *testing.T) {
require.NoError(t, err)
provider := &signupProvider{fail: true}
service.Facts = &identity.Service{Store: bindings, Provider: provider, Logger: zap.NewNop()}
wallet := &signupWallet{err: merrors.ServiceUnavailable("account setup temporarily unavailable")}
service.Wallet = wallet
// Keep the real verification/customer stores and capture only session writes.
sessions := &testRefreshTokenDB{client: &session.Client{ClientID: "web"}}
records, err := factorstore.Open(zap.NewNop(), factory)
@@ -133,6 +146,14 @@ func TestSignupActivationAndLoginEmailFlow(t *testing.T) {
signup["name"] = "Changed retry name"
provider.fail = false
w = post("signup", signup)
require.Equal(t, 503, w.Code, w.Body.String())
require.Empty(t, mail.messages)
require.Len(t, wallet.customers, 1)
w = post("verification/send", map[string]string{"email": "user@example.com", "idempotencyKey": "wallet-retry"})
require.Equal(t, 503, w.Code, w.Body.String())
require.Empty(t, mail.messages)
wallet.err = nil
w = post("signup", signup)
require.Equal(t, 202, w.Code, w.Body.String())
require.Equal(t, 1, provider.registrations)
require.Equal(t, "Customer", provider.last.FullName)
@@ -149,6 +170,7 @@ func TestSignupActivationAndLoginEmailFlow(t *testing.T) {
require.NoError(t, err)
require.False(t, a.IsActive())
require.Equal(t, a.ID.Hex(), provider.last.CustomerRef)
require.Equal(t, []bson.ObjectID{a.ID, a.ID, a.ID}, wallet.customers)
require.NoError(t, service.Facts.Ready(ctx, a.ID))
login := map[string]any{"login": map[string]string{"login": "user@example.com", "password": " correct password "}, "clientId": "web", "deviceId": "browser"}
require.Equal(t, 403, post("login", login).Code)
+50
View File
@@ -0,0 +1,50 @@
package payments
import (
"context"
"time"
customer "git.sendico.io/tech/sxchnge/product/customers/model"
merrors "git.sendico.io/tech/sxchnge/shared/errors"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"go.uber.org/zap"
)
var accountSetup = promauto.NewCounterVec(prometheus.CounterOpts{Name: "sxchnge_currency_accounts_setup_total", Help: "Default currency account setup attempts, including checks of existing accounts."}, []string{"outcome"})
var accountSetupDuration = promauto.NewHistogram(prometheus.HistogramOpts{Name: "sxchnge_currency_accounts_setup_duration_seconds", Help: "Time to confirm all default currency accounts exist."})
func (s *Service) Ensure(ctx context.Context, a *customer.Account) error {
_, err := s.ensureAccounts(ctx, a)
if err != nil {
return merrors.ServiceUnavailableWrap(err, "currency account setup unavailable")
}
return nil
}
func (s *Service) ensureAccounts(ctx context.Context, a *customer.Account) (subject string, result error) {
started := time.Now()
defer func() {
outcome := "ready"
if result != nil {
outcome = "failed"
}
accountSetup.WithLabelValues(outcome).Inc()
accountSetupDuration.Observe(time.Since(started).Seconds())
}()
if s == nil || s.Identity == nil || s.Provider == nil {
return "", Unavailable
}
if a == nil || a.ID.IsZero() {
return "", NotFound
}
subject, result = s.Identity.Subject(ctx, a)
if result != nil {
return "", result
}
if result = s.Provider.EnsureAccounts(ctx, subject); result != nil {
s.Logger.Warn("Currency account setup incomplete", zap.String("account_ref", a.ID.Hex()), zap.Error(result))
return "", result
}
return subject, nil
}
+28
View File
@@ -0,0 +1,28 @@
package payments
import (
"testing"
customer "git.sendico.io/tech/sxchnge/product/customers/model"
"github.com/stretchr/testify/require"
"go.mongodb.org/mongo-driver/v2/bson"
"go.uber.org/zap"
)
func TestSignupAndWalletReadEnsureCustomerCurrencyAccounts(t *testing.T) {
a := &customer.Account{}
a.ID = bson.NewObjectID()
p := &provider{namespace: "provider"}
s := &Service{Provider: p, Identity: identity{}, Logger: zap.NewNop(), Factors: &factors{}, Store: &memoryStore{records: map[bson.ObjectID]Record{}}}
require.NoError(t, s.Ensure(t.Context(), a))
require.Equal(t, "subject", p.setupSubject)
snapshot, err := s.Fetch(t.Context(), a)
require.NoError(t, err)
require.Len(t, snapshot.Accounts, 1)
require.Equal(t, 2, p.setupCalls)
p.setupErr = Unavailable
_, err = s.Fetch(t.Context(), a)
require.ErrorIs(t, err, Unavailable)
require.Error(t, s.Ensure(t.Context(), a))
require.Error(t, s.Ensure(t.Context(), nil))
}
+1
View File
@@ -53,6 +53,7 @@ func (*Record) Collection() string { return "payment_quotes" }
type Provider interface {
Namespace() string
EnsureAccounts(context.Context, string) error
Accounts(context.Context, string) ([]Account, error)
Quote(context.Context, string, Request, []Account) (Quote, error)
Execute(context.Context, string, string) (RemoteOperation, error)
+1 -1
View File
@@ -206,7 +206,7 @@ func (s *Service) Fetch(ctx context.Context, a *customer.Account) (Snapshot, err
if err := s.ready(); err != nil {
return Snapshot{}, err
}
subject, err := s.Identity.Subject(ctx, a)
subject, err := s.ensureAccounts(ctx, a)
if err != nil {
return Snapshot{}, err
}
+16 -6
View File
@@ -76,15 +76,25 @@ func (s *memoryStore) Attach(_ context.Context, r *Record, id string) error {
}
type provider struct {
mu sync.Mutex
accepted int
lost bool
status string
key string
namespace string
mu sync.Mutex
accepted int
lost bool
status string
key string
namespace string
setupCalls int
setupSubject string
setupErr error
}
func (p *provider) Namespace() string { return p.namespace }
func (p *provider) EnsureAccounts(_ context.Context, subject string) error {
p.mu.Lock()
defer p.mu.Unlock()
p.setupCalls++
p.setupSubject = subject
return p.setupErr
}
func (p *provider) Accounts(context.Context, string) ([]Account, error) {
return []Account{{ID: "ledger", Currency: "EUR", Precision: 2, Available: 5000}}, nil
}
@@ -0,0 +1,87 @@
package sendico
import (
"context"
"net/http"
"git.sendico.io/tech/sxchnge/product/payments"
merrors "git.sendico.io/tech/sxchnge/shared/errors"
)
func (c *Client) validCustomerAccount(a accountDTO, subject string) bool {
return validRef(a.ID) && a.ScopeRef == c.config.OrganizationID && a.Owner == "" &&
a.Type == "ACCOUNT_TYPE_ASSET" && a.LedgerRole == "ACCOUNT_ROLE_OPERATING" &&
a.Status == "ACCOUNT_STATUS_ACTIVE" && !a.AllowNegative && a.Metadata["sxchnge_subject_ref"] == subject &&
(a.Metadata["external_subject_ref"] == "" || a.Metadata["external_subject_ref"] == subject)
}
// EnsureAccounts also repairs older customers, using provider-enforced
// organization/subject/currency uniqueness for safe retries after lost responses.
func (c *Client) EnsureAccounts(ctx context.Context, subject string) error {
if !validRef(subject) {
return payments.Unavailable
}
root := "/ledger_accounts/" + c.config.OrganizationID
var data accountListDTO
if err := c.paymentRequest(ctx, http.MethodGet, root, nil, &data); err != nil {
return err
}
existing := map[string]bool{}
reconcile := map[string]bool{}
for _, a := range data.Accounts {
if a.Metadata["sxchnge_subject_ref"] != subject && a.Metadata["external_subject_ref"] != subject {
continue
}
if !c.validCustomerAccount(a, subject) || existing[a.Currency] {
return payments.Unavailable
}
existing[a.Currency] = true
reconcile[a.Currency] = a.Metadata["external_subject_ref"] == subject
}
missing := []string{}
for _, code := range []string{"RUB", "USDT", "USDC"} {
if !existing[code] || reconcile[code] {
missing = append(missing, code)
}
}
if len(missing) == 0 {
return nil
}
// Older providers can accept the metadata but cannot prevent duplicates.
// Require the explicit capability before issuing the first create.
if !data.ExternalSubjectAccounts {
return payments.Unavailable
}
var catalog struct {
Rows []currencyDTO `json:"currencies"`
}
if err := c.paymentRequest(ctx, http.MethodGet, "/payments/fx/currencies/"+c.config.OrganizationID, nil, &catalog); err != nil {
return err
}
for _, code := range missing {
found := false
for _, currency := range catalog.Rows {
if currency.Code == code && currency.Scale != nil && *currency.Scale >= 0 && *currency.Scale <= 8 {
found = true
}
}
if !found {
return merrors.OpaqueWrap(payments.Unavailable, payments.Unsupported, "default currency missing from provider catalog")
}
}
// Replaying a bound create also resumes Sendico's durable wallet/route setup
// if a previous response failed after the ledger itself had been committed.
for _, code := range missing {
var response struct {
Account accountDTO `json:"account"`
}
request := createAccountDTO{Currency: code, Type: "asset", LedgerRole: "operating", Metadata: map[string]string{"external_subject_ref": subject, "sxchnge_subject_ref": subject}}
if err := c.paymentRequest(ctx, http.MethodPost, root, request, &response, http.StatusCreated); err != nil {
return err
}
if !c.validCustomerAccount(response.Account, subject) || response.Account.Currency != code || response.Account.Metadata["external_subject_ref"] != subject {
return payments.Unavailable
}
}
return nil
}
@@ -0,0 +1,200 @@
package sendico
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"github.com/stretchr/testify/require"
"go.mongodb.org/mongo-driver/v2/bson"
)
type accountServer struct {
mu sync.Mutex
rows []accountDTO
posts []string
capable bool
missingCurrency bool
failCurrency string
loseResponse bool
wrongResponse bool
}
func customerLedger(code string) accountDTO {
return accountDTO{ID: bson.NewObjectID().Hex(), ScopeRef: org, Currency: code, Type: "ACCOUNT_TYPE_ASSET", LedgerRole: "ACCOUNT_ROLE_OPERATING", Status: "ACCOUNT_STATUS_ACTIVE", Metadata: map[string]string{"sxchnge_subject_ref": subject, "external_subject_ref": subject}}
}
func (s *accountServer) serve(t *testing.T) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
s.mu.Lock()
defer s.mu.Unlock()
require.NotEmpty(t, r.Header.Get("Authorization"))
var data any
switch {
case strings.Contains(r.URL.Path, "/currencies/"):
rows := []map[string]any{{"code": "RUB", "accountingScale": 2}, {"code": "USDT", "accountingScale": 6}, {"code": "USDC", "accountingScale": 6}}
if s.missingCurrency {
rows = rows[:2]
}
data = map[string]any{"currencies": rows}
case r.Method == http.MethodPost:
var request createAccountDTO
require.NoError(t, json.NewDecoder(r.Body).Decode(&request))
require.Equal(t, "asset", request.Type)
require.Equal(t, "operating", request.LedgerRole)
require.False(t, request.AllowNegative)
require.Equal(t, subject, request.Metadata["external_subject_ref"])
require.Equal(t, subject, request.Metadata["sxchnge_subject_ref"])
s.posts = append(s.posts, request.Currency)
if s.failCurrency == request.Currency {
s.failCurrency = ""
w.WriteHeader(503)
return
}
var row accountDTO
for _, a := range s.rows {
if a.Currency == request.Currency && a.Metadata["external_subject_ref"] == subject {
row = a
}
}
if row.ID == "" {
row = customerLedger(request.Currency)
s.rows = append(s.rows, row)
}
if s.loseResponse {
s.loseResponse = false
w.WriteHeader(503)
return
}
if s.wrongResponse {
row.Currency = "EUR"
}
w.WriteHeader(http.StatusCreated)
data = map[string]any{"account": row}
case strings.HasSuffix(r.URL.Path, "/balance"):
parts := strings.Split(r.URL.Path, "/")
data = map[string]any{"balance": balanceDTO{ID: parts[len(parts)-2], Money: moneyDTO{Amount: "45.00", Currency: "RUB"}}}
default:
data = map[string]any{"accounts": s.rows, "externalSubjectAccounts": s.capable}
}
require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"status": "success", "data": data}))
}
}
func TestEnsureCurrencyAccountsPreservesExistingAndRecoversPartialFailure(t *testing.T) {
for _, lost := range []bool{false, true} {
t.Run(map[bool]string{false: "partial failure", true: "lost response"}[lost], func(t *testing.T) {
s := &accountServer{capable: true, failCurrency: "USDT", loseResponse: lost}
if lost {
s.failCurrency = ""
}
server := httptest.NewServer(s.serve(t))
defer server.Close()
c, err := Open(config(t, server.URL))
require.NoError(t, err)
require.Error(t, c.EnsureAccounts(t.Context(), subject))
require.Len(t, s.rows, 1)
firstID := s.rows[0].ID
require.NoError(t, c.EnsureAccounts(t.Context(), subject))
require.NoError(t, c.EnsureAccounts(t.Context(), subject))
require.Len(t, s.rows, 3)
require.Equal(t, firstID, s.rows[0].ID)
// Bound creates are replayed to finish interrupted wallet/route setup;
// the provider must return the original ledger each time.
require.Equal(t, 3, countCurrency(s.posts, "RUB"))
})
}
}
func countCurrency(rows []string, code string) int {
n := 0
for _, r := range rows {
if r == code {
n++
}
}
return n
}
func TestEnsureAccountsConcurrentRequestsUseStableSubjectBinding(t *testing.T) {
s := &accountServer{capable: true}
server := httptest.NewServer(s.serve(t))
defer server.Close()
c, err := Open(config(t, server.URL))
require.NoError(t, err)
errors := make(chan error, 12)
for range 12 {
go func() { errors <- c.EnsureAccounts(t.Context(), subject) }()
}
for range 12 {
require.NoError(t, <-errors)
}
require.Len(t, s.rows, 3)
}
func TestEnsureAccountsRejectsUnsafeProviderStateBeforeCreating(t *testing.T) {
for _, kind := range []string{"old provider", "missing currency", "frozen", "owned", "negative", "duplicate", "wrong subject"} {
t.Run(kind, func(t *testing.T) {
s := &accountServer{capable: true}
a := customerLedger("RUB")
switch kind {
case "old provider":
s.capable = false
case "missing currency":
s.missingCurrency = true
case "frozen":
a.Status = "ACCOUNT_STATUS_FROZEN"
s.rows = []accountDTO{a}
case "owned":
a.Owner = source
s.rows = []accountDTO{a}
case "negative":
a.AllowNegative = true
s.rows = []accountDTO{a}
case "duplicate":
s.rows = []accountDTO{a, customerLedger("RUB")}
case "wrong subject":
a.Metadata["sxchnge_subject_ref"] = source
s.rows = []accountDTO{a}
}
server := httptest.NewServer(s.serve(t))
defer server.Close()
c, err := Open(config(t, server.URL))
require.NoError(t, err)
require.Error(t, c.EnsureAccounts(t.Context(), subject))
require.Empty(t, s.posts)
})
}
}
func TestEnsureAccountsValidatesCreateReceipt(t *testing.T) {
s := &accountServer{capable: true, wrongResponse: true}
server := httptest.NewServer(s.serve(t))
defer server.Close()
c, err := Open(config(t, server.URL))
require.NoError(t, err)
require.Error(t, c.EnsureAccounts(t.Context(), subject))
require.Len(t, s.posts, 1)
}
func TestEnsureAccountsKeepsExistingBindingsAndBalances(t *testing.T) {
a := customerLedger("RUB")
delete(a.Metadata, "external_subject_ref")
s := &accountServer{capable: true, rows: []accountDTO{a}}
server := httptest.NewServer(s.serve(t))
defer server.Close()
c, err := Open(config(t, server.URL))
require.NoError(t, err)
require.NoError(t, c.EnsureAccounts(t.Context(), subject))
require.Equal(t, []string{"USDT", "USDC"}, s.posts)
// A subsequent read uses the original RUB ledger and its authoritative funds.
s.rows = s.rows[:1]
accounts, err := c.Accounts(t.Context(), subject)
require.NoError(t, err)
require.Len(t, accounts, 1)
require.Equal(t, a.ID, accounts[0].ID)
require.EqualValues(t, 4500, accounts[0].Available)
}
@@ -0,0 +1,14 @@
package sendico
type accountListDTO struct {
Accounts []accountDTO `json:"accounts"`
ExternalSubjectAccounts bool `json:"externalSubjectAccounts"`
}
type createAccountDTO struct {
Currency string `json:"currency"`
Type string `json:"accountType"`
LedgerRole string `json:"role"`
AllowNegative bool `json:"allowNegative"`
Metadata map[string]string `json:"metadata"`
}
@@ -27,9 +27,7 @@ func (c *Client) Accounts(ctx context.Context, subject string) ([]payments.Accou
scales[row.Code] = *row.Scale
}
}
var data struct {
Accounts []accountDTO `json:"accounts"`
}
var data accountListDTO
root := "/ledger_accounts/" + c.config.OrganizationID
if err := c.paymentRequest(ctx, http.MethodGet, root, nil, &data); err != nil {
return nil, err
@@ -39,7 +37,7 @@ func (c *Client) Accounts(ctx context.Context, subject string) ([]payments.Accou
if row.Metadata["sxchnge_subject_ref"] != subject {
continue
}
if !validRef(row.ID) || row.ScopeRef != c.config.OrganizationID || row.Owner != "" || row.Status != "ACCOUNT_STATUS_ACTIVE" || row.Type != "ACCOUNT_TYPE_ASSET" || row.LedgerRole != "ACCOUNT_ROLE_OPERATING" {
if !c.validCustomerAccount(row, subject) {
return nil, payments.Unavailable
}
scale, ok := scales[row.Currency]
@@ -54,14 +54,15 @@ type paymentDTO struct {
Created time.Time `json:"createdAt"`
}
type accountDTO struct {
ID string `json:"ledgerAccountRef"`
ScopeRef string `json:"organizationRef"`
Owner string `json:"ownerRef"`
Currency string `json:"currency"`
Type string `json:"accountType"`
LedgerRole string `json:"role"`
Status string `json:"status"`
Metadata map[string]string `json:"metadata"`
ID string `json:"ledgerAccountRef"`
ScopeRef string `json:"organizationRef"`
Owner string `json:"ownerRef"`
Currency string `json:"currency"`
Type string `json:"accountType"`
LedgerRole string `json:"role"`
Status string `json:"status"`
AllowNegative bool `json:"allowNegative"`
Metadata map[string]string `json:"metadata"`
}
type currencyDTO struct {
Code string `json:"code"`
@@ -72,7 +73,7 @@ type balanceDTO struct {
Money moneyDTO `json:"balance"`
}
func (c *Client) paymentRequest(ctx context.Context, method, path string, payload, dst any) error {
func (c *Client) paymentRequest(ctx context.Context, method, path string, payload, dst any, accepted ...int) error {
token, err := c.token()
if err != nil {
return err
@@ -98,7 +99,11 @@ func (c *Client) paymentRequest(ctx context.Context, method, path string, payloa
}
defer func() { _ = res.Body.Close() }()
// Never forward upstream errors, credentials, or arbitrary error bodies.
if res.StatusCode != http.StatusOK {
expected := http.StatusOK
if len(accepted) != 0 {
expected = accepted[0]
}
if res.StatusCode != expected {
return payments.Unavailable
}
var envelope struct {