Add optional JetStream domain config to the natsstore plugin (PER-13735) (#5)

* feat(natsstore): add optional JetStream domain config (PER-13735)

Adds a `domain` plugin option; when set, the JetStream context is created
with nats.Domain so API calls are scoped to $JS.<domain>.API. Empty preserves
the default domain-less prefix, so existing deployments are unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(natsstore): close NATS connection and defer connected state until JetStream init succeeds

Addresses review comments:
- #5 (comment) (@copilot-pull-request-reviewer)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: omer9564

README.md

@@ -89,6 +89,7 @@ services:
 | `tls_key` | string | `""` | Path to TLS private key |
 | `tls_ca_cert` | string | `""` | Path to TLS CA certificate |
 | `tls_insecure` | bool | `false` | Skip TLS certificate verification |
+| `domain` | string | `""` | JetStream domain; scopes API calls to `$JS.<domain>.API`. Needed when NATS is reached across a leafnode boundary (optional) |
 
 ## Built-in Functions
 

pkg/natsstore/config.go

@@ -42,6 +42,13 @@ type Config struct {
 	TLSCACert    string `json:"tls_ca_cert,omitempty"`
 	TLSInsecure  bool   `json:"tls_insecure,omitempty"`
 
+	// Domain is an optional JetStream domain. When set, JetStream API calls are
+	// scoped to $JS.<domain>.API instead of the default $JS.API. Required when
+	// the target cluster is reached across a leafnode boundary; harmless for
+	// direct connections since the server serves its own domain prefix locally.
+	// Empty preserves the default (domain-less) prefix.
+	Domain string `json:"domain,omitempty"`
+
 	// Cache settings
 	TTL                  Duration `json:"ttl"`
 	RefreshInterval      Duration `json:"refresh_interval"`

pkg/natsstore/config_test.go

@@ -229,6 +229,7 @@ func TestConfig_JSONMarshaling(t *testing.T) {
 		ReconnectWait:        Duration(1 * time.Second),
 		MaxBucketsWatchers:   20,
 		RootTenant:           "test-bucket",
+		Domain:               "cloud",
 	}
 
 	// Marshal to JSON
@@ -248,4 +249,5 @@ func TestConfig_JSONMarshaling(t *testing.T) {
 	assert.Equal(t, config.MaxBucketsWatchers, unmarshaled.MaxBucketsWatchers)
 	assert.Equal(t, config.Bucket, unmarshaled.Bucket)
 	assert.Equal(t, config.RootTenant, unmarshaled.RootTenant)
+	assert.Equal(t, config.Domain, unmarshaled.Domain)
 }

pkg/natsstore/nats_client.go

@@ -101,15 +101,22 @@ func (nc *NATSClient) connect() error {
 		return fmt.Errorf("failed to connect to NATS server: %w", err)
 	}
 
-	nc.conn = conn
-	nc.setConnected(true)
-
-	// Create JetStream context
-	js, err := conn.JetStream()
+	// Create JetStream context, scoped to the configured domain when set
+	var jsOpts []nats.JSOpt
+	if nc.config.Domain != "" {
+		jsOpts = append(jsOpts, nats.Domain(nc.config.Domain))
+	}
+	js, err := conn.JetStream(jsOpts...)
 	if err != nil {
+		// Close the connection so a JetStream init failure doesn't leak it;
+		// only mark the client connected once setup fully succeeds.
+		conn.Close()
 		return fmt.Errorf("failed to create JetStream context: %w", err)
 	}
+
+	nc.conn = conn
 	nc.js = js
+	nc.setConnected(true)
 
 	nc.logger.Info("Connected to NATS at %s", nc.config.ServerURL)
 	return nil