feat(natsjsregistry): fix watcher feature

Signed-off-by: jkoberg <jkoberg@owncloud.com>

vendor/github.com/go-micro/plugins/v4/store/nats-js-kv/nats.go

@@ -335,22 +335,63 @@ func (n *natsStore) String() string {
 	return "NATS JetStream KeyValueStore"
 }
 
+// StoreUpdate is the update type for the store.
+type StoreUpdate struct {
+	Value  KeyValueEnvelope
+	Action string
+}
+
 // WatchAll exposes the watcher interface from the underlying JetStreamContext.
-func (n *natsStore) WatchAll(bucket string, opts ...nats.WatchOpt) (nats.KeyWatcher, error) {
+func (n *natsStore) WatchAll(bucket string, opts ...nats.WatchOpt) (<-chan *StoreUpdate, func() error, error) {
 	if bucket == "" {
-		return nil, errors.New("multi bucket watching is not supported")
+		return nil, nil, errors.New("multi bucket watching is not supported")
 	}
 
 	if err := n.initConn(); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	b, err := n.js.KeyValue(bucket)
 	if err != nil {
-		return nil, errors.Wrap(err, "Failed to get bucket")
+		return nil, nil, errors.Wrap(err, "Failed to get bucket")
 	}
 
-	return b.WatchAll(opts...)
+	w, err := b.WatchAll(opts...)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "Failed to watch bucket")
+	}
+
+	ch := make(chan *StoreUpdate)
+	go func() {
+		for u := range w.Updates() {
+			if u == nil {
+				continue
+			}
+
+			var action string
+			switch u.Operation() {
+			default:
+				action = u.Operation().String()
+			case nats.KeyValuePut:
+				action = "create"
+			case nats.KeyValueDelete:
+				action = "delete"
+			case nats.KeyValuePurge:
+				action = "delete"
+			}
+
+			var kv KeyValueEnvelope
+			if err := json.Unmarshal(u.Value(), &kv); err != nil {
+				continue
+			}
+			ch <- &StoreUpdate{
+				Value:  kv,
+				Action: action,
+			}
+		}
+	}()
+
+	return ch, w.Stop, nil
 }
 
 // thread safe way to initialize the connection.