fix: decode bytea hex from wal2json before writing to postgres

pkg/wal/listener/postgres/bytea.go

@@ -0,0 +1,59 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"encoding/hex"
+	"strings"
+
+	"github.com/xataio/pgstream/pkg/wal"
+)
+
+const byteaType = "bytea"
+
+// decodeByteaColumns converts the hex text wal2json emits for bytea columns
+// into the raw bytes the rest of the pipeline expects.
+//
+// wal2json renders a bytea column as bare hex digits ("deadbeef"); the postgres
+// hex format ("\xdeadbeef") is accepted too, since that is the form
+// pkg/transformers already documents for the replication path. Snapshots are
+// unaffected: pgx hands those values over as []byte to begin with, which is
+// what makes this the point where the two paths converge.
+//
+// Without it the hex text travels all the way to the writer and is handed to
+// pgx as the parameter value for a bytea column, storing its ASCII characters
+// as the column contents — '\xdeadbeef' on the source becomes
+// '\x6465616462656566' on the target. Nothing else about the row changes, so
+// row counts and keys still match and the corruption only surfaces when
+// something parses the column.
+//
+// Both Columns and Identity are covered, so insert values, the SET clause, the
+// WHERE clause and the bulk-delete builders all see decoded bytes.
+//
+// A value that is not valid hex is left untouched rather than dropped, so an
+// unexpected producer format degrades to the previous behaviour instead of
+// failing the batch.
+func decodeByteaColumns(d *wal.Data) {
+	if d == nil {
+		return
+	}
+	decodeByteaValues(d.Columns)
+	decodeByteaValues(d.Identity)
+}
+
+func decodeByteaValues(cols []wal.Column) {
+	for i := range cols {
+		if cols[i].Type != byteaType {
+			continue
+		}
+		strVal, ok := cols[i].Value.(string)
+		if !ok {
+			continue
+		}
+		decoded, err := hex.DecodeString(strings.TrimPrefix(strVal, `\x`))
+		if err != nil {
+			continue
+		}
+		cols[i].Value = decoded
+	}
+}

pkg/wal/listener/postgres/bytea_test.go

@@ -0,0 +1,89 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/pkg/wal"
+)
+
+func Test_decodeByteaColumns(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		data *wal.Data
+
+		want *wal.Data
+	}{
+		{
+			name: "nil data",
+			data: nil,
+			want: nil,
+		},
+		{
+			name: "wal2json bare hex, in columns and identity",
+			data: &wal.Data{
+				Columns:  []wal.Column{{Name: "payload", Type: "bytea", Value: "deadbeef"}},
+				Identity: []wal.Column{{Name: "key", Type: "bytea", Value: "0102"}},
+			},
+			want: &wal.Data{
+				Columns:  []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0xde, 0xad, 0xbe, 0xef}}},
+				Identity: []wal.Column{{Name: "key", Type: "bytea", Value: []byte{0x01, 0x02}}},
+			},
+		},
+		{
+			name: "postgres hex format",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: `\xdeadbeef`}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0xde, 0xad, 0xbe, 0xef}}}},
+		},
+		{
+			name: "empty value",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: ""}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{}}}},
+		},
+		{
+			// snapshots hand bytea over as []byte already
+			name: "already decoded",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0x01}}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: []byte{0x01}}}},
+		},
+		{
+			// degrade to the previous behaviour rather than failing the batch
+			name: "not valid hex",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "not-hex"}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "not-hex"}}},
+		},
+		{
+			name: "odd length hex",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "abc"}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: "abc"}}},
+		},
+		{
+			name: "nil value",
+			data: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: nil}}},
+			want: &wal.Data{Columns: []wal.Column{{Name: "payload", Type: "bytea", Value: nil}}},
+		},
+		{
+			name: "other column types are untouched",
+			data: &wal.Data{Columns: []wal.Column{
+				{Name: "name", Type: "text", Value: "deadbeef"},
+				{Name: "id", Type: "integer", Value: 1},
+			}},
+			want: &wal.Data{Columns: []wal.Column{
+				{Name: "name", Type: "text", Value: "deadbeef"},
+				{Name: "id", Type: "integer", Value: 1},
+			}},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			decodeByteaColumns(tc.data)
+			require.Equal(t, tc.want, tc.data)
+		})
+	}
+}

pkg/wal/listener/postgres/wal_pg_listener.go

@@ -178,6 +178,11 @@ func (l *Listener) processWALEvent(ctx context.Context, msg *replication.Message
 		if err := l.walDataDeserialiser(msg.Data, event.Data); err != nil {
 			return fmt.Errorf("error unmarshaling wal data: %w", err)
 		}
+		// Here rather than in the writer: this is the only place that knows the
+		// value came from wal2json, so one pass covers every downstream call
+		// site, and transformers then see the same []byte the snapshot path
+		// gives them instead of hex text.
+		decodeByteaColumns(event.Data)
 	}
 	event.CommitPosition = wal.CommitPosition(l.lsnParser.ToString(msg.LSN))
 	if isInternalPgstreamDML(event.Data) {