Add configurable worker names for SQL backends

Co-authored-by: cschleiden <[email protected]>

Author: Copilot

backend/mysql/mysql.go

@@ -55,7 +55,7 @@ func NewMysqlBackend(host string, port int, user, password, database string, opt
 	b := &mysqlBackend{
 		dsn:        dsn,
 		db:         db,
-		workerName: fmt.Sprintf("worker-%v", uuid.NewString()),
+		workerName: getWorkerName(options),
 		options:    options,
 	}
 
@@ -974,3 +974,11 @@ func scheduleActivity(ctx context.Context, tx *sql.Tx, queue workflow.Queue, ins
 
 	return err
 }
+
+// getWorkerName returns the worker name from options, or generates a UUID-based name if not set.
+func getWorkerName(options *options) string {
+	if options.WorkerName != "" {
+		return options.WorkerName
+	}
+	return fmt.Sprintf("worker-%v", uuid.NewString())
+}

backend/mysql/mysql_test.go

@@ -162,3 +162,47 @@ func (mb *mysqlBackend) GetFutureEvents(ctx context.Context) ([]*history.Event,
 
 	return f, nil
 }
+
+func Test_MysqlBackend_WorkerName(t *testing.T) {
+	if testing.Short() {
+		t.Skip()
+	}
+
+	t.Run("DefaultWorkerName", func(t *testing.T) {
+		// Create a backend without specifying worker name
+		// Since we can't connect to MySQL without it being available, we'll test the getWorkerName function directly
+		options := &options{}
+		workerName := getWorkerName(options)
+
+		// The default worker name should be in the format "worker-<uuid>"
+		if !strings.Contains(workerName, "worker-") {
+			t.Errorf("Expected worker name to contain 'worker-', got: %s", workerName)
+		}
+		if len(workerName) != 43 { // "worker-" (7) + UUID (36)
+			t.Errorf("Expected worker name length to be 43, got: %d", len(workerName))
+		}
+	})
+
+	t.Run("CustomWorkerName", func(t *testing.T) {
+		customWorkerName := "test-worker-123"
+		options := &options{WorkerName: customWorkerName}
+		workerName := getWorkerName(options)
+
+		if workerName != customWorkerName {
+			t.Errorf("Expected worker name to be '%s', got: %s", customWorkerName, workerName)
+		}
+	})
+
+	t.Run("EmptyWorkerNameUsesDefault", func(t *testing.T) {
+		options := &options{WorkerName: ""}
+		workerName := getWorkerName(options)
+
+		// Empty worker name should fall back to UUID generation
+		if !strings.Contains(workerName, "worker-") {
+			t.Errorf("Expected worker name to contain 'worker-', got: %s", workerName)
+		}
+		if len(workerName) != 43 { // "worker-" (7) + UUID (36)
+			t.Errorf("Expected worker name length to be 43, got: %d", len(workerName))
+		}
+	})
+}

backend/mysql/options.go

@@ -13,10 +13,21 @@ type options struct {
 
 	// ApplyMigrations automatically applies database migrations on startup.
 	ApplyMigrations bool
+
+	// WorkerName allows setting a custom worker name. If not set, a random UUID will be generated.
+	WorkerName string
 }
 
 type option func(*options)
 
+// WithWorkerName sets a custom worker name for the MySQL backend.
+// If not provided, a random UUID will be generated.
+func WithWorkerName(workerName string) option {
+	return func(o *options) {
+		o.WorkerName = workerName
+	}
+}
+
 // WithApplyMigrations automatically applies database migrations on startup.
 func WithApplyMigrations(applyMigrations bool) option {
 	return func(o *options) {

backend/sqlite/options.go

@@ -9,10 +9,21 @@ type options struct {
 
 	// ApplyMigrations automatically applies database migrations on startup.
 	ApplyMigrations bool
+
+	// WorkerName allows setting a custom worker name. If not set, a random UUID will be generated.
+	WorkerName string
 }
 
 type option func(*options)
 
+// WithWorkerName sets a custom worker name for the SQLite backend.
+// If not provided, a random UUID will be generated.
+func WithWorkerName(workerName string) option {
+	return func(o *options) {
+		o.WorkerName = workerName
+	}
+}
+
 // WithApplyMigrations automatically applies database migrations on startup.
 func WithApplyMigrations(applyMigrations bool) option {
 	return func(o *options) {

backend/sqlite/sqlite.go

@@ -76,7 +76,7 @@ func newSqliteBackend(dsn string, opts ...option) *sqliteBackend {
 
 	b := &sqliteBackend{
 		db:         db,
-		workerName: fmt.Sprintf("worker-%v", uuid.NewString()),
+		workerName: getWorkerName(options),
 		options:    options,
 	}
 
@@ -862,3 +862,11 @@ func (sb *sqliteBackend) ExtendActivityTask(ctx context.Context, task *backend.A
 
 	return tx.Commit()
 }
+
+// getWorkerName returns the worker name from options, or generates a UUID-based name if not set.
+func getWorkerName(options *options) string {
+	if options.WorkerName != "" {
+		return options.WorkerName
+	}
+	return fmt.Sprintf("worker-%v", uuid.NewString())
+}

backend/sqlite/sqlite_test.go

@@ -1,10 +1,15 @@
 package sqlite
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/cschleiden/go-workflows/backend"
+	"github.com/cschleiden/go-workflows/backend/history"
 	"github.com/cschleiden/go-workflows/backend/test"
+	"github.com/cschleiden/go-workflows/core"
+	"github.com/cschleiden/go-workflows/workflow"
 	"github.com/stretchr/testify/require"
 )
 
@@ -36,3 +41,74 @@ func Test_EndToEndSqliteBackend(t *testing.T) {
 		require.NoError(t, b.Close())
 	})
 }
+
+func Test_SqliteBackend_WorkerName(t *testing.T) {
+	t.Run("DefaultWorkerName", func(t *testing.T) {
+		backend := NewInMemoryBackend()
+		defer backend.Close()
+
+		// The default worker name should be in the format "worker-<uuid>"
+		require.Contains(t, backend.workerName, "worker-")
+		require.Len(t, backend.workerName, 43) // "worker-" (7) + UUID (36)
+	})
+
+	t.Run("CustomWorkerName", func(t *testing.T) {
+		customWorkerName := "test-worker-123"
+		backend := NewInMemoryBackend(WithWorkerName(customWorkerName))
+		defer backend.Close()
+
+		require.Equal(t, customWorkerName, backend.workerName)
+	})
+
+	t.Run("EmptyWorkerNameUsesDefault", func(t *testing.T) {
+		backend := NewInMemoryBackend(WithWorkerName(""))
+		defer backend.Close()
+
+		// Empty worker name should fall back to UUID generation
+		require.Contains(t, backend.workerName, "worker-")
+		require.Len(t, backend.workerName, 43) // "worker-" (7) + UUID (36)
+	})
+
+	t.Run("CustomWorkerNameIsUsedInDatabase", func(t *testing.T) {
+		customWorkerName := "integration-test-worker"
+		backend := NewInMemoryBackend(WithWorkerName(customWorkerName))
+		defer backend.Close()
+
+		// Verify the worker name is stored correctly
+		require.Equal(t, customWorkerName, backend.workerName)
+
+		// Create a workflow instance and task to ensure the worker name is actually used
+		ctx := context.Background()
+		instance := core.NewWorkflowInstance("test-instance", "test-execution")
+		
+		event := history.NewPendingEvent(
+			time.Now(),
+			history.EventType_WorkflowExecutionStarted,
+			&history.ExecutionStartedAttributes{
+				Queue: "test-queue",
+				Metadata: &workflow.Metadata{},
+			},
+		)
+
+		// Create workflow instance
+		err := backend.CreateWorkflowInstance(ctx, instance, event)
+		require.NoError(t, err)
+
+		// Get a workflow task (this should lock it with our custom worker name)
+		task, err := backend.GetWorkflowTask(ctx, []workflow.Queue{"test-queue"})
+		require.NoError(t, err)
+		require.NotNil(t, task)
+
+		// Query the database to verify our custom worker name is used
+		rows, err := backend.db.Query("SELECT worker FROM instances WHERE id = ? AND execution_id = ?", 
+			instance.InstanceID, instance.ExecutionID)
+		require.NoError(t, err)
+		defer rows.Close()
+
+		var workerNameFromDB string
+		require.True(t, rows.Next())
+		err = rows.Scan(&workerNameFromDB)
+		require.NoError(t, err)
+		require.Equal(t, customWorkerName, workerNameFromDB)
+	})
+}