Make redis backend honor lastSequenceID when retrieving history

Author: cschleiden

backend/redis/events.go

@@ -31,20 +31,21 @@ func addEventToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string,
 // ARGV[1] - event data as serialized strings
 var addEventsToStreamCmd = redis.NewScript(`
 	local msgID = ""
-	for i = 1, #ARGV do
-		msgID = redis.call("XADD", KEYS[1], "*", "event", ARGV[i])
+	for i = 1, #ARGV,2 do
+		msgID = redis.call("XADD", KEYS[1], ARGV[i], "event", ARGV[i + 1])
 	end
 	return msgID
 `)
 
-func addEventsToStreamP(ctx context.Context, p redis.Pipeliner, streamKey string, events []history.Event) error {
+func addEventsToHistoryStreamP(ctx context.Context, p redis.Pipeliner, streamKey string, events []history.Event) error {
 	eventsData := make([]string, 0)
 	for _, event := range events {
 		eventData, err := json.Marshal(event)
 		if err != nil {
 			return err
 		}
 
+		eventsData = append(eventsData, historyID(event.SequenceID))
 		eventsData = append(eventsData, string(eventData))
 	}
 

backend/redis/instance.go

@@ -58,7 +58,13 @@ func (rb *redisBackend) CreateWorkflowInstance(ctx context.Context, instance *wo
 }
 
 func (rb *redisBackend) GetWorkflowInstanceHistory(ctx context.Context, instance *core.WorkflowInstance, lastSequenceID *int64) ([]history.Event, error) {
-	msgs, err := rb.rdb.XRange(ctx, historyKey(instance.InstanceID), "-", "+").Result()
+	start := "-"
+
+	if lastSequenceID != nil {
+		start = "(" + historyID(*lastSequenceID)
+	}
+
+	msgs, err := rb.rdb.XRange(ctx, historyKey(instance.InstanceID), start, "+").Result()
 	if err != nil {
 		return nil, err
 	}

backend/redis/keys.go

@@ -20,6 +20,10 @@ func historyKey(instanceID string) string {
 	return fmt.Sprintf("history:%v", instanceID)
 }
 
+func historyID(sequenceID int64) string {
+	return fmt.Sprintf("%v-0", sequenceID)
+}
+
 func futureEventsKey() string {
 	return "future-events"
 }