mesh-controller/cmd/mesh-controller/delivery_conditions.go

package main

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

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

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
)

// A delivery held past its bound, said by the controller and healed by H2 from mesh-delivery's own table
// (novox/hq ADR 0239 decision 9, to-be 47 Phase B). The conditions are the controller's, one owner: the
// self-check reads the delivery owner's `stalled` (probe D14) and raises `delivery.<id>.stalled`; healer
// H2 calls the owner's `close`, which takes only the transition the table names for that state, and only
// the next observation says whether it worked (ADR 0231).

// probeDeliveriesID is the probe that reads the delivery's owner.
const probeDeliveriesID = "D14"

// kindDeliveryStalled is what a delivery held past its bound raises: H2's kind, in the delivery scope.
const kindDeliveryStalled = "stalled"

// deliveryOwnerWithin is how long the owner is given to answer.
var deliveryOwnerWithin = 10 * time.Second

// stalledLine is one delivery past its bound, as mesh-delivery's `stalled` says it.
type stalledLine struct {
	ID string `json:"id"`
	// Number is the pull request's, for a line of a head the forge never announced (novox/hq issue 347).
	Number int    `json:"number,omitempty"`
	State  string `json:"state"`
	// Walk is the walk a line is about: a waiting walk no live delivery holds, or, asked with `walks`, each walk
	// waiting for the owner's word and what holds it (novox/hq issues 362, 529).
	Walk string `json:"walk,omitempty"`
	// Waiting are the merge checks a line of the state `unanswered` waits on, as mesh-delivery says each:
	// "mesh/check pending since <UTC time>", "mesh/check never set" (novox/hq issue 438).
	Waiting []string `json:"waiting,omitempty"`
	// Checks are the same merge checks as data, which the controller words in the operator's own time (novox/hq
	// issue 443): a time inside a finished sentence cannot be said again in another zone. A mesh-delivery from before
	// says none, and Waiting is said as it reads.
	Checks []waitingCheck `json:"checks,omitempty"`
	// Held is the state of the delivery of a line of the state `unanswered`, ready or rejected, when one is held
	// (novox/hq issues 441, 451): its check answered once, so a recheck is the step. Empty for a head with no
	// delivery, and from a mesh-delivery that does not say it yet; the words are never read for it.
	Held  string `json:"held,omitempty"`
	For   string `json:"for"`
	Bound string `json:"bound"`
	H2    string `json:"h2"`
	Says  string `json:"says"`
	// Severity is "urgent" for a group held or half merged and a refused member (novox/hq issues 1275, 1278); empty
	// or anything else is a warning, as before.
	Severity string `json:"severity,omitempty"`
	// Needs is "new-head" for a delivery whose cure is its author's new head (novox/hq ADR 0505 §5), Author that author
	// as the operator reads it ("the author of feat/x", with the session that pushed it when its pull request names one)
	// and Branch its branch. A mesh-delivery from before says none, and the words are as before.
	Needs  string `json:"needs,omitempty"`
	Author string `json:"author,omitempty"`
	Branch string `json:"branch,omitempty"`
}

// needsNewHead is what a stalled line needs when its cure is its author's new head (ADR 0505 §5).
const needsNewHead = "new-head"

// severity is the condition's severity a stalled line asks: urgent when it says so, a warning otherwise.
func (l stalledLine) severity() conditions.Severity {
	if l.Severity == string(conditions.Urgent) {
		return conditions.Urgent
	}
	return conditions.Warning
}

// waitingCheck is one merge check a stalled line waits on, as mesh-delivery gives it: its context, its state —
// "pending", or "never-set" — and, for a pending one, since when in RFC 3339, empty when the forge did not say.
type waitingCheck struct {
	Context string `json:"context"`
	State   string `json:"state"`
	Since   string `json:"since,omitempty"`
}

// wordsNow is the time the words are said at, which says whether a time needs its day; a seam a test replaces.
var wordsNow = time.Now

// wordsZone is the zone a stalled line's times are said in: the controller's local zone, as every other time in its
// messages. A seam a test replaces, so that no test writes time.Local, which every goroutine of the package reads.
var wordsZone = func() *time.Location { return time.Local }

// operatorsOnly is whether the table leaves H2 nothing to do for the line: the state is the operator's.
func (l stalledLine) operatorsOnly() bool { return l.H2 == "" || strings.HasPrefix(l.H2, "none") }

// askDeliveryOwner asks the holder of the mesh-delivery seat one of the verbs the controller is granted;
// a seam a test replaces. The holder's own refusal is an error naming it.
var askDeliveryOwner = func(ctx context.Context, conn *nats.Conn, verb string, args map[string]any) (json.RawMessage, error) {
	granted := false
	for _, v := range broker.VerbsTheControllerAsksTheDeliveryOwner {
		granted = granted || v.Verb == verb
	}
	if !granted {
		return nil, fmt.Errorf("the controller asks %s.%s, which its grant does not name", catalogue.DeliverySeat, verb)
	}
	if conn == nil {
		return nil, errors.New("this controller is not on the bus")
	}
	answer, err := link.AskMeshSeatTool(ctx, conn, catalogue.DeliverySeat, verb, args, deliveryOwnerWithin)
	if err != nil {
		return nil, err
	}
	if answer.Error != "" {
		return nil, fmt.Errorf("%s.%s refused: %s", catalogue.DeliverySeat, verb, answer.Error)
	}
	return unwrapToolResult(answer.Result), nil
}

// unwrapToolResult is a tool's answer whatever the runtime wrapped it in: the JSON itself, or the
// protocol's content list holding it as text.
func unwrapToolResult(raw json.RawMessage) json.RawMessage {
	var wrapped struct {
		Content []struct {
			Text string `json:"text"`
		} `json:"content"`
	}
	if json.Unmarshal(raw, &wrapped) == nil && len(wrapped.Content) > 0 && json.Valid([]byte(wrapped.Content[0].Text)) {
		return json.RawMessage(wrapped.Content[0].Text)
	}
	return raw
}

// deliveriesStalled is what the owner says is held past its bound; nothing when no holder is on record,
// or none answers (D3 says that one).
func deliveriesStalled(ctx context.Context, conn *nats.Conn, held bool) ([]stalledLine, error) {
	if !held {
		return nil, nil
	}
	raw, err := askDeliveryOwner(ctx, conn, "stalled", map[string]any{})
	if errors.Is(err, link.ErrNothingServes) {
		return nil, nil // the holder not running is D3's holder-silent, said once there
	}
	if err != nil {
		return nil, err
	}
	var lines []stalledLine
	if err := json.Unmarshal(raw, &lines); err != nil {
		return nil, fmt.Errorf("%s.stalled answered something unreadable: %w", catalogue.DeliverySeat, err)
	}
	return lines, nil
}

// walkWaits are, by walk, the lines mesh-delivery says of a walk waiting for its word: what holds it.
type walkWaits map[string][]stalledLine

// walksWaiting asks the delivery's owner what holds each walk waiting for its word (novox/hq issue 529): its
// `stalled` asked with `walks`, so a line of `plans` names a held delivery, its reason and the verb that releases
// it, instead of pointing at a hand start. Nil when no holder is on record or none answers: the line then reads as
// before. An owner from before this answers its stalled lines, of which only those naming a walk are kept.
func walksWaiting(ctx context.Context, conn *nats.Conn, held bool) walkWaits {
	if !held {
		return nil
	}
	raw, err := askDeliveryOwner(ctx, conn, "stalled", map[string]any{"walks": true})
	if err != nil {
		return nil
	}
	var lines []stalledLine
	if json.Unmarshal(raw, &lines) != nil {
		return nil
	}
	out := walkWaits{}
	for _, l := range lines {
		if l.Walk != "" {
			out[l.Walk] = append(out[l.Walk], l)
		}
	}
	return out
}

// stalledObservations are the conditions of what is stalled.
func stalledObservations(lines []stalledLine) []conditions.Observation {
	out := make([]conditions.Observation, 0, len(lines))
	for _, l := range lines {
		o := conditions.Observation{Scope: conditions.ScopeDelivery, ID: l.ID, Kind: kindDeliveryStalled,
			Severity: l.severity(),
			Summary: fmt.Sprintf("the delivery %s has been %s for %s, past its bound of %s (%s): healer H2 may %s — "+
				"`mesh-delivery.show %s`", l.ID, l.State, l.For, l.Bound, l.Says, l.H2, l.ID),
			Said: fmt.Sprintf("%s for %s", l.State, l.For)}
		if l.operatorsOnly() {
			o.Resolver = conditions.ResolverOperator
		}
		o.Headline, o.Explanation, o.Resolved, o.Needs, o.Actions = stalledWords(l, o)
		out = append(out, o)
	}
	return out
}

// probeDeliveries is D14: every delivery held past its state's bound is said.
func probeDeliveries(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
	entries, err := d.open.inventory.Catalogued(ctx)
	if err != nil {
		return nil, err
	}
	var conn *nats.Conn
	if d.js != nil {
		conn = d.js.Conn()
	}
	lines, err := deliveriesStalled(ctx, conn, deliverySeatHeld(entries))
	if err != nil {
		return nil, err
	}
	return stalledObservations(lines), nil
}

// --- H2, for a delivery ---------------------------------------------------------------------------------

// connOf is the bus a healer asks over.
func (h *healing) connOf() *nats.Conn {
	if h.js == nil {
		return nil
	}
	return h.js.Conn()
}

// appliesToAStalledDelivery is H2's for a delivery: the owner still lists it, with a transition the table
// lets H2 take; never one whose state is the operator's.
func appliesToAStalledDelivery(ctx context.Context, h *healing, c conditions.Condition) (string, bool, string, error) {
	lines, err := deliveriesStalled(ctx, h.connOf(), true)
	if err != nil {
		return "", false, "", err
	}
	for _, l := range lines {
		if l.ID != c.Subject.ID {
			continue
		}
		if l.operatorsOnly() {
			return "", false, "the delivery is " + l.State + ", a state the table leaves to the operator", nil
		}
		return c.Key, true, "", nil
	}
	return "", false, "the delivery's owner no longer lists it as stalled: its condition clears on the next look", nil
}

// repairDelivery is H2 for a delivery: the owner's `close`, which reads the walk again and takes only the
// transition its table names. A refusal is no repair, said; the next observation says whether it worked.
func repairDelivery(ctx context.Context, h *healing, c conditions.Condition) (string, string, error) {
	raw, err := askDeliveryOwner(ctx, h.connOf(), "close", map[string]any{"id": c.Subject.ID, "why": c.Key})
	if err != nil {
		if errors.Is(err, link.ErrNothingServes) {
			return "", "", err
		}
		return "closed nothing", oneLine(err.Error()), nil
	}
	var said string
	if json.Unmarshal(raw, &said) != nil {
		said = string(raw)
	}
	return "asked " + catalogue.DeliverySeat + " to close " + c.Subject.ID, said, nil
}