O mecanismo central é uma tabela que registra cada efeito externo aplicado, com uma chave que depende só do evento e do tipo do efeito: evento_id mais tipo_efeito. O consumidor ao vivo e o replay usam a mesma função de chave e a mesma tabela, então o replay descobre, efeito por efeito, o que já aconteceu no dia 14 e o que ficou faltando. O schema abaixo tem também o arquivo dos eventos, o estado de comissão com a versão do agregado que o produziu, o checkpoint e o diff da simulação.
-- schema.sql
create table evento_arquivado (
posicao bigint primary key, -- ordem global do log de origem
id text not null unique,
tipo text not null,
pedido_id text not null,
versao bigint not null, -- versao do agregado (o pedido)
payload jsonb not null,
ocorrido_em timestamptz not null
);
create index on evento_arquivado (ocorrido_em, posicao);
create table comissao (
pedido_id text primary key,
valor_centavos bigint not null,
versao_evento bigint not null,
atualizado_em timestamptz not null default now()
);
-- Um registro por efeito externo, com a mesma chave no consumidor ao vivo e no replay
create table efeito_aplicado (
chave text primary key, -- '<evento_id>:<tipo_efeito>'
evento_id text not null,
tipo_efeito text not null,
estado text not null check (estado in ('reservado', 'concluido', 'incerto')),
execucao_id text not null, -- 'ao-vivo' ou o id do reprocessamento
resultado jsonb,
criado_em timestamptz not null default now(),
atualizado_em timestamptz not null default now()
);
create table reprocessamento_checkpoint (
execucao_id text primary key,
ultima_posicao bigint not null,
atualizado_em timestamptz not null default now()
);
create table reprocessamento_diff (
execucao_id text not null,
pedido_id text not null,
antes bigint,
depois bigint not null,
primary key (execucao_id, pedido_id)
);
// comissao.js
export function calcularComissao({ valorCentavos, percentual }) {
if (!Number.isInteger(valorCentavos) || typeof percentual !== 'number') {
throw new Error('payload de pedido.pago invalido');
}
// A versao com bug dividia por 1.000: percentual 5 virava 0,5%
return Math.round((valorCentavos * percentual) / 100);
}
O aplicador abaixo é o mesmo código para o consumidor ao vivo e para o replay; muda só o execucaoId, a política e a flag de simulação. O estado de comissão usa um upsert com guarda de versão: a linha só é sobrescrita se o evento for da mesma versão do pedido ou mais novo. Isso permite que o replay corrija a versão 1 de um pedido sem apagar a versão 2 que o consumidor ao vivo gravou enquanto o replay ainda estava no meio da manhã.
// aplicador.js
import { calcularComissao } from './comissao.js';
// Reserva a chave do efeito. Uma reserva abandonada so pode ser retomada
// quando o provedor deduplica pela chave ($5 = true).
const RESERVAR = `
insert into efeito_aplicado (chave, evento_id, tipo_efeito, estado, execucao_id)
values ($1, $2, $3, 'reservado', $4)
on conflict (chave) do update
set execucao_id = excluded.execucao_id, atualizado_em = now()
where efeito_aplicado.estado = 'reservado'
and $5::boolean
and efeito_aplicado.atualizado_em < now() - interval '5 minutes'
returning chave`;
export function criarAplicador({
pool, // pg.Pool
gateway,
execucaoId, // 'ao-vivo' ou, por exemplo, 'replay-2026-10-14-v2'
politica, // { [tipoEfeito]: { modo: 'aplicar' | 'suprimir', aoFalhar: 'liberar' | 'manter' } }
simular = false,
metricas,
}) {
async function gravarComissao(evento, valor) {
// So sobrescreve se o evento for da mesma versao ou mais novo que o da linha atual
const { rowCount } = await pool.query(
`insert into comissao (pedido_id, valor_centavos, versao_evento)
values ($1, $2, $3)
on conflict (pedido_id) do update
set valor_centavos = excluded.valor_centavos,
versao_evento = excluded.versao_evento,
atualizado_em = now()
where comissao.versao_evento <= excluded.versao_evento`,
[evento.pedidoId, valor, evento.versao],
);
metricas?.estado(rowCount === 1 ? 'gravado' : 'ignorado-versao-mais-nova');
}
async function registrarDiff(evento, valor) {
const { rows } = await pool.query(
'select valor_centavos, versao_evento from comissao where pedido_id = $1',
[evento.pedidoId],
);
const atual = rows[0];
if (atual && Number(atual.versao_evento) > Number(evento.versao)) return;
await pool.query(
`insert into reprocessamento_diff (execucao_id, pedido_id, antes, depois)
values ($1, $2, $3, $4)
on conflict (execucao_id, pedido_id) do update
set antes = excluded.antes, depois = excluded.depois`,
[execucaoId, evento.pedidoId, atual ? atual.valor_centavos : null, valor],
);
}
async function efeitoExterno(evento, tipo, executar) {
const regra = politica[tipo];
const chave = `${evento.id}:${tipo}`;
if (!regra || regra.modo === 'suprimir') {
metricas?.efeito(tipo, 'suprimido');
return;
}
if (simular) {
const { rowCount } = await pool.query(
'select 1 from efeito_aplicado where chave = $1',
[chave],
);
metricas?.efeito(tipo, rowCount ? 'ja-registrado' : 'seria-aplicado');
return;
}
const reserva = await pool.query(RESERVAR, [
chave,
evento.id,
tipo,
execucaoId,
regra.aoFalhar === 'liberar',
]);
if (reserva.rowCount === 0) {
metricas?.efeito(tipo, 'ja-registrado');
return;
}
try {
const resultado = await executar(chave);
await pool.query(
`update efeito_aplicado
set estado = 'concluido', resultado = $2, atualizado_em = now()
where chave = $1`,
[chave, JSON.stringify(resultado ?? null)],
);
metricas?.efeito(tipo, 'aplicado');
} catch (erro) {
if (regra.aoFalhar === 'liberar') {
// O provedor deduplica pela chave: tentar de novo nao duplica o efeito
await pool.query(
"delete from efeito_aplicado where chave = $1 and estado = 'reservado'",
[chave],
);
throw erro;
}
// Sem idempotencia no provedor nao da para saber se a mensagem saiu
await pool.query(
"update efeito_aplicado set estado = 'incerto', atualizado_em = now() where chave = $1",
[chave],
);
metricas?.efeito(tipo, 'incerto');
}
}
async function aplicar(evento) {
if (evento.tipo !== 'pedido.pago') return;
const valor = calcularComissao(evento.payload);
if (simular) await registrarDiff(evento, valor);
else await gravarComissao(evento, valor);
await efeitoExterno(evento, 'notificar-vendedor', (chave) =>
gateway.notificarVendedor(
{ pedidoId: evento.pedidoId, valorCentavos: valor },
{ chaveIdempotencia: chave },
),
);
}
return { aplicar };
}
O efeito externo segue três passos: reservar a chave, chamar o provedor e marcar como concluído. A reserva é um insert com on conflict, então só uma execução consegue reservar a mesma chave, mesmo com o replay e o consumidor ao vivo processando o mesmo evento ao mesmo tempo. A chave também vai para o provedor como chave de idempotência, e é isso que cobre a janela entre a chamada ter funcionado e a marcação de concluído ter sido gravada.
A diferença entre aoFalhar liberar e manter é a decisão mais importante do ledger. Quando o provedor deduplica pela chave, a falha libera a reserva e o replay tenta de novo; o pior caso é o provedor receber a mesma chave duas vezes e ignorar a segunda. Quando o provedor não deduplica, como a maioria das APIs de mensagem, uma falha depois do envio é indistinguível de uma falha antes dele. O estado incerto existe para isso: o efeito sai do fluxo automático e vai para uma fila de revisão, porque reenviar às cegas troca uma mensagem perdida por uma mensagem duplicada. Pelo mesmo motivo, uma reserva abandonada por um processo que morreu só é retomada automaticamente quando o provedor é idempotente; nos outros casos, um job de varredura move a reserva velha para incerto.