feat: SSE support for OFREP endpoints

core/pkg/service/ofrep/models.go

@@ -22,6 +22,24 @@ type EvaluationSuccess struct {
 type BulkEvaluationResponse struct {
 	Flags    []interface{}  `json:"flags"`
 	Metadata model.Metadata `json:"metadata"`
+	// EventStreams advertises SSE endpoints clients can subscribe to for change
+	// notifications, per OpenFeature protocol ADR-0008. Omitted when SSE is disabled.
+	EventStreams []EventStream `json:"eventStreams,omitempty"`
+}
+
+// EventStream describes a Server-Sent Events endpoint a client can subscribe to in order to
+// be notified (via a `refetchEvaluation` event) when the flag configuration changes.
+type EventStream struct {
+	Type               string               `json:"type"`
+	Endpoint           *EventStreamEndpoint `json:"endpoint"`
+	InactivityDelaySec int                  `json:"inactivityDelaySec,omitempty"`
+}
+
+// EventStreamEndpoint is the ADR-0008 structured form of an event-stream location. Origin is
+// optional; when omitted the client resolves RequestUri against its OFREP base URL origin.
+type EventStreamEndpoint struct {
+	Origin     string `json:"origin,omitempty"`
+	RequestUri string `json:"requestUri"`
 }
 
 type EvaluationError struct {
@@ -54,8 +72,8 @@ func BulkEvaluationResponseFrom(resolutions []evaluator.AnyValue, metadata model
 	}
 
 	return BulkEvaluationResponse{
-		evaluations,
-		metadata,
+		Flags:    evaluations,
+		Metadata: metadata,
 	}
 }
 

core/pkg/store/query.go

@@ -83,6 +83,12 @@ func (s Selector) WithSource(source string) Selector { return s.withIndex(source
 func (s Selector) WithFlagSetId(id string) Selector  { return s.withIndex(flagSetIdIndex, id) }
 func (s Selector) withKey(key string) Selector       { return s.withIndex(keyIndex, key) }
 
+// FlagSetId returns the flagSetId constraint of the selector, or "" if none is set.
+func (s Selector) FlagSetId() string { return s.indexMap[flagSetIdIndex] }
+
+// Source returns the source constraint of the selector, or "" if none is set.
+func (s Selector) Source() string { return s.indexMap[sourceIndex] }
+
 func (s Selector) withIndex(key, value string) Selector {
 	m := maps.Clone(s.indexMap)
 	if m == nil {

docs/reference/flagd-cli/flagd_start.md

@@ -24,6 +24,9 @@ flagd start [flags]
   -R, --max-request-header int               Maximum allowed request header size in bytes. Requests exceeding this are rejected with HTTP 431. Set to 0 to use Go's built-in default (1 MiB). WARNING: setting a very large or zero value may allow memory exhaustion from oversized headers. (default 1000000)
   -t, --metrics-exporter string              Set the metrics exporter. Default(if unset) is Prometheus. Can be override to otel - OpenTelemetry metric exporter. Overriding to otel require otelCollectorURI to be present
   -r, --ofrep-port int32                     ofrep service port (default 8016)
+      --ofrep-sse-enabled                    Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse/{channel} on the ofrep port, where the channel is a selector expression. Defaults to true. (default true)
+      --ofrep-sse-inactivity-delay int       Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120. (default 120)
+      --ofrep-sse-public-url string          Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy.
   -A, --otel-ca-path string                  tls certificate authority path to use with OpenTelemetry collector
   -D, --otel-cert-path string                tls certificate path to use with OpenTelemetry collector
   -o, --otel-collector-uri string            Set the grpc URI of the OpenTelemetry collector for flagd runtime. If unset, the collector setup will be ignored and traces will not be exported.

flagd/cmd/start.go

@@ -23,6 +23,9 @@ const (
 	managementPortFlagName     = "management-port"
 	metricsExporter            = "metrics-exporter"
 	ofrepPortFlagName          = "ofrep-port"
+	ofrepSSEEnabledFlagName    = "ofrep-sse-enabled"
+	ofrepSSEInactivityFlagName = "ofrep-sse-inactivity-delay"
+	ofrepSSEPublicURLFlagName  = "ofrep-sse-public-url"
 	otelCollectorURI           = "otel-collector-uri"
 	otelCertPathFlagName       = "otel-cert-path"
 	otelKeyPathFlagName        = "otel-key-path"
@@ -58,6 +61,9 @@ func init() {
 	flags.Int32P(syncPortFlagName, "g", 8015, "gRPC Sync port")
 	flags.Int32P(ofrepPortFlagName, "r", 8016, "ofrep service port")
 
+	flags.Bool(ofrepSSEEnabledFlagName, true, "Enable the OFREP SSE change-notification endpoint (ADR-0008) at /ofrep/v1/sse/{channel} on the ofrep port, where the channel is a selector expression. Defaults to true.")
+	flags.Int(ofrepSSEInactivityFlagName, 120, "Inactivity delay (seconds) advertised to OFREP SSE clients in the eventStreams block. Clients close idle connections after this. Defaults to 120.")
+	flags.String(ofrepSSEPublicURLFlagName, "", "Origin (scheme://host) advertised as the OFREP SSE eventStreams endpoint.origin. Omitted when empty, so clients resolve the requestUri against the OFREP base URL. Set when flagd is behind a proxy.")
 	flags.StringP(socketPathFlagName, "d", "", "Flagd unix socket path. "+
 		"With grpc the evaluations service will become available on this address. "+
 		"With http(s) the grpc-gateway proxy will use this address internally.")
@@ -123,6 +129,9 @@ func bindFlags(flags *pflag.FlagSet) {
 	_ = viper.BindPFlag(syncPortFlagName, flags.Lookup(syncPortFlagName))
 	_ = viper.BindPFlag(syncSocketPathFlagName, flags.Lookup(syncSocketPathFlagName))
 	_ = viper.BindPFlag(ofrepPortFlagName, flags.Lookup(ofrepPortFlagName))
+	_ = viper.BindPFlag(ofrepSSEEnabledFlagName, flags.Lookup(ofrepSSEEnabledFlagName))
+	_ = viper.BindPFlag(ofrepSSEInactivityFlagName, flags.Lookup(ofrepSSEInactivityFlagName))
+	_ = viper.BindPFlag(ofrepSSEPublicURLFlagName, flags.Lookup(ofrepSSEPublicURLFlagName))
 	_ = viper.BindPFlag(contextValueFlagName, flags.Lookup(contextValueFlagName))
 	_ = viper.BindPFlag(headerToContextKeyFlagName, flags.Lookup(headerToContextKeyFlagName))
 	_ = viper.BindPFlag(streamDeadlineFlagName, flags.Lookup(streamDeadlineFlagName))
@@ -201,6 +210,9 @@ var startCmd = &cobra.Command{
 			MetricExporter:               viper.GetString(metricsExporter),
 			ManagementPort:               viper.GetUint16(managementPortFlagName),
 			OfrepServicePort:             viper.GetUint16(ofrepPortFlagName),
+			OfrepSSEEnabled:              viper.GetBool(ofrepSSEEnabledFlagName),
+			OfrepSSEInactivityDel:        viper.GetInt(ofrepSSEInactivityFlagName),
+			OfrepSSEPublicURL:            viper.GetString(ofrepSSEPublicURLFlagName),
 			OtelCollectorURI:             viper.GetString(otelCollectorURI),
 			OtelCertPath:                 viper.GetString(otelCertPathFlagName),
 			OtelKeyPath:                  viper.GetString(otelKeyPathFlagName),

flagd/go.mod

@@ -9,6 +9,7 @@ require (
 	connectrpc.com/connect v1.19.1
 	github.com/dimiro1/banner v1.1.0
 	github.com/gorilla/mux v1.8.1
+	github.com/launchdarkly/eventsource v1.11.0
 	github.com/mattn/go-colorable v0.1.14
 	github.com/open-feature/flagd/core v0.15.6
 	github.com/prometheus/client_golang v1.23.2

flagd/go.sum

@@ -255,6 +255,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/launchdarkly/eventsource v1.11.0 h1:aAdvh2XmtXA17QsRFL0XKHURMqhxg7J+CceQmhSzBas=
+github.com/launchdarkly/eventsource v1.11.0/go.mod h1:dU+rZxkPOlGPsyJPpiDqiepAcFwIITDUClY9+A6RrMw=
+github.com/launchdarkly/go-test-helpers/v3 v3.1.0 h1:E3bxJMzMoA+cJSF3xxtk2/chr1zshl1ZWa0/oR+8bvg=
+github.com/launchdarkly/go-test-helpers/v3 v3.1.0/go.mod h1:Ake5+hZFS/DmIGKx/cizhn5W9pGA7pplcR7xCxWiLIo=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=

flagd/pkg/runtime/from_config.go

@@ -27,6 +27,9 @@ type Config struct {
 	MetricExporter        string
 	ManagementPort        uint16
 	OfrepServicePort      uint16
+	OfrepSSEEnabled       bool
+	OfrepSSEInactivityDel int
+	OfrepSSEPublicURL     string
 	OtelCollectorURI      string
 	OtelCertPath          string
 	OtelKeyPath           string
@@ -112,13 +115,16 @@ func FromConfig(logger *logger.Logger, version string, config Config) (*Runtime,
 		recorder)
 
 	// ofrep service
-	ofrepService, err := ofrep.NewOfrepService(jsonEvaluator, config.CORS, ofrep.SvcConfiguration{
+	ofrepService, err := ofrep.NewOfrepService(jsonEvaluator, store, config.CORS, ofrep.SvcConfiguration{
 		Logger:                logger.WithFields(zap.String("component", "OFREPService")),
 		Port:                  config.OfrepServicePort,
 		ServiceName:           svcName,
 		MetricsRecorder:       recorder,
 		MaxRequestBodyBytes:   config.MaxRequestBodyBytes,
 		MaxRequestHeaderBytes: config.MaxRequestHeaderBytes,
+		SSEEnabled:            config.OfrepSSEEnabled,
+		SSEInactivityDelaySec: config.OfrepSSEInactivityDel,
+		SSEPublicURL:          config.OfrepSSEPublicURL,
 	},
 		config.ContextValues,
 		config.HeaderToContextKeyMappings,

flagd/pkg/service/flag-evaluation/ofrep/handler.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/gorilla/mux"
 	"github.com/open-feature/flagd/core/pkg/evaluator"
@@ -16,6 +17,7 @@ import (
 	"github.com/open-feature/flagd/core/pkg/telemetry"
 	"github.com/open-feature/flagd/flagd/pkg/service"
 	evalservice "github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation"
+	"github.com/open-feature/flagd/flagd/pkg/service/flag-evaluation/ofrep/sse"
 	metricsmw "github.com/open-feature/flagd/flagd/pkg/service/middleware/metrics"
 	"github.com/rs/xid"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
@@ -29,13 +31,35 @@ const (
 	bulkEvaluation   = "/ofrep/v1/evaluate/{path:flags\\/|flags}"
 )
 
+// configVersioner resolves the current config ETag / last-modified time for an SSE channel so
+// the bulk handler can serve conditional (ETag/304) responses consistent with the SSE stream.
+// Implemented by the OFREP SSE change tracker; nil when SSE is disabled. ok is false whenever no
+// stream for the channel is live, which is the normal state before a client connects.
+type configVersioner interface {
+	Version(channel string) (etag string, lastModified int64, ok bool)
+}
+
 type handler struct {
 	Logger                     *logger.Logger
 	evaluator                  evaluator.IEvaluator
 	contextValues              map[string]any
 	headerToContextKeyMappings map[string]string
 	metricsRecorder            telemetry.IMetricsRecorder
 	tracer                     trace.Tracer
+
+	versioner             configVersioner
+	sseEnabled            bool
+	sseInactivityDelaySec int
+	ssePublicURL          string
+}
+
+// SSEConfig carries the SSE advertisement settings the bulk handler needs to expose the
+// `eventStreams` block and conditional-evaluation ETags.
+type SSEConfig struct {
+	Enabled            bool
+	Versioner          configVersioner
+	InactivityDelaySec int
+	PublicURL          string
 }
 
 func NewOfrepHandler(
@@ -45,6 +69,7 @@ func NewOfrepHandler(
 	headerToContextKeyMappings map[string]string,
 	metricsRecorder telemetry.IMetricsRecorder,
 	serviceName string,
+	sseCfg SSEConfig,
 ) http.Handler {
 	h := handler{
 		Logger:                     logger,
@@ -53,6 +78,10 @@ func NewOfrepHandler(
 		headerToContextKeyMappings: headerToContextKeyMappings,
 		metricsRecorder:            metricsRecorder,
 		tracer:                     otel.Tracer("flagd.ofrep.v1"),
+		versioner:                  sseCfg.Versioner,
+		sseEnabled:                 sseCfg.Enabled,
+		sseInactivityDelaySec:      sseCfg.InactivityDelaySec,
+		ssePublicURL:               sseCfg.PublicURL,
 	}
 
 	router := mux.NewRouter()
@@ -143,6 +172,14 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) {
 	}
 	ctx := context.WithValue(r.Context(), store.SelectorContextKey{}, selector)
 
+	// Conditional evaluation (ADR-0008): short-circuit with 304 when the client already holds
+	// the current config version.
+	lastModified, notModified := h.applyConditionalETag(w, r, selectorExpression)
+	if notModified {
+		w.WriteHeader(http.StatusNotModified)
+		return
+	}
+
 	evaluations, metadata, err := h.evaluator.ResolveAllValues(ctx, requestID, evaluationContext)
 	if h.metricsRecorder != nil {
 		for _, evaluation := range evaluations {
@@ -155,9 +192,84 @@ func (h *handler) HandleBulkEvaluation(w http.ResponseWriter, r *http.Request) {
 		res := ofrep.BulkEvaluationContextErrorFrom(model.GeneralErrorCode,
 			fmt.Sprintf("Bulk evaluation failed. Tracking ID: %s", requestID))
 		h.writeJSONToResponse(http.StatusInternalServerError, res, w)
-	} else {
-		h.writeJSONToResponse(http.StatusOK, ofrep.BulkEvaluationResponseFrom(evaluations, metadata), w)
+		return
+	}
+
+	h.writeJSONToResponse(http.StatusOK,
+		h.bulkResponse(selectorExpression, evaluations, metadata, lastModified), w)
+}
+
+// applyConditionalETag resolves the current config version for the selector, sets the ETag
+// response header, and reports the lastModified time plus whether the request can be answered
+// with 304 Not Modified. It is a no-op (returns notModified=false) when SSE/versioning is off.
+func (h *handler) applyConditionalETag(w http.ResponseWriter, r *http.Request, channel string) (lastModified int64, notModified bool) {
+	if h.versioner == nil {
+		return 0, false
 	}
+	etag, lastModified, ok := h.versioner.Version(channel)
+	if !ok || etag == "" {
+		return lastModified, false
+	}
+	w.Header().Set("ETag", quoteETag(etag))
+
+	if trigger := r.URL.Query().Get(flagConfigEtagParam); trigger != "" {
+		h.Logger.Debug(fmt.Sprintf("bulk refetch triggered by %s=%s", flagConfigEtagParam, trigger))
+	}
+
+	clientCacheETag := r.Header.Get("If-None-Match")
+	return lastModified, clientCacheETag != "" && normalizeETag(clientCacheETag) == etag
+}
+
+// bulkResponse assembles the OFREP bulk response, adding the ADR-0008 eventStreams block and
+// lastModified metadata when SSE is enabled.
+func (h *handler) bulkResponse(selectorExpression string, evaluations []evaluator.AnyValue, metadata model.Metadata, lastModified int64) ofrep.BulkEvaluationResponse {
+	response := ofrep.BulkEvaluationResponseFrom(evaluations, metadata)
+	if !h.sseEnabled {
+		return response
+	}
+	response.EventStreams = h.eventStreams(selectorExpression)
+	if lastModified > 0 {
+		if response.Metadata == nil {
+			response.Metadata = model.Metadata{}
+		}
+		response.Metadata["flagConfigLastModified"] = lastModified
+	}
+	return response
+}
+
+// eventStreams builds the ADR-0008 eventStreams advertisement pointing OFREP clients back at
+// this flagd's SSE endpoint. The advertised channel is the request's own selector expression,
+// carried as the final path segment, so the stream covers exactly the flags the client just
+// evaluated.
+//
+// It uses the structured `endpoint` form and omits origin unless a public URL is configured, so
+// the client resolves the requestUri against the OFREP base URL it is already talking to.
+func (h *handler) eventStreams(selectorExpression string) []ofrep.EventStream {
+	requestURI := sse.ChannelPath(ssePath, selectorExpression)
+
+	return []ofrep.EventStream{{
+		Type:               "sse",
+		InactivityDelaySec: h.sseInactivityDelaySec,
+		Endpoint: &ofrep.EventStreamEndpoint{
+			Origin:     strings.TrimSuffix(h.ssePublicURL, "/"),
+			RequestUri: requestURI,
+		},
+	}}
+}
+
+// flagConfigEtagParam is the ADR-0008 query parameter carrying the config version that an SSE
+// refetch event advertised. It is trigger metadata only; the conditional response is decided by
+// the If-None-Match header. See applyConditionalETag.
+const flagConfigEtagParam = "flagConfigEtag"
+
+// normalizeETag strips optional surrounding quotes so quoted and unquoted forms compare equal.
+func normalizeETag(etag string) string {
+	return strings.Trim(etag, `"`)
+}
+
+// quoteETag wraps a bare ETag value in the double quotes required by the HTTP ETag header.
+func quoteETag(etag string) string {
+	return `"` + etag + `"`
 }
 
 func (h *handler) writeJSONToResponse(status int, payload interface{}, w http.ResponseWriter) {

flagd/pkg/service/flag-evaluation/ofrep/handler_test.go

@@ -472,7 +472,7 @@ func TestHandlerRecordsSingleEvaluationMetrics(t *testing.T) {
 			eval.EXPECT().
 				ResolveAsAnyValue(gomock.Any(), gomock.Any(), flagKey, gomock.Any()).
 				Return(test.evaluation)
-			handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd")
+			handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd", SSEConfig{})
 
 			request := httptest.NewRequest(http.MethodPost, "/ofrep/v1/evaluate/flags/"+flagKey, nil)
 			response := httptest.NewRecorder()
@@ -490,7 +490,7 @@ func TestHandlerRecordsEachBulkEvaluationMetric(t *testing.T) {
 	eval := mock.NewMockIEvaluator(gomock.NewController(t))
 	eval.EXPECT().ResolveAllValues(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(evaluations, model.Metadata{}, nil)
-	handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd")
+	handler := NewOfrepHandler(logger.NewLogger(nil, false), eval, nil, nil, metrics, "flagd", SSEConfig{})
 
 	request := httptest.NewRequest(http.MethodPost, "/ofrep/v1/evaluate/flags", nil)
 	response := httptest.NewRecorder()