refactor: cleanup (#55)

Co-authored-by: Dimy Jeannot <>

Author: dimyjeannot

apps/connectors/poc/network-account/v1alpha/listeners/create.go

@@ -16,6 +16,7 @@ import (
 	typev2pb "github.com/openecosystems/ecosystem/libs/protobuf/go/protobuf/gen/platform/type/v2"
 	ecosystemv2alphapbmodel "github.com/openecosystems/ecosystem/libs/public/go/model/gen/platform/ecosystem/v2alpha"
 	ecosystemv2alphapb "github.com/openecosystems/ecosystem/libs/public/go/sdk/gen/platform/ecosystem/v2alpha"
+	"github.com/openecosystems/ecosystem/libs/public/go/sdk/gen/platform/ecosystem/v2alpha/ecosystemv2alphapbconnect"
 	sdkv2alphalib "github.com/openecosystems/ecosystem/libs/public/go/sdk/v2alpha"
 )
 
@@ -24,16 +25,10 @@ type CreateEcosystemListener struct{}
 
 // GetConfiguration returns the listener configuration for the CreateEcosystemListener, including entity, subject, and queue details.
 func (l *CreateEcosystemListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &ecosystemv2alphapbmodel.EcosystemSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.CommandTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
 	return &natsnodev1.ListenerConfiguration{
 		Entity:     &ecosystemv2alphapbmodel.EcosystemSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
 		StreamType: &natsnodev1.InboundStream{},
+		Procedure:  ecosystemv2alphapbconnect.EcosystemServiceCreateEcosystemProcedure,
 		JetstreamConfiguration: &jetstream.ConsumerConfig{
 			Durable:       "ecosystem-createEcosystem",
 			AckPolicy:     jetstream.AckExplicitPolicy,
@@ -46,7 +41,7 @@ func (l *CreateEcosystemListener) GetConfiguration() *natsnodev1.ListenerConfigu
 
 // Listen starts the listener to process multiplexed spec events synchronously based on the provided context and configuration.
 func (l *CreateEcosystemListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles incoming listener messages to create and store a configuration, ensuring required fields are validated.
@@ -98,5 +93,5 @@ func (l *CreateEcosystemListener) Process(ctx context.Context, request *natsnode
 	}
 	log.Info("Create Ecosystem Response", zap.Any("id", response.Ecosystem.Id))
 
-	natsnodev1.RespondToSyncCommand(ctx, request, &response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, &response)
 }

apps/workloads/public/ecosystem/v2alpha/ecosystem/certificate/sign.go

@@ -3,8 +3,6 @@ package certificate
 import (
 	"context"
 
-	"github.com/nats-io/nats.go/jetstream"
-
 	natsnodev1 "github.com/openecosystems/ecosystem/libs/partner/go/nats"
 	zaploggerv1 "github.com/openecosystems/ecosystem/libs/partner/go/zap"
 	specv2pb "github.com/openecosystems/ecosystem/libs/protobuf/go/protobuf/gen/platform/spec/v2"
@@ -18,29 +16,13 @@ type SignCertificateListener struct{}
 
 // GetConfiguration provides the listener configuration for SignCertificateListener, including subject, queue, and jetstream settings.
 func (l *SignCertificateListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &cryptographyv2alphapb.CertificateSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.CommandTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
-	return &natsnodev1.ListenerConfiguration{
-		Entity:     &cryptographyv2alphapb.CertificateSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
-		StreamType: &natsnodev1.InboundStream{},
-		JetstreamConfiguration: &jetstream.ConsumerConfig{
-			Durable:       "cryptography-signCertificate",
-			AckPolicy:     jetstream.AckExplicitPolicy,
-			MemoryStorage: false,
-			FilterSubject: "inbound-certificate.data.command",
-			Metadata:      nil,
-		},
-	}
+	handler := cryptographyv2alphapb.CertificateServiceHandler{}
+	return handler.GetSignCertificateConfiguration()
 }
 
 // Listen synchronously listens for multiplexed spec events and routes them to the associated handler.
 func (l *SignCertificateListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles the incoming ListenerMessage, processes the request, and sends an appropriate response back to the client.
@@ -81,5 +63,5 @@ func (l *SignCertificateListener) Process(ctx context.Context, request *natsnode
 
 	// log.Info("Signed certificate successfully: " + response.Certificate.Id)
 
-	natsnodev1.RespondToSyncCommand(ctx, request, &response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, &response)
 }

apps/workloads/public/ecosystem/v2alpha/ecosystem/configuration/create.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/nats-io/nats.go/jetstream"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -26,29 +25,13 @@ type CreateConfigurationListener struct{}
 
 // GetConfiguration returns the ListenerConfiguration for CreateConfigurationListener, defining subject, queue, entity, and stream settings.
 func (l *CreateConfigurationListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &configurationv2alphapb.ConfigurationSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.CommandTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
-	return &natsnodev1.ListenerConfiguration{
-		Entity:     &configurationv2alphapb.ConfigurationSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
-		StreamType: &natsnodev1.InboundStream{},
-		JetstreamConfiguration: &jetstream.ConsumerConfig{
-			Durable:       "configuration-createConfiguration",
-			AckPolicy:     jetstream.AckExplicitPolicy,
-			MemoryStorage: false,
-			FilterSubject: "inbound-configuration.data.command",
-			Metadata:      nil,
-		},
-	}
+	handler := configurationv2alphapb.ConfigurationServiceHandler{}
+	return handler.GetCreateConfigurationConfiguration()
 }
 
 // Listen starts the listener to process multiplexed spec events synchronously based on the provided context and configuration.
 func (l *CreateConfigurationListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles incoming listener messages to create and store a configuration, ensuring required fields are validated.
@@ -103,5 +86,5 @@ func (l *CreateConfigurationListener) Process(ctx context.Context, request *nats
 	}
 	log.Info("Create Configuration Response", zap.Any("id", response.Configuration.Id))
 
-	natsnodev1.RespondToSyncCommand(ctx, request, &response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, &response)
 }

apps/workloads/public/ecosystem/v2alpha/ecosystem/configuration/get.go

@@ -3,7 +3,6 @@ package configuration
 import (
 	"context"
 
-	"github.com/nats-io/nats.go/jetstream"
 	"go.uber.org/zap"
 
 	configurationv2alphalib "github.com/openecosystems/ecosystem/libs/partner/go/configuration/v2alpha"
@@ -20,33 +19,13 @@ type GetConfigurationListener struct{}
 
 // GetConfiguration creates and returns a ListenerConfiguration for the GetConfigurationListener.
 func (l *GetConfigurationListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &configurationv2alphapb.ConfigurationSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.EventTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
-	return &natsnodev1.ListenerConfiguration{
-		Entity:     &configurationv2alphapb.ConfigurationSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
-		StreamType: &natsnodev1.InboundStream{},
-		JetstreamConfiguration: &jetstream.ConsumerConfig{
-			Durable: "configuration-getConfiguration",
-			//Durable: natsnodev1.GetListenerGroup(
-			//	&configurationv2alphapb.ConfigurationSpecEntity{},
-			//	&configurationv2alphapb.ConfigurationSpecEntity{},
-			//),
-			AckPolicy:     jetstream.AckExplicitPolicy,
-			MemoryStorage: false,
-			FilterSubject: "inbound-configuration.data.event",
-			Metadata:      nil,
-		},
-	}
+	handler := configurationv2alphapb.ConfigurationServiceHandler{}
+	return handler.GetGetConfigurationConfiguration()
 }
 
 // Listen subscribes the listener to a NATS subject to process multiplexed specification events synchronously.
 func (l *GetConfigurationListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles incoming listener messages, validates the request, retrieves platform configurations, and sends a response.
@@ -82,5 +61,5 @@ func (l *GetConfigurationListener) Process(ctx context.Context, request *natsnod
 
 	log.Info("Get Configuration Response", zap.Any("response", response))
 
-	natsnodev1.RespondToSyncCommand(ctx, request, response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, response)
 }

apps/workloads/public/ecosystem/v2alpha/ecosystem/ecosystem/create.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/nats-io/nats.go/jetstream"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -23,29 +22,13 @@ type CreateEcosystemListener struct{}
 
 // GetConfiguration returns the listener configuration for the CreateEcosystemListener, including entity, subject, and queue details.
 func (l *CreateEcosystemListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &ecosystemv2alphapb.EcosystemSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.CommandTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
-	return &natsnodev1.ListenerConfiguration{
-		Entity:     &ecosystemv2alphapb.EcosystemSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
-		StreamType: &natsnodev1.InboundStream{},
-		JetstreamConfiguration: &jetstream.ConsumerConfig{
-			Durable:       "ecosystem-createEcosystem",
-			AckPolicy:     jetstream.AckExplicitPolicy,
-			MemoryStorage: false,
-			FilterSubject: "inbound-ecosystem.data.command",
-			Metadata:      nil,
-		},
-	}
+	handler := ecosystemv2alphapb.EcosystemServiceHandler{}
+	return handler.GetCreateEcosystemConfiguration()
 }
 
 // Listen starts the listener to process multiplexed spec events synchronously based on the provided context and configuration.
 func (l *CreateEcosystemListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles incoming listener messages to create and store a configuration, ensuring required fields are validated.
@@ -97,5 +80,5 @@ func (l *CreateEcosystemListener) Process(ctx context.Context, request *natsnode
 	}
 	log.Info("Create Ecosystem Response", zap.Any("id", response.Ecosystem.Id))
 
-	natsnodev1.RespondToSyncCommand(ctx, request, &response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, &response)
 }

apps/workloads/public/ecosystem/v2alpha/ecosystem/iam/create.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/nats-io/nats.go/jetstream"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -22,29 +21,13 @@ type CreateAccountListener struct{}
 
 // GetConfiguration returns the listener configuration for the CreateAccountListener, including entity, subject, and queue details.
 func (l *CreateAccountListener) GetConfiguration() *natsnodev1.ListenerConfiguration {
-	entity := &iamv2alphapb.AccountSpecEntity{}
-	streamType := natsnodev1.InboundStream{}
-	subject := natsnodev1.GetMultiplexedRequestSubjectName(streamType.StreamPrefix(), entity.CommandTopic())
-	queue := natsnodev1.GetQueueGroupName(streamType.StreamPrefix(), entity.TypeName())
-
-	return &natsnodev1.ListenerConfiguration{
-		Entity:     &iamv2alphapb.AccountSpecEntity{},
-		Subject:    subject,
-		Queue:      queue,
-		StreamType: &natsnodev1.InboundStream{},
-		JetstreamConfiguration: &jetstream.ConsumerConfig{
-			Durable:       "iam-createAccount",
-			AckPolicy:     jetstream.AckExplicitPolicy,
-			MemoryStorage: false,
-			FilterSubject: "inbound-iam.data.command",
-			Metadata:      nil,
-		},
-	}
+	handler := iamv2alphapb.AccountAuthorityServiceHandler{}
+	return handler.GetCreateAccountAuthorityConfiguration()
 }
 
 // Listen starts the listener to process multiplexed spec events synchronously based on the provided context and configuration.
 func (l *CreateAccountListener) Listen(ctx context.Context, _ chan sdkv2alphalib.SpecListenableErr) {
-	natsnodev1.ListenForMultiplexedSpecEventsSync(ctx, l)
+	natsnodev1.ListenForMultiplexedRequests(ctx, l)
 }
 
 // Process handles incoming listener messages to create and store a configuration, ensuring required fields are validated.
@@ -61,7 +44,6 @@ func (l *CreateAccountListener) Process(ctx context.Context, request *natsnodev1
 		return
 	}
 
-	fmt.Println("REEEEEUUUUUU", req) //nolint:copylocks
 	if req.Name == "" {
 		fmt.Println("Name is required")
 		return
@@ -100,5 +82,5 @@ func (l *CreateAccountListener) Process(ctx context.Context, request *natsnodev1
 	}
 	log.Info("Create Account Response", zap.Any("id", response.Account.Id))
 
-	natsnodev1.RespondToSyncCommand(ctx, request, &response)
+	natsnodev1.RespondToMultiplexedRequest(ctx, request, &response)
 }