Merge pull request 'feat(event-subscriber): NRC event subscriber + rebuildable read projection (closes #7)' (#59) from feat/7-event-subscriber-projection into main
Some checks failed
CI / lint (push) Successful in 1m0s
CI / build (push) Successful in 46s
CI / unit (push) Successful in 53s
CI / mutation (push) Failing after 11m57s
CI / verify-stack (push) Successful in 4m57s

Reviewed-on: #59
This commit was merged in pull request #59.
This commit is contained in:
2026-06-30 13:56:59 +00:00
41 changed files with 1411 additions and 9 deletions

View File

@@ -8,6 +8,13 @@
"dotnet-stryker" "dotnet-stryker"
], ],
"rollForward": false "rollForward": false
},
"dotnet-ef": {
"version": "10.0.0",
"commands": [
"dotnet-ef"
],
"rollForward": false
} }
} }
} }

View File

@@ -62,6 +62,12 @@ jobs:
name: acl-mutation-report name: acl-mutation-report
path: services/acl/StrykerOutput/**/reports/mutation-report.html path: services/acl/StrykerOutput/**/reports/mutation-report.html
if-no-files-found: warn if-no-files-found: warn
- uses: https://github.com/actions/upload-artifact@v3
if: always()
with:
name: event-subscriber-mutation-report
path: services/event-subscriber/StrykerOutput/**/reports/mutation-report.html
if-no-files-found: warn
# One stage for every check that needs the live stack. On the single self-hosted # One stage for every check that needs the live stack. On the single self-hosted
# runner jobs run sequentially, so booting OpenZaak once (instead of once per job) # runner jobs run sequentially, so booting OpenZaak once (instead of once per job)
@@ -80,10 +86,12 @@ jobs:
run: make verify-acl run: make verify-acl
- name: OpenZaak → NRC notification delivery - name: OpenZaak → NRC notification delivery
run: make verify-nrc run: make verify-nrc
- name: OpenZaak → NRC → Event Subscriber → projection-api
run: make verify-projection
# Log dump must precede teardown (which removes the containers). # Log dump must precede teardown (which removes the containers).
- name: Dump container logs on failure - name: Dump container logs on failure
if: failure() if: failure()
run: docker compose -f infra/docker-compose.yml logs --no-color --tail=100 oz-init openzaak nrc-init nrc-web nrc-celery nrc-beat flowable-init keycloak acl bff 2>&1 || true run: docker compose -f infra/docker-compose.yml logs --no-color --tail=100 oz-init openzaak nrc-init nrc-web nrc-celery nrc-beat flowable-init keycloak acl bff projection-db event-subscriber projection-api 2>&1 || true
- name: Tear down - name: Tear down
if: always() if: always()
run: make down run: make down

View File

@@ -10,7 +10,7 @@ COMPOSE := infra/docker-compose.yml
# Long-running services with a healthcheck — the smoke polls these for readiness # Long-running services with a healthcheck — the smoke polls these for readiness
# (infra/wait-healthy.sh). One-shot init jobs (oz-init, nrc-init, flowable-init) # (infra/wait-healthy.sh). One-shot init jobs (oz-init, nrc-init, flowable-init)
# are not polled; they only need to have run. See docs/runbooks/gitea-actions-gotchas.md. # are not polled; they only need to have run. See docs/runbooks/gitea-actions-gotchas.md.
WAIT_SVCS := openzaak nrc-web acl bff WAIT_SVCS := openzaak nrc-web acl bff event-subscriber projection-api
# Config files (OpenZaak data.yaml, Keycloak realms, Flowable BPMN) are streamed # Config files (OpenZaak data.yaml, Keycloak realms, Flowable BPMN) are streamed
# into external named volumes via `docker cp` (infra/seed-config.sh) instead of # into external named volumes via `docker cp` (infra/seed-config.sh) instead of
# bind-mounted, because bind mounts don't reach sibling containers on the # bind-mounted, because bind mounts don't reach sibling containers on the
@@ -43,7 +43,7 @@ export DOCKER_HOST := unix://$(PODMAN_SOCK)
endif endif
endif endif
.PHONY: ci lint build unit mutation integration verify verify-up verify-acl verify-nrc verify-notifications smoke up down local local-down changelog openzaak-up openzaak-smoke openzaak-seed openzaak-down stack-up stack-smoke stack-down keycloak-up keycloak-smoke keycloak-down flowable-up flowable-smoke flowable-down help .PHONY: ci lint build unit mutation integration verify verify-up verify-acl verify-nrc verify-projection verify-notifications smoke up down local local-down changelog openzaak-up openzaak-smoke openzaak-seed openzaak-down stack-up stack-smoke stack-down keycloak-up keycloak-smoke keycloak-down flowable-up flowable-smoke flowable-down help
## ci: run the full pipeline — lint, build, unit, mutation, verify (mirrors Gitea Actions) ## ci: run the full pipeline — lint, build, unit, mutation, verify (mirrors Gitea Actions)
## `verify` is the live-stack stage (full stack up once → ACL + notification checks). ## `verify` is the live-stack stage (full stack up once → ACL + notification checks).
@@ -61,14 +61,15 @@ build:
unit: unit:
dotnet test $(SLN) -c Release --filter "Category!=Integration" dotnet test $(SLN) -c Release --filter "Category!=Integration"
## mutation: run the Stryker.NET ratchet on the ACL (fails below the recorded baseline) ## mutation: run the Stryker.NET ratchet on each service with branching logic (fails below baseline)
# Stryker is pinned as a local dotnet tool (.config/dotnet-tools.json); `tool restore` # Stryker is pinned as a local dotnet tool (.config/dotnet-tools.json); `tool restore`
# makes `make mutation` work from a fresh clone. Config + break threshold (the ratchet, # makes `make mutation` work from a fresh clone. Each service owns its config + break
# CLAUDE.md §5) live in services/acl/stryker-config.json. The ACL is the first service # threshold (the ratchet, CLAUDE.md §5): services/acl/stryker-config.json and
# with branching logic, so it sets the repo-wide baseline; later slices ratchet it up. # services/event-subscriber/stryker-config.json. Scores never regress below baseline.
mutation: mutation:
dotnet tool restore dotnet tool restore
cd services/acl && dotnet stryker cd services/acl && dotnet stryker
cd services/event-subscriber && dotnet stryker
## smoke: seed config, bring the whole stack up, wait for health-checked services, tear down ## smoke: seed config, bring the whole stack up, wait for health-checked services, tear down
# SEED populates the external config volumes first (upstream images used verbatim; # SEED populates the external config volumes first (upstream images used verbatim;
@@ -130,7 +131,12 @@ verify-acl:
verify-nrc: verify-nrc:
bash infra/run-notification-check.sh bash infra/run-notification-check.sh
## verify: local mirror of the CI verify-stack job — full stack up once, both checks, ## verify-projection: OpenZaak → NRC → Event Subscriber → projection-api end-to-end (S-06),
## against the already-running stack.
verify-projection:
bash infra/run-projection-check.sh
## verify: local mirror of the CI verify-stack job — full stack up once, all checks,
## tear down (always). For fast single-concern local iteration use `integration` ## tear down (always). For fast single-concern local iteration use `integration`
## (oz-only) or `verify-notifications` (oz+nrc) instead. ## (oz-only) or `verify-notifications` (oz+nrc) instead.
verify: verify:
@@ -139,7 +145,8 @@ verify:
@bash -c 'set -e; rc=0; \ @bash -c 'set -e; rc=0; \
WAIT_TIMEOUT=420 bash infra/wait-healthy.sh $(WAIT_SVCS) \ WAIT_TIMEOUT=420 bash infra/wait-healthy.sh $(WAIT_SVCS) \
&& bash infra/run-acl-integration.sh \ && bash infra/run-acl-integration.sh \
&& bash infra/run-notification-check.sh || rc=$$?; \ && bash infra/run-notification-check.sh \
&& bash infra/run-projection-check.sh || rc=$$?; \
docker compose -f $(COMPOSE) down --volumes >/dev/null 2>&1; \ docker compose -f $(COMPOSE) down --volumes >/dev/null 2>&1; \
docker volume rm -f $(CFG_VOLS) >/dev/null 2>&1; \ docker volume rm -f $(CFG_VOLS) >/dev/null 2>&1; \
exit $$rc' exit $$rc'

View File

@@ -0,0 +1,89 @@
# ADR-0008: The read projection — a shared, rebuildable store with a writer and a reader
- **Status:** Accepted
- **Date:** 2026-06-30
- **Deciders:** Respellion engineering
- **Relates to:** S-06 (#7); builds on ADR-0001 (loose coupling), ADR-0007 (#56, OZ→NRC wiring); first EF Core usage in the repo
## Context
S-06 (#7) adds the upstream event path's destination: an **Event Subscriber** that consumes
NRC notifications and a **read projection** the openbaar register reads. The walking-skeleton
projection (PRD §8.4) holds one row per zaak — `id`, `bsn`, `naam_placeholder`, `status`
and must be **idempotent** (NRC redelivers and reorders, CLAUDE.md §8.6) and **rebuildable**
(a derived artefact, never a write-only source of truth).
Two design questions had no obvious answer:
1. **Where does `bsn` come from?** The NRC `zaken`/`zaak`/`create` notification carries only the
zaak URL plus the fixed `kenmerken` (`bronorganisatie`, `zaaktype`, `vertrouwelijkheidaanduiding`).
It does **not** carry the bsn. Reading it means calling a ZGW API — which **only the ACL** may
do (CLAUDE.md §8.1). The issue's "Touches" lists only `event-subscriber` + `projection-api`,
not the ACL.
2. **Who owns the projection schema?** The subscriber writes the projection; the projection-api
reads it. CLAUDE.md §8.5 says "no direct DB access across services; each service owns its
schema." Two deployables on one table looks like a violation.
## Decision
**One Postgres database is the read projection. The Event Subscriber writes it (projector) and
the projection-api reads it (query); both are processes of the single "Read Projection" bounded
context and share one schema, defined in a shared `Projection.ReadModel` library. `bsn` is
deferred.**
- **Schema ownership.** The read model — `register_projection` plus the subscriber's
`processed_notifications` log — lives in `services/projection-api/Projection.ReadModel`
(EF Core + Npgsql). Both services reference it. This is the textbook CQRS read-model split
(one writer, one reader over one derived store), **not** the cross-*domain* DB reach §8.5
forbids: no domain owns write-state here; the projection is rebuildable (§8.4). §8.5 still
holds for every domain database.
- **Idempotency** is the primary key on `processed_notifications.key` (a deterministic key
derived from the immutable notification content). A duplicate insert raises a unique violation,
caught and reported as "already recorded", so the duplicate never reaches the projection. The
projection upsert is itself idempotent on the zaak id, a second line of defence.
- **Rebuild replays the log, not OpenZaak.** `POST /admin/rebuild` clears `register_projection`
and reprojects every row in `processed_notifications`. So "rebuildable" needs **no** ZGW access
(§8.1) and no ACL dependency — keeping S-06 within its stated scope.
- **`bsn` and `naam_placeholder` are deferred.** They are columns (nullable) but the minimal slice
populates only `id` + `status` (`INGEDIEND`) from the notification. Populating personal data
requires reading the zaak **through the ACL** (§8.1) and is its own follow-up; the column shape
is in place so that change is additive.
- **New dependency: EF Core 10 + `Npgsql.EntityFrameworkCore.PostgreSQL`.** What it gives us: a
migrated relational schema, LINQ queries, and a clean port implementation. What we'd write
instead: hand-rolled SQL + a migration runner. Risk: ORM complexity and an extra dependency
graph — bounded here to a tiny two-table read model. `dotnet-ef` is pinned as a local tool for
migrations; `NuGetAuditMode=direct` keeps EF's design-time-only tooling transitive out of the
audited, shipped graph.
The end-to-end path is verified by a runner-safe live-stack smoke (`infra/run-projection-check.sh`,
the `verify-projection` step of the `verify-stack` job, #58): register an abonnement at the real
Event Subscriber's callback, create a zaak, assert projection-api serves an `INGEDIEND` row — all
in-network, reaching services by container IP (ADR-0006/0007).
## Consequences
- **Positive:** the upstream event path reaches a queryable projection; idempotent and rebuildable
without OpenZaak; S-06 stays inside its stated touch-set (no ACL change); the projection-api is
ready for S-09 to tighten public-safe field filtering.
- **Negative / deferred:**
- `bsn`/`naam_placeholder` stay empty until a follow-up wires zaak reads via the ACL.
- The abonnement is registered by the verify harness (by container IP), not provisioned
persistently — ADR-0007 already deferred a persistent abonnement, and a single-label service
host is not URL-valid for NRC, so persistent registration needs a dotted network alias. Tracked
as a follow-up; a plain `make up` therefore needs the abonnement registered before the event
path flows.
- Two services share one database. Acceptable for a derived read model; revisit if the read and
write sides ever need independent scaling or storage.
## Alternatives considered
- **Subscriber reads OpenZaak directly to fill `bsn`** — rejected: breaks §8.1 (only the ACL talks
to ZGW) and would need its own ADR to bend the rule.
- **Extend the ACL with a zaak-read operation, consumed as a library** — viable and §8.1-clean, but
it grows S-06 beyond its stated scope (touches the ACL) and pulls personal-data handling forward;
deferred to a follow-up.
- **projection-api owns the DB and exposes an internal write endpoint the subscriber calls** —
rejected for the walking skeleton: adds an HTTP hop and a write surface on a read service for no
current benefit over a shared, rebuildable read model.
- **Separate databases for the log and the projection** — rejected as premature: both are the read
side's private, rebuildable state; one DB is simpler and still honours §8.5's intent.

35
docs/demo-script.md Normal file
View File

@@ -0,0 +1,35 @@
# Demo script
A running log of demoable outcomes, one section per slice. Each entry is a short,
copy-pasteable walkthrough against a local `make up` stack.
---
## S-06 — Event Subscriber + read projection
**Outcome:** a zaak created in OpenZaak flows through NRC to the Event Subscriber, which
projects it into a rebuildable read projection the projection-api serves.
**The path:** OpenZaak → (notification) NRC → (abonnement callback) Event Subscriber →
`register_projection` → projection-api `GET /register`.
```bash
# 1. Bring the full stack up (seeds config, builds our services, waits for health).
make up
# 2. Register the Event Subscriber's abonnement and create a zaak, then read it back.
# (The verify-projection check does exactly this end-to-end and asserts the result.)
make verify-projection # → "OK — projection-api serves zaak <uuid> with status INGEDIEND"
# 3. Observe the projection directly via the read API (host port 8120).
curl -fsS http://localhost:8120/register | jq
# → [ { "id": "<zaak-uuid>", "status": "INGEDIEND", "bsn": null, "naamPlaceholder": null } ]
# 4. Idempotency + rebuild: replays don't duplicate; a rebuild repopulates from the
# notification log (no OpenZaak access needed — ADR-0008).
curl -fsS -X POST http://localhost:8110/admin/rebuild # Event Subscriber, host port 8110
curl -fsS http://localhost:8120/register | jq 'length' # → unchanged
```
> `bsn` / `naam_placeholder` are deferred (ADR-0008) — the notification doesn't carry them and
> the subscriber may not read OpenZaak directly (§8.1). They surface in a later slice.

View File

@@ -305,10 +305,68 @@ services:
start_period: 10s start_period: 10s
networks: [cg] networks: [cg]
# ── Read projection (S-06) ────────────────────────────────────────────────
projection-db:
image: docker.io/library/postgres:16
environment:
POSTGRES_USER: projection
POSTGRES_PASSWORD: projection
POSTGRES_DB: projection
volumes:
- projection-db:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U projection -d projection"]
interval: 5s
timeout: 3s
retries: 10
networks: [cg]
event-subscriber:
build:
context: ..
dockerfile: services/event-subscriber/Dockerfile
image: register-referentie/event-subscriber:dev
environment:
ConnectionStrings__Projection: Host=projection-db;Database=projection;Username=projection;Password=projection
EventSubscriber__Webhook__AuthToken: ${NOTIFICATION_WEBHOOK_TOKEN:-Bearer big-reference-notifications}
ports:
- "8110:8080"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8080/health"]
interval: 5s
timeout: 3s
retries: 5
start_period: 15s
depends_on:
projection-db:
condition: service_healthy
networks: [cg]
projection-api:
build:
context: ..
dockerfile: services/projection-api/Dockerfile
image: register-referentie/projection-api:dev
environment:
ConnectionStrings__Projection: Host=projection-db;Database=projection;Username=projection;Password=projection
ports:
- "8120:8080"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8080/health"]
interval: 5s
timeout: 3s
retries: 5
start_period: 15s
depends_on:
projection-db:
condition: service_healthy
networks: [cg]
volumes: volumes:
oz-db: oz-db:
nrc-db: nrc-db:
flowable-db: flowable-db:
projection-db:
networks: networks:
cg: cg:

View File

@@ -9,6 +9,8 @@
# 8080 BFF GET /health → Healthy # 8080 BFF GET /health → Healthy
# 8090 Flowable REST http://localhost:8090/flowable-rest/service/ # 8090 Flowable REST http://localhost:8090/flowable-rest/service/
# 8100 ACL GET /health → Healthy POST /zaken # 8100 ACL GET /health → Healthy POST /zaken
# 8110 Event Subscriber GET /health → Healthy POST /notifications
# 8120 projection-api GET /health → Healthy GET /register
# 8180 Keycloak (admin: admin / admin) # 8180 Keycloak (admin: admin / admin)
# #
# docker compose -f infra/docker-compose.yml up -d --build --wait # docker compose -f infra/docker-compose.yml up -d --build --wait
@@ -320,10 +322,76 @@ services:
start_period: 10s start_period: 10s
networks: [cg] networks: [cg]
# ── Read projection (S-06) ────────────────────────────────────────────────
# One Postgres DB backing the rebuildable read projection (PRD §8.4): the Event
# Subscriber writes it, projection-api reads it. See ADR-0008.
projection-db:
image: docker.io/library/postgres:16
environment:
POSTGRES_USER: projection
POSTGRES_PASSWORD: projection
POSTGRES_DB: projection
volumes:
- projection-db:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U projection -d projection"]
interval: 5s
timeout: 3s
retries: 10
networks: [cg]
# Consumes NRC notifications (abonnement callback) and projects zaak-created events
# into register_projection. Build context is the repo root: it shares the read model
# in services/projection-api/Projection.ReadModel.
event-subscriber:
build:
context: ..
dockerfile: services/event-subscriber/Dockerfile
image: register-referentie/event-subscriber:dev
environment:
ConnectionStrings__Projection: Host=projection-db;Database=projection;Username=projection;Password=projection
# The bearer Open Notificaties must present on the abonnement callback. NRC's
# registration probe expects a 401 without it (ADR-0007). Dev-only token.
EventSubscriber__Webhook__AuthToken: ${NOTIFICATION_WEBHOOK_TOKEN:-Bearer big-reference-notifications}
ports:
- "8110:8080"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8080/health"]
interval: 5s
timeout: 3s
retries: 5
start_period: 15s
depends_on:
projection-db:
condition: service_healthy
networks: [cg]
# The read side of the projection. Shares Projection.ReadModel, so build context is root.
projection-api:
build:
context: ..
dockerfile: services/projection-api/Dockerfile
image: register-referentie/projection-api:dev
environment:
ConnectionStrings__Projection: Host=projection-db;Database=projection;Username=projection;Password=projection
ports:
- "8120:8080"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8080/health"]
interval: 5s
timeout: 3s
retries: 5
start_period: 15s
depends_on:
projection-db:
condition: service_healthy
networks: [cg]
volumes: volumes:
oz-db: oz-db:
nrc-db: nrc-db:
flowable-db: flowable-db:
projection-db:
# Config volumes — created and populated out-of-band by infra/seed-config.sh # Config volumes — created and populated out-of-band by infra/seed-config.sh
# (docker cp), because bind mounts don't reach sibling containers on the CI # (docker cp), because bind mounts don't reach sibling containers on the CI
# runner. `external` keeps the names deterministic; the seed step manages them. # runner. `external` keeps the names deterministic; the seed step manages them.

68
infra/run-projection-check.sh Executable file
View File

@@ -0,0 +1,68 @@
#!/usr/bin/env bash
#
# Verify the end-to-end read-projection path (S-06) against an ALREADY-RUNNING full stack:
# OpenZaak → NRC → Event Subscriber → projection → projection-api. Seeds a published BIG
# zaaktype (idempotent), registers an abonnement on the `zaken` kanaal pointing at the real
# Event Subscriber's /notifications callback (with the bearer it enforces), creates a zaak,
# and asserts projection-api serves a row for that zaak with status INGEDIEND.
#
# All in-network, reaching services by container IP — single-label hosts aren't URL-valid and
# the runner can't reach published ports (gitea-actions-gotchas.md §5/§6). Reuses the
# notification driver to register the abonnement + create the zaak. Does NOT manage the stack
# lifecycle (the caller owns bring-up + teardown). Plain docker primitives only. See ADR-0007/0008.
set -euo pipefail
here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
WEBHOOK_AUTH="${NOTIFICATION_WEBHOOK_TOKEN:-Bearer big-reference-notifications}"
cleanup() { docker rm -f rr-pverify rr-pquery >/dev/null 2>&1 || true; }
trap cleanup EXIT
ip() { docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$1"; }
oz="$(docker ps -q --filter 'name=[-_]openzaak[-_]' | head -1)"
nrc="$(docker ps -q --filter 'name=nrc-web' | head -1)"
es="$(docker ps -q --filter 'name=event-subscriber' | head -1)"
proj="$(docker ps -q --filter 'name=projection-api' | head -1)"
[ -n "$oz" ] && [ -n "$nrc" ] || { echo "ERROR: OpenZaak and/or NRC not running — bring the stack up first" >&2; exit 1; }
[ -n "$es" ] && [ -n "$proj" ] || { echo "ERROR: event-subscriber and/or projection-api not running — bring the stack up first" >&2; exit 1; }
net="$(docker inspect -f '{{range $k,$_ := .NetworkSettings.Networks}}{{$k}}{{"\n"}}{{end}}' "$oz" | head -1)"
oz_ip="$(ip "$oz")"; nrc_ip="$(ip "$nrc")"; es_ip="$(ip "$es")"; proj_ip="$(ip "$proj")"
echo ">> network=$net openzaak=$oz_ip nrc=$nrc_ip event-subscriber=$es_ip projection-api=$proj_ip"
echo ">> seeding a published BIG zaaktype (idempotent)"
sid="$(docker create --network "$net" -e "OZ_BASE=http://$oz_ip:8000" -e OZ_PUBLISH=1 \
python:3-slim python /seed.py)"
docker cp "$here/openzaak/seed_catalogus.py" "$sid:/seed.py" >/dev/null
docker start -a "$sid"
docker rm -f "$sid" >/dev/null
echo ">> registering abonnement at the Event Subscriber + creating a zaak"
docker rm -f rr-pverify >/dev/null 2>&1 || true
drv="$(docker create --network "$net" --name rr-pverify \
-e "OZ_BASE=http://$oz_ip:8000" -e "NRC_BASE=http://$nrc_ip:8000" \
-e "SINK_CALLBACK=http://$es_ip:8080/notifications" -e "SINK_AUTH=$WEBHOOK_AUTH" \
python:3-slim python /driver.py)"
docker cp "$here/verify-notification-driver.py" "$drv:/driver.py" >/dev/null
docker start -a "$drv"
zaak_url="$(docker logs rr-pverify 2>/dev/null | sed -n 's/^ZAAK_CREATED //p' | head -1)"
docker rm -f rr-pverify >/dev/null
[ -n "$zaak_url" ] || { echo "ERROR: driver did not create a zaak" >&2; exit 1; }
zaak_uuid="${zaak_url##*/}"
echo ">> zaak created: $zaak_url"
echo ">> polling projection-api for the projected row (status INGEDIEND)"
for _ in $(seq 1 30); do
body="$(docker run --rm --network "$net" curlimages/curl:latest \
-fsS "http://$proj_ip:8080/register/$zaak_uuid" 2>/dev/null || true)"
if echo "$body" | grep -q '"INGEDIEND"'; then
echo "OK — projection-api serves zaak $zaak_uuid with status INGEDIEND"
echo "$body" | cut -c1-300
exit 0
fi
sleep 2
done
echo "FAIL — projection-api never served an INGEDIEND row for zaak $zaak_uuid" >&2
echo "--- event-subscriber log ---" >&2; docker logs "$es" 2>&1 | tail -10 >&2
echo "--- projection-api log ---" >&2; docker logs "$proj" 2>&1 | tail -10 >&2
exit 1

View File

@@ -29,7 +29,9 @@ nav:
- "ADR-0005: Mutation testing": architecture/adr-0005-mutation-testing.md - "ADR-0005: Mutation testing": architecture/adr-0005-mutation-testing.md
- "ADR-0006: ACL integration test provisioning": architecture/adr-0006-integration-test-provisioning.md - "ADR-0006: ACL integration test provisioning": architecture/adr-0006-integration-test-provisioning.md
- "ADR-0007: OpenZaak → NRC notification wiring": architecture/adr-0007-notification-wiring.md - "ADR-0007: OpenZaak → NRC notification wiring": architecture/adr-0007-notification-wiring.md
- "ADR-0008: Read projection store": architecture/adr-0008-read-projection-store.md
- Working in Gitea: gitea-workflow.md - Working in Gitea: gitea-workflow.md
- Demo script: demo-script.md
- Runbooks: - Runbooks:
- CI: runbooks/ci.md - CI: runbooks/ci.md

View File

@@ -11,6 +11,15 @@
<Project Path="services/bff/Bff.Api/Bff.Api.csproj" /> <Project Path="services/bff/Bff.Api/Bff.Api.csproj" />
<Project Path="services/bff/Bff.Tests/Bff.Tests.csproj" /> <Project Path="services/bff/Bff.Tests/Bff.Tests.csproj" />
</Folder> </Folder>
<Folder Name="/services/event-subscriber/">
<Project Path="services/event-subscriber/EventSubscriber.Api/EventSubscriber.Api.csproj" />
<Project Path="services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj" />
<Project Path="services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj" />
</Folder>
<Folder Name="/services/projection-api/">
<Project Path="services/projection-api/Projection.ReadModel/Projection.ReadModel.csproj" />
<Project Path="services/projection-api/ProjectionApi.Api/ProjectionApi.Api.csproj" />
</Folder>
<Folder Name="/tests/"> <Folder Name="/tests/">
<Project Path="tests/acceptance/Acceptance.csproj" /> <Project Path="tests/acceptance/Acceptance.csproj" />
</Folder> </Folder>

View File

@@ -0,0 +1,32 @@
# Multi-stage build for the Event Subscriber service (.NET 10).
# Build context is services/event-subscriber, but it references the shared read model in
# services/projection-api, so the context is the repo root (see infra/docker-compose.yml).
FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
WORKDIR /src
# Restore first (cached unless .csproj files change).
COPY services/event-subscriber/EventSubscriber.Api/EventSubscriber.Api.csproj services/event-subscriber/EventSubscriber.Api/
COPY services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj services/event-subscriber/EventSubscriber.Application/
COPY services/projection-api/Projection.ReadModel/Projection.ReadModel.csproj services/projection-api/Projection.ReadModel/
RUN dotnet restore services/event-subscriber/EventSubscriber.Api/EventSubscriber.Api.csproj
COPY services/event-subscriber/ services/event-subscriber/
COPY services/projection-api/Projection.ReadModel/ services/projection-api/Projection.ReadModel/
RUN dotnet publish services/event-subscriber/EventSubscriber.Api/EventSubscriber.Api.csproj -c Release -o /app/publish /p:UseAppHost=false
FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS runtime
WORKDIR /app
RUN apt-get update \
&& apt-get install -y --no-install-recommends curl \
&& rm -rf /var/lib/apt/lists/*
COPY --from=build /app/publish .
ENV ASPNETCORE_URLS=http://+:8080
EXPOSE 8080
HEALTHCHECK --interval=5s --timeout=3s --start-period=15s --retries=5 \
CMD curl -fsS http://localhost:8080/health || exit 1
ENTRYPOINT ["dotnet", "EventSubscriber.Api.dll"]

View File

@@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<ItemGroup>
<ProjectReference Include="..\EventSubscriber.Application\EventSubscriber.Application.csproj" />
<ProjectReference Include="..\..\projection-api\Projection.ReadModel\Projection.ReadModel.csproj" />
</ItemGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,66 @@
using System.Text.Json;
using EventSubscriber.Application;
using Projection.ReadModel;
var builder = WebApplication.CreateBuilder(args);
var connectionString = builder.Configuration.GetConnectionString("Projection")
?? throw new InvalidOperationException("Missing connection string 'ConnectionStrings:Projection'");
// The exact Authorization header value Open Notificaties sends on each abonnement callback.
// NRC probes the callback during registration and refuses it unless it returns 401 without
// this value (ADR-0007), so the webhook enforces it.
var webhookToken = builder.Configuration["EventSubscriber:Webhook:AuthToken"]
?? throw new InvalidOperationException("Missing configuration 'EventSubscriber:Webhook:AuthToken'");
builder.Services.AddProjectionReadModel(connectionString);
builder.Services.AddProjectionWriteSide();
var app = builder.Build();
// Apply migrations on start so a fresh stack reaches a usable schema unattended.
await app.Services.MigrateProjectionAsync();
app.MapGet("/health", () => "Healthy");
// The NRC abonnement callback. Open Notificaties POSTs a notification here; we project it.
// Auth-on-callback is mandatory: the auth check runs *before* the body is read, so NRC's
// registration probe (a POST without the configured Authorization, and without a valid
// notification body) gets the 401 it requires rather than a 400 (ADR-0007).
app.MapPost("/notifications", async (
HttpRequest request,
NotificationProjector projector,
CancellationToken ct) =>
{
if (request.Headers.Authorization != webhookToken)
return Results.Unauthorized();
var dto = await JsonSerializer.DeserializeAsync<NotificationDto>(
request.Body, SerializerOptions, ct);
if (dto is null)
return Results.BadRequest();
await projector.HandleAsync(dto.ToNotification(), ct);
return Results.NoContent();
});
// Admin: rebuild the projection from the durable notification log (PRD §8.4 — rebuildable).
app.MapPost("/admin/rebuild", async (NotificationProjector projector, CancellationToken ct) =>
{
await projector.RebuildAsync(ct);
return Results.NoContent();
});
await app.RunAsync();
/// <summary>The NRC notification body, as Open Notificaties POSTs it. Only the fields the
/// projection needs are bound; <c>aanmaakdatum</c>/<c>kenmerken</c> are ignored for the minimal slice.</summary>
public sealed record NotificationDto(string Kanaal, string Resource, string Actie, Uri ResourceUrl)
{
public Notification ToNotification() => new(Kanaal, Resource, Actie, ResourceUrl);
}
public partial class Program
{
// NRC sends camelCase JSON; match it case-insensitively.
private static readonly JsonSerializerOptions SerializerOptions = new(JsonSerializerDefaults.Web);
}

View File

@@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}

View File

@@ -0,0 +1,9 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,30 @@
namespace EventSubscriber.Application;
/// <summary>
/// An inbound NRC (Open Notificaties) notification, as Open Notificaties POSTs it to an
/// abonnement callback. Only the fields the projection needs are modelled; the full ZGW
/// "Notificatie" resource also carries <c>aanmaakdatum</c> and <c>kenmerken</c> which the
/// minimal projection ignores (bsn is deferred — see ADR-0008). For a <c>zaken</c>/<c>zaak</c>/<c>create</c>
/// notification <c>hoofdObject</c> and <c>resourceUrl</c> are both the created zaak's URL.
/// </summary>
public sealed record Notification(
string Kanaal,
string Resource,
string Actie,
Uri ResourceUrl)
{
/// <summary>The only event the walking-skeleton projection reacts to: a zaak being created.</summary>
public bool IsZaakCreated =>
Kanaal == "zaken" && Resource == "zaak" && Actie == "create";
/// <summary>The zaak UUID — the trailing path segment of the resource URL — used as the projection key.</summary>
public string ZaakId => ResourceUrl.Segments[^1].Trim('/');
/// <summary>
/// A deterministic dedup key. Open Notificaties carries no notification id and may
/// redeliver, so the key is derived from the immutable notification content: two
/// deliveries of the same zaak-create collapse to one. (NRC may also deliver
/// out of order; the projector tolerates that — order does not change the outcome.)
/// </summary>
public string IdempotencyKey => $"{Kanaal}:{Resource}:{Actie}:{ResourceUrl}";
}

View File

@@ -0,0 +1,38 @@
namespace EventSubscriber.Application;
/// <summary>
/// Projects inbound NRC notifications into the read projection. Tolerates duplicate and
/// out-of-order deliveries (CLAUDE.md §8.6): the notification log dedups, and the projection
/// upsert is idempotent on the zaak id. Rebuilds the projection by replaying the log.
/// </summary>
public sealed class NotificationProjector(INotificationLog log, IProjectionStore store)
{
/// <summary>Handle one inbound notification. Ignores anything that is not a zaak-created event.</summary>
public async Task HandleAsync(Notification notification, CancellationToken ct = default)
{
if (!notification.IsZaakCreated)
return;
var recorded = new RecordedNotification(notification.IdempotencyKey, notification.Actie, notification.ZaakId);
// Atomic record-or-skip: a duplicate (or concurrent) delivery is recognised and dropped
// before it touches the projection, so the projection stays a faithful derived artefact.
if (!await log.TryRecordAsync(recorded, ct))
return;
await store.UpsertAsync(ToEntry(recorded), ct);
}
/// <summary>Rebuild the projection from the durable notification log (PRD §8.4).</summary>
public async Task RebuildAsync(CancellationToken ct = default)
{
await store.ClearAsync(ct);
foreach (var recorded in await log.AllAsync(ct))
await store.UpsertAsync(ToEntry(recorded), ct);
}
/// <summary>The minimal projection row for an accepted notification. A zaak-create maps to
/// INGEDIEND; bsn/naam are deferred (ADR-0008).</summary>
private static RegisterEntry ToEntry(RecordedNotification recorded)
=> new(recorded.ZaakId, RegistrationStatus.Ingediend);
}

View File

@@ -0,0 +1,37 @@
namespace EventSubscriber.Application;
/// <summary>
/// The durable log of notifications the subscriber has accepted. It is both the idempotency
/// guard (a replayed notification is recognised and dropped) and the rebuild source: the
/// projection is a derived artefact (PRD §8.4) regenerated by replaying this log, so a rebuild
/// needs no access to OpenZaak (CLAUDE.md §8.1). Implemented in Infrastructure over Postgres.
/// </summary>
public interface INotificationLog
{
/// <summary>
/// Record a notification. Returns <c>true</c> if it was newly recorded, <c>false</c> if an
/// entry with the same key already existed (a duplicate delivery). Must be atomic so that
/// concurrent duplicate deliveries cannot both observe <c>true</c>.
/// </summary>
Task<bool> TryRecordAsync(RecordedNotification notification, CancellationToken ct = default);
/// <summary>Every accepted notification, for rebuilding the projection.</summary>
Task<IReadOnlyList<RecordedNotification>> AllAsync(CancellationToken ct = default);
}
/// <summary>A notification that has been accepted, retaining what a rebuild needs to recompute its projection row.</summary>
public sealed record RecordedNotification(string Key, string Actie, string ZaakId);
/// <summary>The read projection store. Owned by the projection bounded context (ADR-0008); the
/// subscriber writes to it and the projection-api reads it.</summary>
public interface IProjectionStore
{
/// <summary>Insert or update the row for <see cref="RegisterEntry.Id"/>. Idempotent on the id.</summary>
Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default);
/// <summary>Remove every row — the first step of a rebuild.</summary>
Task ClearAsync(CancellationToken ct = default);
/// <summary>Every projection row (used by tests and the rebuild verification).</summary>
Task<IReadOnlyList<RegisterEntry>> AllAsync(CancellationToken ct = default);
}

View File

@@ -0,0 +1,20 @@
namespace EventSubscriber.Application;
/// <summary>
/// A row of the rebuildable read projection (PRD §8.4). For the minimal slice it carries
/// the zaak id and a status; <c>Bsn</c> and <c>NaamPlaceholder</c> are part of the schema but
/// are deferred — the NRC notification does not carry them and the subscriber may not read
/// OpenZaak directly (CLAUDE.md §8.1). Populating them is a follow-up (ADR-0008).
/// </summary>
public sealed record RegisterEntry(
string Id,
string Status,
string? Bsn = null,
string? NaamPlaceholder = null);
/// <summary>The projection statuses the walking skeleton knows about.</summary>
public static class RegistrationStatus
{
/// <summary>A zaak has been created; the registration is submitted.</summary>
public const string Ingediend = "INGEDIEND";
}

View File

@@ -0,0 +1,25 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<IsPackable>false</IsPackable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="coverlet.collector" Version="6.0.4" />
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
<PackageReference Include="xunit" Version="2.9.3" />
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.4" />
</ItemGroup>
<ItemGroup>
<Using Include="Xunit" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\EventSubscriber.Application\EventSubscriber.Application.csproj" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,42 @@
using EventSubscriber.Application;
namespace EventSubscriber.Tests;
/// <summary>In-memory stand-ins for the projection store and notification log, so the
/// projector's behaviour is exercised without Postgres (hand-written stubs, the repo's
/// convention — no mocking library).</summary>
internal sealed class InMemoryNotificationLog : INotificationLog
{
private readonly Dictionary<string, RecordedNotification> _byKey = [];
public Task<bool> TryRecordAsync(RecordedNotification notification, CancellationToken ct = default)
=> Task.FromResult(_byKey.TryAdd(notification.Key, notification));
public Task<IReadOnlyList<RecordedNotification>> AllAsync(CancellationToken ct = default)
=> Task.FromResult<IReadOnlyList<RecordedNotification>>([.. _byKey.Values]);
}
internal sealed class InMemoryProjectionStore : IProjectionStore
{
private readonly Dictionary<string, RegisterEntry> _byId = [];
/// <summary>How many times a write was attempted — lets a test prove a deduped delivery
/// never reaches the store (an idempotent upsert hides the difference in row count alone).</summary>
public int UpsertCount { get; private set; }
public Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default)
{
UpsertCount++;
_byId[entry.Id] = entry;
return Task.CompletedTask;
}
public Task ClearAsync(CancellationToken ct = default)
{
_byId.Clear();
return Task.CompletedTask;
}
public Task<IReadOnlyList<RegisterEntry>> AllAsync(CancellationToken ct = default)
=> Task.FromResult<IReadOnlyList<RegisterEntry>>([.. _byId.Values]);
}

View File

@@ -0,0 +1,89 @@
using EventSubscriber.Application;
namespace EventSubscriber.Tests;
/// <summary>Behaviour of the projector that turns NRC notifications into projection rows.
/// The walking skeleton reacts only to a zaak being created (status INGEDIEND) and must
/// tolerate duplicate and out-of-order deliveries (CLAUDE.md §8.6).</summary>
public sealed class NotificationProjectorTests
{
private const string ZaakUrl = "http://openzaak:8000/zaken/api/v1/zaken/11111111-1111-1111-1111-111111111111";
private readonly InMemoryNotificationLog _log = new();
private readonly InMemoryProjectionStore _store = new();
private NotificationProjector Projector() => new(_log, _store);
private static Notification ZaakCreated(string url = ZaakUrl)
=> new("zaken", "zaak", "create", new Uri(url));
[Fact]
public async Task creating_a_zaak_writes_one_row_with_status_ingediend()
{
await Projector().HandleAsync(ZaakCreated());
var entry = Assert.Single(await _store.AllAsync());
Assert.Equal("11111111-1111-1111-1111-111111111111", entry.Id);
Assert.Equal(RegistrationStatus.Ingediend, entry.Status);
}
[Fact]
public async Task replaying_the_same_notification_keeps_a_single_row()
{
var projector = Projector();
await projector.HandleAsync(ZaakCreated());
await projector.HandleAsync(ZaakCreated());
Assert.Single(await _store.AllAsync());
}
[Fact]
public async Task a_replayed_notification_never_reaches_the_projection_store()
{
var projector = Projector();
await projector.HandleAsync(ZaakCreated());
await projector.HandleAsync(ZaakCreated());
// The duplicate is dropped at the log, before the (idempotent) upsert — so the store
// is written exactly once. Row count alone can't see this; the upsert count can.
Assert.Equal(1, _store.UpsertCount);
}
[Fact]
public async Task two_different_zaken_each_get_their_own_row()
{
var projector = Projector();
await projector.HandleAsync(ZaakCreated());
await projector.HandleAsync(ZaakCreated(ZaakUrl[..^1] + "2")); // a distinct zaak url
Assert.Equal(2, (await _store.AllAsync()).Count);
}
[Theory]
[InlineData("documenten", "enkelvoudiginformatieobject", "create")] // wrong kanaal + resource
[InlineData("documenten", "zaak", "create")] // wrong kanaal only
[InlineData("zaken", "status", "create")] // wrong resource only
[InlineData("zaken", "zaak", "update")] // wrong actie
[InlineData("zaken", "zaak", "destroy")] // wrong actie
public async Task only_a_zaken_zaak_create_notification_is_projected(string kanaal, string resource, string actie)
{
await Projector().HandleAsync(new Notification(kanaal, resource, actie, new Uri(ZaakUrl)));
Assert.Empty(await _store.AllAsync());
}
[Fact]
public async Task rebuild_clears_stale_rows_and_repopulates_from_the_notification_log()
{
var projector = Projector();
await projector.HandleAsync(ZaakCreated());
// A stale row that is not backed by any logged notification must not survive a rebuild.
await _store.UpsertAsync(new RegisterEntry("stale-9999", RegistrationStatus.Ingediend));
await projector.RebuildAsync();
var entry = Assert.Single(await _store.AllAsync());
Assert.Equal("11111111-1111-1111-1111-111111111111", entry.Id);
Assert.Equal(RegistrationStatus.Ingediend, entry.Status);
}
}

View File

@@ -0,0 +1,4 @@
<Solution>
<Project Path="EventSubscriber.Application/EventSubscriber.Application.csproj" />
<Project Path="EventSubscriber.Tests/EventSubscriber.Tests.csproj" />
</Solution>

View File

@@ -0,0 +1,12 @@
{
"stryker-config": {
"solution": "EventSubscriber.slnx",
"test-projects": ["EventSubscriber.Tests/EventSubscriber.Tests.csproj"],
"reporters": ["progress", "html"],
"thresholds": {
"high": 95,
"low": 90,
"break": 90
}
}
}

View File

@@ -0,0 +1,31 @@
# Multi-stage build for the projection-api service (.NET 10).
# Build context is the repo root (it shares Projection.ReadModel — see infra/docker-compose.yml).
FROM mcr.microsoft.com/dotnet/sdk:10.0 AS build
WORKDIR /src
# Restore first (cached unless .csproj files change).
COPY services/projection-api/ProjectionApi.Api/ProjectionApi.Api.csproj services/projection-api/ProjectionApi.Api/
COPY services/projection-api/Projection.ReadModel/Projection.ReadModel.csproj services/projection-api/Projection.ReadModel/
COPY services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj services/event-subscriber/EventSubscriber.Application/
RUN dotnet restore services/projection-api/ProjectionApi.Api/ProjectionApi.Api.csproj
COPY services/projection-api/ services/projection-api/
COPY services/event-subscriber/EventSubscriber.Application/ services/event-subscriber/EventSubscriber.Application/
RUN dotnet publish services/projection-api/ProjectionApi.Api/ProjectionApi.Api.csproj -c Release -o /app/publish /p:UseAppHost=false
FROM mcr.microsoft.com/dotnet/aspnet:10.0 AS runtime
WORKDIR /app
RUN apt-get update \
&& apt-get install -y --no-install-recommends curl \
&& rm -rf /var/lib/apt/lists/*
COPY --from=build /app/publish .
ENV ASPNETCORE_URLS=http://+:8080
EXPOSE 8080
HEALTHCHECK --interval=5s --timeout=3s --start-period=15s --retries=5 \
CMD curl -fsS http://localhost:8080/health || exit 1
ENTRYPOINT ["dotnet", "ProjectionApi.Api.dll"]

View File

@@ -0,0 +1,18 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Design;
namespace Projection.ReadModel;
/// <summary>Lets <c>dotnet ef migrations</c> build the context at design time without a running
/// database or the host's DI. The connection string is a placeholder — migrations only need the
/// provider to emit Postgres-shaped SQL.</summary>
public sealed class DesignTimeProjectionDbContextFactory : IDesignTimeDbContextFactory<ProjectionDbContext>
{
public ProjectionDbContext CreateDbContext(string[] args)
{
var options = new DbContextOptionsBuilder<ProjectionDbContext>()
.UseNpgsql("Host=localhost;Database=projection;Username=projection;Password=projection")
.Options;
return new ProjectionDbContext(options);
}
}

View File

@@ -0,0 +1,39 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>EF Core implementation of the notification log. Idempotency is enforced atomically
/// by the primary key on <c>key</c>: a duplicate insert raises a unique violation, which is
/// caught and reported as "already recorded" rather than failing the request.</summary>
public sealed class EfNotificationLog(ProjectionDbContext db) : INotificationLog
{
public async Task<bool> TryRecordAsync(RecordedNotification notification, CancellationToken ct = default)
{
db.ProcessedNotifications.Add(new ProcessedNotificationRow
{
Key = notification.Key,
Actie = notification.Actie,
ZaakId = notification.ZaakId,
ReceivedAt = DateTimeOffset.UtcNow,
});
try
{
await db.SaveChangesAsync(ct);
return true;
}
catch (DbUpdateException)
{
// Already recorded by an earlier (or concurrent) delivery — drop this duplicate.
db.ChangeTracker.Clear();
return false;
}
}
public async Task<IReadOnlyList<RecordedNotification>> AllAsync(CancellationToken ct = default)
=> await db.ProcessedNotifications
.OrderBy(r => r.ReceivedAt)
.Select(r => new RecordedNotification(r.Key, r.Actie, r.ZaakId))
.ToListAsync(ct);
}

View File

@@ -0,0 +1,40 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>EF Core implementation of the projection store over <see cref="ProjectionDbContext"/>.</summary>
public sealed class EfProjectionStore(ProjectionDbContext db) : IProjectionStore
{
public async Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default)
{
var row = await db.RegisterEntries.FindAsync([entry.Id], ct);
if (row is null)
{
db.RegisterEntries.Add(new RegisterEntryRow
{
Id = entry.Id,
Status = entry.Status,
Bsn = entry.Bsn,
NaamPlaceholder = entry.NaamPlaceholder,
});
}
else
{
row.Status = entry.Status;
row.Bsn = entry.Bsn;
row.NaamPlaceholder = entry.NaamPlaceholder;
}
await db.SaveChangesAsync(ct);
}
public async Task ClearAsync(CancellationToken ct = default)
=> await db.RegisterEntries.ExecuteDeleteAsync(ct);
public async Task<IReadOnlyList<RegisterEntry>> AllAsync(CancellationToken ct = default)
=> await db.RegisterEntries
.OrderBy(r => r.Id)
.Select(r => new RegisterEntry(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.ToListAsync(ct);
}

View File

@@ -0,0 +1,79 @@
// <auto-generated />
using System;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
using Projection.ReadModel;
#nullable disable
namespace Projection.ReadModel.Migrations
{
[DbContext(typeof(ProjectionDbContext))]
[Migration("20260630125055_InitialProjection")]
partial class InitialProjection
{
/// <inheritdoc />
protected override void BuildTargetModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
.HasAnnotation("ProductVersion", "10.0.0")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
modelBuilder.Entity("Projection.ReadModel.ProcessedNotificationRow", b =>
{
b.Property<string>("Key")
.HasColumnType("text")
.HasColumnName("key");
b.Property<string>("Actie")
.IsRequired()
.HasColumnType("text")
.HasColumnName("actie");
b.Property<DateTimeOffset>("ReceivedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("received_at");
b.Property<string>("ZaakId")
.IsRequired()
.HasColumnType("text")
.HasColumnName("zaak_id");
b.HasKey("Key");
b.ToTable("processed_notifications", (string)null);
});
modelBuilder.Entity("Projection.ReadModel.RegisterEntryRow", b =>
{
b.Property<string>("Id")
.HasColumnType("text")
.HasColumnName("id");
b.Property<string>("Bsn")
.HasColumnType("text")
.HasColumnName("bsn");
b.Property<string>("NaamPlaceholder")
.HasColumnType("text")
.HasColumnName("naam_placeholder");
b.Property<string>("Status")
.IsRequired()
.HasColumnType("text")
.HasColumnName("status");
b.HasKey("Id");
b.ToTable("register_projection", (string)null);
});
#pragma warning restore 612, 618
}
}
}

View File

@@ -0,0 +1,53 @@
using System;
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace Projection.ReadModel.Migrations
{
/// <inheritdoc />
public partial class InitialProjection : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.CreateTable(
name: "processed_notifications",
columns: table => new
{
key = table.Column<string>(type: "text", nullable: false),
actie = table.Column<string>(type: "text", nullable: false),
zaak_id = table.Column<string>(type: "text", nullable: false),
received_at = table.Column<DateTimeOffset>(type: "timestamp with time zone", nullable: false)
},
constraints: table =>
{
table.PrimaryKey("PK_processed_notifications", x => x.key);
});
migrationBuilder.CreateTable(
name: "register_projection",
columns: table => new
{
id = table.Column<string>(type: "text", nullable: false),
status = table.Column<string>(type: "text", nullable: false),
bsn = table.Column<string>(type: "text", nullable: true),
naam_placeholder = table.Column<string>(type: "text", nullable: true)
},
constraints: table =>
{
table.PrimaryKey("PK_register_projection", x => x.id);
});
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropTable(
name: "processed_notifications");
migrationBuilder.DropTable(
name: "register_projection");
}
}
}

View File

@@ -0,0 +1,76 @@
// <auto-generated />
using System;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
using Projection.ReadModel;
#nullable disable
namespace Projection.ReadModel.Migrations
{
[DbContext(typeof(ProjectionDbContext))]
partial class ProjectionDbContextModelSnapshot : ModelSnapshot
{
protected override void BuildModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
.HasAnnotation("ProductVersion", "10.0.0")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
modelBuilder.Entity("Projection.ReadModel.ProcessedNotificationRow", b =>
{
b.Property<string>("Key")
.HasColumnType("text")
.HasColumnName("key");
b.Property<string>("Actie")
.IsRequired()
.HasColumnType("text")
.HasColumnName("actie");
b.Property<DateTimeOffset>("ReceivedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("received_at");
b.Property<string>("ZaakId")
.IsRequired()
.HasColumnType("text")
.HasColumnName("zaak_id");
b.HasKey("Key");
b.ToTable("processed_notifications", (string)null);
});
modelBuilder.Entity("Projection.ReadModel.RegisterEntryRow", b =>
{
b.Property<string>("Id")
.HasColumnType("text")
.HasColumnName("id");
b.Property<string>("Bsn")
.HasColumnType("text")
.HasColumnName("bsn");
b.Property<string>("NaamPlaceholder")
.HasColumnType("text")
.HasColumnName("naam_placeholder");
b.Property<string>("Status")
.IsRequired()
.HasColumnType("text")
.HasColumnName("status");
b.HasKey("Id");
b.ToTable("register_projection", (string)null);
});
#pragma warning restore 612, 618
}
}
}

View File

@@ -0,0 +1,28 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<!-- Audit only the packages we choose directly. The one flagged transitive
(System.Security.Cryptography.Xml, NU1903) comes via EF Core's design-time
migration tooling (Microsoft.EntityFrameworkCore.Design, PrivateAssets=all) and
is never shipped at runtime — it is out of our control to upgrade. -->
<NuGetAuditMode>direct</NuGetAuditMode>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.0" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="10.0.0" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.0">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.0.0" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,59 @@
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>
/// The read projection's persistence. This single store is the system's rebuildable read model
/// (PRD §8.4): the Event Subscriber writes to it (projector) and the projection-api reads it
/// (query). Both processes are the one "Read Projection" bounded context and share this schema —
/// not a cross-domain DB share (ADR-0008 reconciles this with CLAUDE.md §8.5).
/// </summary>
public sealed class ProjectionDbContext(DbContextOptions<ProjectionDbContext> options) : DbContext(options)
{
/// <summary>The public-facing register rows (public-safe filtering is tightened in S-09).</summary>
public DbSet<RegisterEntryRow> RegisterEntries => Set<RegisterEntryRow>();
/// <summary>The Event Subscriber's durable log of accepted notifications — idempotency guard and rebuild source.</summary>
public DbSet<ProcessedNotificationRow> ProcessedNotifications => Set<ProcessedNotificationRow>();
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder.Entity<RegisterEntryRow>(e =>
{
e.ToTable("register_projection");
e.HasKey(r => r.Id);
e.Property(r => r.Id).HasColumnName("id");
e.Property(r => r.Status).HasColumnName("status").IsRequired();
e.Property(r => r.Bsn).HasColumnName("bsn");
e.Property(r => r.NaamPlaceholder).HasColumnName("naam_placeholder");
});
modelBuilder.Entity<ProcessedNotificationRow>(e =>
{
e.ToTable("processed_notifications");
e.HasKey(r => r.Key);
e.Property(r => r.Key).HasColumnName("key");
e.Property(r => r.Actie).HasColumnName("actie").IsRequired();
e.Property(r => r.ZaakId).HasColumnName("zaak_id").IsRequired();
e.Property(r => r.ReceivedAt).HasColumnName("received_at");
});
}
}
/// <summary>A register projection row. <c>Bsn</c>/<c>NaamPlaceholder</c> are deferred (ADR-0008).</summary>
public sealed class RegisterEntryRow
{
public required string Id { get; set; }
public required string Status { get; set; }
public string? Bsn { get; set; }
public string? NaamPlaceholder { get; set; }
}
/// <summary>An accepted notification, retained so the projection can be rebuilt without OpenZaak (§8.1).</summary>
public sealed class ProcessedNotificationRow
{
public required string Key { get; set; }
public required string Actie { get; set; }
public required string ZaakId { get; set; }
public DateTimeOffset ReceivedAt { get; set; }
}

View File

@@ -0,0 +1,30 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
namespace Projection.ReadModel;
public static class ServiceCollectionExtensions
{
/// <summary>Register the projection <see cref="ProjectionDbContext"/> against Postgres.</summary>
public static IServiceCollection AddProjectionReadModel(this IServiceCollection services, string connectionString)
=> services.AddDbContext<ProjectionDbContext>(o => o.UseNpgsql(connectionString));
/// <summary>Register the write-side ports (projector store + notification log) used by the Event Subscriber.</summary>
public static IServiceCollection AddProjectionWriteSide(this IServiceCollection services)
{
services.AddScoped<IProjectionStore, EfProjectionStore>();
services.AddScoped<INotificationLog, EfNotificationLog>();
services.AddScoped<NotificationProjector>();
return services;
}
/// <summary>Apply any pending EF migrations. Called once on service start so a fresh stack
/// reaches a usable schema without a manual migration step (DoD: compose up reaches green).</summary>
public static async Task MigrateProjectionAsync(this IServiceProvider services, CancellationToken ct = default)
{
await using var scope = services.CreateAsyncScope();
var db = scope.ServiceProvider.GetRequiredService<ProjectionDbContext>();
await db.Database.MigrateAsync(ct);
}
}

View File

@@ -0,0 +1,43 @@
using Microsoft.EntityFrameworkCore;
using Projection.ReadModel;
var builder = WebApplication.CreateBuilder(args);
var connectionString = builder.Configuration.GetConnectionString("Projection")
?? throw new InvalidOperationException("Missing connection string 'ConnectionStrings:Projection'");
builder.Services.AddProjectionReadModel(connectionString);
var app = builder.Build();
// Ensure the schema exists before serving reads. EF serialises concurrent migrators via the
// migrations-history lock, so it is safe that the Event Subscriber migrates too.
await app.Services.MigrateProjectionAsync();
app.MapGet("/health", () => "Healthy");
// The read side of the projection. Public-safe field filtering is tightened in S-09; for now
// the minimal projection only carries id + status (bsn/naam deferred — ADR-0008).
app.MapGet("/register", async (ProjectionDbContext db, CancellationToken ct) =>
{
var rows = await db.RegisterEntries
.OrderBy(r => r.Id)
.Select(r => new RegisterEntryDto(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.ToListAsync(ct);
return Results.Ok(rows);
});
app.MapGet("/register/{id}", async (string id, ProjectionDbContext db, CancellationToken ct) =>
{
var row = await db.RegisterEntries
.Where(r => r.Id == id)
.Select(r => new RegisterEntryDto(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.SingleOrDefaultAsync(ct);
return row is null ? Results.NotFound() : Results.Ok(row);
});
await app.RunAsync();
public sealed record RegisterEntryDto(string Id, string Status, string? Bsn, string? NaamPlaceholder);
public partial class Program;

View File

@@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<ItemGroup>
<ProjectReference Include="..\Projection.ReadModel\Projection.ReadModel.csproj" />
</ItemGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}

View File

@@ -18,6 +18,7 @@
<ItemGroup> <ItemGroup>
<ProjectReference Include="..\..\services\acl\Acl.Application\Acl.Application.csproj" /> <ProjectReference Include="..\..\services\acl\Acl.Application\Acl.Application.csproj" />
<ProjectReference Include="..\..\services\acl\Acl.Infrastructure\Acl.Infrastructure.csproj" /> <ProjectReference Include="..\..\services\acl\Acl.Infrastructure\Acl.Infrastructure.csproj" />
<ProjectReference Include="..\..\services\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
</ItemGroup> </ItemGroup>
</Project> </Project>

View File

@@ -0,0 +1,19 @@
# language: en
# Drives S-06 (#7). On a zaak-created notification from NRC the Event Subscriber writes a
# rebuildable read-projection row (PRD §8.4). This scenario exercises the use case against an
# in-memory stand-in for the projection store and notification log; real OpenZaak → NRC →
# subscriber delivery is verified by the live-stack check (verify-projection, ADR-0007/#58).
Feature: Register-projectie bijwerken op een zaaknotificatie
Als openbaar register wil ik dat een aangemaakte zaak in de projectie verschijnt
zodat het register de ingediende registratie kan tonen.
Scenario: Een zaaknotificatie levert een rij met status INGEDIEND
Given a zaak is created in OpenZaak with id "11111111-1111-1111-1111-111111111111"
When the NRC notification for that zaak is delivered to the event subscriber
Then the register projection contains a row for "11111111-1111-1111-1111-111111111111" with status "INGEDIEND"
Scenario: Dezelfde notificatie tweemaal levert geen duplicaat
Given a zaak is created in OpenZaak with id "22222222-2222-2222-2222-222222222222"
When the NRC notification for that zaak is delivered to the event subscriber
And the same NRC notification is delivered again
Then the register projection contains exactly one row for "22222222-2222-2222-2222-222222222222"

View File

@@ -0,0 +1,48 @@
using Acceptance.Support;
using EventSubscriber.Application;
using Reqnroll;
using Xunit;
namespace Acceptance.Steps;
/// <summary>Bindings for <c>RegisterProjectieBijwerken.feature</c> (S-06). Reqnroll creates
/// one instance per scenario, so instance fields hold scenario-scoped state.</summary>
[Binding]
public sealed class RegisterProjectieBijwerkenSteps
{
private const string ZaakBase = "http://openzaak:8000/zaken/api/v1/zaken/";
private readonly InMemoryNotificationLog _log = new();
private readonly InMemoryProjectionStore _store = new();
private readonly NotificationProjector _projector;
private Notification? _notification;
public RegisterProjectieBijwerkenSteps() => _projector = new NotificationProjector(_log, _store);
[Given("a zaak is created in OpenZaak with id \"(.*)\"")]
public void GivenAZaakIsCreatedInOpenZaakWithId(string id)
=> _notification = new Notification("zaken", "zaak", "create", new Uri(ZaakBase + id));
[When("the NRC notification for that zaak is delivered to the event subscriber")]
public Task WhenTheNotificationIsDelivered()
=> _projector.HandleAsync(_notification!);
[When("the same NRC notification is delivered again")]
public Task WhenTheSameNotificationIsDeliveredAgain()
=> _projector.HandleAsync(_notification!);
[Then("the register projection contains a row for \"(.*)\" with status \"(.*)\"")]
public async Task ThenTheProjectionContainsARowWithStatus(string id, string status)
{
var entry = Assert.Single(await _store.AllAsync());
Assert.Equal(id, entry.Id);
Assert.Equal(status, entry.Status);
}
[Then("the register projection contains exactly one row for \"(.*)\"")]
public async Task ThenTheProjectionContainsExactlyOneRowFor(string id)
{
await _store.AllAsync();
Assert.Single(_store.RowsFor(id));
}
}

View File

@@ -0,0 +1,40 @@
using EventSubscriber.Application;
namespace Acceptance.Support;
/// <summary>In-memory stand-ins for the Event Subscriber's projection store and notification
/// log, so the S-06 acceptance scenario runs without Postgres (real delivery is a live-stack
/// check). Mirrors the <c>InMemoryZaakGateway</c> approach used for the ACL scenario.</summary>
public sealed class InMemoryNotificationLog : INotificationLog
{
private readonly Dictionary<string, RecordedNotification> _byKey = [];
public Task<bool> TryRecordAsync(RecordedNotification notification, CancellationToken ct = default)
=> Task.FromResult(_byKey.TryAdd(notification.Key, notification));
public Task<IReadOnlyList<RecordedNotification>> AllAsync(CancellationToken ct = default)
=> Task.FromResult<IReadOnlyList<RecordedNotification>>([.. _byKey.Values]);
}
public sealed class InMemoryProjectionStore : IProjectionStore
{
private readonly Dictionary<string, RegisterEntry> _byId = [];
public Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default)
{
_byId[entry.Id] = entry;
return Task.CompletedTask;
}
public Task ClearAsync(CancellationToken ct = default)
{
_byId.Clear();
return Task.CompletedTask;
}
public Task<IReadOnlyList<RegisterEntry>> AllAsync(CancellationToken ct = default)
=> Task.FromResult<IReadOnlyList<RegisterEntry>>([.. _byId.Values]);
public IReadOnlyList<RegisterEntry> RowsFor(string id)
=> [.. _byId.Values.Where(e => e.Id == id)];
}