mesh-host/internal/link/watch.go

package link

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/nats-io/nats.go"
)

// KVUpdate is one change of a key in a bucket the node-engine watches (novox/hq ADR 0539 §9). Caught marks the end of
// the keys that were there when the watch began.
type KVUpdate struct {
	Key      string
	Value    []byte
	Revision uint64
	Deleted  bool
	Caught   bool
}

// KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and
// never polls.
type KVBus interface {
	// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends
	// or the link does (the channel is then closed). A bucket the node's account may not read is an error. The pattern
	// may be several, separated by spaces (WatchPatterns), watched as one: a node's placement in either form of its
	// keys (research 072 row 37).
	KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
}

// KVWatch is the watch on the link open now.
func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
	js := b.JS
	if b.Conn != nil {
		bound, err := b.Conn.JetStream()
		if err != nil {
			return nil, err
		}
		js = bound
	}
	kv, err := js.KeyValue(bucket)
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
	}
	w, err := kv.WatchFiltered(strings.Fields(pattern), nats.Context(ctx))
	if err != nil {
		return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err)
	}
	out := make(chan KVUpdate, 64)
	go func() {
		defer close(out)
		defer func() { _ = w.Stop() }()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				var u KVUpdate
				if e == nil {
					u.Caught = true
				} else {
					u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
						Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
				}
				select {
				case out <- u:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out, nil
}

func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
	return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern)
}

var (
	_ KVBus = OverNATS{}
	_ KVBus = (*natsLink)(nil)
)

// KVHistoryBus is a link that can read the revisions a bucket keeps of one key: the node agent takes the composition its
// want names from them when a newer one replaced it (novox/hq ADR 0539 clause 1).
type KVHistoryBus interface {
	// KVHistory is the key's revisions, oldest first; none when the key has none.
	KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error)
}

// historyWithin bounds one read of a key's revisions.
const historyWithin = 10 * time.Second

// KVHistory reads the key's revisions on the link open now.
func (b OverNATS) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
	js := b.JS
	if b.Conn != nil {
		bound, err := b.Conn.JetStream()
		if err != nil {
			return nil, err
		}
		js = bound
	}
	kv, err := js.KeyValue(bucket)
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
	}
	// Bounded on its own: a read that does not answer must not hold the node agent (the review of issue 1945).
	ctx, cancel := context.WithTimeout(ctx, historyWithin)
	defer cancel()
	entries, err := kv.History(key, nats.Context(ctx))
	if errors.Is(err, nats.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, err)
	}
	out := make([]KVUpdate, 0, len(entries))
	for _, e := range entries {
		out = append(out, KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
			Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge})
	}
	return out, nil
}

func (l *natsLink) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
	return OverNATS{Conn: l.conn, JS: l.js}.KVHistory(ctx, bucket, key)
}

var (
	_ KVHistoryBus = OverNATS{}
	_ KVHistoryBus = (*natsLink)(nil)
)

// KVHistory reads a key's revisions on the link open now.
func (q *Session) KVHistory(ctx context.Context, bucket, key string) ([]KVUpdate, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	h, ok := bus.(KVHistoryBus)
	if bus == nil || !ok {
		return nil, ErrNoLink
	}
	return h.KVHistory(ctx, bucket, key)
}

// ErrNoLink is what a watch asks for with no link open.
var ErrNoLink = errors.New("no link to the bus is open")

// KVWatch watches a bucket on the link open now. It ends when ctx does or the link closes.
func (q *Session) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	w, ok := bus.(KVBus)
	if bus == nil || !ok {
		return nil, ErrNoLink
	}
	return w.KVWatch(ctx, bucket, pattern)
}

// WhenLinked waits for a link to be open and returns a context that ends when that link does (or ctx does): what the
// node agent runs under, so its watches end with the link and begin again with the next.
func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.CancelFunc, error) {
	q.init()
	for {
		q.mu.Lock()
		if q.bus != nil {
			gone := q.gone
			q.mu.Unlock()
			linked, cancel := context.WithCancel(ctx)
			go func() {
				select {
				case <-gone:
					cancel()
				case <-linked.Done():
				}
			}()
			return linked, cancel, nil
		}
		changed := q.changed
		q.mu.Unlock()
		select {
		case <-ctx.Done():
			return nil, nil, ctx.Err()
		case <-changed:
		}
	}
}

// WhenLinkedAs waits for a link open that authenticated as user: after a switch of login, the vault is asked over the new
// connection and no other (ADR 0583 §3).
func (q *Session) WhenLinkedAs(ctx context.Context, user string) error {
	q.init()
	for {
		q.mu.Lock()
		if q.bus != nil && q.linkedUser == user {
			q.mu.Unlock()
			return nil
		}
		changed := q.changed
		q.mu.Unlock()
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-changed:
		}
	}
}

// EventBus is a link that can hear an event on a subject of core NATS: the node agent waits for a backup's end with it,
// and never polls.
type EventBus interface {
	// Hear yields the payloads published on the subject until stop is called or the link ends.
	Hear(subject string) (events <-chan []byte, stop func(), err error)
}

func (b OverNATS) Hear(subject string) (<-chan []byte, func(), error) {
	out := make(chan []byte, 16)
	sub, err := b.Conn.Subscribe(subject, func(m *nats.Msg) {
		select {
		case out <- m.Data:
		default:
		}
	})
	if err != nil {
		return nil, nil, err
	}
	if err := b.Conn.Flush(); err != nil {
		_ = sub.Unsubscribe()
		return nil, nil, err
	}
	return out, func() { _ = sub.Unsubscribe() }, nil
}

func (l *natsLink) Hear(subject string) (<-chan []byte, func(), error) {
	return OverNATS{Conn: l.conn, JS: l.js}.Hear(subject)
}

var (
	_ EventBus = OverNATS{}
	_ EventBus = (*natsLink)(nil)
)

// Hear listens for an event on the link open now, from before the caller asks for what it waits on.
func (q *Session) Hear(subject string) (<-chan []byte, func(), error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	h, ok := bus.(EventBus)
	if bus == nil || !ok {
		return nil, nil, ErrNoLink
	}
	return h.Hear(subject)
}

// CreatedAdvisory is the subject the bus says a bucket's stream was created on (the server's own advisory; the same
// subject mesh-sdk's process.AwaitBucket binds on, and the grants part gives a principal for each bucket it may read).
func CreatedAdvisory(bucket string) string {
	return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket
}

// KVWatchAwait is KVWatch for a bucket that may not be on the bus yet (the placement seat's states, made when their
// holder starts): the channel is returned at once, and while the bucket is missing it yields nothing and waiting is told
// why; when the bus says the bucket's stream was created the watch is made and its updates follow. No timer: the
// advisory is heard before the first try, so a creation between the try and the wait is not missed. Any other error of
// the first try is the caller's, as for KVWatch: a bucket the account may not read is not waited for.
func (q *Session) KVWatchAwait(ctx context.Context, bucket, pattern string, waiting func(error)) (<-chan KVUpdate, error) {
	created, stop, err := q.Hear(CreatedAdvisory(bucket))
	if err != nil {
		return nil, err
	}
	in, err := q.KVWatch(ctx, bucket, pattern)
	if err == nil {
		stop()
		return in, nil
	}
	if !errors.Is(err, nats.ErrBucketNotFound) {
		stop()
		return nil, err
	}
	waiting(err)
	out := make(chan KVUpdate, 64)
	go func() {
		defer close(out)
		defer stop()
		for in == nil {
			select {
			case <-ctx.Done():
				return
			case _, ok := <-created:
				if !ok {
					return
				}
				w, err := q.KVWatch(ctx, bucket, pattern)
				if err != nil {
					waiting(err)
					continue
				}
				in = w
			}
		}
		for u := range in {
			select {
			case out <- u:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out, nil
}

// WatchPatterns is several key patterns as one pattern KVWatch takes.
func WatchPatterns(patterns ...string) string { return strings.Join(patterns, " ") }