mesh-sdk/go/process/gated_integration_test.go

//go:build integration

package process

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

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

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

// busChannel is the operator channel as the test's router: the gate's asks kept in the holder's state approvals, the
// router's record read under the holder's name from messenger_asks (design 47 G §700), the claim as
// operatorchannelclient's Take makes it. asked says each ask raised.
type busChannel struct {
	conn  *Conn
	state State
	asked chan approval.Ask
}

func (b *busChannel) Ask(module string, a approval.Ask) error {
	raw, _ := json.Marshal(a)
	if _, err := b.state.Create("ask."+a.ID, approval.Kept[json.RawMessage, json.RawMessage]{Ask: raw, Asked: time.Now()}); err != nil {
		return err
	}
	b.asked <- a
	return nil
}

func (b *busChannel) Record(module, id string) (*approval.Record, error) {
	e, err := b.conn.Ask("messenger_asks", id)
	if err != nil || e == nil {
		return nil, err
	}
	var r approval.Record
	return &r, json.Unmarshal(e.Value, &r)
}

func (b *busChannel) Claim(module string, w approval.Warrant, act operatorask.Act, run func()) (approval.Claim, error) {
	e, err := b.state.Get("ask." + w.Ask)
	if err != nil || e == nil {
		return approval.ClaimRefused, fmt.Errorf("no ask %s: %v", w.Ask, err)
	}
	var k approval.Kept[json.RawMessage, json.RawMessage]
	_ = json.Unmarshal(e.Value, &k)
	if k.Acted {
		return approval.ClaimDone, nil
	}
	var a approval.Ask
	_ = json.Unmarshal(k.Ask, &a)
	if d, _ := operatorask.ActDigest(act); d != a.About {
		return approval.ClaimRefused, fmt.Errorf("the option binds another act")
	}
	k.Acted = true
	if _, err := b.state.Update("ask."+w.Ask, k, e.Revision); err != nil {
		return approval.ClaimDone, nil
	}
	run()
	return approval.Claimed, nil
}

func (b *busChannel) Cancel(string, string) error { return nil }
func (b *busChannel) Decided(module string) string {
	return "mesh.seat.operator-channel.event.decided." + module
}
func (b *busChannel) Seat() string { return "operator-channel" }

// Design 47 G §677 (lead answers A-158, A-224, A-244), its one whole test: a process holder's marked verb, called by a
// module's process through process.Conn's call, is asked, approved, run once, recorded in the hand-act log and its
// answer heard on the caller's own gated-ended; a declined call runs nothing; a marked verb served through no gate, or
// a gate whose state approvals is not there, fails closed.
func TestAProcessHoldersMarkedVerbIsAskedApprovedRunAndRecorded(t *testing.T) {
	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
		Users: []*server.User{{Username: adminUser, Password: adminPass}, {Username: "n1.holder", Password: "pw"},
			{Username: "n1.caller", Password: "pw"}, {Username: "n1.mesh-hand-acts", Password: "pw"}, {Username: "n1.other", Password: "pw"}}}
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(10 * time.Second) {
		t.Fatal("the test bus did not come up")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	url := s.ClientURL()
	_, js := adminJS(t, url)
	ctx := ctxFor(t)
	records, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "messenger_asks"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "holder_" + approval.StateName, TTL: 48 * time.Hour,
		LimitMarkerTTL: time.Minute}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.operator-channel.event.>"}}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateOrUpdateConsumer(ctx, EventsStream, jetstream.ConsumerConfig{Durable: "n1_holder", AckPolicy: jetstream.AckExplicitPolicy,
		FilterSubjects: []string{"mesh.seat.operator-channel.event.decided.holder"}}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seatdef.KeptStream("wiper", seatdef.Message{Name: approval.GatedEndedEvent}),
		Subjects: []string{"mesh.seat.wiper.event.gated-ended.>"}, MaxMsgsPerSubject: 1, MaxAge: approval.GatedEndedKeptFor}); err != nil {
		t.Fatal(err)
	}
	connect := func(module string) *Conn {
		c, err := Connect(Account{URL: url, Node: "n1", Module: module, User: "n1." + module, Password: "pw"})
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(c.Close)
		return c
	}

	// The seat mesh-hand-acts' holder, recording each warranted run.
	var mu sync.Mutex
	var warranted []string
	handActs := connect("mesh-hand-acts")
	if _, err := handActs.AnswerSeat("mesh-hand-acts", func() (seatdef.Definition, bool) {
		return seatdef.Definition{Seat: "mesh-hand-acts", Verbs: []seatdef.Verb{{Name: "warranted"}}}, true
	}, map[string]SeatHandler{"warranted": func(caller string, body json.RawMessage) (any, error) {
		mu.Lock()
		defer mu.Unlock()
		warranted = append(warranted, caller+" "+string(body))
		return map[string]bool{"recorded": true}, nil
	}}); err != nil {
		t.Fatal(err)
	}

	// The holder: its marked verb wipe behind its gate, its decided heard through its hub, its start read.
	holder := connect("holder")
	ch := &busChannel{conn: holder, state: holder.State(approval.StateName), asked: make(chan approval.Ask, 4)}
	gate := holder.Gate(Gating{Channel: ch, Build: "fp1", Logf: t.Logf})
	ran := 0
	marked := func(seat string) func() (seatdef.Definition, bool) {
		return func() (seatdef.Definition, bool) {
			return seatdef.Definition{Seat: seat, Verbs: []seatdef.Verb{{Name: "wipe", NeedsApproval: true}, {Name: "approvals"}}}, true
		}
	}
	wipe := map[string]SeatHandler{"wipe": func(caller string, body json.RawMessage) (any, error) {
		mu.Lock()
		defer mu.Unlock()
		ran++
		return "wiped", nil
	}}
	if _, err := holder.AnswerSeat("wiper", marked("wiper"), wipe, Gated(gate)); err != nil {
		t.Fatal(err)
	}
	hub := NewEventHub(t.Logf)
	if err := HearDecided(gate, ch, "holder", hub); err != nil {
		t.Fatal(err)
	}
	hjs, _ := jetstream.New(holder.nc)
	if err := hub.Start(ctx, hjs, holder.account); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(hub.Stop)
	if err := gate.StartHolding(); err != nil {
		t.Fatal(err)
	}

	caller := connect("caller")
	type result struct {
		answer json.RawMessage
		err    error
	}
	callWipe := func(args map[string]string) chan result {
		done := make(chan result, 1)
		go func() {
			a, err := caller.Call("seat:wiper.wipe", args)
			done <- result{a, err}
		}()
		return done
	}
	decide := func(a approval.Ask, option string) {
		raw, _ := json.Marshal(approval.Record{State: "chosen", OfferedOn: []string{"telegram"},
			Warrant: &approval.Warrant{Ask: a.ID, Asker: "holder", Option: option, Channel: "telegram", At: time.Now().UTC().Format(time.RFC3339Nano)}})
		if _, err := records.Put(ctx, "holder."+a.ID, raw); err != nil {
			t.Fatal(err)
		}
		// The decided's body is not read beyond the ask's id: the gate reads the router's record.
		if _, err := js.Publish(ctx, ch.Decided("holder"), []byte(`{"ask":"`+a.ID+`","option":"approve"}`)); err != nil {
			t.Fatal(err)
		}
	}
	wait := func(done chan result) result {
		select {
		case r := <-done:
			return r
		case <-time.After(10 * time.Second):
			t.Fatal("the call's end was not heard within 10s")
			return result{}
		}
	}
	asked := func() approval.Ask {
		select {
		case a := <-ch.asked:
			return a
		case <-time.After(10 * time.Second):
			t.Fatal("no ask raised within 10s")
			return approval.Ask{}
		}
	}

	t.Run("asked, approved, run once, recorded, heard on the caller's gated-ended", func(t *testing.T) {
		done := callWipe(map[string]string{"dir": "scratch"})
		a := asked()
		decide(a, approval.OptionApprove)
		r := wait(done)
		var got approval.Answered
		if r.err != nil || json.Unmarshal(r.answer, &got) != nil || got.Answer != "wiped" || !strings.Contains(got.Approval, a.ID) {
			t.Fatalf("answer %s, %v", r.answer, r.err)
		}
		mu.Lock()
		defer mu.Unlock()
		if ran != 1 || len(warranted) != 1 || !strings.Contains(warranted[0], a.ID) || !strings.HasPrefix(warranted[0], "n1.holder ") {
			t.Errorf("ran %d, warranted %v", ran, warranted)
		}
	})
	t.Run("declined, nothing runs", func(t *testing.T) {
		done := callWipe(map[string]string{"dir": "other"})
		decide(asked(), approval.OptionDecline)
		if r := wait(done); r.err == nil || !strings.Contains(r.err.Error(), "declined") {
			t.Fatalf("answer %s, %v", r.answer, r.err)
		}
		mu.Lock()
		defer mu.Unlock()
		if ran != 1 {
			t.Errorf("ran %d", ran)
		}
	})
	t.Run("a waiting call's bound key is kept with its time to live, and its end is heard", func(t *testing.T) {
		// A-255: gate-unanswered rests on a bound key in the state approvals, never a clock wait.
		state := holder.State(approval.StateName)
		kept := make(chan string, 1)
		if err := state.Watch("unanswered.*", func(c stdio.StateChange) error {
			if c.Op == stdio.OpPut && !c.Current {
				kept <- c.Key
			}
			return nil
		}); err != nil {
			t.Fatal(err)
		}
		done := callWipe(map[string]string{"dir": "bound"})
		a := asked()
		var bound string
		select {
		case bound = <-kept:
		case <-time.After(10 * time.Second):
			t.Fatal("no bound key kept within 10s")
		}
		decide(a, approval.OptionDecline)
		wait(done)
		if e, _ := state.Get(bound); e != nil {
			t.Errorf("the bound %s of an ended call is kept", bound)
		}
		heard := make(chan string, 2)
		stop, err := heldBounds{state}.Watch("unanswered.*", func(k string) { heard <- k })
		if err != nil {
			t.Fatal(err)
		}
		defer stop()
		b := heldBounds{state}
		if err := b.CreateTTL("unanswered.gone", map[string]string{}, MinTTL); err != nil {
			t.Fatal(err)
		}
		if err := b.CreateTTL("unanswered.deleted", map[string]string{}, time.Hour); err != nil {
			t.Fatal(err)
		}
		_ = b.Delete("unanswered.deleted")
		select {
		case k := <-heard:
			if k != "unanswered.gone" {
				t.Errorf("heard %s", k)
			}
		case <-time.After(10 * time.Second):
			t.Fatal("the bound key's end was not heard within 10s")
		}
	})
	t.Run("failing closed: no gate, or no state approvals", func(t *testing.T) {
		if _, err := holder.AnswerSeat("bare", marked("bare"), wipe); err != nil {
			t.Fatal(err)
		}
		other := connect("other") // its state approvals is not on the bus
		g := other.Gate(Gating{Channel: &busChannel{conn: other, state: other.State(approval.StateName), asked: make(chan approval.Ask, 1)}, Build: "fp1"})
		if _, err := other.AnswerSeat("keeper", marked("keeper"), wipe, Gated(g)); err != nil {
			t.Fatal(err)
		}
		for _, seat := range []string{"bare", "keeper"} {
			if _, err := caller.Call("seat:"+seat+".wipe", map[string]string{"dir": "x"}); err == nil || !strings.Contains(err.Error(), "Not run") {
				t.Errorf("%s: %v", seat, err)
			}
		}
		mu.Lock()
		defer mu.Unlock()
		if ran != 1 {
			t.Errorf("ran %d", ran)
		}
	})
}