LISTEN / NOTIFY

-- 1) The simplest example
LISTEN orders_changed;        -- subscribe in this session
NOTIFY orders_changed, 'o1 updated';  -- emit from any session
-- Every LISTENing session receives the payload (max 8000 bytes).

-- 2) From an app — node-postgres example
-- const client = new pg.Client(...); await client.connect();
-- await client.query('LISTEN orders_changed');
-- client.on('notification', (n) => console.log(n.channel, n.payload));

-- 3) Wire it up via a trigger so every UPDATE fires a NOTIFY
CREATE OR REPLACE FUNCTION trg_notify_orders_changed()
RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify(
    'orders_changed',
    json_build_object(
      'op',  TG_OP,
      'id',  COALESCE(NEW.id, OLD.id),
      'at',  extract(epoch from now())
    )::text
  );
  RETURN NEW;
END;
$$;

CREATE TRIGGER orders_changed_trg
  AFTER INSERT OR UPDATE OR DELETE ON orders
  FOR EACH ROW
  EXECUTE FUNCTION trg_notify_orders_changed();

-- 4) Consumer pattern — invalidate a cache
-- node-postgres:
--   pgClient.on('notification', async (n) => {
--     const { id } = JSON.parse(n.payload);
--     await redis.del('order:' + id);
--   });

-- 5) Outbox pattern for AT-LEAST-ONCE delivery
-- LISTEN/NOTIFY is fire-and-forget — a listener that disconnects misses
-- events. For at-least-once, persist + notify:
CREATE TABLE outbox (
  id          bigserial PRIMARY KEY,
  topic       text NOT NULL,
  payload     jsonb NOT NULL,
  created_at  timestamptz DEFAULT now(),
  processed_at timestamptz
);

CREATE OR REPLACE FUNCTION trg_orders_outbox()
RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
  INSERT INTO outbox(topic, payload)
  VALUES ('orders_changed',
          json_build_object('op', TG_OP, 'id', COALESCE(NEW.id, OLD.id))::jsonb);
  PERFORM pg_notify('orders_changed', 'wake');
  RETURN NEW;
END;
$$;

CREATE TRIGGER orders_outbox_trg
  AFTER INSERT OR UPDATE OR DELETE ON orders
  FOR EACH ROW EXECUTE FUNCTION trg_orders_outbox();

-- Consumer (single-instance for ordering, multi-instance with FOR UPDATE SKIP LOCKED)
-- 1) LISTEN orders_changed
-- 2) On notification, SELECT * FROM outbox WHERE processed_at IS NULL
--    ORDER BY id LIMIT 100 FOR UPDATE SKIP LOCKED;
-- 3) Do the side effect (HTTP, queue, email, etc.)
-- 4) UPDATE outbox SET processed_at = now() WHERE id IN (...)

-- 6) Gotchas
-- - Payload limited to 8000 bytes (after JSON encoding). Use outbox + id.
-- - Notifications are queued per LISTENing connection; slow consumers can pile up.
-- - If the consumer disconnects, queued notifications are LOST.
-- - The notification is only delivered AFTER the issuing transaction commits.

-- 7) Inspect
-- pg_listening_channels()                          -- channels of this session
-- SELECT * FROM pg_stat_activity WHERE wait_event = 'notify';
-- SELECT pg_notification_queue_usage();             -- backlog ratio (0..1)

-- 8) When to graduate
-- - Multiple services consuming the same events  -> Kafka / Redis Streams / NATS
-- - Cross-region replication of events           -> Debezium + Kafka Connect
-- - Strict ordering with at-most-once            -> Kafka with idempotent producers
-- - Internal cache invalidation, single deploy   -> LISTEN/NOTIFY is the right tool

-- Listener
LISTEN new_order;
-- Publisher
NOTIFY new_order, '{"id":42}';