---
title: Build a webhook receiver
description: A webhook receiver in TypeScript or Python that verifies signatures, answers fast, skips duplicates and survives retries, with a self-test and a cURL script to send it deliveries.
canonical: https://developers.coevera.com/recipes/webhook-receiver/
section: recipes
verification: behaviour
lastVerified: 2026-10-15
publisher: Coevera
---

# Build a webhook receiver

**Answer:** Read the raw body and verify WebHook-Signature in constant time, answering 401 on a mismatch (never 403 or 410). Answer 200 at once and process afterwards. Skip events whose event, entity.id, entity.revision and event_time you have seen, drop events older than 25 hours, and keep the highest revision per record. Accept the DELETE sent when the webhook is deleted.

This receiver accepts the deliveries of a Coevera webhook, such as one subscribed to
`Opportunity.Update` for **Northwind Analytics**' deals, and handles each event once, in the
right order, however often it arrives. It has no dependencies beyond Node.js or Python's
`requests`, and a self-test that proves it handles retries, replays and forgeries.

The receiver itself calls no API. Its self-test reads one opportunity over REST to build a
realistic delivery, because deliveries carry records in the REST API's format; that is why the
samples have no GraphQL version.

## Before you begin

- A webhook with a `signature`, and its value in `WEBHOOK_SECRET`
  ([create a webhook](/webhooks/overview/#create-a-webhook)). The self-test makes one up if it is
  not set.
- For the self-test, an API application that can read opportunities, and its credentials in
  `COEVERA_API_KEY` and `COEVERA_API_SECRET` ([quickstart](/get-started/quickstart/)).
- Node.js 18 or later, or Python 3.9 or later with `requests`. The cURL script needs `curl`, `jq`
  and `openssl`.

The operations this recipe calls, in the API reference: REST
[`GET /entities/Opportunities`](/reference/rest/opportunities/#list).

## 1. Verify the signature

Check the bytes as they arrived, before any JSON parsing, against `WebHook-Signature`, and compare
in constant time. The cURL tab computes the same value with `openssl`, to check a delivery by
hand.

_This sample is REST only._

**REST · cURL**

```bash
# signature_of BODY: lowercase hex HMAC-SHA256 of the exact bytes, keyed with the webhook's
# signature value. verify_signature BODY HEADER: succeeds when HEADER is that signature.
signature_of() {
  printf '%s' "$1" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | sed 's/^.*= //'
}
verify_signature() {
  [ "$(signature_of "$1")" = "$(printf '%s' "$2" | tr -d '[:space:]' | tr 'A-F' 'a-f')" ]
}
```

**REST · TypeScript**

```ts
// WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
// webhook's signature value. Check the bytes as received, before parsing, in constant time.
function verifySignature(rawBody: Buffer, header: string | undefined, secret: string): boolean {
  if (!header) return false;
  const expected = Buffer.from(createHmac("sha256", secret).update(rawBody).digest("hex"));
  const received = Buffer.from(header.trim().toLowerCase());
  return expected.length === received.length && timingSafeEqual(expected, received);
}
```

**REST · Python**

```python
# WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
# webhook's signature value. Check the bytes as received, before parsing, in constant time.
def verify_signature(raw_body: bytes, header: Optional[str], secret: str) -> bool:
    if not header:
        return False
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.strip().lower())
```

**What to check:** a delivery with a wrong signature gets 401. Never answer it with 403 or 410:
Coevera disables the webhook at once on either.

## 2. Skip duplicates and old events

A delivery has no ID, and a retry sends the same body again for about 24 hours. The receiver keys
each event on `event`, `entity.id`, `entity.revision` and `event_time` and skips a key it has
seen, which makes its processing idempotent: an event that arrives twice is handled once. It
drops events older than 25 hours, since a replayed body still verifies. `event_time` is UTC
without an offset: the TypeScript version adds the `Z`, or JavaScript would read it as local time.
In production, keep the keys in your database, not in memory. This is the `deduplicate` step of
the program below: `firstTime` (`first_time` in Python) and `ageMs` (`age`).

## 3. Keep the newest revision

Deliveries are not ordered, and each chunk retries on its own, so an older revision can arrive
after a newer one. The receiver stores a record only if its `revision` is higher than the one it
has; every write is safe to repeat. Keep `id` and `revision` in the record if the webhook trims it
with `load-only`. This is `processEvent` (`process_event`) in the program below.

## 4. Answer first, then process

The handler answers 200 as soon as the signature checks out and the body parses, and only then
processes the events: Coevera waits at most 10 seconds. It answers 400 to a body that is not a
list of events, and 200 to the `DELETE` Coevera sends when the webhook is deleted. In production,
hand the events to a job queue.

The whole program, steps 1 to 4 and the self-test below; the cURL tab is the test-delivery script:

_This sample is REST only._

**REST · cURL**

```bash
COEVERA_BASE="https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}"
SPACE_ID="{space_id}"
RECEIVER_URL="${RECEIVER_URL:-http://localhost:8787/coevera-webhooks}"

# signature_of BODY: lowercase hex HMAC-SHA256 of the exact bytes, keyed with the webhook's
# signature value. verify_signature BODY HEADER: succeeds when HEADER is that signature.
signature_of() {
  printf '%s' "$1" | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" | sed 's/^.*= //'
}
verify_signature() {
  [ "$(signature_of "$1")" = "$(printf '%s' "$2" | tr -d '[:space:]' | tr 'A-F' 'a-f')" ]
}

# The test delivery carries a real record, as the REST API returns it.
record=$(curl -sS --fail-with-body -G "$COEVERA_BASE/entities/Opportunities" \
  -d first=1 \
  -d load-only=id,name,owner_id,step_id,revision \
  -u "$COEVERA_API_KEY:$COEVERA_API_SECRET" | jq -ce '.data[0]')

# One line of JSON, like a delivery; event_time is UTC without an offset.
body=$(jq -cn --argjson entity "$record" --arg url "$RECEIVER_URL" --arg space "$SPACE_ID" \
  --arg time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
  '[{webhook: {url: $url, events: ["Opportunity.Update"]}, event: "Opportunity.Update",
     event_time: $time, team_space_id: $space, entity: $entity, related_entity: null}]')
signature=$(signature_of "$body")
verify_signature "$body" "$signature"
curl -sS --fail-with-body "$RECEIVER_URL" \
  -H "Content-Type: application/json" \
  -H "WebHook-Signature: $signature" \
  -H "WebHook-Signature-Version: v1" \
  -H "Webhook-Delivery-Retry: 0" \
  --data-binary "$body"
echo
```

**REST · TypeScript**

```ts
import { Buffer } from "node:buffer";
import { createHmac, randomUUID, timingSafeEqual } from "node:crypto";
import { createServer, type IncomingMessage, type ServerResponse } from "node:http";
import type { AddressInfo } from "node:net";
import { argv, env } from "node:process";

const base = `https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}`;
const spaceId = "{space_id}";
const auth = Buffer.from(`${env.COEVERA_API_KEY}:${env.COEVERA_API_SECRET}`).toString("base64");
const port = Number(env.PORT ?? 8787);

// WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
// webhook's signature value. Check the bytes as received, before parsing, in constant time.
function verifySignature(rawBody: Buffer, header: string | undefined, secret: string): boolean {
  if (!header) return false;
  const expected = Buffer.from(createHmac("sha256", secret).update(rawBody).digest("hex"));
  const received = Buffer.from(header.trim().toLowerCase());
  return expected.length === received.length && timingSafeEqual(expected, received);
}

interface WebhookEvent {
  event: string;
  event_time: string;
  team_space_id: string;
  entity: { id: string; revision?: number; [field: string]: unknown };
  related_entity: Record<string, unknown> | null;
}

// A delivery has no id, and a retry sends the same body again for about 24 hours. Remember
// what you processed for longer than that (in a database, in production), and drop older
// events: a replayed body still carries a valid signature.
const MAX_AGE_MS = 25 * 60 * 60 * 1000;
const seen = new Map<string, number>();

function ageMs(e: WebhookEvent, now: number): number {
  // event_time is UTC. Without an offset, JavaScript would read it as local time: add the Z.
  const time = /(?:Z|[+-]\d\d:\d\d)$/.test(e.event_time) ? e.event_time : `${e.event_time}Z`;
  return now - Date.parse(time);
}

function firstTime(e: WebhookEvent, now: number): boolean {
  for (const [key, at] of seen) if (now - at > MAX_AGE_MS) seen.delete(key);
  const key = [e.event, e.entity.id, e.entity.revision, e.event_time].join("|");
  if (seen.has(key)) return false;
  seen.set(key, now);
  return true;
}

// Deliveries are not ordered, and each chunk retries on its own: an older revision can arrive
// after a newer one. Keep the newest, and make every write safe to repeat.
const records = new Map<string, { revision: number; deleted: boolean; data: Record<string, unknown> }>();

function processEvent(e: WebhookEvent): string {
  const revision = e.entity.revision ?? 0;
  const current = records.get(e.entity.id);
  if (current && current.revision >= revision) return `kept revision ${current.revision}`;
  records.set(e.entity.id, { revision, deleted: e.event.endsWith(".Delete"), data: e.entity });
  return `stored revision ${revision}`;
}

// The webhook's signature value. The self-test makes one up when WEBHOOK_SECRET is not set.
const secret = env.WEBHOOK_SECRET ?? randomUUID();

function reply(res: ServerResponse, status: number, body: object): void {
  res.writeHead(status, { "Content-Type": "application/json" });
  res.end(JSON.stringify(body));
}

function header(req: IncomingMessage, name: string): string | undefined {
  const value = req.headers[name];
  return Array.isArray(value) ? value[0] : value;
}

// Answer within 10 seconds, then process. Never answer 403 or 410 (Coevera disables the
// webhook at once on either): a bad signature gets 401.
const server = createServer(async (req, res) => {
  // Deleting the webhook sends DELETE to its URL: accept it.
  if (req.method === "DELETE") return reply(res, 200, { removed: true });
  if (req.method !== "POST") return reply(res, 405, { error: "POST only" });
  const chunks: Buffer[] = [];
  for await (const chunk of req) chunks.push(chunk as Buffer);
  const rawBody = Buffer.concat(chunks);
  if (!verifySignature(rawBody, header(req, "webhook-signature"), secret)) return reply(res, 401, { error: "bad signature" });
  let events: WebhookEvent[];
  try {
    events = JSON.parse(rawBody.toString("utf8"));
  } catch {
    return reply(res, 400, { error: "not JSON" });
  }
  if (!Array.isArray(events)) return reply(res, 400, { error: "not a list of events" });
  reply(res, 200, { received: events.length });
  // The answer is on its way. In production, hand the events to a job queue here.
  const now = Date.now();
  for (const e of events) {
    const outcome = ageMs(e, now) > MAX_AGE_MS ? "too old, dropped" : firstTime(e, now) ? processEvent(e) : "duplicate, skipped";
    console.log(`${e.event} ${e.entity.id}: ${outcome}`);
  }
});

// The test delivery carries a real record, as the REST API returns it.
async function findRecord(): Promise<Record<string, unknown>> {
  const res = await fetch(`${base}/entities/Opportunities?first=1&load-only=id,name,owner_id,step_id,revision`, {
    headers: { Authorization: `Basic ${auth}` },
  });
  const body = await res.json();
  if (!res.ok) throw new Error(`HTTP ${res.status} ${body.name}: ${body.message}`);
  if (!body.data.length) throw new Error("The self-test needs one opportunity in the space.");
  return body.data[0];
}

// A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
// deleted webhook look like to the receiver.
async function selfTest(url: string): Promise<void> {
  const record = await findRecord();
  const delivery = (eventTime: Date) =>
    JSON.stringify([
      {
        webhook: { id: randomUUID(), url, events: ["Opportunity.Update"] },
        event: "Opportunity.Update",
        event_time: eventTime.toISOString().slice(0, -1),
        team_space_id: spaceId,
        entity: record,
        related_entity: null,
      },
    ]);
  const send = (body: string, signature: string, retry = 0) =>
    fetch(url, {
      method: "POST",
      headers: { "Content-Type": "application/json", "WebHook-Signature": signature, "WebHook-Signature-Version": "v1", "Webhook-Delivery-Retry": String(retry) },
      body,
    }).then((r) => r.status);
  const sign = (body: string) => createHmac("sha256", secret).update(body).digest("hex");

  const fresh = delivery(new Date());
  const old = delivery(new Date(Date.now() - 48 * 60 * 60 * 1000));
  const answers = {
    first: await send(fresh, sign(fresh)),
    retry: await send(fresh, sign(fresh), 1),
    old: await send(old, sign(old)),
    forged: await send(fresh, "0".repeat(64)),
    deleted: await fetch(url, { method: "DELETE" }).then((r) => r.status),
  };
  console.log("Answers:", JSON.stringify(answers));
  const expected = { first: 200, retry: 200, old: 200, forged: 401, deleted: 200 };
  if (JSON.stringify(answers) !== JSON.stringify(expected)) throw new Error(`Expected ${JSON.stringify(expected)}`);
  if (records.size !== 1 || seen.size !== 1) throw new Error("A retry or an old event was processed.");
}

if (argv.includes("--serve")) {
  server.listen(port, () => console.log(`Listening on http://localhost:${port}/`));
} else {
  server.listen(0, "127.0.0.1");
  await new Promise((resolve) => server.once("listening", resolve));
  const { port: testPort } = server.address() as AddressInfo;
  try {
    await selfTest(`http://127.0.0.1:${testPort}/coevera-webhooks`);
    console.log("Self-test passed.");
  } finally {
    server.close();
    server.closeAllConnections();
  }
}
```

**REST · Python**

```python
import hashlib
import hmac
import json
import os
import queue
import sys
import threading
import uuid
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional

import requests


BASE = "https://{region}.api.coevera.com/api/v100/rest/spaces/{space_id}"
SPACE_ID = "{space_id}"
session = requests.Session()
session.auth = (os.environ["COEVERA_API_KEY"], os.environ["COEVERA_API_SECRET"])
PORT = int(os.environ.get("PORT", "8787"))


# WebHook-Signature is the lowercase hex HMAC-SHA256 of the raw request body, keyed with the
# webhook's signature value. Check the bytes as received, before parsing, in constant time.
def verify_signature(raw_body: bytes, header: Optional[str], secret: str) -> bool:
    if not header:
        return False
    expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header.strip().lower())


# A delivery has no id, and a retry sends the same body again for about 24 hours. Remember
# what you processed for longer than that (in a database, in production), and drop older
# events: a replayed body still carries a valid signature.
MAX_AGE = timedelta(hours=25)
seen = {}
seen_lock = threading.Lock()


def age(event, now):
    # event_time is UTC; read it as UTC whether or not it carries an offset.
    when = datetime.fromisoformat(event["event_time"].replace("Z", "+00:00"))
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    return now - when


def first_time(event, now):
    key = "|".join(str(part) for part in (event["event"], event["entity"]["id"], event["entity"].get("revision"), event["event_time"]))
    with seen_lock:
        for old_key, at in list(seen.items()):
            if now - at > MAX_AGE:
                del seen[old_key]
        if key in seen:
            return False
        seen[key] = now
        return True


# Deliveries are not ordered, and each chunk retries on its own: an older revision can arrive
# after a newer one. Keep the newest, and make every write safe to repeat.
records = {}


def process_event(event):
    entity = event["entity"]
    revision = entity.get("revision") or 0
    current = records.get(entity["id"])
    if current and current["revision"] >= revision:
        return f"kept revision {current['revision']}"
    records[entity["id"]] = {"revision": revision, "deleted": event["event"].endswith(".Delete"), "data": entity}
    return f"stored revision {revision}"


# The webhook's signature value. The self-test makes one up when WEBHOOK_SECRET is not set.
SECRET = os.environ.get("WEBHOOK_SECRET") or str(uuid.uuid4())
jobs = queue.Queue()


def worker():
    while True:
        events = jobs.get()
        now = datetime.now(timezone.utc)
        for event in events:
            if age(event, now) > MAX_AGE:
                outcome = "too old, dropped"
            elif first_time(event, now):
                outcome = process_event(event)
            else:
                outcome = "duplicate, skipped"
            print(f"{event['event']} {event['entity']['id']}: {outcome}", flush=True)
        jobs.task_done()


class Receiver(BaseHTTPRequestHandler):
    # Answer within 10 seconds, then process. Never answer 403 or 410 (Coevera disables the
    # webhook at once on either): a bad signature gets 401.
    def do_POST(self):
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not verify_signature(raw_body, self.headers.get("WebHook-Signature"), SECRET):
            return self.reply(401, {"error": "bad signature"})
        try:
            events = json.loads(raw_body)
        except ValueError:
            return self.reply(400, {"error": "not JSON"})
        if not isinstance(events, list):
            return self.reply(400, {"error": "not a list of events"})
        self.reply(200, {"received": len(events)})
        jobs.put(events)

    def do_DELETE(self):
        # Deleting the webhook sends DELETE to its URL: accept it.
        self.reply(200, {"removed": True})

    def reply(self, status, body):
        payload = json.dumps(body).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, format, *args):
        pass


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


# The test delivery carries a real record, as the REST API returns it.
def find_record():
    response = session.get(
        f"{BASE}/entities/Opportunities",
        params={"first": 1, "load-only": "id,name,owner_id,step_id,revision"},
        timeout=30,
    )
    body = response.json()
    if not response.ok:
        sys.exit(f"HTTP {response.status_code} {body.get('name')}: {body.get('message')}")
    if not body["data"]:
        sys.exit("The self-test needs one opportunity in the space.")
    return body["data"][0]


# A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
# deleted webhook look like to the receiver.
def self_test(url):
    record = find_record()

    def delivery(event_time):
        return json.dumps([{
            "webhook": {"id": str(uuid.uuid4()), "url": url, "events": ["Opportunity.Update"]},
            "event": "Opportunity.Update",
            "event_time": event_time.replace(tzinfo=None).isoformat(),
            "team_space_id": SPACE_ID,
            "entity": record,
            "related_entity": None,
        }]).encode("utf-8")

    def sign(body):
        return hmac.new(SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()

    def send(body, signature, retry=0):
        headers = {
            "Content-Type": "application/json",
            "WebHook-Signature": signature,
            "WebHook-Signature-Version": "v1",
            "Webhook-Delivery-Retry": str(retry),
        }
        status = requests.post(url, data=body, headers=headers, timeout=10).status_code
        jobs.join()  # wait until the receiver has processed it, so the checks below see the result
        return status

    now = datetime.now(timezone.utc)
    fresh = delivery(now)
    old = delivery(now - timedelta(hours=48))
    answers = {
        "first": send(fresh, sign(fresh)),
        "retry": send(fresh, sign(fresh), 1),
        "old": send(old, sign(old)),
        "forged": send(fresh, "0" * 64),
        "deleted": requests.delete(url, timeout=10).status_code,
    }
    print("Answers:", json.dumps(answers))
    expected = {"first": 200, "retry": 200, "old": 200, "forged": 401, "deleted": 200}
    if answers != expected:
        sys.exit(f"Expected {json.dumps(expected)}")
    if len(records) != 1 or len(seen) != 1:
        sys.exit("A retry or an old event was processed.")


if "--serve" in sys.argv:
    port = int(sys.argv[sys.argv.index("--port") + 1]) if "--port" in sys.argv else PORT
    server = ThreadingHTTPServer(("", port), Receiver)
    print(f"Listening on http://localhost:{server.server_port}/", flush=True)
    server.serve_forever()
else:
    server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        self_test(f"http://127.0.0.1:{server.server_port}/coevera-webhooks")
        print("Self-test passed.")
    finally:
        server.shutdown()
```

**What to check:** a delivery gets its 200 before any event is processed, and a `DELETE` gets 200.

## 5. Test it

Run the receiver without arguments for the self-test. It starts on a free port and sends itself
a delivery built from one of your opportunities, the same delivery again, one from two days ago,
a forged one and a `DELETE`:

```bash
npx tsx index.ts
```

It prints `Answers: {"first":200,"retry":200,"old":200,"forged":401,"deleted":200}`, then
`Self-test passed.`; `python main.py` does the same. The self-test builds its delivery from the
first opportunity it finds:

_This sample is REST only._

**REST · cURL**

```bash
# The test delivery carries a real record, as the REST API returns it.
record=$(curl -sS --fail-with-body -G "$COEVERA_BASE/entities/Opportunities" \
  -d first=1 \
  -d load-only=id,name,owner_id,step_id,revision \
  -u "$COEVERA_API_KEY:$COEVERA_API_SECRET" | jq -ce '.data[0]')
```

**REST · TypeScript**

```ts
// The test delivery carries a real record, as the REST API returns it.
async function findRecord(): Promise<Record<string, unknown>> {
  const res = await fetch(`${base}/entities/Opportunities?first=1&load-only=id,name,owner_id,step_id,revision`, {
    headers: { Authorization: `Basic ${auth}` },
  });
  const body = await res.json();
  if (!res.ok) throw new Error(`HTTP ${res.status} ${body.name}: ${body.message}`);
  if (!body.data.length) throw new Error("The self-test needs one opportunity in the space.");
  return body.data[0];
}
```

**REST · Python**

```python
# The test delivery carries a real record, as the REST API returns it.
def find_record():
    response = session.get(
        f"{BASE}/entities/Opportunities",
        params={"first": 1, "load-only": "id,name,owner_id,step_id,revision"},
        timeout=30,
    )
    body = response.json()
    if not response.ok:
        sys.exit(f"HTTP {response.status_code} {body.get('name')}: {body.get('message')}")
    if not body["data"]:
        sys.exit("The self-test needs one opportunity in the space.")
    return body["data"][0]
```

**Recorded response · REST · 200 OK**

```json
{
  "success": true,
  "total": 3,
  "page_info": {
    "start_cursor": "WyIwYjAwMDAwMC0wMDAwLTQwMDAtYTAwMC0wMDAwMDAwMDAwMDEiXQ==",
    "end_cursor": "WyIwYjAwMDAwMC0wMDAwLTQwMDAtYTAwMC0wMDAwMDAwMDAwMDEiXQ==",
    "has_previous_page": false,
    "has_next_page": true
  },
  "data": [
    {
      "id": "0b000000-0000-4000-a000-000000000001",
      "name": "Northwind data platform rollout",
      "owner_id": "0d000000-0000-4000-a000-000000000001",
      "step_id": "07000000-0000-4000-a000-000000000001",
      "revision": 646
    }
  ]
}
```

_Recorded on 2026-10-15 against API 6.5.2 in the docs sandbox, de-identified._

Then it sends the deliveries. The cURL tab is the script that sends one delivery to a running
receiver:

_This sample is REST only._

**REST · cURL**

```bash
# One line of JSON, like a delivery; event_time is UTC without an offset.
body=$(jq -cn --argjson entity "$record" --arg url "$RECEIVER_URL" --arg space "$SPACE_ID" \
  --arg time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
  '[{webhook: {url: $url, events: ["Opportunity.Update"]}, event: "Opportunity.Update",
     event_time: $time, team_space_id: $space, entity: $entity, related_entity: null}]')
signature=$(signature_of "$body")
verify_signature "$body" "$signature"
curl -sS --fail-with-body "$RECEIVER_URL" \
  -H "Content-Type: application/json" \
  -H "WebHook-Signature: $signature" \
  -H "WebHook-Signature-Version: v1" \
  -H "Webhook-Delivery-Retry: 0" \
  --data-binary "$body"
echo
```

**REST · TypeScript**

```ts
// A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
// deleted webhook look like to the receiver.
async function selfTest(url: string): Promise<void> {
  const record = await findRecord();
  const delivery = (eventTime: Date) =>
    JSON.stringify([
      {
        webhook: { id: randomUUID(), url, events: ["Opportunity.Update"] },
        event: "Opportunity.Update",
        event_time: eventTime.toISOString().slice(0, -1),
        team_space_id: spaceId,
        entity: record,
        related_entity: null,
      },
    ]);
  const send = (body: string, signature: string, retry = 0) =>
    fetch(url, {
      method: "POST",
      headers: { "Content-Type": "application/json", "WebHook-Signature": signature, "WebHook-Signature-Version": "v1", "Webhook-Delivery-Retry": String(retry) },
      body,
    }).then((r) => r.status);
  const sign = (body: string) => createHmac("sha256", secret).update(body).digest("hex");

  const fresh = delivery(new Date());
  const old = delivery(new Date(Date.now() - 48 * 60 * 60 * 1000));
  const answers = {
    first: await send(fresh, sign(fresh)),
    retry: await send(fresh, sign(fresh), 1),
    old: await send(old, sign(old)),
    forged: await send(fresh, "0".repeat(64)),
    deleted: await fetch(url, { method: "DELETE" }).then((r) => r.status),
  };
  console.log("Answers:", JSON.stringify(answers));
  const expected = { first: 200, retry: 200, old: 200, forged: 401, deleted: 200 };
  if (JSON.stringify(answers) !== JSON.stringify(expected)) throw new Error(`Expected ${JSON.stringify(expected)}`);
  if (records.size !== 1 || seen.size !== 1) throw new Error("A retry or an old event was processed.");
}
```

**REST · Python**

```python
# A delivery signed the way Coevera signs, then what retries, replays, forgeries and a
# deleted webhook look like to the receiver.
def self_test(url):
    record = find_record()

    def delivery(event_time):
        return json.dumps([{
            "webhook": {"id": str(uuid.uuid4()), "url": url, "events": ["Opportunity.Update"]},
            "event": "Opportunity.Update",
            "event_time": event_time.replace(tzinfo=None).isoformat(),
            "team_space_id": SPACE_ID,
            "entity": record,
            "related_entity": None,
        }]).encode("utf-8")

    def sign(body):
        return hmac.new(SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()

    def send(body, signature, retry=0):
        headers = {
            "Content-Type": "application/json",
            "WebHook-Signature": signature,
            "WebHook-Signature-Version": "v1",
            "Webhook-Delivery-Retry": str(retry),
        }
        status = requests.post(url, data=body, headers=headers, timeout=10).status_code
        jobs.join()  # wait until the receiver has processed it, so the checks below see the result
        return status

    now = datetime.now(timezone.utc)
    fresh = delivery(now)
    old = delivery(now - timedelta(hours=48))
    answers = {
        "first": send(fresh, sign(fresh)),
        "retry": send(fresh, sign(fresh), 1),
        "old": send(old, sign(old)),
        "forged": send(fresh, "0" * 64),
        "deleted": requests.delete(url, timeout=10).status_code,
    }
    print("Answers:", json.dumps(answers))
    expected = {"first": 200, "retry": 200, "old": 200, "forged": 401, "deleted": 200}
    if answers != expected:
        sys.exit(f"Expected {json.dumps(expected)}")
    if len(records) != 1 or len(seen) != 1:
        sys.exit("A retry or an old event was processed.")
```

**What to check:** the receiver logs `stored revision` for the first delivery, `duplicate,
skipped` for the retry and `too old, dropped` for the old one, and stores the record once.

Without arguments the program runs the self-test and stops. To run it for real, start it with
`--serve` and your secret, then point a webhook at it
([testing locally](/webhooks/testing-locally/)):

```bash
WEBHOOK_SECRET="<the webhook's signature value>" npx tsx index.ts --serve
```

## Clean up

The receiver writes nothing to Coevera, and the self-test only reads one opportunity. Its store
is in memory and is gone when it stops.

## Limits

- The seen keys and the records live in memory: a restart forgets them, and two instances do not
  share them. Use a database for both in production.
- The TypeScript version processes events in the request handler after answering, the Python
  version in one worker thread; a busy receiver needs a job queue.
- It does not check `team_space_id`. If one receiver serves several spaces, check it against the
  space each secret belongs to.
- On a `Delete` event the receiver marks the record deleted and keeps its revision.