mesh-sdk/go/buildstep/askrun.go

package buildstep

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"path"
	"slices"
	"sort"
	"strings"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// **One ask, one whole job** (novox/hq ADR 0525 §3). A build module runs a pulled ask with its own step tools' bodies,
// the same ones its request and reply tools run (ADR 0525 §8), in the order the job names: a build ask fetches, then
// builds or copies each artifact; a check ask fetches, digests the level, takes the kept verdicts and tests the rest,
// the packages whose tests the change touches first. The answers are the tools' own, whole, so the asker judges them as
// it judged the tools' answers.

// StepFunc is one step tool's body: its arguments in, its answer out.
type StepFunc func(ctx context.Context, args map[string]any) (any, error)

// AskSteps are the step tools a build module runs an ask with. Copy is the image module's alone; Digest the Go module's.
type AskSteps struct {
	Fetch, Build, Copy, Digest, Test StepFunc
	// Toolchain is the toolchain the module builds with now; a build ask naming another is declined (Declining). Nil checks nothing
	// (an image or an archive names none).
	Toolchain func(ctx context.Context) (delivery.ToolchainUsed, error)
	// Kept looks up a kept verdict by level and digest (the checker's kept bucket, delivery.KeptKey): whether one is
	// kept and how many tests it counted. Nil takes none.
	Kept func(ctx context.Context, level, digest string) (tests int, found bool, err error)
	// Release removes a workspace once the ask's steps on it ended (the module's release tool's body); nil leaves it to
	// the module's sweep.
	Release func(workspace string) error
	Logf    func(format string, args ...any)
	// Has says whether this node has a need a package's tests declare (delivery.AskNeeds, issue 890); a package needing
	// what it lacks is not tested here but named in the level's Unmet, for the asker to place. Nil has nothing.
	Has func(need string) bool
}

// RunAsk runs ask's job with steps and answers a delivery.BuildAnswer or a delivery.CheckAnswer.
func RunAsk(ctx context.Context, steps AskSteps, ask delivery.StepAsk) (any, error) {
	if ask.Build != nil {
		job := *ask.Build
		job.Fetch = withPackages(job.Fetch, ask.Packages)
		return steps.build(ctx, job)
	}
	if ask.Check != nil {
		job := *ask.Check
		job.Fetch = withPackages(job.Fetch, ask.Packages)
		return steps.check(ctx, job)
	}
	return nil, Refuse("the ask %s holds no job", ask.ID)
}

// withPackages is the fetch's arguments with the ask's package versions (novox/hq ADR 0573 §2), which the build module
// writes into the workspace it fetches; a copy, so the ask is left as it was. An ask with none, or a copy alone that
// fetches nothing, is left as it is.
func withPackages(fetch map[string]any, packages delivery.PackageVersions) map[string]any {
	if len(packages) == 0 || fetch == nil {
		return fetch
	}
	out := maps.Clone(fetch)
	out["packages"] = maps.Clone(packages)
	return out
}

func raw(v any) (json.RawMessage, error) {
	if r, ok := v.(json.RawMessage); ok {
		return r, nil
	}
	return json.Marshal(v)
}

// fetched is what an ask reads of a fetch's answer.
type fetched struct {
	Workspace string   `json:"workspace"`
	Merge     string   `json:"merge"`
	Conflicts []string `json:"conflicts"`
	Changed   []string `json:"changed"`
	Modules   []string `json:"modules"`
}

func (s AskSteps) fetch(ctx context.Context, args map[string]any) (json.RawMessage, fetched, error) {
	if s.Fetch == nil {
		return nil, fetched{}, Refuse("this build module fetches nothing")
	}
	if _, merges := args["base"]; merges {
		Phase(ctx, "fetch and merge")
	} else {
		Phase(ctx, "fetch")
	}
	out, err := s.Fetch(ctx, args)
	if err != nil {
		return nil, fetched{}, err
	}
	r, err := raw(out)
	if err != nil {
		return nil, fetched{}, err
	}
	var f fetched
	if err := json.Unmarshal(r, &f); err != nil || f.Workspace == "" {
		return r, f, fmt.Errorf("the fetch answered no workspace: %s", r)
	}
	return r, f, nil
}

// onToolchain is the toolchain this node builds with when the ask names one and this module reports its own, nil when
// either is absent. A node on another toolchain than the ask names answers a Declining: it is not on the agreed one and
// sits out, for a node that is to take the ask (novox/hq ADR 0584 clause 2).
func (s AskSteps) onToolchain(ctx context.Context, want *delivery.ToolchainUsed) (*delivery.ToolchainUsed, error) {
	if want == nil || s.Toolchain == nil {
		return nil, nil
	}
	tc, err := s.Toolchain(ctx)
	if err != nil {
		return nil, err
	}
	if tc.Version != want.Version || tc.Recipe != want.Recipe {
		return nil, &Declining{Lacks: []string{fmt.Sprintf("the agreed toolchain %s (recipe %s)", want.Version, want.Recipe)},
			Text: fmt.Sprintf("the ask is for %s (recipe %s) and this node builds with %s (recipe %s): it sits out, and the ask is left to a node on the agreed toolchain",
				want.Version, want.Recipe, tc.Version, tc.Recipe)}
	}
	return &tc, nil
}

func (s AskSteps) build(ctx context.Context, job delivery.BuildJob) (any, error) {
	answer := delivery.BuildAnswer{}
	tc, err := s.onToolchain(ctx, job.Toolchain)
	if err != nil {
		return nil, err
	}
	answer.Toolchain = tc
	var f fetched
	defer func() { s.release(f.Workspace) }()
	for _, a := range job.Artifacts {
		if a.From == "" && f.Workspace == "" {
			r, got, err := s.fetch(ctx, job.Fetch)
			if err != nil {
				return nil, err
			}
			if _, candidate := job.Fetch["base"]; candidate {
				// A candidate is built from its head merged onto the trunk; a head that does not merge leaves the tree at the
				// trunk, and building that would store the trunk's build under the candidate's fingerprint.
				if len(got.Conflicts) > 0 {
					return nil, fmt.Errorf("the head does not merge into the trunk: files in conflict: %s", strings.Join(got.Conflicts, ", "))
				}
				if got.Merge == "" {
					return nil, fmt.Errorf("the fetch with a base answered no merge: %s", r)
				}
			}
			f = got
		}
		var (
			out any
			err error
		)
		if a.From != "" {
			if s.Copy == nil {
				return nil, Refuse("this build module copies no artifact (%s)", a.Artifact)
			}
			Phase(ctx, "copy "+a.Artifact)
			out, err = s.Copy(ctx, map[string]any{"module": job.Module, "artifact": a.Artifact, "from": a.From})
		} else {
			if s.Build == nil {
				return nil, Refuse("this build module builds no artifact (%s)", a.Artifact)
			}
			args := map[string]any{"workspace": f.Workspace, "module": job.Path, "artifact": a.Artifact}
			if job.Publish {
				args["publish"] = true
			}
			Phase(ctx, "build "+a.Artifact)
			out, err = s.Build(ctx, args)
		}
		if err != nil {
			return nil, fmt.Errorf("%s: %w", a.Artifact, err)
		}
		r, err := raw(out)
		if err != nil {
			return nil, err
		}
		var built struct {
			Digest    string `json:"digest"`
			Reference string `json:"reference"`
		}
		if err := json.Unmarshal(r, &built); err != nil || built.Digest == "" {
			return nil, fmt.Errorf("the build of %s answered no digest: %s", a.Artifact, r)
		}
		answer.Artifacts = append(answer.Artifacts, delivery.AskBuilt{Artifact: a.Artifact, Digest: built.Digest, Reference: built.Reference})
	}
	return answer, nil
}

func (s AskSteps) check(ctx context.Context, job delivery.CheckJob) (any, error) {
	if _, err := s.onToolchain(ctx, job.Toolchain); err != nil {
		return nil, err
	}
	r, f, err := s.fetch(ctx, job.Fetch)
	if err != nil {
		return nil, err
	}
	defer s.release(f.Workspace)
	answer := delivery.CheckAnswer{Fetch: r}
	if len(f.Conflicts) > 0 {
		return answer, nil // the asker reads the conflict from the fetch's answer: a verdict, not an error
	}
	if s.Test == nil {
		return nil, Refuse("this build module tests nothing")
	}
	runner := ""
	if job.Runner != "" && (job.RunnerWhen == "" || changesUnder(f.Changed, job.RunnerWhen)) {
		runner = job.Runner
	}
	for _, level := range job.Levels {
		la := delivery.LevelAnswer{Level: level}
		passed, err := s.level(ctx, job, level, f, runner, &la)
		answer.Levels = append(answer.Levels, la)
		if err != nil {
			return nil, err
		}
		if !passed {
			return answer, nil // the first level that did not pass ends the check (ADR 0484 clause 13)
		}
	}
	return answer, nil
}

// level runs one level of a check ask and says whether it passed.
func (s AskSteps) level(ctx context.Context, job delivery.CheckJob, level string, f fetched, runner string, la *delivery.LevelAnswer) (bool, error) {
	withRunner := func(args map[string]any) map[string]any {
		if runner != "" {
			args["runner"] = runner
		}
		if len(job.NodeNames) > 0 {
			args[delivery.NodeNamesArg] = job.NodeNames // the names check's (ADR 0567 §12)
		}
		return args
	}
	test := func(args map[string]any) (bool, error) {
		what := "the whole level"
		if dir, _ := args["module"].(string); dir != "" {
			what = dir
		} else if pkgs, _ := args["packages"].([]string); len(pkgs) > 0 {
			what = fmt.Sprintf("%d packages", len(pkgs))
		}
		Phase(ctx, fmt.Sprintf("%s: tests of %s", level, what))
		out, err := s.Test(ctx, args)
		if err != nil {
			return false, err
		}
		r, err := raw(out)
		if err != nil {
			return false, err
		}
		la.Tests = append(la.Tests, r)
		var t struct {
			Passed bool `json:"passed"`
		}
		_ = json.Unmarshal(r, &t)
		return t.Passed, nil
	}
	// The packages (or Node directories) whose tests the change touches run first, and a failure among them ends the level
	// there, so a change's own failing test is answered without the rest of the suite (issue 1289; ADR 0444). Past them every
	// test step of the level runs, also after one that did not pass, so one check names every failing test of the rest of
	// the level and not only the first (issue 1921); the level passes only when each of its steps did.
	all := true
	if s.Digest == nil {
		// A module with no digest (Node) tests each module directory of the tree, those whose tests changed first.
		for _, priority := range []bool{true, false} {
			for _, dir := range f.Modules {
				if changesTestIn(f.Changed, dir) != priority {
					continue
				}
				passed, err := test(withRunner(map[string]any{"workspace": f.Workspace, "module": dir, "level": level}))
				if err != nil {
					return false, fmt.Errorf("the %s level of %s: %w", level, dir, err)
				}
				all = all && passed
			}
			if !all {
				return false, nil // the change's own tests failed: the others are not run (issue 1289)
			}
		}
		return all, nil
	}
	todo, whole, keepRunner := s.plan(ctx, job, level, f, runner, withRunner, la)
	if !keepRunner {
		runner = "" // an installed module that ignores the runner judges the level as it judges any other (ADR 0535 §2)
	}
	if whole && len(job.Packages) > 0 {
		todo, whole = job.Packages, false // a placed ask tests its packages alone, whatever the digest said (ADR 0552 §4)
	}
	for _, priority := range []bool{true, false} {
		args := withRunner(map[string]any{"workspace": f.Workspace, "level": level})
		if job.Run != "" {
			args["run"] = job.Run
		}
		if whole {
			if priority {
				continue // nothing to put first: the level runs whole after
			}
		} else {
			var run []string
			for _, p := range todo {
				if changesTestOf(f.Changed, p) == priority {
					run = append(run, p)
				}
			}
			if len(run) == 0 {
				continue
			}
			args["packages"] = run
		}
		passed, err := test(args)
		if err != nil {
			return false, fmt.Errorf("the %s level: %w", level, err)
		}
		all = all && passed // past the change's own packages the others still run (issue 1921)
		if priority && !all {
			return false, nil // the change's own tests failed: the others are not run (issue 1289)
		}
	}
	return all, nil
}

// release removes an ask's workspace, saying a failure: the module's sweep removes it later.
func (s AskSteps) release(workspace string) {
	if workspace == "" || s.Release == nil {
		return
	}
	if err := s.Release(workspace); err != nil && s.Logf != nil {
		s.Logf("the workspace %s is left to the sweep: %v", workspace, err)
	}
}

// plan digests the level and takes the kept verdicts: the packages left to test, or all when the level runs whole, and
// whether the runner stays. A try of some tests runs whole, digesting nothing (a partial run proves no package). Under the
// candidate's runner the digest is asked as the candidate: when the candidate answers it, every package runs and nothing
// is digested for keeping (ADR 0535 §4); when the installed module answers it (it does not take the runner), the level is
// planned as any other and the runner is dropped. The level runs whole and takes nothing when the digest names no package
// or does not cover the siblings the fetch placed (ADR 0444, L5).
func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string, f fetched, runner string, withRunner func(map[string]any) map[string]any, la *delivery.LevelAnswer) ([]string, bool, bool) {
	if job.Run != "" {
		return nil, true, true
	}
	Phase(ctx, level+": digest")
	out, err := s.Digest(ctx, withRunner(map[string]any{"workspace": f.Workspace, "level": level}))
	if err != nil {
		if s.Logf != nil {
			s.Logf("no digest of the %s level, so it runs whole: %v", level, err)
		}
		return nil, true, true
	}
	r, err := raw(out)
	if err != nil {
		return nil, true, true
	}
	var d delivery.LevelDigest // packages kept as a sorted list, never keyed by a package path (issue 1930)
	if err := json.Unmarshal(r, &d); err != nil {
		return nil, true, true
	}
	if runner != "" {
		if strings.HasPrefix(d.JudgedBy, "candidate ") {
			la.Judged = d.JudgedBy
			return nil, true, true // the candidate decides which verdicts would be taken: all run, nothing kept
		}
		runner = ""
	}
	if len(d.Packages) == 0 || !slices.Equal(d.Beside, besideOf(job.Fetch)) {
		return nil, true, runner != "" // a digest that names nothing or not the siblings keys no verdict: the level runs whole
	}
	la.Digest = &d
	only := map[string]bool{}
	for _, p := range job.Packages {
		only[p] = true
	}
	var todo []string
	for _, pd := range d.Packages {
		p := pd.Package
		if len(only) > 0 && !only[p] {
			continue // a placed ask's other packages were judged by the first ask (ADR 0552 §4)
		}
		if job.Kept && s.Kept != nil {
			tests, found, err := s.Kept(ctx, level, pd.Digest)
			if err != nil && s.Logf != nil {
				s.Logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, level, err)
			}
			if err == nil && found {
				la.Kept = append(la.Kept, delivery.KeptTaken{Package: p, Digest: pd.Digest, Tests: tests})
				continue
			}
		}
		if needs := d.NeedsOf(p); len(delivery.Lacks(needs, s.Has)) > 0 {
			// Its tests would only skip here (issue 890): it is left for a node that has what they need.
			la.Unmet = append(la.Unmet, delivery.PackageNeeds{Package: p, Needs: needs})
			if s.Logf != nil {
				s.Logf("the %s level of %s is left to a node with %s: this node lacks %s", level, p,
					strings.Join(needs, ", "), strings.Join(delivery.Lacks(needs, s.Has), ", "))
			}
			continue
		}
		todo = append(todo, p)
	}
	return todo, false, false
}

// besideOf are the siblings a fetch's arguments place, as a digest echoes them: `<owner/repo> <commit>`, sorted.
func besideOf(fetch map[string]any) []string {
	raw, err := json.Marshal(fetch["beside"])
	if err != nil {
		return nil
	}
	var refs []struct {
		Repository string `json:"repository"`
		Commit     string `json:"commit"`
	}
	_ = json.Unmarshal(raw, &refs)
	var out []string
	for _, r := range refs {
		out = append(out, r.Repository+" "+r.Commit)
	}
	sort.Strings(out)
	return out
}

// changesUnder says whether a path the merge changes is under prefix.
func changesUnder(changed []string, prefix string) bool {
	for _, p := range changed {
		if strings.HasPrefix(p, prefix) {
			return true
		}
	}
	return false
}

// isTestFile says whether a repository-relative path is a test file: Go's _test.go, and a TypeScript module's .test.
// and .spec. files and test directories.
func isTestFile(p string) bool {
	switch {
	case strings.HasSuffix(p, "_test.go"):
		return true
	case strings.Contains(p, ".test.") || strings.Contains(p, ".spec."):
		return true
	}
	for _, seg := range strings.Split(path.Dir(p), "/") {
		if seg == "__tests__" || seg == "test" || seg == "tests" {
			return true
		}
	}
	return false
}

// changesTestOf says whether the merge changes a test file of the Go package with the import path pkg: the file's
// directory and the import path agree on their last segments.
func changesTestOf(changed []string, pkg string) bool {
	for _, f := range changed {
		if !strings.HasSuffix(f, "_test.go") {
			continue
		}
		dir := path.Dir(f)
		if dir == pkg || strings.HasSuffix(pkg, "/"+dir) || strings.HasSuffix(dir, "/"+pkg) {
			return true
		}
	}
	return false
}

// changesTestIn says whether the merge changes a test file inside the directory of a Node module.
func changesTestIn(changed []string, dir string) bool {
	for _, f := range changed {
		if !isTestFile(f) || strings.HasSuffix(f, "_test.go") {
			continue
		}
		if dir == "." || dir == "" || strings.HasPrefix(f, strings.TrimSuffix(dir, "/")+"/") {
			return true
		}
	}
	return false
}