mesh-delivery/modules/mesh-delivery-forge-bridge/internal/groups/groups.go

// Package groups is what the forge bridge's process and its tools bundle share of a delivery group's end (novox/hq ADR
// 0633 §4, issue 1585): the keys of the bridge's state performed, and the asks of the state end-asks, through which the
// tool bridge_group_end_again asks the process to say a group's end again (the bridge serves no request subject, ADR 0437
// §4, so a tool asks in a state the process watches, as indexer_take_in asks the indexer in takeins).
package groups

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
	"git.novox.be/novox/mesh-delivery/sdk/trunk"
	"git.novox.be/novox/mesh-sdk/go/keys"
)

// The bridge's states the tools read and write.
const (
	// Module is the forge bridge's own name (its manifest's module, checked by the bridge's names test).
	Module = "mesh-delivery-forge-bridge"
	// PerformedState is the bridge's state of what it performed and published.
	PerformedState = "performed"
	// EndAsksState is the state of asks to say a group's end again, keyed by the group escaped as a key.
	EndAsksState = "end-asks"
	// EndAsksBucket is its bucket on the bus.
	EndAsksBucket = Module + "_" + EndAsksState
)

// Key is the key of performed that says a group's end is written: `groups.<group, escaped as a key>`, its value the last
// member's merge commit the end named.
func Key(group string) (string, error) {
	part, err := KeyOf(group)
	if err != nil {
		return "", err
	}
	return "groups." + part, nil
}

// KeyOf is a group's id (its branch) escaped as a key, as the states performed and end-asks key it.
func KeyOf(group string) (string, error) {
	if strings.TrimSpace(group) == "" {
		return "", errors.New("a group's id is empty")
	}
	return keys.EscapeName(group), nil
}

// PublishedKey is the key of performed that says a member's merged message is written: `published.<16 hex digits of the
// sha256 of <repository>#<number>>`, its value the merge commit the message named (a pull request merges once).
func PublishedKey(repository string, number int) string {
	sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d", strings.ToLower(repository), number)))
	return "published." + hex.EncodeToString(sum[:8])
}

// EndID is the message id of a group's end, and of one said again by hand at a time: the stream's duplicate window drops a
// second publish of the same id.
func EndID(group, lastMerge, again string) string {
	id := "group-end." + group + "." + lastMerge
	if again != "" {
		id += "." + again
	}
	return id
}

// AgainStamp is the time a hand ask appends to a group end's message id.
func AgainStamp(at time.Time) string { return at.UTC().Format("20060102T150405.000000000Z") }

// EndAsk is a value of the end-asks state: which group's end a person asked to be said again, why, who and when, and, once
// the bridge read it, its answer.
type EndAsk struct {
	V        int       `json:"v"`
	Group    string    `json:"group"`
	Why      string    `json:"why,omitempty"`
	By       string    `json:"by,omitempty"`
	At       time.Time `json:"at"`
	Written  bool      `json:"written,omitempty"`
	Answer   string    `json:"answer,omitempty"`
	Answered time.Time `json:"answered,omitzero"`
}

// Done says the ask is answered.
func (a EndAsk) Done() bool { return !a.Answered.IsZero() }

// NewEndAsk is the value bridge_group_end_again writes for a group: its key (the group escaped as a key) and the value.
func NewEndAsk(group, why, by string, at time.Time) (string, []byte, error) {
	group = strings.TrimSpace(group)
	if group == "" {
		return "", nil, errors.New("name the group whose end is said again (group)")
	}
	key, err := KeyOf(group)
	if err != nil {
		return "", nil, err
	}
	data, err := json.Marshal(EndAsk{V: 1, Group: group, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
	return key, data, err
}

// DecodeEndAsk reads an end-asks value.
func DecodeEndAsk(data []byte) (EndAsk, error) {
	var a EndAsk
	if err := json.Unmarshal(data, &a); err != nil {
		return a, err
	}
	if a.V != 1 || a.Group == "" {
		return a, errors.New("an end-asks value of no version 1 naming no group")
	}
	return a, nil
}

// Row is one group as bridge_groups answers it: the seat mesh-delivery's group and what the bridge kept of its end.
type Row struct {
	Key     string      `json:"key"`
	Group   string      `json:"group"`
	State   string      `json:"state,omitempty"`
	Members []RowMember `json:"members"`
	Ended   bool        `json:"ended"`
	// EndedWith is the last member's merge commit the end named, when it is written.
	EndedWith string `json:"ended-with,omitempty"`
	Open      bool   `json:"open"`
	Words     string `json:"words"`
	// LastEndAsk is the last ask to say the group's end again (bridge_group_end_again), its who, why and when, and the
	// bridge's answer once it read it; absent when none was asked. The state end-asks keeps the last per group.
	LastEndAsk *EndAsk `json:"last-end-ask,omitempty"`
}

// RowMember is one member of a group: its delivery state at the seat and, once its merged message is written, the merge
// commit it named.
type RowMember struct {
	Repository string `json:"repository"`
	Number     int    `json:"number"`
	State      string `json:"state"`
	Commit     string `json:"commit,omitempty"`
	Published  bool   `json:"published"`
}

// Rows are the groups the seat mesh-delivery's verb groups answers, each with what the bridge kept: ended (its end written,
// with the last merge it named) or open (some member's merged message is not written, or its end not written). published
// is performed's `published.` values by key (the merge commit), ended its `groups.` values by key; asks are the end-asks
// values by the same key, and one that does not read is named unread as `end-asks.<key>`. A group whose id cannot be a
// key is named unread by its id.
func Rows(of []seat.Group, published, ended map[string]string, asks map[string][]byte) (rows []Row, unread []string) {
	lastAsk := map[string]*EndAsk{}
	for key, raw := range asks {
		a, err := DecodeEndAsk(raw)
		if err != nil {
			unread = append(unread, EndAsksState+"."+key)
			continue
		}
		lastAsk[key] = &a
	}
	for _, g := range of {
		if len(g.Members) < 2 {
			continue
		}
		key, err := KeyOf(g.ID)
		if err != nil {
			unread = append(unread, "a group with no id")
			continue
		}
		r := Row{Key: key, Group: g.ID, State: g.State}
		allPublished := true
		for _, m := range g.Members {
			commit, pub := published[PublishedKey(m.Repository, int(m.Number))]
			r.Members = append(r.Members, RowMember{Repository: m.Repository, Number: int(m.Number), State: m.State, Commit: commit, Published: pub})
			allPublished = allPublished && pub
		}
		r.EndedWith, r.Ended = ended["groups."+key]
		r.LastEndAsk = lastAsk[key]
		r.Open = !r.Ended
		switch {
		case r.Ended:
			r.Words = "its end is written"
		case allPublished:
			r.Words = "every member's merged message is written, and its end is not written: bridge_group_end_again says it"
		default:
			r.Words = "gathered: not every member's merged message is written"
		}
		rows = append(rows, r)
	}
	sortRows(rows)
	sort.Strings(unread)
	return rows, unread
}

func sortRows(rows []Row) { sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key }) }

// MembersOf are a group's members with their heads, in the group's order.
func MembersOf(g seat.Group) []meshdeliveryclient.Head {
	out := make([]meshdeliveryclient.Head, 0, len(g.Members))
	for _, m := range g.Members {
		out = append(out, meshdeliveryclient.Head{Repository: m.Repository, Number: int(m.Number), Head: m.Head})
	}
	return out
}

// Named is the group of id among the groups the seat answered, when it has more than one member and names the pull
// request number of repository among them (a lone pull request is no group).
func Named(of []seat.Group, id, repository string, number int) (seat.Group, bool) {
	for _, g := range of {
		if g.ID != id || len(g.Members) < 2 {
			continue
		}
		if slices.ContainsFunc(g.Members, func(m seat.Member) bool {
			return trunk.SameRepository(m.Repository, repository) && int(m.Number) == number
		}) {
			return g, true
		}
	}
	return seat.Group{}, false
}

// Read is every delivery group the seat mesh-delivery's verb groups answers, the ended ones too (all), called through call
// by the verb's manifest key: the bridge reads a group from the seat, never from the gate's state (design 47 *A group's
// merges are taken in together*). An answer saying an error, or one the contract cannot type, is an error: a failed read
// is never read as no group.
func Read(ctx context.Context, call func(ctx context.Context, key string, body any) (json.RawMessage, error)) ([]seat.Group, error) {
	raw, err := call(ctx, "seat:"+seat.Seat+"."+seat.VerbGroups, seat.GroupsInput{All: seat.GroupsInputAllTrue})
	if err != nil {
		return nil, fmt.Errorf("the seat %s did not answer %s: %w", seat.Seat, seat.VerbGroups, err)
	}
	var a seat.GroupsAnswer
	if err := json.Unmarshal(raw, &a); err != nil {
		return nil, fmt.Errorf("the seat %s's answer of %s is not its contract's: %w", seat.Seat, seat.VerbGroups, err)
	}
	if a.Error != "" {
		return nil, fmt.Errorf("the seat %s refused %s: %s", seat.Seat, seat.VerbGroups, a.Error)
	}
	return a.Groups, nil
}