Implement diagnostics interface for Redis backend

Author: cschleiden

backend/redis/diagnostics.go

@@ -0,0 +1,84 @@
+package redis
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/cschleiden/go-workflows/web"
+	"github.com/go-redis/redis/v8"
+)
+
+var _ web.WebBackend = (*redisBackend)(nil)
+
+func (rb *redisBackend) GetWorkflowInstances(ctx context.Context, afterInstanceID string, count int) ([]*web.WorkflowInstanceRef, error) {
+	max := "+inf"
+
+	if afterInstanceID != "" {
+		scores, err := rb.rdb.ZMScore(ctx, instancesByCreation(), afterInstanceID).Result()
+		if err != nil {
+			return nil, fmt.Errorf("getting instance score for %v: %w", afterInstanceID, err)
+		}
+
+		if len(scores) == 0 {
+			rb.Logger().Error("could not find instance %v", "afterInstanceID", afterInstanceID)
+			return nil, nil
+		}
+
+		max = fmt.Sprintf("(%v", int64(scores[0]))
+	}
+
+	result, err := rb.rdb.ZRangeArgs(ctx, redis.ZRangeArgs{
+		Key:     instancesByCreation(),
+		Stop:    max,
+		Start:   "-inf",
+		ByScore: true,
+		Rev:     true,
+		Count:   int64(count),
+	}).Result()
+	if err != nil {
+		return nil, fmt.Errorf("getting instances after %v: %w", max, err)
+	}
+
+	instanceIDs := make([]string, 0)
+	for _, r := range result {
+		instanceID := r
+		instanceIDs = append(instanceIDs, instanceKey(instanceID))
+	}
+
+	instances, err := rb.rdb.MGet(ctx, instanceIDs...).Result()
+	if err != nil {
+		return nil, fmt.Errorf("getting instances: %w", err)
+	}
+
+	var instanceRefs []*web.WorkflowInstanceRef
+	for _, instance := range instances {
+		var state instanceState
+		if err := json.Unmarshal([]byte(instance.(string)), &state); err != nil {
+			return nil, fmt.Errorf("unmarshaling instance state: %w", err)
+		}
+
+		instanceRefs = append(instanceRefs, &web.WorkflowInstanceRef{
+			Instance:    state.Instance,
+			CreatedAt:   state.CreatedAt,
+			CompletedAt: state.CompletedAt,
+			State:       state.State,
+		})
+	}
+
+	return instanceRefs, nil
+}
+
+func (rb *redisBackend) GetWorkflowInstance(ctx context.Context, instanceID string) (*web.WorkflowInstanceRef, error) {
+	instance, err := readInstance(ctx, rb.rdb, instanceID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &web.WorkflowInstanceRef{
+		Instance:    instance.Instance,
+		CreatedAt:   instance.CreatedAt,
+		CompletedAt: instance.CompletedAt,
+		State:       instance.State,
+	}, nil
+}

backend/redis/instance.go

@@ -119,10 +119,12 @@ type instanceState struct {
 func createInstance(ctx context.Context, rdb redis.UniversalClient, instance *core.WorkflowInstance, ignoreDuplicate bool) error {
 	key := instanceKey(instance.InstanceID)
 
+	createdAt := time.Now()
+
 	b, err := json.Marshal(&instanceState{
 		Instance:  instance,
 		State:     backend.WorkflowStateActive,
-		CreatedAt: time.Now(),
+		CreatedAt: createdAt,
 	})
 	if err != nil {
 		return fmt.Errorf("marshaling instance state: %w", err)
@@ -148,6 +150,13 @@ func createInstance(ctx context.Context, rdb redis.UniversalClient, instance *co
 		}
 	}
 
+	if err := rdb.ZAdd(ctx, instancesByCreation(), &redis.Z{
+		Member: instance.InstanceID,
+		Score:  float64(createdAt.UnixMilli()),
+	}).Err(); err != nil {
+		return fmt.Errorf("storing instance reference: %w", err)
+	}
+
 	return nil
 }
 
@@ -164,6 +173,8 @@ func updateInstance(ctx context.Context, rdb redis.UniversalClient, instanceID s
 		return fmt.Errorf("updating instance: %w", err)
 	}
 
+	// CreatedAt does not change, so skip updating the instancesByCreation() ZSET
+
 	return nil
 }
 

backend/redis/keys.go

@@ -8,6 +8,10 @@ func instanceKey(instanceID string) string {
 	return fmt.Sprintf("instance:%v", instanceID)
 }
 
+func instancesByCreation() string {
+	return "instances-by-creation"
+}
+
 func subInstanceKey(instanceID string) string {
 	return fmt.Sprintf("sub-instance:%v", instanceID)
 }

backend/redis/redis.go

@@ -34,7 +34,7 @@ func WithBackendOptions(opts ...backend.BackendOption) RedisBackendOption {
 	}
 }
 
-func NewRedisBackend(address, username, password string, db int, opts ...RedisBackendOption) (backend.Backend, error) {
+func NewRedisBackend(address, username, password string, db int, opts ...RedisBackendOption) (*redisBackend, error) {
 	client := redis.NewUniversalClient(&redis.UniversalOptions{
 		Addrs:    []string{address},
 		Username: username,