package link

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

	"github.com/novox/mesh-controller/internal/broker"
)

// The order the build queue hands asks out in (novox/hq issue 558).
//
// **A build a walk waits for goes before a pull request's check.** The queue was first come, first served,
// and a merged change's delivery build waited minutes behind checks while its walk, and every walk after
// it, waited on the build. So:
//
//   - a delivery is handed out before any check;
//   - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check
//     waits for ever behind a stream of deliveries;
//   - within a kind, first come, first served, by when it was asked;
//   - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check
//     taken late breaks nothing.
//
// **Kept in the stream, never beside it.** The queue is the seat's work queue, which the server hands out
// in stream order to whichever holder pulls; nothing else holds the order, so a restarted controller
// finds it as it was left. An ask that must go behind another is moved to the tail — deleted from where
// it was and published again, its body unchanged, under the same id — while the worker is paused, so no
// holder takes it in the moment it moves. Only waiting asks move: what a holder took is never touched.

// The kinds of ask, as the queue says them.
const (
	KindDelivery = "delivery"
	KindCheck    = "check"
)

// CheckWaitsAtMost is how long a check waits before it goes ahead of the deliveries asked after it.
var CheckWaitsAtMost = 10 * time.Minute

// KindOf is an ask's kind: a check for a pull request's check — a recheck the merge tool asks included —
// a dry run, or an ask nothing could be read from; a delivery for any other build.
func KindOf(r BuildRequest) string {
	if r.ID == "" || r.Check != nil || r.DryRun {
		return KindCheck
	}
	return KindDelivery
}

// askedOrStored is when an ask was asked, as its id says, or else when the queue took it.
func (a QueuedAsk) askedOrStored() time.Time {
	if !a.AskedAt.IsZero() {
		return a.AskedAt
	}
	return a.Stored
}

// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check
// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked.
func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
	var first, then []QueuedAsk
	for _, a := range waiting {
		if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost {
			then = append(then, a)
			continue
		}
		first = append(first, a)
	}
	byAsking := func(s []QueuedAsk) {
		sort.SliceStable(s, func(i, j int) bool {
			if ti, tj := s[i].askedOrStored(), s[j].askedOrStored(); !ti.Equal(tj) {
				return ti.Before(tj)
			}
			return s[i].Seq < s[j].Seq
		})
	}
	byAsking(first)
	byAsking(then)
	return append(first, then...)
}

// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything
// after the longest start of order that the queue already holds in that order.
func toMove(queue, order []QueuedAsk) []QueuedAsk {
	at := make(map[uint64]int, len(queue))
	for i, a := range queue {
		at[a.Seq] = i
	}
	kept, last := 0, -1
	for kept < len(order) {
		i, ok := at[order[kept].Seq]
		if !ok || i < last {
			break
		}
		last = i
		kept++
	}
	return order[kept:]
}

// orderPasses keeps one pass at a time. Only the acting controller's loop runs one (KeepingInOrder), so
// there is one; the lock holds a test's two at once to that too.
var orderPasses sync.Mutex

// pausedFor bounds a pause the pass could not end itself: a controller that stops mid-pass leaves the
// worker paused this long at most. moveMargin is the part of it no move starts in: a pass stops moving
// with at least this left, and the next pass moves the rest, so no move runs past the pause.
const (
	pausedFor    = 20 * time.Second
	moveMargin   = 5 * time.Second
	publishWaits = 3 * time.Second
)

// QueuedAtHeader carries, on a moved ask, when the stream first took it: a move keeps an ask's place in
// time, its id unread or not.
const QueuedAtHeader = "Mesh-Queued-At"

// MovedFromHeader names, on a move's copy, the sequence of the message it is a copy of: with QueuedAtHeader,
// what proves a second message of one id is a move's copy and not an ask delivered again (ADR 0293), whose
// headers are its original's — and so never name that original's own sequence.
const MovedFromHeader = "Mesh-Moved-From"

// Pass is what one pass did: the asks it moved, and each second copy it removed, in words.
type Pass struct {
	Moved   []string
	Removed []string
}

// asked is the signal an ask sends the acting controller's pass loop, in this process. An ask asked in
// another process — a verb run as its own command — is seen by the loop as the queue's last sequence moving.
var asked = make(chan struct{}, 1)

// signalAsked tells the pass loop an ask joined the queue. Never blocks, never runs a pass.
func signalAsked() {
	select {
	case asked <- struct{}{}:
	default:
	}
}

// PutInOrder moves the seat's waiting asks into the order QueueOrder says, and says each it moved. A queue
// already in order is read and left alone; the worker is paused only while asks move. Run by the acting
// controller alone (KeepingInOrder): two passes in two processes would each resume what the other paused.
func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) (Pass, error) {
	var pass Pass
	orderPasses.Lock()
	defer orderPasses.Unlock()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return pass, err
	}
	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
	waiting, extra, delivered, err := readWaiting(ctx, api, worker, seat)
	if err != nil || (len(extra) == 0 && len(toMove(waiting, QueueOrder(waiting, time.Now()))) == 0) {
		return pass, err
	}
	// Paused, then read again: what the worker handed out up to the pause is not waiting any more.
	pauseEnds := time.Now().Add(pausedFor)
	if delivered >= 0 {
		if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, pauseEnds); err != nil {
			return pass, fmt.Errorf("the %s worker could not be paused to put its queue in order, so it is left "+
				"first come, first served: %w", seat, err)
		}
		defer func() {
			resuming, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			defer cancel()
			if _, err := api.ResumeConsumer(resuming, worker.Stream, worker.Name); err != nil {
				fmt.Printf("the %s worker could not be resumed after its queue was put in order; it resumes by "+
					"itself by %s: %v\n", seat, pauseEnds.Format(time.RFC3339), err)
			}
		}()
		if waiting, extra, _, err = readWaiting(ctx, api, worker, seat); err != nil {
			return pass, err
		}
	}
	stream, err := api.Stream(ctx, worker.Stream)
	if err != nil {
		return pass, err
	}
	// A second copy of one ask — a move that published and stopped before it deleted — goes first.
	for _, a := range extra {
		if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
			return pass, fmt.Errorf("a second copy of %s (seq %d) could not be removed: %w", a.ID, a.Seq, err)
		}
		pass.Removed = append(pass.Removed, fmt.Sprintf("%s at seq %d, a move's second copy of the message at seq %s",
			a.ID, a.Seq, a.movedFrom))
	}
	for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) {
		if delivered >= 0 && time.Until(pauseEnds) < moveMargin {
			// The rest waits for the next pass, under a pause of its own.
			break
		}
		ok, err := moveToTheTail(ctx, js, stream, worker.Name, seat, a)
		if err != nil {
			return pass, err
		}
		if ok {
			pass.Moved = append(pass.Moved, a.ID)
		}
	}
	return pass, nil
}

// readWaiting is every ask the worker has not yet handed out, in queue order, but each second copy of one
// ask; the second copies, to remove; and the last sequence the worker handed out, -1 when there is no
// worker, so nothing is handed out to pause.
//
// **A second copy is proven, never guessed from an id**: a waiting message whose MovedFromHeader names an
// earlier message of the same request id still in the queue, and whose QueuedAtHeader is that message's
// first-queued time — a move that published its copy and stopped before it deleted the original. The
// original stays, waiting or handed out (a holder's then, and the copy would build it twice). Any other
// second message of one id — an ask a dead letter delivered again beside its original (ADR 0293) — is a
// real ask and stays.
func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Consumer, seat string) (
	waiting, extra []QueuedAsk, delivered int64, err error) {
	stream, err := api.Stream(ctx, worker.Stream)
	if err != nil {
		return nil, nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
	}
	delivered = -1
	if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
		info, err := consumer.Info(ctx)
		if err != nil {
			return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
		}
		delivered = int64(info.Delivered.Stream)
	} else if !errors.Is(err, jetstream.ErrConsumerNotFound) {
		return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
	}
	// From the first: the stream holds only what is unsettled, so this is a handful of reads.
	// Every message by sequence, with its id and first-queued time, to prove a copy against.
	type held struct {
		id     string
		stored time.Time
	}
	bySeq := map[string]held{}
	var seq uint64 = 1
	for n := 0; n < 10000; n++ {
		msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(BuildWorkOf(seat)))
		if errors.Is(err, jetstream.ErrMsgNotFound) {
			break
		}
		if err != nil {
			return nil, nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
		}
		seq = msg.Sequence + 1
		a := askOf(msg)
		a.movedFrom = msg.Header.Get(MovedFromHeader)
		from, known := bySeq[a.movedFrom]
		again := a.ID != "" && a.movedFrom != "" && known && from.id == a.ID && from.stored.Equal(a.Stored)
		bySeq[strconv.FormatUint(a.Seq, 10)] = held{id: a.ID, stored: a.Stored}
		switch {
		case int64(a.Seq) <= delivered:
		case again:
			extra = append(extra, a)
		default:
			waiting = append(waiting, a)
		}
	}
	return waiting, extra, delivered, nil
}

// moveToTheTail publishes the ask's body again at the queue's tail and then deletes it where it was. **The
// copy first**: a publish that fails deletes nothing, and a pass that stops between the two leaves two
// copies, which the next pass sees by their one id and makes one (readWaiting). A move never loses an ask.
// Looked at once more just before: one the worker has handed out since is a holder's now and stays. False
// when it was not moved.
func moveToTheTail(ctx context.Context, js *broker.JetStream, stream jetstream.Stream, workerName, seat string,
	a QueuedAsk) (bool, error) {
	if consumer, err := stream.Consumer(ctx, workerName); err == nil {
		info, err := consumer.Info(ctx)
		if err != nil {
			return false, err
		}
		if info.Delivered.Stream >= a.Seq {
			return false, nil
		}
	}
	msg, err := stream.GetMsg(ctx, a.Seq)
	if errors.Is(err, jetstream.ErrMsgNotFound) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	again := &nats.Msg{Subject: msg.Subject, Data: msg.Data, Header: nats.Header{}}
	for k, v := range msg.Header {
		// The server's own headers stay behind: a message id would be taken for a duplicate and dropped.
		if !strings.HasPrefix(k, "Nats-") {
			again.Header[k] = v
		}
	}
	// Its place in time goes with it: when the stream first took it, as askOf read it; and the message it
	// is a copy of, which proves it one.
	again.Header.Set(QueuedAtHeader, askOf(msg).Stored.UTC().Format(time.RFC3339Nano))
	again.Header.Set(MovedFromHeader, strconv.FormatUint(a.Seq, 10))
	// Inside the margin, so the publish ends while the worker is still paused.
	publishing, cancel := context.WithTimeout(ctx, publishWaits)
	ack, err := js.Context().PublishMsg(again, nats.Context(publishing))
	cancel()
	if err != nil {
		return false, fmt.Errorf("%s (seq %d) could not be put behind, and stays where it is: %w", a.ID, a.Seq, err)
	}
	// **Handed out meanwhile** — a bus slow enough that the pause ended: the original is a holder's, and the
	// copy goes instead, so it is built once.
	if consumer, err := stream.Consumer(ctx, workerName); err == nil {
		info, err := consumer.Info(ctx)
		if err != nil || info.Delivered.Stream >= a.Seq {
			if derr := stream.DeleteMsg(ctx, ack.Sequence); derr != nil && !errors.Is(derr, jetstream.ErrMsgNotFound) {
				return false, fmt.Errorf("%s was handed out while it moved and its copy (seq %d) could not be "+
					"removed; the next pass removes it: %w", a.ID, ack.Sequence, derr)
			}
			if err != nil {
				return false, fmt.Errorf("whether %s was handed out while it moved cannot be read, so its copy was "+
					"removed and it stays where it was: %w", a.ID, err)
			}
			if info.Delivered.Stream >= ack.Sequence {
				return false, fmt.Errorf("%s was handed out while it moved, and its copy too: it may be built twice", a.ID)
			}
			return false, nil
		}
	}
	if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
		return false, fmt.Errorf("%s was copied behind and its original (seq %d) not removed; the next pass "+
			"removes the second copy: %w", a.ID, a.Seq, err)
	}
	return true, nil
}

// KeepingInOrder runs the passes, on the acting controller alone, until ctx ends: when an ask in this
// process says so, when the queue's last sequence moved — an ask from another process — and every
// passEvery, for a check that comes of age with nothing asked. seat says the build seat to order; acting
// whether this controller acts (Holding); said what a pass did.
func KeepingInOrder(ctx context.Context, js *broker.JetStream, seat func() string, acting func() bool,
	said func(format string, args ...any)) {
	tick := time.NewTicker(orderLooks)
	defer tick.Stop()
	var lastSeq uint64
	var lastPass time.Time
	for {
		signalled := false
		select {
		case <-ctx.Done():
			return
		case <-asked:
			signalled = true
		case <-tick.C:
		}
		if !acting() {
			continue
		}
		name := seat()
		worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: name, Accepts: []string{"build"}})
		info, err := js.Context().StreamInfo(worker.Stream)
		if err != nil {
			continue
		}
		if !signalled && info.State.LastSeq == lastSeq && time.Since(lastPass) < passEvery {
			continue
		}
		lastSeq, lastPass = info.State.LastSeq, time.Now()
		passing, cancel := context.WithTimeout(ctx, pausedFor+10*time.Second)
		pass, err := PutInOrder(passing, js, name)
		cancel()
		for _, r := range pass.Removed {
			said("the %s queue: removed %s", name, r)
		}
		if len(pass.Moved) > 0 {
			said("the %s queue put in order: %s moved behind what goes first", name, strings.Join(pass.Moved, ", "))
		}
		if err != nil {
			said("the %s queue could not be put in order: %v", name, err)
		}
		if info, err := js.Context().StreamInfo(worker.Stream); err == nil {
			// Its own moves are not an ask.
			lastSeq = info.State.LastSeq
		}
	}
}

// orderLooks is how often the loop looks whether an ask joined from another process; passEvery how often
// it passes with nothing asked.
var (
	orderLooks = 2 * time.Second
	passEvery  = 30 * time.Second
)

// DeleteAsk removes a waiting or dead ask from the queue: at its sequence, or — moved into order since it
// was read — the copy carrying its id. Says the sequence it removed, 0 when neither was there any more.
func DeleteAsk(ctx context.Context, js *broker.JetStream, seat string, ask QueuedAsk) (uint64, error) {
	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
	gone := func(err error) bool {
		return errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgNotFound) ||
			(err != nil && strings.Contains(err.Error(), "no message found"))
	}
	err := js.Context().DeleteMsg(worker.Stream, ask.Seq)
	if err == nil {
		return ask.Seq, nil
	}
	if !gone(err) {
		return 0, err
	}
	if ask.ID == "" {
		return 0, nil
	}
	q, err := ReadQueue(ctx, js, seat)
	if err != nil {
		return 0, err
	}
	for _, a := range q.Asks {
		if a.ID == ask.ID && a.Seq != ask.Seq && a.State != AskInFlight {
			if err := js.Context().DeleteMsg(worker.Stream, a.Seq); err != nil && !gone(err) {
				return 0, err
			} else if err == nil {
				return a.Seq, nil
			}
		}
	}
	return 0, nil
}