mesh-controller/cmd/mesh-controller/kept_helpers.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/micro"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
	"regexp"
	"strings"
	"time"
)

// Small helpers the plan machinery once held and other code still reads.

// silentCallBound is how long a call to the bus may take before it is called unanswered.
var silentCallBound = 20 * time.Second

// clipWords keeps a plain sentence within a bound, at a word.
func clipWords(s string, n int) string {
	if len(s) <= n {
		return s
	}
	cut := strings.LastIndex(s[:n-1], " ")
	if cut <= 0 {
		cut = n - 1
	}
	return s[:cut] + "…"
}

// orWords is s, or the words for when it is empty.
func orWords(s, none string) string {
	if s == "" {
		return none
	}
	return s
}

func boundedCall[T any](ctx context.Context, what string, call func(context.Context) (T, error)) (T, error) {
	ctx, cancel := context.WithTimeout(ctx, silentCallBound)
	defer cancel()
	type answer struct {
		v   T
		err error
	}
	done := make(chan answer, 1)
	go func() { v, err := call(ctx); done <- answer{v, err} }()
	select {
	case a := <-done:
		return a.v, a.err
	case <-ctx.Done():
		var zero T
		return zero, fmt.Errorf("%s: no answer from the bus within %s: %w", what, silentCallBound, ctx.Err())
	}
}

func sameCommitID(a, b string) bool {
	if a == "" || b == "" {
		return false
	}
	if a == b {
		return true
	}
	n := min(len(a), len(b))
	return n >= 7 && a[:n] == b[:n]
}

// 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 }

// 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

var (
	planIDWord = regexp.MustCompile(`\bplan-[0-9]+\b`)
	walkWd     = regexp.MustCompile(`\bwalks\b`)
	walkOne    = regexp.MustCompile(`\bwalk\b`)
)

// plainRefusalOr is plainRefusal with the sentence to say when the refusal's own words cannot be said plainly.
func plainRefusalOr(refusal, fallback string) string {
	r := planIDWord.ReplaceAllString(refusal, "the plan")
	r = walkWd.ReplaceAllString(r, "deliveries")
	r = walkOne.ReplaceAllString(r, "delivery")
	r = strings.TrimSpace(r)
	if _, ok := conditions.Plain(r); !ok {
		return fallback
	}
	return r
}

// 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
}

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

// 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
}

// servedOnTheBus asks the bus's discovery what every machine's node tools answer and which modules'
// tools are served where. A variable so a test needs no runtime.
var servedOnTheBus = func(ctx context.Context, conn *nats.Conn) (map[string]served, error) {
	inbox := conn.NewRespInbox()
	sub, err := conn.SubscribeSync(inbox)
	if err != nil {
		return nil, err
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := conn.PublishRequest("$SRV.INFO", inbox, nil); err != nil {
		return nil, fmt.Errorf("asking the bus who serves what: %w", err)
	}
	out := map[string]served{}
	add := func(node string) served {
		s, ok := out[node]
		if !ok {
			s = served{tools: map[string]bool{}}
		}
		return s
	}
	deadline := time.Now().Add(discoveryPatience)
	for time.Now().Before(deadline) {
		wait, cancel := context.WithTimeout(ctx, discoveryQuiet)
		msg, err := sub.NextMsgWithContext(wait)
		cancel()
		if err != nil {
			if ctx.Err() != nil {
				return nil, ctx.Err()
			}
			break
		}
		var info micro.Info
		if json.Unmarshal(msg.Data, &info) != nil {
			continue
		}
		if info.Name == broker.RuntimeModule {
			node := info.Metadata["node"]
			if node == "" {
				node = info.ID
			}
			s := add(node)
			s.runtime = true
			out[node] = s
		}
		for _, e := range info.Endpoints {
			if e.Metadata["kind"] != "tool" || e.Metadata["module"] == "" {
				continue
			}
			node := e.Metadata["node"]
			if node == "" {
				node = info.ID
			}
			s := add(node)
			s.tools[e.Metadata["module"]] = true
			out[node] = s
		}
	}
	return out, nil
}

// served is what one machine's node tools answered the bus's discovery with.
type served struct {
	runtime bool
	tools   map[string]bool
}

// 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 := readForPlanning(ctx, inv)
	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
}

// refusedWord is the word a build's failure says the reader did not know; empty when it says none.
func refusedWord(failed string) string {
	if m := unknownWord.FindStringSubmatch(failed); m != nil {
		return m[1]
	}
	return ""
}

// unknownWord is how the strict reading of a manifest says a word is not one it knows: encoding/json's own words, which
// the build-agent's answer carries whole. The build's answer has no typed field for it (Failed is the builder's words).
var unknownWord = regexp.MustCompile(`json: unknown field "([^"]+)"`)