Implement workflow engine with YAML support and SSH session management

Author: yasithdev

examples/cs-bridge-workflow.yaml

@@ -0,0 +1,28 @@
+name: "cs-bridge-hpc-setup"
+
+steps:
+  - action: "vscode.create_session"
+    name: "Start SSH server"
+    params:
+      password: "test"
+    outputs:
+      bind_port: "ssh_port"
+
+  - action: "tunnel.devtunnel_create"
+    name: "Create devtunnel"
+    params:
+      tunnel_name: "linkspan-tunnel"
+      expiration: "1d"
+      ports:
+        - "{{.ssh_port}}"
+
+  - action: "tunnel.devtunnel_host"
+    name: "Host devtunnel"
+    params:
+      tunnel_name: "linkspan-tunnel"
+      create_token: true
+
+  - action: "shell.exec"
+    name: "Clone repo"
+    params:
+      command: "git clone https://github.com/org/repo.git /workspace"

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/fatedier/frp v0.67.0
 	github.com/gliderlabs/ssh v0.3.8
 	github.com/gorilla/mux v1.8.1
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (

internal/workflow/actions.go

@@ -0,0 +1,180 @@
+package workflow
+
+import (
+	"fmt"
+	"log"
+	"os/exec"
+	"strings"
+
+	tunnel "github.com/cyber-shuttle/linkspan/subsystems/tunnel"
+	vscode "github.com/cyber-shuttle/linkspan/subsystems/vscode"
+	utils "github.com/cyber-shuttle/linkspan/utils"
+)
+
+// registerBuiltinActions populates a Registry with all built-in action wrappers.
+func registerBuiltinActions(r *Registry) {
+	r.Register("vscode.create_session", actionVSCodeCreateSession)
+	r.Register("tunnel.devtunnel_create", actionDevTunnelCreate)
+	r.Register("tunnel.devtunnel_host", actionDevTunnelHost)
+	r.Register("tunnel.devtunnel_delete", actionDevTunnelDelete)
+	r.Register("tunnel.frp_proxy_create", actionFrpProxyCreate)
+	r.Register("shell.exec", actionShellExec)
+}
+
+// --- vscode.create_session ---
+
+func actionVSCodeCreateSession(params map[string]any) (*ActionResult, error) {
+	password, _ := params["password"].(string)
+
+	port, err := utils.GetAvailablePort()
+	if err != nil {
+		return nil, fmt.Errorf("get available port: %w", err)
+	}
+
+	sessionID := fmt.Sprintf("wf-%d", port)
+	addr := fmt.Sprintf(":%d", port)
+
+	vscode.StartSSHServerForVSCodeConnection(sessionID, addr, password)
+
+	result := ActionResult{
+		"session_id": sessionID,
+		"bind_port":  port,
+	}
+	return &result, nil
+}
+
+// --- tunnel.devtunnel_create ---
+
+func actionDevTunnelCreate(params map[string]any) (*ActionResult, error) {
+	tunnelName, _ := params["tunnel_name"].(string)
+	expiration, _ := params["expiration"].(string)
+	if expiration == "" {
+		expiration = "1d"
+	}
+
+	ports := toIntSlice(params["ports"])
+
+	info, err := tunnel.DevTunnelCreate(tunnelName, expiration, ports)
+	if err != nil {
+		return nil, err
+	}
+
+	result := ActionResult{
+		"tunnel_id":   info.TunnelID,
+		"tunnel_name": info.TunnelName,
+	}
+	return &result, nil
+}
+
+// --- tunnel.devtunnel_host ---
+
+func actionDevTunnelHost(params map[string]any) (*ActionResult, error) {
+	tunnelName, _ := params["tunnel_name"].(string)
+	createToken, _ := params["create_token"].(bool)
+
+	cmdID, conn, err := tunnel.DevTunnelHost(tunnelName, createToken)
+	if err != nil {
+		return nil, err
+	}
+
+	result := ActionResult{
+		"command_id":     cmdID,
+		"connection_url": conn.ConnectionURL,
+		"token":          conn.Token,
+	}
+	return &result, nil
+}
+
+// --- tunnel.devtunnel_delete ---
+
+func actionDevTunnelDelete(params map[string]any) (*ActionResult, error) {
+	tunnelName, _ := params["tunnel_name"].(string)
+	if err := tunnel.DevTunnelDelete(tunnelName); err != nil {
+		return nil, err
+	}
+	return &ActionResult{}, nil
+}
+
+// --- tunnel.frp_proxy_create ---
+
+func actionFrpProxyCreate(params map[string]any) (*ActionResult, error) {
+	tunnelName, _ := params["tunnel_name"].(string)
+	port := toInt(params["port"])
+	tunnelType, _ := params["tunnel_type"].(string)
+	tunnelSecret, _ := params["tunnel_secret"].(string)
+	discoveryHost, _ := params["discovery_host"].(string)
+	discoveryPort := toInt(params["discovery_port"])
+	discoveryToken, _ := params["discovery_token"].(string)
+
+	info, err := tunnel.FrpTunnelProxyCreate(
+		tunnelName, port, tunnelType, tunnelSecret,
+		discoveryHost, discoveryPort, discoveryToken,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	result := ActionResult{
+		"tunnel_name": info.TunnelName,
+		"tunnel_type": info.TunnelType,
+	}
+	return &result, nil
+}
+
+// --- shell.exec ---
+
+func actionShellExec(params map[string]any) (*ActionResult, error) {
+	command, _ := params["command"].(string)
+	if command == "" {
+		return nil, fmt.Errorf("shell.exec: command is required")
+	}
+
+	parts := strings.Fields(command)
+	cmd := exec.Command(parts[0], parts[1:]...)
+
+	output, err := cmd.CombinedOutput()
+	log.Printf("shell.exec: %s\n%s", command, string(output))
+	if err != nil {
+		return nil, fmt.Errorf("shell.exec %q: %w\n%s", command, err, string(output))
+	}
+
+	result := ActionResult{
+		"output": strings.TrimSpace(string(output)),
+	}
+	return &result, nil
+}
+
+// --- helpers ---
+
+// toIntSlice converts a param value to []int, handling YAML-decoded []any.
+func toIntSlice(v any) []int {
+	switch val := v.(type) {
+	case []any:
+		out := make([]int, 0, len(val))
+		for _, elem := range val {
+			out = append(out, toInt(elem))
+		}
+		return out
+	case []int:
+		return val
+	default:
+		return nil
+	}
+}
+
+// toInt converts a param value to int, handling YAML's default float64/int types.
+func toInt(v any) int {
+	switch val := v.(type) {
+	case int:
+		return val
+	case float64:
+		return int(val)
+	case string:
+		// Support template-resolved numeric strings.
+		var n int
+		fmt.Sscanf(val, "%d", &n)
+		return n
+	default:
+		return 0
+	}
+}

internal/workflow/registry.go

@@ -0,0 +1,34 @@
+package workflow
+
+import "fmt"
+
+// Registry maps action names (e.g. "vscode.create_session") to handler functions.
+type Registry struct {
+	actions map[string]ActionFunc
+}
+
+// NewRegistry creates an empty Registry.
+func NewRegistry() *Registry {
+	return &Registry{actions: make(map[string]ActionFunc)}
+}
+
+// Register adds an action handler. Panics on duplicate names to catch
+// programming errors early.
+func (r *Registry) Register(name string, fn ActionFunc) {
+	if _, exists := r.actions[name]; exists {
+		panic(fmt.Sprintf("workflow: duplicate action registered: %s", name))
+	}
+	r.actions[name] = fn
+}
+
+// Get returns the handler for the given action name, or nil if not found.
+func (r *Registry) Get(name string) ActionFunc {
+	return r.actions[name]
+}
+
+// DefaultRegistry returns a Registry pre-loaded with all built-in actions.
+func DefaultRegistry() *Registry {
+	r := NewRegistry()
+	registerBuiltinActions(r)
+	return r
+}