fix(store): scope flag deletion by flagSetId to support per-selector sync

  When an upstream provider sends per-selector stream messages (each with
  a flagSetId in metadata), store.Update now only deletes stale flags
  within the same flagSetId(s), preserving flags from other selector.

Signed-off-by: Eren Atas <eren_atas@hotmail.com>

Author: erenatas

core/pkg/store/query.go

@@ -136,10 +136,17 @@ func (s *Selector) ToMetadata() model.Metadata {
 }
 
 func (s *Selector) ToLogString() string {
-	if s != nil && len(s.indexMap) == 1 {
-		for k, v := range s.indexMap {
-			return fmt.Sprintf("'%s=%s'", k, v)
-		}
+	if s == nil || len(s.indexMap) == 0 {
+		return "<none>"
 	}
-	return "<none>"
+	keys := make([]string, 0, len(s.indexMap))
+	for k := range s.indexMap {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	parts := make([]string, 0, len(keys))
+	for _, k := range keys {
+		parts = append(parts, fmt.Sprintf("%s=%s", k, s.indexMap[k]))
+	}
+	return "'" + strings.Join(parts, ",") + "'"
 }

core/pkg/store/store.go

@@ -254,9 +254,41 @@ func (s *Store) Update(
 	txn := s.db.Txn(true)
 	defer txn.Abort()
 
-	// get all flags for the source we are updating
-	selector := NewSelector(sourceIndex + "=" + source)
-	oldFlags, _, _ := s.GetAll(context.Background(), &selector)
+	// When metadata carries a flagSetId, scope deletion to only the flagSetIds touched by
+	// this update (the metadata-level one plus any flag-level overrides). This allows
+	// per-flagSetId updates (e.g., from per-project stream messages) to accumulate in
+	// the store without deleting flags from unrelated flagSetIds.
+	// When metadata has no flagSetId, fall back to source-scoped deletion (original behavior).
+	// Note: we keep the fsi != "" guard because empty-string flagSetIds are not normalized
+	// to nilFlagSetId at storage time, but NewSelector DOES normalize them, creating a
+	// mismatch that would cause queries to miss the actual flags. Source-scoped fallback
+	// is the correct behavior for empty-string flagSetIds.
+	var oldFlags []model.Flag
+	if fsi, ok := metadata["flagSetId"].(string); ok && fsi != "" {
+		seenFlagSetIds := map[string]struct{}{fsi: {}}
+		for id := range newFlags {
+			seenFlagSetIds[id.flagSetId] = struct{}{}
+		}
+		for fsi := range seenFlagSetIds {
+			sel := NewSelector(flagSetIdIndex + "=" + fsi).WithIndex(sourceIndex, source)
+			indexId, constraints := sel.ToQuery()
+			it, err := txn.Get(flagsTable, indexId, constraints...)
+			if err != nil {
+				s.logger.Error(fmt.Sprintf("unable to query flags for flagSetId %s: %v", fsi, err))
+				continue
+			}
+			oldFlags = append(oldFlags, s.collect(it)...)
+		}
+	} else {
+		sel := NewSelector(sourceIndex + "=" + source)
+		indexId, constraints := sel.ToQuery()
+		it, err := txn.Get(flagsTable, indexId, constraints...)
+		if err != nil {
+			s.logger.Error(fmt.Sprintf("unable to query flags for source %s: %v", source, err))
+		} else {
+			oldFlags = s.collect(it)
+		}
+	}
 
 	for _, oldFlag := range oldFlags {
 		if _, ok := newFlags[flagIdentifier{flagSetId: oldFlag.FlagSetId, key: oldFlag.Key}]; !ok {

core/pkg/store/store_test.go

@@ -618,6 +618,141 @@ func TestWatch(t *testing.T) {
 	}
 }
 
+func TestUpdateFlagSetIdScoping(t *testing.T) {
+	t.Parallel()
+
+	const src = "src1"
+	sources := []string{src}
+
+	type updateStep struct {
+		flags    []model.Flag
+		metadata model.Metadata
+	}
+
+	tests := []struct {
+		name        string
+		updates     []updateStep
+		wantPresent []string // "flagSetId/key" entries expected in the store
+		wantAbsent  []string // "flagSetId/key" entries expected to be gone
+	}{
+		{
+			name: "per-flagSetId update preserves flags from other flagSetIds",
+			updates: []updateStep{
+				{flags: []model.Flag{{Key: "flagA1"}, {Key: "flagA2"}}, metadata: model.Metadata{"flagSetId": "A"}},
+				{flags: []model.Flag{{Key: "flagB1"}}, metadata: model.Metadata{"flagSetId": "B"}},
+				{flags: []model.Flag{{Key: "flagA1"}}, metadata: model.Metadata{"flagSetId": "A"}},
+			},
+			wantPresent: []string{"A/flagA1", "B/flagB1"},
+			wantAbsent:  []string{"A/flagA2"},
+		},
+		{
+			name: "out-of-scope flag-level override persists when not in batch",
+			updates: []updateStep{
+				{flags: []model.Flag{{Key: "kept"}, {Key: "override", Metadata: model.Metadata{"flagSetId": "Y"}}}, metadata: model.Metadata{"flagSetId": "X"}},
+				{flags: []model.Flag{{Key: "kept"}}, metadata: model.Metadata{"flagSetId": "X"}},
+			},
+			wantPresent: []string{"X/kept", "Y/override"},
+		},
+		{
+			name: "stale flag deleted when its flagSetId is in scope",
+			updates: []updateStep{
+				{flags: []model.Flag{{Key: "inX"}, {Key: "inY", Metadata: model.Metadata{"flagSetId": "Y"}}}, metadata: model.Metadata{"flagSetId": "X"}},
+				{flags: []model.Flag{{Key: "inY", Metadata: model.Metadata{"flagSetId": "Y"}}}, metadata: model.Metadata{"flagSetId": "X"}},
+			},
+			wantPresent: []string{"Y/inY"},
+			wantAbsent:  []string{"X/inX"},
+		},
+		{
+			name: "no flagSetId in metadata falls back to source-scoped deletion",
+			updates: []updateStep{
+				{flags: []model.Flag{{Key: "flagA"}}, metadata: model.Metadata{"flagSetId": "A"}},
+				{flags: []model.Flag{{Key: "flagB"}}, metadata: model.Metadata{"flagSetId": "B"}},
+				{flags: []model.Flag{}, metadata: nil},
+			},
+			wantAbsent: []string{"A/flagA", "B/flagB"},
+		},
+		{
+			name: "empty update with flagSetId clears only that set",
+			updates: []updateStep{
+				{flags: []model.Flag{{Key: "flagA"}}, metadata: model.Metadata{"flagSetId": "A"}},
+				{flags: []model.Flag{{Key: "flagB"}}, metadata: model.Metadata{"flagSetId": "B"}},
+				{flags: []model.Flag{}, metadata: model.Metadata{"flagSetId": "A"}},
+			},
+			wantPresent: []string{"B/flagB"},
+			wantAbsent:  []string{"A/flagA"},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			s, err := NewStore(logger.NewLogger(nil, false), sources)
+			require.NoError(t, err)
+
+			for _, step := range tt.updates {
+				s.Update(src, step.flags, step.metadata)
+			}
+
+			allFlags, _, _ := s.GetAll(context.Background(), nil)
+			flagKeys := make(map[string]struct{}, len(allFlags))
+			for _, f := range allFlags {
+				flagKeys[f.FlagSetId+"/"+f.Key] = struct{}{}
+			}
+
+			for _, key := range tt.wantPresent {
+				assert.Contains(t, flagKeys, key)
+			}
+			for _, key := range tt.wantAbsent {
+				assert.NotContains(t, flagKeys, key)
+			}
+		})
+	}
+}
+
+func TestToLogStringCompound(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		selector *Selector
+		want     string
+	}{
+		{
+			name:     "nil selector",
+			selector: nil,
+			want:     "<none>",
+		},
+		{
+			name:     "empty selector",
+			selector: &Selector{indexMap: map[string]string{}},
+			want:     "<none>",
+		},
+		{
+			name:     "single key",
+			selector: &Selector{indexMap: map[string]string{"source": "mySource"}},
+			want:     "'source=mySource'",
+		},
+		{
+			name:     "compound selector",
+			selector: &Selector{indexMap: map[string]string{"flagSetId": "abc", "source": "mySource"}},
+			want:     "'flagSetId=abc,source=mySource'",
+		},
+		{
+			name:     "three keys sorted",
+			selector: &Selector{indexMap: map[string]string{"source": "s", "key": "k", "flagSetId": "f"}},
+			want:     "'flagSetId=f,key=k,source=s'",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.selector.ToLogString()
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
 func TestQueryMetadata(t *testing.T) {
 
 	sourceA := "sourceA"