fix: Hardening 2 (#79)

* fix: added additional logging

* fix: added additional logging

* fix: added additional bad request error handling

---------

Co-authored-by: Dimy Jeannot <>

Author: dimyjeannot

go/oeco-sdk/v2beta/CHANGELOG.md

@@ -2,11 +2,21 @@
 
 ### 🩹 Fixes
 
-- added fail early if proto transcoding fails ([#78](https://github.com/openecosystems/ecosystem/pull/78))
+-   added fail early if proto transcoding fails ([#78](https://github.com/openecosystems/ecosystem/pull/78))
 
 ### ❤️ Thank You
 
-- Dimy Jeannot @dimyjeannot
+-   Dimy Jeannot @dimyjeannot
+
+## 0.20.2 (2025-09-11)
+
+### 🩹 Fixes
+
+-   added fail early if proto transcoding fails ([#78](https://github.com/openecosystems/ecosystem/pull/78))
+
+### ❤️ Thank You
+
+-   Dimy Jeannot @dimyjeannot
 
 ## 0.20.1 (2025-09-09)
 

go/oeco-sdk/v2beta/bindings/nats/listener.go

@@ -18,6 +18,7 @@ import (
 	specv2pb "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta/gen/platform/spec/v2"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/protobuf/types/known/anypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // SpecEventListener is an interface for handling event streaming, listening, and processing for specific configurations.
@@ -126,6 +127,7 @@ func ListenForMultiplexedRequests(_ context.Context, listener SpecEventListener)
 
 // RespondToMultiplexedRequest processes an inbound message, modifies the provided message, and sends a response through NATS.
 func RespondToMultiplexedRequest(_ context.Context, message *ListenerMessage) {
+	log := *zaploggerv1.Bound.Logger
 	js := *Bound.JetStream
 	nm := *message.NatsMessage
 
@@ -135,6 +137,9 @@ func RespondToMultiplexedRequest(_ context.Context, message *ListenerMessage) {
 		return
 	}
 
+	fields := receivedFields(message.Spec, nm.Subject)
+	log.Info("Received multiplexed request", fields...)
+
 	if message.Spec.SpecData != nil && message.Spec.SpecData.Data != nil {
 		fmutils.Filter(message.Spec.SpecData.Data, message.Spec.SpecData.FieldMask.GetPaths())
 	}
@@ -201,6 +206,7 @@ func RespondToMultiplexedRequest(_ context.Context, message *ListenerMessage) {
 
 // respond reply to a NATS message
 func respond(msg *nats.Msg, spec *specv2pb.Spec) {
+	log := *zaploggerv1.Bound.Logger
 	if msg == nil {
 		apexlog.Warn("Received nil message, ignoring")
 		return
@@ -213,6 +219,15 @@ func respond(msg *nats.Msg, spec *specv2pb.Spec) {
 		}
 	}
 
+	spec.CompletedAt = timestamppb.Now()
+
+	fields := completedFields(spec, msg.Subject)
+	var milliseconds int64
+	if spec.CompletedAt != nil && spec.ReceivedAt != nil {
+		milliseconds = spec.CompletedAt.AsTime().Sub(spec.ReceivedAt.AsTime()).Milliseconds()
+	}
+	log.Info(fmt.Sprintf("Completed multiplexed request in %d ms\n", milliseconds), fields...)
+
 	marshal, err := protopb.Marshal(spec)
 	if err != nil {
 		apexlog.Error(sdkv2betalib.ErrServerInternal.WithSpecDetail(spec).WithInternalErrorDetail(errors.New("cannot marshal spec: "), err).Error())

go/oeco-sdk/v2beta/bindings/nats/producer.go

@@ -3,6 +3,7 @@ package natsnodev1
 import (
 	"context"
 	"errors"
+	"fmt"
 	"time"
 
 	"connectrpc.com/connect"
@@ -13,18 +14,23 @@ import (
 	sdkv2betalib "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta"
 	zaploggerv1 "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta/bindings/zap"
 	specv2pb "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta/gen/platform/spec/v2"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // MultiplexCommandSync sends a command synchronously by publishing it to a NATS stream and awaiting a reply.
 // Uses Nats Publish and Subscribe Pattern
-func (b *Binding) MultiplexCommandSync(ctx context.Context, s *specv2pb.Spec, command *SpecCommand) (*nats.Msg, error) {
+func (b *Binding) MultiplexCommandSync(ctx context.Context, spec *specv2pb.Spec, command *SpecCommand) (*nats.Msg, error) {
 	if command == nil {
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("a SpecCommand object is required"))
 	}
 
+	if spec == nil {
+		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("a Spec object is required"))
+	}
+
 	log := *zaploggerv1.Bound.Logger
-	s.SpecEvent = command.CommandName
-	s.SpecType = command.EntityTypeName
+	spec.SpecEvent = command.CommandName
+	spec.SpecType = command.EntityTypeName
 
 	// Encrypt here
 	//fmt.Println(command.Request.ProtoReflect())
@@ -41,77 +47,92 @@ func (b *Binding) MultiplexCommandSync(ctx context.Context, s *specv2pb.Spec, co
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("internal error"))
 	}
 
-	s.Data = data
+	spec.Data = data
 
-	specBytes, err := proto.Marshal(s)
+	specBytes, err := proto.Marshal(spec)
 	if err != nil {
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("could not marshall spec"))
 	}
 
 	subject := GetMultiplexedRequestSubjectName(command.Stream.StreamPrefix(), command.CommandTopic, command.Procedure)
 
-	log.Debug("Issuing a multiplex command: " + command.Procedure + ", on channel: " + subject)
+	fields := receivedFields(spec, subject)
+	log.Info("Issuing a multiplex command: "+command.Procedure, fields...)
 
-	return publish(b.Nats, subject, s, specBytes)
+	return publish(b.Nats, subject, spec, specBytes)
 }
 
 // MultiplexEventSync sends an event to a multiplexed stream and waits for the response or error within the specified timeout.
 // Uses Nats Publish and Subscribe Pattern
-func (b *Binding) MultiplexEventSync(_ context.Context, s *specv2pb.Spec, event *SpecEvent) (*nats.Msg, error) {
+func (b *Binding) MultiplexEventSync(_ context.Context, spec *specv2pb.Spec, event *SpecEvent) (*nats.Msg, error) {
 	if event == nil {
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("a SpecEvent object is required"))
 	}
 
+	if spec == nil {
+		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("a Spec object is required"))
+	}
+
 	log := *zaploggerv1.Bound.Logger
-	s.SpecEvent = event.EventName
-	s.SpecType = event.EntityTypeName
+	spec.SpecEvent = event.EventName
+	spec.SpecType = event.EntityTypeName
 
 	data, err := anypb.New(event.Request)
 	if err != nil {
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("internal error"))
 	}
 
-	s.Data = data
+	spec.Data = data
 
-	specBytes, err := proto.Marshal(s)
+	specBytes, err := proto.Marshal(spec)
 	if err != nil {
 		return nil, sdkv2betalib.ErrServerInternal.WithInternalErrorDetail(errors.New("could not marshall spec"))
 	}
 
 	subject := GetMultiplexedRequestSubjectName(event.Stream.StreamPrefix(), event.EventTopic, event.Procedure)
 
-	log.Debug("Issuing a multiplex event: " + event.Procedure + ", on channel: " + subject)
+	fields := receivedFields(spec, subject)
+	log.Info("Issuing a multiplex event: "+event.Procedure, fields...)
 
-	return publish(b.Nats, subject, s, specBytes)
+	return publish(b.Nats, subject, spec, specBytes)
 }
 
-func publish(n *nats.Conn, subject string, s *specv2pb.Spec, specBytes []byte) (*nats.Msg, error) {
+func publish(n *nats.Conn, subject string, spec *specv2pb.Spec, specBytes []byte) (*nats.Msg, error) {
+	log := *zaploggerv1.Bound.Logger
 	reply, err := n.RequestMsg(&nats.Msg{
 		Subject: subject,
 		Data:    specBytes,
 	}, 10*time.Second)
 	if err != nil {
 		switch {
 		case errors.Is(err, nats.ErrTimeout):
-			return nil, ErrTimeout.WithSpecDetail(s)
+			return nil, ErrTimeout.WithSpecDetail(spec).WithInternalErrorDetail(err)
 		case errors.Is(err, nats.ErrNoResponders):
 			// no responders
-			return nil, ErrNoResponders.WithSpecDetail(s)
+			return nil, ErrNoResponders.WithSpecDetail(spec).WithInternalErrorDetail(err)
 		case errors.Is(err, nats.ErrConnectionClosed):
 			// NATS connection was closed
-			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(s).WithInternalErrorDetail(err)
+			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(spec).WithInternalErrorDetail(err)
 		case errors.Is(err, nats.ErrBadSubscription):
 			// something wrong with the sub
-			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(s).WithInternalErrorDetail(err)
+			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(spec).WithInternalErrorDetail(err)
 		default:
 			// unknown or generic error
-			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(s).WithInternalErrorDetail(errors.New("unhandled NATS error"), err)
+			return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(spec).WithInternalErrorDetail(errors.New("unhandled NATS error"), err)
 		}
 	}
 
 	if reply == nil {
-		return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(s).WithInternalErrorDetail(errors.New("received nil reply from NATS responder"))
+		return nil, sdkv2betalib.ErrServerInternal.WithSpecDetail(spec).WithInternalErrorDetail(errors.New("received nil reply from NATS responder"))
+	}
+
+	spec.CompletedAt = timestamppb.Now()
+	fields := completedFields(spec, subject)
+	var milliseconds int64
+	if spec.CompletedAt != nil && spec.ReceivedAt != nil {
+		milliseconds = spec.CompletedAt.AsTime().Sub(spec.ReceivedAt.AsTime()).Milliseconds()
 	}
+	log.Info(fmt.Sprintf("Completed multiplexed request in %d ms\n", milliseconds), fields...)
 
 	return reply, err
 }

go/oeco-sdk/v2beta/bindings/nats/util.go

@@ -0,0 +1,64 @@
+package natsnodev1
+
+import (
+	"strings"
+
+	specv2pb "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta/gen/platform/spec/v2"
+	"go.uber.org/zap/zapcore"
+)
+
+func baseFields(spec *specv2pb.Spec, subject string) []zapcore.Field {
+	var fields []zapcore.Field
+
+	if spec.Context != nil {
+		if subject != "" {
+			fields = append(fields, zapcore.Field{Key: "subject", Type: zapcore.StringType, String: subject})
+		}
+
+		if spec.SpanContext != nil && spec.SpanContext.TraceId != "" {
+			fields = append(fields, zapcore.Field{Key: "trace_id", Type: zapcore.StringType, String: spec.SpanContext.TraceId})
+		}
+
+		if spec.Context.Validation != nil && spec.Context.Validation.ValidateOnly {
+			fields = append(fields, zapcore.Field{Key: "validate_only", Type: zapcore.StringType, String: "true"})
+		}
+
+		if spec.Context.OrganizationId != "" {
+			fields = append(fields, zapcore.Field{Key: "organization_id", Type: zapcore.StringType, String: spec.Context.OrganizationId})
+		}
+
+		if spec.Context.EcosystemId != "" {
+			fields = append(fields, zapcore.Field{Key: "ecosystem_id", Type: zapcore.StringType, String: spec.Context.EcosystemId})
+		}
+
+		if spec.SpecData != nil && spec.SpecData.FieldMask != nil && len(spec.SpecData.FieldMask.GetPaths()) > 0 {
+			fields = append(fields, zapcore.Field{Key: "field_mask", Type: zapcore.StringType, String: strings.Join(spec.SpecData.FieldMask.GetPaths(), ",")})
+		}
+	}
+
+	return fields
+}
+
+func receivedFields(spec *specv2pb.Spec, subject string) []zapcore.Field {
+	fields := baseFields(spec, subject)
+
+	if spec != nil {
+		if spec.ReceivedAt != nil {
+			fields = append(fields, zapcore.Field{Key: "received_at", Type: zapcore.StringType, String: spec.ReceivedAt.AsTime().String()})
+		}
+	}
+
+	return fields
+}
+
+func completedFields(spec *specv2pb.Spec, subject string) []zapcore.Field {
+	fields := baseFields(spec, subject)
+
+	if spec != nil {
+		if spec.CompletedAt != nil {
+			fields = append(fields, zapcore.Field{Key: "completed_at", Type: zapcore.StringType, String: spec.CompletedAt.AsTime().String()})
+		}
+	}
+
+	return fields
+}

go/oeco-sdk/v2beta/error.go

@@ -186,12 +186,21 @@ func (se *SpecError) WithBadRequest(request *errdetails.BadRequest) SpecErrorabl
 			}
 		}
 		if len(msgs) > 0 {
-			apexlog.WithField("bad_request_errors", strings.Join(msgs, "; ")).Error("captured bad request error details")
+			apexlog.WithField("bad_request_errors", strings.Join(msgs, "; ")).Info("captured bad request error details")
 		}
 	}
 
-	se.ConnectErr.AddDetail(d)
-	return se
+	newErr := *se
+	t := connect.NewError(se.ConnectErr.Code(), fmt.Errorf(se.ConnectErr.Message()))
+	newErr.ConnectErr = *t
+
+	for _, detail := range se.ConnectErr.Details() {
+		newErr.ConnectErr.AddDetail(detail)
+	}
+
+	newErr.ConnectErr.AddDetail(d)
+
+	return &newErr
 }
 
 func (se *SpecError) WithHelp(help *errdetails.Help) SpecErrorable {

go/oeco-sdk/v2beta/error_wire_validation.go

@@ -0,0 +1,45 @@
+package sdkv2betalib
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"buf.build/go/protovalidate"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
+)
+
+func ConvertValidationErrorToFieldValidations(validationErr error) ([]*errdetails.BadRequest_FieldViolation, SpecErrorable) {
+	var err *protovalidate.ValidationError
+	if !errors.As(validationErr, &err) {
+		return nil, ErrServerInternal.WithInternalErrorDetail(
+			fmt.Errorf("expected ValidationError, got: %T", validationErr),
+		)
+	}
+
+	var fv []*errdetails.BadRequest_FieldViolation
+	for _, violation := range err.Violations {
+		fieldBuilder := &strings.Builder{}
+		descriptionBuilder := &strings.Builder{}
+
+		if fieldPath := protovalidate.FieldPathString(violation.Proto.GetField()); fieldPath != "" {
+			fieldBuilder.WriteString(fieldPath)
+		}
+
+		_, _ = fmt.Fprintf(descriptionBuilder, "%s [%s]",
+			violation.Proto.GetMessage(),
+			violation.Proto.GetRuleId())
+
+		fv = append(fv, &errdetails.BadRequest_FieldViolation{
+			Field:       fieldBuilder.String(),
+			Description: descriptionBuilder.String(),
+			Reason:      "WIRE_PROTOCOL_VALIDATION_ERROR",
+			LocalizedMessage: &errdetails.LocalizedMessage{
+				Locale:  LocaleEnglishUS,
+				Message: "Validation failed for this request",
+			},
+		})
+	}
+
+	return fv, nil
+}

libs/plugins/protoc-gen-platform/languages/go/plugins/v2beta/multiplexer/templates/file.go.tmpl

@@ -7,6 +7,7 @@ import (
   "errors"
   "connectrpc.com/connect"
   "google.golang.org/protobuf/types/known/anypb"
+  "google.golang.org/genproto/googleapis/rpc/errdetails"
   {{ if $sm }}"github.com/nats-io/nats.go"{{end}}
   "github.com/nats-io/nats.go/jetstream"
   "github.com/openecosystems/ecosystem/go/oeco-sdk/v2beta/bindings/nats"

libs/plugins/protoc-gen-platform/languages/go/plugins/v2beta/multiplexer/templates/mutation_implemented.go.tmpl

@@ -29,10 +29,14 @@ func (s *{{ $s.Name }}Handler) {{ .Name }}(ctx context.Context, req *connect.Req
 	}
 
 	// Executes top level validation, no business domain validation
-	validationCtx, validationSpan := tracer.Start(ctx, "{{ $methodDashCaseName }}-request-validation", trace.WithSpanKind(trace.SpanKindInternal))
-		v := *protovalidatev0.Bound.Validator
-		if err := v.Validate(req.Msg); err != nil {
-		return nil, sdkv2betalib.ErrServerPreconditionFailed.WithInternalErrorDetail(err)
+	_, validationSpan := tracer.Start(ctx, "{{ $methodDashCaseName }}-request-wire-validation", trace.WithSpanKind(trace.SpanKindInternal))
+	v := *protovalidatev0.Bound.Validator
+	if err := v.Validate(req.Msg); err != nil {
+		fv, serr := sdkv2betalib.ConvertValidationErrorToFieldValidations(err)
+		if serr != nil {
+			return nil, serr
+		}
+		return nil, sdkv2betalib.ErrServerPreconditionFailed.WithBadRequest(&errdetails.BadRequest{FieldViolations: fv})
 	}
 	validationSpan.End()
 
@@ -45,7 +49,7 @@ func (s *{{ $s.Name }}Handler) {{ .Name }}(ctx context.Context, req *connect.Req
 	}
 
 	// Distributed Domain Handler
-	handlerCtx, handlerSpan := tracer.Start(validationCtx, "{{ $methodDashCaseName }}-event-generation", trace.WithSpanKind(trace.SpanKindInternal))
+	handlerCtx, handlerSpan := tracer.Start(ctx, "{{ $methodDashCaseName }}-event-multiplex", trace.WithSpanKind(trace.SpanKindInternal))
 
 	config := s.Get{{ .Name }}Configuration()
 	reply, serr := natsnodev1.Bound.MultiplexCommandSync(handlerCtx, spec, &natsnodev1.SpecCommand{