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 }