Add Nats JetStream as a sink

README.md

@@ -65,6 +65,7 @@ Sequin works great for change data capture use cases like:
 | <img src="docs/images/introduction/kinesis.png" alt="Kinesis" width="24" height="24" /> | [Kinesis](https://sequinstream.com/docs/quickstart/kinesis) | [Reference](https://sequinstream.com/docs/reference/sinks/kinesis) | Send messages to Amazon Kinesis streams queues |
 | <img src="docs/images/introduction/meilisearch.png" alt="Meilisearch" width="24" height="24" /> | [Meilisearch](https://sequinstream.com/docs/quickstart/meilisearch) | [Reference](https://sequinstream.com/docs/reference/sinks/meilisearch) | Index database changes with Meilisearch |
 | <img src="docs/images/introduction/nats.png" alt="NATS" width="24" height="24" /> | [NATS](https://sequinstream.com/docs/quickstart/nats) | [Reference](https://sequinstream.com/docs/reference/sinks/nats) | Stream changes to NATS subjects |
+| <img src="docs/images/introduction/nats.png" alt="NATS JetStream" width="24" height="24" /> | [NATS JetStream](https://sequinstream.com/docs/quickstart/nats-jetstream) | [Reference](https://sequinstream.com/docs/reference/sinks/nats-jetstream) | Stream changes to NATS JetStream |
 | <img src="docs/images/introduction/rabbit.png" alt="RabbitMQ" width="24" height="24" /> | [RabbitMQ](https://sequinstream.com/docs/quickstart/rabbitmq) | [Reference](https://sequinstream.com/docs/reference/sinks/rabbitmq) | Publish messages to RabbitMQ exchanges |
 | <img src="docs/images/introduction/redis-1.png" alt="Redis Stream" width="24" height="24" /> | [Redis Stream](https://sequinstream.com/docs/quickstart/redis-stream) | [Reference](https://sequinstream.com/docs/reference/sinks/redis-stream) | `XADD` to Redis Streams |
 | <img src="docs/images/introduction/redis-2.png" alt="Redis String" width="24" height="24" /> | [Redis String](https://sequinstream.com/docs/quickstart/redis-string) | [Reference](https://sequinstream.com/docs/reference/sinks/redis-string) | `SET` to Redis keys  |

assets/svelte/consumers/ShowSink.svelte

@@ -22,6 +22,7 @@
     SequinStreamConsumer,
     GcpPubsubConsumer,
     NatsConsumer,
+    NatsJetstreamConsumer,
     RabbitMqConsumer,
     TypesenseConsumer,
     ElasticsearchConsumer,
@@ -34,6 +35,7 @@
   import KafkaSinkCard from "../sinks/kafka/KafkaSinkCard.svelte";
   import KinesisSinkCard from "../sinks/kinesis/KinesisSinkCard.svelte";
   import NatsSinkCard from "../sinks/nats/NatsSinkCard.svelte";
+  import NatsJetstreamSinkCard from "../sinks/nats_jetstream/NatsJetstreamSinkCard.svelte";
   import RabbitMqSinkCard from "../sinks/rabbitmq/RabbitMqSinkCard.svelte";
   import RedisStreamSinkCard from "../sinks/redis-stream/RedisStreamSinkCard.svelte";
   import RedisStringSinkCard from "../sinks/redis-string/RedisStringSinkCard.svelte";
@@ -143,6 +145,12 @@
     return consumer.sink.type === "nats";
   }
 
+  function isNatsJetstreamConsumer(
+    consumer: Consumer,
+  ): consumer is NatsJetstreamConsumer {
+    return consumer.sink.type === "nats_jetstream";
+  }
+
   function isGcpPubsubConsumer(
     consumer: Consumer,
   ): consumer is GcpPubsubConsumer {
@@ -1246,6 +1254,8 @@
         <SequinStreamSinkCard {consumer} {apiBaseUrl} {apiTokens} />
       {:else if isNatsConsumer(consumer)}
         <NatsSinkCard {consumer} />
+      {:else if isNatsJetstreamConsumer(consumer)}
+        <NatsJetstreamSinkCard {consumer} />
       {:else if isRabbitMqConsumer(consumer)}
         <RabbitMqSinkCard {consumer} />
       {:else if isTypesenseConsumer(consumer)}

assets/svelte/consumers/ShowSinkHeader.svelte

@@ -27,6 +27,7 @@
   import GcpPubsubIcon from "../sinks/gcp_pubsub/GcpPubsubIcon.svelte";
   import SequinStreamIcon from "../sinks/sequin_stream/SequinStreamIcon.svelte";
   import NatsIcon from "../sinks/nats/NatsIcon.svelte";
+  import NatsJetstreamIcon from "../sinks/nats_jetstream/NatsJetstreamIcon.svelte";
   import MeilisearchIcon from "../sinks/meilisearch/MeilisearchIcon.svelte";
   import RabbitMqIcon from "../sinks/rabbitmq/RabbitMqIcon.svelte";
   import AzureEventHubIcon from "../sinks/azure_event_hub/AzureEventHubIcon.svelte";
@@ -146,6 +147,8 @@
             <SequinStreamIcon class="h-6 w-6 mr-2" />
           {:else if consumer.sink.type === "nats"}
             <NatsIcon class="h-6 w-6 mr-2" />
+          {:else if consumer.sink.type === "nats_jetstream"}
+            <NatsJetstreamIcon class="h-6 w-6 mr-2" />
           {:else if consumer.sink.type === "meilisearch"}
             <MeilisearchIcon class="h-6 w-6 mr-2" />
           {:else if consumer.sink.type === "rabbitmq"}

assets/svelte/consumers/SinkConsumerForm.svelte

@@ -30,6 +30,7 @@
   import GcpPubsubSinkForm from "$lib/sinks/gcp_pubsub/GcpPubsubSinkForm.svelte";
   import SequinStreamSinkForm from "$lib/sinks/sequin_stream/SequinStreamSinkForm.svelte";
   import NatsSinkForm from "$lib/sinks/nats/NatsSinkForm.svelte";
+  import NatsJetstreamSinkForm from "$lib/sinks/nats_jetstream/NatsJetstreamSinkForm.svelte";
   import RabbitMqSinkForm from "$lib/sinks/rabbitmq/RabbitMqSinkForm.svelte";
   import AzureEventHubSinkForm from "$lib/sinks/azure_event_hub/AzureEventHubSinkForm.svelte";
   import { CircleAlert, Info, Plus } from "lucide-svelte";
@@ -752,6 +753,14 @@
         {refreshFunctions}
         bind:functionRefreshState
       />
+    {:else if consumer.type === "nats_jetstream"}
+      <NatsJetstreamSinkForm
+        errors={errors.consumer}
+        bind:form
+        {functions}
+        {refreshFunctions}
+        bind:functionRefreshState
+      />
     {:else if consumer.type === "rabbitmq"}
       <RabbitMqSinkForm
         errors={errors.consumer}

assets/svelte/consumers/SinkIndex.svelte

@@ -28,6 +28,7 @@
   import GcpPubsubIcon from "../sinks/gcp_pubsub/GcpPubsubIcon.svelte";
   import SequinStreamIcon from "../sinks/sequin_stream/SequinStreamIcon.svelte";
   import NatsIcon from "../sinks/nats/NatsIcon.svelte";
+  import NatsJetstreamIcon from "../sinks/nats_jetstream/NatsJetstreamIcon.svelte";
   import RabbitMqIcon from "../sinks/rabbitmq/RabbitMqIcon.svelte";
 
   import TypesenseIcon from "../sinks/typesense/TypesenseIcon.svelte";
@@ -54,6 +55,7 @@
       | "gcp_pubsub"
       | "sequin_stream"
       | "nats"
+      | "nats_jetstream"
       | "rabbitmq"
       | "typesense"
       | "elasticsearch";
@@ -139,6 +141,11 @@
       name: "NATS",
       icon: NatsIcon,
     },
+    {
+      id: "nats_jetstream",
+      name: "NATS JetStream",
+      icon: NatsJetstreamIcon,
+    },
     {
       id: "rabbitmq",
       name: "RabbitMQ",

assets/svelte/consumers/dynamicRoutingDocs.ts

@@ -83,6 +83,22 @@ export const routedSinkDocs: Record<RoutedSinkType, RoutedSinkDocs> = {
       },
     },
   },
+  nats_jetstream: {
+    fields: {
+      subject: {
+        description: "The NATS JetStream subject to publish messages to",
+        staticValue:
+          "sequin.<database_name>.<table_schema>.<table_name>.<action>",
+        dynamicDefault:
+          "sequin.<database_name>.<table_schema>.<table_name>.<action>",
+      },
+      headers: {
+        description: "Map of key value pairs",
+        staticValue: '%{"Nats-Msg-Id" => <idempotency-key>}',
+        dynamicDefault: '%{"Nats-Msg-Id" => <idempotency-key>}',
+      },
+    },
+  },
   kafka: {
     fields: {
       topic: {

assets/svelte/consumers/types.ts

@@ -138,6 +138,23 @@ export type NatsConsumer = BaseConsumer & {
   };
 };
 
+// NATS JetStream specific sink
+export type NatsJetstreamConsumer = BaseConsumer & {
+  sink: {
+    type: "nats_jetstream";
+    host: string;
+    port: number;
+    stream_name: string;
+    domain: string;
+    publish_timeout_ms: number;
+    username: string;
+    password: string;
+    jwt: string;
+    nkey_seed: string;
+    tls: boolean;
+  };
+};
+
 // Azure Event Hub specific sink
 export type AzureEventHubConsumer = BaseConsumer & {
   sink: {
@@ -260,6 +277,7 @@ export type Consumer =
   | SequinStreamConsumer
   | GcpPubsubConsumer
   | NatsConsumer
+  | NatsJetstreamConsumer
   | AzureEventHubConsumer
   | RabbitMqConsumer
   | TypesenseConsumer
@@ -278,6 +296,7 @@ export const SinkTypeValues = [
   "gcp_pubsub",
   "elasticsearch",
   "nats",
+  "nats_jetstream",
   "rabbitmq",
   "typesense",
   "meilisearch",
@@ -292,6 +311,7 @@ export const RoutedSinkTypeValues = [
   "redis_string",
   "redis_stream",
   "nats",
+  "nats_jetstream",
   "kafka",
   "gcp_pubsub",
   "typesense",

assets/svelte/sinks/nats_jetstream/NatsJetstreamIcon.svelte

@@ -0,0 +1,26 @@
+<svg
+  xmlns="http://www.w3.org/2000/svg"
+  class={$$props.class}
+  viewBox="0 0 64 64"
+>
+  <path
+    d="M1.156 50H27.53l15.125 14V50h20.187V0H1.156zm9.937-38.312h11.563l22.25 20.75v-20.75h7.375V38.5H41.094L18.47 17.375V38.5h-7.375z"
+    fill="#8dc63f"
+  />
+  <path
+    d="M18.47 17.375L41.094 38.5H52.28V11.687h-7.374v20.75l-22.25-20.75H11.094V38.5h7.375z"
+    fill="#fff"
+  />
+  <path d="M1.156 0h61.687v50H1.156z" fill="#27aae1" />
+  <path d="M32 0h30.844v25H32z" fill="#34a574" />
+  <path d="M32 25h30.844v25H32z" fill="#8dc63f" />
+  <path d="M1.156 0H32v25H1.156z" fill="#27aae1" />
+  <g fill="#375c93">
+    <path d="M1.156 25H32v25H1.156z" />
+    <path d="M32 50l.154 4.303L27.53 50z" />
+  </g>
+  <path
+    d="M16.367 15.83l26.758 24.984h13.23V9.102h-8.722v24.54L21.32 9.102H7.644v31.71h8.722z"
+    fill="#fff"
+  />
+</svg>

assets/svelte/sinks/nats_jetstream/NatsJetstreamSinkCard.svelte

@@ -0,0 +1,145 @@
+<script lang="ts">
+  import { Card, CardContent } from "$lib/components/ui/card";
+  import { ExternalLink } from "lucide-svelte";
+  import type { NatsJetstreamConsumer } from "../../consumers/types";
+
+  export let consumer: NatsJetstreamConsumer;
+</script>
+
+<Card>
+  <CardContent class="p-6">
+    <div class="flex justify-between items-center mb-4">
+      <h2 class="text-lg font-semibold">NATS JetStream configuration</h2>
+    </div>
+
+    <div class="grid grid-cols-1 lg:grid-cols-2 gap-4">
+      <div>
+        <span class="text-sm text-gray-500">Host</span>
+        <div class="mt-2">
+          <div class="overflow-x-auto">
+            <span
+              class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md inline-block whitespace-nowrap"
+              >{consumer.sink.host}</span
+            >
+          </div>
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Port</span>
+        <div class="mt-2">
+          <div class="overflow-x-auto">
+            <span
+              class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md inline-block whitespace-nowrap"
+              >{consumer.sink.port}</span
+            >
+          </div>
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Stream Name</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.stream_name || "-"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Domain</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.domain || "-"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Username</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.username || "-"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">TLS Enabled</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.tls ? "Yes" : "No"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">JWT</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.jwt ? "********" : "-"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">NKey Seed</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.nkey_seed ? "********" : "-"}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Publish Timeout</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.publish_timeout_ms}ms</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-gray-500">Subject</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md whitespace-nowrap"
+          >
+            {#if consumer.routing_id}
+              Determined by <a
+                href={`/functions/${consumer.routing_id}`}
+                data-phx-link="redirect"
+                data-phx-link-state="push"
+                class="underline">router</a
+              >
+              <ExternalLink class="h-4 w-4 inline" />
+            {:else}
+              sequin.{consumer.database
+                .name}.&lt;table_schema&gt;.&lt;table_name&gt;.&lt;action&gt;
+            {/if}
+          </span>
+        </div>
+      </div>
+    </div>
+
+    {#if consumer.routing}
+      <div class="mt-4">
+        <span class="text-sm text-gray-500">Router</span>
+        <div class="mt-2">
+          <pre
+            class="font-mono bg-slate-50 p-2 border border-slate-100 rounded-md text-sm overflow-x-auto"><code
+              >{consumer.routing.function.code}</code
+            ></pre>
+        </div>
+      </div>
+    {/if}
+  </CardContent>
+</Card>