fix: apply enrichment to sequin_stream messages

sequin_stream was bypassing the SinkPipeline (which applies enrichment for push sinks like NATS). This commit adds enrichment processing in PullController.receive() to match the behavior of push sinks.

Changes:
- Add :enrichment and :postgres_database to cached_sink_consumer_for_sequin_stream preloads
- Add enrich_messages/2 helper that groups messages by table_oid for efficient batched enrichment
- Add test for enrichment in sequin_stream pull API

Author: acco

lib/sequin/consumers/consumers.ex

@@ -414,7 +414,11 @@ defmodule Sequin.Consumers do
     Cache.get_or_store(
       cache_key,
       fn ->
-        find_sink_consumer(account_id, id_or_name: id_or_name, type: :sequin_stream, preload: [:transform])
+        find_sink_consumer(account_id,
+          id_or_name: id_or_name,
+          type: :sequin_stream,
+          preload: [:transform, :enrichment, :postgres_database]
+        )
       end,
       ttl
     )

lib/sequin_web/controllers/pull_controller.ex

@@ -2,6 +2,7 @@ defmodule SequinWeb.PullController do
   use SequinWeb, :controller
 
   alias Sequin.Consumers
+  alias Sequin.Consumers.SinkConsumer
   alias Sequin.Error
   alias Sequin.Runtime.SlotMessageStore
   alias Sequin.String, as: SequinString
@@ -21,7 +22,8 @@ defmodule SequinWeb.PullController do
          :ok <- SlotMessageStore.nack_stale_produced_messages(consumer),
          {:ok, messages} <- SlotMessageStore.produce(consumer, batch_size, :consistent_pid) do
       Logger.metadata(batch_size: batch_size)
-      render(conn, "receive.json", consumer: consumer, messages: messages)
+      enriched_messages = enrich_messages(consumer, messages)
+      render(conn, "receive.json", consumer: consumer, messages: enriched_messages)
     end
   end
 
@@ -169,4 +171,22 @@ defmodule SequinWeb.PullController do
   defp env do
     Application.get_env(:sequin, :env)
   end
+
+  defp enrich_messages(%SinkConsumer{enrichment: nil}, messages), do: messages
+
+  defp enrich_messages(%SinkConsumer{} = consumer, messages) do
+    # Group by table_oid to batch enrichment queries efficiently
+    enriched_by_cursor =
+      messages
+      |> Enum.group_by(& &1.table_oid)
+      |> Enum.flat_map(fn {_table_oid, table_messages} ->
+        Consumers.enrich_messages!(consumer.postgres_database, consumer.enrichment, table_messages)
+      end)
+      |> Map.new(&{{&1.commit_lsn, &1.commit_idx}, &1})
+
+    # Return in original order
+    Enum.map(messages, fn msg ->
+      Map.fetch!(enriched_by_cursor, {msg.commit_lsn, msg.commit_idx})
+    end)
+  end
 end

test/sequin_web/controllers/pull_controller_test.exs

@@ -4,11 +4,14 @@ defmodule SequinWeb.PullControllerTest do
   alias Sequin.Consumers
   alias Sequin.Databases.ConnectionCache
   alias Sequin.Factory.AccountsFactory
+  alias Sequin.Factory.CharacterFactory
   alias Sequin.Factory.ConsumersFactory
   alias Sequin.Factory.DatabasesFactory
+  alias Sequin.Factory.FunctionsFactory
   alias Sequin.Factory.ReplicationFactory
   alias Sequin.Runtime.SlotMessageStore
   alias Sequin.Runtime.SlotMessageStoreSupervisor
+  alias Sequin.TestSupport.Models.Character
 
   setup :authenticated_conn
 
@@ -118,6 +121,66 @@ defmodule SequinWeb.PullControllerTest do
       assert %{"data" => messages} = json_response(conn, 200)
       assert length(messages) == 1
     end
+
+    test "applies enrichment function to messages", %{conn: conn, account: account} do
+      # Create a character in the test database
+      character = CharacterFactory.insert_character!()
+
+      # Create database and replication slot
+      db = DatabasesFactory.insert_configured_postgres_database!(account_id: account.id, tables: :character_tables)
+      ConnectionCache.cache_connection(db, Sequin.Repo)
+      rep_slot = ReplicationFactory.insert_postgres_replication!(postgres_database_id: db.id, account_id: account.id)
+
+      # Create an enrichment function that looks up character data
+      enrichment_function =
+        FunctionsFactory.insert_enrichment_function!(
+          account_id: account.id,
+          function_attrs: [
+            code: """
+            SELECT id, 'enriched_' || name as enriched_name
+            FROM "Characters"
+            WHERE id = ANY($1::int[])
+            """
+          ]
+        )
+
+      # Create consumer with enrichment
+      consumer =
+        ConsumersFactory.insert_sink_consumer!(
+          message_kind: :record,
+          account_id: account.id,
+          backfill_completed_at: DateTime.add(DateTime.utc_now(), -24, :hour),
+          replication_slot_id: rep_slot.id,
+          sink: %{type: :sequin_stream},
+          enrichment_id: enrichment_function.id,
+          source: %{
+            include_table_oids: [Character.table_oid()]
+          }
+        )
+
+      start_supervised!({SlotMessageStoreSupervisor, consumer_id: consumer.id, test_pid: self()})
+
+      # Create a message for the character
+      record =
+        ConsumersFactory.deliverable_consumer_record(
+          consumer_id: consumer.id,
+          table_oid: Character.table_oid(),
+          record_pks: [character.id],
+          data: ConsumersFactory.consumer_record_data(record: %{"id" => character.id, "name" => character.name})
+        )
+
+      SlotMessageStore.put_messages(consumer, [record])
+
+      # Fetch the message and verify enrichment
+      conn = get(conn, ~p"/api/sequin_streams/#{consumer.id}/receive")
+      assert %{"data" => [message]} = json_response(conn, 200)
+
+      # Verify enrichment data is present
+      assert message["data"]["metadata"]["enrichment"] == %{
+               "id" => character.id,
+               "enriched_name" => "enriched_#{character.name}"
+             }
+    end
   end
 
   describe "receive, wait_for behavior" do