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..ac9554937 --- /dev/null +++ b/app/en/build/eventing/page.mdx @@ -0,0 +1,589 @@ +--- +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. + +## 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" +``` + + + + +## Connect a customer-owned realtime provider + +Provider ingress lets agent developers route signed Slack or GitHub events into the existing trigger runtime without operating a separate receiver. Arcade supports one realtime trigger type per project-owned OAuth app for this release. + +Before setup, you need: + +- A Slack or GitHub OAuth app in the selected Arcade project +- An active connected account for that app +- A current realtime trigger type: `slack.message.received` or `github.push.received` +- A provider-reachable HTTPS origin for the Arcade Engine + +Arcade Cloud supplies the public origin. For customer-managed or local deployments, your platform operator must expose the Arcade Engine through HTTPS and set `api.public_host` to the public hostname. Private-only deployments must use polling triggers. + + +The provider callback receives events from Slack or GitHub. It is not the OAuth redirect URI and it is not an outgoing webhook subscription. Arcade does not create the provider-side event subscription or webhook for you. + + + + + + + +### Configure provider ingress + +Open **Auth providers**, select your Slack or GitHub app, and find **Provider ingress**. Choose the realtime trigger type and the connected account that Arcade should use for trigger runs. + +For Slack, enter the app's signing secret. For GitHub, choose whether the webhook belongs to a GitHub App installation, repository, or organization, then enter the webhook secret that you will use in GitHub. + +Select **Configure ingress**, then copy the provider callback URL. Arcade encrypts the secret and does not show it again. + +### Configure the provider + +For Slack, paste the provider callback into **Event Subscriptions > Request URL** and subscribe to message events. Slack sends a signed URL-verification challenge. + +For GitHub, paste the provider callback into **Payload URL**, choose `application/json`, use the same webhook secret, and subscribe to push events. The target kind in Arcade must match where you create the webhook. + +### Confirm the first delivery + +Send a matching Slack message or GitHub push. Return to **Provider ingress** and confirm that **Latest verified delivery** has a timestamp and **Current signing secret** is **Verified**. + +The first matching event pins the Slack workspace or selected GitHub target. A signed event from another target is acknowledged but does not enter Arcade event routing. + + + + + + +Read the provider ingress state to discover current eligible types: + +```bash +export AUTH_PROVIDER_ID="" +export CONNECTION_ID="" + +curl --fail-with-body --silent --show-error \ + --header "Authorization: Bearer $ARCADE_API_KEY" \ + "$SCOPE/auth_providers/$AUTH_PROVIDER_ID/ingress" +``` + +Configure Slack ingress with the current type and version returned by that read: + +```bash +export PROVIDER_SIGNING_SECRET="" + +curl --fail-with-body --silent --show-error \ + --request PUT "$SCOPE/auth_providers/$AUTH_PROVIDER_ID/ingress" \ + --header "Authorization: Bearer $ARCADE_API_KEY" \ + --header "Content-Type: application/json" \ + --data "{\"type\":\"slack.message.received\",\"type_version\":\"1\",\"connection_id\":\"$CONNECTION_ID\",\"signing_secret\":\"$PROVIDER_SIGNING_SECRET\"}" +``` + +The response contains `callback_url` but never contains the signing secret. For GitHub, use `github.push.received` and include `external_scope_kind` with `installation`, `repository`, or `organization`. + +Paste `callback_url` into the provider settings described in the Dashboard tab. After the provider sends a signed challenge, ping, or matching event, repeat the GET request and inspect `last_verified_at` and `current_secret_verified`. + +To change only the connected account, send the stored type and version with the new `connection_id` and `"preserve_signing_secret":true`. To rotate the secret, send `signing_secret` instead. Both operations keep the callback URL and pinned provider target. + +Remove provider ingress explicitly before deleting the auth provider or changing its Slack or GitHub classification: + +```bash +curl --fail-with-body --silent --show-error \ + --request DELETE "$SCOPE/auth_providers/$AUTH_PROVIDER_ID/ingress" \ + --header "Authorization: Bearer $ARCADE_API_KEY" +``` + +Removal keeps the OAuth app, connected accounts, trigger instances, Arcade events, and outgoing webhook subscriptions. + + + + +### Recover provider ingress + +- **Rotate a secret:** pause delivery in Slack or GitHub, update the provider and Arcade with the same secret, resume delivery, then send a new event. Do not set the Arcade auth provider to Disabled for this cutover. That setting acknowledges and drops deliveries instead of buffering them. +- **Replace a connected account:** choose another active account for the same provider and project. Matching deliveries return `503` while no usable account is selected so the provider can retry. +- **Fix a public host:** restore the provider-reachable HTTPS origin. Existing configurations can still rotate a secret or change an account while the host check reports `public_host_required`, and a previously copied callback continues to route. +- **Reset a pinned target:** remove and recreate provider ingress. Slack Enterprise Grid customers need one provider configuration per workspace. GitHub customers must recreate ingress to change the installation, repository, or organization target kind after the first event pins it. + +Provider ingress accepts request bodies up to 5 MiB and suppresses a provider redelivery for seven days within one configuration lifetime. Removing and recreating ingress starts a new configuration lifetime. Deliveries already staged before removal continue through normalization and event routing from their stored snapshot. + + +Arcade Cloud applies edge controls to the public callback. Before exposing a customer-managed or local deployment, configure request-rate and byte-rate limits at your reverse proxy or load balancer. Keep the callback route public, but do not expose private management endpoints through that exception. + + +## 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/public/llms.txt b/public/llms.txt index b0301a8a8..50efe208e 100644 --- a/public/llms.txt +++ b/public/llms.txt @@ -1,4 +1,4 @@ - + # Arcade @@ -93,6 +93,7 @@ Arcade docs serve two audiences. Start with the path that matches your goal: - [Build an AI agent with Arcade and Spring AI](https://docs.arcade.dev/en/get-started/agent-frameworks/springai): Documentation page - [Build an AI Chatbot with Arcade and TanStack AI](https://docs.arcade.dev/en/get-started/agent-frameworks/tanstack-ai): This documentation page guides users through the process of building a browser-based AI chatbot using Arcade tools and TanStack AI, enabling integration with Gmail and Slack for seamless communication. Users will learn how to set up a TanStack Start project, manage chat state, - [Build an AI Chatbot with Arcade and Vercel AI SDK](https://docs.arcade.dev/en/get-started/agent-frameworks/vercelai): This documentation page guides users through the process of building a browser-based AI chatbot using the Vercel AI SDK and Arcade tools for Gmail and Slack integration. Users will learn how to set up a Next.js project, manage chat state, and implement authorization +- [Build event-driven integrations](https://docs.arcade.dev/en/build/eventing): Documentation page - [Build MCP Server QuickStart](https://docs.arcade.dev/en/get-started/quickstarts/mcp-server-quickstart): The "Build MCP Server QuickStart" documentation provides a step-by-step guide for users to create and run a custom MCP Server using the Arcade MCP framework. It covers prerequisites, installation of necessary tools, server setup, and how to implement and call various - [Build with Arcade](https://docs.arcade.dev/en/build): Documentation page - [Build Your Own Contextual Access Server](https://docs.arcade.dev/en/operate/governance/contextual-access/build-your-own): Documentation page diff --git a/tests/eventing-guide.test.ts b/tests/eventing-guide.test.ts new file mode 100644 index 000000000..104c51270 --- /dev/null +++ b/tests/eventing-guide.test.ts @@ -0,0 +1,159 @@ +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 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 RETENTION_WINDOW_RE = + /Keep each recorded `webhook-id` for at least Arcade's configured event-retention period \(90 days by default\)/; +const STABLE_WEBHOOK_ID_RE = + /same `webhook-id` across automatic retries, manual retry, and recovery/; + +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("## Connect a customer-owned realtime provider"); + 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", + "AUTH_PROVIDER_ID", + ]) { + expect(page).toContain(`export ${variable}=`); + } + }); + + test("documents provider ingress setup, proof, recovery, and boundaries", () => { + for (const value of [ + "slack.message.received", + "github.push.received", + '"preserve_signing_secret":true', + "current_secret_verified", + "last_verified_at", + "public_host_required", + "request-rate and byte-rate limits", + "seven days", + "5 MiB", + ]) { + expect(page).toContain(value); + } + expect(page).toContain("$SCOPE/auth_providers/$AUTH_PROVIDER_ID/ingress"); + expect(page).toContain("It is not the OAuth redirect URI"); + expect(page).toContain("acknowledges and drops deliveries"); + }); + + 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 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(STABLE_WEBHOOK_ID_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()