✨ Routing for Elasticsearch

Author: RTLS

assets/svelte/consumers/SinkConsumerForm.svelte

@@ -577,7 +577,13 @@
         bind:functionRefreshState
       />
     {:else if consumer.type === "elasticsearch"}
-      <ElasticsearchSinkForm errors={errors.consumer} bind:form />
+      <ElasticsearchSinkForm
+        errors={errors.consumer}
+        bind:form
+        {functions}
+        {refreshFunctions}
+        bind:functionRefreshState
+      />
     {/if}
 
     <Card>

assets/svelte/consumers/dynamicRoutingDocs.ts

@@ -113,4 +113,14 @@ export const routedSinkDocs: Record<RoutedSinkType, RoutedSinkDocs> = {
       },
     },
   },
+  elasticsearch: {
+    fields: {
+      index_name: {
+        description: "Elasticsearch index name to publish to",
+        staticValue: "<empty>",
+        staticFormField: "index_name",
+        dynamicDefault: "sequin.<database_name>.<table_schema>.<table_name>",
+      },
+    },
+  },
 };

assets/svelte/consumers/types.ts

@@ -187,6 +187,7 @@ export type SequinStreamConsumer = BaseConsumer & {
 export type GcpPubsubConsumer = BaseConsumer & {
   sink: {
     type: "gcp_pubsub";
+    elasticsearch;
     project_id: string;
     topic_id: string;
     use_emulator: boolean;
@@ -234,7 +235,7 @@ export type ElasticsearchConsumer = BaseConsumer & {
     type: "elasticsearch";
     endpoint_url: string;
     index_name: string;
-    auth_type: "api_key" | "basic" | "bearer";
+    auth_type: "none" | "api_key" | "basic" | "bearer";
     auth_value: string;
     batch_size: number;
     timeout_seconds: number;
@@ -268,6 +269,7 @@ export const SinkTypeValues = [
   "kafka",
   "sequin_stream",
   "gcp_pubsub",
+  "elasticsearch",
   "nats",
   "rabbitmq",
   "typesense",
@@ -286,6 +288,7 @@ export const RoutedSinkTypeValues = [
   "gcp_pubsub",
   "typesense",
   "meilisearch",
+  "elasticsearch",
 ] as const;
 
 export type RoutedSinkType = (typeof RoutedSinkTypeValues)[number];

assets/svelte/functions/Edit.svelte

@@ -105,6 +105,7 @@
     gcp_pubsub: "GCP PubSub",
     typesense: "Typesense",
     meilisearch: "Meilisearch",
+    elasticsearch: "Elasticsearch",
   };
 
   let errorKeyOrder = ["description", "snippet", "line", "column"];

assets/svelte/sinks/elasticsearch/ElasticsearchSinkCard.svelte

@@ -1,39 +1,87 @@
 <script lang="ts">
   import { ExternalLink } from "lucide-svelte";
-  import { Card, CardContent } from "$lib/components/ui/card";
+  import {
+    Card,
+    CardContent,
+    CardHeader,
+    CardTitle,
+  } from "$lib/components/ui/card";
   import { Button } from "$lib/components/ui/button";
   import type { ElasticsearchConsumer } from "../../consumers/types";
 
   export let consumer: ElasticsearchConsumer;
 </script>
 
 <Card>
+  <CardHeader>
+    <CardTitle>Elasticsearch Configuration</CardTitle>
+  </CardHeader>
   <CardContent class="p-6">
-    <div class="flex justify-between items-center mb-4">
-      <h2 class="text-lg font-semibold">Elasticsearch Configuration</h2>
-    </div>
-
     <div class="grid grid-cols-2 gap-4">
       <div>
         <span class="text-sm text-gray-500">Endpoint URL</span>
         <div class="mt-2">
           <div
-            class="font-mono bg-slate-50 pl-1 pr-4 py-1 border border-slate-100 rounded-md break-all w-fit"
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md break-all w-fit"
           >
             <span>{consumer.sink.endpoint_url}</span>
           </div>
         </div>
       </div>
 
+      <div>
+        <span class="text-sm text-gray-500">Authentication</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md break-all w-fit"
+          >
+            {#if consumer.sink.auth_type === "none"}
+              No authentication
+            {:else if consumer.sink.auth_type === "api_key"}
+              API Key
+            {:else if consumer.sink.auth_type === "basic"}
+              Basic Auth
+            {:else if consumer.sink.auth_type === "bearer"}
+              Bearer Token
+            {/if}
+          </span>
+        </div>
+      </div>
+    </div>
+  </CardContent>
+</Card>
+<Card>
+  <CardHeader>
+    <CardTitle>Routing</CardTitle>
+  </CardHeader>
+  <CardContent>
+    <div class="grid grid-cols-1 lg:grid-cols-2 gap-4">
       <div>
         <span class="text-sm text-gray-500">Index Name</span>
         <div class="mt-2">
           <span
-            class="font-mono bg-slate-50 pl-1 pr-4 py-1 border border-slate-100 rounded-md whitespace-nowrap"
-            >{consumer.sink.index_name}</span
+            class="font-mono bg-slate-50 px-2 py-1 border border-slate-100 rounded-md break-all w-fit"
           >
+            {#if consumer.routing_id}
+              determined-by-router
+            {:else}
+              {consumer.sink.index_name}
+            {/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 w-full"><code
+              >{consumer.routing.function.code}</code
+            ></pre>
+        </div>
+      </div>
+    {/if}
   </CardContent>
 </Card>

assets/svelte/sinks/elasticsearch/ElasticsearchSinkForm.svelte

@@ -10,6 +10,7 @@
   import type { ElasticsearchConsumer } from "$lib/consumers/types";
   import { Label } from "$lib/components/ui/label";
   import { Eye, EyeOff, Info } from "lucide-svelte";
+  import DynamicRoutingForm from "$lib/consumers/DynamicRoutingForm.svelte";
   import {
     Select,
     SelectContent,
@@ -19,10 +20,15 @@
   } from "$lib/components/ui/select";
 
   export let form: ElasticsearchConsumer;
+  export let functions: Array<any> = [];
+  export let refreshFunctions: () => void;
+  export let functionRefreshState: "idle" | "refreshing" | "done" = "idle";
+  let selectedDynamic = form.routingMode === "dynamic";
   export let errors: any = {};
   let showPassword = false;
 
   const authTypeOptions = [
+    { value: "none", label: "None" },
     { value: "api_key", label: "API Key" },
     { value: "basic", label: "Basic Auth" },
     { value: "bearer", label: "Bearer Token" },
@@ -71,18 +77,6 @@
       </p>
     </div>
 
-    <div class="space-y-2">
-      <Label for="index_name">Index name</Label>
-      <Input
-        id="index_name"
-        bind:value={form.sink.index_name}
-        placeholder="my-index"
-      />
-      {#if errors.sink?.index_name}
-        <p class="text-destructive text-sm">{errors.sink.index_name}</p>
-      {/if}
-    </div>
-
     <div class="space-y-2">
       <Label for="auth_type">Authentication type</Label>
       <Select
@@ -126,11 +120,14 @@
           id="auth_value"
           type={showPassword ? "text" : "password"}
           bind:value={form.sink.auth_value}
-          placeholder={form.sink.auth_type === "api_key"
-            ? "API Key"
-            : form.sink.auth_type === "basic"
-              ? "username:password"
-              : "Bearer Token"}
+          disabled={form.sink.auth_type === "none"}
+          placeholder={form.sink.auth_type === "none"
+            ? "N/A"
+            : form.sink.auth_type === "api_key"
+              ? "API Key"
+              : form.sink.auth_type === "basic"
+                ? "username:password"
+                : "Bearer Token"}
           data-1p-ignore
           autocomplete="off"
         />
@@ -150,7 +147,9 @@
         <p class="text-destructive text-sm">{errors.sink.auth_value}</p>
       {/if}
       <p class="text-sm text-muted-foreground">
-        {#if form.sink.auth_type === "api_key"}
+        {#if form.sink.auth_type === "none"}
+          No authentication required
+        {:else if form.sink.auth_type === "api_key"}
           Your Elasticsearch API key
         {:else if form.sink.auth_type === "basic"}
           Basic auth credentials in format username:password
@@ -179,3 +178,35 @@
     </div>
   </CardContent>
 </Card>
+
+<Card>
+  <CardHeader>
+    <CardTitle>Routing</CardTitle>
+  </CardHeader>
+  <CardContent class="space-y-4">
+    <DynamicRoutingForm
+      bind:form
+      routedSinkType="elasticsearch"
+      {functions}
+      {refreshFunctions}
+      bind:functionRefreshState
+      bind:selectedDynamic
+      {errors}
+    />
+
+    {#if !selectedDynamic}
+      <div class="space-y-2">
+        <Label for="index_name">Index name</Label>
+        <Input
+          id="index_name"
+          name="sink[index_name]"
+          bind:value={form.sink.index_name}
+          placeholder="my-index"
+        />
+        {#if errors.sink?.index_name}
+          <p class="text-destructive text-sm">{errors.sink.index_name}</p>
+        {/if}
+      </div>
+    {/if}
+  </CardContent>
+</Card>

docker-compose.yaml

@@ -120,9 +120,23 @@ services:
       KAFKA_MAX_REQUEST_SIZE: 10485760
       KAFKA_REPLICA_FETCH_MAX_BYTES: 10485760
 
+  elasticsearch:
+    profiles: [elasticsearch]
+    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.1
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
+    ports:
+      - "127.0.0.1:9200:9200"
+    volumes:
+      - sequin_dev_elasticsearch:/usr/share/elasticsearch/data
+
+
 volumes:
   sequin_dev_postgres:
   sequin_dev_nats:
   sequin_dev_rabbitmq:
+  sequin_dev_elasticsearch:
   prometheus_data:
   grafana_data:

docs/reference/routing.mdx

@@ -39,6 +39,7 @@ The following sinks support dynamic routing:
 - GCP PubSub
 - Typesense
 - Meilisearch
+- Elasticsearch
 
 Each sink type has different fields that can be routed:
 
@@ -91,6 +92,14 @@ For [GCP PubSub](/reference/sinks/gcp-pubsub), your routing function must return
 |-----|------|-------------|---------|
 | `topic_id` | String | The topic ID to publish to | `"users.created"` |
 
+### Elasticsearch sink
+
+For [Elasticsearch](/reference/sinks/elasticsearch), your routing function must return a map with these keys:
+
+| Key | Type | Description | Example |
+|-----|------|-------------|---------|
+| `index_name` | String | The index to publish to | `"users"` |
+
 ### Typesense sink
 
 For [Typesense](/reference/sinks/typesense), your routing function must return a map with these keys:

docs/reference/sinks/elasticsearch.mdx

@@ -104,3 +104,19 @@ Errors from Elasticsearch are shown verbatim in the **Messages** tab. For bulk r
 - One sink targets one index; multiple indices require multiple sinks.
 - Only `index` and `delete` bulk operations are issued. Other operations (`update`, `create`, etc.) are not used.
 
+## Routing
+
+The Elasticsearch sink supports dynamic routing of the `index_name` with [routing functions](/reference/routing).
+
+Example routing function:
+
+```elixir
+def route(action, record, changes, metadata) do
+  %{
+    index_name: metadata.table_name
+  }
+end
+```
+
+When not using a routing function, messages will be indexed into the statically configured index.
+

lib/sequin/consumers/elasticsearch_sink.ex

@@ -12,20 +12,48 @@ defmodule Sequin.Consumers.ElasticsearchSink do
     field :type, Ecto.Enum, values: [:elasticsearch], default: :elasticsearch
     field :endpoint_url, :string
     field :index_name, :string
-    field :auth_type, Ecto.Enum, values: [:api_key, :basic, :bearer], default: :api_key
+    field :auth_type, Ecto.Enum, values: [:none, :api_key, :basic, :bearer], default: :api_key
     field :auth_value, Sequin.Encrypted.Binary
     field :batch_size, :integer, default: 100
+    field :routing_mode, Ecto.Enum, values: [:dynamic, :static]
   end
 
   def changeset(struct, params) do
     struct
-    |> cast(params, [:endpoint_url, :index_name, :auth_type, :auth_value, :batch_size])
-    |> validate_required([:endpoint_url, :index_name, :auth_type, :auth_value])
+    |> cast(params, [:endpoint_url, :index_name, :auth_type, :auth_value, :batch_size, :routing_mode])
+    |> validate_required([:endpoint_url, :auth_type])
+    |> validate_auth()
+    |> validate_routing()
     |> validate_endpoint_url()
     |> validate_length(:index_name, max: 1024)
     |> validate_number(:batch_size, greater_than: 0, less_than_or_equal_to: 10_000)
   end
 
+  defp validate_auth(changeset) do
+    auth_type = get_field(changeset, :auth_type)
+
+    if auth_type == :none do
+      put_change(changeset, :auth_value, nil)
+    else
+      validate_required(changeset, [:auth_value])
+    end
+  end
+
+  defp validate_routing(changeset) do
+    routing_mode = get_field(changeset, :routing_mode)
+
+    cond do
+      routing_mode == :dynamic ->
+        put_change(changeset, :index_name, nil)
+
+      routing_mode == :static ->
+        validate_required(changeset, [:index_name])
+
+      true ->
+        add_error(changeset, :routing_mode, "is required")
+    end
+  end
+
   defp validate_endpoint_url(changeset) do
     changeset
     |> validate_change(:endpoint_url, fn :endpoint_url, url ->