260 lines
10 KiB
Go
260 lines
10 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// The operator's Approve of a delivery group registers a moved module from its new repository (novox/hq ADR 0420
|
|
// §7, issue 1025). ADR 0266 §44 lets a build's outcome register a known module only from the repository the
|
|
// catalogue already builds it from. The one exception was a build asked at the controller's terminal. This is a
|
|
// second, for a move: mesh-delivery announces each merge it makes with the ask whose Approve it merged on; the
|
|
// controller keeps the announcement, and when the build of a merge of a known module comes from another repository
|
|
// it reads that ask's warrant from the router and accepts only the operator's Approve of the merged head. A verb
|
|
// alone, or a merge with no such warrant, never moves a module.
|
|
|
|
// theDelivery is the module that merges, and whose asks these are.
|
|
const theDelivery = "mesh-delivery"
|
|
|
|
// optionApprove is the option of mesh-delivery's ask that merges (its OptionApprove).
|
|
const optionApprove = "approve"
|
|
|
|
// deliveryMerge is mesh-delivery's announcement of a merge it made: the pull request and the head it merged, the
|
|
// group it belongs to, and the ask whose answer it merged on with the option chosen and the heads that answer was
|
|
// bound to (delivery id → commit).
|
|
type deliveryMerge struct {
|
|
Owner string `json:"owner"`
|
|
Repo string `json:"repo"`
|
|
Number int `json:"number"`
|
|
HeadSHA string `json:"head_sha"`
|
|
Group string `json:"group,omitempty"`
|
|
AskID string `json:"ask_id"`
|
|
Option string `json:"option"`
|
|
Heads map[string]string `json:"heads"`
|
|
}
|
|
|
|
// parseDeliveryMerge reads an announcement, refusing one that names no pull request, head or ask.
|
|
func parseDeliveryMerge(body []byte) (deliveryMerge, error) {
|
|
var d deliveryMerge
|
|
if err := json.Unmarshal(body, &d); err != nil {
|
|
return d, fmt.Errorf("mesh-delivery's word on a merge cannot be read: %w", err)
|
|
}
|
|
if d.Owner == "" || d.Repo == "" || d.Number <= 0 || d.HeadSHA == "" || d.AskID == "" || d.Option == "" {
|
|
return d, errors.New("mesh-delivery's word on a merge names no pull request, head, ask or option")
|
|
}
|
|
return d, nil
|
|
}
|
|
|
|
// deliveryMergeKey is where a merge is kept: bus keys allow no `#` or `@`, so the pull request and its head are
|
|
// path-like tokens, as the forge spells the repository.
|
|
func deliveryMergeKey(owner, repo string, number int, head string) string {
|
|
return fmt.Sprintf("%s/%s/%d/%s", owner, repo, number, head)
|
|
}
|
|
|
|
// mergeAct is what mesh-delivery's answer binds, spelled exactly as mesh-delivery spells it (its approval.go:
|
|
// mergeAct and headsString): this ask, the option and the heads it covers.
|
|
func mergeAct(ask, option string, heads map[string]string) asks.Act {
|
|
var out []string
|
|
for id, sha := range heads {
|
|
out = append(out, id+"="+sha)
|
|
}
|
|
sort.Strings(out)
|
|
return asks.Act{"verb": theDelivery + ".merge", "ask": ask, "option": option, "heads": strings.Join(out, ",")}
|
|
}
|
|
|
|
// movedOnApprove judges the warrant of the ask a merge was made on: it holds only if it is the operator's Approve,
|
|
// proved by the channel it came through, of this very ask as it was published (digest), bound to the heads the
|
|
// announcement names, and those heads hold the head merged. Answers why it holds, or why not.
|
|
func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, error) {
|
|
switch {
|
|
case ask == nil || w == nil:
|
|
return "", errors.New("the router holds no closed record of the ask the merge was made on")
|
|
case ask.Who != asks.Operator:
|
|
return "", fmt.Errorf("the ask %s is for %q to answer, not the operator", ask.ID, ask.Who)
|
|
case d.Option != optionApprove:
|
|
return "", fmt.Errorf("the merge was made on the option %q, not an Approve", d.Option)
|
|
}
|
|
option, err := w.For(theDelivery, *ask)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if option.ID != optionApprove || w.Option != optionApprove || d.Option != option.ID {
|
|
return "", fmt.Errorf("the operator chose %q on the ask %s, not an Approve", w.Option, w.Ask)
|
|
}
|
|
if w.By == nil || w.By.Who != asks.Operator || w.By.Verified == "" {
|
|
return "", fmt.Errorf("the ask %s was not answered by the operator on a channel that proves who answered", w.Ask)
|
|
}
|
|
if err := option.Performs(mergeAct(d.AskID, option.ID, d.Heads)); err != nil {
|
|
return "", err
|
|
}
|
|
named := false
|
|
for _, sha := range d.Heads {
|
|
named = named || sha == d.HeadSHA
|
|
}
|
|
if !named {
|
|
return "", fmt.Errorf("the Approve covers %v, and the head merged, %.8s, is not among them", d.Heads, d.HeadSHA)
|
|
}
|
|
return w.Says() + " (ask " + w.Ask + ")", nil
|
|
}
|
|
|
|
// ---- keeping what mesh-delivery announced ----------------------------------------------------------------------
|
|
|
|
// keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them.
|
|
type keepingDeliveryMerges struct {
|
|
js func() *broker.JetStream
|
|
}
|
|
|
|
func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error {
|
|
d, err := parseDeliveryMerge(body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
js := k.js()
|
|
if js == nil {
|
|
return fmt.Errorf("%w: the bus is not held", link.ErrTryAgain)
|
|
}
|
|
api, err := jetstream.New(js.Conn())
|
|
if err != nil {
|
|
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
|
|
}
|
|
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err)
|
|
}
|
|
if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
|
|
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// keptDeliveryMerge is the merge mesh-delivery announced for a pull request's head; false when none was kept.
|
|
// A variable so a test can hand it one.
|
|
var keptDeliveryMerge = func(ctx context.Context, owner, repo string, number int, head string) (deliveryMerge, bool, error) {
|
|
serving := servingBus.Load()
|
|
if serving == nil {
|
|
return deliveryMerge{}, false, errors.New("only the serving controller keeps what mesh-delivery announced")
|
|
}
|
|
api, err := jetstream.New(serving.Conn())
|
|
if err != nil {
|
|
return deliveryMerge{}, false, err
|
|
}
|
|
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
|
|
if err != nil {
|
|
return deliveryMerge{}, false, err
|
|
}
|
|
e, err := kv.Get(ctx, deliveryMergeKey(owner, repo, number, head))
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return deliveryMerge{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return deliveryMerge{}, false, err
|
|
}
|
|
d, err := parseDeliveryMerge(e.Value())
|
|
return d, err == nil, err
|
|
}
|
|
|
|
// routerAskOf is the router's record of one of mesh-delivery's asks: the ask as published and its warrant. A
|
|
// variable so a test can hand it one.
|
|
var routerAskOf = func(ctx context.Context, inv *inventory.Inventory, id string) (*asks.Ask, *asks.Warrant, error) {
|
|
serving := servingBus.Load()
|
|
if serving == nil {
|
|
return nil, nil, errors.New("only the serving controller reads the router's record")
|
|
}
|
|
bucket, err := asksRecords(ctx, inv)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if bucket == "" {
|
|
return nil, nil, errors.New("no module declares the operator channel's records")
|
|
}
|
|
_, ask, w, err := readRouterEntry(ctx, serving.Conn(), bucket, theDelivery, id)
|
|
return ask, w, err
|
|
}
|
|
|
|
// readRouterEntry reads the router's record of one asker's ask: its state, the ask as published, and its warrant;
|
|
// "" when there is none. The controller's grant reaches the JetStream API whole (`$JS.API.>`), so it reads any
|
|
// asker's record.
|
|
func readRouterEntry(ctx context.Context, conn *nats.Conn, bucket, asker, ask string) (string, *asks.Ask, *asks.Warrant, error) {
|
|
reply, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+asker+"."+ask, nil)
|
|
if err != nil {
|
|
return "", nil, nil, err
|
|
}
|
|
if status := reply.Header.Get("Status"); status != "" {
|
|
if status == "404" {
|
|
return "", nil, nil, nil
|
|
}
|
|
return "", nil, nil, fmt.Errorf("the router's record could not be read: %s %s", status, reply.Header.Get("Description"))
|
|
}
|
|
var rec struct {
|
|
State string `json:"state"`
|
|
Ask *asks.Ask `json:"ask"`
|
|
Warrant *asks.Warrant `json:"warrant"`
|
|
}
|
|
if err := json.Unmarshal(reply.Data, &rec); err != nil {
|
|
return "", nil, nil, fmt.Errorf("the router's record of %s's ask %s cannot be read: %w", asker, ask, err)
|
|
}
|
|
return rec.State, rec.Ask, rec.Warrant, nil
|
|
}
|
|
|
|
// errNotAMerge says a build was not asked for a merge, so there is no move to look for.
|
|
var errNotAMerge = errors.New("the build is not of a merge")
|
|
|
|
// approvedMove says whether a build of a known module from another repository is the move the operator approved
|
|
// (ADR 0420 §7): the build is of a merge, that merge is a pull request mesh-delivery announced it made, and the
|
|
// ask it made it on holds an Approve by the operator of that head. Answers why it was accepted; an error says why
|
|
// it was not, and the refusal of ADR 0266 stands.
|
|
func approvedMove(ctx context.Context, inv *inventory.Inventory, forms *sourceForms, buildID string,
|
|
built inventory.Source) (string, error) {
|
|
r, found, err := inv.BuildRequestByID(ctx, buildID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if !found || r.For != "merge" || r.Commit == "" {
|
|
return "", errNotAMerge
|
|
}
|
|
owner, name, onForge := forms.onTheForge(built.Repository, built.Seat)
|
|
if !onForge {
|
|
return "", errors.New("the repository is not on the mesh's forge")
|
|
}
|
|
merged, known, err := inv.MergeOf(ctx, owner+"/"+name, r.Commit)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if !known {
|
|
return "", fmt.Errorf("the controller kept no merge of %s/%s at %.8s", owner, name, r.Commit)
|
|
}
|
|
k := announcedOf(merged)
|
|
if k.Number <= 0 || k.HeadSHA == "" {
|
|
return "", errors.New("the forge's announcement of the merge names no pull request or head")
|
|
}
|
|
d, kept, err := keptDeliveryMerge(ctx, owner, name, k.Number, k.HeadSHA)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if !kept {
|
|
return "", fmt.Errorf("mesh-delivery announced no merge of %s/%s#%d at %.8s", owner, name, k.Number, k.HeadSHA)
|
|
}
|
|
ask, w, err := routerAskOf(ctx, inv, d.AskID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
why, err := movedOnApprove(d, ask, w)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return fmt.Sprintf("%s/%s#%d merged at %.8s on %s", owner, name, k.Number, k.HeadSHA, why), nil
|
|
}
|