fn_registry/apps/pipeline_launcher/views/runner.go

package views

import (
	"bytes"
	"fmt"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"time"

	ops "fn-registry/fn_operations"
	"fn-registry/registry"
)

// PipelineFlag describes a CLI flag parsed from -help output.
type PipelineFlag struct {
	Name     string // e.g. "project"
	Type     string // e.g. "string"
	Desc     string // description text
	Default  string // default value, empty if none
	Required bool   // true if no default
}

var flagLineRe = regexp.MustCompile(`^\s+-(\S+)\s+(\S+)$`)
var defaultRe = regexp.MustCompile(`\(default "(.*)"\)`)

// GetPipelineFlags runs `go run . -help` and parses the flag output.
func GetPipelineFlags(fn *registry.Function, registryRoot string) []PipelineFlag {
	absPath := filepath.Join(registryRoot, fn.FilePath)
	dir := filepath.Dir(absPath)

	cmd := exec.Command("go", "run", ".", "-help")
	cmd.Dir = dir
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	cmd.Run() // -help exits with code 2, ignore error

	return parseFlags(stderr.String())
}

func parseFlags(output string) []PipelineFlag {
	var flags []PipelineFlag
	lines := strings.Split(output, "\n")

	for i := 0; i < len(lines); i++ {
		m := flagLineRe.FindStringSubmatch(lines[i])
		if m == nil {
			continue
		}
		f := PipelineFlag{Name: m[1], Type: m[2]}

		// Next line is the description
		if i+1 < len(lines) {
			desc := strings.TrimSpace(lines[i+1])
			if dm := defaultRe.FindStringSubmatch(desc); dm != nil {
				f.Default = dm[1]
				f.Desc = strings.TrimSpace(defaultRe.ReplaceAllString(desc, ""))
			} else {
				f.Desc = desc
			}
			i++
		}

		f.Required = f.Default == "" && !strings.Contains(strings.ToLower(f.Desc), "opcional")
		flags = append(flags, f)
	}
	return flags
}

// RunResult holds the outcome of a pipeline execution.
type RunResult struct {
	Stdout     string
	Stderr     string
	ExecID     string
	PipelineID string
	Status     ops.ExecutionStatus
	DurationMs int64
	Err        error
}

// RunPipeline executes a pipeline as a subprocess and records the execution.
func RunPipeline(fn *registry.Function, registryRoot string, opsDB *ops.DB, args []string) RunResult {
	absPath := filepath.Join(registryRoot, fn.FilePath)
	dir := filepath.Dir(absPath)

	startedAt := time.Now().UTC()

	cmdArgs := append([]string{"run", "."}, args...)
	cmd := exec.Command("go", cmdArgs...)
	cmd.Dir = dir

	var stdout, stderr bytes.Buffer
	cmd.Stdout = &stdout
	cmd.Stderr = &stderr

	err := cmd.Run()
	endedAt := time.Now().UTC()

	status := ops.ExecSuccess
	var execErr string
	if err != nil {
		status = ops.ExecFailure
		execErr = err.Error()
		if stderr.Len() > 0 {
			execErr = stderr.String()
		}
	}

	execID := fmt.Sprintf("exec_%d", time.Now().UnixNano())
	durationMs := endedAt.Sub(startedAt).Milliseconds()

	execution := &ops.Execution{
		ID:         execID,
		PipelineID: fn.ID,
		Status:     status,
		StartedAt:  startedAt,
		EndedAt:    &endedAt,
		DurationMs: &durationMs,
		Error:      execErr,
		CreatedAt:  time.Now().UTC(),
	}

	insertErr := ops.InsertExecutionSafe(opsDB, execution)
	if insertErr != nil {
		return RunResult{
			Stdout:     stdout.String(),
			Stderr:     stderr.String(),
			ExecID:     execID,
			PipelineID: fn.ID,
			Status:     status,
			DurationMs: durationMs,
			Err:        fmt.Errorf("pipeline ran but failed to record: %w", insertErr),
		}
	}

	return RunResult{
		Stdout:     stdout.String(),
		Stderr:     stderr.String(),
		ExecID:     execID,
		PipelineID: fn.ID,
		Status:     status,
		DurationMs: durationMs,
		Err:        err,
	}
}