package link

import (
	"strings"
	"testing"
	"time"

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

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

// A work-queue stream's dead letters are kept and shown as a restricted stream's are — aged with the
// stream, never with their body or headers — and can only be dropped, never delivered again (novox/hq ADR
// 0436 §6, changing ADR 0363 §7 for such a stream).

// review round 1, finding 6 (1178/AC3): a work-queue stream really is restricted on the bus — its own
// Stream() says so, and that is what EnsureStream's config sets the metadata from (jetstream.go's
// streamConfigFor) — not merely a header this test set by hand, which would still pass with
// modstreams.go's `out.Restricted = true` removed.
func TestAWorkQueueStreamIsRestrictedOnTheBus(t *testing.T) {
	mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
		MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
	stream := mailSends.Stream()
	if !stream.Restricted {
		t.Fatalf("a work-queue stream's own Stream() is not restricted: %+v", stream)
	}
	cfg := broker.StreamConfigFor(stream)
	if cfg.Metadata[broker.RestrictedMetadata] != "true" {
		t.Fatalf("the server config EnsureStream would send does not carry the restricted metadata: %+v", cfg.Metadata)
	}
}

// mesh-controller.try fault A, d8249dad: a seat's own work queue (SEAT_*, novox/hq ADR 0190, ADR 0264 §4) is
// a JetStream work queue at the bus level too — exactly one worker takes each ask — but it is never a module
// stream the catalogue declared retention: work-queue (ADR 0436 §6), so it carries none of that clause's
// metadata and its dead letters are still delivered again (ADR 0264 §4), unlike mail-sends'.
func TestASeatsWorkQueueIsNotADR0436WorkQueue(t *testing.T) {
	seatQueue := broker.Stream{Name: "SEAT_NODE_BUILD_AGENT", Retention: broker.RetentionWorkQueue, DiscardNew: true}
	cfg := broker.StreamConfigFor(seatQueue)
	if _, has := cfg.Metadata[broker.WorkQueueMetadata]; has {
		t.Fatalf("a seat's own work queue carries ADR 0436 §6's metadata: %+v", cfg.Metadata)
	}
	if _, has := cfg.Metadata[broker.RestrictedMetadata]; has {
		t.Fatalf("a seat's own work queue is restricted: %+v", cfg.Metadata)
	}
	// sourceAge (deadletters.go) reads exactly these two keys off a real stream's config; neither set means
	// KeepDeadLetter marks a seat's dead letter WorkQueue: false, so AgainTo never reaches clause 6's
	// drop-only case for it (proven directly in TestAWorkQueueDeadLetterIsNeverDeliveredAgain) and a seat's
	// ask is still delivered again (ADR 0264 §4), as it was before this release.
}

// 1178/AC3: a dead letter's headers say it came from a work-queue stream, and reading it back carries that
// (and the restricted flag its aging and hiding already key off, deadLetterOf being the one place both are
// read: see DeadLetter.Restricted's doc).
func TestAWorkQueueDeadLettersHeaderIsReadBack(t *testing.T) {
	agesOut := time.Now().Add(time.Hour).UTC()
	h := nats.Header{}
	h.Set(DeadHeaderStream, "MAIL_SENDS")
	h.Set(DeadHeaderConsumer, "novox_mesh-mail_send")
	h.Set(DeadHeaderSubject, "mesh.stream.mesh-mail.event.mail.send.approved")
	h.Set(DeadHeaderRestricted, "true")
	h.Set(DeadHeaderWorkQueue, "true")
	h.Set(DeadHeaderAgesOut, agesOut.Format(time.RFC3339Nano))

	d := deadLetterOf(42, "mesh.stream.mesh-mail.event.mail.send.approved", h, []byte("secret body"), false)
	if !d.WorkQueue {
		t.Fatalf("a work-queue dead letter read as %+v", d)
	}
	if !d.Restricted {
		t.Fatalf("a work-queue dead letter is not read as restricted, so it would be shown with its body: %+v", d)
	}
	if !d.AgesOut.Equal(agesOut) {
		t.Fatalf("ages out at %v, want %v — it should age with its stream", d.AgesOut, agesOut)
	}
}

// 1178/AC3: asking to deliver a work-queue dead letter again is refused, naming ADR 0436 clause 6 — before
// anything is asked of the bus, so it holds whatever max-deliver or age is left.
func TestAWorkQueueDeadLetterIsNeverDeliveredAgain(t *testing.T) {
	d := DeadLetter{ID: 7, Stream: "MAIL_SENDS", Consumer: "novox_mesh-mail_send",
		Subject: "mesh.stream.mesh-mail.event.mail.send.approved", WorkQueue: true, Restricted: true,
		AgesOut: time.Now().Add(time.Hour)}
	if _, err := AgainTo(nil, d); err == nil || !strings.Contains(err.Error(), "ADR 0436 clause 6") ||
		!strings.Contains(err.Error(), "Drop it") {
		t.Fatalf("delivering a work-queue dead letter again: %v; want a refusal naming ADR 0436 clause 6", err)
	}
}