package events
import (
"testing"
"time"
natsserver "github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
)
// externalNATS starts a real listening JetStream server (the stand-in
// for an operator's external NATS) and returns its client URL.
func externalNATS(t *testing.T) string {
t.Helper()
srv, err := natsserver.NewServer(&natsserver.Options{
Host: "127.0.0.1",
Port: -1, // OS-assigned
JetStream: true,
StoreDir: t.TempDir(),
MaxPayload: 8 << 20,
JetStreamMaxStore: StoreBudget,
})
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("external NATS not ready")
}
t.Cleanup(srv.Shutdown)
return srv.ClientURL()
}
// Two nodes on one external NATS share the event plane: an artifact
// published by one is delivered to the other's Watch, and both see the
// same stream — the substrate multi-node folding and leasing ride on.
func TestExternalNATSSharedPlane(t *testing.T) {
url := externalNATS(t)
a, err := Connect(url)
if err != nil {
t.Fatal(err)
}
defer a.Close()
b, err := Connect(url)
if err != nil {
t.Fatal(err)
}
defer b.Close()
// b watches; a publishes; the tick crosses the wire.
ticks, cancel := b.Watch("repo1")
defer cancel()
if err := a.Indexed("repo1"); err != nil {
t.Fatal(err)
}
select {
case <-ticks:
case <-time.After(5 * time.Second):
t.Fatal("cross-node event not delivered")
}
// Both connected to the same stream (work-queue retention set once).
info, err := b.JS().StreamInfo(StreamName)
if err != nil {
t.Fatal(err)
}
if info.Config.Retention != nats.LimitsPolicy {
t.Fatalf("stream retention = %v", info.Config.Retention)
}
}
// A large parsed artifact must cross an external NATS too — the server
// is required to allow the 8 MiB payload, matching the embedded default.
func TestExternalNATSLargePayload(t *testing.T) {
url := externalNATS(t)
s, err := Connect(url)
if err != nil {
t.Fatal(err)
}
defer s.Close()
big := make([]byte, 3<<20)
if _, err := s.JS().Publish("zit.evt.big.artifact", big); err != nil {
t.Fatalf("publish %d bytes over external NATS: %v", len(big), err)
}
}
// Connect to an unreachable URL fails rather than hanging.
func TestExternalNATSUnreachable(t *testing.T) {
if _, err := Connect("nats://127.0.0.1:1"); err == nil {
t.Fatal("connect to dead port should fail")
}
}