mesh-controller/cmd/mesh-controller/composed_live.go

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"

	"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"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/link"
)

// composeModules composes each named node and writes its compositions that differ (ADR 0539 clause 5): what each act
// that once sent a declaration calls instead. A node that cannot compose is that node's: the others are written, and
// every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that
// does not hold it writes nothing and says so.
func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) {
	out, byNode := composeModulesByNode(ctx, open, nodes)
	names := make([]string, 0, len(byNode))
	for n := range byNode {
		names = append(names, n)
	}
	sort.Strings(names)
	var errs []error
	for _, n := range names {
		errs = append(errs, byNode[n])
	}
	return out, errors.Join(errs...)
}

// composeModulesByNode is composeModules with each node's error apart: a failure before any node is composed is every
// node's (the composer gives each caller only the errors of the nodes it asked for, issue 1945).
func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (map[string][]string, map[string]error) {
	all := func(err error) (map[string][]string, map[string]error) {
		byNode := map[string]error{}
		for _, n := range nodes {
			byNode[n] = err
		}
		return nil, byNode
	}
	if _, err := theLease.epoch(ctx); err != nil {
		return all(fmt.Errorf("nothing is composed: %w", err))
	}
	bus, err := aBus()
	if err != nil {
		return all(fmt.Errorf("nothing is composed: the bus: %w", err))
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return all(err)
	}
	sign, err := controllerSigner(ctx, open)
	if err != nil {
		return all(fmt.Errorf("nothing is composed: the controller's signing key: %w", err))
	}
	fps, err := currentFingerprints(ctx, open.inventory)
	if err != nil {
		return all(err)
	}
	entries, err := open.inventory.Catalogued(ctx)
	if err != nil {
		return all(err)
	}
	manifests := make(map[string]catalogue.Manifest, len(entries))
	for _, e := range entries {
		manifests[e.Manifest.Module] = e.Manifest
	}
	upgrades, err := open.inventory.Upgrades(ctx)
	if err != nil {
		return all(err)
	}
	policy := func(module string) (string, string, string) {
		u, ok := upgrades[module]
		if !ok {
			return "", "", ""
		}
		return u.Policy(), u.From, u.Why
	}
	sort.Strings(nodes)
	out := map[string][]string{}
	errsOf := map[string][]error{}
	for _, node := range nodes {
		// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
		unlock := lockComposing(node)
		func() {
			defer unlock()
			at, err := openComposedPlace(ctx, js, bus.Conn(), node)
			if err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
				return
			}
			at.sign, at.policy = sign, policy
			if composeTrace != nil {
				traceInput(ctx, open, node)
			}
			declared, err := composeSend(ctx, open, node)
			if err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
				return
			}
			body, err := compositionBody(declared)
			if err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
				return
			}
			// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
			if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
			}
			comps, moved, left, err := compositionsOf(node, body, fps, manifests, declared.modules, namedByOwnComposition(ctx, at))
			if err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
				return
			}
			if len(moved) > 0 {
				fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
					node, strings.Join(moved, ", "))
			}
			if len(left) > 0 {
				fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
					"new form for them, and the node's want or have names their own composition (issue 1945)\n",
					node, strings.Join(left, ", "))
			}
			if composeTrace != nil {
				traceComposed(ctx, at, node, comps)
			}
			wrote, err := writeNodeCompositions(ctx, at, comps)
			if len(wrote) > 0 {
				out[node] = wrote
				fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
			}
			// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
			if len(declared.Bindings) > 0 {
				if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
					errsOf[node] = append(errsOf[node], fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
				}
			}
			if err != nil {
				errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
			}
		}()
	}
	byNode := map[string]error{}
	for node, errs := range errsOf {
		if len(errs) > 0 {
			byNode[node] = errors.Join(errs...)
		}
	}
	return out, byNode
}

// namedByOwnComposition says a module's want or have on the node names a composition of its own: a fingerprint of the
// new form (issue 1945).
func namedByOwnComposition(ctx context.Context, at composedPlace) func(module string) (bool, error) {
	return func(module string) (bool, error) {
		if at.wanted != nil {
			fp, err := at.wanted(ctx, module)
			if err != nil {
				return false, err
			}
			if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
				return true, nil
			}
		}
		if at.ran != nil {
			h, found, err := at.ran(ctx, module)
			if err != nil {
				return false, err
			}
			if found && strings.HasPrefix(h.FP, sdkdelivery.SourcePrefix2) {
				return true, nil
			}
		}
		return false, nil
	}
}

// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
// of a verb or an event calls.
func composeAndSay(ctx context.Context, open *stores, nodes []string) error {
	if len(nodes) == 0 {
		return nil
	}
	_, err := composeModules(ctx, open, nodes)
	if err != nil {
		fmt.Printf("compositions: %v\n", err)
	}
	return err
}

// openComposedPlace opens a node's buckets: its compositions, the digests, its files, and reads its want and have.
func openComposedPlace(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, node string) (composedPlace, error) {
	if err := broker.EnsureComposed(ctx, js, node); err != nil {
		return composedPlace{}, err
	}
	name, err := sdkdelivery.ComposedBucketFor(node)
	if err != nil {
		return composedPlace{}, err
	}
	bucket, err := js.KeyValue(ctx, name)
	if err != nil {
		return composedPlace{}, fmt.Errorf("the bucket %s: %w", name, err)
	}
	digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
	if err != nil {
		return composedPlace{}, fmt.Errorf("the bucket %s: %w", sdkdelivery.ControllerComposedBucket, err)
	}
	objects, err := openBucket(ctx, conn, node)
	if err != nil {
		return composedPlace{}, err
	}
	return composedPlace{node: node, bucket: bucket, digests: digests, objects: objects,
		wanted: func(ctx context.Context, module string) (string, error) { return wantedOn(ctx, js, node, module) },
		ran: func(ctx context.Context, module string) (sdkdelivery.Have, bool, error) {
			return haveOn(ctx, js, node, module)
		},
		lastApplied: lastAppliedOn(js, node),
	}, nil
}

// wantedOn is the fingerprint the assigner's want names for a module on a node; "" when there is none or the bucket is
// not there yet.
func wantedOn(ctx context.Context, js jetstream.JetStream, node, module string) (string, error) {
	kv, err := js.KeyValue(ctx, sdkdelivery.AssignerWantsBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return "", nil
	}
	if err != nil {
		return "", err
	}
	key, err := sdkdelivery.WantKey(node, module)
	if err != nil {
		return "", err
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return "", nil
	}
	if err != nil {
		return "", err
	}
	w, err := sdkdelivery.DecodeWant(e.Value())
	if err != nil {
		return "", nil // a want of release 5, naming no composition: no fingerprint a composition must be kept for
	}
	return w.FP, nil
}

// haveOn is what a node's node-engine says it runs of a module.
func haveOn(ctx context.Context, js jetstream.JetStream, node, module string) (sdkdelivery.Have, bool, error) {
	kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return sdkdelivery.Have{}, false, nil
	}
	if err != nil {
		return sdkdelivery.Have{}, false, err
	}
	key, err := sdkdelivery.HaveKey(node, module)
	if err != nil {
		return sdkdelivery.Have{}, false, err
	}
	e, err := kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return sdkdelivery.Have{}, false, nil
	}
	if err != nil {
		return sdkdelivery.Have{}, false, err
	}
	h, err := sdkdelivery.DecodeHave(e.Value())
	if err != nil {
		return sdkdelivery.Have{}, false, nil
	}
	return h, true, nil
}

// wantedOnTheBus is every fingerprint a want names, as `<module> <fingerprint>`: what the keep set keeps the builds of
// (ADR 0539 clause 1). Read from the assigner's bucket on each look; no bucket yet names none.
func wantedOnTheBus(ctx context.Context) (map[string]bool, error) {
	bus, err := aBus()
	if err != nil {
		return nil, err
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return nil, err
	}
	kv, err := js.KeyValue(ctx, sdkdelivery.AssignerWantsBucket)
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return map[string]bool{}, nil
	}
	if err != nil {
		return nil, err
	}
	lister, err := kv.ListKeys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return map[string]bool{}, nil
	}
	if err != nil {
		return nil, err
	}
	out := map[string]bool{}
	for key := range lister.Keys() {
		_, module, err := sdkdelivery.ParseWantKey(key)
		if err != nil {
			continue
		}
		e, err := kv.Get(ctx, key)
		if err != nil {
			continue
		}
		w, err := sdkdelivery.DecodeWant(e.Value())
		if err != nil || w.FP == "" {
			continue
		}
		out[module+" "+w.FP] = true
	}
	return out, nil
}

// plansKept is a composition's provisioner's plans as the store keeps them: by module, each plan and what requires a
// module only pulled in by another (design 57 P6, issue 765); nil when it hands none.
func plansKept(d sendable) []byte {
	if len(d.plans) == 0 {
		return nil
	}
	kept := map[string]composedPlan{}
	for module, p := range d.plans {
		kept[module] = composedPlan{Plan: p, RequiredBy: d.requiredBy[module]}
	}
	raw, _ := json.Marshal(kept) // plain strings and slices: it marshals
	return raw
}

// composedPlan is one handed module's plan as the store keeps it, the shape inventory.SentProvisioningPlans reads.
type composedPlan struct {
	catalogue.Plan
	RequiredBy string `json:"required-by,omitempty"`
}

// lastAppliedOn reads, once, the declaration the node was last sent from the kept stream of declarations (the last
// message on `mesh.node.<node>.declare`), and answers each module's slice of it (its resources that say the module, or
// whose id names it) with the declaration's digest. None when the stream or the node's message is gone.
func lastAppliedOn(js jetstream.JetStream, node string) func(ctx context.Context, module string) ([]json.RawMessage, string, bool, error) {
	var once sync.Once
	var last []byte
	var found bool
	var readErr error
	return func(ctx context.Context, module string) ([]json.RawMessage, string, bool, error) {
		once.Do(func() { last, found, readErr = lastDeclaration(ctx, js, node) })
		if readErr != nil || !found {
			return nil, "", false, readErr
		}
		return sliceOfDeclaration(last, module)
	}
}

// lastDeclaration is the node's last signed declaration on the kept stream, as the stream holds it.
func lastDeclaration(ctx context.Context, js jetstream.JetStream, node string) ([]byte, bool, error) {
	stream, err := js.Stream(ctx, broker.DeclarationsStream)
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	msg, err := stream.GetLastMsgForSubject(ctx, "mesh.node."+node+".declare")
	if errors.Is(err, jetstream.ErrMsgNotFound) {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, err
	}
	return msg.Data, true, nil
}

// sliceOfDeclaration is module's slice of a signed declaration as the stream holds it, and the digest of the
// declaration it was signed over.
func sliceOfDeclaration(data []byte, module string) ([]json.RawMessage, string, bool, error) {
	var signed link.Signed
	if err := json.Unmarshal(data, &signed); err != nil {
		return nil, "", false, fmt.Errorf("the last declaration is not readable: %w", err)
	}
	var envelope struct {
		Resources []json.RawMessage `json:"resources"`
	}
	if err := json.Unmarshal(signed.Declaration, &envelope); err != nil {
		return nil, "", false, fmt.Errorf("the last declaration is not readable: %w", err)
	}
	sum := sha256.Sum256(signed.Declaration)
	return sliceOfModule(envelope.Resources, module), "sha256:" + hex.EncodeToString(sum[:]), true, nil
}

// sliceOfModule is, of a declaration's resources in their order, those of module: saying it, or with no module said
// and an id that names it (sdkdecl.ModuleGuess).
func sliceOfModule(resources []json.RawMessage, module string) []json.RawMessage {
	var out []json.RawMessage
	for _, r := range resources {
		var head struct {
			ID     string `json:"id"`
			Module string `json:"module"`
		}
		if json.Unmarshal(r, &head) != nil {
			continue
		}
		owner := head.Module
		if owner == "" {
			owner, _ = sdkdecl.ModuleGuess(head.ID)
		}
		if owner == module {
			out = append(out, r)
		}
	}
	return out
}

// compositionBody is the body a node's compositions are split from: every file's content inline, never by digest, so
// writeComposed puts each object of a file over 64 KiB in the node's bucket before it writes the composition that names
// it (ADR 0539 §4). A send's body names those files by digest for a node that reads references, and a send put their
// objects itself; with sends gone, a composition split from that body named objects nothing put (issue 1945, live on
// 2026-10-25: novox's nats.bus-users).
func compositionBody(declared sendable) ([]byte, error) {
	declared.fileReferences = false
	return declared.Body()
}

// composingLocks hold one composition of a node at a time in this process (issue 1945).
var composingLocks sync.Map

func lockComposing(node string) func() {
	l, _ := composingLocks.LoadOrStore(node, &sync.Mutex{})
	mu := l.(*sync.Mutex)
	mu.Lock()
	return mu.Unlock
}