//go:build integration

package main

// mesh-mail send over a real bus (novox/hq ADR 0436 §7, issue 1195 AC2): a JetStream server of the test's
// own, nats-server v2.11.17, copied from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58
// (newTestBus) — the SDK's own testbus is internal, so this module keeps its own copy. Compiled by the
// merge gate's `go vet -tags integration`; never run on this machine (the context file's hard rule).

import (
	"context"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// mailSendApprovedSubject is the real wire subject the controller composes for mesh-mail's own stream
// mail-sends' one event, mail.send.approved: "mesh.stream." + module + ".event." + event (mesh-controller
// internal/broker/modstreams.go ModuleStreamSubject, moduleStreamPrefix). The sender's own Consume binds
// by stream and consumer name alone (mesh-sdk go/process queue.go), so this subject matters only to how a
// test builds the stream and publishes to it — it must match what the controller actually composes, or an
// integration test proves nothing about the real path (novox/hq review round 1, finding 7).
const mailSendApprovedSubject = "mesh.stream." + Module + ".event." + MailSendApproved

// mailSendsStreamConfig is the work-queue stream mail-sends, composed exactly as the controller would
// (mesh-controller internal/catalogue/streams.go StreamDeclaration.Stream and
// internal/broker/modstreams.go ModuleStream.Stream): retention work-queue, discard new, module.json's
// max-age-days 1 and max-mib 64, and its "duplicates": 24 — in hours (novox/hq ADR 0436 clause 6) — which
// the controller turns into DuplicatesSeconds = Duplicates * 60 * 60, 24 hours here.
func mailSendsStreamConfig() jetstream.StreamConfig {
	return jetstream.StreamConfig{
		Name:       MailSendsStream,
		Subjects:   []string{mailSendApprovedSubject},
		Retention:  jetstream.WorkQueuePolicy,
		Discard:    jetstream.DiscardNew,
		MaxAge:     24 * time.Hour,
		MaxBytes:   64 << 20,
		Duplicates: 24 * time.Hour,
	}
}

// mailSendsConsumerConfig is the one durable consumer the controller composes for a process's bus.queues
// (mesh-controller internal/broker/modstreams.go QueueConsumer): AckWaitSeconds 30, MaxDeliver from the
// stream's own max-deliveries (module.json: 5, MostAttempts here).
func mailSendsConsumerConfig(durable string) jetstream.ConsumerConfig {
	return jetstream.ConsumerConfig{Durable: durable, AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: MostAttempts}
}

// newTestBus starts a JetStream-enabled nats-server for the test alone, on a random port with its own temp
// store (modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58).
func newTestBus(t *testing.T) string {
	t.Helper()
	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
		NoLog: true, NoSigs: true}
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatalf("a bus for this test could not be made: %v", err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		s.Shutdown()
		t.Fatal("a bus for this test did not come up within 30s")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	return s.ClientURL()
}

// 1195/AC2: given the mail-sends work-queue stream and its consumer <machine>_mesh-mail_send, when the
// process runs with a handler and a message is published, then the handler receives it; when the process
// gets SIGTERM (here, its context cancelled, the same signal main.go's send() reacts to) it stops without
// acknowledging a message in hand.
func TestSendConsumesMailSendsAndStopsCleanlyOnSignal(t *testing.T) {
	url := newTestBus(t)
	ctx := context.Background()

	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, mailSendsStreamConfig()); err != nil {
		t.Fatal(err)
	}

	account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"}
	if _, err := js.CreateConsumer(ctx, MailSendsStream, mailSendsConsumerConfig(account.QueueConsumerName())); err != nil {
		t.Fatal(err)
	}

	received := make(chan *process.Message, 1)
	runCtx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() { done <- runSend(runCtx, account, func(m *process.Message) { received <- m }) }()

	body := []byte(`{"send":"send-1","account":"mesh","ask":"send-1"}`)
	// Published twice under the same message id (the send id, as the tools' own publish path always uses,
	// novox/hq ADR 0436 clause 8): the stream's own 24-hour duplicate window keeps it one message, proving
	// the dedup this test's stream config actually carries (novox/hq review round 1, finding 7).
	if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil {
		t.Fatal(err)
	}
	if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil {
		t.Fatal(err)
	}

	select {
	case msg := <-received:
		if msg.Subject != mailSendApprovedSubject {
			t.Fatalf("received on %q, want %q", msg.Subject, mailSendApprovedSubject)
		}
		// Left exactly here, never acknowledged: the handler's own fate-deciding (Ack, NakWithDelay, Term) is
		// issues 1196-1198's, not this skeleton's.
	case <-time.After(10 * time.Second):
		t.Fatal("the handler never received the published message")
	}
	select {
	case <-received:
		t.Fatal("the duplicate publish under the same message id was delivered a second time")
	case <-time.After(500 * time.Millisecond):
		// The duplicate was dropped by the stream's own window, as it should be.
	}

	cancel() // the signal send() itself reacts to (SIGTERM, or ^C)
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("runSend returned nil, want ctx.Err() once its context was cancelled")
		}
	case <-time.After(10 * time.Second):
		t.Fatal("runSend did not stop once its context was cancelled")
	}

	// The message was never acknowledged: it is still pending (or ack-pending) on the same durable consumer,
	// read fresh over a new connection.
	nc2, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer nc2.Close()
	js2, err := jetstream.New(nc2)
	if err != nil {
		t.Fatal(err)
	}
	cons, err := js2.Consumer(ctx, MailSendsStream, account.QueueConsumerName())
	if err != nil {
		t.Fatal(err)
	}
	info, err := cons.Info(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if info.NumPending == 0 && info.NumAckPending == 0 {
		t.Fatal("the message was acknowledged despite the process stopping mid-handle")
	}
}