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

package versions

import (
	"context"
	"errors"
	"fmt"
	"time"
)

// ErrConflict is what a Store answers when a Create finds the key or an Update finds another revision.
var ErrConflict = errors.New("the key exists or changed")

// Store is the bucket: create and compare-and-set, as the builder's builds bucket (internal/build.Store).
type Store interface {
	Get(ctx context.Context, key string) (value []byte, revision uint64, found bool, err error)
	Create(ctx context.Context, key string, value []byte) (uint64, error)
	Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
	// KeysOf are the keys under a filter (NameFilter).
	KeysOf(ctx context.Context, filter string) ([]string, error)
}

// Versions is the builder's state.
type Versions struct {
	Store Store
	Now   func() time.Time
}

// Of is every entry of name.
func (v Versions) Of(ctx context.Context, name string) ([]Entry, error) {
	keys, err := v.Store.KeysOf(ctx, NameFilter(name))
	if err != nil {
		return nil, fmt.Errorf("the versions of %s cannot be listed: %w", name, err)
	}
	var out []Entry
	for _, k := range keys {
		raw, _, found, err := v.Store.Get(ctx, k)
		if err != nil {
			return nil, fmt.Errorf("%s: %w", k, err)
		}
		if !found {
			continue
		}
		e, err := Decode(raw)
		if err != nil {
			return nil, fmt.Errorf("%s holds what is not a version: %w", k, err)
		}
		out = append(out, e)
	}
	return InOrder(out), nil
}

// Claim is one numbering asked of the builder.
type Claim struct {
	Name, Kind, Series, Module, Fingerprint, Commit, Revision, Digest, Seat string
	// Tag asks a tagged version for this cleaned tag; empty for a trunk version.
	Tag   string
	Merge uint64
	Floor Floor
}

// Number numbers a claim under compare-and-set: two claims of one name never get one number, the second reading the
// first's key and taking the next (ADR 0606 §2). It answers the decision and, when it numbered or kept a version, its
// entry. Attempts are bounded by the claims the name can meet at once; a store that keeps conflicting is an error.
func (v Versions) Number(ctx context.Context, c Claim) (Decision, Entry, error) {
	if c.Series == "" {
		c.Series = DefaultSeries
	}
	for attempt := 0; attempt < 16; attempt++ {
		entries, err := v.Of(ctx, c.Name)
		if err != nil {
			return Decision{}, Entry{}, err
		}
		var d Decision
		if c.Tag != "" {
			if !c.Floor.Known() && len(entries) == 0 {
				return Decision{Wait: true, Why: "the tagged version waits for its floor"}, Entry{}, nil
			}
			ver := Tagged(entries, c.Series, c.Tag, c.Floor)
			d = Decision{Version: ver, Publish: true, Why: "tagged " + ver}
		} else {
			d = Decide(entries, c.Kind, c.Series, c.Fingerprint, c.Revision, c.Floor)
		}
		if d.Version == "" {
			return d, Entry{}, nil
		}
		if d.Keep {
			for _, e := range entries {
				if e.Version == d.Version {
					return d, e, nil
				}
			}
		}
		e := Entry{Name: c.Name, Kind: c.Kind, Series: c.Series, Version: d.Version, Tag: c.Tag, Module: c.Module,
			Fingerprint: c.Fingerprint, Commit: c.Commit, Revision: c.Revision, Digest: c.Digest, Seat: c.Seat,
			State: Numbered, At: v.now(), Merge: c.Merge}
		raw, err := Encode(e)
		if err != nil {
			return Decision{}, Entry{}, err
		}
		_, err = v.Store.Create(ctx, Key(c.Name, d.Version), raw)
		if errors.Is(err, ErrConflict) {
			continue // another claim took this number: read again and take the next
		}
		if err != nil {
			return Decision{}, Entry{}, fmt.Errorf("%s@%s: %w", c.Name, d.Version, err)
		}
		return d, e, nil
	}
	return Decision{}, Entry{}, fmt.Errorf("%s: the versions kept changing under 16 claims", c.Name)
}

// Mark moves one version to a new state by compare-and-set, as the holders' events say (published, refused, removed):
// change edits the entry read; a version the state does not hold is an error naming it.
func (v Versions) Mark(ctx context.Context, name, version string, change func(*Entry)) (Entry, error) {
	key := Key(name, version)
	for attempt := 0; attempt < 16; attempt++ {
		raw, rev, found, err := v.Store.Get(ctx, key)
		if err != nil {
			return Entry{}, fmt.Errorf("%s: %w", key, err)
		}
		if !found {
			return Entry{}, fmt.Errorf("%s@%s is not numbered by the builder", name, version)
		}
		e, err := Decode(raw)
		if err != nil {
			return Entry{}, fmt.Errorf("%s holds what is not a version: %w", key, err)
		}
		change(&e)
		e.At = v.now()
		out, err := Encode(e)
		if err != nil {
			return Entry{}, err
		}
		if _, err := v.Store.Update(ctx, key, out, rev); errors.Is(err, ErrConflict) {
			continue
		} else if err != nil {
			return Entry{}, fmt.Errorf("%s: %w", key, err)
		}
		return e, nil
	}
	return Entry{}, fmt.Errorf("%s@%s kept changing under 16 marks", name, version)
}

// Recorded is the holder's question through builder_versions (ADR 0605 §4): whether name, version and digest are a
// version the builder numbered. It answers the entry when they are, and why not otherwise.
func (v Versions) Recorded(ctx context.Context, name, version, digest string) (Entry, bool, string, error) {
	raw, _, found, err := v.Store.Get(ctx, Key(name, version))
	if err != nil {
		return Entry{}, false, "", err
	}
	if !found {
		return Entry{}, false, fmt.Sprintf("%s@%s is not a version the builder numbered", name, version), nil
	}
	e, err := Decode(raw)
	if err != nil {
		return Entry{}, false, "", err
	}
	switch {
	case e.State == Removed:
		return e, false, fmt.Sprintf("%s@%s was removed (%s)", name, version, e.Reason), nil
	case e.Digest != "" && digest != "" && e.Digest != digest:
		return e, false, fmt.Sprintf("%s@%s is recorded with digest %s, not %s", name, version, e.Digest, digest), nil
	}
	return e, true, "", nil
}

func (v Versions) now() time.Time {
	if v.Now != nil {
		return v.Now().UTC()
	}
	return time.Now().UTC()
}