phylax

README

phylax

Minimal PostgreSQL logical replication in Go — stream row changes to stdout, webhooks, or a live console.

Live demo • Why phylax • Quick start • Features • How it works • CLI • Library • Console • Outbox • Change payload • Performance • Limitations • Project layout

phylax connects to PostgreSQL, creates its own replication slot and publication, and streams every row change (insert / update / delete / truncate) to your code as it happens — to stdout, a webhook, or the embedded live console. It reconnects with backoff and resumes from the slot's saved position, so a restart loses nothing.

Live demo

Try it: phylax · names — a live Postgres-backed demo. Every name and like you see on that page arrived through phylax's logical-replication stream — the browser never talks to the database directly.

The screenshot above is the embedded console the CLI serves at /dashboard (live KPIs, lag sparkline, change feed, dark/light).

[!NOTE] The demo link is a short-lived Cloudflare quick tunnel minted for a one-off post — it may rotate and is not a permanent URL. The demo app itself is small and self-contained: a Postgres source of truth, an app-owned write connection, and phylax keeping an in-memory list in sync. The library section below shows the same OnChange wiring in five lines.

Why phylax?

The replication protocol is the hard part — your business logic isn't. Logical replication is the proper way to watch a database: no trigger overhead on every write, no polling latency, only committed transactions, and the server itself tracks your position. The sharp edges are exactly where hand-rolled clients go wrong: keepalives and wal_sender_timeout, standby-status timing, slot and publication lifecycle, resume semantics, and what happens when a consumer is slow. phylax handles all of that and hands you a Change callback.

Instead of...	phylax gives you...
Wiring pglogrepl yourself	Slot/publication provisioning, keepalives, reconnect with backoff, LSN resume, and error classification — done and tested. Your handler is five lines.
A CDC platform (Debezium, Kafka, …)	No JVM, no broker, no schema registry, no distributed deployment. One small binary or embedded library, delivering to your code, a webhook, or the included console.
Triggers or polling	Changes arrive as they commit — no write-path overhead, no application-code changes.

Not the right fit for exactly-once delivery, multi-slot HA, or frequent schema evolution — see Deliberate limitations.

Quick start

Requirements: PostgreSQL with wal_level = logical and a role with REPLICATION privileges.

docker run -d --name pg -e POSTGRES_PASSWORD=secret -p 5432:5432 postgres:16 -c wal_level=logical

[!NOTE] Not using Docker? The server-side setup is just two psql commands — set wal_level = logical, restart PostgreSQL, and give a role REPLICATION privileges. No container needed.

-- as a superuser, once:
ALTER SYSTEM SET wal_level = logical;

-- wal_level only takes effect at startup, so restart PostgreSQL:
--   sudo systemctl restart postgresql    (Debian/Ubuntu)
--   sudo systemctl restart postgresql    (RHEL/Fedora/Arch)
--   pg_ctl restart -D /var/lib/postgresql/16/main

-- then create a role phylax can connect as:
CREATE ROLE repl WITH LOGIN REPLICATION PASSWORD 'secret';

(For the CLI example below, use your own role and database — e.g. postgres://repl:secret@localhost:5432/mydb if you followed the psql path above.)

Install the CLI, then start it:

go install github.com/codetesla51/phylax/cmd/phylax@latest
phylax --dsn 'postgres://postgres:secret@localhost:5432/mydb' --tables users,orders

Prefer running from source? go run ./cmd/phylax --dsn … behaves identically.

That's it — the CLI creates its own slot and publication, starts replicating, and serves the console:

1. Open http://localhost:8080/dashboard — live KPIs, a lag sparkline, and the change feed.
2. Insert a row: INSERT INTO users (email, name) VALUES ('a@b.c', 'Alice');
3. Watch it appear in the feed, in your terminal, and in the changes_processed counter.

[!TIP] Ctrl-C stops the client gracefully and saves the slot position, so a restart resumes exactly where it left off — nothing lost, nothing replayed.

Features

• Real-time row changes — insert / update / delete / truncate as they commit
• Self-provisioning — creates its own slot + publication, idempotent and restart-safe
• Resume from LSN — backoff reconnect (1s → 30s); picks up exactly where it left off
• Embedded console — a self-contained dashboard at /dashboard: KPI cards, log-scale lag sparkline, live feed with CSV export, dark/light mode
• SSE endpoints — /events and /metrics/stream from one http.Server
• Webhook delivery — POST each change to an endpoint, bounded retries
• Transactional outbox — route inserts on an outbox table to a DeliveryFunc, async with per-topic ordering and bounded concurrency (see Outbox)
• Zero-state safe — a zero-value Server works; slow subscribers never stall the stream

How it works

1. Connect — opens a replication connection and an admin connection to the same database.
2. Resume — picks up from the slot's saved position, or starts fresh if there's none.
3. Provision — creates its slot and publication if they don't exist yet.
4. Stream — decodes WAL into Change values, dispatches them to your handlers, and tells the server to advance the slot.

CLI

Stream changes to stdout:

go run ./cmd/phylax --dsn 'postgres://user:pass@localhost:5432/db' --tables users,orders

Every change prints as one JSON object per line:

{"Table":"users","Operation":"insert","OldRow":null,"NewRow":{"email":"a@b.c","name":"Alice"}}

Flag	Default	Meaning
--dsn	required	libpq connection string
--tables	required	comma-separated tables to replicate
--outbox-table	—	table whose inserts route through the outbox pipeline (see Outbox)
--webhook	—	POST each change to this URL
--slot	my_slot	replication slot name
--publication	my_publication	publication name
--addr	:8080	HTTP address for the console (dashboard + SSE)
--no-http	false	disable the HTTP console
-v	false	debug-level logging (protocol traffic, raw WAL)

--webhook POSTs each change as JSON, retrying up to 3 times (1s/2s/3s backoff) before giving up — a slow webhook must never stall replication.

With --outbox-table set, inserts on that table are consumed by the outbox pipeline — logged and acked by the built-in handler (or your OnOutboxDelivery when driving the library directly) — instead of printed to stdout.

Library

go get github.com/codetesla51/phylax

cdc, err := phylax.New(phylax.Config{
    DSN:    "postgres://user:pass@localhost:5432/db", // required
    Tables: []string{"users", "orders"},
})
if err != nil { log.Fatal(err) }

cdc.OnChange(func(c *phylax.Change) {
    fmt.Printf("%s %s → %v\n", c.Operation, c.Table, c.NewRow)
})

log.Fatal(cdc.Start(context.Background()))

Config has six fields: DSN (required — no default), Tables, SlotName (default my_slot), PublicationName (default my_publication), ChangeBufferSize (per-OnChange-subscriber channel buffer, default 100), and OutboxTable (optional — when set, inserts on that table are routed through the outbox pipeline instead of to OnChange). Low-level tunables (heartbeat interval, per-connection URLs) live in ClientConfig / DefaultClientConfig().

[!NOTE] ChangeBufferSize only needs raising when a consumer verifiably drops changes (changes_dropped climbing in the console) under bursty load — size it for the biggest burst, roughly 1KB per buffered change. A full buffer drops rather than stalls the stream, so this trades a little memory for burst headroom, not correctness.

Outbox

phylax can be the tail of a transactional outbox: write your domain change and an outbox row in the same transaction, and phylax delivers the outbox row to your DeliveryFunc as the WAL commits — no polling, using the same writes your app already makes.

cdc, _ := phylax.New(phylax.Config{
    DSN:         "postgres://user:pass@localhost:5432/db",
    Tables:      []string{"users", "outbox"},
    OutboxTable: "outbox",
})

cdc.OnOutboxDelivery(func(ctx context.Context, row *phylax.OutboxRow) error {
    return broker.Publish(row.Topic, row.Payload)
})

The outbox table needs id (int), topic (text), and payload (JSON) columns; Payload is decoded from JSON into a map[string]any. On success the row is acked (UPDATE outbox SET delivered_at = now()); an error retries with exponential backoff (1s → 16s, 5 attempts) before the row is left pending.

OutboxRow is { ID int64; Topic string; Payload map[string]any }.

Delivery semantics:

• Async & bounded — every row dispatches to a per-topic drainer goroutine, so a slow or down broker never stalls WAL consumption. Topics run in parallel (capped by a semaphore of 64 drainers); within a topic, rows deliver strictly in order.
• At-least-once — phylax resumes from the slot's saved position on restart and replays every outbox insert, including rows already acked. Your DeliveryFunc must be idempotent: it will see the same row.ID more than once.
• Best-effort ack — after exhausting retries a row is left pending (no dead-letter table in v1); it replays on restart and retries again. The slot is the safety net.

[!NOTE] The acker uses its own database connection (separate from the admin connection) and serializes acks. A drainer racing the admin connection would error with conn busy — pgx.Conn is not safe for concurrent use, so the ack path gets a dedicated, mutex-guarded connection. The UPDATE is atomic per statement; Postgres is never the bottleneck, the Go-concurrency discipline is.

Console

cdc.Server() (or phylax.NewServer(broadcaster, metrics)) serves three routes from one http.Server:

Route	What it serves
/events	every change as an SSE event
/metrics/stream	a JSON metrics snapshot every second (changes_processed, changes_dropped, subscribers, replication_lag_bytes, and the outbox counters outbox_delivered / outbox_inflight / outbox_failed)
/dashboard	the embedded console page (go:embed, no static files to ship)

The dashboard's KPI row includes a live outbox card — outbox delivered with in-flight / failed sub-lines (the latter turns red when deliveries are failing) — whenever an outbox table is configured.

srv := cdc.Server()
log.Fatal(srv.ListenAndServe(":8080"))

Shutdown(ctx) stops it gracefully (works even if called before Serve — a later Serve refuses its listener instead of leaking one); Handler() mounts the routes on an existing mux.

Delivery buffers. Every /events subscriber gets a bounded channel — 10 events for SSE subscribers, 100 for the OnChange consumer — sized at Broadcaster.Subscribe(id, bufferSize) time in code (constants in sse.go / cdc.go, not exposed via config or CLI). A full buffer drops the change and counts it in changes_dropped rather than stalling the stream.

[!WARNING] The console endpoints are unauthenticated by design — localhost/dev only. Put them behind an auth proxy if exposed publicly, or use --no-http.

Understanding the change payload

{"Table":"users","Operation":"update","OldRow":null,"NewRow":{"id":"124","email":"a@b.c","name":"Alice"}}

• NewRow — the row as it is now; null for deletes
• OldRow — the row as it was before; null for inserts
• truncate — the whole table was emptied; both rows are null, and a single TRUNCATE a, b, c; emits one event per table (this is the only operation that carries no row data)

[!NOTE] Why OldRow is usually sparse: PostgreSQL only ships old-row data as allowed by the table's replica identity (default = primary key only). Non-key updates send no old tuple at all; deletes and key changes send the key plus null placeholders. This is expected behavior, not a bug. If your consumer needs old values (diffs, audit), switch the table to ALTER TABLE users REPLICA IDENTITY FULL; — at the cost of more WAL per change.

Performance

Normal load first: phylax is built for application write volumes — tens to low thousands of changes per second — and at that scale it has room to spare. Decode keeps pace instantly, lag sits at zero, and even several dashboard subscribers receive everything, no drops. The numbers below are the stress-test ceiling: where the stack finally strained when we deliberately tried to break it. Read them as "we pushed until something gave," not "you'll hit this."

Stress ceiling — the generator walls out first. Measured with the checked-in barrage configs against a local Postgres 16 (Docker): the single-row ladder (500 → 2000 → 5000 writes/s, no subscribers) established the single-row generator ceiling at 4,583 changes/s; 50-row batched inserts raised that wall 8× to ~37,000 changes/s. The target was 50,000 — Postgres's commit rate, not barrage and not phylax, was the cap (barrage hit 917 of its 1,000 statements/s; the failed remainder is DB saturation at concurrency 100, success 90.7–93.4%). Decode consumed 100% of everything generated in every run, lag pinned at 0.

Stress ceiling — the fan-out is the second wall. Each SSE subscriber is a goroutine writing one event per change; under load that path sustains roughly 26k events/s for one subscriber and less per subscriber as the count grows (CPU contention). At ~37k changes/s:

Subscribers	Generated	Decode consumed	Drops	Per-sub received
11 (10 headless + 1 tab)	~1.285M	1,285,500 (100%)	~1.05M/sub	~15%
3 (2 headless + 1 tab)	~1.284M	1,284,900 (100%)	~776k/sub	~27%
1 (dashboard tab)	~1.248M	1,248,200 (100%)	~340k	~73%

Drops are counted in changes_dropped — by design, a slow subscriber never stalls the stream.

Where normal load sits. At ≤2,000 changes/s the same stack ran clean: zero drops with subscribers attached, lag draining to 0. A busy application writing hundreds of changes per second is well below either wall — the delivery path alone handles ~26k events/s per subscriber, and the generator only starts straining near 37k. The no-subscriber ceiling remains unmeasured; the generator walls out first.

Reproduce: barrage run -c benchmarks/barrage-ceiling-5000.yml (single-row ladder) or barrage run -c benchmarks/barrage-ceiling-50k.yml (batched matrix; subscriber counts are recorded live on the console's subscribers gauge). The benchmarks/ directory holds all configs plus a sample report.html from the last run.

[!TIP] On the lag sparkline: a plateau during steady writes is normal pipeline depth; a climb while writes continue means the consumer is falling behind; a quick drain to 0 after writes stop is proof of health.

Deliberate limitations

Each trade-off is a choice, not an oversight — what you give up, why, and when you'd outgrow it.

Limitation	Why it's deliberate	When to outgrow it
Best-effort delivery — 3 webhook retries, then drop; no durable outbox for webhooks	Keeps memory bounded; the slot is the safety net (at-least-once)	Need exactly-once for a broker → use the built-in outbox (its own ack + retry)
Drop-on-full subscribers — SSE subscribers hold a 10-event buffer (100 for the OnChange consumer); sizes are set in code at Subscribe(id, size) time, not via config; a full buffer drops the change (counted)	A subscriber must never stall the stream	Consumers can't keep pace → speed them up or watch changes_dropped
Ghost subscribers — an SSE client that drops its connection without a clean close stays registered until the next write fails, so subscribers can lag reality while idle	Unsubscribe-on-write-error is simple and correct-enough	Need exact live counts → heartbeat or read-side EOF detection
Delivery-bound, not decode-bound — decode consumed 100% at every tested rate (up to ~37k changes/s); the walls are Postgres's commit rate and SSE fan-out (Performance)	Drops are the designed degradation path	Sustained high write rates → batch SSE writes, go binary, or fan out consumers
Text tuples, string values — no binary protocol, no typed decode	Text mode is the simplest correct path	Need typed values → decode binary or convert downstream
Key-only old rows (REPLICA IDENTITY DEFAULT)	Leaner WAL; identity + new state covers most consumers	Need before-images → REPLICA IDENTITY FULL
No auth on the console	Localhost/dev convenience, not a product	Public exposure → auth proxy or --no-http
Single slot / stream / process — no sharding, no HA	Simplest correct model for a minimal client	Scale-out → partition by slot or add leader election
No DDL handling — schema changes can desync decoding	Out of scope for a minimal client	Frequent schema evolution → use a battle-tested CDC (Debezium)

Project layout

File	Responsibility
cdc.go	Public CDC wrapper: Config, New, OnChange, Start, Server
decode.go	WAL bytes → Change: Decode and tupleToMap
stream.go	Long-running loop: keepalives, standby status, lag
broadcast.go	Fan-out Broadcaster (subscribe/unsubscribe, drop-on-full)
sse.go	SSE Server: /events + /metrics/stream, Handler/Shutdown
dashboard.go/html	Embedded console served at /dashboard
metrics.go	Live counters (Metrics) and MetricsSnapshot
cmd/phylax/main.go	CLI entry point: flags, webhook client, console server
benchmarks/	Barrage load-test configs (single-row ladder + 50k batched matrix) and a sample report.html from the last run

Documentation

Index

• func EnsurePublication(ctx context.Context, adminConn *pgx.Conn, publicationName string, ...) (bool, error)
• func EnsureReplicationSlot(ctx context.Context, replConn *pgconn.PgConn, adminConn *pgx.Conn, ...) (bool, error)
• func IdentifySystem(ctx context.Context, conn *pgconn.PgConn) (pglogrepl.IdentifySystemResult, error)
• func OpenAdminConnection(ctx context.Context, adminURL string) (*pgx.Conn, error)
• func OpenReplicationConnection(ctx context.Context, databaseURL string) (*pgconn.PgConn, error)
• func PublicationExists(ctx context.Context, conn *pgx.Conn, publicationName string) (bool, error)
• func SlotConfirmedFlushLSN(ctx context.Context, conn *pgx.Conn, slotName string) (pglogrepl.LSN, bool, error)
• func SlotExists(ctx context.Context, conn *pgx.Conn, slotName string) (bool, error)
• type Broadcaster
  • func NewBroadcaster() *Broadcaster
  • func (b *Broadcaster) ChangesDropped() int64
  • func (b *Broadcaster) Publish(change *Change)
  • func (b *Broadcaster) Subscribe(id string, bufferSize int) chan *Change
  • func (b *Broadcaster) SubscriberCount() int
  • func (b *Broadcaster) Unsubscribe(id string)
• type CDC
  • func New(cfg Config) (*CDC, error)
  • func (c *CDC) Broadcaster() *Broadcaster
  • func (c *CDC) MetricsSnapshot() MetricsSnapshot
  • func (c *CDC) OnChange(fn func(*Change))
  • func (c *CDC) OnOutboxDelivery(fn func(context.Context, *OutboxRow) error)
  • func (c *CDC) Server() *Server
  • func (c *CDC) Start(ctx context.Context) error
• type Change
  • func Decode(walData []byte, relations map[uint32]*pglogrepl.RelationMessage, ...) ([]*Change, error)
• type ChangeHandler
• type ClientConfig
  • func DefaultClientConfig() ClientConfig
• type Config
• type DeliveryFunc
• type Metrics
• type MetricsProvider
• type MetricsSnapshot
• type OutboxConsumer
  • func NewOutboxConsumer(db *pgx.Conn, deliver DeliveryFunc, tableName string) *OutboxConsumer
  • func (oc *OutboxConsumer) Handle(ctx context.Context, c *Change) bool
  • func (oc *OutboxConsumer) Stats() (delivered, inflight, failed int64)
• type OutboxRow
  • func ToOutboxRow(c *Change, tableName string) (row *OutboxRow, ok bool, err error)
• type ReplicationStream
  • func NewReplicationStream(ctx context.Context, conn *pgconn.PgConn, cfg ClientConfig, ...) (*ReplicationStream, error)
  • func (s *ReplicationStream) Broadcaster() *Broadcaster
  • func (s *ReplicationStream) OutboxStats() (delivered, inflight, failed int64)
  • func (s *ReplicationStream) ReplicationLag() uint64
  • func (s *ReplicationStream) Run(ctx context.Context) error
• type Server
  • func NewServer(broadcaster *Broadcaster, metrics MetricsProvider) *Server
  • func (s *Server) Handler() http.Handler
  • func (s *Server) ListenAndServe(addr string) error
  • func (s *Server) NewMetricsHandler(w http.ResponseWriter, r *http.Request)
  • func (s *Server) NewSSEHandler(w http.ResponseWriter, r *http.Request)
  • func (s *Server) Serve(ln net.Listener) error
  • func (s *Server) Shutdown(ctx context.Context) error

Functions

func EnsurePublication

func EnsurePublication(ctx context.Context, adminConn *pgx.Conn, publicationName string, tables []string) (bool, error)

EnsurePublication creates the publication for the configured tables if it does not exist yet. It returns true when the publication was created, false when it already existed.

func EnsureReplicationSlot

func EnsureReplicationSlot(ctx context.Context, replConn *pgconn.PgConn, adminConn *pgx.Conn, slotName string) (bool, error)

EnsureReplicationSlot creates the slot if it does not exist yet. It returns true when the slot was created, false when it already existed. Slot creation must happen on the replication connection; the existence check uses the admin connection.

func IdentifySystem

func IdentifySystem(ctx context.Context, conn *pgconn.PgConn) (pglogrepl.IdentifySystemResult, error)

IdentifySystem asks the server for its identity: the system ID, the timeline ID, and the current WAL position. The WAL position is used as the starting point for replication, so the client only sees changes that happen after it starts.

func OpenAdminConnection

func OpenAdminConnection(ctx context.Context, adminURL string) (*pgx.Conn, error)

OpenAdminConnection opens a regular pgx connection used for ordinary queries (checking and creating replication slots and publications).

func OpenReplicationConnection

func OpenReplicationConnection(ctx context.Context, databaseURL string) (*pgconn.PgConn, error)

OpenReplicationConnection opens a dedicated connection for logical replication. The connection URL must include `replication=database`; the returned connection is used for IDENTIFY_SYSTEM, START_REPLICATION and the streaming of changes.

func PublicationExists

func PublicationExists(ctx context.Context, conn *pgx.Conn, publicationName string) (bool, error)

PublicationExists reports whether a publication with the given name already exists in the database.

func SlotConfirmedFlushLSN

func SlotConfirmedFlushLSN(ctx context.Context, conn *pgx.Conn, slotName string) (pglogrepl.LSN, bool, error)

SlotConfirmedFlushLSN returns the slot's confirmed flush position — the point up to which a previous run of the client consumed the WAL. The second return value is false when the slot has no usable saved position yet (a freshly created slot, or one whose position is 0/0).

func SlotExists

func SlotExists(ctx context.Context, conn *pgx.Conn, slotName string) (bool, error)

SlotExists reports whether a replication slot with the given name already exists in the database.

Types

type Broadcaster

type Broadcaster struct {
	// contains filtered or unexported fields
}

func NewBroadcaster

func NewBroadcaster() *Broadcaster

func (*Broadcaster) ChangesDropped

func (b *Broadcaster) ChangesDropped() int64

ChangesDropped returns the total number of changes dropped for active subscribers because their buffers were full.

func (*Broadcaster) Publish

func (b *Broadcaster) Publish(change *Change)

func (*Broadcaster) Subscribe

func (b *Broadcaster) Subscribe(id string, bufferSize int) chan *Change

func (*Broadcaster) SubscriberCount

func (b *Broadcaster) SubscriberCount() int

SubscriberCount returns the number of active subscribers.

func (*Broadcaster) Unsubscribe

func (b *Broadcaster) Unsubscribe(id string)

type CDC

type CDC struct {
	// contains filtered or unexported fields
}

CDC is a change data capture client wrapping the phylax building blocks.

func New

func New(cfg Config) (*CDC, error)

New validates cfg and returns a CDC client. It does not connect; the connections are established by Start.

func (*CDC) Broadcaster

func (c *CDC) Broadcaster() *Broadcaster

Broadcaster returns the client's change broadcaster, shared by every OnChange registration. Use it to plug the client into a phylax.Server so SSE clients receive the same changes as OnChange subscribers.

func (*CDC) MetricsSnapshot

func (c *CDC) MetricsSnapshot() MetricsSnapshot

MetricsSnapshot returns a point-in-time reading of the live metrics: changes processed by the current stream, changes dropped for the broadcaster's subscribers, the active subscriber count, and the current replication lag. It reads only in-memory state — no database or network access. With no stream running yet, processed and lag are zero.

func (*CDC) OnChange

func (c *CDC) OnChange(fn func(*Change))

OnChange registers fn to be called for every decoded change. Each registration runs in its own goroutine; the goroutine exits when Start shuts down and unsubscribes the registration.

func (*CDC) OnOutboxDelivery

func (c *CDC) OnOutboxDelivery(fn func(context.Context, *OutboxRow) error)

OnOutboxDelivery registers the handler called for every outbox row (inserts on Config.OutboxTable). The handler must be idempotent — phylax delivers at-least-once, so the same row id may arrive more than once. Without a handler, phylax logs each row and acks it.

func (*CDC) Server

func (c *CDC) Server() *Server

Server returns a phylax.Server wired to this client: /events fans out the client's changes, /metrics/stream reports the live metrics, and /dashboard serves the embedded Phylax Console.

func (*CDC) Start

func (c *CDC) Start(ctx context.Context) error

Start connects, ensures the slot and publication exist, and runs the read/decode/broadcast loop until ctx is cancelled.

If the replication connection drops mid-stream, Start reconnects with exponential backoff (1s doubling up to 30s) and resumes from the slot's confirmed position, logging each retry. It keeps retrying — even when the database is unreachable at startup — until ctx is cancelled.

Only transient failures (connection loss, server restart) are retried. Permanent failures — e.g. unknown tables or bad credentials — stop Start immediately and are returned as errors.

On ctx cancellation Start shuts down gracefully: the connections are closed and every OnChange registration is unsubscribed, so their goroutines exit via the closed subscriber channel. A clean shutdown returns nil.

type Change

type Change struct {
	// Table is the name of the table the change happened on.
	Table string

	// Operation is one of "insert", "update", "delete" or "truncate".
	// A truncate change means every row in Table was removed: it carries no
	// row data (OldRow and NewRow are nil), and TRUNCATE a, b, c emits one
	// truncate change per table.
	Operation string

	// OldRow holds the pre-change column values (nil for inserts and
	// truncates).
	OldRow map[string]any

	// NewRow holds the post-change column values (nil for deletes and
	// truncates).
	NewRow map[string]any
}

Change describes one logical change on a single row — or, for a TRUNCATE, on a whole table.

func Decode

func Decode(walData []byte, relations map[uint32]*pglogrepl.RelationMessage, metrics *Metrics) ([]*Change, error)

Decode parses one chunk of WAL data and returns the Changes it describes. It returns an empty slice when the message carries no row change (relation metadata, transaction begin/commit, keepalives, ...). Most messages yield at most one Change; a TRUNCATE statement can truncate several tables at once, so it yields one Change per truncated table.

Every successfully decoded Change increments metrics.ChangesProcessed, if metrics is non-nil.

Relation metadata is cached in `relations` and reused across calls: the server sends a RelationMessage the first time a table is referenced, and afterwards row data is compact — tuples reference columns by index and rely on the cached relation for the column names.

type ChangeHandler

type ChangeHandler func(*Change) error

ChangeHandler receives every decoded change. Returning an error aborts the replication stream.

type ClientConfig

type ClientConfig struct {
	// DatabaseURL is a libpq connection string for the *replication*
	// connection. Logical replication requires the special parameter
	// `replication=database`; pgx only allows it on a single dedicated
	// connection, not a pool.
	DatabaseURL string

	// AdminURL is a normal connection string used for administrative
	// queries such as checking and creating slots and publications.
	AdminURL string

	// SlotName is the name of the logical replication slot.
	SlotName string

	// PublicationName is the publication whose changes we subscribe to.
	// The pgoutput plugin is told this name when replication starts.
	PublicationName string

	// Tables lists the tables the publication is created for. It is only
	// used the first time the client runs, when the publication does not
	// exist yet.
	Tables []string

	// HeartbeatInterval controls how often the client sends a standby
	// status update back to the server. These updates tell the server how
	// far the client has consumed the WAL; without them the connection is
	// dropped after wal_sender_timeout and the slot never advances.
	HeartbeatInterval time.Duration
}

ClientConfig bundles every runtime setting the client needs. It is the per-connection configuration used internally by the replication stream; the public CDC wrapper (see cdc.go) exposes a simpler Config on top.

func DefaultClientConfig

func DefaultClientConfig() ClientConfig

DefaultClientConfig returns an example configuration for local development. The connection strings are placeholders — replace them with your own DSNs; the other values are the phylax defaults (slot my_slot, publication my_publication).

type Config

type Config struct {
	// DSN is a libpq connection string. It is used for the admin connection
	// as-is, and for the replication connection with `replication=database`
	// appended.
	DSN string

	// Tables lists the tables the publication is created for. It is only
	// used the first time the client runs, when the publication does not
	// exist yet.
	Tables []string

	// SlotName is the name of the logical replication slot. Defaults to
	// "my_slot" when empty.
	SlotName string

	// PublicationName is the publication whose changes are replicated.
	// Defaults to "my_publication" when empty.
	PublicationName string

	// ChangeBufferSize is the per-subscriber channel buffer used by
	// OnChange. A slow consumer with a full buffer drops changes
	// (counted in MetricsSnapshot.ChangesDropped) rather than stalling
	// the stream, so size this for your biggest burst. Defaults to 100
	// when <= 0. Each buffered change costs roughly a kilobyte.
	ChangeBufferSize int

	OutboxTable string
}

Config configures a CDC client.

type DeliveryFunc

type DeliveryFunc func(ctx context.Context, row *OutboxRow) error

DeliveryFunc is the user-supplied handler called once per outbox row. Returning nil marks the row delivered; returning an error triggers retry.

DELIVERY MUST BE IDEMPOTENT. phylax delivers at-least-once: on restart it resumes from the slot's saved position and replays every outbox insert, including rows already acked with delivered_at, and in-flight retry state is lost on crash. The same row may therefore be delivered more than once — design the handler (and the broker it talks to) to tolerate duplicate deliveries of the same row ID.

type Metrics

type Metrics struct {
	// ChangesProcessed counts every successfully decoded, non-nil Change.
	ChangesProcessed atomic.Int64
}

Metrics holds the counters the metrics stream reports. It is owned by a ReplicationStream and updated by the decode path.

type MetricsProvider

type MetricsProvider interface {
	MetricsSnapshot() MetricsSnapshot
}

MetricsProvider supplies the live metrics snapshot. It is implemented by the CDC client, which owns the current stream and the change broadcaster.

type MetricsSnapshot

type MetricsSnapshot struct {
	ChangesProcessed int64  `json:"changes_processed"`
	ChangesDropped   int64  `json:"changes_dropped"`
	Subscribers      int    `json:"subscribers"`
	ReplicationLag   uint64 `json:"replication_lag_bytes"`
	OutboxDelivered  int64  `json:"outbox_delivered"`
	OutboxInflight   int64  `json:"outbox_inflight"`
	OutboxFailed     int64  `json:"outbox_failed"`
}

MetricsSnapshot is a point-in-time reading of the live metrics.

type OutboxConsumer

type OutboxConsumer struct {
	// contains filtered or unexported fields
}

OutboxConsumer reads outbox-table inserts off the WAL stream, delivers them via a user-supplied DeliveryFunc, retries failures with backoff, and acks (marks delivered) on success.

Delivery is asynchronous and bounded: each row is dispatched to a per-topic drainer goroutine, so a slow or down broker never blocks WAL consumption. Within a topic, rows are delivered strictly in order; across topics, delivery runs in parallel. A global semaphore caps the number of concurrent drainers so a burst of distinct topics can't spin up unbounded goroutines.

func NewOutboxConsumer

func NewOutboxConsumer(db *pgx.Conn, deliver DeliveryFunc, tableName string) *OutboxConsumer

NewOutboxConsumer wires up a consumer against the given connection and handler. tableName is the table whose inserts are treated as outbox events.

func (*OutboxConsumer) Handle

func (oc *OutboxConsumer) Handle(ctx context.Context, c *Change) bool

Handle is the entrypoint the stream calls for every decoded Change. It returns handled=true if the change belonged to the outbox (regardless of delivery success) — the caller should skip broadcaster fan-out for handled changes.

func (*OutboxConsumer) Stats

func (oc *OutboxConsumer) Stats() (delivered, inflight, failed int64)

Stats returns the outbox consumer's live counters for the metrics stream: cumulative delivered, currently in-flight, and cumulative failed (rows that exhausted retries and were left pending).

type OutboxRow

type OutboxRow struct {
	ID      int64
	Topic   string
	Payload map[string]any
}

OutboxRow is a single pending outbox row decoded from a WAL Change.

func ToOutboxRow

func ToOutboxRow(c *Change, tableName string) (row *OutboxRow, ok bool, err error)

ToOutboxRow converts a decoded Change into an OutboxRow if it is an insert on the outbox table. ok is false if the change isn't relevant (wrong table/operation) — that's not an error, just "not for you." A malformed but relevant row reports ok=true with a non-nil err.

type ReplicationStream

type ReplicationStream struct {
	// contains filtered or unexported fields
}

ReplicationStream consumes messages from a running replication session.

func NewReplicationStream

func NewReplicationStream(ctx context.Context, conn *pgconn.PgConn, cfg ClientConfig, startLSN pglogrepl.LSN, logger *slog.Logger, handle ChangeHandler, db *pgx.Conn, delivery DeliveryFunc, outboxTable string) (*ReplicationStream, error)

NewReplicationStream issues START_REPLICATION for the configured slot and returns a stream ready to consume. The start LSN comes from IDENTIFY_SYSTEM; the server resumes from there. The publication name is passed to the pgoutput plugin so it knows which tables to send.

func (*ReplicationStream) Broadcaster

func (s *ReplicationStream) Broadcaster() *Broadcaster

Broadcaster returns the stream's fan-out broadcaster. Every decoded change is published to it, so subscribers receive each change as it streams in.

func (*ReplicationStream) OutboxStats

func (s *ReplicationStream) OutboxStats() (delivered, inflight, failed int64)

OutboxStats returns the outbox consumer's live counters (0 when outbox is disabled). Read by CDC.MetricsSnapshot for the dashboard.

func (*ReplicationStream) ReplicationLag

func (s *ReplicationStream) ReplicationLag() uint64

func (*ReplicationStream) Run

func (s *ReplicationStream) Run(ctx context.Context) error

Run consumes the replication stream until the context is cancelled, the server returns an error, or the change handler fails.

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer(broadcaster *Broadcaster, metrics MetricsProvider) *Server

NewServer returns a Server that fans decoded changes out through the given broadcaster — typically the CDC client's broadcaster, so SSE clients see every change OnChange subscribers see. metrics supplies the live counters for the /metrics/stream endpoint; a nil provider reports an all-zero snapshot.

func (*Server) Handler

func (s *Server) Handler() http.Handler

Handler returns an http.Handler serving /events, /metrics/stream, and /dashboard on a single mux.

func (*Server) ListenAndServe

func (s *Server) ListenAndServe(addr string) error

ListenAndServe serves both endpoints on addr until Shutdown is called.

func (*Server) NewMetricsHandler

func (s *Server) NewMetricsHandler(w http.ResponseWriter, r *http.Request)

NewMetricsHandler streams a JSON metrics snapshot once per tick as one SSE `data: ...` event, exiting when the client disconnects. It reads only in-memory counters — it does not subscribe to the change broadcaster and never touches Postgres.

func (*Server) NewSSEHandler

func (s *Server) NewSSEHandler(w http.ResponseWriter, r *http.Request)

NewSSEHandler streams every change the broadcaster fans out as one SSE `data: ...` event per change. It exits when the client disconnects.

func (*Server) Serve

func (s *Server) Serve(ln net.Listener) error

Serve serves both endpoints on ln until Shutdown is called. It returns http.ErrServerClosed after a graceful shutdown.

func (*Server) Shutdown

func (s *Server) Shutdown(ctx context.Context) error

Shutdown gracefully stops the HTTP server started by Serve or ListenAndServe: it waits for in-flight requests to finish or ctx to expire. It is a no-op when the server is not running.

Shutdown also works when it is called before Serve has started: it still marks the server shut down, so a Serve that starts afterwards refuses its listener and returns http.ErrServerClosed instead of serving forever without ever being stoppable.