PostgreSQL LISTEN/NOTIFY — Redis 없이 DB 하나로 실시간 이벤트를 클라이언트까지 전달하는 법

-- INSERT/UPDATE에만 적용되므로 NEW가 항상 존재합니다.
-- DELETE 이벤트를 추가하려면 NEW 대신 OLD를 참조해야 합니다.
CREATE OR REPLACE FUNCTION notify_order_change() RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify(
    'orders_channel',
    json_build_object(
      'op', TG_OP,
      'id', NEW.id,
      'status', NEW.status
    )::text
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
CREATE TRIGGER orders_after_change
AFTER INSERT OR UPDATE ON orders
FOR EACH ROW EXECUTE FUNCTION notify_order_change();

import asyncio
import asyncpg
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
 
app = FastAPI()
 
DATABASE_URL = "postgresql://user:password@localhost/mydb"
 
async def get_listen_connection():
    return await asyncpg.connect(DATABASE_URL)
 
async def event_generator(order_id: int):
    conn = await get_listen_connection()
    queue: asyncio.Queue = asyncio.Queue()
 
    # 외부 스코프의 conn과 구분하기 위해 _conn으로 명명
    async def handle_notification(_conn, pid, channel, payload):
        data = json.loads(payload)
        if data.get("id") == order_id:
            await queue.put(payload)
 
    await conn.add_listener("orders_channel", handle_notification)
 
    try:
        while True:
            try:
                payload = await asyncio.wait_for(queue.get(), timeout=30.0)
                yield f"data: {payload}\n\n"
            except asyncio.TimeoutError:
                # SSE keep-alive: 연결 유지를 위한 ping
                yield ": ping\n\n"
    finally:
        await conn.remove_listener("orders_channel", handle_notification)
        await conn.close()
 
@app.get("/orders/{order_id}/events")
async def order_events(order_id: int):
    return StreamingResponse(
        event_generator(order_id),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            # Nginx 전용 헤더. Caddy, HAProxy, AWS ALB 등 다른 프록시는
            # 각자의 SSE 버퍼링 비활성화 설정을 별도로 확인해야 합니다.
            "X-Accel-Buffering": "no",
        }
    )

import createSubscriber from "pg-listen";
import { Pool } from "pg";
 
const databaseUrl = process.env.DATABASE_URL;
 
// LISTEN 전용 연결 - 일반 풀과 분리
const subscriber = createSubscriber({ connectionString: databaseUrl });
 
// WebSocket 클라이언트를 Set으로 추적
const wsClients = new Set();
 
subscriber.notifications.on("orders_channel", (payload) => {
  console.log("주문 변경 감지:", payload);
  for (const client of wsClients) {
    if (client.readyState === WebSocket.OPEN) {
      client.send(JSON.stringify(payload));
    }
  }
});
 
subscriber.events.on("error", (error) => {
  console.error("LISTEN 연결 오류:", error);
  // pg-listen이 자동으로 재연결 시도
});
 
await subscriber.connect();
await subscriber.listenTo("orders_channel");
 
// 일반 쿼리용 풀은 별도로 유지
const pool = new Pool({ connectionString: databaseUrl });

CREATE OR REPLACE FUNCTION notify_tenant_event() RETURNS trigger AS $$
BEGIN
  PERFORM pg_notify(
    'tenant_' || NEW.tenant_id::text,
    json_build_object(
      'event', TG_OP,
      'table', TG_TABLE_NAME,
      'id', NEW.id
    )::text
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

channel_name = f"tenant_{tenant_id}"
await conn.add_listener(channel_name, handle_notification)

CREATE TABLE outbox (
    id BIGSERIAL PRIMARY KEY,
    channel TEXT NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    processed_at TIMESTAMPTZ
);
 
-- NOTIFY는 "신호"로만 쓰고, 실제 데이터는 아웃박스에 저장
CREATE OR REPLACE FUNCTION notify_with_outbox() RETURNS trigger AS $$
DECLARE
  payload JSONB;
BEGIN
  payload := json_build_object('op', TG_OP, 'id', NEW.id);
  INSERT INTO outbox (channel, payload) VALUES ('orders_channel', payload);
  PERFORM pg_notify('orders_channel', '{"type":"new_event"}');
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

async def process_outbox(conn):
    # SKIP LOCKED로 다중 인스턴스 환경에서의 중복 처리 방지
    rows = await conn.fetch("""
        SELECT id, channel, payload
        FROM outbox
        WHERE processed_at IS NULL
        ORDER BY id
        FOR UPDATE SKIP LOCKED
    """)
    for row in rows:
        await handle_event(row["channel"], row["payload"])
        await conn.execute(
            "UPDATE outbox SET processed_at = NOW() WHERE id = $1",
            row["id"]
        )
 
async def handle_notification(_conn, pid, channel, payload):
    async with pool.acquire() as conn:
        await process_outbox(conn)