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

package main

import (
	"encoding/json"
	"fmt"
	"regexp"
	"sort"
	"strings"
	"time"
)

// What a transition owes outside the state (novox/hq ADR 0239 decision 5), each by its owner: the event on
// the bus (said by this module), a line on the commit's note and the pull request's view and statuses (by
// the forge's holder, asked through its tools). Owed, never lost: kept with the delivery before anything is
// tried, and tried again until done.

// transitionEvent is what `mesh-delivery.transition` says: no secret, no address. opened is when its pull
// request was first heard (zero when not known); with its class and its merge's moment it is what the issue
// tracker measures a fix's time by (novox/hq ADR 0313 §2, §4).
func transitionEvent(d *Delivery, t Transition, opened time.Time) map[string]any {
	e := map[string]any{"id": d.ID, "repository": d.Repository, "commit": d.Commit, "from": t.From, "to": t.To,
		"event": t.Event, "why": t.Why, "at": t.At}
	if t.By != "" {
		e["by"] = t.By
	}
	if d.Number > 0 {
		e["number"] = d.Number
	}
	if d.Group != "" {
		e["group"] = d.Group
	}
	if d.MergedAs != "" {
		e["merged_as"] = d.MergedAs
	}
	if d.Walk != nil {
		e["walk"] = d.Walk.ID
	}
	if d.Plan != nil {
		e["plan"] = d.Plan.Summary
	}
	if c := fixClass(d); c != "" {
		e["class"] = c
	}
	if !opened.IsZero() {
		e["opened"] = opened.UTC()
	}
	if !d.MergedAt.IsZero() {
		e["merged_at"] = d.MergedAt.UTC()
	}
	return e
}

// fixClass is the class of what a delivery moves (novox/hq ADR 0313 §2): its plan's, as the planner said it;
// none for a plan that moves no module, from a planner that did not say a class; else its walk's class once its
// delivery time is known; empty when nothing says it.
func fixClass(d *Delivery) string {
	switch {
	case d.Plan != nil && d.Plan.Class != "":
		return d.Plan.Class
	case d.Plan != nil && len(d.Plan.Moved) == 0 && len(d.Plan.Dependents) == 0 && len(d.Plan.New) == 0:
		return classNone
	case d.Time != nil:
		return d.Time.Class
	}
	return ""
}

// classNone is the class of a delivery that moves no module: nothing for a walk to move, and no fix budget.
const classNone = "none"

// openedOf is when a delivery's pull request was first heard: the earliest head of it this owner holds, never
// one made from the forge's word at its merge; zero when none is held or it has no pull request. Under the lock.
func (h *Holder) openedOf(d *Delivery) time.Time {
	if d.Number <= 0 {
		return time.Time{}
	}
	var first time.Time
	for _, c := range h.deliveries {
		if c.FromForge || c.Number != d.Number || !strings.EqualFold(c.Repository, d.Repository) || c.Created.IsZero() {
			continue
		}
		if first.IsZero() || c.Created.Before(first) {
			first = c.Created
		}
	}
	if first.IsZero() && !d.FromForge {
		first = d.Created
	}
	return first
}

// noteLine is one transition as the commit's note keeps it.
func noteLine(d *Delivery, t Transition) string {
	by := ""
	if t.By != "" {
		by = " by " + t.By
	}
	line := fmt.Sprintf("%s delivery %s: %s -> %s (%s%s): %s", t.At.UTC().Format(time.RFC3339), d.ID, orNothing(t.From),
		t.To, t.Event, by, t.Why)
	if t.To == Proposed && d.Plan != nil {
		line += "; predicted: " + d.Plan.Summary
	}
	if d.Group != "" && (t.To == Proposed || t.To == Published) {
		line += "; group " + d.Group
	}
	return oneLine(line)
}

// executedLine is what a delivery did, written on the commit it landed as when it ends.
func executedLine(d *Delivery) string {
	var parts []string
	for _, s := range d.Steps {
		parts = append(parts, fmt.Sprintf("%s on %s %s", s.Module, s.Machine, s.State))
	}
	walk := ""
	if d.Walk != nil {
		walk = " (walk " + d.Walk.ID + ")"
	}
	return oneLine(fmt.Sprintf("%s delivery %s executed%s: %s", time.Now().UTC().Format(time.RFC3339), d.ID, walk,
		strings.Join(parts, "; ")))
}

func oneLine(s string) string { return strings.Join(strings.Fields(s), " ") }

// statusOf is the commit status `mesh/delivery`: where the delivery stands, in the forge's words.
func statusOf(d *Delivery) (string, string) {
	why := ""
	if len(d.Transitions) > 0 {
		why = d.Transitions[len(d.Transitions)-1].Why
	}
	switch d.State {
	case Proposed, Checked:
		return "pending", "checking"
	case Ready:
		return "success", "ready: it delivers once merged"
	case Rejected:
		return "failure", clip("rejected: "+why, 140)
	case Published:
		return "pending", "published: its walk waits for its turn"
	case Held:
		return "pending", clip("held for a person: "+d.HeldWhy, 140)
	case Delivering:
		passed := 0
		for _, s := range d.Steps {
			if s.State == StepPassed {
				passed++
			}
		}
		return "pending", fmt.Sprintf("delivering: %d machine step(s) passed", passed)
	case Delivered:
		return "success", "delivered"
	case Failed:
		if d.coveredStop() {
			return "success", clip(d.coveredStopSaid(), 140) // its change reached the machines (novox/hq issue 747)
		}
		return "failure", clip("failed: "+why, 140)
	case Superseded:
		return "warning", clip("superseded: "+why, 140)
	case Stopped:
		if d.coveredStop() {
			return "success", clip(d.coveredStopSaid(), 140) // its change reached the machines (novox/hq issue 480)
		}
		return "error", clip("stopped: "+why, 140)
	case Retired:
		return "success", "history: merged before this owner delivered anything, never walked"
	case Covered:
		by := d.CoveredBy
		if by == "" {
			by = d.Carrier
		}
		return "success", clip("covered: a later merge that contains it was delivered: "+by, 140)
	}
	return "pending", string(d.State)
}

// viewMarker is how the forge's holder finds the one view comment of a pull request.
const viewMarker = "<!-- mesh-delivery:view -->"

// ViewBody is a delivery's view on its pull request, written from the delivery whole.
func ViewBody(d *Delivery, g *Group) string {
	var b strings.Builder
	fmt.Fprintf(&b, "%s\n**Delivery** `%s` — **%s** since %s\n\n", viewMarker, d.ID, d.State, d.Since.UTC().Format(time.RFC3339))
	if d.coveredStop() {
		fmt.Fprintf(&b, "**Covered**: %s\n\n", d.coveredStopSaid())
	}
	if d.HeldWhy != "" && d.State == Held {
		fmt.Fprintf(&b, "Held for a person: %s — `mesh-delivery.release` with why lets it go on.\n\n", d.HeldWhy)
	}
	if d.Plan != nil {
		fmt.Fprintf(&b, "**Delivery plan** — %s\n", d.Plan.Summary)
		for i, tier := range d.Plan.Tiers {
			fmt.Fprintf(&b, "- tier %d: %s\n", i, strings.Join(tier, ", "))
		}
		for _, m := range d.Plan.Machines {
			var parts []string
			if len(m.Receives) > 0 {
				parts = append(parts, "receives "+strings.Join(m.Receives, ", "))
			}
			if len(m.Waits) > 0 {
				parts = append(parts, "waits for a person: "+strings.Join(m.Waits, ", "))
			}
			fmt.Fprintf(&b, "- %s: %s\n", m.Machine, strings.Join(parts, "; "))
		}
		for _, s := range d.Plan.Steps {
			fmt.Fprintf(&b, "- %s\n", s)
		}
		b.WriteString("\n")
	}
	if g != nil {
		fmt.Fprintf(&b, "**Group** `%s`: %s\n", g.ID, strings.Join(g.Members, ", "))
		if g.Check != nil && g.Check.Verdict != "" {
			fmt.Fprintf(&b, "- composed together: %s — %s\n", g.Check.Verdict, g.Check.Summary)
		}
		b.WriteString("\n")
	}
	if len(d.Steps) > 0 {
		b.WriteString("**Machines**\n")
		for _, s := range d.Steps {
			fmt.Fprintf(&b, "- %s on %s: %s", s.Module, s.Machine, s.State)
			if s.Why != "" {
				fmt.Fprintf(&b, " — %s", s.Why)
			}
			b.WriteString("\n")
		}
		b.WriteString("\n")
	}
	b.WriteString("**Transitions**\n")
	from := 0
	if len(d.Transitions) > 12 {
		from = len(d.Transitions) - 12
	}
	for _, t := range d.Transitions[from:] {
		fmt.Fprintf(&b, "- %s %s → %s (%s): %s\n", t.At.UTC().Format("2006-01-02 15:04"), orNothing(t.From), t.To, t.Event, t.Why)
	}
	b.WriteString("\nThe commit's note under `refs/notes/mesh-plan` keeps every transition: `git log --notes=mesh-plan`.\n")
	return b.String()
}

// Flush does what is owed, kept deliveries first: a delivery not yet kept says nothing.
func (h *Holder) Flush() {
	type job struct {
		id    string
		group bool
		e     Effect
		body  string
		repo  [2]string
		num   int
		url   string
	}
	h.mu.Lock()
	now := h.Now()
	for id := range h.dirty {
		if d := h.deliveries[id]; d != nil {
			h.keep(d)
		}
	}
	for id := range h.dirtyGroup {
		if g := h.groups[id]; g != nil {
			h.keepGroup(g)
		}
	}
	var jobs []job
	ids := make([]string, 0, len(h.deliveries))
	for id := range h.deliveries {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	for _, id := range ids {
		d := h.deliveries[id]
		if h.dirty[id] {
			continue
		}
		for _, e := range d.Owed {
			if e.Next != nil && now.Before(*e.Next) {
				continue
			}
			j := job{id: id, e: e, repo: [2]string{d.Owner(), d.Repo()}, num: d.Number, url: d.HTMLURL}
			if e.Kind == EffectView {
				j.body = ViewBody(d, h.groups[d.Group])
			}
			jobs = append(jobs, j)
		}
	}
	for id, g := range h.groups {
		if h.dirtyGroup[id] {
			continue
		}
		for _, e := range g.Owed {
			if e.Next != nil && now.Before(*e.Next) {
				continue
			}
			jobs = append(jobs, job{id: id, group: true, e: e})
		}
	}
	h.mu.Unlock()

	for _, j := range jobs {
		err := h.do(j.e, j.repo, j.num, j.body, j.url, j.group)
		h.mu.Lock()
		// Why it is given up, when it is: the forge's word that what it is for does not exist is final, and
		// so is a failure tried retryLimit times, the last ones retryCap apart.
		dropped := ""
		switch {
		case err != nil && forgeKinds[j.e.Kind] && doesNotExist(err):
			dropped = "the forge answers that what it is for does not exist"
		case err != nil && j.e.Tries+1 >= retryLimit:
			dropped = fmt.Sprintf("tried %d times since %s", j.e.Tries+1, j.e.Since.UTC().Format(time.RFC3339))
		}
		owed := func(list []Effect) []Effect {
			for i, o := range list {
				if o.Kind == j.e.Kind && o.Commit == j.e.Commit && o.Context == j.e.Context && o.Line == j.e.Line &&
					o.Since.Equal(j.e.Since) {
					if err == nil || dropped != "" {
						return append(list[:i:i], list[i+1:]...)
					}
					list[i].Tries++
					list[i].Last = err.Error()
					next := h.Now().Add(retryAfter(list[i].Tries))
					list[i].Next = &next
					return list
				}
			}
			return list
		}
		if j.group {
			if g := h.groups[j.id]; g != nil {
				g.Owed = owed(g.Owed)
				h.keepGroup(g)
			}
		} else if d := h.deliveries[j.id]; d != nil {
			d.Owed = owed(d.Owed)
			h.keep(d)
		}
		// Counted for delivery_status, so a window's writes can be judged against its tries (novox/hq issue 407).
		h.tried.last = h.Now()
		switch {
		case err == nil:
			h.tried.done++
		case dropped != "":
			h.tried.givenUp++
		default:
			h.tried.failed++
		}
		if dropped != "" {
			h.refuse(fmt.Errorf("%s: its owed %s given up — %s: %v", j.id, j.e.Kind, dropped, err))
		}
		h.mu.Unlock()
		if err != nil && dropped == "" && j.e.Tries%10 == 0 {
			h.Logf("[mesh-delivery] %s: %s not done yet (%v); tried again in %s", j.id, j.e.Kind, err,
				retryAfter(j.e.Tries+1))
		}
	}
	h.owedAgain()
}

// owedAgain says when what is still owed is tried again: an effect the forge or the bus could not take, at the time
// retryAfter gave it, and a delivery or group not yet kept, after retryFirst. Flush is the bounded retry of both.
func (h *Holder) owedAgain() {
	h.mu.Lock()
	defer h.mu.Unlock()
	for id := range h.dirty {
		if h.deliveries[id] != nil {
			h.lookAt(h.Now().Add(retryFirst), "a delivery not yet kept, kept again before it is said")
		}
	}
	for id := range h.dirtyGroup {
		if h.groups[id] != nil {
			h.lookAt(h.Now().Add(retryFirst), "a group not yet kept, kept again before it is said")
		}
	}
	for _, d := range h.deliveries {
		for _, e := range d.Owed {
			if e.Next != nil {
				h.lookAt(*e.Next, "an owed effect the forge or the bus could not take")
			}
		}
	}
	for _, g := range h.groups {
		for _, e := range g.Owed {
			if e.Next != nil {
				h.lookAt(*e.Next, "an owed effect the forge or the bus could not take")
			}
		}
	}
}

// An owed effect that fails is tried again later each time: retryFirst after the first failure, doubling up
// to retryCap, and given up after retryLimit tries — a little over two days — saying so (novox/hq issue 313).
const (
	retryFirst = 10 * time.Second
	retryCap   = time.Hour
	retryLimit = 60
)

// retryAfter is how long after its tries-th failure an effect is tried again.
func retryAfter(tries int) time.Duration {
	wait := retryFirst
	for i := 1; i < tries && wait < retryCap; i++ {
		wait *= 2
	}
	return min(wait, retryCap)
}

// forgeKinds are the effects done by the forge's holder.
var forgeKinds = map[string]bool{EffectNote: true, EffectView: true, EffectStatus: true}

// notThere is the forge's own answer that what was asked of does not exist: a pull request whose number the
// forge no longer holds, a commit it does not know. Tried again it answers the same, for ever.
var notThere = regexp.MustCompile(`(?i)Gitea API \S+: (404\b|\d{3}\b.*\bdoes not exist\b)`)

func doesNotExist(err error) bool { return err != nil && notThere.MatchString(err.Error()) }

// do is one owed effect.
func (h *Holder) do(e Effect, repo [2]string, number int, body, url string, group bool) error {
	switch e.Kind {
	case EffectEmit:
		var payload any
		if err := json.Unmarshal([]byte(e.Line), &payload); err != nil {
			return nil // nothing sayable: dropped
		}
		event := "transition"
		if group {
			event = "group"
		}
		if e.Event != "" {
			event = e.Event
		}
		return h.Emit(event, payload)
	case EffectNote:
		if h.Forge == nil {
			return fmt.Errorf("no forge")
		}
		return h.Forge.Note(repo[0], repo[1], e.Commit, e.Line)
	case EffectView:
		if h.Forge == nil || number == 0 {
			return nil
		}
		return h.Forge.View(repo[0], repo[1], number, body)
	case EffectStatus:
		if h.Forge == nil {
			return fmt.Errorf("no forge")
		}
		if group {
			// A group's status names its member's repository and page in Last.
			where, page, _ := strings.Cut(e.Last, " ")
			ownerRepo, _, _ := strings.Cut(where, "#")
			owner, name, _ := strings.Cut(ownerRepo, "/")
			return h.Forge.Status(owner, name, e.Commit, e.Context, e.State, e.Line, page)
		}
		return h.Forge.Status(repo[0], repo[1], e.Commit, e.Context, e.State, e.Line, url)
	}
	return nil
}