//go:build integration

package process

import (
	"context"
	"testing"
	"time"

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

// mailSendsApprovedSubject is where the runtime publishes a work-queue message on mesh-mail's own stream
// (mesh-controller internal/broker/modstreams.go ModuleStreamSubject, mail-sends' one event).
const mailSendsApprovedSubject = "mesh.stream.mesh-mail.event.mail.send.approved"

// makeQueue sets up MAIL_SENDS as a work-queue stream and binds the one durable consumer a process's
// bus.queues names it (QueueConsumerName), both as the admin, mirroring what the controller composes
// (mesh-controller internal/broker/modstreams.go QueueConsumer), named <node>_<module>_<process id>.
// ackWait is the controller's 30s in most tests, shortened in the Term test so waiting past it (to prove
// no redelivery) does not make that test slow.
func makeQueue(t *testing.T, url string, account Account, ackWait time.Duration) {
	t.Helper()
	_, adminJetStream := adminJS(t, url)
	if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
		Name: "MAIL_SENDS", Subjects: []string{mailSendsApprovedSubject}, Retention: jetstream.WorkQueuePolicy,
		Discard: jetstream.DiscardNew,
	}); err != nil {
		t.Fatalf("the admin could not make the work-queue stream: %v", err)
	}
	if _, err := adminJetStream.CreateConsumer(ctxFor(t), "MAIL_SENDS", jetstream.ConsumerConfig{
		Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait,
		MaxDeliver: 5,
	}); err != nil {
		t.Fatalf("the admin could not make the queue consumer: %v", err)
	}
}

func publishApproved(t *testing.T, url string, body string) {
	t.Helper()
	nc, _ := adminJS(t, url)
	if err := nc.Publish(mailSendsApprovedSubject, []byte(body)); err != nil {
		t.Fatal(err)
	}
	if err := nc.FlushTimeout(3 * time.Second); err != nil {
		t.Fatal(err)
	}
}

// recv reads one value off ch, failing the test rather than hanging forever if nothing arrives within
// timeout.
func recv[T any](t *testing.T, ch <-chan T, timeout time.Duration, what string) T {
	t.Helper()
	select {
	case v := <-ch:
		return v
	case <-time.After(timeout):
		t.Fatalf("%s: nothing arrived within %s", what, timeout)
		var zero T
		return zero
	}
}

// streamMsgs is how many messages MAIL_SENDS still holds.
func streamMsgs(t *testing.T, url string) uint64 {
	t.Helper()
	_, adminJetStream := adminJS(t, url)
	stream, err := adminJetStream.Stream(ctxFor(t), "MAIL_SENDS")
	if err != nil {
		t.Fatal(err)
	}
	info, err := stream.Info(ctxFor(t))
	if err != nil {
		t.Fatal(err)
	}
	return info.State.Msgs
}

// 1190/AC1: given a work-queue stream with one durable pull consumer and a process connection, when a
// message is consumed and acknowledged, then it is gone from the stream.
func TestConsumeAckRemovesTheMessageFromTheStream(t *testing.T) {
	url, account := grantedSendProcess(t)
	makeQueue(t, url, account, 30*time.Second)
	publishApproved(t, url, `{"send":"s1"}`)

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	got := make(chan *Message, 1)
	go func() {
		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
			got <- m
			cancel()
		})
	}()
	msg := recv(t, got, 8*time.Second, "the message")
	<-ctx.Done()
	if err := msg.Ack(); err != nil {
		t.Fatalf("Ack: %v", err)
	}

	deadline := time.Now().Add(5 * time.Second)
	for {
		if n := streamMsgs(t, url); n == 0 {
			return
		} else if time.Now().After(deadline) {
			t.Fatalf("the stream still holds %d messages after the ack", n)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

// 1190/AC2: given a consumed message, when the handler asks for it to be delivered again after a delay,
// then it comes back no sooner than the delay with its delivery count raised by one.
func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T) {
	url, account := grantedSendProcess(t)
	makeQueue(t, url, account, 30*time.Second)
	publishApproved(t, url, `{"send":"s2"}`)

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	const delay = 2 * time.Second
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	type delivery struct {
		at        time.Time
		delivered uint64
	}
	deliveries := make(chan delivery, 2)
	go func() {
		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
			deliveries <- delivery{time.Now(), m.Delivered}
			if m.Delivered == 1 {
				_ = m.NakWithDelay(delay)
			} else {
				_ = m.Ack()
				cancel()
			}
		})
	}()

	first := recv(t, deliveries, 8*time.Second, "the first delivery")
	if first.delivered != 1 {
		t.Fatalf("first delivery count: %d, want 1", first.delivered)
	}
	second := recv(t, deliveries, delay+8*time.Second, "the redelivery")
	if second.delivered != 2 {
		t.Fatalf("second delivery count: %d, want 2", second.delivered)
	}
	if elapsed := second.at.Sub(first.at); elapsed < delay {
		t.Fatalf("redelivered after %s, want no sooner than %s", elapsed, delay)
	}
}

// 1190/AC3: given a consumed message, when the handler terminates it, then it is never delivered again.
// A short AckWait (well under the 4s this then waits past) is what would otherwise make this pass
// whether or not Term did anything at all: without it, the message simply sits unredelivered until the
// controller's ordinary 30s AckWait, and the test would see the same "nothing arrived" result either way.
func TestConsumeTermNeverDeliversTheMessageAgain(t *testing.T) {
	url, account := grantedSendProcess(t)
	const ackWait = time.Second
	makeQueue(t, url, account, ackWait)
	publishApproved(t, url, `{"send":"s3"}`)

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	deliveries := make(chan *Message, 4)
	go func() {
		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
			deliveries <- m
			_ = m.Term()
		})
	}()

	first := recv(t, deliveries, 8*time.Second, "the message")
	if first.Delivered != 1 {
		t.Fatalf("first delivery count: %d, want 1", first.Delivered)
	}

	// Term removes it from the work-queue stream and leaves the consumer with nothing pending — not just
	// "has not come back yet", which a message merely sitting past an un-acted AckWait would also show.
	deadline := time.Now().Add(5 * time.Second)
	for {
		if n := streamMsgs(t, url); n == 0 {
			break
		} else if time.Now().After(deadline) {
			t.Fatalf("the stream still holds %d messages after Term", n)
		}
		time.Sleep(100 * time.Millisecond)
	}
	_, adminJetStream := adminJS(t, url)
	cons, err := adminJetStream.Consumer(ctxFor(t), "MAIL_SENDS", account.QueueConsumerName())
	if err != nil {
		t.Fatal(err)
	}
	info, err := cons.Info(ctxFor(t))
	if err != nil {
		t.Fatal(err)
	}
	if info.NumAckPending != 0 {
		t.Fatalf("the consumer still has %d acks pending after Term", info.NumAckPending)
	}

	select {
	case again := <-deliveries:
		t.Fatalf("a terminated message was delivered again: %+v", again)
	case <-time.After(4 * ackWait):
	}
}