Build an Event Pipeline

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.

Choose a push channel

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 on the raw body and answering 200 within a few seconds:

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 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.

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 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 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.
• 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.