mesh-tools/node-tools/internal/bus/bindings_test.go

package bus_test

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"strings"
	"testing"

	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

// 1966/AC12: given a bundle whose membership binds objects `files`, when it puts and gets `a`, then it reads back what it
// put; a local name it was not bound is refused with the reason, and so is another module's binding and a binding of
// another storage type. A stream and a work queue bound alike answer their read-back, fill and size.
func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) {
	mesh := mt.New(t)
	nc, err := nats.Connect(mt.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := jetstream.New(nc)
	ctx := context.Background()
	if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_files"}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "m_events", Subjects: []string{"mesh.state.m.events.>"}}); err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{"mesh.state.m.events.one", "mesh.state.m.events.two"} {
		if _, err := js.Publish(ctx, s, []byte(`{"n":1}`)); err != nil {
			t.Fatal(err)
		}
	}
	m := mt.MembershipOf("m", "anchor", false, nil)
	m.Bindings = []membership.BusBinding{
		{Local: "files", Storage: catalogue.StorageObjectStore, Template: catalogue.TemplateObjects, ObjectStore: "m_files"},
		{Local: "events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, Stream: "m_events",
			Subject: "mesh.state.m.events"},
	}
	other := mt.MembershipOf("other", "anchor", false, nil)
	mesh.Issue(t, m)
	mesh.Issue(t, other)
	conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
	if err != nil {
		t.Fatal(err)
	}
	conn.Logf = func(string, ...any) {}
	t.Cleanup(conn.Close)
	conn.Follow("m")
	conn.Follow("other")
	mt.Until(t, func() error {
		if conn.Membership("m") == nil || conn.Membership("other") == nil {
			return errors.New("not issued yet")
		}
		return nil
	})

	if _, err := conn.ObjectsPut("m", buscalls.ObjectsPutRequest{Objects: "files", Name: "a", Data: []byte("what was put")}); err != nil {
		t.Fatal(err)
	}
	got, err := conn.ObjectsGet("m", buscalls.ObjectsNameRequest{Objects: "files", Name: "a"})
	if err != nil || got == nil || string(got.Data) != "what was put" || got.Info.Name != "a" {
		t.Fatalf("read back %+v, %v", got, err)
	}
	// Design 47 G §518: a value over one chunk is streamed into the store and read back out of it, its description kept.
	big := strings.Repeat("0123456789", 120<<10)
	info, err := conn.ObjectsPutStream(ctx, "m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"}, "streamed", strings.NewReader(big))
	if err != nil || info.Size != uint64(len(big)) || info.Description != "streamed" {
		t.Fatalf("streamed put %+v, %v", info, err)
	}
	opened, rd, found, err := conn.ObjectsOpen(ctx, "m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"})
	if err != nil || !found || opened.Size != uint64(len(big)) {
		t.Fatalf("opened %+v %v %v", opened, found, err)
	}
	if back, _ := io.ReadAll(rd); string(back) != big {
		t.Fatalf("read back %d bytes, want %d", len(back), len(big))
	}
	_ = rd.Close()
	if err := conn.ObjectsDelete("m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"}); err != nil {
		t.Fatal(err)
	}
	if listed, err := conn.ObjectsList("m", buscalls.ObjectsRequest{Objects: "files"}); err != nil || len(listed.Objects) != 1 {
		t.Fatalf("listed %+v, %v", listed, err)
	}
	page, err := conn.StreamRead("m", buscalls.StreamReadRequest{Stream: "events", Limit: 1})
	if err != nil || len(page.Messages) != 1 || page.Next != 2 || page.Last != 2 {
		t.Fatalf("read %+v, %v", page, err)
	}
	if fill, err := conn.StreamFill("m", buscalls.StreamRequest{Stream: "events"}); err != nil || fill.Count != 2 {
		t.Fatalf("fill %+v, %v", fill, err)
	}

	// A-42: an event said on the module's own stream lands on the binding's subject plus the event, acknowledged, and a
	// second say with the same id is the same message.
	said, err := conn.StreamPublish("m", buscalls.StreamPublishRequest{Stream: "events", Event: "scrub.requested",
		Body: json.RawMessage(`{"pool":"tank"}`), ID: "id-1"})
	if err != nil || said.Sequence != 3 {
		t.Fatalf("said %+v, %v", said, err)
	}
	if again, err := conn.StreamPublish("m", buscalls.StreamPublishRequest{Stream: "events", Event: "scrub.requested",
		Body: json.RawMessage(`{"pool":"tank"}`), ID: "id-1"}); err != nil || again.Sequence != 3 {
		t.Fatalf("said again %+v, %v", again, err)
	}
	if page, err := conn.StreamRead("m", buscalls.StreamReadRequest{Stream: "events", From: 3,
		Subjects: "mesh.state.m.events.scrub.requested"}); err != nil || len(page.Messages) != 1 ||
		string(page.Messages[0].Body) != `{"pool":"tank"}` {
		t.Fatalf("read back %+v, %v", page, err)
	}

	for _, c := range []struct {
		name, module, refusal string
		call                  func(module string) error
	}{
		{"a local name not bound", "m", `bound no object-store called "photos"`, func(module string) error {
			_, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "photos", Name: "a"})
			return err
		}},
		{"another module's binding", "other", `other was bound no object-store called "files"`, func(module string) error {
			_, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "files", Name: "a"})
			return err
		}},
		{"a binding of another storage type", "m", `"events" is bound with the storage type stream, not object-store`, func(module string) error {
			_, err := conn.ObjectsList(module, buscalls.ObjectsRequest{Objects: "events"})
			return err
		}},
		{"say on another module's stream", "other", `other was bound no stream called "events"`, func(module string) error {
			_, err := conn.StreamPublish(module, buscalls.StreamPublishRequest{Stream: "events", Event: "x"})
			return err
		}},
		{"say on a binding that is not a stream", "m", `"files" is bound with the storage type object-store, not stream`, func(module string) error {
			_, err := conn.StreamPublish(module, buscalls.StreamPublishRequest{Stream: "files", Event: "x"})
			return err
		}},
		{"a work queue not bound", "m", `bound no stream called "jobs"`, func(module string) error {
			_, err := conn.WorkQueueSize(module, buscalls.WorkQueueRequest{Queue: "jobs"})
			return err
		}},
	} {
		t.Run(c.name, func(t *testing.T) {
			err := c.call(c.module)
			if err == nil || !strings.Contains(err.Error(), c.refusal) {
				t.Fatalf("refused with %v, want the reason %q", err, c.refusal)
			}
		})
	}
}