mesh-controller/cmd/mesh-controller/try.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"net/url"
	"os"
	"regexp"
	"strings"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/link"
)

// A branch is tried before its pull request (novox/hq issue 686).
//
// A repository's store tests skip without a store, and no agent may start a database on its machine. So a
// builder pushed, a reviewer read, and only the pull request's merge check ran those tests for the first time:
// each failure there cost a round of fifteen to thirty minutes (mesh-controller #283, #310, #312).
//
// `try <owner>/<repository> <ref>` asks the build seat for the repository's own check of a pushed branch or
// commit, as the merge check runs it (internal/builder/check.go): the ref merged into the trunk's head, the
// trunk's merge-check.sh, each part in its toolchain's container, against a throwaway store and bus. It is a
// check and nothing more:
//
//   - **no pull request, so no status**: its outcome is never said as `checked`, which is what the forge's
//     holder sets statuses from (isTry, in Built);
//   - **nothing registered, nothing built**: it names no module of the graph, so the merge gate does not run and
//     no build is made for a merge (novox/hq ADR 0350's scratch builds are a pull request's check's alone);
//   - **a check in the build queue**: after deliveries (link.KindOf, novox/hq issue 558), and bounded by the
//     check's own bound (checkTimeout);
//   - **the code a pull request's check would run anyway**: only a repository of the mesh, whose every pull
//     request the build seat already checks, in containers that hold no runtime socket.
//
// A check outlasts a seat verb's one answer, so the ask answers its id at once and `try --id <id>` reads what
// came of it from the build's log, where the build seat says it whole (link.StepTried): holding its answer up
// to tryWait for a verdict, like `plans wait` (novox/hq issue 229). Nothing is kept beside the log.

// tryWait is how long reading a try holds its answer for the verdict. A variable so a test need not wait.
var tryWait = 20 * time.Second

// A ref as the forge is asked for it — a branch or a commit, never an option — and a `go test -run` pattern.
var (
	tryRef   = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]*$`)
	tryTests = regexp.MustCompile(`^[A-Za-z0-9_^$|().*+?/\[\]-]+$`)
)

// tryAnswer is what `try` answers, asking or reading.
type tryAnswer struct {
	ID string `json:"id"`
	// State is asked (just now), waiting (nothing said of it yet: in the build queue), running, or the verdict:
	// pass, warning (the repository declares no merge-check.sh), fail, or error (it could not run).
	State      string   `json:"state"`
	Said       string   `json:"said"`
	Repository string   `json:"repository,omitempty"`
	Ref        string   `json:"ref,omitempty"`
	Commit     string   `json:"commit,omitempty"`
	Base       string   `json:"base,omitempty"`
	Trunk      string   `json:"trunk,omitempty"`
	On         string   `json:"on,omitempty"`
	Took       string   `json:"took,omitempty"`
	Failed     []string `json:"failed,omitempty"`
	// Last is the last thing the build seat said of one still running.
	Last string `json:"last,omitempty"`
}

// isTry says a check's outcome is a try's: marked so, or — echoed by a build seat from before the mark — a
// check of no pull request and no group, which is nobody's status either.
func isTry(result link.BuildResult) bool {
	c := result.Checked
	return c != nil && (c.Try || (c.Number == 0 && c.Group == ""))
}

// asATry makes a check's ask a try's: the repository's own script alone, for no pull request.
func asATry(request link.BuildRequest, tests string, storeOnly bool) link.BuildRequest {
	c := *request.Check
	c.Number, c.Try, c.Tests, c.StoreOnly = 0, true, tests, storeOnly
	c.Paths, c.Modules, c.New, c.Dependents, c.Manifests, c.Builds = nil, nil, nil, nil, nil, nil
	c.Judge, c.Plan, c.Group, c.Members = "", nil, "", nil
	request.Check = &c
	return request
}

// maxTries is how many tries may wait or run at once, mesh-wide: each holds a build machine for minutes.
const maxTries = 4

// triesIn reads the build queue for a new try of one commit: the id of a try of that commit already waiting or
// running — the answer to a second ask — and whether there is room for another.
func triesIn(q link.Queue, owner, repo, commit string) (same string, room bool) {
	n := 0
	for _, a := range q.Asks {
		if !link.IsTry(a.Request) || (a.State != link.AskWaiting && a.State != link.AskInFlight) {
			continue
		}
		n++
		c := a.Request.Check
		if same == "" && strings.EqualFold(c.Owner, owner) && strings.EqualFold(c.Repo, repo) && a.Request.Ref == commit {
			same = a.ID
		}
	}
	return same, n < maxTries
}

// forgeCommit asks the forge, through its module's tools on the bus, which commit a branch or commit of a
// repository is, and the repository's default branch. A variable so a test needs no forge.
var forgeCommit = func(ctx context.Context, owner, repo, ref string) (commit, trunk string, err error) {
	js, err := aBus()
	if err != nil {
		return "", "", err
	}
	defer js.Close()
	bus := link.OverNATS{Conn: js.Conn()}
	ask := func(path string, into any) error {
		raw, _ := json.Marshal(map[string]any{"path": path})
		answer, err := link.Ask(ctx, bus, "gitea", "gitea_api", raw, 30*time.Second)
		if err != nil {
			return err
		}
		if answer.Error != "" {
			return errors.New(answer.Error)
		}
		return json.Unmarshal(answer.Result, into)
	}
	var commits struct {
		Result []struct {
			SHA string `json:"sha"`
		} `json:"result"`
	}
	if err := ask("/repos/"+owner+"/"+repo+"/commits?limit=1&stat=false&verification=false&files=false&sha="+
		url.QueryEscape(ref), &commits); err != nil {
		return "", "", err
	}
	if len(commits.Result) == 0 || commits.Result[0].SHA == "" {
		return "", "", errors.New("the forge named no commit for it")
	}
	var found struct {
		Result struct {
			DefaultBranch string `json:"default_branch"`
		} `json:"result"`
	}
	if err := ask("/repos/"+owner+"/"+repo, &found); err != nil {
		return "", "", err
	}
	return commits.Result[0].SHA, found.Result.DefaultBranch, nil
}

// tryCommand is `try <owner>/<repository> <ref> [--base <branch>] [--tests <pattern>] [--store-only]`, and
// `try --id <id>` to read what came of one.
func tryCommand(ctx context.Context, args []string) error {
	set := flag.NewFlagSet("try", flag.ContinueOnError)
	base := set.String("base", "", "the branch it is merged into (the repository's default branch when none)")
	tests := set.String("tests", "", "only the tests this `go test -run` pattern names")
	storeOnly := set.Bool("store-only", false, "only the repository's store tests")
	id := set.String("id", "", "a try's id: what came of it")
	positionals, err := parseAround(set, args)
	if err != nil {
		return err
	}
	usage := errors.New("try <owner>/<repository> <branch or commit> [--base <branch>] [--tests <pattern>] " +
		"[--store-only], or try --id <id>")
	if *id != "" {
		if len(positionals) != 0 {
			return usage
		}
		// Before it is a token of a bus subject: an id, never a pattern.
		if !link.SafeBuildID(*id) {
			return fmt.Errorf("%q is not a try's id", *id)
		}
		return tryRead(ctx, *id)
	}
	if len(positionals) != 2 {
		return usage
	}
	owner, repo, isPath := strings.Cut(positionals[0], "/")
	ref := positionals[1]
	switch {
	case !isPath || !tryRef.MatchString(owner) || !tryRef.MatchString(repo) || strings.Contains(repo, "/"):
		return fmt.Errorf("%q is not a repository on the forge, owner/name; nothing was asked", positionals[0])
	case !tryRef.MatchString(ref):
		return fmt.Errorf("%q is not a branch or a commit; nothing was asked", ref)
	case *base != "" && !tryRef.MatchString(*base):
		return fmt.Errorf("%q is not a branch; nothing was asked", *base)
	case *tests != "" && !tryTests.MatchString(*tests):
		return fmt.Errorf("%q is not a `go test -run` pattern of test names; nothing was asked", *tests)
	}
	// **An unknown ref is refused here, loudly** — not sent to a machine to fail at its clone minutes later.
	commit, trunk, err := forgeCommit(ctx, owner, repo, ref)
	if err != nil {
		return fmt.Errorf("the forge holds no branch or commit %s in %s/%s — is it pushed? — so nothing was asked: %w",
			ref, owner, repo, err)
	}
	if *base != "" {
		trunk = *base
	}
	if trunk == "" {
		return fmt.Errorf("the forge named no default branch for %s/%s, and no --base says what %s is merged into; "+
			"nothing was asked", owner, repo, ref)
	}

	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	entries, read, edges, err := theGraph(ctx, open.inventory)
	if err != nil {
		return err
	}
	p := link.PullUpdated{Owner: owner, Repo: repo, Base: trunk, Commit: commit}
	// Whose repository it is, as a pull request's check reckons it: only the mesh's is tried, the code the
	// build seat runs for every pull request already.
	mesh := pullScope(p, entries, read, edges)
	if !mesh.Mesh {
		return fmt.Errorf("%s/%s is no repository of the mesh — no module is built from it and its owner is not "+
			"the core's — and only the mesh's are tried on the build seat; nothing was asked", owner, repo)
	}
	request, err := checkRequestFor(ctx, open, p, checkScope{Mesh: true, From: mesh.From}, entries)
	if err != nil {
		return err
	}
	request = asATry(request, *tests, *storeOnly)
	seat := buildSeatHeld(ctx)
	// **Bounded in number** (novox/hq issue 686): one try of a commit at a time, and maxTries in all.
	js, err := dialTheBus()
	if err != nil {
		return err
	}
	queue, err := link.ReadQueue(ctx, js, seat)
	js.Close()
	if err != nil {
		return err
	}
	same, room := triesIn(queue, owner, repo, commit)
	if same != "" {
		return printTry(tryAnswer{ID: same, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
			On: seat, Said: fmt.Sprintf("a try of %s/%s at %.12s already waits or runs, as %s: nothing new was asked. "+
				"Read it with try id=%s", owner, repo, commit, same, same)})
	}
	if !room {
		return fmt.Errorf("%d tries already wait or run on the build seat, the most at once (`queue` lists them); "+
			"nothing was asked — read one of yours, or ask again when one has ended", maxTries)
	}
	ask, err := askOverOn(seat)
	if err != nil {
		return err
	}
	defer ask.Close()
	if err := ask.Ask(ctx, request); err != nil {
		return err
	}
	return printTry(tryAnswer{ID: request.ID, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
		Base: trunk, On: seat, Said: fmt.Sprintf("asked %s to try %s/%s at %.12s (%s) merged into %s: its own "+
			"merge-check.sh, as a check in the build queue, bounded at %s; no status is set and nothing is registered. "+
			"Read it with try id=%s — a full run of this kind takes minutes", seat, owner, repo, commit, ref, trunk,
			checkTimeout, request.ID)})
}

// tryRead answers what came of a try, holding the answer up to tryWait for its verdict.
func tryRead(ctx context.Context, id string) error {
	until := time.Now().Add(tryWait)
	for {
		lines, err := buildLines(ctx, id)
		if err != nil {
			return err
		}
		a := tryAnswerOf(id, lines)
		if (a.State != "waiting" && a.State != "running") || !time.Now().Add(2*time.Second).Before(until) {
			return printTry(a)
		}
		if err := sleepOrDone(ctx, 2*time.Second); err != nil {
			return err
		}
	}
}

// tryAnswerOf reads a try's answer from what the build seat said of it: its outcome under StepTried; `failed`
// for one that could not run — an error, never a pass; else how far it is.
func tryAnswerOf(id string, lines []link.BuildLine) tryAnswer {
	a := tryAnswer{ID: id, State: "waiting", Said: "nothing is said of " + id + " yet: it waits in the build queue " +
		"(`queue` says where), or no try has that id; ask again"}
	for _, l := range lines {
		switch l.Step {
		case link.StepTried:
			var t link.Tried
			if err := json.Unmarshal([]byte(l.Message), &t); err != nil || t.Verdict == "" {
				return tryAnswer{ID: id, State: "error", Said: "the build seat's word on it cannot be read: " + l.Message}
			}
			return tryAnswer{ID: id, State: t.Verdict, Said: t.Summary, Took: t.Took, Commit: t.Head, Trunk: t.Trunk,
				Failed: t.Failed}
		case "failed":
			return tryAnswer{ID: id, State: "error", Said: "it could not run: " + l.Message}
		case "checked":
			// A check that is no try, or a build seat from before the verb: its verdict is not a try's.
			return tryAnswer{ID: id, State: "error", Said: "the build seat checked it and said no try's outcome — " +
				"it predates the verb (novox/hq issue 686), or " + id + " is no try: " + l.Message + "; `builds` with " +
				"the id says every line"}
		}
		a.State, a.Last = "running", l.Message
		a.Said = "still running; ask again. `builds` with the id says every line so far"
	}
	return a
}

func printTry(a tryAnswer) error {
	fmt.Fprintln(os.Stderr, a.Said)
	for _, l := range a.Failed {
		fmt.Fprintln(os.Stderr, "    "+l)
	}
	out, err := json.MarshalIndent(a, "", "  ")
	if err != nil {
		return err
	}
	fmt.Println(string(out))
	return nil
}

// buildLines is everything the build seat said of one build, as the bus keeps it, in order.
func buildLines(ctx context.Context, id string) ([]link.BuildLine, error) {
	js, err := aBus()
	if err != nil {
		return nil, fmt.Errorf("cannot reach the bus to read a build's log: %w", err)
	}
	defer js.Close()
	subject := link.BuildLogOf("*", id)
	sub, err := js.Context().PullSubscribe(subject, "",
		nats.BindStream(broker.EventsStream), nats.DeliverAll(), nats.AckNone())
	if err != nil {
		return nil, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	var lines []link.BuildLine
	for {
		batch, err := sub.Fetch(200, nats.MaxWait(2*time.Second))
		if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
			return nil, fmt.Errorf("reading a build's log: %w", err)
		}
		for _, msg := range batch {
			var line link.BuildLine
			if json.Unmarshal(msg.Data, &line) == nil {
				lines = append(lines, line)
			}
		}
		if len(batch) < 200 {
			return lines, nil
		}
	}
}