//go:build integration

package broker

import (
	"fmt"
	"testing"
	"time"

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

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

// A consumer that reacts to announcements, made on a stream that keeps a week of them, starts from now:
// made from the start it replays every merge and build of that week (novox/hq issue 248). And a reset
// re-makes a stuck one from now, its configuration otherwise kept, and refuses a work queue.
//
//	go test ./internal/broker/ -run TestAConsumerMadeFromNow   (each test on a bus of its own: internal/testbus)
func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
	url := testbus.URL(t)
	js, err := Dial(url)
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()
	const stream = "HISTORY_TEST"
	_ = js.js.DeleteStream(stream)
	if _, err := js.js.AddStream(&nats.StreamConfig{Name: stream, Subjects: []string{"history.>"}, Storage: nats.MemoryStorage}); err != nil {
		t.Fatal(err)
	}
	defer func() { _ = js.js.DeleteStream(stream) }()
	for i := 0; i < 5; i++ {
		if _, err := js.js.Publish("history.merged", []byte(fmt.Sprint(i))); err != nil {
			t.Fatal(err)
		}
	}

	fresh := Consumer{Name: "fresh", Stream: stream, Filters: []string{"history.merged"}, Push: true,
		AckWaitSeconds: 30, MaxDeliver: 5, MaxAckPending: 1, FromNow: true}
	if err := js.EnsureConsumer(fresh); err != nil {
		t.Fatal(err)
	}
	info, _ := js.js.ConsumerInfo(stream, "fresh")
	if info.NumPending != 0 || info.Config.DeliverPolicy != nats.DeliverNewPolicy {
		t.Fatalf("a consumer made from now holds %d of the stream's past (policy %v)", info.NumPending, info.Config.DeliverPolicy)
	}
	_, _ = js.js.Publish("history.merged", []byte("new"))
	time.Sleep(100 * time.Millisecond)
	if info, _ = js.js.ConsumerInfo(stream, "fresh"); info.NumPending != 1 {
		t.Fatalf("a new announcement is not pending: %d", info.NumPending)
	}
	// Asserted again, it keeps where it is.
	if err := js.EnsureConsumer(fresh); err != nil {
		t.Fatal(err)
	}

	// One made from the start, as the server's default makes it, and stuck behind its history.
	old := fresh
	old.Name, old.FromNow = "old", false
	if err := js.EnsureConsumer(old); err != nil {
		t.Fatal(err)
	}
	if info, _ = js.js.ConsumerInfo(stream, "old"); info.NumPending != 6 {
		t.Fatalf("the default should replay all six: %d", info.NumPending)
	}
	before, after, err := js.ResetConsumer(stream, "old")
	if err != nil {
		t.Fatal(err)
	}
	info, _ = js.js.ConsumerInfo(stream, "old")
	if before.Pending != 6 || after.Pending != 0 || info.Config.MaxAckPending != 1 || info.Config.MaxDeliver != 5 ||
		info.Config.AckWait != 30*time.Second || info.Config.DeliverSubject == "" {
		t.Fatalf("before %+v after %+v config %+v", before, after, info.Config)
	}

	// Never a work queue: what is pending there is work.
	const queue = "QUEUE_TEST"
	_ = js.js.DeleteStream(queue)
	if _, err := js.js.AddStream(&nats.StreamConfig{Name: queue, Subjects: []string{"queue.>"}, Retention: nats.WorkQueuePolicy, Storage: nats.MemoryStorage}); err != nil {
		t.Fatal(err)
	}
	defer func() { _ = js.js.DeleteStream(queue) }()
	if _, err := js.js.AddConsumer(queue, &nats.ConsumerConfig{Durable: "w", AckPolicy: nats.AckExplicitPolicy}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := js.ResetConsumer(queue, "w"); err == nil {
		t.Fatal("a work queue's consumer was reset")
	}
}

// pendingOf polls a consumer's pending count until it is want or a second has passed, and returns what it saw last.
func pendingOf(js *JetStream, stream, name string, want uint64) uint64 {
	var got uint64
	for deadline := time.Now().Add(time.Second); ; time.Sleep(20 * time.Millisecond) {
		if info, err := js.js.ConsumerInfo(stream, name); err == nil {
			got = info.NumPending
		}
		if got == want || time.Now().After(deadline) {
			return got
		}
	}
}

// 1279/AC2 (ADR 0443 §1): on a bus whose EVENTS holds earlier events, the consumer the controller makes for a module
// delivers none of them and delivers one published after its making; the same on another module's stream.
func TestAModulesConsumerDeliversNothingOfTheStreamsPast(t *testing.T) { // 1279/AC2
	js := aLiveBus(t)
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := js.Context().PurgeStream(EventsStream); err != nil {
		t.Fatal(err)
	}
	const subject = "mesh.mod.shop.event.order.placed"
	for i := 0; i < 3; i++ {
		if _, err := js.Context().Publish(subject, []byte(fmt.Sprint(i))); err != nil {
			t.Fatal(err)
		}
	}
	c, ok := ConsumerFor(Principal{Kind: KindModule, Node: "one", Module: "audit", Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	if !ok {
		t.Fatal("no consumer derived")
	}
	t.Cleanup(func() { _ = js.Context().DeleteConsumer(c.Stream, c.Name) })
	if err := js.EnsureConsumer(c); err != nil {
		t.Fatal(err)
	}
	if got := pendingOf(js, c.Stream, c.Name, 0); got != 0 {
		t.Fatalf("a module's consumer made after 3 events holds %d of them", got)
	}
	if _, err := js.Context().Publish(subject, []byte("after")); err != nil {
		t.Fatal(err)
	}
	if got := pendingOf(js, c.Stream, c.Name, 1); got != 1 {
		t.Fatalf("the event published after its making is not pending: %d", got)
	}

	// Another module's stream.
	s := mailStream.Stream()
	if err := js.EnsureStream(s); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = js.js.DeleteStream(s.Name) })
	for _, m := range []string{"added", "changed"} {
		if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail.mesh.message."+m), []byte("{}")); err != nil {
			t.Fatal(err)
		}
	}
	reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
	h := heardConsumer(reader, mailHeard)
	t.Cleanup(func() { _ = js.js.DeleteConsumer(h.Stream, h.Name) })
	if err := js.EnsureConsumer(h); err != nil {
		t.Fatal(err)
	}
	if got := pendingOf(js, h.Stream, h.Name, 0); got != 0 {
		t.Fatalf("a module's consumer on another module's stream holds %d of its past", got)
	}
	if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail.mesh.message.added"), []byte("{}")); err != nil {
		t.Fatal(err)
	}
	if got := pendingOf(js, h.Stream, h.Name, 1); got != 1 {
		t.Fatalf("the event published after its making is not pending on the module's stream: %d", got)
	}
}

// 1279/AC3 (ADR 0443 §2): a consumer that exists, made from the start, asserted again as the controller now derives it,
// keeps its deliver policy and delivers its next message from where it was.
func TestAnExistingConsumerKeepsItsPositionWhenAssertedAgain(t *testing.T) { // 1279/AC3
	js := aLiveBus(t)
	if err := AssertMeshStreams(js); err != nil {
		t.Fatal(err)
	}
	if err := js.Context().PurgeStream(EventsStream); err != nil {
		t.Fatal(err)
	}
	const subject = "mesh.mod.shop.event.order.placed"
	for i := 0; i < 3; i++ {
		if _, err := js.Context().Publish(subject, []byte(fmt.Sprint(i))); err != nil {
			t.Fatal(err)
		}
	}
	c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "one", Module: "audit", Consumes: []string{"shop.order.placed"}, PasswordHash: "x"})
	t.Cleanup(func() { _ = js.Context().DeleteConsumer(c.Stream, c.Name) })
	old := c
	old.FromNow = false // as the controller made it before ADR 0443
	if err := js.EnsureConsumer(old); err != nil {
		t.Fatal(err)
	}
	next := func() string {
		t.Helper()
		sub, err := js.Context().PullSubscribe("", c.Name, nats.Bind(c.Stream, c.Name))
		if err != nil {
			t.Fatal(err)
		}
		defer func() { _ = sub.Unsubscribe() }()
		msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
		if err != nil || len(msgs) != 1 {
			t.Fatalf("fetch: %v (%d messages)", err, len(msgs))
		}
		_ = msgs[0].Ack()
		return string(msgs[0].Data)
	}
	if got := next(); got != "0" {
		t.Fatalf("the first message is %q", got)
	}
	if err := js.EnsureConsumer(c); err != nil { // now derived from now
		t.Fatal(err)
	}
	info, err := js.Context().ConsumerInfo(c.Stream, c.Name)
	if err != nil {
		t.Fatal(err)
	}
	if info.Config.DeliverPolicy != nats.DeliverAllPolicy {
		t.Fatalf("asserting moved the deliver policy to %v", info.Config.DeliverPolicy)
	}
	if got := next(); got != "1" {
		t.Fatalf("after asserting again the next message is %q, want 1: the position moved", got)
	}
}