helper to list workflow registration (#157)

Adds exported methods to list registered workflows and scheduled
workflows with their registration parameters.

Resolves #150

---------

Co-authored-by: Max dml <[email protected]>

Author: PhakornKiong

dbos/dbos.go

@@ -121,12 +121,13 @@ type DBOSContext interface {
 	GetStepID() (int, error)                                                                                    // Get the current step ID (only available within workflows)
 
 	// Workflow management
-	RetrieveWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)     // Get a handle to an existing workflow
-	CancelWorkflow(_ DBOSContext, workflowID string) error                              // Cancel a workflow by setting its status to CANCELLED
-	ResumeWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)       // Resume a cancelled workflow
-	ForkWorkflow(_ DBOSContext, input ForkWorkflowInput) (WorkflowHandle[any], error)   // Fork a workflow from a specific step
-	ListWorkflows(_ DBOSContext, opts ...ListWorkflowsOption) ([]WorkflowStatus, error) // List workflows based on filtering criteria
-	GetWorkflowSteps(_ DBOSContext, workflowID string) ([]StepInfo, error)              // Get the execution steps of a workflow
+	RetrieveWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)                                // Get a handle to an existing workflow
+	CancelWorkflow(_ DBOSContext, workflowID string) error                                                         // Cancel a workflow by setting its status to CANCELLED
+	ResumeWorkflow(_ DBOSContext, workflowID string) (WorkflowHandle[any], error)                                  // Resume a cancelled workflow
+	ForkWorkflow(_ DBOSContext, input ForkWorkflowInput) (WorkflowHandle[any], error)                              // Fork a workflow from a specific step
+	ListWorkflows(_ DBOSContext, opts ...ListWorkflowsOption) ([]WorkflowStatus, error)                            // List workflows based on filtering criteria
+	GetWorkflowSteps(_ DBOSContext, workflowID string) ([]StepInfo, error)                                         // Get the execution steps of a workflow
+	ListRegisteredWorkflows(_ DBOSContext, opts ...ListRegisteredWorkflowsOption) ([]WorkflowRegistryEntry, error) // List registered workflows with filtering options
 
 	// Accessors
 	GetApplicationVersion() string // Get the application version for this context
@@ -159,7 +160,7 @@ type dbosContext struct {
 	workflowsWg *sync.WaitGroup
 
 	// Workflow registry - read-mostly sync.Map since registration happens only before launch
-	workflowRegistry        *sync.Map // map[string]workflowRegistryEntry
+	workflowRegistry        *sync.Map // map[string]WorkflowRegistryEntry
 	workflowCustomNametoFQN *sync.Map // Maps fully qualified workflow names to custom names. Usefor when client enqueues a workflow by name because registry is indexed by FQN.
 
 	// Workflow scheduler
@@ -275,6 +276,34 @@ func (c *dbosContext) GetApplicationID() string {
 	return c.applicationID
 }
 
+// ListRegisteredWorkflows returns information about registered workflows with their registration parameters.
+// Supports filtering using functional options.
+func (c *dbosContext) ListRegisteredWorkflows(_ DBOSContext, opts ...ListRegisteredWorkflowsOption) ([]WorkflowRegistryEntry, error) {
+	// Initialize parameters with defaults
+	params := &listRegisteredWorkflowsOptions{}
+
+	// Apply all provided options
+	for _, opt := range opts {
+		opt(params)
+	}
+
+	// Get all registered workflows and apply filters
+	var filteredWorkflows []WorkflowRegistryEntry
+	c.workflowRegistry.Range(func(key, value interface{}) bool {
+		workflow := value.(WorkflowRegistryEntry)
+
+		// Filter by scheduled only
+		if params.scheduledOnly && workflow.CronSchedule == "" {
+			return true
+		}
+
+		filteredWorkflows = append(filteredWorkflows, workflow)
+		return true
+	})
+
+	return filteredWorkflows, nil
+}
+
 // NewDBOSContext creates a new DBOS context with the provided configuration.
 // The context must be launched with Launch() for workflow execution and should be shut down with Shutdown().
 // This function initializes the DBOS system database, sets up the queue sub-system, and prepares the workflow registry.

dbos/queue.go

@@ -221,7 +221,7 @@ func (qr *queueRunner) run(ctx *dbosContext) {
 					qr.logger.Error("workflow function not found in registry", "workflow_name", workflow.name)
 					continue
 				}
-				registeredWorkflow, ok := registeredWorkflowAny.(workflowRegistryEntry)
+				registeredWorkflow, ok := registeredWorkflowAny.(WorkflowRegistryEntry)
 				if !ok {
 					qr.logger.Error("invalid workflow registry entry type", "workflow_name", workflow.name)
 					continue

dbos/recovery.go

@@ -48,7 +48,7 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 			ctx.logger.Error("Workflow function not found in registry", "workflow_id", workflow.ID, "name", workflow.Name)
 			continue
 		}
-		registeredWorkflow, ok := registeredWorkflowAny.(workflowRegistryEntry)
+		registeredWorkflow, ok := registeredWorkflowAny.(WorkflowRegistryEntry)
 		if !ok {
 			ctx.logger.Error("invalid workflow registry entry type", "workflow_id", workflow.ID, "name", workflow.Name)
 			continue

dbos/workflow.go

@@ -269,10 +269,12 @@ func (h *workflowHandleProxy[R]) GetWorkflowID() string {
 /**********************************/
 type wrappedWorkflowFunc func(ctx DBOSContext, input any, opts ...WorkflowOption) (WorkflowHandle[any], error)
 
-type workflowRegistryEntry struct {
+type WorkflowRegistryEntry struct {
 	wrappedFunction wrappedWorkflowFunc
-	maxRetries      int
-	name            string
+	MaxRetries      int
+	Name            string
+	FQN             string // Fully qualified name of the workflow function
+	CronSchedule    string // Empty string for non-scheduled workflows
 }
 
 func registerWorkflow(ctx DBOSContext, workflowFQN string, fn wrappedWorkflowFunc, maxRetries int, customName string) {
@@ -287,10 +289,12 @@ func registerWorkflow(ctx DBOSContext, workflowFQN string, fn wrappedWorkflowFun
 	}
 
 	// Check if workflow already exists and store atomically using LoadOrStore
-	entry := workflowRegistryEntry{
+	entry := WorkflowRegistryEntry{
 		wrappedFunction: fn,
-		maxRetries:      maxRetries,
-		name:            customName,
+		FQN:             workflowFQN,
+		MaxRetries:      maxRetries,
+		Name:            customName,
+		CronSchedule:    "",
 	}
 
 	if _, exists := c.workflowRegistry.LoadOrStore(workflowFQN, entry); exists {
@@ -321,6 +325,15 @@ func registerScheduledWorkflow(ctx DBOSContext, workflowName string, fn Workflow
 		panic("Cannot register scheduled workflow after DBOS has launched")
 	}
 
+	// Update the existing workflow entry with the cron schedule
+	registryEntryAny, exists := c.workflowRegistry.Load(workflowName)
+	if !exists {
+		panic(fmt.Sprintf("workflow %s must be registered before scheduling", workflowName))
+	}
+	registryEntry := registryEntryAny.(WorkflowRegistryEntry)
+	registryEntry.CronSchedule = cronSchedule
+	c.workflowRegistry.Store(workflowName, registryEntry)
+
 	var entryID cron.EntryID
 	entryID, err := c.getWorkflowScheduler().AddFunc(cronSchedule, func() {
 		// Execute the workflow on the cron schedule once DBOS is launched
@@ -658,15 +671,15 @@ func (c *dbosContext) RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opt
 	if !exists {
 		return nil, newNonExistentWorkflowError(params.workflowName)
 	}
-	registeredWorkflow, ok := registeredWorkflowAny.(workflowRegistryEntry)
+	registeredWorkflow, ok := registeredWorkflowAny.(WorkflowRegistryEntry)
 	if !ok {
 		return nil, fmt.Errorf("invalid workflow registry entry type for workflow %s", params.workflowName)
 	}
-	if registeredWorkflow.maxRetries > 0 {
-		params.maxRetries = registeredWorkflow.maxRetries
+	if registeredWorkflow.MaxRetries > 0 {
+		params.maxRetries = registeredWorkflow.MaxRetries
 	}
-	if len(registeredWorkflow.name) > 0 {
-		params.workflowName = registeredWorkflow.name
+	if len(registeredWorkflow.Name) > 0 {
+		params.workflowName = registeredWorkflow.Name
 	}
 
 	// Check if we are within a workflow (and thus a child workflow)
@@ -1951,3 +1964,48 @@ func GetWorkflowSteps(ctx DBOSContext, workflowID string) ([]StepInfo, error) {
 	}
 	return ctx.GetWorkflowSteps(ctx, workflowID)
 }
+
+// listRegisteredWorkflowsOptions holds configuration parameters for listing registered workflows
+type listRegisteredWorkflowsOptions struct {
+	scheduledOnly bool
+}
+
+// ListRegisteredWorkflowsOption is a functional option for configuring registered workflow listing parameters.
+type ListRegisteredWorkflowsOption func(*listRegisteredWorkflowsOptions)
+
+// WithScheduledOnly filters to only return scheduled workflows (those with a cron schedule).
+func WithScheduledOnly() ListRegisteredWorkflowsOption {
+	return func(p *listRegisteredWorkflowsOptions) {
+		p.scheduledOnly = true
+	}
+}
+
+// ListRegisteredWorkflows returns information about workflows registered with DBOS.
+// Each WorkflowRegistryEntry contains:
+// - MaxRetries: Maximum number of retry attempts for workflow recovery
+// - Name: Custom name if provided during registration, otherwise empty
+// - FQN: Fully qualified name of the workflow function (always present)
+// - CronSchedule: Empty string for non-scheduled workflows
+//
+// The function supports filtering using functional options:
+// - WithScheduledOnly(): Return only scheduled workflows
+//
+// Example:
+//
+//	// List all registered workflows
+//	workflows, err := dbos.ListRegisteredWorkflows(ctx)
+//	if err != nil {
+//	    log.Fatal(err)
+//	}
+//
+//	// List only scheduled workflows
+//	scheduled, err := dbos.ListRegisteredWorkflows(ctx, dbos.WithScheduledOnly())
+//	if err != nil {
+//	    log.Fatal(err)
+//	}
+func ListRegisteredWorkflows(ctx DBOSContext, opts ...ListRegisteredWorkflowsOption) ([]WorkflowRegistryEntry, error) {
+	if ctx == nil {
+		return nil, errors.New("ctx cannot be nil")
+	}
+	return ctx.ListRegisteredWorkflows(ctx, opts...)
+}

dbos/workflows_test.go

@@ -47,6 +47,10 @@ func simpleWorkflowWithStepError(dbosCtx DBOSContext, input string) (string, err
 	})
 }
 
+func simpleWorkflowWithSchedule(dbosCtx DBOSContext, scheduledTime time.Time) (time.Time, error) {
+	return scheduledTime, nil
+}
+
 // idempotencyWorkflow increments a global counter and returns the input
 func incrementCounter(_ context.Context, value int64) (int64, error) {
 	idempotencyCounter += value
@@ -4238,6 +4242,73 @@ func TestSpecialSteps(t *testing.T) {
 		require.Equal(t, "success", result, "workflow should return success")
 	})
 }
+
+func TestRegisteredWorkflowListing(t *testing.T) {
+	dbosCtx := setupDBOS(t, true, true)
+
+	// Register some regular workflows
+	RegisterWorkflow(dbosCtx, simpleWorkflow)
+	RegisterWorkflow(dbosCtx, simpleWorkflowError, WithMaxRetries(5))
+	RegisterWorkflow(dbosCtx, simpleWorkflowWithStep, WithWorkflowName("CustomStepWorkflow"))
+	RegisterWorkflow(dbosCtx, simpleWorkflowWithSchedule, WithWorkflowName("ScheduledWorkflow"), WithSchedule("0 0 * * * *"))
+
+	err := Launch(dbosCtx)
+	require.NoError(t, err, "failed to launch DBOS")
+
+	t.Run("ListRegisteredWorkflows", func(t *testing.T) {
+		workflows, err := ListRegisteredWorkflows(dbosCtx)
+		require.NoError(t, err, "ListRegisteredWorkflows should not return an error")
+
+		// Should have 4 workflows (3 regular + 1 scheduled)
+		require.GreaterOrEqual(t, len(workflows), 4, "Should have 4 registered workflows")
+
+		// Create a map for easier lookup
+		workflowMap := make(map[string]WorkflowRegistryEntry)
+		for _, wf := range workflows {
+			workflowMap[wf.FQN] = wf
+		}
+
+		// Check that simpleWorkflow is registered
+		simpleWorkflowFQN := runtime.FuncForPC(reflect.ValueOf(simpleWorkflow).Pointer()).Name()
+		simpleWf, exists := workflowMap[simpleWorkflowFQN]
+		require.True(t, exists, "simpleWorkflow should be registered")
+		require.Equal(t, _DEFAULT_MAX_RECOVERY_ATTEMPTS, simpleWf.MaxRetries, "simpleWorkflow should have default max retries")
+		require.Empty(t, simpleWf.CronSchedule, "simpleWorkflow should not have cron schedule")
+
+		// Check that simpleWorkflowError is registered with custom max retries
+		simpleWorkflowErrorFQN := runtime.FuncForPC(reflect.ValueOf(simpleWorkflowError).Pointer()).Name()
+		errorWf, exists := workflowMap[simpleWorkflowErrorFQN]
+		require.True(t, exists, "simpleWorkflowError should be registered")
+		require.Equal(t, 5, errorWf.MaxRetries, "simpleWorkflowError should have custom max retries")
+		require.Empty(t, errorWf.CronSchedule, "simpleWorkflowError should not have cron schedule")
+
+		// Check that custom named workflow is registered
+		customStepWorkflowFQN := runtime.FuncForPC(reflect.ValueOf(simpleWorkflowWithStep).Pointer()).Name()
+		customWf, exists := workflowMap[customStepWorkflowFQN]
+		require.True(t, exists, "CustomStepWorkflow should be found")
+		require.Equal(t, "CustomStepWorkflow", customWf.Name, "CustomStepWorkflow should have the correct name")
+		require.Empty(t, customWf.CronSchedule, "CustomStepWorkflow should not have cron schedule")
+
+		// Check that scheduled workflow is registered
+		scheduledWorkflowFQN := runtime.FuncForPC(reflect.ValueOf(simpleWorkflowWithSchedule).Pointer()).Name()
+		scheduledWf, exists := workflowMap[scheduledWorkflowFQN]
+		require.True(t, exists, "ScheduledWorkflow should be found")
+		require.Equal(t, "ScheduledWorkflow", scheduledWf.Name, "ScheduledWorkflow should have the correct name")
+		require.Equal(t, "0 0 * * * *", scheduledWf.CronSchedule, "ScheduledWorkflow should have the correct cron schedule")
+	})
+
+	t.Run("ListRegisteredWorkflowsWithScheduledOnly", func(t *testing.T) {
+		scheduledWorkflows, err := ListRegisteredWorkflows(dbosCtx, WithScheduledOnly())
+		require.NoError(t, err, "ListRegisteredWorkflows with WithScheduledOnly should not return an error")
+		require.Equal(t, 1, len(scheduledWorkflows), "Should have exactly 1 scheduled workflow")
+
+		entry := scheduledWorkflows[0]
+		scheduledWorkflowFQN := runtime.FuncForPC(reflect.ValueOf(simpleWorkflowWithSchedule).Pointer()).Name()
+		require.Equal(t, scheduledWorkflowFQN, entry.FQN, "ScheduledWorkflow should have the correct FQN")
+		require.Equal(t, "0 0 * * * *", entry.CronSchedule, "ScheduledWorkflow should have the correct cron schedule")
+	})
+}
+
 func TestWorkflowIdentity(t *testing.T) {
 	dbosCtx := setupDBOS(t, true, true)
 	RegisterWorkflow(dbosCtx, simpleWorkflow)