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

package versions

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

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
)

// **A reserved version left unpublished is published again by the builder, on an event or by a person's tool** (design 47
// G §588, ADR 0674): on the node-engine seat's `running` of a module whose catalog graph entry claims the registry seat
// of the missing artifacts' kind, and on its `manifest-reads-changed` with `why` `returned` of the node such a holder
// runs on. Each message is acted on once, by its stream sequence (state `acted`); each holder has one queue of versions,
// kept in the state `publish-queue`, sent one at a time. A send is the registry seat's `publish` with the digest and no
// parts (A-73): the holder takes the bytes from the artifact store by digest, and refuses what no store holds.

const (
	// PublishQueueState is the builder's old state section of the queues, capped at PublishQueueMiB.
	PublishQueueState  = "publish-queue"
	PublishQueueBucket = "mesh-delivery-builder_" + PublishQueueState
	PublishQueueMiB    = 4
	// ActedState holds, one key per subject, the last stream sequence the builder acted on; capped at ActedMiB.
	ActedState  = "acted"
	ActedBucket = "mesh-delivery-builder_" + ActedState
	ActedMiB    = 1
	// PublishQueueTool is builder_publish_queue.
	PublishQueueTool = "builder_publish_queue"
	// DefaultRepublishMax is the setting republish-max-versions unless set (§633).
	DefaultRepublishMax = 50
	// QueueFull is what version-unreleased says of a version the queue's cap refused.
	QueueFull = "queue full"
	// NotStoredWords opens what a refused artifact is said as.
	NotStoredWords = "not stored: "
	// VerbPublish is the registry seats' verb a publish again sends.
	VerbPublish = "publish"
)

// RegistrySeats are the four registry seats a version is published to.
var RegistrySeats = []string{SeatImages, SeatArtifacts, SeatGoPackages, SeatNPM}

// SeatOfArtifact is the registry seat of one artifact of the catalog graph entry: from its package's ecosystem when the
// entry names one (go, npm, image), else from its kind (SeatOf).
func SeatOfArtifact(a meshcatalog.Artifact) (string, bool) {
	if a.Package != nil {
		switch a.Package.Ecosystem {
		case "go":
			return SeatGoPackages, true
		case "npm":
			return SeatNPM, true
		case "image":
			return SeatImages, true
		}
	}
	seat, _, ok := SeatOf(a.Kind, a.Language)
	return seat, ok
}

// QueuedPublish is the value of a publish-queue key `<holder module>.<module>.<version>` (each token escaped, §73): the
// artifacts of the version this holder publishes, by reference with their digests, when it was queued and whether its
// publish has begun.
type QueuedPublish struct {
	V         int               `json:"v"`
	Holder    string            `json:"holder"`
	Module    string            `json:"module"`
	Version   string            `json:"version"`
	Artifacts map[string]string `json:"artifacts"`
	Queued    time.Time         `json:"queued"`
	Begun     bool              `json:"begun,omitempty"`
}

// QueueKey is the key of holder's queued version of module.
func QueueKey(holder, module, version string) string {
	return EscapeToken(holder) + "." + EscapeToken(module) + "." + EscapeToken(version)
}

// Acted is the value of an acted key: the last stream sequence acted on for its subject.
type Acted struct {
	Sequence uint64 `json:"sequence"`
}

// QueueStore is the queue's bucket: compare-and-set, deletion and how many bytes it holds, for its cap.
type QueueStore interface {
	ReserveStore
	Bytes(ctx context.Context) (uint64, error)
}

// Publish is one artifact's publish again: what the registry seat's publish takes, without parts.
type Publish struct {
	Seat, Kind, Name, Version, Digest, Fingerprint, Commit, Transfer string
}

// Republisher publishes reserved versions again (design 47 G §588).
type Republisher struct {
	Reserved *Reserver
	Queue    QueueStore
	Acted    ReserveStore
	// Entry reads the catalog graph entry of a fingerprint (the seat mesh-catalog's verb entry).
	Entry func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error)
	// Send sends one publish to a registry seat; an error is the holder's refusal or the call's fault, in its words.
	Send func(ctx context.Context, p Publish) error
	// Unreleased says version-unreleased about a version with the holder's words.
	Unreleased func(ctx context.Context, v ReservedVersion, why string)
	// Max is the setting republish-max-versions: the most versions one event queues.
	Max  int
	Now  func() time.Time
	Logf func(string, ...any)
	// Go starts the sending of a holder's queue; nil starts a goroutine.
	Go func(func())

	mu   sync.Mutex
	runs map[string]map[string][]string // node -> holder module -> the registry seats it claims
	busy map[string]bool
	// lastQueued is the newest time a version was queued at: a clock that has not moved between two versions (a coarse clock,
	// or a fixed one) must not make the queue's order the order of its keys, so each is queued after the last.
	lastQueued time.Time
}

// queuedAt is the time a version is queued at: now, or a nanosecond after the last one queued when now is not after it, so the
// order the versions were queued in (newest reservation first) is the order they are sent in.
func (r *Republisher) queuedAt() time.Time {
	r.mu.Lock()
	defer r.mu.Unlock()
	at := r.now()
	if !at.After(r.lastQueued) {
		at = r.lastQueued.Add(time.Nanosecond)
	}
	r.lastQueued = at
	return at
}

func (r *Republisher) now() time.Time {
	if r.Now != nil {
		return r.Now().UTC()
	}
	return time.Now().UTC()
}

func (r *Republisher) logf(f string, a ...any) {
	if r.Logf != nil {
		r.Logf(f, a...)
	}
}

// ActedKey is the acted key of a subject.
func ActedKey(subject string) string { return EscapeToken(subject) }

// Heard takes one message of the node-engine seat's running or manifest-reads-changed in, at stream sequence seq: it acts
// on it only when its sequence is newer than the one kept for its subject, and keeps it after. A subject that is neither is
// not its; an error leaves the message not acted on, so it is read again.
func (r *Republisher) Heard(ctx context.Context, subject string, data []byte, seq uint64) (bool, error) {
	running := strings.HasPrefix(subject, strings.TrimSuffix(nodeengine.SubjectRunningAll, "*.*"))
	changed := strings.HasPrefix(subject, "mesh.seat.node-engine.event."+nodeengine.EventManifestReadsChanged+".")
	if !running && !changed {
		return false, nil
	}
	key := ActedKey(subject)
	raw, rev, found, err := r.Acted.Get(ctx, key)
	if err != nil {
		return true, fmt.Errorf("the acted sequence of %s cannot be read: %w", subject, err)
	}
	if found {
		var a Acted
		if json.Unmarshal(raw, &a) == nil && a.Sequence >= seq {
			return true, nil
		}
	}
	if running {
		err = r.running(ctx, data)
	} else {
		err = r.returned(ctx, data)
	}
	if err != nil {
		return true, err
	}
	val, _ := json.Marshal(Acted{Sequence: seq})
	if found {
		_, err = r.Acted.Update(ctx, key, val, rev)
	} else {
		_, err = r.Acted.Create(ctx, key, val)
	}
	if err != nil && !errors.Is(err, ErrConflict) {
		return true, fmt.Errorf("the acted sequence of %s cannot be kept: %w", subject, err)
	}
	return true, nil
}

// registrySeatsOf are the registry seats the module of the catalog entry fingerprint claims, by their publish verb.
func (r *Republisher) registrySeatsOf(ctx context.Context, fingerprint string) ([]string, error) {
	e, err := r.Entry(ctx, fingerprint)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, seat := range RegistrySeats {
		if slices.Contains(e.Serves, seat+"."+VerbPublish) {
			out = append(out, seat)
		}
	}
	return out, nil
}

// running takes a running in: a module stopped is forgotten; one running whose entry claims a registry seat is a holder on
// its node, and its versions are queued.
func (r *Republisher) running(ctx context.Context, data []byte) error {
	var ev nodeengine.Running
	if err := json.Unmarshal(data, &ev); err != nil || ev.Module == "" || ev.Node == "" {
		r.logf("a running that does not read is not acted on: %v", err)
		return nil
	}
	if ev.Stopped || ev.Fingerprint == "" {
		r.mu.Lock()
		delete(r.runs[ev.Node], ev.Module)
		r.mu.Unlock()
		return nil
	}
	seats, err := r.registrySeatsOf(ctx, ev.Fingerprint)
	if err != nil {
		return fmt.Errorf("the catalog entry %s of %s cannot be read: %w", ev.Fingerprint, ev.Module, err)
	}
	r.mu.Lock()
	if r.runs == nil {
		r.runs = map[string]map[string][]string{}
	}
	if r.runs[ev.Node] == nil {
		r.runs[ev.Node] = map[string][]string{}
	}
	if len(seats) == 0 {
		delete(r.runs[ev.Node], ev.Module)
	} else {
		r.runs[ev.Node][ev.Module] = seats
	}
	r.mu.Unlock()
	if len(seats) == 0 {
		return nil
	}
	_, err = r.queueEvent(ctx, map[string][]string{ev.Module: seats})
	return err
}

// returned takes a manifest-reads-changed in: only one with why returned acts, for every holder known to run on its node.
func (r *Republisher) returned(ctx context.Context, data []byte) error {
	var ev nodeengine.ManifestReadsChanged
	if err := json.Unmarshal(data, &ev); err != nil || ev.Node == "" {
		r.logf("a manifest-reads-changed that does not read is not acted on: %v", err)
		return nil
	}
	if ev.Why != nodeengine.ManifestReadsChangedWhyReturned {
		return nil
	}
	r.mu.Lock()
	holders := map[string][]string{}
	for m, seats := range r.runs[ev.Node] {
		holders[m] = seats
	}
	r.mu.Unlock()
	if len(holders) == 0 {
		return nil
	}
	_, err := r.queueEvent(ctx, holders)
	return err
}

// view is a reserved version with the artifacts of it a holder would publish.
type view struct {
	v    ReservedVersion
	arts map[string]meshcatalog.Artifact // reference -> its catalog artifact
}

// queueEvent queues, for each holder (module -> the seats it claims), the unreleased versions whose missing artifacts
// that holder's seats take: newest keys first, at most Max versions in all (one key per module and MAJOR.MINOR is the
// reservation's own).
func (r *Republisher) queueEvent(ctx context.Context, holders map[string][]string) ([]string, error) {
	over, err := r.Reserved.Overdue(ctx, nil)
	if err != nil {
		return nil, err
	}
	var vs []ReservedVersion
	for _, o := range over {
		if v, ok, err := r.Reserved.Of(ctx, o.Module, o.Version); err == nil && ok {
			vs = append(vs, v)
		}
	}
	return r.queue(ctx, vs, func(seat string) []string { return holdersOf(holders, seat) }, r.Max)
}

func holdersOf(holders map[string][]string, seat string) []string {
	var out []string
	for m, seats := range holders {
		if slices.Contains(seats, seat) {
			out = append(out, m)
		}
	}
	sort.Strings(out)
	return out
}

// queue queues vs newest first, at most max of them (0 is no limit), each artifact to the first holder holdersOf names for
// its seat. It answers the versions queued, `<module> <version>`.
func (r *Republisher) queue(ctx context.Context, vs []ReservedVersion, holdersOf func(seat string) []string, max int) ([]string, error) {
	sort.SliceStable(vs, func(i, j int) bool { return vs[i].At.After(vs[j].At) })
	var queued []string
	started := map[string]bool{}
	for _, v := range vs {
		if max > 0 && len(queued) >= max {
			break
		}
		if v.Fingerprint == "" || v.BuiltAt.IsZero() {
			continue
		}
		e, err := r.Entry(ctx, v.Fingerprint)
		if err != nil {
			return queued, fmt.Errorf("the catalog entry %s of %s %s cannot be read: %w", v.Fingerprint, v.Module, v.Version, err)
		}
		byHolder := map[string]map[string]string{}
		for _, ref := range v.Missing() {
			if _, no := v.Unstored[ref]; no {
				continue
			}
			for _, a := range e.Artifacts {
				seat, ok := SeatOfArtifact(a)
				if a.Ref != ref || !ok {
					continue
				}
				if hs := holdersOf(seat); len(hs) > 0 {
					if byHolder[hs[0]] == nil {
						byHolder[hs[0]] = map[string]string{}
					}
					byHolder[hs[0]][ref] = v.Artifacts[ref]
				}
			}
		}
		did := false
		for holder, arts := range byHolder {
			ok, err := r.enqueue(ctx, v, holder, arts)
			if err != nil {
				return queued, err
			}
			did = did || ok
			started[holder] = true
		}
		if did {
			queued = append(queued, v.Module+" "+v.Version)
		}
	}
	for holder := range started {
		r.drainLater(ctx, holder)
	}
	return queued, nil
}

// enqueue puts one holder's artifacts of a version in its queue: not again when it is queued already, and not past the
// section's cap (said as queue full).
func (r *Republisher) enqueue(ctx context.Context, v ReservedVersion, holder string, arts map[string]string) (bool, error) {
	key := QueueKey(holder, v.Module, v.Version)
	if _, _, found, err := r.Queue.Get(ctx, key); err != nil {
		return false, err
	} else if found {
		return false, nil
	}
	raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: holder, Module: v.Module, Version: v.Version, Artifacts: arts, Queued: r.queuedAt()})
	used, err := r.Queue.Bytes(ctx)
	if err != nil {
		return false, fmt.Errorf("the size of %s cannot be read: %w", PublishQueueBucket, err)
	}
	if used+uint64(len(raw)) > PublishQueueMiB<<20 {
		if r.Unreleased != nil {
			r.Unreleased(ctx, v, QueueFull)
		}
		return false, nil
	}
	if _, err := r.Queue.Create(ctx, key, raw); err != nil {
		if errors.Is(err, ErrConflict) {
			return false, nil
		}
		return false, fmt.Errorf("%s cannot be queued: %w", key, err)
	}
	return true, nil
}

// Republish is builder_publish_again: with module and version, that reserved version's missing artifacts; with module
// alone, each reserved version of it whose artifacts are not all published; with neither, every version the finding
// version-unreleased names. It queues as the events do (no limit but the cap), for the holders known to run, and answers
// the versions it queued.
func (r *Republisher) Republish(ctx context.Context, module, version string) ([]string, error) {
	var vs []ReservedVersion
	if module == "" {
		over, err := r.Reserved.Overdue(ctx, nil)
		if err != nil {
			return nil, err
		}
		for _, o := range over {
			if v, ok, err := r.Reserved.Of(ctx, o.Module, o.Version); err == nil && ok {
				vs = append(vs, v)
			}
		}
	} else {
		all, err := r.Reserved.All(ctx)
		if err != nil {
			return nil, err
		}
		for _, v := range all {
			if v.Module == module && (version == "" || v.Version == version) && len(v.Missing()) > 0 {
				vs = append(vs, v)
			}
		}
		if len(vs) == 0 {
			return nil, fmt.Errorf("%s holds no reserved version%s with an artifact unpublished", module, map[bool]string{true: " " + version, false: ""}[version != ""])
		}
	}
	r.mu.Lock()
	known := map[string][]string{}
	for _, mods := range r.runs {
		for m, seats := range mods {
			known[m] = seats
		}
	}
	r.mu.Unlock()
	return r.queue(ctx, vs, func(seat string) []string { return holdersOf(known, seat) }, 0)
}

// Queued are the queued versions, per holder in the order they are sent, and the acted keys with their sequences:
// builder_publish_queue.
type Queued struct {
	Queue []QueuedPublish   `json:"queue"`
	Acted map[string]uint64 `json:"acted"`
}

// AnswerQueue is builder_publish_queue over the two sections as the tools bundle reads them.
func AnswerQueue(queue, acted Reader) (Queued, error) {
	out := Queued{Queue: []QueuedPublish{}, Acted: map[string]uint64{}}
	keys, err := queue.Keys()
	if err != nil {
		return out, fmt.Errorf("the state %s cannot be listed: %w", PublishQueueState, err)
	}
	for _, k := range keys {
		raw, ok, err := queue.Get(k)
		if err != nil || !ok {
			continue
		}
		var q QueuedPublish
		if json.Unmarshal(raw, &q) == nil {
			out.Queue = append(out.Queue, q)
		}
	}
	sort.SliceStable(out.Queue, func(i, j int) bool {
		if out.Queue[i].Holder != out.Queue[j].Holder {
			return out.Queue[i].Holder < out.Queue[j].Holder
		}
		return out.Queue[i].Queued.Before(out.Queue[j].Queued)
	})
	akeys, err := acted.Keys()
	if err != nil {
		return out, fmt.Errorf("the state %s cannot be listed: %w", ActedState, err)
	}
	for _, k := range akeys {
		raw, ok, err := acted.Get(k)
		if err != nil || !ok {
			continue
		}
		var a Acted
		if json.Unmarshal(raw, &a) == nil {
			out.Acted[k] = a.Sequence
		}
	}
	return out, nil
}

// PublishQueueDescription is builder_publish_queue's description.
const PublishQueueDescription = "The versions the builder will publish again (its state " + PublishQueueState + ", design 47 G section 588): per registry " +
	"seat holder, in the order they are sent, each with its artifacts' digests, when it was queued and whether its publish has begun; and the " +
	"keys of the state " + ActedState + ", the last stream sequence the builder acted on for each running and manifest-reads-changed subject. Read-only."

// PublishAgainDescription is builder_publish_again's description.
const PublishAgainDescription = "Publish reserved versions again (design 47 G section 588), which builds nothing: with module and version, that reserved " +
	"version's missing artifacts; with module alone, each reserved version of that module whose artifacts are not all published; with neither, every " +
	"version the finding version-unreleased names, newest first. It queues them as the builder does on a registry holder's running, and answers at once " +
	"with the versions it queued; each outcome is said as package-published, or left in version-unreleased with the holder's words. A hand act, " +
	"recorded in the hand-act log."

// Resume takes up, at the builder's start, every queued version again, a begun one sent anew (§588).
func (r *Republisher) Resume(ctx context.Context) error {
	keys, err := r.Queue.KeysOf(ctx, AllFilter)
	if err != nil {
		return fmt.Errorf("the state %s cannot be listed: %w", PublishQueueState, err)
	}
	holders := map[string]bool{}
	for _, k := range keys {
		raw, _, found, err := r.Queue.Get(ctx, k)
		if err != nil || !found {
			continue
		}
		var q QueuedPublish
		if json.Unmarshal(raw, &q) == nil {
			holders[q.Holder] = true
		}
	}
	for h := range holders {
		r.drainLater(ctx, h)
	}
	return nil
}

func (r *Republisher) drainLater(ctx context.Context, holder string) {
	r.mu.Lock()
	if r.busy == nil {
		r.busy = map[string]bool{}
	}
	if r.busy[holder] {
		r.mu.Unlock()
		return
	}
	r.busy[holder] = true
	r.mu.Unlock()
	run := func() {
		defer func() { r.mu.Lock(); delete(r.busy, holder); r.mu.Unlock() }()
		r.Drain(ctx, holder)
	}
	if r.Go != nil {
		r.Go(run)
		return
	}
	go run()
}

// Drain sends holder's queued versions one at a time, oldest first, each bounded by UnreleasedBound; a key is deleted when
// its outcome is said.
func (r *Republisher) Drain(ctx context.Context, holder string) {
	for ctx.Err() == nil {
		keys, err := r.Queue.KeysOf(ctx, EscapeToken(holder)+".*.*")
		if err != nil {
			r.logf("the queue of %s cannot be listed: %v", holder, err)
			return
		}
		var items []QueuedPublish
		for _, k := range keys {
			if raw, _, found, err := r.Queue.Get(ctx, k); err == nil && found {
				var q QueuedPublish
				if json.Unmarshal(raw, &q) == nil {
					items = append(items, q)
				}
			}
		}
		if len(items) == 0 {
			return
		}
		sort.SliceStable(items, func(i, j int) bool { return items[i].Queued.Before(items[j].Queued) })
		r.send(ctx, items[0])
	}
}

// send publishes one queued version's artifacts and says the outcome, then deletes its key.
func (r *Republisher) send(ctx context.Context, q QueuedPublish) {
	key := QueueKey(q.Holder, q.Module, q.Version)
	defer func() {
		if err := r.Queue.Delete(context.WithoutCancel(ctx), key); err != nil {
			r.logf("%s cannot be deleted from %s: %v", key, PublishQueueState, err)
		}
	}()
	if !q.Begun {
		q.Begun = true
		raw, _ := json.Marshal(q)
		if _, rev, found, err := r.Queue.Get(ctx, key); err == nil && found {
			_, _ = r.Queue.Update(ctx, key, raw, rev)
		}
	}
	v, found, err := r.Reserved.Of(ctx, q.Module, q.Version)
	if err != nil || !found {
		return // released meanwhile, or dropped
	}
	e, err := r.Entry(ctx, v.Fingerprint)
	if err != nil {
		r.say(ctx, v, fmt.Sprintf("the catalog entry %s cannot be read: %v", v.Fingerprint, err))
		return
	}
	refs := make([]string, 0, len(q.Artifacts))
	for ref := range q.Artifacts {
		refs = append(refs, ref)
	}
	sort.Strings(refs)
	for _, ref := range refs {
		var art *meshcatalog.Artifact
		for i := range e.Artifacts {
			if e.Artifacts[i].Ref == ref {
				art = &e.Artifacts[i]
			}
		}
		seat, ok := "", false
		if art != nil {
			seat, ok = SeatOfArtifact(*art)
		}
		if !ok {
			r.say(ctx, v, ref+": the catalog entry names no registry seat for it")
			return
		}
		name := NameOf(q.Module, strings.TrimPrefix(ref, q.Module+":"))
		if art.Package != nil && art.Package.Name != "" {
			name = art.Package.Name
		}
		kind := art.Kind
		bound, cancel := context.WithTimeout(ctx, UnreleasedBound) // a context's bound, no timer (ADR 0437 §8)
		err := r.Send(bound, Publish{Seat: seat, Kind: kind, Name: name, Version: v.Version, Digest: q.Artifacts[ref], Fingerprint: v.Fingerprint,
			Commit: v.Commit, Transfer: "republish-" + q.Module + "-" + v.Version + "-" + strings.TrimPrefix(ref, q.Module+":")})
		cancel()
		if err != nil {
			words := err.Error()
			if strings.Contains(words, strings.TrimSuffix(NotStoredWords, ": ")) {
				if ok, merr := r.Reserved.MarkUnstored(ctx, q.Module, q.Version, ref, words); merr != nil {
					r.logf("%s %s: the unstored %s is not kept: %v", q.Module, q.Version, ref, merr)
				} else if ok {
					r.say(ctx, v, ref+" "+NotStoredWords+words)
				}
				continue
			}
			r.say(ctx, v, ref+": "+words)
		}
	}
}

func (r *Republisher) say(ctx context.Context, v ReservedVersion, why string) {
	r.logf("%s %s is not published again: %s", v.Module, v.Version, why)
	if r.Unreleased != nil {
		r.Unreleased(ctx, v, why)
	}
}

// MarkUnstored records that a holder refused artifact of module's reserved version as not stored; it answers whether that
// was new (said once).
func (r *Reserver) MarkUnstored(ctx context.Context, module, version, artifact, words string) (bool, error) {
	series, _, err := SplitVersion(version)
	if err != nil {
		return false, err
	}
	key := ReservedKey(module, series)
	for {
		cur, rev, found, err := r.get(ctx, key)
		if err != nil || !found || cur.Version != version {
			return false, err
		}
		if _, was := cur.Unstored[artifact]; was {
			return false, nil
		}
		if cur.Unstored == nil {
			cur.Unstored = map[string]string{}
		}
		cur.Unstored[artifact] = words
		raw, _ := json.Marshal(cur)
		if _, err := r.Store.Update(ctx, key, raw, rev); errors.Is(err, ErrConflict) {
			continue
		} else if err != nil {
			return false, fmt.Errorf("%s cannot be written: %w", key, err)
		}
		return true, nil
	}
}