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 the changes, until ctx ends
	// or the link does (the channel is then closed). It never holds the bus client back: while the reader does not take,
	// a key's changes fold into its newest (a delete included), so a reader sees each key's latest state, not every
	// change, and a key still queued from before Caught is handed before Caught with its newest value; across keys the
	// order is the order each key was first queued. 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 assignment 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() }()
		forwardCoalescing(ctx, w.Updates(), entryUpdate, out)
	}()
	return out, nil
}

// entryUpdate is the update a client's entry is; the nil entry the client sends after the last key there was is Caught.
func entryUpdate(e nats.KeyValueEntry) KVUpdate {
	if e == nil {
		return KVUpdate{Caught: true}
	}
	return KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
		Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
}

// forwardCoalescing hands in's items, as updates, to out, and never lets out hold in back: the client's watch sends into
// its channel from inside the subscription's callback, so a reader that is not draining (a node-engine composing for
// minutes) overflows the subscription's pending limit and the bus drops messages (issue 1966). While out is not taken,
// the pending updates are kept by key (the newest value and revision of each, in the order the keys were first seen) and
// the Caught marker keeps its place among the keys queued before it. It returns when ctx ends, or when in is closed and
// what is pending is handed over.
func forwardCoalescing[T any](ctx context.Context, in <-chan T, conv func(T) KVUpdate, out chan<- KVUpdate) {
	var queue []*KVUpdate
	byKey := map[string]*KVUpdate{}
	add := func(u KVUpdate) {
		if !u.Caught {
			if q, ok := byKey[u.Key]; ok {
				*q = u
				return
			}
		}
		q := &u
		queue = append(queue, q)
		if !u.Caught {
			byKey[u.Key] = q
		}
	}
	source := in
	for source != nil || len(queue) > 0 {
		var send chan<- KVUpdate
		var head KVUpdate
		if len(queue) > 0 {
			send, head = out, *queue[0]
		}
		select {
		case <-ctx.Done():
			return
		case x, ok := <-source:
			if !ok {
				source = nil
				continue
			}
			add(conv(x))
		case send <- head:
			if !head.Caught {
				delete(byKey, head.Key)
			}
			queue = queue[1:]
		}
	}
}

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 assignment 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, " ") }