feat(jsapi): add support for message scheduling

- Introduced `allow_msg_schedules` in stream configurations to enable scheduling messages.
- Added `ScheduleOptions` type to define scheduling parameters (e.g., specification, target, source, ttl).
- Updated headers to include scheduling fields (`Nats-Schedule`, `Nats-Schedule-Target`, etc.).
- Added tests to validate message scheduling functionality.

Signed-off-by: Alberto Ricart <[email protected]>

Author: aricart

jetstream/src/jsapi_types.ts

@@ -183,6 +183,11 @@ export type StreamConfig = StreamUpdateConfig & {
    * Enables a NATS stream implementation of CRDT operations
    */
   "allow_msg_counter": boolean;
+
+  /**
+   * Enables the scheduling of messages in a stream.
+   */
+  "allow_msg_schedules": boolean;
 };
 
 /**
@@ -1270,5 +1275,9 @@ export const PubHeaders = {
    * enable {@link StreamConfig#allow_msg_ttl}.
    */
   MessageTTL: "Nats-TTL",
+  Schedule: "Nats-Schedule",
+  ScheduleTarget: "Nats-Schedule-Target",
+  ScheduleSource: "Nats-Schedule-Source",
+  ScheduleTTL: "Nats-Schedule-TTL",
 } as const;
 export type PubHeaders = typeof PubHeaders[keyof typeof PubHeaders];

jetstream/src/jsclient.ts

@@ -212,6 +212,29 @@ export class JetStreamClientImpl extends BaseApiClientImpl
           `${opts.ttl}`,
         );
       }
+
+      if (opts.schedule) {
+        const so = opts.schedule;
+        if (so.specification) {
+          if (typeof so.specification === "string") {
+            mh.set(PubHeaders.Schedule, so.specification);
+          } else if (so.specification instanceof Date) {
+            mh.set(
+              PubHeaders.Schedule,
+              "@at " + so.specification.toISOString(),
+            );
+          }
+        }
+        if (so.target) {
+          mh.set(PubHeaders.ScheduleTarget, so.target);
+        }
+        if (so.source) {
+          mh.set(PubHeaders.ScheduleSource, so.source);
+        }
+        if (so.ttl) {
+          mh.set(PubHeaders.ScheduleTTL, so.ttl);
+        }
+      }
     }
 
     const to = opts.timeout || this.timeout;

jetstream/src/types.ts

@@ -98,6 +98,31 @@ export type PubAck = {
   duplicate: boolean;
 };
 
+export type ScheduleOptions = {
+  /**
+   * The schedule specification format.
+   * Currently only supports:
+   * "@at <date in isostring format>" - e.g., `new Date(Date.now() + 60_000).toISOString()`
+   *
+   * Note: The ADR-51 specification defines additional formats that may be supported in future
+   * server versions.
+   */
+  specification: string | Date;
+
+  /**
+   * The subject the message will be delivered to
+   */
+  target: string;
+  /**
+   * Instructs the schedule to read the last message on the given subject and publish. If the subject is empty, nothing is published. Wildcards are NOT supported.
+   */
+  source?: string;
+  /**
+   * Sets a message TTL if the stream supports per-message TTL.
+   */
+  ttl?: string;
+};
+
 /**
  * Options for messages published to JetStream
  */
@@ -170,6 +195,12 @@ export type JetStreamPublishOptions = {
      * └────────────────────┴────────┘
      */
     lastSubjectSequenceSubject: string;
+
+    /**
+     * The expected last sequence on the stream for a message with this subject
+     * and this value.
+     */
+    lastSubjectSequenceValue: number;
   }>;
 
   /**
@@ -184,6 +215,11 @@ export type JetStreamPublishOptions = {
    * Default is 1.
    */
   retries?: number;
+
+  /**
+   * Specifies the schedule for the message.
+   */
+  schedule?: ScheduleOptions;
 };
 
 /**

jetstream/tests/schedules_test.ts

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2025 The NATS Authors
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { jetstreamManager } from "../src/jsclient.ts";
+import { deferred, nanos } from "@nats-io/nats-core";
+import {
+  cleanup,
+  jetstreamServerConf,
+  notCompatible,
+  setup,
+} from "test_helpers";
+
+Deno.test("schedules - basics", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}));
+
+  if (await notCompatible(ns, nc, "2.11.0")) {
+    return;
+  }
+
+  const jsm = await jetstreamManager(nc);
+  await jsm.streams.add({
+    name: "schedules",
+    allow_msg_schedules: true,
+    subjects: ["schedules.>", "target.>"],
+    allow_msg_ttl: true,
+  });
+
+  await jsm.consumers.add("schedules", {
+    flow_control: true,
+    idle_heartbeat: nanos(60_000),
+    deliver_subject: "cron",
+    filter_subject: "target.>",
+  });
+
+  const d3000 = deferred();
+  const d5000 = deferred();
+
+  nc.subscribe("cron", {
+    callback: (err, m) => {
+      if (m.subject.endsWith("5000")) {
+        d5000.resolve();
+      } else if (m.subject.endsWith("3000")) {
+        d3000.resolve();
+      }
+    },
+  });
+
+  const js = jsm.jetstream();
+
+  const specification = "@at " + new Date(Date.now() + 5000).toISOString();
+  await js.publish("schedules.a", "5000", {
+    schedule: {
+      specification,
+      target: "target.5000",
+      ttl: "5m",
+    },
+  });
+
+  await js.publish("schedules.b", "3000", {
+    schedule: {
+      specification: new Date(Date.now() + 3000),
+      target: "target.3000",
+      ttl: "5m",
+    },
+  });
+
+  await Promise.all([d3000, d5000]);
+
+  await cleanup(ns, nc);
+});