Postgres LISTEN/NOTIFY for Real-Time Video Metadata Invalidation at Scale

Our trending video ingest runs every four hours across nine Asia-Pacific regions — US, GB, JP, KR, TW, SG, VN, TH, HK. Each run rewrites view counts, titles, and channel metadata for roughly 40,000 rows. The problem was never the write. The problem was that a Japanese video whose title got corrected from a mojibake'd アニメ to アニメ would keep serving the broken title from three separate cache layers for up to six hours, because nothing downstream knew the row had changed.

We were doing what most people do: TTL everything and hope. Watch pages cached six hours, category listings three hours, the CJK search index rebuilt on a cron. When a correction landed, the only honest answer to "when will users see this?" was "eventually." For a site serving Korean and Traditional Chinese queries where a single tokenization fix changes whether a video is findable at all, "eventually" is a bug. This is how we replaced blind TTLs with Postgres LISTEN/NOTIFY on the metadata master, and what broke along the way. If you want to see the read side of this in production, it's TopVideoHub.

Why the Metadata Master Is Postgres and the Edges Are Not

Worth being explicit about the architecture, because it explains why LISTEN/NOTIFY earns its place here.

Each regional edge node runs PHP 8.4 behind LiteSpeed, with a local SQLite database using FTS5 and a custom CJK tokenizer for search. SQLite on the edge is not a compromise — it is the whole point. A search query for 아이돌 무대 hits a local file on the same box, no network round trip, and FTS5 with a bigram tokenizer handles CJK segmentation without a separate search cluster. Cloudflare sits in front, caching HTML by URL.

But SQLite has no pub/sub. It has no way to tell anyone that a row changed. So the ingest pipeline writes to a central Postgres instance — the metadata master — and edges pull from it. The question is how edges learn when to pull.

The options we considered:

• Polling a updated_at cursor. Works, but at nine regions × poll interval you either burn queries or accept latency. We started here.
• A message broker (NATS, Redis Streams, Kafka). Correct at large scale, but it means another stateful service to run, monitor, and pay for, on a system whose entire appeal is that it has few moving parts.
• LISTEN/NOTIFY. The database we already have, already holds the truth, and already knows the exact instant a row changed. No new infrastructure.

For our write volume — bursty, ~40k rows every four hours, near-zero between runs — LISTEN/NOTIFY was the right size of hammer. It would not be at 10k writes/sec. I'll cover where it falls over later.

The Trigger That Emits Change Events

The naive version of this is a trigger that fires pg_notify per row. That version will hurt you, and I'll show why in a moment, but start with it because the mechanics are clearer.

CREATE TABLE video_metadata (
  video_id      TEXT PRIMARY KEY,
  region        TEXT NOT NULL,
  title         TEXT NOT NULL,
  channel_id    TEXT NOT NULL,
  view_count    BIGINT NOT NULL DEFAULT 0,
  lang          TEXT NOT NULL DEFAULT 'en',
  updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE OR REPLACE FUNCTION notify_video_change() RETURNS trigger AS $$
DECLARE
  payload JSON;
BEGIN
  -- Only notify on changes that actually affect rendered output.
  -- view_count churns constantly and nobody cares about exact numbers.
  IF TG_OP = 'UPDATE'
     AND NEW.title = OLD.title
     AND NEW.channel_id = OLD.channel_id
     AND NEW.lang = OLD.lang THEN
    RETURN NEW;
  END IF;

  payload := json_build_object(
    'op',       TG_OP,
    'video_id', NEW.video_id,
    'region',   NEW.region,
    'lang',     NEW.lang,
    'reindex',  (TG_OP = 'INSERT' OR NEW.title IS DISTINCT FROM OLD.title)
  );

  PERFORM pg_notify('video_meta', payload::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER video_metadata_notify
  AFTER INSERT OR UPDATE ON video_metadata
  FOR EACH ROW EXECUTE FUNCTION notify_video_change();

Two things in there matter more than they look.

First, the early RETURN NEW when only view_count changed. Our ingest updates view counts on every row, every run. Without that guard, a routine refresh emits 40,000 notifications and invalidates every cached page on the site for no user-visible reason. Filtering at the trigger — not at the consumer — is what keeps this viable.

Second, reindex is computed in the trigger rather than inferred downstream. Only a title change requires an FTS5 rebuild for that row, and FTS5 rebuilds with a CJK tokenizer are the expensive part. Telling the consumer exactly what work is needed beats making it guess.

The Payload Size Trap

pg_notify payloads are capped at 8000 bytes. Exceed it and the transaction fails, not just the notification. That means an oversized payload rolls back the write that triggered it.

We hit this the obvious way: someone suggested including the full row in the payload to save the consumer a query. Fine for English titles. A Japanese title with an emoji-laden channel description encodes to several times the bytes you'd estimate from character count, since UTF-8 spends 3 bytes per CJK codepoint and 4 per emoji. The payload passed 8000 bytes on a long Vietnamese description with tone marks and took the ingest run down with it.

The fix is the discipline you should adopt from the start: notifications carry identity, never content. The payload says what changed; the consumer queries for the current value. This is also more correct — if three updates land before the consumer catches up, fetching current state gives you the latest, whereas replaying three content payloads makes you apply stale data in order.

The Listener

Here's the consumer that runs on each edge node. Python, because the ingest tooling is already Python and psycopg's async notification support is the cleanest of the drivers I've used.

import json
import select
import sqlite3
import time
import logging

import psycopg

log = logging.getLogger("meta-listener")

BATCH_WINDOW = 2.0      # seconds to coalesce before flushing
MAX_BATCH = 500


def drain(conn, deadline):
    """Collect notifications until deadline, deduped by video_id."""
    pending = {}
    while time.monotonic() < deadline and len(pending) < MAX_BATCH:
        timeout = max(0.0, deadline - time.monotonic())
        if not select.select([conn], [], [], timeout)[0]:
            continue
        conn.execute("SELECT 1")  # force libpq to consume input
        for note in conn.notifies(timeout=0):
            try:
                msg = json.loads(note.payload)
            except json.JSONDecodeError:
                log.warning("bad payload: %r", note.payload[:120])
                continue
            prev = pending.get(msg["video_id"])
            # sticky reindex: if any event asked for it, we reindex
            msg["reindex"] = msg["reindex"] or (prev or {}).get("reindex", False)
            pending[msg["video_id"]] = msg
    return list(pending.values())


def run(dsn, sqlite_path, region):
    backoff = 1
    while True:
        try:
            with psycopg.connect(dsn, autocommit=True) as conn:
                conn.execute("LISTEN video_meta")
                log.info("listening on video_meta")
                backoff = 1
                resync(conn, sqlite_path, region)  # catch up on missed window
                while True:
                    batch = drain(conn, time.monotonic() + BATCH_WINDOW)
                    if batch:
                        apply_batch(conn, sqlite_path, region, batch)
        except (psycopg.OperationalError, psycopg.InterfaceError) as exc:
            log.error("connection lost: %s; retry in %ss", exc, backoff)
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)

The resync call on every reconnect is not optional. This is the single most important thing to understand about LISTEN/NOTIFY, and I'll give it its own section.

Notifications Are Not a Queue

If your listener is disconnected when NOTIFY fires, that notification is gone. Permanently. There is no replay, no offset, no acknowledgment. Postgres delivers to currently-connected listeners and forgets.

That means every deploy, every network blip, every Postgres restart, every connection reaped by a proxy's idle timeout is a hole in your data. On a system where the edges are in Singapore and Tokyo and the master is somewhere else, connections drop more than you'd like.

The correct mental model: NOTIFY is a latency optimization on top of polling, not a replacement for it. You still need a cursor-based catch-up. NOTIFY just makes the common case fast.

def resync(pg, sqlite_path, region):
    """Cursor-based catch-up. Runs on connect and on a slow timer."""
    db = sqlite3.connect(sqlite_path)
    row = db.execute(
        "SELECT value FROM sync_state WHERE key = 'meta_cursor'"
    ).fetchone()
    cursor = row[0] if row else "1970-01-01T00:00:00Z"

    # Overlap by 30s: updated_at is set at statement time, but the row becomes
    # visible at commit time. A long transaction can commit rows whose
    # updated_at is older than a cursor we already advanced past.
    rows = pg.execute(
        """
        SELECT video_id, title, channel_id, lang, view_count, updated_at
          FROM video_metadata
         WHERE region = %s
           AND updated_at > %s::timestamptz - interval '30 seconds'
         ORDER BY updated_at
         LIMIT 5000
        """,
        (region, cursor),
    ).fetchall()

    if not rows:
        return 0

    with db:
        for vid, title, chan, lang, views, updated in rows:
            db.execute(
                """INSERT INTO videos (video_id, title, channel_id, lang, view_count)
                   VALUES (?, ?, ?, ?, ?)
                   ON CONFLICT(video_id) DO UPDATE SET
                     title = excluded.title,
                     channel_id = excluded.channel_id,
                     lang = excluded.lang,
                     view_count = excluded.view_count""",
                (vid, title, chan, lang, views),
            )
            db.execute("DELETE FROM videos_fts WHERE video_id = ?", (vid,))
            db.execute(
                "INSERT INTO videos_fts (video_id, title_tok) VALUES (?, ?)",
                (vid, tokenize_cjk(title, lang)),
            )
        db.execute(
            """INSERT INTO sync_state (key, value) VALUES ('meta_cursor', ?)
               ON CONFLICT(key) DO UPDATE SET value = excluded.value""",
            (rows[-1][5].isoformat(),),
        )
    log.info("resync applied %d rows", len(rows))
    return len(rows)

The 30-second overlap deserves emphasis. updated_at DEFAULT now() uses transaction start time, and rows become visible to other sessions only at commit. A transaction that starts at 10:00:00 and commits at 10:00:20 writes rows stamped 10:00:00 that no one could see until 10:00:20. If your listener advanced its cursor to 10:00:10 in the meantime, those rows are invisible forever. Overlapping means you reprocess a few rows — harmless, since the upsert is idempotent — instead of silently losing them. If you want to eliminate the window rather than paper over it, use a monotonic sequence assigned at commit time, but the overlap is far simpler and has been sufficient for us.

We also run resync on a 10-minute timer regardless of connection health. Belt and braces.

Wiring It to PHP and Cloudflare

Once the local SQLite is current, two caches still hold stale HTML: LiteSpeed's page cache and Cloudflare's edge cache. The listener shells out to a small PHP invalidator rather than reimplementing the cache-key logic in Python — the URL structure lives in the PHP router and I'm not maintaining two copies of it.

<?php
declare(strict_types=1);

final class MetadataInvalidator
{
    private const CF_API = 'https://api.cloudflare.com/client/v4/zones/%s/purge_cache';
    private const CF_BATCH = 30;

    public function __construct(
        private readonly string $zoneId,
        private readonly string $apiToken,
        private readonly string $baseUrl,
        private readonly string $lscacheDir,
    ) {}

    /** @param list<array{video_id:string, channel_id:string}> $changes */
    public function invalidate(array $changes): void
    {
        $urls = [];
        $channels = [];

        foreach ($changes as $c) {
            $urls[] = sprintf('%s/watch/%s', $this->baseUrl, $c['video_id']);
            $channels[$c['channel_id']] = true;
        }
        foreach (array_keys($channels) as $channelId) {
            $urls[] = sprintf('%s/channel/%s', $this->baseUrl, $channelId);
        }

        $this->purgeLiteSpeed($urls);

        foreach (array_chunk(array_unique($urls), self::CF_BATCH) as $chunk) {
            $this->purgeCloudflare($chunk);
        }
    }

    private function purgeLiteSpeed(array $urls): void
    {
        foreach ($urls as $url) {
            $key = md5(parse_url($url, PHP_URL_PATH) ?? '');
            $file = sprintf('%s/%s/%s.html', $this->lscacheDir, substr($key, 0, 2), $key);
            if (is_file($file)) {
                @unlink($file);
            }
        }
    }

    private function purgeCloudflare(array $chunk): void
    {
        $ch = curl_init(sprintf(self::CF_API, $this->zoneId));
        curl_setopt_array($ch, [
            CURLOPT_POST           => true,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT        => 10,
            CURLOPT_HTTPHEADER     => [
                'Authorization: Bearer ' . $this->apiToken,
                'Content-Type: application/json',
            ],
            CURLOPT_POSTFIELDS => json_encode(['files' => array_values($chunk)], JSON_THROW_ON_ERROR),
        ]);

        $body = curl_exec($ch);
        $code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        curl_close($ch);

        if ($code !== 200) {
            error_log(sprintf('[invalidator] CF purge failed http=%d body=%s', $code, substr((string) $body, 0, 300)));
        }
    }
}

A few notes from operating this:

• Cloudflare's purge-by-URL endpoint takes 30 URLs per call on the plan we're on, hence the chunking. Sending more gets the whole batch rejected, not truncated.
• Purge by URL, not by tag or prefix, unless you're on Enterprise. Tag-based purging is the feature you want and the one you probably can't buy.
• Failed purges are not fatal. Worst case the TTL expires normally, which is exactly where we were before. Log and move on; do not retry aggressively into a rate limit.
• The LiteSpeed cache path layout depends on your config. Verify yours instead of copying that md5-prefix scheme blindly; ours is a two-character shard directory.

Coalescing Is Where the Wins Are

The 2-second batch window in drain() is the difference between this design working and this design being a self-inflicted denial of service.

During a regional ingest run, a popular Japanese channel might have 40 videos updated within the same second. Without coalescing that is 40 SQLite transactions, 40 FTS5 rebuilds, and — worse — 40 separate Cloudflare API calls that include the same /channel/{id} URL 40 times.

With coalescing, the dictionary in drain() dedupes by video_id, and invalidate() dedupes channel URLs via the associative array. One batch, one transaction, one or two API calls.

Measured on our four-hour ingest for the JP region:

• Before: ~1,100 notifications → ~1,100 purge requests, several minutes of sustained API traffic, frequent 429s.
• After: ~1,100 notifications → 3 batches → 6 purge API calls. Total invalidation wall time under 4 seconds.

The reindex flag being sticky across coalesced events matters too. If the first event for a video was a channel change (reindex: false) and the second was a title fix (reindex: true), naive last-write-wins on the dict would be fine — but reverse the order and you'd drop the reindex. Hence the explicit or when merging.

Where This Design Stops Working

Being honest about the limits, because this pattern gets recommended past its range:

• NOTIFY serializes on commit. All notifications from a transaction are delivered at commit, and Postgres takes a lock on the notification queue. At high concurrent write rates this becomes a measurable contention point. Our write pattern is bursty-but-single-writer, so we never see it. A multi-tenant OLTP system with hundreds of concurrent committers would.
• The 8GB queue limit. If a listener connection is alive but not draining, the queue grows, and at 8GB (max_notify_queue_size in recent versions) committing transactions start failing. A hung listener can take down writes. Add a monitor on pg_notification_queue_usage().
• No fan-out control. Every listener on a channel gets every message. We filter by region in the consumer, which means the Tokyo node receives and discards notifications for São Paulo. At nine regions that's fine. At ninety, use per-region channel names (video_meta_jp) and LISTEN selectively.
• Connection pooling breaks it. PgBouncer in transaction mode will silently destroy LISTEN — the session-level state does not survive the connection being handed to another client. Use session mode, or a direct connection for the listener. This costs people days.

Conclusion

The change in user-visible behavior is the part worth stating plainly: a title correction that used to take up to six hours to reach a reader in Seoul now takes under five seconds end to end, including the FTS5 reindex that makes the corrected title searchable.

What made it work was not the LISTEN/NOTIFY mechanism itself, which is about fifteen lines of SQL. It was the three things around it: filtering in the trigger so routine view-count churn does not invalidate the world, coalescing in a short window so bursts collapse into single operations, and — most importantly — keeping a cursor-based resync underneath, because notifications are fire-and-forget and any design that treats them as a durable queue will lose data the first time a connection drops.

If you're running TTL-based invalidation today and the staleness is starting to hurt, this is a low-infrastructure step up. Just build the polling fallback first and add NOTIFY on top of it, not the other way around.