feat: Normalize selector handling across sync, evaluation, and OFREP services

flagd/pkg/service/flag-sync/handler.go

@@ -11,12 +11,14 @@
 	"github.com/open-feature/flagd/core/pkg/model"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 
 	"buf.build/gen/go/open-feature/flagd/grpc/go/flagd/sync/v1/syncv1grpc"
 	syncv1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/store"
+	flagdService "github.com/open-feature/flagd/flagd/pkg/service"
 	"google.golang.org/protobuf/types/known/structpb"
 )
 
@@ -32,7 +34,7 @@
 
 func (s syncHandler) SyncFlags(req *syncv1.SyncFlagsRequest, server syncv1grpc.FlagSyncService_SyncFlagsServer) error {
 	watcher := make(chan store.FlagQueryResult, 1)
-	selectorExpression := req.GetSelector()
+	selectorExpression := s.getSelectorExpression(server.Context(), req)
 	selector := store.NewSelector(selectorExpression)
 	ctx := server.Context()
 
@@ -85,6 +87,39 @@
 	}
 }
 
+// getSelectorExpression extracts the selector expression from the request.
+// It first checks the Flagd-Selector header (metadata), then falls back to the request body selector.
+// A deprecation warning is logged when the request body selector is used.
+//
+// The req parameter accepts *syncv1.SyncFlagsRequest or *syncv1.FetchAllFlagsRequest.
+// Using interface{} here is intentional as both protobuf-generated types implement GetSelector()
+// but do not share a common interface.
+func (s syncHandler) getSelectorExpression(ctx context.Context, req interface{}) string {
+	// Try to get selector from metadata (header)
+	if md, ok := metadata.FromIncomingContext(ctx); ok {
+		if values := md.Get(flagdService.FLAGD_SELECTOR_HEADER); len(values) > 0 {
+			return values[0]
+		}
+	}
+
+	// Fall back to request body selector for backward compatibility
+	var bodySelector string
+	switch r := req.(type) {
+	case *syncv1.SyncFlagsRequest:
+		bodySelector = r.GetSelector()
+	case *syncv1.FetchAllFlagsRequest:
+		bodySelector = r.GetSelector()
+	}
+
+	// Log deprecation warning if using request body selector
+	if bodySelector != "" {
+		s.log.Warn("Using selector from request body is deprecated. Please use the 'Flagd-Selector' header instead. " +
+			"Request body selector support will be removed in a future major version.")
+	}
+
+	return bodySelector
+}
+
 func (s syncHandler) convertMap(flags []model.Flag) map[string]model.Flag {
 	flagMap := make(map[string]model.Flag, len(flags))
 	for _, flag := range flags {
@@ -96,7 +131,7 @@
 func (s syncHandler) FetchAllFlags(ctx context.Context, req *syncv1.FetchAllFlagsRequest) (
 	*syncv1.FetchAllFlagsResponse, error,
 ) {
-	selectorExpression := req.GetSelector()
+	selectorExpression := s.getSelectorExpression(ctx, req)
 	selector := store.NewSelector(selectorExpression)
 	flags, _, err := s.store.GetAll(ctx, &selector)
 	if err != nil {
@@ -117,8 +152,8 @@
 
 // Deprecated - GetMetadata is deprecated and will be removed in a future release.
 // Use the sync_context field in syncv1.SyncFlagsResponse, providing same info.
 func (s syncHandler) GetMetadata(_ context.Context, _ *syncv1.GetMetadataRequest) (
 	*syncv1.GetMetadataResponse, error,
 ) {
 	if s.disableSyncMetadata {
 		return nil, status.Error(codes.Unimplemented, "metadata endpoint disabled")
@@ -134,7 +169,7 @@
 		return nil, fmt.Errorf("error constructing metadata response")
 	}
 
 	return &syncv1.GetMetadataResponse{
 			Metadata: metadata,
 		},
 		nil

flagd/pkg/service/flag-sync/handler_test.go

@@ -10,8 +10,13 @@ import (
 	syncv1 "buf.build/gen/go/open-feature/flagd/protocolbuffers/go/flagd/sync/v1"
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/store"
+	flagdService "github.com/open-feature/flagd/flagd/pkg/service"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+	"google.golang.org/grpc/metadata"
 )
 
 func TestSyncHandler_SyncFlags(t *testing.T) {
@@ -128,3 +133,220 @@ func (m *mockSyncFlagsServer) GetLastResponse() *syncv1.SyncFlagsResponse {
 	defer m.mu.Unlock()
 	return m.lastResp
 }
+
+// TestSyncHandler_SelectorFromHeader tests that the selector is correctly extracted from the header
+func TestSyncHandler_SelectorFromHeader(t *testing.T) {
+	flagStore, err := store.NewStore(logger.NewLogger(nil, false), []string{})
+	require.NoError(t, err)
+
+	// Create a logger with observer to capture log messages
+	observedZapCore, observedLogs := observer.New(zapcore.WarnLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, false)
+
+	handler := syncHandler{
+		store:         flagStore,
+		log:           log,
+		contextValues: map[string]any{},
+	}
+
+	// Create context with metadata containing the selector header
+	md := metadata.New(map[string]string{
+		flagdService.FLAGD_SELECTOR_HEADER: "source:my-source",
+	})
+	ctx := metadata.NewIncomingContext(context.Background(), md)
+
+	// Test with SyncFlags
+	stream := &mockSyncFlagsServer{
+		ctx:       ctx,
+		mu:        sync.Mutex{},
+		respReady: make(chan struct{}, 1),
+	}
+
+	go func() {
+		// Use empty request body selector to verify header is used
+		err := handler.SyncFlags(&syncv1.SyncFlagsRequest{Selector: ""}, stream)
+		assert.NoError(t, err)
+	}()
+
+	select {
+	case <-stream.respReady:
+		// Verify no deprecation warning was logged
+		logs := observedLogs.All()
+		for _, log := range logs {
+			assert.NotContains(t, log.Message, "deprecated", "Should not log deprecation warning when using header")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for response")
+	}
+}
+
+// TestSyncHandler_SelectorFromRequestBody tests backward compatibility with request body selector
+func TestSyncHandler_SelectorFromRequestBody(t *testing.T) {
+	flagStore, err := store.NewStore(logger.NewLogger(nil, false), []string{})
+	require.NoError(t, err)
+
+	// Create a logger with observer to capture log messages
+	observedZapCore, observedLogs := observer.New(zapcore.WarnLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, false)
+
+	handler := syncHandler{
+		store:         flagStore,
+		log:           log,
+		contextValues: map[string]any{},
+	}
+
+	// Create context without metadata (no header)
+	ctx := context.Background()
+
+	// Test with SyncFlags
+	stream := &mockSyncFlagsServer{
+		ctx:       ctx,
+		mu:        sync.Mutex{},
+		respReady: make(chan struct{}, 1),
+	}
+
+	go func() {
+		// Use request body selector
+		err := handler.SyncFlags(&syncv1.SyncFlagsRequest{Selector: "source:legacy-source"}, stream)
+		assert.NoError(t, err)
+	}()
+
+	select {
+	case <-stream.respReady:
+		// Verify deprecation warning was logged
+		logs := observedLogs.All()
+		require.Greater(t, len(logs), 0, "Expected at least one log entry")
+		found := false
+		for _, log := range logs {
+			if log.Level == zapcore.WarnLevel {
+				assert.Contains(t, log.Message, "deprecated", "Should log deprecation warning when using request body selector")
+				assert.Contains(t, log.Message, "Flagd-Selector", "Deprecation message should mention the header name")
+				found = true
+				break
+			}
+		}
+		assert.True(t, found, "Expected to find deprecation warning in logs")
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for response")
+	}
+}
+
+// TestSyncHandler_SelectorHeaderTakesPrecedence tests that header takes precedence over request body
+func TestSyncHandler_SelectorHeaderTakesPrecedence(t *testing.T) {
+	flagStore, err := store.NewStore(logger.NewLogger(nil, false), []string{})
+	require.NoError(t, err)
+
+	// Create a logger with observer to capture log messages
+	observedZapCore, observedLogs := observer.New(zapcore.WarnLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, false)
+
+	handler := syncHandler{
+		store:         flagStore,
+		log:           log,
+		contextValues: map[string]any{},
+	}
+
+	// Create context with metadata containing the selector header
+	md := metadata.New(map[string]string{
+		flagdService.FLAGD_SELECTOR_HEADER: "source:header-source",
+	})
+	ctx := metadata.NewIncomingContext(context.Background(), md)
+
+	// Test with SyncFlags
+	stream := &mockSyncFlagsServer{
+		ctx:       ctx,
+		mu:        sync.Mutex{},
+		respReady: make(chan struct{}, 1),
+	}
+
+	go func() {
+		// Provide both header and request body selector
+		err := handler.SyncFlags(&syncv1.SyncFlagsRequest{Selector: "source:body-source"}, stream)
+		assert.NoError(t, err)
+	}()
+
+	select {
+	case <-stream.respReady:
+		// Verify no deprecation warning was logged (header was used)
+		logs := observedLogs.All()
+		for _, log := range logs {
+			assert.NotContains(t, log.Message, "deprecated", "Should not log deprecation warning when header is present")
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for response")
+	}
+}
+
+// TestSyncHandler_FetchAllFlags_SelectorFromHeader tests FetchAllFlags with header selector
+func TestSyncHandler_FetchAllFlags_SelectorFromHeader(t *testing.T) {
+	flagStore, err := store.NewStore(logger.NewLogger(nil, false), []string{})
+	require.NoError(t, err)
+
+	// Create a logger with observer to capture log messages
+	observedZapCore, observedLogs := observer.New(zapcore.WarnLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, false)
+
+	handler := syncHandler{
+		store:         flagStore,
+		log:           log,
+		contextValues: map[string]any{},
+	}
+
+	// Create context with metadata containing the selector header
+	md := metadata.New(map[string]string{
+		flagdService.FLAGD_SELECTOR_HEADER: "source:my-source",
+	})
+	ctx := metadata.NewIncomingContext(context.Background(), md)
+
+	// Call FetchAllFlags with empty request body selector
+	_, err = handler.FetchAllFlags(ctx, &syncv1.FetchAllFlagsRequest{Selector: ""})
+	require.NoError(t, err)
+
+	// Verify no deprecation warning was logged
+	logs := observedLogs.All()
+	for _, log := range logs {
+		assert.NotContains(t, log.Message, "deprecated", "Should not log deprecation warning when using header")
+	}
+}
+
+// TestSyncHandler_FetchAllFlags_SelectorFromRequestBody tests FetchAllFlags with request body selector
+func TestSyncHandler_FetchAllFlags_SelectorFromRequestBody(t *testing.T) {
+	flagStore, err := store.NewStore(logger.NewLogger(nil, false), []string{})
+	require.NoError(t, err)
+
+	// Create a logger with observer to capture log messages
+	observedZapCore, observedLogs := observer.New(zapcore.WarnLevel)
+	observedLogger := zap.New(observedZapCore)
+	log := logger.NewLogger(observedLogger, false)
+
+	handler := syncHandler{
+		store:         flagStore,
+		log:           log,
+		contextValues: map[string]any{},
+	}
+
+	// Create context without metadata (no header)
+	ctx := context.Background()
+
+	// Call FetchAllFlags with request body selector
+	_, err = handler.FetchAllFlags(ctx, &syncv1.FetchAllFlagsRequest{Selector: "source:legacy-source"})
+	require.NoError(t, err)
+
+	// Verify deprecation warning was logged
+	logs := observedLogs.All()
+	require.Greater(t, len(logs), 0, "Expected at least one log entry")
+	found := false
+	for _, log := range logs {
+		if log.Level == zapcore.WarnLevel {
+			assert.Contains(t, log.Message, "deprecated", "Should log deprecation warning when using request body selector")
+			assert.Contains(t, log.Message, "Flagd-Selector", "Deprecation message should mention the header name")
+			found = true
+			break
+		}
+	}
+	assert.True(t, found, "Expected to find deprecation warning in logs")
+}