mesh-controller/cmd/mesh-controller/first_builds_integration_test.go

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"slices"
	"sync"
	"testing"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
	"github.com/novox/mesh-controller/internal/link"
	"github.com/novox/mesh-controller/internal/testbus"
)

// 1146/AC7, whole against a real bus and the store, in both orders of the two events (novox/hq ADR 0432, issue 1146):
// mesh-delivery #7 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's merge,
// heard as the controller hears it (SourceMoved), and mesh-delivery's word, published on the subject the controller
// consumes with the router's record of the ask in its bucket, come in either order; each way the first build of the
// module the merge commit holds is asked exactly once, its outcome registers the module, and it is in the catalogue.
// Without the Approve in the record nothing is asked. Store-backed: skips without one.
func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule(t *testing.T) {
	const head, commit = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", "0177ea7a00000000"
	for _, c := range []struct {
		what        string
		wordFirst   bool
		approved    bool // the router's record holds the operator's Approve of the merged head
		wantRegiter bool
	}{
		{what: "the forge's merge before mesh-delivery's word", approved: true, wantRegiter: true},
		{what: "mesh-delivery's word before the forge's merge", wordFirst: true, approved: true, wantRegiter: true},
		{what: "a word whose record holds no Approve", wordFirst: true},
	} {
		t.Run(c.what, func(t *testing.T) {
			open := aMesh(t)
			ctx := t.Context()
			inv := open.inventory
			// The real ask, guard and queue included: only its answers are kept, so the build's outcome can be matched.
			aForgeAndABuildSeat(t, open)
			// Written by the controller's merge path on its own goroutine, read by the test: under one lock.
			var mu sync.Mutex
			var asks [][3]string
			var ids []string
			asked := func() [][3]string { mu.Lock(); defer mu.Unlock(); return slices.Clone(asks) }
			realAsk := askABuild
			askABuild = func(c context.Context, source buildSource, path, ref string) (string, error) {
				id, err := realAsk(c, source, path, ref)
				if err == nil {
					mu.Lock()
					asks = append(asks, [3]string{source.Repository, path, ref})
					ids = append(ids, id)
					mu.Unlock()
				}
				return id, err
			}
			t.Cleanup(func() { askABuild = realAsk })
			if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "operator-channel", Version: "1",
				DefinesSeats: []catalogue.SeatDeclaration{{Name: broker.AsksSeat, Records: []string{"asks"}}}},
				inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/operator-channel", BuiltFrom: "c0ffee"}); err != nil {
				t.Fatal(err)
			}
			js, err := broker.Dial(testbus.URL(t))
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(js.Close)
			if err := broker.AssertMeshStreams(js); err != nil {
				t.Fatal(err)
			}
			if err := js.EnsureControllerBuckets(); err != nil {
				t.Fatal(err)
			}
			api, err := jetstream.New(js.Conn())
			if err != nil {
				t.Fatal(err)
			}
			records, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.BucketName("operator-channel", "asks")})
			if err != nil {
				t.Fatal(err)
			}
			a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head})
			if c.approved {
				record, _ := json.Marshal(map[string]any{"state": "closed", "ask": a.ask, "warrant": a.warrant})
				if _, err := records.Put(ctx, theDelivery+"."+a.merge.AskID, record); err != nil {
					t.Fatal(err)
				}
			}

			following := following{open: open}
			server := link.ConnectNats(js, nil, noListener{})
			if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims, kept: following.resumeFirstBuilds}); err != nil {
				t.Fatal(err)
			}
			if err := broker.AssertServingConsumers(js); err != nil {
				t.Fatal(err)
			}
			serving, stop := context.WithCancel(context.Background())
			done := make(chan struct{})
			go func() { defer close(done); _ = server.Serve(serving) }()
			t.Cleanup(func() { stop(); <-done })
			servingBus.Store(js)
			t.Cleanup(func() { servingBus.Store(nil) })

			word := func() {
				t.Helper()
				body, _ := json.Marshal(a.merge)
				if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil {
					t.Fatal(err)
				}
				becomes(t, 15*time.Second, "the word being kept", func() bool {
					_, kept, err := keptDeliveryMerge(ctx, "novox", "mesh-delivery", 7, head)
					return err == nil && kept
				})
				time.Sleep(500 * time.Millisecond) // the keeper's hook runs as the word is kept
			}
			forge := func() {
				t.Helper()
				m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 7,
					MergedAt: time.Now().Add(-time.Second).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"},
					ModuleDirs: []string{"."}, ModuleDirsSaid: true}
				if err := following.SourceMoved(ctx, m); err != nil {
					t.Fatal(err)
				}
			}
			if c.wordFirst {
				word()
				forge()
			} else {
				forge()
				if len(asked()) != 0 {
					t.Fatalf("1146/AC7: the merge asked %v before mesh-delivery's word was kept", asked())
				}
				word()
			}
			if !c.wantRegiter {
				if len(asked()) != 0 {
					t.Fatalf("1146/AC7: a merge whose record holds no Approve asked %v", asked())
				}
				return
			}
			want := [3]string{"novox/mesh-delivery", "", "main"}
			becomes(t, 10*time.Second, "the first build asked", func() bool { return len(asked()) >= 1 })
			if got := asked(); len(got) != 1 || got[0] != want {
				t.Fatalf("1146/AC7: the merge asked %v, want %v once", got, want)
			}
			mu.Lock()
			first := ids[0]
			mu.Unlock()
			build := onTrunk(first, "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"})
			build.Commit = commit
			if _, _, err := takeIn(ctx, inv, build); err != nil {
				t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err)
			}
			src, err := inv.SourceOf(ctx, "mesh-delivery-view")
			if err != nil || src.Repository != "novox/mesh-delivery" {
				t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err)
			}
			if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known {
				t.Fatal("1146/AC7: the module is not in the catalogue")
			}
		})
	}
}

func inventoryHolds(t *testing.T, inv *inventory.Inventory, module string) (inventory.Entry, bool) {
	t.Helper()
	entries, err := inv.Catalogued(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	for _, e := range entries {
		if e.Manifest.Module == module {
			return e, true
		}
	}
	return inventory.Entry{}, false
}