mesh-catalog/modules/distribution/cmd/distribution/whole_integration_test.go

//go:build integration

package main

// distribution's program whole against a real bus (CAT-D7; design 51 *Acceptance tests*, the holder's cases): its
// run() as the node starts it, its broker account on a nats-server of this test alone, the registry container a fake
// OCI store on the loopback, the status view's node_haves and the node-engine seat's manifest-reads answered on their
// call subjects. An away node holds the old name open; its return (manifest-reads-changed) closes it, said once on
// distribution.events, after which a direct write is refused naming publish; reads stay open; a waiter on a seat's
// `published` hears none of the windows' events; and the build modules counted are the modules claiming a `node-build-*` seat
// in the catalog half the seat mesh-catalog answers (design 47 G §610, A-156), not a list of names.

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	neturl "net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

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

	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-sdk/go/testbus"

	"distribution/internal/holder"
)

func TestDistributionWhole(t *testing.T) {
	url := testbus.URL(t)
	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, _ := jetstream.New(nc)
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()
	for _, b := range []string{"distribution_windows", "distribution_findings"} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "DISTRIBUTION_EVENTS", Subjects: []string{"mesh.stream.distribution.event.>"}, Duplicates: time.Hour}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "novox_distribution", AckPolicy: jetstream.AckExplicitPolicy,
		FilterSubjects: []string{"mesh.mod.mesh-delivery-view.event.delivered", "mesh.seat.node-engine.event.manifest-reads-changed.*",
			"mesh.seat.mesh-catalog.event.catalog-revision.>", "mesh.seat.mesh-catalog.event.registration-changed", "mesh.seat.mesh-delivery.event.source-moved"}}); err != nil {
		t.Fatal(err)
	}

	// The status view and the node-engines: g14 away (its node-engine answers nothing) and running the old build.
	var mu sync.Mutex
	g14 := false
	answer := func(m *nats.Msg, result any) {
		b, _ := json.Marshal(map[string]any{"result": result, "node": "novox"})
		_ = m.Respond(b)
	}
	if _, err := nc.Subscribe("mesh.mod.mesh-delivery-view.>", func(m *nats.Msg) {
		mu.Lock()
		defer mu.Unlock()
		fp := "fp-old"
		if g14 {
			fp = "fp-new"
		}
		answer(m, map[string]any{"keys": map[string]any{
			"ace.mesh-build-go": map[string]any{"fp": "fp-new"}, "g14.mesh-build-go": map[string]any{"fp": fp},
			"ace.gitea": map[string]any{"fp": "x"}, "g14.gitea": map[string]any{"fp": "x"},
			// A module claiming no build seat in the catalog half: it holds no window.
			"g14.mesh-build-node": map[string]any{"fp": "fp-node-old"},
		}})
	}); err != nil {
		t.Fatal(err)
	}
	// The catalog half, as the seats mesh-catalog and mesh-delivery answer it: one watched branch, main, whose trunk revision
	// has mesh-build-go claiming the build seat node-build-go; mesh-build-node claims no build seat, so it holds no window.
	entryOf := func(module string, claims ...string) catalog.CatalogEntry {
		var cs []any
		for _, c := range claims {
			cs = append(cs, map[string]any{"name": c, "scope": "node"})
		}
		return catalog.CatalogEntry{Module: module, Manifest: map[string]any{"claims": cs}}
	}
	catalogHalf := func(m *nats.Msg) {
		switch {
		case strings.Contains(m.Subject, ".call.registrations."):
			answer(m, catalog.RegistrationsAnswer{Registrations: []catalog.Registration{{Module: "mesh-build-go", Branches: []string{"main"}}}})
		case strings.Contains(m.Subject, ".call.revision."):
			answer(m, catalog.RevisionAnswer{Revision: &catalog.CatalogRevision{Revision: 1, Kind: meshcatalogclient.KindTrunk, Branch: "main"},
				Entries: []catalog.CatalogEntry{entryOf("mesh-build-go", "node-build-go"), entryOf("mesh-build-node"), entryOf("gitea", "git")}})
		case strings.Contains(m.Subject, ".call.source."):
			answer(m, map[string]any{"branches": []string{"main"}})
		}
	}
	for _, subject := range []string{"mesh.seat.mesh-catalog.>", "mesh.seat.mesh-delivery.call.>"} {
		if _, err := nc.Subscribe(subject, catalogHalf); err != nil {
			t.Fatal(err)
		}
	}
	engine := func(node string) {
		for _, subj := range []string{"mesh.seat.node-engine.call.manifest-reads." + node + ".>", "mesh.seat.node-engine.tool.manifest-reads." + node} {
			if _, err := nc.Subscribe(subj, func(m *nats.Msg) {
				answer(m, map[string]any{"v": 1, "manifest-reads": []string{holder.NewNamesCapability}})
			}); err != nil {
				t.Fatal(err)
			}
		}
		_ = nc.Flush()
	}
	engine("ace") // g14 is away: nothing answers its node-engine, so it states nothing
	// A waiter on a seat's published: it must hear none of the windows' events.
	var woke []string
	if _, err := nc.Subscribe("mesh.seat.container-image-registry.event.published", func(m *nats.Msg) { woke = append(woke, m.Subject) }); err != nil {
		t.Fatal(err)
	}
	pub, _ := json.Marshal(holder.Publishing{Name: holder.PublishingKey, Modules: map[string][]string{"mesh-build-go": {"fp-new"}}})
	windows, _ := js.KeyValue(ctx, "distribution_windows")
	if _, err := windows.Put(ctx, holder.PublishingKey, pub); err != nil {
		t.Fatal(err)
	}

	// The registry container on the loopback, and the program's port.
	store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if strings.HasSuffix(r.URL.Path, "/tags/list") {
			_, _ = io.WriteString(w, `{"tags":[]}`)
			return
		}
		if r.Method == http.MethodHead {
			w.WriteHeader(http.StatusNotFound) // no tag published through the seat
			return
		}
		w.WriteHeader(http.StatusCreated)
	}))
	defer store.Close()
	l, _ := net.Listen("tcp", "127.0.0.1:0")
	listen := l.Addr().String()
	l.Close()
	dir := t.TempDir()
	acct, _ := json.Marshal(map[string]any{"url": url, "node": "novox", "module": "distribution", "user": "novox.distribution", "password": "unused"})
	_ = os.WriteFile(filepath.Join(dir, "broker"), acct, 0o600)
	env := map[string]string{"MESH_BROKER_FILE": filepath.Join(dir, "broker"), "MESH_REGISTRY_LOOPBACK": store.URL, "MESH_REGISTRY_LISTEN": listen}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }) }()
	defer func() { stop(); <-done }()

	window := func(name string) holder.Window {
		var w holder.Window
		if e, err := windows.Get(ctx, name); err == nil {
			_ = json.Unmarshal(e.Value(), &w)
		}
		return w
	}
	eventually := func(what string, ok func() bool) {
		t.Helper()
		for end := time.Now().Add(20 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
			if ok() {
				return
			}
			select {
			case err := <-done:
				t.Fatalf("the program ended: %v", err)
			default:
			}
		}
		t.Fatalf("not within 20s: %s (old-name %+v, direct-writes %+v)", what, window(holder.OldNameWindow), window(holder.DirectWritesWindow))
	}
	write := func() int {
		req, _ := http.NewRequest(http.MethodPut, "http://"+listen+"/v2/m/i/manifests/0.1.9", nil)
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			return 0
		}
		resp.Body.Close()
		return resp.StatusCode
	}
	eventually("the start's wake: g14 holds both windows", func() bool {
		o, d := window(holder.OldNameWindow), window(holder.DirectWritesWindow)
		return o.Open && fmt.Sprint(o.Holding) == "[g14]" && d.Open && fmt.Sprint(d.Holding) == "[g14]"
	})
	if got := write(); got != http.StatusCreated {
		t.Errorf("a direct write while the window is open = %d, want it through", got)
	}

	// g14 returns with the new node-engine and the new build: its manifest-reads-changed closes both windows.
	mu.Lock()
	g14 = true
	mu.Unlock()
	engine("g14")
	if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.manifest-reads-changed.g14", []byte(`{"v":1,"node":"g14","why":"returned"}`)); err != nil {
		t.Fatal(err)
	}
	eventually("g14's return closes both windows", func() bool {
		return !window(holder.OldNameWindow).Closed.IsZero() && !window(holder.DirectWritesWindow).Closed.IsZero()
	})
	// A delivered after the close says nothing again.
	if _, err := js.Publish(ctx, "mesh.mod.mesh-delivery-view.event.delivered", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	eventually("a direct write refused naming publish once closed", func() bool { return write() == http.StatusForbidden })
	resp, err := http.Get("http://" + listen + "/v2/m/i/tags/list")
	if err != nil || resp.StatusCode != http.StatusOK {
		t.Errorf("a read once the window closed: %v %v", resp, err)
	}
	if _, err := holder.Health(&holder.OCI{Base: mustURL("http://" + listen)}, nil); err != nil {
		t.Errorf("the health check through the program's port: %v", err)
	}
	s, _ := js.Stream(ctx, "DISTRIBUTION_EVENTS")
	eventually("the program heard every waking event, the delivered after the close included", func() bool {
		c, err := js.Consumer(ctx, "EVENTS", "novox_distribution")
		if err != nil {
			return false
		}
		i, err := c.Info(ctx)
		return err == nil && i.NumPending == 0 && i.NumAckPending == 0 && i.Delivered.Consumer >= 2
	})
	// The stream holds the module's own events once each; the finding writer's (design 47 G A-30) are said beside them.
	info, _ := s.Info(ctx, jetstream.WithSubjectFilter("mesh.stream.distribution.event.>"))
	for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} {
		if n := info.State.Subjects["mesh.stream.distribution.event."+ev]; n != 1 {
			t.Errorf("distribution.events holds %s %d times, want once", ev, n)
		}
	}
	for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} {
		if _, err := s.GetLastMsgForSubject(ctx, "mesh.stream.distribution.event."+ev); err != nil {
			t.Errorf("%s not said: %v", ev, err)
		}
	}
	if len(woke) != 0 {
		t.Errorf("a waiter on published woke on %v", woke)
	}
}

func mustURL(s string) *neturl.URL {
	u, err := neturl.Parse(s)
	if err != nil {
		panic(err)
	}
	return u
}