mesh-tools/node-tools/internal/firstview/firstview_integration_test.go

//go:build integration

package firstview

import (
	"context"
	"encoding/json"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/views"
	"github.com/nats-io/nats.go"

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

// issue 1966 TLS-5 [i] (ADR 0615 §2, §4, §6): against the real bus, with the owners down and then up, the runner's own
// consumers on the owners' own views streams carry the manifest's four `consumes`: the first views are asked until answered,
// the unanswered ones are findings that the answer clears, and each owner's next event reaches its picture.
func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThenUp(t *testing.T) {
	mesh := mt.New(t)
	url := mt.URL(t)

	// The owners publish their views on their own streams (views.Emit: `mesh.stream.<owner>.event.views.<state>`, on
	// assigner-views and indexer-views), never on EVENTS. The consumers on them are the mesh's, one filter per family the
	// manifest's `consumes` names, and the runner's membership names them as the mesh issues it (ADR 0363 §8).
	raw, err := os.ReadFile("../../module.json")
	if err != nil {
		t.Fatal(err)
	}
	var manifest struct {
		Consumes []string `json:"consumes"`
	}
	if err := json.Unmarshal(raw, &manifest); err != nil {
		t.Fatal(err)
	}
	streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views"), Indexer: catalogue.StreamBusName("indexer-views")}
	filters := map[string][]string{}
	families := 0
	for _, c := range manifest.Consumes {
		emitter, event, isView := strings.Cut(c, ".views.")
		if !isView { // the seat mesh-catalog's events: the seat list's, on EVENTS (cmd/node-tools/seatlist.go)
			continue
		}
		stream := streamOf[emitter]
		if stream == "" {
			t.Fatalf("the manifest consumes %s, whose views stream the test does not know", c)
		}
		filters[stream] = append(filters[stream], process.ModuleStreamSubject(emitter, views.EventPrefix+event))
		families++
	}
	if families != len(Specs) {
		t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs))
	}
	runner := mt.MembershipOf("node-tools", "anchor", false, nil)
	for _, owner := range []string{Assigner, Indexer} {
		stream := streamOf[owner]
		mesh.ModuleStream(t, owner, stream)
		mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second)
		runner.Hears = append(runner.Hears, bus.HeardIssued{Stream: stream, Consumer: "anchor_node-tools"})
	}
	mesh.Consumer(t, "anchor", "node-tools", []string{"mesh.mod.mesh-knowledge.event.reviewed"}, 2*time.Second) // its `waits`, on EVENTS
	mesh.Issue(t, runner)

	conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	conn.Logf = func(string, ...any) {}
	t.Cleanup(conn.Close)
	conn.Follow("node-tools")
	mt.Until(t, func() error {
		if conn.Membership("node-tools") == nil {
			return errFindings(-1)
		}
		return nil
	})

	var mu sync.Mutex
	findings := map[string]bool{}
	deps := DepsOf(conn, t.Logf)
	deps.Bound, deps.Wait = 100*time.Millisecond, 20*time.Millisecond
	deps.Findings = writes{&mu, findings}
	deps.StartFindings = nil
	v, stop := StartWith(context.Background(), deps, func(string, ...any) {})
	t.Cleanup(stop)

	// The owners are down: no tool answers, nothing is claimed, and past the bound each view is a finding.
	if _, err := v.State(Assigner + ".assignments"); err == nil {
		t.Fatal("a state its owner has not answered is claimed")
	}
	mt.Until(t, func() error {
		mu.Lock()
		defer mu.Unlock()
		if len(findings) != len(Specs) {
			return errFindings(len(findings))
		}
		return nil
	})

	// The owners come up and answer their first views.
	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	for _, s := range Specs {
		s := s
		if _, err := nc.Subscribe("mesh.mod."+s.Owner+".tool."+s.Tool, func(m *nats.Msg) {
			answer, _ := json.Marshal(views.Answer{V: 1, Owner: s.Owner, State: s.State, Revision: 4, Created: time.Unix(1000, 0).UTC(),
				Values: map[string]json.RawMessage{"first": json.RawMessage(`{"n":1}`)}})
			body, _ := json.Marshal(map[string]json.RawMessage{"result": answer})
			_ = m.Respond(body)
		}); err != nil {
			t.Fatal(err)
		}
	}
	_ = nc.Flush()
	for _, s := range Specs {
		name := s.Name()
		mt.Until(t, func() error { _, err := v.Entry(name, "first"); return err })
	}
	mt.Until(t, func() error {
		mu.Lock()
		defer mu.Unlock()
		if len(findings) != 0 {
			return errFindings(len(findings))
		}
		return nil
	})

	// Each owner's next event arrives on the runner's consumer: all four families are bound.
	for _, s := range Specs {
		ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(),
			Value: json.RawMessage(`{"n":2}`)})
		mesh.Emit(t, process.ModuleStreamSubject(s.Owner, views.EventPrefix+s.State), json.RawMessage(ev))
	}
	for _, s := range Specs {
		name := s.Name()
		mt.Until(t, func() error {
			got, err := v.Entry(name, "next")
			if err != nil || string(got) != `{"n":2}` {
				return errFindings(-1)
			}
			return nil
		})
	}
}

type errFindings int

func (e errFindings) Error() string { return "not yet" }

// writes is the runner's finding writer in this test: the findings written and not cleared.
type writes struct {
	mu *sync.Mutex
	by map[string]bool
}

func (w writes) Write(_ context.Context, f condseat.Finding) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.by[views.FindingKey(f.Scope, f.ID, f.Kind)] = true
	return nil
}

func (w writes) Clear(_ context.Context, scope, id, kind, _ string) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	delete(w.by, views.FindingKey(scope, id, kind))
	return nil
}