mesh-delivery/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

	occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/process"

	"git.novox.be/novox/mesh-sdk/go/views/viewstest"

	"mesh-delivery-view/internal/view"
)

// serveOwners serves the six states of the indexer and the assigner the view's process follows by their first views
// (ADR 0615 §1, DLV-M4): each owner's read tool and views family, with the view's durable consumer (`n_mesh-delivery-view`).
// A bucket the test made is kept; the others are made empty.
func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
	t.Helper()
	for _, o := range ownersFollowed {
		viewstest.Serve(t, ctx, js, nc, o.owner, o.state, o.tool, o.stream, "n_"+delivery.ViewModule)
	}
}

// TestProcessWholeSeesTheBuckets starts the program the way the node does (its entry, run, with its environment and a
// broker file) against an embedded bus holding the three buckets, and asks its socket.
func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)

	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	kvs := map[string]jetstream.KeyValue{}
	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, delivery.PresenceBucket, delivery.StuckBucket} {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[b] = kv
	}
	put := func(bucket, key string, value []byte) {
		t.Helper()
		if _, err := kvs[bucket].Put(ctx, key, value); err != nil {
			t.Fatal(err)
		}
	}
	haveValue := func(build string, phase delivery.Phase, health delivery.Health) []byte {
		b, err := delivery.EncodeHave(delivery.Have{Build: build, Phase: phase, Health: health})
		if err != nil {
			t.Fatal(err)
		}
		return b
	}
	// The seat's placement (issue 774, ADR 0549 §3): x and y on a, nothing on b.
	placeValue := func(node string, modules ...string) []byte {
		p := delivery.Placement{Node: node}
		for _, m := range modules {
			p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
		}
		v, err := delivery.EncodePlacement(p)
		if err != nil {
			t.Fatal(err)
		}
		return v
	}
	put(delivery.PlacementBucket, "placement.a", placeValue("a", "x", "y"))
	put(delivery.PlacementBucket, "placement.b", placeValue("b"))
	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseChecking, delivery.HealthWaits))
	put(delivery.HaveBucket, "b.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK))

	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	rowsMatch := func(check func(view.Snapshot) bool) view.Snapshot {
		t.Helper()
		var last view.Snapshot
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if s, err := view.Ask(socket); err == nil {
				last = s
				if check(s) {
					return s
				}
			}
		}
		t.Fatalf("the socket never showed the expected rows; last: %+v", last)
		return last
	}
	byKey := func(s view.Snapshot, node, module string) view.Row {
		for _, r := range s.Rows {
			if r.Node == node && r.Module == module {
				return r
			}
		}
		return view.Row{}
	}
	s := rowsMatch(func(s view.Snapshot) bool {
		return byKey(s, "a", "x").Health == "waits" && byKey(s, "a", "y").Waits == "no have yet" && byKey(s, "b", "x").Status == view.StatusLeftBehind
	})
	if s.Counted() != 2 {
		t.Fatalf("counted = %d, want 2 (a.x, a.y): %+v", s.Counted(), s.Rows)
	}

	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK))
	rowsMatch(func(s view.Snapshot) bool { r := byKey(s, "a", "x"); return r.Phase == "running" && r.Health == "ok" })
}

// TestProcessWholeShowsWhatTheWatchmanSays starts the program whole and writes an offline presence and a stuck key to the
// watchman's buckets, sees both shown, then writes them online and moved and sees both gone.
func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	kvs := map[string]jetstream.KeyValue{}
	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, delivery.PresenceBucket, delivery.StuckBucket} {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[b] = kv
	}
	presence := func(key string, online bool) {
		t.Helper()
		b, _ := delivery.EncodePresence(delivery.Presence{Online: online})
		if _, err := kvs[delivery.PresenceBucket].Put(ctx, key, b); err != nil {
			t.Fatal(err)
		}
	}
	stuck := func(moved bool) {
		t.Helper()
		b, _ := delivery.EncodeStuck(delivery.Stuck{Bucket: delivery.HaveBucket, Key: "a.x", Waited: "a report from a", Since: time.Now(), Moved: moved})
		if _, err := kvs[delivery.StuckBucket].Put(ctx, delivery.HaveBucket+".a.x", b); err != nil {
			t.Fatal(err)
		}
	}
	presence("a", false)
	presence("part.mesh-delivery-view.i1", false)
	stuck(false)

	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	wait := func(check func(view.Snapshot) bool) {
		t.Helper()
		var last view.Snapshot
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if s, err := view.Ask(socket); err == nil {
				last = s
				if check(s) {
					return
				}
			}
		}
		t.Fatalf("the socket never showed the expected lists; last: %+v", last)
	}
	wait(func(s view.Snapshot) bool { return len(s.Offline) == 2 && len(s.Stuck) == 1 && s.Stuck[0].Key == "a.x" })
	presence("a", true)
	presence("part.mesh-delivery-view.i1", true)
	stuck(true)
	wait(func(s view.Snapshot) bool { return len(s.Offline) == 0 && len(s.Stuck) == 0 })
}

// TestProcessWholeShowsTheIndexersBucketsOnceTheyExist starts the program whole before the indexer's buckets exist, then
// creates them and puts keys: the view's absent-bucket retry (follow, unchanged) finds them and the snapshot holds them.
// The retry is the view's own, every 5 s, so the wait here is 15 s, not 5.
func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 1241/AC5
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, delivery.PresenceBucket, delivery.StuckBucket} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5}); err != nil {
			t.Fatal(err)
		}
	}
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	// The view is up and the indexer's buckets are not there.
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
		if _, err := view.Ask(socket); err == nil {
			break
		}
	}
	cands, err := js.KeyValue(ctx, delivery.CandidatesBucket)
	if err != nil {
		t.Fatal(err)
	}
	mods, err := js.KeyValue(ctx, delivery.ModulesBucket)
	if err != nil {
		t.Fatal(err)
	}
	const head = "0123456789abcdef0123456789abcdef01234567"
	hash, _ := delivery.HeadsHash([]delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: head}})
	cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-delivery", Number: 12, Head: head}, Modules: []string{"a"}}}})
	if err != nil {
		t.Fatal(err)
	}
	mod, err := delivery.EncodeModule(delivery.Module{None: "no tree", Repository: "novox/mesh-delivery", Commit: head})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := cands.Put(ctx, "feat/x", cand); err != nil {
		t.Fatal(err)
	}
	if _, err := mods.Put(ctx, "a", mod); err != nil {
		t.Fatal(err)
	}
	var last view.Snapshot
	for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
		if s, err := view.Ask(socket); err == nil {
			last = s
			if len(s.Groups) == 1 && len(s.Fingerprints) == 1 {
				if s.Groups[0].Group != "feat/x" || s.Fingerprints[0].Module != "a" || s.Fingerprints[0].State != view.FPNone {
					t.Fatalf("snapshot = %+v", s)
				}
				return
			}
		}
	}
	t.Fatalf("the snapshot never held the indexer's keys; last: %+v", last)
}

// TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes starts the program whole before the gate's asks bucket and the
// teller's answers bucket exist, then creates them: an ask appears in the snapshot, and leaves it once the teller has an
// answer under its key (1345/AC1 against a real bus; the retry is the view's own, every 5 s).
func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { // 1345/AC1
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	gate, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AsksBucket, History: delivery.AsksHistory})
	if err != nil {
		t.Fatal(err)
	}
	teller, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AnswersBucket, History: delivery.AnswersHistory})
	if err != nil {
		t.Fatal(err)
	}
	heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}}
	key, err := delivery.MergeAskKey("feat/x", heads)
	if err != nil {
		t.Fatal(err)
	}
	ask := oc.Ask{ID: delivery.AskID(key), Headline: "Merge into main: mesh-delivery#12", Explanation: "Merges one pull request.", Who: occ.Operator,
		Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "the merge stays held",
		Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}}
	value, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Ask: rawAsk(t, ask)})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := gate.Put(ctx, key, value); err != nil {
		t.Fatal(err)
	}
	wait := func(check func(view.Snapshot) bool, what string) {
		t.Helper()
		var last view.Snapshot
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if s, err := view.Ask(socket); err == nil {
				last = s
				if check(s) {
					return
				}
			}
		}
		t.Fatalf("%s; last: %+v", what, last)
	}
	wait(func(s view.Snapshot) bool {
		return len(s.Asks) == 1 && s.Asks[0].Key == key && s.Asks[0].How == "answer on the phone"
	}, "the open ask never showed")
	answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "approve", Warrant: ask.ID + "@2026-10-18T12:00:00Z", Words: "the operator, via telegram, chose Approve"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := teller.Put(ctx, key, answer); err != nil {
		t.Fatal(err)
	}
	wait(func(s view.Snapshot) bool { return len(s.Asks) == 0 }, "the answered ask never left the list")
}

// TestProcessWholeListsEveryDifference starts the program whole before the assigner's differences bucket exists, then
// creates it and writes a difference explained by a rule and one unexplained: the socket's snapshot lists both with their
// rule, and a difference taken away leaves the list (1297/B8: the view lists every difference with its rule).
func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	kv, err := js.KeyValue(ctx, delivery.AssignerWantsBucket)
	if err != nil {
		t.Fatal(err)
	}
	const fpOld = "src2:1111111111111111111111111111111111111111111111111111111111111111"
	const fpNew = "src2:2222222222222222222222222222222222222222222222222222222222222222"
	for key, d := range map[string]delivery.Want{
		"want.ace.plex": {FP: fpNew, Composed: "sha256:" + strings.Repeat("c", 64)},
		"want.g14.plex": {FP: fpOld, Composed: "sha256:" + strings.Repeat("c", 64), Newest: fpNew, Rule: delivery.RuleCanary, Words: "the canary ace does not run it healthy yet"},
	} {
		v, err := delivery.EncodeWant(d)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := kv.Put(ctx, key, v); err != nil {
			t.Fatal(err)
		}
	}
	wait := func(check func(view.Snapshot) bool, what string) {
		t.Helper()
		var last view.Snapshot
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if s, err := view.Ask(socket); err == nil {
				last = s
				if check(s) {
					return
				}
			}
		}
		t.Fatalf("%s; last: %+v", what, last)
	}
	wait(func(s view.Snapshot) bool {
		d := s.Wants
		return len(d) == 2 && d[0].Node == "ace" && d[0].State == view.WantPending && d[1].Node == "g14" && d[1].State == view.WantHeldBack && d[1].Rule == delivery.RuleCanary
	}, "the two wants never showed with where they stand")
	if err := kv.Delete(ctx, "want.g14.plex"); err != nil {
		t.Fatal(err)
	}
	wait(func(s view.Snapshot) bool { return len(s.Wants) == 1 }, "the want taken away never left the list")
}

// TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof starts the program whole, with the assigner's asks, the view's
// answers and the teller's answers buckets, and a fixture hold ask (in release 5 the assigner writes none yet). The
// operator's answer over the socket is written under `<ask key>.view` in the view's bucket; the ask shows the choice, and
// leaves the list once the teller's answer is under its key. The merge's Approve is refused with its reason and writes
// nothing (1297/B9a AC).
func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 1297/B9a AC
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	bucket := func(name string, history uint8) jetstream.KeyValue {
		t.Helper()
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history})
		if err != nil {
			t.Fatal(err)
		}
		return kv
	}
	askBucket := bucket(delivery.AssignerAsksBucket, delivery.AssignerAsksHistory)
	viewBucket := bucket(delivery.ViewAnswersBucket, delivery.ViewAnswersHistory)
	teller := bucket(delivery.AnswersBucket, delivery.AnswersHistory)
	gate := bucket(delivery.AsksBucket, delivery.AsksHistory)

	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	const fp = "src2:2222222222222222222222222222222222222222222222222222222222222222"
	key, err := delivery.HoldAskKey("plex", fp)
	if err != nil {
		t.Fatal(err)
	}
	ask := oc.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: occ.Operator,
		Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "the fingerprint stays held",
		Options: []oc.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: occ.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: occ.Approve}}}
	value, _ := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: rawAsk(t, ask)})
	if _, err := askBucket.Put(ctx, key, value); err != nil {
		t.Fatal(err)
	}
	wait := func(check func(view.Snapshot) bool, what string) {
		t.Helper()
		var last view.Snapshot
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if s, err := view.Ask(socket); err == nil {
				last = s
				if check(s) {
					return
				}
			}
		}
		t.Fatalf("%s; last: %+v", what, last)
	}
	wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].How == view.AnswerInTheView }, "the open ask never showed")

	if err := view.AnswerAsk(socket, key, "release", "the check was a flake"); err != nil {
		t.Fatal(err)
	}
	e, err := viewBucket.Get(ctx, key+".view")
	if err != nil {
		t.Fatalf("the view wrote no answer under %s.view: %v", key, err)
	}
	if a, err := delivery.DecodeViewAnswer(e.Value()); err != nil || a.Option != "release" {
		t.Fatalf("answer = %+v, %v", a, err)
	}
	wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Chosen == "release" }, "the ask never showed the view's choice")

	// The merge's Approve is answered on the phone alone.
	heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}}
	mkey, err := delivery.MergeAskKey("feat/x", heads)
	if err != nil {
		t.Fatal(err)
	}
	merge := oc.Ask{ID: delivery.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: occ.Operator, Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "held",
		Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}}
	mv, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Ask: rawAsk(t, merge)})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := gate.Put(ctx, mkey, mv); err != nil {
		t.Fatal(err)
	}
	wait(func(s view.Snapshot) bool { return len(s.Asks) == 2 }, "the merge ask never showed")
	if err := view.AnswerAsk(socket, mkey, "approve", ""); err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") {
		t.Errorf("merge Approve = %v, want the refusal naming the phone", err)
	}
	if _, err := viewBucket.Get(ctx, mkey+".view"); err == nil {
		t.Error("the view wrote an answer to the merge's ask")
	}

	// The teller fills the ask: it leaves the list.
	answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-23T12:00:00Z", Words: "answered in the view"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := teller.Put(ctx, key, answer); err != nil {
		t.Fatal(err)
	}
	wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask never left the list")
}

// TestProcessWholeShowsTheBacklogOfBuildAsks starts the program whole against a bus holding the go seat's work-queue stream
// with its three shared workers, three check asks waiting and one held by a node: the socket's backlog row names three
// waiting, one held, and the oldest's age (1887/A8a AC; compiled in the unit run, run by the merge check).
func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	stream, _ := delivery.AskStream(delivery.AskGo)
	st, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat.node-build-go.accept.>"}, Retention: jetstream.WorkQueuePolicy})
	if err != nil {
		t.Fatal(err)
	}
	var checkWorker jetstream.Consumer
	for _, kind := range delivery.AskKinds {
		name, _ := delivery.AskWorker(delivery.AskGo, kind)
		subject, _ := delivery.AskSubject(delivery.AskGo, kind)
		c, err := st.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{Durable: name, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy})
		if err != nil {
			t.Fatal(err)
		}
		if kind == delivery.AskCheck {
			checkWorker = c
		}
	}
	subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
	oldest := time.Now().UTC().Add(-2 * time.Minute)
	for i, id := range []string{"m0000001aaaaaa", "m0000002bbbbbb", "m0000003cccccc", "m0000004dddddd"} {
		body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: oldest.Add(time.Duration(i) * time.Second),
			Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := js.Publish(ctx, subject, body); err != nil {
			t.Fatal(err)
		}
	}
	// A node pulls one and holds it unacknowledged.
	held, err := checkWorker.Fetch(1)
	if err != nil {
		t.Fatal(err)
	}
	for range held.Messages() {
	}

	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	socket := filepath.Join(dir, "run", "cli.sock")
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	var last view.Snapshot
	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
		s, err := view.Ask(socket)
		if err != nil {
			continue
		}
		last = s
		for _, b := range s.Backlog {
			if b.Language == "go" && b.Kind == "check" && b.Waiting == 3 && b.Held == 1 && b.OldestSecs >= 118 {
				return
			}
		}
	}
	t.Fatalf("the backlog never showed three waiting asks with the oldest's age; last: %+v", last.Backlog)
}

// 1906/D9, whole on the bus: when the wants and the haves make the source revision delivered, the view says it once, as its event
// `delivered` with the revision, the group and the heads, which the issue tracker derives "fix delivered" from.
func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}})
	if err != nil {
		t.Fatal(err)
	}
	kvs := map[string]jetstream.KeyValue{}
	for _, name := range []string{delivery.SourceBucket, delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket} {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[name] = kv
	}
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })

	fp := "src2:" + strings.Repeat("2", 64)
	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
	want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
	failed, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed})
	good, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	src, _ := delivery.EncodeSource(delivery.Source{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
	for bucket, kv := range map[string][]byte{delivery.ModulesBucket: mod} {
		if _, err := kvs[bucket].Put(ctx, "web", kv); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := kvs[delivery.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil {
		t.Fatal(err)
	}
	if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", failed); err != nil {
		t.Fatal(err)
	}
	rev, err := kvs[delivery.SourceBucket].Put(ctx, delivery.SourceMainKey, src)
	if err != nil {
		t.Fatal(err)
	}
	time.Sleep(time.Second)
	if info, err := events.Info(ctx); err != nil || info.State.Msgs != 0 {
		t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State)
	}
	if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", good); err != nil {
		t.Fatal(err)
	}
	var msg *jetstream.RawStreamMsg
	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && msg == nil; time.Sleep(100 * time.Millisecond) {
		msg, _ = events.GetLastMsgForSubject(ctx, "mesh.mod.mesh-delivery-view.event."+DeliveredEventType)
	}
	if msg == nil {
		t.Fatal("the delivery of the source revision was never said")
	}
	var e view.DeliveredEvent
	if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) {
		t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev)
	}
	if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" {
		t.Errorf("the headers = %v", msg.Header)
	}
}

// 1906/D10, whole on the bus (through run): a want that differs from its have is said `moving`, the have reaching it running and
// healthy `settled`, a have held or failed `stuck`, each once with the body the messenger reads and an event id per type, node, module
// and composition.
func TestProcessWholeSaysModulesInMotion(t *testing.T) {
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}})
	if err != nil {
		t.Fatal(err)
	}
	kvs := map[string]jetstream.KeyValue{}
	for _, name := range []string{delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.StuckBucket} {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[name] = kv
	}
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	serveOwners(t, ctx, js, nc)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(func() { stop(); <-done })
	time.Sleep(2 * time.Second) // the view has read what stood: nothing

	fpOld, fpNew := "src2:"+strings.Repeat("1", 64), "src2:"+strings.Repeat("2", 64)
	digest := "sha256:" + strings.Repeat("c", 64)
	put := func(bucket, key string, v []byte, err error) {
		t.Helper()
		if err != nil {
			t.Fatal(err)
		}
		if _, err := kvs[bucket].Put(ctx, key, v); err != nil {
			t.Fatal(err)
		}
	}
	want := func(node, module string) {
		v, err := delivery.EncodeWant(delivery.Want{FP: fpNew, Composed: digest})
		put(delivery.AssignerWantsBucket, "want."+node+"."+module, v, err)
	}
	have := func(node, module, fp string, phase delivery.Phase, health delivery.Health, detail string) {
		v, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: phase, Health: health, Detail: detail})
		put(delivery.HaveBucket, node+"."+module, v, err)
	}
	have("ace", "web", fpOld, delivery.PhaseRunning, delivery.HealthOK, "")
	want("ace", "web")
	have("g14", "web", fpOld, delivery.PhaseRunning, delivery.HealthOK, "")
	want("g14", "web")

	said := func() map[string]jetstream.RawStreamMsg {
		out := map[string]jetstream.RawStreamMsg{}
		for seq := uint64(1); ; seq++ {
			m, err := events.GetMsg(ctx, seq)
			if err != nil {
				return out
			}
			out[m.Header.Get("x-event-id")] = *m
		}
	}
	waitFor := func(id string) jetstream.RawStreamMsg {
		t.Helper()
		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
			if m, ok := said()[id]; ok {
				return m
			}
		}
		t.Fatalf("never said %s; said %v", id, func() []string {
			var ids []string
			for k := range said() {
				ids = append(ids, k)
			}
			return ids
		}())
		return jetstream.RawStreamMsg{}
	}
	hex := strings.Repeat("c", 64)
	mv := waitFor("moving.ace.web." + hex)
	var body view.MotionBody
	if err := json.Unmarshal(mv.Data, &body); err != nil || body.Node != "ace" || body.Module != "web" || body.FromFP != fpOld || body.ToFP != fpNew || body.Composed != digest || body.BoundSeconds != view.MotionBoundSeconds || body.Since.IsZero() {
		t.Errorf("moving = %+v, %v", body, err)
	}
	if mv.Subject != "mesh.mod.mesh-delivery-view.event.moving" {
		t.Errorf("subject = %s", mv.Subject)
	}
	have("ace", "web", fpNew, delivery.PhaseRunning, delivery.HealthOK, "")
	st := waitFor("settled.ace.web." + hex)
	if err := json.Unmarshal(st.Data, &body); err != nil || body.Outcome != "running" {
		t.Errorf("settled = %+v, %v", body, err)
	}
	have("g14", "web", fpOld, delivery.PhaseHeld, delivery.HealthWaits, "no space left")
	stk := waitFor("stuck.g14.web." + hex)
	if err := json.Unmarshal(stk.Data, &body); err != nil || body.Outcome != "stopped" || body.Words != "no space left" {
		t.Errorf("stuck = %+v, %v", body, err)
	}
	time.Sleep(time.Second)
	count := map[string]int{}
	for id := range said() {
		count[strings.SplitN(id, ".", 2)[0]]++
	}
	if count["moving"] != 2 || count["settled"] != 1 || count["stuck"] != 1 {
		t.Errorf("events said by type = %v, want moving 2, settled 1, stuck 1", count)
	}
}