mesh-sdk/go/process/call_integration_test.go

//go:build integration

package process

import (
	"context"
	"encoding/json"
	"errors"
	"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")
	}
}