refactor: remove legacy_event_singleton_transform

lib/sequin/consumers/consumers.ex

@@ -448,20 +448,12 @@ defmodule Sequin.Consumers do
     |> Repo.all()
   end
 
-  @legacy_event_singleton_transform_cutoff_date ~D[2024-11-06]
   def consumer_features(%SinkConsumer{} = consumer) do
     consumer = Repo.lazy_preload(consumer, [:account])
 
-    cond do
-      Accounts.has_feature?(consumer.account, :legacy_event_transform) ->
-        [legacy_event_transform: true]
-
-      Date.before?(consumer.account.inserted_at, @legacy_event_singleton_transform_cutoff_date) ->
-        [legacy_event_singleton_transform: true]
-
-      true ->
-        []
-    end
+    if Accounts.has_feature?(consumer.account, :legacy_event_transform),
+      do: [legacy_event_transform: true],
+      else: []
   end
 
   # ConsumerEvent

lib/sequin/runtime/http_push_pipeline.ex

@@ -108,10 +108,6 @@ defmodule Sequin.Runtime.HttpPushPipeline do
         [message] = messages
         legacy_event_transform_message(consumer, message.data)
 
-      features[:legacy_event_singleton_transform] && length(messages) == 1 ->
-        [message] = messages
-        message.data
-
       consumer.sink.batch == false ->
         [message] = messages
         Transforms.Message.to_external(consumer, message)

priv/repo/migrations/20250522134759_remove_legacy_event_singleton_transform_config.exs

@@ -0,0 +1,24 @@
+defmodule Sequin.Repo.Migrations.RemoveLegacyEventSingletonTransformConfig do
+  use Ecto.Migration
+
+  @config_schema Application.compile_env(:sequin, [Sequin.Repo, :config_schema_prefix])
+
+  def change do
+    execute(
+      """
+      update #{@config_schema}.sink_consumers
+      set
+        sink = jsonb_set(sink, '{batch}', 'false'::jsonb),
+        batch_size = 1
+      where
+        type = 'http_push'
+        and account_id in (
+          select id
+          from #{@config_schema}.accounts
+          where inserted_at <= '2024-11-06'
+        )
+      """,
+      ""
+    )
+  end
+end

test/sequin/consumers_test.exs

@@ -2271,7 +2271,7 @@ defmodule Sequin.ConsumersTest do
       assert Consumers.consumer_features(consumer) == []
     end
 
-    test "returns legacy_event_singleton_transform when account is old enough" do
+    test "not return legacy_event_singleton_transform when account is old enough" do
       account = AccountsFactory.account(features: [], inserted_at: ~D[2024-11-01])
 
       event_table = DatabasesFactory.event_table()
@@ -2285,7 +2285,7 @@ defmodule Sequin.ConsumersTest do
           type: :http_push
         )
 
-      assert Consumers.consumer_features(consumer) == [{:legacy_event_singleton_transform, true}]
+      refute Consumers.consumer_features(consumer) == [{:legacy_event_singleton_transform, true}]
     end
   end
 

test/sequin/http_push_pipeline_test.exs

@@ -139,33 +139,6 @@ defmodule Sequin.Runtime.HttpPushPipelineTest do
       assert_receive :sent, 1_000
     end
 
-    test "legacy_event_singleton_transform sends unwrapped single messages", %{
-      consumer: consumer,
-      http_endpoint: http_endpoint
-    } do
-      test_pid = self()
-      event = ConsumersFactory.insert_consumer_event!(consumer_id: consumer.id, action: :insert)
-
-      adapter = fn %Req.Request{} = req ->
-        assert to_string(req.url) == HttpEndpoint.url(http_endpoint)
-        json = Jason.decode!(req.body)
-
-        # Should NOT be wrapped in a list
-        refute is_list(json)
-        assert json["action"] == "insert"
-
-        send(test_pid, :sent)
-        {req, Req.Response.new(status: 200)}
-      end
-
-      # Start pipeline with legacy_event_singleton_transform enabled
-      start_pipeline!(consumer, adapter, features: [legacy_event_singleton_transform: true])
-
-      ref = send_test_event(consumer, event)
-      assert_receive {:ack, ^ref, [%{data: %{data: %{action: :insert}}}], []}, 1_000
-      assert_receive :sent, 1_000
-    end
-
     test "when all messages are rejected due to idempotency, the pipeline does not invoke the adapter", %{
       consumer: consumer
     } do