Webhooks

{
  "name": "fernbrook-webhooks",
  "private": true,
  "type": "module",
  "engines": { "node": ">=22.13" },
  "dependencies": {
    "express": "5.3.0",
    "webflow-api": "3.3.4"
  }
}

// server.mjs
import crypto from "node:crypto";
import express from "express";

const SECRET = process.env.WEBFLOW_WEBHOOK_SECRET; // per-webhook secret, or the app client secret
const MAX_AGE_MS = 5 * 60 * 1000;

if (!SECRET) throw new Error("WEBFLOW_WEBHOOK_SECRET is not set");

/** Returns true when the request was signed by Webflow and is recent. */
function verify(headers, rawBody, now = Date.now()) {
  const timestamp = headers["x-webflow-timestamp"];
  const signature = headers["x-webflow-signature"];
  if (typeof timestamp !== "string" || typeof signature !== "string") return false;

  const sent = Number(timestamp);
  if (!Number.isFinite(sent)) return false;
  const sentMs = sent > 1e12 ? sent : sent * 1000; // accept seconds or milliseconds
  if (Math.abs(now - sentMs) > MAX_AGE_MS) return false;

  const expected = crypto
    .createHmac("sha256", SECRET)
    .update(`${timestamp}:`)
    .update(rawBody)
    .digest("hex");

  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(signature, "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// ---- idempotency and queue (in memory; see the production section) ----
const seen = new Map(); // key -> expiry
const TTL_MS = 24 * 60 * 60 * 1000;
const queue = [];
let draining = false;

function alreadySeen(key) {
  const now = Date.now();
  for (const [k, exp] of seen) if (exp < now) seen.delete(k);
  if (seen.has(key)) return true;
  seen.set(key, now + TTL_MS);
  return false;
}

function enqueue(job) {
  queue.push(job);
  if (!draining) void drain();
}

async function drain() {
  draining = true;
  while (queue.length > 0) {
    const job = queue.shift();
    try {
      await handle(job);
    } catch (err) {
      console.error("handler failed", job.key, err);
      seen.delete(job.key); // allow a Webflow retry to be processed again
    }
  }
  draining = false;
}

// ---- Fernbrook handlers ----
async function handleFormSubmission(payload) {
  const { name, data, submittedAt, localeId } = payload;
  if (name !== "Book a guided tour") return;
  const visitor = {
    fullName: data["Full name"],
    email: data["Email"],
    visitDate: data["Visit date"],
    groupSize: Number(data["Group size"] ?? 1),
    language: localeId ? "non-default locale" : "English",
  };
  // Replace with your mail or CRM client. Never log the full payload: it holds personal data.
  console.log("tour request", { visitDate: visitor.visitDate, groupSize: visitor.groupSize, submittedAt });
}

async function handleItemPublished(payload) {
  const { collectionId, id, fieldData } = payload;
  if (collectionId !== process.env.JOURNAL_COLLECTION_ID) return;
  const res = await fetch(process.env.REVALIDATE_URL, {
    method: "POST",
    headers: { "content-type": "application/json", "x-revalidate-token": process.env.REVALIDATE_TOKEN },
    body: JSON.stringify({ path: `/journal/${fieldData.slug}`, itemId: id }),
  });
  if (!res.ok) throw new Error(`revalidate failed: ${res.status}`);
}

async function handleSitePublish(payload) {
  console.log("site published", payload.publishedOn ?? "", payload.domains ?? []);
}

async function handle({ triggerType, payload }) {
  switch (triggerType) {
    case "form_submission":
      return handleFormSubmission(payload);
    case "collection_item_published":
      return handleItemPublished(payload);
    case "site_publish":
      return handleSitePublish(payload);
    default:
      console.log("ignored trigger", triggerType);
  }
}

// ---- HTTP ----
const app = express();

app.post("/webflow/:kind", express.raw({ type: "application/json", limit: "1mb" }), (req, res) => {
  const raw = req.body; // Buffer, because of express.raw
  if (!Buffer.isBuffer(raw) || !verify(req.headers, raw)) {
    return res.status(400).send("invalid signature");
  }

  let event;
  try {
    event = JSON.parse(raw.toString("utf8"));
  } catch {
    return res.status(400).send("invalid json");
  }

  const p = event.payload ?? {};
  const key = [
    event.triggerType,
    p.id ?? p.formId ?? "",
    p.lastPublished ?? p.lastUpdated ?? p.submittedAt ?? "",
  ].join("|");

  if (!alreadySeen(key)) enqueue({ key, ...event });
  return res.status(200).send("ok"); // exactly 200
});

app.get("/healthz", (_req, res) => res.status(200).send("ok"));

const port = Number(process.env.PORT ?? 8080);
app.listen(port, () => console.log(`listening on ${port}`));

npm install
export WEBFLOW_WEBHOOK_SECRET="REPLACE_WITH_SECRET"
export JOURNAL_COLLECTION_ID="REPLACE_WITH_JOURNAL_COLLECTION_ID"
export REVALIDATE_URL="https://www.fernbrook-gardens.example/api/revalidate"
export REVALIDATE_TOKEN="REPLACE_WITH_TOKEN"
node server.mjs

flask==3.1.0
webflow==2.0.0
gunicorn==23.0.0

# app.py
import hashlib
import hmac
import json
import os
import queue
import threading
import time

from flask import Flask, request

SECRET = os.environ["WEBFLOW_WEBHOOK_SECRET"].encode("utf-8")
MAX_AGE_SECONDS = 5 * 60
TTL_SECONDS = 24 * 60 * 60

app = Flask(__name__)
jobs: "queue.Queue[dict]" = queue.Queue()
seen: dict[str, float] = {}
seen_lock = threading.Lock()


def verify(timestamp: str | None, signature: str | None, raw_body: bytes) -> bool:
    """True when the request was signed by Webflow and is recent."""
    if not timestamp or not signature:
        return False
    try:
        sent = float(timestamp)
    except ValueError:
        return False
    sent_seconds = sent / 1000 if sent > 1e12 else sent  # accept seconds or milliseconds
    if abs(time.time() - sent_seconds) > MAX_AGE_SECONDS:
        return False

    message = timestamp.encode("utf-8") + b":" + raw_body
    digest = hmac.new(SECRET, message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(digest, signature)


def already_seen(key: str) -> bool:
    now = time.time()
    with seen_lock:
        for k in [k for k, exp in seen.items() if exp < now]:
            del seen[k]
        if key in seen:
            return True
        seen[key] = now + TTL_SECONDS
        return False


def forget(key: str) -> None:
    with seen_lock:
        seen.pop(key, None)


def handle_form_submission(payload: dict) -> None:
    if payload.get("name") != "Book a guided tour":
        return
    data = payload.get("data", {})
    # Replace with your mail or CRM client. Do not log the whole payload: it is personal data.
    print("tour request", data.get("Visit date"), data.get("Group size"), payload.get("submittedAt"))


def handle_item_published(payload: dict) -> None:
    import urllib.request

    if payload.get("collectionId") != os.environ["JOURNAL_COLLECTION_ID"]:
        return
    slug = payload.get("fieldData", {}).get("slug", "")
    body = json.dumps({"path": f"/journal/{slug}", "itemId": payload.get("id")}).encode("utf-8")
    req = urllib.request.Request(
        os.environ["REVALIDATE_URL"],
        data=body,
        method="POST",
        headers={
            "content-type": "application/json",
            "x-revalidate-token": os.environ["REVALIDATE_TOKEN"],
        },
    )
    with urllib.request.urlopen(req, timeout=10) as res:
        if res.status >= 300:
            raise RuntimeError(f"revalidate failed: {res.status}")


def handle_site_publish(payload: dict) -> None:
    print("site published", payload.get("publishedOn"), payload.get("domains"))


HANDLERS = {
    "form_submission": handle_form_submission,
    "collection_item_published": handle_item_published,
    "site_publish": handle_site_publish,
}


def worker() -> None:
    while True:
        job = jobs.get()
        try:
            handler = HANDLERS.get(job["triggerType"])
            if handler:
                handler(job.get("payload", {}))
            else:
                print("ignored trigger", job["triggerType"])
        except Exception as exc:  # noqa: BLE001
            print("handler failed", job["key"], exc)
            forget(job["key"])  # let a retry be processed
        finally:
            jobs.task_done()


threading.Thread(target=worker, daemon=True).start()


@app.post("/webflow/<kind>")
def webflow_hook(kind: str):
    raw = request.get_data()  # bytes, before any JSON parsing
    if not verify(
        request.headers.get("x-webflow-timestamp"),
        request.headers.get("x-webflow-signature"),
        raw,
    ):
        return "invalid signature", 400

    try:
        event = json.loads(raw)
    except json.JSONDecodeError:
        return "invalid json", 400

    p = event.get("payload", {}) or {}
    key = "|".join(
        [
            str(event.get("triggerType", "")),
            str(p.get("id") or p.get("formId") or ""),
            str(p.get("lastPublished") or p.get("lastUpdated") or p.get("submittedAt") or ""),
        ]
    )
    if not already_seen(key):
        jobs.put({"key": key, **event})
    return "ok", 200


@app.get("/healthz")
def healthz():
    return "ok", 200

pip install -r requirements.txt
export WEBFLOW_WEBHOOK_SECRET="REPLACE_WITH_SECRET"
export JOURNAL_COLLECTION_ID="REPLACE_WITH_JOURNAL_COLLECTION_ID"
export REVALIDATE_URL="https://www.fernbrook-gardens.example/api/revalidate"
export REVALIDATE_TOKEN="REPLACE_WITH_TOKEN"
gunicorn --workers 1 --threads 4 --bind 0.0.0.0:8080 app:app

CREATE TABLE webhook_events (
  key         TEXT PRIMARY KEY,
  trigger     TEXT        NOT NULL,
  body        JSONB       NOT NULL,
  received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  status      TEXT        NOT NULL DEFAULT 'pending',
  attempts    INTEGER     NOT NULL DEFAULT 0
);

// inside the route, replacing alreadySeen and enqueue
const result = await db.query(
  "INSERT INTO webhook_events (key, trigger, body) VALUES ($1, $2, $3) ON CONFLICT (key) DO NOTHING",
  [key, event.triggerType, event],
);
// result.rowCount is 0 for a duplicate; either way answer 200
return res.status(200).send("ok");