diff --git a/.gitignore b/.gitignore
index 596a0104d..25ab1369b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,6 +8,10 @@ public/sitemap*.xml
# TypeScript
*.tsbuildinfo
+# Python examples and tests
+__pycache__/
+*.pyc
+
*.bak
# Vale synced packages (re-sync with `vale sync`)
diff --git a/app/en/build/_meta.tsx b/app/en/build/_meta.tsx
index 1ac006542..b11956e5d 100644
--- a/app/en/build/_meta.tsx
+++ b/app/en/build/_meta.tsx
@@ -12,6 +12,9 @@ export const meta: MetaRecord = {
title: "Quickstart: build an MCP server",
href: "/get-started/quickstarts/mcp-server-quickstart",
},
+ eventing: {
+ title: "Events and webhooks",
+ },
"tool-calling": {
title: "Call tools",
},
diff --git a/app/en/build/eventing/page.mdx b/app/en/build/eventing/page.mdx
new file mode 100644
index 000000000..b3a56e076
--- /dev/null
+++ b/app/en/build/eventing/page.mdx
@@ -0,0 +1,551 @@
+---
+title: "Build event-driven integrations"
+description: "Produce, inspect, and deliver Arcade events with triggers, schedules, and signed webhooks"
+---
+
+import { Callout, Steps, Tabs } from "nextra/components";
+
+# Build event-driven integrations
+
+Use Arcade eventing when your application or agent should react after something happens. This guide follows one fact from its producer, through project event history, to a signed request at your receiver.
+
+## The eventing model
+
+The producer side is separate from the delivery side:
+
+```text
+trigger instance ─┐
+schedule ─────────┼─> Arcade event ─> webhook subscription ─> webhook delivery ─> receiver
+provider ingress ─┘
+```
+
+| Term | Responsibility | Source | Destination |
+| --- | --- | --- | --- |
+| Arcade event | Store | Trigger instance, schedule, or provider ingress | Project history |
+| Trigger type | Configure | Toolkit declaration | Trigger instance |
+| Trigger instance | Produce | Connected-account observation | Arcade event |
+| Schedule | Produce | Time rule | Arcade event |
+| Provider ingress | Produce | Verified provider callback | Arcade event |
+| Webhook subscription | Route | Matching Arcade event | Webhook delivery |
+| Webhook delivery | Deliver | Webhook subscription | Configured receiver |
+
+A trigger type is a reusable toolkit declaration. A trigger instance binds that type to one user, connection, and filter configuration. Poll trigger instances, such as Gmail, ask the provider for changes on a declared cadence. Provider ingress instead begins realtime processing when Arcade receives a verified provider callback; it does not run an Arcade polling loop.
+
+Provider ingress setup is specific to a realtime provider. Find supported providers in [Integrations](/resources/integrations); do not configure provider ingress as an outgoing webhook.
+
+A webhook subscription is outgoing: it selects Arcade events and routes them to your URL. A webhook delivery is one event-subscription pairing and owns its attempt history. Neither is an OAuth callback URL.
+
+## Choose your deployment origin
+
+The origin changes by deployment mode. The scoped REST path does not.
+
+| Mode | API origin | Dashboard origin |
+| --- | --- | --- |
+| Arcade Cloud | `https://api.arcade.dev` | `https://app.arcade.dev` |
+| Customer-managed | `$ARCADE_ENGINE_URL` | `$ARCADE_ENGINE_URL/dashboard` |
+| Local | `http://localhost:9099` | `http://localhost:9099/dashboard` |
+
+The examples default to Arcade Cloud. For customer-managed or local Arcade,
+replace the API origin with the value in the table.
+
+```bash
+export ARCADE_API_ORIGIN="https://api.arcade.dev"
+export ARCADE_ORG_ID="your-org-id"
+export ARCADE_PROJECT_ID="your-project-id"
+export ARCADE_API_KEY="your-api-key"
+export RECEIVER_URL="https://receiver.example.com/events"
+export SCOPE="$ARCADE_API_ORIGIN/v1/orgs/$ARCADE_ORG_ID/projects/$ARCADE_PROJECT_ID"
+```
+
+Every REST request below sends `Authorization: Bearer $ARCADE_API_KEY` and stays under `/v1/orgs/$ARCADE_ORG_ID/projects/$ARCADE_PROJECT_ID`.
+
+
+Credentials for one organization and project cannot access another scope's resources. Keep the organization, project, and API key from the same selected project. Do not accept organization or project authority from an event payload.
+
+
+The [Arcade API reference](/references/api) owns the complete request and response schemas. This guide keeps only the fields needed for the two journeys.
+
+## Verify the receiver before processing
+
+Arcade returns a webhook signing secret only when the subscription is created or its secret is rotated. It uses the `whsec_` prefix followed by padded standard base64. Store it as a secret and verify the exact raw request body before parsing JSON or checking for duplicates.
+
+The following Python 3.10+ example has no framework dependency. Call `receive` from your HTTP handler with the unchanged raw request bytes and lowercase or mixed-case headers, then use its return value as the HTTP response status.
+
+```python filename="receiver.py"
+import base64
+import hashlib
+import hmac
+import json
+import logging
+import sqlite3
+import time
+from collections.abc import Callable, Mapping
+
+TOLERANCE_SECONDS = 300
+SECRET_FORMAT_ERROR = (
+ "webhook secrets must use whsec_ followed by padded standard base64"
+)
+WebhookSecrets = list[str] | tuple[str, ...]
+logger = logging.getLogger(__name__)
+
+
+class VerificationError(Exception):
+ pass
+
+
+class ConfigurationError(Exception):
+ pass
+
+
+def verify_request(
+ body: bytes,
+ headers: Mapping[str, str],
+ secrets: WebhookSecrets,
+ now: int | None = None,
+) -> tuple[dict, str]:
+ if not isinstance(secrets, (list, tuple)) or any(
+ not isinstance(secret, str) for secret in secrets
+ ):
+ raise ConfigurationError("webhook secrets must be a list or tuple of strings")
+
+ normalized = {key.lower(): value for key, value in headers.items()}
+ try:
+ delivery_id = normalized["webhook-id"]
+ timestamp_text = normalized["webhook-timestamp"]
+ supplied = normalized["webhook-signature"].split()
+ except KeyError as error:
+ raise VerificationError(f"missing {error.args[0]}") from error
+
+ try:
+ timestamp = int(timestamp_text)
+ except ValueError as error:
+ raise VerificationError("invalid webhook-timestamp") from error
+
+ verification_time = int(time.time()) if now is None else now
+ if abs(verification_time - timestamp) > TOLERANCE_SECONDS:
+ raise VerificationError("webhook-timestamp outside tolerance")
+
+ signed = (
+ delivery_id.encode()
+ + b"."
+ + timestamp_text.encode()
+ + b"."
+ + body
+ )
+ keys: list[bytes] = []
+ for secret in secrets:
+ if not secret.startswith("whsec_"):
+ raise ConfigurationError(SECRET_FORMAT_ERROR)
+ try:
+ key = base64.b64decode(secret.removeprefix("whsec_"), validate=True)
+ except ValueError as error:
+ raise ConfigurationError(SECRET_FORMAT_ERROR) from error
+ if not key:
+ raise ConfigurationError(SECRET_FORMAT_ERROR)
+ keys.append(key)
+ if not keys:
+ raise ConfigurationError("no webhook secrets configured")
+
+ matched = False
+ for key in keys:
+ digest = hmac.new(key, signed, hashlib.sha256).digest()
+ expected = b"v1," + base64.b64encode(digest)
+ for candidate in supplied:
+ try:
+ encoded = candidate.encode("ascii")
+ except UnicodeEncodeError:
+ continue
+ matched |= hmac.compare_digest(expected, encoded)
+ if not matched:
+ raise VerificationError("invalid webhook-signature")
+
+ try:
+ event = json.loads(body)
+ except json.JSONDecodeError as error:
+ raise VerificationError("invalid JSON") from error
+ if not isinstance(event, dict):
+ raise VerificationError("event must be a JSON object")
+ return event, delivery_id
+
+
+class SQLiteInbox:
+ """A durable idempotency inbox for one receiver process."""
+
+ def __init__(self, path: str):
+ self.path = path
+ connection = sqlite3.connect(path, timeout=30, isolation_level=None)
+ try:
+ connection.execute(
+ """CREATE TABLE IF NOT EXISTS webhook_inbox (
+ webhook_id TEXT PRIMARY KEY,
+ received_at INTEGER NOT NULL
+ )"""
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ def handle(
+ self,
+ delivery_id: str,
+ event: dict,
+ handler: Callable[[sqlite3.Connection, dict], None],
+ ) -> bool:
+ connection = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+ try:
+ connection.execute("BEGIN IMMEDIATE")
+ inserted = connection.execute(
+ """INSERT OR IGNORE INTO webhook_inbox(webhook_id, received_at)
+ VALUES (?, ?)""",
+ (delivery_id, int(time.time())),
+ ).rowcount
+ if inserted == 0:
+ connection.commit()
+ return False
+ handler(connection, event)
+ connection.commit()
+ return True
+ except Exception:
+ connection.rollback()
+ raise
+ finally:
+ connection.close()
+
+
+def receive(
+ body: bytes,
+ headers: Mapping[str, str],
+ subscription_secrets: WebhookSecrets,
+ inbox: SQLiteInbox,
+ authorize: Callable[[dict], bool],
+ handler: Callable[[sqlite3.Connection, dict], None],
+ now: int | None = None,
+) -> int:
+ try:
+ event, delivery_id = verify_request(body, headers, subscription_secrets, now)
+ except VerificationError:
+ return 400
+ except ConfigurationError:
+ return 500
+
+ try:
+ # Build this callback from server-side subscription configuration. Do not
+ # accept event types or tenant IDs merely because they appear in the payload.
+ if not authorize(event):
+ return 403
+ inbox.handle(delivery_id, event, handler)
+ except Exception:
+ logger.exception("webhook handler failed")
+ return 500
+ return 204
+```
+
+The inbox claim and your business writes must share one transaction. If the handler fails, both roll back and the non-2xx response lets Arcade retry. A valid duplicate returns `204` without rerunning the handler. During secret rotation, pass both active secrets; remove the retired secret after the grace period.
+
+Resolve one subscription from your HTTP route before calling `receive`, and pass only that subscription's current and previous secrets as `subscription_secrets`. Build `authorize` from that server-side subscription: allow its configured event types and, when the event schema carries tenant identifiers, compare them with the route's organization and project. The payload itself is not authority. Arcade event envelopes use `type`, `timestamp`, and `data`:
+
+```json
+{
+ "type": "tenant.audit_recorded",
+ "timestamp": "2026-08-28T21:00:00Z",
+ "data": {"organization_id":"org_123","project_id":"proj_123"}
+}
+```
+
+An authorization rejection returns `403`, so Arcade retries and eventually records a dead delivery; use that behavior to surface a mis-scoped subscription rather than silently dropping it. On an otherwise-valid request, any malformed current or previous secret makes the receiver return `500`, including during rotation. Configure a request-body size limit in the HTTP framework before reading the raw body.
+
+
+Use a persistent store in production. Keep each recorded `webhook-id` for at least Arcade's configured event-retention period (90 days by default), plus operational margin. Manual retry and recovery reuse the original delivery ID after automatic attempts end, so sizing the inbox only to the 27-hour automatic retry span can repeat business side effects.
+
+
+The sample records `received_at` for that cleanup policy but does not schedule the cleanup job for you. SQLite serializes these write transactions, so keep the handler's transactional work short and local; use a concurrent durable inbox for production receivers with parallel or network-bound work.
+
+## Recover expired connected accounts
+
+Subscribe to connection lifecycle events when your app should tell a user that an OAuth connection needs attention:
+
+- `connected_account.created`—the connection became active for the first time.
+- `connected_account.expired`—Arcade can no longer use the connection.
+- `connected_account.reconnected`—reauthorization restored an expired connection.
+
+Create the project-scoped subscription:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/webhooks" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data "{\"url\":\"$RECEIVER_URL\",\"event_types\":[\"connected_account.expired\"]}"
+```
+
+Arcade sends the Standard Webhooks envelope. Verify the Standard Webhooks signature against the raw body before reading it:
+
+```json
+{
+ "type": "connected_account.expired",
+ "timestamp": "2026-08-28T21:00:00Z",
+ "data": {
+ "organization_id": "org_123",
+ "project_id": "proj_123",
+ "user_id": "user@example.com",
+ "provider_id": "google",
+ "connection_id": "ac_123",
+ "status": "expired",
+ "reason": "refresh_failed"
+ }
+}
+```
+
+All three events identify the `organization_id`, `user_id`, `provider_id`, `connection_id`, and `status`. Project-bound events also include `project_id`. Only `connected_account.expired` includes `reason`:
+
+- `no_refresh_token` means the access token expired and no refresh token was available.
+- `refresh_failed` means the provider permanently rejected the refresh grant.
+
+Each lifecycle event can originate from a project-bound or organization-bound connection. The public subscription API is project-scoped, so it delivers the project-bound events in the preceding example. Organization-bound lifecycle events omit `project_id` and are not delivered to a project subscription.
+
+In the Dashboard, open the expired connected user, and choose **Reconnect**. The same action is available through `POST /v1/orgs/{org_id}/projects/{project_id}/auth/authorize`:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/auth/authorize" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data '{
+ "user_id":"user@example.com",
+ "auth_requirement":{
+ "provider_id":"google",
+ "provider_type":"oauth2",
+ "oauth2":{"scopes":["https://www.googleapis.com/auth/gmail.readonly"]}
+ }
+ }'
+```
+
+Open the returned authorization URL. A successful flow restores the same connection and emits `connected_account.reconnected`. Triggers pinned to it become healthy again. Use the scopes on the connected-account record; Dashboard does this for you. Arcade also applies scopes configured on the provider. A reconnected event reports the OAuth state change, but normal scope checks still apply if the provider returns a reduced grant.
+
+Engine does not observe MCP-managed OAuth refresh outcomes, so MCP-managed OAuth connections do not emit this lifecycle or use the standard OAuth Reconnect action.
+
+## Try a scheduled event
+
+A schedule is a configurable producer; every fire creates a separate retained Arcade event with its own delivery history. Use an interval for this check.
+
+
+
+
+
+
+### Create the outgoing webhook
+
+Open the selected project's **Webhooks** page, create a URL endpoint for your receiver, and set **Events to send** to `demo.follow_up`. Copy the signing secret when it appears.
+
+### Create the schedule
+
+Open **Schedules**, create `Demo follow-up`, choose an interval of 60 seconds, set the event type to `demo.follow_up`, and use this payload:
+
+```json
+{"customer_id":"demo-123","action":"follow_up"}
+```
+
+Record the schedule ID and **Next fire** time.
+
+### Inspect the event and delivery
+
+After the due time, poll **Events** for up to 120 seconds and select the row whose source is that schedule. The event detail shows the subscription, delivery ID, status, and attempts. Confirm the delivery succeeds and your receiver records a signature-valid request with the same `webhook-id`.
+
+### Clean up
+
+Delete the demo schedule and webhook subscription. Return to **Events** and confirm the event is still present.
+
+
+
+
+
+
+Create the subscription and save the returned `id` and one-time `secret`:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/webhooks" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data "{\"url\":\"$RECEIVER_URL\",\"event_types\":[\"demo.follow_up\"]}"
+
+export WEBHOOK_ID=""
+export WEBHOOK_SECRET=""
+```
+
+Load `WEBHOOK_SECRET` into this receiver route's secret store and pass `[WEBHOOK_SECRET]` as `subscription_secrets` to `receive`. Do not combine secrets from unrelated subscriptions.
+
+Create a schedule and save its `id` and `next_fire_at`:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/schedules" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data '{
+ "name":"Demo follow-up",
+ "event_type":"demo.follow_up",
+ "interval_seconds":60,
+ "payload":{"customer_id":"demo-123","action":"follow_up"}
+ }'
+
+export SCHEDULE_ID=""
+```
+
+After `next_fire_at`, poll for up to 120 seconds and list only that schedule's events. Save the returned event ID, then inspect that event and its delivery attempts:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/events?source_type=schedule&source_id=$SCHEDULE_ID"
+
+export EVENT_ID=""
+
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/events/$EVENT_ID"
+```
+
+Delete the resources, then repeat `GET /events/{event_id}` to confirm the retained event remains:
+
+```bash
+curl --fail-with-body --silent --show-error --request DELETE \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/schedules/$SCHEDULE_ID"
+
+curl --fail-with-body --silent --show-error --request DELETE \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/webhooks/$WEBHOOK_ID"
+
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/events/$EVENT_ID"
+```
+
+
+
+
+## Try a filtered Gmail trigger
+
+The `gmail.message.received` trigger declares a 60-second polling interval. Its optional filters are `subject_contains`, `from`, and `to`; `to` matches both To and Cc recipients. Filters inspect message metadata, not message bodies.
+
+Before starting, connect the Gmail account for the user who owns the trigger. Create a webhook subscription for `gmail.message.received` and keep the receiver serving. For REST, use the webhook request above with `gmail.message.received` as its event type, then set `WEBHOOK_ID` to the returned ID.
+
+
+
+
+
+
+### Create the trigger
+
+Open **Triggers**, choose **Email received**, and choose **Myself** for an initial test. Enter a run-unique value such as `CUSTOMER-20260827-1048` in **Subject contains**, then create the trigger.
+
+Wait for its first successful poll before sending the test messages. Send one message whose subject contains the token and one whose subject does not.
+
+### Inspect the result
+
+Open the trigger's **View delivery history** sheet. After a completed poll newer than both messages, the matching Gmail message ID appears in an event and the non-matching ID does not. Open the event to confirm its linked delivery succeeded, then check the same `webhook-id` at your receiver.
+
+### Clean up
+
+Delete the trigger and its demo webhook subscription. The emitted event remains in **Events** until event retention removes it.
+
+
+
+
+
+
+Set the connected user's stable ID and the connection to observe.
+
+```bash
+export ARCADE_USER_ID="connected-user@example.com"
+export CONNECTION_ID=""
+export SUBJECT_TOKEN="CUSTOMER-20260827-1048"
+```
+
+Create the trigger and save its `id`:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/triggers" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data "{\"type_slug\":\"gmail.message.received\",\"user_id\":\"$ARCADE_USER_ID\",\"connection_id\":\"$CONNECTION_ID\",\"config\":{\"subject_contains\":\"$SUBJECT_TOKEN\"}}"
+
+export TRIGGER_ID=""
+```
+
+After the first successful poll, send matching and non-matching messages. Inspect the trigger event list, event payload, and linked deliveries:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/triggers/$TRIGGER_ID/events"
+
+export EVENT_ID=""
+
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/triggers/$TRIGGER_ID/events/$EVENT_ID"
+```
+
+Delete the trigger and webhook subscription, then confirm the event remains:
+
+```bash
+curl --fail-with-body --silent --show-error --request DELETE \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/triggers/$TRIGGER_ID"
+
+curl --fail-with-body --silent --show-error --request DELETE \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/webhooks/$WEBHOOK_ID"
+
+curl --fail-with-body --silent --show-error \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ "$SCOPE/events/$EVENT_ID"
+```
+
+
+
+
+## Operate each resource
+
+Every route below is relative to the project-scoped `$SCOPE`. The Dashboard uses the same public project resources. Recovery and replay belong to the webhook subscription: recovery requeues existing dead deliveries; replay creates deliveries for retained matching events that subscription never received.
+
+| Resource | Dashboard | REST API |
+| --- | --- | --- |
+| Trigger instance | Inspect, enable, disable, delete | `GET /triggers/{trigger_id}`, `PATCH /triggers/{trigger_id}`, `DELETE /triggers/{trigger_id}`, `GET /triggers/{trigger_id}/events`, `GET /triggers/{trigger_id}/events/{event_id}` |
+| Schedule | Inspect, enable, disable, delete | `GET /schedules/{schedule_id}`, `PATCH /schedules/{schedule_id}`, `DELETE /schedules/{schedule_id}` |
+| Arcade event | List and inspect delivery traces | `GET /events`, `GET /events/{event_id}` |
+| Webhook subscription | Inspect, enable, disable, rotate signing secret, delete, recover, replay | `GET /webhooks/{webhook_id}`, `PATCH /webhooks/{webhook_id}`, `POST /webhooks/{webhook_id}/rotate_secret`, `DELETE /webhooks/{webhook_id}`, `POST /webhooks/{webhook_id}/recover_deliveries`, `POST /webhooks/{webhook_id}/replay_missing` |
+| Webhook delivery | Inspect and retry a dead delivery | `GET /webhooks/{webhook_id}/deliveries/{delivery_id}`, `POST /webhooks/{webhook_id}/deliveries/{delivery_id}/retry` |
+
+Deleting a trigger, schedule, or webhook subscription does not delete events already retained in project history.
+
+Recovery and replay take an RFC 3339 `since` timestamp in the JSON body:
+
+```bash
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/webhooks/$WEBHOOK_ID/recover_deliveries" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data '{"since":"2026-08-28T00:00:00Z"}'
+
+curl --fail-with-body --silent --show-error \
+ --request POST "$SCOPE/webhooks/$WEBHOOK_ID/replay_missing" \
+ --header "Authorization: Bearer $ARCADE_API_KEY" \
+ --header "Content-Type: application/json" \
+ --data '{"since":"2026-08-28T00:00:00Z"}'
+```
+
+## Reliability boundaries
+
+- **Schedule identity:** Arcade publishes one Arcade event per scheduled fire, keyed by schedule ID and scheduled fire time. A schedule and the events it publishes have separate lifecycles.
+- **Schedule timing:** A due schedule is published no later than 60 seconds after its due time. The local 120-second delivery observation in this guide is a readiness check, not a production delivery-latency guarantee.
+- **Delivery:** Outgoing webhook delivery is at-least-once. Arcade makes 8 attempts: immediately, then after 5 seconds, 5 minutes, 30 minutes, 2 hours, 5 hours, 10 hours, and 10 hours. The configured delay from attempt 1 through attempt 8 totals 27 hours, 35 minutes, and 5 seconds.
+- **Identity:** One delivery keeps the same `webhook-id` across automatic retries, manual retry, and recovery. That value is also the `{delivery_id}` used by the delivery detail and retry APIs. Different events, and one event delivered through different subscriptions, receive different IDs; if two subscriptions share a receiver, deduplicate each delivery independently.
+- **Signatures:** Arcade signs every retry with a fresh `webhook-timestamp`, so a short receiver tolerance remains compatible with the full retry schedule. Verify the raw body, `webhook-id`, and `webhook-timestamp` before deduplication. The sample receiver chooses a 300-second clock-skew tolerance: it accepts timestamps through 300 seconds in either direction and rejects timestamps 301 seconds away.
+- **Retention:** Project events are retained for 90 days by default. An event exactly at the retention cutoff remains replayable; an older event is unavailable for replay.
+- **Recovery and replay:** `since` is inclusive. Recovery selects existing failed deliveries at or after the boundary. Replay selects retained matching events at or after the boundary that were never delivered to that subscription.
+
+When a delivery is dead, retry that delivery. When several existing deliveries failed, recover them from a chosen time. When a subscription was absent or disabled, replay retained missing events from a chosen time.
diff --git a/examples/eventing/receiver.py b/examples/eventing/receiver.py
new file mode 100644
index 000000000..d16bfed41
--- /dev/null
+++ b/examples/eventing/receiver.py
@@ -0,0 +1,166 @@
+import base64
+import hashlib
+import hmac
+import json
+import logging
+import sqlite3
+import time
+from collections.abc import Callable, Mapping
+
+TOLERANCE_SECONDS = 300
+SECRET_FORMAT_ERROR = (
+ "webhook secrets must use whsec_ followed by padded standard base64"
+)
+WebhookSecrets = list[str] | tuple[str, ...]
+logger = logging.getLogger(__name__)
+
+
+class VerificationError(Exception):
+ pass
+
+
+class ConfigurationError(Exception):
+ pass
+
+
+def verify_request(
+ body: bytes,
+ headers: Mapping[str, str],
+ secrets: WebhookSecrets,
+ now: int | None = None,
+) -> tuple[dict, str]:
+ if not isinstance(secrets, (list, tuple)) or any(
+ not isinstance(secret, str) for secret in secrets
+ ):
+ raise ConfigurationError("webhook secrets must be a list or tuple of strings")
+
+ normalized = {key.lower(): value for key, value in headers.items()}
+ try:
+ delivery_id = normalized["webhook-id"]
+ timestamp_text = normalized["webhook-timestamp"]
+ supplied = normalized["webhook-signature"].split()
+ except KeyError as error:
+ raise VerificationError(f"missing {error.args[0]}") from error
+
+ try:
+ timestamp = int(timestamp_text)
+ except ValueError as error:
+ raise VerificationError("invalid webhook-timestamp") from error
+
+ verification_time = int(time.time()) if now is None else now
+ if abs(verification_time - timestamp) > TOLERANCE_SECONDS:
+ raise VerificationError("webhook-timestamp outside tolerance")
+
+ signed = (
+ delivery_id.encode()
+ + b"."
+ + timestamp_text.encode()
+ + b"."
+ + body
+ )
+ keys: list[bytes] = []
+ for secret in secrets:
+ if not secret.startswith("whsec_"):
+ raise ConfigurationError(SECRET_FORMAT_ERROR)
+ try:
+ key = base64.b64decode(secret.removeprefix("whsec_"), validate=True)
+ except ValueError as error:
+ raise ConfigurationError(SECRET_FORMAT_ERROR) from error
+ if not key:
+ raise ConfigurationError(SECRET_FORMAT_ERROR)
+ keys.append(key)
+ if not keys:
+ raise ConfigurationError("no webhook secrets configured")
+
+ matched = False
+ for key in keys:
+ digest = hmac.new(key, signed, hashlib.sha256).digest()
+ expected = b"v1," + base64.b64encode(digest)
+ for candidate in supplied:
+ try:
+ encoded = candidate.encode("ascii")
+ except UnicodeEncodeError:
+ continue
+ matched |= hmac.compare_digest(expected, encoded)
+ if not matched:
+ raise VerificationError("invalid webhook-signature")
+
+ try:
+ event = json.loads(body)
+ except json.JSONDecodeError as error:
+ raise VerificationError("invalid JSON") from error
+ if not isinstance(event, dict):
+ raise VerificationError("event must be a JSON object")
+ return event, delivery_id
+
+
+class SQLiteInbox:
+ """A durable idempotency inbox for one receiver process."""
+
+ def __init__(self, path: str):
+ self.path = path
+ connection = sqlite3.connect(path, timeout=30, isolation_level=None)
+ try:
+ connection.execute(
+ """CREATE TABLE IF NOT EXISTS webhook_inbox (
+ webhook_id TEXT PRIMARY KEY,
+ received_at INTEGER NOT NULL
+ )"""
+ )
+ connection.commit()
+ finally:
+ connection.close()
+
+ def handle(
+ self,
+ delivery_id: str,
+ event: dict,
+ handler: Callable[[sqlite3.Connection, dict], None],
+ ) -> bool:
+ connection = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+ try:
+ connection.execute("BEGIN IMMEDIATE")
+ inserted = connection.execute(
+ """INSERT OR IGNORE INTO webhook_inbox(webhook_id, received_at)
+ VALUES (?, ?)""",
+ (delivery_id, int(time.time())),
+ ).rowcount
+ if inserted == 0:
+ connection.commit()
+ return False
+ handler(connection, event)
+ connection.commit()
+ return True
+ except Exception:
+ connection.rollback()
+ raise
+ finally:
+ connection.close()
+
+
+def receive(
+ body: bytes,
+ headers: Mapping[str, str],
+ subscription_secrets: WebhookSecrets,
+ inbox: SQLiteInbox,
+ authorize: Callable[[dict], bool],
+ handler: Callable[[sqlite3.Connection, dict], None],
+ now: int | None = None,
+) -> int:
+ try:
+ event, delivery_id = verify_request(body, headers, subscription_secrets, now)
+ except VerificationError:
+ return 400
+ except ConfigurationError:
+ return 500
+
+ try:
+ # Build this callback from server-side subscription configuration. Do not
+ # accept event types or tenant IDs merely because they appear in the payload.
+ if not authorize(event):
+ return 403
+ inbox.handle(delivery_id, event, handler)
+ except Exception:
+ logger.exception("webhook handler failed")
+ return 500
+ return 204
diff --git a/tests/eventing-guide.test.ts b/tests/eventing-guide.test.ts
new file mode 100644
index 000000000..215c03a45
--- /dev/null
+++ b/tests/eventing-guide.test.ts
@@ -0,0 +1,177 @@
+import { readdirSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import { describe, expect, test } from "vitest";
+import { meta } from "../app/en/build/_meta";
+
+const PAGE = "app/en/build/eventing/page.mdx";
+const TITLE_RE = /title:\s*"Build event-driven integrations"/;
+const MODEL_SECTION_RE =
+ /## The eventing model([\s\S]*?)## Choose your deployment origin/;
+const TABLE_DATA_ROW_RE = /^\| (?!Term \|)(?!-)[^|]+\|/gm;
+const ACCOUNT_LIFECYCLE_SECTION_RE =
+ /## Recover expired connected accounts([\s\S]*?)## Try a scheduled event/;
+const JSON_BLOCK_RE = /```json\n([\s\S]*?)\n```/;
+const RETENTION_WINDOW_RE =
+ /Keep each recorded `webhook-id` for at least Arcade's configured event-retention period \(90 days by default\)/;
+const DELIVERY_ID_REUSE_RE =
+ /same `webhook-id` across automatic retries, manual retry, and recovery/;
+const TIMESTAMP_TOLERANCE_RE = /through (\d+) seconds/;
+const TIMESTAMP_REJECTION_RE = /timestamps (\d+) seconds away/;
+const TOLERANCE_CONSTANT_RE = /TOLERANCE_SECONDS = (\d+)/;
+const RETRY_DELAYS_RE =
+ /Arcade makes 8 attempts: immediately, then after ([^.]+)\. The configured delay/;
+const RETRY_TOTAL_RE = /totals (\d+) hours, (\d+) minutes, and (\d+) seconds/;
+const RETRY_DELAY_SEPARATOR_RE = /,\s*(?:and\s+)?/;
+const RETRY_DELAY_RE = /(\d+) (second|minute|hour)s?/;
+
+const page = readFileSync(join(process.cwd(), PAGE), "utf8");
+
+describe("unified eventing guide", () => {
+ test("registers the eventing content directory in Build navigation", () => {
+ const contentDirectories = readdirSync(
+ join(process.cwd(), "app/en/build"),
+ { withFileTypes: true }
+ )
+ .filter((entry) => entry.isDirectory())
+ .map((entry) => entry.name);
+ expect(contentDirectories).toContain("eventing");
+ expect(meta.eventing).toEqual({
+ title: "Events and webhooks",
+ });
+ expect(page).toMatch(TITLE_RE);
+ });
+
+ test("defines the exact seven-term source-to-destination model", () => {
+ const model = page.match(MODEL_SECTION_RE)?.[1] ?? "";
+ const rows = [
+ "| Arcade event | Store | Trigger instance, schedule, or provider ingress | Project history |",
+ "| Trigger type | Configure | Toolkit declaration | Trigger instance |",
+ "| Trigger instance | Produce | Connected-account observation | Arcade event |",
+ "| Schedule | Produce | Time rule | Arcade event |",
+ "| Provider ingress | Produce | Verified provider callback | Arcade event |",
+ "| Webhook subscription | Route | Matching Arcade event | Webhook delivery |",
+ "| Webhook delivery | Deliver | Webhook subscription | Configured receiver |",
+ ];
+ for (const row of rows) {
+ expect(model).toContain(row);
+ }
+ expect(model.match(TABLE_DATA_ROW_RE)).toHaveLength(7);
+ });
+
+ test("keeps examples on Dashboard and scoped REST surfaces", () => {
+ expect(page).toContain("## Try a scheduled event");
+ expect(page).toContain("## Try a filtered Gmail trigger");
+ expect(page).toContain('Tabs items={["Dashboard", "REST API"]}');
+ expect(page).toContain("Authorization: Bearer $ARCADE_API_KEY");
+ expect(page).toContain(
+ "/v1/orgs/$ARCADE_ORG_ID/projects/$ARCADE_PROJECT_ID"
+ );
+ for (const variable of [
+ "WEBHOOK_ID",
+ "SCHEDULE_ID",
+ "TRIGGER_ID",
+ "EVENT_ID",
+ ]) {
+ expect(page).toContain(`export ${variable}=`);
+ }
+ });
+
+ test("pins origins, tenant isolation, and the reference boundary", () => {
+ for (const value of [
+ "https://api.arcade.dev",
+ "https://app.arcade.dev",
+ "$ARCADE_ENGINE_URL/dashboard",
+ "http://localhost:9099",
+ "http://localhost:9099/dashboard",
+ ]) {
+ expect(page).toContain(value);
+ }
+ expect(page).toContain("[Arcade API reference](/references/api)");
+ });
+
+ test("documents the complete connected-account recovery contract", () => {
+ const section = page.match(ACCOUNT_LIFECYCLE_SECTION_RE)?.[1] ?? "";
+ for (const value of [
+ "connected_account.created",
+ "connected_account.expired",
+ "connected_account.reconnected",
+ '--request POST "$SCOPE/webhooks"',
+ "event_types",
+ "url",
+ "no_refresh_token",
+ "refresh_failed",
+ "POST /v1/orgs/{org_id}/projects/{project_id}/auth/authorize",
+ "MCP-managed OAuth",
+ ]) {
+ expect(section).toContain(value);
+ }
+ expect(section).toContain("**Reconnect**");
+ expect(section).toContain("public subscription API is project-scoped");
+ expect(section).toContain("Organization-bound lifecycle events omit");
+ expect(section).toContain("Verify the Standard Webhooks signature");
+ expect(section).toContain("reduced grant");
+
+ const envelope = JSON.parse(section.match(JSON_BLOCK_RE)?.[1] ?? "{}");
+ expect(Object.keys(envelope).sort()).toEqual(["data", "timestamp", "type"]);
+ expect(envelope.type).toBe("connected_account.expired");
+ expect(Object.keys(envelope.data).sort()).toEqual([
+ "connection_id",
+ "organization_id",
+ "project_id",
+ "provider_id",
+ "reason",
+ "status",
+ "user_id",
+ ]);
+ });
+
+ test("documents the supported lifecycle without hiding retained events", () => {
+ for (const route of [
+ "GET /triggers/{trigger_id}",
+ "PATCH /triggers/{trigger_id}",
+ "DELETE /triggers/{trigger_id}",
+ "GET /schedules/{schedule_id}",
+ "PATCH /schedules/{schedule_id}",
+ "DELETE /schedules/{schedule_id}",
+ "GET /events/{event_id}",
+ "POST /webhooks/{webhook_id}/rotate_secret",
+ "POST /webhooks/{webhook_id}/recover_deliveries",
+ "POST /webhooks/{webhook_id}/replay_missing",
+ "POST /webhooks/{webhook_id}/deliveries/{delivery_id}/retry",
+ ]) {
+ expect(page).toContain(route);
+ }
+ });
+
+ test("states only reviewed delivery guarantees and resource boundaries", () => {
+ const tolerance = Number(page.match(TIMESTAMP_TOLERANCE_RE)?.[1]);
+ const rejection = Number(page.match(TIMESTAMP_REJECTION_RE)?.[1]);
+ const receiverTolerance = Number(page.match(TOLERANCE_CONSTANT_RE)?.[1]);
+ expect(Number.isInteger(tolerance)).toBe(true);
+ expect(Number.isInteger(rejection)).toBe(true);
+ expect(Number.isInteger(receiverTolerance)).toBe(true);
+ expect(receiverTolerance).toBe(tolerance);
+ expect(rejection).toBe(tolerance + 1);
+
+ const unitSeconds = { second: 1, minute: 60, hour: 3600 };
+ const delays = page
+ .match(RETRY_DELAYS_RE)?.[1]
+ .split(RETRY_DELAY_SEPARATOR_RE)
+ .map((delay) => {
+ const [, amount, unit] = delay.match(RETRY_DELAY_RE) ?? [];
+ return Number(amount) * unitSeconds[unit as keyof typeof unitSeconds];
+ });
+ expect(delays).toHaveLength(7);
+ const [, hours, minutes, seconds] = page.match(RETRY_TOTAL_RE) ?? [];
+ const statedTotal =
+ Number(hours) * 3600 + Number(minutes) * 60 + Number(seconds);
+ expect(delays?.reduce((total, delay) => total + delay, 0)).toBe(
+ statedTotal
+ );
+ });
+
+ test("keeps deduplication through the manual recovery window", () => {
+ expect(page).toMatch(RETENTION_WINDOW_RE);
+ expect(page).toMatch(DELIVERY_ID_REUSE_RE);
+ });
+});
diff --git a/tests/eventing-receiver-executable.test.ts b/tests/eventing-receiver-executable.test.ts
new file mode 100644
index 000000000..763c8a6d3
--- /dev/null
+++ b/tests/eventing-receiver-executable.test.ts
@@ -0,0 +1,31 @@
+import { spawnSync } from "node:child_process";
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { describe, expect, test } from "vitest";
+
+const PAGE = "app/en/build/eventing/page.mdx";
+const RECEIVER = "examples/eventing/receiver.py";
+const RECEIVER_BLOCK_RE = /```python filename="receiver\.py"\n([\s\S]*?)\n```/;
+
+describe("eventing receiver example", () => {
+ test("the published snippet is the executable example", () => {
+ const page = readFileSync(join(process.cwd(), PAGE), "utf8");
+ const published = page.match(RECEIVER_BLOCK_RE)?.[1];
+ const executable = readFileSync(
+ join(process.cwd(), RECEIVER),
+ "utf8"
+ ).trim();
+
+ expect(published).toBe(executable);
+ });
+
+ test("executes signature, rotation, boundary, duplicate, and rollback proofs", () => {
+ const result = spawnSync("python3", ["tests/eventing_receiver_test.py"], {
+ cwd: process.cwd(),
+ encoding: "utf8",
+ env: { ...process.env, PYTHONPATH: process.cwd() },
+ });
+ expect(result.error?.message ?? "", result.stderr).toBe("");
+ expect(result.status, result.stderr || result.stdout).toBe(0);
+ });
+});
diff --git a/tests/eventing_receiver_test.py b/tests/eventing_receiver_test.py
new file mode 100644
index 000000000..694589e02
--- /dev/null
+++ b/tests/eventing_receiver_test.py
@@ -0,0 +1,320 @@
+import base64
+import hashlib
+import hmac
+import json
+import sqlite3
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from examples.eventing.receiver import (
+ ConfigurationError,
+ SQLiteInbox,
+ VerificationError,
+ receive,
+ verify_request,
+)
+
+
+NOW = 2_000_000_000
+BODY = json.dumps(
+ {
+ "type": "demo.follow_up",
+ "data": {
+ "id": "evt_1",
+ "organization_id": "org_1",
+ "project_id": "project_1",
+ },
+ }
+).encode()
+
+
+def authorize(event: dict) -> bool:
+ data = event.get("data")
+ return (
+ event.get("type") == "demo.follow_up"
+ and isinstance(data, dict)
+ and data.get("organization_id") == "org_1"
+ and data.get("project_id") == "project_1"
+ )
+
+
+def secret(key: bytes) -> str:
+ return "whsec_" + base64.b64encode(key).decode()
+
+
+def headers(key: bytes, *, delivery_id: str = "msg_1", timestamp: int = NOW, body: bytes = BODY) -> dict[str, str]:
+ timestamp_text = str(timestamp)
+ signed = delivery_id.encode() + b"." + timestamp_text.encode() + b"." + body
+ signature = "v1," + base64.b64encode(
+ hmac.new(key, signed, hashlib.sha256).digest()
+ ).decode()
+ return {
+ "webhook-id": delivery_id,
+ "webhook-timestamp": timestamp_text,
+ "webhook-signature": signature,
+ }
+
+
+class ReceiverTest(unittest.TestCase):
+ def test_accepts_the_canonical_standard_webhooks_vector(self) -> None:
+ body = b'{"event_type":"ping","data":{"success":true}}'
+ event, delivery_id = verify_request(
+ body,
+ {
+ "webhook-id": "msg_loFOjxBNrRLzqYUf",
+ "webhook-timestamp": "1731705121",
+ "webhook-signature": "v1,rAvfW3dJ/X/qxhsaXPOyyCGmRKsaKWcsNccKXlIktD0=",
+ },
+ ["whsec_plJ3nmyCDGBKInavdOK15jsl"],
+ 1731705121,
+ )
+ self.assertEqual("msg_loFOjxBNrRLzqYUf", delivery_id)
+ self.assertEqual("ping", event["event_type"])
+
+ def test_rejects_tampering_and_unknown_secrets(self) -> None:
+ key = b"current-secret"
+ with self.assertRaises(VerificationError):
+ verify_request(BODY + b" ", headers(key), [secret(key)], NOW)
+ with self.assertRaises(VerificationError):
+ verify_request(BODY, headers(key), [secret(b"other-secret")], NOW)
+
+ def test_accepts_exact_timestamp_boundary_and_rejects_beyond_it(self) -> None:
+ key = b"current-secret"
+ for offset in (-300, 300):
+ event, delivery_id = verify_request(
+ BODY, headers(key, timestamp=NOW + offset), [secret(key)], NOW
+ )
+ self.assertEqual("demo.follow_up", event["type"])
+ self.assertEqual("msg_1", delivery_id)
+ for offset in (-301, 301):
+ with self.assertRaises(VerificationError):
+ verify_request(
+ BODY, headers(key, timestamp=NOW + offset), [secret(key)], NOW
+ )
+
+ def test_accepts_either_active_rotation_secret(self) -> None:
+ old_key, new_key = b"old-secret", b"new-secret"
+ active = [secret(old_key), secret(new_key)]
+ verify_request(BODY, headers(old_key), active, NOW)
+ verify_request(BODY, headers(new_key), active, NOW)
+
+ combined = headers(old_key)
+ combined["webhook-signature"] += " " + headers(new_key)["webhook-signature"]
+ verify_request(BODY, combined, active, NOW)
+
+ def test_rejects_malformed_secret_collections_as_configuration(self) -> None:
+ key = b"current-secret"
+ with self.assertRaisesRegex(ConfigurationError, "list or tuple of strings"):
+ verify_request(BODY, headers(key), secret(key), NOW) # type: ignore[arg-type]
+ with self.assertRaisesRegex(ConfigurationError, "list or tuple of strings"):
+ verify_request(BODY, headers(key), [None], NOW) # type: ignore[list-item]
+ with self.assertRaisesRegex(ConfigurationError, "list or tuple of strings"):
+ verify_request(BODY, headers(key), {secret(key)}, NOW) # type: ignore[arg-type]
+
+ def test_rejects_malformed_headers_and_non_object_json(self) -> None:
+ key = b"current-secret"
+ valid = headers(key)
+ for required in ("webhook-id", "webhook-timestamp", "webhook-signature"):
+ candidate = dict(valid)
+ candidate.pop(required)
+ with self.assertRaises(VerificationError):
+ verify_request(BODY, candidate, [secret(key)], NOW)
+ with self.assertRaises(VerificationError):
+ verify_request(BODY, {**valid, "webhook-timestamp": "nope"}, [secret(key)], NOW)
+ with self.assertRaises(VerificationError):
+ verify_request(BODY, {**valid, "webhook-signature": "v1,é"}, [secret(key)], NOW)
+
+ scalar = b'"signed but not an event"'
+ with self.assertRaises(VerificationError):
+ verify_request(scalar, headers(key, body=scalar), [secret(key)], NOW)
+
+ def test_header_names_are_case_insensitive(self) -> None:
+ key = b"current-secret"
+ mixed_case = {
+ name.title(): value for name, value in headers(key).items()
+ }
+ event, delivery_id = verify_request(BODY, mixed_case, [secret(key)], NOW)
+ self.assertEqual("demo.follow_up", event["type"])
+ self.assertEqual("msg_1", delivery_id)
+
+ def test_malformed_rotation_secret_fails_configuration(self) -> None:
+ key = b"current-secret"
+ with self.assertRaises(ConfigurationError):
+ verify_request(BODY, headers(key), ["whsec_not-base64!", secret(key)], NOW)
+
+ def test_rejects_invalid_or_empty_prefixed_secrets(self) -> None:
+ key = b"current-secret"
+ for configured in (["not-prefixed"], ["whsec_not-base64!"], ["whsec_"]):
+ with self.assertRaisesRegex(
+ ConfigurationError,
+ "webhook secrets must use whsec_ followed by padded standard base64",
+ ):
+ verify_request(BODY, headers(key), configured, NOW)
+
+ def test_receive_maps_failures_and_keeps_duplicate_and_retry_contracts(self) -> None:
+ key = b"current-secret"
+ active = [secret(key)]
+ with tempfile.TemporaryDirectory() as directory:
+ inbox = SQLiteInbox(str(Path(directory) / "inbox.sqlite"))
+ handled: list[str] = []
+ connection = sqlite3.connect(inbox.path)
+ connection.execute("CREATE TABLE business_events (event_type TEXT NOT NULL)")
+ connection.commit()
+ connection.close()
+
+ def succeed(_: sqlite3.Connection, event: dict) -> None:
+ handled.append(event["type"])
+
+ duplicate_headers = headers(key, delivery_id="msg_duplicate")
+ self.assertEqual(
+ 204,
+ receive(BODY, duplicate_headers, active, inbox, authorize, succeed, NOW),
+ )
+ self.assertEqual(
+ 204,
+ receive(BODY, duplicate_headers, active, inbox, authorize, succeed, NOW),
+ )
+ self.assertEqual(["demo.follow_up"], handled)
+ self.assertEqual(
+ 400,
+ receive(
+ BODY + b" ",
+ duplicate_headers,
+ active,
+ inbox,
+ authorize,
+ succeed,
+ NOW,
+ ),
+ )
+ self.assertEqual(
+ 500,
+ receive(BODY, duplicate_headers, [], inbox, authorize, succeed, NOW),
+ )
+
+ unauthorized = json.dumps(
+ {
+ "type": "demo.follow_up",
+ "data": {
+ "organization_id": "org_2",
+ "project_id": "project_1",
+ },
+ }
+ ).encode()
+ self.assertEqual(
+ 403,
+ receive(
+ unauthorized,
+ headers(key, delivery_id="msg_wrong_tenant", body=unauthorized),
+ active,
+ inbox,
+ authorize,
+ succeed,
+ NOW,
+ ),
+ )
+ self.assertEqual(["demo.follow_up"], handled)
+
+ attempts = 0
+
+ def fail_once(connection: sqlite3.Connection, event: dict) -> None:
+ nonlocal attempts
+ attempts += 1
+ connection.execute(
+ "INSERT INTO business_events(event_type) VALUES (?)",
+ (event["type"],),
+ )
+ if attempts == 1:
+ raise RuntimeError("retry me")
+
+ retry_headers = headers(key, delivery_id="msg_retry")
+ self.assertEqual(
+ 500,
+ receive(BODY, retry_headers, active, inbox, authorize, fail_once, NOW),
+ )
+ connection = sqlite3.connect(inbox.path)
+ self.assertEqual([], connection.execute("SELECT * FROM business_events").fetchall())
+ self.assertEqual(
+ [],
+ connection.execute(
+ "SELECT * FROM webhook_inbox WHERE webhook_id = 'msg_retry'"
+ ).fetchall(),
+ )
+ connection.close()
+ self.assertEqual(
+ 204,
+ receive(BODY, retry_headers, active, inbox, authorize, fail_once, NOW),
+ )
+ self.assertEqual(2, attempts)
+ connection = sqlite3.connect(inbox.path)
+ self.assertEqual(
+ [("demo.follow_up",)],
+ connection.execute("SELECT event_type FROM business_events").fetchall(),
+ )
+ self.assertEqual(
+ [("msg_retry",)],
+ connection.execute(
+ "SELECT webhook_id FROM webhook_inbox WHERE webhook_id = 'msg_retry'"
+ ).fetchall(),
+ )
+ connection.close()
+
+ def test_distinct_ids_fan_out_and_concurrent_duplicates_run_once(self) -> None:
+ key = b"current-secret"
+ active = [secret(key)]
+ with tempfile.TemporaryDirectory() as directory:
+ inbox = SQLiteInbox(str(Path(directory) / "inbox.sqlite"))
+ handled: list[str] = []
+ lock = threading.Lock()
+
+ def succeed(_: sqlite3.Connection, event: dict) -> None:
+ with lock:
+ handled.append(event["type"])
+
+ for delivery_id in ("msg_a", "msg_b"):
+ self.assertEqual(
+ 204,
+ receive(
+ BODY,
+ headers(key, delivery_id=delivery_id),
+ active,
+ inbox,
+ authorize,
+ succeed,
+ NOW,
+ ),
+ )
+ self.assertEqual(2, len(handled))
+
+ barrier = threading.Barrier(3)
+ statuses: list[int] = []
+
+ def send_duplicate() -> None:
+ barrier.wait()
+ status = receive(
+ BODY,
+ headers(key, delivery_id="msg_race"),
+ active,
+ inbox,
+ authorize,
+ succeed,
+ NOW,
+ )
+ with lock:
+ statuses.append(status)
+
+ threads = [threading.Thread(target=send_duplicate) for _ in range(2)]
+ for thread in threads:
+ thread.start()
+ barrier.wait()
+ for thread in threads:
+ thread.join()
+
+ self.assertEqual([204, 204], sorted(statuses))
+ self.assertEqual(3, len(handled))
+
+
+if __name__ == "__main__":
+ unittest.main()