payments: add modulr transfer and payouts

components/payments/internal/app/connectors/mangopay/client/transfer.go

@@ -50,7 +50,7 @@ func (c *Client) InitiateWalletTransfer(ctx context.Context, transferRequest *Tr
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewBuffer(body))
 	if err != nil {
-		return nil, fmt.Errorf("failed to create login request: %w", err)
+		return nil, fmt.Errorf("failed to create transfer request: %w", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
 

components/payments/internal/app/connectors/mangopay/task_payments.go

@@ -42,6 +42,8 @@ func taskInitiatePayment(logger logging.Logger, mangopayClient *client.Client, t
 		var err error
 		defer func() {
 			if err != nil {
+				ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+				defer cancel()
 				metricsRegistry.ConnectorObjectsErrors().Add(ctx, 1, attrs)
 				if err := ingester.UpdateTransferInitiationStatus(ctx, transferInitiationID, models.TransferInitiationStatusFailed, err.Error(), time.Now()); err != nil {
 					logger.Error("failed to update transfer initiation status: %v", err)

components/payments/internal/app/connectors/modulr/client/accounts.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"encoding/json"
-	"fmt"
 	"net/http"
 )
 
@@ -28,11 +27,10 @@ func (m *Client) GetAccounts() ([]*Account, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res responseWrapper[[]*Account]

components/payments/internal/app/connectors/modulr/client/beneficiaries.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"encoding/json"
-	"fmt"
 	"net/http"
 )
 
@@ -21,7 +20,7 @@ func (m *Client) GetBeneficiaries() ([]*Beneficiary, error) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res responseWrapper[[]*Beneficiary]

components/payments/internal/app/connectors/modulr/client/client.go

@@ -1,8 +1,10 @@
 package client
 
 import (
+	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/formancehq/payments/internal/app/connectors/modulr/hmac"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
@@ -34,7 +36,8 @@ type Client struct {
 }
 
 func (m *Client) buildEndpoint(path string, args ...interface{}) string {
-	return fmt.Sprintf("%s/%s", m.endpoint, fmt.Sprintf(path, args...))
+	endpoint := strings.TrimSuffix(m.endpoint, "/")
+	return fmt.Sprintf("%s/%s", endpoint, fmt.Sprintf(path, args...))
 }
 
 const sandboxAPIEndpoint = "https://api-sandbox.modulrfinance.com/api-sandbox-token"
@@ -60,3 +63,20 @@ func NewClient(apiKey, apiSecret, endpoint string) (*Client, error) {
 		endpoint: endpoint,
 	}, nil
 }
+
+type ErrorResponse struct {
+	Field         string `json:"field"`
+	Code          string `json:"code"`
+	Message       string `json:"message"`
+	ErrorCode     string `json:"errorCode"`
+	SourceService string `json:"sourceService"`
+}
+
+func getError(resp *http.Response) error {
+	var errorResponse ErrorResponse
+	if err := json.NewDecoder(resp.Body).Decode(&errorResponse); err != nil {
+		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
+
+	return fmt.Errorf("%s: %s", errorResponse.ErrorCode, errorResponse.Message)
+}

components/payments/internal/app/connectors/modulr/client/payout.go

@@ -3,7 +3,6 @@ package client
 import (
 	"bytes"
 	"encoding/json"
-	"fmt"
 	"math/big"
 	"net/http"
 )
@@ -14,10 +13,10 @@ type PayoutRequest struct {
 		Type string `json:"type"`
 		ID   string `json:"id"`
 	} `json:"destination"`
-	Currency          string   `json:"currency"`
-	Amount            *big.Int `json:"amount"`
-	Reference         string   `json:"reference"`
-	ExternalReference string   `json:"externalReference"`
+	Currency          string     `json:"currency"`
+	Amount            *big.Float `json:"amount"`
+	Reference         string     `json:"reference"`
+	ExternalReference string     `json:"externalReference"`
 }
 
 type PayoutResponse struct {
@@ -42,7 +41,7 @@ func (c *Client) InitiatePayout(payoutRequest *PayoutRequest) (*PayoutResponse,
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusCreated {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res PayoutResponse
@@ -61,7 +60,7 @@ func (c *Client) GetPayout(payoutID string) (*PayoutResponse, error) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res PayoutResponse

components/payments/internal/app/connectors/modulr/client/transactions.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"encoding/json"
-	"fmt"
 	"net/http"
 )
 
@@ -29,7 +28,7 @@ func (m *Client) GetTransactions(accountID string) ([]Transaction, error) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res responseWrapper[[]Transaction]

components/payments/internal/app/connectors/modulr/client/transfer.go

@@ -8,16 +8,30 @@ import (
 	"net/http"
 )
 
+type DestinationType string
+
+const (
+	DestinationTypeAccount     DestinationType = "ACCOUNT"
+	DestinationTypeBeneficiary DestinationType = "BENEFICIARY"
+)
+
+type Destination struct {
+	Type string `json:"type"`
+	ID   string `json:"id"`
+}
+
 type TransferRequest struct {
-	SourceAccountID string `json:"sourceAccountId"`
-	Destination     struct {
-		Type string `json:"type"`
-		ID   string `json:"id"`
-	} `json:"destination"`
-	Currency          string   `json:"currency"`
-	Amount            *big.Int `json:"amount"`
-	Reference         string   `json:"reference"`
-	ExternalReference string   `json:"externalReference"`
+	SourceAccountID   string      `json:"sourceAccountId"`
+	Destination       Destination `json:"destination"`
+	Currency          string      `json:"currency"`
+	Amount            *big.Float  `json:"amount"`
+	Reference         string      `json:"reference"`
+	ExternalReference string      `json:"externalReference"`
+	PaymentDate       string      `json:"paymentDate"`
+}
+
+type getTransferResponse struct {
+	Content []*TransferResponse `json:"content"`
 }
 
 type TransferResponse struct {
@@ -35,14 +49,20 @@ func (c *Client) InitiateTransfer(transferRequest *TransferRequest) (*TransferRe
 		return nil, err
 	}
 
-	resp, err := c.httpClient.Post(c.buildEndpoint("payments"), "application/json", bytes.NewBuffer(body))
+	req, err := http.NewRequest(http.MethodPost, c.buildEndpoint("payments"), bytes.NewBuffer(body))
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to create transfer request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initiate transfer: %w", err)
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusCreated {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
 	var res TransferResponse
@@ -61,13 +81,17 @@ func (c *Client) GetTransfer(transferID string) (*TransferResponse, error) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, getError(resp)
 	}
 
-	var res TransferResponse
+	var res getTransferResponse
 	if err = json.NewDecoder(resp.Body).Decode(&res); err != nil {
 		return nil, err
 	}
 
-	return &res, nil
+	if len(res.Content) == 0 {
+		return nil, fmt.Errorf("transfer not found")
+	}
+
+	return res.Content[0], nil
 }

components/payments/internal/app/connectors/modulr/connector.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/formancehq/payments/internal/app/integration"
 	"github.com/formancehq/payments/internal/app/task"
+	"github.com/formancehq/stack/libs/go-libs/contextutil"
 	"github.com/formancehq/stack/libs/go-libs/logging"
 )
 
@@ -26,8 +27,31 @@ type Connector struct {
 }
 
 func (c *Connector) InitiatePayment(ctx task.ConnectorContext, transfer *models.TransferInitiation) error {
-	// TODO implement me
-	return errors.New("not implemented")
+	// Detach the context since we're launching an async task and we're mostly
+	// coming from a HTTP request.
+	detachedCtx, _ := contextutil.Detached(ctx.Context())
+	taskDescriptor, err := models.EncodeTaskDescriptor(TaskDescriptor{
+		Name:       "Initiate payment",
+		Key:        taskNameInitiatePayment,
+		TransferID: transfer.ID.String(),
+	})
+	if err != nil {
+		return err
+	}
+
+	err = ctx.Scheduler().Schedule(detachedCtx, taskDescriptor, models.TaskSchedulerOptions{
+		// We want to polling every c.cfg.PollingPeriod.Duration seconds the users
+		// and their transactions.
+		ScheduleOption: models.OPTIONS_RUN_NOW,
+		// No need to restart this task, since the connector is not existing or
+		// was uninstalled previously, the task does not exists in the database
+		Restart: true,
+	})
+	if err != nil && !errors.Is(err, task.ErrAlreadyScheduled) {
+		return err
+	}
+
+	return nil
 }
 
 func (c *Connector) Install(ctx task.ConnectorContext) error {