feat: multi-project support via selectors and flagSetId namespacing (#1702)

Sorry for how big this PR seems (a lot of the change lines are not
significant (DB schema or tests), so I've tried to highlight the
important parts. This is a substantial PR that builds on our recent
changes to use go-memdb for storage. It primarily supports 2 new crucial
features:

- support for duplicate flag keys from multiple sources (namespaced by
`flagSetId`) as described in [this
ADR](https://github.com/open-feature/flagd/blob/main/docs/architecture-decisions/duplicate-flag-keys.md)
- support for a robust query syntax in the "selector" fields and
headers, as proposed by @tangenti in [this
ADR](https://github.com/open-feature/flagd/blob/main/docs/architecture-decisions/decouple-flag-source-and-set.md)

Both of these were accomplished using the new go-memdb module. **The
built-in "watcher" functionality also allows us to _completely delete_
our "mux" layer, which was responsible for fanning out changes from
sync-sources to listeners**; this is something the go-memdb module
providers for free (the ability to watch a query for changes). Now, we
completely rely on this feature for all change notifications.
Additionally, unlike before, change notifications are now scoped to
particular _selectors_ (ie: if a client is only interested in changes
for flags from `flagSetId: x` or `source: y`, they will only get change
notifications pertaining to that selection. Currently, the only
supported query fields for the `selector` are `"source"` and
`"flagSetId"`, but this functionality can easily be extended. By
default, if no `selector` is specified, the previous key-overwrite by
source priority apples (this logic has also been simplified using the
new database). Most of the new functionality is tested
[here](https://github.com/open-feature/flagd/pull/1702/files#diff-0cdd4fe716f4b8a94466279fd1b11187fcf4d74e6434727c33d57ed78c89fe27R163-R478).

Selector in action for `Sync` API:


![demo](https://github.com/user-attachments/assets/9a4fd33e-80ef-4b5b-80d1-4fa3fc92b8e7)

Selector in action for `Evaluation` API:


![demo2](https://github.com/user-attachments/assets/abf49f9c-2247-4a03-9b8f-15b6f540aad1)

To test, run the new make target `make run-flagd-selector-demo`, then
use the OFREP or gRPC endpoints to experiment. This new functionality is
available on all APIs and endpoints. The command I ran in the gifs above
are:

streaming:
```shell
grpcurl -d '{"selector":"flagSetId=example"}' -import-path schemas/protobuf/flagd/sync/v1/ -proto sync.proto -plaintext localhost:8015 flagd.sync.v1.FlagSyncService/SyncFlags | jq
grpcurl -d '{"selector":"flagSetId=example,source=../config/samples/example_flags.flagd.json"}' -import-path schemas/protobuf/flagd/sync/v1/ -proto sync.proto -plaintext localhost:8015 flagd.sync.v1.FlagSyncService/SyncFlags | jq
grpcurl -d '{"selector":"flagSetId=other"}' -import-path schemas/protobuf/flagd/sync/v1/ -proto sync.proto -plaintext localhost:8015 flagd.sync.v1.FlagSyncService/SyncFlags | jq
```

single-evaluations:
```shell
curl -X POST  -d '{"context":{}}' 'http://localhost:8016/ofrep/v1/evaluate/flags' | jq
curl -X POST -H 'flagd-selector:flagSetId=other'  -d '{"context":{}}' 'http://localhost:8016/ofrep/v1/evaluate/flags' | jq
```

:warning: There's no breaking changes here. Besides the new features,
there is one behavioral change - the top level "metadata" object
returned for bulk evaluations (and failed evaluations) was previously
very nonsensical in it's behavior (we basically just aggregated the
metadata from all sources, discarding duplicates, and sent it back. This
field was used by providers for telemetry purposes. Now, since flag
evaluations and subscripts are "query based" and can aggregate data from
multiple sources, we've opted to simply reflect the selector queries
contents here.

So if you used a selector like `"flagSetId=1234,source=../my/source"`,
the top-level metadata object in the response would be:

```
"metadata": {
  "flagSetId": 1234,
  "source": "../my/source"
}
```

This is useful for the provider's ability to report errors, etc in
telemetry.

Fixes: #1675
Fixes: #1695
Fixes: #1611
Fixes: #1700
Fixes: #1610

---------

Signed-off-by: Todd Baert <[email protected]>
Co-authored-by: chrfwow <[email protected]>
Co-authored-by: Giovanni Liva <[email protected]>

Author: 3 people

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

Makefile

@@ -47,8 +47,6 @@ test-flagd:
 	go test -race -covermode=atomic -cover -short ./flagd/pkg/... -coverprofile=flagd-coverage.out
 test-flagd-proxy:
 	go test -race -covermode=atomic -cover -short ./flagd-proxy/pkg/... -coverprofile=flagd-proxy-coverage.out
-flagd-integration-test: # dependent on ./bin/flagd start -f file:test-harness/flags/testing-flags.json -f file:test-harness/flags/custom-ops.json -f file:test-harness/flags/evaluator-refs.json -f file:test-harness/flags/zero-flags.json
-	go test -cover ./test/integration $(ARGS)
 flagd-benchmark-test:
 	go test -bench=Bench -short -benchtime=5s -benchmem ./core/... | tee benchmark.txt
 flagd-integration-test-harness:
@@ -59,7 +57,9 @@ flagd-integration-test: # dependent on flagd-e2e-test-harness if not running in
 run: # default to flagd
 	make run-flagd
 run-flagd:
-	cd flagd; go run main.go start -f file:../config/samples/example_flags.flagd.json
+	cd flagd; go run main.go start -f file:../config/samples/example_flags.flagd.json 
+run-flagd-selector-demo:
+	cd flagd; go run main.go start -f file:../config/samples/example_flags.flagd.json -f file:../config/samples/example_flags.flagd.2.json
 install:
 	cp systemd/flagd.service /etc/systemd/system/flagd.service
 	mkdir -p /etc/flagd

config/samples/example_flags.flagd.2.json

@@ -0,0 +1,17 @@
+{
+  "$schema": "https://flagd.dev/schema/v0/flags.json",
+  "metadata": {
+    "flagSetId": "other",
+    "version": "v1"
+  },
+  "flags": {
+    "myStringFlag": {
+      "state": "ENABLED",
+      "variants": {
+        "dupe1": "dupe1",
+        "dupe2": "dupe2"
+      },
+      "defaultVariant": "dupe1"
+    }
+  }
+}

core/pkg/evaluator/fractional_test.go

@@ -11,6 +11,8 @@ import (
 )
 
 func TestFractionalEvaluation(t *testing.T) {
+	const source = "testSource"
+	var sources = []string{source}
 	ctx := context.Background()
 
 	commonFlags := Flags{
@@ -458,8 +460,13 @@ func TestFractionalEvaluation(t *testing.T) {
 	for name, tt := range tests {
 		t.Run(name, func(t *testing.T) {
 			log := logger.NewLogger(nil, false)
-			je := NewJSON(log, store.NewFlags())
-			je.store.Update("", "", tt.flags.Flags, model.Metadata{})
+			s, err := store.NewStore(log, sources)
+			if err != nil {
+				t.Fatalf("NewStore failed: %v", err)
+			}
+
+			je := NewJSON(log, s)
+			je.store.Update(source, tt.flags.Flags, model.Metadata{})
 
 			value, variant, reason, _, err := resolve[string](ctx, reqID, tt.flagKey, tt.context, je.evaluateVariant)
 
@@ -486,6 +493,8 @@ func TestFractionalEvaluation(t *testing.T) {
 }
 
 func BenchmarkFractionalEvaluation(b *testing.B) {
+	const source = "testSource"
+	var sources = []string{source}
 	ctx := context.Background()
 
 	flags := Flags{
@@ -587,8 +596,13 @@ func BenchmarkFractionalEvaluation(b *testing.B) {
 	for name, tt := range tests {
 		b.Run(name, func(b *testing.B) {
 			log := logger.NewLogger(nil, false)
-			je := NewJSON(log, store.NewFlags())
-			je.store.Update("", "", tt.flags.Flags, model.Metadata{})
+			s, err := store.NewStore(log, sources)
+			if err != nil {
+				b.Fatalf("NewStore failed: %v", err)
+			}
+			je := NewJSON(log, s)
+			je.store.Update(source, tt.flags.Flags, model.Metadata{})
+
 			for i := 0; i < b.N; i++ {
 				value, variant, reason, _, err := resolve[string](
 					ctx, reqID, tt.flagKey, tt.context, je.evaluateVariant)

core/pkg/evaluator/ievaluator.go

@@ -35,7 +35,7 @@ IEvaluator is an extension of IResolver, allowing storage updates and retrievals
 */
 type IEvaluator interface {
 	GetState() (string, error)
-	SetState(payload sync.DataSync) (model.Metadata, bool, error)
+	SetState(payload sync.DataSync) (map[string]interface{}, bool, error)
 	IResolver
 }
 

core/pkg/evaluator/json.go

@@ -118,7 +118,7 @@ func (je *JSON) SetState(payload sync.DataSync) (map[string]interface{}, bool, e
 	var events map[string]interface{}
 	var reSync bool
 
-	events, reSync = je.store.Update(payload.Source, payload.Selector, definition.Flags, definition.Metadata)
+	events, reSync = je.store.Update(payload.Source, definition.Flags, definition.Metadata)
 
 	// Number of events correlates to the number of flags changed through this sync, record it
 	span.SetAttributes(attribute.Int("feature_flag.change_count", len(events)))
@@ -149,8 +149,12 @@ func (je *Resolver) ResolveAllValues(ctx context.Context, reqID string, context
 	_, span := je.tracer.Start(ctx, "resolveAll")
 	defer span.End()
 
-	var err error
-	allFlags, flagSetMetadata, err := je.store.GetAll(ctx)
+	var selector store.Selector
+	s := ctx.Value(store.SelectorContextKey{})
+	if s != nil {
+		selector = s.(store.Selector)
+	}
+	allFlags, flagSetMetadata, err := je.store.GetAll(ctx, &selector)
 	if err != nil {
 		return nil, flagSetMetadata, fmt.Errorf("error retreiving flags from the store: %w", err)
 	}
@@ -301,19 +305,19 @@ func resolve[T constraints](ctx context.Context, reqID string, key string, conte
 func (je *Resolver) evaluateVariant(ctx context.Context, reqID string, flagKey string, evalCtx map[string]any) (
 	variant string, variants map[string]interface{}, reason string, metadata map[string]interface{}, err error,
 ) {
-	flag, metadata, ok := je.store.Get(ctx, flagKey)
-	if !ok {
+
+	var selector store.Selector
+	s := ctx.Value(store.SelectorContextKey{})
+	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)
 	}
 
-	// add selector to evaluation metadata
-	selector := je.store.SelectorForFlag(ctx, flag)
-	if selector != "" {
-		metadata[SelectorMetadataKey] = selector
-	}
-
 	for key, value := range flag.Metadata {
 		// If value is not nil or empty, copy to metadata
 		if value != nil {