mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/branchend.go

package main

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

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/health"

	"mesh-delivery-builder/internal/build"
	"mesh-delivery-builder/internal/versions"
)

// forgeStream is the forge bridge's stream `forge` as the bus names it, where the builder's durable consumer hears a
// branch's end (design 51 *Tagged versions*).
const forgeStream = "FORGE"

// forgeBranchDeleted is the forge bridge's kind for a branch's end (delivery.ForgeBranchDeleted in the group's mesh-sdk).
const forgeBranchDeleted = "branch-deleted"

// checkerAsksStream is the checker's family `asks` as the bus names it, and tagWantedSubject its event tag-wanted
// (design 51 *Tagged versions*, DLV-C3): on the checker's own stream, so its subject is mesh.stream.<module>.event.*.
const (
	checkerAsksStream = "ASKS"
	tagWantedSubject  = "mesh.stream." + delivery.CheckerModule + ".event.tag-wanted"
)

// forgeEnd is the part of a forge message the builder reads.
type forgeEnd struct {
	Kind       string `json:"kind"`
	Repository string `json:"repository"`
	Branch     string `json:"branch"`
	Action     string `json:"action"`
}

// ended says whether the message ends a branch: merged, a pull closed, or branch-deleted.
func (f forgeEnd) ended() bool {
	switch f.Kind {
	case delivery.ForgeMerged, forgeBranchDeleted:
		return f.Branch != ""
	case delivery.ForgePull:
		return f.Action == delivery.PullClosed && f.Branch != ""
	}
	return false
}

// branchEnds hears the forge bridge's stream: every message wakes a floor whose bridge_tags was silent, and a branch's
// end removes its tagged versions, acknowledged only once all are removed.
type branchEnds struct {
	acts     versions.Acts
	findings *findings
	tagAsks  func() jetstream.KeyValue
	wake     func(ctx context.Context)
	logf     func(string, ...any)
	now      func() time.Time
}

func (h *branchEnds) heard(ctx context.Context, m jetstream.Msg) {
	if h.wake != nil {
		h.wake(ctx)
	}
	var f forgeEnd
	if err := json.Unmarshal(m.Data(), &f); err != nil || !f.ended() {
		h.ack(m)
		return
	}
	end, err := h.acts.BranchEnded(ctx, f.Repository, f.Branch)
	if err != nil {
		h.logf("the end of %s %s: the versions cannot be read, heard again: %v", f.Repository, f.Branch, err)
		h.again(m)
		return
	}
	id := f.Repository + "/" + f.Branch
	if end.Done {
		h.findings.clear(ctx, versions.RemovalNotDone, health.ScopeModule, id)
		if kv := h.tagAsks(); kv != nil {
			if err := kv.Delete(ctx, versions.TagAskKey(f.Repository, f.Branch)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
				h.logf("the tag ask of %s %s stays in %s: %v", f.Repository, f.Branch, versions.TagAsksState, err)
			}
		}
		h.ack(m)
		return
	}
	since := h.now()
	if md, err := m.Metadata(); err == nil {
		since = md.Timestamp
	}
	if h.now().Sub(since) >= versions.RemovalFindingAfter {
		h.findings.raise(ctx, finding{kind: versions.RemovalNotDone, scope: health.ScopeModule, id: id,
			summary:     end.Finding(f.Repository, f.Branch, since),
			headline:    "A branch's tagged versions are not removed",
			explanation: "The branch ended, and the registry seats did not say removed for every version tagged for it within 15 minutes.",
			resolved:    "Every seat said removed.",
			needs:       "Look at the seats named: " + strings.Join(end.Seats(), ", ")})
	}
	h.logf("the end of %s %s: remove asked of %d tagged versions, heard again in a minute", f.Repository, f.Branch, len(end.Left))
	h.again(m)
}

func (h *branchEnds) ack(m jetstream.Msg) {
	if err := m.Ack(); err != nil {
		h.logf("acknowledging %s: %v", m.Subject(), err)
	}
}

// again leaves the message for the bus to give again in a minute: the retry is the bus's redelivery, not a timer here.
func (h *branchEnds) again(m jetstream.Msg) {
	if err := m.NakWithDelay(versions.RemoveAgainAfter); err != nil {
		h.logf("leaving %s for the bus to give again: %v", m.Subject(), err)
	}
}

// tagWanted takes the checker's tag-wanted in: written to tag-asks as builder_tag writes it, answered by its watch.
func tagWanted(ctx context.Context, kv jetstream.KeyValue, m jetstream.Msg, by string, now time.Time, logf func(string, ...any)) {
	if m.Subject() != tagWantedSubject {
		_ = m.Ack() // the family's other events (the verdicts) are not the builder's
		return
	}
	var t versions.TagWanted
	err := json.Unmarshal(m.Data(), &t)
	var key string
	var value []byte
	if err == nil {
		key, value, err = versions.TagWantedAsk(t, by, now)
	}
	if err != nil {
		logf("%s holds what is no tag-wanted, left: %v", m.Subject(), err)
		_ = m.Ack()
		return
	}
	if kv == nil {
		logf("tag-wanted of %s %s is heard again: %s is not on the bus yet", t.Repository, t.Branch, versions.TagAsksState)
		_ = m.NakWithDelay(versions.RemoveAgainAfter)
		return
	}
	if _, err := kv.Put(ctx, key, value); err != nil {
		logf("tag-wanted of %s %s is heard again: %v", t.Repository, t.Branch, err)
		_ = m.Nak()
		return
	}
	_ = m.Ack()
}

// tagsVia is build.Builder.Tags over the state tag-asks: the asks of the repository, read when a check's build is
// looked at; none while the state is not bound.
func tagsVia(kv func() jetstream.KeyValue, logf func(string, ...any)) build.TagOf {
	return func(ctx context.Context, repository, head string) (string, bool) {
		store := kv()
		if store == nil {
			return "", false
		}
		lister, err := store.ListKeysFiltered(ctx, versions.EscapeToken(repository)+".>")
		if err != nil {
			logf("the tag asks of %s cannot be read, so its check builds no tagged version: %v", repository, err)
			return "", false
		}
		var asks []versions.TagAsk
		for key := range lister.Keys() {
			e, err := store.Get(ctx, key)
			if err != nil {
				continue
			}
			if t, err := versions.DecodeTagAsk(e.Value()); err == nil {
				asks = append(asks, t)
			}
		}
		return versions.TagFor(asks, repository, head)
	}
}