mesh-controller/cmd/mesh-controller/group_wait.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"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"
)

// **A delivery group still assembles** (novox/hq ADR 0306 decision 2): a merge whose delivery group has a member
// not yet merged waits for the merge window, whether or not a batch is assembling, so the group's members walk in
// one batch, in their order (ADR 0276 decision 5). Which groups exist and where each member stands is
// mesh-delivery's record (ADR 0239): a group is the pull requests of two repositories or more sharing a head
// branch name, which is its id. The controller asks it, through its `groups` verb, when a merge from a head is
// heard; a member whose pull request is open (proposed, checked, ready or rejected) is still to merge. When
// mesh-delivery does not answer within groupAskWithin, or answers something unreadable, the merge waits: the safe
// error is to wait, and its batch says why.

// groupAskWithin is how long mesh-delivery is given to say a merge's group, while the merge is being heard.
var groupAskWithin = 3 * time.Second

// openPullStates are the states of a delivery whose pull request is open and not merged (mesh-delivery's table):
// a member in one of them is still to merge. Closed unmerged is stopped, final, and waits for nothing.
var openPullStates = map[string]bool{"proposed": true, "checked": true, "ready": true, "rejected": true}

// groupMember is one member of a delivery group as mesh-delivery's `groups` says it: `<owner/repo>@<commit> (state)`.
type groupMember struct {
	Repository, State string
}

// groupsLine is one group as mesh-delivery's `groups` says it: its id (the head branch name), its members as text and,
// from a mesh-delivery that knows novox/hq ADR 0382, each member again as an object (`member_details`) naming its pull
// request's number, whole head, base branch and whether it is open.
type groupsLine struct {
	ID      string       `json:"id"`
	Members []string     `json:"members"`
	Details []memberLine `json:"member_details,omitempty"`
}

// memberLine is one member of a group, as `groups` says it beside its text.
type memberLine struct {
	ID         string `json:"id"`
	Repository string `json:"repository"`
	State      string `json:"state"`
	Number     int    `json:"number"`
	Head       string `json:"head"`
	Base       string `json:"base"`
	Open       bool   `json:"open"`
}

// groupsAsked is mesh-delivery's `groups`: every group it keeps, or an error when it cannot be asked or did not
// answer within groupAskWithin. A seam a test replaces.
var groupsAsked = func(ctx context.Context, held bool) ([]groupsLine, error) {
	if !held {
		return nil, fmt.Errorf("no holder of %s is on record to say it", catalogue.DeliverySeat)
	}
	// On the bus as this process is: the serving controller's own connection, lent, or — in a verb that runs as a command
	// of its own, such as a recheck's `delivery check` — one dialled for the moment (onTheBus).
	var raw json.RawMessage
	err := onTheBus(func(conn *nats.Conn) error {
		asking, cancel := context.WithTimeout(ctx, groupAskWithin)
		defer cancel()
		var err error
		raw, err = askDeliveryOwner(asking, conn, "groups", map[string]any{})
		return err
	})
	if err != nil {
		if errors.Is(err, link.ErrNothingServes) {
			return nil, fmt.Errorf("nothing serves %s.groups", catalogue.DeliverySeat)
		}
		return nil, err
	}
	var lines []groupsLine
	if err := json.Unmarshal(raw, &lines); err != nil {
		return nil, fmt.Errorf("%s.groups answered something unreadable: %w", catalogue.DeliverySeat, err)
	}
	return lines, nil
}

// deliveryGroupOf is what mesh-delivery says of the group a head names: its members, and whether the head names a
// group at all; an error when it cannot be asked or did not answer. A seam a test replaces.
var deliveryGroupOf = func(ctx context.Context, held bool, head string) ([]groupMember, bool, error) {
	lines, err := groupsAsked(ctx, held)
	if err != nil {
		return nil, false, err
	}
	for _, l := range lines {
		if l.ID != head {
			continue
		}
		var members []groupMember
		for _, s := range l.Members {
			members = append(members, memberOf(s))
		}
		return members, true, nil
	}
	return nil, false, nil
}

// memberOf reads one member as `groups` says it: `novox/mesh-sdk@d41d255f28ad (held)`.
func memberOf(s string) groupMember {
	id, state := strings.TrimSpace(s), ""
	if i := strings.LastIndex(id, " ("); i >= 0 && strings.HasSuffix(id, ")") {
		id, state = id[:i], id[i+2:len(id)-1]
	}
	repository, _, _ := strings.Cut(id, "@")
	return groupMember{Repository: repository, State: state}
}

// groupWait says whether a merge heard now waits for the window for its delivery group, and why in words; nothing
// when it names no head, its head names no group, or every other member is merged or closed. own is the merge's
// kind (ownPath).
//
// A member of another repository whose merge this controller already keeps in a batch not yet cut of the same
// kind is merged, whatever mesh-delivery says yet (it marks a member merged once its walk opened): the merge joins
// that batch by their head. One kept in a batch of the other kind keeps the merge waiting, whatever mesh-delivery
// says (review of #250): kinds never share a batch, so the merge's batch is a new one, and cut at once it would
// walk before that member's batch is cut and break the group's order (ADR 0276 decision 5). Waiting, the two
// batches are cut in the order their windows close, an own-path one first at one look, as cutBatchesHeld cuts
// every pair of kinds.
//
// Asked before the plans are held: a slow answer holds up no walk.
func groupWait(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, held, own bool) (string, error) {
	if m.Head == "" {
		return "", nil
	}
	same, other, err := keptFromHead(ctx, inv, m.Head, own)
	if err != nil {
		return "", err
	}
	if len(other) > 0 {
		return "group " + m.Head + " has a member waiting in a batch of the other kind: " + strings.Join(other, ", "), nil
	}
	members, isGroup, err := deliveryGroupOf(ctx, held, m.Head)
	if err != nil {
		return fmt.Sprintf("%s did not say whether group %s has a member still to merge (%v): it waits for the "+
			"window, the safe side", catalogue.DeliverySeat, m.Head, err), nil
	}
	if !isGroup {
		return "", nil
	}
	var open []string
	for _, g := range members {
		if strings.EqualFold(g.Repository, m.Owner+"/"+m.Repo) || same[strings.ToLower(g.Repository)] ||
			!openPullStates[g.State] {
			continue
		}
		open = append(open, g.Repository+" ("+g.State+")")
	}
	if len(open) == 0 {
		return "", nil
	}
	return "group " + m.Head + " has a member still to merge: " + strings.Join(open, ", "), nil
}

// keptFromHead are the merges from a head that batches not yet cut hold: the repositories (lower case) of those in
// a batch of the kind own says, and, in words, each of the other kind with its batch.
func keptFromHead(ctx context.Context, inv *inventory.Inventory, head string, own bool) (map[string]bool, []string, error) {
	batches, err := inv.Batches(ctx)
	if err != nil {
		return nil, nil, err
	}
	same := map[string]bool{}
	var other []string
	for _, b := range batches {
		merges, err := inv.MergesOf(ctx, b.ID)
		if err != nil {
			return nil, nil, err
		}
		for _, k := range merges {
			if announcedOf(k).Head != head {
				continue
			}
			if b.OwnPath() == own {
				same[strings.ToLower(k.Repository)] = true
			} else {
				other = append(other, fmt.Sprintf("%s@%s in %s", k.Repository, short(k.Commit), b.ID))
			}
		}
	}
	return same, other, nil
}