Usando postgres como message broker
Tags: postgres, node, mensageria, webdev
Published: 2023-08-13 19:23:52

Em muitos cenários de desenvolvimento, a necessidade de um _message broker_ é inegável. No entanto, adotar soluções como **Kafka** ou **RabbitMQ** nem sempre é a opção mais viável ou desejável. Neste tutorial, vamos explorar uma abordagem alternativa e interessante: transformar uma tabela comum do **Postgres** em um _message broker_ robusto, garantindo a integridade e consistência das mensagens processadas.

O que é um _message broker_? Trata-se de um sistema que gerencia e armazena uma fila de mensagens, sendo crucial para a comunicação assíncrona entre diferentes partes de um sistema. No nosso caso, o objetivo é criar um sistema onde diferentes sistemas e threads possam consumir mensagens da fila, assegurando que cada mensagem seja processada apenas uma vez, evitando corrupções ou duplicações indesejadas.

##Preparando o Terreno
A ideia central é simples: um _producer_ envia ordens de pagamento para uma tabela que atua como fila, com status inicial "_PENDING_". Em seguida, três clusters de uma aplicação Node executam cron jobs a cada 1 minuto, desempenhando o papel de _consumers_. Cada job verifica as ordens de pagamento pendentes na tabela, processa o pagamento e atualiza o status para "_COMPLETED_". O desafio é garantir que uma ordem seja processada apenas uma vez, mesmo com múltiplos consumidores operando simultaneamente.

O objetivo deste artigo não é ensinar como criar uma aplicação com Node do zero, mas apenas mostrar como utilizar o **Postgres** como _message broker_ numa API feita como Node, por isso iremos pular alguns passos de setup básico.

Também, para fins didáticos, usaremos [NestJS](https://docs.nestjs.com/) como framework para a API, [pm2](https://github.com/Unitech/pm2) para o gerenciamento dos clusters da aplicação e [prisma](https://docs.nestjs.com/recipes/prisma#set-up-prisma) para modelagem e como camada de comunicação com nosso banco de dados.

Você pode obter o projeto-base no repositório [anti-duhring/postgres-queue-broker](https://github.com/anti-duhring/postgres-queue-broker).

Primeiro de tudo é necessário iniciar um novo projeto Nest:
```
nest new postgres-queue-broker
```
Após isso, iremos instalar os pacotes `pm2`, `@nestjs/schedule` para gerenciamento dos cron jobs e o `@prisma/client`:
```
npm i pm2 @nestjs/schedule @prisma/client
```
Esse será o schema da nossa tabela **Order**, que conterá as ordens de pagamentos:
```
// This is your Prisma schema file,
// learn more about it in the docs: https://pris.ly/d/prisma-schema

generator client {
  provider = "prisma-client-js"
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model Order {
  id        String    @id             @default(uuid())
  message   String
  status    String
  createdAt DateTime  @default(now()) @map("created_at")
  updatedAt DateTime  @updatedAt      @map("updated_at")
}
```

Após isso, iremos criar um módulo para ser o nosso _Producer_, ele será responsável por criar uma nova ordem de pagamentos e enviar para a nossa tabela **Order**.
```
nest g resource order
```
**order.service.ts**:
```typescript
import { Injectable } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service'
import { orderStatus } from '../../common/orderStatus.enum'

@Injectable()
export class OrderService {
    constructor(private readonly prisma: PrismaService) {}

    async create(message: string) {
        const order = await this.prisma.order.create({
            data: {
                message,
                status: orderStatus.PENDING
            }
        })

        return order
    }
}
```

##Trabalhando com Locks Inteligentes
Para alcançar esse objetivo, iremos explorar as funcionalidades do **Postgres**: as cláusulas [`FOR UPDATE`](https://www.postgresql.org/docs/9.0/sql-select.html#SQL-FOR-UPDATE-SHARE) e [`SKIP LOCKED`](https://www.postgresql.org/docs/current/sql-select.html). O `FOR UPDATE` bloqueia as linhas retornadas por uma consulta, garantindo que nenhuma outra transação possa modificá-las até que a transação atual seja concluída. Por sua vez, o `SKIP LOCKED` permite que uma consulta ignore linhas que estejam bloqueadas, possibilitando o salto sobre linhas já bloqueadas por outros consumidores.

Sabendo isto, iremos criar o módulo do nosso _consumer_, que consistirá num _cron job__ que roda a cada 1min, obtendo as ordens de pagamento pendentes, processando e atualizando seu status:
```
nest g resource consumer
```
**consumer.service.ts**
```typescript
import { Injectable } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service';
import { orderStatus } from '../../common/orderStatus.enum';
import { Cron } from '@nestjs/schedule';

@Injectable()
export class ConsumerService {
  constructor(private readonly prisma: PrismaService) {}

  async getPendingOrders() {
    const pendingOrders = await this.prisma.$queryRaw`
            	select o.*
                from "Order" o 
                where o.status = 'PENDING'
                order by o.created_at asc 
                for update skip locked
        `;

    return pendingOrders as any[];
  }

  @Cron('0 * * * * *')
  async processOrders() {
    const pendingOrders = await this.getPendingOrders();

    for await (const order of pendingOrders) {
      await this.prisma.order.update({
        where: {
          id: order.id,
        },
        data: {
          status: orderStatus.COMPLETED,
        },
      });
      console.log(`Order ${order.id} has been processed`);
    }
  }
}
```

Esta consulta permite que cada _job consumer_ obtenha uma ordem de pagamento pendente e a bloqueie, impedindo que outros _jobs_ acessem a mesma ordem simultaneamente. Uma vez que a ordem é processada e o status é atualizado, ela é liberada para futuros consumidores.

Importante mencionar que, num cenário real, é necessário ter cautela com a quantidade de itens carregados, por isso um `LIMIT` para limitar a quantidade de itens obtidos por cada _job_ pode evitar gargalos e erros de _out of memory_.

##Rodando nossa aplicação
Por fim, basta realizar a build da aplicação.
```
npm run build
```
E executar os 3 clusters com o pm2:
```
pm2 start dist/main.js -i 3
```

É possível monitorar os logs dos clusters com seguinte comando:
```
pm2 logs main
```
Dessa forma podemos ver que cada ordem de pagamento é processada por apenas um cluster:

![Log mostrando que cada ordem de pagamento foi processada por um cluster diferente](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/s3ar6pahiwmvdxa7lkl3.png)



##Conclusão
O uso do **Postgres** como um _message broker_ alternativo pode ser uma solução valiosa em cenários onde a complexidade do **Kafka** ou do **RabbitMQ** não é justificada. Aproveitar as funcionalidades nativas do **Postgres**, como `FOR UPDATE` e `SKIP LOCKED`, possibilita a criação de um sistema robusto que garante a integridade e consistência das mensagens processadas.

Ao explorar essa abordagem, você poderá aprimorar a comunicação assíncrona em seus sistemas, evitando problemas de duplicação e corrupção de mensagens. Dessa forma, o **Postgres** se mostra não apenas como um excelente banco de dados, mas também como um aliado poderoso na construção de soluções de integração confiáveis. Experimente essa abordagem em seus projetos e desfrute dos benefícios de um _message broker_ sólido e eficiente.

##Para saber mais
Esse artigo foi fortemente inspirado pela palestra do **Rafael Ponte** no canal da **ZUP**, fica aqui a recomendação para entender a implementação com mais detalhes:
https://www.youtube.com/watch?v=FF6Am0N6eq4&t=3275s