Board Engine

Board Engine

WebSocket Flow Execution - Real-time flow execution engine with pause, resume, and input capabilities.

---

Overview

The Board engine (backend/ion/board/) provides real-time flow execution through WebSocket communication. It enables:

• Live flow execution monitoring
• Pause/Resume/Stop controls
• External input during execution
• Progress and error streaming

---

Directory Structure

board/
├── engine.go           # ExecutionEngine implementation
├── hub.go              # WebSocket hub management
├── server.go           # WebSocket server handling
├── types.go            # Types and interfaces
├── company_dev_flow.go # Company development mode
├── company_live_flow.go# Company production mode
├── account_dev_flow.go # Account development mode
├── account_live_flow.go# Account production mode
├── company_session.go  # Company session management
└── account_session.go  # Account session management

---

Execution Engine

Creating an Engine

engine, err := board.NewExecutionEngine(
    sessionID,      // Unique session identifier
    executor,       // Executor implementation
    context,        // Execution context
    messageSender,  // WebSocket message callback
)

Lifecycle Methods

Method	Description
Start()	Begin execution in goroutine
Pause()	Pause execution
Resume()	Resume from pause
Stop()	Stop execution

// Start execution
err := engine.Start()

// Pause
err := engine.Pause()

// Resume
err := engine.Resume()

// Stop
engine.Stop()

---

Execution States

stateDiagram-v2
    [*] --> pending: NewEngine
    pending --> running: Start
    running --> paused: Pause
    paused --> running: Resume
    running --> waiting_input: WaitForInput
    waiting_input --> running: Input received
    running --> completed: Success
    running --> error: Error
    running --> stopped: Stop
    paused --> stopped: Stop
    waiting_input --> stopped: Stop

Status Values

Status	Description
pending	Engine created, waiting to start
running	Actively executing
paused	Execution paused
waiting_input	Waiting for external input
stopped	Manually stopped
completed	Successfully finished
error	Failed with error

---

Message Types

Output Messages

Type	Description
ready	Engine ready to execute
log	Log message
progress	Execution progress update
error	Error occurred
waiting	Waiting for input
complete	Execution completed
status_update	Status changed

Input Commands

Command	Description
start	Start execution
pause	Pause execution
resume	Resume execution
stop	Stop execution
input	Provide input data

---

WebSocket Message Format

Client to Server

{
    "command": "start",
    "payload": {
        "flowId": "flow-123",
        "nodeId": "n_1"
    }
}

Server to Client

{
    "command": "progress",
    "payload": {
        "node_id": "n_1",
        "status": "completed",
        "data": {...}
    }
}

---

Executor Interface

Custom executors implement this interface:

type Executor interface {
    // GetType returns executor type (e.g., "flow", "integration")
    GetType() string

    // Execute performs the main execution logic
    Execute(ctx ExecutionContext, control ExecutionControl) (*ExecutionResult, error)

    // Cleanup performs cleanup after execution
    Cleanup()
}

ExecutionControl Interface

Available to executors during execution:

type ExecutionControl interface {
    // Pause handling
    CheckPause() error

    // Input handling
    WaitForInput(timeout time.Duration, payload any) (any, error)

    // Message sending
    SendProgress(payload any)
    SendLog(message string)
    SendError(message string, errors ...string)

    // Context access
    GetContext() context.Context
    GetData() any
    SetData(data any)

    // Unit tracking
    ConsumeUnits(units uint64)
    GetUnitsConsumed() uint64

    // State access
    GetExecutionState() *ExecutionState
    GetExecutionContext() *ExecutionContext
}

---

Flow Executor Example

type FlowExecutor struct {
    flowId   string
    flow     *models.Flow
    startNode string
}

func (e *FlowExecutor) GetType() string {
    return "company_dev_flow"
}

func (e *FlowExecutor) Execute(ctx ExecutionContext, control ExecutionControl) (*ExecutionResult, error) {
    // Load flow
    err := flow.LoadFlow(e.flowId, *e.flow)
    if err != nil {
        return nil, err
    }

    // Run initial node
    response, err := flow.RunNode(e.flowId, e.startNode, "output")
    if err != nil {
        return nil, err
    }

    // Execution loop
    for response != nil && response.Queue != nil {
        // Check for pause
        if err := control.CheckPause(); err != nil {
            return nil, err
        }

        // Send progress
        control.SendProgress(map[string]any{
            "node_id": response.Node.ID,
            "type": response.Node.Type,
        })

        // Process node...
        outputs := processNode(response)

        // Push results
        result, _ := flow.PushNode(e.flowId, outputs, "ok", status.Success, "")

        // Continue
        response, _ = flow.ContinueNode(e.flowId)
    }

    return &ExecutionResult{
        Status: StatusCompleted,
        Message: "Flow completed",
    }, nil
}

func (e *FlowExecutor) Cleanup() {
    // Clean up flow workspace
}

---

Hub Management

The Hub manages active WebSocket sessions:

// Setup hub
board.SetupHub()

// Register session
board.CHub.Register <- session

// Unregister session
board.CHub.Unregister <- session

// Broadcast to all
board.CHub.Broadcast <- message

// Shutdown
board.CHub.Shutdown()

---

Input Waiting

Wait for external input during execution:

func (e *FlowExecutor) Execute(...) (*ExecutionResult, error) {
    // ...

    if needsInput {
        // Wait for input (0 = indefinite)
        input, err := control.WaitForInput(0, map[string]any{
            "message": "Please provide approval",
            "fields": []string{"approved", "reason"},
        })

        if err != nil {
            return nil, err
        }

        // Use input
        approved := input.(map[string]any)["approved"].(bool)
    }

    // ...
}

---

Related Documentation

• Execution Engine - Engine details
• Company Dev Flow - Development mode
• Company Live Flow - Production mode
• Sessions - Session management
• Hub - Hub management