//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): AckWaitSeconds 30, named
// <node>_<module>_<process id>.
func makeQueue(t *testing.T, url string, account Account) {
	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: 30 * time.Second,
		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)
	}
}

// 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)
	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()
		})
	}()
	var msg *Message
	select {
	case msg = <-got:
	case <-time.After(8 * time.Second):
		t.Fatal("the message was never consumed")
	}
	<-ctx.Done()
	if err := msg.Ack(); err != nil {
		t.Fatalf("Ack: %v", err)
	}

	_, adminJetStream := adminJS(t, url)
	deadline := time.Now().Add(5 * time.Second)
	for {
		info, err := adminJetStream.Stream(ctxFor(t), "MAIL_SENDS")
		if err != nil {
			t.Fatal(err)
		}
		state, err := info.Info(ctxFor(t))
		if err != nil {
			t.Fatal(err)
		}
		if state.State.Msgs == 0 {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("the stream still holds %d messages after the ack", state.State.Msgs)
		}
		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)
	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()
	deliveries := make(chan struct {
		at        time.Time
		delivered uint64
	}, 2)
	go func() {
		_ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
			deliveries <- struct {
				at        time.Time
				delivered uint64
			}{time.Now(), m.Delivered}
			if m.Delivered == 1 {
				_ = m.NakWithDelay(delay)
			} else {
				_ = m.Ack()
				cancel()
			}
		})
	}()

	first := <-deliveries
	if first.delivered != 1 {
		t.Fatalf("first delivery count: %d, want 1", first.delivered)
	}
	second := <-deliveries
	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.
func TestConsumeTermNeverDeliversTheMessageAgain(t *testing.T) {
	url, account := grantedSendProcess(t)
	makeQueue(t, url, account)
	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 := <-deliveries
	if first.Delivered != 1 {
		t.Fatalf("first delivery count: %d, want 1", first.Delivered)
	}
	select {
	case again := <-deliveries:
		t.Fatalf("a terminated message was delivered again: %+v", again)
	case <-time.After(4 * time.Second):
	}
}