mesh-controller/internal/link/deadletters_integration_test.go

//go:build integration

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/novox/mesh-controller/internal/broker"
)

func TestADeadLetterIsDeliveredAgainToItsConsumerAlone(t *testing.T) {
	js := aBus(t)
	keeping(t, js)
	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
	handling := aModuleConsumer(t, js, "media", "radarr", "gitea.pull.merged")

	const subject = "mesh.mod.gitea.event.pull.merged"
	msg := &nats.Msg{Subject: subject, Data: []byte(`{"n":1}`), Header: nats.Header{}}
	msg.Header.Set("x-node", "forge")
	msg.Header.Set(nats.MsgIdHdr, "event-1")
	if _, err := js.Context().PublishMsg(msg); err != nil {
		t.Fatal(err)
	}
	if m := next(t, handling); m == nil {
		t.Fatal("the other module was not handed the event")
	} else {
		_ = m.Ack()
	}
	d := givenUp(t, js, failing)
	if d.Consumer != "media_sonarr" || d.Stream != "EVENTS" || d.Subject != subject || d.Deliveries != 5 ||
		d.GaveUp.IsZero() || d.Published.IsZero() || d.Lost != "" {
		t.Fatalf("kept as %+v", d)
	}
	held, err := HeldDeadLetters(js.Context())
	if err != nil || held["EVENTS.media_sonarr"] != 1 {
		t.Fatalf("held %v (%v)", held, err)
	}

	whole, err := DeadLetterNamed(js.Context(), d.ID)
	if err != nil || whole.Body != `{"n":1}` || len(whole.Headers["x-node"]) != 1 || whole.Headers["x-node"][0] != "forge" {
		t.Fatalf("one asked whole is %+v (%v)", whole, err)
	}

	_, to, err := DeliverAgain(js.Context(), d.ID)
	if err != nil {
		t.Fatal(err)
	}
	if to != "mesh.again.media_sonarr.mod.gitea.event.pull.merged" {
		t.Fatalf("delivered again on %s", to)
	}
	again := next(t, failing)
	if again == nil {
		t.Fatal("the consumer that gave it up was not handed it again")
	}
	if string(again.Data()) != `{"n":1}` || again.Headers().Get("x-node") != "forge" ||
		again.Headers().Get(AgainHeader) == "" {
		t.Fatalf("handed again as %s %v", again.Data(), again.Headers())
	}
	if original, ok := broker.OriginalOfAgain(again.Subject()); !ok || original != subject {
		t.Fatalf("delivered again on %s, which does not say the event's own subject", again.Subject())
	}
	_ = again.Ack()
	if m := next(t, handling); m != nil {
		t.Fatalf("the module that handled it was handed it twice: %s", m.Subject())
	}
	if _, err := DeadLetterNamed(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) {
		t.Fatalf("still kept after it was delivered again: %v", err)
	}
	held, _ = HeldDeadLetters(js.Context())
	if held["EVENTS.media_sonarr"] != 0 {
		t.Fatalf("still held: %v", held)
	}
	// Asked twice, delivered once: the second finds nothing kept.
	if _, _, err := DeliverAgain(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) {
		t.Fatalf("a second delivery answered %v", err)
	}
}

func TestADeadLetterDroppedIsGone(t *testing.T) {
	js := aBus(t)
	keeping(t, js)
	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
	if _, err := js.Context().Publish("mesh.mod.gitea.event.pull.merged", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	d := givenUp(t, js, failing)
	if _, err := DropDeadLetter(js.Context(), d.ID); err != nil {
		t.Fatal(err)
	}
	if left, total, err := DeadLetters(js.Context(), "", 0); err != nil || total != 0 || len(left) != 0 {
		t.Fatalf("after the drop: %v %d %v", left, total, err)
	}
	if m := next(t, failing); m != nil {
		t.Fatalf("a dropped event was handed over: %s", m.Subject())
	}
}

func TestANoticeWhoseMessageIsGoneIsKeptAndSaysSo(t *testing.T) {
	js := aBus(t)
	d, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+
		`"deliveries":5,"timestamp":"2026-10-06T15:51:00Z"}`))
	if err != nil {
		t.Fatal(err)
	}
	if !strings.Contains(d.Lost, "no longer held message 42") || d.Consumer != "media_sonarr" {
		t.Fatalf("kept as %+v", d)
	}
	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "Drop it") {
		t.Fatalf("a record without its message was delivered again: %v", err)
	}
	// The same notice taken twice is kept once.
	if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+
		`"deliveries":5}`)); err != nil {
		t.Fatal(err)
	}
	if _, total, _ := DeadLetters(js.Context(), "media_sonarr", 0); total != 1 {
		t.Fatalf("one notice taken twice is kept %d times", total)
	}
}

func TestADeadLetterIsDeliveredAgainOnlyWhereItIsReceived(t *testing.T) {
	js := aBus(t)
	keeping(t, js)
	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
	if _, err := js.Context().Publish("mesh.mod.gitea.event.pull.merged", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	d := givenUp(t, js, failing)

	// A consumer as it was before this fix: its filters do not take what is delivered again.
	c, _ := broker.ConsumerFor(broker.Principal{Kind: broker.KindModule, Node: "media", Module: "sonarr",
		Consumes: []string{"gitea.pull.merged"}, PasswordHash: "x"})
	c.Filters = c.Filters[:len(c.Filters)-1]
	if err := js.EnsureConsumer(c); err != nil {
		t.Fatal(err)
	}
	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "does not yet take") {
		t.Fatalf("delivered to a consumer that does not filter its again subject: %v", err)
	}
	if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil {
		t.Fatalf("a refused delivery let the dead letter go: %v", err)
	}

	// The consumer gone: nothing would receive it.
	if err := js.Context().DeleteConsumer(broker.EventsStream, c.Name); err != nil {
		t.Fatal(err)
	}
	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "no longer on the bus") {
		t.Fatalf("delivered to a consumer that is gone: %v", err)
	}
	if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil {
		t.Fatalf("a refused delivery let the dead letter go: %v", err)
	}

	// Not an event: refused, whatever stream it is from.
	for _, other := range []DeadLetter{
		{ID: 2, Stream: "SEAT_TELEGRAM_SENDER", Consumer: "SEAT_TELEGRAM_SENDER_worker", Subject: "mesh.seat.telegram-sender.accept.send"},
		{ID: 3, Stream: "KV_x", Consumer: "y", Subject: "$KV.x.k"},
	} {
		if _, err := AgainTo(js.Context(), other); err == nil {
			t.Errorf("%s was given a subject to be delivered again on", other.Stream)
		}
	}
}

func TestTheListSaysTheTotalAndStopsAtItsLimit(t *testing.T) {
	js := aBus(t)
	for seq := 1; seq <= 5; seq++ {
		if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":"media_sonarr",`+
			`"stream_seq":%d,"deliveries":5}`, seq))); err != nil {
			t.Fatal(err)
		}
	}
	list, total, err := DeadLetters(js.Context(), "media_sonarr", 2)
	if err != nil || total != 5 || len(list) != 2 || list[0].Sequence != 5 {
		t.Fatalf("%d of %d (%v): %+v", len(list), total, err, list)
	}
	if _, total, _ := DeadLetters(js.Context(), "another_one", 2); total != 0 {
		t.Fatalf("another consumer's total is %d", total)
	}
}

func TestTheControllerActsOnAnEventDeliveredAgain(t *testing.T) {
	js := aBus(t)
	told := &toldAbout{}
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
	if err := s.Follows(told); err != nil {
		t.Fatal(err)
	}
	if err := s.Answers(replaysWith{}); err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithCancel(context.Background())
	defer stop()
	go func() { _ = s.Serve(ctx) }()
	eventually(t, "the controller's event consumer being made", func() bool {
		_, err := js.Context().ConsumerInfo("EVENTS", broker.ControllerName)
		return err == nil
	})
	moved, _ := json.Marshal(Upgraded{Module: "gitea", Commit: "abcdef0123"})
	if _, err := js.Context().Publish(broker.AgainSubject(broker.ControllerName, broker.ControllerFollows[0]), moved); err != nil {
		t.Fatal(err)
	}
	eventually(t, "the upgrade delivered again reaching the controller", func() bool { return told.count() == 1 })
}

func TestNoticesThatCannotBeTakenAreSaidAndServingGoesOn(t *testing.T) {
	js := aBus(t)
	if err := js.Context().DeleteStream(broker.DeadLetterNoticesStream); err != nil {
		t.Fatal(err)
	}
	stop := keepingGivenUp(context.Background(), js, quiet())
	defer stop()
	eventually(t, "the failure said", func() bool {
		for _, a := range Advisories.Since(time.Now().Add(-time.Minute)) {
			if a.Token == AdvisoryNotKept && a.Stream == broker.DeadLetterNoticesStream {
				return true
			}
		}
		return false
	})
	held := &counted{}
	_, stopServing := servingOn(t, js, held)
	defer stopServing()
	if _, err := js.Context().Publish("mesh.control.anchor.report", []byte(`{"node":"anchor"}`)); err != nil {
		t.Fatal(err)
	}
	eventually(t, "a report heard while the notices cannot be taken", func() bool { return held.count() == 1 })
}

func TestTheNewestHeldDeadLetterIsWhenItWasKept(t *testing.T) {
	js := aBus(t)
	gaveUp := time.Date(2026, 10, 10, 21, 4, 0, 0, time.UTC)
	keep := func(consumer string, seq int, at time.Time) {
		t.Helper()
		if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":%q,`+
			`"stream_seq":%d,"deliveries":5,"timestamp":%q}`, consumer, seq, at.Format(time.RFC3339Nano)))); err != nil {
			t.Fatal(err)
		}
	}
	look := func() map[string]time.Time {
		t.Helper()
		held, err := HeldDeadLetters(js.Context())
		if err != nil {
			t.Fatal(err)
		}
		newest, err := NewestDeadLetters(js.Context(), held)
		if err != nil {
			t.Fatal(err)
		}
		return newest
	}

	// Sonarr gave up at 21:05 and that is kept first; radarr's letter is kept after it.
	keep("media_sonarr", 2, gaveUp.Add(time.Minute))
	keep("media_radarr", 3, gaveUp)
	before := look()
	if len(before) != 2 || before["EVENTS.media_sonarr"].IsZero() ||
		!before["EVENTS.media_radarr"].After(before["EVENTS.media_sonarr"]) {
		t.Fatalf("%v", before)
	}
	// Sonarr's give-up of 21:04, kept late, is newer than anything the condition has seen.
	keep("media_sonarr", 1, gaveUp)
	after := look()
	if !after["EVENTS.media_sonarr"].After(before["EVENTS.media_sonarr"]) ||
		!after["EVENTS.media_sonarr"].After(before["EVENTS.media_radarr"]) {
		t.Fatalf("a letter kept late does not read as newer: before %v, after %v", before, after)
	}
	if !after["EVENTS.media_radarr"].Equal(before["EVENTS.media_radarr"]) {
		t.Fatalf("radarr's newest moved: before %v, after %v", before, after)
	}
}