package link import ( "context" "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/delivery" "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-` over the link open now: bound, never created (the // grants part creates it, create-only, history delivery.ComposedHistory). func composedStore(js nats.JetStreamContext, node string) (nats.KeyValue, error) { bucket, err := delivery.ComposedBucketFor(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) }