Add single worker mode

Author: cschleiden

docs/source/index.html.md

@@ -58,7 +58,7 @@ func Workflow1(ctx workflow.Context, input string) error {
 }
 ```
 
-Let's first write a simple workflows. Our workflow executes two _activities_ in sequence waiting for each result. Both workflows and activities are written in plain Go. Workflows can be long-running and have to be deterministic so that they can be interrupted and resumed. Activities are functions that can have side-effects and don't have to be deterministic.
+Let's first write a simple workflow. Our workflow executes two _activities_ in sequence waiting for each result. Both workflows and activities are written in plain Go. Workflows can be long-running and have to be deterministic so that they can be interrupted and resumed. Activities are functions that can have side-effects and don't have to be deterministic.
 
 Both workflows and activities support arbitrary inputs and outputs as long as those are serializable.
 
@@ -135,3 +135,71 @@ func main() {
 To finish the example, we create the backend, start a worker in a separate go-routine. We then create a `Client` instance which we then  use to create a new _workflow instance_. A workflow instance is just one running instance of a previously registered workflow.
 
 With the exception of the in-memory backend, we do not have to start the workflow from the same process the worker runs in, we could create the client from another process and create/wait for/cancel/... workflow instances from there.
+
+## Using the WorkflowOrchestrator
+
+For simpler scenarios where you don't need the separation between client and worker, you can use the `WorkflowOrchestrator` which combines both:
+
+```go
+// Define the workflow with activities that will be auto-registered
+func AutoRegisteredWorkflow(ctx workflow.Context, input string) (string, error) {
+	// When using WorkflowOrchestrator, activities are automatically registered
+	// You can directly use any activity function without registering it first
+	r1, err := workflow.ExecuteActivity[int](ctx, workflow.DefaultActivityOptions, Activity1).Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	// Another activity that will be auto-registered
+	r2, err := workflow.ExecuteActivity[string](ctx, workflow.DefaultActivityOptions, FormatResult, input, r1).Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return r2, nil
+}
+
+// Activities are automatically registered when using WorkflowOrchestrator
+func Activity1(ctx context.Context) (int, error) {
+	return 42, nil
+}
+
+func FormatResult(ctx context.Context, input string, value int) (string, error) {
+	return fmt.Sprintf("Processed %s with result %d", input, value), nil
+}
+
+func main() {
+	ctx := context.Background()
+
+	b := sqlite.NewSqliteBackend("simple.sqlite")
+
+	// Create orchestrator instead of separate client and worker
+	orchestrator := worker.NewWorkflowOrchestrator(b, nil)
+
+	// Workflows are automatically registered
+	// Activities defined with InlineActivity don't need registration
+
+	// Start the orchestrator
+	if err := orchestrator.Start(ctx); err != nil {
+		panic("could not start orchestrator")
+	}
+
+	// Create workflow instance
+	wf, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: uuid.NewString(),
+	}, AutoRegisteredWorkflow, "input-for-workflow")
+	if err != nil {
+		panic("could not start workflow")
+	}
+
+	// Get result directly
+	result, err := client.GetWorkflowResult[string](ctx, orchestrator.Client, wf, 5*time.Second)
+	if err != nil {
+		panic("error getting workflow result: " + err.Error())
+	}
+
+	fmt.Println("Workflow completed with result:", result)
+}
+```
+
+The `WorkflowOrchestrator` provides automatic registration of workflows when passing workflow functions directly to `CreateWorkflowInstance`. For activities, the orchestrator automatically registers any activities used in the workflows via `ExecuteActivity` without explicit registration. Similarly, any sub-workflows created with `CreateSubWorkflowInstance` are registered automatically. This approach offers a unified API for workflow creation and execution in a single component, simplifying the development experience for simpler scenarios.

internal/contextvalue/registry.go

@@ -0,0 +1,20 @@
+package contextvalue
+
+import (
+	"github.com/cschleiden/go-workflows/internal/sync"
+	"github.com/cschleiden/go-workflows/registry"
+)
+
+type registryKey struct{}
+
+func WithRegistry(ctx sync.Context, r *registry.Registry) sync.Context {
+	return sync.WithValue(ctx, registryKey{}, r)
+}
+
+func GetRegistry(ctx sync.Context) *registry.Registry {
+	if v := ctx.Value(registryKey{}); v != nil {
+		return v.(*registry.Registry)
+	}
+
+	return nil
+}

internal/contextvalue/singleworkermode.go

@@ -0,0 +1,21 @@
+package contextvalue
+
+import (
+	"github.com/cschleiden/go-workflows/internal/sync"
+)
+
+type singleWorkerModeKey struct{}
+
+// WithSingleWorkerMode sets the single worker mode flag in the context.
+// When this flag is set, activities and workflows will be automatically registered when needed.
+func WithSingleWorkerMode(ctx sync.Context) sync.Context {
+	return sync.WithValue(ctx, singleWorkerModeKey{}, true)
+}
+
+// IsSingleWorkerMode checks if the context is in single worker mode.
+func IsSingleWorkerMode(ctx sync.Context) bool {
+	if v := ctx.Value(singleWorkerModeKey{}); v != nil {
+		return v.(bool)
+	}
+	return false
+}

internal/worker/workflow.go

@@ -27,6 +27,7 @@ type WorkflowWorkerOptions struct {
 	WorkflowExecutorCache     executor.Cache
 	WorkflowExecutorCacheSize int
 	WorkflowExecutorCacheTTL  time.Duration
+	SingleWorkerMode          bool
 }
 
 func NewWorkflowWorker(
@@ -39,20 +40,22 @@ func NewWorkflowWorker(
 	}
 
 	tw := &WorkflowTaskWorker{
-		backend:  b,
-		registry: registry,
-		cache:    options.WorkflowExecutorCache,
-		logger:   b.Options().Logger,
+		backend:          b,
+		registry:         registry,
+		cache:            options.WorkflowExecutorCache,
+		logger:           b.Options().Logger,
+		singleWorkerMode: options.SingleWorkerMode,
 	}
 
 	return NewWorker(b, tw, &options.WorkerOptions)
 }
 
 type WorkflowTaskWorker struct {
-	backend  backend.Backend
-	registry *registry.Registry
-	cache    executor.Cache
-	logger   *slog.Logger
+	backend          backend.Backend
+	registry         *registry.Registry
+	cache            executor.Cache
+	logger           *slog.Logger
+	singleWorkerMode bool
 }
 
 func (wtw *WorkflowTaskWorker) Start(ctx context.Context, queues []workflow.Queue) error {
@@ -181,6 +184,7 @@ func (wtw *WorkflowTaskWorker) getExecutor(ctx context.Context, t *backend.Workf
 			t.Metadata,
 			clock.New(),
 			wtw.backend.Options().MaxHistorySize,
+			wtw.singleWorkerMode,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("creating workflow task executor: %w", err)

registry/auto_registry.go

@@ -0,0 +1,50 @@
+package registry
+
+import (
+	"fmt"
+)
+
+// AutoRegisteringRegistry extends the regular Registry with auto-registration capabilities
+// It can be used as an alternative to context-based SingleWorkerMode
+type AutoRegisteringRegistry struct {
+	*Registry
+}
+
+// NewAutoRegistering creates a new auto-registering registry
+func NewAutoRegistering() *AutoRegisteringRegistry {
+	return &AutoRegisteringRegistry{
+		Registry: New(),
+	}
+}
+
+// GetWorkflowWithAutoRegister returns the workflow function for the given name.
+// If the workflow is not found and wf is provided, it will attempt to register it.
+func (r *AutoRegisteringRegistry) GetWorkflowWithAutoRegister(name string, wf interface{}) (interface{}, error) {
+	v, err := r.Registry.GetWorkflow(name)
+
+	// If not found, try to register the workflow
+	if err != nil && wf != nil {
+		if err := r.RegisterWorkflow(wf); err != nil {
+			return nil, fmt.Errorf("auto-registering workflow %s: %w", name, err)
+		}
+		return r.Registry.GetWorkflow(name)
+	}
+
+	return v, err
+}
+
+// GetActivityWithAutoRegister returns the activity function for the given name.
+// If the activity is not found and activity is provided, it will attempt to register it.
+func (r *AutoRegisteringRegistry) GetActivityWithAutoRegister(name string, activity interface{}) (interface{}, error) {
+	v, err := r.Registry.GetActivity(name)
+
+	// If not found, try to register the activity
+	if err != nil && activity != nil {
+		if err := r.RegisterActivity(activity); err != nil {
+			return nil, fmt.Errorf("auto-registering activity %s: %w", name, err)
+		}
+		return r.Registry.GetActivity(name)
+	}
+
+	return v, err
+}

registry/registry.go

@@ -8,29 +8,38 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/args"
 	"github.com/cschleiden/go-workflows/internal/fn"
-	wf "github.com/cschleiden/go-workflows/workflow"
 )
 
 type Registry struct {
+	sealed bool
+
 	sync.Mutex
 
-	workflowMap map[string]wf.Workflow
-	activityMap map[string]interface{}
+	workflowMap map[string]any
+	activityMap map[string]any
 }
 
 // New creates a new registry instance.
 func New() *Registry {
 	return &Registry{
-		workflowMap: make(map[string]wf.Workflow),
-		activityMap: make(map[string]interface{}),
+		workflowMap: make(map[string]any),
+		activityMap: make(map[string]any),
 	}
 }
 
 type registerConfig struct {
 	Name string
 }
 
-func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption) error {
+func (r *Registry) Seal() {
+	r.sealed = true
+}
+
+func (r *Registry) Sealed() bool {
+	return r.sealed
+}
+
+func (r *Registry) RegisterWorkflow(workflow any, opts ...RegisterOption) error {
 	cfg := registerOptions(opts).applyRegisterOptions(registerConfig{})
 	name := cfg.Name
 	if name == "" {
@@ -75,7 +84,7 @@ func (r *Registry) RegisterWorkflow(workflow wf.Workflow, opts ...RegisterOption
 	return nil
 }
 
-func (r *Registry) RegisterActivity(activity wf.Activity, opts ...RegisterOption) error {
+func (r *Registry) RegisterActivity(activity any, opts ...RegisterOption) error {
 	cfg := registerOptions(opts).applyRegisterOptions(registerConfig{})
 
 	t := reflect.TypeOf(activity)
@@ -151,7 +160,7 @@ func checkActivity(actType reflect.Type) error {
 	return nil
 }
 
-func (r *Registry) GetWorkflow(name string) (wf.Workflow, error) {
+func (r *Registry) GetWorkflow(name string) (any, error) {
 	r.Lock()
 	defer r.Unlock()
 

registry/registry_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/cschleiden/go-workflows/internal/fn"
 	"github.com/cschleiden/go-workflows/internal/sync"
-	wf "github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,7 +16,7 @@ func reg_workflow1(ctx sync.Context) error {
 func TestRegistry_RegisterWorkflow(t *testing.T) {
 	type args struct {
 		name     string
-		workflow wf.Workflow
+		workflow any
 	}
 	tests := []struct {
 		name     string

samples/orchestrator/main.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/cschleiden/go-workflows/client"
+	"github.com/cschleiden/go-workflows/samples"
+	"github.com/cschleiden/go-workflows/worker"
+	"github.com/cschleiden/go-workflows/workflow"
+
+	"github.com/google/uuid"
+)
+
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	backend := samples.GetBackend("orchestrator")
+
+	orchestrator := worker.NewWorkflowOrchestrator(
+		backend,
+		nil,
+	)
+
+	if err := orchestrator.Start(ctx); err != nil {
+		panic("could not start orchestrator: " + err.Error())
+	}
+
+	// Create instance ID
+	instanceID := uuid.NewString()
+
+	// Create and run workflow using the orchestrator - no explicit registration needed
+	instance, err := orchestrator.CreateWorkflowInstance(ctx, client.WorkflowInstanceOptions{
+		InstanceID: instanceID,
+	}, SimpleWorkflow, "Hello from orchestrator!")
+	if err != nil {
+		panic("could not create workflow instance: " + err.Error())
+	}
+
+	// Wait for result using client package directly
+	result, err := client.GetWorkflowResult[string](ctx, orchestrator.Client, instance, 10*time.Second)
+	if err != nil {
+		panic("error getting workflow result: " + err.Error())
+	}
+
+	log.Printf("Workflow result: %s\n", result)
+
+	// Clean shutdown
+	cancel()
+
+	if err := orchestrator.WaitForCompletion(); err != nil {
+		panic("could not stop orchestrator: " + err.Error())
+	}
+}
+
+// SimpleWorkflow is a basic workflow that calls an activity and returns its result
+func SimpleWorkflow(ctx workflow.Context, message string) (string, error) {
+	future := workflow.ExecuteActivity[string](ctx, workflow.DefaultActivityOptions, ProcessMessage, message)
+
+	result, err := future.Get(ctx)
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// ProcessMessage is a simple activity that processes a message
+// When orchestrator mode is enabled, this will be automatically registered
+func ProcessMessage(ctx context.Context, message string) (string, error) {
+	return message + " (processed by activity)", nil
+}
+
+// Note: No separate activity function needed when using InlineActivity