mesh-controller/cmd/mesh-controller/delivery.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"os"
	"slices"
	"sort"
	"strings"
	"time"

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

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

// The controller's part in a delivery (novox/hq ADR 0239, to-be 47).
//
// **A delivery is mesh-delivery's; the walk is the controller's.** A delivery is one commit in one
// repository from its pull request's head to every machine, and a delivery group a few of them sharing a
// branch name; their states, order, holds and record belong to the module holding the `mesh-delivery`
// seat. The controller keeps what it already owned — the planner, the gate, registration, sending, the
// rollback — and the **walk**: one trunk commit's plan, tier by tier across the machines, one machine
// first and judged at the gate (ADR 0236). What changes here is when a walk starts.
//
//   - A walk that moves the controller, the node-engine, the node tools, the bus or mesh-delivery itself is
//     **on the controller's own path**: started by the merge, as every plan was. mesh-delivery cannot gate
//     or deliver itself, and nothing the core needs to be repaired may wait for it.
//   - Any other walk, **while the seat has a holder on record**, is opened by the merge and waits — nothing
//     asked, nothing registered, nothing sent — until mesh-delivery says `deliver`, or a person says
//     `plans go`. Nothing of it is registered before then, so no other send can carry it to a machine
//     before its turn (ADR 0236 §4a carries everything registered).
//   - With no holder on record, every walk starts at the merge, exactly as before this existed.
//
// The verbs mesh-delivery asks with are here: `delivery plan` (the planner's one answer for a diffset),
// `delivery order` (a group's order from the graph), `delivery check` (a group's heads composed), `delivery
// go`, `delivery stop` and `delivery walks`.

// onTheControllersPath are the modules whose walk never waits for the delivery's owner, and what each is.
var onTheControllersPath = map[string]string{
	"mesh-controller":      "the controller",
	"mesh-host":            "the node-engine",
	"node-tools":           "the node tools",
	"nats":                 "the bus",
	catalogue.DeliverySeat: "the delivery's owner",
}

// deliverySeatHeld is whether a module claiming the delivery seat is assigned somewhere: the seat has a
// holder on record. Read from the catalogue the merge handler already holds; never from whether the
// holder answers, so a walk does not start by itself because mesh-delivery is down — that wait is said.
func deliverySeatHeld(entries []inventory.Entry) bool {
	for _, e := range entries {
		if len(e.On) > 0 && e.Manifest.ClaimsSeat(catalogue.DeliverySeat) {
			return true
		}
	}
	return false
}

// awaitsFor is what a new walk waits for: nil when it starts at once — no holder on record, or a module
// on the controller's own path among those it moves.
func awaitsFor(entries []inventory.Entry, modules []string) *inventory.PlanDelivery {
	if !deliverySeatHeld(entries) {
		return nil
	}
	for _, m := range modules {
		if _, own := onTheControllersPath[m]; own {
			return nil
		}
	}
	return &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat}
}

// waitingNote is what a walk waiting for its word says, wherever it is read.
func waitingNote(p inventory.Plan) string {
	return fmt.Sprintf("published; its walk waits for %s's word — `plans go %s --why …` starts it by hand",
		p.Delivery.Awaits, p.ID)
}

// letGo gives a waiting walk its word: by the delivery's owner, or a person. The next advance asks its
// first tier. Refused for a walk that does not wait.
func letGo(ctx context.Context, inv *inventory.Inventory, id, by, why string) (inventory.Plan, error) {
	release, err := inv.HoldPlans(ctx, true)
	if err != nil {
		return inventory.Plan{}, err
	}
	defer release()
	p, err := inv.PlanByID(ctx, id)
	if err != nil {
		return inventory.Plan{}, err
	}
	switch {
	case !p.Open():
		return p, fmt.Errorf("%s is %s: there is no walk to start", p.ID, p.State)
	case p.Delivery == nil || p.Delivery.Awaits == "":
		return p, fmt.Errorf("%s waits for nobody: it started at its merge", p.ID)
	case p.Delivery.Go != nil:
		return p, fmt.Errorf("%s was let go by %s at %s already", p.ID, p.Delivery.By,
			p.Delivery.Go.Local().Format("15:04:05"))
	}
	now := time.Now().UTC()
	p.Delivery.Go, p.Delivery.By, p.Delivery.Why = &now, by, why
	p.Note = "let go by " + by + "; its first tier is asked next"
	if err := inv.SavePlan(ctx, &p); err != nil {
		return p, err
	}
	return p, nil
}

// stopWalk ends a walk on its delivery's word: failed, said as stopped by whom, why. What it asked still
// builds and registers; nothing further is asked or sent.
func stopWalk(ctx context.Context, inv *inventory.Inventory, id, by, why string) (inventory.Plan, error) {
	release, err := inv.HoldPlans(ctx, true)
	if err != nil {
		return inventory.Plan{}, err
	}
	defer release()
	p, err := inv.PlanByID(ctx, id)
	if err != nil {
		return inventory.Plan{}, err
	}
	if !p.Open() {
		return p, fmt.Errorf("%s is already %s", p.ID, p.State)
	}
	if p.Delivery == nil {
		p.Delivery = &inventory.PlanDelivery{}
	}
	p.Delivery.Stopped, p.Delivery.StoppedWhy = by, why
	p.State = inventory.PlanFailed
	p.Note = fmt.Sprintf("stopped by %s at tier %d: %s", by, p.Tier, why)
	if err := inv.SavePlan(ctx, &p); err != nil {
		return p, err
	}
	return p, nil
}

// orderMember is one member of a group, as mesh-delivery says it.
type orderMember struct {
	ID             string   `json:"id"`
	Repository     string   `json:"repository"`
	Base           string   `json:"base,omitempty"`
	Head           string   `json:"head,omitempty"`
	Number         int      `json:"number,omitempty"`
	Paths          []string `json:"paths,omitempty"`
	PathsTruncated bool     `json:"paths_truncated,omitempty"`
	Removed        []string `json:"removed,omitempty"`
	ModuleDirs     []string `json:"module_dirs,omitempty"`
	ModuleDirsSaid bool     `json:"module_dirs_said,omitempty"`
	CloneURL       string   `json:"clone_url,omitempty"`
	// After are the repositories its pull request says it goes after (`after: <repository>`).
	After []string `json:"after,omitempty"`
}

// pull is the member as the forge announced it.
func (m orderMember) pull() link.PullUpdated {
	owner, repo, _ := strings.Cut(m.Repository, "/")
	base := m.Base
	if base == "" {
		base = "main"
	}
	return link.PullUpdated{Owner: owner, Repo: repo, Number: m.Number, Base: base, Commit: m.Head,
		CloneURL: m.CloneURL, Paths: m.Paths, PathsTruncated: m.PathsTruncated, Removed: m.Removed,
		ModuleDirs: m.ModuleDirs, ModuleDirsSaid: m.ModuleDirsSaid}
}

// orderPair is one "before" among a group's members, and why.
type orderPair struct {
	Before string `json:"before"`
	After  string `json:"after"`
	Why    string `json:"why"`
}

// orderReach is what a member moves, as the planner says.
type orderReach struct {
	Moved      []string `json:"moved,omitempty"`
	Dependents []string `json:"dependents,omitempty"`
	New        []string `json:"new,omitempty"`
	Manifests  []string `json:"manifests,omitempty"`
}

// groupOrder is a group's order: the members in it, every pair and why, and the cycle when there is one.
type groupOrder struct {
	Order []string              `json:"order"`
	Pairs []orderPair           `json:"pairs,omitempty"`
	Cycle []string              `json:"cycle,omitempty"`
	Reach map[string]orderReach `json:"reach,omitempty"`
}

// The reasons a pair is ordered, in the words the delivery plan shows.
const (
	orderDeclared    = "declared"
	orderBuiltBy     = "built by"
	orderVersionSkew = "version skew"
	orderEngineFirst = "engine before controller"
)

// orderOf is a group's order (novox/hq ADR 0239 decision 4), pure. A member goes before another when:
//
//   - its pull request is named in the other's `after:` lines (declared);
//   - it moves a module the other's moved modules are built by, stand on, package or declare (built by);
//   - it moves the controller and the other changes any module's manifest (version skew: the newer
//     controller parses what the newer manifest says) — the node-engine's excepted, which goes first;
//   - it moves the node-engine and the other moves the controller (the witness reads nothing the controller
//     does not yet grant, and a controller sends nothing an older engine would refuse — ADR 0236's rollout
//     order).
//
// Otherwise, by repository then id, so every reading gives one order. A pair both ways is a cycle: the
// members in it are named, and none of them is ordered.
func orderOf(members []orderMember, reach map[string]orderReach, edges []inventory.Edge) groupOrder {
	out := groupOrder{Reach: reach}
	byID := map[string]orderMember{}
	for _, m := range members {
		byID[m.ID] = m
	}
	add := func(before, after, why string) {
		if before == after {
			return
		}
		for _, p := range out.Pairs {
			if p.Before == before && p.After == after {
				return
			}
		}
		out.Pairs = append(out.Pairs, orderPair{Before: before, After: after, Why: why})
	}
	named := func(said, repository string) bool {
		said = strings.TrimSuffix(strings.TrimSpace(said), ".git")
		if strings.EqualFold(said, repository) {
			return true
		}
		_, repo, _ := strings.Cut(repository, "/")
		return strings.EqualFold(said, repo)
	}
	for _, a := range members {
		for _, b := range members {
			if a.ID == b.ID {
				continue
			}
			ra, rb := reach[a.ID], reach[b.ID]
			for _, said := range b.After {
				if named(said, a.Repository) {
					add(a.ID, b.ID, orderDeclared)
				}
			}
			for _, e := range edges {
				if slices.Contains(ra.Moved, e.To) && slices.Contains(rb.Moved, e.From) && e.From != e.To {
					add(a.ID, b.ID, orderBuiltBy)
					break
				}
			}
			if slices.Contains(ra.Moved, "mesh-controller") && len(rb.Manifests) > 0 &&
				!slices.Contains(rb.Moved, "mesh-controller") && !slices.Contains(rb.Moved, "mesh-host") {
				add(a.ID, b.ID, orderVersionSkew)
			}
			if slices.Contains(ra.Moved, "mesh-host") && slices.Contains(rb.Moved, "mesh-controller") &&
				!slices.Contains(ra.Moved, "mesh-controller") {
				add(a.ID, b.ID, orderEngineFirst)
			}
		}
	}
	sort.Slice(out.Pairs, func(i, j int) bool {
		if out.Pairs[i].Before != out.Pairs[j].Before {
			return out.Pairs[i].Before < out.Pairs[j].Before
		}
		return out.Pairs[i].After < out.Pairs[j].After
	})
	// Kahn's walk, the next always the first by repository and id among those with nothing before them.
	waiting := map[string]int{}
	for _, m := range members {
		waiting[m.ID] = 0
	}
	for _, p := range out.Pairs {
		waiting[p.After]++
	}
	done := map[string]bool{}
	for len(done) < len(members) {
		var ready []orderMember
		for _, m := range members {
			if !done[m.ID] && waiting[m.ID] == 0 {
				ready = append(ready, m)
			}
		}
		if len(ready) == 0 {
			for _, m := range members {
				if !done[m.ID] {
					out.Cycle = append(out.Cycle, m.ID)
				}
			}
			sort.Strings(out.Cycle)
			return out
		}
		sort.Slice(ready, func(i, j int) bool {
			if ready[i].Repository != ready[j].Repository {
				return ready[i].Repository < ready[j].Repository
			}
			return ready[i].ID < ready[j].ID
		})
		next := ready[0]
		done[next.ID] = true
		out.Order = append(out.Order, next.ID)
		for _, p := range out.Pairs {
			if p.Before == next.ID {
				waiting[p.After]--
			}
		}
	}
	return out
}

// reachOfMembers is each member's reach, as the planner says it.
func reachOfMembers(members []orderMember, entries []inventory.Entry, read map[string][]inventory.ReadRepository,
	edges []inventory.Edge) map[string]orderReach {
	out := map[string]orderReach{}
	for _, m := range members {
		s := pullScope(m.pull(), entries, read, edges)
		out[m.ID] = orderReach{Moved: s.Modules, Dependents: s.Dependents, New: s.New, Manifests: s.Manifests}
	}
	return out
}

// deliveryCommand is `delivery`, the controller's verbs for the delivery's owner:
//
//	delivery plan --repository owner/repo --head <commit> [--base main] --paths a,b [--module-dirs …] [--removed …]
//	delivery order --members <json>
//	delivery check --group <id> --members <json>
//	delivery go <plan> [--by <who>] [--why <text>]
//	delivery stop <plan> --why <text> [--by <who>]
//	delivery walks [-n 50] [--plan <id>]
func deliveryCommand(ctx context.Context, args []string) error {
	if len(args) == 0 {
		return errors.New("delivery plan|order|check|go|stop|walks")
	}
	sub, rest := args[0], args[1:]
	set := flag.NewFlagSet("delivery "+sub, flag.ContinueOnError)
	repository := set.String("repository", "", "owner/repository")
	base := set.String("base", "main", "the branch it merges into")
	head := set.String("head", "", "the commit at hand")
	paths := set.String("paths", "", "the files it changes, comma-separated")
	moduleDirs := set.String("module-dirs", "", "the directories holding a module.json at the head, comma-separated")
	removed := set.String("removed", "", "the files it deletes, comma-separated")
	membersJSON := set.String("members", "", "a group's members, as JSON")
	group := set.String("group", "", "a delivery group's id")
	by := set.String("by", catalogue.DeliverySeat, "who says it")
	why := set.String("why", "", "why")
	limit := set.Int("n", 50, "how many ended walks to answer beside the open ones")
	planID := set.String("plan", "", "one walk")
	positionals, err := parseAround(set, rest)
	if err != nil {
		return err
	}
	var members []orderMember
	if *membersJSON != "" {
		if err := json.Unmarshal([]byte(*membersJSON), &members); err != nil {
			return fmt.Errorf("the members are not readable: %w", err)
		}
	}
	open, err := openStores(ctx)
	if err != nil {
		return err
	}
	defer open.Close()
	inv := open.inventory
	answer := func(v any) error {
		enc := json.NewEncoder(os.Stdout)
		enc.SetIndent("", "  ")
		return enc.Encode(v)
	}
	switch sub {
	case "plan":
		if *repository == "" || *head == "" && *paths == "" {
			return errors.New("delivery plan --repository owner/repo --head <commit> --paths a,b")
		}
		m := orderMember{ID: *repository + "@" + *head, Repository: *repository, Base: *base, Head: *head,
			Paths: splitList(*paths), Removed: splitList(*removed)}
		if d := moduleDirsOf(*moduleDirs); d != nil {
			m.ModuleDirs, m.ModuleDirsSaid = *d, true
		}
		entries, read, edges, err := theGraph(ctx, inv)
		if err != nil {
			return err
		}
		s := pullScope(m.pull(), entries, read, edges)
		plan := changePlanOf(*repository, *base, *head, s.Reach, entries, func(module string) (inventory.Upgrade, bool) {
			u, err := inv.UpgradeOf(ctx, module)
			return u, err == nil
		})
		return answer(map[string]any{"plan": plan, "reach": orderReach{Moved: s.Modules, Dependents: s.Dependents,
			New: s.New, Manifests: s.Manifests}, "mesh": s.Mesh, "gated": s.gated()})
	case "order":
		if len(members) == 0 {
			return errors.New("delivery order --members <json>: a group has members")
		}
		entries, read, edges, err := theGraph(ctx, inv)
		if err != nil {
			return err
		}
		return answer(orderOf(members, reachOfMembers(members, entries, read, edges), edges))
	case "check":
		if *group == "" && len(members) == 1 {
			// One head, checked again as the forge's announcement would have it (a recheck): the controller's
			// own path for a pull request, run because the delivery's owner asked.
			if err := sayingOnTheBus(ctx, func() error { return (following{open}).PullUpdated(ctx, members[0].pull()) }); err != nil {
				return err
			}
			return answer(map[string]any{"rechecked": members[0].ID})
		}
		if *group == "" || len(members) < 2 {
			return errors.New("delivery check --group <id> --members <json> (two heads or more), or --members <one head>")
		}
		id, err := askGroupCheck(ctx, open, *group, members)
		if err != nil {
			return err
		}
		return answer(map[string]any{"asked": id, "group": *group})
	case "go":
		if len(positionals) != 1 {
			return errors.New("delivery go <plan> [--by <who>] [--why <text>]")
		}
		var p inventory.Plan
		if err := sayingOnTheBus(ctx, func() (err error) {
			p, err = letGo(ctx, inv, positionals[0], *by, strings.TrimSpace(*why))
			return err
		}); err != nil {
			return err
		}
		fmt.Printf("%s (%s at %s) is let go by %s; its first tier is asked at the next pass\n", p.ID, p.Repository,
			short(p.Commit), *by)
		return nil
	case "stop":
		if len(positionals) != 1 || strings.TrimSpace(*why) == "" {
			return errors.New("delivery stop <plan> --why <text> [--by <who>]")
		}
		var p inventory.Plan
		if err := sayingOnTheBus(ctx, func() (err error) {
			p, err = stopWalk(ctx, inv, positionals[0], *by, strings.TrimSpace(*why))
			return err
		}); err != nil {
			return err
		}
		fmt.Printf("%s stopped by %s at tier %d of %d; what was asked still builds and registers, nothing further "+
			"is asked or sent\n", p.ID, *by, p.Tier, len(p.Tiers))
		return nil
	case "walks":
		var walks []inventory.Plan
		if *planID != "" {
			p, err := inv.PlanByID(ctx, *planID)
			if err != nil {
				return err
			}
			walks = []inventory.Plan{p}
		} else {
			if walks, err = inv.OpenPlans(ctx); err != nil {
				return err
			}
			recent, err := inv.RecentPlans(ctx, *limit)
			if err != nil {
				return err
			}
			for _, p := range recent {
				if !slices.ContainsFunc(walks, func(w inventory.Plan) bool { return w.ID == p.ID }) {
					walks = append(walks, p)
				}
			}
		}
		entries, err := inv.Catalogued(ctx)
		if err != nil {
			return err
		}
		return answer(map[string]any{"held": deliverySeatHeld(entries), "walks": walks,
			"own-path": sortedKeysOf(ownPathWords())})
	}
	return fmt.Errorf("delivery %s: plan, order, check, go, stop or walks", sub)
}

// ownPathWords is the controller's own path as words, for an answer.
func ownPathWords() map[string]string { return onTheControllersPath }

// theGraph is what the planner reads.
func theGraph(ctx context.Context, inv *inventory.Inventory) ([]inventory.Entry, map[string][]inventory.ReadRepository,
	[]inventory.Edge, error) {
	entries, err := inv.Catalogued(ctx)
	if err != nil {
		return nil, nil, nil, err
	}
	read, err := inv.ReadRepositories(ctx)
	if err != nil {
		return nil, nil, nil, err
	}
	edges, err := inv.Dependencies(ctx)
	if err != nil {
		return nil, nil, nil, err
	}
	return entries, read, edges, nil
}

// groupPrimary is the head a group's composed check is run from: the one moving the controller, which then
// judges the group by itself; else the one moving the node-engine, whose validator judges; else the first.
func groupPrimary(members []orderMember, reach map[string]orderReach) int {
	for _, want := range []string{"mesh-controller", "mesh-host"} {
		for i, m := range members {
			if slices.Contains(reach[m.ID].Moved, want) {
				return i
			}
		}
	}
	return 0
}

// askGroupCheck asks the build seat for a group's composed check: every head laid over the mesh in turn,
// judged as one future state (novox/hq ADR 0239). The verdict comes back as `checked` with the group's id,
// for mesh-delivery; the forge's holder sets no status from it.
func askGroupCheck(ctx context.Context, open *stores, group string, members []orderMember) (string, error) {
	entries, read, edges, err := theGraph(ctx, open.inventory)
	if err != nil {
		return "", err
	}
	reach := reachOfMembers(members, entries, read, edges)
	primary := groupPrimary(members, reach)
	p := members[primary].pull()
	scope := pullScope(p, entries, read, edges)
	// The gate runs over what any member moves; a judge from the primary alone.
	for _, m := range members {
		r := reach[m.ID]
		scope.Modules = appendNew(scope.Modules, r.Moved...)
		scope.Dependents = appendNew(scope.Dependents, r.Dependents...)
	}
	request, err := checkRequestFor(ctx, open, p, scope, entries)
	if err != nil {
		return "", err
	}
	request.Check.Group = group
	world, err := theRestOfTheMesh(ctx, open.inventory, shelfOf(entries), "")
	if err != nil {
		return "", err
	}
	for i, m := range members {
		if i == primary {
			continue
		}
		mp := m.pull()
		// As the mesh clones a module built from it; a repository no module is built from, from the forge.
		source := inventory.Source{Seat: gitSeat, Repository: mp.Owner + "/" + mp.Repo}
		for _, e := range entries {
			if !e.Provided && sameRepository(e.Source.Repository, link.SourceMoved{Owner: mp.Owner, Repo: mp.Repo}) {
				source = e.Source
				break
			}
		}
		url := source.Repository
		if source.Seat != "" {
			url, err = clonedFromSeat(world, source.Seat, source.Repository)
		}
		if err != nil {
			return "", fmt.Errorf("%s cannot be cloned for the group's check: %w", m.ID, err)
		}
		request.Check.Members = append(request.Check.Members, link.GroupMember{Owner: mp.Owner, Repo: mp.Repo,
			Number: mp.Number, Repository: url, Ref: mp.Commit, Paths: mp.Paths})
	}
	seat := buildSeatHeld(ctx)
	ask, err := askOverOn(seat)
	if err != nil {
		return "", err
	}
	defer ask.Close()
	if err := ask.Ask(ctx, request); err != nil {
		return "", err
	}
	fmt.Fprintf(os.Stderr, "group %s: asked %s to check %d head(s) composed together, as %s\n", group, seat,
		len(members), request.ID)
	return request.ID, nil
}

// appendNew appends what is not there yet.
func appendNew(to []string, items ...string) []string {
	for _, i := range items {
		if !slices.Contains(to, i) {
			to = append(to, i)
		}
	}
	return to
}

// shelfOf is the catalogue's manifests by name.
func shelfOf(entries []inventory.Entry) map[string]catalogue.Manifest {
	shelf := map[string]catalogue.Manifest{}
	for _, e := range entries {
		shelf[e.Manifest.Module] = e.Manifest
	}
	return shelf
}

// sayPlanMoved says a walk kept in a new state as the controller's `plan-moved`, the plan whole: what the
// delivery it walks reads its steps from. Never in the way of the walk: said beside it, and a save that could
// not be said is logged — the delivery's owner, which also asks for the walks it follows, finds it by
// comparison. Records arriving out of order are told apart by the plan's revision.
func sayPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) {
	go publishPlanMoved(ctx, bus, p)
}

func publishPlanMoved(ctx context.Context, bus link.Bus, p inventory.Plan) {
	body, err := json.Marshal(p)
	if err != nil {
		return
	}
	stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
	defer stop()
	if err := bus.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeyPlanMoved, body); err != nil {
		fmt.Fprintf(os.Stderr, "%s: kept, and could not be said as plan-moved: %v\n", p.ID, err)
	}
}

// sayingOnTheBus runs a command that keeps walks or says verdicts with the bus to say them on: the serving
// controller's, or a connection of the command's own — a verb runs as a command of its own, and what it kept
// would otherwise be said by nobody until the delivery's owner read the walks back. Without a bus the command
// still runs; what it did is found by comparison.
func sayingOnTheBus(ctx context.Context, f func() error) error {
	if checkEvents != nil {
		return f()
	}
	ran := false
	err := onTheBus(func(conn *nats.Conn) error {
		js, err := conn.JetStream()
		if err != nil {
			return err
		}
		bus := link.OverNATS{Conn: conn, JS: js}
		checkEvents = bus
		inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, p) }
		defer func() { checkEvents, inventory.PlanSaved = nil, nil }()
		ran = true
		return f()
	})
	if !ran {
		fmt.Fprintf(os.Stderr, "the bus cannot be reached (%v): what this does is not said, and is found by comparison\n", err)
		return f()
	}
	return err
}