Building a Kafka-free CDC Pipeline in Go (and the DDL bug that almost broke my brain)

So I've been obsessed with Postgres internals for a while now - WAL, replication slots, MVCC, all of it. I built pgstream a while back just to read the WAL and stream changes in real time, and it kind of took off (34 stars, got mentioned on Golang News Twitter which honestly made my whole week). But pgstream was just a reader. It didn't actually do anything with the data besides print it or forward it raw.

So I started building Rift. The idea was simple: I wanted a single Go binary that reads Postgres WAL changes and ships them to wherever you want - a webhook, another Postgres database, Redis - without needing Kafka, Zookeeper, Debezium, or any of that JVM-based infra that honestly feels like overkill for 90% of side projects and even a lot of production use cases.

Why not just use Debezium?

Look, Debezium is great. It's battle-tested, it's what actual companies run But spinning up Kafka Connect just to sync a couple tables between two Postgres instances feels like using a shipping container to move a backpack. I wanted something you could go install and run in like 30 seconds, with one yaml file.

So that's what Rift is. Here's roughly what it does:

~Reads the WAL using logical replication (pglogrepl + pgx, no C bindings, no cgo headaches)
~Decodes INSERT/UPDATE/DELETE events
~Optionally filters/transforms them with a JS script (using goja, so no need to shell out to Node)
~Ships them to one or more destinations - webhook, Postgres, Redis
~If a destination is down, events get queued to disk (BoltDB) instead of getting dropped or blocking everything

That last part matters a lot more than I expected once I actually started testing it.

The DDL bug that ate my weekend

Once the basic INSERT/UPDATE/DELETE pipeline was working, I wanted to go further — track schema changes too. Like if someone runs ALTER TABLE users ADD COLUMN city TEXT on the source db, Rift should know about it, not just silently break next time it tries to sync a row with a new column it's never seen.

The way I did this was with a Postgres event trigger. Basically:

Rift installs a function + event trigger in the source database
Any DDL command (CREATE TABLE, ALTER TABLE, CREATE INDEX, whatever) fires the trigger
The trigger writes a row into an internal table called rift_ddl_log
Since rift_ddl_log is a normal table, its INSERTs show up in the WAL too
Rift picks those up and forwards them to destinations as DDL events

Sounds clean right? Except I completely missed something obvious: the INSERT into rift_ddl_log is itself a WAL event, and Rift was treating it exactly like any other table's insert - meaning it tried to sync rift_ddl_log rows into my destination databases too. Which obviously don't have a rift_ddl_log table. So I'd get:

ERROR: relation "rift_ddl_log" does not exist

Over and over. Every 5 seconds. Forever. Because the failed event kept getting shoved into the disk queue and retried.

My first fix attempt was dumb - just checking if event.Table == "rift_ddl_log" and skipping it. Didn't work. Turns out the table name coming through was schema-qualified in some paths (public.rift_ddl_log) and not in others, so a strict equality check missed half the cases. Took me embarrassingly long to figure that out because I was staring at the filter logic instead of actually logging what event.Table looked like at the point it failed.

Fix ended up being a small isInternalTable() helper that checks both exact match and suffix match, so it doesn't matter if the schema prefix is there or not:

func isInternalTable(table string) bool {
    internal := []string{"rift_ddl_log", "public.rift_ddl_log"}
    for _, t := range internal {
        if strings.EqualFold(table, t) || strings.HasSuffix(table, t) {
            return true
        }
    }
    return false
}

Once that was in, rift_ddl_log inserts got filtered out cleanly ([FILTERED] INSERT rift_ddl_log in the logs now, instead of an infinite error loop), while actual schema changes like CREATE TABLE, ALTER TABLE, CREATE INDEX still got captured and forwarded correctly.

Honestly this was a good reminder that "just add a string check" bugs are rarely actually about the check - they're about not knowing exactly what shape your data is in at that point in the pipeline. Should've logged the raw value first instead of guessing.

The other thing I learned: DDL isn't the same as schema replication

While testing this, I hit something that made me realize a gap in the whole design. I inserted a row into a table Rift didn't know about yet, and got:

error sending to analytics-db: ERROR: relation "users" does not exist

At first I thought "wait, didn't the DDL tracking just handle this?" But no — DDL tracking detects and forwards schema changes as metadata (like "hey, someone ran CREATE TABLE users"), it doesn't automatically apply that DDL to the destination database. Those are two very different things. Rift knows the schema changed, but it doesn't (yet) go run CREATE TABLE users (...) on the destination for you.

Right now that's a real limitation I'm noting honestly in the release notes instead of pretending it's not there. Auto-applying DDL to Postgres destinations is genuinely tricky - you have to worry about column type differences, existing data conflicts, permissions, and probably a dozen edge cases I haven't thought of yet. It's on my list for a future version, but I'd rather ship something honest than something that pretends to do more than it does.

Making it feel like an actual tool

For the first while I was just running everything with go run main.go, which is fine for testing but feels pretty amateur if you're trying to get someone else to actually use your project. So I rebuilt the CLI using Cobra — now it's:

rift run
rift run -c custom-config.yaml
rift version

instead of a raw script. go install github.com/mujib77/rift@latest and you're done - no cloning, no build step, it just becomes a command on your machine like git or docker. Small change but it genuinely changes how the project feels, both to use and honestly just to work on.

Where it's at now

Rift can currently:

-Stream INSERT/UPDATE/DELETE from Postgres to webhook, Postgres, or Redis destinations
-Filter/transform events with a JS filter script
-Track DDL changes via event triggers (with the filtering bug above, finally, actually fixed)
-Survive destination downtime using an embedded disk queue, no external broker needed
-Run as a proper installed CLI binary

It's still early - 1 star as of writing this, so if you're reading this and it sounds useful, go kick the tires and tell me what breaks. That's genuinely the fastest way I learn what's actually wrong with it versus what I assume is wrong with it.

Repo's here: https://github.com/mujib77/rift

If you've built something similar or have thoughts on how you'd handle the DDL-to-destination sync problem, I'd love to hear it in the comments - still figuring that part out.https://github.com/mujib77

Comments

[Mads Hansen]

The raw-value logging lesson is exactly right. One caution: suffix matching identifiers can create a new false positive, for example a user table whose name ends in rift_ddl_log. I would normalize schema and relation as separate parsed fields, or better classify the internal relation by OID from replication metadata. The deeper production boundary is an LSN/schema barrier: quarantine row events that require a DDL version the destination has not acknowledged, then dead-letter them after bounded retries. Detecting DDL, applying it, and proving the destination schema is ready should be three explicit states.