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

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"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/asks"
	"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 teller's own principal <node>.<module>.
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-teller", password: "pw" }
    ]
  }
}
`

type syncBuffer struct {
	mu sync.Mutex
	b  strings.Builder
}

func (s *syncBuffer) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.b.Write(p)
}
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }

// TestTellerWaitsForTheGate starts the program the way the node does (its entry, run, with MESH_BROKER_FILE) against an
// embedded bus where mesh-delivery-gate_asks does not exist: it keeps running until stopped, says once that it waits for
// the gate's asks, and writes and publishes nothing (the revised 1240/AC2: the watch is release 3d's).
func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
	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)

	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(), 30*time.Second)
	defer cancel()

	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker}
	logs := &syncBuffer{}
	rn := startRunner(ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
	t.Cleanup(func() { rn.finish(t, 10*time.Second) })

	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline) && !strings.Contains(logs.String(), "waiting for the gate's asks"); time.Sleep(20 * time.Millisecond) {
	}
	if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
		t.Fatalf("the teller said it waits %d times, want once: %q", n, logs.String())
	}
	select {
	case err := <-rn.done:
		t.Fatalf("the teller stopped: %v", err)
	case <-time.After(500 * time.Millisecond):
	}
	if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
		t.Errorf("the teller said it waits %d times after a while, want once", n)
	}
	// It writes and publishes nothing: no stream, no bucket exists.
	names := js.StreamNames(ctx)
	for name := range names.Name() {
		t.Errorf("a stream %s exists: the teller wrote something", name)
	}
	if _, err := js.KeyValue(ctx, "mesh-delivery-gate_asks"); err == nil {
		t.Error("the gate's asks bucket exists")
	}
	if err := rn.finish(t, 10*time.Second); err != nil {
		t.Errorf("the teller stopped with %v, want a clean stop", err)
	}
}

// The principals of the composed accounts file for the teller's tests, with the permissions mesh-controller composes for
// them (internal/broker): a module's own state is the read grants of its bucket and `$KV.<bucket>.>` to write it
// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses
// the seat operator-channel submits its own `accept.ask.<module>`, `accept.cancel.<module>` and `accept.condition.<module>`,
// hears its own `event.decided.<module>` and reads `$KV.asks.<module>.>` of the router's record (seattraffic.go
// SeatTrafficOf, with the messenger's seat declaring accepts ask, cancel, condition, emits decided, by-caller and records
// asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'.
// The permissions are these lists written out, not the controller's function: the broker tests of issue 1243 prove that.
func readOf(bucket string) []string {
	stream := "KV_" + bucket
	return []string{"$JS.API.STREAM.INFO." + stream, "$JS.API.DIRECT.GET." + stream + ".>", "$JS.API.CONSUMER.CREATE." + stream + ".>",
		"$JS.API.CONSUMER.DELETE." + stream + ".>", "$JS.FC." + stream + ".>"}
}

func keepsOf(bucket string) []string { return append(readOf(bucket), "$KV."+bucket+".>") }

func user(name string, pub, sub []string) string {
	q := func(l []string) string {
		var b strings.Builder
		for _, s := range l {
			b.WriteString(fmt.Sprintf("            %q\n", s))
		}
		return b.String()
	}
	return fmt.Sprintf(`      { user: %q, password: "pw"
        permissions {
          publish { allow: [
%s          ] }
          subscribe { allow: [
%s          ] }
          allow_responses: true
        }
      }
`, name, q(pub), q(append([]string{"_INBOX." + name + ".>"}, sub...)))
}

const (
	tellerAnswers = "mesh-delivery-teller_answers"
	tellerAsked   = "mesh-delivery-teller_asked"
	gateAsksKV    = "mesh-delivery-gate_asks"
	seatAsk       = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller"
	seatCancel    = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
	seatCondition = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller"
	seatDecided   = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller"
)

func fullConf(dir string) string {
	teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...),
		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_asks.$KV.asks.mesh-delivery-teller.>")
	view := append(readOf(tellerAnswers), readOf(gateAsksKV)...)
	tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
	return fmt.Sprintf(`
listen: 127.0.0.1:-1
jetstream { store_dir: %q }
accounts {
  MESH {
    jetstream: enabled
    users = [
      { user: "ctl", password: "pw" }
%s%s%s    ]
  }
}
`, filepath.Join(dir, "js"),
		user("n1.mesh-delivery-teller", teller, []string{seatDecided}),
		user("n1.mesh-delivery-view", view, nil),
		user("n1.node-tools", tools, nil))
}

type fullBed struct {
	srv *server.Server
	ctl *nats.Conn
	js  jetstream.JetStream
	ctx context.Context
}

// startFull starts the bus with the principals above and what the controller makes before any module starts: the buckets
// of the gate's asks, the teller's answers and asked, the router's work queue SEAT_OPERATOR_CHANNEL.
func startFull(t *testing.T) *fullBed {
	t.Helper()
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	conf := filepath.Join(dir, "bus.conf")
	if err := os.WriteFile(conf, []byte(fullConf(dir)), 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)
	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)
	for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"},
		Retention: jetstream.WorkQueuePolicy}); err != nil {
		t.Fatal(err)
	}
	return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
}

func (b *fullBed) as(t *testing.T, name string) jetstream.JetStream {
	t.Helper()
	c, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo(name, "pw"), nats.CustomInboxPrefix("_INBOX."+name))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(c.Close)
	js, err := jetstream.New(c)
	if err != nil {
		t.Fatal(err)
	}
	return js
}

// put writes a key as a principal, within a bounded time: a write the permissions refuse reaches the writer as a timeout.
func (b *fullBed) put(t *testing.T, js jetstream.JetStream, bucket, key string) error {
	t.Helper()
	ctx, cancel := context.WithTimeout(b.ctx, 3*time.Second)
	defer cancel()
	kv, err := js.KeyValue(ctx, bucket)
	if err != nil {
		t.Fatalf("%s cannot be bound: %v", bucket, err)
	}
	_, err = kv.Put(ctx, key, []byte(`{}`))
	return err
}

// TestCredentialsWriteOnlyTheirOwnBuckets: the teller's credential is refused writing the gate's asks, and the gate's
// (the tool runner's) and the view's are refused writing the teller's answers; each writes where it may, so a refusal is the
// permissions' and not a bus that does not take writes (1343/AC4, ADR 0437 §6's check).
func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
	b := startFull(t)
	teller, view, tools := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.node-tools")
	if err := b.put(t, teller, tellerAnswers, "merge.g.0123456789ab"); err != nil {
		t.Errorf("the teller cannot write its own answers: %v", err)
	}
	if err := b.put(t, tools, gateAsksKV, "merge.g.0123456789ab"); err != nil {
		t.Errorf("the gate's account cannot write the gate's asks: %v", err)
	}
	if err := b.put(t, teller, gateAsksKV, "merge.g.ffffffffffff"); err == nil {
		t.Error("the teller's credential wrote the gate's asks")
	}
	if err := b.put(t, tools, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
		t.Error("the gate's account wrote the teller's answers")
	}
	if err := b.put(t, view, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
		t.Error("the view's credential wrote the teller's answers")
	}
	// What was refused is not there.
	for _, c := range []struct{ bucket, key string }{{gateAsksKV, "merge.g.ffffffffffff"}, {tellerAnswers, "merge.g.ffffffffffff"}} {
		kv, _ := b.js.KeyValue(b.ctx, c.bucket)
		if _, err := kv.Get(b.ctx, c.key); err == nil {
			t.Errorf("%s holds %s though its write was refused", c.bucket, c.key)
		}
	}
}

// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the gate's account
// writes an ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the
// warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a
// second ask cancels it (1343/AC1, AC2).
func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
	now = time.Now().UTC()
	b := startFull(t)
	tools := b.as(t, "n1.node-tools")
	broker := filepath.Join(t.TempDir(), "broker")
	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker}
	logs := &syncBuffer{}
	rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
	t.Cleanup(func() { rn.finish(t, 10*time.Second) })

	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
	kv, err := tools.KeyValue(b.ctx, gateAsksKV)
	if err != nil {
		t.Fatal(err)
	}
	// The program hears warrants before it reads the asks; give it its subscription before the ask is written.
	time.Sleep(500 * time.Millisecond)
	if _, err := kv.Put(b.ctx, key, value); err != nil {
		t.Fatal(err)
	}
	seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
	if err != nil {
		t.Fatal(err)
	}
	var raw *jetstream.RawStreamMsg
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil {
			break
		}
	}
	if raw == nil {
		t.Fatalf("no ask on the seat's queue: %v\n%s", err, logs.String())
	}
	var asked asks.Ask
	if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.Digest() != ga.Ask.Digest() {
		t.Fatalf("the ask on the queue is %+v (%v), want the gate's", asked, err)
	}
	// The router says the warrant.
	w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
	if err := b.ctl.Publish(seatDecided, w); err != nil {
		t.Fatal(err)
	}
	answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
	if err != nil {
		t.Fatal(err)
	}
	var e jetstream.KeyValueEntry
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if e, err = answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil {
			break
		}
	}
	if e == nil {
		t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String())
	}
	if a, err := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
		t.Errorf("answer %+v err %v", a, err)
	}
	// A second ask, withdrawn by the gate, is cancelled at the router.
	other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
	key2, value2, ga2 := gateAsk(t, other, other)
	if _, err := kv.Put(b.ctx, key2, value2); err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if _, err := seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil && strings.Contains(logs.String(), ga2.Ask.ID) {
			break
		}
	}
	if err := kv.Delete(b.ctx, key2); err != nil {
		t.Fatal(err)
	}
	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil {
			break
		}
	}
	if err != nil || string(raw.Data) != `{"id":"`+ga2.Ask.ID+`"}` {
		t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String())
	}
}