fixup: review feedback from giovanni

Signed-off-by: Todd Baert <[email protected]>

Author: toddbaert

CONTRIBUTING.md

@@ -83,6 +83,13 @@ curl -X POST  -d '{"context":{}}' 'http://localhost:8016/ofrep/v1/evaluate/flags
 grpcurl -import-path schemas/protobuf/flagd/evaluation/v1/ -proto evaluation.proto -plaintext -d '{}' localhost:8013 flagd.evaluation.v1.Service/ResolveAll | jq
 ```
 
+#### Remote event streaming via gRPC
+
+```sh
+# notifies of flag changes (but does not evaluate)
+grpcurl -import-path schemas/protobuf/flagd/evaluation/v1/ -proto evaluation.proto -plaintext -d '{}' localhost:8013 flagd.evaluation.v1.Service/EventStream
+```
+
 #### Flag configuration fetch via gRPC
 
 ```sh

core/pkg/evaluator/json.go

@@ -308,13 +308,18 @@ func (je *Resolver) evaluateVariant(ctx context.Context, reqID string, flagKey s
 	variant string, variants map[string]interface{}, reason string, metadata map[string]interface{}, err error,
 ) {
 
-	selector := store.NewSelector("")
+	var selector store.Selector
 	s := ctx.Value(store.SelectorContextKey{})
 	if s != nil {
 		selector = s.(store.Selector)
+	} else {
+		selector = store.NewSelector("")
 	}
-	flag, metadata, ok := je.store.Get(ctx, flagKey, selector)
-	if !ok {
+	if s != nil {
+		selector = s.(store.Selector)
+	}
+	flag, metadata, err := je.store.Get(ctx, flagKey, selector)
+	if err != nil {
 		// flag not found
 		je.Logger.DebugWithID(reqID, fmt.Sprintf("requested flag could not be found: %s", flagKey))
 		return "", map[string]interface{}{}, model.ErrorReason, metadata, errors.New(model.FlagNotFoundErrorCode)

core/pkg/store/store.go

@@ -22,7 +22,7 @@ type FlagQueryResult struct {
 }
 
 type IStore interface {
-	Get(ctx context.Context, key string, selector Selector) (model.Flag, model.Metadata, bool)
+	Get(ctx context.Context, key string, selector Selector) (model.Flag, model.Metadata, error)
 	GetAll(ctx context.Context, selector Selector) (map[string]model.Flag, model.Metadata, error)
 	Watch(ctx context.Context, selector Selector, watcher chan FlagQueryResult)
 }
@@ -151,7 +151,7 @@ func NewFlags() *Store {
 	return state
 }
 
-func (s *Store) Get(_ context.Context, key string, selector Selector) (model.Flag, model.Metadata, bool) {
+func (s *Store) Get(_ context.Context, key string, selector Selector) (model.Flag, model.Metadata, error) {
 	s.logger.Debug(fmt.Sprintf("getting flag %s", key))
 	txn := s.db.Txn(false)
 	queryMeta := model.Metadata{}
@@ -165,33 +165,36 @@ func (s *Store) Get(_ context.Context, key string, selector Selector) (model.Fla
 		raw, err := txn.First(flagsTable, indexId, constraints...)
 		flag, ok := raw.(model.Flag)
 		if (err != nil) || !ok {
-			return model.Flag{}, queryMeta, false
+			return model.Flag{}, queryMeta, fmt.Errorf("flag %s not found: %w", key, err)
 		}
-		return flag, queryMeta, true
+		return flag, queryMeta, nil
 
 	}
 	// otherwise, get all flags with the given key, and keep the last one with the highest priority
 	s.logger.Debug(fmt.Sprintf("getting highest priority flag with key: %s", key))
 	it, err := txn.Get(flagsTable, keyIndex, key)
 	if err != nil {
-		return model.Flag{}, queryMeta, false
+		return model.Flag{}, queryMeta, fmt.Errorf("flag %s not found: %w", key, err)
 	}
 	flag := model.Flag{}
-	var found bool
+	found := false
 	for raw := it.Next(); raw != nil; raw = it.Next() {
-		s.logger.Debug(fmt.Sprintf("got range scan: %v", raw))
 		nextFlag, ok := raw.(model.Flag)
-		found = true
 		if !ok {
 			continue
 		}
+		found = true
 		if nextFlag.Priority >= flag.Priority {
 			flag = nextFlag
 		} else {
 			s.logger.Debug(fmt.Sprintf("discarding flag %s from lower priority source %s in favor of flag from source %s", nextFlag.Key, s.sources[nextFlag.Priority], s.sources[flag.Priority]))
 		}
 	}
-	return flag, queryMeta, found
+
+	if !found {
+		return flag, queryMeta, fmt.Errorf("flag %s not found", key)
+	}
+	return flag, queryMeta, nil
 }
 
 func (f *Store) String() (string, error) {

core/pkg/store/store_test.go

@@ -174,46 +174,46 @@ func TestGet(t *testing.T) {
 
 	t.Parallel()
 	tests := []struct {
-		name      string
-		key       string
-		selector  Selector
-		wantFlag  model.Flag
-		wantFound bool
+		name     string
+		key      string
+		selector Selector
+		wantFlag model.Flag
+		wantErr  bool
 	}{
 		{
-			name:      "nil selector",
-			key:       "flagA",
-			selector:  nil,
-			wantFlag:  model.Flag{Key: "flagA", DefaultVariant: "off", Source: sourceA, FlagSetId: nilFlagSetId, Priority: 0},
-			wantFound: true,
+			name:     "nil selector",
+			key:      "flagA",
+			selector: nil,
+			wantFlag: model.Flag{Key: "flagA", DefaultVariant: "off", Source: sourceA, FlagSetId: nilFlagSetId, Priority: 0},
+			wantErr:  false,
 		},
 		{
-			name:      "flagSetId selector",
-			key:       "dupe",
-			selector:  flagSetIdCSelector,
-			wantFlag:  model.Flag{Key: "dupe", DefaultVariant: "off", Source: sourceC, FlagSetId: flagSetIdC, Priority: 2, Metadata: model.Metadata{"flagSetId": flagSetIdC}},
-			wantFound: true,
+			name:     "flagSetId selector",
+			key:      "dupe",
+			selector: flagSetIdCSelector,
+			wantFlag: model.Flag{Key: "dupe", DefaultVariant: "off", Source: sourceC, FlagSetId: flagSetIdC, Priority: 2, Metadata: model.Metadata{"flagSetId": flagSetIdC}},
+			wantErr:  false,
 		},
 		{
-			name:      "source selector",
-			key:       "dupe",
-			selector:  sourceASelector,
-			wantFlag:  model.Flag{Key: "dupe", DefaultVariant: "on", Source: sourceA, FlagSetId: nilFlagSetId, Priority: 0},
-			wantFound: true,
+			name:     "source selector",
+			key:      "dupe",
+			selector: sourceASelector,
+			wantFlag: model.Flag{Key: "dupe", DefaultVariant: "on", Source: sourceA, FlagSetId: nilFlagSetId, Priority: 0},
+			wantErr:  false,
 		},
 		{
-			name:      "flag not found with source selector",
-			key:       "flagB",
-			selector:  sourceASelector,
-			wantFlag:  model.Flag{Key: "flagB", DefaultVariant: "off", Source: sourceB, FlagSetId: flagSetIdB, Priority: 1, Metadata: model.Metadata{"flagSetId": flagSetIdB}},
-			wantFound: false,
+			name:     "flag not found with source selector",
+			key:      "flagB",
+			selector: sourceASelector,
+			wantFlag: model.Flag{Key: "flagB", DefaultVariant: "off", Source: sourceB, FlagSetId: flagSetIdB, Priority: 1, Metadata: model.Metadata{"flagSetId": flagSetIdB}},
+			wantErr:  true,
 		},
 		{
-			name:      "flag not found with flagSetId selector",
-			key:       "flagB",
-			selector:  flagSetIdCSelector,
-			wantFlag:  model.Flag{Key: "flagB", DefaultVariant: "off", Source: sourceB, FlagSetId: flagSetIdB, Priority: 1, Metadata: model.Metadata{"flagSetId": flagSetIdB}},
-			wantFound: false,
+			name:     "flag not found with flagSetId selector",
+			key:      "flagB",
+			selector: flagSetIdCSelector,
+			wantFlag: model.Flag{Key: "flagB", DefaultVariant: "off", Source: sourceB, FlagSetId: flagSetIdB, Priority: 1, Metadata: model.Metadata{"flagSetId": flagSetIdB}},
+			wantErr:  true,
 		},
 	}
 
@@ -242,12 +242,12 @@ func TestGet(t *testing.T) {
 			store.Update(sourceA, sourceAFlags, nil)
 			store.Update(sourceB, sourceBFlags, nil)
 			store.Update(sourceC, sourceCFlags, nil)
-			gotFlag, _, found := store.Get(context.Background(), tt.key, tt.selector)
-
-			require.Equal(t, tt.wantFound, found, "expected found to be %v, got %v", tt.wantFound, found)
+			gotFlag, _, err := store.Get(context.Background(), tt.key, tt.selector)
 
-			if tt.wantFound {
+			if !tt.wantErr {
 				require.Equal(t, tt.wantFlag, gotFlag)
+			} else {
+				require.Error(t, err, "expected an error for key %s with selector %v", tt.key, tt.selector)
 			}
 		})
 	}

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

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -214,6 +215,53 @@ func TestConnectServiceNotify(t *testing.T) {
 	}
 }
 
+func TestConnectServiceWatcher(t *testing.T) {
+	sources := []string{"source1", "source2"}
+	log := logger.NewLogger(nil, false)
+	s, err := store.NewStore(log, sources)
+
+	if err != nil {
+		t.Fatalf("NewStore failed: %v", err)
+	}
+
+	sChan := make(chan iservice.Notification, 1)
+	eventing := eventingConfiguration{
+		store:  s,
+		logger: log,
+		mu:     &sync.RWMutex{},
+		subs:   make(map[any]chan iservice.Notification),
+	}
+
+	// subscribe and wait for for the sub to be active
+	eventing.Subscribe(context.Background(), "anything", nil, sChan)
+	time.Sleep(100 * time.Millisecond)
+
+	// make a change
+	s.Update(sources[0], map[string]model.Flag{
+		"flag1": {
+			Key:            "flag1",
+			DefaultVariant: "off",
+		},
+	}, model.Metadata{})
+
+	// notification type
+	ofType := iservice.ConfigurationChange
+
+	timeout, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	select {
+	case n := <-sChan:
+		require.Equal(t, ofType, n.Type, "expected notification type: %s, but received %s", ofType, n.Type)
+		notifications := n.Data["flags"].(map[string]interface{})
+		flag1, ok := notifications["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():
+		t.Error("timeout while waiting for notifications")
+	}
+}
+
 func TestConnectServiceShutdown(t *testing.T) {
 	// given
 	ctrl := gomock.NewController(t)

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

@@ -37,17 +37,21 @@ func (eventing *eventingConfiguration) Subscribe(ctx context.Context, id any, se
 	watcher := make(chan store.FlagQueryResult, 1)
 	go func() {
 		// store the previous flags to compare against new notifications, to compute proper diffs for RPC mode
-		oldFlags := map[string]model.Flag{}
-		for payload := range watcher {
-			newFlags := payload.Flags
-			notifications := notifications.NewFromFlags(oldFlags, newFlags)
-			notifier <- iservice.Notification{
-				Type: iservice.ConfigurationChange,
-				Data: map[string]interface{}{
-					"flags": notifications,
-				},
+		var oldFlags map[string]model.Flag
+		for result := range watcher {
+			newFlags := result.Flags
+
+			// 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)
+				notifier <- iservice.Notification{
+					Type: iservice.ConfigurationChange,
+					Data: map[string]interface{}{
+						"flags": notifications,
+					},
+				}
 			}
-			oldFlags = newFlags
+			oldFlags = result.Flags
 		}
 
 		eventing.logger.Debug(fmt.Sprintf("closing notify channel for id %v", id))