stream/consumer info timestamps, stream configuration `

jetstream/jsapi_types.ts

@@ -84,6 +84,11 @@ export interface StreamInfo extends ApiPaged {
    * closer and faster to access.
    */
   alternates?: StreamAlternate[];
+  /**
+   * The ISO timestamp when the StreamInfo was generated. This field is only available
+   * on servers 2.10.x or better
+   */
+  "ts"?: string;
 }
 
 export interface SubjectTransformConfig {
@@ -125,6 +130,13 @@ export interface StreamConfig extends StreamUpdateConfig {
    * Can only be set on already created streams via the Update API
    */
   sealed: boolean;
+
+  /**
+   * Sets the first sequence number used by the stream. This property can only be
+   * specified when creating the stream, and likely is not valid on mirrors etc,
+   * as it may disrupt the synchronization logic.
+   */
+  "first_seq": number;
 }
 
 /**
@@ -623,9 +635,9 @@ export interface ConsumerInfo {
    */
   name: string;
   /**
-   * The time the Consumer was created
+   * The ISO timestamp when the Consumer was created
    */
-  created: Nanos;
+  created: string;
   /**
    * The consumer configuration
    */
@@ -663,6 +675,11 @@ export interface ConsumerInfo {
    * Indicates if any client is connected and receiving messages from a push consumer
    */
   "push_bound": boolean;
+  /**
+   * The ISO timestamp when the ConsumerInfo was generated. This field is only available
+   * on servers 2.10.x or better
+   */
+  "ts"?: string;
 }
 
 export interface ConsumerListResponse extends ApiResponse, ApiPaged {

jetstream/jsmstream_api.ts

@@ -231,6 +231,12 @@ export class StreamAPIImpl extends BaseApiClient implements StreamAPI {
         throw new Error(`stream 'metadata' requires server ${min}`);
       }
     }
+    if (cfg.first_seq) {
+      const { min, ok } = nci.features.get(Feature.JS_STREAM_FIRST_SEQ);
+      if (!ok) {
+        throw new Error(`stream 'first_seq' requires server ${min}`);
+      }
+    }
     validateStreamName(cfg.name);
     cfg.mirror = convertStreamSourceDomain(cfg.mirror);
     //@ts-ignore: the sources are either set or not - so no item should be undefined in the list

jetstream/tests/consumers_test.ts

@@ -144,8 +144,11 @@ Deno.test("consumers - info", async () => {
   const c = await js.consumers.get(stream, "b");
   // retrieve the cached consumer - no messages
   const cached = await c.info(false);
+  cached.ts = "";
   assertEquals(cached.num_pending, 0);
-  assertEquals(await jsm.consumers.info(stream, "b"), cached);
+  const updated = await jsm.consumers.info(stream, "b");
+  updated.ts = "";
+  assertEquals(updated, cached);
 
   // add a message, retrieve the cached one - still not updated
   await js.publish(subj);

jetstream/tests/streams_test.ts

@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-import { NatsServer } from "../../tests/helpers/mod.ts";
+import { NatsServer, notCompatible } from "../../tests/helpers/mod.ts";
 import { AckPolicy, connect, JSONCodec } from "../../src/mod.ts";
 import {
   assertEquals,
@@ -27,6 +27,7 @@ import {
   setup,
 } from "../../tests/helpers/mod.ts";
 import { initStream } from "./jstest_util.ts";
+import { NatsConnectionImpl } from "../../nats-base-client/nats.ts";
 
 Deno.test("streams - get", async () => {
   const { ns, nc } = await setup(jetstreamServerConf({}, true));
@@ -186,3 +187,44 @@ Deno.test("streams - delete message", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("streams - first_seq", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+  if (await notCompatible(ns, nc, "2.10.0")) {
+    return;
+  }
+
+  const jsm = await nc.jetstreamManager();
+  const si = await jsm.streams.add({
+    name: "test",
+    first_seq: 50,
+    subjects: ["foo"],
+  });
+  assertEquals(si.config.first_seq, 50);
+
+  const pa = await nc.jetstream().publish("foo");
+  assertEquals(pa.seq, 50);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("streams - first_seq fails if wrong server", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf({}, true));
+  const nci = nc as NatsConnectionImpl;
+  nci.features.update("2.9.2");
+
+  const jsm = await nc.jetstreamManager();
+  await assertRejects(
+    async () => {
+      await jsm.streams.add({
+        name: "test",
+        first_seq: 50,
+        subjects: ["foo"],
+      });
+    },
+    Error,
+    "stream 'first_seq' requires server 2.10.0",
+  );
+
+  await cleanup(ns, nc);
+});

nats-base-client/semver.ts

@@ -47,6 +47,7 @@ export enum Feature {
   JS_SIMPLIFICATION = "js_simplification",
   JS_STREAM_CONSUMER_METADATA = "js_stream_consumer_metadata",
   JS_CONSUMER_FILTER_SUBJECTS = "js_consumer_filter_subjects",
+  JS_STREAM_FIRST_SEQ = "js_stream_first_seq",
 }
 
 type FeatureVersion = {
@@ -99,6 +100,7 @@ export class Features {
     this.set(Feature.JS_SIMPLIFICATION, "2.9.4");
     this.set(Feature.JS_STREAM_CONSUMER_METADATA, "2.10.0");
     this.set(Feature.JS_CONSUMER_FILTER_SUBJECTS, "2.10.0");
+    this.set(Feature.JS_STREAM_FIRST_SEQ, "2.10.0");
 
     this.disabled.forEach((f) => {
       this.features.delete(f);