fix(jetstream): fixed possible dereference of undefined value when va…

…lidating stream sources

fix(kv): fixed handling of kv sources by adding a necessary transform

--
This fix was ported from nats-io/nats.deno#721

deno.json

@@ -32,7 +32,8 @@
     "check-versions-services": "bin/check-bundle-version.ts --module services",
     "check-versions-transport-node": "bin/check-bundle-version.ts --module transport-node",
     "check-versions-transport-deno": "bin/check-bundle-version.ts --module transport-deno",
-    "check-dependencies": "bin/check-dep-versions.ts"
+    "sync-dependencies": "deno task chk-deps && deno fmt",
+    "chk-deps": "bin/check-dep-versions.ts"
   },
   "fmt": {
     "include": ["transport-deno/", "bin/", "core/", "debug/", "jetstream/", "kv/", "obj/", "services/", "*.md", "transport-node/"],

jetstream/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.0-3",
+  "version": "3.0.0-4",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"

jetstream/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.0-3",
+  "version": "3.0.0-4",
   "files": [
     "lib/",
     "build/src/"

jetstream/src/jsmstream_api.ts

@@ -255,7 +255,7 @@ export class StreamAPIImpl extends BaseApiClientImpl implements StreamAPI {
       context: string,
       src: Partial<StreamSource>,
     ): void {
-      const count = src.subject_transforms?.length || 0;
+      const count = src?.subject_transforms?.length || 0;
       if (count > 0) {
         const { min, ok } = nci.features.get(
           Feature.JS_STREAM_SOURCE_SUBJECT_TRANSFORM,

kv/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/kv",
-  "version": "3.0.0-2",
+  "version": "3.0.0-3",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"
@@ -29,6 +29,6 @@
   },
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-18",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected]3"
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-4"
   }
 }

kv/import_map.json

@@ -2,8 +2,8 @@
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-18",
     "@nats-io/nats-core/internal": "jsr:@nats-io/nats-core@~3.0.0-18/internal",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected]3",
-    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-3/internal",
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-4",
+    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-4/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",

kv/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/kv",
-  "version": "3.0.0-2",
+  "version": "3.0.0-3",
   "files": [
     "lib/",
     "build/src/"
@@ -27,7 +27,7 @@
   "description": "kv library - this library implements all the base functionality for NATS KV javascript clients",
   "dependencies": {
     "@nats-io/nats-core": "~3.0.0-18",
-    "@nats-io/jetstream": "3.0.0-3"
+    "@nats-io/jetstream": "~3.0.0-4"
   },
   "devDependencies": {
     "@types/node": "^20.11.30",

kv/src/kv.ts

@@ -364,11 +364,21 @@ export class Bucket implements KV, KvRemove {
     } else if (opts.sources) {
       const sources = opts.sources.map((s) => {
         const c = Object.assign({}, s) as StreamSource;
+        const srcBucketName = c.name.startsWith(kvPrefix)
+          ? c.name.substring(kvPrefix.length)
+          : c.name;
         if (!c.name.startsWith(kvPrefix)) {
           c.name = `${kvPrefix}${c.name}`;
         }
+        if (!s.external && srcBucketName !== this.bucket) {
+          c.subject_transforms = [
+            { src: `$KV.${srcBucketName}.>`, dest: `$KV.${this.bucket}.>` },
+          ];
+        }
+        return c;
       });
       sc.sources = sources as unknown[] as StreamSource[];
+      sc.subjects = [this.subjectForBucket()];
     } else {
       sc.subjects = [this.subjectForBucket()];
     }

kv/tests/kv_test.ts

@@ -2269,3 +2269,32 @@ Deno.test("kv - replicas", async () => {
   await nc.close();
   await NatsServer.stopAll(servers, true);
 });
+
+Deno.test("kv - sourced", async () => {
+  const { ns, nc } = await _setup(
+    connect,
+    jetstreamServerConf({}),
+  );
+  if (await notCompatible(ns, nc, "2.6.3")) {
+    return;
+  }
+
+  const js = jetstream(nc);
+  const kvm = await new Kvm(js);
+  const source = await kvm.create("source");
+  const target = await kvm.create("target", {
+    sources: [{ name: "source" }],
+  });
+
+  await source.put("hello", "world");
+  for (let i = 0; i < 10; i++) {
+    const v = await target.get("hello");
+    if (v === null) {
+      await delay(250);
+      continue;
+    }
+    assertEquals(v.string(), "world");
+  }
+
+  await cleanup(ns, nc);
+});

obj/deno.json

@@ -28,6 +28,6 @@
   },
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-18",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected]3"
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-4"
   }
 }

obj/import_map.json

@@ -2,8 +2,8 @@
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-18",
     "@nats-io/nats-core/internal": "jsr:@nats-io/nats-core@~3.0.0-18/internal",
-    "@nats-io/jetstream": "jsr:@nats-io/[email protected]3",
-    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-3/internal",
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-4",
+    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-4/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",

obj/package.json

@@ -27,7 +27,7 @@
   "description": "obj library - this library implements all the base functionality for NATS objectstore for javascript clients",
   "dependencies": {
     "@nats-io/nats-core": "~3.0.0-18",
-    "@nats-io/jetstream": "3.0.0-3"
+    "@nats-io/jetstream": "~3.0.0-4"
   },
   "devDependencies": {
     "@types/node": "^20.11.30",