Fix/improve event grouping

Author: cschleiden

backend/mysql/mysql.go

@@ -517,7 +517,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 

backend/redis/workflow.go

@@ -210,7 +210,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				// Create new instance
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
-				if err := createInstanceP(ctx, p, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstanceP(ctx, p, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 			}
@@ -222,7 +222,7 @@ func (rb *redisBackend) CompleteWorkflowTask(
 		}
 
 		// Try to queue workflow task
-		if targetInstance != instance {
+		if targetInstance != *instance {
 			if err := rb.workflowQueue.Enqueue(ctx, p, targetInstance.InstanceID, nil); err != nil {
 				return fmt.Errorf("enqueuing workflow task: %w", err)
 			}

backend/sqlite/sqlite.go

@@ -406,7 +406,7 @@ func (sb *sqliteBackend) CompleteWorkflowTask(
 			if event.Type == history.EventType_WorkflowExecutionStarted {
 				a := event.Attributes.(*history.ExecutionStartedAttributes)
 				// Create new instance
-				if err := createInstance(ctx, tx, targetInstance, a.Metadata, true); err != nil {
+				if err := createInstance(ctx, tx, &targetInstance, a.Metadata, true); err != nil {
 					return err
 				}
 

internal/history/grouping.go

@@ -2,15 +2,17 @@ package history
 
 import "github.com/cschleiden/go-workflows/internal/core"
 
-func EventsByWorkflowInstance(events []WorkflowEvent) map[*core.WorkflowInstance][]Event {
-	groupedEvents := make(map[*core.WorkflowInstance][]Event)
+func EventsByWorkflowInstance(events []WorkflowEvent) map[core.WorkflowInstance][]Event {
+	groupedEvents := make(map[core.WorkflowInstance][]Event)
 
 	for _, m := range events {
-		if _, ok := groupedEvents[m.WorkflowInstance]; !ok {
-			groupedEvents[m.WorkflowInstance] = []Event{}
+		instance := *m.WorkflowInstance
+
+		if _, ok := groupedEvents[instance]; !ok {
+			groupedEvents[instance] = []Event{}
 		}
 
-		groupedEvents[m.WorkflowInstance] = append(groupedEvents[m.WorkflowInstance], m.HistoryEvent)
+		groupedEvents[instance] = append(groupedEvents[instance], m.HistoryEvent)
 	}
 
 	return groupedEvents

internal/history/grouping_test.go

@@ -0,0 +1,28 @@
+package history
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cschleiden/go-workflows/internal/core"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGrouping_MultipleEventsSameInstance(t *testing.T) {
+	id := uuid.NewString()
+
+	r := EventsByWorkflowInstance([]WorkflowEvent{
+		{
+			WorkflowInstance: core.NewWorkflowInstance(id, "exid"),
+			HistoryEvent:     NewPendingEvent(time.Now(), EventType_SubWorkflowScheduled, &SubWorkflowScheduledAttributes{}),
+		},
+		{
+			WorkflowInstance: core.NewWorkflowInstance(id, "exid"),
+			HistoryEvent:     NewPendingEvent(time.Now(), EventType_SubWorkflowScheduled, &SubWorkflowScheduledAttributes{}),
+		},
+	})
+
+	require.Len(t, r, 1)
+	require.Len(t, r[*core.NewWorkflowInstance(id, "exid")], 2)
+}