MG-3486 - RE Loop Prevention (#3559)

Signed-off-by: dusan <borovcanindusan1@gmail.com>

pkg/messaging/README.md

@@ -26,16 +26,17 @@ The `$queue/` prefix lets any publisher force delivery into the durable stream q
 
 ### Stream queues
 
-On startup, every publisher and pubsub client declares a durable stream queue named after its prefix. Stream subscribers use consumer groups, so each group receives every message exactly once. The default stream queue is named `m`.
+On startup, publishers and pubsub clients normally declare a durable stream queue named after their prefix. Stream subscribers use consumer groups, so each group receives every message exactly once. The default stream queue is named `m`. `InternalMetadata` instead requires that stream to be pre-provisioned by the broker and never attempts to create or modify it.
 
 ### Subscription
 
-`Subscribe` attaches to the durable stream queue via a consumer group filtered by topic. Optionally (when `DirectTopicIngress` is enabled), it also subscribes to the raw MQTT topic so that messages published directly by MQTT clients — bypassing the queue — are also received.
+`Subscribe` attaches to the durable stream queue via a consumer group filtered by topic. Optionally (when `DirectTopicIngress` is enabled), it also subscribes to the raw MQTT topic so that messages published directly by MQTT clients — bypassing the queue — are also received. A deployment using `InternalMetadata` must authorize the requested subscriptions explicitly; the Rules Engine local principal authorizes only pre-provisioned stream `m`.
 
 ### Options
 
-| Option                 | Description                                            |
-| ---------------------- | ------------------------------------------------------ |
-| `Prefix(p)`            | Set topic prefix (default: `m`)                        |
-| `ConnectionName(n)`    | Human-readable broker connection name                  |
-| `DirectTopicIngress()` | Also consume raw MQTT topic messages (subscriber only) |
+| Option                                 | Description                                                                                       |
+| -------------------------------------- | ------------------------------------------------------------------------------------------------- |
+| `Prefix(p)`                            | Set topic prefix (default: `m`)                                                                   |
+| `ConnectionName(n)`                    | Human-readable broker connection name                                                             |
+| `DirectTopicIngress()`                 | Also consume raw MQTT topic messages (subscriber only)                                            |
+| `InternalMetadata(cert, key, ca)`      | Require mTLS, carry reserved internal metadata, and use a broker-provisioned stream                 |

pkg/messaging/brokers/brokers_fluxmq.go

@@ -28,6 +28,13 @@ func ConnectionName(name string) messaging.Option {
 	return fluxmq.ConnectionName(name)
 }
 
+// InternalMetadata returns an option for a trusted local-service connection
+// that carries internal metadata over mTLS and consumes the broker-provisioned
+// message stream.
+func InternalMetadata(certFile, keyFile, caFile string) messaging.Option {
+	return fluxmq.InternalMetadata(certFile, keyFile, caFile)
+}
+
 func NewPublisher(ctx context.Context, url string, opts ...messaging.Option) (messaging.Publisher, error) {
 	pb, err := fluxmq.NewPublisher(ctx, url, opts...)
 	if err != nil {

pkg/messaging/brokers/brokers_nats.go

@@ -28,6 +28,12 @@ func ConnectionName(_ string) messaging.Option {
 	return func(_ any) error { return nil }
 }
 
+// InternalMetadata is a no-op for the NATS backend. It exists for compile-time
+// compatibility with FluxMQ; NATS carries metadata in the protobuf message.
+func InternalMetadata(_, _, _ string) messaging.Option {
+	return func(_ any) error { return nil }
+}
+
 func NewPublisher(ctx context.Context, url string, opts ...messaging.Option) (messaging.Publisher, error) {
 	pb, err := nats.NewPublisher(ctx, url, opts...)
 	if err != nil {

pkg/messaging/fluxmq/internal_metadata_integration_test.go

@@ -0,0 +1,65 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package fluxmq
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/absmach/magistrala/pkg/messaging"
+	"github.com/stretchr/testify/require"
+)
+
+type integrationHandler struct{}
+
+func (*integrationHandler) Handle(*messaging.Message) error { return nil }
+
+func (*integrationHandler) Cancel() error { return nil }
+
+func TestInternalMetadataPreprovisionedStream(t *testing.T) {
+	brokerURL := os.Getenv("FLUXMQ_INTERNAL_METADATA_URL")
+	if brokerURL == "" {
+		t.Skip("set FLUXMQ_INTERNAL_METADATA_URL to run the trusted stream integration test")
+	}
+
+	ctx := context.Background()
+	ps, err := NewPubSub(ctx, brokerURL, nil, InternalMetadata(
+		os.Getenv("FLUXMQ_INTERNAL_METADATA_CERT"),
+		os.Getenv("FLUXMQ_INTERNAL_METADATA_KEY"),
+		os.Getenv("FLUXMQ_INTERNAL_METADATA_CA"),
+	))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = ps.Close() })
+
+	stamp := time.Now().UnixNano()
+	topic := "m/integration-domain/c/integration-channel/" + time.Unix(0, stamp).Format("150405.000000000")
+	handler := &integrationHandler{}
+	subscribeErr := ps.Subscribe(ctx, messaging.SubscriberConfig{
+		ID:             "internal-metadata-integration",
+		Topic:          "m/#",
+		DeliveryPolicy: messaging.DeliverNewPolicy,
+		Handler:        handler,
+	})
+	if os.Getenv("FLUXMQ_INTERNAL_METADATA_EXPECT_SUBSCRIBE_ERROR") == "true" {
+		require.Error(t, subscribeErr)
+		return
+	}
+	require.NoError(t, subscribeErr)
+
+	msg := &messaging.Message{
+		Domain:    "integration-domain",
+		Channel:   "integration-channel",
+		Subtopic:  time.Unix(0, stamp).Format("150405.000000000"),
+		Payload:   []byte("metadata-round-trip"),
+		Publisher: "rules-engine",
+		Protocol:  "internal",
+		Metadata: map[string]string{
+			"magistrala.re.trace": "signed-trace",
+			"other":               "preserved",
+		},
+	}
+	require.NoError(t, ps.Publish(ctx, topic, msg))
+}

pkg/messaging/fluxmq/options.go

@@ -4,7 +4,11 @@
 package fluxmq
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
+	"fmt"
+	"os"
 	"strings"
 
 	"github.com/absmach/magistrala/pkg/messaging"
@@ -21,6 +25,8 @@ type options struct {
 	connectionName     string
 	directTopicIngress bool
 	directTopicOnly    bool
+	preprovisioned     bool
+	tlsConfig          *tls.Config
 }
 
 func defaultOptions() options {
@@ -51,6 +57,58 @@ func Prefix(prefix string) messaging.Option {
 	}
 }
 
+// InternalMetadata configures a trusted FluxMQ service connection that can
+// exchange broker-internal message metadata. It presents a client certificate,
+// verifies the broker against caFile, and uses the broker-provisioned stream
+// instead of trying to declare it with the service principal's restricted ACL.
+//
+// All three paths are required: a half-configured client would silently connect
+// without the identity the broker authorizes against.
+func InternalMetadata(certFile, keyFile, caFile string) messaging.Option {
+	return func(val any) error {
+		cfg, err := mtlsConfig(certFile, keyFile, caFile)
+		if err != nil {
+			return err
+		}
+		switch v := val.(type) {
+		case *publisher:
+			v.tlsConfig = cfg
+			v.preprovisioned = true
+		case *pubsub:
+			v.tlsConfig = cfg
+			v.preprovisioned = true
+		default:
+			return ErrInvalidType
+		}
+
+		return nil
+	}
+}
+
+func mtlsConfig(certFile, keyFile, caFile string) (*tls.Config, error) {
+	if certFile == "" || keyFile == "" || caFile == "" {
+		return nil, fmt.Errorf("%w: mTLS needs a certificate, a key, and a CA", ErrInvalidType)
+	}
+	certificate, err := tls.LoadX509KeyPair(certFile, keyFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load FluxMQ client certificate: %w", err)
+	}
+	ca, err := os.ReadFile(caFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read FluxMQ CA certificate: %w", err)
+	}
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(ca) {
+		return nil, fmt.Errorf("failed to parse FluxMQ CA certificate %q", caFile)
+	}
+
+	return &tls.Config{
+		Certificates: []tls.Certificate{certificate},
+		RootCAs:      pool,
+		MinVersion:   tls.VersionTLS12,
+	}, nil
+}
+
 // ConnectionName sets a human-readable connection name sent to FluxMQ
 // for identifying this client in the broker's admin UI.
 func ConnectionName(name string) messaging.Option {

pkg/messaging/fluxmq/options_test.go

@@ -0,0 +1,148 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package fluxmq
+
+import (
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestInternalMetadataRequiresMTLS(t *testing.T) {
+	if err := InternalMetadata("", "", "")(&pubsub{}); err == nil {
+		t.Fatal("expected empty certificate paths to be rejected")
+	}
+
+	dir := t.TempDir()
+	certFile, keyFile, caFile := writeTestCertificate(t, dir)
+	for name, paths := range map[string][3]string{
+		"missing certificate": {"", keyFile, caFile},
+		"missing key":         {certFile, "", caFile},
+		"missing CA":          {certFile, keyFile, ""},
+	} {
+		t.Run(name, func(t *testing.T) {
+			if err := InternalMetadata(paths[0], paths[1], paths[2])(&pubsub{}); err == nil {
+				t.Fatal("expected partial mTLS configuration to be rejected")
+			}
+		})
+	}
+
+	tests := []struct {
+		name string
+		get  func() options
+	}{
+		{
+			name: "publisher",
+			get: func() options {
+				var pub publisher
+				if err := InternalMetadata(certFile, keyFile, caFile)(&pub); err != nil {
+					t.Fatalf("configure publisher: %v", err)
+				}
+				return pub.options
+			},
+		},
+		{
+			name: "pubsub",
+			get: func() options {
+				var ps pubsub
+				if err := InternalMetadata(certFile, keyFile, caFile)(&ps); err != nil {
+					t.Fatalf("configure pubsub: %v", err)
+				}
+				return ps.options
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			opts := tc.get()
+			if opts.tlsConfig == nil {
+				t.Fatal("expected TLS configuration")
+			}
+			if !opts.preprovisioned {
+				t.Fatal("expected broker-provisioned stream mode")
+			}
+		})
+	}
+}
+
+func TestMQTTTopicToWireTopic(t *testing.T) {
+	for input, want := range map[string]string{
+		"m/domain/c/channel/subtopic":  "m.domain.c.channel.subtopic",
+		"m/domain/c/channel/sub.topic": "m.domain.c.channel.sub%2Etopic",
+	} {
+		if got := mqttTopicToWireTopic(input); got != want {
+			t.Fatalf("wire topic = %q, want %q", got, want)
+		}
+	}
+}
+
+func writeTestCertificate(t *testing.T, dir string) (certFile, keyFile, caFile string) {
+	t.Helper()
+
+	now := time.Now()
+	caPublic, caPrivate, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("generate CA key: %v", err)
+	}
+	caTemplate := &x509.Certificate{
+		SerialNumber:          big.NewInt(1),
+		Subject:               pkix.Name{CommonName: "test-ca"},
+		NotBefore:             now.Add(-time.Minute),
+		NotAfter:              now.Add(time.Hour),
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+		KeyUsage:              x509.KeyUsageCertSign,
+	}
+	caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, caPublic, caPrivate)
+	if err != nil {
+		t.Fatalf("create CA certificate: %v", err)
+	}
+
+	clientPublic, clientPrivate, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("generate client key: %v", err)
+	}
+	clientTemplate := &x509.Certificate{
+		SerialNumber: big.NewInt(2),
+		Subject:      pkix.Name{CommonName: "rules-engine"},
+		NotBefore:    now.Add(-time.Minute),
+		NotAfter:     now.Add(time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+	}
+	clientDER, err := x509.CreateCertificate(rand.Reader, clientTemplate, caTemplate, clientPublic, caPrivate)
+	if err != nil {
+		t.Fatalf("create client certificate: %v", err)
+	}
+	clientKey, err := x509.MarshalPKCS8PrivateKey(clientPrivate)
+	if err != nil {
+		t.Fatalf("marshal client key: %v", err)
+	}
+
+	certFile = filepath.Join(dir, "client.crt")
+	keyFile = filepath.Join(dir, "client.key")
+	caFile = filepath.Join(dir, "ca.crt")
+	writePEM(t, certFile, "CERTIFICATE", clientDER)
+	writePEM(t, keyFile, "PRIVATE KEY", clientKey)
+	writePEM(t, caFile, "CERTIFICATE", caDER)
+
+	return certFile, keyFile, caFile
+}
+
+func writePEM(t *testing.T, path, typ string, contents []byte) {
+	t.Helper()
+
+	data := pem.EncodeToMemory(&pem.Block{Type: typ, Bytes: contents})
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		t.Fatalf("write %s: %v", path, err)
+	}
+}

pkg/messaging/fluxmq/publisher.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	fluxamqp "github.com/absmach/fluxmq/client/amqp"
+	fluxtopics "github.com/absmach/fluxmq/topics"
 	"github.com/absmach/magistrala/pkg/messaging"
 )
 
@@ -18,7 +19,13 @@ var _ messaging.Publisher = (*publisher)(nil)
 const (
 	headerExternalID = "external_id"
 	headerProtocol   = "protocol"
-	protocolMQTT     = "mqtt"
+	// headerMetadataPrefix sits inside FluxMQ's reserved "_flux." namespace, so
+	// the broker treats message metadata as internal state passed between
+	// first-party services: it is dropped when an untrusted connection sets it
+	// and omitted when one subscribes. A service therefore cannot be fed forged
+	// metadata by a device, and a device cannot read metadata a service set.
+	headerMetadataPrefix = "_flux.mg."
+	protocolMQTT         = "mqtt"
 )
 
 type publisher struct {
@@ -60,6 +67,9 @@ func newPublisher(_ context.Context, url string, declare bool, opts ...messaging
 		SetOnConnect(func() {
 			logger.Info("FluxMQ message publisher connected")
 		})
+	if pub.tlsConfig != nil {
+		amqpOpts = amqpOpts.SetTLSConfig(pub.tlsConfig)
+	}
 
 	client, err := fluxamqp.New(amqpOpts)
 	if err != nil {
@@ -68,7 +78,7 @@ func newPublisher(_ context.Context, url string, declare bool, opts ...messaging
 	if err := client.Connect(); err != nil {
 		return nil, err
 	}
-	if declare {
+	if declare && !pub.preprovisioned {
 		if err := declareStream(client, pub.prefix); err != nil {
 			_ = client.Close()
 			return nil, err
@@ -85,16 +95,7 @@ func (pub *publisher) Publish(ctx context.Context, topic string, msg *messaging.
 		return ErrEmptyTopic
 	}
 
-	props := map[string]string{
-		headerExternalID: msg.GetPublisher(),
-		headerProtocol:   msg.GetProtocol(),
-	}
-	if clientID := msg.ClientIdentity(); clientID != "" {
-		props["client_id"] = clientID
-	}
-	if msg.GetCreated() != 0 {
-		props["created"] = strconv.FormatInt(msg.GetCreated(), 10)
-	}
+	props := messageProperties(msg)
 
 	cleanTopic := strings.TrimPrefix(strings.TrimSpace(topic), "/")
 	if cleanTopic == "" {
@@ -123,6 +124,9 @@ func (pub *publisher) Publish(ctx context.Context, topic string, msg *messaging.
 			Properties: props,
 		})
 	}
+	if pub.preprovisioned {
+		publishTopic = mqttTopicToWireTopic(publishTopic)
+	}
 
 	return pub.client.PublishWithOptionsContext(ctx, &fluxamqp.PublishOptions{
 		Topic:      publishTopic,
@@ -131,6 +135,32 @@ func (pub *publisher) Publish(ctx context.Context, topic string, msg *messaging.
 	})
 }
 
+func mqttTopicToWireTopic(topic string) string {
+	// AMQP uses dots as segment separators, while MQTT permits dots as ordinary
+	// topic characters. Escape literal dots before translating slashes so the
+	// subscriber can URL-decode them back instead of treating them as slashes.
+	topic = strings.ReplaceAll(topic, ".", "%2E")
+	return fluxtopics.MQTTTopicToAMQP(topic)
+}
+
+func messageProperties(msg *messaging.Message) map[string]string {
+	props := map[string]string{
+		headerExternalID: msg.GetPublisher(),
+		headerProtocol:   msg.GetProtocol(),
+	}
+	if clientID := msg.ClientIdentity(); clientID != "" {
+		props["client_id"] = clientID
+	}
+	if msg.GetCreated() != 0 {
+		props["created"] = strconv.FormatInt(msg.GetCreated(), 10)
+	}
+	for key, value := range msg.GetMetadata() {
+		props[headerMetadataPrefix+key] = value
+	}
+
+	return props
+}
+
 func (pub *publisher) Close() error {
 	return pub.client.Close()
 }

pkg/messaging/fluxmq/pubsub.go

@@ -64,8 +64,11 @@ func NewPubSub(_ context.Context, url string, logger *slog.Logger, opts ...messa
 			ps.logInfo("FluxMQ message pub/sub reconnecting", "attempt", attempt)
 		}).
 		SetOnConnect(func() {
-			ps.logInfo("FluxMQ message pub/sub connected", url, ps.prefix)
+			ps.logInfo("FluxMQ message pub/sub connected", "prefix", ps.prefix)
 		})
+	if ps.tlsConfig != nil {
+		amqpOpts = amqpOpts.SetTLSConfig(ps.tlsConfig)
+	}
 
 	client, err := fluxamqp.New(amqpOpts)
 	if err != nil {
@@ -74,9 +77,11 @@ func NewPubSub(_ context.Context, url string, logger *slog.Logger, opts ...messa
 	if err := client.Connect(); err != nil {
 		return nil, err
 	}
-	if err := declareStream(client, ps.prefix); err != nil {
-		_ = client.Close()
-		return nil, err
+	if !ps.preprovisioned {
+		if err := declareStream(client, ps.prefix); err != nil {
+			_ = client.Close()
+			return nil, err
+		}
 	}
 
 	ps.client = client
@@ -240,6 +245,24 @@ func messageFromDelivery(body []byte, headers map[string]any, ts time.Time, pref
 		created = v
 	}
 
+	// Allocated lazily: this runs for every delivered message, and carrying
+	// metadata is the exception rather than the rule.
+	var metadata map[string]string
+	for key, value := range headers {
+		metadataKey, ok := strings.CutPrefix(key, headerMetadataPrefix)
+		if !ok || metadataKey == "" {
+			continue
+		}
+		metadataValue, ok := value.(string)
+		if !ok {
+			continue
+		}
+		if metadata == nil {
+			metadata = make(map[string]string)
+		}
+		metadata[metadataKey] = metadataValue
+	}
+
 	return &messaging.Message{
 		Domain:    domain,
 		Channel:   channel,
@@ -249,6 +272,7 @@ func messageFromDelivery(body []byte, headers map[string]any, ts time.Time, pref
 		ClientId:  clientID,
 		Protocol:  protocol,
 		Created:   created,
+		Metadata:  metadata,
 	}, nil
 }
 

pkg/messaging/fluxmq/pubsub_test.go

@@ -4,6 +4,7 @@
 package fluxmq
 
 import (
+	"reflect"
 	"testing"
 	"time"
 
@@ -119,13 +120,22 @@ func TestMessageFromDelivery(t *testing.T) {
 		wantErr   bool
 	}{
 		{
-			name:      "use explicit publisher and client_id headers",
-			body:      []byte(`{"temperature":22.5}`),
-			headers:   map[string]any{"external_id": "ext-1", "client_id": "client-1", "protocol": "mqtt", "created": "1710000000000000123"},
+			name: "use explicit publisher and client_id headers",
+			body: []byte(`{"temperature":22.5}`),
+			headers: map[string]any{
+				"external_id": "ext-1",
+				"client_id":   "client-1",
+				"protocol":    "mqtt",
+				"created":     "1710000000000000123",
+				headerMetadataPrefix + "magistrala.re.trace": `["rule-1"]`,
+				headerMetadataPrefix + "invalid":             int64(1),
+				"ordinary_header":                            "ignored",
+			},
 			ts:        time.Unix(1710000000, 0),
 			prefix:    "writers",
 			mqttTopic: "writers/domain/c/channel/temp",
 			want: &messaging.Message{
+				Metadata:  map[string]string{"magistrala.re.trace": `["rule-1"]`},
 				Domain:    "domain",
 				Channel:   "channel",
 				Subtopic:  "temp",
@@ -211,10 +221,71 @@ func TestMessageFromDelivery(t *testing.T) {
 			if got.Created != tc.want.Created {
 				t.Fatalf("created mismatch: got %d, want %d", got.Created, tc.want.Created)
 			}
+			if len(got.GetMetadata()) != len(tc.want.GetMetadata()) || got.GetMetadata()["magistrala.re.trace"] != tc.want.GetMetadata()["magistrala.re.trace"] {
+				t.Fatalf("metadata mismatch: got %#v, want %#v", got.GetMetadata(), tc.want.GetMetadata())
+			}
 		})
 	}
 }
 
+func TestMessagePropertiesIncludesMetadata(t *testing.T) {
+	msg := &messaging.Message{
+		Publisher: "publisher",
+		Protocol:  "mqtt",
+		ClientId:  "client",
+		Created:   1710000000000000123,
+		Metadata: map[string]string{
+			"magistrala.re.trace": `["rule-1"]`,
+		},
+	}
+
+	got := messageProperties(msg)
+
+	want := map[string]string{
+		headerExternalID: "publisher",
+		headerProtocol:   "mqtt",
+		"client_id":      "publisher",
+		"created":        "1710000000000000123",
+		headerMetadataPrefix + "magistrala.re.trace": `["rule-1"]`,
+	}
+	for key, value := range want {
+		if got[key] != value {
+			t.Errorf("property %q mismatch: got %q, want %q", key, got[key], value)
+		}
+	}
+	if len(got) != len(want) {
+		t.Errorf("unexpected properties: %#v", got)
+	}
+}
+
+func TestMetadataPropertyRoundTrip(t *testing.T) {
+	want := &messaging.Message{
+		Publisher: "rules-engine",
+		Protocol:  "internal",
+		ClientId:  "origin-client",
+		Created:   1710000000000000123,
+		Payload:   []byte("payload"),
+		Metadata: map[string]string{
+			"magistrala.re.trace": "signed-trace",
+			"other":               "preserved",
+		},
+	}
+
+	properties := messageProperties(want)
+	headers := make(map[string]any, len(properties))
+	for key, value := range properties {
+		headers[key] = value
+	}
+
+	got, err := messageFromDelivery(want.Payload, headers, time.Time{}, "m", "m/domain/c/channel/subtopic")
+	if err != nil {
+		t.Fatalf("reconstruct message: %v", err)
+	}
+	if !reflect.DeepEqual(got.Metadata, want.Metadata) {
+		t.Fatalf("metadata mismatch: got %#v, want %#v", got.Metadata, want.Metadata)
+	}
+}
+
 func TestMessageFromDeliveryZeroTimestampFallsBackToNow(t *testing.T) {
 	before := time.Now().UnixNano()
 	got, err := messageFromDelivery([]byte("raw"), nil, time.Time{}, "m", "m/dom/c/ch")

pkg/messaging/message.pb.go

@@ -26,15 +26,17 @@ const (
 
 // Message represents a message emitted by the Magistrala adapters layer.
 type Message struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Channel       string                 `protobuf:"bytes,1,opt,name=channel,proto3" json:"channel,omitempty"`
-	Domain        string                 `protobuf:"bytes,2,opt,name=domain,proto3" json:"domain,omitempty"`
-	Subtopic      string                 `protobuf:"bytes,3,opt,name=subtopic,proto3" json:"subtopic,omitempty"`
-	Publisher     string                 `protobuf:"bytes,4,opt,name=publisher,proto3" json:"publisher,omitempty"`
-	Protocol      string                 `protobuf:"bytes,5,opt,name=protocol,proto3" json:"protocol,omitempty"`
-	Payload       []byte                 `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"`
-	Created       int64                  `protobuf:"varint,7,opt,name=created,proto3" json:"created,omitempty"`                  // Unix timestamp in nanoseconds
-	ClientId      string                 `protobuf:"bytes,8,opt,name=client_id,json=clientId,proto3" json:"client_id,omitempty"` // Transport-level client identifier
+	state     protoimpl.MessageState `protogen:"open.v1"`
+	Channel   string                 `protobuf:"bytes,1,opt,name=channel,proto3" json:"channel,omitempty"`
+	Domain    string                 `protobuf:"bytes,2,opt,name=domain,proto3" json:"domain,omitempty"`
+	Subtopic  string                 `protobuf:"bytes,3,opt,name=subtopic,proto3" json:"subtopic,omitempty"`
+	Publisher string                 `protobuf:"bytes,4,opt,name=publisher,proto3" json:"publisher,omitempty"`
+	Protocol  string                 `protobuf:"bytes,5,opt,name=protocol,proto3" json:"protocol,omitempty"`
+	Payload   []byte                 `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"`
+	Created   int64                  `protobuf:"varint,7,opt,name=created,proto3" json:"created,omitempty"`                  // Unix timestamp in nanoseconds
+	ClientId  string                 `protobuf:"bytes,8,opt,name=client_id,json=clientId,proto3" json:"client_id,omitempty"` // Transport-level client identifier
+	// Internal metadata propagated between services
+	Metadata      map[string]string `protobuf:"bytes,9,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -125,11 +127,18 @@ func (x *Message) GetClientId() string {
 	return ""
 }
 
+func (x *Message) GetMetadata() map[string]string {
+	if x != nil {
+		return x.Metadata
+	}
+	return nil
+}
+
 var File_pkg_messaging_message_proto protoreflect.FileDescriptor
 
 const file_pkg_messaging_message_proto_rawDesc = "" +
 	"\n" +
-	"\x1bpkg/messaging/message.proto\x12\tmessaging\"\xe2\x01\n" +
+	"\x1bpkg/messaging/message.proto\x12\tmessaging\"\xdd\x02\n" +
 	"\aMessage\x12\x18\n" +
 	"\achannel\x18\x01 \x01(\tR\achannel\x12\x16\n" +
 	"\x06domain\x18\x02 \x01(\tR\x06domain\x12\x1a\n" +
@@ -138,7 +147,11 @@ const file_pkg_messaging_message_proto_rawDesc = "" +
 	"\bprotocol\x18\x05 \x01(\tR\bprotocol\x12\x18\n" +
 	"\apayload\x18\x06 \x01(\fR\apayload\x12\x18\n" +
 	"\acreated\x18\a \x01(\x03R\acreated\x12\x1b\n" +
-	"\tclient_id\x18\b \x01(\tR\bclientIdB\rZ\v./messagingb\x06proto3"
+	"\tclient_id\x18\b \x01(\tR\bclientId\x12<\n" +
+	"\bmetadata\x18\t \x03(\v2 .messaging.Message.MetadataEntryR\bmetadata\x1a;\n" +
+	"\rMetadataEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01B\rZ\v./messagingb\x06proto3"
 
 var (
 	file_pkg_messaging_message_proto_rawDescOnce sync.Once
@@ -152,16 +165,18 @@ func file_pkg_messaging_message_proto_rawDescGZIP() []byte {
 	return file_pkg_messaging_message_proto_rawDescData
 }
 
-var file_pkg_messaging_message_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_pkg_messaging_message_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_pkg_messaging_message_proto_goTypes = []any{
 	(*Message)(nil), // 0: messaging.Message
+	nil,             // 1: messaging.Message.MetadataEntry
 }
 var file_pkg_messaging_message_proto_depIdxs = []int32{
-	0, // [0:0] is the sub-list for method output_type
-	0, // [0:0] is the sub-list for method input_type
-	0, // [0:0] is the sub-list for extension type_name
-	0, // [0:0] is the sub-list for extension extendee
-	0, // [0:0] is the sub-list for field type_name
+	1, // 0: messaging.Message.metadata:type_name -> messaging.Message.MetadataEntry
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
 }
 
 func init() { file_pkg_messaging_message_proto_init() }
@@ -175,7 +190,7 @@ func file_pkg_messaging_message_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_pkg_messaging_message_proto_rawDesc), len(file_pkg_messaging_message_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   1,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   0,
 		},

pkg/messaging/message.proto

@@ -16,4 +16,6 @@ message Message {
   bytes payload = 6;
   int64 created = 7; // Unix timestamp in nanoseconds
   string client_id = 8; // Transport-level client identifier
+  // Internal metadata propagated between services
+  map<string, string> metadata = 9;
 }

pkg/messaging/nats/pubsub_test.go

@@ -31,6 +31,7 @@ var (
 		Protocol:  "mqtt",
 		Payload:   []byte("payload"),
 		Created:   time.Now().UnixNano(),
+		Metadata:  map[string]string{"magistrala.re.trace": `["rule-1"]`},
 	}
 )
 
@@ -100,6 +101,7 @@ func TestPublisher(t *testing.T) {
 			assert.Equal(t, tc.message.Publisher, receivedMsg.Publisher, fmt.Sprintf("%s: expected %+v got %+v\n", tc.desc, &tc.message, receivedMsg))
 			assert.Equal(t, tc.message.Subtopic, receivedMsg.Subtopic, fmt.Sprintf("%s: expected %+v got %+v\n", tc.desc, &tc.message, receivedMsg))
 			assert.Equal(t, tc.message.Payload, receivedMsg.Payload, fmt.Sprintf("%s: expected %+v got %+v\n", tc.desc, &tc.message, receivedMsg))
+			assert.Equal(t, tc.message.Metadata, receivedMsg.Metadata, fmt.Sprintf("%s: expected %+v got %+v\n", tc.desc, &tc.message, receivedMsg))
 		}
 	}
 }