jschoubben
5906e18028
The engine's Composed keeps the catalogue's Plan and the declaration types and goes through nodeengineclient's codec; the fingerprint key part is meshcatalogclient's. The candidate answer is nodeengine.CandidateAnswer; the have still holds it through an interim JSON bridge (link.interimAnswers) until the Have group moves. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
104 lines
2.9 KiB
Go
104 lines
2.9 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
engcomp "github.com/novox/mesh-host/internal/compositions"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
|
|
"github.com/novox/mesh-host/internal/selfcompose"
|
|
)
|
|
|
|
// ---- this node's composition state, in the bus's own storage (novox/hq ADR 0593 §6) --------------------------------
|
|
|
|
// composedStore binds the node's composition state `composed-<node>` over the link open now: bound, never created (the
|
|
// grants part creates it, create-only, history engcomp.History).
|
|
func composedStore(js nats.JetStreamContext, node string) (nats.KeyValue, error) {
|
|
bucket, err := engcomp.BucketFor(node)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kv, err := js.KeyValue(bucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("this node's composition state %s cannot be bound (the grants part creates it; written "+
|
|
"again at the next pass that composes): %w", bucket, err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
func (l *natsLink) composedStore(node string) (nats.KeyValue, error) {
|
|
return composedStore(l.js, node)
|
|
}
|
|
|
|
func (b OverNATS) composedStore(node string) (nats.KeyValue, error) { return composedStore(b.JS, node) }
|
|
|
|
// EngineComposed is this node's composition state over whichever link the session has open (selfcompose.ComposedStore);
|
|
// with none open every read and write is refused, and nothing is kept anywhere else.
|
|
type EngineComposed struct {
|
|
Session *Session
|
|
Node string
|
|
}
|
|
|
|
var _ selfcompose.ComposedStore = EngineComposed{}
|
|
|
|
func (e EngineComposed) kv() (nats.KeyValue, error) {
|
|
e.Session.init()
|
|
e.Session.mu.Lock()
|
|
bus := e.Session.bus
|
|
e.Session.mu.Unlock()
|
|
l, ok := bus.(interface {
|
|
composedStore(node string) (nats.KeyValue, error)
|
|
})
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return l.composedStore(e.Node)
|
|
}
|
|
|
|
// Entries is every key the state holds with its latest value and revision.
|
|
func (e EngineComposed) Entries(ctx context.Context) (map[string]selfcompose.ComposedEntry, error) {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
keys, err := kv.Keys(nats.Context(ctx))
|
|
if errors.Is(err, nats.ErrNoKeysFound) {
|
|
return map[string]selfcompose.ComposedEntry{}, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make(map[string]selfcompose.ComposedEntry, len(keys))
|
|
for _, key := range keys {
|
|
got, err := kv.Get(key)
|
|
if errors.Is(err, nats.ErrKeyNotFound) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out[key] = selfcompose.ComposedEntry{Value: got.Value(), Revision: got.Revision()}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// Put writes a key and answers its revision.
|
|
func (e EngineComposed) Put(_ context.Context, key string, value []byte) (uint64, error) {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return kv.Put(key, value)
|
|
}
|
|
|
|
// Delete deletes a key (a delete marker, kept with the key's history).
|
|
func (e EngineComposed) Delete(_ context.Context, key string) error {
|
|
kv, err := e.kv()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return kv.Delete(key)
|
|
}
|