//go:build integration

package broker

import (
	"testing"
	"time"

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

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

// 1178/AC4, on a real bus: the controller ensuring mesh-mail's work-queue stream and its one consumer gives
// the bus's own guarantees — a duplicate publish within the window is taken once, a message naked past
// max-deliver is not handed over again, and a full stream refuses a new publish (ADR 0436 §6).
//
//	go test -tags integration ./internal/broker/ -run TestAWorkQueueStreamOnARealBus
func TestAWorkQueueStreamOnARealBus(t *testing.T) {
	js, err := Dial(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer js.Close()

	// Small enough that a handful of messages fill it, so "the stream is full" is reachable within a test.
	stream := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
		MaxAgeDays: 1, MaxMiB: 1, WorkQueue: true, Duplicates: 1}
	if err := js.EnsureStream(stream.Stream()); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = js.Context().DeleteStream(stream.Name) })

	owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5}
	sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"}
	consumer := QueueConsumer(sender, owner)
	consumer.AckWaitSeconds = 1
	if err := js.EnsureConsumer(consumer); err != nil {
		t.Fatal(err)
	}

	subject := ModuleStreamSubject("mesh-mail", "mail.send.approved")

	// A duplicate publish within the window is kept once (ADR 0436 §6's duplicate window).
	publish := func(id string, size int) (*nats.PubAck, error) {
		return js.Context().Publish(subject, make([]byte, size), nats.MsgId(id))
	}
	if _, err := publish("dup-1", 16); err != nil {
		t.Fatal(err)
	}
	if _, err := publish("dup-1", 16); err != nil {
		t.Fatal(err)
	}
	if info, err := js.Context().StreamInfo(stream.Name); err != nil || info.State.Msgs != 1 {
		t.Fatalf("a message published twice with the same id: %v msgs, err %v; want 1", info, err)
	}

	// Naked five times, the consumer is not handed the message a sixth time.
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	reading, err := api.Consumer(t.Context(), stream.Name, consumer.Name)
	if err != nil {
		t.Fatal(err)
	}
	for i := 0; i < owner.MaxDeliver; i++ {
		batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
		if err != nil {
			t.Fatal(err)
		}
		got := false
		for msg := range batch.Messages() {
			got = true
			_ = msg.Nak()
		}
		if !got {
			t.Fatalf("delivery %d of %d never arrived", i+1, owner.MaxDeliver)
		}
	}
	batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
	if err != nil {
		t.Fatal(err)
	}
	for range batch.Messages() {
		t.Fatalf("the message was delivered a %dth time, past its max-deliver of %d", owner.MaxDeliver+1, owner.MaxDeliver)
	}

	// Full, the stream refuses a new publish rather than dropping what it holds (DiscardNew).
	var fullErr error
	for i := 0; i < 1<<16; i++ {
		if _, err := publish("fill-"+time.Now().String(), 1<<16); err != nil {
			fullErr = err
			break
		}
	}
	if fullErr == nil {
		t.Fatal("a work-queue stream accepted every publish; want it to refuse one once full")
	}
}