PostgreSQL LISTEN/NOTIFY for pub-sub messaging

-- WebSocket push notifications via LISTEN/NOTIFY
/*
// Node.js WebSocket server
const WebSocket = require('ws');
const { Client } = require('pg');

const wss = new WebSocket.Server({ port: 8080 });
const dbClient = new Client();
await dbClient.connect();

// Subscribe to database notifications
dbClient.on('notification', (msg) => {
  const payload = JSON.parse(msg.payload);

  // Broadcast to all connected WebSocket clients
  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(JSON.stringify({
        channel: msg.channel,
        data: payload
      }));
    }
  });
});

await dbClient.query('LISTEN new_order');
await dbClient.query('LISTEN user_updates');
*/

-- Job queue pattern
CREATE TABLE job_queue (
  id SERIAL PRIMARY KEY,
  job_type VARCHAR(50),
  payload JSONB,
  status VARCHAR(20) DEFAULT 'pending',
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  started_at TIMESTAMP,
  completed_at TIMESTAMP
);

CREATE OR REPLACE FUNCTION notify_new_job()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify(
    'job_queue',
    json_build_object(
      'job_id', NEW.id,
      'job_type', NEW.job_type
    )::TEXT
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER job_notification
AFTER INSERT ON job_queue
FOR EACH ROW
EXECUTE FUNCTION notify_new_job();

-- Worker process listens for jobs
/*
client.on('notification', async (msg) => {
  const { job_id, job_type } = JSON.parse(msg.payload);

  // Claim job
  const result = await client.query(`
    UPDATE job_queue
    SET status = 'processing', started_at = NOW()
    WHERE id = $1 AND status = 'pending'
    RETURNING *
  `, [job_id]);

  if (result.rows.length === 0) return; // Already claimed

  // Process job
  await processJob(job_type, result.rows[0].payload);

  // Mark complete
  await client.query(`
    UPDATE job_queue
    SET status = 'completed', completed_at = NOW()
    WHERE id = $1
  `, [job_id]);
});

await client.query('LISTEN job_queue');
*/

-- Rate limiting notifications (prevent spam)
CREATE TABLE notification_throttle (
  channel VARCHAR(100) PRIMARY KEY,
  last_notification TIMESTAMP
);

CREATE OR REPLACE FUNCTION throttled_notify(
  channel_name TEXT,
  message TEXT,
  throttle_seconds INT DEFAULT 1
)
RETURNS VOID AS $$
DECLARE
  last_sent TIMESTAMP;
BEGIN
  SELECT last_notification INTO last_sent
  FROM notification_throttle
  WHERE channel = channel_name;

  IF last_sent IS NULL OR
     NOW() - last_sent > (throttle_seconds || ' seconds')::INTERVAL THEN
    PERFORM pg_notify(channel_name, message);

    INSERT INTO notification_throttle (channel, last_notification)
    VALUES (channel_name, NOW())
    ON CONFLICT (channel)
    DO UPDATE SET last_notification = NOW();
  END IF;
END;
$$ LANGUAGE plpgsql;

-- Usage: Only send notification every 5 seconds
SELECT throttled_notify('metrics_update', 'New metrics available', 5);

-- Bi-directional communication
-- Application -> Database (via INSERT/UPDATE)
-- Database -> Application (via NOTIFY)

-- Command pattern
CREATE TABLE commands (
  id SERIAL PRIMARY KEY,
  command_type VARCHAR(50),
  parameters JSONB,
  result JSONB,
  status VARCHAR(20) DEFAULT 'pending',
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Application sends command
INSERT INTO commands (command_type, parameters)
VALUES ('send_email', '{"to": "user@example.com", "subject": "Hello"}');

-- Trigger notifies workers
CREATE TRIGGER command_notification
AFTER INSERT ON commands
FOR EACH ROW
EXECUTE FUNCTION notify_new_command();

-- Worker processes command and updates result
UPDATE commands
SET status = 'completed',
    result = '{"sent": true, "message_id": "abc123"}'
WHERE id = 123;

-- Monitoring active listeners
SELECT
  pid,
  usename,
  application_name,
  client_addr,
  backend_start,
  state
FROM pg_stat_activity
WHERE query LIKE '%LISTEN%';