Skip to content

feat: [FFM-12441]: Optimise SSE events #404

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jul 10, 2025
Merged

feat: [FFM-12441]: Optimise SSE events #404

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ae3bc88
feat: [FFM-12441] Optimise SSE events
jcox250 Jul 9, 2025
90c547e
update tests
jcox250 Jul 9, 2025
3cce332
update tests
jcox250 Jul 9, 2025
4a5df75
update tests
jcox250 Jul 9, 2025
92b7057
run go imports
jcox250 Jul 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 16 additions & 14 deletions cache/refresher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,14 +188,14 @@ func (s Refresher) handleAddEnvironmentEvent(ctx context.Context, environments [
return err
}
// update key inventory for environment.
if err := s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
if err := s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
newAssets, err := s.inventory.BuildAssetListFromConfig(proxyConfig)
if err != nil {
return newAssets, err
}
for k := range newAssets {
for k, version := range newAssets {
if _, ok := assets[k]; !ok {
assets[k] = ""
assets[k] = version
}
}
return assets, nil
Expand Down Expand Up @@ -244,7 +244,7 @@ func (s Refresher) removeAssets(ctx context.Context, env string) error {
return err
}

if err := s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
if err := s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
// remove deleted keys from the assets
for k := range assetsToDelete {
delete(assets, k)
Expand Down Expand Up @@ -299,7 +299,7 @@ func (s Refresher) handleAddAPIKeyEvent(ctx context.Context, env, apiKey string)
}

// add key to the invetnory if does not exits
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
apiKeyEntry := string(domain.NewAuthAPIKey(apiKey))
apiConfigsEntry := string(domain.NewAPIConfigsKey(env))
return s.addItems(assets, apiKeyEntry, apiConfigsEntry)
Expand All @@ -321,7 +321,7 @@ func (s Refresher) handleRemoveAPIKeyEvent(ctx context.Context, env, apiKey stri
return err
}

return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
_, ok := assets[apiKeyEntry]
if ok {
delete(assets, apiKeyEntry)
Expand Down Expand Up @@ -353,7 +353,7 @@ func (s Refresher) handleFetchFeatureEvent(ctx context.Context, env, id string)
return err
}
// patch the inventory
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
featureConfigEntry := string(domain.NewFeatureConfigKey(env, id))
featureConfigsEntry := string(domain.NewFeatureConfigsKey(env))
return s.addItems(assets, featureConfigEntry, featureConfigsEntry)
Expand Down Expand Up @@ -387,7 +387,7 @@ func (s Refresher) handleDeleteFeatureEvent(ctx context.Context, env, identifier
}
}

return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
_, ok := assets[featureConfigEntry]
if ok {
delete(assets, featureConfigEntry)
Expand Down Expand Up @@ -433,7 +433,7 @@ func (s Refresher) handleFetchSegmentEvent(ctx context.Context, env, id string)
return err
}
// patch the inventory
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
segmentConfigEntry := string(domain.NewSegmentKey(env, id))
segmentConfigsEntry := string(domain.NewSegmentsKey(env))
return s.addItems(assets, segmentConfigEntry, segmentConfigsEntry)
Expand Down Expand Up @@ -465,7 +465,7 @@ func (s Refresher) handleDeleteSegmentEvent(ctx context.Context, env, identifier
}
}

return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]string) (map[string]string, error) {
return s.inventory.Patch(ctx, s.config.Key(), func(assets map[string]int64) (map[string]int64, error) {
_, ok := assets[segmentConfig]
if ok {
delete(assets, segmentConfig)
Expand All @@ -492,14 +492,16 @@ func (s Refresher) updateSegmentConfigsEntry(ctx context.Context, env string, id
})
}

func (s Refresher) addItems(assets map[string]string, configKey, configsKey string) (map[string]string, error) {
_, ok := assets[configKey]
func (s Refresher) addItems(assets map[string]int64, configKey, configsKey string) (map[string]int64, error) {
version, ok := assets[configKey]
if !ok {
assets[configKey] = ""
assets[configKey] = version
}
_, ok = assets[configsKey]
if !ok {
assets[configsKey] = ""
// Configs Keys aren't versioned the way keys for individual items are e.g. Flags, Segments are individually versioned,
// but the Flag/Segment config for an entire environment isn't versioned, so we can just set this to 0.
assets[configKey] = 0
}
return assets, nil
}
Expand Down
42 changes: 21 additions & 21 deletions cache/refresher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,20 +260,20 @@ func TestRefresher_HandleMessage(t *testing.T) {
setProxyConfigFn: func(proxyConfig []domain.ProxyConfig) {},
}
inventoryRepo := mockInventoryRepo{
addFn: func(ctx context.Context, key string, assets map[string]string) error {
addFn: func(ctx context.Context, key string, assets map[string]int64) error {
return nil
},
removeFn: func(ctx context.Context, key string) error {
return nil
},
getFn: func(ctx context.Context, key string) (map[string]string, error) {
return map[string]string{}, nil
getFn: func(ctx context.Context, key string) (map[string]int64, error) {
return map[string]int64{}, nil
},
patchFn: func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
patchFn: func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return nil
},
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]string, error) {
return map[string]string{}, nil
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]int64, error) {
return map[string]int64{}, nil
},
cleanupFn: func(ctx context.Context, key string, config []domain.ProxyConfig) ([]domain.SSEMessage, error) {
return []domain.SSEMessage{}, nil
Expand Down Expand Up @@ -370,12 +370,12 @@ func TestRefresher_handleAddEnvironmentEvent(t *testing.T) {
}

inventoryRepo := mockInventoryRepo{
patchFn: func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
patchFn: func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return nil
},

buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]string, error) {
return map[string]string{}, nil
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]int64, error) {
return map[string]int64{}, nil
},
}
for desc, tc := range testCases {
Expand Down Expand Up @@ -438,11 +438,11 @@ func TestRefresher_handleRemoveEnvironmentEvent(t *testing.T) {
},
}
inventoryRepo := mockInventoryRepo{
patchFn: func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
patchFn: func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return nil
},
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]string, error) {
return map[string]string{}, nil
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]int64, error) {
return map[string]int64{}, nil
},
}

Expand Down Expand Up @@ -531,7 +531,7 @@ func TestRefresher_handleRemoveEnvironmentEvent(t *testing.T) {
},
},
inventoryRepo: mockInventoryRepo{
patchFn: func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
patchFn: func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return domain.ErrCacheInternal
},
getKeysForEnvironmentFn: func(ctx context.Context, env string) (map[string]string, error) {
Expand Down Expand Up @@ -581,33 +581,33 @@ type mockConfig struct {
}

type mockInventoryRepo struct {
addFn func(ctx context.Context, key string, assets map[string]string) error
addFn func(ctx context.Context, key string, assets map[string]int64) error
removeFn func(ctx context.Context, key string) error
getFn func(ctx context.Context, key string) (map[string]string, error)
patchFn func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error
buildAssetListFromConfigFn func(config []domain.ProxyConfig) (map[string]string, error)
getFn func(ctx context.Context, key string) (map[string]int64, error)
patchFn func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error
buildAssetListFromConfigFn func(config []domain.ProxyConfig) (map[string]int64, error)
cleanupFn func(ctx context.Context, key string, config []domain.ProxyConfig) ([]domain.SSEMessage, error)
keyExistsFn func(ctx context.Context, key string) bool
getKeysForEnvironmentFn func(ctx context.Context, env string) (map[string]string, error)
}

func (m mockInventoryRepo) Add(ctx context.Context, key string, assets map[string]string) error {
func (m mockInventoryRepo) Add(ctx context.Context, key string, assets map[string]int64) error {
return m.addFn(ctx, key, assets)
}

func (m mockInventoryRepo) Remove(ctx context.Context, key string) error {
return m.removeFn(ctx, key)
}

func (m mockInventoryRepo) Get(ctx context.Context, key string) (map[string]string, error) {
func (m mockInventoryRepo) Get(ctx context.Context, key string) (map[string]int64, error) {
return m.getFn(ctx, key)
}

func (m mockInventoryRepo) Patch(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
func (m mockInventoryRepo) Patch(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return m.patchFn(ctx, key, patch)
}

func (m mockInventoryRepo) BuildAssetListFromConfig(config []domain.ProxyConfig) (map[string]string, error) {
func (m mockInventoryRepo) BuildAssetListFromConfig(config []domain.ProxyConfig) (map[string]int64, error) {
return m.buildAssetListFromConfigFn(config)
}

Expand Down
28 changes: 14 additions & 14 deletions config/remote/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,33 +26,33 @@ import (
)

type mockInventoryRepo struct {
addFn func(ctx context.Context, key string, assets map[string]string) error
addFn func(ctx context.Context, key string, assets map[string]int64) error
removeFn func(ctx context.Context, key string) error
getFn func(ctx context.Context, key string) (map[string]string, error)
patchFn func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error
buildAssetListFromConfigFn func(config []domain.ProxyConfig) (map[string]string, error)
getFn func(ctx context.Context, key string) (map[string]int64, error)
patchFn func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error
buildAssetListFromConfigFn func(config []domain.ProxyConfig) (map[string]int64, error)
cleanupFn func(ctx context.Context, key string, config []domain.ProxyConfig) ([]domain.SSEMessage, error)
keyExistsFn func(ctx context.Context, key string) bool
getKeysForEnvironmentFn func(ctx context.Context, env string) (map[string]string, error)
}

func (m mockInventoryRepo) Add(ctx context.Context, key string, assets map[string]string) error {
func (m mockInventoryRepo) Add(ctx context.Context, key string, assets map[string]int64) error {
return m.addFn(ctx, key, assets)
}

func (m mockInventoryRepo) Remove(ctx context.Context, key string) error {
return m.removeFn(ctx, key)
}

func (m mockInventoryRepo) Get(ctx context.Context, key string) (map[string]string, error) {
func (m mockInventoryRepo) Get(ctx context.Context, key string) (map[string]int64, error) {
return m.getFn(ctx, key)
}

func (m mockInventoryRepo) Patch(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
func (m mockInventoryRepo) Patch(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return m.patchFn(ctx, key, patch)
}

func (m mockInventoryRepo) BuildAssetListFromConfig(config []domain.ProxyConfig) (map[string]string, error) {
func (m mockInventoryRepo) BuildAssetListFromConfig(config []domain.ProxyConfig) (map[string]int64, error) {
return m.buildAssetListFromConfigFn(config)
}

Expand Down Expand Up @@ -510,20 +510,20 @@ func TestConfig_Populate(t *testing.T) {
}

inventoryRepo := mockInventoryRepo{
addFn: func(ctx context.Context, key string, assets map[string]string) error {
addFn: func(ctx context.Context, key string, assets map[string]int64) error {
return nil
},
removeFn: func(ctx context.Context, key string) error {
return nil
},
getFn: func(ctx context.Context, key string) (map[string]string, error) {
return map[string]string{}, nil
getFn: func(ctx context.Context, key string) (map[string]int64, error) {
return map[string]int64{}, nil
},
patchFn: func(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error {
patchFn: func(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error {
return nil
},
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]string, error) {
return map[string]string{}, nil
buildAssetListFromConfigFn: func(config []domain.ProxyConfig) (map[string]int64, error) {
return map[string]int64{}, nil
},
cleanupFn: func(ctx context.Context, key string, config []domain.ProxyConfig) ([]domain.SSEMessage, error) {
return []domain.SSEMessage{}, nil
Expand Down
6 changes: 3 additions & 3 deletions domain/inventory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,9 @@ import (
type KeyInventory string

type Assets struct {
Deleted map[string]string
Created map[string]string
Patched map[string]string
Deleted map[string]int64
Created map[string]int64
Patched map[string]int64
}

// NewKeyInventory creates a key inventory entry for the proxy key. This key contains all the entries associated with the proxy key.
Expand Down
8 changes: 4 additions & 4 deletions domain/repository.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,11 @@ import (

// KeyRepo the interface for keyRepository.
type InventoryRepo interface {
Add(ctx context.Context, key string, assets map[string]string) error
Add(ctx context.Context, key string, assets map[string]int64) error
Remove(ctx context.Context, key string) error
Get(ctx context.Context, key string) (map[string]string, error)
Patch(ctx context.Context, key string, patch func(assets map[string]string) (map[string]string, error)) error
BuildAssetListFromConfig(config []ProxyConfig) (map[string]string, error)
Get(ctx context.Context, key string) (map[string]int64, error)
Patch(ctx context.Context, key string, patch func(assets map[string]int64) (map[string]int64, error)) error
BuildAssetListFromConfig(config []ProxyConfig) (map[string]int64, error)
Cleanup(ctx context.Context, key string, config []ProxyConfig) ([]SSEMessage, error)
KeyExists(ctx context.Context, key string) bool
GetKeysForEnvironment(ctx context.Context, env string) (map[string]string, error)
Expand Down
Loading