Mover a consulta para um worker sem deduplicar só troca o lugar do congestionamento: cada clique vira um job, e a fila enche com cópias do mesmo trabalho. A chave do job deve ser derivada dos parâmetros normalizados, de modo que o mesmo pedido, feito pelo mesmo tenant, sempre caia na mesma linha. Normalizar significa reduzir as variações que não mudam o resultado: datas em formato canônico, listas ordenadas e sem repetição, valores padrão explícitos.
O tenant precisa fazer parte da chave. Uma chave baseada apenas nos filtros faria dois clientes com o mesmo período dividirem o mesmo resultado, o que é um vazamento de dados, não uma otimização. Pelo mesmo motivo, o GET de acompanhamento filtra por id e por tenant: o id sequencial é fácil de adivinhar.
CREATE TABLE relatorio_jobs (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
chave text NOT NULL UNIQUE,
tenant_id bigint NOT NULL,
tipo text NOT NULL,
params jsonb NOT NULL,
status text NOT NULL
CHECK (status IN ('pendente', 'executando', 'pronto', 'falhou')),
tentativas int NOT NULL DEFAULT 0,
erro text,
resultado jsonb,
criado_em timestamptz NOT NULL DEFAULT now(),
iniciado_em timestamptz,
heartbeat_em timestamptz,
concluido_em timestamptz,
expira_em timestamptz
);
-- A fila so olha para os pendentes; o indice parcial fica pequeno mesmo com milhoes de jobs antigos.
CREATE INDEX relatorio_jobs_fila ON relatorio_jobs (criado_em) WHERE status = 'pendente';
-- Usado no limite de concorrencia por tenant e na varredura de jobs orfaos.
CREATE INDEX relatorio_jobs_executando ON relatorio_jobs (tenant_id) WHERE status = 'executando';
O INSERT com ON CONFLICT resolve a corrida sem trava na aplicação. Se dois pedidos idênticos chegam no mesmo milissegundo, a restrição UNIQUE garante que só um cria a linha, e o outro recebe o mesmo job. A cláusula WHERE do DO UPDATE define quando um job existente pode ser reiniciado: apenas se falhou ou se o resultado expirou. Em qualquer outro caso o RETURNING volta vazio, e a API lê o job atual, que pode já estar pronto.
// relatorios/api.js - aceita o pedido, deduplica e responde sem tocar na consulta pesada.
// Express 5: erros lancados em handlers async chegam ao middleware de erro sem wrapper.
import crypto from 'node:crypto';
import express from 'express';
import { pool } from '../db.js';
export const router = express.Router();
const TIPOS = new Set(['vendas', 'comissoes', 'estoque']);
function normalizar(params) {
const inicio = new Date(params.inicio);
const fim = new Date(params.fim);
if (Number.isNaN(inicio.getTime()) || Number.isNaN(fim.getTime()) || inicio > fim) {
throw Object.assign(new Error('periodo invalido'), { status: 400 });
}
return {
inicio: inicio.toISOString().slice(0, 10),
fim: fim.toISOString().slice(0, 10),
agrupar: params.agrupar === 'mes' ? 'mes' : 'dia',
lojas: [...new Set((params.lojas ?? []).map(String))].sort(),
};
}
// O tenant faz parte da chave: sem ele, dois clientes com o mesmo filtro dividiriam o resultado.
function chaveDoPedido(tenantId, tipo, normalizado) {
const texto = JSON.stringify([tenantId, tipo, normalizado]);
return crypto.createHash('sha256').update(texto).digest('hex');
}
router.post('/relatorios/:tipo', async (req, res) => {
const { tipo } = req.params;
if (!TIPOS.has(tipo)) return res.status(404).end();
const tenantId = req.auth.tenantId;
const params = normalizar(req.body);
const chave = chaveDoPedido(tenantId, tipo, params);
// Cria o job ou reaproveita o existente. So reinicia quando falhou ou quando o resultado expirou.
const { rows } = await pool.query(
`INSERT INTO relatorio_jobs (chave, tenant_id, tipo, params, status)
VALUES ($1, $2, $3, $4, 'pendente')
ON CONFLICT (chave) DO UPDATE
SET status = 'pendente', tentativas = 0, erro = NULL, resultado = NULL,
criado_em = now(), concluido_em = NULL, expira_em = NULL
WHERE relatorio_jobs.status = 'falhou'
OR (relatorio_jobs.status = 'pronto' AND relatorio_jobs.expira_em < now())
RETURNING id, status`,
[chave, tenantId, tipo, params],
);
// Sem linha no RETURNING: o job ja existe e esta pendente, executando ou pronto e valido.
const job =
rows[0] ??
(
await pool.query('SELECT id, status, erro, resultado FROM relatorio_jobs WHERE chave = $1', [
chave,
])
).rows[0];
const local = `/relatorios/jobs/${job.id}`;
if (job.status === 'pronto') return res.status(200).location(local).json(job);
res.status(202).location(local).set('Retry-After', '2').json(job);
});
router.get('/relatorios/jobs/:id', async (req, res) => {
const { rows } = await pool.query(
`SELECT id, status, erro, resultado, concluido_em, expira_em
FROM relatorio_jobs
WHERE id = $1 AND tenant_id = $2`,
[req.params.id, req.auth.tenantId],
);
const job = rows[0];
if (!job) return res.status(404).end();
if (job.status === 'pendente' || job.status === 'executando') {
return res.set('Retry-After', '2').json({ id: job.id, status: job.status });
}
res.json(job);
});
No ERP do exemplo, 61% dos pedidos da segunda-feira passaram a ser atendidos por um job já existente, pendente, em execução ou pronto. Gerentes da mesma rede abrem o mesmo painel com o mesmo filtro, e esse padrão é comum em qualquer produto com relatórios padronizados.