feat: add mysql sink

assets/svelte/consumers/ShowSink.svelte

@@ -27,12 +27,14 @@
     ElasticsearchConsumer,
     RedisStringConsumer,
     AzureEventHubConsumer,
+    MysqlConsumer,
   } from "./types";
   import AzureEventHubSinkCard from "../sinks/azure_event_hub/AzureEventHubSinkCard.svelte";
   import ElasticsearchSinkCard from "../sinks/elasticsearch/ElasticsearchSinkCard.svelte";
   import GcpPubsubSinkCard from "../sinks/gcp_pubsub/GcpPubsubSinkCard.svelte";
   import KafkaSinkCard from "../sinks/kafka/KafkaSinkCard.svelte";
   import KinesisSinkCard from "../sinks/kinesis/KinesisSinkCard.svelte";
+  import MysqlSinkCard from "../sinks/mysql/MysqlSinkCard.svelte";
   import NatsSinkCard from "../sinks/nats/NatsSinkCard.svelte";
   import RabbitMqSinkCard from "../sinks/rabbitmq/RabbitMqSinkCard.svelte";
   import RedisStreamSinkCard from "../sinks/redis-stream/RedisStreamSinkCard.svelte";
@@ -161,6 +163,10 @@
     return consumer.sink.type === "rabbitmq";
   }
 
+  function isMysqlConsumer(consumer: Consumer): consumer is MysqlConsumer {
+    return consumer.sink.type === "mysql";
+  }
+
   let chartElement;
   let updateChart;
   let resizeObserver;
@@ -1254,6 +1260,8 @@
         <ElasticsearchSinkCard {consumer} />
       {:else if isRedisStringConsumer(consumer)}
         <RedisStringSinkCard {consumer} />
+      {:else if isMysqlConsumer(consumer)}
+        <MysqlSinkCard {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 MysqlIcon from "../sinks/mysql/MysqlIcon.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 === "mysql"}
+            <MysqlIcon 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 MysqlSinkForm from "$lib/sinks/mysql/MysqlSinkForm.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";
@@ -797,6 +798,14 @@
         {refreshFunctions}
         bind:functionRefreshState
       />
+    {:else if consumer.type === "mysql"}
+      <MysqlSinkForm
+        errors={errors.consumer}
+        bind:form
+        {functions}
+        {refreshFunctions}
+        bind:functionRefreshState
+      />
     {/if}
 
     <Card>

assets/svelte/consumers/SinkIndex.svelte

@@ -33,6 +33,7 @@
   import TypesenseIcon from "../sinks/typesense/TypesenseIcon.svelte";
   import MeilisearchIcon from "../sinks/meilisearch/MeilisearchIcon.svelte";
   import ElasticsearchIcon from "../sinks/elasticsearch/ElasticsearchIcon.svelte";
+  import MysqlIcon from "../sinks/mysql/MysqlIcon.svelte";
 
   import { Badge } from "$lib/components/ui/badge";
   import * as d3 from "d3";
@@ -56,7 +57,8 @@
       | "nats"
       | "rabbitmq"
       | "typesense"
-      | "elasticsearch";
+      | "elasticsearch"
+      | "mysql";
 
     status: "active" | "disabled" | "paused";
     database_name: string;
@@ -71,6 +73,7 @@
   export let live: any;
   export let hasDatabases: boolean;
   export let selfHosted: boolean;
+  export let isMysqlSinkEnabled: boolean;
   export let page: number;
   export let pageSize: number;
   export let totalCount: number;
@@ -83,6 +86,15 @@
   $: startIndex = page * pageSize + 1;
   $: endIndex = Math.min((page + 1) * pageSize, totalCount);
 
+  function isSinkEnabled(sink: { id: string }): boolean {
+    switch (sink.id) {
+      case "mysql":
+        return isMysqlSinkEnabled;
+      default:
+        return true;
+    }
+  }
+
   const sinks = [
     {
       id: "http_push",
@@ -164,7 +176,12 @@
       name: "Elasticsearch",
       icon: ElasticsearchIcon,
     },
-  ];
+    {
+      id: "mysql",
+      name: "MySQL",
+      icon: MysqlIcon,
+    },
+  ].filter(isSinkEnabled);
 
   function handleConsumerClick(id: string, type: string) {
     // Store current page before navigation

assets/svelte/consumers/dynamicRoutingDocs.ts

@@ -231,4 +231,14 @@ export const routedSinkDocs: Record<RoutedSinkType, RoutedSinkDocs> = {
       },
     },
   },
+  mysql: {
+    fields: {
+      table_name: {
+        description: "MySQL table name to write records to",
+        staticValue: "<configured-table-name>",
+        staticFormField: "table_name",
+        dynamicDefault: "<source-schema>_<source-table>",
+      },
+    },
+  },
 };

assets/svelte/consumers/types.ts

@@ -247,6 +247,24 @@ export type ElasticsearchConsumer = BaseConsumer & {
   };
 };
 
+// MySQL specific sink
+export type MysqlConsumer = BaseConsumer & {
+  sink: {
+    type: "mysql";
+    host: string;
+    port: number;
+    database: string;
+    table_name: string;
+    username: string;
+    password?: string;
+    ssl: boolean;
+    batch_size: number;
+    timeout_seconds: number;
+    upsert_on_duplicate: boolean;
+    routing_mode: "static" | "dynamic";
+  };
+};
+
 // Union type for all consumer types
 export type Consumer =
   | HttpPushConsumer
@@ -263,24 +281,27 @@ export type Consumer =
   | TypesenseConsumer
   | SnsConsumer
   | ElasticsearchConsumer
+  | MysqlConsumer
   | RedisStringConsumer;
 
 export const SinkTypeValues = [
   "http_push",
   "sqs",
   "sns",
   "kinesis",
+  "s2",
   "redis_stream",
+  "redis_string",
   "kafka",
   "sequin_stream",
   "gcp_pubsub",
-  "elasticsearch",
   "nats",
   "rabbitmq",
+  "azure_event_hub",
   "typesense",
   "meilisearch",
   "elasticsearch",
-  "redis_string",
+  "mysql",
 ] as const;
 
 export type SinkType = (typeof SinkTypeValues)[number];
@@ -301,6 +322,7 @@ export const RoutedSinkTypeValues = [
   "sqs",
   "sns",
   "kinesis",
+  "mysql",
 ] as const;
 
 export type RoutedSinkType = (typeof RoutedSinkTypeValues)[number];

assets/svelte/sinks/mysql/MysqlIcon.svelte

@@ -0,0 +1,25 @@
+<svg
+  width="800px"
+  height="800px"
+  viewBox="-18.458 -22.75 191.151 191.151"
+  xmlns="http://www.w3.org/2000/svg"
+  class={$$props.class}
+>
+  <path d="M-18.458 6.58h191.151v132.49H-18.458V6.58z" fill="none" />
+  <path
+    d="M40.054 113.583h-5.175c-.183-8.735-.687-16.947-1.511-24.642h-.046l-7.879 24.642h-3.94l-7.832-24.642h-.045c-.581 7.388-.947 15.602-1.099 24.642H7.81c.304-10.993 1.068-21.299 2.289-30.919h6.414l7.465 22.719h.046l7.511-22.719h6.137c1.344 11.268 2.138 21.575 2.382 30.919M62.497 90.771c-2.107 11.434-4.887 19.742-8.337 24.928-2.688 3.992-5.633 5.99-8.84 5.99-.855 0-1.91-.258-3.16-.77v-2.757c.611.088 1.328.138 2.152.138 1.498 0 2.702-.412 3.62-1.238 1.098-1.006 1.647-2.137 1.647-3.388 0-.858-.428-2.612-1.282-5.268L42.618 90.77h5.084l4.076 13.19c.916 2.995 1.298 5.086 1.145 6.277 2.229-5.953 3.786-12.444 4.673-19.468h4.901v.002z"
+    fill="#5d87a1"
+  />
+  <path
+    d="M131.382 113.583h-14.7V82.664h4.945v27.113h9.755v3.806zM112.834 114.33l-5.684-2.805c.504-.414.986-.862 1.42-1.381 2.416-2.838 3.621-7.035 3.621-12.594 0-10.229-4.014-15.346-12.045-15.346-3.938 0-7.01 1.298-9.207 3.895-2.414 2.84-3.619 7.022-3.619 12.551 0 5.435 1.068 9.422 3.205 11.951 1.955 2.291 4.902 3.438 8.843 3.438 1.47 0 2.819-.18 4.048-.543l7.4 4.308 2.018-3.474zm-18.413-6.934c-1.252-2.014-1.878-5.248-1.878-9.707 0-7.785 2.365-11.682 7.1-11.682 2.475 0 4.289.932 5.449 2.792 1.25 2.017 1.879 5.222 1.879 9.619 0 7.849-2.367 11.774-7.099 11.774-2.476.001-4.29-.928-5.451-2.796M85.165 105.013c0 2.622-.962 4.773-2.884 6.458-1.924 1.678-4.504 2.519-7.737 2.519-3.024 0-5.956-.966-8.794-2.888l1.329-2.655c2.442 1.223 4.653 1.831 6.638 1.831 1.863 0 3.319-.413 4.375-1.232 1.055-.822 1.684-1.975 1.684-3.433 0-1.837-1.281-3.407-3.631-4.722-2.167-1.19-6.501-3.678-6.501-3.678-2.349-1.712-3.525-3.55-3.525-6.578 0-2.506.877-4.529 2.632-6.068 1.757-1.545 4.024-2.315 6.803-2.315 2.87 0 5.479.769 7.829 2.291l-1.192 2.656c-2.01-.854-3.994-1.281-5.951-1.281-1.585 0-2.809.381-3.66 1.146-.858.762-1.387 1.737-1.387 2.933 0 1.828 1.308 3.418 3.722 4.759 2.196 1.192 6.638 3.723 6.638 3.723 2.409 1.709 3.612 3.53 3.612 6.534"
+    fill="#f8981d"
+  />
+  <path
+    d="M137.59 72.308c-2.99-.076-5.305.225-7.248 1.047-.561.224-1.453.224-1.531.933.303.3.338.784.601 1.198.448.747 1.229 1.752 1.942 2.276.783.6 1.569 1.194 2.393 1.717 1.453.899 3.1 1.422 4.516 2.318.825.521 1.645 1.195 2.471 1.756.406.299.666.784 1.193.971v-.114c-.264-.336-.339-.822-.598-1.196l-1.122-1.082c-1.084-1.456-2.431-2.727-3.884-3.771-1.196-.824-3.812-1.944-4.297-3.322l-.076-.076c.822-.077 1.797-.375 2.578-.604 1.271-.335 2.43-.259 3.734-.594.6-.15 1.195-.338 1.797-.523v-.337c-.676-.673-1.158-1.567-1.869-2.203-1.902-1.643-3.998-3.25-6.164-4.595-1.16-.749-2.652-1.231-3.887-1.868-.445-.225-1.195-.336-1.457-.71-.67-.822-1.047-1.904-1.533-2.877-1.08-2.053-2.129-4.331-3.061-6.502-.674-1.456-1.084-2.91-1.906-4.257-3.85-6.35-8.031-10.196-14.457-13.971-1.381-.786-3.024-1.121-4.779-1.533l-2.803-.148c-.598-.262-1.197-.973-1.719-1.309-2.132-1.344-7.621-4.257-9.189-.411-1.01 2.431 1.494 4.821 2.354 6.054.635.856 1.458 1.83 1.902 2.802.263.635.337 1.309.6 1.98.598 1.644 1.157 3.473 1.943 5.007.41.782.857 1.604 1.381 2.312.3.414.822.597.936 1.272-.521.744-.562 1.867-.861 2.801-1.344 4.221-.819 9.45 1.086 12.552.596.934 2.018 2.99 3.92 2.202 1.684-.672 1.311-2.801 1.795-4.668.111-.451.038-.747.262-1.043v.073c.521 1.045 1.047 2.052 1.53 3.1 1.159 1.829 3.177 3.735 4.858 5.002.895.676 1.604 1.832 2.725 2.245V74.1h-.074c-.227-.335-.559-.485-.857-.745-.674-.673-1.42-1.495-1.943-2.241-1.566-2.093-2.952-4.41-4.182-6.801-.602-1.16-1.121-2.428-1.606-3.586-.226-.447-.226-1.121-.601-1.346-.562.821-1.381 1.532-1.791 2.538-.711 1.609-.785 3.588-1.049 5.646l-.147.072c-1.19-.299-1.604-1.53-2.056-2.575-1.119-2.654-1.307-6.914-.336-9.976.26-.783 1.385-3.249.936-3.995-.225-.715-.973-1.122-1.383-1.685-.482-.708-1.01-1.604-1.346-2.39-.896-2.091-1.347-4.408-2.312-6.498-.451-.974-1.234-1.982-1.868-2.879-.712-1.008-1.495-1.718-2.058-2.913-.186-.411-.447-1.083-.148-1.53.073-.3.225-.412.523-.487.484-.409 1.867.111 2.352.336 1.385.56 2.543 1.083 3.699 1.867.523.375 1.084 1.085 1.755 1.272h.786c1.193.26 2.538.072 3.661.41 1.979.636 3.772 1.569 5.38 2.576 4.893 3.103 8.928 7.512 11.652 12.778.447.858.637 1.644 1.045 2.539.787 1.832 1.76 3.7 2.541 5.493.785 1.755 1.533 3.547 2.654 5.005.559.784 2.805 1.195 3.812 1.606.745.335 1.905.633 2.577 1.044 1.271.783 2.537 1.682 3.732 2.543.595.448 2.465 1.382 2.576 2.13M99.484 39.844a5.82 5.82 0 0 0-1.529.188v.075h.072c.301.597.824 1.011 1.197 1.532.301.599.562 1.193.857 1.791l.072-.074c.527-.373.789-.971.789-1.868-.227-.264-.262-.522-.451-.784-.22-.374-.705-.56-1.007-.86"
+    fill="#5d87a1"
+  />
+  <path
+    d="M141.148 113.578h.774v-3.788h-1.161l-.947 2.585-1.029-2.585h-1.118v3.788h.731v-2.882h.041l1.078 2.882h.557l1.074-2.882v2.882zm-6.235 0h.819v-3.146h1.072v-.643h-3.008v.643h1.115l.002 3.146z"
+    fill="#f8981d"
+  />
+</svg>

assets/svelte/sinks/mysql/MysqlSinkCard.svelte

@@ -0,0 +1,137 @@
+<script lang="ts">
+  import { ExternalLink } from "lucide-svelte";
+  import {
+    Card,
+    CardContent,
+    CardHeader,
+    CardTitle,
+  } from "$lib/components/ui/card";
+  import type { MysqlConsumer } from "../../consumers/types";
+
+  export let consumer: MysqlConsumer;
+
+  function getRoutingCode(consumer: MysqlConsumer): string | null {
+    if (!consumer.routing || !consumer.routing.function) return null;
+    const func = consumer.routing.function;
+    if (func.type === "routing") {
+      return (func as any).code || null;
+    }
+    return null;
+  }
+</script>
+
+<Card>
+  <CardHeader>
+    <CardTitle>MySQL Configuration</CardTitle>
+  </CardHeader>
+  <CardContent class="p-6">
+    <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 px-2 py-1 border border-slate-100 rounded-md break-all w-fit"
+          >
+            <span>{consumer.sink.host}:{consumer.sink.port}</span>
+          </div>
+        </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">Username</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.username}
+          </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>
+        <span class="text-sm text-muted-foreground">Batch Size</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.batch_size}
+          </span>
+        </div>
+      </div>
+
+      <div>
+        <span class="text-sm text-muted-foreground">Upsert Mode</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.upsert_on_duplicate ? "Enabled" : "Disabled"}
+          </span>
+        </div>
+      </div>
+    </div>
+  </CardContent>
+</Card>
+
+<Card>
+  <CardHeader>
+    <CardTitle>Routing</CardTitle>
+  </CardHeader>
+  <CardContent>
+    <div>
+      <span class="text-sm text-muted-foreground">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-muted-foreground">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>