feat: clean connectors + add polling for connectors (#417)

components/operator/internal/controllers/stack/testdata/monopod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-post-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-post-upgrade
+    namespace: monopod-latest
+spec:
+    configuration: monopod-latest
+    module: payments
+    postUpgrade: true
+    targetedVersion: v0.7.0
+    version: monopod-latest
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-pre-upgrade
+    namespace: monopod-latest
+spec:
+    configuration: monopod-latest
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.7.0
+    version: monopod-latest
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-ledgerv1/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-post-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-post-upgrade
+    namespace: monopod-ledgerv1
+spec:
+    configuration: monopod-ledgerv1
+    module: payments
+    postUpgrade: true
+    targetedVersion: v0.7.0
+    version: monopod-ledgerv1
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-ledgerv1/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-pre-upgrade
+    namespace: monopod-ledgerv1
+spec:
+    configuration: monopod-ledgerv1
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.7.0
+    version: monopod-ledgerv1
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-search-before-v0.7.0/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-post-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-post-upgrade
+    namespace: monopod-search-before-v0-7-0
+spec:
+    configuration: monopod-search-before-v0-7-0
+    module: payments
+    postUpgrade: true
+    targetedVersion: v0.7.0
+    version: monopod-search-before-v0-7-0
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/monopod-search-before-v0.7.0/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-pre-upgrade
+    namespace: monopod-search-before-v0-7-0
+spec:
+    configuration: monopod-search-before-v0-7-0
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.7.0
+    version: monopod-search-before-v0-7-0
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/multipod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-post-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-post-upgrade
+    namespace: multipod-latest
+spec:
+    configuration: multipod-latest
+    module: payments
+    postUpgrade: true
+    targetedVersion: v0.7.0
+    version: multipod-latest
+status:
+    terminated: true

components/operator/internal/controllers/stack/testdata/multipod-latest/results/migrations-stack.formance.com-v1beta3/payments-v0.7.0-pre-upgrade.yaml

@@ -0,0 +1,18 @@
+apiVersion: stack.formance.com/v1beta3
+kind: Migration
+metadata:
+    annotations:
+        reloader.stakater.com/auto: "true"
+    generation: 1
+    labels:
+        stack: "true"
+    name: payments-v0.7.0-pre-upgrade
+    namespace: multipod-latest
+spec:
+    configuration: multipod-latest
+    module: payments
+    postUpgrade: false
+    targetedVersion: v0.7.0
+    version: multipod-latest
+status:
+    terminated: true

components/operator/internal/handlers/handler_payments.go

@@ -10,6 +10,7 @@ import (
 	"github.com/formancehq/payments/cmd"
 	"github.com/uptrace/bun"
 	"github.com/uptrace/bun/dialect/pgdialect"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 func init() {
@@ -62,24 +63,12 @@ func init() {
 					return paymentsPreUpgradeMigration(ctx)
 				},
 				PostUpgrade: func(ctx modules.PostInstallContext) error {
-					if err := resetConnectors(ctx, "stripe"); err != nil {
-						return err
-					}
-					if err := resetConnectors(ctx, "wise"); err != nil {
-						return err
-					}
-					if err := resetConnectors(ctx, "modulr"); err != nil {
-						return err
-					}
-					if err := resetConnectors(ctx, "banking-circle"); err != nil {
-						return err
-					}
-					if err := resetConnectors(ctx, "currency-cloud"); err != nil {
-						return err
-					}
-					if err := resetConnectors(ctx, "dummy-pay"); err != nil {
-						return err
-					}
+					resetConnectors(ctx, "stripe")
+					resetConnectors(ctx, "wise")
+					resetConnectors(ctx, "modulr")
+					resetConnectors(ctx, "banking-circle")
+					resetConnectors(ctx, "currency-cloud")
+					resetConnectors(ctx, "dummy-pay")
 					return nil
 				},
 				Services: func(ctx modules.ModuleContext) modules.Services {
@@ -102,6 +91,25 @@ func init() {
 					return paymentsServices(ctx, env)
 				},
 			},
+			"v0.7.0": {
+				PreUpgrade: func(ctx modules.Context) error {
+					return paymentsPreUpgradeMigration(ctx)
+				},
+				PostUpgrade: func(ctx modules.PostInstallContext) error {
+					resetConnectors(ctx, "stripe")
+					resetConnectors(ctx, "wise")
+					resetConnectors(ctx, "modulr")
+					resetConnectors(ctx, "banking-circle")
+					resetConnectors(ctx, "currency-cloud")
+					resetConnectors(ctx, "dummy-pay")
+					resetConnectors(ctx, "mangopay")
+					resetConnectors(ctx, "moneycorp")
+					return nil
+				},
+				Services: func(ctx modules.ModuleContext) modules.Services {
+					return paymentsServices(ctx, env)
+				},
+			},
 		},
 	})
 }
@@ -148,13 +156,17 @@ func paymentsServices(
 	}}
 }
 
-func resetConnectors(ctx modules.PostInstallContext, connector string) error {
+func resetConnectors(ctx modules.PostInstallContext, connector string) {
 	endpoint := fmt.Sprintf(
 		"http://payments.%s.svc:%d/connectors/%s/reset",
 		ctx.Stack.Name,
 		ctx.Stack.Status.Ports[ctx.ModuleName]["payments"],
 		connector,
 	)
 	_, err := http.Post(endpoint, "", nil)
-	return err
+	if err != nil {
+		logger := log.FromContext(ctx)
+		logger.WithValues("endpoint", endpoint).Error(err, "failed to reset connector")
+		// Do not return any error here, as the connector is not required to be installed
+	}
 }

components/payments/internal/app/connectors/bankingcircle/client/auth.go

@@ -0,0 +1,69 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"time"
+)
+
+func (c *Client) login(ctx context.Context) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet,
+		c.authorizationEndpoint+"/api/v1/authorizations/authorize", http.NoBody)
+	if err != nil {
+		return fmt.Errorf("failed to create login request: %w", err)
+	}
+
+	req.SetBasicAuth(c.username, c.password)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("failed to login: %w", err)
+	}
+
+	defer func() {
+		err = resp.Body.Close()
+		if err != nil {
+			c.logger.Error(err)
+		}
+	}()
+
+	responseBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read login response body: %w", err)
+	}
+
+	//nolint:tagliatelle // allow for client-side structures
+	type response struct {
+		AccessToken string `json:"access_token"`
+		ExpiresIn   string `json:"expires_in"`
+	}
+
+	var res response
+
+	if err = json.Unmarshal(responseBody, &res); err != nil {
+		return fmt.Errorf("failed to unmarshal login response: %w", err)
+	}
+
+	c.accessToken = res.AccessToken
+
+	expiresIn, err := strconv.Atoi(res.ExpiresIn)
+	if err != nil {
+		return fmt.Errorf("failed to convert expires_in to int: %w", err)
+	}
+
+	c.accessTokenExpiresAt = time.Now().Add(time.Duration(expiresIn) * time.Second)
+
+	return nil
+}
+
+func (c *Client) ensureAccessTokenIsValid(ctx context.Context) error {
+	if c.accessTokenExpiresAt.After(time.Now().Add(5 * time.Second)) {
+		return nil
+	}
+
+	return c.login(ctx)
+}

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

@@ -0,0 +1,71 @@
+package client
+
+import (
+	"context"
+	"crypto/tls"
+	"net/http"
+	"time"
+
+	"github.com/formancehq/stack/libs/go-libs/logging"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+)
+
+type Client struct {
+	httpClient *http.Client
+
+	username string
+	password string
+
+	endpoint              string
+	authorizationEndpoint string
+
+	logger logging.Logger
+
+	accessToken          string
+	accessTokenExpiresAt time.Time
+}
+
+func newHTTPClient(userCertificate, userCertificateKey string) (*http.Client, error) {
+	cert, err := tls.X509KeyPair([]byte(userCertificate), []byte(userCertificateKey))
+	if err != nil {
+		return nil, err
+	}
+
+	tr := http.DefaultTransport.(*http.Transport).Clone()
+	tr.TLSClientConfig = &tls.Config{
+		Certificates: []tls.Certificate{cert},
+	}
+
+	return &http.Client{
+		Timeout:   10 * time.Second,
+		Transport: otelhttp.NewTransport(tr),
+	}, nil
+}
+
+func NewClient(
+	username, password,
+	endpoint, authorizationEndpoint,
+	uCertificate, uCertificateKey string,
+	logger logging.Logger) (*Client, error) {
+	httpClient, err := newHTTPClient(uCertificate, uCertificateKey)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &Client{
+		httpClient: httpClient,
+
+		username:              username,
+		password:              password,
+		endpoint:              endpoint,
+		authorizationEndpoint: authorizationEndpoint,
+
+		logger: logger,
+	}
+
+	if err := c.login(context.TODO()); err != nil {
+		return nil, err
+	}
+
+	return c, nil
+}