mesh-catalog/modules/messenger/cmd/messenger/condition_bus_test.go

//go:build integration

package main

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

	"git.novox.be/novox/mesh-sdk/go/asks"
	"github.com/nats-io/nats.go"
)

// A condition said on the operator channel's one subject over the real bus (novox/hq ADR 0437 §9, issue 1218),
// in the bed of bus_test.go. The fake runtime enforces no grant: that a module may say a condition only under its
// own name is the broker's, proven in the controller (issue 1216).

// conditionBed starts the router with the desk able to send, hands it the conditions said on the one subject as
// work, and keeps what the desk is sent and whether each work item was acknowledged.
type conditionBed struct {
	r    *busRuntime
	mu   sync.Mutex
	desk []asks.Message
	acks []string // the answer of the router to each work item: "" when acknowledged
}

func newConditionBed(t *testing.T) *conditionBed {
	t.Helper()
	b := &conditionBed{r: newBusRuntime(t)}
	b.r.setKinds(map[string][]string{"desktop": {"choice", "edit"}})
	if _, err := b.r.nc.Subscribe("mesh.seat."+asks.ChannelSeat+".accept.*.desktop", func(m *nats.Msg) {
		var msg asks.Message
		if json.Unmarshal(m.Data, &msg) == nil {
			b.mu.Lock()
			b.desk = append(b.desk, msg)
			b.mu.Unlock()
		}
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := b.r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.condition.*", func(m *nats.Msg) {
		answer, err := b.r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}})
		var a struct {
			Error *struct {
				Message string `json:"message"`
			} `json:"error"`
		}
		_ = json.Unmarshal(answer, &a)
		res := ""
		if err != nil {
			res = err.Error()
		} else if a.Error != nil {
			res = a.Error.Message
		}
		b.mu.Lock()
		b.acks = append(b.acks, res)
		b.mu.Unlock()
	}); err != nil {
		t.Fatal(err)
	}
	b.forwardEvents(t)
	b.r.start(t.TempDir())
	b.r.until("the bundle takes work and read the kinds", 20*time.Second, func() bool {
		b.r.mu.Lock()
		defer b.r.mu.Unlock()
		return b.r.taking && b.r.kindsRead > 0
	})
	b.r.until("the bundle subscribed to the channels' word", 20*time.Second, func() bool {
		b.r.mu.Lock()
		defer b.r.mu.Unlock()
		return b.r.subscribed
	})
	standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()})
	if err := b.r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "desktop"), standing); err != nil {
		t.Fatal(err)
	}
	time.Sleep(time.Second)
	return b
}

// forwardEvents hands the bundle the channels' word, as bus_test.go's forward does for it.
func (b *conditionBed) forwardEvents(t *testing.T) {
	t.Helper()
	if _, err := b.r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) {
		key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1)
		_, _ = b.r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}})
	}); err != nil {
		t.Fatal(err)
	}
}

func (b *conditionBed) say(t *testing.T, publisher string, c asks.Condition) {
	t.Helper()
	body, err := asks.EncodeCondition(c)
	if err != nil {
		t.Fatal(err)
	}
	subject, err := asks.ConditionSubject(publisher)
	if err != nil {
		t.Fatal(err)
	}
	if err := b.r.nc.Publish(subject, body); err != nil {
		t.Fatal(err)
	}
}

func (b *conditionBed) record(key string) *Record {
	e, err := b.r.bucket("open").Get(context.Background(), kvKey(key))
	if err != nil {
		return nil
	}
	var rec Record
	if json.Unmarshal(e.Value(), &rec) != nil {
		return nil
	}
	return &rec
}

func (b *conditionBed) told() []asks.Message {
	b.mu.Lock()
	defer b.mu.Unlock()
	return append([]asks.Message(nil), b.desk...)
}

func (b *conditionBed) acknowledged(n int) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.acks) != n {
		return false
	}
	for _, a := range b.acks {
		if a != "" {
			return false
		}
	}
	return true
}

func busCondition(key, id, headline, needs string) asks.Condition {
	verdict := asks.NothingToDo
	if needs != "" {
		verdict = asks.NeedsYou + needs
	}
	return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id},
		State: asks.StateRaised, Summary: "the merge waits on a check", Headline: headline, Needs: needs,
		Explanation: verdict + " It waits on a check."}
}

// 1218/AC1 and 1218/AC2: a raised condition said on the one subject reaches the desk, is recorded and its work item
// is acknowledged; its cleared form ends it. The bundle starts with a short grace (bus_test.go).
func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *testing.T) {
	b := newConditionBed(t)
	b.say(t, "mesh-delivery-teller", busCondition("merge.d-7.stalled", "d-7", "A merge waits on a check", "release it, or stop it."))
	// 1218/AC1: recorded, acknowledged, and told to the desk.
	b.r.until("the condition is recorded open", 30*time.Second, func() bool {
		rec := b.record("merge.d-7.stalled")
		return rec != nil && rec.Origin == "condition" && rec.Cleared.IsZero()
	})
	b.r.until("the work item is acknowledged", 30*time.Second, func() bool { return b.acknowledged(1) })
	b.r.until("the desk is sent the condition", 15*time.Second, func() bool {
		for _, m := range b.told() {
			if strings.Contains(m.Title+m.Body, "A merge waits on a check") {
				return true
			}
		}
		return false
	})
	time.Sleep(3 * time.Second)
	if n := b.nAcks(); n != 1 {
		t.Errorf("the work item was offered %d times, want once: it was not acknowledged", n)
	}
	// 1218/AC2: the cleared form, on the same subject, ends it.
	b.say(t, "mesh-delivery-teller", asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared})
	b.r.until("the condition is recorded cleared", 30*time.Second, func() bool {
		rec := b.record("merge.d-7.stalled")
		return rec == nil || !rec.Cleared.IsZero()
	})
}

func (b *conditionBed) nAcks() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return len(b.acks)
}

// 1218/AC3: a warning whose needs is empty is recorded and reaches no channel. A second warning that needs the
// operator, said after it, is told: when it is, the quiet one's turn has passed and it was not sent.
func TestAWarningThatNeedsNothingReachesNoChannelOverTheBus(t *testing.T) {
	b := newConditionBed(t)
	b.say(t, "mesh-delivery-teller", busCondition("merge.d-8.quiet", "d-8", "A quiet merge note", ""))
	b.r.until("the quiet warning is recorded", 30*time.Second, func() bool { return b.record("merge.d-8.quiet") != nil })
	b.say(t, "mesh-delivery-teller", busCondition("merge.d-9.loud", "d-9", "A loud merge note", "release it, or stop it."))
	b.r.until("the warning that needs the operator is sent to the desk", 15*time.Second, func() bool {
		for _, m := range b.told() {
			if strings.Contains(m.Title+m.Body, "A loud merge note") {
				return true
			}
		}
		return false
	})
	for _, m := range b.told() {
		if strings.Contains(m.Title+m.Body, "A quiet merge note") {
			t.Errorf("a warning that needs nothing was sent: %+v", m)
		}
	}
}