Connect customer quotations and payment execution to Sendico (#50)

backend/product/payments/http/v1/dto.go

@@ -0,0 +1,84 @@
+package paymenthttp
+
+import (
+	"git.sendico.io/tech/sxchnge/product/payments"
+	factor "git.sendico.io/tech/sxchnge/product/twofactor/model"
+	"time"
+)
+
+type requestDTO struct {
+	Kind        string `json:"kind"`
+	Source      string `json:"source"`
+	Destination string `json:"destination"`
+	Recipient   string `json:"recipient"`
+	Amount      int64  `json:"amount"`
+}
+
+func (d requestDTO) model() payments.Request {
+	return payments.Request{Kind: d.Kind, Source: d.Source, Destination: d.Destination, Recipient: d.Recipient, Amount: d.Amount}
+}
+func fromRequest(r payments.Request) requestDTO {
+	return requestDTO{r.Kind, r.Source, r.Destination, r.Recipient, r.Amount}
+}
+
+type quoteDTO struct {
+	ID          string     `json:"id"`
+	Request     requestDTO `json:"request"`
+	Fee         int64      `json:"fee"`
+	Debit       int64      `json:"debit"`
+	Credit      int64      `json:"credit"`
+	ExpiresAt   time.Time  `json:"expiresAt"`
+	Numerator   int64      `json:"rateNumerator"`
+	Denominator int64      `json:"rateDenominator"`
+}
+
+func fromQuote(q payments.Quote) quoteDTO {
+	return quoteDTO{q.ID, fromRequest(q.Request), q.Fee, q.Debit, q.Credit, q.ExpiresAt, q.RateNumerator, q.RateDenominator}
+}
+
+type operationDTO struct {
+	ID        string    `json:"id"`
+	Quote     quoteDTO  `json:"quote"`
+	Status    string    `json:"status"`
+	CreatedAt time.Time `json:"createdAt"`
+}
+
+func fromOperation(o payments.Operation) operationDTO {
+	return operationDTO{o.ID, fromQuote(o.Quote), o.Status, o.CreatedAt}
+}
+
+type accountDTO struct {
+	ID           string `json:"id"`
+	Currency     string `json:"currency"`
+	Precision    int    `json:"precision"`
+	Available    int64  `json:"available"`
+	PayoutMethod string `json:"payoutMethod"`
+	PayoutLabel  string `json:"payoutLabel"`
+	CanFund      bool   `json:"canFund"`
+}
+type snapshotDTO struct {
+	Accounts   []accountDTO   `json:"accounts"`
+	Operations []operationDTO `json:"operations"`
+}
+
+func fromSnapshot(s payments.Snapshot) snapshotDTO {
+	v := snapshotDTO{Accounts: []accountDTO{}, Operations: []operationDTO{}}
+	for _, a := range s.Accounts {
+		v.Accounts = append(v.Accounts, accountDTO{a.ID, a.Currency, a.Precision, a.Available, a.PayoutMethod, a.PayoutLabel, a.FundingMethod != ""})
+	}
+	for i := len(s.Operations) - 1; i >= 0; i-- {
+		v.Operations = append(v.Operations, fromOperation(s.Operations[i]))
+	}
+	return v
+}
+
+type challengeDTO struct {
+	ID        string    `json:"challengeId"`
+	Target    string    `json:"target"`
+	ExpiresAt time.Time `json:"expiresAt"`
+	ResendAt  time.Time `json:"resendAt"`
+}
+
+func fromChallenge(c factor.Challenge) challengeDTO {
+	return challengeDTO{c.ID, c.Target, c.ExpiresAt, c.ResendAt}
+}

backend/product/payments/http/v1/router.go

@@ -0,0 +1,164 @@
+package paymenthttp
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strconv"
+
+	sresponse "git.sendico.io/tech/sxchnge/product/auth/http/v1/response"
+	"git.sendico.io/tech/sxchnge/product/auth/ipguard"
+	customer "git.sendico.io/tech/sxchnge/product/customers/model"
+	"git.sendico.io/tech/sxchnge/product/payments"
+	verification "git.sendico.io/tech/sxchnge/product/verification/store"
+	merrors "git.sendico.io/tech/sxchnge/shared/errors"
+	api "git.sendico.io/tech/sxchnge/shared/http"
+	"git.sendico.io/tech/sxchnge/shared/http/response"
+	mlogger "git.sendico.io/tech/sxchnge/shared/logging"
+	mservice "git.sendico.io/tech/sxchnge/shared/service"
+)
+
+type Registrar interface {
+	AccountHandler(mservice.Type, string, api.HTTPMethod, sresponse.AccountHandlerFunc)
+}
+type router struct {
+	service *payments.Service
+	logger  mlogger.Logger
+}
+
+func Install(reg Registrar, s *payments.Service, logger mlogger.Logger) {
+	h := &router{s, logger.Named("payment_http")}
+	reg.AccountHandler(mservice.Accounts, "/wallet/profile", api.Get, h.profile)
+	reg.AccountHandler(mservice.Accounts, "/wallet", api.Get, h.fetch)
+	reg.AccountHandler(mservice.Accounts, "/wallet/quote", api.Post, h.quote)
+	reg.AccountHandler(mservice.Accounts, "/wallet/submit", api.Post, h.submit)
+	reg.AccountHandler(mservice.Accounts, "/wallet/submission", api.Post, h.reconcile)
+	reg.AccountHandler(mservice.Accounts, "/wallet/challenge", api.Post, h.challenge)
+	reg.AccountHandler(mservice.Accounts, "/wallet/resend", api.Post, h.resend)
+	reg.AccountHandler(mservice.Accounts, "/wallet/verify", api.Post, h.verify)
+}
+func decode(r *http.Request, dst any) error {
+	d := json.NewDecoder(io.LimitReader(r.Body, 8193))
+	d.DisallowUnknownFields()
+	if d.Decode(dst) != nil {
+		return payments.InvalidAmount
+	}
+	var extra any
+	if d.Decode(&extra) != io.EOF {
+		return payments.InvalidAmount
+	}
+	return nil
+}
+func (h *router) result(data any, err error) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Cache-Control", "no-store")
+		if err == nil {
+			response.Ok(h.logger, data)(w, r)
+			return
+		}
+		code := "unavailable"
+		status := http.StatusServiceUnavailable
+		var f payments.Failure
+		if errors.As(err, &f) {
+			code = string(f)
+			if f != payments.Unavailable && f != payments.Uncertain {
+				status = http.StatusBadRequest
+			}
+		}
+		switch {
+		case errors.Is(err, verification.ErrTokenExpired):
+			code = "factorExpired"
+			status = 400
+		case errors.Is(err, verification.ErrTokenAttemptsExceeded):
+			code = "factorLocked"
+			status = 400
+		case errors.Is(err, verification.ErrTokenNotFound), errors.Is(err, verification.ErrTokenAlreadyUsed), errors.Is(err, merrors.ErrUnauthorized):
+			code = "factorInvalid"
+			status = 400
+		}
+		if throttle, ok := verification.Throttle(err); ok {
+			w.Header().Set("Retry-After", strconv.Itoa(max(1, int(throttle.RetryAfter.Seconds()))))
+			code = "factorCooldown"
+			status = 429
+		}
+		response.Error(h.logger, mservice.Accounts, status, code, "Unable to complete wallet operation")(w, r)
+	}
+}
+func (h *router) fetch(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	v, e := h.service.Fetch(r.Context(), a)
+	return h.result(fromSnapshot(v), e)
+}
+func (h *router) quote(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d requestDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	v, e := h.service.Quote(r.Context(), a, d.model())
+	return h.result(fromQuote(v), e)
+}
+
+type actionDTO struct {
+	Quote     string `json:"quoteId"`
+	Key       string `json:"idempotencyKey"`
+	Challenge string `json:"challengeId"`
+	Code      string `json:"code"`
+}
+
+func (h *router) submit(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d actionDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	v, e := h.service.Submit(r.Context(), a, d.Quote, d.Key)
+	return h.result(fromOperation(v), e)
+}
+func (h *router) reconcile(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d actionDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	v, e := h.service.Reconcile(r.Context(), a, d.Key)
+	if v == nil {
+		return h.result(nil, e)
+	}
+	return h.result(fromOperation(*v), e)
+}
+func requestIP(r *http.Request) string {
+	if ip := ipguard.ClientIP(r); ip != nil {
+		return ip.String()
+	}
+	return "unresolved"
+}
+func (h *router) challenge(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d actionDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	v, e := h.service.Challenge(r.Context(), a, d.Quote, requestIP(r))
+	return h.result(fromChallenge(v), e)
+}
+func (h *router) resend(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d actionDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	v, e := h.service.Resend(r.Context(), a, d.Quote, d.Challenge, d.Key, requestIP(r))
+	return h.result(fromChallenge(v), e)
+}
+func (h *router) verify(r *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	var d actionDTO
+	if e := decode(r, &d); e != nil {
+		return h.result(nil, e)
+	}
+	e := h.service.Authorize(r.Context(), a, d.Quote, d.Challenge, d.Code, d.Key)
+	return h.result(struct {
+		Key string `json:"idempotencyKey"`
+	}{d.Key}, e)
+}
+
+func (h *router) profile(_ *http.Request, a *customer.Account, _ *sresponse.TokenData) http.HandlerFunc {
+	return h.result(struct {
+		Locale string `json:"locale"`
+	}{a.Locale}, nil)
+}

backend/product/payments/http/v1/router_test.go

@@ -0,0 +1,36 @@
+package paymenthttp
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"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 TestUnconfiguredWalletCannotReturnDemoFunds(t *testing.T) {
+	h := &router{logger: zap.NewNop()}
+	a := &customer.Account{}
+	a.ID = bson.NewObjectID()
+	request := httptest.NewRequest(http.MethodGet, "/accounts/wallet", nil)
+	recorder := httptest.NewRecorder()
+	h.fetch(request, a, nil)(recorder, request)
+	require.Equal(t, http.StatusServiceUnavailable, recorder.Code)
+	require.Equal(t, "no-store", recorder.Header().Get("Cache-Control"))
+	require.NotContains(t, recorder.Body.String(), "7500")
+}
+func TestQuoteRejectsUnknownFieldsAndTrailingRequests(t *testing.T) {
+	h := &router{logger: zap.NewNop()}
+	a := &customer.Account{}
+	a.ID = bson.NewObjectID()
+	for _, body := range []string{`{"kind":"payout","subject":"foreign"}`, `{"amount":10} {"amount":20}`, `{"amount":1.2}`} {
+		request := httptest.NewRequest(http.MethodPost, "/accounts/wallet/quote", strings.NewReader(body))
+		recorder := httptest.NewRecorder()
+		h.quote(request, a, nil)(recorder, request)
+		require.Equal(t, http.StatusBadRequest, recorder.Code)
+	}
+}

backend/product/payments/model.go

@@ -0,0 +1,83 @@
+// Package payments owns customer quotation ownership and durable execution intent.
+// Money and settlement remain authoritative in the provider.
+package payments
+
+import (
+	"context"
+	"time"
+
+	customer "git.sendico.io/tech/sxchnge/product/customers/model"
+	"git.sendico.io/tech/sxchnge/shared/database/record"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+type Account struct {
+	ID, Currency                             string
+	Precision                                int
+	Available                                int64
+	PayoutMethod, PayoutLabel, FundingMethod string
+}
+type Request struct {
+	Kind, Source, Destination, Recipient string
+	Amount                               int64
+}
+type Quote struct {
+	ID                             string
+	Request                        Request
+	Fee, Debit, Credit             int64
+	ExpiresAt                      time.Time
+	RateNumerator, RateDenominator int64
+}
+type Operation struct {
+	ID        string
+	Quote     Quote
+	Status    string
+	CreatedAt time.Time
+}
+type Snapshot struct {
+	Accounts   []Account
+	Operations []Operation
+}
+type Record struct {
+	storable.Base `bson:",inline"`
+	Customer      bson.ObjectID `bson:"customer"`
+	Namespace     string        `bson:"namespace"`
+	Subject       string        `bson:"subject"`
+	Quote         Quote         `bson:"quote"`
+	ProviderQuote string        `bson:"providerQuote"`
+	Key           string        `bson:"key"`
+	Payment       string        `bson:"payment"`
+}
+
+func (*Record) Collection() string { return "payment_quotes" }
+
+type Provider interface {
+	Namespace() string
+	Accounts(context.Context, string) ([]Account, error)
+	Quote(context.Context, string, Request, []Account) (Quote, error)
+	Execute(context.Context, string, string) (RemoteOperation, error)
+	Payment(context.Context, string) (RemoteOperation, error)
+}
+type RemoteOperation struct {
+	ID, QuoteRef, Status string
+	CreatedAt            time.Time
+}
+type Identity interface {
+	Subject(context.Context, *customer.Account) (string, error)
+}
+type Failure string
+
+func (e Failure) Error() string { return string(e) }
+
+const (
+	Unavailable       Failure = "unavailable"
+	InvalidAmount     Failure = "invalidAmount"
+	InvalidRecipient  Failure = "invalidRecipient"
+	InsufficientFunds Failure = "insufficientFunds"
+	LimitExceeded     Failure = "limitExceeded"
+	ExpiredQuote      Failure = "expiredQuote"
+	Unsupported       Failure = "unsupported"
+	NotFound          Failure = "notFound"
+	Uncertain         Failure = "uncertain"
+	FactorRequired    Failure = "factorRequired"
+)

backend/product/payments/service.go

@@ -0,0 +1,237 @@
+package payments
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	customer "git.sendico.io/tech/sxchnge/product/customers/model"
+	"git.sendico.io/tech/sxchnge/product/twofactor"
+	factor "git.sendico.io/tech/sxchnge/product/twofactor/model"
+	verification "git.sendico.io/tech/sxchnge/product/verification/model"
+	merrors "git.sendico.io/tech/sxchnge/shared/errors"
+	mlogger "git.sendico.io/tech/sxchnge/shared/logging"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.uber.org/zap"
+)
+
+type Service struct {
+	Store    Store
+	Provider Provider
+	Identity Identity
+	Factors  twofactor.API
+	Logger   mlogger.Logger
+}
+
+func (s *Service) ready() error {
+	if s == nil || s.Store == nil || s.Provider == nil || s.Identity == nil || s.Factors == nil {
+		return Unavailable
+	}
+	return nil
+}
+func (s *Service) record(ctx context.Context, a *customer.Account, id string) (*Record, error) {
+	if err := s.ready(); err != nil {
+		return nil, err
+	}
+	if a == nil || a.ID.IsZero() {
+		return nil, NotFound
+	}
+	ref, err := bson.ObjectIDFromHex(id)
+	if err != nil {
+		return nil, NotFound
+	}
+	r, err := s.Store.Get(ctx, ref, a.ID, s.Provider.Namespace())
+	if errors.Is(err, merrors.ErrNoData) {
+		return nil, NotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	subject, err := s.Identity.Subject(ctx, a)
+	if err != nil {
+		return nil, err
+	}
+	if subject != r.Subject {
+		return nil, Unavailable
+	}
+	return r, nil
+}
+func (s *Service) Quote(ctx context.Context, a *customer.Account, request Request) (Quote, error) {
+	if err := s.ready(); err != nil {
+		return Quote{}, err
+	}
+	subject, err := s.Identity.Subject(ctx, a)
+	if err != nil {
+		return Quote{}, err
+	}
+	accounts, err := s.Provider.Accounts(ctx, subject)
+	if err != nil {
+		return Quote{}, err
+	}
+	q, err := s.Provider.Quote(ctx, subject, request, accounts)
+	if err != nil {
+		return Quote{}, err
+	}
+	providerRef := q.ID
+	q.ID = bson.NewObjectID().Hex()
+	r := &Record{Customer: a.ID, Namespace: s.Provider.Namespace(), Subject: subject, Quote: q, ProviderQuote: providerRef}
+	r.SetID(mustID(q.ID))
+	if err = s.Store.Create(ctx, r); err != nil {
+		return Quote{}, err
+	}
+	s.Logger.Info("Customer quotation stored", zap.String("quote_ref", q.ID), zap.String("kind", request.Kind))
+	return q, nil
+}
+func mustID(id string) bson.ObjectID { value, _ := bson.ObjectIDFromHex(id); return value }
+func operationScope(r *Record) factor.Scope {
+	return factor.Scope{Purpose: verification.PurposePayment, ContextRef: r.Quote.ID}
+}
+func (s *Service) Challenge(ctx context.Context, a *customer.Account, id, ip string) (factor.Challenge, error) {
+	r, err := s.record(ctx, a, id)
+	if err != nil {
+		return factor.Challenge{}, err
+	}
+	if !time.Now().Before(r.Quote.ExpiresAt) || r.Key != "" {
+		return factor.Challenge{}, ExpiredQuote
+	}
+	return s.Factors.Begin(ctx, factor.Request{Subject: a.ID, Scope: operationScope(r), Deadline: r.Quote.ExpiresAt, IdempotencyKey: "payment:" + r.Quote.ID, IP: ip})
+}
+func (s *Service) Resend(ctx context.Context, a *customer.Account, id, challenge, key, ip string) (factor.Challenge, error) {
+	r, err := s.record(ctx, a, id)
+	if err != nil {
+		return factor.Challenge{}, err
+	}
+	if !time.Now().Before(r.Quote.ExpiresAt) || r.Key != "" {
+		return factor.Challenge{}, ExpiredQuote
+	}
+	if !validKey(key) {
+		return factor.Challenge{}, InvalidAmount
+	}
+	return s.Factors.Resend(ctx, factor.Request{Subject: a.ID, Scope: operationScope(r), Deadline: r.Quote.ExpiresAt, IdempotencyKey: key, IP: ip}, challenge)
+}
+func validKey(key string) bool {
+	if len(key) < 16 || len(key) > 128 {
+		return false
+	}
+	for _, c := range key {
+		if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-' || c == '_') {
+			return false
+		}
+	}
+	return true
+}
+
+// Authorize consumes the challenge atomically with a durable execution intent.
+// No upstream I/O occurs inside the transaction. Replaying the same authorization
+// recovers a lost response without reusing the one-use verification code.
+func (s *Service) Authorize(ctx context.Context, a *customer.Account, id, challenge, code, key string) error {
+	r, err := s.record(ctx, a, id)
+	if err != nil {
+		return err
+	}
+	if !validKey(key) {
+		return InvalidAmount
+	}
+	if r.Key != "" {
+		if r.Key == key {
+			return nil
+		}
+		return ExpiredQuote
+	}
+	return s.Factors.Complete(ctx, a.ID, operationScope(r), challenge, code, func(tx context.Context) error {
+		if !time.Now().Before(r.Quote.ExpiresAt) {
+			return ExpiredQuote
+		}
+		return s.Store.Authorize(tx, r, key)
+	})
+}
+func (s *Service) execute(ctx context.Context, r *Record) (Operation, error) {
+	var remote RemoteOperation
+	var err error
+	if r.Payment != "" {
+		remote, err = s.Provider.Payment(ctx, r.Payment)
+	} else {
+		remote, err = s.Provider.Execute(ctx, r.ProviderQuote, "sxchnge:payment:"+r.ID.Hex())
+	}
+	if err != nil {
+		s.Logger.Warn("Payment execution needs reconciliation", zap.String("quote_ref", r.Quote.ID))
+		return Operation{}, merrors.OpaqueWrap(Uncertain, err, "provider payment outcome unknown")
+	}
+	if remote.QuoteRef != r.ProviderQuote || remote.ID == "" || (r.Payment != "" && r.Payment != remote.ID) {
+		return Operation{}, Uncertain
+	}
+	if r.Payment == "" {
+		if err = s.Store.Attach(ctx, r, remote.ID); err != nil {
+			return Operation{}, merrors.OpaqueWrap(Uncertain, err, "persist provider payment reference")
+		}
+	}
+	s.Logger.Info("Payment execution reconciled", zap.String("quote_ref", r.Quote.ID), zap.String("payment_ref", remote.ID), zap.String("state", remote.Status))
+	return Operation{ID: r.Quote.ID, Quote: r.Quote, Status: remote.Status, CreatedAt: remote.CreatedAt}, nil
+}
+func (s *Service) Submit(ctx context.Context, a *customer.Account, id, key string) (Operation, error) {
+	r, err := s.record(ctx, a, id)
+	if err != nil {
+		return Operation{}, err
+	}
+	if !validKey(key) || r.Key == "" || r.Key != key {
+		return Operation{}, FactorRequired
+	}
+	return s.execute(ctx, r)
+}
+
+// Reconcile replays the already authorized durable intent with the same provider
+// key. A timeout or an empty upstream list is never treated as proof of absence.
+func (s *Service) Reconcile(ctx context.Context, a *customer.Account, key string) (*Operation, error) {
+	if err := s.ready(); err != nil {
+		return nil, err
+	}
+	if !validKey(key) {
+		return nil, NotFound
+	}
+	r, err := s.Store.ByKey(ctx, a.ID, s.Provider.Namespace(), key)
+	if errors.Is(err, merrors.ErrNoData) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	r, err = s.record(ctx, a, r.ID.Hex())
+	if err != nil {
+		return nil, err
+	}
+	op, err := s.execute(ctx, r)
+	return &op, err
+}
+func (s *Service) Fetch(ctx context.Context, a *customer.Account) (Snapshot, error) {
+	if err := s.ready(); err != nil {
+		return Snapshot{}, err
+	}
+	subject, err := s.Identity.Subject(ctx, a)
+	if err != nil {
+		return Snapshot{}, err
+	}
+	rows, err := s.Store.List(ctx, a.ID, s.Provider.Namespace())
+	if err != nil {
+		return Snapshot{}, err
+	}
+	result := Snapshot{Operations: []Operation{}}
+	for _, r := range rows {
+		if r.Subject != subject {
+			return Snapshot{}, Unavailable
+		}
+		op, err := s.execute(ctx, &r)
+		if errors.Is(err, Uncertain) {
+			// A durable authorization with an unknown provider outcome remains
+			// pending. It must not hide unrelated balances or report failure.
+			op = Operation{ID: r.Quote.ID, Quote: r.Quote, Status: "pending", CreatedAt: r.CreatedAt}
+		} else if err != nil {
+			return Snapshot{}, err
+		}
+		result.Operations = append(result.Operations, op)
+	}
+	result.Accounts, err = s.Provider.Accounts(ctx, subject)
+	if err != nil {
+		return Snapshot{}, err
+	}
+	return result, nil
+}

backend/product/payments/service_test.go

@@ -0,0 +1,189 @@
+package payments
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	customer "git.sendico.io/tech/sxchnge/product/customers/model"
+	factor "git.sendico.io/tech/sxchnge/product/twofactor/model"
+	merrors "git.sendico.io/tech/sxchnge/shared/errors"
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.uber.org/zap"
+)
+
+type memoryStore struct {
+	mu      sync.Mutex
+	records map[bson.ObjectID]Record
+}
+
+func (s *memoryStore) Create(_ context.Context, r *Record) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.records[r.ID] = *r
+	return nil
+}
+func (s *memoryStore) Get(_ context.Context, id, customer bson.ObjectID, namespace string) (*Record, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	r, ok := s.records[id]
+	if !ok || r.Customer != customer || r.Namespace != namespace {
+		return nil, merrors.ErrNoData
+	}
+	return &r, nil
+}
+func (s *memoryStore) ByKey(_ context.Context, customer bson.ObjectID, namespace, key string) (*Record, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, r := range s.records {
+		if r.Customer == customer && r.Namespace == namespace && r.Key == key {
+			return &r, nil
+		}
+	}
+	return nil, merrors.ErrNoData
+}
+func (s *memoryStore) List(_ context.Context, customer bson.ObjectID, namespace string) ([]Record, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	rows := []Record{}
+	for _, r := range s.records {
+		if r.Customer == customer && r.Namespace == namespace && r.Key != "" {
+			rows = append(rows, r)
+		}
+	}
+	return rows, nil
+}
+func (s *memoryStore) Authorize(_ context.Context, r *Record, key string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	current := s.records[r.ID]
+	if current.Key != "" {
+		return Uncertain
+	}
+	current.Key = key
+	s.records[r.ID] = current
+	return nil
+}
+func (s *memoryStore) Attach(_ context.Context, r *Record, id string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	current := s.records[r.ID]
+	current.Payment = id
+	s.records[r.ID] = current
+	return nil
+}
+
+type provider struct {
+	mu        sync.Mutex
+	accepted  int
+	lost      bool
+	status    string
+	key       string
+	namespace string
+}
+
+func (p *provider) Namespace() string { return p.namespace }
+func (p *provider) Accounts(context.Context, string) ([]Account, error) {
+	return []Account{{ID: "ledger", Currency: "EUR", Precision: 2, Available: 5000}}, nil
+}
+func (p *provider) Quote(_ context.Context, _ string, r Request, _ []Account) (Quote, error) {
+	return Quote{ID: "provider-quote", Request: r, ExpiresAt: time.Now().Add(time.Minute)}, nil
+}
+func (p *provider) Execute(_ context.Context, quote, key string) (RemoteOperation, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.key == "" {
+		p.accepted++
+		p.key = key
+	}
+	if p.key != key {
+		return RemoteOperation{}, Unavailable
+	}
+	if p.lost {
+		p.lost = false
+		return RemoteOperation{}, Unavailable
+	}
+	return RemoteOperation{ID: "payment", QuoteRef: quote, Status: p.status, CreatedAt: time.Now()}, nil
+}
+func (p *provider) Payment(context.Context, string) (RemoteOperation, error) {
+	return RemoteOperation{ID: "payment", QuoteRef: "provider-quote", Status: p.status, CreatedAt: time.Now()}, nil
+}
+
+type identity struct{}
+
+func (identity) Subject(context.Context, *customer.Account) (string, error) { return "subject", nil }
+
+type factors struct {
+	calls   int
+	subject bson.ObjectID
+	scope   factor.Scope
+}
+
+func (f *factors) Begin(_ context.Context, r factor.Request) (factor.Challenge, error) {
+	f.subject = r.Subject
+	f.scope = r.Scope
+	return factor.Challenge{ID: "challenge"}, nil
+}
+func (f *factors) Resend(ctx context.Context, r factor.Request, _ string) (factor.Challenge, error) {
+	return f.Begin(ctx, r)
+}
+func (f *factors) Complete(ctx context.Context, subject bson.ObjectID, scope factor.Scope, id, code string, action func(context.Context) error) error {
+	if subject != f.subject || scope != f.scope || id != "challenge" || code != "123456" {
+		return FactorRequired
+	}
+	f.calls++
+	return action(ctx)
+}
+func TestCustomerExecutionOwnershipAndLostResponse(t *testing.T) {
+	ctx := context.Background()
+	a := &customer.Account{}
+	a.ID = bson.NewObjectID()
+	other := &customer.Account{}
+	other.ID = bson.NewObjectID()
+	store := &memoryStore{records: map[bson.ObjectID]Record{}}
+	p := &provider{lost: true, status: "pending", namespace: "provider"}
+	f := &factors{}
+	s := &Service{Store: store, Provider: p, Identity: identity{}, Factors: f, Logger: zap.NewNop()}
+	q, err := s.Quote(ctx, a, Request{Kind: "payout", Source: "ledger", Amount: 1000})
+	require.NoError(t, err)
+	_, err = s.Submit(ctx, a, q.ID, "unapproved-key-123")
+	require.ErrorIs(t, err, FactorRequired)
+	require.Zero(t, p.accepted)
+	_, err = s.Challenge(ctx, other, q.ID, "")
+	require.ErrorIs(t, err, NotFound)
+	_, err = s.Challenge(ctx, a, q.ID, "")
+	require.NoError(t, err)
+	require.ErrorIs(t, s.Authorize(ctx, a, q.ID, "challenge", "bad-code", "authorized-key-123"), FactorRequired)
+	require.NoError(t, s.Authorize(ctx, a, q.ID, "challenge", "123456", "authorized-key-123"))
+	require.NoError(t, s.Authorize(ctx, a, q.ID, "challenge", "123456", "authorized-key-123"))
+	require.Equal(t, 1, f.calls)
+	_, err = s.Submit(ctx, a, q.ID, "authorized-key-123")
+	require.ErrorIs(t, err, Uncertain)
+	op, err := s.Reconcile(ctx, other, "authorized-key-123")
+	require.NoError(t, err)
+	require.Nil(t, op)
+	op, err = s.Reconcile(ctx, a, "authorized-key-123")
+	require.NoError(t, err)
+	require.Equal(t, "pending", op.Status)
+	require.Equal(t, 1, p.accepted)
+	p.status = "completed"
+	snapshot, err := s.Fetch(ctx, a)
+	require.NoError(t, err)
+	require.Equal(t, "completed", snapshot.Operations[0].Status)
+	require.ErrorIs(t, s.Authorize(ctx, a, q.ID, "challenge", "123456", "different-key-1234"), ExpiredQuote)
+	// A changed provider namespace cannot execute the old quote.
+	p.namespace = "rotated-provider"
+	_, err = s.Submit(ctx, a, q.ID, "authorized-key-123")
+	require.ErrorIs(t, err, NotFound)
+}
+func TestExpiredQuoteCannotAuthorize(t *testing.T) {
+	a := &customer.Account{}
+	a.ID = bson.NewObjectID()
+	r := Record{Customer: a.ID, Namespace: "p", Subject: "subject", Quote: Quote{ID: bson.NewObjectID().Hex(), ExpiresAt: time.Now().Add(-time.Second)}}
+	r.SetID(mustID(r.Quote.ID))
+	s := &Service{Store: &memoryStore{records: map[bson.ObjectID]Record{r.ID: r}}, Provider: &provider{namespace: "p"}, Identity: identity{}, Factors: &factors{}, Logger: zap.NewNop()}
+	_, err := s.Challenge(context.Background(), a, r.Quote.ID, "")
+	require.ErrorIs(t, err, ExpiredQuote)
+}

backend/product/payments/store.go

@@ -0,0 +1,88 @@
+package payments
+
+import (
+	"context"
+
+	db "git.sendico.io/tech/sxchnge/shared/database"
+	deps "git.sendico.io/tech/sxchnge/shared/database/dependencies"
+	"git.sendico.io/tech/sxchnge/shared/database/repository"
+	"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
+	ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
+	"git.sendico.io/tech/sxchnge/shared/database/source"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+type Store interface {
+	Create(context.Context, *Record) error
+	Get(context.Context, bson.ObjectID, bson.ObjectID, string) (*Record, error)
+	ByKey(context.Context, bson.ObjectID, string, string) (*Record, error)
+	List(context.Context, bson.ObjectID, string) ([]Record, error)
+	Authorize(context.Context, *Record, string) error
+	Attach(context.Context, *Record, string) error
+}
+type mongoStore struct{ records repository.Repository }
+
+func OpenStore(factory db.Factory) (Store, error) {
+	d, err := deps.ResolveDependencies(factory)
+	if err != nil {
+		return nil, err
+	}
+	r := repository.CreateRepository(source.Majority(d.Database), "payment_quotes")
+	err = r.CreateIndex(&ri.Definition{Name: "customer_execution_key", Unique: true, Keys: []ri.Key{{Field: "customer", Sort: ri.Asc}, {Field: "namespace", Sort: ri.Asc}, {Field: "key", Sort: ri.Asc}}, PartialFilter: repository.Query().Comparison(repository.Field("key"), builder.Gt, "")})
+	if err != nil {
+		return nil, err
+	}
+	if err = r.CreateIndex(&ri.Definition{Name: "provider_quote_unique", Unique: true, Keys: []ri.Key{{Field: "namespace", Sort: ri.Asc}, {Field: "providerQuote", Sort: ri.Asc}}, PartialFilter: repository.Query().Comparison(repository.Field("providerQuote"), builder.Gt, "")}); err != nil {
+		return nil, err
+	}
+	return &mongoStore{r}, nil
+}
+func scope(customer bson.ObjectID, namespace string) repository.FilterQuery {
+	return repository.Query().Filter(repository.Field("customer"), customer).Filter(repository.Field("namespace"), namespace)
+}
+func (s *mongoStore) Create(ctx context.Context, r *Record) error {
+	return s.records.Insert(ctx, r, nil)
+}
+func (s *mongoStore) Get(ctx context.Context, id, customer bson.ObjectID, namespace string) (*Record, error) {
+	r := &Record{}
+	err := s.records.FindOneByFilter(ctx, scope(customer, namespace).Filter(repository.IDField(), id), r)
+	return r, err
+}
+func (s *mongoStore) ByKey(ctx context.Context, customer bson.ObjectID, namespace, key string) (*Record, error) {
+	r := &Record{}
+	err := s.records.FindOneByFilter(ctx, scope(customer, namespace).Filter(repository.Field("key"), key), r)
+	return r, err
+}
+func (s *mongoStore) List(ctx context.Context, customer bson.ObjectID, namespace string) ([]Record, error) {
+	var rows []Record
+	err := s.records.FindAllByFilter(ctx, scope(customer, namespace).Comparison(repository.Field("key"), builder.Gt, "").Sort(repository.Field("createdAt"), true), &rows)
+	return rows, err
+}
+func (s *mongoStore) Authorize(ctx context.Context, r *Record, key string) error {
+	matched, err := s.records.PatchOneByFilter(ctx, scope(r.Customer, r.Namespace).Filter(repository.IDField(), r.ID).Filter(repository.Field("key"), ""), repository.Patch().Set(repository.Field("key"), key), &Record{})
+	if err != nil {
+		return err
+	}
+	if !matched {
+		return Uncertain
+	}
+	return nil
+}
+
+func (s *mongoStore) Attach(ctx context.Context, r *Record, id string) error {
+	matched, err := s.records.PatchOneByFilter(ctx, scope(r.Customer, r.Namespace).Filter(repository.IDField(), r.ID).Filter(repository.Field("payment"), ""), repository.Patch().Set(repository.Field("payment"), id), &Record{})
+	if err != nil {
+		return err
+	}
+	if matched {
+		return nil
+	}
+	current, err := s.Get(ctx, r.ID, r.Customer, r.Namespace)
+	if err != nil {
+		return err
+	}
+	if current.Payment != id {
+		return Uncertain
+	}
+	return nil
+}

backend/product/payments/store_test.go

@@ -0,0 +1,66 @@
+package payments
+
+import (
+	"context"
+	"net/url"
+	"testing"
+	"time"
+
+	db "git.sendico.io/tech/sxchnge/shared/database"
+	merrors "git.sendico.io/tech/sxchnge/shared/errors"
+	sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go/modules/mongodb"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.uber.org/zap"
+)
+
+func TestDurableAuthorizationTransactionAndCustomerIsolation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	container, err := mongodb.Run(ctx, "mongo:latest", mongodb.WithReplicaSet("rs"))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, container.Terminate(context.Background())) })
+	uri, err := container.ConnectionString(ctx)
+	require.NoError(t, err)
+	u, err := url.Parse(uri)
+	require.NoError(t, err)
+	q := u.Query()
+	q.Set("directConnection", "true")
+	u.RawQuery = q.Encode()
+	factory, err := db.NewConnection(zap.NewNop(), &db.Config{Driver: db.Mongo, Settings: sharedmodel.SettingsT{"uri": u.String(), "database": "payments"}})
+	require.NoError(t, err)
+	t.Cleanup(factory.CloseConnection)
+	store, err := OpenStore(factory)
+	require.NoError(t, err)
+	customer := bson.NewObjectID()
+	r := &Record{Customer: customer, Namespace: "provider", Subject: "subject"}
+	r.SetID(bson.NewObjectID())
+	require.NoError(t, store.Create(ctx, r))
+	_, err = store.Get(ctx, r.ID, bson.NewObjectID(), r.Namespace)
+	require.ErrorIs(t, err, merrors.ErrNoData)
+	_, err = factory.TransactionFactory().CreateTransaction().Execute(ctx, func(tx context.Context) (any, error) {
+		require.NoError(t, store.Authorize(tx, r, "authorized-key-123"))
+		return nil, Uncertain
+	})
+	require.ErrorIs(t, err, Uncertain)
+	stored, err := store.Get(ctx, r.ID, customer, r.Namespace)
+	require.NoError(t, err)
+	require.Empty(t, stored.Key)
+	require.NoError(t, store.Authorize(ctx, r, "authorized-key-123"))
+	require.ErrorIs(t, store.Authorize(ctx, r, "other-key-123456"), Uncertain)
+	another := &Record{Customer: customer, Namespace: r.Namespace}
+	another.SetID(bson.NewObjectID())
+	require.NoError(t, store.Create(ctx, another))
+	require.ErrorIs(t, store.Authorize(ctx, another, "authorized-key-123"), merrors.ErrDataConflict)
+	require.NoError(t, store.Attach(ctx, r, "payment-1"))
+	require.NoError(t, store.Attach(ctx, r, "payment-1"))
+	require.ErrorIs(t, store.Attach(ctx, r, "different-payment"), Uncertain)
+	reopened, err := OpenStore(factory)
+	require.NoError(t, err)
+	stored, err = reopened.ByKey(ctx, customer, r.Namespace, "authorized-key-123")
+	require.NoError(t, err)
+	require.Equal(t, "payment-1", stored.Payment)
+	_, err = reopened.ByKey(ctx, customer, "other-provider", "authorized-key-123")
+	require.ErrorIs(t, err, merrors.ErrNoData)
+}