Receive webhooks reliably

import express from "express";
import { verifyFluxSignature } from "./verify-flux-signature.js"; // from "Verify signatures"

const app = express();

app.post("/flux", express.raw({ type: "application/json" }), async (req, res) => {
  // 1. Verify against the raw body.
  if (!verifyFluxSignature(process.env.FLUX_WEBHOOK_SECRET, req.get("Flux-Signature"), req.body)) {
    return res.sendStatus(400);
  }
  const event = JSON.parse(req.body);

  // 2. Record the event id once. A unique index on event_id turns duplicates into no-ops.
  const isNew = await db.query(
    "INSERT INTO flux_events (event_id, type, body) VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING",
    [event.id, event.type, req.body.toString("utf8")],
  ).then((r) => r.rowCount === 1);

  // 3. Queue the work and answer at once (well inside 10 seconds).
  if (isNew) await queue.add("flux-event", { eventId: event.id });
  res.sendStatus(200);
});

// 4. The worker re-reads current state when order matters.
queue.process("flux-event", async ({ eventId }) => {
  const event = await loadEvent(eventId);
  if (event.type.startsWith("invoice.")) {
    const invoice = await flux(`/invoices/${event.data.object.id}`).then((r) => r.json());
    await saveInvoiceState(invoice);
  }
  // ...other types; ignore the ones you don't handle.
});

import json
import os

from flask import Flask, request

from verify import verify_flux_signature  # from "Verify signatures"

app = Flask(__name__)


@app.post("/flux")
def flux_webhook():
    raw = request.get_data()
    # 1. Verify against the raw body.
    if not verify_flux_signature(os.environ["FLUX_WEBHOOK_SECRET"], request.headers.get("Flux-Signature"), raw):
        return "", 400
    event = json.loads(raw)

    # 2. Record the event id once. A unique index on event_id turns duplicates into no-ops.
    is_new = db.execute(
        "INSERT INTO flux_events (event_id, type, body) VALUES (%s, %s, %s) ON CONFLICT (event_id) DO NOTHING",
        (event["id"], event["type"], raw.decode()),
    ).rowcount == 1

    # 3. Queue the work and answer at once (well inside 10 seconds).
    if is_new:
        queue.enqueue(process_flux_event, event["id"])
    return "", 200


# 4. The worker re-reads current state when order matters.
def process_flux_event(event_id):
    event = load_event(event_id)
    if event["type"].startswith("invoice."):
        invoice = flux("GET", f"/invoices/{event['data']['object']['id']}")
        save_invoice_state(invoice)
    # ...other types; ignore the ones you don't handle.