> ## Documentation index
> Fetch the complete documentation index at: https://developer.rhombus.com/llms.txt
> Use it to discover all available pages before exploring further.

# Build an Event Pipeline

> Get every Rhombus alert into your own systems without gaps: webhooks or the event WebSocket for speed, a reconciliation job over the REST API for completeness, and de-duplication between them.

Rhombus pushes events as they happen, but neither push channel keeps a backlog: a webhook gets one delivery attempt, and the event WebSocket doesn't replay what you missed while disconnected. A pipeline that must not lose events therefore has two paths:

1. **A push path for speed**: webhooks or the event WebSocket deliver new alerts within seconds.
2. **A reconciliation path for completeness**: a scheduled job reads recent alerts from the REST API and fills any gaps.

Both paths write to the same store, keyed by the alert's `uuid`, so an alert that arrives twice is stored once.

```mermaid
flowchart LR
    R[Rhombus] -- webhook or WebSocket --> I[Receiver]
    I --> Q[Queue]
    J[Reconciliation job] -- getPolicyAlerts --> R
    J --> Q
    Q --> W[Worker: skip seen uuids] --> S[(Your store)]
```

## Choose a push channel

|                         | Webhooks                                                                                           | Event WebSocket                                                        |
| ----------------------- | -------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------- |
| Connection              | Rhombus calls your HTTPS endpoint                                                                  | Your process keeps a STOMP connection open                             |
| Needs a public endpoint | Yes                                                                                                | No (outbound connection only)                                          |
| Missed events           | Not resent; one attempt per event                                                                  | Not replayed after a disconnect                                        |
| Content                 | Rule webhooks: the rule's events. Organization webhooks: every Activity or Diagnostic notification | Changes across the organization, including policy alerts               |
| Set up                  | [Webhooks](https://developer.rhombus.com/webhooks)                                                 | [WebSocket overview](https://developer.rhombus.com/websocket/overview) |

Either way, the reconciliation job below is what makes the pipeline complete.

## Receive and queue

Whichever channel you use, keep the receiver small: check the event is genuine, put it on a queue, and return. For webhooks that means [verifying the signature](https://developer.rhombus.com/webhooks#verify-a-request) on the raw body and answering `200` within a few seconds:

```python Python (Flask)
import hashlib, hmac, json, os
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["RHOMBUS_WEBHOOK_SECRET"].encode()

@app.post("/rhombus/webhook")
def webhook():
    body = request.get_data()  # raw bytes: the signature covers them exactly
    sent = request.headers.get("x-rhombus-signature-sha256", "")
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sent, expected):  # constant time
        abort(401)
    queue.put(json.loads(body))  # your queue: SQS, Pub/Sub, Kafka, a database table
    return "", 200
```

Rule webhook actions and organization webhooks both send `x-rhombus-signature-sha256`, so the same check covers either kind.

## Reconcile on a schedule

Every few minutes, read the alerts since your last run with [`getPolicyAlerts`](https://developer.rhombus.com/api-reference/event-webservice/get-recent-policy-alerts-for-organization) and queue each one. Start a little before the previous run's high-water mark so events near the boundary aren't missed; the worker drops the ones you already have.

```python Python
import os, time, requests

URL = "https://api2.rhombussystems.com/api/event/getPolicyAlerts"
HEADERS = {"x-auth-scheme": "api-token", "x-auth-apikey": os.environ["RHOMBUS_API_KEY"]}
OVERLAP_MS = 10 * 60 * 1000

def reconcile(since_ms):
    """Queue every alert newer than since_ms, newest first; returns the newest timestamp seen."""
    body = {"afterTimestampMs": since_ms - OVERLAP_MS, "maxResults": 500}
    newest = since_ms
    while True:
        page = requests.post(URL, headers=HEADERS, json=body).json()
        if page.get("error"):
            raise RuntimeError(page.get("errorMsg"))
        alerts = page.get("policyAlerts", [])
        if not alerts:
            return newest
        for alert in alerts:
            queue.put(alert)
            newest = max(newest, alert["timestampMs"])
        body["lastTimestampMs"] = alerts[-1]["timestampMs"]  # continue after the last alert
        body["lastUuid"] = alerts[-1]["uuid"]
```

Store the returned timestamp and pass it to the next run. [Pagination](https://developer.rhombus.com/pagination#policy-alerts) explains the `lastTimestampMs` and `lastUuid` cursor.

## De-duplicate in the worker

The worker processes each queued alert once, keyed by `uuid`:

- **Skip alerts you've stored.** Keep the `uuid` of every alert you've processed, for at least as long as your reconciliation overlap.
- **Order by `timestampMs`**, not by arrival: the push path and the reconciliation job deliver in different orders.
- **Keep the latest version.** The push path and the reconciliation job can deliver different versions of the same alert; when a `uuid` arrives again, update your stored copy rather than discarding it.

## Run it in production

- **Use a dedicated API key** with a read-only role covering the locations you collect, so its access and its [audit trail](https://developer.rhombus.com/authentication/permissions#audit-trail) are its own.
- **Stay under your rate limit.** A reconciliation run makes one request per page; large pages and a schedule of a few minutes keep it far below the [organization limit](https://developer.rhombus.com/rate-limits).
- **Alert on the pipeline itself**: a reconciliation run that fails, or a run that finds many alerts the push path missed, means your receiver or connection needs attention.
- **Reconnect the WebSocket** after any disconnect and run a reconciliation for the gap; see [Connection lifecycle](https://developer.rhombus.com/websocket/connection-lifecycle).