package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

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

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

// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687 and 1688).
//
// **The layout other parts read** (ADR 0484 clause 12 has the checker read the same key; a layout shared between modules
// belongs in mesh-sdk, and this one is not there yet):
//
//   - bucket `mesh-controller_plans-building` (broker.PlansBuildingBucket): one key per delivery plan in its build phase, named
//     by the plan's id (`plan-<digits>`), value a PlanBuilding as JSON. Written by the controller (lease holder) alone: put when
//     the plan's first build is asked, put again about every thirty seconds, deleted when the phase ends; the bucket's age
//     (two minutes) removes the key of a controller that stopped. A reader lists the keys and needs no value to decide.
//   - key `let-through` in bucket `mesh-controller_checks-running` (broker.CheckRunningBucket), value a LetThrough as JSON:
//     created by the holder that takes a check which waited LetThroughAfter, refreshed at each heartbeat of that check, deleted
//     when it ends. At most one such check runs at a time on the build seat, because the create fails while the key is there.

// LetThroughAfter is how long a check waits, while a delivery builds, before it is let through (ADR 0483 clause 3).
const LetThroughAfter = 30 * time.Minute

// LetThroughKey is the key of the one check let through, in the bucket of the checks running. It has no dot, so it is no
// pull request's key (`<owner>.<repo>.<number>`) and no group's (`group.<name>`).
const LetThroughKey = "let-through"

// PlanBuilding is the value of a plan's key in the bucket of the plans building.
type PlanBuilding struct {
	Plan  string    `json:"plan"`
	Since time.Time `json:"since"`
	Seen  time.Time `json:"seen"`
}

// LetThrough is the value of the let-through key: the builder that runs the check and when it took it.
type LetThrough struct {
	Builder string    `json:"builder"`
	Since   time.Time `json:"since"`
}

// checkGate is what a taking loop asks of the bus to hold checks while a delivery builds; a variable of the machine so a test
// hands out against fakes.
type checkGate interface {
	// Building is the ids of the plans in their build phase: the keys of the bucket. An error is a bucket that cannot be read.
	Building(ctx context.Context) ([]string, error)
	// OldestWaiting is when the first waiting ask of a check queue was asked, read without taking it.
	OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error)
	// TakeLetThrough creates the let-through key for a builder: false when it is there (another builder runs a check let through).
	TakeLetThrough(ctx context.Context, builder string) (bool, error)
	// KeepLetThrough puts the key again, and DropLetThrough deletes it.
	KeepLetThrough(ctx context.Context, builder string) error
	DropLetThrough(ctx context.Context) error
}

// busGate is the gate over a connection.
type busGate struct{ conn *nats.Conn }

func (g busGate) Building(ctx context.Context) ([]string, error) {
	return BuildingPlans(ctx, g.conn)
}

// BuildingPlans is the ids of the delivery plans in their build phase, from the controller's bucket. A reader that is not the
// build seat's (the checker, ADR 0484 clause 12) calls this. A bucket that cannot be read is an error; none is none building.
func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket)
	if err != nil {
		return nil, fmt.Errorf("the plans building: %w", err)
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the plans building: %w", err)
	}
	return keys, nil
}

// OldestWaiting is when the check a fetch of this queue would return was asked, read without taking anything.
//
// **It sees redeliveries** (review of issue 1688). A work queue keeps only what is not acknowledged: what the worker handed out
// (sequence up to its delivered mark) is running or waiting to be handed out again, and a fetch returns those before anything
// new. An ask there that is not said running (a key of the checks running names its build) is such a redelivery. If one is
// younger than the bound, the queue's next fetch would return it, and a young check is never let through: the queue is not a
// candidate, and nothing is fetched, so nothing is handed back and no delivery is spent. Otherwise the candidate is the oldest
// of the redeliveries and the first ask after the delivered mark.
func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
	info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
	if err != nil {
		return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
	}
	var oldest time.Time
	note := func(at time.Time) {
		if oldest.IsZero() || at.Before(oldest) {
			oldest = at
		}
	}
	for seq := uint64(1); seq <= info.Delivered.Stream; {
		data, stored, got, err := g.directGet(ctx, q, seq)
		if err != nil {
			return time.Time{}, false, err
		}
		if got == 0 || got > info.Delivered.Stream {
			break
		}
		seq = got + 1
		var r BuildRequest
		if json.Unmarshal(data, &r) != nil {
			continue
		}
		running, ok, err := ReadRunning(ctx, g.conn, r)
		if err != nil {
			return time.Time{}, false, fmt.Errorf("the checks running: %w", err)
		}
		if ok && running.Build == r.ID {
			continue // running now
		}
		at := askedAtOf(data, stored)
		if time.Since(at) < LetThroughAfter {
			return time.Time{}, false, nil // a young redelivery is next in line
		}
		note(at)
	}
	data, stored, got, err := g.directGet(ctx, q, info.Delivered.Stream+1)
	if err != nil {
		return time.Time{}, false, err
	}
	if got != 0 {
		note(askedAtOf(data, stored))
	}
	return oldest, !oldest.IsZero(), nil
}

// directGet reads the first ask at or after a sequence of a check queue's stream, without taking it: its body, the time the
// stream stored it, and its sequence (0 when there is none).
func (g busGate) directGet(ctx context.Context, q *takeQueue, seq uint64) ([]byte, string, uint64, error) {
	body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": q.worker.Filters[0]})
	reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
	if err != nil {
		return nil, "", 0, fmt.Errorf("the asks of the %s queue: %w", q.kind, err)
	}
	if st := reply.Header.Get("Status"); st == "404" || st == "408" {
		return nil, "", 0, nil
	}
	got, _ := strconv.ParseUint(reply.Header.Get("Nats-Sequence"), 10, 64)
	return reply.Data, reply.Header.Get("Nats-Time-Stamp"), got, nil
}

// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
// neither is taken as asked now, so it is never let through on a guess.
func askedAtOf(data []byte, stored string) time.Time {
	var r BuildRequest
	if json.Unmarshal(data, &r) == nil {
		if at, ok := BuildAskedAt(r.ID); ok {
			return at
		}
	}
	if at, err := time.Parse(time.RFC3339Nano, stored); err == nil {
		return at
	}
	return time.Now()
}

func jsOf(conn *nats.Conn) nats.JetStreamContext {
	js, _ := conn.JetStream()
	return js
}

func (g busGate) TakeLetThrough(ctx context.Context, builder string) (bool, error) {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return false, err
	}
	body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
	if _, err := kv.Create(ctx, LetThroughKey, body); err != nil {
		if errors.Is(err, jetstream.ErrKeyExists) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (g busGate) KeepLetThrough(ctx context.Context, builder string) error {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return err
	}
	body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
	_, err = kv.Put(ctx, LetThroughKey, body)
	return err
}

func (g busGate) DropLetThrough(ctx context.Context) error {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return err
	}
	if err := kv.Delete(ctx, LetThroughKey); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
		return err
	}
	return nil
}

// CheckHold is what a reader of the queue (the controller's own verb) says of why checks wait: the plans building and the check
// let through now, if one.
type CheckHold struct {
	Plans      []string
	LetThrough *LetThrough
}

// ReadCheckHold reads the plans building and the let-through key. A bucket that cannot be read is an error the caller says.
func ReadCheckHold(ctx context.Context, conn *nats.Conn) (CheckHold, error) {
	var h CheckHold
	plans, err := BuildingPlans(ctx, conn)
	if err != nil {
		return h, err
	}
	h.Plans = plans
	kv, err := runningBucket(ctx, conn)
	if err != nil {
		return h, err
	}
	e, err := kv.Get(ctx, LetThroughKey)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return h, nil
	}
	if err != nil {
		return h, err
	}
	var l LetThrough
	if json.Unmarshal(e.Value(), &l) == nil {
		h.LetThrough = &l
	}
	return h, nil
}

// gateOf is the machine's gate: the one a test set, or the bus's.
func (m *natsMachine) gateOf() checkGate {
	if m.gate != nil {
		return m.gate
	}
	return busGate{conn: m.js.Conn()}
}

// sayOnce says a fault of the hold the first time it happens and again once it has recovered and happened anew, so a bucket
// that cannot be read is in the log once and not every second.
func (m *natsMachine) sayOnce(kind string, err error) {
	if m.said == nil {
		m.said = map[string]bool{}
	}
	if err == nil {
		delete(m.said, kind)
		return
	}
	if !m.said[kind] {
		m.said[kind] = true
		fmt.Fprintf(os.Stderr, "%s; read as empty, so no check is held for it: %v\n", kind, err)
	}
}

// building says whether a delivery builds: any key in the bucket of the plans building. **A bucket that cannot be read is said
// and read as empty** (ADR 0483 clause 2): a check wrongly held costs a pull request its turn, a fault that held every check
// would stop the mesh checking.
func (m *natsMachine) building(ctx context.Context) []string {
	plans, err := m.gateOf().Building(ctx)
	m.sayOnce("the bucket of the plans building cannot be read", err)
	if err != nil {
		// empty-on-error: ADR 0483 clause 2 — a bucket that cannot be read is said (sayOnce) and read as empty, because a check wrongly held costs a pull request its turn and a fault that held every check would stop the mesh checking.
		return nil
	}
	return plans
}

// letOneThrough takes the oldest waiting check of the two check queues once it waited LetThroughAfter, by the one let-through
// key: nil when none waited that long, when another holder runs the check let through, or when the check was taken by
// another holder in the moment. The caller owns the key while it returns a message and releases it with dropLetThrough.
func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue) {
	gate := m.gateOf()
	var oldest *takeQueue
	var oldestAt time.Time
	for _, q := range m.subs[1:] {
		if !q.bound {
			continue
		}
		at, found, err := gate.OldestWaiting(ctx, q)
		m.sayOnce("the first waiting check of the "+q.kind+" queue cannot be read", err)
		if err != nil || !found {
			continue
		}
		if oldest == nil || at.Before(oldestAt) {
			oldest, oldestAt = q, at
		}
	}
	if oldest == nil || time.Since(oldestAt) < m.letThroughAfter() {
		return nil, nil
	}
	created, err := gate.TakeLetThrough(ctx, m.on)
	m.sayOnce("the let-through key cannot be taken", err)
	if err != nil || !created {
		return nil, nil
	}
	trying, end := context.WithTimeout(ctx, checkTakeWait)
	defer end()
	fetched, err := oldest.pull.Fetch(1, nats.Context(trying))
	if len(fetched) == 0 {
		if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.Canceled) {
			m.sayOnce("the "+oldest.kind+" check queue cannot be taken from", err)
		}
		m.dropLetThrough(ctx)
		return nil, nil
	}
	// Another holder took the one that was oldest in the moment between the read and the fetch: what came is a younger check,
	// handed back and the key released, never run during a delivery.
	if askedAtOf(fetched[0].Data, "").Add(m.letThroughAfter()).After(time.Now()) {
		_ = fetched[0].NakWithDelay(time.Minute) // a delay keeps it off the queue of redeliveries a fetch reads first
		m.dropLetThrough(ctx)
		return nil, nil
	}
	m.letKey = true
	return fetched[0], oldest
}

func (m *natsMachine) letThroughAfter() time.Duration {
	if m.opts.LetThroughAfter > 0 {
		return m.opts.LetThroughAfter
	}
	return LetThroughAfter
}

// dropLetThrough deletes the let-through key this holder created.
func (m *natsMachine) dropLetThrough(ctx context.Context) {
	m.letKey = false
	if err := m.gateOf().DropLetThrough(context.WithoutCancel(ctx)); err != nil {
		fmt.Fprintf(os.Stderr, "could not take the let-through key away; it ages out: %v\n", err)
	}
}

// keepLetThrough refreshes the key at every heartbeat while the check it was created for runs, and returns the end of that: it
// stops the refreshing and deletes the key.
func (m *natsMachine) keepLetThrough(ctx context.Context, every time.Duration) (end func()) {
	done := make(chan struct{})
	finished := make(chan struct{})
	go func() {
		defer close(finished)
		tick := time.NewTicker(every)
		defer tick.Stop()
		for {
			select {
			case <-done:
				return
			case <-tick.C:
				putting, stop := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
				if err := m.gateOf().KeepLetThrough(putting, m.on); err != nil {
					fmt.Fprintf(os.Stderr, "could not keep the let-through key: %v\n", err)
				}
				stop()
			}
		}
	}()
	return func() {
		close(done)
		<-finished
		m.dropLetThrough(ctx)
	}
}

// whyHeld words the plans a check is held for, for the log.
func whyHeld(plans []string) string { return strings.Join(plans, ", ") }