Colas invisibles: el truco de PostgreSQL para gestionar tareas

Este segundo artículo sobre “colear sin cola” se centra en una implementación basada en PostgreSQL. Esta solución traslada la gestión de colas desde el sistema de nodos hacia la base de datos, donde cuenta con almacenamiento dedicado para guardar el estado de las colas. El protagonista de este post será pg-boss, una librería de NPM que implementa el sistema de colas gracias a pg y PostgreSQL.

pg-boss

pg-boss es una cola de trabajos construida en Node.js sobre PostgreSQL, diseñada para proporcionar procesamiento en segundo plano y ejecución asíncrona confiable en aplicaciones Node.js.

Cuenta con una API sencilla y ofrece muchas funcionalidades útiles para manejar colas en nuestro sistema.

Para trabajar con pg-boss, es necesario permitirle crear un esquema de base de datos personalizado para gestionar las colas. Como veremos pronto, una simple instalación desde NPM y algunos conocimientos básicos son suficientes para usarlo sin problemas.

Comenzando con pg-boss

Para empezar con pg-boss, instálalo en tu proyecto Node.js.
Ya he creado un proyecto sencillo para que sigas el post; lo puedes encontrar aquí.
El primer paso simple, después de hacer el git clone y npm install, es ejecutar los comandos:

npm install pg pg-boss
npm install -D @types/pg

Esos comandos instalan las dependencias en tu proyecto.
Con esto, ya estás listo para usar pg-boss en tu proyecto.
Como puedes ver, la configuración de este paquete es sencilla y fácil de hacer.

Antes de empezar a programar, necesitamos configurar una base de datos PostgreSQL. Ya la he preparado para ti. Si tienes Docker instalado en tu equipo, escribe:

docker compose up -d

y tu base de datos estará lista para usar.

---

Crear la cola

Primero, necesitas crear una conexión con la base de datos para que pg-boss pueda conectarse a ella. pg-boss simplifica este paso y permite pasar las opciones de conexión.
Crea un archivo en esta ruta:

src/pgBoss.ts

y añade este contenido:

import PgBoss from 'pg-boss';
  
const pgBossInstance = new PgBoss({
  host: process.env.POSTGRES_HOST!,
  port: Number(process.env.POSTGRES_PORT!),
  user: process.env.POSTGRES_USER!,
  password: process.env.POSTGRES_PASSWORD!,
  database: process.env.POSTGRES_DB!,
});pgBossInstance.on('error', console.error)await pgBossInstance.start()export default pgBossInstance;Lenguaje del código: JavaScript (javascript)

Este código crea una instancia de pg-boss y la conecta a la base de datos. Como puedes ver, crear una instancia de pg-boss es fácil.

Ahora es momento de crear la cola.
Comencemos añadiendo el nombre de la cola en el archivo src/common.ts:

export const QUEUE_NAME = 'user-creation-queue';

Es momento de crear tu cola. Crea un nuevo archivo en esta ruta: src/queue.ts, y añade este código:

import pgBossInstance from "pgBoss.js";
import { QUEUE_NAME, UserCreatedTask } from "./common.js";await pgBossInstance.createQueue(QUEUE_NAME, {
  name: QUEUE_NAME,
  retryLimit: 2
});export function enqueueJob(job: UserCreatedTask) {
  return pgBossInstance.send(QUEUE_NAME, job);
}Lenguaje del código: JavaScript (javascript)

Como puedes notar, este código también es fácil de entender.
Primero, importas el nombre de la cola. Luego, usando el método createQueue, creas una cola pg-boss con el nombre y un retryLimit igual a dos; esto significa que si el mismo mensaje falla dos veces, la cola lo descarta. Si quieres, también puedes definir una dead letter queue y encontrar más información aquí.
Por último, se exporta la función enqueueJob para enviar tareas a la cola.

El último paso es enviar datos a la cola, así que ve al archivo src/index.ts y añade estas dos importaciones después de randomUUID:

import logger from 'logger.js';
import { enqueueJob } from 'queue.js';Lenguaje del código: JavaScript (javascript)

y luego reemplaza el TODO con el siguiente código:

const idTask = await enqueueJob(task);
logger.info(task, `Task with id ${idTask} has been pushed`);

Este código permite enviar eventos dentro de la cola.
Probémoslo.
Abre la terminal y escribe npm run start; el resultado debería ser algo como esto:

[19:07:34.036] INFO (23215): Task with id a5a01088-25b9-424c-8a1e-4ec1e678ae96 has been pushed
    id: "8cd828c2-4028-45bb-bea4-6cdaf3627496"
[19:07:34.037] INFO (23215): Task with id 92071cf5-d278-4654-8fda-4076763e62d1 has been pushed
    id: "13e6b091-b85e-48ba-9d2c-a376fc5ada8d"
[19:07:34.038] INFO (23215): Task with id 1567a32b-daa2-43f7-a016-2729f5511f86 has been pushed
    id: "0589cb0e-9423-48dc-ab05-85ac560aa886"
[19:07:34.039] INFO (23215): Task with id 5aecbd45-a6d5-4913-816f-8bb495792569 has been pushed
    id: "8b0218d1-22eb-4869-9bf8-7d9bc42b84a0"
[19:07:34.040] INFO (23215): Task with id adc92a04-ff85-49a6-9539-ea993b77a31c has been pushed
    id: "1269f0d5-bbcd-49fd-89d5-7102be8237be"Lenguaje del código: JavaScript (javascript)

¡Genial! Acabas de completar la primera parte, y ahora sabes cómo enviar datos a una cola pg-boss.
Pasemos al procesamiento.

---

Procesando mensajes con pg-boss

Para procesar los mensajes, ya he preparado un archivo src/worker.js, donde escribirás el código.
Abre el archivo src/worker.js y añade las siguientes importaciones:

import { setTimeout } from "timers/promises";
import { QUEUE_NAME, UserCreatedTask } from "./common.js";
import pgBossInstance from "./pgBoss.js";Lenguaje del código: JavaScript (javascript)

y reemplaza el TODO con el siguiente código:

pgBossInstance.work<UserCreatedTask>(QUEUE_NAME, async ([job]) => {
  if (!messagesHandled[job.id]) {
    messagesHandled[job.id] = { retries: 0, result: 'created' }; 
 } else {
    messagesHandled[job.id].retries++;
 }
  const resultType = Math.random() > 0.6;
  const fakeImplementation = resultType ? 'success' : 'error'
  const timeout = fakeImplementation === 'success' ? 2000 : 1000;
  await setTimeout(timeout);
  if (fakeImplementation === 'error') {
    messagesHandled[job.id].result = 'error';
    printMessagesHandled();
    const message = `User created task got error with id: ${job.id}`
    throw new Error(message);
 } else {
    messagesHandled[job.id].result = 'success';
    printMessagesHandled();
 }
});Lenguaje del código: JavaScript (javascript)

En este código estás simulando el procesamiento de mensajes. Primero, gestionas un mapa con todos los mensajes procesados para imprimir el resultado en la consola. Luego, usando Math.random(), determinas si el mensaje falla o tiene éxito y esperas dos o un segundo según el resultado. Por último, si el mensaje falla, lanzas un error; de lo contrario, imprimes el resultado en la consola (solo para mostrar el resultado de este ejemplo).

Como puedes entender, si la función lanza un error, pg-boss marcará el mensaje como fallido; de lo contrario, se interpretará como completado.

Ejecutemos el código con npm run start.
El resultado final debería verse así:

Unique messages handled: 5
940c9d32-3b83-4034-ba81-3b2d88035305: success (2 retries)
a19a224f-20c1-4481-b54b-81d73b81c691: success (1 retries)
701f1a33-bde2-4013-9bd3-d4a4e0bedcf7: success (2 retries)
d00b4a03-2e11-4f05-b2c3-8e60f2efe4f6: success (1 retries)
6d660781-0fdf-4414-8bc6-d791be0588dc: success (0 retries)

Como puedes ver, el sistema ha procesado cinco mensajes en este caso. Algunos tuvieron reintentos, otros no.
¡Perfecto! Ahora también aprendiste cómo procesar mensajes con pg-boss.

---

Conclusión

¡Es hora de cerrar este post!

Como puedes ver, trabajar con pg-boss es sencillo; la API es simple y el código se mantiene simple.
Este enfoque puede ser una excelente solución si no tienes un sistema de colas específico en tu arquitectura, pero debes prestar atención a la carga que puede soportar la base de datos para manejar tus colas; por eso, en el siguiente artículo veremos otra solución: la solución que mi equipo eligió para equilibrar rendimiento y simplicidad.

Bueno, eso es todo por pg-boss. Espero que hayas disfrutado del artículo y nos vemos en el próximo.

¡Adiós! 👋

p.d. Puedes encontrar el código completo del artículo aquí.