cleanup

Author: maxdml

dbos/client.go

@@ -149,7 +149,7 @@ func (c *client) Enqueue(queueName, workflowName string, input any, opts ...Enqu
 	}
 
 	// Serialize input before storing in workflow status
-	encodedInputStr, err := serialize(dbosCtx.serializer, params.workflowInput)
+	encodedInputStr, err := dbosCtx.serializer.Encode(params.workflowInput)
 	if err != nil {
 		return nil, fmt.Errorf("failed to serialize workflow input: %w", err)
 	}

dbos/recovery.go

@@ -14,17 +14,6 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 	}
 
 	for _, workflow := range pendingWorkflows {
-		// Pass encoded input directly - decoding will happen in workflow wrapper when we know the target type
-		var encodedInput *string
-		if workflow.Input != nil {
-			inputString, ok := workflow.Input.(*string)
-			if !ok {
-				ctx.logger.Warn("Skipping workflow recovery due to invalid input type", "workflow_id", workflow.ID, "name", workflow.Name, "input_type", workflow.Input)
-				continue
-			}
-			encodedInput = inputString
-		}
-
 		if workflow.QueueName != "" {
 			cleared, err := ctx.systemDB.clearQueueAssignment(ctx, workflow.ID)
 			if err != nil {
@@ -59,7 +48,8 @@ func recoverPendingWorkflows(ctx *dbosContext, executorIDs []string) ([]Workflow
 			WithWorkflowID(workflow.ID),
 		}
 		// Create a workflow context from the executor context
-		handle, err := registeredWorkflow.wrappedFunction(ctx, encodedInput, opts...)
+		// Pass encoded input directly - decoding will happen in workflow wrapper when we know the target type
+		handle, err := registeredWorkflow.wrappedFunction(ctx, workflow.Input, opts...)
 		if err != nil {
 			return nil, err
 		}

dbos/serialization.go

@@ -58,14 +58,6 @@ func (j *JSONSerializer) Decode(data *string) (any, error) {
 	return result, nil
 }
 
-// serialize serializes data using the provided serializer
-func serialize[T any](serializer Serializer, data T) (string, error) {
-	if serializer == nil {
-		return "", fmt.Errorf("serializer cannot be nil")
-	}
-	return serializer.Encode(data)
-}
-
 // deserialize decodes an encoded string directly into a typed variable.
 // For JSON serializer, this decodes directly into the target type, preserving type information.
 // For other serializers, it decodes into any and then type-asserts.

dbos/workflow.go

@@ -217,7 +217,7 @@ func (h *workflowHandle[R]) processOutcome(outcome workflowOutcome[R]) (R, error
 		if dbosCtx.serializer == nil {
 			return *new(R), newWorkflowExecutionError(workflowState.workflowID, fmt.Errorf("no serializer configured in DBOSContext"))
 		}
-		encodedOutputStr, encErr := serialize(dbosCtx.serializer, typedResult)
+		encodedOutputStr, encErr := dbosCtx.serializer.Encode(typedResult)
 		if encErr != nil {
 			return *new(R), newWorkflowExecutionError(workflowState.workflowID, fmt.Errorf("serializing child workflow result: %w", encErr))
 		}
@@ -286,7 +286,7 @@ func (h *workflowPollingHandle[R]) GetResult(opts ...GetResultOption) (R, error)
 			if dbosCtx.serializer == nil {
 				return *new(R), newWorkflowExecutionError(workflowState.workflowID, fmt.Errorf("no serializer configured in DBOSContext"))
 			}
-			encodedOutputStr, encErr := serialize(dbosCtx.serializer, typedResult)
+			encodedOutputStr, encErr := dbosCtx.serializer.Encode(typedResult)
 			if encErr != nil {
 				return *new(R), newWorkflowExecutionError(workflowState.workflowID, fmt.Errorf("serializing child workflow result: %w", encErr))
 			}
@@ -876,7 +876,7 @@ func (c *dbosContext) RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opt
 	}
 
 	// Serialize input before storing in workflow status
-	encodedInputStr, serErr := serialize(c.serializer, input)
+	encodedInputStr, serErr := c.serializer.Encode(input)
 	if serErr != nil {
 		return nil, newWorkflowExecutionError(workflowID, fmt.Errorf("failed to serialize workflow input: %w", serErr))
 	}
@@ -1044,7 +1044,7 @@ func (c *dbosContext) RunWorkflow(_ DBOSContext, fn WorkflowFunc, input any, opt
 			}
 
 			// Serialize the output before recording
-			encodedOutput, serErr := serialize(c.serializer, result)
+			encodedOutput, serErr := c.serializer.Encode(result)
 			if serErr != nil {
 				c.logger.Error("Failed to serialize workflow output", "workflow_id", workflowID, "error", serErr)
 				outcomeChan <- workflowOutcome[any]{result: nil, err: fmt.Errorf("failed to serialize output: %w", serErr)}
@@ -1343,7 +1343,7 @@ func (c *dbosContext) RunAsStep(_ DBOSContext, fn StepFunc, opts ...StepOption)
 	}
 
 	// Serialize step output before recording
-	encodedStepOutput, serErr := serialize(c.serializer, stepOutput)
+	encodedStepOutput, serErr := c.serializer.Encode(stepOutput)
 	if serErr != nil {
 		return nil, newStepExecutionError(stepState.workflowID, stepOpts.stepName, fmt.Errorf("failed to serialize step output: %w", serErr))
 	}
@@ -1372,7 +1372,7 @@ func (c *dbosContext) RunAsStep(_ DBOSContext, fn StepFunc, opts ...StepOption)
 
 func (c *dbosContext) Send(_ DBOSContext, destinationID string, message any, topic string) error {
 	// Serialize the message before sending
-	encodedMessageStr, err := serialize(c.serializer, message)
+	encodedMessageStr, err := c.serializer.Encode(message)
 	if err != nil {
 		return fmt.Errorf("failed to serialize message: %w", err)
 	}
@@ -1483,7 +1483,7 @@ func Recv[T any](ctx DBOSContext, topic string, timeout time.Duration) (T, error
 
 func (c *dbosContext) SetEvent(_ DBOSContext, key string, message any) error {
 	// Serialize the event value before storing
-	encodedMessageStr, err := serialize(c.serializer, message)
+	encodedMessageStr, err := c.serializer.Encode(message)
 	if err != nil {
 		return fmt.Errorf("failed to serialize event value: %w", err)
 	}