Webhooks overview

import express from "express";
import { Pool } from "pg";

const app = express();
const db = new Pool();

app.post(
  "/webhooks/obsidian",
  express.raw({ type: "application/json" }), // raw body, not express.json()
  async (req, res) => {
    const sig = req.header("X-Webhook-Signature") ?? "";
    if (!verify(process.env.TF_WEBHOOK_SECRET!, sig, req.body)) {
      return res.status(400).send("bad signature");
    }

    const event = JSON.parse(req.body.toString("utf8"));

    // Ack fast — do the real work after responding.
    res.status(200).end();

    try {
      await handleIdempotently(event);
    } catch (err) {
      // Log and let the delivery fail naturally if you returned non-2xx —
      // here we've already acked, so failures need their own alerting path
      // (dead-letter queue, error tracker) rather than relying on Obsidian's retry.
      console.error("webhook processing failed", event.idempotency_key, err);
    }
  },
);

async function handleIdempotently(event: any) {
  const client = await db.connect();
  try {
    await client.query("BEGIN");

    // Unique constraint on idempotency_key does the real dedupe work —
    // this insert either succeeds once, or throws a unique-violation.
    const inserted = await client.query(
      `INSERT INTO processed_webhooks (idempotency_key, event_type, received_at)
       VALUES ($1, $2, now())
       ON CONFLICT (idempotency_key) DO NOTHING
       RETURNING idempotency_key`,
      [event.idempotency_key, event.event_type],
    );

    if (inserted.rowCount === 0) {
      // Already processed this exact delivery — no-op.
      await client.query("ROLLBACK");
      return;
    }

    // Apply the actual business effect in the same transaction.
    if (event.event_type === "identity.kyc.session.approved") {
      await client.query(
        `UPDATE users SET kyc_status = 'approved' WHERE external_ref = $1`,
        [event.vendor_data],
      );
    }

    await client.query("COMMIT");
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

from fastapi import FastAPI, Request, HTTPException
from sqlalchemy import text
from db import engine  # your SQLAlchemy engine

app = FastAPI()

@app.post("/webhooks/obsidian")
async def receive_webhook(request: Request):
    raw_body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")

    if not verify(WEBHOOK_SECRET, signature, raw_body):
        raise HTTPException(status_code=400, detail="bad signature")

    event = await request.json()

    # Ack fast: hand off to a background task/queue and return 200 immediately
    # in real deployments. Shown inline here for clarity.
    handle_idempotently(event)
    return {"ok": True}


def handle_idempotently(event: dict) -> None:
    with engine.begin() as conn:  # one transaction for dedupe + effect
        result = conn.execute(
            text("""
                INSERT INTO processed_webhooks (idempotency_key, event_type, received_at)
                VALUES (:key, :event_type, now())
                ON CONFLICT (idempotency_key) DO NOTHING
                RETURNING idempotency_key
            """),
            {"key": event["idempotency_key"], "event_type": event["event_type"]},
        )
        if result.rowcount == 0:
            return  # already processed this exact delivery

        if event["event_type"] == "identity.kyc.session.approved":
            conn.execute(
                text("UPDATE users SET kyc_status = 'approved' WHERE external_ref = :ref"),
                {"ref": event.get("vendor_data")},
            )