# Runs started by events

> Start a durable run from a webhook, an S3 notification or an inbound email, with an idempotency key derived from the event so a redelivered event does not run twice.

Source: https://docs.pyrula.io/getting-started/recipes/event-triggered-starts/

Webhooks are delivered at least once: the sender retries until it sees a 2xx, so the same event
can arrive twice. The route that receives one does three things: check it came from the sender,
submit a run with an idempotency key derived from the event's id, and return quickly. `submit`
only queues the run, so the response does not wait for the work.

The route below sits beside the one `pyrula init --fastapi` writes
([Start from a scaffold](https://docs.pyrula.io/getting-started/existing-app.md#start-from-a-scaffold)); in that file,
call `worker.client.submit` instead of building a second client. This version builds its own
client so it runs on its own.

```python
import hashlib
import hmac
import json
import tempfile

from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.testclient import TestClient
from pyrula.workflows import WorkflowClient, open_store, workflow

SECRET = b"example-signing-secret"  # from your secret manager, never the source


@workflow(name="docs_recipe_process_document")
async def process_document(ctx, document_id: str) -> dict:
    return await ctx.step("extract", lambda: {"document_id": document_id})


# A fresh file per run of this example; an app keeps one store path for good.
store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/events.db")
client = WorkflowClient(store)
app = FastAPI()


def verify(body: bytes, signature: str) -> None:
    # Most senders sign the raw body with a shared secret; follow your sender's scheme exactly.
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Unauthorized")


@app.post("/webhooks/documents", status_code=202)
async def document_received(request: Request, x_signature: str = Header(default="")) -> dict:
    body = await request.body()
    verify(body, x_signature)
    try:
        event = json.loads(body)
        event_id, document_id = event["id"], event["document"]["id"]
    except (ValueError, KeyError, TypeError) as err:
        raise HTTPException(status_code=400, detail="Bad event") from err
    run = await client.submit(
        "docs_recipe_process_document",
        {"document_id": document_id},
        idempotency_key=f"document-event:{event_id}",
    )
    return {"run_id": run.run_id}


def post(http: TestClient, event: dict, secret: bytes = SECRET):
    body = json.dumps(event).encode()
    signature = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return http.post("/webhooks/documents", content=body, headers={"X-Signature": signature})


with TestClient(app) as http:
    event = {"id": "evt_1", "document": {"id": "doc_9"}}
    first = post(http, event).json()
    assert post(http, event).json() == first  # the sender retried: same run
    assert post(http, {**event, "id": "evt_2"}).json() != first  # a new event: a new run
    assert post(http, event, secret=b"wrong").status_code == 401
    assert http.post("/webhooks/documents", content=b"{}", headers={"X-Signature": hmac.new(
        SECRET, b"{}", hashlib.sha256).hexdigest()}).status_code == 400
```

A key lives 24 hours. A redelivery after that starts a second run, and within the window a repeat
returns the first run even if that run failed ([Starting and talking to
runs](https://docs.pyrula.io/workflows/guides/client.md)). If a sender retries for longer than that, record handled event ids yourself.

## What to use as the key

The key must be the same for every delivery of one event and different for different events.

| Source | Key from | Check |
|---|---|---|
| A SaaS webhook | The event id in the payload or headers | The sender's docs say whether the id is stable across retries |
| Inbound email through a mail-to-webhook service | The message's `Message-ID` header, or the service's own event id | Redelivery by the service keeps the same one |
| S3 notification | Bucket, object key and version id (`sequencer` on an unversioned bucket) | S3 notifications reach HTTP through SNS or EventBridge; the record has no single event id |

This table is a starting point, not a tested integration: Pyrula has no S3 or email code, and the
claims in the last column are the senders' to confirm. A key made of fields that identify the
work, such as `invoice:{invoice_id}:{version}`, is as good as an event id and survives a sender
that mints a new event id per retry.

## Django, Flask and other sync code

A sync view uses the blocking client in `pyrula.workflows.sync`. Create it once per process and
share it between threads ([From sync code](https://docs.pyrula.io/workflows/guides/client.md#from-sync-code)). The
function that starts the run is the same whatever the framework:

```python
import tempfile

from pyrula.workflows import open_store
from pyrula.workflows.sync import WorkflowClient

store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/sync-events.db")
client = WorkflowClient(store)


def start_from_event(event: dict) -> str:
    run = client.submit(
        "docs_recipe_process_document",
        {"document_id": event["document"]["id"]},
        idempotency_key=f"document-event:{event['id']}",
    )
    return run.run_id


event = {"id": "evt_1", "document": {"id": "doc_9"}}
assert start_from_event(event) == start_from_event(event)
client.close()
```

Called from a Flask view, with the signature check done first:

```python
from flask import Flask, abort, jsonify, request

app = Flask(__name__)


@app.post("/webhooks/documents")
def document_received():
    if not signature_is_valid(request.get_data(), request.headers.get("X-Signature", "")):
        abort(401)
    return jsonify(run_id=start_from_event(request.get_json())), 202
```

`signature_is_valid` is your check, like `verify` above. A Django view is the same call. The
sync client cannot be used from a thread running an event loop: use the async client there.

## What this does not do

- No Pyrula code reads a bucket, a mailbox or a queue. A trigger that polls one belongs in a
  worker process or a scheduled run that calls `submit`. For Kafka, see
  [Kafka sources](https://docs.pyrula.io/workflows/guides/kafka.md).
- The route is yours, so its authentication is too. The run endpoints that `RunRouter` serves
  take their own `AuthConfig` ([Authentication](https://docs.pyrula.io/workflows/guides/http.md#authentication)).