mesh-catalog/modules/distribution/internal/holder/verbs.go

package holder

import (
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"time"

	reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
)

// Catalog is what the store answers of what it holds.
type Catalog interface {
	// Versions are the versions (tags) of name, sorted.
	Versions(name string) ([]string, error)
	// Digest is the digest name holds at version, empty when none.
	Digest(name, version string) (string, error)
	// Remove deletes name at version by its manifest's digest, every tag naming that digest with it; the store's
	// collector frees its blobs.
	Remove(name, version string) error
	// RemoveTag deletes the tag version of name alone, the manifest staying under every other tag naming it; a store
	// that cannot delete a tag alone answers ErrNoTagAlone and removes nothing.
	RemoveTag(name, version string) error
	// RemoveDigest deletes an untagged manifest of name by its digest (retention).
	RemoveDigest(name, digest string) error
	// Platforms are the manifests an index of name names; none for a manifest that is not an index or is not held.
	Platforms(name, digest string) ([]string, error)
	// Mirror copies an outside image, pinned by digest, into upstream/<host>/<path>, answering the reference.
	Mirror(source, digest string) (string, error)
	// Reference is the seat's answer for name at digest, known before anything is copied.
	Reference(name, digest string) string
}

// MirrorsState names the state of the copies `mirror` took on.
const MirrorsState = "mirrors"

// Mirrors is the state mirrors: each copy `mirror` took on, kept until its outcome is said, so a copy cut by a
// restart is taken on again from the state's watch.
type Mirrors interface {
	// Ask keeps r under its key; an ask already kept is no fault (the copy is taken on once).
	Ask(key string, r MirrorRequest) error
	// Claim writes r (its Claim set) under key at revision, compare-and-set: false, and no error, when another
	// writer wrote the key first.
	Claim(key string, r MirrorRequest, revision uint64) (bool, error)
	Delete(key string) error
	// List is every ask kept, by its key: what registry_mirrors reads.
	List() (map[string]MirrorRequest, error)
	// Put writes r under key whatever it held: an ask asked again by hand, its claim let go.
	Put(key string, r MirrorRequest) error
}

// HoldState is distribution's own state holds (bus state, template state; ADR 0617 §2).
type HoldState interface {
	Get(key string) (Hold, bool, error)
	Put(h Hold) error
	Delete(key string) error
	List() ([]Hold, error)
}

// Hold keeps one untagged bundle, archive or copied base from retention's removal.
type Hold struct {
	Seat   string    `json:"seat"`
	Name   string    `json:"name"`
	Kind   string    `json:"kind"`
	Digest string    `json:"digest"`
	By     string    `json:"by"`
	Why    string    `json:"why,omitempty"`
	Since  time.Time `json:"since"`
}

// Key is the hold's key in the state.
func (h Hold) Key() string {
	return h.Seat + "." + strings.ReplaceAll(strings.ReplaceAll(h.Name, ".", "="), "@", "=") + "." + strings.TrimPrefix(h.Digest, "sha256:")
}

// Verbs answers the verbs other than the publish: versions, remove, hold, release, tags and mirror.
type Verbs struct {
	*Holder
	Catalog Catalog
	Holds   HoldState
	Mirrors Mirrors
}

// VersionsAnswer is `versions`: the versions the seat holds of name.
type VersionsAnswer struct {
	Seat     string   `json:"seat"`
	Name     string   `json:"name"`
	Versions []string `json:"versions"`
}

// Versions is `versions`.
func (v *Verbs) Versions(seat, caller, name string) (VersionsAnswer, error) {
	if err := v.Callers.Allowed(VerbVersions, caller); err != nil {
		return VersionsAnswer{}, err
	}
	if strings.TrimSpace(name) == "" {
		return VersionsAnswer{}, errors.New("versions: no name")
	}
	got, err := v.Catalog.Versions(name)
	if err != nil {
		return VersionsAnswer{}, fmt.Errorf("versions: the store cannot list %s: %w", name, err)
	}
	if got == nil {
		got = []string{}
	}
	return VersionsAnswer{Seat: seat, Name: name, Versions: got}, nil
}

// RemoveRequest is `remove`: a tagged version, or with the reason retention an untagged one by its digest.
type RemoveRequest struct {
	Name    string `json:"name"`
	Version string `json:"version,omitempty"`
	Digest  string `json:"digest,omitempty"`
	Kind    string `json:"kind,omitempty"`
	Reason  string `json:"reason"`
}

// ReasonRetention is the reason of the sweep's removal of an untagged version (ADR 0610 §3).
// The seat definitions name it (the removal reasons' enum), so the sweep and the holder share one value.
const ReasonRetention = reg.RemoveInputReasonRetention

// Remove is `remove`: the builder removes a tagged version; the artifact sweep's holder an untagged bundle, archive or
// copied base with the reason retention, which a hold keeps. Every removal says `removed` with its reason.
func (v *Verbs) Remove(seat, caller string, r RemoveRequest) error {
	if err := v.Callers.Allowed(VerbRemove, caller); err != nil {
		return err
	}
	if strings.TrimSpace(r.Name) == "" || strings.TrimSpace(r.Reason) == "" {
		return errors.New("remove: a name and a reason are required")
	}
	ev := Event{Name: r.Name, Version: r.Version, Kind: r.Kind, Digest: r.Digest, Holder: Self, Reason: r.Reason, By: caller}
	switch {
	case r.Reason == ReasonRetention:
		if CallerModule(caller) != SweepHolder {
			return fmt.Errorf("remove: refused for %s: a removal for retention is the artifact sweep's alone", caller)
		}
		if r.Digest == "" {
			return errors.New("remove: a removal for retention names the digest")
		}
		if _, held, err := v.Holds.Get(Hold{Seat: seat, Name: r.Name, Digest: r.Digest}.Key()); err != nil {
			return fmt.Errorf("remove: the holds cannot be read, so nothing is removed: %w", err)
		} else if held {
			return fmt.Errorf("remove: refused: %s@%s is held; release it first", r.Name, r.Digest)
		}
		if err := v.removeForRetention(seat, r.Name, r.Digest); err != nil {
			return err
		}
	default:
		if r.Version == "" || !strings.Contains(r.Version, "-") {
			return fmt.Errorf("remove: refused: %s@%s has no tag; only a tagged version is removed, but for retention", r.Name, r.Version)
		}
		d, err := v.Catalog.Digest(r.Name, r.Version)
		if err != nil {
			return fmt.Errorf("remove: the store cannot read %s@%s: %w", r.Name, r.Version, err)
		}
		if d != "" {
			// Reproducible builds share a digest between a branch's tagged version and the trunk's numbered one: a
			// manifest deleted by its digest would take the trunk's version with it (security review F1). When another
			// tag names the digest, the tag alone goes. A store that cannot remove a tag alone (registry 2.8.3) keeps the
			// tag: its bytes are kept under the other tags anyway, so the removal is said removed with that reason and
			// the builder lets the version go from its keeping, no refusal and no retry. Any other failure is refused.
			others, err := v.otherTags(r.Name, r.Version, d)
			if err != nil {
				return fmt.Errorf("remove: the store cannot say which other tags of %s name %s, so nothing is removed: %w", r.Name, d, err)
			}
			if len(others) > 0 {
				if err := v.Catalog.RemoveTag(r.Name, r.Version); errors.Is(err, ErrNoTagAlone) {
					sort.Strings(others)
					ev.Reason = fmt.Sprintf("%s; the store cannot remove a tag alone, so the tag %s is kept and its bytes stay under %s", r.Reason, r.Version, strings.Join(others, ", "))
				} else if err != nil {
					return fmt.Errorf("remove: refused: %s@%s shares its digest %s with %v, and the store did not remove the tag alone, so nothing is removed: %w", r.Name, r.Version, d, others, err)
				}
			} else if err := v.Catalog.Remove(r.Name, r.Version); err != nil {
				return fmt.Errorf("remove: the store did not remove %s@%s: %w", r.Name, r.Version, err)
			}
		}
		ev.Digest = d // a version already gone is said removed all the same, so the builder's retry ends
	}
	return v.Say(seat, EventRemoved, ev)
}

// otherTags are the tags of name but version that name digest.
func (v *Verbs) otherTags(name, version, digest string) ([]string, error) {
	tags, err := v.Catalog.Versions(name)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, t := range tags {
		if t == version {
			continue
		}
		d, err := v.Catalog.Digest(name, t)
		if err != nil {
			return nil, err
		}
		if d == digest {
			out = append(out, t)
		}
	}
	return out, nil
}

// HoldRequest is `hold` and `release`.
type HoldRequest struct {
	Name   string `json:"name"`
	Kind   string `json:"kind"`
	Digest string `json:"digest"`
	Why    string `json:"why,omitempty"`
}

// HoldOne is `hold`: the artifact sweep's holder alone; an image is never held (its own rules keep images).
func (v *Verbs) HoldOne(seat, caller string, r HoldRequest) error {
	if err := v.Callers.Allowed(VerbHold, caller); err != nil {
		return err
	}
	// The seat's contract names no kind: on the image seat a hold is of a copied base (upstream/…), never an image of
	// its own; on the artifact seat of a bundle or an archive.
	if r.Kind == "" && seat == ImageSeat && strings.HasPrefix(r.Name, "upstream/") {
		r.Kind = KindUpstream
	}
	switch {
	case r.Kind == "" && seat == ArtifactSeat:
	case r.Kind == KindBundle, r.Kind == KindArchive, r.Kind == KindUpstream:
	default:
		return fmt.Errorf("hold: refused: a %s is never held; only an untagged bundle, archive or copied base", r.Kind)
	}
	if r.Name == "" || !strings.HasPrefix(r.Digest, "sha256:") {
		return errors.New("hold: a name and a sha256 digest are required")
	}
	h := Hold{Seat: seat, Name: r.Name, Kind: r.Kind, Digest: r.Digest, By: caller, Why: r.Why, Since: v.now()}
	if err := v.Holds.Put(h); err != nil {
		return fmt.Errorf("hold: the holds did not take it: %w", err)
	}
	return v.Say(seat, EventHeld, Event{Name: r.Name, Kind: r.Kind, Digest: r.Digest, Holder: Self, By: caller})
}

// Release is `release`: the artifact sweep's holder, or the operator's session through registry_holds (a hand act,
// recorded by the tool); distribution's own account is never on its list.
func (v *Verbs) Release(seat, caller string, r HoldRequest) (bool, error) {
	if CallerModule(caller) == Self && !strings.HasPrefix(caller, "person.") {
		return false, fmt.Errorf("release: refused for %s: distribution cannot release in its own name", caller)
	}
	if err := v.Callers.Allowed(VerbRelease, caller); err != nil {
		return false, err
	}
	key := Hold{Seat: seat, Name: r.Name, Digest: r.Digest}.Key()
	h, held, err := v.Holds.Get(key)
	if err != nil {
		return false, err
	}
	if !held {
		return false, nil // nothing held: the outcome asked holds already, and nothing is said (the seat's definition)
	}
	if err := v.Holds.Delete(key); err != nil {
		return false, err
	}
	return true, v.Say(seat, EventReleased, Event{Name: h.Name, Kind: h.Kind, Digest: h.Digest, Holder: Self, By: caller, Reason: r.Why})
}

// Tags is `tags` on the image seat: the tags of an image.
func (v *Verbs) Tags(seat, caller, name string) (VersionsAnswer, error) {
	if seat != ImageSeat {
		return VersionsAnswer{}, fmt.Errorf("tags: %s serves no tags", seat)
	}
	if err := v.Callers.Allowed(VerbTags, caller); err != nil {
		return VersionsAnswer{}, err
	}
	got, err := v.Catalog.Versions(name)
	if err != nil {
		return VersionsAnswer{}, err
	}
	return VersionsAnswer{Seat: seat, Name: name, Versions: got}, nil
}

// MirrorRequest is `mirror`: an outside image by its pinned digest; By is the proved caller, kept with the ask.
type MirrorRequest struct {
	Source string `json:"source"`
	Digest string `json:"digest"`
	By     string `json:"by,omitempty"`
	// ID is the ask's own id, made when it is asked: the id of the event that says its end.
	ID string `json:"id,omitempty"`
	// Claim is the follower that took the copy on: none until one claims it.
	Claim string `json:"claim,omitempty"`
}

// MirrorsKey is an ask's key in the state mirrors: the copy's name and digest, in a key's characters.
func MirrorsKey(name, digest string) string {
	return strings.NewReplacer(":", "_", "@", "_").Replace(name + "/" + digest)
}

// Mirror is `mirror` on the image seat: mesh-build-image alone; an image pinned by no digest is refused. It answers at
// once with the reference the copy will have and keeps the ask in the state mirrors; the copy runs on that state's
// watch (Mirrored), which says published or refused on the seat's events when it ends. The caller waits on that
// event, not on the call, so a large base never outlasts a call's bound.
func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) {
	if seat != ImageSeat {
		return "", fmt.Errorf("mirror: %s serves no mirror", seat)
	}
	if err := v.Callers.Allowed(VerbMirror, caller); err != nil {
		return "", err
	}
	if r.Source == "" || !strings.HasPrefix(r.Digest, "sha256:") {
		return "", errors.New("mirror: refused: an upstream image is copied only by its pinned digest")
	}
	src, err := ParseSource(r.Source)
	if err != nil {
		return "", err
	}
	r.By, r.Claim = caller, ""
	if r.ID == "" {
		id := make([]byte, 8)
		_, _ = rand.Read(id)
		r.ID = "mirror-" + hex.EncodeToString(id)
	}
	if err := v.Mirrors.Ask(MirrorsKey(src.Name(), r.Digest), r); err != nil {
		return "", fmt.Errorf("mirror: the copy of %s@%s cannot be taken on: %w", r.Source, r.Digest, err)
	}
	return SeatReference(seat, v.Catalog.Reference(src.Name(), r.Digest)), nil
}

// Mirrored runs one asked copy and says its outcome on the image seat's events: published with the reference, or
// refused with the reason; the ask leaves the state either way. An outcome that cannot be said keeps the ask, so the
// next watch takes it on again.
func (v *Verbs) Mirrored(r MirrorRequest) error {
	src, err := ParseSource(r.Source)
	if err != nil {
		return err
	}
	e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: r.By, ID: r.ID}
	event := EventPublished
	path, err := v.Catalog.Mirror(r.Source, r.Digest)
	e.Reference = SeatReference(ImageSeat, path)
	if err != nil {
		event, e.Reference, e.Reason = EventRefused, "", fmt.Sprintf("%s@%s was not copied: %v", r.Source, r.Digest, err)
	}
	if err := v.Say(ImageSeat, event, e); err != nil {
		return fmt.Errorf("mirror: the outcome of %s@%s cannot be said: %w", r.Source, r.Digest, err)
	}
	return v.Mirrors.Delete(MirrorsKey(src.Name(), r.Digest))
}

// Take decides one change of the state mirrors for the follower self: an ask no follower claimed, or one claimed
// before this follower's watch began (its claimant may have been cut by a restart), is claimed by compare-and-set at
// revision and copied by the follower that won; an ask claimed since, by self or by another, is passed. Two followers
// that both start on the same cut ask can both win in turn; the outcome then carries the ask's id, so it is said once.
func (v *Verbs) Take(key string, r MirrorRequest, revision uint64, current bool, self string) error {
	if r.Claim != "" && (r.Claim == self || !current) {
		return nil
	}
	r.Claim = self
	won, err := v.Mirrors.Claim(key, r, revision)
	if err != nil {
		return fmt.Errorf("mirror: the ask %s cannot be claimed: %w", key, err)
	}
	if !won {
		return nil
	}
	return v.Mirrored(r)
}

// MirrorsNow lists every copy `mirror` took on and has not ended, by its key, for registry_mirrors: what waits (no
// claim) and what a follower took on (its claim).
func (v *Verbs) MirrorsNow() (map[string]MirrorRequest, error) { return v.Mirrors.List() }

// MirrorAgain asks a kept copy again by hand (registry_mirrors, a hand act): its claim is let go, so the state's watch
// hands it to a follower, which takes it on as a new ask. A copy whose follower was cut, or that a follower holds and
// does not end, goes on this way without a restart.
func (v *Verbs) MirrorAgain(key string) (MirrorRequest, error) {
	all, err := v.Mirrors.List()
	if err != nil {
		return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err)
	}
	r, ok := all[key]
	if !ok {
		return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key)
	}
	r.Claim = ""
	if err := v.Mirrors.Put(key, r); err != nil {
		return MirrorRequest{}, fmt.Errorf("mirror: %s is not asked again: %w", key, err)
	}
	return r, nil
}

// MirrorDrop drops a kept copy by hand (registry_mirrors, a hand act with who and why): it says refused on the image
// seat's events with that reason, under the ask's id, so a waiter on the copy ends, and the ask leaves the state.
func (v *Verbs) MirrorDrop(key, by, why string) (MirrorRequest, error) {
	if strings.TrimSpace(why) == "" {
		return MirrorRequest{}, errors.New("mirror: a drop is a hand act: say why")
	}
	all, err := v.Mirrors.List()
	if err != nil {
		return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err)
	}
	r, ok := all[key]
	if !ok {
		return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key)
	}
	src, err := ParseSource(r.Source)
	if err != nil {
		return MirrorRequest{}, err
	}
	e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: by, ID: r.ID,
		Reason: fmt.Sprintf("%s@%s was not copied: dropped by hand by %s: %s", r.Source, r.Digest, by, why)}
	if err := v.Say(ImageSeat, EventRefused, e); err != nil {
		return MirrorRequest{}, fmt.Errorf("mirror: the drop of %s cannot be said, so the ask stays: %w", key, err)
	}
	return r, v.Mirrors.Delete(key)
}

// HeldNow lists the holds, for registry_holds.
func (v *Verbs) HeldNow() ([]Hold, error) {
	all, err := v.Holds.List()
	if err != nil {
		return nil, err
	}
	slices.SortFunc(all, func(a, b Hold) int { return strings.Compare(a.Key(), b.Key()) })
	return all, nil
}