Documentação do Broker v1

O que é o broker

O broker é um serviço de integração assíncrona baseada em fila: eventos são enfileirados no Postgres (via pg-boss, sem Redis/RabbitMQ/Kafka) e processados por um ou mais processos worker, que chamam uma API externa por HTTP, com retry/backoff nativos, circuit breaker e rate limiting próprios.

Desde a versão atual, o broker suporta múltiplos endpoints externos ao mesmo tempo — cada integração tem sua própria fila, credencial, circuit breaker e rate limiter, isolados uns dos outros. Um dashboard web dá visão em tempo real (SSE) de todas as filas, permite criar/editar endpoints, escalar workers e atribuir filas específicas a workers específicos.

Princípios de arquitetura

• Postgres como única fonte de verdade. Sem infraestrutura extra — a fila, o estado de circuit breaker, a configuração de endpoints e os logs estruturados moram todos no mesmo Postgres.
• Retry/backoff 100% nativos do pg-boss. O código da aplicação nunca reimplementa retry — só classifica erros (permanente vs. transitório) para decidir o que conta pro circuit breaker.
• Isolamento por endpoint. Cada integração externa tem fila, client HTTP, circuit breaker e rate limiter próprios — uma API externa degradada nunca consome a capacidade de processamento de outra.
• Live-reconfig sem restart. Endpoints, parâmetros, atribuição de filas por worker — tudo é lido de tabelas do Postgres e aplicado nos próximos ciclos de poll do worker, sem precisar reiniciar nada.
• Observabilidade em primeiro lugar. Todo estado relevante (fila, circuit breaker, workers vivos) é visível no dashboard em tempo real via Server-Sent Events.

Arquitetura de ambiente

Em desenvolvimento/produção local, tudo roda via docker-compose, numa rede privada própria (broker_default), com um Postgres dedicado (não compartilhado com outras aplicações).

Serviços do docker-compose.yml

Subindo o ambiente completo

# stack de producao/uso normal (postgres + worker + dashboard)
docker-compose up -d --build

# + laboratorio de testes (mock-api)
docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d --build

# escalando workers horizontalmente
docker-compose up -d --scale worker=3

Arquitetura de componentes

Fluxo de dados entre os componentes lógicos da aplicação (independente de estarem em containers separados ou não):

Worker por dentro (src/worker/)

Dashboard por dentro (src/dashboard/)

• server.js — servidor HTTP com todas as rotas /api/*, SSE (/api/events), e serve os arquivos estáticos de public/ (incluindo esta página).
• queries.js — todas as consultas SQL de leitura em pgboss.job e tabelas auxiliares.
• public/index.html — SPA de uma página só (sem build step, sem framework), com todas as abas.

Tabelas do Postgres

Múltiplos endpoints

O broker integra com N endpoints externos ao mesmo tempo, cada um com sua própria fila pg-boss (nomeada ${QUEUE_NAME}.<endpoint-id>), credencial, circuit breaker, rate limiter e política de retry.

Limitação conhecida: tipos de credencial

O client HTTP manda um header estático por chamada — cobre API key, Bearer token e HTTP Basic (pré-codificado). Não cobre esquemas que exigem assinar o corpo por requisição (ex: HMAC) sem uma extensão pontual do client.

Escalonamento e workers

Duas dimensões de escala

1. Vertical — aumentar concurrency de um endpoint (ao vivo, aba Admin).
2. Horizontal — múltiplas réplicas do processo worker (docker-compose --scale worker=N, scripts/run-local.sh worker=N, ou o botão + na aba Arquitetura).

Dedicar um worker a filas específicas

Por padrão todo worker consome todas as filas habilitadas. Pelo ícone de lápis (✎) no card de cada instância (aba Arquitetura), dá para desmarcar "Todas as filas" e escolher um subconjunto — o worker deixa de consumir as demais imediatamente (a atribuição é lida a cada ciclo de poll, sem restart), sem afetar circuit breaker/rate limiter das filas onde ele continua atuando, nem o consumo de outros workers.

A tabela "Todos os Endpoints" mostra quantos workers estão atribuídos a cada fila, com aviso se for 0 (fila parada) ou >1 (rate limit/circuit breaker multiplicado).

Instalação

Pré-requisitos

• Docker + Docker Compose (recomendado) ou Node.js 20+ para rodar nativo.
• Nenhum outro serviço já usando as portas escolhidas (por padrão 5544, 8080, 8090 — ajustável no .env).

Passo a passo (Docker — recomendado)

git clone <repo> broker && cd broker
cp .env.example .env
# edite .env: senha do Postgres, credenciais de cada endpoint, etc.

docker-compose up -d --build
# abre o dashboard em http://localhost:8080

Rodando nativo (sem Docker)

npm install
cp .env.example .env
cp .env.example .env.local   # ajuste DATABASE_URL/EXTERNAL_API_URL para "localhost"

scripts/run-local.sh              # 1 worker + dashboard
scripts/run-local.sh worker=3     # 3 workers nativos
scripts/stop-local.sh             # para tudo (nativo E docker-compose, se estiver rodando)

Criação do banco de dados

Não existe um script de migração para rodar manualmente — o schema é self-provisioning. Tanto o schema do pg-boss quanto as tabelas próprias da aplicação são criados sozinhos, automaticamente, na primeira vez que um processo sobe:

• Schema pgboss (fila, agendamentos, versões) — criado pelo próprio pg-boss dentro de boss.start(), chamado no boot do worker, do dashboard e do console de controle. Não é código da aplicação.
• Tabelas app_* (endpoints, config, workers, circuit breaker, logs) — cada módulo chama seu próprio ensure<X>Table() no boot, usando o helper createTableIfNotExists() de src/config/db.js (CREATE TABLE IF NOT EXISTS, com proteção contra a corrida de duas sessões criando a mesma tabela ao mesmo tempo).

Ou seja: o "processo de criar o banco" é simplesmente subir a aplicação pela primeira vez — não existe npm run migrate nem SQL para rodar à mão. O que precisa existir antes disso é só a role e o database vazios:

POSTGRES_USER=broker
POSTGRES_PASSWORD=change-me
POSTGRES_DB=broker_queue

psql -U postgres -c "CREATE ROLE broker WITH LOGIN PASSWORD 'change-me';"
psql -U postgres -c "CREATE DATABASE broker_queue OWNER broker;"

Verificando a instalação

1. docker-compose ps — todos os serviços Up e postgres healthy.
2. Abrir http://localhost:8080 — deve carregar a tela "Todos os Endpoints".
3. Popular jobs de teste: docker-compose exec worker node scripts/seed.js 10.
4. Confirmar no dashboard que os jobs aparecem e completam.

Manual de configuração

Duas camadas de configuração: variáveis de ambiente (infra/boot) e configuração dinâmica no Postgres, editável ao vivo pelo dashboard sem reiniciar nada.

Endpoints (por integração externa)

Cada endpoint é um registro em app_endpoints, gerenciado 100% pelo dashboard (aba Parâmetros) ou pela API:

GET    /api/endpoints
POST   /api/endpoints            { id, label, baseUrl, authHeaderName, authHeaderValue, ... }
PUT    /api/endpoints/:id        (parcial - so os campos enviados sao atualizados)
DELETE /api/endpoints/:id        (remove so a config; nao apaga jobs ja enfileirados)

Parâmetros globais

Editáveis na aba Parâmetros, aplicados ao vivo (releitura a cada workerConfigPollMs):

Variáveis de ambiente (.env)

Lista completa e comentada em .env.example. Principais grupos:

• Postgres: POSTGRES_USER/PASSWORD/DB/PORT, DATABASE_URL
• pg-boss: QUEUE_NAME, PGBOSS_EXPIRE_IN_SECONDS, PGBOSS_MAINTENANCE_INTERVAL_SECONDS
• Dashboard: DASHBOARD_PORT
• Worker: WORKER_CONTROL_PORT, WORKER_CONTROL_URL
• Laboratório de testes (dev): MOCK_API_*

Como usar: enviando dados para a fila

Existem duas formas de enfileirar um evento. Para qualquer sistema externo (de qualquer linguagem), use a API HTTP de ingestão — é o caminho recomendado. Se o produtor já é um processo Node.js rodando dentro da mesma infraestrutura do broker, também é possível usar o cliente pg-boss diretamente (biblioteca), sem passar pela rede.

API HTTP de ingestão

POST /api/ingest/:endpointId, servida pelo mesmo processo do dashboard. O corpo da requisição é exatamente o JSON que o worker vai enviar (via POST) para o baseUrl configurado nesse endpoint — o broker não adiciona nem remove campos.

Passo a passo

1. Cadastre o endpoint de destino (aba Admin ou POST /api/endpoints) — é o id desse registro que vira :endpointId na URL.
2. Na mesma aba, clique em Gerar chave na coluna "Ingestão HTTP" desse endpoint. A chave aparece uma única vez — copie na hora.
3. Chame a API mandando essa chave no header X-Ingest-Key.

Contrato

Exemplo

curl -X POST https://seu-dominio/api/ingest/orders \
  -H "Content-Type: application/json" \
  -H "X-Ingest-Key: <chave gerada na aba Parametros>" \
  -d '{"orderId": "A-1029", "status": "paid", "total": 149.9}'

# 202 Accepted
# {"jobId":"9f2c...","queueName":"external-api-events.orders"}

Funciona de qualquer linguagem — só precisa fazer um POST com os dois headers. Em Python, por exemplo:

import requests

requests.post(
    "https://seu-dominio/api/ingest/orders",
    headers={"X-Ingest-Key": "..."},
    json={"orderId": "A-1029", "status": "paid", "total": 149.9},
)

Alternativa avançada: biblioteca pg-boss direto (só Node.js)

Se o produtor já é Node.js e prefere não depender de rede/HTTP (ex: outro processo dentro da mesma infraestrutura, com acesso direto ao Postgres do broker), dá para inserir o job direto na fila com o cliente pg-boss — exatamente como scripts/seed.js e scripts/console.js já fazem. Essa via não passa pela autenticação/limite de tamanho da API HTTP (é o mesmo nível de confiança que o próprio worker).

import pg from 'pg';
import { env } from './src/config/env.js';
import { createBoss } from './src/config/pgboss.js';
import { queueNameFor, listEndpoints, buildEndpointSendOptions } from './src/config/endpoints.js';

const ENDPOINT_ID = 'orders'; // id cadastrado na aba Parametros

const pool = new pg.Pool({ connectionString: env.databaseUrl });
const [endpoint] = await listEndpoints(pool, { onlyEnabled: true })
  .then((rows) => rows.filter((e) => e.id === ENDPOINT_ID));
await pool.end();
if (!endpoint) throw new Error(`endpoint "${ENDPOINT_ID}" nao encontrado ou desativado`);

const boss = createBoss();
await boss.start();

const jobId = await boss.send(
  queueNameFor(endpoint.id),
  { orderId: 'A-1029', status: 'paid', total: 149.9 },
  buildEndpointSendOptions(endpoint), // aplica o retryLimit/retryDelay/backoff DESSE endpoint
);

console.log(`job ${jobId} enfileirado na fila ${queueNameFor(endpoint.id)}`);
await boss.stop();

Testando sem uma API real

Para popular filas com dados fake e validar o fluxo de ponta a ponta antes de conectar um produtor de verdade, use o script pronto (ver Laboratório de testes):

# popula TODOS os endpoints habilitados com 20 jobs cada
node scripts/seed.js 20

# popula so um endpoint especifico
node scripts/seed.js 20 orders

Laboratório de testes (mock-api)

Serviço à parte, só Node core, com 5 endpoints — um por tipo de credencial — que sorteiam sucesso/erro para exercitar circuit breaker, retry e rate limit sem depender de internet.

⚠️ Nunca sobe em produção — só existe no overlay docker-compose.dev.yml.

Guia do dashboard

Tour pelas abas do dashboard, com screenshots reais da instância local. O ícone de sol/lua no cabeçalho alterna entre tema claro e escuro (preferência salva no navegador, detecta o tema do sistema operacional na primeira visita) — as capturas abaixo estão no tema escuro (padrão), mas toda a interface tem equivalente claro.

Todos os Endpoints (tela inicial)

Visão agregada de todas as integrações lado a lado: diagrama (Worker → cada endpoint), totais consolidados, e uma tabela com circuit breaker, contagem por estado, volume e latência de cada fila — inclusive quantos workers estão atribuídos a cada uma.

Overview (por endpoint)

Escopada pelo seletor de endpoint no topo: contagem por estado, circuit breaker, backlog, taxa de erro, throughput com seletor de período, gráficos de completados x falhados e criados x resolvidos, distribuição de retries, diagnóstico automático de gargalo (fila vs. API externa vs. rate limit).

Jobs

Grid paginada com todos os jobs de qualquer estado. Filtro de Fila independente do seletor do topo (endpoint selecionado, todas as filas juntas, ou uma específica), filtro por estado e busca no conteúdo. Jobs failed podem ser reprocessados; created/retry podem ser editados antes de rodar.

Tabelas

Browser genérico das tabelas internas do pg-boss e das tabelas próprias da aplicação. Tabelas com identidade única têm checkbox de seleção e os botões Excluir selecionados (protegido — nunca apaga jobs pendentes/em processamento) e Limpar tabela inteira (sem exceção).

Parâmetros

Edição em tempo real dos parâmetros globais (alertas, refresh do dashboard) e a seção de manutenção para limpar dados de teste. Gestão de endpoints, usuários e o envio de carga de teste ficaram numa aba própria — ver Admin — visível só para administradores.

Admin só administradores

Aba inteira invisível para o perfil user (nem aparece na barra de abas) — ver Autenticação e controle de acesso. Reúne tudo que é gestão de acesso/provisionamento, separado da aba Parâmetros (que ficou só com ajuste fino):

• Endpoints — o mesmo CRUD que existia em Parâmetros (criar, editar, ativar/desativar, excluir, gerar/regenerar chave de ingestão).
• Usuários do dashboard — criar/excluir usuários e definir o perfil (admin/user).
• Enviar carga de teste — checklist com todos os endpoints cadastrados (nome + status ativo/desativado, seleção múltipla, atalhos "todos"/ "nenhum"), um campo de quantidade (padrão 10, editável), botão Enviar, e ao lado um contador ao vivo ("Na fila agora") somando os jobs pendentes (created + retry) só dos endpoints marcados — atualizado via SSE, sem polling próprio.

Equivale a POST /api/admin/seed com { count, endpointIds } — envia pra cada endpoint marcado, e devolve separadamente o que foi enfileirado (seeded) e o que foi ignorado (skipped, com o motivo: endpoint_disabled ou endpoint_not_found) — selecionar um endpoint desativado não derruba o envio dos demais.

Arquitetura (ao vivo)

Diagrama animado: Dashboard → Postgres → Worker(s) → API Externa, com o estado real do circuit breaker, slots de concorrência ocupados, e a lista de todas as instâncias de worker vivas — cada uma com suas filas atribuídas e os botões + (subir novo worker) e − (encerrar uma instância específica).

Autenticação e controle de acesso

O dashboard inteiro fica atrás de login. Dois perfis:

Usuário master

Existe um usuário "master" sempre disponível, independente do que estiver cadastrado em app_users — vem de variáveis de ambiente e nunca é gravado como linha comum na tabela:

ADMIN_USERNAME=julio      # default se omitido do .env
ADMIN_PASSWORD=102938     # default se omitido - TROQUE em producao

Sessão e senha

• Sessão em Postgres (tabela app_sessions, cookie HttpOnly, 24h) — não é um cookie assinado sem estado. Isso importa porque permite revogar na hora: quando um admin exclui um usuário, a sessão dele morre imediatamente, não só quando o cookie expira sozinho.
• Senha com crypto.scryptSync nativo do Node (memory-hard, resistente a GPU/ASIC) — sem trazer bcrypt/argon2 como dependência nova só por causa disso. Comparação sempre via timingSafeEqual.

Gerenciar usuários

Aba Admin → "Usuários do dashboard", ou via API:

GET    /api/users                       (admin)
POST   /api/users        { username, password, role }   (admin)
DELETE /api/users/:id                                     (admin - revoga a sessao na hora)

Tokens de ingestão continuam por-endpoint

Decisão deliberada: não criamos um segundo mecanismo de token por-usuário. Cada endpoint já é uma fila isolada — "1 token = 1 endpoint" já dá o isolamento certo (comprometer uma chave expõe só aquela fila, não várias). O que mudou com o login: gerar/regenerar a X-Ingest-Key agora exige sessão de admin, e fica registrado quem gerou (coluna ingest_key_created_by, visível na aba Admin) — auditoria sem duplicar o mecanismo de autenticação. Ver API HTTP de ingestão.

Swagger (admin)

Link no menu Ajuda (só aparece logado como admin) → /admin/swagger.html — Swagger UI carregada via CDN, documentando só a API de ingestão (POST /api/ingest/:endpointId e o endpoint de regeneração de chave). É o único contrato que um sistema externo de fato consome; o resto do dashboard é operação interna do próprio time, não vale a pena formalizar em OpenAPI. O spec (GET /api/openapi.json) também exige sessão de admin — a página HTML em si carrega pra qualquer um, mas fica vazia/nega acesso sem um fetch autenticado bem-sucedido.

Botão de teste (seed)

Aba Admin → "Enviar carga de teste" — checklist de endpoints (seleção múltipla, mostra nome + status ativo/desativado), campo de quantidade (padrão 10) e um contador ao vivo de jobs pendentes nos endpoints marcados. Mesma lógica de scripts/seed.js, mas roda dentro do processo do dashboard (reaproveitando o sendBoss já existente ali para reprocessamento), sem subir um subprocesso. Equivale a POST /api/admin/seed (admin-only), que aceita { count, endpointIds } no corpo — sem eles, usa 10 jobs em todos os endpoints habilitados; endpoints desativados ou inexistentes na lista voltam em skipped em vez de derrubar a chamada inteira. Ver tela Admin.

Segurança e boas práticas

• Credenciais nunca são reexibidas. authHeaderValue é write-only — a API só retorna {configured: true/false}.
• Postgres dedicado. Não compartilhar a instância com outras aplicações.
• Sem socket do Docker no dashboard. O botão "+" de subir worker roda um processo filho dentro do próprio container do dashboard, deliberadamente — não monta o socket do Docker, que daria ao dashboard controle total sobre containers do host.
• Exclusão de tabelas é protegida por padrão. pgboss.job nunca é apagado em estado pendente/ativo pela grid, exceto na ação explícita "Limpar tabela inteira".
• mock-api nunca em produção — trava por NODE_ENV e por estar isolado num overlay de compose separado.

Sobre esta documentação

Esta página (src/dashboard/public/docs.html) é servida pelo próprio dashboard — acessível pelo menu Ajuda → Documentação no cabeçalho. Ela é parte do código do projeto, não um documento à parte: toda vez que a arquitetura, as telas ou o processo de configuração mudarem, esta página deve ser atualizada na mesma alteração (mesmo princípio já seguido para o README.md).

Os screenshots em docs-assets/ foram capturados via Chrome headless (--headless --screenshot) direto da instância local rodando — para atualizá-los depois de uma mudança visual relevante:

"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \
  --headless --disable-gpu --window-size=1400,1000 \
  --screenshot=src/dashboard/public/docs-assets/overview.png \
  "http://localhost:8080/#overview"

Trocando #overview pela aba desejada (#all-endpoints, #jobs, #tables, #params, #architecture) — o hash na URL já aciona a troca de aba automaticamente (mesmo mecanismo usado por link direto no dashboard).