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

package main

import (
	"context"
	"errors"
	"log"
	"slices"

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

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	groupkeys "mesh-delivery-forge-bridge/internal/groups"
)

// A delivery group's merges reach the indexer together (novox/hq ADR 0633 §4): the bridge sets the group on each
// member's merged message, from the gate's merges state it reads (ADR 0437 §8), and once every approved member is
// merged and its merged message written on FORGE it says the group's end, forge.group.merged, once.
//
// What it keeps of it is in its state performed, beside the gate's keys it acted on: `published.<hash>` for each
// member's merged message written (the hash is the first 16 hex digits of the sha256 of `<repository>#<number>@<merge>`),
// and `groups.<merges key>` once the group's end is written, its value the revision of the gate's merges key it was
// worked out from. A group whose end is not written yet is open (bridge_groups reads them).

// merges reads one key of the gate's merges state.
type mergesRead func(ctx context.Context, key string) (value []byte, rev uint64, found bool, err error)

// grouper sets a member's group and says a group's end.
type grouper struct {
	merges mergesRead
	send   func(ctx context.Context, subject string, data []byte, id string) error // writes on FORGE with the message id
	done   revisions
	logger *log.Logger
}

// sendVia writes on FORGE through pub, with the message id.
func sendVia(pub publisher) func(context.Context, string, []byte, string) error {
	return func(ctx context.Context, subject string, data []byte, id string) error {
		_, err := pub.Publish(ctx, subject, data, jetstream.WithMsgID(id))
		return err
	}
}

func kvMerges(kv jetstream.KeyValue) mergesRead {
	return func(ctx context.Context, key string) ([]byte, uint64, bool, error) {
		e, err := kv.Get(ctx, key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			return nil, 0, false, nil
		}
		if err != nil {
			return nil, 0, false, err
		}
		return e.Value(), e.Revision(), true, nil
	}
}

// groupOf is the gate's merge of the group a merged message's branch names, when the message's pull request is one of
// its approved members and the group has more than one: a lone pull request is no group.
func (g *grouper) groupOf(ctx context.Context, f forgemsg.Forge) (delivery.Merge, uint64, bool) {
	if g == nil || g.merges == nil || f.Branch == "" {
		return delivery.Merge{}, 0, false
	}
	key, err := delivery.MergesKey(f.Branch)
	if err != nil {
		return delivery.Merge{}, 0, false
	}
	value, rev, found, err := g.merges(ctx, key)
	if err != nil {
		g.logger.Printf("%s#%d: the gate's merges.%s cannot be read, so its merged message names no group: %v", f.Repository, f.Number, key, err)
		return delivery.Merge{}, 0, false
	}
	if !found {
		return delivery.Merge{}, 0, false
	}
	m, err := delivery.DecodeMerge(value)
	if err != nil {
		g.logger.Printf("%s#%d: the gate's merges.%s is not a merge, so its merged message names no group: %v", f.Repository, f.Number, key, err)
		return delivery.Merge{}, 0, false
	}
	if _, ok := m.Approved[delivery.ApprovedKey(f.Repository, f.Number)]; !ok || len(m.Approved) < 2 {
		return delivery.Merge{}, 0, false
	}
	return m, rev, true
}

// member sets the group and its approved members on a merged message's body (ADR 0633 §4); any other body, and a
// merge of no group, is left as it is.
func (g *grouper) member(ctx context.Context, data []byte) ([]byte, forgemsg.Forge, bool) {
	f, err := forgemsg.DecodeForge(data)
	if err != nil || f.Kind != forgemsg.ForgeMerged {
		return data, f, false
	}
	m, _, ok := g.groupOf(ctx, f)
	if !ok {
		return data, f, false
	}
	f.Group, f.Members = m.Group, groupkeys.MembersOf(m)
	out, err := forgemsg.EncodeForge(f)
	if err != nil {
		g.logger.Printf("%s#%d: its merged message cannot name the group %q, written without: %v", f.Repository, f.Number, m.Group, err)
		return data, f, false
	}
	return out, f, true
}

// published keeps that a member's merged message is written, and says the group's end when it was the last.
func (g *grouper) published(ctx context.Context, f forgemsg.Forge) {
	if err := g.done.Put(ctx, groupkeys.PublishedKey(f.Repository, f.Number, f.Merge), 1); err != nil {
		g.logger.Printf("%s#%d: that its merged message is written is not kept, so the group's end waits for the next act: %v", f.Repository, f.Number, err)
		return
	}
	key, err := delivery.MergesKey(f.Branch)
	if err != nil {
		return
	}
	value, rev, found, err := g.merges(ctx, key)
	if err != nil || !found {
		return
	}
	g.ended(ctx, key, rev, value)
}

// ended says the group's end (forge.group.merged) once every approved member of the gate's merge is merged and its
// merged message written, after the last of them; never on a refusal (ADR 0441 §3: a held group merges on from it).
// It is written once per group: `groups.<key>` in performed says it was. again says it is written again by hand
// (bridge_group_end_again), with the time appended to the message id, so the duplicate window does not drop it.
func (g *grouper) ended(ctx context.Context, key string, rev uint64, value []byte) {
	g.end(ctx, key, rev, value, "")
}

func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, again string) bool {
	m, err := delivery.DecodeMerge(value)
	if err != nil || len(m.Approved) < 2 {
		return false
	}
	if again == "" {
		if _, seen, err := g.done.Get(ctx, groupkeys.Key(key)); err == nil && seen {
			return false
		}
	}
	end := forgemsg.ForgeGroupEnd{Group: m.Group}
	for _, member := range groupkeys.MembersOf(m) {
		at := slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool {
			return delivery.SameRepository(p.Repository, member.Repository) && p.Number == member.Number
		})
		if at < 0 || m.Pulls[at].Commit == "" {
			return false // not every member is merged (a merge commit is written only for a merged pull): the group stays gathered
		}
		if _, ok, err := g.done.Get(ctx, groupkeys.PublishedKey(member.Repository, member.Number, m.Pulls[at].Commit)); err != nil || !ok {
			return false // its merged message is not written yet: said when it is (published)
		}
		end.Members = append(end.Members, forgemsg.GroupMerge{Repository: member.Repository, Number: member.Number, Merge: m.Pulls[at].Commit})
	}
	// The members in the order the gate merged them, so the last commit names the end.
	slices.SortStableFunc(end.Members, func(a, b forgemsg.GroupMerge) int {
		return slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { return p.Commit == a.Merge }) -
			slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { return p.Commit == b.Merge })
	})
	data, err := forgemsg.EncodeForgeGroupEnd(end)
	if err != nil {
		g.logger.Printf("the end of the group %q cannot be written: %v", m.Group, err)
		return false
	}
	id := groupkeys.EndID(m.Group, end.Members[len(end.Members)-1].Merge, "")
	if again != "" {
		id += "." + again
	}
	if err := g.send(ctx, forgeSubject(forgemsg.ForgeGroupEndEvent), data, id); err != nil {
		g.logger.Printf("the end of the group %q is not written on FORGE, written at the gate's next write or by bridge_group_end_again: %v", m.Group, err)
		return false
	}
	if err := g.done.Put(ctx, groupkeys.Key(key), rev); err != nil {
		g.logger.Printf("the end of the group %q is written and not kept: the stream's duplicate window drops a second one: %v", m.Group, err)
	}
	g.logger.Printf("the end of the group %q is written: %d members merged, the last %s", m.Group, len(end.Members), end.Members[len(end.Members)-1].Merge)
	return true
}