Postgrex Integration#

Sequin uses Postgrex as its PostgreSQL driver for both regular query execution and logical replication streaming. The integration spans four areas: a unified query interface, compile-time type registration, runtime type-safety gating, and JSON serialization of Postgrex structs.

Key Source Files#

Compile-Time Type Registration#

Sequin defines a single custom Postgrex type module at compile time 1:

Postgrex.Types.define(Sequin.Postgres.PostgrexTypes,
  Pgvector.extensions() ++ Ecto.Adapters.Postgres.extensions(),
  allow_infinite_timestamps: true
)

This registers Sequin.Postgres.PostgrexTypes with:

• Pgvector.extensions() — adds pgvector support for vector columns
• Ecto.Adapters.Postgres.extensions() — all standard Ecto/Postgres type extensions
• allow_infinite_timestamps: true — accepts PostgreSQL's infinity / -infinity timestamp values without error

No application-level config is needed; this module is used wherever a Postgrex connection is opened.

Unified Query Interface (Sequin.Postgres.query/4)#

Sequin.Postgres.query/4 dispatches queries across three connection types:

• PID / DBConnection struct — calls Postgrex.query/4 directly, normalizing timeout errors into {:error, Error.service(..., code: :query_timeout)}
• Module atom — delegates to mod.query/3 (e.g., Sequin.Repo in tests)
• %PostgresDatabase{} — resolves a connection via ConnectionCache, and on :undefined_column errors, triggers DatabaseUpdateWorker to refresh the schema cache

All three paths return {:ok, %Postgrex.Result{}} or {:error, error}. The bang variant query!/4 raises on error 2.

Parameter Passing#

$N-style placeholders (Postgrex native)#

Postgrex uses positional $1, $2, ... placeholders. Two helpers in Sequin.Postgres assist with parameterization:

• parameterize_sql/1 — converts ?-style SQL to $1-style (used when building dynamic SQL that is more readable with ?)
• parameterized_tuple/2 — generates ($1, $2, ...) for multi-value inserts; accepts an offset for composing multi-clause queries

OID type encoding gotcha#

When passing sequence names as parameters, the explicit cast $1::text::regclass is required 3. See the Postgrex OID type encoding docs — Postgrex cannot infer OID type from a string parameter alone.

Type Safety: has_encoder? and safe_select_columns#

Before issuing a backfill SELECT *, Sequin gates which columns are included using has_encoder?/1. This predicate:

1. Allows all time* variants via a prefix match 4
2. Passes composite types (c), domains (d), and enums (e) through optimistically — Postgrex is assumed to handle them 5
3. Validates against a @safe_types whitelist of ~27 known-safe types

safe_select_columns/1 uses has_encoder? to build a SELECT list that omits unsupported columns rather than crashing at decode time.

Row Serialization: load_rows/2#

load_rows/2 converts each %Postgrex.Result{} row into a JSON-safe map during backfills. Per-type handling:

The no-encoder fallback (Jason.Encoder.impl_for(value) == Jason.Encoder.Any) prevents runtime crashes but silently drops the value. If a column type is new and falls through here, check load_rows and add handling.

Custom Jason / String.Chars Encoders#

lib/sequin/postgrex/encoders.ex adds protocol implementations for types not covered by default Jason encoding:

• Postgrex.Lexeme → %{type: :lexeme, word: ..., positions: [[pos, weight], ...]} 13
• Postgrex.Range → String.Chars implementation for ISO 8601 bound formatting 14, plus a Jason.Encoder that delegates to to_string/1 15

These encoders are loaded globally at compile time and apply to all JSON serialization in the application.

Logical Replication Backend#

Sequin.Postgres.PostgrexBackend wraps Postgrex.Protocol as the default backend for SlotProducer's replication connection. It delegates connect, handle_streaming, handle_copy_recv, handle_copy_send, handle_simple, checkin, and disconnect directly to Postgrex.Protocol. The Backend behaviour allows this to be swapped with a test double in non-production environments.