mesh-sdk/go/process/event_hub_integration_test.go

//go:build integration

package process_test

import (
	"context"
	"strings"
	"testing"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"git.novox.be/novox/mesh-sdk/go/testbus"
)

// A module whose consumer the bus makes after the module started is not lost for that order: the hub tries again after
// each wait and calls live once it is bound (the checker's "until it starts again" hole).
func subTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
		t.Fatal(err)
	}
	cred := process.Account{Node: "n1", Module: "late"}
	hub := process.NewEventHub(t.Logf)
	waits, live := make(chan chan bool), make(chan struct{}, 1)
	wait := func(ctx context.Context) bool {
		step := make(chan bool)
		select {
		case waits <- step:
		case <-ctx.Done():
			return false
		}
		return <-step
	}
	if err := hub.StartUntilBound(ctx, js, cred, wait, func() { live <- struct{}{} }); err == nil {
		t.Fatal("no consumer was there, and the start said nothing")
	}
	select {
	case <-live:
		t.Fatal("live before the consumer was there")
	default:
	}
	step := <-waits
	step <- true // tried again: still no consumer
	step = <-waits
	if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: cred.EventConsumerName(), AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
		t.Fatal(err)
	}
	step <- true
	select {
	case <-live:
	case <-ctx.Done():
		t.Fatal("the consumer was made and the hub never said it is live")
	}
	hub.Stop()
}

// W1.12: a module hears a seat's event kept last per subject on its own stream through its durable, made by the maker
// after the events were said, and is handed the last of each subject; without the consumer the call says so.
func subAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	kept, subjects := seatdef.KeptStream("node-engine", seatdef.Message{Name: "running"}), []string{"mesh.seat.node-engine.event.running.>"}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil {
		t.Fatal(err)
	}
	for _, m := range []struct{ subject, fp string }{
		{"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp1"},
		{"mesh.seat.node-engine.event.running.ace.mesh-tools", "fp2"},
		{"mesh.seat.node-engine.event.running.g14.mesh-tools", "fp3"},
	} {
		if _, err := js.Publish(ctx, m.subject, []byte(m.fp)); err != nil {
			t.Fatal(err)
		}
	}
	cred := process.Account{Node: "ace", Module: "mesh-grants"}
	got := make(chan string, 4)
	hear := func(m jetstream.Msg) { got <- m.Subject() + "=" + string(m.Data()); _ = m.Ack() }
	if _, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf); err == nil {
		t.Fatal("no consumer was made, and hearing said nothing")
	}
	// The durable has gone past every message (it starts from now), so only the start read, an ordered consumer on the stream
	// kept last per subject, gives the last of each subject (design 47 G §641, A-19).
	durable := cred.SeatEventConsumerConfig(subjects, true)
	durable.DeliverPolicy = jetstream.DeliverNewPolicy
	if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil {
		t.Fatal(err)
	}
	var reads []string
	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) }))
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Stop()
	heard := map[string]bool{}
	for len(heard) < 2 {
		select {
		case h := <-got:
			heard[h] = true
		case <-ctx.Done():
			t.Fatalf("heard %v", heard)
		}
	}
	if !heard["mesh.seat.node-engine.event.running.ace.mesh-tools=fp2"] || !heard["mesh.seat.node-engine.event.running.g14.mesh-tools=fp3"] {
		t.Errorf("heard %v, want the last of each subject", heard)
	}
	if len(reads) != 2 || reads[0] != process.ReadWaiting || reads[1] != process.ReadDone {
		t.Errorf("the read said %v, want waiting then read", reads)
	}
	// then the durable is followed
	if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.running.ace.mesh-tools", []byte("fp4")); err != nil {
		t.Fatal(err)
	}
	select {
	case h := <-got:
		if h != "mesh.seat.node-engine.event.running.ace.mesh-tools=fp4" {
			t.Errorf("the durable heard %s", h)
		}
	case <-ctx.Done():
		t.Fatal("the durable was not followed after the read")
	}
}

// A-181: a module hears another module's own stream kept last per subject (a manifest's `streams` entry with `last-per-subject`) as
// it hears a seat's: the last message of each subject read whole at its start, though its durable began after them, then the durable.
func subAModuleHearsAnotherModulesStreamKeptLastPerSubject(t *testing.T) {
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	stream, subject := "BUILDER_BUILDS", process.ModuleStreamSubject("builder", "build.ended.>")
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
		t.Fatal(err)
	}
	for _, m := range []struct{ event, v string }{{"build.ended.web.fp1", "old"}, {"build.ended.web.fp1", "new"}, {"build.ended.web.fp2", "two"}} {
		if _, err := js.Publish(ctx, process.ModuleStreamSubject("builder", m.event), []byte(m.v)); err != nil {
			t.Fatal(err)
		}
	}
	cred := process.Account{Node: "ace", Module: "assigner"}
	durable := cred.SeatEventConsumerConfig([]string{subject}, true)
	durable.DeliverPolicy = jetstream.DeliverNewPolicy
	if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil {
		t.Fatal(err)
	}
	got := make(chan string, 4)
	var reads []string
	cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() }, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) }))
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Stop()
	heard := map[string]bool{}
	for len(heard) < 2 {
		select {
		case h := <-got:
			heard[h] = true
		case <-ctx.Done():
			t.Fatalf("heard %v", heard)
		}
	}
	if !heard["new"] || !heard["two"] || heard["old"] || len(reads) != 2 {
		t.Errorf("heard %v with reads %v, want the last of each subject, read whole", heard, reads)
	}
}

// A-233: a module hears a stream kept by age with FromFirst: at every start the stream is read whole from its first message, though
// its durable began after them, then the durable is followed; without FromFirst only the durable is.
func subAModuleHearsAStreamKeptByAgeFromItsFirstMessage(t *testing.T) {
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	kept, subjects := seatdef.KeptStream("mesh-catalog", seatdef.Message{Name: "version-released"}), []string{"mesh.seat.mesh-catalog.event.version-released"}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxAge: 90 * 24 * time.Hour}); err != nil {
		t.Fatal(err)
	}
	for _, v := range []string{"web 1.0.0", "web 1.1.0", "api 2.0.0"} {
		if _, err := js.Publish(ctx, subjects[0], []byte(v)); err != nil {
			t.Fatal(err)
		}
	}
	cred := process.Account{Node: "ace", Module: "assigner"}
	durable := cred.SeatEventConsumerConfig(subjects, false)
	durable.DeliverPolicy = jetstream.DeliverNewPolicy
	if _, err := js.CreateOrUpdateConsumer(ctx, kept, durable); err != nil {
		t.Fatal(err)
	}
	got := make(chan string, 8)
	hear := func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() }
	var reads []string
	cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf, process.FromFirst(), process.OnRead(func(s string) { reads = append(reads, s) }))
	if err != nil {
		t.Fatal(err)
	}
	var heard []string
	for len(heard) < 3 {
		select {
		case h := <-got:
			heard = append(heard, h)
		case <-ctx.Done():
			t.Fatalf("heard %v, want every message from the first", heard)
		}
	}
	if strings.Join(heard, ",") != "web 1.0.0,web 1.1.0,api 2.0.0" || len(reads) != 2 {
		t.Errorf("heard %v with reads %v, want every message in order, read whole", heard, reads)
	}
	if _, err := js.Publish(ctx, subjects[0], []byte("web 1.2.0")); err != nil {
		t.Fatal(err)
	}
	select {
	case h := <-got:
		if h != "web 1.2.0" {
			t.Errorf("the durable heard %s", h)
		}
	case <-ctx.Done():
		t.Fatal("the durable was not followed after the read")
	}
	cc.Stop()
	// Without FromFirst, a stream kept by age is not read: the durable alone, past every message, hands the next one first.
	cc, err = cred.HearSeatEvents(ctx, js, kept, hear, t.Logf)
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Stop()
	if _, err := js.Publish(ctx, subjects[0], []byte("next")); err != nil {
		t.Fatal(err)
	}
	select {
	case h := <-got:
		if h != "next" {
			t.Errorf("without FromFirst %s was handed before the next message", h)
		}
	case <-ctx.Done():
		t.Fatal("the durable was not followed")
	}
}

// TestTheHubBindsALateConsumerAndSaysItIsLive runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
	t.Run("TheHubBindsALateConsumerAndSaysItIsLive", subTheHubBindsALateConsumerAndSaysItIsLive)
	t.Run("AModuleHearsASeatEventKeptOnItsOwnStream", subAModuleHearsASeatEventKeptOnItsOwnStream)
	t.Run("AModuleHearsAnotherModulesStreamKeptLastPerSubject", subAModuleHearsAnotherModulesStreamKeptLastPerSubject)
	t.Run("AModuleHearsAStreamKeptByAgeFromItsFirstMessage", subAModuleHearsAStreamKeptByAgeFromItsFirstMessage)
}