mesh-host/internal/link/enginepackages.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go"
)

// ---- the packages this node's compositions declare, in the bus's own storage (novox/hq ADR 0565 §6) ---------------

// PackagesBucket is the node-engine's declared state packages, created by the grants part from the engine's manifest as
// its states health and data are: one key, the node's name, written by that node's engine alone, read by pacman as
// mesh-host.packages (delivery.EnginePackagesRead).
const PackagesBucket = "mesh-host_" + delivery.EnginePackagesState

// MirrorPackagesBucket is the controller's declared-packages mirror (ADR 0507 §5), which the engine reads its own node's
// key of until the lease write, to compare (the node seat's verb packages).
const MirrorPackagesBucket = "mesh-controller_declared-packages"

// packagesStore binds PackagesBucket over the link open now: bound, never created.
func (l *natsLink) packagesStore() (nats.KeyValue, error) {
	kv, err := l.js.KeyValue(PackagesBucket)
	if err != nil {
		return nil, fmt.Errorf("the node-engine's state %s cannot be bound (it is created from the node-engine's "+
			"manifest; written again at the next composition that changes, or by the node seat's packages with write): %w",
			PackagesBucket, err)
	}
	return kv, nil
}

// packagesStore binds PackagesBucket over a link made of a connection the caller holds (OverNATS), as natsLink does.
func (b OverNATS) packagesStore() (nats.KeyValue, error) {
	kv, err := b.JS.KeyValue(PackagesBucket)
	if err != nil {
		return nil, fmt.Errorf("the node-engine's state %s cannot be bound: %w", PackagesBucket, err)
	}
	return kv, nil
}

// EnginePackages is this node's key of PackagesBucket over whichever link the session has open
// (selfcompose.PackagesStore); with none open every read and write is refused, and nothing is kept anywhere else.
type EnginePackages struct{ Session *Session }

func (p EnginePackages) kv() (nats.KeyValue, error) {
	p.Session.init()
	p.Session.mu.Lock()
	bus := p.Session.bus
	p.Session.mu.Unlock()
	l, ok := bus.(interface{ packagesStore() (nats.KeyValue, error) })
	if bus == nil || !ok {
		return nil, ErrNoLink
	}
	return l.packagesStore()
}

// Get is the key's value, nil when it has none.
func (p EnginePackages) Get(_ context.Context, key string) ([]byte, error) {
	kv, err := p.kv()
	if err != nil {
		return nil, err
	}
	e, err := kv.Get(key)
	if errors.Is(err, nats.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return e.Value(), nil
}

// Put writes the key.
func (p EnginePackages) Put(_ context.Context, key string, value []byte) error {
	kv, err := p.kv()
	if err != nil {
		return err
	}
	_, err = kv.Put(key, value)
	return err
}

// PackagesView is the node seat's verb packages' answer (ADR 0565 §6): this node's key of the engine's state packages,
// and until the lease write the controller's mirror's key for the node and every package they hold differently.
type PackagesView struct {
	Node string `json:"node"`
	// Key is the node's key as the bus holds it (Source); null when the engine has written none.
	Source   string                     `json:"source"`
	Key      *delivery.DeclaredPackages `json:"key"`
	KeyError string                     `json:"key_error,omitempty"`
	// Mirror is the controller's mirror's key for the node (MirrorSource); null when it holds none or cannot be read.
	MirrorSource string                     `json:"mirror_source"`
	Mirror       *delivery.DeclaredPackages `json:"mirror"`
	MirrorError  string                     `json:"mirror_error,omitempty"`
	// Differences is every package the two hold differently, read only when both are; Agree when there is none.
	Differences []delivery.PackageDifference `json:"differences"`
	Agree       bool                         `json:"agree"`
	// Wrote says, for write, whether the key was written again (false: it already held what the engine composed).
	Wrote *bool  `json:"wrote,omitempty"`
	Note  string `json:"note,omitempty"`
}

// PackagesAnswer is the verb's answer from the two keys as read (nil: no key), each with its read's error.
func PackagesAnswer(node string, key, mirror []byte, keyErr, mirrorErr error) PackagesView {
	v := PackagesView{Node: node, Source: delivery.EnginePackagesRead, MirrorSource: "mesh-controller.declared-packages",
		Differences: []delivery.PackageDifference{}}
	read := func(raw []byte, err error, into **delivery.DeclaredPackages, said *string) {
		if err == nil && raw != nil {
			var d delivery.DeclaredPackages
			if err = json.Unmarshal(raw, &d); err == nil {
				*into = &d
			}
		}
		if err != nil {
			*said = err.Error()
		}
	}
	read(key, keyErr, &v.Key, &v.KeyError)
	read(mirror, mirrorErr, &v.Mirror, &v.MirrorError)
	switch {
	case v.Key != nil && v.Mirror != nil:
		v.Differences = delivery.DiffDeclaredPackages(*v.Key, *v.Mirror)
		v.Agree = len(v.Differences) == 0
	case v.Key == nil && v.KeyError == "":
		v.Note = "the engine has written no key for this node yet: pacman reads the controller's mirror for it"
	case v.Mirror == nil && v.MirrorError == "":
		v.Note = "the controller's mirror holds no key for this node: nothing to compare"
	}
	return v
}