feat: Sync publishes streaming update

internal/dev_server/api/server.go

@@ -56,7 +56,7 @@ func (s Server) GetDevProjectsProjectKey(ctx context.Context, request GetDevProj
 		LastSyncedFromSource: project.LastSyncTime.Unix(),
 		Context:              project.Context,
 		SourceEnvironmentKey: project.SourceEnvironmentKey,
-		FlagsState:           &project.FlagState,
+		FlagsState:           &project.AllFlagsState,
 	}
 
 	if request.Params.Expand != nil {
@@ -98,7 +98,7 @@ func (s Server) PostDevProjectsProjectKey(ctx context.Context, request PostDevPr
 		LastSyncedFromSource: project.LastSyncTime.Unix(),
 		Context:              project.Context,
 		SourceEnvironmentKey: project.SourceEnvironmentKey,
-		FlagsState:           &project.FlagState,
+		FlagsState:           &project.AllFlagsState,
 	}
 
 	if request.Params.Expand != nil {
@@ -143,7 +143,7 @@ func (s Server) PatchDevProjectsProjectKey(ctx context.Context, request PatchDev
 		LastSyncedFromSource: project.LastSyncTime.Unix(),
 		Context:              project.Context,
 		SourceEnvironmentKey: project.SourceEnvironmentKey,
-		FlagsState:           &project.FlagState,
+		FlagsState:           &project.AllFlagsState,
 	}
 
 	if request.Params.Expand != nil {
@@ -188,7 +188,7 @@ func (s Server) PatchDevProjectsProjectKeySync(ctx context.Context, request Patc
 		LastSyncedFromSource: project.LastSyncTime.Unix(),
 		Context:              project.Context,
 		SourceEnvironmentKey: project.SourceEnvironmentKey,
-		FlagsState:           &project.FlagState,
+		FlagsState:           &project.AllFlagsState,
 	}
 
 	if request.Params.Expand != nil {

internal/dev_server/db/sqlite.go

@@ -57,15 +57,15 @@ func (s Sqlite) GetDevProject(ctx context.Context, key string) (*model.Project,
 	}
 
 	// Parse the flag state JSON string
-	if err := json.Unmarshal([]byte(flagStateData), &project.FlagState); err != nil {
+	if err := json.Unmarshal([]byte(flagStateData), &project.AllFlagsState); err != nil {
 		return nil, errors.Wrap(err, "unable to unmarshal flag state data")
 	}
 
 	return &project, nil
 }
 
 func (s Sqlite) UpdateProject(ctx context.Context, project model.Project) (bool, error) {
-	flagsStateJson, err := json.Marshal(project.FlagState)
+	flagsStateJson, err := json.Marshal(project.AllFlagsState)
 	if err != nil {
 		return false, errors.Wrap(err, "unable to marshal flags state when updating project")
 	}
@@ -107,7 +107,7 @@ func (s Sqlite) DeleteDevProject(ctx context.Context, key string) (bool, error)
 }
 
 func (s Sqlite) InsertProject(ctx context.Context, project model.Project) error {
-	flagsStateJson, err := json.Marshal(project.FlagState)
+	flagsStateJson, err := json.Marshal(project.AllFlagsState)
 	if err != nil {
 		return errors.Wrap(err, "unable to marshal flags state when writing project")
 	}

internal/dev_server/model/events.go

@@ -0,0 +1,14 @@
+package model
+
+// Event for individual flag overrides
+type UpsertOverrideEvent struct {
+	FlagKey    string
+	ProjectKey string
+	FlagState  FlagState
+}
+
+// Event for full project sync
+type SyncEvent struct {
+	ProjectKey    string
+	AllFlagsState FlagsState
+}

internal/dev_server/model/override.go

@@ -14,12 +14,6 @@ type Override struct {
 	Version    int
 }
 
-type UpsertOverrideEvent struct {
-	FlagKey    string
-	ProjectKey string
-	FlagState  FlagState
-}
-
 func UpsertOverride(ctx context.Context, projectKey, flagKey string, value ldvalue.Value) (Override, error) {
 	// TODO: validate if the flag type matches
 
@@ -31,7 +25,7 @@ func UpsertOverride(ctx context.Context, projectKey, flagKey string, value ldval
 	}
 
 	var flagExists bool
-	for flag := range project.FlagState {
+	for flag := range project.AllFlagsState {
 		if flagKey == flag {
 			flagExists = true
 			break
@@ -54,9 +48,8 @@ func UpsertOverride(ctx context.Context, projectKey, flagKey string, value ldval
 		return Override{}, err
 	}
 
-	observers := GetObserversFromContext(ctx)
-	flagState := override.Apply(project.FlagState[flagKey])
-	observers.Notify(UpsertOverrideEvent{
+	flagState := override.Apply(project.AllFlagsState[flagKey])
+	GetObserversFromContext(ctx).Notify(UpsertOverrideEvent{
 		FlagKey:    flagKey,
 		ProjectKey: projectKey,
 		FlagState:  flagState,

internal/dev_server/model/override_test.go

@@ -28,8 +28,8 @@ func TestUpsertOverride(t *testing.T) {
 	}
 
 	project := &model.Project{
-		Key:       projKey,
-		FlagState: model.FlagsState{flagKey: model.FlagState{Value: ldvalue.Bool(false), Version: 1}},
+		Key:           projKey,
+		AllFlagsState: model.FlagsState{flagKey: model.FlagState{Value: ldvalue.Bool(false), Version: 1}},
 	}
 
 	ctx = model.ContextWithStore(ctx, store)
@@ -50,8 +50,8 @@ func TestUpsertOverride(t *testing.T) {
 
 	t.Run("Returns error if flag does not exist in project", func(t *testing.T) {
 		badProj := model.Project{
-			Key:       projKey,
-			FlagState: model.FlagsState{},
+			Key:           projKey,
+			AllFlagsState: model.FlagsState{},
 		}
 		store.EXPECT().GetDevProject(gomock.Any(), projKey).Return(&badProj, nil)
 

internal/dev_server/model/project.go

@@ -14,7 +14,7 @@ type Project struct {
 	SourceEnvironmentKey string
 	Context              ldcontext.Context
 	LastSyncTime         time.Time
-	FlagState            FlagsState
+	AllFlagsState        FlagsState
 }
 
 // CreateProject creates a project and adds it to the database.
@@ -35,7 +35,7 @@ func CreateProject(ctx context.Context, projectKey, sourceEnvironmentKey string,
 		return Project{}, err
 	}
 
-	project.FlagState = flagsState
+	project.AllFlagsState = flagsState
 	project.LastSyncTime = time.Now()
 
 	store := StoreFromContext(ctx)
@@ -65,7 +65,7 @@ func UpdateProject(ctx context.Context, projectKey string, context *ldcontext.Co
 		if err != nil {
 			return Project{}, err
 		}
-		project.FlagState = flagsState
+		project.AllFlagsState = flagsState
 		project.LastSyncTime = time.Now()
 	}
 
@@ -76,6 +76,7 @@ func UpdateProject(ctx context.Context, projectKey string, context *ldcontext.Co
 	if !updated {
 		return Project{}, errors.New("Project not updated")
 	}
+
 	return *project, nil
 }
 
@@ -90,15 +91,26 @@ func SyncProject(ctx context.Context, projectKey string) (Project, error) {
 		return Project{}, err
 	}
 
-	project.FlagState = flagsState
+	project.AllFlagsState = flagsState
 	project.LastSyncTime = time.Now()
+
 	updated, err := store.UpdateProject(ctx, *project)
 	if err != nil {
 		return Project{}, err
 	}
 	if !updated {
 		return Project{}, errors.New("Project not updated")
 	}
+
+	allFlagsWithOverrides, err := project.GetFlagStateWithOverridesForProject(ctx)
+	if err != nil {
+		return Project{}, errors.Wrapf(err, "unable to get overrides for project, %s", projectKey)
+	}
+
+	GetObserversFromContext(ctx).Notify(SyncEvent{
+		ProjectKey:    project.Key,
+		AllFlagsState: allFlagsWithOverrides,
+	})
 	return *project, nil
 }
 
@@ -108,8 +120,8 @@ func (p Project) GetFlagStateWithOverridesForProject(ctx context.Context) (Flags
 	if err != nil {
 		return FlagsState{}, errors.Wrapf(err, "unable to fetch overrides for project %s", p.Key)
 	}
-	withOverrides := make(FlagsState, len(p.FlagState))
-	for flagKey, flagState := range p.FlagState {
+	withOverrides := make(FlagsState, len(p.AllFlagsState))
+	for flagKey, flagState := range p.AllFlagsState {
 		if override, ok := overrides.GetFlag(flagKey); ok {
 			flagState = override.Apply(flagState)
 		}

internal/dev_server/model/project_test.go

@@ -58,13 +58,13 @@ func TestCreateProject(t *testing.T) {
 			Key:                  projKey,
 			SourceEnvironmentKey: sourceEnvKey,
 			Context:              ldcontext.NewBuilder("user").Key("dev-environment").Build(),
-			FlagState:            model.FromAllFlags(allFlags),
+			AllFlagsState:        model.FromAllFlags(allFlags),
 		}
 
 		assert.Equal(t, expectedProj.Key, p.Key)
 		assert.Equal(t, expectedProj.SourceEnvironmentKey, p.SourceEnvironmentKey)
 		assert.Equal(t, expectedProj.Context, p.Context)
-		assert.Equal(t, expectedProj.FlagState, p.FlagState)
+		assert.Equal(t, expectedProj.AllFlagsState, p.AllFlagsState)
 	})
 }
 
@@ -134,6 +134,11 @@ func TestSyncProject(t *testing.T) {
 	ctx := model.ContextWithStore(context.Background(), store)
 	ctx, api, sdk := adapters_mocks.WithMockApiAndSdk(ctx, mockController)
 
+	observer := mocks.NewMockObserver(mockController)
+	observers := model.NewObservers()
+	observers.RegisterObserver(observer)
+	ctx = model.SetObserversOnContext(ctx, observers)
+
 	proj := model.Project{
 		Key:                  "projKey",
 		SourceEnvironmentKey: "srcEnvKey",
@@ -183,6 +188,12 @@ func TestSyncProject(t *testing.T) {
 		api.EXPECT().GetSdkKey(gomock.Any(), proj.Key, proj.SourceEnvironmentKey).Return("sdkKey", nil)
 		sdk.EXPECT().GetAllFlagsState(gomock.Any(), gomock.Any(), "sdkKey").Return(flagstate.AllFlags{}, nil)
 		store.EXPECT().UpdateProject(gomock.Any(), gomock.Any()).Return(true, nil)
+		observer.
+			EXPECT().
+			Handle(model.SyncEvent{
+				ProjectKey:    proj.Key,
+				AllFlagsState: model.FlagsState{},
+			})
 
 		project, err := model.SyncProject(ctx, proj.Key)
 		assert.Nil(t, err)
@@ -196,8 +207,8 @@ func TestGetFlagStateWithOverridesForProject(t *testing.T) {
 	ctx := model.ContextWithStore(context.Background(), store)
 	flagKey := "flg"
 	proj := model.Project{
-		Key:       "projKey",
-		FlagState: model.FlagsState{flagKey: model.FlagState{Value: ldvalue.Bool(false), Version: 1}},
+		Key:           "projKey",
+		AllFlagsState: model.FlagsState{flagKey: model.FlagState{Value: ldvalue.Bool(false), Version: 1}},
 	}
 
 	t.Run("Returns error if store fetch fails", func(t *testing.T) {

internal/dev_server/sdk/go_sdk_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/launchdarkly/go-sdk-common/v3/ldcontext"
 	"github.com/launchdarkly/go-sdk-common/v3/ldvalue"
 	ldclient "github.com/launchdarkly/go-server-sdk/v7"
+	"github.com/launchdarkly/go-server-sdk/v7/interfaces"
 	"github.com/launchdarkly/go-server-sdk/v7/interfaces/flagstate"
 	"github.com/launchdarkly/ldcli/internal/dev_server/adapters/mocks"
 	"github.com/launchdarkly/ldcli/internal/dev_server/db"
@@ -62,7 +63,7 @@ func TestSDKRoutesViaGoSDK(t *testing.T) {
 		AddFlag("jsonFlag", flagstate.FlagState{Value: ldvalue.CopyArbitraryValue(map[string]any{"cat": "hat"})}).
 		Build()
 
-	sdk.EXPECT().GetAllFlagsState(gomock.Any(), gomock.Any(), testSdkKey).Return(allFlags, nil)
+	sdk.EXPECT().GetAllFlagsState(gomock.Any(), gomock.Any(), testSdkKey).Return(allFlags, nil).Times(2)
 	_, err = model.CreateProject(ctx, projectKey, environmentKey, nil)
 	require.NoError(t, err)
 
@@ -123,4 +124,43 @@ func TestSDKRoutesViaGoSDK(t *testing.T) {
 			assert.Equal(t, value.AsArbitraryValue(), flagUpdate.NewValue.AsArbitraryValue())
 		})
 	}
+
+	newUpdates := map[string]ldvalue.Value{
+		"boolFlag":   ldvalue.Bool(true),
+		"stringFlag": ldvalue.String("fool"),
+		"intFlag":    ldvalue.Int(567),
+		"doubleFlag": ldvalue.Float64(99.99),
+		"jsonFlag":   ldvalue.CopyArbitraryValue(map[string]any{"tortoise": "hare"}),
+	}
+
+	// This test is testing the "put" payload in a roundabout way by verifying each of the flags are in there.
+	t.Run("Sync sends full flag payload for project", func(t *testing.T) {
+		trackers := make(map[string]*<-chan interfaces.FlagValueChangeEvent, len(updates))
+
+		for flagKey, newValue := range newUpdates {
+			flagUpdateChan := ld.GetFlagTracker().AddFlagValueChangeListener(flagKey, ldContext, ldvalue.String("uh-oh"))
+			defer ld.GetFlagTracker().RemoveFlagValueChangeListener(flagUpdateChan)
+			trackers[flagKey] = &flagUpdateChan
+
+			_, err := store.UpsertOverride(ctx, model.Override{
+				ProjectKey: projectKey,
+				FlagKey:    flagKey,
+				Value:      newValue,
+				Active:     true,
+				Version:    1,
+			})
+			require.NoError(t, err)
+		}
+
+		_, err := model.SyncProject(ctx, projectKey)
+		require.NoError(t, err)
+
+		for flagKey, value := range newUpdates {
+			updateTracker, ok := trackers[flagKey]
+			require.True(t, ok)
+
+			update := <-*updateTracker
+			assert.Equal(t, value.AsArbitraryValue(), update.NewValue.AsArbitraryValue())
+		}
+	})
 }

internal/dev_server/sdk/stream_client_flags.go

@@ -22,7 +22,11 @@ func StreamClientFlags(w http.ResponseWriter, r *http.Request) {
 		WriteError(ctx, w, errors.Wrap(err, "failed to marshal flag state"))
 		return
 	}
-	updateChan, doneChan := OpenStream(w, r.Context().Done(), Message{"put", jsonBody})
+	updateChan, doneChan := OpenStream(
+		w,
+		r.Context().Done(),
+		Message{Event: TYPE_PUT, Data: jsonBody},
+	)
 	defer close(updateChan)
 	projectKey := GetProjectKeyFromContext(ctx)
 	observer := clientFlagsObserver{updateChan, projectKey}
@@ -50,23 +54,34 @@ func (c clientFlagsObserver) Handle(event interface{}) {
 	log.Printf("clientFlagsObserver: handling flag state event: %v", event)
 	switch event := event.(type) {
 	case model.UpsertOverrideEvent:
-		data, err := json.Marshal(clientSidePatchData{
+		err := SendMessage(c.updateChan, TYPE_PATCH, clientFlag{
 			Key:     event.FlagKey,
 			Version: event.FlagState.Version,
 			Value:   event.FlagState.Value,
 		})
 		if err != nil {
 			panic(errors.Wrap(err, "failed to marshal flag state in observer"))
 		}
-		c.updateChan <- Message{
-			Event: "patch",
-			Data:  data,
+	case model.SyncEvent:
+		clientFlags := clientFlags{}
+		for flagKey, flagState := range event.AllFlagsState {
+			clientFlags[flagKey] = clientFlag{
+				Version: flagState.Version,
+				Value:   flagState.Value,
+			}
+		}
+
+		err := SendMessage(c.updateChan, TYPE_PUT, clientFlags)
+		if err != nil {
+			panic(errors.Wrap(err, "failed to marshal flag state in observer"))
 		}
 	}
 }
 
-type clientSidePatchData struct {
-	Key     string        `json:"key"`
+type clientFlag struct {
+	Key     string        `json:"key,omitempty"`
 	Version int           `json:"version"`
 	Value   ldvalue.Value `json:"value"`
 }
+
+type clientFlags map[string]clientFlag