mesh-controller/internal/link/receive_nats_test.go

package link

import (
	"context"
	"io"
	"log"
	"sync"
	"testing"
	"time"

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

// The consume side against a real server, because what is being checked is what the server does.
//
// Reasoning cannot answer any of these: whether a nak-with-delay really comes back, whether the
// delay is honoured, whether terminating a delivery really stops it, or whether an answer published
// to an address carried in the payload reaches a caller waiting on its own inbox. Each is a claim
// about a server, so each is asked of one:
//
//	go test ./internal/link/ -run TestNats   (each test on a bus of its own: internal/testbus)

// serving1 is a controller reading from a real bus, and a way to stop it.
func servingOn(t *testing.T, js *broker.JetStream, l Listener) (*Server, func()) {
	t.Helper()
	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()},
		listener: l, log: quiet()}
	ctx, stop := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { defer close(done); _ = s.Serve(ctx) }()
	return s, func() {
		stop()
		<-done
	}
}

// counted records reports and can be told to refuse them, from another goroutine.
type counted struct {
	mu    sync.Mutex
	err   error
	heard []Report
}

func (c *counted) Heard(_ context.Context, r Report) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.err != nil {
		return false, c.err
	}
	c.heard = append(c.heard, r)
	// News, so what the mesh states about a report is exercised wherever a report is.
	return true, nil
}

func (c *counted) refusing(err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.err = err
}

func (c *counted) count() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return len(c.heard)
}

func eventually(t *testing.T, what string, is func() bool) {
	t.Helper()
	deadline := time.Now().Add(8 * time.Second)
	for time.Now().Before(deadline) {
		if is() {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("%s did not happen within the wait", what)
}

// A report published by a node reaches the controller, is recorded, and is acknowledged — so the
// stream does not hold it. A work queue is the check: what is acknowledged leaves it.
func quiet() *log.Logger { return log.New(io.Discard, "", 0) }

// **The store window, in the server.** A report the store cannot take is naked with a delay and
// comes back; once the store is there it is recorded and leaves the stream. The controller holds
// nothing in the meantime — which is what the sequence check below is for: the message is still on
// the server while it waits.

// A report about a declaration the mesh has moved past is settled without being acted on, and
// leaves the stream rather than coming back for ever.

// sentAndHeardSafely is sentAndHeard, read from two goroutines.
type sentAndHeardSafely struct {
	mu    sync.Mutex
	sent  string
	heard []Report
}

func (s *sentAndHeardSafely) Heard(_ context.Context, r Report) (bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.heard = append(s.heard, r)
	return true, nil
}

func (s *sentAndHeardSafely) Outstanding(context.Context, string) (string, error) {
	return s.sent, nil
}

func (s *sentAndHeardSafely) count() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.heard)
}

// **An enrolment answered through a reply address the stream would have eaten.**
//
// The caller waits on its own inbox and states that address in the request's payload. The check is
// that the answer arrives there — which is the whole reason the address is a field rather than the
// transport's reply, and this is the test design 25 §2 asks for so the reason cannot quietly become
// folklore.

// enrolsAs answers every request the same way.
type enrolsAs struct{ reply EnrolReply }

func (e enrolsAs) Enrol(context.Context, EnrolRequest) (EnrolReply, error) { return e.reply, nil }

// The two events the controller follows arrive over one durable consumer with two filters, and it
// can acknowledge them.
//
// **Both halves are the point.** A consumer with several filter subjects is a 2.10 feature and this
// is the first thing in the mesh to use one; and a delivery from the events stream is acknowledged
// on a different ack subject from a delivery from the control stream, which the controller's own
// permission list has to cover or every announcement is redelivered for ever.

type toldAbout struct {
	mu   sync.Mutex
	saw  []Upgraded
	fail error
}

func (u *toldAbout) Upgraded(_ context.Context, m Upgraded) error {
	u.mu.Lock()
	defer u.mu.Unlock()
	if u.fail != nil {
		return u.fail
	}
	u.saw = append(u.saw, m)
	return nil
}

func (u *toldAbout) count() int {
	u.mu.Lock()
	defer u.mu.Unlock()
	return len(u.saw)
}

// A store that never comes back: the report is let go once the bound passes, and it leaves the
// stream rather than being held for ever. The bound is the controller's, not the server's — nothing
// here sets max-deliver, and that is deliberate (streams.go).

// A merge announcement is not what these tests are about; taken and forgotten.
func (t *toldAbout) SourceMoved(context.Context, SourceMoved) error { return nil }

// Every subject the controller follows decodes to a kind, and decoding never reaches past the
// list: the day the list was three long and the decoder named a fourth, every message panicked
// the control plane (2026-09-28).
func TestEverySubjectTheControllerFollowsDecodesToAKind(t *testing.T) {
	for _, subject := range broker.ControllerFollows {
		if _, ok := kindOfSubject(subject); !ok {
			t.Errorf("%s is followed and decodes to nothing", subject)
		}
	}
	if _, ok := kindOfSubject("mesh.mod.nobody.event.nothing"); ok {
		t.Error("a subject nobody follows decoded to a kind")
	}
}

// **A handler slower than the acknowledgement window is not handed its message again.**
//
// The bus waits a fixed time to be told a message was taken and then redelivers, which is right for
// a consumer that died and wrong for one that is busy. Acting on a merge builds modules — minutes
// against a thirty-second window — and the same merge was handed over five times while the first
// build was still running (2026-09-28). Here the window is a second and the work takes three.

// slowly is a listener that runs whatever it was given.
type slowly struct{ work func() }

func (s slowly) Heard(context.Context, Report) (bool, error) { s.work(); return true, nil }

// **The mesh says what it applied** (novox/hq ADR 0134), under the seat the control plane holds — and
// says nothing when a report is the same state said again, which is what a machine reconciling every
// minute sends.

// And a report that is not news is not a fact. A machine reconciles every minute; a fact per report
// would be a fact per minute per machine, which is a stream nobody reads.

// sameAgain records a report and says it was the same state said again.
type sameAgain struct{}

func (sameAgain) Heard(context.Context, Report) (bool, error) { return false, nil }