mesh-delivery/modules/mesh-delivery-assigner/internal/assigngraph/keep.go

package assigngraph

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

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

// RevisionsObjects is the local name of the kept data in the assigner's manifest's old `objects` section (design 47
// lines 1362-1366: both foundation parts keep their revisions as objects in the old form; the seat mesh-wants' state
// assignment-revisions names its key forms).
const RevisionsObjects = "assignment-revisions"

// ErrNoObject is an object the bucket does not hold.
var ErrNoObject = errors.New("no such object")

// Objects is an object bucket as the assigner keeps its revisions in it.
type Objects interface {
	// Put stores data whole under name, replacing one of that name.
	Put(ctx context.Context, name string, data []byte) error
	// Get is the object's bytes; ErrNoObject when it holds none of that name.
	Get(ctx context.Context, name string) ([]byte, error)
	// Names is every object's name; none is an empty list.
	Names(ctx context.Context) ([]string, error)
}

// EntryObject is the object name of the entry of that digest (the seat's key form entry.<digest>).
func EntryObject(digest string) string { return "entry." + digest }

// TrunkObject is the object name of a trunk revision (the seat's key form trunk.<revision>).
func TrunkObject(revision int64) string { return "trunk." + strconv.FormatInt(revision, 10) }

// trunkOf is the revision number a trunk object's name holds, and whether it is one.
func trunkOf(name string) (int64, bool) {
	n, ok := strings.CutPrefix(name, "trunk.")
	if !ok {
		return 0, false
	}
	r, err := strconv.ParseInt(n, 10, 64)
	return r, err == nil && r > 0
}

// Newest is the newest trunk revision kept, its list only; none kept is false.
func Newest(ctx context.Context, o Objects) (meshwants.AssignmentRevision, bool, error) {
	names, err := o.Names(ctx)
	if err != nil {
		return meshwants.AssignmentRevision{}, false, fmt.Errorf("the objects %s: %w", RevisionsObjects, err)
	}
	var newest int64
	for _, n := range names {
		if r, ok := trunkOf(n); ok && r > newest {
			newest = r
		}
	}
	if newest == 0 {
		return meshwants.AssignmentRevision{}, false, nil
	}
	rev, err := ReadRevision(ctx, o, newest)
	return rev, err == nil, err
}

// ReadRevision is the trunk revision of that number, its list only.
func ReadRevision(ctx context.Context, o Objects, revision int64) (meshwants.AssignmentRevision, error) {
	var rev meshwants.AssignmentRevision
	data, err := o.Get(ctx, TrunkObject(revision))
	if err != nil {
		return rev, fmt.Errorf("revision %d: %w", revision, err)
	}
	if err := json.Unmarshal(data, &rev); err != nil {
		return rev, fmt.Errorf("revision %d: %w", revision, err)
	}
	return rev, nil
}

// ReadEntry is the entry kept under that digest.
func ReadEntry(ctx context.Context, o Objects, digest string) (meshwants.AssignmentEntry, error) {
	var e meshwants.AssignmentEntry
	data, err := o.Get(ctx, EntryObject(digest))
	if err != nil {
		return e, fmt.Errorf("the entry %s: %w", digest, err)
	}
	if err := json.Unmarshal(data, &e); err != nil {
		return e, fmt.Errorf("the entry %s: %w", digest, err)
	}
	return e, nil
}

// Keep keeps c as the next trunk revision, numbered after the newest kept, unless the newest kept lists the same entries
// from the same inputs (Same): then it keeps nothing and answers the newest and false. Each entry object not held yet is
// written before the list, so a list is never kept whose entries are not, and an entry two revisions share is stored
// once (design 47 *An unchanged entry is stored once*).
func Keep(ctx context.Context, o Objects, c Computed, cause string, at time.Time) (meshwants.AssignmentRevision, bool, error) {
	newest, has, err := Newest(ctx, o)
	if err != nil {
		return newest, false, err
	}
	if has && Same(newest, c.Revision) {
		return newest, false, nil
	}
	rev := c.Revision
	rev.Revision = newest.Revision + 1
	rev.Cause = cause
	t := at.UTC()
	rev.At = &t
	if err := writeEntries(ctx, o, c, rev); err != nil {
		return newest, false, err
	}
	list, err := json.Marshal(rev)
	if err != nil {
		return newest, false, err
	}
	if err := o.Put(ctx, TrunkObject(rev.Revision), list); err != nil {
		return newest, false, fmt.Errorf("revision %d in %s: %w", rev.Revision, RevisionsObjects, err)
	}
	return rev, true, nil
}

// writeEntries writes each entry object rev lists that the bucket does not hold yet, from c, before its list is kept.
func writeEntries(ctx context.Context, o Objects, c Computed, rev meshwants.AssignmentRevision) error {
	names, err := o.Names(ctx)
	if err != nil {
		return fmt.Errorf("the objects %s: %w", RevisionsObjects, err)
	}
	held := make(map[string]bool, len(names))
	for _, n := range names {
		held[n] = true
	}
	for _, ref := range rev.Entries {
		name := EntryObject(ref.Digest)
		if held[name] {
			continue
		}
		e, ok := c.Entries[ref.Digest]
		if !ok {
			return fmt.Errorf("revision %d lists the entry of %s/%s (%s), which it does not carry", rev.Revision, ref.Node, ref.Module, ref.Digest)
		}
		data, err := json.Marshal(e)
		if err != nil {
			return err
		}
		if err := o.Put(ctx, name, data); err != nil {
			return fmt.Errorf("the entry of %s/%s in %s: %w", ref.Node, ref.Module, RevisionsObjects, err)
		}
		held[name] = true
	}
	return nil
}

// Event is the seat's event assignment-revision of a kept revision.
func Event(rev meshwants.AssignmentRevision) meshwants.AssignmentRevisionEvent {
	return meshwants.AssignmentRevisionEvent{V: rev.V, Revision: rev.Revision, Kind: rev.Kind, Cause: rev.Cause, Group: rev.Group,
		Heads: rev.Heads, Catalogs: rev.Catalogs, At: rev.At}
}