mesh-sdk/go/process/call_integration_test.go

//go:build integration

package process

import (
	"context"
	"encoding/json"
	"errors"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

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

// 1191/AC1: given a stand-in holder answering seat operator-channel verb tell on a test bus, when the
// process connection of novox.mesh-mail.send calls seat:operator-channel.tell with {headline, lines,
// subject}, then the request goes to mesh.seat.operator-channel.call.tell.novox~mesh-mail~send with the
// arguments as the body, and the holder's {result} comes back; a reply with error set comes back as an
// error with those words.
func TestCallReachesTheCallerNamedSubjectAndReadsTheReply(t *testing.T) {
	url, account := grantedSendProcess(t)
	adminNC, _ := adminJS(t, url)

	heard := make(chan *nats.Msg, 1)
	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
		heard <- m
		_ = m.Respond([]byte(`{"result":{"sent":true}}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := adminNC.Flush(); err != nil {
		t.Fatal(err)
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	result, err := conn.Call("seat:operator-channel.tell",
		map[string]any{"headline": "h", "lines": []string{"l1"}, "subject": "s"})
	if err != nil {
		t.Fatalf("Call: %v", err)
	}
	if string(result) != `{"sent":true}` {
		t.Fatalf("result: %s", result)
	}

	select {
	case msg := <-heard:
		var body map[string]any
		if err := json.Unmarshal(msg.Data, &body); err != nil {
			t.Fatal(err)
		}
		if body["headline"] != "h" || body["subject"] != "s" {
			t.Errorf("the arguments were not the call's body: %v", body)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("the holder never heard the call")
	}
}

// 1191/AC1: a reply with error set comes back as an error with those words.
func TestCallReadsAnErrorReplyAsAnError(t *testing.T) {
	url, account := grantedSendProcess(t)
	adminNC, _ := adminJS(t, url)
	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
		_ = m.Respond([]byte(`{"error":"refused: no such warrant"}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := adminNC.Flush(); err != nil {
		t.Fatal(err)
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	_, err = conn.Call("seat:operator-channel.tell", map[string]any{})
	if err == nil || err.Error() != "refused: no such warrant" {
		t.Fatalf("Call with an error reply: %v, want the words of the error", err)
	}
}

// 1191/AC2: given no responder on the caller-named subject, when the call is made, then it is asked
// again on mesh.seat.operator-channel.tool.tell (the tool runner's fallback).
func TestCallFallsBackToThePlainSubjectWhenNothingAnswersTheCallerNamedOne(t *testing.T) {
	url, account := grantedSendProcess(t)
	adminNC, _ := adminJS(t, url)
	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.tool.tell", func(m *nats.Msg) {
		_ = m.Respond([]byte(`{"result":{"via":"fallback"}}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := adminNC.Flush(); err != nil {
		t.Fatal(err)
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	result, err := conn.Call("seat:operator-channel.tell", map[string]any{})
	if err != nil {
		t.Fatalf("Call: %v", err)
	}
	if string(result) != `{"via":"fallback"}` {
		t.Fatalf("result: %s, want the fallback holder's answer", result)
	}
}

// 1191/AC4, release 2's acceptance test: given the test bus with accounts composed as the controller
// composes a process account (the subjects of mesh-controller internal/broker/process_grants.go), when
// one process connection emits, updates a state, reads an own ask's record, consumes and acknowledges its
// queue and calls tell, then each succeeds; and its attempt to subscribe a tool-call subject is refused
// by the bus.
func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)

	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
		t.Fatalf("the admin could not make the state bucket: %v", err)
	}
	asksKV, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"})
	if err != nil {
		t.Fatalf("the admin could not make the records bucket: %v", err)
	}
	if _, err := asksKV.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)); err != nil {
		t.Fatal(err)
	}
	if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
		Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
	}); err != nil {
		t.Fatalf("the admin could not make the events stream: %v", err)
	}
	makeQueue(t, url, account, 30*time.Second)
	publishApproved(t, url, `{"send":"s1"}`)

	adminNC, _ := adminJS(t, url)
	holderSub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
		_ = m.Respond([]byte(`{"result":{}}`))
	})
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = holderSub.Unsubscribe() }()
	if err := adminNC.Flush(); err != nil {
		t.Fatal(err)
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	// 1. Emits.
	if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "ev-1"}); err != nil {
		t.Fatalf("Emit: %v", err)
	}

	// 2. Updates a state.
	s := conn.State("sends")
	rev, err := s.Create("k", map[string]any{"v": 1})
	if err != nil {
		t.Fatalf("State.Create: %v", err)
	}
	if _, err := s.Update("k", map[string]any{"v": 2}, rev); err != nil {
		t.Fatalf("State.Update: %v", err)
	}

	// 3. Reads an own ask's record.
	if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e == nil {
		t.Fatalf("Ask: %+v, %v", e, err)
	}

	// 4. Consumes and acknowledges its queue.
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	consumed := make(chan error, 1)
	go func() {
		consumed <- conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
			if err := m.Ack(); err != nil {
				t.Errorf("Ack: %v", err)
			}
			cancel()
		})
	}()
	select {
	case <-consumed:
	case <-time.After(8 * time.Second):
		t.Fatal("the process did not consume its own queue")
	}

	// 5. Calls tell.
	if _, err := conn.Call("seat:operator-channel.tell", map[string]any{}); err != nil {
		t.Fatalf("Call: %v", err)
	}

	// 6. Refused: subscribing a tool-call subject — a fresh connection, so the error handler catches
	// exactly this one refusal and nothing earlier.
	refusals := make(chan error, 1)
	watched, err := Connect(account, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
	if err != nil {
		t.Fatal(err)
	}
	defer watched.Close()
	if _, err := watched.nc.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
		t.Fatal(err)
	}
	if err := watched.nc.Flush(); err != nil {
		t.Fatal(err)
	}
	select {
	case err := <-refusals:
		if !errors.Is(err, nats.ErrPermissionViolation) {
			t.Errorf("subscribing a tool-call subject was not refused as a permissions violation: %v", err)
		}
	case <-time.After(3 * time.Second):
		t.Error("subscribing a tool-call subject the process does not serve was not refused")
	}
}

// 1470/AC4: two answerers of the same verb on the subjects of nodes ace and g14; CallOn naming g14 gets g14's
// answer, and ace's answerer hears nothing. Issue 774 notes 84-87: Call with the key `<verb>@g14` is published on the
// same subject, never on one holding the '@' that nobody serves.
func TestCallOnReachesTheNamedNodeAlone(t *testing.T) {
	url, account := grantedSendProcess(t)
	adminNC, _ := adminJS(t, url)
	aceHeard := make(chan struct{}, 1)
	for _, node := range []string{"ace", "g14"} {
		node := node
		sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell."+node+".novox~mesh-mail~send", func(m *nats.Msg) {
			if node == "ace" {
				aceHeard <- struct{}{}
			}
			_ = m.Respond([]byte(`{"result":{"node":"` + node + `"}}`))
		})
		if err != nil {
			t.Fatal(err)
		}
		defer func() { _ = sub.Unsubscribe() }()
	}
	if err := adminNC.Flush(); err != nil {
		t.Fatal(err)
	}
	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	result, err := conn.CallOn("g14", "seat:operator-channel.tell", map[string]any{})
	if err != nil {
		t.Fatalf("CallOn: %v", err)
	}
	if string(result) != `{"node":"g14"}` {
		t.Fatalf("result: %s, want g14's answer", result)
	}
	if result, err = conn.Call("seat:operator-channel.tell@g14", map[string]any{}); err != nil || string(result) != `{"node":"g14"}` {
		t.Fatalf("Call with @g14: %s, %v, want g14's answer", result, err)
	}
	select {
	case <-aceHeard:
		t.Error("ace's answerer heard a call named for g14")
	default:
	}
}

// brokerSeat is the fixture seat's definition as a seat list holds it: its three verbs, each its class's default.
func brokerSeat() (seatdef.Definition, bool) {
	return seatdef.Definition{Seat: "broker-seat", Verbs: []seatdef.Verb{{Name: "ping"}, {Name: "fail"}, {Name: "boom"}}}, true
}

// A-107: a process answers its seat's verbs over the bus. Given two instances of one holder answering seat
// `broker-seat`'s verbs, a call on the caller-named subject reaches exactly one, which is handed the caller the
// subject names (dots written `~`) and replies {result, node}; a call on the subject naming none is refused; a
// handler's error is the reply's error, a panic too; and after stop nothing answers.
func TestAnswerSeatServesVerbsWithTheProvedCallerInOneQueueGroup(t *testing.T) {
	url, account := grantedSendProcess(t)
	adminNC, _ := adminJS(t, url)
	account.User, account.Password, account.Node = adminUser, adminPass, "novox"

	var heard sync.Map
	count := func(name string) SeatHandler {
		return func(caller string, body json.RawMessage) (any, error) {
			n, _ := heard.LoadOrStore(name, new(int32))
			atomic.AddInt32(n.(*int32), 1)
			return map[string]string{"caller": caller, "body": string(body)}, nil
		}
	}
	verbs := func(name string) map[string]SeatHandler {
		return map[string]SeatHandler{
			"ping": count(name),
			"fail": func(string, json.RawMessage) (any, error) { return nil, errors.New("refused: not yours") },
			"boom": func(string, json.RawMessage) (any, error) { panic("oops") },
		}
	}
	var stops []func()
	for _, name := range []string{"a", "b"} {
		conn, err := Connect(account)
		if err != nil {
			t.Fatal(err)
		}
		defer conn.Close()
		stop, err := conn.AnswerSeat("broker-seat", brokerSeat, verbs(name))
		if err != nil {
			t.Fatalf("AnswerSeat: %v", err)
		}
		stops = append(stops, stop)
	}
	ask := func(subject string) (struct {
		Result map[string]string `json:"result"`
		Error  string            `json:"error"`
		Node   string            `json:"node"`
	}, error) {
		var r struct {
			Result map[string]string `json:"result"`
			Error  string            `json:"error"`
			Node   string            `json:"node"`
		}
		msg, err := adminNC.Request(subject, []byte(`{"x":1}`), 5*time.Second)
		if err != nil {
			return r, err
		}
		return r, json.Unmarshal(msg.Data, &r)
	}

	r, err := ask("mesh.seat.broker-seat.call.ping.novox~mesh-delivery-builder")
	if err != nil || r.Result["caller"] != "novox.mesh-delivery-builder" || r.Result["body"] != `{"x":1}` || r.Node != "novox" {
		t.Fatalf("a caller-named call: %+v, %v", r, err)
	}
	// An unproven caller is refused before the handler runs (design 47 G §610; lead answers A-165, A-179, A-186).
	r, err = ask("mesh.seat.broker-seat.tool.ping")
	if err != nil || r.Result != nil || !strings.Contains(r.Error, "broker-seat.ping takes no unproven caller") {
		t.Fatalf("an unproven call: %+v, %v", r, err)
	}
	if _, err := ask("mesh.seat.broker-seat.call.ping.novox~mesh-delivery-builder"); err != nil {
		t.Fatal(err)
	}
	total := int32(0)
	heard.Range(func(_, v any) bool { total += atomic.LoadInt32(v.(*int32)); return true })
	if total != 2 {
		t.Errorf("two proven calls were answered %d times in all, want once each (one queue group) and the unproven none", total)
	}
	if r, err = ask("mesh.seat.broker-seat.call.fail.x"); err != nil || r.Error != "refused: not yours" {
		t.Errorf("a handler's error: %+v, %v", r, err)
	}
	if r, err = ask("mesh.seat.broker-seat.call.boom.x"); err != nil || !strings.Contains(r.Error, "oops") {
		t.Errorf("a handler's panic: %+v, %v", r, err)
	}
	for _, stop := range stops {
		stop()
	}
	_ = adminNC.Flush()
	if _, err := adminNC.Request("mesh.seat.broker-seat.call.ping.x", nil, time.Second); !errors.Is(err, nats.ErrNoResponders) {
		t.Errorf("after stop a call reached someone: %v", err)
	}
}