PostgreSQL NOTIFY to RabbitMQ Bridge & Event Outbox

Overview

Many events in this system originate at the database level. When a user taps a campaign link, for example, a row is inserted into tracking_log and a Postgres trigger fires pg_notify.

The pg-listener profile is the process that holds a dedicated connection open, LISTENs on 5 channels, and converts every notification into a RabbitMQ message for the trigger worker to process.

Because pg_notify is fire-and-forget — events are lost if the connection drops — the system also keeps an event_outbox table as a durable backup. The pg-listener catches up after each reconnect, sweeps stalled rows, and periodically cleans out old ones.

Business Flow

Listening and publishing

1. Open a dedicated connection rather than using the shared pool, because LISTEN must stay bound to a single connection for its entire lifetime.

2. LISTEN on all 5 channels, each mapped to a destination queue:

   pg_notify channel	RabbitMQ queue
   attribute_changed	attribute_change
   friend_track_event_inserted	friend_track_event_trigger
   form_submitted	form_submitted_trigger
   campaign_click	campaign_click_trigger
   booking_event	booking_event_trigger

3. On each notification, unmarshal the payload and re-marshal it with encoding/json so the bytes match V8's canonical form, then publish it to the mapped queue verbatim, with no envelope wrapper.

4. If the connection drops, reconnect with exponential backoff of min(2^n seconds, 30s) for up to 100 attempts, then call os.Exit(1) so the orchestrator restarts the pod.

Managing the event outbox

• Catch-up — after a successful reconnect, fetch rows from event_outbox created after the disconnect timestamp using FOR UPDATE SKIP LOCKED and republish them, so no event is lost during downtime.
• Sweep — every 5 minutes, pick up rows that have been pending for more than 30 seconds without being delivered.
• Cleanup — every hour, delete rows older than 24 hours.

Health checks

• Registers a readiness probe named pg-listener; if the LISTEN connection drops, /readyz fails immediately.
• Exposes GET /health on the health port (PG_LISTENER_HEALTH_PORT, default 6010), returning status, pg.connected, pg.disconnectedAt, and timestamp.

Key Files & Functions

• internal/pglistener/service.go
  • New(cfg, pub, log) — constructs the bridge; the channel-to-queue mapping comes from config.Queues.* rather than being hardcoded
  • Run(ctx), listenLoop(), waitLoop(), handleNotification()
  • catchupFromOutbox(), sweepOutbox(), cleanupOutbox()
  • backoff(), IsConnected(), DisconnectedAt()
• cmd/worker/main.go — runPgListener(), which registers the readiness probe and the /health handler
• Published queues: attribute_change, friend_track_event_trigger, form_submitted_trigger, campaign_click_trigger, and booking_event_trigger

Connections to Other Services

• PostgreSQL — a dedicated pgx.Conn for LISTEN, plus the event_outbox table accessed through SQL copied verbatim from the original implementation
• Database-side Postgres triggers — for example tracking_log_campaign_click_notify, which fires pg_notify on the campaign_click channel and inserts into event_outbox in the same statement (see the database project documentation for the schema)
• RabbitMQ — this profile only publishes; it never consumes
• Downstream consumers — all 5 queues are consumed by the trigger-worker profile, except booking_event_trigger, which is handled by cron-scheduler