mesh-sdk/go/process/state_integration_test.go

//go:build integration

package process

import (
	"errors"
	"strings"
	"testing"
	"time"

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/kvlimit"
)

// 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a
// key, reads it, and updates it with the revision it read, then each succeeds and returns the new
// revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails
// as stdio.ErrChanged.
//
// 1790: and before that, the bucket is not on the bus when the process opens it (a delivery restarted it before the
// grants part created its new state): it is bound when the bus says the bucket was created, as the process's own
// account hears it, and the value written while it was missing is there.
func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)

	nc, err := nats.Connect(url, nats.UserInfo(sendUser, sendPassword), nats.CustomInboxPrefix("_INBOX."+sendUser))
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	awaited := AwaitBucket(ctxFor(t), js, nc, "mesh-mail_sends", "nothing is kept", t.Logf)
	appeared := make(chan struct{})
	awaited.Ready(func(jetstream.KeyValue) { close(appeared) })
	if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":false}`)); !errors.Is(err, ErrNotYet) {
		t.Fatalf("a write before the bucket exists: %v, want ErrNotYet", err)
	}
	if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":true}`)); !errors.Is(err, ErrNotYet) {
		t.Fatalf("a second write before the bucket exists: %v, want ErrNotYet", err)
	}
	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", LimitMarkerTTL: time.Second}); err != nil {
		t.Fatalf("the admin could not make the bucket: %v", err)
	}
	select {
	case <-appeared:
	case <-time.After(5 * time.Second):
		t.Fatal("the bucket was not bound within 5s of its creation, with no use since")
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	s := conn.State("sends")
	if kept, err := s.Get("hearing"); err != nil || kept == nil || string(kept.Value) != `{"heard":true}` {
		t.Fatalf("the value written while the bucket was missing: %+v, %v; want the latest", kept, err)
	}

	rev, err := s.Create("key", map[string]any{"v": 1})
	if err != nil {
		t.Fatalf("Create on a granted bucket: %v", err)
	}
	if rev == 0 {
		t.Fatal("Create returned no revision")
	}

	e, err := s.Get("key")
	if err != nil || e == nil {
		t.Fatalf("Get after Create: %+v, %v", e, err)
	}
	if e.Revision != rev || string(e.Value) != `{"v":1}` {
		t.Fatalf("Get read back %+v, want revision %d and {\"v\":1}", e, rev)
	}

	rev2, err := s.Update("key", map[string]any{"v": 2}, rev)
	if err != nil {
		t.Fatalf("Update at the revision read: %v", err)
	}
	if rev2 == rev {
		t.Fatal("Update did not return a new revision")
	}

	if _, err := s.Update("key", map[string]any{"v": 3}, rev); !errors.Is(err, stdio.ErrChanged) {
		t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err)
	}

	if _, err := s.Create("key", map[string]any{"v": 4}); !errors.Is(err, stdio.ErrChanged) {
		t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err)
	}

	// 1952 S6 (ADR 0564 clause 4): a key created with its own time to live is gone once it passes, on the bus itself.
	if _, err := s.CreateTTL("pulse", map[string]any{"v": 5}, time.Second); err != nil {
		t.Fatalf("CreateTTL on a bucket that takes keys with a time to live: %v", err)
	}
	deadline := time.Now().Add(5 * time.Second)
	for {
		e, err := s.Get("pulse")
		if err == nil && e == nil {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("the key created with a second to live still reads %+v, %v after 5s", e, err)
		}
		time.Sleep(100 * time.Millisecond)
	}
}

// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes
// back as an error naming the bucket. The bucket exists (made by the admin) so the only reason Create
// can fail is the missing grant, never "bucket not found"; and Conn's own ErrorHandler (conn.go
// withRefusal/publishRefused) catches the denied publish at once, so this returns promptly rather than
// after a timeout.
func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)
	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil {
		t.Fatalf("the admin could not make the bucket: %v", err)
	}

	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()

	s := conn.State("unrelated")
	_, err = s.Create("key", map[string]any{"v": 1})
	if err == nil {
		t.Fatal("writing to an ungranted bucket was not refused")
	}
	if !errors.Is(err, nats.ErrPermissionViolation) {
		t.Errorf("the refusal %v does not wrap nats.ErrPermissionViolation", err)
	}
	if !strings.Contains(err.Error(), "mesh-mail_unrelated") {
		t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
	}
}

// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put
// returns a new revision each time and Keys lists them sorted (none for a bucket with none).
func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)
	admins, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", MaxValueSize: 1024})
	if err != nil {
		t.Fatalf("the admin could not make the bucket: %v", err)
	}
	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	s := conn.State("sends")

	// Issue 774 (sweep item 5): a value over the bucket's largest is refused before it is put, naming the bucket, the key,
	// its size and the limit, as kvlimit.ErrTooLarge, so a writer stops on it; nothing reaches the bucket.
	big := strings.Repeat("x", 2000)
	for name, write := range map[string]func() (uint64, error){
		"put":    func() (uint64, error) { return s.Put("big", big) },
		"create": func() (uint64, error) { return s.Create("big", big) },
		"update": func() (uint64, error) { return s.Update("big", big, 1) },
	} {
		_, err := write()
		var tooLarge *kvlimit.TooLarge
		if !errors.Is(err, kvlimit.ErrTooLarge) || !errors.As(err, &tooLarge) || tooLarge.Limit != 1024 ||
			tooLarge.Bucket != "mesh-mail_sends" || tooLarge.Key != "big" || tooLarge.Size != 2002 || strings.Contains(err.Error(), "the bus:") {
			t.Errorf("a %s of 2002 bytes into a bucket of 1024: %v", name, err)
		}
	}
	// The bus's own refusal (JetStream 10054), as a writer that put past the check meets it, is the same refusal.
	_, err = admins.Put(ctxFor(t), "big", []byte(big))
	if err == nil || !errors.Is(kvlimit.Refused("mesh-mail_sends", "big", 2000, 1024, err), kvlimit.ErrTooLarge) {
		t.Errorf("the bus's refusal is not ErrTooLarge: %v", err)
	}

	if keys, err := s.Keys(); err != nil || len(keys) != 0 {
		t.Fatalf("Keys of an empty bucket: %v, %v", keys, err)
	}
	rev, err := s.Put("b", map[string]any{"v": 1})
	if err != nil || rev == 0 {
		t.Fatalf("Put on a granted bucket: %d, %v", rev, err)
	}
	rev2, err := s.Put("b", map[string]any{"v": 2})
	if err != nil || rev2 <= rev {
		t.Fatalf("Put over a key: %d, %v", rev2, err)
	}
	if _, err := s.Put("a", 1); err != nil {
		t.Fatal(err)
	}
	keys, err := s.Keys()
	if err != nil || strings.Join(keys, ",") != "a,b" {
		t.Fatalf("Keys: %v, %v, want a,b", keys, err)
	}
}

// 1840/S2: given a process connection granted its bucket, which already holds two keys and a deleted one,
// when it watches, then it gets the two current values marked Current, and then a later put and a delete in
// order; a narrowed watch gets only the keys it names.
func TestStateWatchOnAGrantedBucket(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)
	kv, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"})
	if err != nil {
		t.Fatalf("the admin could not make the bucket: %v", err)
	}
	ctx := ctxFor(t)
	for k, v := range map[string]string{"a.x": `1`, "a.y": `2`, "b.z": `3`, "gone": `4`} {
		if _, err := kv.Put(ctx, k, []byte(v)); err != nil {
			t.Fatal(err)
		}
	}
	if err := kv.Delete(ctx, "gone"); err != nil {
		t.Fatal(err)
	}
	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	s := conn.State("sends")

	all := make(chan stdio.StateChange, 16)
	if err := s.Watch("", func(c stdio.StateChange) error { all <- c; return nil }); err != nil {
		t.Fatalf("Watch on a granted bucket: %v", err)
	}
	current := map[string]bool{}
	for len(all) > 0 {
		c := <-all
		if !c.Current || c.Op != "put" {
			t.Fatalf("a current value arrived as %+v", c)
		}
		current[c.Key] = true
	}
	if len(current) != 3 || current["gone"] {
		t.Fatalf("current values %v, want a.x, a.y, b.z and not gone", current)
	}

	narrow := make(chan stdio.StateChange, 16)
	if err := s.Watch("a.*", func(c stdio.StateChange) error { narrow <- c; return nil }); err != nil {
		t.Fatal(err)
	}
	if len(narrow) != 2 {
		t.Fatalf("a narrowed watch handed %d current values, want 2", len(narrow))
	}
	// the current values come in the order of their last revision, not of their keys: taken as a set.
	narrowCurrent := map[string]bool{}
	for len(narrow) > 0 {
		c := <-narrow
		if !c.Current || c.Op != "put" {
			t.Fatalf("a narrowed watch's current value arrived as %+v", c)
		}
		narrowCurrent[c.Key] = true
	}
	if !narrowCurrent["a.x"] || !narrowCurrent["a.y"] {
		t.Fatalf("a narrowed watch's current values %v, want a.x and a.y", narrowCurrent)
	}

	if _, err := s.Put("a.x", 5); err != nil {
		t.Fatal(err)
	}
	if err := kv.Delete(ctx, "a.y"); err != nil {
		t.Fatal(err)
	}
	for _, want := range []struct{ key, op string }{{"a.x", "put"}, {"a.y", "delete"}} {
		select {
		case c := <-all:
			if c.Key != want.key || c.Op != want.op || c.Current {
				t.Fatalf("got %+v, want %s %s, not current", c, want.op, want.key)
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("no %s of %s", want.op, want.key)
		}
	}
	if _, err := s.Put("b.q", 6); err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{"a.x", "a.y"} {
		select {
		case c := <-narrow:
			if c.Key != want || c.Current {
				t.Fatalf("narrowed watch got %+v, want the change of %s", c, want)
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("narrowed watch missed %s", want)
		}
	}
	select {
	case c := <-narrow:
		t.Fatalf("narrowed watch got %+v for a key it did not name", c)
	case <-time.After(500 * time.Millisecond):
	}

	// ADR 0571 clause 1: another module's log, read on the process's own credential with exactly the read grants: the
	// newest per key, every entry from a time, and a watch from the newest that then hears an appended entry.
	if _, err := adminJetStream.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_mesh-delivery-builder_build-log",
		Subjects: []string{"mesh.log.mesh-delivery-builder.build-log.>"}, AllowDirect: true}); err != nil {
		t.Fatal(err)
	}
	for _, k := range []string{"letta", "gitea", "letta"} {
		if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log."+k, []byte(`{"module":"`+k+`"}`)); err != nil {
			t.Fatal(err)
		}
		time.Sleep(5 * time.Millisecond)
	}
	builds := conn.ReadLog("mesh-delivery-builder", "build-log")
	newest, err := builds.Newest("")
	if err != nil || len(newest) != 2 || newest[0].Key != "gitea" || newest[1].Sequence != 3 {
		t.Fatalf("the newest per key are %+v: %v", newest, err)
	}
	since, err := builds.Since(newest[0].At, "letta")
	if err != nil || len(since) != 1 || since[0].Sequence != 3 {
		t.Fatalf("letta since gitea's build is %+v: %v", since, err)
	}
	heard := make(chan stdio.LogEntry, 4)
	if err := builds.Watch("letta", stdio.LogFromNewest, func(e stdio.LogEntry) error { heard <- e; return nil }); err != nil || len(heard) != 1 {
		t.Fatalf("a watch from the newest handed %d at its start: %v", len(heard), err)
	}
	<-heard
	if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log.letta", []byte(`{}`)); err != nil {
		t.Fatal(err)
	}
	select {
	case e := <-heard:
		if e.Sequence != 4 || e.Key != "letta" {
			t.Fatalf("the watch heard %+v", e)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the watch did not hear the appended entry")
	}
}

// 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket.
func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) {
	url, account := grantedSendProcess(t)
	_, adminJetStream := adminJS(t, url)
	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil {
		t.Fatalf("the admin could not make the bucket: %v", err)
	}
	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	err = conn.State("unrelated").Watch("", func(stdio.StateChange) error { return nil })
	if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") {
		t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err)
	}
}

// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and
// deleting it again is no error.
func TestStateDeleteOnAGrantedBucket(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 bucket: %v", err)
	}
	conn, err := Connect(account)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	s := conn.State("sends")
	if _, err := s.Put("k", 1); err != nil {
		t.Fatal(err)
	}
	if err := s.Delete("k"); err != nil {
		t.Fatalf("Delete on a granted bucket: %v", err)
	}
	if e, err := s.Get("k"); err != nil || e != nil {
		t.Fatalf("Get after Delete: %+v, %v", e, err)
	}
	if err := s.Delete("k"); err != nil {
		t.Fatalf("Delete of an absent key: %v", err)
	}
}