Method InternalRun

pocketflow.go:822–859

InternalRun executes the BatchFlow lifecycle: PrepBatch, Exec(orchestrate per batch), PostBatch.

(ctx *PfContext)

// InternalRun executes the BatchFlow lifecycle: PrepBatch, Exec(orchestrate per batch), PostBatch.
func (bf *BatchFlow) InternalRun(ctx *PfContext) (string, error) {
	// 1. Run BatchFlow's Prep phase (PrepBatchFunc)
	// Should return []*pfContext
	prepBatchResultAny, err := bf.Prep(ctx)
	if err != nil {
		return "", newPocketFlowError(fmt.Sprintf("PrepBatch phase failed for BatchFlow %T", bf), err)
	}

	batchParamsList, ok := prepBatchResultAny.([]map[string]any)
	if prepBatchResultAny != nil && !ok {
		return "", newPocketFlowError(fmt.Sprintf("Internal error: PrepBatch phase in BatchFlow %T did not return []*pfContext (%T)", bf, prepBatchResultAny), nil)
	}
	if batchParamsList == nil {
		batchParamsList = []map[string]any{}
	}

	// 2. Run the orchestration for each item in batchParamsList
	for i, batchParams := range batchParamsList {
		// Run the embedded Flow's orchestration logic for each parameter set.
		// Pass the *original* shared context and current batchParams.
		_, err := bf.Flow.orchestrate(ctx, batchParams)
		if err != nil {
			// If one batch run fails, fail the whole BatchFlow execution
			return "", newPocketFlowError(fmt.Sprintf("Orchestration failed for batch item %d in %T", i, bf), err)
		}
		// Result (lastAction) of individual orchestrate runs is ignored here; side effects matter.
	}

	// 3. Run BatchFlow's Post phase (PostBatchFunc)
	// The result of the "Exec" phase semantically is the list itself.
	execResult := batchParamsList
	finalAction, err := bf.Post(ctx, prepBatchResultAny, execResult)
	if err != nil {
		return "", newPocketFlowError(fmt.Sprintf("PostBatch phase failed for BatchFlow %T", bf), err)
	}

	return finalAction, nil
}