Fix: Ensure benchmark sinks not production in BenchmarkSink

Author: RTLS

lib/sequin/consumers/benchmark_sink.ex

@@ -22,5 +22,14 @@ defmodule Sequin.Consumers.BenchmarkSink do
     struct
     |> cast(params, [:partition_count])
     |> validate_number(:partition_count, greater_than: 0)
+    |> ensure_not_production()
+  end
+
+  defp ensure_not_production(changeset) do
+    if Application.get_env(:sequin, :env) == :prod do
+      add_error(changeset, :type, "benchmark consumers are not allowed in production")
+    else
+      changeset
+    end
   end
 end

lib/sequin/consumers/sink_consumer.ex

@@ -207,7 +207,6 @@ defmodule Sequin.Consumers.SinkConsumer do
     |> cast_embed(:source, required: true)
     |> cast_embed(:source_tables)
     |> put_defaults()
-    |> validate_type()
     |> validate_message_grouping()
     |> validate_enrichment()
     |> validate_required([:name, :status, :replication_slot_id, :batch_size])
@@ -227,16 +226,6 @@ defmodule Sequin.Consumers.SinkConsumer do
     |> Sequin.Changeset.annotations_check_constraint()
   end
 
-  defp validate_type(changeset) do
-    sink = get_field(changeset, :sink)
-
-    if sink && sink.type == :benchmark && Application.get_env(:sequin, :env) == :prod do
-      add_error(changeset, :type, "invalid type: #{inspect(sink.type)}")
-    else
-      changeset
-    end
-  end
-
   defp validate_message_grouping(changeset) do
     message_grouping = get_field(changeset, :message_grouping)
     source_tables = get_field(changeset, :source_tables) || []