Handle a webhook

1. Verify signature                              (reject 400 if bad)
2. Parse JSON
3. Dedupe by the Soxara-Delivery header          (return 200 if seen)
4. Enqueue work + persist the delivery row       (transactional)
5. Return 200
6. Workers process the queue asynchronously

import express from 'express';
import crypto from 'node:crypto';
import { Pool } from 'pg';
import { Queue } from 'bullmq';
 
const app = express();
const db = new Pool({ connectionString: process.env.DATABASE_URL });
const queue = new Queue('soxara-events', { connection: { url: process.env.REDIS_URL } });
const SECRET = process.env.SOXARA_WEBHOOK_SECRET;
 
function verifySignature(rawBody, header) {
  if (!header) return false;
  const parts = Object.fromEntries(header.split(',').map(p => p.split('=')));
  const t = parts.t, v1 = parts.v1;
  if (!t || !v1) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - parseInt(t, 10)) > 300) return false;
  const expected = crypto
    .createHmac('sha256', SECRET)
    .update(`${t}.${rawBody}`)
    .digest('hex');
  try {
    return crypto.timingSafeEqual(
      Buffer.from(expected, 'hex'),
      Buffer.from(v1, 'hex'),
    );
  } catch { return false; }
}
 
app.post('/webhooks/soxara',
  express.raw({ type: 'application/json' }),
  async (req, res) => {
    // 1. Verify
    if (!verifySignature(req.body, req.headers['soxara-signature'])) {
      return res.status(400).send('bad signature');
    }
 
    // 2. Parse
    let event;
    try {
      event = JSON.parse(req.body.toString());
    } catch {
      return res.status(400).send('bad json');
    }
 
    // 3 + 4. Dedupe on Soxara-Delivery — it's stable across retries of the
    //        SAME event (a payment.completed and a payroll_run.completed
    //        for different resources get different ids, but a retried
    //        delivery of one you already saw keeps the same id).
    //        Insert with ON CONFLICT DO NOTHING gives us idempotent dedupe;
    //        0 rows back means it's a duplicate — return 200 without
    //        re-enqueueing.
    const deliveryId = req.headers['soxara-delivery'];
    const result = await db.query(
      `INSERT INTO soxara_deliveries (delivery_id, type, payload)
       VALUES ($1, $2, $3)
       ON CONFLICT (delivery_id) DO NOTHING`,
      [deliveryId, event.type, event],
    );
 
    if (result.rowCount === 1) {
      // New delivery — enqueue work
      await queue.add(event.type, event, { jobId: deliveryId });
    }
 
    // 5. Always return 200 once verified + persisted
    res.status(200).send('ok');
  },
);
 
app.listen(3000);

CREATE TABLE soxara_deliveries (
    delivery_id  TEXT PRIMARY KEY,
    type         TEXT NOT NULL,
    payload      JSONB NOT NULL,
    received_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

import os, json, time, hmac, hashlib
from flask import Flask, request
import psycopg
from rq import Queue
from redis import Redis
 
app = Flask(__name__)
db = psycopg.connect(os.environ["DATABASE_URL"])
q = Queue("soxara-events", connection=Redis.from_url(os.environ["REDIS_URL"]))
SECRET = os.environ["SOXARA_WEBHOOK_SECRET"]
 
 
def verify(raw: bytes, sig: str) -> bool:
    if not sig:
        return False
    parts = dict(p.split("=", 1) for p in sig.split(","))
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1:
        return False
    if abs(int(time.time()) - int(t)) > 300:
        return False
    expected = hmac.new(
        SECRET.encode(),
        f"{t}.".encode() + raw,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, v1)
 
 
@app.post("/webhooks/soxara")
def handler():
    raw = request.get_data()
    if not verify(raw, request.headers.get("Soxara-Signature")):
        return "bad signature", 400
 
    try:
        event = json.loads(raw)
    except json.JSONDecodeError:
        return "bad json", 400
 
    delivery_id = request.headers.get("Soxara-Delivery")
 
    # Idempotent insert; rowcount tells us if we'd seen it before
    with db.cursor() as cur:
        cur.execute(
            """
            INSERT INTO soxara_deliveries (delivery_id, type, payload)
            VALUES (%s, %s, %s)
            ON CONFLICT (delivery_id) DO NOTHING
            """,
            (delivery_id, event["type"], json.dumps(event)),
        )
        is_new = cur.rowcount == 1
    db.commit()
 
    if is_new:
        q.enqueue("worker.process_event", event, job_id=delivery_id)
 
    return "ok", 200