Refactor/data processor

.env.example

@@ -186,6 +186,7 @@ CACHE_REDIS_CONN_MAX_LIFETIME=15m
 # Supported modes: embedded, queue
 TASK_RUNNER_MODE=embedded
 TASK_RUNNER_SHUTDOWN_TIMEOUT=30s
+TASK_RUNNER_MAX_EXECUTION_TIME=1h
 TASK_RUNNER_RECOVERY_INTERVAL=30s
 TASK_RUNNER_HEARTBEAT_INTERVAL=30s
 TASK_RUNNER_MAX_CONCURRENT_TASKS=5

ARCHITECTURE.md

@@ -117,7 +117,7 @@ Service ports are fixed (`3000/3100/3200/3300/4000`); container-to-container URL
 
 - **Modular monolith over microservices** — one database, one deploy unit set, hard module boundaries enforced by convention (service interfaces only, no shared repositories). Scale comes from splitting *binaries*, not codebases.
 - **Transactional outbox** for anything that must not be lost; plain pub/sub for anything that may be.
-- **Opinionated workflow model** — batch is the default: every node takes an array in and returns an array out, and references resolve per item (`$node.field`), by position (`$node[2].field`) or across the whole set (`$node[*].field`). What the canvas deliberately lacks is control-flow machinery — no loop construct, no sub-workflows, no expression language. Flows stay simple, readable, and predictable for non-developer users; that constraint is a feature.
+- **Opinionated workflow model** — batch is the default: every node takes an array in and returns an array out, and references resolve per item (`$node.field` or `$input.field` for whatever feeds the node), by position (`$node[2].field`) or across the whole set (`$node[*].field`). Branching is per item: a condition sends each item down the branch its own comparison chose, and an output with no items skips whatever hangs off it. What the canvas deliberately lacks is control-flow machinery — no loop construct, no sub-workflows, no expression language. Flows stay simple, readable, and predictable for non-developer users; that constraint is a feature.
 - **stdlib-first Go** — `log/slog`, `database/sql`, small focused packages in `go-packages` instead of frameworks.
 - **Runtime env for the UI** — one image per release, environment decided at container start.
 - **Scale-out infrastructure is opt-in** — Redis buys cross-instance coordination, which a fresh single-instance install does not need. Every Redis-backed subsystem ships an in-process provider and defaults to it; Compose only starts what is configured.

apps/platform-api/internal/config/task_runner.go

@@ -107,6 +107,7 @@ type TaskRunnerOptions struct {
 	Mode TaskRunnerMode `env:"MODE"`
 
 	ShutdownTimeout    time.Duration `env:"SHUTDOWN_TIMEOUT"`
+	MaxExecutionTime   time.Duration `env:"MAX_EXECUTION_TIME"`
 	RecoveryInterval   time.Duration `env:"RECOVERY_INTERVAL"`
 	HeartbeatInterval  time.Duration `env:"HEARTBEAT_INTERVAL"`
 	MaxConcurrentTasks int64         `env:"MAX_CONCURRENT_TASKS"`

apps/platform-api/internal/nodeengine/domain/executors/executor.go

@@ -9,6 +9,10 @@ type Executor struct {
 	Disabled bool   `json:"disabled,omitempty"`
 }
 
+type BranchingExecutor interface {
+	ExecuteBranches(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, map[string][]int, error)
+}
+
 type ExecutorManager interface {
 	GetID() string
 	GetDisabled() bool

apps/platform-api/internal/nodeengine/nodes/system/condition/branch.go

@@ -0,0 +1,6 @@
+package condition
+
+const (
+	BranchTrue  = "true"
+	BranchFalse = "false"
+)

apps/platform-api/internal/nodeengine/nodes/system/condition/executor.go

@@ -38,49 +38,70 @@ func NewConditionExecutor(
 }
 
 func (e *ConditionExecutor) ExecuteWithContext(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, error) {
+	outputs, _, err := e.ExecuteBranches(ctx, credentials, data)
+	return outputs, err
+}
+
+func (e *ConditionExecutor) ExecuteBranches(ctx context.Context, credentials map[string]any, data []map[string]any) ([]map[string]any, map[string][]int, error) {
 	select {
 	case <-ctx.Done():
-		return nil, ctx.Err()
+		return nil, nil, ctx.Err()
 	default:
-		// TODO: only the first item is evaluated, so a node fed 10 items decides
-		// the whole flow from item 0. Routing items per branch needs the runner
-		// to carry item identity and keep outputs per edge; see the branching
-		// notes before changing this.
-		input, err := e.validator.Parse(data[0])
-		if err != nil {
-			return nil, err
+		outputs := make([]map[string]any, 0, len(data))
+		branches := map[string][]int{
+			BranchTrue:  make([]int, 0, len(data)),
+			BranchFalse: make([]int, 0, len(data)),
 		}
 
-		result := false
-		switch input.Operator {
-		case "eq":
-			result = input.LeftValue == input.RightValue
-		case "neq":
-			result = input.LeftValue != input.RightValue
-		case "gt":
-			result = cast.ToFloat(input.LeftValue) > cast.ToFloat(input.RightValue)
-		case "gte":
-			result = cast.ToFloat(input.LeftValue) >= cast.ToFloat(input.RightValue)
-		case "lt":
-			result = cast.ToFloat(input.LeftValue) < cast.ToFloat(input.RightValue)
-		case "lte":
-			result = cast.ToFloat(input.LeftValue) <= cast.ToFloat(input.RightValue)
-		case "contains":
-			result = strings.Contains(input.LeftValue, input.RightValue)
-		case "not_contains":
-			result = !strings.Contains(input.LeftValue, input.RightValue)
-		case "is_empty":
-			result = input.LeftValue == ""
-		case "is_not_empty":
-			result = input.LeftValue != ""
-		default:
-			return nil, ErrInvalidOperator
+		for index, item := range data {
+			result, err := e.evaluate(item)
+			if err != nil {
+				return nil, nil, err
+			}
+
+			outputs = append(outputs, map[string]any{"status": result})
+			if result {
+				branches[BranchTrue] = append(branches[BranchTrue], index)
+				continue
+			}
+			branches[BranchFalse] = append(branches[BranchFalse], index)
 		}
 
-		return []map[string]any{
-			{
-				"status": result,
-			},
-		}, nil
+		return outputs, branches, nil
+	}
+}
+
+func (e *ConditionExecutor) evaluate(item map[string]any) (bool, error) {
+	input, err := e.validator.Parse(item)
+	if err != nil {
+		return false, err
+	}
+
+	left := strings.TrimSpace(input.LeftValue)
+	right := strings.TrimSpace(input.RightValue)
+
+	switch input.Operator {
+	case "eq":
+		return left == right, nil
+	case "neq":
+		return left != right, nil
+	case "gt":
+		return cast.ToFloat(left) > cast.ToFloat(right), nil
+	case "gte":
+		return cast.ToFloat(left) >= cast.ToFloat(right), nil
+	case "lt":
+		return cast.ToFloat(left) < cast.ToFloat(right), nil
+	case "lte":
+		return cast.ToFloat(left) <= cast.ToFloat(right), nil
+	case "contains":
+		return strings.Contains(left, right), nil
+	case "not_contains":
+		return !strings.Contains(left, right), nil
+	case "is_empty":
+		return left == "", nil
+	case "is_not_empty":
+		return left != "", nil
+	default:
+		return false, ErrInvalidOperator
 	}
 }

apps/platform-api/internal/nodeengine/nodes/system/condition/node.go

@@ -25,8 +25,8 @@ func NewConditionNode(nodeID string) *ConditionNode {
 				{Key: "in"},
 			},
 			Outputs: []nodes.NodeHandle{
-				{Key: "true", Label: "True"},
-				{Key: "false", Label: "False"},
+				{Key: BranchTrue, Label: "True"},
+				{Key: BranchFalse, Label: "False"},
 			},
 			Categories:    []string{"System"},
 			SubCategories: []string{"System"},

apps/platform-api/internal/taskrunner/application/taskrunner/branching.go

@@ -0,0 +1,186 @@
+package taskrunner
+
+import (
+	"strings"
+
+	"github.com/blocknextai/go-packages/dag"
+	nodeEngineDomainExecutors "github.com/blocknextai/platform-api/internal/nodeengine/domain/executors"
+	taskRunnerDomainTask "github.com/blocknextai/platform-api/internal/taskrunner/domain/task"
+	"github.com/google/uuid"
+)
+
+const branchKeySeparator = "#"
+
+func BuildBranchKey(nodeKey, handle string) string {
+	var builder strings.Builder
+	builder.WriteString(nodeKey)
+	builder.WriteString(branchKeySeparator)
+	builder.WriteString(handle)
+	return builder.String()
+}
+
+type BranchView struct {
+	TaskID    uuid.UUID
+	Keys      map[string]string
+	Absolute  []int
+	InputKey  string
+	collected map[string][]any
+	matches   map[string][][]string
+}
+
+func (v *BranchView) Matches(pattern, template string, find func(string) [][]string) [][]string {
+	if v == nil {
+		return find(template)
+	}
+	if v.matches == nil {
+		v.matches = make(map[string][][]string)
+	}
+	key := pattern + "\x00" + template
+	if cached, ok := v.matches[key]; ok {
+		return cached
+	}
+	found := find(template)
+	v.matches[key] = found
+	return found
+}
+
+func (v *BranchView) Collected(key string) ([]any, bool) {
+	if v == nil || v.collected == nil {
+		return nil, false
+	}
+	values, ok := v.collected[key]
+	return values, ok
+}
+
+func (v *BranchView) Collect(key string, values []any) {
+	if v == nil {
+		return
+	}
+	if v.collected == nil {
+		v.collected = make(map[string][]any)
+	}
+	v.collected[key] = values
+}
+
+func (v *BranchView) Input() string {
+	if v == nil {
+		return ""
+	}
+	return v.InputKey
+}
+
+func (v *BranchView) Task() uuid.UUID {
+	if v == nil {
+		return uuid.Nil
+	}
+	return v.TaskID
+}
+
+func (v *BranchView) StoreKey(nodeKey string) string {
+	if v == nil {
+		return nodeKey
+	}
+	if key, ok := v.Keys[nodeKey]; ok {
+		return key
+	}
+	return nodeKey
+}
+
+func (v *BranchView) AbsoluteIndex(itemIndex int) int {
+	if v == nil || v.Absolute == nil {
+		return itemIndex
+	}
+	if itemIndex < 0 || itemIndex >= len(v.Absolute) {
+		return itemIndex
+	}
+	return v.Absolute[itemIndex]
+}
+
+func (v *BranchView) ItemIndex(nodeKey string, itemIndex int) int {
+	if v == nil {
+		return itemIndex
+	}
+	if _, isBranched := v.Keys[nodeKey]; isBranched {
+		return itemIndex
+	}
+	return v.AbsoluteIndex(itemIndex)
+}
+
+func edgeBetween(d *dag.DAG, parentID, childID string) *dag.Edge {
+	for _, edge := range d.NodeEdges(parentID) {
+		if edge.Target == childID {
+			return &edge
+		}
+	}
+	return nil
+}
+
+func (e *NodeExecutor) buildBranchView(task *taskRunnerDomainTask.Task, node *dag.Node) *BranchView {
+	if task.DAG == nil {
+		return nil
+	}
+
+	view := &BranchView{TaskID: task.ID, Keys: make(map[string]string)}
+	var builder strings.Builder
+
+	parents := task.DAG.NodeParents(node.ID)
+	for _, parentID := range parents {
+		parentNode := task.DAG.NodeByID(parentID)
+		if parentNode == nil {
+			continue
+		}
+
+		parentKey := BuildNodeKeyWithBuilder(&builder, parentNode.NodeID, parentNode.ID)
+		storeKey := parentKey
+
+		edge := edgeBetween(task.DAG, parentID, node.ID)
+		if edge != nil && strings.TrimSpace(edge.SourceHandle) != "" {
+			branchKey := BuildBranchKey(parentKey, edge.SourceHandle)
+			if _, ok := e.outputStore.Get(task.ID, branchKey); ok {
+				storeKey = branchKey
+				view.Keys[parentKey] = branchKey
+				if view.Absolute == nil {
+					if absolute, ok := e.outputStore.Indexes(task.ID, branchKey); ok {
+						view.Absolute = absolute
+					}
+				}
+			}
+		}
+
+		if len(parents) == 1 {
+			view.InputKey = storeKey
+		}
+	}
+
+	return view
+}
+
+func branchReaches(store *OutputStore, taskID uuid.UUID, d *dag.DAG, parent *dag.Node, childID string) (bool, error) {
+	edge := edgeBetween(d, parent.ID, childID)
+	if edge == nil || strings.TrimSpace(edge.SourceHandle) == "" {
+		return true, nil
+	}
+
+	executor, ok := nodeEngineDomainExecutors.GetExecutor(parent.NodeID)
+	if !ok {
+		return true, nil
+	}
+	if _, routes := executor.(nodeEngineDomainExecutors.BranchingExecutor); !routes {
+		return true, nil
+	}
+
+	items, ok := store.Get(taskID, BuildBranchKey(BuildNodeKey(parent.NodeID, parent.ID), edge.SourceHandle))
+	if !ok {
+		return false, ErrBranchOutputsUnavailable
+	}
+	return len(items) > 0, nil
+}
+
+func routesItems(node *dag.Node) bool {
+	executor, ok := nodeEngineDomainExecutors.GetExecutor(node.NodeID)
+	if !ok {
+		return false
+	}
+	_, routes := executor.(nodeEngineDomainExecutors.BranchingExecutor)
+	return routes
+}