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

package main

import (
	"encoding/json"
	"strings"
	"testing"
	"time"
)

const connz = `{"connections":[
 {"cid":1,"ip":"192.0.2.1","name":"mesh-host/anchor","authorized_user":"node.anchor","out_bytes":10,"in_bytes":5,"pending_bytes":0,"subscriptions":2,"in_msgs":3,"out_msgs":4},
 {"cid":2,"ip":"192.0.2.2","name":"laptop.node-tools","authorized_user":"laptop.node-tools","out_bytes":900,"in_bytes":50,"pending_bytes":7,"subscriptions":300,"in_msgs":30,"out_msgs":40}]}`

const subsz = `{"num_subscriptions":3,"subscriptions_list":[
 {"subject":"$SRV.INFO","msgs":4,"cid":2},
 {"subject":"mesh.seat.x.tool.current","qgroup":"","msgs":9,"cid":2},
 {"subject":"mesh.node.anchor.declare","msgs":1,"cid":1}]}`

const jsz = `{"streams":2,"consumers":2,"messages":12,"bytes":300,"account_details":[{"stream_detail":[
 {"name":"EVENTS","config":{"subjects":["mesh.mod.*.event.>"],"retention":"limits","max_age":604800000000000,"max_msgs_per_subject":10000},
  "state":{"messages":10,"bytes":200,"num_subjects":3,"consumer_count":2,"last_ts":"2026-10-04T10:00:00Z"},
  "consumer_detail":[
   {"name":"anchor_audit","config":{"filter_subject":"mesh.mod.*.event.>"},"num_pending":5,"num_ack_pending":1,"num_redelivered":2,"delivered":{"last_active":"2026-10-04T10:00:00Z"}},
   {"name":"anchor_quiet","config":{"filter_subjects":["mesh.mod.a.event.b"]},"num_pending":0,"num_ack_pending":0}]},
 {"name":"KV_claude-code_servers","config":{"subjects":["$KV.claude-code_servers.>"],"retention":"limits","max_msgs_per_subject":1},
  "state":{"messages":2,"bytes":100,"num_subjects":2,"consumer_count":1,"last_ts":"2026-10-04T11:00:00Z"}}]}]}`

const users = `accounts {
  MESH {
    jetstream: enabled
    users = [
      { user: "controller", password: "$2a$10$secret-hash-one", permissions: {
        publish: { allow: ["mesh.control.>", "$JS.API.>"] }
        subscribe: { allow: ["mesh.control.>", "_INBOX.controller.>"] }
        allow_responses: { max: 1, ttl: "1m" }
      } }
      { user: "laptop.node-tools", password: "$2a$10$secret-hash-two", permissions: {
        publish: { allow: ["$KV.claude-code_servers.>", "mesh.mod.*.tool.>"] }
        subscribe: { allow: ["mesh.mod.claude-code.tool.>"] }
      } }
    ]
  }
}`

func TestConnectionsAreNamedByTheirUserAndSorted(t *testing.T) {
	out, err := Connections([]byte(connz), "", "", 10)
	if err != nil {
		t.Fatal(err)
	}
	cs := out["connections"].([]Connection)
	if len(cs) != 2 || cs[0].User != "laptop.node-tools" || cs[1].User != "node.anchor" {
		t.Fatalf("%+v", cs)
	}
	narrowed, _ := Connections([]byte(connz), "anchor", "pending", 10)
	if n := narrowed["total"].(int); n != 1 {
		t.Fatalf("narrowed to %d", n)
	}
}

func TestSubscriptionsNameTheirUser(t *testing.T) {
	out, err := Subscriptions([]byte(subsz), []byte(connz), "SRV", 10)
	if err != nil {
		t.Fatal(err)
	}
	raw, _ := json.Marshal(out)
	if !strings.Contains(string(raw), `"user":"laptop.node-tools"`) || out["matched"].(int) != 1 {
		t.Fatalf("%s", raw)
	}
}

func TestStreamsBacklogAndBuckets(t *testing.T) {
	all, err := Streams([]byte(jsz), "")
	if err != nil || len(all["detail"].([]map[string]any)) != 2 {
		t.Fatalf("%v %v", all, err)
	}
	one, _ := Streams([]byte(jsz), "EVENTS")
	row := one["detail"].([]map[string]any)[0]
	if row["max_age_hours"].(int64) != 168 || len(row["consumer_detail"].([]map[string]any)) != 2 {
		t.Fatalf("%v", row)
	}
	if _, err := Streams([]byte(jsz), "NOPE"); err == nil {
		t.Fatal("an unknown stream answered")
	}
	b, _ := Backlog([]byte(jsz))
	if b["consumers_behind"].(int) != 1 {
		t.Fatalf("%v", b)
	}
	k, _ := Buckets([]byte(jsz))
	rows := k["buckets"].([]map[string]any)
	if len(rows) != 1 || rows[0]["module"] != "claude-code" || rows[0]["state"] != "servers" {
		t.Fatalf("%v", rows)
	}
}

func TestUsersNeverCarryAPasswordHash(t *testing.T) {
	for _, out := range []map[string]any{Users([]byte(users), ""), Users([]byte(users), "laptop")} {
		raw, _ := json.Marshal(out)
		if strings.Contains(string(raw), "secret-hash") || strings.Contains(string(raw), "$2a$") {
			t.Fatalf("a hash in %s", raw)
		}
	}
	us := ParseUsers([]byte(users))
	if len(us) != 2 || len(us[0].Publish) != 2 || !us[0].AllowResponses || us[1].AllowResponses {
		t.Fatalf("%+v", us)
	}
}

func TestWhetherAUserMayDoSomething(t *testing.T) {
	ok, err := UserCan([]byte(users), "laptop.node-tools", "publish", "$KV.claude-code_servers.all.x")
	if err != nil || ok["allowed"] != true || ok["by"] != "$KV.claude-code_servers.>" {
		t.Fatalf("%v %v", ok, err)
	}
	no, _ := UserCan([]byte(users), "laptop.node-tools", "publish", "$KV.claude-code_holdings.laptop")
	if no["allowed"] != false {
		t.Fatalf("%v", no)
	}
	if _, err := UserCan([]byte(users), "nobody", "publish", "x"); err == nil {
		t.Fatal("an unknown user answered")
	}
	for _, c := range []struct {
		p, s string
		want bool
	}{{"a.*.c", "a.b.c", true}, {"a.>", "a.b.c", true}, {"a.>", "a", false}, {"a.b", "a.b.c", false}, {"a.*", "a.b.c", false}} {
		if SubjectMatches(c.p, c.s) != c.want {
			t.Errorf("%s ~ %s", c.p, c.s)
		}
	}
}

// novox/hq issue 327: who opens and closes connections, per user and name, and which were dropped rather
// than closed by their client — read from the server's record of closed connections.
func TestClosedConnectionsArePerUserAndNameAndSayWhichWereDropped(t *testing.T) {
	const closed = `{"connections":[
 {"cid":10,"name":"mesh-controller verb conditions","authorized_user":"controller","start":"2026-10-08T14:12:00Z","stop":"2026-10-08T14:12:01Z","reason":"Client Closed"},
 {"cid":11,"name":"mesh-controller verb conditions","authorized_user":"controller","start":"2026-10-08T14:13:00Z","stop":"2026-10-08T14:13:03Z","reason":"Client Closed"},
 {"cid":12,"name":"ace.node-tools","authorized_user":"ace.node-tools","start":"2026-10-08T13:00:00Z","stop":"2026-10-08T14:14:00Z","reason":"Stale Connection"},
 {"cid":13,"name":"ace.node-tools","authorized_user":"ace.node-tools","start":"2026-10-08T14:14:05Z","stop":"2026-10-08T14:15:00Z","reason":"Read Error"},
 {"cid":5,"name":"old","authorized_user":"controller","start":"2026-10-08T10:00:00Z","stop":"2026-10-08T10:00:01Z","reason":"Client Closed"},
 {"cid":14,"name":"","start":"2026-10-08T14:16:00Z","stop":"2026-10-08T14:16:00Z","reason":"Authentication Failure"}]}`
	now, _ := time.Parse(time.RFC3339, "2026-10-08T14:20:00Z")
	out, err := ClosedConnections([]byte(closed), now, 1, "")
	if err != nil {
		t.Fatal(err)
	}
	users := out["users"].([]ClosedUser)
	if len(users) != 3 {
		t.Fatalf("%+v", users)
	}
	byUser := map[string]ClosedUser{}
	for _, u := range users {
		byUser[u.User] = u
	}
	c := byUser["controller"]
	if c.Closed != 2 || c.Dropped != 0 || c.ClosedPerHour != 2 || len(c.Names) != 1 ||
		c.Names[0].Name != "mesh-controller verb conditions" || c.Names[0].Lived != 3 {
		t.Errorf("the controller's: %+v", c)
	}
	tools := byUser["ace.node-tools"]
	if tools.Closed != 2 || tools.Dropped != 2 || tools.Names[0].Reasons["Stale Connection"] != 1 {
		t.Errorf("the runtime's: %+v", tools)
	}
	if byUser["(no user: refused before it logged in)"].Dropped != 1 {
		t.Errorf("a refused login is not said: %+v", users)
	}
	if out["reaches"] != true {
		t.Error("a record older than the window does not reach back the whole window")
	}
	narrowed, _ := ClosedConnections([]byte(closed), now, 1, "node-tools")
	if got := narrowed["users"].([]ClosedUser); len(got) != 1 || got[0].User != "ace.node-tools" {
		t.Errorf("narrowed: %+v", got)
	}
}