How to Build a Real-Time Indexing Pipeline With Redis and PostgreSQL 19

A search box that's 30 seconds behind reality feels broken even when nothing is technically wrong. This is the plumbing that fixes that — with the honest limits included, not glossed over.

Why "just query the database again" stops working

Most apps start with the simplest possible design: every few seconds, a background job asks the database "anything new since last time?" and rebuilds whatever index needs updating. This works at small scale. The problem is that it ties freshness to an arbitrary timer instead of to the moment something actually happened, and the repeated SELECT ... WHERE updated_at > ? gets more expensive as the table grows.

An event-driven pipeline flips this: Postgres announces a change the moment it happens, and a message system carries that announcement to whoever's listening. Nobody polls.

What NOTIFY does, what changed in PG19, and the 8KB ceiling you must design around

Postgres has had a "tell me when something happens" mechanism for years: any session can run LISTEN orders_changed, and any other session can run NOTIFY orders_changed (usually via pg_notify() inside a trigger) to broadcast a message to every session listening on that channel — no polling required.

Two things matter here. First, PostgreSQL 19 — currently in beta, with Beta 2 released July 16, 2026 — changes NOTIFY so it wakes only backends actually listening on that channel, instead of most connected backends. That's a real scalability fix for pipelines like this one.

Second, and this is easy to miss in a first draft: pg_notify payloads are capped at 8000 bytes. Sending a full row as JSON through NOTIFY works fine in a demo and then fails in production the day someone adds a large text or JSONB column, or the row itself grows. The fix is to never send the row — send only an identifier and an operation type, and let the consumer fetch the current state when it processes the event:

CREATE OR REPLACE FUNCTION notify_order_change() RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify(
    'orders_changed',
    json_build_object('id', NEW.id, 'op', TG_OP)::text
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER orders_notify
AFTER INSERT OR UPDATE ON orders
FOR EACH ROW EXECUTE FUNCTION notify_order_change();

This has a second benefit beyond staying under 8KB: because the consumer re-reads the row by id instead of trusting a stale snapshot, it naturally handles the case where two updates to the same row arrive close together — the second read just picks up whatever is current.

Redis Pub/Sub versus Streams — and where Redis's guarantee actually starts

Redis Pub/Sub looks like the obvious carrier for these events: subscribe, get messages as they arrive. The problem is that Pub/Sub stores nothing — if your worker restarts, deploys, or drops its connection, every message sent during that gap is gone. Redis Streams fix this: events written with XADD stay in the log until trimmed, so a worker that reconnects can pick up where it left off, and a consumer group lets multiple workers split the load without double-processing.

But it's worth being precise about where this durability guarantee begins. Once an event is written to the stream, Streams will not lose it. What Streams cannot protect is the moment before that — the gap between Postgres firing NOTIFY and some relay process actually being connected and writing to the stream. That gap is the subject of the Caveats section below.

Wiring it together, with the failure modes handled instead of ignored

The relay turns each NOTIFY into a durable stream entry. It is a single, always-on process — and it is honest to say up front that it has no persistence of its own: if it's disconnected or restarting at the exact moment NOTIFY fires, that event never reaches Redis at all. Postgres does not buffer it for you. This is the pipeline's real weak point, addressed properly in the Caveats section below rather than hidden.

import asyncio
import asyncpg
import redis.asyncio as redis

async def relay():
    pg = await asyncpg.connect("postgresql://user:pass@db/app")
    r = redis.Redis(host="redis", port=6379)

    async def on_notify(conn, pid, channel, payload):
        # payload is small: {"id": 123, "op": "UPDATE"} — never the full row,
        # to stay under Postgres's 8KB NOTIFY limit.
        await r.xadd("orders:changes", {"payload": payload})

    await pg.add_listener("orders_changed", on_notify)
    await asyncio.Future()  # keep the connection open; monitor this process's uptime closely

asyncio.run(relay())

The consumer side needs two things a first draft usually skips: it must re-fetch the current row instead of trusting the tiny event payload, and it must periodically reclaim messages left behind by a worker that crashed mid-processing — otherwise those messages sit forever in the Pending Entries List, acknowledged by no one:

import json
import redis
import psycopg2

r = redis.Redis(host="redis", port=6379)
pg = psycopg2.connect("postgresql://user:pass@db/app")
r.xgroup_create("orders:changes", "indexers", id="0", mkstream=True)

CONSUMER = "worker-1"
IDLE_MS = 30_000  # reclaim anything stuck for 30 seconds

def apply_change(order_id):
    with pg.cursor() as cur:
        cur.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
        row = cur.fetchone()
        update_search_index(row)  # idempotent: always writes current state, never the stale event

while True:
    # Reclaim messages abandoned by a crashed worker before reading anything new
    _, claimed, _ = r.xautoclaim(
        "orders:changes", "indexers", CONSUMER, min_idle_time=IDLE_MS, start_id="0"
    )
    for msg_id, fields in claimed:
        apply_change(json.loads(fields[b"payload"])["id"])
        r.xack("orders:changes", "indexers", msg_id)

    resp = r.xreadgroup("indexers", CONSUMER, {"orders:changes": ">"}, count=10, block=5000)
    for stream, messages in resp or []:
        for msg_id, fields in messages:
            apply_change(json.loads(fields[b"payload"])["id"])
            r.xack("orders:changes", "indexers", msg_id)

services:
  db:
    image: postgres:19
    environment:
      POSTGRES_PASSWORD: pass
  redis:
    image: redis:8.8
  relay:
    build: ./relay
    restart: always
    depends_on: [db, redis]
  indexer:
    build: ./indexer
    restart: always
    depends_on: [redis]

Caveats & Edge Cases: when this pattern isn't enough

Three limits are worth stating plainly rather than discovering in an incident:

The 8KB NOTIFY ceiling. Never send a full row through pg_notify; send an id and let the consumer read current state. The code above already does this, but it's worth restating as a rule, not just a code comment.

The relay is a real single point of failure. LISTEN/NOTIFY has no buffering on Postgres's side. If the relay process is disconnected, crashing, or mid-restart at the instant NOTIFY fires, that specific event is gone permanently — Redis Streams never see it, because nothing wrote it there. Monitoring the relay's uptime and alerting on gaps is not optional hardening; it's part of making this pattern trustworthy. For a search index, a stale cache, or a non-critical read model, an occasional missed event that a later write will naturally correct is usually an acceptable trade. For financial reconciliation, audit trails, or anything where every single change must be captured with zero exceptions, it is not.

When zero-loss actually matters, replace NOTIFY with logical replication. Postgres's write-ahead log is itself a durable, ordered record of every change, independent of whether any listener happened to be connected at the time. Tools that decode that log directly — Debezium, wal2json, or a custom consumer of Postgres's native pgoutput plugin — don't have the relay's blind spot, because they read from the log after the fact rather than catching a live broadcast. The trade-off is more operational complexity: replication slots to manage, a schema to keep in sync, and generally a heavier piece of infrastructure than a NOTIFY listener. Reach for it when the cost of losing an event is genuinely higher than the cost of running it.

One more decision sits alongside this, and it's about your business rather than your database: which Redis you're allowed to run. Since 2024 Redis's license has moved from BSD to source-available SSPL/RSAL to a tri-license that added the OSI-approved AGPLv3 back in Redis 8.0, while Valkey — the Linux Foundation's fork — stayed BSD-3 throughout. The full comparison across Redis's forks covers which one fits a given business model. And if the reason you're building this pipeline is to avoid a NoSQL bill in the first place, the cost math of Postgres versus a managed store like DynamoDB is worth reading before you commit either way.

Every piece here is deliberate: NOTIFY replaces the timer, the id-only payload respects Postgres's own 8KB limit, Streams replace Pub/Sub's memory loss, and XAUTOCLAIM replaces a single point of silent failure inside the consumer. What none of it replaces is the relay's exposure to its own downtime — that's a property of LISTEN/NOTIFY itself, and the only real fix is either operational (monitor it tightly) or architectural (move to logical replication). The same "don't poll, get told" instinct shows up at a very different scale in Discord's WebSocket gateway, worth a look if this pipeline's next step is pushing updates straight to a browser.