feat: add stargate server and client (#297)

Co-authored-by: Maxence Maireaux <maxence@maireaux.fr>

.pre-commit-config.yaml

@@ -1,5 +1,5 @@
 fail_fast: true
-exclude: ^sdks/|openapi/build|components/payments/client|components/auth/authclient|components/agent/chart|components/operator/helm
+exclude: ^sdks/|openapi/build|components/payments/client|components/auth/authclient|components/agent/chart|components/stargate/internal/api|components/operator/helm
 repos:
 - repo: https://github.com/pre-commit/pre-commit-hooks
   rev: v4.4.0
@@ -16,6 +16,7 @@ repos:
   - id: check-merge-conflict
   - id: check-symlinks
   - id: detect-aws-credentials
+    args: [--allow-missing-credentials]
   - id: detect-private-key
     exclude: components/operator/garden|components/auth/cmd/serve.go
 - repo: local

Taskfile.yaml

@@ -38,6 +38,7 @@ tasks:
       - task orchestration:tests
       - task payments:tests
       - task search:tests
+      - task stargate:tests
       - task wallets:tests
       - task webhooks:tests
       - task ledger:tests
@@ -51,6 +52,7 @@ tasks:
       - task orchestration:lint
       - task payments:lint
       - task search:lint
+      - task stargate:lint
       - task wallets:lint
       - task webhooks:lint
       - task ledger:lint
@@ -83,6 +85,9 @@ includes:
   wallets:
     taskfile: ./components/wallets
     dir: ./components/wallets
+  stargate:
+    taskfile: ./components/stargate
+    dir: ./components/stargate
   webhooks:
     taskfile: ./components/webhooks
     dir: ./components/webhooks

components/ledger/libs/api/utils.go

@@ -55,6 +55,13 @@ func Ok(w io.Writer, v any) {
 	}
 }
 
+func WriteResponse(w http.ResponseWriter, status int, body []byte) {
+	w.WriteHeader(status)
+	if _, err := w.Write(body); err != nil {
+		panic(err)
+	}
+}
+
 func RenderCursor[T any](w io.Writer, v Cursor[T]) {
 	if err := json.NewEncoder(w).Encode(BaseResponse[T]{
 		Cursor: &v,

components/ledger/libs/publish/cli.go

@@ -66,7 +66,7 @@ func CLIPublisherModule(v *viper.Viper, serviceName string) fx.Option {
 		// Currently don't expose http listener, so pass addr == ""
 		options = append(options, httpModule(""))
 	case v.GetBool(PublisherNatsEnabledFlag):
-		options = append(options, natsModule(
+		options = append(options, NatsModule(
 			v.GetString(PublisherNatsClientIDFlag), v.GetString(PublisherNatsURLFlag), serviceName))
 	case v.GetBool(PublisherKafkaEnabledFlag):
 		options = append(options,

components/ledger/libs/publish/module_test.go

@@ -137,7 +137,7 @@ func TestModule(t *testing.T) {
 				t.Cleanup(server.Shutdown)
 
 				return fx.Options(
-					natsModule("example", "nats://127.0.0.1:4322", "testing"),
+					NatsModule("example", "nats://127.0.0.1:4322", "testing"),
 				)
 			},
 			topicMapping: map[string]string{},

components/ledger/libs/publish/nats.go

@@ -7,18 +7,32 @@ import (
 	wNats "github.com/ThreeDotsLabs/watermill-nats/v2/pkg/nats"
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/nats-io/nats.go"
+	"github.com/pkg/errors"
 	"go.uber.org/fx"
 )
 
-func newNatsPublisher(logger watermill.LoggerAdapter, config wNats.PublisherConfig) (*wNats.Publisher, error) {
-	return wNats.NewPublisher(config, logger)
+func newNatsConn(config wNats.PublisherConfig) (*nats.Conn, error) {
+	if err := config.Validate(); err != nil {
+		return nil, err
+	}
+
+	conn, err := nats.Connect(config.URL, config.NatsOptions...)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot connect to nats-core")
+	}
+
+	return conn, nil
+}
+
+func newNatsPublisherWithConn(conn *nats.Conn, logger watermill.LoggerAdapter, config wNats.PublisherConfig) (*wNats.Publisher, error) {
+	return wNats.NewPublisherWithNatsConn(conn, config.GetPublisherPublishConfig(), logger)
 }
 
-func newNatsSubscriber(logger watermill.LoggerAdapter, config wNats.SubscriberConfig) (*wNats.Subscriber, error) {
-	return wNats.NewSubscriber(config, logger)
+func newNatsSubscriberWithConn(conn *nats.Conn, logger watermill.LoggerAdapter, config wNats.SubscriberConfig) (*wNats.Subscriber, error) {
+	return wNats.NewSubscriberWithNatsConn(conn, config.GetSubscriberSubscriptionConfig(), logger)
 }
 
-func natsModule(clientID, url, serviceName string) fx.Option {
+func NatsModule(clientID, url, serviceName string) fx.Option {
 	jetStreamConfig := wNats.JetStreamConfig{
 		AutoProvision: true,
 		DurablePrefix: serviceName,
@@ -27,20 +41,23 @@ func natsModule(clientID, url, serviceName string) fx.Option {
 		nats.Name(clientID),
 	}
 	return fx.Options(
-		fx.Provide(newNatsPublisher),
-		fx.Provide(newNatsSubscriber),
+		fx.Provide(newNatsConn),
+		fx.Provide(newNatsPublisherWithConn),
+		fx.Provide(newNatsSubscriberWithConn),
 		fx.Supply(wNats.PublisherConfig{
-			NatsOptions: natsOptions,
-			URL:         url,
-			Marshaler:   &wNats.NATSMarshaler{},
-			JetStream:   jetStreamConfig,
+			NatsOptions:       natsOptions,
+			URL:               url,
+			Marshaler:         &wNats.NATSMarshaler{},
+			JetStream:         jetStreamConfig,
+			SubjectCalculator: wNats.DefaultSubjectCalculator,
 		}),
 		fx.Supply(wNats.SubscriberConfig{
-			NatsOptions:      natsOptions,
-			Unmarshaler:      &wNats.NATSMarshaler{},
-			URL:              url,
-			QueueGroupPrefix: serviceName,
-			JetStream:        jetStreamConfig,
+			NatsOptions:       natsOptions,
+			Unmarshaler:       &wNats.NATSMarshaler{},
+			URL:               url,
+			QueueGroupPrefix:  serviceName,
+			JetStream:         jetStreamConfig,
+			SubjectCalculator: wNats.DefaultSubjectCalculator,
 		}),
 		fx.Provide(func(publisher *wNats.Publisher) message.Publisher {
 			return publisher

components/operator/PROJECT

@@ -40,7 +40,16 @@ resources:
   controller: true
   domain: formance.com
   group: components
-  kind: Payments
+  kind: Ledger
+  path: github.com/formancehq/operator/apis/components/v1beta3
+  version: v1beta3
+- api:
+    crdVersion: v1
+    namespaced: true
+  controller: true
+  domain: formance.com
+  group: components
+  kind: Stargate
   path: github.com/formancehq/operator/apis/components/v1beta3
   version: v1beta3
 - api:

components/operator/apis/stack/v1beta3/configuration_types.go

@@ -36,6 +36,9 @@ type ConfigurationServicesSpec struct {
 	Orchestration OrchestrationSpec `json:"orchestration,omitempty"`
 	Search        SearchSpec        `json:"search,omitempty"`
 	Auth          AuthSpec          `json:"auth,omitempty"`
+
+	// +optional
+	Stargate StargateSpec `json:"stargate,omitempty"`
 }
 
 func (in *ConfigurationServicesSpec) List() []string {

components/operator/apis/stack/v1beta3/stack_types.go

@@ -92,6 +92,13 @@ type StackAuthSpec struct {
 	StaticClients []StaticClient `json:"staticClients,omitempty"`
 }
 
+type StackStargateConfig struct {
+	StargateServerURL string `json:"stargateServerURL"`
+	AuthClientID      string `json:"authClientId"`
+	AuthClientSecret  string `json:"authClientSecret"`
+	AuthIssuerURL     string `json:"authIssuerURL"`
+}
+
 // StackSpec defines the desired state of Stack
 type StackSpec struct {
 	DevProperties `json:",inline"`
@@ -100,6 +107,9 @@ type StackSpec struct {
 	Host string        `json:"host"`
 	Auth StackAuthSpec `json:"auth"`
 
+	// +optional
+	Stargate *StackStargateConfig `json:"stargate"`
+
 	// +optional
 	Versions string `json:"versions"`
 

components/operator/apis/stack/v1beta3/stargate_types.go

@@ -0,0 +1,5 @@
+package v1beta3
+
+type StargateSpec struct {
+	DevProperties `json:",inline"`
+}

components/operator/apis/stack/v1beta3/versions_types.go

@@ -43,6 +43,8 @@ type VersionsSpec struct {
 	Orchestration string `json:"orchestration"`
 	// +optional
 	Gateway string `json:"gateway"`
+	// +optional
+	Stargate string `json:"stargate"`
 }
 
 // VersionsStatus defines the observed state of Versions