Crear una canalización de eventos

Rhombus envía los eventos en cuanto ocurren, pero ningún canal de envío guarda un historial: un webhook tiene un solo intento de entrega, y el WebSocket de eventos no reenvía lo que te perdiste mientras estabas desconectado. Por eso, una canalización que no puede perder eventos tiene dos vías:

1. Una vía de envío para la rapidez: los webhooks o el WebSocket de eventos entregan las alertas nuevas en segundos.
2. Una vía de conciliación para la integridad: una tarea programada lee las alertas recientes con la API REST y completa los huecos.

Ambas vías escriben en el mismo almacén, con el uuid de la alerta como clave, así que una alerta que llega dos veces se guarda una sola vez.

Elige un canal de envío

En ambos casos, la tarea de conciliación de abajo es lo que completa la canalización.

Recibe y encola

Uses el canal que uses, mantén el receptor pequeño: comprueba que el evento es auténtico, ponlo en una cola y responde. En los webhooks eso significa verificar la firma sobre el cuerpo sin procesar y responder 200 en pocos segundos:

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

Las acciones de webhook de reglas y los webhooks de la organización envían x-rhombus-signature-sha256, así que la misma verificación sirve para ambos tipos.

Concilia de forma programada

Cada pocos minutos, lee las alertas desde la ejecución anterior con getPolicyAlerts y encola cada una. Empieza un poco antes de la marca de la ejecución anterior para no perder eventos cerca del límite; el trabajador descarta las que ya tienes.

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"]

Guarda la marca de tiempo devuelta y pásala a la siguiente ejecución. Paginación explica el cursor lastTimestampMs y lastUuid.

Elimina duplicados en el trabajador

El trabajador procesa cada alerta encolada una sola vez, con su uuid como clave:

• Omite las alertas que ya guardaste. Conserva el uuid de cada alerta procesada al menos durante el margen de tu conciliación.
• Ordena por timestampMs, no por orden de llegada: la vía de envío y la tarea de conciliación entregan en distinto orden.
• Conserva la versión más reciente. La vía de envío y la tarea de conciliación pueden entregar versiones distintas de la misma alerta; cuando un uuid vuelve a llegar, actualiza tu copia en lugar de descartarla.

Ponlo en producción

• Usa una clave de API dedicada con un rol de solo lectura que cubra las ubicaciones que recopilas, para que su acceso y su registro de auditoría sean propios.
• Mantente por debajo de tu límite de tasa. Cada ejecución de conciliación hace una solicitud por página; páginas grandes y una frecuencia de unos minutos la mantienen muy por debajo del límite de la organización.
• Vigila la propia canalización: una ejecución de conciliación que falla, o una que encuentra muchas alertas que la vía de envío no entregó, indica que tu receptor o tu conexión necesitan atención.
• Reconecta el WebSocket después de cualquier desconexión y ejecuta una conciliación para el hueco; consulta Ciclo de vida de la conexión.