LISTEN e NOTIFY no Node.js com PostgreSQL

LISTEN e NOTIFY no Node.js permitem que processos conectados ao PostgreSQL troquem sinais leves por canais. Uma sessão executa LISTEN canal, outra envia NOTIFY canal, 'payload', e o servidor entrega a notificação aos ouvintes depois do commit.

O recurso é útil para invalidar cache, acordar workers, informar que uma tabela recebeu novos dados ou coordenar mudanças pequenas. Ele não é uma fila durável: mensagens não ficam armazenadas para consumidores desconectados, o payload é limitado e a entrega não substitui persistência.

Primeiro listener com node-postgres

import pg from 'pg';

const client = new pg.Client({
  connectionString: process.env.DATABASE_URL
});

await client.connect();
await client.query('LISTEN cache_invalidation');

client.on('notification', (message) => {
  console.log(message.channel, message.payload);
});

O listener precisa manter uma conexão dedicada. Não use uma conexão devolvida ao pool enquanto espera eventos, pois outra requisição pode alterar seu estado de sessão.

Enviando uma notificação

await pool.query(
  `SELECT pg_notify($1, $2)`,
  [
    'cache_invalidation',
    JSON.stringify({ type: 'product.updated', id: productId })
  ]
);

pg_notify aceita parâmetros, evitando concatenar canal e payload. Canais ainda devem seguir uma convenção controlada; não permita que o usuário escolha qualquer canal.

Entrega após commit

await client.query('BEGIN');
await client.query(
  `UPDATE produtos SET nome = $2 WHERE id = $1`,
  [productId, newName]
);
await client.query(
  `SELECT pg_notify('cache_invalidation', $1)`,
  [JSON.stringify({ id: productId })]
);
await client.query('COMMIT');

A notificação só é entregue quando a transação confirma. Se houver rollback, ela não é enviada. Isso ajuda a evitar sinais sobre dados que não foram persistidos.

Não é uma fila

LISTEN/NOTIFY não oferece:

• replay para consumidores offline;
• ack individual;
• retenção de mensagens;
• dead letter queue;
• particionamento;
• garantia de processamento;
• payloads grandes.

Para eventos de negócio que não podem ser perdidos, use uma tabela Outbox ou broker. Consulte Outbox Pattern no Node.js e RabbitMQ no Node.js.

Padrão “acorde e consulte”

Uma abordagem robusta é persistir o trabalho em uma tabela e usar NOTIFY apenas para acordar o worker.

await client.query('BEGIN');

await client.query(
  `INSERT INTO jobs (id, type, payload, status)
   VALUES ($1, $2, $3::jsonb, 'pending')`,
  [jobId, jobType, JSON.stringify(payload)]
);

await client.query(
  `SELECT pg_notify('jobs_available', $1)`,
  [jobId]
);

await client.query('COMMIT');

Se o listener estiver desconectado, o job continua na tabela e pode ser encontrado por polling periódico.

Worker com fallback de polling

let wakeRequested = false;

listener.on('notification', () => {
  wakeRequested = true;
});

setInterval(async () => {
  if (!wakeRequested) {
    await processarPendentes();
    return;
  }

  wakeRequested = false;
  await processarPendentes();
}, 1000);

Use controle de concorrência para não executar o loop várias vezes simultaneamente.

Consumindo jobs com lock

const result = await client.query(
  `SELECT id, payload
   FROM jobs
   WHERE status = 'pending'
   ORDER BY created_at
   FOR UPDATE SKIP LOCKED
   LIMIT 10`
);

SKIP LOCKED permite que vários workers processem linhas diferentes. O NOTIFY reduz latência, enquanto a tabela fornece durabilidade.

Payload pequeno

Envie somente identificador, tipo e versão. O consumidor deve buscar o estado completo no banco.

{
  "type": "invoice.changed.v1",
  "id": "8f...",
  "tenantId": "t-1"
}

Payloads grandes aumentam tráfego e atingem o limite do PostgreSQL. Dados sensíveis também não devem circular desnecessariamente.

Validando notificações

function parseNotification(message) {
  if (message.channel !== 'cache_invalidation') {
    return null;
  }

  try {
    const data = JSON.parse(message.payload ?? '{}');

    if (typeof data.id !== 'string' || data.id.length > 100) {
      return null;
    }

    return data;
  } catch {
    return null;
  }
}

Mesmo sendo originado pelo banco, o payload pode vir de outra versão da aplicação ou de operação manual. Valide antes de usar.

Reconexão

Quando a conexão cai, as inscrições são perdidas. O processo precisa reconectar e executar LISTEN novamente.

async function connectListener() {
  const client = new pg.Client(config);
  await client.connect();
  await client.query('LISTEN jobs_available');

  client.on('error', (error) => {
    logger.error({ error }, 'listener PostgreSQL desconectado');
    scheduleReconnect();
  });

  return client;
}

Use backoff com jitter para evitar tempestade de reconexões. Após reconectar, execute uma varredura dos itens pendentes porque notificações ocorridas durante a queda foram perdidas.

Conexão dedicada e PgBouncer

LISTEN depende de sessão. Em PgBouncer com transaction pooling, a aplicação não mantém a mesma sessão entre comandos. Use uma conexão direta ou session pooling para listeners.

Consulte PgBouncer com Node.js.

Invalidando cache

listener.on('notification', (message) => {
  const event = parseNotification(message);
  if (!event) return;

  cache.delete(`product:${event.tenantId}:${event.id}`);
});

A chave precisa incluir tenant e contexto. Considere TTL mesmo com invalidação, pois uma notificação perdida não deve manter dados obsoletos para sempre.

Múltiplas instâncias

Todas as sessões que executaram LISTEN recebem o sinal. Isso é útil para invalidar cache local em cada instância. Para distribuir um job a apenas um consumidor, mantenha o job numa tabela com lock ou use um broker.

Debounce e coalescing

Uma transação pode alterar milhares de linhas. Em vez de enviar uma notificação por linha, envie uma notificação por agregado, tenant ou versão de cache.

SELECT pg_notify(
  'catalog_changed',
  json_build_object('tenantId', $1, 'version', $2)::text
);

O consumidor pode invalidar um namespace inteiro ou carregar um lote de mudanças.

Canais por tenant?

Evite criar um canal dinâmico para cada usuário ou tenant pequeno. Um número grande de canais aumenta complexidade. Use um canal estável e inclua o tenant no payload, validando autorização e isolamento.

Segurança

• não exponha uma rota que envie NOTIFY arbitrário;
• restrinja permissões de banco;
• limite tamanho e formato do payload;
• não inclua segredos;
• valide tenant e tipo;
• registre falhas sem vazar dados;
• trate canal como configuração controlada.

Observabilidade

Monitore estado da conexão, número de reconexões, notificações recebidas, payloads inválidos, atraso entre persistência e processamento e quantidade de itens encontrados no polling de recuperação.

Um contador alto de notificações com poucos itens pendentes pode indicar duplicação ou sinalização excessiva.

Graceful shutdown

async function shutdown() {
  await listener.query('UNLISTEN *').catch(() => {});
  await listener.end().catch(() => {});
  await pool.end();
}

Pare de aceitar novos trabalhos, conclua tarefas em andamento e feche a conexão dedicada.

Testes

Use duas conexões reais: uma executa LISTEN, outra envia NOTIFY dentro de uma transação. Verifique que:

• não há entrega antes do commit;
• rollback não entrega;
• payload chega corretamente;
• reconexão refaz LISTEN;
• polling recupera notificações perdidas;
• múltiplos listeners recebem invalidação;
• jobs são processados uma vez com lock e idempotência.

Erros comuns

• tratar NOTIFY como fila durável;
• usar conexão comum do pool;
• não refazer LISTEN após reconexão;
• enviar payload grande;
• não ter polling de recuperação;
• usar transaction pooling para listener;
• processar job diretamente sem persistência;
• criar canais dinâmicos demais;
• não validar JSON.

Checklist

• conexão dedicada;
• canal controlado;
• payload pequeno;
• estado durável no banco;
• polling de fallback;
• reconexão com jitter;
• LISTEN refeito;
• idempotência aplicada;
• métricas de conexão e atraso;
• shutdown gracioso.

Conclusão

LISTEN e NOTIFY no Node.js fornecem sinalização rápida e simples quando PostgreSQL já faz parte da arquitetura. O recurso funciona muito bem para invalidar caches e acordar workers.

Não dependa dele para durabilidade. Persista jobs e eventos, use NOTIFY como otimização de latência e mantenha polling de recuperação. Com conexão dedicada, reconexão e validação, o mecanismo oferece coordenação leve sem adicionar outro serviço.

Consulte a documentação de LISTEN, a documentação de NOTIFY e a API Client do node-postgres.