Webhooks: idempotency, failures and catching up

# modtager.py (FastAPI): verificér, dedupliker, gem i en indbakke, svar 204
import hashlib
import hmac
import json
import os
import sqlite3

from fastapi import FastAPI, Header, HTTPException, Request, Response

app = FastAPI()
SECRET = os.environ["DANADRESSE_WEBHOOK_SECRET"]

db = sqlite3.connect("adresser.db", check_same_thread=False)
db.execute("""CREATE TABLE IF NOT EXISTS webhook_inbox (
    entity        TEXT    NOT NULL,
    sekvensnummer INTEGER NOT NULL,
    body          TEXT    NOT NULL,
    modtaget      TEXT    NOT NULL DEFAULT (datetime('now')),
    behandlet     INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (entity, sekvensnummer)
)""")
db.commit()


@app.post("/hooks/danadresse", status_code=204)
async def hook(request: Request, x_danadresse_signature: str | None = Header(default=None)) -> Response:
    raw = await request.body()
    expected = hmac.new(SECRET.encode(), raw, hashlib.sha256).hexdigest()
    if not x_danadresse_signature or not hmac.compare_digest(
            expected.encode(), x_danadresse_signature.strip().lower().encode()):
        raise HTTPException(status_code=401)

    p = json.loads(raw)
    # Primærnøglen (entity, sekvensnummer) gør det idempotent: en gentaget
    # levering rammer INSERT OR IGNORE og ændrer intet.
    db.execute("INSERT OR IGNORE INTO webhook_inbox (entity, sekvensnummer, body) VALUES (?, ?, ?)",
               (p["entity"], p["sekvensnummer"], raw.decode()))
    db.commit()
    return Response(status_code=204)  # hurtigt svar: arbejdet sker i worker.py

# worker.py: behandl indbakken ved at hente hændelser fra din egen cursor.
# Genbruger synk.py fra guiden "Replikering, del 2". Kør fx hvert minut fra cron.
import sqlite3

from synk import ENTITETER, synk_entitet

db = sqlite3.connect("adresser.db")

ventende = db.execute(
    "SELECT entity, max(sekvensnummer) FROM webhook_inbox WHERE behandlet = 0 GROUP BY entity"
).fetchall()

for entity, hoejeste in ventende:
    if entity not in ENTITETER:
        continue
    # Hent ALT fra din cursor op til leveringens sekvensnummer, ikke kun de id'er,
    # der stod i `changes`. Så fanger du også det, en tabt levering skulle have sagt.
    antal = synk_entitet(db, entity, ENTITETER[entity], hoejeste)
    db.execute("UPDATE webhook_inbox SET behandlet = 1 WHERE entity = ? AND sekvensnummer <= ?",
               (entity, hoejeste))
    db.commit()
    print(f"{entity}: {antal} hændelser til og med {hoejeste}")

# hvert minut: behandl indbakken (ingen API-kald, når den er tom)
* * * * *  cd /srv/adresser && flock -n worker.lock python worker.py >> worker.log 2>&1
# hver time: fuld indhentning som sikkerhedsnet
7 * * * *  cd /srv/adresser && flock -n synk.lock python synk.py >> synk.log 2>&1

# overvaag_webhooks.py: kør dagligt; exit 1 (og en linje pr. problem), hvis noget er galt
import datetime as dt
import os
import sys

import requests

BASE_URL = os.environ.get("DANADRESSE_BASE_URL", "https://api.danadresse.dk")
STILLE_TIMER = 48  # alarmér, hvis en aktiv webhook ikke har fået noget i så lang tid

token = requests.post(f"{BASE_URL}/api/v1/auth/login", timeout=10, json={
    "email": os.environ["DANADRESSE_EMAIL"],
    "password": os.environ["DANADRESSE_PASSWORD"],
}).json()["access_token"]

r = requests.get(f"{BASE_URL}/api/v1/webhooks", headers={"Authorization": f"Bearer {token}"}, timeout=10)
r.raise_for_status()

problemer = []
nu = dt.datetime.now(dt.timezone.utc)
for w in r.json()["webhooks"]:
    navn = f"{w['url']} ({','.join(w['entities'])})"
    if not w["is_active"]:
        problemer.append(f"SLÅET FRA efter {w['failure_count']} fejl i træk: {navn}. Slet og opret igen.")
    elif w["failure_count"] > 0:
        problemer.append(f"{w['failure_count']} fejl i træk: {navn}")
    if w["is_active"] and w["last_delivery_at"]:
        sidst = dt.datetime.fromisoformat(w["last_delivery_at"])
        if sidst.tzinfo is None:
            sidst = sidst.replace(tzinfo=dt.timezone.utc)
        if nu - sidst > dt.timedelta(hours=STILLE_TIMER):
            problemer.append(f"ingen levering siden {sidst:%Y-%m-%d %H:%M} UTC: {navn}")

for p in problemer:
    print(p)
sys.exit(1 if problemer else 0)