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

package holder

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

	"git.novox.be/novox/mesh-sdk/go/parts"
)

// The bounds of incoming (design 51 *Publishing*, ADR 0611): a part is at most 1 MiB, the store 20 GiB, a transfer
// with no part for 24 hours is dropped, and past 80 % of the cap the holder writes a finding.
const (
	MaxPart         = 1 << 20
	IncomingCap     = 20 << 30
	TransferIdle    = 24 * time.Hour
	IncomingFullPct = 80
	// IncomingObjects and TransfersState are the object store and the state distribution asks as bus state.
	IncomingObjects = "incoming"
	TransfersState  = "transfers"
)

// The kinds a seat holds.
const (
	KindImage    = "image"
	KindBundle   = "bundle"
	KindArchive  = "archive"
	KindUpstream = "upstream"
)

// Part is one file of a publish: an image's layer, config or manifest, a bundle's or an archive's blob, by its digest.
type Part struct {
	Digest string `json:"digest"`
	Size   int64  `json:"size"`
	// MediaType says what the part is to the store; an image's manifest names its own.
	MediaType string `json:"media_type,omitempty"`
}

// PublishRequest is the body of `publish` (ADR 0611 §1).
type PublishRequest struct {
	Transfer    string `json:"transfer"`
	Name        string `json:"name"`
	Version     string `json:"version,omitempty"`
	Tag         string `json:"tag,omitempty"`
	Kind        string `json:"kind"`
	Fingerprint string `json:"fingerprint,omitempty"`
	Commit      string `json:"commit,omitempty"`
	// Digest is the digest the build recorded: an image's manifest, a bundle's or an archive's blob.
	Digest string `json:"digest"`
	Parts  []Part `json:"parts"`
}

// PublishAnswer is what `publish` answers: the parts the holder lacks, which the build module sends; none when it
// holds them all and the publish ended (Published names how).
type PublishAnswer struct {
	Transfer string   `json:"transfer"`
	Missing  []string `json:"missing"`
	// Ended says the publish ended at once: everything was held, `published` or `refused` is said.
	Ended bool `json:"ended,omitempty"`
}

// PartRequest is the body of `publish-part` (ADR 0611 §2): one chunk of one part.
type PartRequest struct {
	Transfer string `json:"transfer"`
	Digest   string `json:"digest"`
	Sequence int    `json:"sequence"`
	Last     bool   `json:"last"`
	Data     []byte `json:"data"`
}

// Opening is what distribution keeps of a publish beside the receiving side's own fields: the request and its seat.
type Opening struct {
	PublishRequest
	Seat string `json:"seat"`
}

// Transfer is one open publish, kept in the state transfers; its chunks are objects of incoming. The receiving side is
// mesh-sdk's seat-neutral parts.Receiver (ADR 0627 §2); distribution binds it to the seats it holds.
type Transfer = parts.Transfer[Opening]

// Event is the body of `published`, `refused` and `removed` on `<seat>.events` (design 51 *One contract*): the name,
// version, kind, tag, digest, fingerprint, commit and holder; `refused` its reason, `removed` its reason too.
type Event struct {
	Name        string `json:"name"`
	Version     string `json:"version,omitempty"`
	Kind        string `json:"kind"`
	Tag         string `json:"tag,omitempty"`
	Digest      string `json:"digest"`
	Fingerprint string `json:"fingerprint,omitempty"`
	Commit      string `json:"commit,omitempty"`
	Holder      string `json:"holder"`
	Transfer    string `json:"transfer,omitempty"`
	// Reference is the seat's answer: `<registry>/<module>/<artifact>@sha256:<digest>` (ADR 0605 §9).
	Reference string `json:"reference,omitempty"`
	Reason    string `json:"reason,omitempty"`
	// By is who asked a removal, hold or release: the caller the tool runner proved.
	By string `json:"by,omitempty"`
	// ID is the event's id on the bus when the outcome has one of its own (a mirror ask's): the same outcome said
	// twice, by a redelivery or a second follower, is one message. Empty: the runtime makes one.
	ID string `json:"-"`
}

// The events of the seats.
const (
	EventPublished = "published"
	EventRefused   = "refused"
	EventRemoved   = "removed"
	EventHeld      = "held"
	EventReleased  = "released"
)

// Objects is distribution's own object store incoming.
type Objects = parts.Objects

// TransferState is the state transfers.
type TransferState = parts.Transfers[Opening]

// Builder is the builder's tool builder_versions, asked before anything is stored (ADR 0605 §4).
type Builder interface {
	Recorded(name, version, digest string) (ok bool, why string, err error)
}

// Store is the artifact store behind the loopback, written by distribution's own process alone.
type Store interface {
	// Has says whether the store holds a blob of name.
	Has(name, digest string) (bool, error)
	// Put stores a publish's parts under its name and version and answers the reference's path (SeatReference).
	Put(r PublishRequest, parts map[string][]byte) (reference string, err error)
	// Held is the reference's path of name at version when the store already holds digest under it: a rebuild of a held
	// fingerprint answers it and stores nothing.
	Held(name, version, digest string) (reference string, ok bool, err error)
}

// SeatReference is a reference in the form the seat answers (novox/hq ADR 0605 §9): the seat's name, then the path the
// store answers (`<name>@sha256:<digest>`), as catalogue.InRegistrySeat reads it back. Nothing set on the process names
// a reach: the seat's name is the registry part.
func SeatReference(seat, path string) string { return seat + "/" + path }

// Say says an event on a seat's family `<seat>.events`.
type Say func(seat, event string, body Event) error

// Finding writes (raised true) or clears distribution's finding of a kind for a subject.
type Finding func(kind, subject, words string, raised bool)

// Holder answers the verbs.
type Holder struct {
	Callers   Callers
	Objects   Objects
	Transfers TransferState
	Builder   Builder
	Store     Store
	Say       Say
	Finding   Finding
	Now       func() time.Time
	// Cap is incoming's cap; IncomingCap when zero.
	Cap uint64
}

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

func (h *Holder) cap() uint64 {
	if h.Cap > 0 {
		return h.Cap
	}
	return IncomingCap
}

// receiver binds mesh-sdk's receiving side to distribution's incoming, its transfers state, the builder's check, its
// store and its seats' events.
func (h *Holder) receiver() *parts.Receiver[Opening] {
	return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: h.cap(), MaxChunk: MaxPart,
		Idle: TransferIdle, Full: IncomingFullPct, Now: h.now,
		// Only the missing layers are sent: a part the store already holds under the name counts as held.
		Holds: func(t Transfer, digest string) bool {
			ok, err := h.Store.Has(t.Shape.Name, digest)
			return err == nil && ok
		},
		// builder_versions is asked before anything is stored (ADR 0605 §4).
		Check: func(t Transfer) error {
			if t.Shape.Version == "" {
				return nil
			}
			ok, why, err := h.Builder.Recorded(t.Shape.Name, t.Shape.Version, t.Shape.Digest)
			if err != nil {
				return errors.New("builder_versions did not answer, so nothing is stored: " + err.Error())
			}
			if !ok {
				return errors.New("the builder did not number it: " + why)
			}
			return nil
		},
		Store: func(t Transfer, held map[string][]byte) error {
			path, err := h.Store.Put(t.Shape.PublishRequest, held)
			ref := SeatReference(t.Shape.Seat, path)
			if err != nil {
				reason := "the store did not take it: " + err.Error()
				return errors.Join(errors.New("refused: "+reason), h.refused(t, reason))
			}
			h.Finding("publish-refused", t.Shape.Name+"@"+t.Shape.Version, "", false)
			// Room was found again: a refusal for want of it clears with the first publish the holder takes.
			h.Finding("publish-refused", IncomingObjects, "", false)
			return h.Say(t.Shape.Seat, EventPublished, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, ref, ""))
		},
		Refuse: h.refused,
		Filled: func(used, cap uint64, full bool) {
			h.Finding("incoming-full", IncomingObjects, fmt.Sprintf("incoming holds %d of its %d bytes (%d %%); registry_incoming lists and drops transfers", used, cap, used*100/cap), full)
		}}
}

// refused writes the finding and says refused on the transfer's seat. A refusal for want of room is the holder's,
// not the file's: its finding stands on incoming and clears when the holder takes a publish again.
func (h *Holder) refused(t Transfer, reason string) error {
	subject := t.Shape.Name + "@" + t.Shape.Version
	if parts.RefusedAtCap(reason) {
		subject = IncomingObjects
	}
	h.Finding("publish-refused", subject, reason, true)
	return h.Say(t.Shape.Seat, EventRefused, h.eventOf(t.Shape.Seat, t.Shape.PublishRequest, "", reason))
}

func answerOf(r parts.Received) PublishAnswer {
	return PublishAnswer{Transfer: r.ID, Missing: r.Lacks, Ended: r.Ended}
}

// Publish is `publish` on seat for the proved caller.
func (h *Holder) Publish(seat, caller string, r PublishRequest) (PublishAnswer, error) {
	if err := h.Callers.Allowed(VerbPublish, caller); err != nil {
		return PublishAnswer{}, err
	}
	if err := checkPublish(seat, r); err != nil {
		return PublishAnswer{}, err
	}
	// A rebuild of a held fingerprint answers the held version and stores nothing.
	if r.Version != "" {
		if ref, ok, err := h.Store.Held(r.Name, r.Version, r.Digest); err != nil {
			return PublishAnswer{}, fmt.Errorf("publish: the store cannot say whether it holds %s@%s: %w", r.Name, r.Version, err)
		} else if ok {
			return PublishAnswer{Transfer: r.Transfer, Ended: true}, h.Say(seat, EventPublished, h.eventOf(seat, r, SeatReference(seat, ref), ""))
		}
	}
	wants := make([]parts.Want, 0, len(r.Parts))
	for _, p := range r.Parts {
		wants = append(wants, parts.Want{Digest: p.Digest, Size: p.Size})
	}
	got, err := h.receiver().Open(r.Transfer, caller, wants, Opening{PublishRequest: r, Seat: seat})
	if err != nil {
		return answerOf(got), fmt.Errorf("publish: %w", err)
	}
	return answerOf(got), nil
}

// PublishPart is `publish-part`: one chunk, from the caller that opened the transfer alone.
func (h *Holder) PublishPart(seat, caller string, r PartRequest) (PublishAnswer, error) {
	if err := h.Callers.Allowed(VerbPublishPart, caller); err != nil {
		return PublishAnswer{}, err
	}
	got, err := h.receiver().Take(r.Transfer, caller, r.Digest, r.Sequence, r.Last, r.Data, func(t Transfer) error {
		if t.Shape.Seat != seat {
			return fmt.Errorf("transfer %s is %s's, not %s's", r.Transfer, t.Shape.Seat, seat)
		}
		return nil
	})
	if err != nil {
		return answerOf(got), fmt.Errorf("publish-part: %w", err)
	}
	return answerOf(got), nil
}

func (h *Holder) eventOf(seat string, r PublishRequest, ref, reason string) Event {
	return Event{Name: r.Name, Version: r.Version, Kind: r.Kind, Tag: r.Tag, Digest: r.Digest, Fingerprint: r.Fingerprint,
		Commit: r.Commit, Holder: Self, Transfer: r.Transfer, Reference: ref, Reason: reason}
}

// Drop drops a transfer by hand (registry_incoming) or because it was idle for TransferIdle: its chunks and its key.
func (h *Holder) Drop(id string) (Transfer, error) { return h.receiver().Drop(id) }

// DropIdle drops every transfer with no part for TransferIdle, answering them; it runs on each publish's open and each
// registry_incoming call, never on a timer (the object store's own lifetime of 24 hours drops the bytes too).
func (h *Holder) DropIdle() ([]string, error) { return h.receiver().DropIdle() }

func checkPublish(seat string, r PublishRequest) error {
	switch {
	case !slices.Contains(Seats, seat):
		return fmt.Errorf("publish: distribution holds no seat %s", seat)
	case strings.TrimSpace(r.Name) == "":
		return errors.New("publish: no name")
	case !strings.HasPrefix(r.Digest, "sha256:"):
		return fmt.Errorf("publish: digest %q is no sha256 digest", r.Digest)
	}
	switch {
	case seat == ImageSeat && r.Kind != KindImage && r.Kind != KindUpstream:
		return fmt.Errorf("publish: %s holds images, not a %s", seat, r.Kind)
	case seat == ArtifactSeat && r.Kind != KindBundle && r.Kind != KindArchive:
		return fmt.Errorf("publish: %s holds bundles and archives, not a %s", seat, r.Kind)
	}
	return nil
}

func sum(b []byte) string {
	s := sha256.Sum256(b)
	return hex.EncodeToString(s[:])
}