mesh-sdk/go/process/event_hub_integration_test.go

//go:build integration

package process_test

import (
	"context"
	"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")
	}
}

// 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)
}