mesh-tools/node-tools/cmd/node-tools/approvals_integration_test.go

//go:build integration

package main

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

	"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-tools/node-tools/internal/approval"
	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// marked is the fixture marked module's catalog half: its build src2:1 marks its tool wipe (design 46 7-approval-mark §1).
type marked struct{}

var wipeEntry = catalogseat.CatalogEntry{Module: "shell", Fp: "src2:1", Manifest: map[string]any{"needs-approval": []any{"wipe"}}}

func (marked) Ready() error     { return nil }
func (marked) Revision() string { return "7" }
func (marked) Served(fp string) (catalogseat.CatalogEntry, bool, string) {
	return wipeEntry, fp == wipeEntry.Fp, "not-found"
}
func (marked) Newest(m string) (catalogseat.CatalogEntry, bool) { return wipeEntry, m == "shell" }
func (marked) Definitions() map[string]seatdef.Definition       { return map[string]seatdef.Definition{} }
func (marked) Entries() []catalogseat.CatalogEntry              { return []catalogseat.CatalogEntry{wipeEntry} }

// bed is the runner's gate wired as main wires it, over the runner's own connection to a real bus, with stand-ins of the
// operator channel's router (it takes each ask into its record asks, as the router writes it) and of the hand-act seat.
type bed struct {
	t     *testing.T
	conn  *bus.Conn
	asks  nats.KeyValue
	taken chan struct{}
	done  chan struct{}

	mu       sync.Mutex
	asked    []seat.Ask
	recorded []string
	holders  map[string][]byte
}

func newBed(t *testing.T, bindState bool) *bed {
	t.Helper()
	mesh := mt.New(t)
	nc, err := nats.Connect(mt.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := nc.JetStream()
	if _, err := js.AddStream(&nats.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"},
		Retention: nats.WorkQueuePolicy}); err != nil {
		t.Fatal(err)
	}
	b := &bed{t: t, asks: mesh.Bucket(t, "messenger_asks"), taken: make(chan struct{}, 1), done: make(chan struct{}),
		holders: map[string][]byte{"operator-channel": []byte(`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"messenger"}]}`)}}
	t.Cleanup(func() { close(b.done) })
	mesh.Bucket(t, "node-tools_approvals")
	mesh.BucketKeyTTL(t, "node-tools_bound")

	// The router's stand-in: each ask accepted under the marked module's name is taken into its record, state open.
	sub, err := js.PullSubscribe("mesh.seat.operator-channel.accept.>", "router", nats.BindStream("SEAT_OPERATOR_CHANNEL"))
	if err != nil {
		t.Fatal(err)
	}
	go func() {
		for {
			select {
			case <-b.done:
				return
			default:
			}
			msgs, _ := sub.Fetch(1, nats.MaxWait(200*time.Millisecond))
			for _, m := range msgs {
				var a seat.Ask
				if err := json.Unmarshal(m.Data, &a); err == nil {
					module := strings.TrimPrefix(m.Subject, "mesh.seat.operator-channel.accept.ask.")
					b.mu.Lock()
					b.asked = append(b.asked, a)
					b.mu.Unlock()
					b.write(seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", Headline: a.Headline,
						Expires: a.Expires, Opened: operatorchannelclient.FormatTime(time.Now()), OfferedOn: []string{"telegram"}})
					select {
					case b.taken <- struct{}{}:
					default:
					}
				}
				_ = m.Ack()
			}
		}
	}()
	// The hand-act seat's stand-in: warranted {asker, ask} answered, each one kept.
	if _, err := nc.Subscribe("mesh.seat.mesh-hand-acts.>", func(m *nats.Msg) {
		b.mu.Lock()
		b.recorded = append(b.recorded, string(m.Data))
		b.mu.Unlock()
		_ = m.Respond([]byte(`{"result":{"recorded":true}}`))
	}); err != nil {
		t.Fatal(err)
	}

	runner := mt.MembershipOf("node-tools", "n1", false, nil)
	if bindState {
		runner.State = []bus.StateIssued{{Name: approval.StateName, Bucket: "node-tools_approvals", Writes: true},
			{Name: BoundState, Bucket: "node-tools_bound", Writes: true}}
	}
	shell := mt.MembershipOf("shell", "n1", false, nil)
	shell.SeatTraffic = &bus.SeatTraffic{Publish: []string{seat.AcceptSubjectAsk("shell")},
		Records: []string{"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.shell.>"}}
	mesh.Issue(t, runner)
	mesh.Issue(t, shell)
	b.conn, err = bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "n1"})
	if err != nil {
		t.Fatal(err)
	}
	b.conn.Logf = func(string, ...any) {}
	t.Cleanup(b.conn.Close)
	b.conn.Follow("node-tools")
	b.conn.Follow("shell")
	mt.Until(t, func() error {
		if b.conn.Membership("node-tools") == nil || b.conn.Membership("shell") == nil {
			return errors.New("not issued yet")
		}
		return nil
	})
	return b
}

// gate is a runner's gate as main makes it; record says how the run's recording goes beside the call (nil: at once).
func (b *bed) gate(record func(func())) *approval.Gate {
	if record == nil {
		record = func(f func()) { f() }
	}
	g := approval.New(approval.Deps{Node: "n1", Runner: "node-tools", Catalog: marked{}, Logf: b.t.Logf,
		State: approvalsState{conn: b.conn, module: "node-tools"}, Publish: publishAsk(b.conn), Record: routerRecord(b.conn),
		Warranted: warranted(b.conn), Bounds: boundState{conn: b.conn, module: "node-tools"}, Go: record,
		Holder: approval.HoldersIn(func() (map[string][]byte, error) {
			b.mu.Lock()
			defer b.mu.Unlock()
			return b.holders, nil
		}),
		// The pause while the router takes the ask ends when the stand-in has taken one: no sleeping.
		After: func(time.Duration) <-chan time.Time {
			c := make(chan time.Time, 1)
			go func() {
				select {
				case <-b.taken:
				case <-b.done:
				}
				c <- time.Now()
			}()
			return c
		}})
	_ = g.Unserved("shell", "src2:1", []string{"wipe"})
	return g
}

func (b *bed) write(rec seat.AsksByAskerIDValue) {
	raw, _ := json.Marshal(rec)
	if _, err := b.asks.Put(rec.Key, raw); err != nil {
		b.t.Fatal(err)
	}
}

// decide writes the router's record of the newest ask as the router writes the operator's choice: chosen, with its warrant.
func (b *bed) decide(option string) seat.Ask {
	b.mu.Lock()
	a := b.asked[len(b.asked)-1]
	b.mu.Unlock()
	e, err := b.asks.Get("shell." + a.ID)
	if err != nil {
		b.t.Fatal(err)
	}
	var rec seat.AsksByAskerIDValue
	_ = json.Unmarshal(e.Value(), &rec)
	rec.State = "chosen"
	rec.Warrant = &seat.Warrant{Ask: a.ID, Asker: "shell", Outcome: operatorchannelclient.OutcomeChosen, Option: option,
		Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(time.Now()), AskDigest: operatorchannelclient.Digest(a),
		Channel: "telegram", By: &seat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}}
	b.write(rec)
	return a
}

func wipe(args string) approval.Call {
	return approval.Call{Address: "shell.wipe", Module: "shell", Name: "wipe", Fingerprint: "src2:1", Caller: "agent",
		Args: json.RawMessage(args), Description: "Wipes a scratch directory."}
}

// Design 46 7-approval-mark §22: the runner's gate against the real bus with a fixture marked module: a call asked and taken
// by the operator channel's stand-in, the operator's Approve written as the router's record, the call again run once and
// recorded in the hand-act log, mesh_approvals' list, a restart between the run and warranted's answer recorded once; a
// Decline runs nothing; a gate missing a part of its check runs nothing.
func TestAMarkedCallOnTheBusIsAskedRunOnTheApprovalAndRecorded(t *testing.T) {
	t.Run("approved: asked, run once, recorded, listed", func(t *testing.T) {
		b := newBed(t, true)
		g := b.gate(nil)
		runs := 0
		run := func() (any, error) { runs++; return "wiped", nil }
		if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil ||
			!strings.HasPrefix(err.Error(), "Not run: this needs the operator's approval. Asked on telegram at ") || runs != 0 {
			t.Fatalf("first call: %v, runs %d", err, runs)
		}
		if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "still waiting for the operator's answer") {
			t.Fatalf("before the answer: %v", err)
		}
		a := b.decide(approval.OptionApprove)
		got, err := g.Run(wipe(`{"dir":"scratch"}`), run)
		answered, ok := got.(approval.Answered)
		if err != nil || !ok || answered.Answer != "wiped" || !strings.Contains(answered.Approval, "ask "+a.ID) || runs != 1 {
			t.Fatalf("on the approval: %+v, %v, runs %d", got, err, runs)
		}
		if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || runs != 1 {
			t.Fatalf("the warrant ran twice: %v, runs %d", err, runs)
		}
		b.mu.Lock()
		recorded := strings.Join(b.recorded, "\n")
		b.mu.Unlock()
		if !strings.Contains(recorded, a.ID) || !strings.Contains(recorded, "shell") {
			t.Fatalf("hand-act log heard %q", recorded)
		}
		list, err := g.List()
		if err != nil || len(list) == 0 || list[len(list)-1].Ask != a.ID || list[len(list)-1].State != "run, recorded" {
			t.Fatalf("mesh_approvals lists %+v, %v", list, err)
		}
	})
	t.Run("a restart between the run and warranted's answer records the run once", func(t *testing.T) {
		b := newBed(t, true)
		stopped := b.gate(func(func()) {}) // the runner stops before warranted answers
		runs := 0
		run := func() (any, error) { runs++; return "wiped", nil }
		_, _ = stopped.Run(wipe(`{"dir":"tmp"}`), run)
		a := b.decide(approval.OptionApprove)
		if _, err := stopped.Run(wipe(`{"dir":"tmp"}`), run); err != nil || runs != 1 {
			t.Fatalf("run: %v, runs %d", err, runs)
		}
		if list, _ := stopped.List(); len(list) != 1 || list[0].State != "run, not yet recorded" {
			t.Fatalf("before the restart: %+v", list)
		}
		started := b.gate(nil)
		started.Recover()
		started.Recover()
		b.mu.Lock()
		n := len(b.recorded)
		b.mu.Unlock()
		if list, _ := started.List(); n != 1 || len(list) != 1 || list[0].Ask != a.ID || list[0].State != "run, recorded" || runs != 1 {
			t.Fatalf("after the restart: recorded %d, %+v, runs %d", n, list, runs)
		}
	})
	t.Run("declined: refused, nothing runs", func(t *testing.T) {
		b := newBed(t, true)
		g := b.gate(nil)
		runs := 0
		run := func() (any, error) { runs++; return nil, nil }
		_, _ = g.Run(wipe(`{"dir":"scratch"}`), run)
		b.decide(approval.OptionDecline)
		if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "the operator declined ask") || runs != 0 {
			t.Fatalf("declined: %v, runs %d", err, runs)
		}
	})
	t.Run("failing closed: no state bound, or nobody to ask, runs nothing and asks nothing", func(t *testing.T) {
		unbound := newBed(t, false)
		runs := 0
		run := func() (any, error) { runs++; return nil, nil }
		if _, err := unbound.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil ||
			!strings.Contains(err.Error(), "binds no state approvals") || runs != 0 || len(unbound.asked) != 0 {
			t.Fatalf("no state: %v, runs %d", err, runs)
		}
		b := newBed(t, true)
		b.holders = map[string][]byte{}
		if _, err := b.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil ||
			!strings.Contains(err.Error(), "no module holds operator-channel") || runs != 0 {
			t.Fatalf("no holder: %v, runs %d", err, runs)
		}
		if got, err := b.gate(nil).Run(approval.Call{Address: "shell.tidy", Module: "shell", Name: "tidy", Fingerprint: "src2:1",
			Args: json.RawMessage(`{}`)}, func() (any, error) { return "tidied", nil }); err != nil || got != "tidied" {
			t.Fatalf("an unmarked tool: %v, %v", got, err)
		}
	})
}