//go:build integration

package link

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"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, 1688), against a real bus.

// aBusWithTheBuckets is the bus with the build role and the controller's buckets, the bucket of the checks running with an age
// of two seconds so a key a holder stopped refreshing is seen to expire.
func aBusWithTheBuckets(t *testing.T) (*broker.JetStream, jetstream.KeyValue) {
	t.Helper()
	js := aBusWithTheBuildRole(t)
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	ctx := t.Context()
	if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.CheckRunningBucket, History: 1, TTL: 2 * time.Second}); err != nil {
		t.Fatal(err)
	}
	plans, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.PlansBuildingBucket, History: 1, TTL: broker.PlansBuildingKeptFor})
	if err != nil {
		t.Fatal(err)
	}
	return js, plans
}

func aCheckAskedAgo(ago time.Duration, number int, ready bool) BuildRequest {
	r := aPullCheck(fmt.Sprintf("build-%d", time.Now().Add(-ago).UnixNano()), number, fmt.Sprintf("head-%d-0000000", number), "trunk-0-000000", ready)
	return r
}

// 1652/AC6: a plan's builds asked in two tiers with a check waiting, the check is handed out only after the last tier's builds,
// and not in the gap between the tiers.
func TestNatsACheckIsHandedOutOnlyAfterTheLastTierOfAPlan(t *testing.T) {
	js, plans := aBusWithTheBuckets(t)
	ctx, stop := context.WithCancel(t.Context())
	defer stop()
	asks := BuildsOn(js, TheBuildMachine)

	if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
		t.Fatal(err)
	}
	if err := asks.Ask(ctx, aCheckAskedAgo(time.Minute, 7, true)); err != nil {
		t.Fatal(err)
	}
	if err := asks.Ask(ctx, BuildRequest{ID: "tier-1-a", Repository: "/a"}); err != nil {
		t.Fatal(err)
	}
	took := make(chan string, 16)
	machine := MachineOverNATS(js, "anchor")
	defer machine.Close()
	go func() {
		_ = machine.Take(ctx, func(ctx context.Context, work Build) {
			took <- work.Request().ID
			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
			_ = work.Done()
		})
	}()
	expect := func(want string) {
		t.Helper()
		select {
		case got := <-took:
			if got != want {
				t.Fatalf("taken %s, want %s", got, want)
			}
		case <-time.After(15 * time.Second):
			t.Fatalf("%s was not taken", want)
		}
	}
	expect("tier-1-a")
	// The gap between two tiers: the delivery's queue is empty, the plan's key is there, and the check is not taken.
	select {
	case got := <-took:
		t.Fatalf("%s was taken in the gap between two tiers", got)
	case <-time.After(3 * time.Second):
	}
	if err := asks.Ask(ctx, BuildRequest{ID: "tier-2-a", Repository: "/b"}); err != nil {
		t.Fatal(err)
	}
	expect("tier-2-a")
	select {
	case got := <-took:
		t.Fatalf("%s was taken before the plan's last build was built and its key deleted", got)
	case <-time.After(2 * time.Second):
	}
	// The phase ends: the key is deleted, and the check is handed out.
	if err := plans.Delete(ctx, "plan-1"); err != nil {
		t.Fatal(err)
	}
	select {
	case got := <-took:
		if got == "tier-1-a" || got == "tier-2-a" {
			t.Fatalf("a build was taken twice: %s", got)
		}
	case <-time.After(15 * time.Second):
		t.Fatal("the check was not handed out once the plan's key was deleted")
	}
}

// 1688: two holders racing for checks that waited past the bound: exactly one is let through at a time, by a create that fails
// while the key is there, and the key is deleted when its check ends, after which the next oldest goes.
func TestNatsTwoHoldersRacingForOldChecksLetExactlyOneThrough(t *testing.T) {
	js, plans := aBusWithTheBuckets(t)
	ctx, stop := context.WithCancel(t.Context())
	defer stop()
	// The buckets' age is two seconds here, so the keys are refreshed well within it.
	was := heartbeatWhileWorking
	heartbeatWhileWorking = 400 * time.Millisecond
	t.Cleanup(func() { heartbeatWhileWorking = was })
	asks := BuildsOn(js, TheBuildMachine)
	if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
		t.Fatal(err)
	}
	for i, ago := range []time.Duration{50 * time.Minute, 40 * time.Minute, 35 * time.Minute} {
		if err := asks.Ask(ctx, aCheckAskedAgo(ago, i+1, false)); err != nil {
			t.Fatal(err)
		}
	}
	var mu sync.Mutex
	running, most := 0, 0
	started := make(chan string, 8)
	release := make(chan struct{})
	do := func(ctx context.Context, work Build) {
		mu.Lock()
		running++
		most = max(most, running)
		mu.Unlock()
		started <- work.Request().Check.Repo + fmt.Sprint(work.Request().Check.Number)
		<-release
		mu.Lock()
		running--
		mu.Unlock()
		_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "x", Failed: "no"})
		_ = work.Done()
	}
	for _, on := range []string{"ace", "shanks"} {
		machine := MachineOverNATS(js, on)
		defer machine.Close()
		go func() { _ = machine.Take(ctx, do) }()
	}
	var order []string
	for range 3 {
		select {
		case id := <-started:
			order = append(order, id)
		case <-time.After(20 * time.Second):
			t.Fatalf("started so far: %v", order)
		}
		// While it runs, no second check starts and the key is there.
		select {
		case id := <-started:
			t.Fatalf("%s started beside a check let through", id)
		case <-time.After(2500 * time.Millisecond):
		}
		kv := runningKV(t, js)
		if _, err := kv.Get(ctx, LetThroughKey); err != nil {
			t.Fatalf("the let-through key is not there while the check runs: %v", err)
		}
		release <- struct{}{}
		// It is deleted when the check ends: the next check let through creates the key again at once, and a create fails
		// while the key is there, so its start (the next round) is the proof; after the last one the key stays gone.
		if len(order) == 3 {
			deadline := time.Now().Add(10 * time.Second)
			for {
				if _, err := kv.Get(ctx, LetThroughKey); err != nil {
					break
				}
				if time.Now().After(deadline) {
					t.Fatal("the let-through key was not deleted when the last check ended")
				}
				time.Sleep(50 * time.Millisecond)
			}
		}
	}
	if want := []string{"a1", "a2", "a3"}; !reflect.DeepEqual(order, want) {
		t.Fatalf("checks went %v, want the oldest first %v", order, want)
	}
	mu.Lock()
	defer mu.Unlock()
	if most != 1 {
		t.Fatalf("%d checks ran at once while a delivery built", most)
	}
}

func runningKV(t *testing.T, js *broker.JetStream) jetstream.KeyValue {
	t.Helper()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	kv, err := api.KeyValue(t.Context(), broker.CheckRunningBucket)
	if err != nil {
		t.Fatal(err)
	}
	return kv
}

// 1688: the create of the let-through key succeeds for exactly one of holders asking at once, and a key its holder stopped
// refreshing expires, after which the next create succeeds.
func TestNatsOneCreateOfTheLetThroughKeySucceedsAndAStoppedHoldersKeyExpires(t *testing.T) {
	js, _ := aBusWithTheBuckets(t)
	gate := busGate{conn: js.Conn()}
	ctx := t.Context()
	var wg sync.WaitGroup
	var mu sync.Mutex
	won := 0
	for i := range 8 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ok, err := gate.TakeLetThrough(ctx, fmt.Sprint("holder-", i))
			if err != nil {
				t.Error(err)
				return
			}
			if ok {
				mu.Lock()
				won++
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	if won != 1 {
		t.Fatalf("%d of 8 holders created the let-through key, want exactly one", won)
	}
	// The holder stopped: nothing refreshes the key, and the bucket's age (two seconds here) removes it.
	deadline := time.Now().Add(15 * time.Second)
	for {
		ok, err := gate.TakeLetThrough(ctx, "next")
		if err != nil {
			t.Fatal(err)
		}
		if ok {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the key of a stopped holder did not expire")
		}
		time.Sleep(250 * time.Millisecond)
	}
	// And a refreshed key does not expire.
	for range 6 {
		if err := gate.KeepLetThrough(ctx, "next"); err != nil {
			t.Fatal(err)
		}
		time.Sleep(700 * time.Millisecond)
	}
	if ok, _ := gate.TakeLetThrough(ctx, "other"); ok {
		t.Fatal("a refreshed key expired")
	}
	var v LetThrough
	e, err := runningKV(t, js).Get(ctx, LetThroughKey)
	if err != nil || json.Unmarshal(e.Value(), &v) != nil || v.Builder != "next" {
		t.Fatalf("the key says %+v (%v)", v, err)
	}
}

// 1687: the oldest waiting ask is read without being taken: no delivery is counted against it.
func TestNatsTheOldestWaitingAskIsReadWithoutTakingIt(t *testing.T) {
	js, _ := aBusWithTheBuckets(t)
	ctx := t.Context()
	asks := BuildsOn(js, TheBuildMachine)
	old := aCheckAskedAgo(45*time.Minute, 1, false)
	if err := asks.Ask(ctx, old); err != nil {
		t.Fatal(err)
	}
	gate := busGate{conn: js.Conn()}
	q := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]}
	at, found, err := gate.OldestWaiting(ctx, q)
	if err != nil || !found {
		t.Fatalf("the first waiting ask was not read: %v %v", found, err)
	}
	if want, _ := BuildAskedAt(old.ID); !at.Equal(want) {
		t.Fatalf("read %v, asked %v", at, want)
	}
	for range 3 {
		if _, _, err := gate.OldestWaiting(ctx, q); err != nil {
			t.Fatal(err)
		}
	}
	info, err := js.Context().ConsumerInfo(q.worker.Stream, q.worker.Name)
	if err != nil {
		t.Fatal(err)
	}
	if info.Delivered.Consumer != 0 || info.NumAckPending != 0 || info.NumPending != 1 {
		t.Fatalf("reading the age took the ask: delivered %d, pending acks %d, waiting %d", info.Delivered.Consumer, info.NumAckPending, info.NumPending)
	}
	// And an empty queue reads as none.
	if _, found, err := gate.OldestWaiting(ctx, &takeQueue{kind: queueReady, worker: broker.CheckWorkers()[0]}); err != nil || found {
		t.Fatalf("an empty queue read as %v (%v)", found, err)
	}
}

// Issue 1747: a message that reaches a builder's subscription outside a fetch is not held while the builder runs another ask. A
// stray check lands in the other queue's subscription of builder A, which then runs a long check; the server hands the stray
// out again when its wait for an acknowledgement ends (two seconds here), builder B runs it, and A, finished, does not run it a
// second time from the subscription it held.
func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAsk(t *testing.T) { // 1747
	js := aBusWithTheBuildRole(t)
	ctx, stop := context.WithCancel(t.Context())
	defer stop()
	for _, w := range broker.CheckWorkers() {
		w.AckWaitSeconds = 2
		if err := js.EnsureConsumer(w); err != nil {
			t.Fatal(err)
		}
	}
	asks := BuildsOn(js, TheBuildMachine)
	long := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()), 1, "head-1-0000000", "trunk-0-000000", true)
	stray := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()+1), 2, "head-2-0000000", "trunk-0-000000", false)
	for _, r := range []BuildRequest{long, stray} {
		if err := asks.Ask(ctx, r); err != nil {
			t.Fatal(err)
		}
	}

	var mu sync.Mutex
	runs := map[string][]string{} // ask id -> the machines that ran it
	release := make(chan struct{})
	started := make(chan string, 8)
	do := func(on string) func(context.Context, Build) {
		return func(ctx context.Context, work Build) {
			id := work.Request().ID
			mu.Lock()
			runs[id] = append(runs[id], on)
			mu.Unlock()
			started <- on + ":" + id
			if id == long.ID {
				<-release
			}
			_ = work.Announce(ctx, BuildResult{ID: id, On: on, Failed: "no"})
			_ = work.Done()
		}
	}

	var a *natsMachine
	strayLanded := make(chan error, 1)
	a = MachineOverNATSWith(js, "ace", TheBuildMachine, MachineOptions{Claim: func() bool {
		// The moment the long check is claimed: the server hands the other queue's waiting check to A's subscription there, as
		// one that arrives after a fetch gave up does. Done once.
		select {
		case <-strayLanded:
			return true
		default:
		}
		other := a.subs[2]
		inbox := strings.TrimSuffix(other.sub.Subject, "*") + "stray"
		body, _ := json.Marshal(map[string]any{"batch": 1, "expires": int64(30 * time.Second)})
		if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, inbox, body); err != nil {
			strayLanded <- err
			return true
		}
		_ = js.Conn().Flush()
		for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
			if n, _, _ := other.sub.Pending(); n > 0 {
				break
			}
		}
		strayLanded <- nil
		return true
	}}).(*natsMachine)
	defer a.Close()
	go func() { _ = a.Take(ctx, do("ace")) }()

	select {
	case got := <-started:
		if got != "ace:"+long.ID {
			t.Fatalf("started %s, want the long check on ace first", got)
		}
	case <-time.After(15 * time.Second):
		t.Fatal("the long check did not start")
	}
	if err := <-strayLanded; err != nil {
		t.Fatal(err)
	}

	// B, free, runs the stray once the server hands it out again.
	b := MachineOverNATS(js, "shanks")
	defer b.Close()
	go func() { _ = b.Take(ctx, do("shanks")) }()
	select {
	case got := <-started:
		if got != "shanks:"+stray.ID {
			t.Fatalf("started %s, want the stray on shanks", got)
		}
	case <-time.After(20 * time.Second):
		t.Fatal("the stray was held by the builder that ran the long check and nobody ran it")
	}
	// A finishes, and does not run the stray from the subscription it held.
	release <- struct{}{}
	select {
	case got := <-started:
		t.Fatalf("%s started after the long check: the stray ran twice", got)
	case <-time.After(6 * time.Second):
	}
	mu.Lock()
	defer mu.Unlock()
	if len(runs[stray.ID]) != 1 {
		t.Fatalf("the stray ran on %v, want once", runs[stray.ID])
	}
}