Require type label on custmsg BaseMessage emissions

pkg/custmsg/custom_message.go

@@ -11,13 +11,22 @@ import (
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder/pb"
 )
 
+const LabelKeyType = "type"
+
 type MessageEmitter interface {
 	// Emit sends a message to the labeler's destination.
 	Emit(context.Context, string) error
 
 	// WithMapLabels sets the labels for the message to be emitted.  Labels are cumulative.
 	WithMapLabels(map[string]string) MessageEmitter
 
+	// WithType sets the required message type label (BaseMessage.Labels["type"]).
+	WithType(msgType string) MessageEmitter
+
+	// WithLabelsAndType sets labels and the required message type label.
+	// msgType wins over any "type" key already present in labels.
+	WithLabelsAndType(labels map[string]string, msgType string) MessageEmitter
+
 	// With adds multiple key-value pairs to the emission.
 	With(keyValues ...string) MessageEmitter
 
@@ -47,6 +56,22 @@ func (l Labeler) WithMapLabels(labels map[string]string) MessageEmitter {
 	return newCustomMessageLabeler
 }
 
+func (l Labeler) WithType(msgType string) MessageEmitter {
+	newCustomMessageLabeler := NewLabeler()
+	maps.Copy(newCustomMessageLabeler.labels, l.labels)
+	newCustomMessageLabeler.labels[LabelKeyType] = msgType
+	return newCustomMessageLabeler
+}
+
+func (l Labeler) WithLabelsAndType(labels map[string]string, msgType string) MessageEmitter {
+	newCustomMessageLabeler := NewLabeler()
+	maps.Copy(newCustomMessageLabeler.labels, l.labels)
+	maps.Copy(newCustomMessageLabeler.labels, labels)
+	newCustomMessageLabeler.labels[LabelKeyType] = msgType
+	return newCustomMessageLabeler
+}
+
+
 // With adds multiple key-value pairs to the CustomMessageLabeler for transmission With SendLogAsCustomMessage
 func (l Labeler) With(keyValues ...string) MessageEmitter {
 	newCustomMessageLabeler := NewLabeler()
@@ -88,6 +113,10 @@ func (l Labeler) SendLogAsCustomMessage(ctx context.Context, msg string) error {
 }
 
 func sendLogAsCustomMessageW(ctx context.Context, msg string, labels map[string]string) error {
+	if labels[LabelKeyType] == "" {
+		return fmt.Errorf("custmsg: missing required label %q", LabelKeyType)
+	}
+
 	// TODO un-comment after INFOPLAT-1386
 	// cast to map[string]any
 	//newLabels := map[string]any{}

pkg/custmsg/custom_message_test.go

@@ -1,26 +1,49 @@
 package custmsg
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // tests CustomMessageAgent does not share state across new instances created by `With`
 func Test_CustomMessageAgent(t *testing.T) {
 	cma := NewLabeler()
-	cma1 := cma.With("key1", "value1")
+	cma1 := cma.With("key1", "value1").WithType("TestType")
 	cma2 := cma1.With("key2", "value2")
 
 	assert.NotEqual(t, cma1.Labels(), cma2.Labels())
 }
 
 func Test_CustomMessageAgent_With(t *testing.T) {
-	cma := NewLabeler().With("key1", "value1")
-	assert.Equal(t, map[string]string{"key1": "value1"}, cma.Labels())
+	cma := NewLabeler().WithType("TestType").With("key1", "value1")
+	assert.Equal(t, map[string]string{"key1": "value1", LabelKeyType: "TestType"}, cma.Labels())
 }
 
 func Test_CustomMessageAgent_WithMapLabels(t *testing.T) {
-	cma := NewLabeler().WithMapLabels(map[string]string{"key1": "value1"})
-	assert.Equal(t, map[string]string{"key1": "value1"}, cma.Labels())
+	cma := NewLabeler().WithType("TestType").WithMapLabels(map[string]string{"key1": "value1"})
+	assert.Equal(t, map[string]string{"key1": "value1", LabelKeyType: "TestType"}, cma.Labels())
+}
+
+func Test_CustomMessageAgent_WithType(t *testing.T) {
+	cma := NewLabeler().WithType("NodeConfig")
+	assert.Equal(t, map[string]string{LabelKeyType: "NodeConfig"}, cma.Labels())
+}
+
+func Test_CustomMessageAgent_WithLabelsAndType(t *testing.T) {
+	cma := NewLabeler().WithLabelsAndType(map[string]string{
+		"system": "Application",
+		"type":   "ignored",
+	}, "NodeConfig")
+	assert.Equal(t, map[string]string{
+		"system":       "Application",
+		LabelKeyType:   "NodeConfig",
+	}, cma.Labels())
+}
+
+func Test_CustomMessageAgent_EmitRequiresType(t *testing.T) {
+	err := NewLabeler().Emit(context.Background(), "msg")
+	require.ErrorContains(t, err, `missing required label "type"`)
 }

pkg/workflows/wasm/host/module.go

@@ -1062,6 +1062,14 @@ func (u *unimplementedMessageEmitter) WithMapLabels(map[string]string) custmsg.M
 	return u
 }
 
+func (u *unimplementedMessageEmitter) WithType(string) custmsg.MessageEmitter {
+	return u
+}
+
+func (u *unimplementedMessageEmitter) WithLabelsAndType(map[string]string, string) custmsg.MessageEmitter {
+	return u
+}
+
 func (u *unimplementedMessageEmitter) With(kvs ...string) custmsg.MessageEmitter {
 	return u
 }

pkg/workflows/wasm/host/module_test.go

@@ -39,6 +39,23 @@ func (m *mockMessageEmitter) WithMapLabels(labels map[string]string) custmsg.Mes
 	return m
 }
 
+func (m *mockMessageEmitter) WithType(msgType string) custmsg.MessageEmitter {
+	if m.labels == nil {
+		m.labels = map[string]string{}
+	}
+	m.labels[custmsg.LabelKeyType] = msgType
+	return m
+}
+
+func (m *mockMessageEmitter) WithLabelsAndType(labels map[string]string, msgType string) custmsg.MessageEmitter {
+	m.labels = labels
+	if m.labels == nil {
+		m.labels = map[string]string{}
+	}
+	m.labels[custmsg.LabelKeyType] = msgType
+	return m
+}
+
 func (m *mockMessageEmitter) With(keyValues ...string) custmsg.MessageEmitter {
 	// do nothing
 	return m