Postgres as a Message Bus: Implementing Durable Event Queues with LISTEN/NOTIFY and Logical Decoding

You’ve got a small-ish system: a monolith plus a couple of background workers, maybe one or two sidecar services.

You want events. You’d like to:

Emit “order.created” when an order row is inserted
Run background jobs when a user signs up
Invalidate cache entries when products change

And immediately someone says: “We should add Kafka.”

For many teams, that’s the moment when architecture complexity starts climbing faster than actual business value. You now have another clustered system to deploy, monitor, secure, and pay for.

But you already have a distributed log with strong durability guarantees sitting right there: PostgreSQL.

In this post we’ll walk through how to treat Postgres as your event backbone, using:

LISTEN / NOTIFY for lightweight, low-latency signaling (PostgreSQL)
Tables + row locking as durable queues
Logical decoding to stream changes out of Postgres like a real message bus (PostgreSQL)

The goal isn’t to “replace Kafka forever.” It’s to show you how far you can get with the database you already have before you introduce another moving part.

What Do You Actually Need From a “Message Bus”?

Before we dive into Postgres specifics, let’s translate “I need Kafka” into concrete requirements.

Most event-driven systems actually need some subset of:

Cross-process signaling “When this happens over there, notify code running over here.”
Durability & replay If a consumer is down for a bit, it can catch up later.
Ordering At least per key ordering (e.g., events for the same order_id).
Backpressure Events pile up somewhere safe instead of knocking your app over.
At-least-once semantics You can tolerate duplicate processing as long as you don’t lose events.
Transactional coupling (the “outbox problem”) Business changes and “events about those changes” should commit together.

For a ton of small and medium systems, you don’t actually need:

Tens of thousands of messages per second
Dozens of independent consumer groups
Cross-region event pipelines

You just need a durable queue with wake-up notifications, plus a way to fan out data to a few services.

And that’s exactly the niche where Postgres shines.

A Primer on LISTEN / NOTIFY: Postgres as Pub/Sub

Postgres includes a built-in asynchronous notification mechanism:

NOTIFY channel, 'payload'
LISTEN channel

When a session calls NOTIFY on a channel, all sessions that previously issued LISTEN on that channel get an asynchronous notification with an optional payload string. (PostgreSQL)

The payload:

Is just a string (often JSON or an ID)
In default config must be less than 8000 bytes (PostgreSQL)

Crucially:

Notifications are transactional. If you NOTIFY inside a transaction, listeners only see it after the transaction commits. If the transaction rolls back, no notification is sent. (Falcon)
Postgres preserves order: notifications from a single transaction are delivered in send order; notifications from different transactions are delivered in commit order. (postgrespro.com)

Tiny demo

Open two psql sessions connected to the same database.

Session A:

LISTEN app_events;
-- psql will now print async notifications as they arrive

Session B:

NOTIFY app_events, 'hello from B';

Session A will print something like:

Asynchronous notification "app_events" with payload "hello from B" received from server process with PID 12345.

You’ve just done lightweight pub/sub inside Postgres—no extra components.

So… why doesn’t everyone just use this as a queue?

Why LISTEN/NOTIFY Alone Is Not a Queue

There are a few deal-breakers if you try to treat notifications themselves as the message store:

Notifications are not persisted They live in an in-memory queue. If a client is disconnected when a notification is sent, it simply never sees it. (Compile N Run)
No offsets or acks You can’t ask Postgres: “What notifications did I miss between 14:03 and 14:05?” There’s no built-in replay or consumer offset mechanism.
Payload size is capped (~8 KB) You really shouldn’t be stuffing large payloads into the notification itself. The official docs explicitly recommend storing big data in a table and only sending a key. (PostgreSQL)
All listeners see all notifications It’s pub/sub, not consumer groups. If you have multiple workers, each one gets the same notification and must coordinate in userland.

The pattern we want is:

Use LISTEN/NOTIFY as the “doorbell”, and a table as the durable queue.

Let’s build that.

Pattern #1: Durable Event Queues with Tables + LISTEN/NOTIFY

We’ll implement a minimal but production-grade event queue on top of Postgres primitives.

Conceptually:

Producers insert rows into an event_outbox table inside their business transactions.
The same transaction also calls pg_notify(...).
Consumers:
- LISTEN on a channel for wake-ups
- Pull rows from event_outbox using FOR UPDATE SKIP LOCKED to distribute work
- Mark events as processed

Step 1: The Outbox Table

Start with a schema like this:

CREATE TABLE event_outbox (
    id           bigserial PRIMARY KEY,
    stream_name  text        NOT NULL, -- e.g. 'order'
    stream_key   text        NOT NULL, -- e.g. 'order:123'
    type         text        NOT NULL, -- e.g. 'order.created'
    payload      jsonb       NOT NULL,
    created_at   timestamptz NOT NULL DEFAULT now(),
    processed_at timestamptz,
    -- optional metadata for locking & tracing
    locked_by    text,
    locked_at    timestamptz
);

CREATE INDEX ON event_outbox (processed_at, id);
CREATE INDEX ON event_outbox (stream_name, stream_key, id);

This table is your durable log:

id gives you global ordering
stream_name + stream_key let you reason about per-aggregate ordering
processed_at marks completion for a given consumer (we’ll start with a single consumer group)

Step 2: Enqueue Function with NOTIFY

Next, write a helper function that inserts an event and notifies listeners.

CREATE OR REPLACE FUNCTION enqueue_event(
    p_stream_name text,
    p_stream_key  text,
    p_type        text,
    p_payload     jsonb
) RETURNS bigint AS $$
DECLARE
    v_id bigint;
BEGIN
    INSERT INTO event_outbox (stream_name, stream_key, type, payload)
    VALUES (p_stream_name, p_stream_key, p_type, p_payload)
    RETURNING id INTO v_id;

    -- Fire an async notification; listeners will see it after commit
    PERFORM pg_notify('event_outbox', v_id::text);

    RETURN v_id;
END;
$$ LANGUAGE plpgsql;

Because both the insert and pg_notify run inside your application transaction:

If the transaction commits, listeners see both the durable row and the notification.
If it rolls back, neither the row nor the notification exist.

This solves the classic “outbox problem” more cleanly than “write to DB, then send Kafka message in app code.” (PostgreSQL)

Step 3: Hook Into Your Domain Writes

Example: when an order is created, you want to emit an order.created event.

You might have something like:

CREATE TABLE orders (
    id          uuid PRIMARY KEY,
    customer_id uuid NOT NULL,
    total_cents integer NOT NULL,
    created_at  timestamptz NOT NULL DEFAULT now()
);

You can call enqueue_event in your application code after inserting the orders row within the same transaction, or use a trigger:

CREATE OR REPLACE FUNCTION orders_outbox_trigger()
RETURNS trigger AS $$
BEGIN
    PERFORM enqueue_event(
        'order',
        NEW.id::text,
        'order.created',
        to_jsonb(NEW)
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER orders_outbox_after_insert
AFTER INSERT ON orders
FOR EACH ROW EXECUTE FUNCTION orders_outbox_trigger();

Now every new order automatically emits an event row and a notification.

Step 4: A Worker That Listens and Drains

Let’s use asyncpg in Python (the same idea works in Go, Node, etc.). asyncpg exposes an add_listener method that registers a callback for Postgres notifications. (yinternational.co.kr)

We’ll use:

One connection dedicated to LISTEN
Another connection (or pool) to pull and process events

import asyncio
import asyncpg
import json
import os
from contextlib import asynccontextmanager

DSN = os.getenv("DATABASE_URL", "postgres://user:pass@localhost/appdb")

CHANNEL = "event_outbox"
CONSUMER_NAME = "worker-1"
BATCH_SIZE = 10


@asynccontextmanager
async def create_pool():
    pool = await asyncpg.create_pool(DSN, min_size=1, max_size=5)
    try:
        yield pool
    finally:
        await pool.close()


async def process_event(row):
    event_type = row["type"]
    payload = row["payload"]
    print(f"[{CONSUMER_NAME}] handling {event_type} id={row['id']}")
    # do your business logic here
    await asyncio.sleep(0.1)  # pretend work


async def drain_once(conn):
    rows = await conn.fetch(
        """
        UPDATE event_outbox
        SET locked_by = $1,
            locked_at = now()
        WHERE id IN (
            SELECT id
            FROM event_outbox
            WHERE processed_at IS NULL
            ORDER BY id
            FOR UPDATE SKIP LOCKED
            LIMIT $2
        )
        RETURNING id, type, payload
        """,
        CONSUMER_NAME,
        BATCH_SIZE,
    )

    if not rows:
        return False

    for row in rows:
        try:
            await process_event(row)
            await conn.execute(
                "UPDATE event_outbox SET processed_at = now() WHERE id = $1",
                row["id"],
            )
        except Exception as e:
            # In a real system you’d log and maybe move to a DLQ table
            print(f"Error processing event {row['id']}: {e}")

    return True


async def worker_loop(pool, wakeup_queue: asyncio.Queue):
    async with pool.acquire() as conn:
        while True:
            # Drain as long as there’s work
            while await drain_once(conn):
                pass

            # Sleep a bit or wait for the doorbell
            try:
                await asyncio.wait_for(wakeup_queue.get(), timeout=5.0)
            except asyncio.TimeoutError:
                # periodic poll in case we missed a notification
                continue


async def listen_loop(wakeup_queue: asyncio.Queue):
    conn = await asyncpg.connect(DSN)
    await conn.execute(f"LISTEN {CHANNEL};")

    def _callback(connection, pid, channel, payload):
        # We don't care about which id specifically; just wake the worker
        wakeup_queue.put_nowait(None)

    await conn.add_listener(CHANNEL, _callback)
    print(f"Listening on channel {CHANNEL}...")

    try:
        while True:
            await asyncio.sleep(3600)
    finally:
        await conn.close()


async def main():
    wakeup_queue = asyncio.Queue(maxsize=1000)

    async with create_pool() as pool:
        await asyncio.gather(
            listen_loop(wakeup_queue),
            worker_loop(pool, wakeup_queue),
        )

if __name__ == "__main__":
    asyncio.run(main())

What this setup gives you:

Low latency: workers are woken up almost immediately by NOTIFY.
Durability: events live in event_outbox until marked processed.
Backpressure: if workers are slow, events simply accumulate in the table.
Safe concurrency: multiple workers can compete for events thanks to FOR UPDATE SKIP LOCKED, which is a standard Postgres pattern for job queues. (PostgreSQL Korea)

If a worker dies mid-batch, the locks are released on transaction rollback and other workers can pick them up.

Extending the Pattern: Consumer Groups & Retention

So far we’ve assumed a single logical consumer.

If you want multiple independent consumer groups (e.g. email-service, search-indexer, billing), you have a few options:

Per-consumer outbox tables Each consumer gets its own event_outbox_* table & channel. Simple but more schema.

Per-consumer “offset” table Keep event_outbox as the central log, and track consumer positions in a separate table:

 CREATE TABLE event_consumer_offsets (
     consumer_name text PRIMARY KEY,
     last_seen_id  bigint NOT NULL
 );

Workers then do:

 SELECT ...
 FROM event_outbox
 WHERE id > (SELECT last_seen_id FROM event_consumer_offsets WHERE consumer_name = $1)
 ORDER BY id
 LIMIT 100;

Update last_seen_id as part of the same transaction when marking events processed.

Hybrid Use #2 for critical consumers, and “fire-and-forget” notifications for best-effort subscribers (e.g. metrics dashboards).

For retention, you can periodically delete old, fully processed events (e.g., where all consumer offsets are greater than event_outbox.id). For small systems, even a single DELETE or partition-based pruning job run nightly is plenty.

Pattern #2: Postgres Logical Decoding as a Streaming Bus

The table + LISTEN/NOTIFY approach works great as an internal queue. But what if:

You want to stream changes into external systems (data warehouse, search index, another service)?
You’d like a more “log-like” API that doesn’t require polling the DB with ad-hoc queries?

Enter logical decoding.

Logical decoding is Postgres’s mechanism for turning the Write-Ahead Log (WAL)—the low-level stream of physical changes—into a stream of logical change events (insert/update/delete), consumable by external programs. (PostgreSQL)

Under the hood:

You create a logical replication slot, which represents a named stream of WAL changes. (PostgreSQL)
An output plugin decides how to format those changes (e.g., a JSON blob, SQL-like statements, etc.). Postgres ships with pgoutput and some example plugins; community plugins like wal2json are common. (DEV Community)

Enabling Logical Decoding

On the Postgres server, you need:

wal_level = logical
max_replication_slots = 4     # or more, depending on how many streams you need
max_wal_senders = 4

These are standard configuration knobs for logical replication and logical decoding. (postgrespro.com)

After reloading Postgres, create a replication slot:

SELECT * FROM pg_create_logical_replication_slot('events_slot', 'pgoutput');

Or with wal2json if installed:

SELECT * FROM pg_create_logical_replication_slot('events_slot', 'wal2json');

Now you have a durable “cursor” into the WAL named events_slot.

Streaming Changes Out

The Postgres distribution includes a helper CLI, pg_recvlogical, which can stream logical changes from a slot. (PostgreSQL)

Example:

pg_recvlogical -d appdb \
  --slot=events_slot \
  --plugin=wal2json \
  --start -f -

You’ll see a JSON stream with inserts/updates/deletes as they commit.

In code, you’d usually:

Use a library that can open a replication connection
Consume changes from the slot
Persist your own “offsets” or simply let the replication slot track progress

Logical decoding gives you:

Ordering and durability tied directly to WAL
A single source of truth for all committed changes in specific tables
A natural foundation for Change Data Capture (CDC) pipelines (blog.peerdb.io)

Combining Outbox + Logical Decoding

A powerful pattern is:

Keep the outbox table as above.
Configure logical decoding to only stream changes from that outbox table (e.g., via publication filters or by letting your consumer filter rows by schema/table).
External services consume the WAL-based stream instead of polling.

This is effectively “Kafka, but stored in Postgres’s WAL”:

You still get atomic coupling between domain writes and outbox inserts.
Consumers read a forward-only append log.
You haven’t introduced another infra component yet.

Later, if you do add Kafka, Debezium, etc., they can hang off the same logical decoding feed without changing the core app.

When Is “Postgres as Message Bus” the Right Tool?

This approach works extremely well when:

You already have a single Postgres cluster that is the primary system of record.
Event throughput is moderate (hundreds or thousands of messages per second, not hundreds of thousands).
Most consumers are running in the same environment as the DB.
The most important property is transactional correctness between your domain state and emitted events.

You’ll likely be happy with:

LISTEN/NOTIFY + outbox table for internal workers
Optional logical decoding for streaming to a small number of external consumers

On the other hand, you probably want Kafka, Redpanda, Pulsar, NATS, etc. when:

You have many independent consumer groups with long-lived positions.
You need to store months or years of event history independent of your OLTP database.
There’s a strong requirement for cross-region, multi-datacenter event pipelines.
You’re approaching very high throughput and need specialized storage & compaction strategies.

Think of Postgres as your small-systems event backbone:

It’s simple.
It’s operationally friendly (one system instead of two).
It reuses the guarantees you already trust for your data.

And when you eventually outgrow it, you’ll have a clean path to plug in a dedicated message bus.

Summary & Further Reading

We covered:

How LISTEN/NOTIFY gives you low-latency pub/sub inside Postgres, and why notifications alone aren’t a full queue. (PostgreSQL)
Building a durable event queue using:
- An event_outbox table as the event log
- A PL/pgSQL enqueue_event function
- LISTEN/NOTIFY as a wake-up signal
- FOR UPDATE SKIP LOCKED to safely scale concurrent workers (PostgreSQL Korea)
Using logical decoding to turn the WAL into a structured change stream that looks and feels like a message bus. (PostgreSQL)

If you’re running a small or mid-sized system and you’re on the fence about adding Kafka “because that’s what everyone does,” try this first. There’s a good chance Postgres can be your message bus for quite a while.