Ls Notif

CONCEPT — Let op: Deze skill is geen officieel product van Logius. De beschrijvingen zijn informatieve samenvattingen — niet de officiële standaarden zelf. De definities op forumstandaardisatie.nl en Logius zijn altijd leidend. Overheidsorganisaties die generatieve AI inzetten dienen te voldoen aan het Overheidsbreed standpunt voor de inzet van generatieve AI. Zie DISCLAIMER.md en onze verantwoording.

CloudEvents & Notificaties

Agent-instructie: Deze skill helpt bij het implementeren van event-driven communicatie met CloudEvents NL GOV profiel. Gebruik de cloudevents SDK voor Python/JavaScript. Source moet URN nld-notatie gebruiken: urn:nld:oin:<OIN>:systeem:<naam>.

De notificatiestandaarden definiëren hoe overheidsorganisaties gebeurtenissen (events) kunnen publiceren en daarop kunnen abonneren. Gebaseerd op de internationale CNCF CloudEvents specificatie (v1.0.1) met een NL GOV profiel dat specifieke eisen stelt aan naamgeving, identificatie en het gebruik van context-attributen binnen de Nederlandse overheid.

Versiemodel

Net als andere Logius-standaarden kennen de notificatiestandaarden twee publicatiekanalen (vergelijkbaar met W3C):

• Vastgestelde versie (DEF): officieel goedgekeurd, gepubliceerd op gitdocumentatie.logius.nl
• Werkversie (draft): werk-in-uitvoering, gepubliceerd op logius-standaarden.github.io

Het NL GOV profiel voor CloudEvents heeft een vastgestelde versie (v1.1 DEF). De Guidelines zijn een toelichting zonder eigen publicatie. Notificatieservices is een werkrepository zonder publicatie. Abonneren heeft een werkversie (v0.0.1 WV) op gitdocumentatie maar is nog niet vastgesteld.

Op het Forum Standaardisatie staat CloudEvents v1.0 als verplicht ('pas-toe-of-leg-uit', goedgekeurd OBDO 25-11-2025). Versie v1.1 is de meest recente DEF van Logius.

Repositories

¹ Abonneren WV v0.0.1 is gepubliceerd op gitdocumentatie.logius.nl, wat normaal voor vastgestelde versies (DEF/VV) is. Dit is een uitzondering; de standaard is nog in ontwikkeling.

CloudEvents NL GOV Profiel

Het NL GOV profiel scherpt de CNCF CloudEvents specificatie aan voor gebruik binnen de Nederlandse overheid. Hieronder de volledige set verplichte en optionele attributen.

Verplichte attributen

Optionele attributen

Notificatieservices API

De Notificatieservices API is gespecificeerd als OpenAPI 3.x en biedt een centraal endpoint voor het publiceren van events.

Event publiceren

POST /api/v1/notifications
Authorization: Bearer <JWT>
Content-Type: application/json

{
  "id": "f3dce042-cd6e-4977-844d-05be8dce7cea",
  "source": "urn:nld:oin:00000001823288444000:systeem:BRP-component",
  "specversion": "1.0",
  "type": "nl.brp.persoon-verhuisd",
  "subject": "999990342",
  "time": "2024-01-15T10:30:00Z",
  "datacontenttype": "application/json",
  "data": {
    "oud_adres": "Keizersgracht 100, Amsterdam",
    "nieuw_adres": "Herengracht 200, Amsterdam"
  }
}

Abonneren

Het Abonneren-component beschrijft hoe afnemers zich kunnen registreren om specifieke events te ontvangen. Twee hoofdmodellen worden onderscheiden:

Push-model (aanbevolen): Event-driven aflevering via webhooks. De notificatieservice stuurt events actief naar een door de afnemer opgegeven endpoint. Dit is het voorkeursmodel vanwege lage latency en efficiënter resourcegebruik.

Pull-model: Polling-gebaseerde aflevering. De afnemer bevraagt periodiek de notificatieservice op nieuwe events. Geschikt voor situaties waar de afnemer geen inkomend endpoint kan aanbieden (bijv. vanwege firewallrestricties).

Let op: De huidige Notificatieservices OpenAPI-specificatie bevat alleen push-endpoints voor events (POST /events). Er zijn geen pull/polling-endpoints voor het ophalen van events. De Abonneren-werkversie beschrijft push als het voorkeursmodel; pull/polling wordt alleen genoemd als het onwenselijke alternatief dat de abonnementsfunctie overbodig maakt — het is niet uitgewerkt als een ondersteund aflevermodel.

Keuzehulp: Push of Pull?

Kan de afnemer een bereikbaar HTTPS endpoint aanbieden?
  JA  → Push-model (webhook) — aanbevolen, lage latency
  NEE → Pull-model (polling) — nog niet gespecificeerd in de API, check status in de Abonneren-repo

Security & Privacy

Bij het werken met notificaties gelden strikte eisen op het gebied van beveiliging en privacy:

• Geen gevoelige data in context-attributen: Context-attributen (zoals source, type, subject) zijn inspecteerbaar en worden gelogd door tussenliggende systemen. Plaats nooit persoonsgegevens of gevoelige informatie in deze velden. Gebruik hiervoor de versleutelde data-payload.
• Versleuteling van event data: Wanneer de payload gevoelige gegevens bevat, moet deze versleuteld worden (end-to-end encryptie). De context-attributen blijven leesbaar voor routering.
• TLS verplicht: Alle communicatie tussen bronnen, notificatieservices en afnemers moet plaatsvinden over TLS (minimaal versie 1.2).
• Claim Check Pattern: Bij grote of gevoelige payloads kan het dataref-attribuut verwijzen naar een beveiligd endpoint. De afnemer haalt de payload separaat op met eigen autorisatie.

Implementatievoorbeelden

Event Publiceren met CloudEvents SDK (Python)

from cloudevents.http import CloudEvent
from cloudevents.conversion import to_json
import requests, uuid
from datetime import datetime, timezone

# CloudEvent aanmaken conform NL GOV profiel
event = CloudEvent({
    "id": str(uuid.uuid4()),
    "source": "urn:nld:oin:00000001823288444000:systeem:BRP-component",
    "type": "nl.brp.persoon-verhuisd",
    "specversion": "1.0",
    "subject": "999990342",  # BSN van betrokkene
    "time": datetime.now(timezone.utc).isoformat(),
    "datacontenttype": "application/json",
}, data={
    "oud_adres": {"straat": "Keizersgracht", "huisnummer": "100", "plaats": "Amsterdam"},
    "nieuw_adres": {"straat": "Herengracht", "huisnummer": "200", "plaats": "Amsterdam"},
})

# Publiceer naar notificatieservice
response = requests.post(
    "https://notificaties.example.com/api/v1/notifications",
    headers={
        "Authorization": "Bearer eyJ...",
        "Content-Type": "application/cloudevents+json",
    },
    data=to_json(event),
)
print(f"Status: {response.status_code}")  # 200 OK

Webhook Ontvanger (FastAPI)

from fastapi import FastAPI, Request, HTTPException
from cloudevents.http import from_http

app = FastAPI(title="CloudEvents Webhook Ontvanger")

@app.post("/webhooks/cloudevents")
async def receive_event(request: Request):
    """Ontvang en verwerk CloudEvents conform NL GOV profiel."""
    body = await request.body()
    headers = dict(request.headers)

    try:
        event = from_http(headers, body)
    except Exception:
        raise HTTPException(400, "Ongeldig CloudEvent formaat")

    # Verplichte attributen valideren (NL GOV profiel)
    assert event["specversion"] == "1.0"
    assert event["source"].startswith("urn:nld:")  # URN met nld namespace
    assert "." in event["type"]  # Reverse Domain Name Notation

    # Event verwerken op basis van type
    match event["type"]:
        case "nl.brp.persoon-verhuisd":
            await verwerk_verhuizing(event)
        case "nl.kvk.inschrijving-gewijzigd":
            await verwerk_kvk_wijziging(event)
        case _:
            print(f"Onbekend event type: {event['type']}")

    return {"status": "accepted", "event_id": event["id"]}

async def verwerk_verhuizing(event):
    """Verwerk een BRP verhuizing notificatie."""
    subject = event["subject"]  # BSN
    data = event.data
    print(f"Persoon {subject} verhuisd naar {data['nieuw_adres']}")

Event Publiceren met curl

# CloudEvent publiceren in structured content mode
curl -X POST https://notificaties.example.com/api/v1/notifications \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "id": "f3dce042-cd6e-4977-844d-05be8dce7cea",
    "source": "urn:nld:oin:00000001823288444000:systeem:BRP-component",
    "type": "nl.brp.persoon-verhuisd",
    "subject": "999990342",
    "time": "2024-01-15T10:30:00Z",
    "datacontenttype": "application/json",
    "data": {"nieuw_adres": "Herengracht 200, Amsterdam"}
  }'

# Binary content mode (attributen als HTTP headers)
curl -X POST https://notificaties.example.com/api/v1/notifications \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -H "ce-specversion: 1.0" \
  -H "ce-id: f3dce042-cd6e-4977-844d-05be8dce7cea" \
  -H "ce-source: urn:nld:oin:00000001823288444000:systeem:BRP-component" \
  -H "ce-type: nl.brp.persoon-verhuisd" \
  -H "ce-subject: 999990342" \
  -H "ce-time: 2024-01-15T10:30:00Z" \
  -d '{"nieuw_adres": "Herengracht 200, Amsterdam"}'

Express.js Webhook Ontvanger

const express = require('express');
const { HTTP } = require('cloudevents');

const app = express();
app.use(express.json());

app.post('/webhooks/cloudevents', (req, res) => {
  try {
    const event = HTTP.toEvent({ headers: req.headers, body: req.body });

    // NL GOV profiel validatie
    if (!event.source.startsWith('urn:nld:')) {
      return res.status(400).json({ error: 'Source moet URN nld namespace gebruiken' });
    }

    console.log(`Event ontvangen: ${event.type} van ${event.source}`);
    console.log(`Subject: ${event.subject}, Data:`, event.data);

    // Verwerk event
    switch (event.type) {
      case 'nl.brp.persoon-verhuisd':
        handleVerhuizing(event);
        break;
      default:
        console.log(`Onbekend type: ${event.type}`);
    }

    res.status(200).json({ status: 'accepted', event_id: event.id });
  } catch (err) {
    res.status(400).json({ error: 'Ongeldig CloudEvent' });
  }
});

app.listen(8080, () => console.log('Webhook listener op poort 8080'));

Claim Check Pattern voor Grote Payloads

# Wanneer event data te groot is voor het bericht zelf,
# gebruik het dataref attribuut om naar een extern endpoint te verwijzen.

event = CloudEvent({
    "id": str(uuid.uuid4()),
    "source": "urn:nld:oin:00000001823288444000:systeem:DMS",
    "type": "nl.dms.document-beschikbaar",
    "specversion": "1.0",
    "subject": "doc-2024-001",
    "dataref": "https://dms.example.com/api/v1/documents/doc-2024-001",  # claim check
    # Geen 'data' veld - payload wordt separaat opgehaald door ontvanger
})

# Ontvanger haalt payload op via dataref met eigen autorisatie
document = requests.get(
    event["dataref"],
    headers={"Authorization": "Bearer receiver_token"},
    cert=("pkio_cert.pem", "pkio_key.pem"),
)

Foutafhandeling

Notificatieservice Error Responses

Webhook Delivery Retry Patroon

Bij het afleveren van events aan webhook endpoints geldt een retry-strategie:

import time

def deliver_event(webhook_url: str, event: dict, max_retries: int = 5):
    """Lever event af aan webhook met exponential backoff."""
    for attempt in range(max_retries):
        try:
            response = requests.post(webhook_url, json=event, timeout=10)
            if response.status_code == 200:
                return True
            if response.status_code >= 500:
                # Server error - retry met backoff
                wait = min(2 ** attempt * 5, 300)  # max 5 minuten
                time.sleep(wait)
                continue
            if response.status_code >= 400:
                # Client error - niet opnieuw proberen
                log.error(f"Webhook afgewezen: {response.status_code}")
                return False
        except requests.Timeout:
            wait = min(2 ** attempt * 5, 300)
            time.sleep(wait)
    return False  # Alle pogingen mislukt

Achtergrondinfo

Zie reference.md voor het type systeem, kernconcepten, en aanbevolen technologieën. Zie conflicts.md voor bronconflicten en gemaakte keuzes.