mesh-delivery/modules/mesh-delivery-builder/internal/versions/floor.go

package versions

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"
)

// FloorSources are the three sources of a name's floor (design 51 *The floor*): each a tool call that answers the
// versions it knows of a name, or an error when it does not answer.
type FloorSources struct {
	// Seat is `seat:<registry>.versions` of the registry seat the name is published to.
	Seat func(ctx context.Context, seat, name string) ([]string, error)
	// Tags is the forge bridge's bridge_tags: the git tags of the repository the name is built from, under prefix.
	Tags func(ctx context.Context, repository, prefix string) ([]string, error)
	// Requirements is the indexer's indexer_requirements: the highest pseudo-versions consumers require.
	Requirements func(ctx context.Context, name string) ([]string, error)
}

// FloorOf names what a floor is asked for: the name on its seat, and the repository and tag prefix of its git tags.
type FloorOf struct{ Seat, Name, Repository, Prefix string }

// Ask asks the three sources of a name; a source that errs, or is not wired, is named in Silent with its error in why.
func (s FloorSources) Ask(ctx context.Context, of FloorOf) (Floor, map[string]string) {
	seat, name := of.Seat, of.Name
	var f Floor
	why := map[string]string{}
	ask := func(source string, call func() ([]string, error), into *[]string) {
		if call == nil {
			f.Silent, why[source] = append(f.Silent, source), "not asked: the builder has no path to it"
			return
		}
		got, err := call()
		if err != nil {
			f.Silent, why[source] = append(f.Silent, source), err.Error()
			return
		}
		*into = got
	}
	var seatCall, tagsCall, reqCall func() ([]string, error)
	if s.Seat != nil {
		seatCall = func() ([]string, error) { return s.Seat(ctx, seat, name) }
	}
	if s.Tags != nil {
		tagsCall = func() ([]string, error) { return s.Tags(ctx, of.Repository, of.Prefix) }
	}
	if s.Requirements != nil {
		reqCall = func() ([]string, error) { return s.Requirements(ctx, name) }
	}
	ask(SourceSeat, seatCall, &f.Seat)
	ask(SourceTags, tagsCall, &f.Tags)
	ask(SourceRequirements, reqCall, &f.Requirements)
	return f, why
}

// What wakes a name waiting on its floor, never a timer (design 51, DLV-B3b): the source that was silent saying
// something, or its module's health condition clearing.
const (
	// WakePublished is the holder's `published` on the builder's durable consumer of `<seat>.events`.
	WakePublished = "published"
	// WakeForge is the forge bridge's next message on its stream `forge`.
	WakeForge = "forge"
	// WakeAccepted is the indexer's next `manifest-accepted`.
	WakeAccepted = "manifest-accepted"
	// WakeCleared is the condition keeper's clear of a silent module's health condition on `mesh-conditions`.
	WakeCleared = "condition-cleared"
	// WakeByHand is builder_floor_again, a hand act.
	WakeByHand = "by hand"
)

// sourceOfWake is the floor source each event says something of; a clear names the source by its module.
var sourceOfWake = map[string]string{WakePublished: SourceSeat, WakeForge: SourceTags, WakeAccepted: SourceRequirements}

// ModuleOfSource is the module whose health condition's clear wakes a silent source.
var ModuleOfSource = map[string]string{
	SourceTags:         "mesh-delivery-forge-bridge",
	SourceRequirements: "mesh-delivery-indexer",
}

// FloorWait is one name waiting on its floor: the sources still silent and why, since when; the finding the builder
// writes names them, cleared once all three answer.
type FloorWait struct {
	Name   string            `json:"name"`
	Seat   string            `json:"seat"`
	Silent []string          `json:"silent"`
	Why    map[string]string `json:"why,omitempty"`
	Since  time.Time         `json:"since"`
	// Holder is the module holding the name's seat: its condition's clear wakes the seat source.
	Holder string `json:"holder,omitempty"`
}

// Finding is the floor finding's words.
func (w FloorWait) Finding() string {
	var parts []string
	for _, s := range w.Silent {
		p := s
		if why := w.Why[s]; why != "" {
			p += " (" + why + ")"
		}
		parts = append(parts, p)
	}
	return fmt.Sprintf("the floor of %s is not known since %s: %s did not answer; nothing of it is numbered until all three answer",
		w.Name, w.Since.UTC().Format(time.RFC3339), strings.Join(parts, ", "))
}

// The builder's state floor-waits (DLV-B4): each name waiting on its floor, keyed by its escaped name, so a restarted
// builder still knows which names its silent sources' events wake; builder_floor_waits reads it.
const (
	FloorWaitsState  = "floor-waits"
	FloorWaitsBucket = "mesh-delivery-builder_floor-waits"
	FloorWaitsTool   = "builder_floor_waits"
)

// FloorWaitKey is a wait's key in floor-waits.
func FloorWaitKey(name string) string { return EscapeToken(name) }

// FloorKeeper keeps the waits on the bus: Put a wait's value, Delete its key.
type FloorKeeper interface {
	Put(key string, value []byte) error
	Delete(key string) error
}

// FloorWaits are the names waiting on their floor. Wake answers which names to ask again for an event, Answered drops
// a name once all three answered; the builder's finding for each is the wait's Finding. Once Kept, every change is
// written to the state floor-waits as well.
type FloorWaits struct {
	mu     sync.Mutex
	waits  map[string]FloorWait
	keeper FloorKeeper
	// Logf says a wait that could not be kept; nil says nothing.
	Logf func(string, ...any)
	// Said is told each wait recorded (raised) and each answered (not raised): the builder's finding floor-silent.
	Said func(w FloorWait, raised bool)
}

// FloorSilent is the kind of the floor finding.
const FloorSilent = "floor-silent"

// Keep binds the state: the waits it holds that are not known here are taken in (a restarted builder's), every wait
// known here is written to it, and every change from now on is.
func (fw *FloorWaits) Keep(k FloorKeeper, held []FloorWait) {
	fw.mu.Lock()
	if fw.waits == nil {
		fw.waits = map[string]FloorWait{}
	}
	for _, w := range held {
		if _, had := fw.waits[w.Name]; !had && w.Name != "" {
			fw.waits[w.Name] = w
		}
	}
	fw.keeper = k
	all := make([]FloorWait, 0, len(fw.waits))
	for _, w := range fw.waits {
		all = append(all, w)
	}
	fw.mu.Unlock()
	for _, w := range all {
		fw.put(k, w)
	}
}

func (fw *FloorWaits) put(k FloorKeeper, w FloorWait) {
	if k == nil {
		return
	}
	data, err := json.Marshal(w)
	if err == nil {
		err = k.Put(FloorWaitKey(w.Name), data)
	}
	if err != nil && fw.Logf != nil {
		fw.Logf("the floor wait of %s is not kept in %s (a restart forgets it until its next claim): %v", w.Name, FloorWaitsState, err)
	}
}

// DecodeFloorWait reads one value of floor-waits.
func DecodeFloorWait(raw []byte) (FloorWait, error) {
	var w FloorWait
	if err := json.Unmarshal(raw, &w); err != nil {
		return FloorWait{}, err
	}
	if w.Name == "" {
		return FloorWait{}, errors.New("a floor wait names no name")
	}
	return w, nil
}

// Wait records that name's floor is not known.
func (fw *FloorWaits) Wait(name, seat, holder string, f Floor, why map[string]string, now time.Time) FloorWait {
	fw.mu.Lock()
	if fw.waits == nil {
		fw.waits = map[string]FloorWait{}
	}
	w, had := fw.waits[name]
	if !had {
		w.Since = now
	}
	w.Name, w.Seat, w.Holder, w.Silent, w.Why = name, seat, holder, slices.Clone(f.Silent), why
	fw.waits[name] = w
	k := fw.keeper
	fw.mu.Unlock()
	fw.put(k, w)
	if fw.Said != nil {
		fw.Said(w, true)
	}
	return w
}

// Answered drops name: its floor is known. It says whether the name waited, so its finding is cleared.
func (fw *FloorWaits) Answered(name string) bool {
	fw.mu.Lock()
	w, had := fw.waits[name]
	delete(fw.waits, name)
	k := fw.keeper
	fw.mu.Unlock()
	if had && fw.Said != nil {
		fw.Said(w, false)
	}
	if had && k != nil {
		if err := k.Delete(FloorWaitKey(name)); err != nil && fw.Logf != nil {
			fw.Logf("the floor wait of %s is not cleared from %s: %v", name, FloorWaitsState, err)
		}
	}
	return had
}

// FloorWaitsAnswer is what builder_floor_waits answers: every name waiting on its floor, oldest first, with its finding.
type FloorWaitsAnswer struct {
	Waits []FloorWaitRead `json:"waits"`
}

// FloorWaitRead is one wait as builder_floor_waits shows it.
type FloorWaitRead struct {
	FloorWait
	Finding string `json:"finding"`
}

// AnswerFloorWaits is builder_floor_waits, read from the state floor-waits.
func AnswerFloorWaits(r Reader) (FloorWaitsAnswer, error) {
	keys, err := r.Keys()
	if err != nil {
		return FloorWaitsAnswer{}, err
	}
	out := FloorWaitsAnswer{Waits: []FloorWaitRead{}}
	for _, k := range keys {
		raw, ok, err := r.Get(k)
		if err != nil {
			return FloorWaitsAnswer{}, err
		}
		if !ok {
			continue
		}
		w, err := DecodeFloorWait(raw)
		if err != nil {
			return FloorWaitsAnswer{}, fmt.Errorf("%s.%s: %w", FloorWaitsState, k, err)
		}
		out.Waits = append(out.Waits, FloorWaitRead{FloorWait: w, Finding: w.Finding()})
	}
	sort.Slice(out.Waits, func(i, j int) bool {
		if !out.Waits[i].Since.Equal(out.Waits[j].Since) {
			return out.Waits[i].Since.Before(out.Waits[j].Since)
		}
		return out.Waits[i].Name < out.Waits[j].Name
	})
	return out, nil
}

// FloorWaitsDescription is builder_floor_waits' description.
const FloorWaitsDescription = "The names waiting on their floor (the builder's state floor-waits, design 51 *The floor*), oldest first: " +
	"each name, its registry seat, the sources still silent and why, since when, and the finding the builder says for it. " +
	"Nothing of a waiting name is numbered until all three sources answer; builder_floor_again asks one again by hand. Read-only."

// Wake answers the names to ask their floor again on an event: kind one of the Wake constants; seat the seat a
// `published` came on; module the module whose condition cleared. A name is woken only when the event says something
// of a source it waits on; WakeByHand with a name wakes that name alone.
func (fw *FloorWaits) Wake(kind, seat, module, name string) []string {
	fw.mu.Lock()
	defer fw.mu.Unlock()
	var out []string
	for n, w := range fw.waits {
		switch kind {
		case WakeByHand:
			if n == name {
				out = append(out, n)
			}
		case WakeCleared:
			for _, s := range w.Silent {
				if ModuleOfSource[s] == module || (s == SourceSeat && w.Holder != "" && w.Holder == module) {
					out = append(out, n)
					break
				}
			}
		default:
			src := sourceOfWake[kind]
			if slices.Contains(w.Silent, src) && (kind != WakePublished || seat == "" || seat == w.Seat) {
				out = append(out, n)
			}
		}
	}
	sort.Strings(out)
	return out
}

// Waiting is every wait, by name.
func (fw *FloorWaits) Waiting() []FloorWait {
	fw.mu.Lock()
	defer fw.mu.Unlock()
	out := make([]FloorWait, 0, len(fw.waits))
	for _, w := range fw.waits {
		out = append(out, w)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}