fixed fee direction

api/gateway/aurora/storage/mongo/repository.go

@@ -0,0 +1,88 @@
+package mongo
+
+import (
+	"context"
+	"time"
+
+	"github.com/tech/sendico/gateway/aurora/storage"
+	"github.com/tech/sendico/gateway/aurora/storage/mongo/store"
+	gatewayoutbox "github.com/tech/sendico/gateway/common/outbox"
+	"github.com/tech/sendico/pkg/db"
+	"github.com/tech/sendico/pkg/db/transaction"
+	"github.com/tech/sendico/pkg/merrors"
+	"github.com/tech/sendico/pkg/mlogger"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.uber.org/zap"
+)
+
+type Repository struct {
+	logger    mlogger.Logger
+	conn      *db.MongoConnection
+	db        *mongo.Database
+	txFactory transaction.Factory
+
+	payouts storage.PayoutsStore
+	outbox  gatewayoutbox.Store
+}
+
+func New(logger mlogger.Logger, conn *db.MongoConnection) (*Repository, error) {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	if conn == nil {
+		return nil, merrors.InvalidArgument("mongo connection is nil")
+	}
+	client := conn.Client()
+	if client == nil {
+		return nil, merrors.Internal("mongo client is not initialised")
+	}
+	db := conn.Database()
+	if db == nil {
+		return nil, merrors.Internal("mongo database is not initialised")
+	}
+	dbName := db.Name()
+	logger = logger.Named("storage").Named("mongo")
+	if dbName != "" {
+		logger = logger.With(zap.String("database", dbName))
+	}
+	result := &Repository{
+		logger:    logger,
+		conn:      conn,
+		db:        db,
+		txFactory: newMongoTransactionFactory(client),
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := result.conn.Ping(ctx); err != nil {
+		result.logger.Error("Mongo ping failed during repository initialisation", zap.Error(err))
+		return nil, err
+	}
+	payoutsStore, err := store.NewPayouts(result.logger, result.db)
+	if err != nil {
+		result.logger.Error("Failed to initialise payouts store", zap.Error(err), zap.String("store", "payments"))
+		return nil, err
+	}
+	outboxStore, err := gatewayoutbox.NewMongoStore(result.logger, result.db)
+	if err != nil {
+		result.logger.Error("Failed to initialise outbox store", zap.Error(err), zap.String("store", "outbox"))
+		return nil, err
+	}
+	result.payouts = payoutsStore
+	result.outbox = outboxStore
+	result.logger.Info("Payouts gateway MongoDB storage initialised")
+	return result, nil
+}
+
+func (r *Repository) Payouts() storage.PayoutsStore {
+	return r.payouts
+}
+
+func (r *Repository) Outbox() gatewayoutbox.Store {
+	return r.outbox
+}
+
+func (r *Repository) TransactionFactory() transaction.Factory {
+	return r.txFactory
+}
+
+var _ storage.Repository = (*Repository)(nil)

api/gateway/aurora/storage/mongo/store/payouts.go

@@ -0,0 +1,108 @@
+package store
+
+import (
+	"context"
+	"strings"
+
+	storage "github.com/tech/sendico/gateway/aurora/storage"
+	"github.com/tech/sendico/gateway/aurora/storage/model"
+	"github.com/tech/sendico/pkg/db/repository"
+	ri "github.com/tech/sendico/pkg/db/repository/index"
+	"github.com/tech/sendico/pkg/merrors"
+	"github.com/tech/sendico/pkg/mlogger"
+	"github.com/tech/sendico/pkg/mutil/mzap"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.uber.org/zap"
+)
+
+const (
+	payoutsCollection = "card_payouts"
+	payoutIdemField   = "idempotencyKey"
+	payoutIdField     = "paymentRef"
+	payoutOpField     = "operationRef"
+)
+
+type Payouts struct {
+	logger     mlogger.Logger
+	repository repository.Repository
+}
+
+func NewPayouts(logger mlogger.Logger, db *mongo.Database) (*Payouts, error) {
+	if db == nil {
+		return nil, merrors.InvalidArgument("mongo database is nil")
+	}
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	logger = logger.Named("payouts").With(zap.String("collection", payoutsCollection))
+
+	repo := repository.CreateMongoRepository(db, payoutsCollection)
+	if err := repo.CreateIndex(&ri.Definition{
+		Keys:   []ri.Key{{Field: payoutOpField, Sort: ri.Asc}},
+		Unique: true,
+		Sparse: true,
+	}); err != nil {
+		logger.Error("Failed to create payouts operation index",
+			zap.Error(err), zap.String("index_field", payoutOpField))
+		return nil, err
+	}
+	if err := repo.CreateIndex(&ri.Definition{
+		Keys:   []ri.Key{{Field: payoutIdemField, Sort: ri.Asc}},
+		Unique: true,
+	}); err != nil {
+		logger.Error("Failed to create payouts idempotency index",
+			zap.Error(err), zap.String("index_field", payoutIdemField))
+		return nil, err
+	}
+
+	p := &Payouts{
+		logger:     logger,
+		repository: repo,
+	}
+	p.logger.Debug("Payouts store initialised")
+	return p, nil
+}
+
+func (p *Payouts) findOneByField(ctx context.Context, field, value string) (*model.CardPayout, error) {
+	var res model.CardPayout
+	return &res, p.repository.FindOneByFilter(ctx, repository.Filter(field, value), &res)
+}
+
+func (p *Payouts) FindByIdempotencyKey(ctx context.Context, key string) (*model.CardPayout, error) {
+	return p.findOneByField(ctx, payoutIdemField, key)
+}
+
+func (p *Payouts) FindByOperationRef(ctx context.Context, operationRef string) (*model.CardPayout, error) {
+	return p.findOneByField(ctx, payoutOpField, operationRef)
+}
+
+func (p *Payouts) FindByPaymentID(ctx context.Context, paymentID string) (*model.CardPayout, error) {
+	return p.findOneByField(ctx, payoutIdField, paymentID)
+}
+
+func (p *Payouts) Upsert(ctx context.Context, record *model.CardPayout) error {
+	if record == nil {
+		p.logger.Warn("Invalid argument provided: nil record")
+		return merrors.InvalidArgument("payout record is nil", "record")
+	}
+
+	record.OperationRef = strings.TrimSpace(record.OperationRef)
+	record.PaymentRef = strings.TrimSpace(record.PaymentRef)
+	record.CustomerID = strings.TrimSpace(record.CustomerID)
+	record.ProviderCode = strings.TrimSpace(record.ProviderCode)
+	record.ProviderPaymentID = strings.TrimSpace(record.ProviderPaymentID)
+
+	if record.OperationRef == "" {
+		p.logger.Warn("Invalid argument provided: operation reference missing")
+		return merrors.InvalidArgument("operation ref is required", "operation_ref")
+	}
+
+	if err := p.repository.Upsert(ctx, record); err != nil {
+		p.logger.Warn("Failed to upsert payout record", zap.Error(err), mzap.ObjRef("payout_ref", record.ID),
+			zap.String("operation_ref", record.OperationRef), zap.String("payment_ref", record.PaymentRef))
+		return err
+	}
+	return nil
+}
+
+var _ storage.PayoutsStore = (*Payouts)(nil)

api/gateway/aurora/storage/mongo/transaction.go

@@ -0,0 +1,38 @@
+package mongo
+
+import (
+	"context"
+
+	"github.com/tech/sendico/pkg/db/transaction"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+type mongoTransactionFactory struct {
+	client *mongo.Client
+}
+
+func (f *mongoTransactionFactory) CreateTransaction() transaction.Transaction {
+	return &mongoTransaction{client: f.client}
+}
+
+type mongoTransaction struct {
+	client *mongo.Client
+}
+
+func (t *mongoTransaction) Execute(ctx context.Context, cb transaction.Callback) (any, error) {
+	session, err := t.client.StartSession()
+	if err != nil {
+		return nil, err
+	}
+	defer session.EndSession(ctx)
+
+	run := func(sessCtx context.Context) (any, error) {
+		return cb(sessCtx)
+	}
+
+	return session.WithTransaction(ctx, run)
+}
+
+func newMongoTransactionFactory(client *mongo.Client) transaction.Factory {
+	return &mongoTransactionFactory{client: client}
+}