Repository: nats-io/nats-server

Issue:
Jetstream cross-account mirror doesn't work with API filter
### Observed behavior

When I add a <stream> filter in API export, messages do not appear in the mirror stream. 

### Expected behavior

I should be able to export API to the specific stream and create a mirror in the different account.

### Server and client version

 2.10.24

### Host environment

Docker

### Steps to reproduce


Here is my working NATS config file:

```
websocket: {
  port: 8222
  no_tls: true
}

jetstream {}

accounts: {  
  PUBLIC_ACCOUNT: {
    jetstream: enabled
    exports: [
      { service: "$JS.API.CONSUMER.CREATE.>", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
      { service: "$JS.FC.>" },
      { stream: "shared.public.>", accounts: ["INTERNAL_ACCOUNT"] }
    ]
    users: [
      {user: "frontend1", password: "pass123"}
    ]
  },
  
  INTERNAL_ACCOUNT: {
    jetstream: enabled
    imports: [
      { service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.>" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.>" },
      { service: { account: "PUBLIC_ACCOUNT", subject: "$JS.FC.>" }, to: "$JS.FC.>" },
      { stream: { account: "PUBLIC_ACCOUNT", subject: "shared.public.>" }, to: "shared.public.>" }
    ]
    users: [
      {user: "backend1", password: "pass123"}
    ]
  }
}
```

With the user **frontend1** I create a stream called "**SHARED_STREAM**" and publish a message on the "**public.a**" subject.
With the user **backend1** I create a stream mirror called "**SHARED_STREAM_M**", with deliver set to "**shared.public**" and API to "**JS.PUBLIC_ACCOUNT**".
I get messages from **SHARED_STREAM** into **SHARED_STREAM_M**, all good. 

Here is how the SHARED_STREAM_M looks like:

```
{
    "config": {
        "name": "SHARED_STREAM_M",
        "retention": "limits",
        "max_consumers": -1,
        "max_msgs": -1,
        "max_bytes": -1,
        "max_age": 0,
        "max_msgs_per_subject": -1,
        "max_msg_size": -1,
        "discard": "old",
        "storage": "file",
        "num_replicas": 1,
        "mirror": {
            "name": "SHARED_STREAM",
            "external": {
                "api": "JS.PUBLIC_ACCOUNT",
                "deliver": "shared.public"
            }
        },
        "compression": "none",
        "allow_direct": false,
        "mirror_direct": false,
        "sealed": false,
        "deny_delete": false,
        "deny_purge": false,
        "allow_rollup_hdrs": false,
        "consumer_limits": {}
    },
    "created": "2025-03-26T21:44:51.145469375Z",
    "state": {
        "messages": 1,
        "bytes": 56,
        "first_seq": 1,
        "first_ts": "2025-03-26T21:45:20.347194318Z",
        "last_seq": 1,
        "last_ts": "2025-03-26T21:45:20.347194318Z",
        "num_subjects": 1,
        "consumer_count": 1
    },
    "mirror": {
        "name": "SHARED_STREAM",
        "external": {
            "api": "JS.PUBLIC_ACCOUNT",
            "deliver": "shared.public"
        },
        "lag": 0,
        "active": 623464906
    },
    "ts": "2025-03-26T21:45:36.981488539Z"
}
```

But if I try to limit mirroring to the specific stream (and/or subject) and apply the filter like this:

```
websocket: {
  port: 8222
  no_tls: true
}

jetstream {}

accounts: {  
  PUBLIC_ACCOUNT: {
    jetstream: enabled
    exports: [
      { service: "$JS.API.CONSUMER.CREATE.SHARED_STREAM.*.public.a", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
      { service: "$JS.FC.>" },
      { stream: "shared.public.>", accounts: ["INTERNAL_ACCOUNT"] }
    ]
    users: [
      {user: "frontend1", password: "pass123"}
    ]
  },
  
  INTERNAL_ACCOUNT: {
    jetstream: enabled
    imports: [
      { service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.SHARED_STREAM.*.public.a" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.SHARED_STREAM.*.public.a" },
      { service: { account: "PUBLIC_ACCOUNT", subject: "$JS.FC.>" }, to: "$JS.FC.>" },
      { stream: { account: "PUBLIC_ACCOUNT", subject: "shared.public.>" }, to: "shared.public.>" }
    ]
    users: [
      {user: "backend1", password: "pass123"}
    ]
  }
}
```

The mirror stream is always empty. I was using mostly this https://github.com/synadia-labs/cross-account-jetstream-sourcing/tree/main as a reference and the fact that the API is formatted as: `$JS.API.CONSUMER.CREATE.<stream>.<consumer>.<filter>`.

Here is how the SHARED_STREAM_M looks like now:

```
{
    "config": {
        "name": "SHARED_STREAM_M",
        "retention": "limits",
        "max_consumers": -1,
        "max_msgs": -1,
        "max_bytes": -1,
        "max_age": 0,
        "max_msgs_per_subject": -1,
        "max_msg_size": -1,
        "discard": "old",
        "storage": "file",
        "num_replicas": 1,
        "mirror": {
            "name": "SHARED_STREAM",
            "external": {
                "api": "JS.PUBLIC_ACCOUNT",
                "deliver": "shared.public"
            }
        },
        "compression": "none",
        "allow_direct": false,
        "mirror_direct": false,
        "sealed": false,
        "deny_delete": false,
        "deny_purge": false,
        "allow_rollup_hdrs": false,
        "consumer_limits": {}
    },
    "created": "2025-03-26T21:47:05.182241461Z",
    "state": {
        "messages": 0,
        "bytes": 0,
        "first_seq": 0,
        "first_ts": "0001-01-01T00:00:00Z",
        "last_seq": 0,
        "last_ts": "0001-01-01T00:00:00Z",
        "consumer_count": 1
    },
    "mirror": {
        "name": "SHARED_STREAM",
        "external": {
            "api": "JS.PUBLIC_ACCOUNT",
            "deliver": "shared.public"
        },
        "lag": 0,
        "active": -1
    },
    "ts": "2025-03-26T21:47:27.641547768Z"
}
```

Note that replacing
`"$JS.API.CONSUMER.CREATE.SHARED_STREAM.*.public.a"`
with the
`"$JS.API.CONSUMER.CREATE.SHARED_STREAM.>"`
or with the
`"$JS.API.CONSUMER.CREATE.*.*.*"`

doesn't work as well. If I add anything after the CREATE it will not work.


Hints from issue comments:
@kozlovic: The request to create a JetStream consumer is sent to `JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.SHARED_STREAM`, not `JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.SHARED_STREAM.<some extra tokens>`, this is why this does not work. You can limit to a given stream by adding `SHARED_STREAM` in the consumer create API, but that's it.

That is, in `PUBLIC_ACCOUNT`, you could use:
```
{ service: "$JS.API.CONSUMER.CREATE.SHARED_STREAM", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
```
and in `INTERNAL_ACCOUNT`:
```
{ service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.SHARED_STREAM" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.SHARED_STREAM" },
```@zoran-petrovic-87: @kozlovic Thank you, it works with using only the stream filter...

I was confused by this example: https://github.com/synadia-labs/cross-account-jetstream-sourcing/tree/main
It clearly states that exports with a subject filter are possible:

> The security imports and exports can be set in two different ways, depending on whether you want to control on a per (single) subject filter basis what messages are allowed for export/import or simply grant access to the whole stream with however many subject filters the sourcing streams want to use.

These are exports and imports from their config file:
```
exports: [
            { service: "$JS.API.CONSUMER.CREATE.foo.*.foo.a", accounts: [ "accountB" ] }
imports: [
            { service: { account: accountA, subject: "$JS.API.CONSUMER.CREATE.foo.*.foo.a" }, to: "$JS.accountA.API.CONSUMER.CREATE.foo.*.foo.a" }
```

Also in the docs: https://docs.nats.io/reference/reference-protocols/nats_api_reference 
it says that this is the format for the admin api: 
`$JS.API.CONSUMER.CREATE.<stream>.<consumer>.<filter>`

Are these all wrong or am I missing something?
@kozlovic: @zoran-petrovic-87 Ok, I think I know what the problem is. I will research a bit more and will update this issue later today. Thanks!@kozlovic: @zoran-petrovic-87 I have posted a PR because the server was not handling the case of a single subject transform, like we do for source, but based on the JSON output of your mirror stream configuration, it looks like you did not use a filter subject, which is likely why it did not work in your case. However, it looks like `nats` utility uses a subject transform for the filter subject (but maybe there is an option that I missed). If you were to create the stream with the `FilterSubject: "public.a"` when creating the mirror stream, it would have worked (as my new test demonstrated before adding the fix for the `SubjectTransforms` cases).@zoran-petrovic-87: @kozlovic Thanks! Now I understand what is going on :) 

Solution patch:
diff --git a/server/stream.go b/server/stream.go
index 7341bf24..9ce3098d 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2836,7 +2836,14 @@ func (mset *stream) setupMirrorConsumer() error {
 		}
 		mirror.sfs = sfs
 		mirror.trs = trs
-		req.Config.FilterSubjects = sfs
+		// If there was no explicit FilterSubject defined and we have a single
+		// subject transform, use Config.FilterSubject instead of FilterSubjects
+		// so that we can use the extended consumer create API down below.
+		if req.Config.FilterSubject == _EMPTY_ && len(sfs) == 1 {
+			req.Config.FilterSubject = sfs[0]
+		} else {
+			req.Config.FilterSubjects = sfs
+		}
 	}
 
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
@@ -2862,8 +2869,6 @@ func (mset *stream) setupMirrorConsumer() error {
 		return nil
 	}
 
-	b, _ := json.Marshal(req)
-
 	var subject string
 	if req.Config.FilterSubject != _EMPTY_ {
 		req.Config.Name = fmt.Sprintf("mirror-%s", createConsumerName())
@@ -2876,6 +2881,9 @@ func (mset *stream) setupMirrorConsumer() error {
 		subject = strings.ReplaceAll(subject, "..", ".")
 	}
 
+	// Marshal now that we are done with `req`.
+	b, _ := json.Marshal(req)
+
 	// Reset
 	mirror.msgs = nil
 	mirror.err = nil



Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index d45aa328..bbc3aced 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -26124,3 +26124,142 @@ func TestJetStreamUpgradeConsumerVersioning(t *testing.T) {
 		})
 	}
 }
+
+func TestJetStreamMirrorCrossAccountWithFilteredSubjectAndSubjectTransform(t *testing.T) {
+	conf := createConfFile(t, fmt.Appendf(nil, `
+		listen: "127.0.0.1:-1"
+		jetstream: {store_dir: %q}
+		accounts: {
+			PUBLIC_ACCOUNT: {
+				jetstream: enabled
+				exports: [
+					{ service: "$JS.API.CONSUMER.CREATE.S.*.public.a", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
+					{ service: "$JS.API.CONSUMER.CREATE.S.*.public.b", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
+					{ service: "$JS.API.CONSUMER.CREATE.S.*.public.c", response_type: stream, accounts: ["INTERNAL_ACCOUNT"] },
+					{ service: "$JS.FC.>" },
+					{ stream: "shared.public.>", accounts: ["INTERNAL_ACCOUNT"] }
+				]
+				users: [ {user: "public", password: "pwd"} ]
+			},
+
+			INTERNAL_ACCOUNT: {
+				jetstream: enabled
+				imports: [
+					{ service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.S.*.public.a" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.S.*.public.a" },
+					{ service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.S.*.public.b" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.S.*.public.b" },
+					{ service: { account: "PUBLIC_ACCOUNT", subject: "$JS.API.CONSUMER.CREATE.S.*.public.c" }, to: "JS.PUBLIC_ACCOUNT.CONSUMER.CREATE.S.*.public.c" },
+					{ service: { account: "PUBLIC_ACCOUNT", subject: "$JS.FC.>" }, to: "$JS.FC.>" },
+					{ stream: { account: "PUBLIC_ACCOUNT", subject: "shared.public.>" }, to: "shared.public.>" }
+				]
+				users: [ {user: "internal", password: "pwd"} ]
+			}
+		}
+	`, t.TempDir()))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	pc, pjs := jsClientConnect(t, s, nats.UserInfo("public", "pwd"))
+	defer pc.Close()
+
+	_, err := pjs.AddStream(&nats.StreamConfig{
+		Name:     "S",
+		Subjects: []string{"public.>"},
+	})
+	require_NoError(t, err)
+
+	ic, ijs := jsClientConnect(t, s, nats.UserInfo("internal", "pwd"))
+	defer ic.Close()
+
+	addStream := func(name, fs string, transform *nats.SubjectTransformConfig) {
+		t.Helper()
+		sc := &nats.StreamConfig{
+			Name: name,
+			Mirror: &nats.StreamSource{
+				Name: "S",
+				External: &nats.ExternalStream{
+					APIPrefix:     "JS.PUBLIC_ACCOUNT",
+					DeliverPrefix: "shared.public",
+				},
+			},
+		}
+		if fs != _EMPTY_ {
+			sc.Mirror.FilterSubject = fs
+		} else {
+			sc.Mirror.SubjectTransforms = append(sc.Mirror.SubjectTransforms, *transform)
+		}
+		_, err = ijs.AddStream(sc)
+		require_NoError(t, err)
+	}
+
+	checkMirror := func(name string, expected int) {
+		t.Helper()
+		checkFor(t, time.Second, 15*time.Millisecond, func() error {
+			si, err := ijs.StreamInfo(name)
+			if err != nil {
+				return err
+			}
+			if n := int(si.State.Msgs); n != expected {
+				return fmt.Errorf("Expected %v mirrored message, got %v", expected, n)
+			}
+			return nil
+		})
+	}
+
+	// Just a subject filter
+	addStream("M1", "public.a", nil)
+	natsPub(t, pc, "public.a", []byte("hello"))
+	checkMirror("M1", 1)
+
+	// Now try with equivalent subject transform (no destination).
+	addStream("M2", _EMPTY_, &nats.SubjectTransformConfig{Source: "public.b"})
+	natsPub(t, pc, "public.b", []byte("hello"))
+	checkMirror("M2", 1)
+
+	// And now with a transform destination.
+	addStream("M3", _EMPTY_, &nats.SubjectTransformConfig{Source: "public.c", Destination: "public.d"})
+	natsPub(t, pc, "public.c", []byte("hello"))
+	checkMirror("M3", 1)
+
+	checkMsg := func(stream, subj string, seq uint64) {
+		t.Helper()
+		msg, err := ijs.GetMsg(stream, seq)
+		require_NoError(t, err)
+		require_Equal(t, msg.Subject, subj)
+	}
+	checkMsg("M1", "public.a", 1)
+	checkMsg("M2", "public.b", 2)
+	checkMsg("M3", "public.d", 3)
+
+	ic.Close()
+	pc.Close()
+	s.Shutdown()
+	s, _ = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	ic, ijs = jsClientConnect(t, s, nats.UserInfo("internal", "pwd"))
+	defer ic.Close()
+
+	checkMirror("M1", 1)
+	checkMirror("M2", 1)
+	checkMirror("M3", 1)
+	checkMsg("M1", "public.a", 1)
+	checkMsg("M2", "public.b", 2)
+	checkMsg("M3", "public.d", 3)
+
+	pc, _ = jsClientConnect(t, s, nats.UserInfo("public", "pwd"))
+	defer pc.Close()
+	natsPub(t, pc, "public.a", []byte("hello"))
+	natsPub(t, pc, "public.b", []byte("hello"))
+	natsPub(t, pc, "public.c", []byte("hello"))
+	natsFlush(t, pc)
+
+	checkMirror("M1", 2)
+	checkMirror("M2", 2)
+	checkMirror("M3", 2)
+	checkMsg("M1", "public.a", 1)
+	checkMsg("M2", "public.b", 2)
+	checkMsg("M3", "public.d", 3)
+	checkMsg("M1", "public.a", 4)
+	checkMsg("M2", "public.b", 5)
+	checkMsg("M3", "public.d", 6)
+}