mesh-tools/node-tools/internal/bus/streams_test.go

package bus_test

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

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// streamMesh is a mesh with the mail module's own stream MAIL, and a connection on anchor following the
// memberships of the modules named.
func streamMesh(t *testing.T, memberships ...bus.Membership) (*mt.Mesh, *bus.Conn) {
	t.Helper()
	mesh := mt.New(t)
	mesh.ModuleStream(t, "mail", "MAIL")
	for _, m := range memberships {
		mesh.Issue(t, m)
	}
	conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	conn.Logf = func(string, ...any) {}
	t.Cleanup(conn.Close)
	for _, m := range memberships {
		conn.Follow(m.Module)
	}
	mt.Until(t, func() error {
		for _, m := range memberships {
			if conn.Membership(m.Module) == nil {
				return errors.New(m.Module + " not issued yet")
			}
		}
		return nil
	})
	return mesh, conn
}

// novox/hq ADR 0363 §2, §8 (issue 725): an event one of the module's streams carries is published on that
// stream's subject and kept by that stream alone; any other event of it goes to EVENTS as before.
func TestAnEventADeclaredStreamCarriesIsPublishedOnItsSubjectAlone(t *testing.T) {
	mail := mt.MembershipOf("mail", "anchor", true, nil)
	mail.Streams = []bus.StreamIssued{{Stream: "MAIL", Events: []string{"mail.*.message.added"},
		Subject: "mesh.stream.mail.event.{event}"}}
	mesh, conn := streamMesh(t, mail)

	landed := mesh.NextEvent(t, "mesh.stream.mail.event.>")
	if err := conn.PublishAs("mail", bus.Envelope{Key: "mail.home.message.added", Body: json.RawMessage(`{"id":1}`)}); err != nil {
		t.Fatal(err)
	}
	select {
	case got := <-landed:
		if got != "mesh.stream.mail.event.mail.home.message.added" {
			t.Errorf("published on %s", got)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the event did not land on its stream's subject")
	}
	if err := conn.PublishAs("mail", bus.Envelope{Key: "status", Body: json.RawMessage(`{}`)}); err != nil {
		t.Fatal(err)
	}
	if n := mesh.Messages(t, "MAIL"); n != 1 {
		t.Errorf("MAIL holds %d messages, want the one event it carries", n)
	}
	if n := mesh.Messages(t, "EVENTS"); n != 1 {
		t.Errorf("EVENTS holds %d messages, want only the event no stream carries", n)
	}
}

// novox/hq ADR 0363 §8 (issue 725): a module hears, beside EVENTS, every module stream its membership names,
// one event at a time, with the key it reads on EVENTS; a membership naming none hears EVENTS alone.
func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
	reader := mt.MembershipOf("reader", "anchor", false, nil)
	mesh, conn := streamMesh(t, reader)
	mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)
	mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.mail.home.>"}, 5*time.Second)

	var mu sync.Mutex
	heard := map[string]int{}
	handling := 0
	stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
		mu.Lock()
		handling++
		if handling > 1 {
			t.Error("two events were handed to the module at once")
		}
		mu.Unlock()
		time.Sleep(50 * time.Millisecond)
		mu.Lock()
		handling--
		heard[env.Key]++
		mu.Unlock()
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(stop)
	count := func(key string) int {
		mu.Lock()
		defer mu.Unlock()
		return heard[key]
	}

	// Its membership names no stream: EVENTS alone, as before.
	mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": 1})
	mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1})
	mt.Until(t, func() error {
		if count("alpha.happened") != 1 {
			return errors.New("the event on EVENTS was not heard")
		}
		return nil
	})
	if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet != 1 || count("mail.mail.home.message.added") != 0 {
		t.Fatalf("a module whose membership names no stream heard MAIL (%d not yet delivered)", notYet)
	}

	// Its membership now names its consumer on MAIL: bound, and what waited there is heard.
	reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
	mesh.Issue(t, reader)
	mt.Until(t, func() error {
		if count("mail.mail.home.message.added") != 1 {
			return errors.New("the event on MAIL was not heard")
		}
		return nil
	})
	for n := 2; n <= 4; n++ {
		mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": n})
		mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": n})
	}
	mt.Until(t, func() error {
		if a, m := count("alpha.happened"), count("mail.mail.home.message.added"); a != 4 || m != 4 {
			return fmt.Errorf("heard %d on EVENTS and %d on MAIL, want 4 and 4", a, m)
		}
		for _, stream := range []string{"EVENTS", "MAIL"} {
			if ack, notYet := mesh.PendingOn(t, stream, "anchor", "reader"); ack != 0 || notYet != 0 {
				return fmt.Errorf("%s still holds %d unacknowledged, %d undelivered", stream, ack, notYet)
			}
		}
		return nil
	})

	// A membership that stops naming it stops what it reads there.
	reader.Hears = nil
	mesh.Issue(t, reader)
	mt.Until(t, func() error {
		if mesh.Consumers(t, "MAIL") != 1 {
			return errors.New("MAIL's consumer is gone")
		}
		mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 0})
		time.Sleep(300 * time.Millisecond)
		if _, notYet := mesh.PendingOn(t, "MAIL", "anchor", "reader"); notYet == 0 {
			return errors.New("MAIL is still read")
		}
		return nil
	})

	// The membership as the grants part keeps it (ADR 0571 clause 7, 10): read first and applied live, and nothing
	// published on the controller's subject is heard.
	granted := reader
	granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
	mesh.Issue(t, granted)
	again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	again.Logf = func(string, ...any) {}
	t.Cleanup(again.Close)
	again.Follow("reader")
	consumerOf := func() string {
		if m := again.Membership("reader"); m != nil && len(m.Hears) == 1 {
			return m.Hears[0].Consumer
		}
		return ""
	}
	if got := consumerOf(); got != "from-grants" {
		t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
	}
	granted.Hears[0].Consumer = "from-grants-2"
	mesh.Issue(t, granted)
	mt.Until(t, func() error {
		if got := consumerOf(); got != "from-grants-2" {
			return fmt.Errorf("the membership names %q, want the grants part's change", got)
		}
		return nil
	})
	// 774/S32/AC1: a copy said on the controller's subject (mesh.assignment) changes nothing; there is no stream to read it from.
	streamed := reader
	streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-subject"}}
	mesh.Publish(t, "mesh.assignment.anchor.reader", streamed)
	again.Flush()
	time.Sleep(200 * time.Millisecond)
	if got := consumerOf(); got != "from-grants-2" {
		t.Fatalf("a membership said on the controller's subject was applied: the membership names %q", got)
	}
	// Review g1 round 1: memberships that cannot be read at the follow (a bus not ready after its boot) are read again until
	// they can be; the follow, which runs once per module, does not give the module up for good.
	was := bus.GrantsRetry
	bus.GrantsRetry = 50 * time.Millisecond
	t.Cleanup(func() { bus.GrantsRetry = was })
	mesh.DropGrants(t)
	late, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	late.Logf = func(string, ...any) {}
	t.Cleanup(late.Close)
	late.Follow("reader")
	if m := late.Membership("reader"); m != nil {
		t.Fatalf("a membership was served from a state not there: %+v", m)
	}
	mesh.Issue(t, granted)
	mt.Until(t, func() error {
		if m := late.Membership("reader"); m == nil || len(m.Hears) != 1 || m.Hears[0].Consumer != "from-grants-2" {
			return fmt.Errorf("the membership is %+v, want the one written once the state is there", m)
		}
		return nil
	})
}

// novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once
// it exists, without the module's events on EVENTS waiting for it.
func TestAConsumerNamedBeforeItExistsIsBoundOnceItDoes(t *testing.T) {
	restore := bus.RebindEvery
	bus.RebindEvery = 200 * time.Millisecond
	t.Cleanup(func() { bus.RebindEvery = restore })
	reader := mt.MembershipOf("reader", "anchor", false, nil)
	reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
	mesh, conn := streamMesh(t, reader)
	mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)

	var mu sync.Mutex
	heard := map[string]int{}
	stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
		mu.Lock()
		heard[env.Key]++
		mu.Unlock()
		return nil
	})
	if err != nil {
		t.Fatalf("a consumer on a module stream not made yet kept the module from hearing EVENTS: %v", err)
	}
	t.Cleanup(stop)
	mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{})
	mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
	mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{})
	mt.Until(t, func() error {
		mu.Lock()
		defer mu.Unlock()
		if heard["alpha.happened"] != 1 || heard["mail.mail.home.message.added"] != 1 {
			return fmt.Errorf("heard %v", heard)
		}
		return nil
	})
}

// novox/hq ADR 0363 §6 (issue 725): a consumer the controller deletes and makes again under the same name, its
// membership unchanged, is bound again, and what is published after is heard — the module does not go deaf.
func TestAConsumerDeletedAndMadeAgainIsHeardAgain(t *testing.T) {
	restore := bus.RebindEvery
	bus.RebindEvery = 200 * time.Millisecond
	t.Cleanup(func() { bus.RebindEvery = restore })
	reader := mt.MembershipOf("reader", "anchor", false, nil)
	reader.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "anchor_reader"}}
	mesh, conn := streamMesh(t, reader)
	mesh.Consumer(t, "anchor", "reader", []string{"mesh.mod.alpha.event.>"}, 5*time.Second)
	mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
	logs := &syncLogs{}
	conn.Logf = logs.Logf

	var mu sync.Mutex
	heard := 0
	stop, err := conn.ConsumeAs("reader", func(env bus.Envelope) error {
		if env.Key == "mail.mail.home.message.added" {
			mu.Lock()
			heard++
			mu.Unlock()
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(stop)
	count := func() int {
		mu.Lock()
		defer mu.Unlock()
		return heard
	}
	mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 1})
	mt.Until(t, func() error {
		if count() != 1 {
			return errors.New("the first event on MAIL was not heard")
		}
		return nil
	})

	mesh.RemoveConsumer(t, "MAIL", "anchor", "reader")
	mt.Until(t, func() error {
		if !logs.Has("stopped hearing MAIL") {
			return errors.New("the end of the loop was not said")
		}
		return nil
	})
	mesh.ConsumerOn(t, "MAIL", "anchor", "reader", []string{"mesh.stream.mail.event.>"}, 5*time.Second)
	mesh.Emit(t, "mesh.stream.mail.event.mail.home.message.added", map[string]any{"n": 2})
	mt.Until(t, func() error {
		if count() != 2 {
			return fmt.Errorf("heard %d events on MAIL after its consumer was made again, want 2", count())
		}
		return nil
	})
}

// syncLogs keeps what the runtime says, from any goroutine.
type syncLogs struct {
	mu    sync.Mutex
	lines []string
}

func (l *syncLogs) Logf(format string, args ...any) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.lines = append(l.lines, fmt.Sprintf(format, args...))
}

func (l *syncLogs) Has(fragment string) bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	for _, line := range l.lines {
		if strings.Contains(line, fragment) {
			return true
		}
	}
	return false
}