feat: stripe transfers integration

internal/app/api/connectormodule.go

@@ -4,15 +4,13 @@ import (
 	"context"
 	"net/http"
 
+	"github.com/formancehq/go-libs/sharedlogging"
+	"github.com/formancehq/go-libs/sharedpublish"
 	"github.com/formancehq/payments/internal/pkg/ingestion"
 	"github.com/formancehq/payments/internal/pkg/integration"
 	"github.com/formancehq/payments/internal/pkg/payments"
 	"github.com/formancehq/payments/internal/pkg/task"
 	"github.com/formancehq/payments/internal/pkg/writeonly"
-	"github.com/gorilla/mux"
-
-	"github.com/formancehq/go-libs/sharedlogging"
-	"github.com/formancehq/go-libs/sharedpublish"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.uber.org/dig"
 	"go.uber.org/fx"
@@ -87,24 +85,3 @@ func addConnector[
 		}),
 	)
 }
-
-func connectorRouter[Config payments.ConnectorConfigObject, Descriptor payments.TaskDescriptor](
-	name string,
-	manager *integration.ConnectorManager[Config, Descriptor],
-) *mux.Router {
-	r := mux.NewRouter()
-
-	r.Path("/" + name).Methods(http.MethodPost).Handler(install(manager))
-
-	r.Path("/" + name + "/reset").Methods(http.MethodPost).Handler(reset(manager))
-
-	r.Path("/" + name).Methods(http.MethodDelete).Handler(uninstall(manager))
-
-	r.Path("/" + name + "/config").Methods(http.MethodGet).Handler(readConfig(manager))
-
-	r.Path("/" + name + "/tasks").Methods(http.MethodGet).Handler(listTasks(manager))
-
-	r.Path("/" + name + "/tasks/{taskID}").Methods(http.MethodGet).Handler(readTask(manager))
-
-	return r
-}

internal/app/api/module.go

@@ -26,7 +26,6 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"go.mongodb.org/mongo-driver/mongo"
-	"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
 	"go.uber.org/fx"
 )
 
@@ -77,42 +76,6 @@ func HTTPModule() fx.Option {
 	)
 }
 
-func httpRouter(db *mongo.Database, client *mongo.Client, handlers []connectorHandler) (*mux.Router, error) {
-	rootMux := mux.NewRouter()
-
-	if viper.GetBool(otelTracesFlag) {
-		rootMux.Use(otelmux.Middleware(serviceName))
-	}
-
-	rootMux.Use(recoveryHandler(httpRecoveryFunc))
-	rootMux.Use(httpCorsHandler())
-	rootMux.Use(httpServeFunc)
-
-	rootMux.Path("/_health").Handler(healthHandler(client))
-	rootMux.Path("/_live").Handler(liveHandler())
-
-	authGroup := rootMux.Name("authenticated").Subrouter()
-
-	if methods := sharedAuthMethods(); len(methods) > 0 {
-		authGroup.Use(sharedauth.Middleware(methods...))
-	}
-
-	authGroup.HandleFunc("/connectors", readConnectorsHandler(db))
-	connectorGroup := authGroup.PathPrefix("/connectors").Subrouter()
-
-	connectorGroup.Path("/configs").Handler(connectorConfigsHandler())
-
-	for _, h := range handlers {
-		connectorGroup.PathPrefix("/" + h.Name).Handler(
-			http.StripPrefix("/connectors", h.Handler),
-		)
-	}
-
-	authGroup.PathPrefix("/").Handler(paymentsRouter(db))
-
-	return rootMux, nil
-}
-
 func httpRecoveryFunc(ctx context.Context, e interface{}) {
 	if viper.GetBool(otelTracesFlag) {
 		sharedotlp.RecordAsError(ctx, e)

internal/app/api/router.go

@@ -0,0 +1,76 @@
+package api
+
+import (
+	"net/http"
+
+	"github.com/formancehq/payments/internal/pkg/integration"
+	"github.com/formancehq/payments/internal/pkg/payments"
+
+	"github.com/formancehq/go-libs/sharedauth"
+	"github.com/gorilla/mux"
+	"github.com/spf13/viper"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/gorilla/mux/otelmux"
+)
+
+func httpRouter(db *mongo.Database, client *mongo.Client, connectorHandlers []connectorHandler) (*mux.Router, error) {
+	rootMux := mux.NewRouter()
+
+	if viper.GetBool(otelTracesFlag) {
+		rootMux.Use(otelmux.Middleware(serviceName))
+	}
+
+	rootMux.Use(recoveryHandler(httpRecoveryFunc))
+	rootMux.Use(httpCorsHandler())
+	rootMux.Use(httpServeFunc)
+
+	rootMux.Path("/_health").Handler(healthHandler(client))
+	rootMux.Path("/_live").Handler(liveHandler())
+
+	authGroup := rootMux.Name("authenticated").Subrouter()
+
+	if methods := sharedAuthMethods(); len(methods) > 0 {
+		authGroup.Use(sharedauth.Middleware(methods...))
+	}
+
+	authGroup.HandleFunc("/connectors", readConnectorsHandler(db))
+	connectorGroup := authGroup.PathPrefix("/connectors").Subrouter()
+
+	connectorGroup.Path("/configs").Handler(connectorConfigsHandler())
+
+	for _, h := range connectorHandlers {
+		connectorGroup.PathPrefix("/" + h.Name).Handler(
+			http.StripPrefix("/connectors", h.Handler),
+		)
+	}
+
+	// TODO: It's not ideal to define it explicitly here
+	// Refactor it when refactoring the HTTP lib.
+	connectorGroup.Path("/stripe/transfers").Methods(http.MethodPost).
+		Handler(handleStripeTransfers(db))
+
+	authGroup.PathPrefix("/").Handler(paymentsRouter(db))
+
+	return rootMux, nil
+}
+
+func connectorRouter[Config payments.ConnectorConfigObject, Descriptor payments.TaskDescriptor](
+	name string,
+	manager *integration.ConnectorManager[Config, Descriptor],
+) *mux.Router {
+	r := mux.NewRouter()
+
+	r.Path("/" + name).Methods(http.MethodPost).Handler(install(manager))
+
+	r.Path("/" + name + "/reset").Methods(http.MethodPost).Handler(reset(manager))
+
+	r.Path("/" + name).Methods(http.MethodDelete).Handler(uninstall(manager))
+
+	r.Path("/" + name + "/config").Methods(http.MethodGet).Handler(readConfig(manager))
+
+	r.Path("/" + name + "/tasks").Methods(http.MethodGet).Handler(listTasks(manager))
+
+	r.Path("/" + name + "/tasks/{taskID}").Methods(http.MethodGet).Handler(readTask(manager))
+
+	return r
+}

internal/app/api/stripe.go

@@ -0,0 +1,109 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/formancehq/payments/internal/pkg/integration"
+
+	"go.mongodb.org/mongo-driver/mongo"
+
+	"github.com/pkg/errors"
+
+	stripeConnector "github.com/formancehq/payments/internal/pkg/connectors/stripe"
+	"github.com/stripe/stripe-go/v72"
+	"github.com/stripe/stripe-go/v72/transfer"
+)
+
+type stripeTransferRequest struct {
+	Amount      int64             `json:"amount"`
+	Asset       string            `json:"asset"`
+	Destination string            `json:"destination"`
+	Metadata    map[string]string `json:"metadata"`
+
+	currency string
+}
+
+func (req *stripeTransferRequest) validate() error {
+	if req.Amount <= 0 {
+		return errors.New("amount must be greater than 0")
+	}
+
+	if req.Asset == "" {
+		return errors.New("asset is required")
+	}
+
+	if req.Asset != "USD/2" && req.Asset != "EUR/2" {
+		return errors.New("asset must be USD/2 or EUR/2")
+	}
+
+	req.currency = req.Asset[:3]
+
+	if req.Destination == "" {
+		return errors.New("destination is required")
+	}
+
+	return nil
+}
+
+func handleStripeTransfers(db *mongo.Database) http.HandlerFunc {
+	connectorStore := integration.NewMongoDBConnectorStore(db)
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		var cfg stripeConnector.Config
+
+		err := connectorStore.ReadConfig(r.Context(), stripeConnector.Name, &cfg)
+		if err != nil {
+			handleError(w, r, err)
+
+			return
+		}
+
+		stripe.Key = cfg.APIKey
+
+		var transferRequest stripeTransferRequest
+
+		err = json.NewDecoder(r.Body).Decode(&transferRequest)
+		if err != nil {
+			handleError(w, r, err)
+
+			return
+		}
+
+		err = transferRequest.validate()
+		if err != nil {
+			handleError(w, r, err)
+
+			return
+		}
+
+		params := &stripe.TransferParams{
+			Params: stripe.Params{
+				Context: r.Context(),
+			},
+			Amount:      stripe.Int64(transferRequest.Amount),
+			Currency:    stripe.String(transferRequest.currency),
+			Destination: stripe.String(transferRequest.Destination),
+		}
+
+		for k, v := range transferRequest.Metadata {
+			params.AddMetadata(k, v)
+		}
+
+		transferResponse, err := transfer.New(params)
+		if err != nil {
+			handleServerError(w, r, err)
+
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+
+		err = json.NewEncoder(w).Encode(transferResponse)
+		if err != nil {
+			handleServerError(w, r, err)
+
+			return
+		}
+	}
+}

internal/pkg/connectors/bankingcircle/config.go

@@ -37,5 +37,5 @@ func (c Config) BuildTemplate() (string, configtemplate.Config) {
 	cfg.AddParameter("endpoint", configtemplate.TypeString, true)
 	cfg.AddParameter("authorizationEndpoint", configtemplate.TypeString, true)
 
-	return connectorName, cfg
+	return Name, cfg
 }

internal/pkg/connectors/bankingcircle/loader.go

@@ -6,11 +6,11 @@ import (
 	"github.com/formancehq/payments/internal/pkg/task"
 )
 
-const connectorName = "bankingcircle"
+const Name = "bankingcircle"
 
 // NewLoader creates a new loader.
 func NewLoader() integration.Loader[Config, TaskDescriptor] {
-	loader := integration.NewLoaderBuilder[Config, TaskDescriptor](connectorName).
+	loader := integration.NewLoaderBuilder[Config, TaskDescriptor](Name).
 		WithLoad(func(logger sharedlogging.Logger, config Config) integration.Connector[TaskDescriptor] {
 			return integration.NewConnectorBuilder[TaskDescriptor]().
 				WithInstall(func(ctx task.ConnectorContext[TaskDescriptor]) error {

internal/pkg/connectors/currencycloud/config.go

@@ -78,5 +78,5 @@ func (c Config) BuildTemplate() (string, configtemplate.Config) {
 	cfg.AddParameter("endpoint", configtemplate.TypeString, false)
 	cfg.AddParameter("pollingPeriod", configtemplate.TypeDurationNs, true)
 
-	return connectorName, cfg
+	return Name, cfg
 }

internal/pkg/connectors/currencycloud/connector.go

@@ -8,7 +8,7 @@ import (
 	"github.com/formancehq/payments/internal/pkg/task"
 )
 
-const connectorName = "currencycloud"
+const Name = "currencycloud"
 
 type Connector struct {
 	logger sharedlogging.Logger

internal/pkg/connectors/currencycloud/loader.go

@@ -14,7 +14,7 @@ func (l *Loader) AllowTasks() int {
 }
 
 func (l *Loader) Name() string {
-	return connectorName
+	return Name
 }
 
 func (l *Loader) Load(logger sharedlogging.Logger, config Config) integration.Connector[TaskDescriptor] {

internal/pkg/connectors/dummypay/config.go

@@ -55,5 +55,5 @@ func (c Config) BuildTemplate() (string, configtemplate.Config) {
 	cfg.AddParameter("filePollingPeriod", configtemplate.TypeDurationNs, true)
 	cfg.AddParameter("fileGenerationPeriod", configtemplate.TypeDurationNs, false)
 
-	return connectorName, cfg
+	return Name, cfg
 }

internal/pkg/connectors/dummypay/connector.go

@@ -9,8 +9,8 @@ import (
 	"github.com/formancehq/go-libs/sharedlogging"
 )
 
-// connectorName is the name of the connector.
-const connectorName = "dummypay"
+// Name is the name of the connector.
+const Name = "dummypay"
 
 // Connector is the connector for the dummy payment connector.
 type Connector struct {

internal/pkg/connectors/dummypay/loader.go

@@ -12,7 +12,7 @@ type Loader struct{}
 
 // Name returns the name of the connector.
 func (l *Loader) Name() string {
-	return connectorName
+	return Name
 }
 
 // AllowTasks returns the amount of tasks that are allowed to be scheduled.

internal/pkg/connectors/dummypay/loader_test.go

@@ -19,7 +19,7 @@ func TestLoader(t *testing.T) {
 
 	loader := NewLoader()
 
-	assert.Equal(t, connectorName, loader.Name())
+	assert.Equal(t, Name, loader.Name())
 	assert.Equal(t, 10, loader.AllowTasks())
 	assert.Equal(t, Config{
 		FilePollingPeriod:    connectors.Duration{Duration: 10 * time.Second},

internal/pkg/connectors/modulr/config.go

@@ -27,5 +27,5 @@ func (c Config) BuildTemplate() (string, configtemplate.Config) {
 	cfg.AddParameter("apiSecret", configtemplate.TypeString, true)
 	cfg.AddParameter("endpoint", configtemplate.TypeString, false)
 
-	return connectorName, cfg
+	return Name, cfg
 }

internal/pkg/connectors/modulr/connector.go

@@ -8,7 +8,7 @@ import (
 	"github.com/formancehq/payments/internal/pkg/task"
 )
 
-const connectorName = "modulr"
+const Name = "modulr"
 
 type Connector struct {
 	logger sharedlogging.Logger

internal/pkg/connectors/modulr/loader.go

@@ -14,7 +14,7 @@ func (l *Loader) AllowTasks() int {
 }
 
 func (l *Loader) Name() string {
-	return connectorName
+	return Name
 }
 
 func (l *Loader) Load(logger sharedlogging.Logger, config Config) integration.Connector[TaskDescriptor] {