package main

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/conditions"
)

// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's
// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look
// of a controller whose consumer is new, and nothing at a restart but what the controller before it left
// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once
// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is
// refused with a delay that grows with each refusal, and the bus delivers it again.
//
// **The bound on the store is the one take-in loop**, a batch of at most startTakeIns keys at a time (issues 1945 and
// 1277: a burst of take-ins once filled the store until it refused every connection). The consumers' own limit on
// unacknowledged keys is far above it, because a refused key stays unacknowledged until its delay ends: with a limit of
// a batch, a handful of keys refused for long (a manifest word this controller does not read yet) would stop a consumer
// for good, the build that reads the word among what it holds back (the review of ADR 0544).

// startTakeIns is the most a batch takes: the take-ins that may wait on the store at once.
const startTakeIns = 16

// takeInPending is how many keys each consumer may leave unacknowledged: delivered and being taken in, or refused and
// waiting for their delay. Far above a batch, so refused keys never stop the others.
const takeInPending = 1000

// takeInBuffered is how many delivered keys each consumer's reader holds ahead of the loop.
const takeInBuffered = startTakeIns / 2

// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch
// takes, its two composer passes among it.
const takeInAckWait = 10 * time.Minute

// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch.
var takeInGather = 200 * time.Millisecond

// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6).
var takeInStall = 5 * time.Minute

// takeInMsg is one delivered key, as the take-in uses it.
type takeInMsg struct {
	delivered
	ack func()
	nak func(delay time.Duration)
}

// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in.
func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) {
	if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" {
		return delivered{}, false
	}
	key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".")
	module, _, _ := strings.Cut(key, ".")
	if module == "" {
		return delivered{}, false
	}
	var deliveries uint64 = 1
	if meta, err := m.Metadata(); err == nil {
		deliveries = meta.NumDelivered
	}
	return delivered{module: module, again: deliveries > 1, deliveries: deliveries}, true
}

// takeInConsumer is one bucket's consumer of the take-in.
type takeInConsumer struct {
	bucket   string
	consumer jetstream.Consumer
}

// takeInConfig is the take-in's consumer on bucket's stream.
func takeInConfig(bucket string) jetstream.ConsumerConfig {
	deliver := jetstream.DeliverNewPolicy
	if bucket == sdkdelivery.ModulesBucket {
		deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once
	}
	return jetstream.ConsumerConfig{
		Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver,
		AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: takeInPending,
		Description: "the controller's take-in of done keys (novox/hq ADR 0544)",
	}
}

// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and
// the other is still read.
func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer {
	var out []takeInConsumer
	for _, bucket := range broker.TakeInBuckets {
		c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, takeInConfig(bucket))
		o := takeInNotConsumedObservation(bucket, err)
		if err != nil {
			// Loud: a consumer whose deliver policy differs cannot be updated, and nothing of its bucket is taken in.
			doneKeysSay("done keys: " + o.Summary)
			if keeper := conditionsFrom; keeper != nil {
				_, _ = keeper.Observe(ctx, o)
			}
			continue
		}
		// Made: a condition a controller before this one raised for it is cleared (a restart that works clears it).
		if keeper := conditionsFrom; keeper != nil {
			_, _ = keeper.Clear(ctx, o.Key(), "the take-in's consumer on "+bucket+" is made")
		}
		out = append(out, takeInConsumer{bucket: bucket, consumer: c})
	}
	return out
}

// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their
// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start.
func watchDoneKeys(ctx context.Context, open *stores) {
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err)
		return
	}
	consumers := takeInConsumers(ctx, keys.JS)
	if len(consumers) == 0 {
		closeKeys()
		return
	}
	in := make(chan takeInMsg)
	// moving counts the keys handed to the loop and the keys it settled (acknowledged or refused, each as the bus took
	// it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6).
	moving := &takeInMoving{}
	read := make([]jetstream.Consumer, 0, len(consumers))
	// The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach
	// the bus (the third review of ADR 0544).
	var users sync.WaitGroup
	for _, c := range consumers {
		bucket := c.bucket
		read = append(read, c.consumer)
		it, err := c.consumer.Messages(jetstream.PullMaxMessages(takeInBuffered))
		if err != nil {
			fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err)
			continue
		}
		goServing(ctx, func() {
			<-ctx.Done()
			it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait
		})
		users.Add(1)
		goServing(ctx, func() {
			defer users.Done()
			for {
				m, err := it.Next()
				if err != nil {
					if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) {
						fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err)
					}
					return
				}
				d, ok := takeInKeyOf(bucket, m)
				if !ok {
					_ = m.Ack()
					continue
				}
				// Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start,
				// until the bus node is composed) is delivered again and settled again, and never counts as work held
				// (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said.
				msg := takeInMsg{delivered: d, ack: func() {
					acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
					defer cancel()
					if err := m.DoubleAck(acking); err != nil {
						doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err))
					}
					moving.settled.Add(1)
				},
					nak: func(delay time.Duration) {
						_ = m.NakWithDelay(delay)
						moving.settled.Add(1)
					}}
				if ctx.Err() != nil {
					_ = m.Nak()
					continue // drained: each key it held, refused at once
				}
				select {
				case in <- msg:
					moving.handed.Add(1)
				case <-ctx.Done():
					_ = m.Nak() // delivered again at once to the controller after this one, not after the ack wait
					continue
				}
			}
		})
	}
	backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) }
	steps := liveTakeInSteps(open, keys, keys.JS)
	users.Add(1)
	goServing(ctx, func() {
		defer users.Done()
		takeInLoop(ctx, in, steps, backlog)
	})
	goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) })
	goServing(ctx, func() {
		<-ctx.Done()
		users.Wait()
		closeKeys()
	})
}

// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends.
func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps,
	backlog func(context.Context) (takeInBacklog, error)) {
	t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}}
	for {
		var batch []takeInMsg
		select {
		case <-ctx.Done():
			return
		case m := <-in:
			batch = append(batch, m)
		}
		gather := time.NewTimer(takeInGather)
	more:
		for len(batch) < startTakeIns {
			select {
			case m := <-in:
				batch = append(batch, m)
			case <-gather.C:
				break more
			case <-ctx.Done():
				gather.Stop()
				for _, m := range batch {
					m.nak(0)
				}
				return
			}
		}
		gather.Stop()
		said := t.take(ctx, steps, batch)
		if b, err := backlog(ctx); err == nil {
			said += fmt.Sprintf("; backlog %d", b.Total())
		}
		doneKeysSay("done keys: " + said)
	}
}

// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at,
// the modules whose condition it raised, and the wait last said of each.
type takeInState struct {
	memo     map[string]string
	raised   map[string]bool
	saidWait map[string]string
}

// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers
// the batch in a line.
func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string {
	keys := make([]delivered, len(batch))
	for i, m := range batch {
		keys[i] = m.delivered
	}
	settled, passes := steps.batch(ctx, keys, t.memo)
	verdicts := make(map[string]verdict, len(settled))
	var taken, nothing, waiting, failed int
	for _, r := range settled {
		verdicts[r.module] = r.verdict
		o := doneKeyObservation(r.module)
		switch {
		case r.err != nil:
			failed++
			doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err))
			if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil &&
				!errors.Is(r.err, context.Canceled) {
				o.Summary += r.err.Error()
				if _, err := keeper.Observe(ctx, o); err == nil {
					t.raised[r.module] = true
				}
			}
		case r.waits != "":
			waiting++
			if t.saidWait[r.module] != r.waits {
				doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits))
				t.saidWait[r.module] = r.waits
			}
		case r.taken:
			taken++
			delete(t.saidWait, r.module)
			if r.said != "" {
				doneKeysSay("done keys: " + r.said)
			}
			// Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again.
			if keeper := conditionsFrom; keeper != nil {
				if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil {
					delete(t.raised, r.module)
				}
			}
		case r.nothingNew:
			nothing++
		}
	}
	for _, m := range batch {
		if verdicts[m.module] == refused {
			if ctx.Err() != nil {
				m.nak(0) // cut by the stop: the next controller takes it at once
			} else {
				m.nak(refusalDelay(m.deliveries))
			}
		} else {
			m.ack()
		}
	}
	return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)",
		len(batch), len(settled), taken, nothing, waiting, failed, passes)
}

// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, and keys delivered and not acknowledged
// (being taken in, or refused and waiting for their delay).
type takeInBacklog struct {
	Pending uint64 `json:"pending"`
	Taking  uint64 `json:"taking"`
}

// Total is every key the take-in has not acknowledged.
func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking }

// backlogOf reads the consumers' own counts from the bus.
func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) {
	var b takeInBacklog
	for _, c := range consumers {
		info, err := c.Info(ctx)
		if err != nil {
			return takeInBacklog{}, err
		}
		b.Pending += info.NumPending
		b.Taking += uint64(info.NumAckPending)
	}
	return b, nil
}

// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name.
func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) {
	keys, closeKeys, err := openBuilderKeys()
	if err != nil {
		return takeInBacklog{}, err
	}
	defer closeKeys()
	var consumers []jetstream.Consumer
	for _, bucket := range broker.TakeInBuckets {
		c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer)
		if err != nil {
			return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err)
		}
		consumers = append(consumers, c)
	}
	return backlogOf(ctx, consumers)
}

// kindTakeInStalled is the condition of a take-in that holds work and settles none (ADR 0544 §6); kindTakeInNotConsumed
// that of a bucket whose consumer could not be made.
const (
	kindTakeInStalled     = "take-in-stalled"
	kindTakeInNotConsumed = "take-in-not-consumed"
)

func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation {
	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled,
		Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys",
		Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has settled none since %s",
			b.Total(), since.UTC().Format(time.RFC3339))}
}

// takeInNotConsumedObservation is the condition of bucket's consumer not made, with its own id, so it is raised and
// cleared apart from a stall and from the other bucket's.
func takeInNotConsumedObservation(bucket string, err error) conditions.Observation {
	summary := "the take-in's consumer on " + bucket + " is made"
	if err != nil {
		summary = fmt.Sprintf("the take-in's consumer on %s is not made, so its keys are not taken in until the next "+
			"start: %v", bucket, err)
	}
	return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in-" + bucket, Token: kindTakeInNotConsumed,
		Kind: kindTakeInNotConsumed, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator,
		Source: "done keys", Summary: summary}
}

// takeInMoving counts the keys handed to the take-in loop and the keys it settled.
type takeInMoving struct {
	handed, settled atomic.Uint64
}

// stallWatch judges each look: stalled once there was work to do (keys not delivered yet, or keys handed to the loop
// and not settled) and nothing settled for takeInStall. A refused key waiting for its delay is no work of the loop's:
// with nothing else, a quiet mesh is not a stall; a loop that hangs on its batch is (the review of ADR 0544). Its first
// look always says it changed, so a condition a controller before this one raised is cleared once this one moves.
type stallWatch struct {
	mu      sync.Mutex
	settled uint64
	movedAt time.Time
	stalled bool
	known   bool
}

// look takes a reading at now: the backlog, and how many keys were handed and settled; it answers whether the take-in
// stalls and whether that changed (always on the first look).
func (w *stallWatch) look(b takeInBacklog, handed, settled uint64, now time.Time) (stalled, changed bool) {
	w.mu.Lock()
	defer w.mu.Unlock()
	work := b.Pending > 0 || handed > settled
	if !work || settled != w.settled || w.movedAt.IsZero() {
		w.movedAt = now
	}
	w.settled = settled
	nowStalled := work && now.Sub(w.movedAt) >= takeInStall
	changed = !w.known || nowStalled != w.stalled
	w.stalled, w.known = nowStalled, true
	return nowStalled, changed
}

// watchTakeInStall reads the backlog every minute and raises take-in-stalled while the take-in holds work and settles
// none, clearing it when it moves, and at its first look when it does not stall: a watchdog's timer, as research 072 §2
// keeps them until the health part leaves the controller.
func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error), moving *takeInMoving) {
	w := &stallWatch{}
	tick := time.NewTicker(time.Minute)
	defer tick.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
		b, err := backlog(ctx)
		if err != nil {
			continue
		}
		stalled, changed := w.look(b, moving.handed.Load(), moving.settled.Load(), time.Now())
		keeper := conditionsFrom
		if !changed || keeper == nil {
			continue
		}
		w.mu.Lock()
		since := w.movedAt
		w.mu.Unlock()
		o := takeInStalledObservation(b, since)
		if stalled {
			doneKeysSay("done keys: " + o.Summary)
			_, _ = keeper.Observe(ctx, o)
		} else {
			_, _ = keeper.Clear(ctx, o.Key(), "the take-in moves")
		}
	}
}