mesh-delivery/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go

//go:build integration

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream,
// the controller's user, and the bridge's own principal <node>.<module> with the permissions the controller composes for
// it, and no others, so a permission the bridge lacks fails a test here and not on the mesh (issue 1279).
//
// The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the
// bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342).
//
// The bridge neither reads the checker's owed nor hears its stream (ADR 0521, R4e finding R2-1): the forge module writes
// mesh/check from it. The stream checks is made here so a test can say the checker wrote a status.
//
// The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal
// with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls,
// gitea.gitea_commit_trees and gitea.gitea_get_file, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go
// WirePattern). The consumer's durable is <node>_<module> on EVENTS (nats.go consumerDurable). The controller's own
// user keeps all rights, as it has.
const busConf = `
listen: 127.0.0.1:-1
jetstream { store_dir: %q }
accounts {
  MESH {
    jetstream: enabled
    users = [
      { user: "ctl", password: "pw" }
      { user: "n1.mesh-delivery-forge-bridge", password: "pw"
        permissions {
          publish { allow: [
            "$JS.ACK.EVENTS.n1_mesh-delivery-forge-bridge.>"
            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_heads.>"
            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_heads.>"
            "$JS.API.CONSUMER.INFO.EVENTS.n1_mesh-delivery-forge-bridge"
            "$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_mesh-delivery-forge-bridge"
            "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_heads.>"
            "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_heads"
            "$JS.FC.KV_mesh-delivery-forge-bridge_heads.>"
            "$KV.mesh-delivery-forge-bridge_heads.>"
            "mesh.mod.gitea.call.gitea_commit_trees.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_commit_trees.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_open_pulls.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_open_pulls.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_commit_trees"
            "mesh.mod.gitea.tool.gitea_commit_trees.*"
            "mesh.mod.gitea.call.gitea_get_file.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_get_file.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_get_file"
            "mesh.mod.gitea.tool.gitea_get_file.*"
            "mesh.mod.gitea.tool.gitea_open_pulls"
            "mesh.mod.gitea.tool.gitea_open_pulls.*"
            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_performed.>"
            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_performed.>"
            "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_performed.>"
            "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_performed"
            "$JS.FC.KV_mesh-delivery-forge-bridge_performed.>"
            "$KV.mesh-delivery-forge-bridge_performed.>"
            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_merges.>"
            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_merges.>"
            "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_merges.>"
            "$JS.API.STREAM.INFO.KV_mesh-delivery-gate_merges"
            "$JS.FC.KV_mesh-delivery-gate_merges.>"
            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_owed.>"
            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_owed.>"
            "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_owed.>"
            "$JS.API.STREAM.INFO.KV_mesh-delivery-gate_owed"
            "$JS.FC.KV_mesh-delivery-gate_owed.>"
            "mesh.mod.gitea.call.gitea_merge_pull_request.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_merge_pull_request.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_merge_pull_request"
            "mesh.mod.gitea.tool.gitea_merge_pull_request.*"
            "mesh.mod.gitea.call.gitea_commit_status.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_commit_status.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_commit_status"
            "mesh.mod.gitea.tool.gitea_commit_status.*"
            "mesh.mod.gitea.call.gitea_delivery_view.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_delivery_view.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_delivery_view"
            "mesh.mod.gitea.tool.gitea_delivery_view.*"
            "mesh.mod.gitea.call.gitea_note_append.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_note_append.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_note_append"
            "mesh.mod.gitea.tool.gitea_note_append.*"
            "mesh.mod.gitea.call.gitea_add_comment.*.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.call.gitea_add_comment.n1~mesh-delivery-forge-bridge"
            "mesh.mod.gitea.tool.gitea_add_comment"
            "mesh.mod.gitea.tool.gitea_add_comment.*"
            "mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
          ] }
          subscribe { allow: [
            "_INBOX.n1.mesh-delivery-forge-bridge.>"
            "mesh.assignment.n1.mesh-delivery-forge-bridge"
            "mesh.mod.gitea.event.pull.closed"
            "mesh.mod.gitea.event.pull.merged"
            "mesh.mod.gitea.event.pull.updated"
            "mesh.mod.mesh-delivery-forge-bridge.call.>"
            "mesh.mod.mesh-delivery-forge-bridge.tool.>"
          ] }
          allow_responses: true
        }
      }
    ]
  }
}
`

func startBus(t *testing.T) *server.Server {
	t.Helper()
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "b")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	conf := filepath.Join(dir, "bus.conf")
	if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
		t.Fatal(err)
	}
	opts, err := server.ProcessConfigFile(conf)
	if err != nil {
		t.Fatal(err)
	}
	opts.NoLog, opts.NoSigs = true, true
	srv, err := server.NewServer(opts)
	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)
	return srv
}

// startBridge runs the program against an embedded bus with the stream EVENTS, the bridge's stream FORGE and its durable
// consumer <node>_<module> on EVENTS, as the controller makes them (mesh-control internal/broker streams.go:192,
// nats.go:1387-1396, modstreams.go). It returns a control connection, its JetStream and a context.
func startBridge(t *testing.T, openReply string) (*nats.Conn, jetstream.JetStream, context.Context, <-chan error) {
	b := startBed(t, openReply, nil)
	return b.ctl, b.js, b.ctx, b.done
}

// bed is a started bridge with the bus it runs against.
type bed struct {
	srv   *server.Server
	ctl   *nats.Conn
	js    jetstream.JetStream
	ctx   context.Context
	done  <-chan error
	reply func(string)
	// files sets what the stand-in forge answers to gitea_get_file from then on; it is asked with the tool's arguments.
	files func(func(owner, repo, path, ref string) (string, bool))
}

// getFileReply is the reply envelope for gitea_get_file as mesh-catalog modules/gitea tools/index.ts and client.ts getFile
// answer it ({file: {path, ref, sha, size, content}}), and for a path the forge does not have the error client.ts request
// writes: `Gitea API <path>: 404 <body>`.
func getFileReply(path, ref, content string, found bool) []byte {
	if !found {
		b, _ := json.Marshal(map[string]any{"error": "Gitea API /repos/novox/mesh-delivery/contents/" + path + "?ref=" + ref + `: 404 {"message":"GetContentsOrList","url":"https://forge.example/api/swagger"}`})
		return b
	}
	b, _ := json.Marshal(map[string]any{"result": map[string]any{"file": map[string]any{"path": path, "ref": ref, "sha": "0123456789abcdef0123456789abcdef01234567",
		"size": len(content), "content": content}}, "node": "n0"})
	return b
}

// startBed is startBridge with a step between the streams and the consumer: prepare writes what the streams hold before
// the bridge's consumer is made, as the controller makes it, from now (ADR 0443 §1), and reply sets the answer of the
// forge's gitea_open_pulls from then on.
func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, js jetstream.JetStream)) *bed {
	srv := startBus(t)
	ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ctl.Close)
	js, err := jetstream.New(ctl)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	t.Cleanup(cancel)
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
		t.Fatal(err)
	}
	// The checker's stream checks, which the bridge no longer hears (ADR 0521): a test says on it that the checker wrote a status.
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil {
		t.Fatal(err)
	}
	// The bridge's state bucket, made by the controller from the manifest's state entry before the process starts.
	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
		t.Fatal(err)
	}
	// Its other state bucket, and the gate's buckets the bridge reads, made by the controller from the gate's manifest
	// (mesh-sdk go/delivery MergesBucket, OwedBucket with their histories).
	for name, history := range map[string]uint8{performedBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
			t.Fatal(err)
		}
	}
	if prepare != nil {
		prepare(ctx, js)
	}
	if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{
		Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy, DeliverPolicy: jetstream.DeliverNewPolicy,
		FilterSubjects: []string{"mesh.mod.gitea.event.pull.updated", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.closed"},
	}); err != nil {
		t.Fatal(err)
	}

	// A stand-in answers the forge's gitea_open_pulls, which the bridge asks at its start and after a reconnect, on its tool
	// subject (mesh-control internal/broker/nats.go invokedSubjects) in the reply envelope of mesh-sdk go/process/call.go.
	var mu sync.Mutex
	current := openReply
	open, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_open_pulls", func(m *nats.Msg) {
		mu.Lock()
		defer mu.Unlock()
		_ = m.Respond([]byte(current))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = open.Unsubscribe() })
	if err := ctl.Flush(); err != nil {
		t.Fatal(err)
	}

	// A stand-in answers the forge's gitea_get_file: by default a module.json of {} at every path and no other file.
	var fmu sync.Mutex
	filesOf := func(_, _, path, _ string) (string, bool) { return "{}", strings.HasSuffix(path, "module.json") }
	getFile, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_get_file", func(m *nats.Msg) {
		var a struct{ Owner, Repo, Path, Ref string }
		_ = json.Unmarshal(m.Data, &a)
		fmu.Lock()
		f := filesOf
		fmu.Unlock()
		content, found := f(a.Owner, a.Repo, a.Path, a.Ref)
		_ = m.Respond(getFileReply(a.Path, a.Ref, content, found))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = getFile.Unsubscribe() })
	if err := ctl.Flush(); err != nil {
		t.Fatal(err)
	}

	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	t.Cleanup(stop)
	return &bed{srv: srv, ctl: ctl, js: js, ctx: ctx, done: done, reply: func(s string) { mu.Lock(); current = s; mu.Unlock() },
		files: func(f func(owner, repo, path, ref string) (string, bool)) { fmu.Lock(); filesOf = f; fmu.Unlock() }}
}

// TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE).
func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2
	_, js, ctx, _ := startBridge(t, `{"result":{"pulls":[]}}`)

	// gitea.pull.updated as the forge module publishes it: the body of trunk.ts:233-250 on mesh.mod.gitea.event.pull.updated.
	ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "t", "body": "b", "base": "main",
		"head": "feat/x", "head_sha": head, "clone_url": "u", "html_url": "h", "paths": []string{"modules/a/x.go"}, "paths_truncated": false,
		"removed": []string{}, "module_dirs": []string{"modules/a"}, "module_dirs_said": true})
	if _, err := js.Publish(ctx, "mesh.mod.gitea.event.pull.updated", ev); err != nil {
		t.Fatal(err)
	}

	forge, err := js.Stream(ctx, "FORGE")
	if err != nil {
		t.Fatal(err)
	}
	var info *jetstream.StreamInfo
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 {
			break
		}
	}
	// The open message the bridge writes at its start (ADR 0443 §5), and the event's message.
	if info.State.Msgs != 2 {
		t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs)
	}
	raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull")
	if err != nil {
		t.Fatal(err)
	}
	f, err := delivery.DecodeForge(raw.Data)
	if err != nil {
		t.Fatal(err)
	}
	want := delivery.Forge{V: delivery.Version, Kind: "pull", Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head,
		Action: "updated", Dirs: []string{"modules/a"}, DirsKnown: true, Manifests: map[string]string{"modules/a": "{}"}}
	if !delivery.SameForge(f, want) {
		t.Errorf("message = %+v, want %+v", f, want)
	}

	// The event is acknowledged: the consumer owes nothing and has nothing pending.
	cons, err := js.Consumer(ctx, "EVENTS", "n1_mesh-delivery-forge-bridge")
	if err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		ci, err := cons.Info(ctx)
		if err == nil && ci.NumAckPending == 0 && ci.NumPending == 0 && ci.Delivered.Consumer >= 1 {
			return
		}
	}
	ci, _ := cons.Info(ctx)
	t.Errorf("the event is not acknowledged: %+v", ci)
}

const merge = "89abcdef0123456789abcdef0123456789abcdef"

// TestBridgeWritesMergedMessage has a stand-in answer the forge's gitea_commit_trees on its tool subject
// (mesh-control internal/broker/nats.go invokedSubjects: mesh.mod.<module>.tool.<tool>), in the reply envelope of
// mesh-sdk go/process/call.go ({result, error, node}).
func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5
	bed := startBed(t, `{"result":{"pulls":[]}}`, nil)
	ctl, js, ctx := bed.ctl, bed.js, bed.ctx
	// The stand-in forge's files are two real files of this repository (this module's module.json and go.sum, read from the
	// working tree), in gitea_get_file's real answer shape: the root has a manifest only, modules/a a manifest and go.sum.
	manifest, err := os.ReadFile("../../module.json")
	if err != nil {
		t.Fatal(err)
	}
	sum, err := os.ReadFile("../../go.sum")
	if err != nil {
		t.Fatal(err)
	}
	bed.files(func(_, _, path, ref string) (string, bool) {
		if ref != merge {
			t.Errorf("the forge was asked for %s at %s, want the merge commit %s", path, ref, merge)
		}
		switch path {
		case "module.json", "modules/a/module.json":
			return string(manifest), true
		case "modules/a/go.sum":
			return string(sum), true
		}
		return "", false
	})
	var asked chan []byte = make(chan []byte, 4)
	sub, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_trees", func(m *nats.Msg) {
		asked <- m.Data
		_ = m.Respond([]byte(`{"result":{"root":"r0","trees":{"":"t0","modules/a":"t1"}},"node":"n0"}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	if err := ctl.Flush(); err != nil {
		t.Fatal(err)
	}

	// gitea.pull.merged as mesh-catalog modules/gitea/index.ts:186-207 publishes it.
	ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-delivery", "number": 12, "title": "t", "body": "b",
		"head_sha": head, "head": "feat/x", "base": "main", "merge_commit_sha": merge, "merged_at": "2026-10-19T12:00:00Z",
		"clone_url": "u", "html_url": "h", "paths": []string{"modules/a/x.go"}, "paths_truncated": false, "removed": []string{},
		"module_dirs": []string{"modules/a"}, "module_dirs_said": true})
	if _, err := js.Publish(ctx, "mesh.mod.gitea.event.pull.merged", ev); err != nil {
		t.Fatal(err)
	}

	forge, err := js.Stream(ctx, "FORGE")
	if err != nil {
		t.Fatal(err)
	}
	var info *jetstream.StreamInfo
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 {
			break
		}
	}
	// The open message the bridge writes at its start (ADR 0443 §5), and the event's message.
	if info.State.Msgs != 2 {
		t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs)
	}
	raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged")
	if err != nil {
		t.Fatal(err)
	}
	f, err := delivery.DecodeForge(raw.Data)
	if err != nil {
		t.Fatal(err)
	}
	want := delivery.Forge{V: delivery.Version, Kind: "merged", Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head,
		Merge: merge, Root: "r0", Trees: map[string]string{"": "t0", "modules/a": "t1"},
		Manifests: map[string]string{"": string(manifest), "modules/a": string(manifest)},
		Locks:     map[string]map[string]string{"modules/a": {"go.sum": hex.EncodeToString(sha256Sum(sum))}}}
	if !delivery.SameForge(f, want) {
		t.Errorf("message = %+v, want %+v", f, want)
	}
	select {
	case body := <-asked:
		if string(body) != `{"owner":"novox","repo":"mesh-delivery","sha":"`+merge+`"}` {
			t.Errorf("the tool was asked %s", body)
		}
	default:
		t.Error("the stand-in was not asked")
	}
	if len(asked) != 0 {
		t.Error("the tool was asked more than once")
	}
	cons, err := js.Consumer(ctx, "EVENTS", "n1_mesh-delivery-forge-bridge")
	if err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		ci, err := cons.Info(ctx)
		if err == nil && ci.NumAckPending == 0 && ci.NumPending == 0 && ci.Delivered.Consumer >= 1 {
			return
		}
	}
	ci, _ := cons.Info(ctx)
	t.Errorf("the event is not acknowledged: %+v", ci)
}

// TestBridgeSaysTheOpenPullsAtStart: the forge holds two open pull requests when the bridge starts; one message each is on
// FORGE, and the open message naming both after them (ADR 0443 §5).
func TestBridgeSaysTheOpenPullsAtStart(t *testing.T) { // 1237/AC1
	// The entries are gitea_open_pulls's (mesh-catalog modules/gitea/client.ts:369-381, head the branch).
	_, js, ctx, _ := startBridge(t, `{"result":{"pulls":[
	 {"owner":"novox","repo":"mesh-delivery","number":12,"base":"main","head":"feat/x","head_sha":"`+head+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]},
	 {"owner":"novox","repo":"mesh-catalog","number":300,"base":"main","head":"feat/x","head_sha":"`+strings.Repeat("a", 40)+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}]},"node":"n0"}`)
	forge, err := js.Stream(ctx, "FORGE")
	if err != nil {
		t.Fatal(err)
	}
	var info *jetstream.StreamInfo
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 3 {
			break
		}
	}
	if info.State.Msgs != 3 {
		t.Fatalf("FORGE holds %d messages within 5 s, want 3", info.State.Msgs)
	}
	for _, repo := range []string{"mesh-delivery", "mesh-catalog"} {
		raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".pull")
		if err != nil {
			t.Fatal(err)
		}
		if f, err := delivery.DecodeForge(raw.Data); err != nil || f.Action != "updated" || f.DirsKnown || f.Branch != "feat/x" {
			t.Errorf("%s: %+v err %v", repo, f, err)
		}
	}
}

// TestBridgeExitsWhenTheOpenPullsCannotBeRead: the process manager starts it again.
func TestBridgeExitsWhenTheOpenPullsCannotBeRead(t *testing.T) { // 1237/AC3
	_, _, _, done := startBridge(t, `{"error":"the forge said 502"}`)
	select {
	case err := <-done:
		if err == nil || !strings.Contains(err.Error(), "gitea.gitea_open_pulls") || !strings.Contains(err.Error(), "the forge said 502") {
			t.Errorf("run returned %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the bridge did not exit")
	}
}

// TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent starts the bridge whole against a fake forge with two
// open pull requests, one of them with its head already kept in the bucket heads, and a bus whose EVENTS holds older pull events from
// before its consumer was made, under the permissions the controller composes for the bridge. At start it writes the one pull message that differs and the open message naming both;
// the bus connection closed, it reads the forge again and writes the open message alone, since every head is held now;
// for the older events it writes nothing (ADR 0443 §1, §5).
func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *testing.T) { // 1279/AC8
	h2 := strings.Repeat("a", 40)
	pullSubject := func(repo string) string {
		return "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox." + repo + ".pull"
	}
	openSubject := "mesh.stream.mesh-delivery-forge-bridge.event.forge.open"
	b := startBed(t, `{"result":{"pulls":[
	 {"owner":"novox","repo":"mesh-delivery","number":12,"base":"main","head":"feat/x","head_sha":"`+head+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]},
	 {"owner":"novox","repo":"mesh-catalog","number":300,"base":"main","head":"feat/x","head_sha":"`+h2+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}]},"node":"n0"}`,
		func(ctx context.Context, js jetstream.JetStream) {
			// Older events: a pull request long closed and one long updated, from before the bridge's consumer.
			for subject, ev := range map[string]map[string]any{
				"mesh.mod.gitea.event.pull.updated": {"owner": "novox", "repo": "mesh-old", "number": 5, "head": "feat/old", "head_sha": h2},
				"mesh.mod.gitea.event.pull.closed":  {"owner": "novox", "repo": "mesh-older", "number": 6, "head": "feat/older", "head_sha": h2},
			} {
				raw, _ := json.Marshal(ev)
				if _, err := js.Publish(ctx, subject, raw); err != nil {
					t.Fatal(err)
				}
			}
			// What the bridge kept of an earlier write: the head of mesh-delivery#12.
			kv, err := js.KeyValue(ctx, headsBucket)
			if err != nil {
				t.Fatal(err)
			}
			if _, err := kv.PutString(ctx, "novox.mesh-delivery.12", head); err != nil {
				t.Fatal(err)
			}
		})
	forge, err := b.js.Stream(b.ctx, "FORGE")
	if err != nil {
		t.Fatal(err)
	}
	waitMsgs := func(want uint64) {
		t.Helper()
		var info *jetstream.StreamInfo
		for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
			if info, err = forge.Info(b.ctx); err == nil && info.State.Msgs >= want {
				break
			}
		}
		time.Sleep(300 * time.Millisecond) // and nothing more arrives
		if info, err = forge.Info(b.ctx); err != nil || info.State.Msgs != want {
			t.Fatalf("FORGE holds %d messages (%v), want %d", info.State.Msgs, err, want)
		}
	}
	named := func(seq uint64) []delivery.OpenPull {
		t.Helper()
		raw, err := forge.GetMsg(b.ctx, seq)
		if err != nil || raw.Subject != openSubject {
			t.Fatalf("message %d: %v %v, want the open message", seq, raw, err)
		}
		f, err := delivery.DecodeForge(raw.Data)
		if err != nil {
			t.Fatal(err)
		}
		return f.Open
	}
	bothOpen := []delivery.OpenPull{{Repository: "novox/mesh-catalog", Number: 300}, {Repository: "novox/mesh-delivery", Number: 12}}
	same := func(got []delivery.OpenPull) bool {
		return len(got) == 2 && got[0] == bothOpen[0] && got[1] == bothOpen[1]
	}

	// At start: the one pull message that differs (1), the open message naming both (2).
	waitMsgs(2)
	if raw, err := forge.GetMsg(b.ctx, 1); err != nil || raw.Subject != pullSubject("mesh-catalog") {
		t.Fatalf("message 1 = %v %v, want mesh-catalog's pull message", raw, err)
	}
	if got := named(2); !same(got) {
		t.Fatalf("the open message names %v, want %v", got, bothOpen)
	}

	// The connection closes and reopens: the forge is read again, every head is held, the open message alone follows.
	conns, err := b.srv.Connz(&server.ConnzOptions{})
	if err != nil {
		t.Fatal(err)
	}
	closed := 0
	for _, c := range conns.Conns {
		if c.Name == "n1.mesh-delivery-forge-bridge" {
			if err := b.srv.DisconnectClientByID(c.Cid); err != nil {
				t.Fatal(err)
			}
			closed++
		}
	}
	if closed != 1 {
		t.Fatalf("%d connections of the bridge found", closed)
	}
	waitMsgs(3)
	if got := named(3); !same(got) {
		t.Fatalf("after the reconnect the open message names %v, want %v", got, bothOpen)
	}

	// For the older events nothing was written.
	for _, repo := range []string{"mesh-old", "mesh-older"} {
		if raw, err := forge.GetLastMsgForSubject(b.ctx, pullSubject(repo)); err == nil {
			t.Errorf("a message for %s's older event: %+v", repo, raw)
		}
	}
}

// TestBridgeMergesAGroupOneMemberAtATime: the gate (played by the test) writes merges.g with member A handed and the
// Approve's heads of A and B; a stand-in forge merges A and announces it; the gate then hands B; both end merged on FORGE
// (1342/AC3, ADR 0437 §8's check).
func TestBridgeMergesAGroupOneMemberAtATime(t *testing.T) { // 1342/AC3
	b := startBed(t, `{"result":{"pulls":[]}}`, nil)
	const headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	const mergeA, mergeB = "1111111111111111111111111111111111111111", "2222222222222222222222222222222222222222"
	merges := make(chan string, 4)
	// The stand-in for gitea_merge_pull_request answers in the reply envelope of mesh-sdk go/process/call.go and publishes
	// gitea.pull.merged as mesh-catalog modules/gitea/index.ts:186-207 does, with the merge commit of the member merged.
	sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_merge_pull_request", func(m *nats.Msg) {
		var a struct {
			Repo string `json:"repo"`
			Head string `json:"head"`
		}
		_ = json.Unmarshal(m.Data, &a)
		merges <- string(m.Data)
		commit, number := mergeA, 12
		if a.Repo == "mesh-sdk" {
			commit, number = mergeB, 4
		}
		ev, _ := json.Marshal(map[string]any{"owner": "novox", "repo": a.Repo, "number": number, "head_sha": a.Head, "head": "feat/x",
			"base": "main", "merge_commit_sha": commit, "module_dirs": []string{"modules/a"}, "module_dirs_said": true})
		_, _ = b.js.Publish(b.ctx, "mesh.mod.gitea.event.pull.merged", ev)
		_ = m.Respond([]byte(`{"result":{"merged":true},"node":"n0"}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	trees, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_trees", func(m *nats.Msg) {
		_ = m.Respond([]byte(`{"result":{"root":"r0","trees":{"":"t0","modules/a":"t1"}},"node":"n0"}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = trees.Unsubscribe() })
	if err := b.ctl.Flush(); err != nil {
		t.Fatal(err)
	}

	kv, err := b.js.KeyValue(b.ctx, delivery.MergesBucket)
	if err != nil {
		t.Fatal(err)
	}
	approved := map[string]string{delivery.ApprovedKey("novox/mesh-delivery", 12): head, delivery.ApprovedKey("novox/mesh-sdk", 4): headB}
	pullA := delivery.MergePull{Head: delivery.Head{Repository: "novox/mesh-delivery", Number: 12, Head: head}, State: delivery.MergeMerging}
	pullB := delivery.MergePull{Head: delivery.Head{Repository: "novox/mesh-sdk", Number: 4, Head: headB}, State: delivery.MergeMerging}
	put := func(pulls ...delivery.MergePull) {
		v, err := delivery.EncodeMerge(delivery.Merge{Group: "g", Ask: "merge.g.0123456789ab", Approved: approved, Pulls: pulls})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := kv.Put(b.ctx, "g", v); err != nil {
			t.Fatal(err)
		}
	}
	next := func(repo string) {
		select {
		case got := <-merges:
			if !strings.Contains(got, `"repo":"`+repo+`"`) {
				t.Errorf("the forge was asked %s, want %s", got, repo)
			}
			for _, want := range []string{"novox/mesh-delivery@" + head, "novox/mesh-sdk@" + headB} {
				if !strings.Contains(got, want) {
					t.Errorf("the merge %s does not carry the approved head %s", got, want)
				}
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("%s was not merged within 10 s", repo)
		}
	}
	put(pullA)
	next("mesh-delivery")
	// The gate hands the next member only after the bridge reported the one before merged (ADR 0441 §5).
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && len(forgeMerged(t, b)) < 1; time.Sleep(50 * time.Millisecond) {
	}
	pullA.State, pullA.Commit = delivery.MergeMerged, mergeA
	put(pullA, pullB)
	next("mesh-sdk")
	var got []string
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if got = forgeMerged(t, b); len(got) == 2 {
			break
		}
	}
	if len(got) != 2 {
		t.Fatalf("FORGE holds merged messages for %v, want both members", got)
	}
	select {
	case extra := <-merges:
		t.Errorf("a third merge was asked: %s", extra)
	default:
	}
}

// forgeMerged lists the repositories FORGE holds a merged message for, sorted.
func forgeMerged(t *testing.T, b *bed) []string {
	t.Helper()
	forge, err := b.js.Stream(b.ctx, "FORGE")
	if err != nil {
		t.Fatal(err)
	}
	var out []string
	for _, repo := range []string{"mesh-delivery", "mesh-sdk"} {
		raw, err := forge.GetLastMsgForSubject(b.ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".merged")
		if err != nil {
			continue
		}
		if f, err := delivery.DecodeForge(raw.Data); err == nil && f.Kind == delivery.ForgeMerged {
			out = append(out, f.Repository)
		}
	}
	return out
}

func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }

// TestBridgePerformsNoMeshCheckAct: the checker owes mesh/check on a head and says so on its stream; the bridge calls no
// tool with mesh/check and publishes nothing of its own on EVENTS for it. The forge module writes mesh/check from the
// checker's owed state itself (ADR 0521; R4e round 2 finding R2-1).
func TestBridgePerformsNoMeshCheckAct(t *testing.T) { // R2-1, ADR 0521
	b := startBed(t, `{"result":{"pulls":[]}}`, nil)
	calls := make(chan string, 8)
	// Every forge call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
	sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) {
		if strings.Contains(string(m.Data), "mesh/check") {
			calls <- m.Subject + " " + string(m.Data)
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	if err := b.ctl.Flush(); err != nil {
		t.Fatal(err)
	}
	kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.CheckerOwedBucket, History: delivery.CheckerOwedHistory})
	if err != nil {
		t.Fatal(err)
	}
	key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, head)
	if err != nil {
		t.Fatal(err)
	}
	v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.0123456789ab"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := kv.Put(b.ctx, key, v); err != nil {
		t.Fatal(err)
	}
	if _, err := b.js.Publish(b.ctx, "mesh.stream.mesh-delivery-checker.event.owed.written", []byte(`{"key":"`+key+`"}`)); err != nil {
		t.Fatal(err)
	}
	select {
	case c := <-calls:
		t.Errorf("a forge tool was called with mesh/check: %s", c)
	case <-time.After(2 * time.Second):
	}
	events, err := b.js.Stream(b.ctx, "EVENTS")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-forge-bridge.event.>"); err == nil {
		t.Error("the bridge published an event of its own for the checker's mesh/check")
	}
}