mesh-catalog/modules/nats/cmd/nats-tools/consumer_test.go

package main

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"
)

const consumerJsz = `{"account_details":[{"stream_detail":[{"name":"EVENTS","state":{"messages":9,"first_seq":1,"last_seq":50},
 "consumer_detail":[{"name":"ace_assigner","created":"2026-11-09T10:00:00Z",
  "config":{"filter_subjects":["mesh.mod.a.event.b"],"deliver_policy":"new","ack_policy":"explicit","ack_wait":30000000000,"max_ack_pending":1000,"max_deliver":5},
  "delivered":{"consumer_seq":40,"stream_seq":48,"last_active":"2026-11-09T11:00:00Z"},"ack_floor":{"consumer_seq":39,"stream_seq":47},
  "num_pending":2,"num_ack_pending":1,"num_redelivered":3,"num_waiting":0}]}]}]}`

// fake is a stream of subjects by sequence 1..n.
func fake(subjects map[uint64]string, fail error) MsgGet {
	return func(_ context.Context, _ string, seq uint64, subject string) (*Stored, error) {
		if fail != nil {
			return nil, fail
		}
		for s := seq; s <= 50; s++ {
			if sub, ok := subjects[s]; ok && (subject == "" || sub == subject) {
				return &Stored{Seq: s, Subject: sub, Time: time.Unix(int64(s), 0).UTC(), Data: []byte(fmt.Sprintf("body-%d", s))}, nil
			}
		}
		return nil, errNoMessage
	}
}

func TestAConsumerIsReadWithItsConfigAndItsState(t *testing.T) {
	got, err := Consumer([]byte(consumerJsz), "EVENTS", "ace_assigner")
	if err != nil {
		t.Fatal(err)
	}
	cfg, st := got["config"].(map[string]any), got["state"].(map[string]any)
	if cfg["ack_policy"] != "explicit" || cfg["deliver_policy"] != "new" || cfg["max_ack_pending"].(int64) != 1000 || cfg["ack_wait_seconds"].(int64) != 30 {
		t.Fatalf("%v", cfg)
	}
	if st["delivered_stream_seq"].(int64) != 48 || st["ack_floor_stream_seq"].(int64) != 47 || st["num_pending"].(int64) != 2 || st["num_redelivered"].(int64) != 3 {
		t.Fatalf("%v", st)
	}
	for _, c := range [][2]string{{"NOPE", "x"}, {"EVENTS", "nope"}} {
		if _, err := Consumer([]byte(consumerJsz), c[0], c[1]); err == nil {
			t.Fatalf("%v answered", c)
		}
	}
}

func TestMessagesComeNewestFirstBoundedAndBodiesOnlyWhenAsked(t *testing.T) {
	ctx := context.Background()
	subjects := map[uint64]string{}
	for s := uint64(1); s <= 50; s++ {
		subjects[s] = "a.b"
		if s%10 == 0 {
			subjects[s] = "a.c"
		}
	}
	got, _ := NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 3, 0, false)
	rows := got["messages"].([]MessageRow)
	if len(rows) != 3 || rows[0].Seq != 50 || rows[1].Seq != 40 || rows[2].Seq != 30 || rows[0].Body != "" || rows[0].Size != 7 || got["stopped"] == nil {
		t.Fatalf("%+v", got)
	}
	got, _ = NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 20, 0, true)
	rows = got["messages"].([]MessageRow)
	if len(rows) != 5 || rows[4].Seq != 10 || rows[0].Body != "body-50" || got["stopped"] != nil {
		t.Fatalf("%+v", got)
	}
	got, _ = NewestMessages(ctx, fake(subjects, nil), "EVENTS", "a.c", 1, 50, 20, 16, false)
	if n := len(got["messages"].([]MessageRow)); n != 2 || !strings.Contains(got["stopped"].(string), "bytes") {
		t.Fatalf("%+v", got)
	}
	big := func(_ context.Context, _ string, seq uint64, _ string) (*Stored, error) {
		if seq > 5 {
			return nil, errNoMessage
		}
		return &Stored{Seq: 5, Subject: "x", Data: []byte(strings.Repeat("z", bodyLimit+1))}, nil
	}
	got, _ = NewestMessages(ctx, big, "EVENTS", "x", 1, 5, 5, 0, true)
	if r := got["messages"].([]MessageRow)[0]; r.Body != "" || r.BodyOmitted == "" {
		t.Fatalf("%+v", r)
	}
}

func TestMessagesScanBackAcrossBlocksAndSayAnUnansweredRead(t *testing.T) {
	ctx := context.Background()
	got, _ := NewestMessages(ctx, func(_ context.Context, _ string, seq uint64, _ string) (*Stored, error) {
		if seq <= 3 {
			return &Stored{Seq: 3, Subject: "old"}, nil
		}
		return nil, errNoMessage
	}, "EVENTS", "old", 1, 3*block, 5, 0, false)
	if rows := got["messages"].([]MessageRow); len(rows) != 1 || rows[0].Seq != 3 {
		t.Fatalf("a message three blocks back was not found: %+v", got)
	}
	_, err := NewestMessages(ctx, fake(nil, context.DeadlineExceeded), "EVENTS", "a", 1, 50, 5, 0, false)
	if err == nil || !strings.Contains(err.Error(), "grant") {
		t.Fatalf("%v", err)
	}
	if _, err := NewestMessages(ctx, fake(nil, errors.New("boom")), "EVENTS", "a", 1, 50, 5, 0, false); err == nil {
		t.Fatal("a failed read answered")
	}
}

func TestTheReadToolsAreServedAndRefuseBadArguments(t *testing.T) {
	ts := readTools(func() ([]byte, error) { return []byte(consumerJsz), nil },
		func(context.Context) (MsgGet, func(), error) {
			return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil
		})
	by := map[string]func(map[string]any) (any, error){}
	for _, tl := range ts {
		by[tl.Name] = tl.Run
	}
	if _, err := by["nats_consumer"](map[string]any{"stream": "EVENTS", "consumer": "ace_assigner"}); err != nil {
		t.Fatal(err)
	}
	if _, err := by["nats_consumer"](map[string]any{"stream": "EVENTS.x"}); err == nil {
		t.Fatal("bad names answered")
	}
	out, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS", "subject": "a.>"})
	if err != nil || out.(map[string]any)["returned"].(int) != 0 && out.(map[string]any)["returned"].(int) != 1 {
		t.Fatalf("%v %v", out, err)
	}
	for _, st := range []string{"KV_mesh-vault_signers", "OBJ_mesh-knowledge_sections"} {
		if _, err := by["nats_stream_messages"](map[string]any{"stream": st, "subject": ">", "body": true}); err == nil || !strings.Contains(err.Error(), "kept state") {
			t.Fatalf("%s was read: %v", st, err)
		}
	}
	jszOf := strings.Replace(consumerJsz, `"name":"EVENTS"`, `"name":"ASKS"`, 1)
	ask := readTools(func() ([]byte, error) { return []byte(jszOf), nil },
		func(context.Context) (MsgGet, func(), error) {
			return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil
		})
	for _, tl := range ask {
		if tl.Name != "nats_stream_messages" {
			continue
		}
		o, err := tl.Run(map[string]any{"stream": "ASKS", "subject": "a.b", "body": true})
		m := o.(map[string]any)
		if err != nil || m["body_not_shown"] == nil || m["messages"].([]MessageRow)[0].Body != "" {
			t.Fatalf("an ask stream showed a body: %v %v", o, err)
		}
	}
	ev, _ := by["nats_stream_messages"](map[string]any{"stream": "EVENTS", "subject": "a.b", "body": true})
	if rows := ev.(map[string]any)["messages"].([]MessageRow); len(rows) != 1 || rows[0].Body != "body-50" {
		t.Fatalf("an event stream showed no body: %v", ev)
	}
	seat := strings.Replace(consumerJsz, `"name":"EVENTS"`, `"name":"SEAT_node-engine_running"`, 1)
	for _, tl := range readTools(func() ([]byte, error) { return []byte(seat), nil },
		func(context.Context) (MsgGet, func(), error) {
			return fake(map[uint64]string{50: "a.b"}, nil), func() {}, nil
		}) {
		if tl.Name != "nats_stream_messages" {
			continue
		}
		o, err := tl.Run(map[string]any{"stream": "SEAT_node-engine_running", "subject": "a.b", "body": true})
		if err != nil || o.(map[string]any)["messages"].([]MessageRow)[0].Body != "" {
			t.Fatalf("a kept event stream showed a body: %v %v", o, err)
		}
	}
	if _, err := by["nats_stream_messages"](map[string]any{"stream": "EVENTS"}); err == nil {
		t.Fatal("no subject answered")
	}
}