mesh-delivery/modules/mesh-delivery-checker/internal/check/programs.go

package check

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"slices"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
	"git.novox.be/novox/mesh-delivery/sdk/buildstep"

	"mesh-delivery-checker/rules"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
)

// The checking program (novox/hq ADR 0644 §1, ADR 0656 §5; design 47's bootstrap delivery; issue 1921): every check has
// the program built from the group's mesh-delivery member head, else from mesh-delivery's trunk, and run in the build
// seat's sandbox through the seat node-build-go's ask `run`, with every member's head laid for it. It answers the heads'
// seat list first (`-seats`), which the manifest validator's and the composer's requests carry, and then the check's
// verdict of the facts the asking gathered (`-verdict`). The running checker asks, keeps and records; it judges nothing.

const (
	// InputsStateName is the checker's own state of run inputs and InputsBucket its bucket: every input by the sha256 of
	// its exact bytes, read by the checker alone. InputVerb is the seat verb a run ask's input names
	// (buildstep.RunInput.Verb), which the checker's tools serve from that state (Reader.Input): the Go build module asks it
	// as its manifest's invokes grants it and checks the digest (buildstep.RunInput.Read), never reading the state (ADR 0596
	// §2; the operator's answer on milestone 47/B0).
	InputsStateName = "inputs"
	InputsBucket    = CheckerModule + "_" + InputsStateName
	InputVerb       = meshdeliverycheckerclient.InputVerb
	// CheckingProgramDir and CheckingProgramPackage are where in mesh-delivery the checking program is built.
	CheckingProgramDir     = "modules/mesh-delivery-checker"
	CheckingProgramPackage = "mesh-delivery-checker/cmd/mesh-checking-program"
	// DefaultBeginBound is how long a run ask may wait for a build node to begin its step before it is put once more, and
	// after that once more before the check fails naming the job: a build module that does not know the job `run` drops
	// it unanswered (the lead's ruling on milestone 47/B0, 2026-11-06).
	DefaultBeginBound = 5 * time.Minute
)

// Inputs keeps a run's input under its digest, exactly its bytes.
type Inputs interface {
	Keep(ctx context.Context, digest string, body []byte) error
}

func (j *Examiner) beginBound() time.Duration {
	if j.BeginBound > 0 {
		return j.BeginBound
	}
	return DefaultBeginBound
}

// checkingProgram is where this check's checking program is built: the group's mesh-delivery member head, else
// mesh-delivery's trunk; false when neither is known.
func (st *state) checkingProgram() (buildstep.RunProgram, string, bool) {
	p := buildstep.RunProgram{Repository: DeliveryRepository, Dir: CheckingProgramDir, Package: CheckingProgramPackage}
	for _, pull := range st.c.Pulls {
		if pull.Repository == DeliveryRepository {
			p.Commit = pull.Head
			return p, "the group's mesh-delivery head " + pull.Head, true
		}
	}
	if st.j.Trunks != nil {
		p.Commit = st.j.Trunks.Heads()[DeliveryRepository]
	}
	return p, "mesh-delivery's trunk " + p.Commit, p.Commit != ""
}

// runHeads are the members' heads, as the program reads them.
func (st *state) runHeads() []buildstep.RunHead {
	out := make([]buildstep.RunHead, 0, len(st.c.Pulls))
	for _, p := range st.c.Pulls {
		out = append(out, buildstep.RunHead{Repository: p.Repository, Commit: p.Head})
	}
	return out
}

// runProgram has the checking program built and run with args over body, and answers its output and what judged; an
// error says why it did not answer: its input over its bound, no build node beginning the job, a step that did not
// pass, or a program that exited otherwise than 0.
func (st *state) runProgram(ctx context.Context, args []string, body []byte) (json.RawMessage, *meshdeliverychecker.VerdictsByGroupHeadsValueJudgedBy, error) {
	prog, at, ok := st.checkingProgram()
	if !ok {
		return nil, nil, fmt.Errorf("it cannot be built: the trunk head of %s is not known", DeliveryRepository)
	}
	if st.j.Inputs == nil {
		return nil, nil, fmt.Errorf("its input cannot be kept: the checker keeps no state %s", InputsStateName)
	}
	in, err := buildstep.RunInputOf(InputVerb, body)
	if err != nil {
		return nil, nil, err
	}
	if err := st.j.Inputs.Keep(ctx, in.Digest, body); err != nil {
		return nil, nil, fmt.Errorf("its input could not be kept in %s: %w", InputsStateName, err)
	}
	job := buildstep.RunJob{Program: prog, Heads: st.runHeads(), Input: in, Args: args, Reader: "the checking program at " + at}
	var ids []string
	for range 2 {
		a := buildstep.StepAsk{ID: buildstep.NewID(), Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: CheckerModule,
			At: st.j.now().UTC(), Packages: st.packagesOf(DeliveryRepository), Run: &job}
		ids = append(ids, a.ID)
		st.c.Progress.Set(fmt.Sprintf("the run ask %s of the checking program %s", a.ID, strings.Join(args, " ")), "")
		st.c.Asked(a.ID)
		if err := st.j.Asks.Put(ctx, a); err != nil {
			return nil, nil, fmt.Errorf("its run ask could not be put: %w", err)
		}
		waiting := ctx
		if watch := st.c.Watch; watch != nil {
			id := a.ID
			watch(id, nil) // a try names it at once (issue 1890)
			waiting = buildrun.WithStepSeen(ctx, func(s buildstep.Step) { watch(id, &s) })
		}
		step, begun, err := st.waitBegun(waiting, a)
		if !begun {
			continue
		}
		if err != nil {
			return nil, nil, fmt.Errorf("its run ask %s: %w", a.ID, err)
		}
		if step.State != buildstep.StepPassed {
			return nil, nil, fmt.Errorf("its run ask %s ended %s on %s: %s", a.ID, step.State, step.Node, step.Reason)
		}
		var answer buildstep.RunAnswer
		if err := json.Unmarshal(step.Result, &answer); err != nil {
			return nil, nil, fmt.Errorf("its run ask %s answered no run answer: %w", a.ID, err)
		}
		if answer.Exit != 0 {
			return nil, nil, fmt.Errorf("it exited %d on %s: %s", answer.Exit, step.Node, answer.Said)
		}
		return answer.Output, judgedByOf(prog, answer, job.Heads), nil
	}
	return nil, nil, fmt.Errorf("no build node began the job run of the asks %s within %s each: a build module that does not know the job run drops it",
		strings.Join(ids, " and "), st.j.beginBound())
}

// waitBegun waits for a's step, and stops waiting, begun false, when no build node began it within the begin bound: a
// bound on the wait's context, no timer of its own.
func (st *state) waitBegun(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, bool, error) {
	wait, cancel := context.WithCancel(ctx)
	defer cancel()
	begin, stop := context.WithTimeout(ctx, st.j.beginBound())
	defer stop()
	seen := make(chan struct{})
	var once sync.Once
	type ended struct {
		step buildstep.Step
		err  error
	}
	done := make(chan ended, 1)
	go func() {
		step, err := st.j.Asks.Wait(buildrun.WithStepSeen(wait, func(buildstep.Step) { once.Do(func() { close(seen) }) }), a)
		done <- ended{step, err}
	}()
	select {
	case e := <-done:
		return e.step, true, e.err
	case <-seen:
	case <-begin.Done():
		if ctx.Err() != nil {
			return buildstep.Step{}, true, ctx.Err()
		}
		return buildstep.Step{}, false, nil
	}
	e := <-done
	return e.step, true, e.err
}

// seatList asks the checking program for the heads' seat list, which the validator's and the composer's requests carry
// (ADR 0636 §4); the list and what it left out are the check's notes. A list it cannot make fails the check, naming why,
// and answers false: the check ends there, with no verdict asked of a program that could not answer its list (review
// round 1 of milestone 47/B0: a build node with no job run would hold the check two begin bounds more).
func (st *state) seatList(ctx context.Context) bool {
	if st.narrowTry() {
		return true
	}
	// The list the checker's reader holds from the seat mesh-catalog: a reader that has not read it acts on nothing new
	// (ADR 0636 §1), so the check ends here, said.
	var base []json.RawMessage
	if st.j.SeatList != nil {
		defs, err := st.j.SeatList.Definitions()
		if err != nil {
			st.fail(st.goModule(), "the change's manifests cannot be judged: the seat list is not read ("+err.Error()+")")
			return false
		}
		base = defs
	}
	// The program answers only the heads' definitions the list does not hold as they are: the list's digests ride in its
	// input, never the list, whose definitions are larger than a run's input or answer may be (review finding 3 of G).
	head := st.factsHead()
	if len(base) > 0 {
		head.SeatDigests = map[string]string{}
		for _, raw := range base {
			var d struct {
				Seat string `json:"seat"`
			}
			if json.Unmarshal(raw, &d) == nil && d.Seat != "" {
				head.SeatDigests[d.Seat] = rules.SeatDigest(raw)
			}
		}
	}
	body, err := json.Marshal(head)
	if err == nil {
		var out json.RawMessage
		if out, _, err = st.runProgram(ctx, []string{"-seats"}, body); err == nil {
			var a rules.SeatsAnswer
			if err = json.Unmarshal(out, &a); err == nil && a.V != rules.SeatsVersion {
				err = fmt.Errorf("it answered a seat list of version %d, not %d", a.V, rules.SeatsVersion)
			}
			if err == nil {
				st.c.Seats = overSeats(base, a.Seats)
				if len(a.Seats) > 0 {
					st.c.note(fmt.Sprintf("the heads carry %d seat definitions new or changed (and %d the seat list holds as they are), laid over the %d the seat list holds", len(a.Seats), a.Same, len(base)))
				}
				for _, l := range a.Left {
					st.c.note("the heads' seat list leaves out " + l.Path + ": " + l.Why)
				}
				return true
			}
		}
	}
	st.fail(st.goModule(), "the heads' seat list could not be made by the checking program: "+err.Error())
	return false
}

// overSeats is the list's definitions with each of the heads' put in place of the one of its seat, or added, sorted by
// seat: the checker reads of a definition only its seat's name.
func overSeats(list, heads []json.RawMessage) []json.RawMessage {
	by := map[string]json.RawMessage{}
	for _, l := range [][]json.RawMessage{list, heads} {
		for _, raw := range l {
			var d struct {
				Seat string `json:"seat"`
			}
			if json.Unmarshal(raw, &d) == nil && d.Seat != "" {
				by[d.Seat] = raw
			}
		}
	}
	var out []json.RawMessage // none when neither holds one: the requests then carry no list
	for _, seat := range slices.Sorted(maps.Keys(by)) {
		out = append(out, by[seat])
	}
	return out
}

// factsHead are the facts that name the check, without what the asking gathered.
func (st *state) factsHead() rules.Facts {
	f := rules.Facts{V: rules.FactsVersion, Group: st.c.Group, Heads: st.c.Heads, Trunk: st.c.Trunk}
	for _, p := range st.c.Pulls {
		f.Members = append(f.Members, rules.Member{Repository: p.Repository, Number: p.Number, Head: p.Head, Trunk: p.Trunk})
	}
	return f
}

// judged is the checking program's answer to the check's facts (`-verdict`), with what judged; a program that cannot be
// asked, built or run, or does not answer a verdict, fails the check naming the checking program, never the running
// checker's rules in its place: the facts gain the failure and the verdict is worded from them as a failure.
func (st *state) judged(ctx context.Context) (rules.Answer, *meshdeliverychecker.VerdictsByGroupHeadsValueJudgedBy) {
	body, err := json.Marshal(st.factsOf())
	var by *meshdeliverychecker.VerdictsByGroupHeadsValueJudgedBy
	if err == nil {
		var out json.RawMessage
		if out, by, err = st.runProgram(ctx, []string{"-verdict"}, body); err == nil {
			var a rules.Answer
			if err = json.Unmarshal(out, &a); err == nil && a.V != rules.AnswerVersion {
				err = fmt.Errorf("it answered a verdict of version %d, not %d", a.V, rules.AnswerVersion)
			}
			if err == nil {
				return a, by
			}
		}
	}
	st.fail(st.goModule(), "the check could not be judged by the checking program: "+err.Error())
	return rules.Answer{V: rules.AnswerVersion, Verdict: rules.Judge(st.factsOf())}, by
}

// judgedByOf is the verdict's judged-by of a run of the checking program: where it was built, the key its build is kept
// under, whether this run built it, and the heads it read.
func judgedByOf(prog buildstep.RunProgram, answer buildstep.RunAnswer, heads []buildstep.RunHead) *meshdeliverychecker.VerdictsByGroupHeadsValueJudgedBy {
	by := &meshdeliverychecker.VerdictsByGroupHeadsValueJudgedBy{Program: meshdeliverychecker.VerdictsByGroupHeadsValueJudgedByProgram{
		Repository: prog.Repository, Commit: prog.Commit, Dir: prog.Dir, Package: prog.Package}, Key: answer.Key, Built: answer.Built}
	for _, h := range heads {
		by.Heads = append(by.Heads, meshdeliverychecker.VerdictsByGroupHeadsValueJudgedByHeadsItem{Repository: h.Repository, Commit: h.Commit})
	}
	return by
}