mesh-controller/cmd/mesh-controller/composed_writer.go

package main

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
)

// The compositions (novox/hq ADR 0539 clauses 1 to 5). Every act that sent a declaration writes compositions instead:
// the controller composes a node as it composed a send (composeSend: the same composition, the same makes), splits the
// declaration by module into one composition per module at the fingerprint the module is registered at, and the node's
// own resources, signs each with its key, and writes those that differ, by compare-and-set while it holds the lease:
// the composition into the node's bucket `composed-<node>`, which only that node reads, and its digest with the
// judgement of the move (composed_judge.go) into `mesh-controller_composed`, which the assigner reads. A file over 64
// KiB goes by digest into the node's object bucket first. Nothing is published on a node's declaration subject.
//
// **A module with no fingerprint of the new form goes with the node's own composition.** The mesh composes resources
// of modules it provides itself, built by nobody (the private network's mesh-wireguard), and a module registered
// before fingerprints of the new form has none either: no want can ever name such a module (the assigner's first rule
// asks for its done build), so its resources ride in `<node>.node`, which the node's own want names, their `module`
// taken off (a composition holds only its own module's resources); their ids still name the module.
//
// **Derived state, pruned** (clause 4): a module's compositions on a node are kept for its five newest fingerprints and
// for any a want or the node's have names; the rest are deleted with their digests.

// composedBucketsKV is what the writer needs of a key-value bucket: jetstream.KeyValue, or a test's.
type composedBucketsKV interface {
	Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)
	Create(ctx context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error)
	Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
	Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error
	ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error)
}

// composedPlace is where the writer writes a node's compositions and reads what decides them: the node's bucket, the
// digests, the node's object bucket, the wants and the haves.
type composedPlace struct {
	node    string
	bucket  composedBucketsKV
	digests composedBucketsKV
	objects objectStore
	// wanted is the fingerprint the node's want names for a module, "" when none; ran the composition digest its have
	// names (the move's judgement reads it) and the fingerprint it runs.
	wanted func(ctx context.Context, module string) (string, error)
	ran    func(ctx context.Context, module string) (sdkdelivery.Have, bool, error)
	// lastApplied is the module's slice of the declaration the node last applied, from the kept stream of declarations
	// (ADR 0540 clause 9), and that declaration's digest; false when there is none to read. Nil judges without it.
	lastApplied func(ctx context.Context, module string) (slice []json.RawMessage, digest string, found bool, err error)
	// policy is a module's upgrade policy, where it comes from and why (catalogue.UpgradeOf), carried beside the
	// judgement for the assigner's hold (ADR 0539 clause 14).
	policy func(module string) (policy, from, why string)
	// sign signs a composition with the controller's key.
	sign func(c sdkdelivery.Composed) (sdkdelivery.Composed, error)
}

// compositionsOf splits a node's declaration body, as a send composed it, into its compositions: each module's at the
// fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names
// no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest
// declares data.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string) (
	out map[string]sdkdelivery.Composed, moved []string, err error) {

	var envelope struct {
		Resources   []map[string]json.RawMessage `json:"resources"`
		Adoption    *sdkdecl.Adoption            `json:"adoption"`
		Filled      []string                     `json:"filled_files"`
		Provisioner string                       `json:"provisioner_account"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		return nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
	}
	inModule := map[string]bool{}
	resources := make([]json.RawMessage, 0, len(envelope.Resources))
	for _, r := range envelope.Resources {
		var module string
		if raw, ok := r["module"]; ok {
			_ = json.Unmarshal(raw, &module)
		}
		if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
			delete(r, "module")
			if !inModule[module] {
				inModule[module] = true
				moved = append(moved, module)
			}
		}
		raw, err := json.Marshal(r)
		if err != nil {
			return nil, nil, err
		}
		resources = append(resources, raw)
	}
	// A filled file of a module moved into the node's own composition is no module's file any more: the provisioner
	// fills it all the same, but a composition carries only its own module's filled paths.
	var filled []string
	for _, p := range envelope.Filled {
		if !filledByMoved(envelope.Resources, p, inModule) {
			filled = append(filled, p)
		}
	}
	sort.Strings(moved)
	out, err = sdkdelivery.SplitComposed(node, resources, fps, filled,
		sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
	if err != nil {
		return nil, nil, err
	}
	// A module the node runs that composes no resource of its own (all its needs met by others) still has its
	// composition, empty, at its fingerprint: its want names that composition (the assigner's seventh rule), and without
	// one no want could ever be written.
	for _, module := range runs {
		fp := fps[module]
		if !strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
			continue
		}
		key, err := sdkdelivery.ComposedKey(node, module, fp)
		if err != nil {
			continue
		}
		if _, ok := out[key]; !ok {
			out[key] = sdkdelivery.Composed{Node: node, Module: module, FP: fp, Resources: []json.RawMessage{}}
		}
	}
	for key, c := range out {
		if m, ok := manifests[c.Module]; ok && c.Module != "" {
			c.Data = len(m.DataItems()) > 0
			out[key] = c
		}
	}
	return out, moved, nil
}

// filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.
func filledByMoved(resources []map[string]json.RawMessage, path string, moved map[string]bool) bool {
	for _, r := range resources {
		var kind, at, id string
		_ = json.Unmarshal(r["type"], &kind)
		_ = json.Unmarshal(r["path"], &at)
		_ = json.Unmarshal(r["id"], &id)
		if kind == string(sdkdecl.TypeFile) && at == path {
			if module, ok := sdkdecl.ModuleGuess(id); ok && moved[module] {
				return true
			}
		}
	}
	return false
}

// writeComposed writes one composition and its digest where they differ from what the buckets hold: the files by
// digest first, then the composition, then its digest and judgement, so a reader that finds the digest finds the
// composition and its files. It answers whether it wrote anything.
func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdelivery.Composed) (bool, error) {
	c, objects, err := c.ByDigest()
	if err != nil {
		return false, err
	}
	if len(objects) > 0 {
		if at.objects == nil {
			return false, fmt.Errorf("%s holds a file over 64 KiB and %s's object bucket is not open", key, at.node)
		}
		named := make(map[string][]byte, len(objects))
		for digest, body := range objects {
			name, err := sdkdecl.ObjectName(digest)
			if err != nil {
				return false, err
			}
			named[name] = body
		}
		if err := putObjects(ctx, at.objects, named); err != nil {
			return false, err
		}
	}
	digest, err := sdkdelivery.ComposedDigestOf(c)
	if err != nil {
		return false, err
	}
	wrote := false
	held, revision, err := readComposed(ctx, at.bucket, key)
	if err != nil {
		return false, err
	}
	if heldDigest, _ := sdkdelivery.ComposedDigestOf(held); revision == 0 || heldDigest != digest {
		signed, err := at.sign(c)
		if err != nil {
			return false, err
		}
		value, err := sdkdelivery.EncodeComposed(signed)
		if err != nil {
			return false, err
		}
		if err := casWrite(ctx, at.bucket, key, value, revision); err != nil {
			return false, err
		}
		wrote = true
	}
	judged, err := judgeMove(ctx, at, key, c, digest)
	if composeTrace != nil {
		composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote,
			judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
	}
	if err != nil {
		return wrote, err
	}
	written, err := keepDigest(ctx, at.digests, key, judged)
	return wrote || written, err
}

// readComposed is the composition a bucket holds under key and its revision; revision 0 when there is none.
func readComposed(ctx context.Context, kv composedBucketsKV, key string) (sdkdelivery.Composed, uint64, error) {
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return sdkdelivery.Composed{}, 0, nil
	}
	if err != nil {
		return sdkdelivery.Composed{}, 0, err
	}
	c, err := sdkdelivery.DecodeComposed(e.Value())
	if err != nil {
		// A value this controller cannot read is written over: it is derived, and composed again here.
		return sdkdelivery.Composed{}, e.Revision(), nil
	}
	return c, e.Revision(), nil
}

// keepDigest writes the digest and judgement beside a composition when it differs from what the bucket holds.
func keepDigest(ctx context.Context, kv composedBucketsKV, key string, v sdkdelivery.ComposedDigest) (bool, error) {
	value, err := sdkdelivery.EncodeComposedDigest(v)
	if err != nil {
		return false, err
	}
	v.V = sdkdelivery.Version
	var revision uint64
	e, err := kv.Get(ctx, key)
	switch {
	case errors.Is(err, jetstream.ErrKeyNotFound):
	case err != nil:
		return false, err
	default:
		revision = e.Revision()
		if held, derr := sdkdelivery.DecodeComposedDigest(e.Value()); derr == nil && sdkdelivery.SameComposedDigest(held, v) {
			return false, nil
		}
	}
	return true, casWrite(ctx, kv, key, value, revision)
}

// casWrite creates key when revision is 0 and otherwise replaces the revision read: a second writer that wrote in
// between makes this one fail, said, and the next event composes again.
func casWrite(ctx context.Context, kv composedBucketsKV, key string, value []byte, revision uint64) error {
	var err error
	if revision == 0 {
		_, err = kv.Create(ctx, key, value)
	} else {
		_, err = kv.Update(ctx, key, value, revision)
	}
	if err != nil {
		return fmt.Errorf("writing %s: %w", key, err)
	}
	return nil
}

// writeNodeCompositions writes every composition of a node that differs, and prunes what is no longer kept. It answers
// the keys it wrote.
func writeNodeCompositions(ctx context.Context, at composedPlace, comps map[string]sdkdelivery.Composed) ([]string, error) {
	keys := make([]string, 0, len(comps))
	for k := range comps {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	var wrote []string
	var errs []error
	for _, key := range keys {
		w, err := writeComposed(ctx, at, key, comps[key])
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if w {
			wrote = append(wrote, key)
		}
	}
	if err := pruneCompositions(ctx, at); err != nil {
		errs = append(errs, err)
	}
	return wrote, errors.Join(errs...)
}

// pruneCompositions deletes, for each module of the node, the compositions outside its ComposedKeep newest fingerprints
// that neither its want nor its have names, each with its digest (ADR 0539 clause 4). Newest is by when each was last
// written: the controller writes only the fingerprint a module is registered at, so an older one is never written again.
func pruneCompositions(ctx context.Context, at composedPlace) error {
	lister, err := at.bucket.ListKeys(ctx)
	if err != nil {
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			return nil
		}
		return err
	}
	type kept struct {
		key, fp string
		at      time.Time
	}
	byModule := map[string][]kept{}
	for key := range lister.Keys() {
		node, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
		if err != nil || isNode || node != at.node {
			continue
		}
		e, err := at.bucket.Get(ctx, key)
		if err != nil {
			continue
		}
		byModule[module] = append(byModule[module], kept{key, fp, e.Created()})
	}
	var errs []error
	for module, ks := range byModule {
		if len(ks) <= sdkdelivery.ComposedKeep {
			continue
		}
		sort.Slice(ks, func(i, j int) bool { return ks[i].at.After(ks[j].at) })
		wanted, err := at.wanted(ctx, module)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		have, _, err := at.ran(ctx, module)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		for _, k := range ks[sdkdelivery.ComposedKeep:] {
			if k.fp == wanted || k.fp == have.FP {
				continue
			}
			if err := at.bucket.Delete(ctx, k.key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
				errs = append(errs, err)
				continue
			}
			if err := at.digests.Delete(ctx, k.key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
				errs = append(errs, err)
			}
		}
	}
	return errors.Join(errs...)
}

// controllerSigner signs a composition with the controller's active key, which its identity store keeps and never hands
// out (as it signed declarations).
func controllerSigner(ctx context.Context, open *stores) (func(sdkdelivery.Composed) (sdkdelivery.Composed, error), error) {
	ident, err := open.Identity(ctx)
	if err != nil {
		return nil, err
	}
	key, err := ident.Active(ctx)
	if err != nil {
		return nil, err
	}
	public := ed25519.PublicKey(key.Public)
	return func(c sdkdelivery.Composed) (sdkdelivery.Composed, error) {
		return sdkdelivery.SignComposedBy(c, public, func(m []byte) ([]byte, error) { return ident.Sign(ctx, m) })
	}, nil
}