Create NATS resources only on missing lookups

ts/packages/base/src/__tests__/nats-backend.test.ts

@@ -5,6 +5,27 @@ const natsMock = vi.hoisted(() => {
   const encoder = new TextEncoder();
   const decoder = new TextDecoder();
 
+  class MockNatsError extends Error {
+    readonly code: string;
+    private readonly apiCode: number | undefined;
+
+    constructor(code: string, apiCode?: number) {
+      super(code);
+      this.name = "NatsError";
+      this.code = code;
+      this.apiCode = apiCode;
+    }
+
+    jsError() {
+      return this.apiCode === undefined
+        ? null
+        : {
+            code: this.apiCode,
+            description: this.code,
+          };
+    }
+  }
+
   const publish = vi.fn();
   const consumersGet = vi.fn();
   const consumersAdd = vi.fn();
@@ -28,6 +49,7 @@ const natsMock = vi.hoisted(() => {
     encoder,
     headerAppend,
     headers,
+    NatsError: MockNatsError,
     nak,
     next,
     publish,
@@ -39,14 +61,20 @@ const natsMock = vi.hoisted(() => {
 vi.mock("nats", () => ({
   AckPolicy: { Explicit: "explicit" },
   DeliverPolicy: { All: "all", New: "new" },
+  ErrorCode: { JetStream404NoMessages: "404" },
   StringCodec: () => ({
     decode: (input: Uint8Array) => natsMock.decoder.decode(input),
     encode: (input: string) => natsMock.encoder.encode(input),
   }),
   connect: natsMock.connect,
   headers: natsMock.headers,
+  NatsError: natsMock.NatsError,
 }));
 
+function makeNatsError(code: string, apiCode?: number) {
+  return new natsMock.NatsError(code, apiCode);
+}
+
 function resetNatsMock(): void {
   vi.clearAllMocks();
 
@@ -88,6 +116,66 @@ describe("NATS backend", () => {
     resetNatsMock();
   });
 
+  it("creates streams only when stream lookup returns a JetStream 404", async () => {
+    natsMock.streamsInfo.mockRejectedValueOnce(makeNatsError("404", 404));
+    const backend = makeNatsBackend("nats://test");
+
+    await backend.createProducer<string>({ topic: "tg.test.topic" });
+
+    expect(natsMock.streamsAdd).toHaveBeenCalledWith({
+      name: "tg_test",
+      subjects: ["tg.test.>"],
+    });
+  });
+
+  it("does not create streams for non-missing lookup failures", async () => {
+    natsMock.streamsInfo.mockRejectedValueOnce(makeNatsError("PERMISSIONS_VIOLATION"));
+    const backend = makeNatsBackend("nats://test");
+
+    const error = await backend.createProducer<string>({ topic: "tg.test.topic" }).catch((caught: unknown) => caught);
+
+    expect(error).toMatchObject({
+      _tag: "PubSubError",
+      operation: "stream-info:tg_test",
+    });
+    expect(natsMock.streamsAdd).not.toHaveBeenCalled();
+  });
+
+  it("creates durable consumers only when consumer lookup returns a JetStream 404", async () => {
+    natsMock.consumersGet
+      .mockRejectedValueOnce(makeNatsError("404", 404))
+      .mockResolvedValueOnce({ next: natsMock.next });
+    const backend = makeNatsBackend("nats://test");
+
+    await backend.createConsumer<string>({
+      topic: "tg.test.topic",
+      subscription: "worker",
+    });
+
+    expect(natsMock.consumersAdd).toHaveBeenCalledWith("tg_test", {
+      ack_policy: "explicit",
+      deliver_policy: "new",
+      durable_name: "worker",
+      filter_subject: "tg.test.topic",
+    });
+  });
+
+  it("does not create durable consumers for non-missing lookup failures", async () => {
+    natsMock.consumersGet.mockRejectedValueOnce(makeNatsError("PERMISSIONS_VIOLATION"));
+    const backend = makeNatsBackend("nats://test");
+
+    const error = await backend.createConsumer<string>({
+      topic: "tg.test.topic",
+      subscription: "worker",
+    }).catch((caught: unknown) => caught);
+
+    expect(error).toMatchObject({
+      _tag: "PubSubError",
+      operation: "init-consumer:tg.test.topic",
+    });
+    expect(natsMock.consumersAdd).not.toHaveBeenCalled();
+  });
+
   it("maps invalid publish headers to tagged PubSubError", async () => {
     natsMock.headerAppend.mockImplementation(() => {
       throw "invalid header";

ts/packages/base/src/backend/nats.ts

@@ -10,6 +10,7 @@
 
 import {
   connect,
+  ErrorCode,
   type NatsConnection,
   type JetStreamClient,
   type JetStreamManager,
@@ -17,6 +18,7 @@ import {
   headers,
   type JsMsg,
   type JetStreamPublishOptions,
+  NatsError,
   StringCodec,
   AckPolicy,
   DeliverPolicy,
@@ -64,6 +66,18 @@ function makeNatsMessage<T>(msg: JsMsg, decoded: T): NatsMessage<T> {
 
 const hasJsMsg = Predicate.hasProperty("_jsMsg");
 
+class NatsLookupError extends S.TaggedErrorClass<NatsLookupError>()(
+  "NatsLookupError",
+  {
+    cause: S.Unknown,
+    operation: S.String,
+  },
+) {}
+
+function natsLookupError(operation: string, cause: unknown): NatsLookupError {
+  return NatsLookupError.make({ cause, operation });
+}
+
 function isAckableJsMsg(value: unknown): value is Pick<JsMsg, "ack" | "nak"> {
   if (!Predicate.isObject(value)) return false;
   if (!Predicate.hasProperty(value, "ack")) return false;
@@ -75,6 +89,22 @@ function isNatsMessage<T>(message: Message<T>): message is NatsMessage<T> {
   return hasJsMsg(message) && isAckableJsMsg(message._jsMsg);
 }
 
+function isJetStreamMissingResource(error: unknown): boolean {
+  if (!(error instanceof NatsError)) {
+    return false;
+  }
+  if (error.code === ErrorCode.JetStream404NoMessages) {
+    return true;
+  }
+
+  const jsError = error.jsError();
+  return jsError?.code === 404;
+}
+
+function isMissingLookupError(error: NatsLookupError): boolean {
+  return isJetStreamMissingResource(error.cause);
+}
+
 function makeNatsProducer<T>(
   js: JetStreamClient,
   subject: string,
@@ -148,31 +178,34 @@ function makeNatsConsumer<T>(
         Effect.gen(function* () {
           const existing = yield* Effect.tryPromise({
             try: () => js.consumers.get(streamName, subscription),
-            catch: (error) => pubSubError(`get-consumer:${streamName}:${subscription}`, error),
+            catch: (error) => natsLookupError(`get-consumer:${streamName}:${subscription}`, error),
           }).pipe(
-            Effect.catch(() =>
-              Effect.gen(function* () {
-                const deliverPolicy =
-                  initialPosition === "earliest"
-                    ? DeliverPolicy.All
-                    : DeliverPolicy.New;
+            Effect.catchIf(
+              isMissingLookupError,
+              () =>
+                Effect.gen(function* () {
+                  const deliverPolicy =
+                    initialPosition === "earliest"
+                      ? DeliverPolicy.All
+                      : DeliverPolicy.New;
 
-                yield* Effect.tryPromise({
-                  try: () =>
-                    jsm.consumers.add(streamName, {
-                      durable_name: subscription,
-                      ack_policy: AckPolicy.Explicit,
-                      deliver_policy: deliverPolicy,
-                      filter_subject: subject,
-                    }),
-                  catch: (error) => pubSubError(`add-consumer:${streamName}:${subscription}`, error),
-                });
+                  yield* Effect.tryPromise({
+                    try: () =>
+                      jsm.consumers.add(streamName, {
+                        durable_name: subscription,
+                        ack_policy: AckPolicy.Explicit,
+                        deliver_policy: deliverPolicy,
+                        filter_subject: subject,
+                      }),
+                    catch: (error) => pubSubError(`add-consumer:${streamName}:${subscription}`, error),
+                  });
 
-                return yield* Effect.tryPromise({
-                  try: () => js.consumers.get(streamName, subscription),
-                  catch: (error) => pubSubError(`get-consumer:${streamName}:${subscription}`, error),
-                });
-              }),
+                  return yield* Effect.tryPromise({
+                    try: () => js.consumers.get(streamName, subscription),
+                    catch: (error) => pubSubError(`get-consumer:${streamName}:${subscription}`, error),
+                  });
+                }),
+              (error) => Effect.fail(pubSubError(error.operation, error.cause)),
             ),
           );
           consumer = existing;
@@ -289,17 +322,20 @@ export function makeNatsBackend(url = "nats://localhost:4222"): PubSubBackend {
 
     yield* Effect.tryPromise({
       try: () => manager.streams.info(streamName),
-      catch: (error) => pubSubError(`stream-info:${streamName}`, error),
+      catch: (error) => natsLookupError(`stream-info:${streamName}`, error),
     }).pipe(
-      Effect.catch(() =>
-        Effect.tryPromise({
-          try: () =>
-            manager.streams.add({
-              name: streamName,
-              subjects: [wildcardSubject],
-            }),
-          catch: (error) => pubSubError(`stream-add:${streamName}`, error),
-        }),
+      Effect.catchIf(
+        isMissingLookupError,
+        () =>
+          Effect.tryPromise({
+            try: () =>
+              manager.streams.add({
+                name: streamName,
+                subjects: [wildcardSubject],
+              }),
+            catch: (error) => pubSubError(`stream-add:${streamName}`, error),
+          }),
+        (error) => Effect.fail(pubSubError(error.operation, error.cause)),
       ),
     );
     initializedStreams.add(streamName);