mesh-catalog/modules/mesh-issues/cmd/mesh-issues/delivery.go

package main

// Whether an issue's fix is delivered (novox/hq ADR 0281, cut by ADR 0539 §17). The tracker hears the gate's `transition`
// event for every pull request an issue's fixed-by names and keeps, per pull request, the state the gate last said — merged
// or not, held and why, closed — and the delivery view's `delivered` event for what reached the machines: the gate keeps no
// state after the merge, so a source revision's delivered state is the view's, said once per revision with the group that
// moved it and the merged heads. From that it derives one line for `get` and
// `list`: a measured fact, never set by a verb. Delivered is not resolved: when every fix of an issue not yet
// resolved is delivered, the issue gains a note and the event `ready-to-verify`, and an agent proves its
// acceptance live before resolving it. A fix merged and not delivered past a bound is said on the issue and to
// the operator's channel.

import (
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

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

// EventReadyToVerify is said once when every fix of an issue not yet resolved is delivered.
const EventReadyToVerify = "ready-to-verify"

// DefaultStuckAfter is how long a fix may be merged and not delivered before it is said stuck: a walk takes
// minutes, a gate ten at most, and a hold waits for a person; two hours is past every ordinary delivery and
// still the same working day.
const DefaultStuckAfter = 2 * time.Hour

// Fix is one pull request of an issue's fixed-by, as the delivery last said it.
type Fix struct {
	PullRequest string     `json:"pull-request"`
	State       string     `json:"state"`
	Why         string     `json:"why,omitempty"`
	Merged      bool       `json:"merged,omitempty"`
	MergedAt    *time.Time `json:"merged-at,omitempty"`
	Since       time.Time  `json:"since"`
	// StuckSaid says it was said stuck, on the issue and to the operator's channel.
	StuckSaid bool `json:"stuck-said,omitempty"`
	// Class is the class of what its delivery moves, as mesh-delivery says it: core, leaf, or none when it moves
	// no module; Opened when its pull request was first heard (novox/hq ADR 0313 §2, §4).
	Class  string     `json:"class,omitempty"`
	Opened *time.Time `json:"opened,omitempty"`
	// Repository, Group and MergedAs are what the gate said of its merge, which the view's `delivered` event is matched to:
	// the repository (owner/name), the group that moved the source, and the commit the merge landed as.
	Repository string `json:"repository,omitempty"`
	Group      string `json:"group,omitempty"`
	MergedAs   string `json:"merged-as,omitempty"`
}

// heardFix is what the delivery said of a pull request at one moment: its state, why, whether it is merged, and
// what a fix's time is read by — the class of what it moves, when its pull request opened and when it merged,
// each when said.
type heardFix struct {
	State    string
	Why      string
	Merged   bool
	At       time.Time
	Class    string
	Opened   *time.Time
	MergedAt *time.Time
	// Repository, Group and MergedAs: see Fix.
	Repository string
	Group      string
	MergedAs   string
}

// news says whether what was heard adds anything to what a fix keeps.
func (h heardFix) news(f *Fix, kept string) bool {
	if f.State != h.State || f.Why != kept || (h.Merged && !f.Merged) || (h.Class != "" && h.Class != f.Class) {
		return true
	}
	if h.Opened != nil && (f.Opened == nil || !f.Opened.Equal(*h.Opened)) {
		return true
	}
	if h.Group != "" && f.Group != h.Group || h.MergedAs != "" && f.MergedAs != h.MergedAs || h.Repository != "" && f.Repository != h.Repository {
		return true
	}
	return h.MergedAt != nil && (f.MergedAt == nil || !f.MergedAt.Equal(*h.MergedAt))
}

// reached says whether a fix's state says the change reached the machines: only the view's `delivered` event says it
// (OnDelivered); the gate's states end at the merge.
func reached(state string) bool { return state == "delivered" }

// Whether a pull request is merged is what the delivery says of the pull request itself — `merged_as` on a
// transition, the pull request's `merged` in checks — never read from the state: `stopped` is also what a pull
// request closed without merging becomes (mesh-delivery's `closed` event), and read as merged it was said stuck to
// the operator for good (novox/hq issue 483).

// withdrawn says whether a fix is a pull request closed without merging: no fix at all, neither awaiting delivery,
// delivered, nor open.
func withdrawn(f *Fix) bool {
	return f != nil && (f.State == "closed" || f.State == "stopped") && !f.Merged
}

// liveFixes is an issue's fixes less those withdrawn: what its delivery, readiness to verify, the stuck sweep and a
// abandoned issue are measured on. fixPullRequests still names the withdrawn ones, so a pull request reopened is heard.
func liveFixes(is *Issue) []string {
	var out []string
	for _, pr := range fixPullRequests(is) {
		if !withdrawn(fixOf(is, pr)) {
			out = append(out, pr)
		}
	}
	return out
}

// withdrawnFixes is the pull requests of an issue closed without merging.
func withdrawnFixes(is *Issue) []string {
	var out []string
	for _, pr := range fixPullRequests(is) {
		if withdrawn(fixOf(is, pr)) {
			out = append(out, pr)
		}
	}
	return out
}

func fixOf(is *Issue, pr string) *Fix {
	for i := range is.Fixes {
		if is.Fixes[i].PullRequest == pr {
			return &is.Fixes[i]
		}
	}
	return nil
}

// allDelivered says whether an issue names a pull request and every one is delivered.
func allDelivered(is *Issue) bool {
	prs := liveFixes(is)
	if len(prs) == 0 {
		return false
	}
	for _, pr := range prs {
		if !contains(is.Delivered, pr) {
			return false
		}
	}
	return true
}

// DeliveryLine is the derived delivery state of an issue's fix, in words; empty when fixed-by names no pull
// request.
func DeliveryLine(is *Issue) string {
	prs := liveFixes(is)
	if len(prs) == 0 {
		if w := withdrawnFixes(is); len(w) > 0 {
			return "fix withdrawn: " + strings.Join(w, ", ") + " closed without merging"
		}
		return ""
	}
	var held, failed, merged, open, unheard []string
	for _, pr := range prs {
		f := fixOf(is, pr)
		switch {
		case contains(is.Delivered, pr):
		case f == nil:
			unheard = append(unheard, pr)
		case f.State == "held":
			held = append(held, strings.TrimSpace(pr+": "+f.Why))
		case f.State == "failed" || f.State == "stopped" || f.State == "closed":
			failed = append(failed, strings.TrimSpace(pr+" "+f.State+": "+f.Why))
		case f.Merged:
			merged = append(merged, pr)
		default:
			open = append(open, pr)
		}
	}
	switch {
	case len(failed) > 0:
		return "fix not delivered: " + strings.Join(failed, "; ")
	case len(held) > 0:
		return "held: " + strings.Join(held, "; ")
	case len(merged) > 0:
		return "fix merged, not delivered: " + strings.Join(merged, ", ")
	case len(open) > 0:
		return "fix not merged: " + strings.Join(open, ", ")
	case len(unheard) > 0:
		return "nothing heard from the delivery of " + strings.Join(unheard, ", ")
	}
	return "delivered to every machine it should reach"
}

// recordFix keeps what the delivery said of a pull request on every issue whose fixed-by names it (only, when
// given, on that one issue), and answers the issues it changed.
func (t *Tracker) recordFix(pr string, h heardFix, only uint64) ([]uint64, error) {
	state, why, merged, at := h.State, h.Why, h.Merged, h.At
	all, err := t.allIssues()
	if err != nil {
		return nil, err
	}
	var changed []uint64
	for n, is := range all {
		if (only != 0 && n != only) || !contains(fixPullRequests(is), pr) {
			continue
		}
		kept := cut(why, MostReasonKept)
		if f := fixOf(is, pr); f != nil && (at.Before(f.Since) || !h.news(f, kept)) {
			// Older than what is kept, or nothing new: transitions can arrive out of order.
			continue
		}
		if contains(is.Delivered, pr) && !reached(state) {
			// Delivered stays delivered: a later state that has not reached the machines is a newer delivery of
			// the same pull request at most, and an older one is history.
			continue
		}
		if _, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error {
			f := fixOf(i, pr)
			if f == nil {
				i.Fixes = append(i.Fixes, Fix{PullRequest: pr})
				f = &i.Fixes[len(i.Fixes)-1]
			}
			f.State, f.Why, f.Since = state, kept, at
			if merged && !f.Merged {
				f.Merged = true
				m := at
				f.MergedAt = &m
			}
			if h.MergedAt != nil && (merged || f.Merged) {
				m := h.MergedAt.UTC()
				f.MergedAt = &m // the delivery's own moment of the merge, not when it was heard
			}
			if h.Class != "" {
				f.Class = h.Class
			}
			if h.Repository != "" {
				f.Repository = h.Repository
			}
			if h.Group != "" {
				f.Group = h.Group
			}
			if h.MergedAs != "" {
				f.MergedAs = h.MergedAs
			}
			if h.Opened != nil {
				o := h.Opened.UTC()
				f.Opened = &o
			}
			if reached(state) && !contains(i.Delivered, pr) {
				i.Delivered = append(i.Delivered, pr)
			}
			return nil
		}); err != nil {
			return changed, err
		}
		changed = append(changed, n)
	}
	sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] })
	return changed, nil
}

// afterFix is what follows a change of an issue's fixes: ready to verify, once, when every fix is delivered and
// the issue is not yet resolved; a stuck fix delivered at last cleared on the operator's channel; and what
// waited on it let go.
func (t *Tracker) afterFix(n uint64) error {
	is, _, err := read(t.Store, n)
	if err != nil {
		return err
	}
	if err := t.clearStuck(is); err != nil {
		t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel (tried again at the next sweep): %v", n, err)
	}
	if err := t.clearNotResolved(is); err != nil {
		t.Logf("[mesh-issues] the warning that %d is delivered and not resolved could not be cleared on the operator's channel (tried again at the next sweep): %v", n, err)
	}
	if allDelivered(is) {
		// Its fix time, final once its last fix is delivered (novox/hq ADR 0313).
		if err := t.timeFix(n); err != nil {
			return err
		}
		if unresolved(is) {
			prs := liveFixes(is)
			sort.Strings(prs)
			if strings.Join(is.VerifyFor, ",") != strings.Join(prs, ",") {
				// The note first, kept once by its words; then the claim; then the event. A failure before the claim
				// is tried again at the next offer, and the note is not written twice. It names when each fix was
				// delivered and who works on the issue: no channel of the tracker reaches a worker, so the note
				// and the event are how the worker hears it (novox/hq issue 483).
				var when []string
				for _, pr := range prs {
					if f := fixOf(is, pr); f != nil {
						when = append(when, pr+" at "+f.Since.Format(time.RFC3339))
					} else {
						when = append(when, pr)
					}
				}
				prefix := fmt.Sprintf("Ready to verify: every fix of this issue is delivered (%s).", strings.Join(when, ", "))
				worker := ""
				if is.Worker != nil {
					worker = is.Worker.By
				}
				if !noted(is, prefix) {
					text := prefix
					if worker != "" {
						text += fmt.Sprintf(" %s works on it: prove the acceptance live and resolve it, or say what is left. "+
							"In fixing and not resolved %s after its delivery, it is said to the operator.", worker,
							t.deliveredNotResolvedAfter())
					}
					text += " Delivered is not resolved: prove the acceptance live, then resolve it, from fixing: move it " +
						"to fixing with by first if it is not there. (novox/hq ADR 0281, ADR 0284, issue 483)"
					if _, err := t.note(GroomBy, n, text); err != nil {
						return err
					}
				}
				claimed := false
				if _, err := t.change(n, Act{Op: EventReadyToVerify, By: Seat}, func(i *Issue) error {
					claimed = false
					if strings.Join(i.VerifyFor, ",") == strings.Join(prs, ",") {
						return nil
					}
					i.VerifyFor, claimed = prs, true
					return nil
				}); err != nil {
					return err
				}
				if claimed {
					body := map[string]any{"number": n, "fixed-by": prs, "at": t.now(), "delivered-at": deliveredAt(is)}
					if worker != "" {
						body["worker"] = worker
					}
					t.emit(EventReadyToVerify, body)
				}
			}
		}
	}
	return t.letGo(n)
}

// stillStuck says whether an issue has a fix said stuck that is still named, not delivered, and the issue not
// closed. A closed issue is the end of its warning: resolved means its worker proved the acceptance live, so a fix
// the tracker never heard delivered is not waited for any more (a failed walk that a later merge carried stays
// failed, novox/hq issue 547), and wontfix waits for nothing.
func stillStuck(is *Issue) bool {
	if closedStatus(is.Status) {
		return false
	}
	named := liveFixes(is)
	for _, f := range is.Fixes {
		if f.StuckSaid && contains(named, f.PullRequest) && !contains(is.Delivered, f.PullRequest) {
			return true
		}
	}
	return false
}

// clearStuck clears the warning said for an issue on the operator's channel once nothing it was said for is
// stuck — delivered, no longer named, or the issue closed (resolved or wontfix); one that fails is tried again at
// the next sweep.
func (t *Tracker) clearStuck(is *Issue) error {
	if !is.StuckOpen || stillStuck(is) || t.Notify == nil {
		return nil
	}
	if err := t.Notify(stuckKey(is.Number), "", "", "", "", stuckEnded(is), true); err != nil {
		return err
	}
	_, err := t.change(is.Number, Act{Op: "warned", By: Seat}, func(i *Issue) error { i.StuckOpen = false; return nil })
	return err
}

// stuckEnded is what the operator's channel says when a stuck warning ends.
func stuckEnded(is *Issue) string {
	switch is.Status {
	case StatusResolved:
		return "resolved: issue " + keyOf(is.Number) + " was proven live by its worker"
	case StatusWontfix:
		return "issue " + keyOf(is.Number) + " was closed as wontfix: its fix is no longer waited for"
	}
	return "the fix of issue " + keyOf(is.Number) + " is delivered or no longer waited for"
}

func stuckKey(n uint64) string { return "issue-tracker.fix-stuck." + keyOf(n) }

func prKey(repository string, number int) string {
	_, repo, ok := strings.Cut(repository, "/")
	if !ok {
		repo = repository
	}
	return fmt.Sprintf("%s#%d", strings.ToLower(repo), number)
}

// OnTransition is the handler of mesh-delivery's `transition` event: every state a pull request's delivery moves
// to is kept on each issue whose fixed-by names it.
func (t *Tracker) OnTransition(e stdio.Envelope) error {
	var body struct {
		Repository string     `json:"repository"`
		Number     int        `json:"number"`
		To         string     `json:"to"`
		Why        string     `json:"why"`
		MergedAs   string     `json:"merged_as"`
		At         time.Time  `json:"at"`
		Class      string     `json:"class"`
		Opened     *time.Time `json:"opened"`
		MergedAt   *time.Time `json:"merged_at"`
		Group      string     `json:"group"`
	}
	if err := json.Unmarshal(e.Body, &body); err != nil || body.To == "" || body.Number <= 0 {
		return nil
	}
	if body.To == "delivered" {
		return nil // the gate keeps no state after the merge; what reached the machines is the view's (OnDelivered)
	}
	at := body.At.UTC()
	if at.IsZero() {
		at = t.now()
	}
	merged := body.MergedAs != ""
	changed, err := t.recordFix(prKey(body.Repository, body.Number), heardFix{State: body.To, Why: body.Why, Merged: merged,
		At: at, Class: body.Class, Opened: nonZero(body.Opened), MergedAt: nonZero(body.MergedAt),
		Repository: strings.ToLower(body.Repository), Group: body.Group, MergedAs: body.MergedAs}, 0)
	if err != nil {
		return err
	}
	for _, n := range changed {
		if err := t.afterFix(n); err != nil {
			return err
		}
	}
	return nil
}

// DeliveredEvent is the view's `delivered` (mesh-delivery-view, ADR 0539 §17 and design 47 R6): said once per source revision
// when every want, have and assignment is at the fingerprint and healthy.
const DeliveredEvent = "mesh-delivery-view.delivered"

// OnDelivered is the handler of the view's `delivered` event: {revision, group, heads}. Every merged fix of a pull request the
// group moved, or whose merge is the head the revision holds for its repository, reached the machines at that revision. The same
// revision twice changes nothing (a delivered fix stays delivered). A fix is matched by the group it merged in, by its own merge
// commit, or because its merge is an ancestor of the head the revision holds for its repository (coveredBy): a newer revision
// covers the older ones, which the view may never have said. The view also says each repository's reached head in this form,
// with no group and that repository's head alone (novox/hq ADR 0556 clause 3): a merge at or before it is on the machines.
func (t *Tracker) OnDelivered(e stdio.Envelope) error {
	var body struct {
		Revision uint64            `json:"revision"`
		Group    string            `json:"group"`
		Heads    map[string]string `json:"heads"`
		// At is when the view judged the revision delivered, when it says so; else this tracker's moment of hearing it.
		At time.Time `json:"at"`
		// NotYet are, in the per-repository form, merges at or before the head that are not yet on the machines (a module a node
		// wants and runs nothing countable of, ADR 0556 clause 2): a fix merged as one of them waits for a later event.
		NotYet []string `json:"not_yet"`
	}
	if err := json.Unmarshal(e.Body, &body); err != nil || (body.Group == "" && len(body.Heads) == 0) {
		return nil
	}
	all, err := t.allIssues()
	if err != nil {
		return err
	}
	at := body.At.UTC()
	if at.IsZero() {
		at = t.now()
	}
	var changed []uint64
	learned := 0
	for n, is := range all {
		for _, pr := range liveFixes(is) {
			f := fixOf(is, pr)
			if f == nil && learned < MostSweepAsks {
				// A pull request named and kept no record for (merged before the tracker heard deliveries): it learns its
				// merge now, so this very event can cover it (issue 1950).
				learned++
				if ok, err := t.learnFix(n, pr); err != nil {
					t.Logf("[mesh-issues] the merge of %s (issue %d) could not be learned (asked again at the next event): %v", pr, n, err)
				} else if ok {
					if fresh, _, err := read(t.Store, n); err == nil {
						is, f = fresh, fixOf(fresh, pr)
					}
				}
			}
			if f != nil && f.Merged && !contains(is.Delivered, pr) && f.MergedAs == "" {
				t.fillMerge(n, f, pr)
			}
			if f == nil || !f.Merged || contains(is.Delivered, pr) || notYet(f, body.NotYet) ||
				!(deliveredBy(f, body.Group, body.Heads) || t.coveredBy(f, body.Heads)) {
				continue
			}
			if err := t.markDelivered(n, pr, fmt.Sprintf("source revision %d reached the machines", body.Revision), at); err != nil {
				return err
			}
			if len(changed) == 0 || changed[len(changed)-1] != n {
				changed = append(changed, n)
			}
		}
	}
	sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] })
	for _, n := range changed {
		if err := t.afterFix(n); err != nil {
			return err
		}
	}
	return nil
}

// fillMerge learns what the gate says of the merge of a fix kept without it (heard through `checks`, or kept before release 6
// with only its merged flag): the repository, the group and the commit its merge landed as, asked of the delivery. A delivery
// that cannot be asked, or does not say, leaves the fix as it was; it is asked again at the next event. f is updated in place.
func (t *Tracker) fillMerge(n uint64, f *Fix, pr string) {
	if t.Deliveries == nil {
		return
	}
	repo, number, _ := strings.Cut(pr, "#")
	d, err := t.Deliveries(repo, number)
	if err != nil {
		t.Logf("[mesh-issues] the merge of %s (issue %d) could not be asked of the delivery: %v", pr, n, err)
		return
	}
	if d.MergedAs == "" {
		return
	}
	if _, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error {
		if g := fixOf(i, pr); g != nil {
			g.MergedAs, g.Group = d.MergedAs, d.Group
			if d.Repository != "" {
				g.Repository = d.Repository
			}
		}
		return nil
	}); err != nil {
		t.Logf("[mesh-issues] the merge of %s (issue %d) could not be kept: %v", pr, n, err)
		return
	}
	f.MergedAs, f.Group = d.MergedAs, d.Group
	if d.Repository != "" {
		f.Repository = d.Repository
	}
}

// busContains asks the seat `git`'s `contains` over the bus: whether commit is an ancestor of in, or it.
func busContains(owner, repo, commit, in string) (bool, error) {
	raw, err := stdio.Ask("seat:git.contains", map[string]any{"repository": owner + "/" + repo, "commit": commit, "in": in})
	if err != nil {
		return false, err
	}
	return parseContains(raw)
}

func parseContains(raw json.RawMessage) (bool, error) {
	m, err := answerOf(raw, "git.contains")
	if err != nil {
		return false, err
	}
	var yes *bool
	if json.Unmarshal(m["contains"], &yes) != nil || yes == nil {
		return false, errors.New("git.contains answered no verdict")
	}
	return *yes, nil
}

// coveredBy says whether a merged fix's merge is an ancestor of, or equal to, the head a delivered revision holds for its
// repository: the view says each revision once and may skip one delivered together with a later one, so a newer revision
// covers the older merges of its repository. The forge decides ancestry; a forge that cannot say leaves the fix for the next event.
func (t *Tracker) coveredBy(f *Fix, heads map[string]string) bool {
	if t.Contains == nil || f.MergedAs == "" {
		return false
	}
	for repository, head := range heads {
		if !strings.EqualFold(repository, f.Repository) {
			continue
		}
		owner, repo, ok := strings.Cut(repository, "/")
		if !ok {
			continue
		}
		yes, err := t.Contains(owner, repo, f.MergedAs, head)
		if err != nil {
			t.Logf("[mesh-issues] whether %s contains %s is not known yet: %v", shortCommit(head), shortCommit(f.MergedAs), err)
			continue
		}
		return yes
	}
	return false
}

func shortCommit(c string) string { return c[:min(len(c), 12)] }

// deliveredBy says whether a merged fix is one a delivered revision names: the group it merged in is the group that moved the
// source, or its merge commit is the head the revision holds for its repository.
func deliveredBy(f *Fix, group string, heads map[string]string) bool {
	if group != "" && f.Group == group {
		return true
	}
	for repository, head := range heads {
		if f.MergedAs != "" && strings.EqualFold(repository, f.Repository) && sameCommitIn(f.MergedAs, head) {
			return true
		}
	}
	return false
}

// notYet is whether a fix's merge is among the merges an event says are not yet on the machines.
func notYet(f *Fix, merges []string) bool {
	for _, m := range merges {
		if f.MergedAs != "" && sameCommitIn(f.MergedAs, m) {
			return true
		}
	}
	return false
}

// sameCommitIn is whether two commits are one, however either is abbreviated (at least seven characters).
func sameCommitIn(a, b string) bool {
	a, b = strings.ToLower(a), strings.ToLower(b)
	if len(a) > len(b) {
		a, b = b, a
	}
	return len(a) >= 7 && strings.HasPrefix(b, a)
}

// Delivery is a pull request's delivery as the tracker learns it (from the seat `git`'s `pull` and from the gate's events); a test stubs the asking.
type Delivery struct {
	State string
	Why   string
	// Since is when the delivery entered its state, as it says; zero when it does not.
	Since time.Time
	// Merged is the pull request's own merged flag, as the forge says it.
	Merged bool
	// Class, Opened and MergedAt are what a fix's time is read by (novox/hq ADR 0313), when the delivery says them.
	Class    string
	Opened   *time.Time
	MergedAt *time.Time
	// Repository (owner/name), Group and MergedAs are what the gate says of the merge, which the view's `delivered` is matched by.
	Repository string
	Group      string
	MergedAs   string
	// Pulled says the answer carried the forge's pull request (merged, closed, the commit its merge landed as); Closed that it
	// was closed without merging.
	Pulled bool
	Closed bool
}

// heard is what a delivery answered, as a fix keeps it.
func (d Delivery) heard(since time.Time) heardFix {
	return heardFix{State: d.State, Why: d.Why, Merged: d.Merged, At: since, Class: d.Class, Opened: d.Opened, MergedAt: d.MergedAt,
		Repository: d.Repository, Group: d.Group, MergedAs: d.MergedAs}
}

// nonZero is a moment said, or nil for one not said.
func nonZero(t *time.Time) *time.Time {
	if t == nil || t.IsZero() {
		return nil
	}
	u := t.UTC()
	return &u
}

// DeliveryState reads one pull request's facts from the seat `git`.
type DeliveryState func(repository string, number string) (Delivery, error)

// OnMoved is the handler of the tracker's own `moved` event: each pull request of the issue's fix not heard of
// yet is asked of mesh-delivery, at most a few per handling, for a delivery that moved before the issue named
// it; then what follows its fixes, and an issue closed lets go what waited on it.
func (t *Tracker) OnMoved(e stdio.Envelope) error {
	var body struct {
		Number uint64 `json:"number"`
	}
	if err := json.Unmarshal(e.Body, &body); err != nil || body.Number == 0 {
		return nil
	}
	is, _, err := read(t.Store, body.Number)
	if errors.Is(err, ErrNoIssue) {
		return nil
	}
	if err != nil {
		return err
	}
	if t.Deliveries != nil {
		// Each ask up to the bus's 30 seconds, under the runtime's two minutes; one that fails, or more left to
		// ask, has the event offered again, so a delivery never asked ends in max-deliveries, not in silence.
		asked, _, err := t.learnUnheard(body.Number, is, MostDeliveryAsks)
		if err != nil {
			return err
		}
		if asked == MostDeliveryAsks && len(unheardOf(is)) > asked {
			return fmt.Errorf("issue %d names more pull requests than are asked at once; the rest at the next offer", body.Number)
		}
	}
	return t.afterFix(body.Number)
}

// Notifier tells the operator's channel something (operator-channel.notify), or clears it.
type Notifier func(key, severity, summary, headline, explanation, resolved string, clear bool) error

// busNotifier asks the operator's channel over the bus.
func busNotifier() Notifier {
	return func(key, severity, summary, headline, explanation, resolved string, clear bool) error {
		body := map[string]any{"key": key, "subject": "the issue tracker"}
		if clear {
			body["clear"] = true
			body["resolved"] = resolved
		} else {
			body["severity"], body["summary"], body["headline"], body["explanation"] = severity, summary, headline, explanation
		}
		_, err := stdio.Ask("seat:operator-channel.notify", body)
		return err
	}
}

// stuckAfter is the setting's bound, or the default.
func (t *Tracker) stuckAfter() time.Duration {
	if s, err := t.Settings(); err == nil && strings.TrimSpace(s.StuckAfter) != "" {
		if d, err := time.ParseDuration(strings.TrimSpace(s.StuckAfter)); err == nil && d > 0 {
			return d
		}
	}
	return DefaultStuckAfter
}

// Sweep says every fix of an issue not closed merged and not delivered past the bound: a note on its issue and a warning on the
// operator's channel, once; and every issue in fixing whose fixes were all delivered past its own bound and is not
// resolved (novox/hq issue 483). First it asks the delivery of a few fixes never heard of. It answers how many
// warnings it said. Run every few minutes.
func (t *Tracker) Sweep() (int, error) {
	if err := t.askUnheard(); err != nil {
		t.Logf("[mesh-issues] the sweep could not ask a fix's delivery (asked again at the next sweep): %v", err)
	}
	if err := t.coverReached(); err != nil {
		t.Logf("[mesh-issues] the sweep could not match the fixes to what reached the machines (tried again at the next sweep): %v", err)
	}
	all, err := t.allIssues()
	if err != nil {
		return 0, err
	}
	bound, now := t.stuckAfter(), t.now()
	said := 0
	for n, is := range all {
		// A fix time missed at its delivery (a failure, or a restart between) is kept now; one of an issue moved to
		// fixing before the tracker kept the moment is kept only when its fix is heard of again.
		if is.FixingFrom != nil && is.FixTime == nil && allDelivered(is) {
			if err := t.timeFix(n); err != nil {
				t.Logf("[mesh-issues] the fix time of %d could not be kept (tried again at the next sweep): %v", n, err)
			}
		}
		ok, err := t.sayNotResolved(is, now)
		if err != nil {
			t.Logf("[mesh-issues] that %d is delivered and not resolved could not be said on the operator's channel (said at the next sweep): %v", n, err)
		} else if ok {
			said++
		}
		if err := t.clearStuck(is); err != nil {
			t.Logf("[mesh-issues] the stuck fix of %d could not be cleared on the operator's channel: %v", n, err)
		}
		if closedStatus(is.Status) {
			// A resolved or wontfix issue waits for no fix: never said stuck (novox/hq issue 547).
			continue
		}
		named := liveFixes(is)
		for _, f := range is.Fixes {
			if !contains(named, f.PullRequest) || contains(is.Delivered, f.PullRequest) || !f.Merged || f.MergedAt == nil ||
				reached(f.State) || f.StuckSaid || now.Sub(*f.MergedAt) < bound || now.Sub(f.Since) < bound {
				continue
			}
			state := f.State
			if f.Why != "" {
				state += ": " + f.Why
			}
			prefix := fmt.Sprintf("Stuck: the fix %s", f.PullRequest)
			if !noted(is, prefix) {
				if _, err := t.note(GroomBy, n, fmt.Sprintf("%s was merged %s ago and is not delivered (%s). A merged fix "+
					"that does not reach the machines is said on the operator's channel. (novox/hq ADR 0281)", prefix,
					now.Sub(*f.MergedAt).Round(time.Minute), state)); err != nil {
					return said, err
				}
			}
			if t.Notify != nil {
				if err := t.Notify(stuckKey(n), "warning",
					fmt.Sprintf("issue %d: fix %s merged %s ago, not delivered", n, f.PullRequest, now.Sub(*f.MergedAt).Round(time.Minute)),
					fmt.Sprintf("A fix is merged and not delivered (issue %d)", n),
					fmt.Sprintf("The fix of issue %d was merged %s ago and has not reached the machines: %s. Look at its "+
						"delivery; nothing else is needed from you unless it waits for a person.", n,
						now.Sub(*f.MergedAt).Round(time.Minute), state), "", false); err != nil {
					// Tried again at the next sweep: the note is not written twice.
					t.Logf("[mesh-issues] the stuck fix of %d could not be said on the operator's channel: %v", n, err)
					continue
				}
			}
			pr := f.PullRequest
			if _, err := t.change(n, Act{Op: "warned", By: Seat}, func(i *Issue) error {
				if g := fixOf(i, pr); g != nil {
					g.StuckSaid = true
				}
				i.StuckOpen = t.Notify != nil
				return nil
			}); err != nil {
				return said, err
			}
			said++
		}
	}
	if err := t.judgeFixBudgets(); err != nil {
		t.Logf("[mesh-issues] a fix over its budget could not be said or cleared on the operator's channel (tried again at the next sweep): %v", err)
	}
	return said, nil
}

// markDelivered keeps a fix as delivered, with the reason the delivery gave.
func (t *Tracker) markDelivered(n uint64, pr, why string, at time.Time) error {
	_, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error {
		g := fixOf(i, pr)
		if g == nil || contains(i.Delivered, pr) {
			return nil
		}
		g.State, g.Why, g.Since = "delivered", why, at
		i.Delivered = append(i.Delivered, pr)
		return nil
	})
	return err
}

// Reach is what the delivery view says of one repository now (mesh-delivery-view.view_reach, ADR 0556): the newest merge on
// the machines, the source revision that named it, and the merges at or before it that are not yet there.
type Reach struct {
	Head     string
	Revision uint64
	NotYet   []string
}

// busReach asks the delivery view for one repository's reached head.
func busReach(repository string) (Reach, error) {
	raw, err := stdio.Ask("mesh-delivery-view.view_reach", map[string]any{"repository": repository})
	if err != nil {
		return Reach{}, err
	}
	return parseReach(raw)
}

// parseReach reads the view's answer, bare or inside an `answer`.
func parseReach(raw []byte) (Reach, error) {
	type body struct {
		Reached    string `json:"reached"`
		ReachedSrc uint64 `json:"reached_src"`
		Modules    map[string]struct {
			Changes []struct {
				Commit string `json:"commit"`
			} `json:"changes"`
			Nodes int    `json:"nodes"`
			Floor string `json:"floor"`
		} `json:"modules"`
	}
	var out struct {
		body
		Answer *body `json:"answer"`
	}
	if err := json.Unmarshal(raw, &out); err != nil {
		return Reach{}, fmt.Errorf("the view's view_reach is not readable: %.200s", raw)
	}
	b := out.body
	if out.Answer != nil {
		b = *out.Answer
	}
	r := Reach{Head: b.Reached, Revision: b.ReachedSrc}
	for _, m := range b.Modules {
		if m.Nodes == 0 || m.Floor != "" {
			continue // as the view's own notYetOf: a module a node wants and runs nothing countable of
		}
		for _, c := range m.Changes {
			r.NotYet = append(r.NotYet, c.Commit)
		}
	}
	sort.Strings(r.NotYet)
	return r, nil
}

// coverReached is the level-triggered half of delivery: the view says a repository's reached head once per source revision, so a
// fix whose record came after that saying would wait for ever (issue 1950, met twice). Every merged fix not delivered is matched
// against what the view says now of its repository, as OnDelivered matches an event. It answers the issues it changed.
func (t *Tracker) coverReached() error {
	if t.Reached == nil {
		return nil
	}
	all, err := t.allIssues()
	if err != nil {
		return err
	}
	reach := map[string]*Reach{}
	var changed []uint64
	for n, is := range all {
		for _, pr := range liveFixes(is) {
			f := fixOf(is, pr)
			if f != nil && f.Merged && !contains(is.Delivered, pr) && f.MergedAs == "" {
				t.fillMerge(n, f, pr)
			}
			if f == nil || !f.Merged || contains(is.Delivered, pr) || f.MergedAs == "" || f.Repository == "" {
				continue
			}
			key := strings.ToLower(f.Repository)
			r, asked := reach[key]
			if !asked {
				got, err := t.Reached(key)
				if err != nil {
					t.Logf("[mesh-issues] what reached the machines of %s is not known yet (asked again at the next sweep): %v", key, err)
				} else if got.Head != "" {
					r = &got
				}
				reach[key] = r
			}
			if r == nil || notYet(f, r.NotYet) {
				continue
			}
			if !(sameCommitIn(f.MergedAs, r.Head) || t.coveredBy(f, map[string]string{key: r.Head})) {
				continue
			}
			if err := t.markDelivered(n, pr, fmt.Sprintf("source revision %d reached the machines", r.Revision), t.now()); err != nil {
				return err
			}
			if len(changed) == 0 || changed[len(changed)-1] != n {
				changed = append(changed, n)
			}
		}
	}
	sort.Slice(changed, func(i, j int) bool { return changed[i] < changed[j] })
	for _, n := range changed {
		if err := t.afterFix(n); err != nil {
			return err
		}
	}
	return nil
}