Receptor de webhook

Um receptor que cumpre as regras do Webhook de venda:

1. Confere a assinatura sobre o corpo cru e responde 401 se não conferir.
2. Grava o evento numa tabela com o id como chave única. Um id repetido responde 2xx sem efeito novo.
3. Responde 200 na hora; o processamento roda depois, fora da requisição.
4. No processamento, descarta o evento cujo sequence for menor que o último aplicado para a mesma venda.

A tabela de entrada pode ser assim:

create table etech_evento (
  id          uuid primary key,          -- o id do evento: deduplicação
  sale_id     bigint,                    -- data.saleId; nulo no integration.ping
  sequence    bigint not null,
  tipo        text   not null,
  corpo       jsonb  not null,
  recebido_em timestamptz not null default now(),
  processado  boolean not null default false
);

create table etech_venda_aplicada (
  sale_id       bigint primary key,
  ultimo_sequence bigint not null
);

C#

app.MapPost("/webhooks/etech", async (HttpRequest request, EventoRepositorio eventos, IConfiguration config) =>
{
    using var leitor = new StreamReader(request.Body, Encoding.UTF8);
    var corpoCru = await leitor.ReadToEndAsync();

    var valida = AssinaturaWebhook.Valida(
        config["Etech:WebhookSecret"]!,
        request.Headers["X-Etech-Timestamp"],
        corpoCru,
        request.Headers["X-Etech-Signature"],
        DateTimeOffset.UtcNow);
    if (!valida)
        return Results.Unauthorized();

    using var json = JsonDocument.Parse(corpoCru);
    var raiz = json.RootElement;
    long? saleId = raiz.GetProperty("data").TryGetProperty("saleId", out var s) ? s.GetInt64() : null;

    // INSERT ... ON CONFLICT (id) DO NOTHING: o repetido não muda nada e também recebe 200.
    await eventos.GravarSeNovoAsync(
        raiz.GetProperty("id").GetGuid(),
        saleId,
        raiz.GetProperty("sequence").GetInt64(),
        raiz.GetProperty("type").GetString()!,
        corpoCru);

    return Results.Ok();
});

No processamento em segundo plano:

foreach (var evento in await eventos.PendentesAsync())
{
    if (evento.SaleId is { } saleId)
    {
        var ultimo = await eventos.UltimoSequenceAplicadoAsync(saleId);
        if (ultimo is not null && evento.Sequence < ultimo)
        {
            await eventos.MarcarProcessadoAsync(evento.Id); // evento mais velho que o já aplicado
            continue;
        }

        await AplicarAsync(evento);                         // emitir NFC-e, debitar saldo...
        await eventos.GuardarSequenceAsync(saleId, evento.Sequence);
    }

    await eventos.MarcarProcessadoAsync(evento.Id);
}

Node.js

import express from 'express';
import { assinaturaValida } from './assinatura.mjs';
import { db } from './db.mjs';

const app = express();

app.post('/webhooks/etech', express.raw({ type: 'application/json', limit: '1mb' }), async (req, res) => {
  const valida = assinaturaValida(
    process.env.ETECH_WEBHOOK_SECRET,
    req.get('X-Etech-Timestamp'),
    req.body,
    req.get('X-Etech-Signature'),
  );
  if (!valida) return res.sendStatus(401);

  const evento = JSON.parse(req.body.toString('utf8'));

  // O repetido não muda nada e também recebe 200.
  await db.query(
    `insert into etech_evento (id, sale_id, sequence, tipo, corpo)
     values ($1, $2, $3, $4, $5)
     on conflict (id) do nothing`,
    [evento.id, evento.data.saleId ?? null, evento.sequence, evento.type, evento],
  );

  res.sendStatus(200);
});

No processamento em segundo plano:

for (const evento of await pendentes()) {
  const saleId = evento.sale_id;
  if (saleId !== null) {
    const ultimo = await ultimoSequenceAplicado(saleId);
    if (ultimo !== null && evento.sequence < ultimo) {
      await marcarProcessado(evento.id); // evento mais velho que o já aplicado
      continue;
    }
    await aplicar(evento.corpo);         // emitir NFC-e, debitar saldo...
    await guardarSequence(saleId, evento.sequence);
  }
  await marcarProcessado(evento.id);
}

Python

import json
import os

from flask import Flask, request

from assinatura import assinatura_valida
from db import conexao

app = Flask(__name__)


@app.post("/webhooks/etech")
def receber_evento():
    corpo_cru = request.get_data()
    if not assinatura_valida(
        os.environ["ETECH_WEBHOOK_SECRET"],
        request.headers.get("X-Etech-Timestamp"),
        corpo_cru,
        request.headers.get("X-Etech-Signature"),
    ):
        return "", 401

    evento = json.loads(corpo_cru)

    # O repetido não muda nada e também recebe 200.
    with conexao() as c:
        c.execute(
            """insert into etech_evento (id, sale_id, sequence, tipo, corpo)
               values (%s, %s, %s, %s, %s)
               on conflict (id) do nothing""",
            (evento["id"], evento["data"].get("saleId"), evento["sequence"],
             evento["type"], corpo_cru.decode("utf-8")),
        )
    return "", 200

Java

@RestController
public class EtechWebhookController {
    private final EventoRepositorio eventos;
    private final ObjectMapper json;
    private final String segredo;

    public EtechWebhookController(EventoRepositorio eventos, ObjectMapper json,
                                  @Value("${etech.webhook-secret}") String segredo) {
        this.eventos = eventos;
        this.json = json;
        this.segredo = segredo;
    }

    @PostMapping(path = "/webhooks/etech", consumes = "application/json")
    public ResponseEntity<Void> receber(
            @RequestHeader(value = "X-Etech-Timestamp", required = false) String timestamp,
            @RequestHeader(value = "X-Etech-Signature", required = false) String assinatura,
            @RequestBody byte[] corpoCru) throws Exception {

        long agora = Instant.now().getEpochSecond();
        if (!AssinaturaWebhook.valida(segredo, timestamp, corpoCru, assinatura, agora)) {
            return ResponseEntity.status(401).build();
        }

        JsonNode evento = json.readTree(corpoCru);
        JsonNode saleId = evento.path("data").path("saleId");

        // INSERT ... ON CONFLICT (id) DO NOTHING: o repetido não muda nada e também recebe 200.
        eventos.gravarSeNovo(
            UUID.fromString(evento.get("id").asText()),
            saleId.isNumber() ? saleId.asLong() : null,
            evento.get("sequence").asLong(),
            evento.get("type").asText(),
            new String(corpoCru, StandardCharsets.UTF_8));

        return ResponseEntity.ok().build();
    }
}

Reconciliação

Rode o feed de eventos de tempos em tempos (por exemplo a cada 15 minutos) e grave pela mesma rotina. Como a gravação deduplica pelo id, o que já chegou pelo webhook é ignorado e o que se perdeu entra.