feat: Add workflows

Author: jwillp

cmd/smallflow/main.go

@@ -3,83 +3,109 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/alitto/pond/v2"
+	"github.com/morebec/go-misas/misas"
+	"github.com/morebec/go-misas/mpostgres"
 	"github.com/morebec/go-misas/muuid"
 	"github.com/morebec/go-misas/mx"
-	"github.com/morebec/smallflow/internal/core"
-	"github.com/morebec/smallflow/internal/core/adapters"
 	"github.com/morebec/smallflow/internal/orchestrator"
 	adapters2 "github.com/morebec/smallflow/internal/orchestrator/adapters"
+	"github.com/morebec/smallflow/internal/workflowmgmt"
+	"github.com/morebec/smallflow/internal/workflowmgmt/adapters"
 	"time"
 )
 
 func main() {
 	fmt.Println("Build, run, and observe workflows without the overhead!")
 
-	eventStore := &adapters.InMemoryEventStore{}
 	clock := mx.NewRealTimeClock(time.UTC)
+
+	dbConn, err := mpostgres.OpenConn("postgres://smallflow:smallflow@localhost:5432/postgres?sslmode=disable")
+	if err != nil {
+		panic(err)
+	}
+
+	var eventStore misas.EventStore
+	eventStore, err = mpostgres.NewEventStore(clock, dbConn)
+	if err != nil {
+		panic(err)
+	}
+
+	eventRegistry := mx.NewMessageRegistry[misas.EventTypeName, misas.Event]()
+
+	eventRegistry.Register(workflowmgmt.WorkflowEnabledEventTypeName, workflowmgmt.WorkflowEnabledEvent{})
+	eventRegistry.Register(workflowmgmt.WorkflowDisabledEventTypeName, workflowmgmt.WorkflowDisabledEvent{})
+	eventRegistry.Register(workflowmgmt.WorkflowTriggeredEventTypeName, workflowmgmt.WorkflowTriggeredEvent{})
+	eventRegistry.Register(workflowmgmt.WorkflowStartedEventTypeName, workflowmgmt.WorkflowStartedEvent{})
+	eventRegistry.Register(workflowmgmt.WorkflowEndedEventTypeName, workflowmgmt.WorkflowEndedEvent{})
+	eventRegistry.Register(workflowmgmt.StepStartedEventTypeName, workflowmgmt.StepStartedEvent{})
+	eventRegistry.Register(workflowmgmt.StepEndedEventTypeName, workflowmgmt.StepEndedEvent{})
+
+	eventStore = mx.NewEventStoreDeserializerDecorator(eventStore, eventRegistry)
+
 	workflowRepo := &adapters.EventStoreWorkflowRepository{
-		EventStore: eventStore,
+		EventStore:    eventStore,
+		EventRegistry: eventRegistry,
+		UUIDGenerator: muuid.NewRandomUUIDGenerator(),
 	}
 	runRepo := &adapters.EventStoreRunRepository{
-		EventStore: eventStore,
+		EventStore:    eventStore,
+		EventRegistry: eventRegistry,
+		UUIDGenerator: muuid.NewRandomUUIDGenerator(),
 	}
 
-	api := core.NewSubsystem(clock, workflowRepo, runRepo, muuid.NewRandomUUIDGenerator()).API
-	orch := &orchestrator.WorkflowOrchestrator{
-		Clock: clock,
-		API:   api,
-		LeaseManager: orchestrator.WorkflowLeaseManager{
-			Clock:      clock,
-			Repository: adapters2.NewInMemoryWorkflowLeaseRepository(),
-		},
-		Pool: pond.NewPool(10),
+	api := workflowmgmt.NewSubsystem(clock, workflowRepo, runRepo, muuid.NewRandomUUIDGenerator()).API
+	workflowLeaseRepository, err := adapters2.NewPostgresWorkflowLeaseRepository(dbConn)
+	if err != nil {
+		panic(err)
 	}
 
-	ctx := context.Background()
-
-	fmt.Println("Enabling workflow...")
-	if result := api.HandleCommand(ctx, core.EnableWorkflowCommand{
-		WorkflowID: "my-workflow",
-	}); result.Error != nil {
-		panic(result.Error)
+	leaseManager := orchestrator.WorkflowLeaseManager{
+		Clock:      clock,
+		Repository: workflowLeaseRepository,
 	}
 
-	fmt.Println("Triggering workflow...")
-	if result := api.HandleCommand(ctx, core.TriggerWorkflowCommand{
-		WorkflowID: "my-workflow",
-		RunID:      muuid.NewRandomUUIDGenerator().Generate().String(),
-	}); result.Error != nil {
-		panic(result.Error)
+	checkpointStore, err := mpostgres.NewPostgreSQLCheckpointStore(dbConn)
+	if err != nil {
+		panic(err)
 	}
+	orch := orchestrator.NewWorkflowOrchestrator(
+		clock,
+		api,
+		leaseManager,
+		muuid.NewRandomUUIDGenerator(),
+		eventStore,
+		checkpointStore,
+	)
+	orch.Start()
+	defer orch.Stop()
+
+	ctx := context.Background()
 
-	fmt.Println("Disabling workflow...")
-	if result := api.HandleCommand(ctx, core.DisableWorkflowCommand{
+	fmt.Println("Enabling workflow...")
+	if result := api.HandleCommand(ctx, workflowmgmt.EnableWorkflowCommand{
 		WorkflowID: "my-workflow",
 	}); result.Error != nil {
 		panic(result.Error)
 	}
 
-	fmt.Println("Dispatching events through the orchestrator...")
-	orch.Start()
-	defer orch.Stop()
-
-	for _, event := range eventStore.Events() {
-		if err := orch.HandleEvent(ctx, event); err != nil {
-			panic(err)
+	for i := range 1 {
+		fmt.Printf("Triggering workflow #%d...\n", i+1)
+		if result := api.HandleCommand(ctx, workflowmgmt.TriggerWorkflowCommand{
+			WorkflowID: "my-workflow",
+			RunID:      muuid.NewRandomUUIDGenerator().Generate().String(),
+		}); result.Error != nil {
+			panic(result.Error)
 		}
 	}
 
-	for orch.IsRunning() {
-		select {
-		case <-time.After(15 * time.Second):
-			fmt.Println("Stopping orchestrator after 15 seconds...")
-			orch.Stop()
-		}
+	<-time.After(30 * time.Second)
+	fmt.Println("Current events in the event store:")
+	stream, err := eventStore.ReadFromStream(ctx, eventStore.GlobalStreamID(), misas.ReadFromEventStreamOptions{}.FromStart().Forward())
+	if err != nil {
+		panic(err)
 	}
 
-	fmt.Println("Current events in the event store:")
-	for i, event := range eventStore.Events() {
+	for i, event := range stream.Events {
 		fmt.Printf("Event %d: %T → %+v\n", i, event, event)
 	}
 }

compose.debug.yaml

@@ -1,8 +1,22 @@
 services:
-  smallflow:
-    image: smallflow
-    build:
-      context: .
-      dockerfile: ./Dockerfile
+#  smallflow:
+#    image: smallflow
+#    build:
+#      context: .
+#      dockerfile: ./Dockerfile
+#    ports:
+#      - 3000:3000
+
+  postgres:
+    image: postgres:18-alpine
+    environment:
+      POSTGRES_USER: smallflow
+      POSTGRES_PASSWORD: smallflow
+      POSTGRES_DB: smallflow
     ports:
-      - 3000:3000
+      - "5432:5432"
+    volumes:
+      - postgres:/var/lib/postgresql/data
+
+volumes:
+  postgres:

compose.yaml

@@ -7,11 +7,12 @@ replace github.com/morebec/go-misas => ./../go-misas-back
 require (
 	github.com/alitto/pond/v2 v2.5.0
 	github.com/morebec/go-misas v0.0.0-00010101000000-000000000000
+	github.com/samber/lo v1.49.1
 )
 
 require (
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/samber/lo v1.49.1 // indirect
+	github.com/lib/pq v1.10.9 // indirect
 	go.opentelemetry.io/otel v1.35.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	golang.org/x/text v0.24.0 // indirect

go.mod

@@ -1,3 +1,5 @@
+github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU=
+github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU=
 github.com/alitto/pond/v2 v2.5.0 h1:vPzS5GnvSDRhWQidmj2djHllOmjFExVFbDGCw1jdqDw=
 github.com/alitto/pond/v2 v2.5.0/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE=
 github.com/cucumber/gherkin/go/v26 v26.2.0 h1:EgIjePLWiPeslwIWmNQ3XHcypPsWAHoMCz/YEBKP4GI=
@@ -20,6 +22,8 @@ github.com/hashicorp/go-memdb v1.3.4 h1:XSL3NR682X/cVk2IeV0d70N4DZ9ljI885xAEU8Io
 github.com/hashicorp/go-memdb v1.3.4/go.mod h1:uBTr1oQbtuMgd1SSGoR8YV27eT3sBHbYiNm53bMpgSg=
 github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
 github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/samber/lo v1.49.1 h1:4BIFyVfuQSEpluc7Fua+j1NolZHiEHEpaSEKdsH0tew=