fix: RPC event serialization error, dont send empty messages (#1871)

## This PR
- adds a test for the notification error that I encountered in
#1869

### Related Issues
None that I'm aware of

### Notes
vibe coding warning: I don't know go well enough to fix this, but I did
use TDD and also did a manual test to verify that my problem went away.
If any part of my solution is technicaly or doesn't fit in the codebase
for any reason: feel free to comment, I'll do my best to improve it.

### How to test
Running the test suite should be sufficient

Signed-off-by: Karel Vervaeke <karel@vervaeke.info>

Author: karel1980

flagd/pkg/service/flag-evaluation/connect_service_test.go

@@ -15,7 +15,6 @@ import (
 	mock "github.com/open-feature/flagd/core/pkg/evaluator/mock"
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/model"
-	"github.com/open-feature/flagd/core/pkg/notifications"
 	iservice "github.com/open-feature/flagd/core/pkg/service"
 	"github.com/open-feature/flagd/core/pkg/store"
 	"github.com/open-feature/flagd/core/pkg/telemetry"
@@ -254,8 +253,8 @@ func TestConnectServiceWatcher(t *testing.T) {
 	select {
 	case n := <-sChan:
 		require.Equal(t, ofType, n.Type, "expected notification type: %s, but received %s", ofType, n.Type)
-		notifications := n.Data["flags"].(notifications.Notifications)
-		flag1, ok := notifications["flag1"].(map[string]interface{})
+		flags := n.Data["flags"].(map[string]interface{})
+		flag1, ok := flags["flag1"].(map[string]interface{})
 		require.True(t, ok, "flag1 notification should be a map[string]interface{}")
 		require.Equal(t, flag1["type"], string(model.NotificationCreate), "expected notification type: %s, but received %s", model.NotificationCreate, flag1["type"])
 	case <-timeout.Done():

flagd/pkg/service/flag-evaluation/eventing.go

@@ -41,17 +41,23 @@ func (eventing *eventingConfiguration) Subscribe(ctx context.Context, id any, se
 		for result := range watcher {
 			newFlags := make(map[string]model.Flag)
 			for _, flag := range result.Flags {
-                // we should be either selecting on a flag set here, or using the source-priority - duplicates are already handled, so we don't have to worry about overwrites
+				// we should be either selecting on a flag set here, or using the source-priority - duplicates are already handled, so we don't have to worry about overwrites
 				newFlags[flag.Key] = flag
 			}
 
 			// ignore the first notification (nil old flags), the watcher emits on initialization, but for RPC we don't care until there's a change
 			if oldFlags != nil {
 				notifications := notifications.NewFromFlags(oldFlags, newFlags)
+				// if there are no changes, don't emit a notification
+				if len(notifications) == 0 {
+					oldFlags = newFlags
+					continue
+				}
 				notifier <- iservice.Notification{
 					Type: iservice.ConfigurationChange,
 					Data: map[string]interface{}{
-						"flags": notifications,
+						// don't use our custom type or it cannot be serialized, convert to map
+						"flags": map[string]interface{}(notifications),
 					},
 				}
 			}

flagd/pkg/service/flag-evaluation/eventing_test.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/open-feature/flagd/core/pkg/logger"
+	"github.com/open-feature/flagd/core/pkg/model"
 	iservice "github.com/open-feature/flagd/core/pkg/service"
 	"github.com/open-feature/flagd/core/pkg/store"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 func TestSubscribe(t *testing.T) {
@@ -71,3 +74,90 @@ func TestUnsubscribe(t *testing.T) {
 		"expected subscription cleared, but value present: %v", eventing.subs[idA])
 	require.Equal(t, chanB, eventing.subs[idB], "incorrect subscription association")
 }
+
+// TestNotificationCompatibleWithStructpb verifies that notification data from
+// flag change events can be converted to protobuf structs, as required by the
+// EventStream handlers. This is a regression test for
+// https://github.com/open-feature/flagd/discussions/1869
+func TestNotificationCompatibleWithStructpb(t *testing.T) {
+	sources := []string{"source1"}
+	log := logger.NewLogger(nil, false)
+	s, err := store.NewStore(log, sources)
+	require.NoError(t, err)
+
+	eventing := &eventingConfiguration{
+		subs:   make(map[interface{}]chan iservice.Notification),
+		mu:     &sync.RWMutex{},
+		store:  s,
+		logger: log,
+	}
+
+	notifyChan := make(chan iservice.Notification, 1)
+	eventing.Subscribe(context.Background(), "test", nil, notifyChan)
+	// allow the subscription goroutine to process the initial watch result
+	time.Sleep(100 * time.Millisecond)
+
+	// first update sets up oldFlags.
+	s.Update(sources[0], []model.Flag{
+		{Key: "flag1", DefaultVariant: "off"},
+	}, model.Metadata{})
+
+	// second update triggers a ConfigurationChange with a real diff.
+	s.Update(sources[0], []model.Flag{
+		{Key: "flag1", DefaultVariant: "on"},
+	}, model.Metadata{})
+
+	select {
+	case n := <-notifyChan:
+		require.Equal(t, iservice.ConfigurationChange, n.Type)
+		// contains a named map type instead of plain map[string]interface{}.
+		_, err := structpb.NewStruct(n.Data)
+		require.NoError(t, err, "notification data must be compatible with structpb.NewStruct")
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for notification")
+	}
+}
+
+// TestNoNotificationWhenFlagsUnchanged verifies that no ConfigurationChange
+// notification is sent when a store update contains the same flags as before.
+func TestNoNotificationWhenFlagsUnchanged(t *testing.T) {
+	sources := []string{"source1"}
+	log := logger.NewLogger(nil, false)
+	s, err := store.NewStore(log, sources)
+	require.NoError(t, err)
+
+	eventing := &eventingConfiguration{
+		subs:   make(map[interface{}]chan iservice.Notification),
+		mu:     &sync.RWMutex{},
+		store:  s,
+		logger: log,
+	}
+
+	notifyChan := make(chan iservice.Notification, 1)
+	eventing.Subscribe(context.Background(), "test", nil, notifyChan)
+	time.Sleep(100 * time.Millisecond)
+
+	// first update creates flag1 — this produces a notification (create).
+	s.Update(sources[0], []model.Flag{
+		{Key: "flag1", DefaultVariant: "off"},
+	}, model.Metadata{})
+
+	// drain the first notification (flag creation).
+	select {
+	case <-notifyChan:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for first notification")
+	}
+
+	// second update with the same flags — should not produce a notification.
+	s.Update(sources[0], []model.Flag{
+		{Key: "flag1", DefaultVariant: "off"},
+	}, model.Metadata{})
+
+	select {
+	case n := <-notifyChan:
+		t.Fatalf("unexpected notification received: %v", n)
+	case <-time.After(500 * time.Millisecond):
+		// expected: no notification sent
+	}
+}