✨ Postgres sink

Author: RTLS

assets/svelte/consumers/ShowSink.svelte

@@ -27,6 +27,7 @@
     ElasticsearchConsumer,
     RedisStringConsumer,
     AzureEventHubConsumer,
+    PostgresSinkConsumer,
   } from "./types";
   import AzureEventHubSinkCard from "../sinks/azure_event_hub/AzureEventHubSinkCard.svelte";
   import ElasticsearchSinkCard from "../sinks/elasticsearch/ElasticsearchSinkCard.svelte";
@@ -48,6 +49,8 @@
   import HealthAlerts from "$lib/health/HealthAlerts.svelte";
   import { Button } from "$lib/components/ui/button";
   import CollapsibleCode from "../components/CollapsibleCode.svelte";
+  import PostgresSinkCard from "../sinks/postgres/PostgresSinkCard.svelte";
+  import type { Table } from "../databases/types";
 
   export let live;
   export let parent;
@@ -161,6 +164,12 @@
     return consumer.sink.type === "rabbitmq";
   }
 
+  function isPostgresSinkConsumer(
+    consumer: Consumer,
+  ): consumer is PostgresSinkConsumer {
+    return consumer.sink.type === "postgres";
+  }
+
   let chartElement;
   let updateChart;
   let resizeObserver;
@@ -761,6 +770,15 @@
       unit: units[unitIndex],
     };
   }
+
+  function getRoutingCode(consumer: Consumer): string | null {
+    if (!consumer.routing || !consumer.routing.function) return null;
+    const func = consumer.routing.function;
+    if (func.type === "routing" && "code" in func) {
+      return func.code;
+    }
+    return null;
+  }
 </script>
 
 <div class="flex flex-col flex-1">
@@ -1254,6 +1272,8 @@
         <ElasticsearchSinkCard {consumer} />
       {:else if isRedisStringConsumer(consumer)}
         <RedisStringSinkCard {consumer} />
+      {:else if isPostgresSinkConsumer(consumer)}
+        <PostgresSinkCard {consumer} />
       {/if}
 
       <ShowSource {consumer} {tables} />

assets/svelte/consumers/ShowSinkHeader.svelte

@@ -32,6 +32,7 @@
   import AzureEventHubIcon from "../sinks/azure_event_hub/AzureEventHubIcon.svelte";
   import TypesenseIcon from "../sinks/typesense/TypesenseIcon.svelte";
   import ElasticsearchIcon from "../sinks/elasticsearch/ElasticsearchIcon.svelte";
+  import PostgresIcon from "../sinks/postgres/PostgresIcon.svelte";
   import StopSinkModal from "./StopSinkModal.svelte";
   import { Badge } from "$lib/components/ui/badge";
 
@@ -186,6 +187,8 @@
             <KinesisIcon class="h-6 w-6 mr-2" />
           {:else if consumer.sink.type === "s2"}
             <S2Icon class="h-6 w-6 mr-2" />
+          {:else if consumer.sink.type === "postgres"}
+            <PostgresIcon class="h-6 w-6 mr-2" />
           {/if}
           <h1 class="text-xl font-semibold">
             {consumer.name}

assets/svelte/consumers/SinkConsumerForm.svelte

@@ -36,6 +36,7 @@
   import TypesenseSinkForm from "$lib/sinks/typesense/TypesenseSinkForm.svelte";
   import MeilisearchSinkForm from "$lib/sinks/meilisearch/MeilisearchSinkForm.svelte";
   import ElasticsearchSinkForm from "$lib/sinks/elasticsearch/ElasticsearchSinkForm.svelte";
+  import PostgresSinkForm from "$lib/sinks/postgres/PostgresSinkForm.svelte";
   import * as Alert from "$lib/components/ui/alert/index.js";
   import SchemaTableSelector from "../components/SchemaTableSelector.svelte";
   import * as Tooltip from "$lib/components/ui/tooltip";
@@ -731,6 +732,14 @@
         {functions}
         {functionRefreshState}
       />
+    {:else if consumer.type === "postgres"}
+      <PostgresSinkForm
+        errors={errors.consumer}
+        bind:form
+        {functions}
+        {refreshFunctions}
+        bind:functionRefreshState
+      />
     {:else if consumer.type === "kafka"}
       <KafkaSinkForm
         errors={errors.consumer}

assets/svelte/consumers/SinkIndex.svelte

@@ -29,10 +29,10 @@
   import SequinStreamIcon from "../sinks/sequin_stream/SequinStreamIcon.svelte";
   import NatsIcon from "../sinks/nats/NatsIcon.svelte";
   import RabbitMqIcon from "../sinks/rabbitmq/RabbitMqIcon.svelte";
-
   import TypesenseIcon from "../sinks/typesense/TypesenseIcon.svelte";
   import MeilisearchIcon from "../sinks/meilisearch/MeilisearchIcon.svelte";
   import ElasticsearchIcon from "../sinks/elasticsearch/ElasticsearchIcon.svelte";
+  import PostgresIcon from "../sinks/postgres/PostgresIcon.svelte";
 
   import { Badge } from "$lib/components/ui/badge";
   import * as d3 from "d3";
@@ -56,7 +56,8 @@
       | "nats"
       | "rabbitmq"
       | "typesense"
-      | "elasticsearch";
+      | "elasticsearch"
+      | "postgres";
 
     status: "active" | "disabled" | "paused";
     database_name: string;
@@ -164,6 +165,11 @@
       name: "Elasticsearch",
       icon: ElasticsearchIcon,
     },
+    {
+      id: "postgres",
+      name: "Postgres",
+      icon: PostgresIcon,
+    },
   ];
 
   function handleConsumerClick(id: string, type: string) {

assets/svelte/consumers/dynamicRoutingDocs.ts

@@ -231,4 +231,14 @@ export const routedSinkDocs: Record<RoutedSinkType, RoutedSinkDocs> = {
       },
     },
   },
+  postgres: {
+    fields: {
+      table_name: {
+        description: "Postgres table name",
+        staticValue: "<empty>",
+        staticFormField: "table_name",
+        dynamicDefault: "<empty>",
+      },
+    },
+  },
 };

assets/svelte/consumers/types.ts

@@ -33,6 +33,7 @@ export type BaseConsumer = {
   message_grouping: boolean;
   ack_wait_ms: number;
   max_ack_pending: number;
+  max_retry_count: number;
   max_deliver: number;
   max_waiting: number;
   inserted_at: string;
@@ -247,6 +248,20 @@ export type ElasticsearchConsumer = BaseConsumer & {
   };
 };
 
+// PostgreSQL specific sink
+export type PostgresSinkConsumer = BaseConsumer & {
+  sink: {
+    type: "postgres";
+    host: string;
+    port: number;
+    database: string;
+    table_name: string;
+    username: string;
+    ssl: boolean;
+    routing_mode: "dynamic" | "static";
+  };
+};
+
 // Union type for all consumer types
 export type Consumer =
   | HttpPushConsumer
@@ -263,7 +278,8 @@ export type Consumer =
   | TypesenseConsumer
   | SnsConsumer
   | ElasticsearchConsumer
-  | RedisStringConsumer;
+  | RedisStringConsumer
+  | PostgresSinkConsumer;
 
 export const SinkTypeValues = [
   "http_push",
@@ -281,6 +297,7 @@ export const SinkTypeValues = [
   "meilisearch",
   "elasticsearch",
   "redis_string",
+  "postgres",
 ] as const;
 
 export type SinkType = (typeof SinkTypeValues)[number];
@@ -301,6 +318,7 @@ export const RoutedSinkTypeValues = [
   "sqs",
   "sns",
   "kinesis",
+  "postgres",
 ] as const;
 
 export type RoutedSinkType = (typeof RoutedSinkTypeValues)[number];

assets/svelte/sinks/postgres/PostgresIcon.svelte

@@ -0,0 +1,58 @@
+<svg
+  width="432.071pt"
+  height="445.383pt"
+  viewBox="0 0 432.071 445.383"
+  xml:space="preserve"
+  xmlns="http://www.w3.org/2000/svg"
+  class={$$props.class}
+>
+  <g
+    id="orginal"
+    style="fill-rule:nonzero;clip-rule:nonzero;stroke:#000000;stroke-miterlimit:4;"
+  >
+  </g>
+  <g
+    id="Layer_x0020_3"
+    style="fill-rule:nonzero;clip-rule:nonzero;fill:none;stroke:#FFFFFF;stroke-width:12.4651;stroke-linecap:round;stroke-linejoin:round;stroke-miterlimit:4;"
+  >
+    <path
+      style="fill:#000000;stroke:#000000;stroke-width:37.3953;stroke-linecap:butt;stroke-linejoin:miter;"
+      d="M323.205,324.227c2.833-23.601,1.984-27.062,19.563-23.239l4.463,0.392c13.517,0.615,31.199-2.174,41.587-7c22.362-10.376,35.622-27.7,13.572-23.148c-50.297,10.376-53.755-6.655-53.755-6.655c53.111-78.803,75.313-178.836,56.149-203.322    C352.514-5.534,262.036,26.049,260.522,26.869l-0.482,0.089c-9.938-2.062-21.06-3.294-33.554-3.496c-22.761-0.374-40.032,5.967-53.133,15.904c0,0-161.408-66.498-153.899,83.628c1.597,31.936,45.777,241.655,98.47,178.31    c19.259-23.163,37.871-42.748,37.871-42.748c9.242,6.14,20.307,9.272,31.912,8.147l0.897-0.765c-0.281,2.876-0.157,5.689,0.359,9.019c-13.572,15.167-9.584,17.83-36.723,23.416c-27.457,5.659-11.326,15.734-0.797,18.367c12.768,3.193,42.305,7.716,62.268-20.224    l-0.795,3.188c5.325,4.26,4.965,30.619,5.72,49.452c0.756,18.834,2.017,36.409,5.856,46.771c3.839,10.36,8.369,37.05,44.036,29.406c29.809-6.388,52.6-15.582,54.677-101.107"
+    />
+    <path
+      style="fill:#336791;stroke:none;"
+      d="M402.395,271.23c-50.302,10.376-53.76-6.655-53.76-6.655c53.111-78.808,75.313-178.843,56.153-203.326c-52.27-66.785-142.752-35.2-144.262-34.38l-0.486,0.087c-9.938-2.063-21.06-3.292-33.56-3.496c-22.761-0.373-40.026,5.967-53.127,15.902    c0,0-161.411-66.495-153.904,83.63c1.597,31.938,45.776,241.657,98.471,178.312c19.26-23.163,37.869-42.748,37.869-42.748c9.243,6.14,20.308,9.272,31.908,8.147l0.901-0.765c-0.28,2.876-0.152,5.689,0.361,9.019c-13.575,15.167-9.586,17.83-36.723,23.416    c-27.459,5.659-11.328,15.734-0.796,18.367c12.768,3.193,42.307,7.716,62.266-20.224l-0.796,3.188c5.319,4.26,9.054,27.711,8.428,48.969c-0.626,21.259-1.044,35.854,3.147,47.254c4.191,11.4,8.368,37.05,44.042,29.406c29.809-6.388,45.256-22.942,47.405-50.555    c1.525-19.631,4.976-16.729,5.194-34.28l2.768-8.309c3.192-26.611,0.507-35.196,18.872-31.203l4.463,0.392c13.517,0.615,31.208-2.174,41.591-7c22.358-10.376,35.618-27.7,13.573-23.148z"
+    />
+    <path
+      d="M215.866,286.484c-1.385,49.516,0.348,99.377,5.193,111.495c4.848,12.118,15.223,35.688,50.9,28.045c29.806-6.39,40.651-18.756,45.357-46.051c3.466-20.082,10.148-75.854,11.005-87.281"
+    />
+    <path
+      d="M173.104,38.256c0,0-161.521-66.016-154.012,84.109c1.597,31.938,45.779,241.664,98.473,178.316c19.256-23.166,36.671-41.335,36.671-41.335"
+    />
+    <path
+      d="M260.349,26.207c-5.591,1.753,89.848-34.889,144.087,34.417c19.159,24.484-3.043,124.519-56.153,203.329"
+    />
+    <path
+      style="stroke-linejoin:bevel;"
+      d="M348.282,263.953c0,0,3.461,17.036,53.764,6.653c22.04-4.552,8.776,12.774-13.577,23.155c-18.345,8.514-59.474,10.696-60.146-1.069c-1.729-30.355,21.647-21.133,19.96-28.739c-1.525-6.85-11.979-13.573-18.894-30.338    c-6.037-14.633-82.796-126.849,21.287-110.183c3.813-0.789-27.146-99.002-124.553-100.599c-97.385-1.597-94.19,119.762-94.19,119.762"
+    />
+    <path
+      d="M188.604,274.334c-13.577,15.166-9.584,17.829-36.723,23.417c-27.459,5.66-11.326,15.733-0.797,18.365c12.768,3.195,42.307,7.718,62.266-20.229c6.078-8.509-0.036-22.086-8.385-25.547c-4.034-1.671-9.428-3.765-16.361,3.994z"
+    />
+    <path
+      d="M187.715,274.069c-1.368-8.917,2.93-19.528,7.536-31.942c6.922-18.626,22.893-37.255,10.117-96.339c-9.523-44.029-73.396-9.163-73.436-3.193c-0.039,5.968,2.889,30.26-1.067,58.548c-5.162,36.913,23.488,68.132,56.479,64.938"
+    />
+    <path
+      style="fill:#FFFFFF;stroke-width:4.155;stroke-linecap:butt;stroke-linejoin:miter;"
+      d="M172.517,141.7c-0.288,2.039,3.733,7.48,8.976,8.207c5.234,0.73,9.714-3.522,9.998-5.559c0.284-2.039-3.732-4.285-8.977-5.015c-5.237-0.731-9.719,0.333-9.996,2.367z"
+    />
+    <path
+      style="fill:#FFFFFF;stroke-width:2.0775;stroke-linecap:butt;stroke-linejoin:miter;"
+      d="M331.941,137.543c0.284,2.039-3.732,7.48-8.976,8.207c-5.238,0.73-9.718-3.522-10.005-5.559c-0.277-2.039,3.74-4.285,8.979-5.015c5.239-0.73,9.718,0.333,10.002,2.368z"
+    />
+    <path
+      d="M350.676,123.432c0.863,15.994-3.445,26.888-3.988,43.914c-0.804,24.748,11.799,53.074-7.191,81.435"
+    />
+    <path style="stroke-width:3;" d="M0,60.232" />
+  </g>
+</svg>

assets/svelte/sinks/postgres/PostgresSinkCard.svelte

@@ -0,0 +1,119 @@
+<script lang="ts">
+  import { ExternalLink } from "lucide-svelte";
+  import { Card, CardContent, CardTitle } from "$lib/components/ui/card";
+  import CardHeader from "$lib/components/ui/card/card-header.svelte";
+  import type { PostgresSinkConsumer } from "../../consumers/types";
+
+  export let consumer: PostgresSinkConsumer;
+
+  function getRoutingCode(consumer: PostgresSinkConsumer): string | null {
+    if (!consumer.routing || !consumer.routing.function) return null;
+    const func = consumer.routing.function;
+    if (func.type === "routing" && "code" in func) {
+      return func.code;
+    }
+    return null;
+  }
+</script>
+
+<Card>
+  <CardContent class="p-6">
+    <div class="flex justify-between items-center mb-4">
+      <h2 class="text-lg font-semibold">PostgreSQL Configuration</h2>
+    </div>
+
+    <div class="grid gap-4">
+      <div>
+        <span class="text-sm text-muted-foreground">Host</span>
+        <div class="mt-2">
+          <div
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md break-all w-fit"
+          >
+            <span>{consumer.sink.host}</span>
+          </div>
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">Port</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.port}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">Database</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.database}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">Table Name</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.table_name}</span
+          >
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">SSL Enabled</span>
+        <div class="mt-2">
+          <span
+            class="font-mono bg-slate-50 py-1 px-2 border border-slate-100 rounded-md whitespace-nowrap"
+            >{consumer.sink.ssl ? "Yes" : "No"}</span
+          >
+        </div>
+      </div>
+    </div>
+  </CardContent>
+</Card>
+
+<Card>
+  <CardHeader>
+    <CardTitle>Routing</CardTitle>
+  </CardHeader>
+  <CardContent>
+    <div>
+      <span class="text-sm text-gray-500">Table</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}
+            {consumer.sink.table_name}
+          {/if}
+        </span>
+      </div>
+    </div>
+    {#if consumer.routing}
+      {#if getRoutingCode(consumer)}
+        <div class="mt-2">
+          <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
+                >{getRoutingCode(consumer)}</code
+              ></pre>
+          </div>
+        </div>
+      {/if}
+    {/if}
+  </CardContent>
+</Card>