helios/pkg/job/job.go

package job

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"time"

	"git.bazzel.dev/bmallen/helios/pkg/state"
	"git.bazzel.dev/bmallen/helios/pkg/symbols"
	"github.com/go-git/go-git/v6"
	"github.com/go-git/go-git/v6/plumbing"
	"github.com/go-git/go-git/v6/plumbing/transport/http"
	"github.com/otiai10/copy"
	"github.com/sirupsen/logrus"
	"github.com/traefik/yaegi/interp"
	"github.com/traefik/yaegi/stdlib"
	"golang.org/x/mod/modfile"
	"gopkg.in/yaml.v3"
)

type Job struct {
	state            *state.State
	sourceDir        string
	files            []string
	source           string
	ref              string
	user             string
	password         string
	globals          map[string]any
	timeout          time.Duration
	autoimport       bool
	autocleanup      bool
	module           string
	errors           []error
	log              *logrus.Logger
	logs             []string
	workspaceCreated bool
	workspacedir     string
	i                *interp.Interpreter
}

type BlankFormatter struct{}

// Format renders a log entry to only the message part.
func (f *BlankFormatter) Format(entry *logrus.Entry) ([]byte, error) {
	// Return the message string as a byte slice, adding a newline character.
	return []byte(entry.Message + "\n"), nil
}

func New() *Job {
	j := &Job{
		state:        state.New(),
		timeout:      10 * time.Second,
		log:          logrus.New(),
		globals:      map[string]any{},
		workspacedir: "workspace",
	}

	j.log.Formatter = &BlankFormatter{}
	j.log.AddHook(j)

	return j
}

func NewWithState(state *state.State) *Job {
	j := New()
	j.state = state
	return j
}

func (j *Job) Levels() []logrus.Level {
	return logrus.AllLevels
}
func (j *Job) Fire(e *logrus.Entry) error {
	// j.logs = append(j.logs, fmt.Sprintf("%s: %s", e.Level.String(), e.Message))
	j.logs = append(j.logs, e.Message)
	return nil
}

func (j *Job) error(err error) *Job {
	if err != nil {
		j.log.Error(err)

		if j.errors == nil {
			j.errors = []error{}
		}

		j.errors = append(j.errors, err)
	}

	return j
}

func (j *Job) SourceDir(dir string) *Job {
	j.sourceDir = dir
	return j
}

func (j *Job) AutoCleanup() *Job {
	j.autocleanup = true
	return j
}

func (j *Job) Module(module string) *Job {
	j.module = module
	return j
}

func (j *Job) Source(url, ref string) *Job {
	if j.source != "" {
		return j.error(errors.New("source has already been set"))
	}

	j.source = url
	j.ref = url

	err := os.MkdirAll(filepath.Join(j.workspacedir, j.ID()), 0770)
	if err != nil {
		return j.error(err)
	}

	j.log.Info("Cloning...")
	_, err = git.PlainClone(
		filepath.Join(j.workspacedir, j.ID(), "repo"),
		&git.CloneOptions{
			URL:           j.source,
			ReferenceName: plumbing.ReferenceName(ref),
			Progress:      j.log.WriterLevel(logrus.DebugLevel),
			Auth: &http.BasicAuth{
				Username: j.user, // yes, this can be anything except an empty string
				Password: j.password,
			},
		})

	j.error(err)
	j.sourceDir = filepath.Join(j.workspacedir, j.ID(), "repo")

	return j
}

func (j *Job) User(user string) *Job {
	j.user = user
	return j
}

func (j *Job) Password(password string) *Job {
	j.password = password
	return j
}

func (j *Job) Timeout(timeout time.Duration) *Job {
	j.timeout = timeout
	return j
}

func (j *Job) AutoImport() *Job {
	j.autoimport = true

	return j
}

func (j *Job) Run(f string) *Job {
	if j.sourceDir == "" {
		return j.error(errors.New("must provide source"))
	}

	start := time.Now()
	workdir := filepath.Join(j.workspacedir, j.state.ID(), "src", j.module)

	if !j.workspaceCreated {
		if j.module == "" {
			modfiledata, err := os.ReadFile(filepath.Join(j.sourceDir, "go.mod"))
			if err != nil {
				j.module = "pipeline"
			} else {
				fff, err := modfile.Parse("go.mod", modfiledata, nil)
				if err != nil {
					return j.error(err)
				}

				// for _, r := range fff.Require {
				// 	j.log.Info("  ", r.Mod.String())
				// }

				j.module = fff.Module.Mod.Path
			}
		}

		j.log.Info("Creating workspace...")
		err := os.MkdirAll(filepath.Join(j.workspacedir, j.state.ID(), "src"), 0770)
		if err != nil {
			return j.error(err)
		}

		workdir = filepath.Join(j.workspacedir, j.state.ID(), "src", j.module)

		err = copy.Copy(j.sourceDir, workdir)
		if err != nil {
			return j.error(err)
		}

		j.workspaceCreated = true
	}

	if j.autocleanup {
		defer func() {

		}()
	}

	j.log.Info("Setting up new interpreter...")
	workspace, err := filepath.Abs(filepath.Join(j.workspacedir, j.state.ID()))
	if err != nil {
		return j.error(err)
	}

	if j.autocleanup {
		defer func() {
			err := os.RemoveAll(workspace)
			if err != nil {
				j.error(err)
			}
		}()
	}

	j.i = interp.New(interp.Options{
		GoPath:       workspace,
		Unrestricted: true,
		Stdin:        nil,
		Stdout:       j.log.WriterLevel(logrus.InfoLevel),
		Stderr:       j.log.WriterLevel(logrus.WarnLevel),
	})

	err = j.i.Use(stdlib.Symbols)
	if err != nil {
		return j.error(err)
	}

	err = j.i.Use(symbols.Symbols)
	if err != nil {
		return j.error(err)
	}

	err = j.i.Use(j.state.Symbols())
	if err != nil {
		return j.error(err)
	}

	if j.autoimport {
		j.i.ImportUsed()
	}

	j.log.Info("Starting job...")
	err = os.Chdir(workdir)
	if err != nil {
		return j.error(err)
	}

	j.files = append(j.files, f)

	if strings.HasSuffix(f, ".go") {
		data, err := os.ReadFile(f)
		if err != nil {
			return j.error(err)
		}

		ctx, _ := context.WithTimeout(context.Background(), j.timeout)
		_, err = j.i.EvalWithContext(ctx, string(data))
		if err != nil {
			j.error(err)
		}
	} else {
		ctx, _ := context.WithTimeout(context.Background(), j.timeout)
		_, err := j.i.EvalPathWithContext(ctx, filepath.Join(j.module, f))
		if err != nil {
			j.error(err)
		}
	}
	// for p, vv := range j.i.Symbols("pipeline") {
	// 	for k, v := range vv {
	// 		fmt.Println(p, k, v.Type(), v)
	// 		if v.Type() == reflect.TypeOf(func() *helios.Workflow { return nil }) {
	// 			fmt.Println(p, k)
	// 			_, err := j.i.Eval(`import "log"`)
	// 			if err != nil {
	// 				j.error(err)
	// 			}
	// 			fmt.Println(vv)
	// 			vv, err := j.i.Eval("pipeline."+k + "().Logger(log.Writer()).Run()")
	// 			if err != nil {
	// 				j.error(err)
	// 			}
	// 			fmt.Println(vv)
	// 		}
	// 	}
	// }

	for k, v := range j.i.Globals() {
		if v.CanInterface() {
			if v.Kind() == reflect.Func {
				continue
			}
			j.globals[k] = v.Interface()
		}
	}

	j.log.Info("Job finished in: ", time.Now().Sub(start))

	return j
}

func (j *Job) Globals() map[string]any {
	return j.globals
}

func (j *Job) State() *state.State {
	return j.state
}

func (j *Job) ID() string {
	return j.state.ID()
}

func (j *Job) Error() error {
	if j.errors != nil {
		err := []string{}

		if len(j.errors) > 1 {
			err = append(err, fmt.Sprintf("!!! %d: errors in stack !!!", len(j.errors)))
		}

		for i, e := range j.errors {
			err = append(err, fmt.Sprintf("%d: %s", i, e.Error()))
		}

		return errors.New(strings.Join(err, "\n"))
	}
	return nil
}

type Report struct {
	Module  string
	ID      string
	Source  string
	Files   []string
	Globals map[string]any
	State   map[string]any
	Errors  []string
	Logs    []string
}

func (j *Job) Report() *Job {
	report := Report{
		ID:      j.ID(),
		Source:  j.source,
		Files:   j.files,
		Module:  j.module,
		Globals: map[string]any{},
		State:   map[string]any{},
		Logs:    j.logs,
	}

	for k, v := range j.Globals() {
		report.Globals[k] = v
	}

	for _, v := range j.errors {
		report.Errors = append(report.Errors, v.Error())
	}

	j.state.Range(func(key string, value any) bool {
		report.State[key] = value
		return true
	})

	out, err := yaml.Marshal(report)
	if err != nil {
		j.log.Error(err)
	} else {
		fmt.Print(string(out))
	}

	return j
}