From 926527e033f764d112c1d6144b8fa848925a838b Mon Sep 17 00:00:00 2001
From: entity-0-0-2 <entity-0-0-2@users.noreply.github.com>
Date: Fri, 9 Oct 2026 17:29:34 -0700
Subject: [PATCH] fix: cast composite type columns to text in backfill selects

Backfill selects picked custom composite type columns as-is, Postgrex
decoded them into values without a Jason.Encoder, and load_rows/2
silently turned them into nil, while the WAL replication path emitted
Postgres text representation like (EUR,42.00) for the same column.

Select composite columns (pg_typtype c) with an explicit ::text cast
and alias so backfilled rows carry the same text representation the
stream path produces.
---
 lib/sequin/postgres/postgres.ex |  8 +++++-
 test/sequin/postgres_test.exs   | 43 +++++++++++++++++++++++++++++++++
 2 files changed, 50 insertions(+), 1 deletion(-)

diff --git a/lib/sequin/postgres/postgres.ex b/lib/sequin/postgres/postgres.ex
index 0eac28669..64b59c887 100644
--- a/lib/sequin/postgres/postgres.ex
+++ b/lib/sequin/postgres/postgres.ex
@@ -716,7 +716,13 @@ defmodule Sequin.Postgres do
   def safe_select_columns(%PostgresDatabaseTable{} = table) do
     table.columns
     |> Enum.filter(&has_encoder?/1)
-    |> Enum.map_join(", ", &quote_name(&1.name))
+    |> Enum.map_join(", ", fn
+      %PostgresDatabaseTable.Column{pg_typtype: "c"} = column ->
+        "#{quote_name(column.name)}::text as #{quote_name(column.name)}"
+
+      column ->
+        quote_name(column.name)
+    end)
   end
 
   def sequence_nextval(sequence_name) do
diff --git a/test/sequin/postgres_test.exs b/test/sequin/postgres_test.exs
index b4edb3635..fb9d1cba9 100644
--- a/test/sequin/postgres_test.exs
+++ b/test/sequin/postgres_test.exs
@@ -323,4 +323,47 @@ defmodule Sequin.PostgresTest do
       assert :ok = Postgres.create_publication(Repo, pub_name)
     end
   end
+
+  describe "safe_select_columns/1 with composite types" do
+    test "backfill select casts composite columns to text so values survive load_rows", %{conn: conn} do
+      Postgres.query!(conn, "DROP TABLE IF EXISTS payments_composite_backfill_test", [])
+      Postgres.query!(conn, "DROP TYPE IF EXISTS money_with_currency_backfill_test CASCADE", [])
+      Postgres.query!(conn, "CREATE TYPE money_with_currency_backfill_test AS (currency_code varchar, amount numeric)", [])
+
+      Postgres.query!(
+        conn,
+        "CREATE TABLE payments_composite_backfill_test (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), amount money_with_currency_backfill_test NOT NULL)",
+        []
+      )
+
+      Postgres.query!(
+        conn,
+        "INSERT INTO payments_composite_backfill_test (amount) VALUES (ROW('EUR', 42.00)::money_with_currency_backfill_test)",
+        []
+      )
+
+      {:ok, fresh_conn} = Postgrex.start_link(UnboxedRepo.config())
+
+      table =
+        DatabasesFactory.table(
+          schema: "public",
+          name: "payments_composite_backfill_test",
+          columns: [
+            DatabasesFactory.column(name: "id", type: "uuid", is_pk?: true, pg_typtype: "b"),
+            DatabasesFactory.column(name: "amount", type: "money_with_currency_backfill_test", pg_typtype: "c")
+          ]
+        )
+
+      sql =
+        "select #{Postgres.safe_select_columns(table)} from #{Postgres.quote_name("public", "payments_composite_backfill_test")}"
+
+      {:ok, result} = Postgres.query(fresh_conn, sql, [])
+      [row] = Postgres.load_rows(table, Postgres.result_to_maps(result))
+
+      assert row["amount"] == "(EUR,42.00)"
+
+      Postgres.query!(conn, "DROP TABLE IF EXISTS payments_composite_backfill_test", [])
+      Postgres.query!(conn, "DROP TYPE IF EXISTS money_with_currency_backfill_test CASCADE", [])
+    end
+  end
 end