feat: allow setting key prefix for progress manager (#3615)

Author: turip

app/common/progressmanager.go

@@ -28,6 +28,7 @@ func NewProgressManager(logger *slog.Logger, conf config.ProgressManagerConfigur
 
 	pm, err := adapter.New(adapter.Config{
 		Expiration: conf.Expiration,
+		KeyPrefix:  conf.KeyPrefix,
 		Logger:     logger,
 		Redis:      redisClient,
 	})

app/config/progressmanager.go

@@ -13,6 +13,7 @@ import (
 // ProgressManagerConfiguration stores the configuration parameters for the progress manager
 type ProgressManagerConfiguration struct {
 	Enabled    bool
+	KeyPrefix  string
 	Expiration time.Duration
 	Redis      redis.Config
 }
@@ -39,5 +40,6 @@ func (c ProgressManagerConfiguration) Validate() error {
 // ConfigureProgressManager sets the default values for the progress manager configuration
 func ConfigureProgressManager(v *viper.Viper) {
 	v.SetDefault("progressManager.expiration", "5m")
+	v.SetDefault("progressManager.keyPrefix", "")
 	redis.Configure(v, "progressManager.redis")
 }

openmeter/progressmanager/adapter/adapter.go

@@ -14,6 +14,7 @@ type Config struct {
 	Expiration time.Duration
 	Redis      *redis.Client
 	Logger     *slog.Logger
+	KeyPrefix  string
 }
 
 func (c Config) Validate() error {
@@ -41,12 +42,15 @@ func New(config Config) (progressmanager.Adapter, error) {
 		expiration: config.Expiration,
 		redis:      config.Redis,
 		logger:     config.Logger,
+		keyPrefix:  config.KeyPrefix,
 	}, nil
 }
 
 var _ progressmanager.Adapter = (*adapter)(nil)
 
 type adapter struct {
+	// keyPrefix is the prefix for progress data in the Redis store, if needed, the key format will be "<keyPrefix>:progress:<namespace>:<id>" or "progress:<namespace>:<id>" if the prefix is empty
+	keyPrefix string
 	// expiration defines how long progress data is stored in Redis before automatic removal
 	expiration time.Duration
 	// redis is the client for storing and retrieving progress data

openmeter/progressmanager/adapter/progress.go

@@ -13,7 +13,7 @@ import (
 
 // keyPrefix is the prefix for progress data in the Redis store.
 // All progress keys will be stored as "progress:<namespace>:<id>"
-const keyPrefix = "progress:"
+const staticKeyPrefix = "progress:"
 
 // GetProgress retrieves the progress
 func (a *adapter) GetProgress(ctx context.Context, input entity.GetProgressInput) (*entity.Progress, error) {
@@ -23,7 +23,7 @@ func (a *adapter) GetProgress(ctx context.Context, input entity.GetProgressInput
 
 	var progress entity.Progress
 
-	cmd := a.redis.Get(ctx, getKey(input.ProgressID))
+	cmd := a.redis.Get(ctx, a.getKey(input.ProgressID))
 
 	if cmd.Err() != nil {
 		if cmd.Err() == redis.Nil {
@@ -53,7 +53,7 @@ func (a *adapter) UpsertProgress(ctx context.Context, input entity.UpsertProgres
 		return fmt.Errorf("marshal progress: %w", err)
 	}
 
-	cmd := a.redis.Set(ctx, getKey(input.ProgressID), data, a.expiration)
+	cmd := a.redis.Set(ctx, a.getKey(input.ProgressID), data, a.expiration)
 	if cmd.Err() != nil {
 		return fmt.Errorf("set progress: %w", cmd.Err())
 	}
@@ -62,6 +62,10 @@ func (a *adapter) UpsertProgress(ctx context.Context, input entity.UpsertProgres
 }
 
 // getKey returns the key for the KV store
-func getKey(id entity.ProgressID) string {
-	return fmt.Sprintf("%s:%s:%s", keyPrefix, id.Namespace, id.ID)
+func (a *adapter) getKey(id entity.ProgressID) string {
+	if a.keyPrefix == "" {
+		return fmt.Sprintf("%s:%s:%s", staticKeyPrefix, id.Namespace, id.ID)
+	}
+
+	return fmt.Sprintf("%s:%s:%s:%s", a.keyPrefix, staticKeyPrefix, id.Namespace, id.ID)
 }