//go:build integration

package link

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/testbus"
)

func aBus(t *testing.T) *broker.JetStream {
	t.Helper()
	url := testbus.URL(t)
	js, err := broker.Dial(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(js.Close)

	// **Streams purged, consumers removed.** Both halves, and each was learned by getting it wrong.
	//
	// The streams are emptied rather than deleted and recreated, because delete-then-add is not a
	// reset: the server's teardown races the creation, and a test then inherits the previous one's
	// messages — which reads as a redelivery bug in the code under test.
	//
	// The consumers are removed, because deleting a stream used to take them with it and purging
	// does not. A durable *push* consumer that survives between tests keeps pushing to a delivery
	// subject the previous test's subscription has gone from: the messages count as delivered, go
	// nowhere, and the next test waits out its timeout for an announcement the server believes it
	// already sent. The controller recreates what it needs on start, so leaving none is correct.
	if err := broker.AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	clean := func() {
		for _, c := range broker.MeshConsumers() {
			_ = js.Context().DeleteConsumer(c.Stream, c.Name)
		}
		for _, s := range broker.MeshStreams() {
			_ = js.Context().PurgeStream(s.Name)
		}
	}
	clean()
	t.Cleanup(clean)
	return js
}

func TestNatsAReportIsHeardAndLeavesTheStream(t *testing.T) {
	js := aBus(t)
	store := &counted{}
	_, stop := servingOn(t, js, store)
	defer stop()

	body, _ := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a report being recorded", func() bool { return store.count() == 1 })
	eventually(t, "the report leaving the work queue", func() bool {
		info, err := js.Context().StreamInfo("CONTROL")
		return err == nil && info.State.Msgs == 0
	})
}

func TestNatsAReportTheStoreCannotTakeIsHeldByTheServerAndComesBack(t *testing.T) {
	js := aBus(t)
	store := &counted{}
	store.refusing(errors.Join(ErrTryAgain, errors.New("the database system is starting up")))
	_, stop := servingOn(t, js, store)
	defer stop()

	body, _ := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}

	// Held: the message is the server's, unacknowledged, and still in the stream.
	eventually(t, "the report being redelivered at least once", func() bool {
		info, err := js.Context().ConsumerInfo("CONTROL", broker.ControllerName)
		return err == nil && info.NumRedelivered >= 1
	})
	info, err := js.Context().StreamInfo("CONTROL")
	if err != nil || info.State.Msgs != 1 {
		t.Fatalf("a held report did not stay on the server: %+v, %v", info, err)
	}
	if store.count() != 0 {
		t.Fatalf("a report was recorded by a store that was refusing it")
	}

	store.refusing(nil)
	eventually(t, "the report being recorded once the store was back",
		func() bool { return store.count() == 1 })
	eventually(t, "the recorded report leaving the work queue", func() bool {
		info, err := js.Context().StreamInfo("CONTROL")
		return err == nil && info.State.Msgs == 0
	})
}

func TestNatsASupersededReportIsSettledAndNotActedOn(t *testing.T) {
	js := aBus(t)
	store := &sentAndHeardSafely{sent: "d2"}
	_, stop := servingOn(t, js, store)
	defer stop()

	body, _ := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the superseded report leaving the stream", func() bool {
		info, err := js.Context().StreamInfo("CONTROL")
		return err == nil && info.State.Msgs == 0
	})
	if store.count() != 0 {
		t.Fatalf("a report about a superseded declaration was acted on")
	}
}

func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {
	js := aBus(t)
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()},
		enroller: enrolsAs{reply: EnrolReply{Accepted: true, Node: "anchor"}}, log: quiet()}
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go func() { _ = s.Serve(ctx) }()

	inbox := nats.NewInbox()
	answers, err := js.Conn().SubscribeSync(inbox)
	if err != nil {
		t.Fatal(err)
	}
	body, _ := json.Marshal(EnrolRequest{Node: "anchor", Secret: "t", ReplyTo: inbox})
	if _, err := js.Context().Publish(EnrolSubject, body); err != nil {
		t.Fatal(err)
	}

	msg, err := answers.NextMsg(8 * time.Second)
	if err != nil {
		t.Fatalf("no answer reached the address the request named: %v", err)
	}
	var reply EnrolReply
	if err := json.Unmarshal(msg.Data, &reply); err != nil {
		t.Fatal(err)
	}
	if !reply.Accepted || reply.Node != "anchor" {
		t.Fatalf("the answer was not the mesh's: %+v", reply)
	}
	// And the address really is not the one the transport carried: what the consumer saw was its
	// own ack subject, which is why this had to travel in the payload.
	if msg.Subject != inbox {
		t.Fatalf("the answer arrived on %s, not the address the request named", msg.Subject)
	}
}

func TestNatsAHeartbeatIsHeardAndNothingIsKept(t *testing.T) {
	js := aBus(t)
	store := &counted{}
	_, stop := servingOn(t, js, store)
	defer stop()

	// Given time to subscribe: a core subscription that is not yet up misses what is published,
	// which is the guarantee a heartbeat has and not a fault.
	eventually(t, "the heartbeat subscription coming up", func() bool {
		body, _ := json.Marshal(Alive{Node: "anchor"})
		_ = js.Conn().Publish(AliveSubject("anchor"), body)
		_ = js.Conn().Flush()
		return store.count() >= 1
	})
	info, err := js.Context().StreamInfo("CONTROL")
	if err != nil || info.State.Msgs != 0 {
		t.Fatalf("a heartbeat was persisted, and the mesh's least valuable message now competes "+
			"for retention with its most valuable: %+v, %v", info, err)
	}
}

func TestNatsTheEventsTheControllerFollowsArriveAndAreAcknowledged(t *testing.T) {
	js := aBus(t)
	told := &toldAbout{}
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
	if err := s.Follows(told); err != nil {
		t.Fatal(err)
	}
	if err := s.Answers(replaysWith{}); err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go func() { _ = s.Serve(ctx) }()
	// The controller's event consumer is made from now (novox/hq issue 248): what was published before
	// it existed is history it never replays. So the announcements are made once it is there.
	eventually(t, "the controller's event consumer being made", func() bool {
		_, err := js.Context().ConsumerInfo("EVENTS", broker.ControllerName)
		return err == nil
	})

	moved, _ := json.Marshal(Upgraded{Module: "gitea", Commit: "abcdef0123"})
	if _, err := js.Context().Publish(broker.ControllerFollows[0], moved); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(broker.ControllerFollows[1], []byte(`{}`)); err != nil {
		t.Fatal(err)
	}

	eventually(t, "the catalogue's upgrade reaching the controller",
		func() bool { return told.count() == 1 })
	eventually(t, "both announcements being acknowledged", func() bool {
		info, err := js.Context().ConsumerInfo("EVENTS", broker.ControllerName)
		return err == nil && info.NumAckPending == 0 && info.Delivered.Consumer == 2
	})
}

func TestNatsAReportIsLetGoOnceTheStoreHasBeenGoneTooLong(t *testing.T) {
	js := aBus(t)
	store := &counted{}
	store.refusing(errors.Join(ErrTryAgain, errors.New("connection refused")))
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()},
		listener: store, log: quiet(), giveUp: 1500 * time.Millisecond}
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go func() { _ = s.Serve(ctx) }()

	body, _ := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the report being let go once the bound passed", func() bool {
		info, err := js.Context().StreamInfo("CONTROL")
		return err == nil && info.State.Msgs == 0
	})
	if store.count() != 0 {
		t.Fatalf("a report was recorded by a store that never came back")
	}
}

func TestNatsWorkSlowerThanTheWindowIsNotHandedOverAgain(t *testing.T) {
	// Work that outlives a window of a second three times over, said while it runs four times a window (novox/hq
	// issue 717: the test once set its window before the serving loop asserted the controller's own of thirty
	// seconds, so it waited nine seconds and passed with the heartbeat removed).
	before := heartbeatWhileWorking
	heartbeatWhileWorking = 250 * time.Millisecond
	t.Cleanup(func() { heartbeatWhileWorking = before })
	js := aBus(t)
	var mu sync.Mutex
	handled := 0
	slow := make(chan struct{})
	s, stop := servingOn(t, js, nil)
	defer stop()
	s.listener = slowly{func() {
		mu.Lock()
		handled++
		first := handled == 1
		mu.Unlock()
		if first {
			time.Sleep(3 * time.Second)
			close(slow)
		}
	}}
	// The controller's own consumer, given a window short enough to outlive in a test once the serving loop has
	// asserted it and bound to it: asserted after this, it would be the controller's window of thirty seconds,
	// which is how this test once waited out a window it never had.
	for deadline := time.Now().Add(10 * time.Second); ; {
		info, err := js.Context().ConsumerInfo("CONTROL", broker.ControllerName)
		if err == nil && info.PushBound {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("the serving loop never bound the controller's consumer: %v", err)
		}
		time.Sleep(10 * time.Millisecond)
	}
	if err := js.EnsureConsumer(broker.Consumer{
		Name: broker.ControllerName, Stream: "CONTROL", Push: true, AckWaitSeconds: 1,
		Why: "a window short enough to outlive in a test",
	}); err != nil {
		t.Fatal(err)
	}
	if info, err := js.Context().ConsumerInfo("CONTROL", broker.ControllerName); err != nil || info.Config.AckWait != time.Second {
		t.Fatalf("the controller's consumer does not have the test's window of a second: %v", err)
	}

	body, err := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	select {
	case <-slow:
	case <-time.After(30 * time.Second):
		t.Fatal("the slow work never finished")
	}
	// A moment for a redelivery to arrive, if the bus were going to send one.
	time.Sleep(1500 * time.Millisecond)
	mu.Lock()
	defer mu.Unlock()
	if handled != 1 {
		t.Fatalf("one report was handled %d times, so slow work is run again while it is running", handled)
	}
	// Not handed over again at all: the loop drops a copy of what it is already doing, so the count above alone
	// would pass on a bus that redelivered the work every window; the bus's own count says it did not.
	info, err := js.Context().ConsumerInfo("CONTROL", broker.ControllerName)
	if err != nil {
		t.Fatal(err)
	}
	if info.NumRedelivered != 0 {
		t.Fatalf("the bus handed the report over again %d time(s) while it was being worked on", info.NumRedelivered)
	}
}

func TestNatsTheMeshSaysWhatAMachineApplied(t *testing.T) {
	js := aBus(t)
	heard := make(chan *nats.Msg, 4)
	sub, err := js.Conn().Subscribe(SeatEventSubject(MeshControllerSeat, ">"), func(m *nats.Msg) {
		heard <- m
	})
	if err != nil {
		t.Fatal(err)
	}
	defer sub.Unsubscribe() //nolint:errcheck // the subscription dies with the connection

	_, stop := servingOn(t, js, &counted{})
	defer stop()

	// A report that changed something: the store says it was news.
	body, err := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store", "broker"}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	select {
	case m := <-heard:
		if m.Subject != SeatEventSubject(MeshControllerSeat, KeyApplied) {
			t.Fatalf("the mesh stated %q", m.Subject)
		}
		var said Applied
		if err := json.Unmarshal(m.Data, &said); err != nil {
			t.Fatal(err)
		}
		if said.Node != "anchor" || said.Declared != "d1" || said.Resources != 2 {
			t.Fatalf("it said %+v", said)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the mesh said nothing about a machine that now runs something else")
	}

	// A refusal is its own fact, with the reason in it rather than only in a log.
	refusal, err := json.Marshal(Report{Node: "anchor", Declared: "d2",
		Failed: map[string]string{"gitea.server": "no such image"}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ReportSubject("anchor"), refusal); err != nil {
		t.Fatal(err)
	}
	select {
	case m := <-heard:
		if m.Subject != SeatEventSubject(MeshControllerSeat, KeyRefused) {
			t.Fatalf("a refusal was stated as %q", m.Subject)
		}
		var said Refused
		if err := json.Unmarshal(m.Data, &said); err != nil {
			t.Fatal(err)
		}
		if said.Failed["gitea.server"] == "" {
			t.Fatalf("the refusal does not say which resource or why: %+v", said)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the mesh said nothing about a machine that refused what it was sent")
	}
}

func TestNatsAReportThatIsNotNewsIsNotStated(t *testing.T) {
	js := aBus(t)
	heard := make(chan *nats.Msg, 4)
	sub, err := js.Conn().Subscribe(SeatEventSubject(MeshControllerSeat, ">"), func(m *nats.Msg) {
		heard <- m
	})
	if err != nil {
		t.Fatal(err)
	}
	defer sub.Unsubscribe() //nolint:errcheck // the subscription dies with the connection

	// A store that records the report and says it was nothing new — which is what the mesh's own
	// store says about a machine repeating itself.
	_, stop := servingOn(t, js, sameAgain{})
	defer stop()

	body, err := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
		t.Fatal(err)
	}
	select {
	case m := <-heard:
		t.Fatalf("the mesh stated %q about a machine that changed nothing", m.Subject)
	case <-time.After(3 * time.Second):
	}
}

func TestNatsAMergeHeldMoreThanFiveTimesIsStillHandedOverAndRecorded(t *testing.T) {
	first, most := MergeHeldFirst, MergeHeldMost
	MergeHeldFirst, MergeHeldMost = 5*time.Millisecond, 40*time.Millisecond
	t.Cleanup(func() { MergeHeldFirst, MergeHeldMost = first, most })
	js := aBus(t)
	up := &manyTimesBusy{busy: 8}
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
	if err := s.Follows(up); err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go func() { _ = s.Serve(ctx) }()
	eventually(t, "the merges consumer being made", func() bool {
		_, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
		return err == nil
	})

	body, _ := json.Marshal(SourceMoved{Owner: "novox", Repo: "one", Base: "main", Commit: "abcdef0123"})
	if _, err := js.Context().Publish(MergedSubject, body); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the ninth hand-over recording the merge", func() bool { return up.called() == 9 })
	eventually(t, "the merge being acknowledged", func() bool {
		info, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName)
		return err == nil && info.NumAckPending == 0 && info.NumPending == 0
	})
	if info, err := js.Context().StreamInfo(broker.DeadLettersStream); err == nil && info.State.Msgs != 0 {
		t.Fatalf("%d message(s) in dead letters: a merge was given up on", info.State.Msgs)
	}
}