swap glide SDK for valkey-go SDK

Author: Derk Schooltink

backend/redis/redis.go

@@ -19,7 +19,7 @@ import (
 var _ backend.Backend = (*redisBackend)(nil)
 
 //go:embed scripts
-var luaScripts embed.FS
+var Luas embed.FS
 
 var (
 	createWorkflowInstanceCmd *redis.Script
@@ -85,7 +85,7 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 
 func loadScripts(ctx context.Context, rdb redis.UniversalClient, cmdMapping map[string]**redis.Script) error {
 	for scriptFile, cmd := range cmdMapping {
-		scriptContent, err := fs.ReadFile(luaScripts, "scripts/"+scriptFile)
+		scriptContent, err := fs.ReadFile(Luas, "scripts/"+scriptFile)
 		if err != nil {
 			return fmt.Errorf("reading Lua script %s: %w", scriptFile, err)
 		}

backend/valkey/activity.go

@@ -7,7 +7,6 @@ import (
 	"github.com/cschleiden/go-workflows/backend"
 	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/workflow"
-	"github.com/valkey-io/valkey-glide/go/v2/options"
 )
 
 func (vb *valkeyBackend) PrepareActivityQueues(ctx context.Context, queues []workflow.Queue) error {
@@ -56,26 +55,23 @@ func (vb *valkeyBackend) CompleteActivityTask(ctx context.Context, task *backend
 	activityQueueKeys := vb.activityQueue.Keys(task.Queue)
 	workflowQueueKeys := vb.workflowQueue.Keys(workflow.Queue(instance.Queue))
 
-	_, err = vb.client.InvokeScriptWithOptions(ctx, completeActivityTaskScript, options.ScriptOptions{
-		Keys: []string{
-			activityQueueKeys.SetKey,
-			activityQueueKeys.StreamKey,
-			vb.keys.pendingEventsKey(task.WorkflowInstance),
-			vb.keys.payloadKey(task.WorkflowInstance),
-			vb.workflowQueue.queueSetKey,
-			workflowQueueKeys.SetKey,
-			workflowQueueKeys.StreamKey,
-		},
-		Args: []string{
-			task.ID,
-			vb.activityQueue.groupName,
-			result.ID,
-			eventData,
-			payload,
-			vb.workflowQueue.groupName,
-			instanceSegment(task.WorkflowInstance),
-		},
-	})
+	err = completeActivityTaskScript.Exec(ctx, vb.client, []string{
+		activityQueueKeys.SetKey,
+		activityQueueKeys.StreamKey,
+		vb.keys.pendingEventsKey(task.WorkflowInstance),
+		vb.keys.payloadKey(task.WorkflowInstance),
+		vb.workflowQueue.queueSetKey,
+		workflowQueueKeys.SetKey,
+		workflowQueueKeys.StreamKey,
+	}, []string{
+		task.ID,
+		vb.activityQueue.groupName,
+		result.ID,
+		eventData,
+		payload,
+		vb.workflowQueue.groupName,
+		instanceSegment(task.WorkflowInstance),
+	}).Error()
 
 	if err != nil {
 		return fmt.Errorf("completing activity task: %w", err)

backend/valkey/delete.go

@@ -5,25 +5,23 @@ import (
 	"fmt"
 
 	"github.com/cschleiden/go-workflows/core"
-	"github.com/valkey-io/valkey-glide/go/v2/options"
 )
 
 // deleteInstance deletes an instance from Valkey. It does not attempt to remove any future events or pending
 // workflow tasks. It's assumed that the instance is in the finished state.
 //
 // Note: might want to revisit this in the future if we want to support removing hung instances.
 func (vb *valkeyBackend) deleteInstance(ctx context.Context, instance *core.WorkflowInstance) error {
-	_, err := vb.client.InvokeScriptWithOptions(ctx, deleteInstanceScript, options.ScriptOptions{
-		Keys: []string{
-			vb.keys.instanceKey(instance),
-			vb.keys.pendingEventsKey(instance),
-			vb.keys.historyKey(instance),
-			vb.keys.payloadKey(instance),
-			vb.keys.activeInstanceExecutionKey(instance.InstanceID),
-			vb.keys.instancesByCreation(),
-		},
-		Args: []string{instanceSegment(instance)},
-	})
+	err := deleteInstanceScript.Exec(ctx, vb.client, []string{
+		vb.keys.instanceKey(instance),
+		vb.keys.pendingEventsKey(instance),
+		vb.keys.historyKey(instance),
+		vb.keys.payloadKey(instance),
+		vb.keys.activeInstanceExecutionKey(instance.InstanceID),
+		vb.keys.instancesByCreation(),
+	}, []string{
+		instanceSegment(instance),
+	}).Error()
 
 	if err != nil {
 		return fmt.Errorf("failed to delete instance: %w", err)

backend/valkey/diagnostics.go

@@ -8,46 +8,32 @@ import (
 	"github.com/cschleiden/go-workflows/core"
 	"github.com/cschleiden/go-workflows/diag"
 	"github.com/cschleiden/go-workflows/internal/log"
-	"github.com/valkey-io/valkey-glide/go/v2/constants"
-	"github.com/valkey-io/valkey-glide/go/v2/options"
 )
 
 var _ diag.Backend = (*valkeyBackend)(nil)
 
 func (vb *valkeyBackend) GetWorkflowInstances(ctx context.Context, afterInstanceID, afterExecutionID string, count int) ([]*diag.WorkflowInstanceRef, error) {
-	start := options.NewInclusiveScoreBoundary(0)
-	end := options.NewInfiniteScoreBoundary(constants.PositiveInfinity)
-
-	zrangeInput := &options.RangeByScore{
-		Start:   start,
-		End:     end,
-		Reverse: true,
-		Limit: &options.Limit{
-			Offset: 0,
-			Count:  int64(count),
-		},
-	}
+	zrangeCmd := vb.client.B().Zrange().Key(vb.keys.instancesByCreation()).Min("0").Max("-1").Rev().Limit(0, int64(count))
 
 	if afterInstanceID != "" {
 		afterSegmentID := instanceSegment(core.NewWorkflowInstance(afterInstanceID, afterExecutionID))
-		scores, err := vb.client.ZMScore(ctx, vb.keys.instancesByCreation(), []string{afterSegmentID})
+		scores, err := vb.client.Do(ctx, vb.client.B().Zscore().Key(vb.keys.instancesByCreation()).Member(afterSegmentID).Build()).AsFloat64()
 		if err != nil {
 			return nil, fmt.Errorf("getting instance score for %v: %w", afterSegmentID, err)
 		}
 
-		if len(scores) == 0 {
+		if scores == 0 {
 			vb.Options().Logger.Error("could not find instance %v",
 				log.NamespaceKey+".valkey.afterInstanceID", afterInstanceID,
 				log.NamespaceKey+".valkey.afterExecutionID", afterExecutionID,
 			)
 			return nil, nil
 		}
 
-		end := options.NewScoreBoundary(scores[0].Value(), false)
-		zrangeInput.End = end
+		zrangeCmd = vb.client.B().Zrange().Key(vb.keys.instancesByCreation()).Min("-inf").Max(fmt.Sprintf("(%f", scores)).Rev().Limit(0, int64(count))
 	}
 
-	instanceSegments, err := vb.client.ZRange(ctx, vb.keys.instancesByCreation(), zrangeInput)
+	instanceSegments, err := vb.client.Do(ctx, zrangeCmd.Build()).AsStrSlice()
 	if err != nil {
 		return nil, fmt.Errorf("getting instances: %w", err)
 	}
@@ -61,19 +47,20 @@ func (vb *valkeyBackend) GetWorkflowInstances(ctx context.Context, afterInstance
 		instanceKeys = append(instanceKeys, vb.keys.instanceKeyFromSegment(r))
 	}
 
-	instances, err := vb.client.MGet(ctx, instanceKeys)
+	cmd := vb.client.B().Mget().Key(instanceKeys...)
+	instances, err := vb.client.Do(ctx, cmd.Build()).AsStrSlice()
 	if err != nil {
 		return nil, fmt.Errorf("getting instances: %w", err)
 	}
 
 	instanceRefs := make([]*diag.WorkflowInstanceRef, 0, len(instances))
 	for _, instance := range instances {
-		if instance.IsNil() {
+		if instance == "" {
 			continue
 		}
 
 		var state instanceState
-		if err := json.Unmarshal([]byte(instance.Value()), &state); err != nil {
+		if err := json.Unmarshal([]byte(instance), &state); err != nil {
 			return nil, fmt.Errorf("unmarshaling instance state: %w", err)
 		}
 

backend/valkey/events_future.go

@@ -5,19 +5,12 @@ import (
 	"fmt"
 	"strconv"
 	"time"
-
-	"github.com/valkey-io/valkey-glide/go/v2/options"
 )
 
 func scheduleFutureEvents(ctx context.Context, vb *valkeyBackend) error {
 	now := time.Now().UnixMilli()
 	nowStr := strconv.FormatInt(now, 10)
-	_, err := vb.client.InvokeScriptWithOptions(ctx, futureEventsScript, options.ScriptOptions{
-		Keys: []string{
-			vb.keys.futureEventsKey(),
-		},
-		Args: []string{nowStr, vb.keys.prefix},
-	})
+	err := futureEventsScript.Exec(ctx, vb.client, []string{vb.keys.futureEventsKey()}, []string{nowStr, vb.keys.prefix}).Error()
 
 	if err != nil {
 		return fmt.Errorf("checking future events: %w", err)

backend/valkey/expire.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/cschleiden/go-workflows/core"
-	"github.com/valkey-io/valkey-glide/go/v2/options"
 )
 
 func (vb *valkeyBackend) setWorkflowInstanceExpiration(ctx context.Context, instance *core.WorkflowInstance, expiration time.Duration) error {
@@ -17,22 +16,19 @@ func (vb *valkeyBackend) setWorkflowInstanceExpiration(ctx context.Context, inst
 	exp := time.Now().Add(expiration).UnixMilli()
 	expStr := strconv.FormatInt(exp, 10)
 
-	_, err := vb.client.InvokeScriptWithOptions(ctx, expireWorkflowInstanceScript, options.ScriptOptions{
-		Keys: []string{
-			vb.keys.instancesByCreation(),
-			vb.keys.instancesExpiring(),
-			vb.keys.instanceKey(instance),
-			vb.keys.pendingEventsKey(instance),
-			vb.keys.historyKey(instance),
-			vb.keys.payloadKey(instance),
-		},
-		Args: []string{
-			nowStr,
-			fmt.Sprintf("%.0f", expiration.Seconds()),
-			expStr,
-			instanceSegment(instance),
-		},
-	})
+	err := expireWorkflowInstanceScript.Exec(ctx, vb.client, []string{
+		vb.keys.instancesByCreation(),
+		vb.keys.instancesExpiring(),
+		vb.keys.instanceKey(instance),
+		vb.keys.pendingEventsKey(instance),
+		vb.keys.historyKey(instance),
+		vb.keys.payloadKey(instance),
+	}, []string{
+		nowStr,
+		fmt.Sprintf("%.0f", expiration.Seconds()),
+		expStr,
+		instanceSegment(instance),
+	}).Error()
 
 	return err
 }