O receptor passa a ter uma única responsabilidade síncrona: validar a assinatura, gravar o evento pela chave do provedor e responder. A gravação com ON CONFLICT DO NOTHING já resolve as reentregas duplicadas, e responder sem esperar o ERP elimina o timeout que disparava a maioria das reentregas fora de ordem. O resto do trabalho acontece depois, em uma transação por evento.
-- Caixa de entrada: todo evento e gravado antes de qualquer decisao.
CREATE TABLE webhook_eventos (
evento_id text PRIMARY KEY, -- id do evento no provedor: deduplica reentregas
recurso_id text NOT NULL, -- id do pedido a que o evento se refere
tipo text NOT NULL,
payload jsonb NOT NULL,
recebido_em timestamptz NOT NULL DEFAULT now(),
aplicado_em timestamptz,
resultado text -- aplicado, evento_antigo, regressao_de_terminal...
);
-- A varredura so procura o que ainda nao foi aplicado.
CREATE INDEX webhook_eventos_pendentes
ON webhook_eventos (recebido_em) WHERE aplicado_em IS NULL;
CREATE TABLE pedidos (
id text PRIMARY KEY, -- id do pedido no provedor
status text NOT NULL
CHECK (status IN ('criado', 'pago', 'enviado', 'cancelado', 'reembolsado')),
total_centavos bigint, -- pode chegar depois: o primeiro fato nem sempre traz tudo
cliente_email text,
versao_origem bigint NOT NULL, -- chave de ordenacao do ultimo fato aplicado
atualizado_em timestamptz NOT NULL DEFAULT now()
);
-- Efeitos colaterais nascem da transicao de estado, na mesma transacao.
CREATE TABLE outbox (
pedido_id text NOT NULL,
efeito text NOT NULL,
criado_em timestamptz NOT NULL DEFAULT now(),
enviado_em timestamptz,
PRIMARY KEY (pedido_id, efeito)
);
-- Ponto de partida da reconciliacao periodica.
CREATE TABLE cursores (
nome text PRIMARY KEY,
valor timestamptz NOT NULL
);
INSERT INTO cursores (nome, valor) VALUES ('pedidos_loja', now() - interval '1 day');
A aplicação usa um advisory lock por pedido, obtido com pg_advisory_xact_lock sobre o hash do id. Isso serializa os fatos de um mesmo pedido mesmo quando a linha ainda não existe, situação em que um SELECT FOR UPDATE não teria o que travar e dois eventos simultâneos poderiam ambos concluir que o pedido é novo. O lock dura só até o fim da transação e não bloqueia pedidos diferentes.
// webhooks/aplicar.js - aplica fatos sobre pedidos sem depender da ordem de chegada.
import { pool } from '../db.js';
// Desempate quando dois fatos tem o mesmo instante: o estado mais avancado vence.
export const RANK = { criado: 1, pago: 2, enviado: 3, cancelado: 4, reembolsado: 5 };
// Estados de onde so se sai para os destinos listados. Fato mais novo fora disso e anomalia.
const TERMINAIS = { cancelado: ['reembolsado'], reembolsado: [] };
const STATUS_DO_TIPO = {
'order.created': 'criado',
'order.paid': 'pago',
'order.shipped': 'enviado',
'order.canceled': 'cancelado',
'order.refunded': 'reembolsado',
};
const STATUS_DA_API = {
open: 'criado',
paid: 'pago',
shipped: 'enviado',
canceled: 'cancelado',
refunded: 'reembolsado',
};
// Um provedor, um criterio: sequencia por recurso quando existe; senao, instante do fato em ms
// vezes 10 mais o rank, o que cabe com folga em Number.MAX_SAFE_INTEGER.
export function versaoDoFato({ sequence, occurredAt }, status) {
if (Number.isSafeInteger(sequence)) return sequence;
const ms = Date.parse(occurredAt);
if (Number.isNaN(ms)) throw new Error(`instante invalido: ${occurredAt}`);
return ms * 10 + RANK[status];
}
// Funcao pura: o resultado depende do estado e do fato, nunca de quem chegou primeiro.
export function decidir(atual, status, versao) {
if (!atual) return { aplicar: true, de: null };
if (versao <= Number(atual.versao_origem)) return { aplicar: false, motivo: 'evento_antigo' };
const saidas = TERMINAIS[atual.status];
if (saidas && !saidas.includes(status)) {
return { aplicar: false, motivo: 'regressao_de_terminal' };
}
return { aplicar: true, de: atual.status };
}
// Efeitos saem da transicao real, nao do tipo do evento recebido.
function efeitosDaTransicao(de, para) {
if (para === 'pago') return ['liberar_separacao'];
if (para === 'cancelado' && de === 'pago') return ['interromper_separacao'];
if (para === 'reembolsado') return ['notificar_reembolso'];
return [];
}
// Precisa rodar dentro de uma transacao aberta pelo chamador.
export async function aplicarFato(client, { recursoId, status, versao, dados }) {
if (!status) throw new Error(`status desconhecido para o pedido ${recursoId}`);
// Serializa os fatos do mesmo pedido, inclusive quando a linha ainda nao existe.
await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1, 0))', [recursoId]);
const { rows } = await client.query('SELECT status, versao_origem FROM pedidos WHERE id = $1', [
recursoId,
]);
const decisao = decidir(rows[0], status, versao);
const total = dados.total_cents ?? null;
const email = dados.customer_email ?? null;
if (!decisao.aplicar) {
// Fato antigo nao muda o status, mas ainda completa campos que o fato novo nao trouxe.
await client.query(
`UPDATE pedidos
SET total_centavos = COALESCE(total_centavos, $2),
cliente_email = COALESCE(cliente_email, $3)
WHERE id = $1`,
[recursoId, total, email],
);
return decisao.motivo;
}
await client.query(
`INSERT INTO pedidos (id, status, total_centavos, cliente_email, versao_origem)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE
SET status = EXCLUDED.status,
total_centavos = COALESCE(EXCLUDED.total_centavos, pedidos.total_centavos),
cliente_email = COALESCE(EXCLUDED.cliente_email, pedidos.cliente_email),
versao_origem = EXCLUDED.versao_origem,
atualizado_em = now()`,
[recursoId, status, total, email, versao],
);
for (const efeito of efeitosDaTransicao(decisao.de, status)) {
await client.query(
'INSERT INTO outbox (pedido_id, efeito) VALUES ($1, $2) ON CONFLICT DO NOTHING',
[recursoId, efeito],
);
}
return 'aplicado';
}
export async function aplicarEvento(eventoId) {
const client = await pool.connect();
try {
await client.query('BEGIN');
const { rows } = await client.query(
`SELECT evento_id, recurso_id, tipo, payload
FROM webhook_eventos
WHERE evento_id = $1 AND aplicado_em IS NULL
FOR UPDATE SKIP LOCKED`,
[eventoId],
);
const evento = rows[0];
if (!evento) {
await client.query('ROLLBACK');
return null; // ja aplicado ou sendo aplicado por outro processo
}
const status = STATUS_DO_TIPO[evento.tipo];
const resultado = status
? await aplicarFato(client, {
recursoId: evento.recurso_id,
status,
versao: versaoDoFato(
{ sequence: evento.payload.sequence, occurredAt: evento.payload.occurred_at },
status,
),
dados: evento.payload.data,
})
: 'tipo_ignorado';
await client.query(
'UPDATE webhook_eventos SET aplicado_em = now(), resultado = $2 WHERE evento_id = $1',
[evento.evento_id, resultado],
);
await client.query('COMMIT');
return resultado;
} catch (err) {
await client.query('ROLLBACK');
throw err;
} finally {
client.release();
}
}
// Mesmo formato de fato a partir do pedido lido na API do provedor.
export function fatoDaApi(pedido) {
const status = STATUS_DA_API[pedido.status];
return {
recursoId: pedido.id,
status,
versao: status
? versaoDoFato({ sequence: pedido.sequence, occurredAt: pedido.updated_at }, status)
: null,
dados: pedido,
};
}
Os efeitos colaterais deixaram de depender do tipo do evento. A separação é liberada quando o pedido transiciona para pago, e essa transição só acontece se o fato for mais novo que o estado atual. Por isso o pagamento atrasado, que chega depois do cancelamento, não gera nada. A intenção é gravada na tabela outbox na mesma transação, com chave por pedido e efeito, e um publicador separado envia para o sistema de separação; se a transação falhar, nem o estado nem o efeito existem.
// webhooks/receptor.js - valida, grava, responde e so depois aplica.
// 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';
import { aplicarEvento } from './aplicar.js';
export const router = express.Router();
function assinaturaValida(corpo, assinatura) {
if (typeof assinatura !== 'string') return false;
const esperada = crypto
.createHmac('sha256', process.env.LOJA_WEBHOOK_SECRET)
.update(corpo)
.digest('hex');
const a = Buffer.from(assinatura, 'utf8');
const b = Buffer.from(esperada, 'utf8');
return a.length === b.length && crypto.timingSafeEqual(a, b);
}
router.post('/webhooks/loja', express.raw({ type: 'application/json' }), async (req, res) => {
if (!assinaturaValida(req.body, req.get('x-loja-signature'))) return res.status(401).end();
const evento = JSON.parse(req.body.toString('utf8'));
await pool.query(
`INSERT INTO webhook_eventos (evento_id, recurso_id, tipo, payload)
VALUES ($1, $2, $3, $4)
ON CONFLICT (evento_id) DO NOTHING`,
[evento.id, evento.data.id, evento.type, evento],
);
res.status(200).end();
// Fora da resposta: se falhar aqui, o evento continua gravado e a varredura tenta de novo.
aplicarEvento(evento.id).catch((err) =>
console.error('falha ao aplicar evento', evento.id, err.message),
);
});
// Varredura: reaplica o que ficou pendente por erro, deploy no meio ou queda do processo.
setInterval(async () => {
try {
const { rows } = await pool.query(
`SELECT evento_id
FROM webhook_eventos
WHERE aplicado_em IS NULL
AND recebido_em < now() - interval '30 seconds'
ORDER BY recebido_em
LIMIT 200`,
);
for (const { evento_id: eventoId } of rows) {
await aplicarEvento(eventoId).catch((err) =>
console.error('varredura: falha no evento', eventoId, err.message),
);
}
} catch (err) {
console.error('varredura falhou', err.message);
}
}, 15_000).unref();