fn_registry/apps/pipeline_launcher/views/runner.go

package views

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

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

// 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) RunResult {
	absPath := filepath.Join(registryRoot, fn.FilePath)
	dir := filepath.Dir(absPath)

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

	cmd := exec.Command("go", "run", ".")
	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,
	}
}