mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/ledger_bus_test.go

//go:build integration

package main

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// jsKV is one of the module's states on a real JetStream bucket.
func jsonMarshal(v any) ([]byte, error) { return json.Marshal(v) }

type jsKV struct {
	kv jetstream.KeyValue
}

func changed(err error) error {
	if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && strings.Contains(err.Error(), "wrong last sequence")) {
		return stdio.ErrChanged
	}
	return err
}

func (k jsKV) Get(key string) (*stdio.StateEntry, error) {
	e, err := k.kv.Get(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
}
func (k jsKV) Put(key string, v any) (uint64, error) {
	b, _ := jsonMarshal(v)
	return k.kv.Put(context.Background(), key, b)
}
func (k jsKV) Create(key string, v any) (uint64, error) {
	b, _ := jsonMarshal(v)
	r, err := k.kv.Create(context.Background(), key, b)
	return r, changed(err)
}
func (k jsKV) Update(key string, v any, rev uint64) (uint64, error) {
	b, _ := jsonMarshal(v)
	r, err := k.kv.Update(context.Background(), key, b, rev)
	return r, changed(err)
}
func (k jsKV) Delete(key string) error { return k.kv.Delete(context.Background(), key) }
func (k jsKV) Keys() ([]string, error) {
	keys, err := k.kv.Keys(context.Background())
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

type jsStates struct {
	mu sync.Mutex
	js jetstream.JetStream
	by map[string]jsKV
	t  *testing.T
}

func (s *jsStates) State(name string) KV {
	s.mu.Lock()
	defer s.mu.Unlock()
	if k, ok := s.by[name]; ok {
		return k
	}
	kv, err := s.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: "mesh_knowledge_" + name, History: 1})
	if err != nil {
		s.t.Fatal(err)
	}
	k := jsKV{kv}
	s.by[name] = k
	return k
}

// The knowledge base running whole on the ledger against a real bus (to-be 55 §10, AC57): the states are JetStream buckets,
// its events are published under their message ids to a stream that drops a repeat, the ledger is played with the module's
// own types. It moves its records, proposes, is reviewed and accepts a record, loses an event between an operation and its send,
// restarts, sends what the stop lost exactly once, builds its index and checks the database within 40 seconds, and is healthy.
func TestTheKnowledgeBaseRunsWholeOnTheLedgerOverARealBus(t *testing.T) {
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		t.Fatal("no bus within 30s")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := jetstream.New(nc)
	ctx := context.Background()
	events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "KB_EVENTS", Subjects: []string{"kb.event.>"}, Duplicates: 24 * time.Hour})
	if err != nil {
		t.Fatal(err)
	}
	states := &jsStates{js: js, by: map[string]jsKV{}, t: t}
	publish := func(event string, body any, id string) error {
		b, _ := jsonMarshal(body)
		_, err := js.Publish(ctx, "kb.event."+event, b, jetstream.WithMsgID(id))
		return err
	}
	fakeBus := newFakeLedger(t) // the ledger of the tool runner, played with the module's own types
	database := newMemStore()
	seed := imported(t, database)
	seed.accepted(t, "author@g14", valid(seed.reserved(t, "author@g14")))

	start := func() (*world, *movableStore) {
		led := newLedgerStore(fakeBus, states)
		m := newMovableStore(database, led, states)
		w := newWorld(t, m, fixture())
		w.k.States = states
		w.k.EmitID = publish
		w.tracker.next[KindDecision] = 400
		return w, m
	}
	w, m := start()
	moved, err := m.Move(ctx, MoveAsk{Commit: true}, nil)
	if err != nil || !moved.Switched {
		t.Fatalf("%v %+v", err, moved)
	}
	// A record is proposed, reviewed in parts and accepted, writing through the ledger and saying its events on the bus.
	n := w.reserved(t, "author@g14")
	w.accepted(t, "author@g14", valid(n))
	if info, _ := events.Info(ctx); info.State.Msgs == 0 {
		t.Fatal("no event reached the bus")
	}
	// An event lost between the operation and its send: the bus is cut, the operation is in the history.
	cut := w.k.EmitID
	w.k.EmitID = func(string, any, string) error { return errors.New("the bus is cut") }
	lost := w.reserved(t, "author@g14")
	if w.k.sentUpTo(Key{KindDecision, lost}) != 0 {
		t.Fatal("an event never sent was marked sent")
	}
	w.k.EmitID = cut
	before, _ := events.Info(ctx)

	// Restart: new objects over the same ledger and buckets. The start builds the index, checks, and sends what the stop lost.
	began := time.Now()
	w2, m2 := start()
	if m2.Source() != SourceLedger {
		t.Fatalf("a restart serves from %s: the switch key was not kept", m2.Source())
	}
	if err := m2.led.BuildIndex(ctx); err != nil {
		t.Fatal(err)
	}
	if diffs, err := m2.Check(ctx); err != nil || len(diffs) != 0 {
		t.Fatalf("%v %v", err, diffs)
	}
	sent, err := w2.k.ResendEvents(ctx)
	if err != nil || sent != 1 {
		t.Fatalf("a restart sent %d events: %v", sent, err)
	}
	if took := time.Since(began); took > 40*time.Second {
		t.Fatalf("the start took %s; at most 40 s", took)
	}
	after, _ := events.Info(ctx)
	if after.State.Msgs != before.State.Msgs+1 {
		t.Fatalf("the bus holds %d messages, was %d: the lost event is sent once", after.State.Msgs, before.State.Msgs)
	}
	// Sent again under the same id (the mark lost too), the bus drops the repeat.
	_ = states.State(sentState).Delete(sentKey(Key{KindDecision, lost}))
	_, _ = w2.k.ResendEvents(ctx)
	again, _ := events.Info(ctx)
	if again.State.Msgs != after.State.Msgs {
		t.Fatalf("a repeat was kept: %d messages, was %d", again.State.Msgs, after.State.Msgs)
	}
	// The records are served, searched and healthy.
	if e, err := w2.k.Store.Get(ctx, Key{KindDecision, n}); err != nil || e == nil || !e.Current.Accepted() {
		t.Fatalf("%v %v", err, e)
	}
	if hits, err := m2.Search(ctx, "merge window", 3); err != nil || len(hits) == 0 {
		t.Fatalf("%v %v", err, hits)
	}
	fakeBus.fill.MaxBytes = 512 << 20
	if h := w2.k.ledgerHealth(HealthAnswer{State: "healthy", Healthy: true}); !h.Healthy || !strings.Contains(h.Why, "served from the ledger") {
		t.Fatalf("%+v", h)
	}
}