package link

import (
	"context"
	"errors"
	"fmt"

	"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.
	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.Watch(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)
)

// 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:
		}
	}
}

// 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)
}