package link import ( "context" "errors" "fmt" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go" ) // ---- the data items of the modules placed on this node, in the bus's own storage (novox/hq ADR 0577 clause 1) ------ // DataBucket is the node-engine's declared state data (delivery.EngineDataBucket), created by the grants part from the // engine's manifest as its states acts and health are: one key `.` per module placed on the node that // declares data items, written by that node's engine alone, read by that node's disk-load. const DataBucket = delivery.EngineDataBucket // dataStore binds DataBucket over the link open now: bound, never created. func (l *natsLink) dataStore() (nats.KeyValue, error) { kv, err := l.js.KeyValue(DataBucket) 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): %w", DataBucket, err) } return kv, nil } // EngineData is this node's keys of DataBucket over whichever link the session has open (selfcompose.DataStore); with // none open every read and write is refused, and nothing is kept anywhere else. type EngineData struct{ Session *Session } func (d EngineData) kv() (nats.KeyValue, error) { d.Session.init() d.Session.mu.Lock() bus := d.Session.bus d.Session.mu.Unlock() l, ok := bus.(interface{ dataStore() (nats.KeyValue, error) }) if bus == nil || !ok { return nil, ErrNoLink } return l.dataStore() } // Get is a key's value, nil when it has none. func (d EngineData) Get(_ context.Context, key string) ([]byte, error) { kv, err := d.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 a key. func (d EngineData) Put(_ context.Context, key string, value []byte) error { kv, err := d.kv() if err != nil { return err } _, err = kv.Put(key, value) return err } // Delete removes a key; one that was not there is no error. func (d EngineData) Delete(_ context.Context, key string) error { kv, err := d.kv() if err != nil { return err } if err := kv.Delete(key); err != nil && !errors.Is(err, nats.ErrKeyNotFound) { return err } return nil } // Keys is every key matching pattern that holds a value: read by a watch of the pattern alone, the only read the // engine's grant gives of its own node's keys. func (d EngineData) Keys(_ context.Context, pattern string) ([]string, error) { kv, err := d.kv() if err != nil { return nil, err } w, err := kv.WatchFiltered([]string{pattern}, nats.IgnoreDeletes(), nats.MetaOnly()) if err != nil { return nil, err } defer func() { _ = w.Stop() }() var keys []string for e := range w.Updates() { if e == nil { break } keys = append(keys, e.Key()) } return keys, nil }