Compare commits

...

21 Commits

Author SHA1 Message Date
67f0ffb88d docs(domain): demo note for submitting a registration (S-05) (refs #6)
Some checks failed
CI / lint (pull_request) Successful in 1m4s
CI / build (pull_request) Successful in 48s
CI / unit (pull_request) Successful in 58s
CI / mutation (pull_request) Failing after 17m2s
CI / verify-stack (pull_request) Successful in 6m2s
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:32:29 +02:00
5a3f28ac6d ci(domain): containerize, wire into compose, and verify end-to-end (refs #6)
Dockerfile (multi-stage, .NET 10) + .dockerignore for the BIG Domain Service; a
'domain' service in infra/docker-compose.yml (health-checked, depends on acl healthy
and flowable-init completed). run-domain-check.sh drives the full path against the up
stack — seed a published zaaktype, recreate the acl pointed at it (host-consistent),
POST /registrations, and assert the worker opens a zaak and records it. Wired as the
verify-domain Makefile target + a verify-stack CI step; domain added to WAIT_SVCS and
the log dump. seed_catalogus.py now emits a machine-readable ZAAKTYPE_URL line.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:31:45 +02:00
e9a873c152 test(domain): mutation baseline 90 (achieved 97.7%) + CI/Makefile wiring (refs #6)
Stryker.NET config for the domain service (break 90, the repo's ratchet floor),
excluding the OpenZaakJobPump hosted-shell from mutation. Hardened the unit tests
to kill survivors — Basic-credential value, variable types, null/failure response
paths, option defaults, guard clauses, save counts and log output — leaving only
two documented equivalent mutants (Stryker-disabled). make mutation runs the domain
ratchet and CI uploads its report alongside the others.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:24:56 +02:00
79dcd8f14b test(domain): acceptance scenario for submitting a registration (refs #6)
Use-case-level BDD (Reqnroll) for S-05: a zorgprofessional submits a registration;
the Domain Service starts the registratie process and the OpenZaakAanmaken external
task opens a zaak via the ACL, recorded on the aggregate (ADR-0009). Driven against
in-memory Workflow Client and ACL stand-ins; real Flowable+ACL+OpenZaak delivery is
the live-stack verify-domain check.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:15:23 +02:00
22ab38f328 feat(domain): expose POST /registrations and the read endpoint (refs #6)
The BIG Domain Service Api wires the use cases and the hosted job worker:
POST /registrations creates the aggregate and starts the registratie process,
returning 202 with a location; GET /registrations/{id} reads the aggregate so
the eventually-opened zaak URL can be observed (ADR-0009). The Workflow Client
is registered once behind both Flowable ports; the ACL client and in-memory
store complete the wiring. Verified against a live flowable-rest: submit starts
a parked process and the worker polls it.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:13:27 +02:00
0d34d60797 feat(domain): implement the Flowable Workflow Client and ACL client (refs #6)
FlowableWorkflowClient speaks flowable-rest's REST API (Basic auth): start a
registratie process with the registrationId variable, acquire OpenZaakAanmaken
external-worker jobs and parse their registrationId, complete a job with the
zaakUrl variable — the contract verified against a live engine. AclHttpClient
POSTs the bsn to the ACL and returns the zaak URL. InMemoryRegistrationStore is
a concurrent-dictionary upsert. OpenZaakJobProcessor drains parked jobs, opening
a zaak per job and completing it, leaving failures for redelivery; OpenZaakJobPump
is the hosted polling shell that drives it on an interval (ADR-0009).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:10:21 +02:00
6d4adaf957 test(domain): Workflow Client, ACL client, store and job processor (refs #6)
Failing infrastructure unit tests (stub HttpMessageHandler, fakes):
- FlowableWorkflowClient starts a process with the registrationId variable and
  returns the instance id; acquires OpenZaakAanmaken jobs (topic/workerId/lock)
  and parses their registrationId; completes a job with the zaakUrl variable —
  request URIs match flowable-rest's service/ and external-job-api/ paths.
- AclHttpClient POSTs the bsn to the ACL and returns the zaak URL.
- InMemoryRegistrationStore saves/reads/upserts by id.
- OpenZaakJobProcessor acquires, opens a zaak, completes the job; leaves a failing
  job uncompleted for redelivery; polls harmlessly when idle.

Adapters are stubs so the tests compile and fail on their assertions; the green
commit implements them against the REST contract verified on a live Flowable.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:08:56 +02:00
39b2388a9d feat(domain): implement SubmitRegistration and OpenZaakWorker (refs #6)
SubmitRegistration creates the aggregate, persists it, starts the registratie
process via the Workflow Client, records the instance id and upserts. OpenZaakWorker
loads the correlated registration, opens a zaak via the ACL, attaches it and saves;
an unknown registration throws (job redelivered), and an already-opened zaak short-
circuits without opening a second one (§8.6).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 17:00:05 +02:00
8d176c2603 test(domain): SubmitRegistration + OpenZaakWorker use cases (refs #6)
Failing application-layer tests over fake ports (IWorkflowClient, IAclClient,
IRegistrationStore):
- Submit persists an INGEDIEND registration and starts the registratie process,
  recording the instance id — and persists *before* starting, so the worker can
  correlate the OpenZaakAanmaken job back to its aggregate (ADR-0009).
- The worker opens a zaak via the ACL and attaches it; an unknown registration
  throws (job left for redelivery); a redelivered job is idempotent and opens no
  second zaak (§8.6).

Handlers are stubs (no persistence / no ACL call) so the tests compile and fail
on their assertions; the green commit implements them.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:59:27 +02:00
53751fd1bc feat(domain): implement the Registration aggregate invariants (refs #6)
Submit requires a bsn and starts the aggregate in INGEDIEND; the started
process-instance id is recorded; AttachZaak stores the ACL's zaak URL,
tolerating a duplicate (at-least-once worker delivery) and rejecting a
conflicting URL, without advancing the status.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:56:31 +02:00
cc9e7852e1 test(domain): Registration aggregate invariants (refs #6)
Failing unit tests for the Registration aggregate root: a submission starts
in INGEDIEND carrying its bsn, an empty/whitespace/null bsn is rejected, the
started process-instance id is remembered, and attaching the zaak the ACL
opened records its URL idempotently (a conflicting URL is rejected) while the
status stays INGEDIEND.

The aggregate is a stub (no-op mutators, empty bsn) so the tests compile and
fail on their assertions; the green commit implements the invariants.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:54:59 +02:00
c9edf27a48 arch(domain): ADR-0009 external-task job-worker pattern (refs #6, #60)
The Domain Service drives the OpenZaakAanmaken external-worker task as a
hosted job worker (PRD §36): POST /registrations starts the registratie
process and returns; a polling worker acquires the job, opens a zaak via
the ACL (§8.1), attaches the zaak URL to the aggregate, and completes the
job. The Workflow Client is the only Flowable client (§8.2); the worker
logic is an Application service over ports. Registration state is in-memory
for the minimal slice (the read path is the projection, S-06).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 16:51:40 +02:00
c3ccffe417 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
2026-06-30 13:56:59 +00:00
0d0778036e docs(arch): ADR-0008 read projection store + demo note for the event path (refs #7)
All checks were successful
CI / lint (pull_request) Successful in 1m0s
CI / build (pull_request) Successful in 45s
CI / unit (pull_request) Successful in 54s
CI / mutation (pull_request) Successful in 2m16s
CI / verify-stack (pull_request) Successful in 5m34s
ADR-0008 records the read-projection design: one rebuildable store shared by the Event
Subscriber (writer) and projection-api (reader) as one CQRS bounded context (reconciled
with §8.5), idempotency + rebuild from the notification log (no OpenZaak access, §8.1),
the deferred bsn/naam, and the new EF Core + Npgsql dependency. Add a demo-script entry
walking the OZ→NRC→subscriber→projection-api path and wire both into the MkDocs nav.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 15:11:36 +02:00
fa8382fc02 ci(infra): run the Event Subscriber + projection-api in compose and verify end-to-end (refs #7)
Add projection-db + the two services to both compose files (host ports 8110/8120), their
Dockerfiles (repo-root context — they share Projection.ReadModel), and a runner-safe
verify-projection check (infra/run-projection-check.sh) that registers the abonnement at the
real subscriber, creates a zaak and asserts projection-api serves an INGEDIEND row. Wire it
into make (verify-projection, verify, WAIT_SVCS) and the CI verify-stack job, and run the
event-subscriber Stryker ratchet in `make mutation` + upload its report.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 15:11:24 +02:00
06d8d13e19 feat(event-subscriber): enforce the callback bearer before reading the body (refs #7)
Check the Authorization header before deserializing the notification, and read/parse the
body manually. NRC probes a new abonnement's callback with a request that has neither the
configured auth nor a valid notification body, and refuses to register unless it gets a 401
(not a 400) — ADR-0007. Mirrors the verify harness's sink contract.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 15:11:14 +02:00
a111e5cc20 test(event-subscriber): ratchet projector mutation baseline to 100% (refs #7)
Sharpen the projector tests so Stryker has no survivors (was 75%): assert a replayed
delivery never reaches the store (upsert count, not just row count), that two distinct
zaken get distinct rows (pins the idempotency key), that rebuild clears stale rows, and
a Theory over wrong kanaal/resource/actie combinations (pins the zaken/zaak/create guard).
Add the per-service Stryker config + solution; break threshold 90 (CLAUDE.md §5).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 15:11:06 +02:00
7ef63c7ae9 feat(projection): persist the read projection and expose webhook + read APIs (refs #7)
Add the projection persistence and the two services around it:

- Projection.ReadModel: a shared EF Core (Npgsql) read model owning the projection
  schema — register_projection + the subscriber's processed_notifications log — plus
  EfProjectionStore / EfNotificationLog (atomic record-or-skip on the PK for idempotency)
  and the initial migration. One rebuildable store, written by the subscriber and read
  by projection-api (ADR-0008).
- EventSubscriber.Api: POST /notifications NRC callback (enforces the abonnement bearer,
  401 without it per ADR-0007), POST /admin/rebuild, /health. Migrates on start.
- ProjectionApi.Api: GET /register, GET /register/{id}, /health — the read side.

dotnet-ef pinned as a local tool for migrations; NuGetAuditMode=direct so EF's
design-time-only tooling transitive doesn't flag the shipped build.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 14:55:04 +02:00
017cd5e66b feat(event-subscriber): project zaak-created notifications into the read projection (refs #7)
Implement NotificationProjector: a zaken/zaak/create notification records the delivery
in the notification log (atomic record-or-skip for idempotency, §8.6) and upserts an
INGEDIEND projection row keyed by zaak id; other channels/actions are ignored. Rebuild
clears the projection and replays the log — no OpenZaak access needed (§8.1). bsn/naam
are deferred (ADR-0008).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 14:48:08 +02:00
c70840e5b7 test(event-subscriber): project zaak-created notifications into the read projection (refs #7)
Failing unit + acceptance tests for the Event Subscriber's NotificationProjector:
a zaken/zaak/create notification yields one INGEDIEND projection row, duplicate
deliveries collapse to one row, non-zaak/non-create notifications are ignored, and
a rebuild repopulates the projection from the durable notification log (PRD §8.4).

The projector is a no-op stub so the tests compile and fail on the assertions; the
implementation follows in the green commit. The notification log doubles as the
idempotency guard and rebuild source so a rebuild needs no OpenZaak access (§8.1).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 14:47:16 +02:00
32c98f00db Merge pull request 'feat(infra): wire OpenZaak → Open Notificaties notifications (closes #56)' (#57) from feat/56-nrc-notification-wiring into main
All checks were successful
CI / lint (push) Successful in 52s
CI / build (push) Successful in 40s
CI / unit (push) Successful in 48s
CI / mutation (push) Successful in 1m35s
CI / verify-stack (push) Successful in 4m44s
Reviewed-on: #57
2026-06-30 12:29:43 +00:00
81 changed files with 3163 additions and 10 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,18 @@ 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
- uses: https://github.com/actions/upload-artifact@v3
if: always()
with:
name: domain-mutation-report
path: services/domain/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 +92,14 @@ 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
- name: Domain → Flowable → ACL → OpenZaak
run: make verify-domain
# 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-db flowable-rest flowable-init keycloak acl bff domain 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 domain 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,17 @@ 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,
# with branching logic, so it sets the repo-wide baseline; later slices ratchet it up. # services/event-subscriber/stryker-config.json and services/domain/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
cd services/domain && 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 +133,17 @@ 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-domain: domain → Flowable → ACL → OpenZaak end-to-end (S-05), against the
## already-running stack. Recreates the acl service to inject the seeded zaaktype URL.
verify-domain:
bash infra/run-domain-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 +152,9 @@ 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 \
&& bash infra/run-domain-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.

View File

@@ -0,0 +1,88 @@
# ADR-0009: The Domain Service drives Flowable as an external-task job worker
- **Status:** Accepted
- **Date:** 2026-06-30
- **Deciders:** Respellion engineering
- **Relates to:** S-05 (#6); proposal #60; builds on ADR-0001 (loose coupling, §8.1/§8.2), S-03 (#4, the `registratie` BPMN), S-04 (#5, the ACL `OpenZaak` operation)
## Context
S-05 (#6) adds the **BIG Domain Service**. Submitting a registration must: create a
`Registration` aggregate, **start the Flowable `registratie` process** (S-03), have the
`OpenZaakAanmaken` task **open a zaak via the ACL** (S-04), and store the resulting zaak URL
back on the aggregate.
`OpenZaakAanmaken` is a Flowable **external-worker** service task (`flowable:type="external-worker"`,
topic `OpenZaakAanmaken`). Flowable does not push it anywhere — it parks the job and waits for a
worker to **acquire and lock** it, do the work, and **complete** it. Two coupling rules constrain
who may do what:
- **§8.2 — the Workflow Client is the only code that talks to Flowable.** BPMN models never embed
OpenZaak knowledge; they ask the Workflow Client to execute external tasks.
- **§8.1 — the ACL is the only code that talks to ZGW.** The worker opens the zaak *through the ACL*,
never by constructing ZGW URLs itself.
This is an ADR-worthy moment (§14): a service boundary is defined and both coupling rules are
exercised. The open question is *how* the external task is driven.
## Decision
**The Domain Service drives the `OpenZaakAanmaken` task as a hosted external-task job worker
(PRD §36). Orchestration is eventually consistent, not request-synchronous.**
- **`POST /registrations` is fast and side-effecting only on the domain side.** It creates the
`Registration` aggregate in state `INGEDIEND`, persists it, and asks the Workflow Client to start
one `registratie` process instance, recording the process-instance id on the aggregate. It returns
immediately; it does **not** wait for the zaak to be opened.
- **A hosted worker polls Flowable for `OpenZaakAanmaken` jobs.** It acquires and locks a job, calls
the ACL `OpenZaak` operation (§8.1), attaches the returned zaak URL to the matching aggregate
(`Registration.AttachZaak`), and completes the job in Flowable. The process then runs to its end
event.
- **The Workflow Client is the only Flowable client (§8.2).** It lives in the Domain Service's
`Infrastructure` layer and speaks Flowable's REST API (start process-instance; acquire/lock/complete
external-worker jobs). No other code — not the Application layer, not the BPMN — knows Flowable
exists.
- **The worker *logic* is an Application service over ports**, not Flowable-aware code. `OpenZaakWorker`
takes an acquired job (topic + the registration id it carries), calls `IAclClient` and
`IRegistrationStore`, and returns the zaak URL to complete with. The **polling loop** is a thin
`BackgroundService` in `Infrastructure` that fetches jobs via the Workflow Client and feeds them to
the worker. So the orchestration is covered by fast unit tests against fakes; only the REST framing
needs a container integration test.
## Scope decisions for the minimal slice
- **Registration persistence is in-memory.** The walking skeleton's *read* path is fed by
NRC → Event Subscriber → projection (S-06, #7), not by the domain database. An EF-backed domain
store buys nothing the demo needs yet, so it is a documented follow-up; the `IRegistrationStore`
port keeps that change additive. (PRD §88 envisions EF Core for the domain DB eventually.)
- **The aggregate's state machine is minimal:** `INGEDIEND` on submission. Later flows (withdrawal,
beoordeling, herregistratie) add states in their own slices — they are out of scope here.
- **No bsn flows to ZGW yet.** The ACL `OpenZaak` operation already default-fills the ZGW-mandatory
fields (ADR-0003) and takes the bsn as its domain payload; the domain hands it through unchanged.
## Consequences
- **Positive:** the submit request is decoupled from ACL/OpenZaak latency; the documented Common
Ground pattern (external-task worker) is realised; both coupling rules (§8.1, §8.2) hold with the
Flowable knowledge isolated to one Infrastructure class; the orchestration is unit-testable.
- **Negative / deferred:**
- Eventual consistency: immediately after `POST /registrations` the aggregate has no zaak URL yet.
Acceptable — the read side is the projection, not the domain store.
- In-memory registration state is lost on restart; fine for the skeleton, replaced by an EF store
in a follow-up.
- The worker polls (no push); poll interval is a tuning knob, not a correctness concern, since
Flowable holds the job until completed.
## Alternatives considered
- **Synchronous acquire+complete inside the `POST /registrations` request** — rejected: simpler and
deterministic, but couples the submit request to ACL/OpenZaak latency and failure, and is not the
external-task worker pattern PRD §36 mandates. It would also make the request fail if OpenZaak is
briefly down, instead of the job simply staying parked for the worker to retry.
- **A standalone Workflow Client service, separate from the Domain Service** — rejected for this
slice: the worker needs the domain's aggregate store and the ACL client anyway, and PRD §9 places
the Workflow Client inside the Domain Service deployment. A separate process adds a hop and a
shared store for no current benefit.
- **Flowable pushes to a webhook instead of being polled** — rejected: Flowable's external-worker
model is pull-based (acquire/lock/complete); a push shim would re-implement it with weaker
delivery guarantees.

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

@@ -0,0 +1,67 @@
# 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-05 — BIG Domain Service: submit a registration
**Outcome:** submitting a registration starts a Flowable process; the external-task worker
opens a zaak via the ACL and records it on the aggregate — the upstream half of the skeleton
that produces the zaak S-06 then projects.
**The path:** domain `POST /registrations` → Flowable `registratie` process → `OpenZaakAanmaken`
worker → ACL → OpenZaak; `GET /registrations/{id}` shows the opened zaak (ADR-0009).
```bash
# 1. Bring the full stack up (seeds config, builds our services, waits for health).
make up
# 2. Drive the full path end-to-end. This also seeds a published BIG zaaktype and points the
# ACL at it (the zaak's zaaktype URL is server-assigned, so it isn't known at bring-up).
make verify-domain # → "OK — the domain opened a zaak and recorded it on the registration"
# 3. Submit one yourself (domain on host port 8130). Returns 202 + a Location to read back.
loc=$(curl -fsS -D - -o /dev/null -X POST http://localhost:8130/registrations \
-H 'Content-Type: application/json' -d '{"bsn":"123456782"}' | sed -n 's/\r$//; s/^[Ll]ocation: //p')
# 4. The worker opens the zaak off the request path (eventual consistency, ADR-0009); poll
# until zaakUrl is filled. (Step 2 must have run first, so the ACL knows the zaaktype.)
curl -fsS "http://localhost:8130$loc" | jq
# → { "registrationId": "...", "status": "Ingediend", "zaakUrl": "http://.../zaken/api/v1/zaken/<uuid>" }
```
> Registration state is in-memory for this slice (ADR-0009); the rebuildable read model is the
> projection (S-06), fed by the very zaak this flow opens.
---
## 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
@@ -283,7 +285,9 @@ services:
dockerfile: Dockerfile dockerfile: Dockerfile
image: register-referentie/acl:dev image: register-referentie/acl:dev
environment: environment:
Acl__OpenZaak__BaseUrl: http://openzaak:8000/ # Overridable so verify-domain can point the ACL at the same OpenZaak host that
# owns the seeded zaaktype URL (host-consistent zaak creation, ADR-0009).
Acl__OpenZaak__BaseUrl: ${ACL_OPENZAAK_BASEURL:-http://openzaak:8000/}
Acl__OpenZaak__ClientId: big-reference-seed Acl__OpenZaak__ClientId: big-reference-seed
Acl__OpenZaak__Secret: insecure-dev-secret-change-me Acl__OpenZaak__Secret: insecure-dev-secret-change-me
Acl__Defaults__Bronorganisatie: "517439943" Acl__Defaults__Bronorganisatie: "517439943"
@@ -304,6 +308,36 @@ services:
condition: service_healthy condition: service_healthy
networks: [cg] networks: [cg]
# ── BIG Domain Service (S-05) ──────────────────────────────────────────────
# Orchestrates a registration: POST /registrations creates the aggregate and
# starts the registratie Flowable process; a hosted worker acquires the
# OpenZaakAanmaken job, opens a zaak via the ACL and completes it (ADR-0009).
# Talks only to Flowable (Workflow Client, §8.2) and the ACL (§8.1).
domain:
build:
context: ../services/domain
dockerfile: Dockerfile
image: register-referentie/domain:dev
environment:
Flowable__BaseUrl: http://flowable-rest:8080/flowable-rest/
Flowable__Username: rest-admin
Flowable__Password: test
Acl__BaseUrl: http://acl:8080/
ports:
- "8130:8080"
healthcheck:
test: ["CMD", "curl", "-fsS", "http://localhost:8080/health"]
interval: 5s
timeout: 3s
retries: 5
start_period: 10s
depends_on:
acl:
condition: service_healthy
flowable-init:
condition: service_completed_successfully
networks: [cg]
# ── BFF ────────────────────────────────────────────────────────────────── # ── BFF ──────────────────────────────────────────────────────────────────
bff: bff:
build: build:
@@ -320,10 +354,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.

View File

@@ -210,6 +210,10 @@ def main():
print(f"zaaktypen in BIG: {names}") print(f"zaaktypen in BIG: {names}")
assert "BIG-REGISTRATIE" in names, "BIG-REGISTRATIE not listed" assert "BIG-REGISTRATIE" in names, "BIG-REGISTRATIE not listed"
state = "published" if PUBLISH else "concept" state = "published" if PUBLISH else "concept"
# Machine-readable line so callers (e.g. infra/run-domain-check.sh) can capture the
# zaaktype URL to configure the ACL's default-fill (ADR-0003/0009).
zt_url = next(z["url"] for z in zaaktypen if z.get("identificatie") == "BIG-REGISTRATIE")
print(f"ZAAKTYPE_URL {zt_url}")
print(f"OK — BIG catalogus seeded (BIG-REGISTRATIE {state} + bsn eigenschap)") print(f"OK — BIG catalogus seeded (BIG-REGISTRATIE {state} + bsn eigenschap)")

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

@@ -0,0 +1,68 @@
#!/usr/bin/env bash
#
# Verify the BIG Domain Service end-to-end (S-05) against an ALREADY-RUNNING full stack:
# domain → Flowable (start the registratie process + external-task worker) → ACL → OpenZaak.
# Submits a registration to the domain and asserts the worker opens a zaak in OpenZaak and
# records its URL on the aggregate (ADR-0009).
#
# The seeded zaaktype URL is server-assigned, so it isn't knowable at initial bring-up. This
# script therefore seeds a published BIG zaaktype and recreates the `acl` service configured to
# default-fill it — pointing the ACL at the SAME OpenZaak host that owns the URL, so zaak creation
# is host-consistent (exactly the configuration the ACL integration test proves, ADR-0006). That
# one recreate aside, the caller owns stack bring-up + teardown.
#
# All in-network, reaching services by container IP (a single-label host isn't URL-valid; the
# runner can't reach published ports — gitea-actions-gotchas.md §5/§6). Plain docker primitives.
set -euo pipefail
here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
root="$(cd "$here/.." && pwd)"
compose="$root/infra/docker-compose.yml"
ip() { docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$1"; }
oz="$(docker ps -q --filter 'name=[-_]openzaak[-_]' | head -1)"
dom="$(docker ps -q --filter 'name=domain' | head -1)"
[ -n "$oz" ] || { echo "ERROR: no running OpenZaak container — bring the stack up first" >&2; exit 1; }
[ -n "$dom" ] || { echo "ERROR: no running domain container — 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")"; dom_ip="$(ip "$dom")"
oz_base="http://$oz_ip:8000"
echo ">> openzaak=$oz_ip domain=$dom_ip network=$net"
echo ">> seeding a published BIG zaaktype (idempotent) and capturing its URL"
sid="$(docker create --network "$net" -e "OZ_BASE=$oz_base" -e OZ_PUBLISH=1 python:3-slim python /seed.py)"
docker cp "$here/openzaak/seed_catalogus.py" "$sid:/seed.py" >/dev/null
zt_url="$(docker start -a "$sid" | sed -n 's/^ZAAKTYPE_URL //p' | head -1)"
docker rm -f "$sid" >/dev/null
[ -n "$zt_url" ] || { echo "ERROR: seed did not report a ZAAKTYPE_URL" >&2; exit 1; }
echo ">> zaaktype: $zt_url"
echo ">> recreating the acl service pointed at the seeded zaaktype (host-consistent)"
ACL_ZAAKTYPE_URL="$zt_url" ACL_OPENZAAK_BASEURL="$oz_base/" docker compose -f "$compose" up -d acl
WAIT_TIMEOUT="${WAIT_TIMEOUT:-120}" bash "$here/wait-healthy.sh" acl
echo ">> submitting a registration to the domain"
loc="$(docker run --rm --network "$net" curlimages/curl:latest \
-fsS -D - -o /dev/null -X POST "http://$dom_ip:8080/registrations" \
-H 'Content-Type: application/json' -d '{"bsn":"123456782"}' \
| sed -n 's/\r$//; s/^[Ll]ocation: //p' | head -1)"
[ -n "$loc" ] || { echo "ERROR: POST /registrations returned no Location" >&2; exit 1; }
echo ">> registration accepted at $loc"
echo ">> polling the domain until the worker records the opened zaak"
for _ in $(seq 1 30); do
body="$(docker run --rm --network "$net" curlimages/curl:latest \
-fsS "http://$dom_ip:8080$loc" 2>/dev/null || true)"
if echo "$body" | grep -q '/zaken/api/v1/zaken/'; then
echo "OK — the domain opened a zaak and recorded it on the registration:"
echo "$body" | cut -c1-300
exit 0
fi
sleep 2
done
echo "FAIL — the registration never received a zaak URL" >&2
echo "--- domain log ---" >&2; docker logs "$dom" 2>&1 | tail -15 >&2
acl="$(docker ps -q --filter 'name=[-_]acl[-_]' | head -1)"
[ -n "$acl" ] && { echo "--- acl log ---" >&2; docker logs "$acl" 2>&1 | tail -15 >&2; }
exit 1

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,10 @@ 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
- "ADR-0009: External-task job worker": architecture/adr-0009-external-task-job-worker.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

@@ -7,10 +7,26 @@
<Project Path="services/acl/Acl.IntegrationTests/Acl.IntegrationTests.csproj" /> <Project Path="services/acl/Acl.IntegrationTests/Acl.IntegrationTests.csproj" />
<Project Path="services/acl/Acl.Tests/Acl.Tests.csproj" /> <Project Path="services/acl/Acl.Tests/Acl.Tests.csproj" />
</Folder> </Folder>
<Folder Name="/services/domain/">
<Project Path="services/domain/Big.Domain/Big.Domain.csproj" />
<Project Path="services/domain/Big.Application/Big.Application.csproj" />
<Project Path="services/domain/Big.Infrastructure/Big.Infrastructure.csproj" />
<Project Path="services/domain/Big.Api/Big.Api.csproj" />
<Project Path="services/domain/Big.Tests/Big.Tests.csproj" />
</Folder>
<Folder Name="/services/bff/"> <Folder Name="/services/bff/">
<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,4 @@
**/bin
**/obj
**/*.user
StrykerOutput

View File

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

View File

@@ -0,0 +1,64 @@
using Big.Application;
using Big.Domain;
using Big.Infrastructure;
var builder = WebApplication.CreateBuilder(args);
// Options bound from configuration (compose sets Flowable__* and Acl__* env vars).
builder.Services.AddSingleton(sp => sp.GetRequiredService<IConfiguration>()
.GetSection("Flowable").Get<FlowableOptions>()
?? throw new InvalidOperationException("Missing configuration section 'Flowable'"));
builder.Services.AddSingleton(sp => sp.GetRequiredService<IConfiguration>()
.GetSection("Acl").Get<AclOptions>()
?? throw new InvalidOperationException("Missing configuration section 'Acl'"));
// The in-memory registration store is shared between the submit endpoint and the worker (ADR-0009).
builder.Services.AddSingleton<IRegistrationStore, InMemoryRegistrationStore>();
// The Workflow Client is one type behind two ports (start side + worker side); both resolve to the
// same HttpClient-backed implementation — the only code that talks to Flowable (§8.2).
builder.Services.AddHttpClient<FlowableWorkflowClient>();
builder.Services.AddTransient<IWorkflowClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
builder.Services.AddTransient<IExternalWorkerClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
builder.Services.AddHttpClient<IAclClient, AclHttpClient>();
builder.Services.AddScoped<SubmitRegistration>();
builder.Services.AddScoped<OpenZaakWorker>();
builder.Services.AddScoped<OpenZaakJobProcessor>();
// The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion.
builder.Services.AddHostedService<OpenZaakJobPump>();
var app = builder.Build();
app.MapGet("/health", () => "Healthy");
// Submit a registration. The aggregate is created (INGEDIEND) and the registratie process started;
// the zaak is opened later, off the request path, by the worker — so this returns 202 Accepted with
// a location to read the registration's progress (ADR-0009, eventual consistency).
app.MapPost("/registrations", async (SubmitRegistrationRequest body, SubmitRegistration submit, CancellationToken ct) =>
{
var id = await submit.HandleAsync(new SubmitRegistrationCommand(body.Bsn), ct);
return Results.Accepted($"/registrations/{id}", new RegistrationResponse(id.ToString(), RegistrationStatus.Ingediend.ToString(), null));
});
// Read a registration. Its zaak URL appears once the worker has opened the zaak (eventually).
app.MapGet("/registrations/{id}", async (string id, IRegistrationStore store, CancellationToken ct) =>
{
if (!Guid.TryParse(id, out var guid))
return Results.NotFound();
var registration = await store.GetAsync(new RegistrationId(guid), ct);
return registration is null
? Results.NotFound()
: Results.Ok(new RegistrationResponse(
registration.Id.ToString(), registration.Status.ToString(), registration.ZaakUrl?.ToString()));
});
await app.RunAsync();
public sealed record SubmitRegistrationRequest(string Bsn);
public sealed record RegistrationResponse(string RegistrationId, string Status, string? ZaakUrl);
public partial class Program;

View File

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

View File

@@ -0,0 +1,15 @@
<Project Sdk="Microsoft.NET.Sdk">
<!-- The application layer (CLAUDE.md §9): use cases over ports. Depends on Domain only;
Infrastructure implements the ports. -->
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Big.Domain\Big.Domain.csproj" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,36 @@
using Big.Domain;
namespace Big.Application;
/// <summary>
/// Handles one acquired <c>OpenZaakAanmaken</c> external-worker job (ADR-0009): load the registration
/// the job correlates to, open a zaak for it via the ACL (§8.1), attach the zaak to the aggregate, and
/// return the zaak URL so the caller can complete the Flowable job. Pure application logic over ports —
/// it knows nothing of Flowable; the polling loop that feeds it jobs lives in Infrastructure.
/// </summary>
public sealed class OpenZaakWorker(IRegistrationStore store, IAclClient acl)
{
/// <summary>
/// Process the job and return the URL of the (existing or newly opened) zaak. Idempotent: if the
/// registration already has a zaak — a job redelivered after its completion was lost — it returns
/// that zaak without opening a second one (§8.6, at-least-once delivery). An unknown registration
/// is an error: it throws, leaving the job un-completed for Flowable to redeliver.
/// </summary>
public async Task<Uri> HandleAsync(OpenZaakJob job, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(job);
var registration = await store.GetAsync(job.RegistrationId, ct)
?? throw new InvalidOperationException(
$"No registration {job.RegistrationId} for OpenZaakAanmaken job {job.JobId}.");
// A redelivered job whose zaak was already opened completes without opening a second one.
if (registration.ZaakUrl is not null)
return registration.ZaakUrl;
var zaakUrl = await acl.OpenZaakAsync(registration.Bsn, ct);
registration.AttachZaak(zaakUrl);
await store.SaveAsync(registration, ct);
return zaakUrl;
}
}

View File

@@ -0,0 +1,47 @@
using Big.Domain;
namespace Big.Application;
/// <summary>
/// The port to the workflow engine. Implemented by the Workflow Client in Infrastructure — the
/// <em>only</em> code that talks to Flowable (CLAUDE.md §8.2). The application asks it to start the
/// registratie process; it never knows Flowable exists.
/// </summary>
public interface IWorkflowClient
{
/// <summary>
/// Start one <c>registratie</c> process instance for the given registration, carrying the
/// registration id so the <c>OpenZaakAanmaken</c> external task can be correlated back to its
/// aggregate. Returns the process instance id.
/// </summary>
Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default);
}
/// <summary>
/// The port to the Anti-Corruption Layer. Implemented in Infrastructure by an HTTP client to the
/// ACL service — the <em>only</em> code that talks to ZGW (CLAUDE.md §8.1). The domain hands over a
/// bsn; the ACL default-fills the ZGW-mandatory fields (ADR-0003) and returns the created zaak URL.
/// </summary>
public interface IAclClient
{
Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default);
}
/// <summary>
/// Persistence port for the <see cref="Registration"/> aggregate. In-memory for the minimal slice
/// (ADR-0009); an EF-backed store is a documented follow-up, and this port keeps that change additive.
/// </summary>
public interface IRegistrationStore
{
/// <summary>Insert or update the registration, keyed on its id.</summary>
Task SaveAsync(Registration registration, CancellationToken ct = default);
/// <summary>Load a registration by id, or <c>null</c> if none exists.</summary>
Task<Registration?> GetAsync(RegistrationId id, CancellationToken ct = default);
}
/// <summary>
/// An acquired <c>OpenZaakAanmaken</c> external-worker job: the Flowable job id (needed to complete
/// it) and the registration id it carries as a process variable.
/// </summary>
public sealed record OpenZaakJob(string JobId, RegistrationId RegistrationId);

View File

@@ -0,0 +1,33 @@
using Big.Domain;
namespace Big.Application;
/// <summary>A zorgprofessional's request to register, in domain language. No ZGW concepts.</summary>
public sealed record SubmitRegistrationCommand(string Bsn);
/// <summary>
/// The submit use case: create the <see cref="Registration"/> aggregate (INGEDIEND), persist it,
/// then start the registratie workflow process and record its instance id. It returns as soon as
/// the process is started — opening the zaak happens later, off the request path, in the external-task
/// worker (ADR-0009). Persisting <em>before</em> starting the process closes the race where the worker
/// acquires the OpenZaakAanmaken job before the aggregate it correlates to exists.
/// </summary>
public sealed class SubmitRegistration(IRegistrationStore store, IWorkflowClient workflow)
{
public async Task<RegistrationId> HandleAsync(SubmitRegistrationCommand command, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(command);
var registration = Registration.Submit(command.Bsn);
// Persist before starting the process so the worker can correlate the OpenZaakAanmaken
// job back to an aggregate that already exists (ADR-0009).
await store.SaveAsync(registration, ct);
var processInstanceId = await workflow.StartRegistrationProcessAsync(registration.Id, ct);
registration.RecordProcessStarted(processInstanceId);
await store.SaveAsync(registration, ct);
return registration.Id;
}
}

View File

@@ -0,0 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<!-- The domain layer (CLAUDE.md §9): aggregates, value objects, invariants.
Pure C# — no external dependencies, no infrastructure concerns. -->
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,67 @@
namespace Big.Domain;
/// <summary>
/// The Registration aggregate root (CLAUDE.md §2.2): a zorgprofessional's submission to the BIG
/// register. It owns its lifecycle invariants — it starts <see cref="RegistrationStatus.Ingediend"/>
/// on submission, remembers the Flowable process that drives it, and records the zaak the ACL opens.
/// </summary>
public sealed class Registration
{
private Registration(RegistrationId id, string bsn)
{
Id = id;
Bsn = bsn;
Status = RegistrationStatus.Ingediend;
}
public RegistrationId Id { get; }
/// <summary>The citizen-service number of the submitting zorgprofessional. Handed to the ACL
/// as the domain payload; the domain never constructs ZGW concepts from it (§8.1).</summary>
public string Bsn { get; }
public RegistrationStatus Status { get; private set; }
/// <summary>The Flowable process instance driving this registration, once started.</summary>
public string? ProcessInstanceId { get; private set; }
/// <summary>The zaak the ACL opened for this registration, once the external task has run.</summary>
public Uri? ZaakUrl { get; private set; }
/// <summary>Submit a new registration. It begins in <see cref="RegistrationStatus.Ingediend"/>.</summary>
public static Registration Submit(string bsn)
{
ArgumentException.ThrowIfNullOrWhiteSpace(bsn);
return new Registration(RegistrationId.New(), bsn);
}
/// <summary>Record that the registratie workflow process has been started for this registration.</summary>
public void RecordProcessStarted(string processInstanceId)
{
ArgumentException.ThrowIfNullOrWhiteSpace(processInstanceId);
ProcessInstanceId = processInstanceId;
}
/// <summary>
/// Attach the zaak the ACL opened. The external-task worker may deliver the same job more than
/// once (at-least-once), so re-attaching the identical URL is a no-op; a different URL signals a
/// genuine conflict and is rejected. The status stays <see cref="RegistrationStatus.Ingediend"/>:
/// opening the zaak does not advance the registration's lifecycle in this slice.
/// </summary>
public void AttachZaak(Uri zaakUrl)
{
ArgumentNullException.ThrowIfNull(zaakUrl);
if (ZaakUrl is not null)
{
if (ZaakUrl != zaakUrl)
throw new InvalidOperationException(
$"Registration {Id} already has zaak {ZaakUrl}; cannot attach a different zaak {zaakUrl}.");
// Stryker disable once Statement : equivalent — re-assigning the identical URL below is a
// no-op, so removing this early return is behaviourally indistinguishable.
return;
}
ZaakUrl = zaakUrl;
}
}

View File

@@ -0,0 +1,15 @@
namespace Big.Domain;
/// <summary>The identity of a <see cref="Registration"/> aggregate — opaque, server-assigned,
/// and distinct from the OpenZaak zaak id (which the projection keys on). A value object so the
/// id can travel as a Flowable process variable and back without ever being a bare string.</summary>
public readonly record struct RegistrationId(Guid Value)
{
/// <summary>Mint a fresh identity for a newly submitted registration.</summary>
public static RegistrationId New() => new(Guid.NewGuid());
/// <summary>Rehydrate an id carried as a string (e.g. a Flowable process variable).</summary>
public static RegistrationId Parse(string value) => new(Guid.Parse(value));
public override string ToString() => Value.ToString();
}

View File

@@ -0,0 +1,10 @@
namespace Big.Domain;
/// <summary>The lifecycle states a <see cref="Registration"/> moves through. The walking
/// skeleton knows only <see cref="Ingediend"/>; withdrawal, beoordeling and herregistratie
/// states arrive in their own slices (Iteration 2+).</summary>
public enum RegistrationStatus
{
/// <summary>Submitted by the zorgprofessional; the registratie process has been started.</summary>
Ingediend,
}

View File

@@ -0,0 +1,28 @@
using System.Net.Http.Json;
using System.Text.Json.Serialization;
using Big.Application;
namespace Big.Infrastructure;
/// <summary>
/// HTTP client to the ACL service — the boundary the domain crosses to open a zaak (§8.1). It POSTs
/// the bsn to the ACL's <c>/zaken</c> endpoint and returns the created zaak URL; it never constructs
/// ZGW URLs or talks to OpenZaak itself.
/// </summary>
public sealed class AclHttpClient(HttpClient http, AclOptions options) : IAclClient
{
public async Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default)
{
using var response = await http.PostAsJsonAsync(
new Uri(options.BaseUrl, "zaken"), new OpenZaakRequest(bsn), ct);
response.EnsureSuccessStatusCode();
var opened = await response.Content.ReadFromJsonAsync<OpenZaakResponse>(ct)
?? throw new InvalidOperationException("The ACL returned an empty zaak response.");
return new Uri(opened.ZaakUrl);
}
private sealed record OpenZaakRequest([property: JsonPropertyName("bsn")] string Bsn);
private sealed record OpenZaakResponse([property: JsonPropertyName("zaakUrl")] string ZaakUrl);
}

View File

@@ -0,0 +1,24 @@
<Project Sdk="Microsoft.NET.Sdk">
<!-- The infrastructure layer (CLAUDE.md §9): adapters implementing the Application ports.
The Workflow Client (Flowable REST) and the ACL HTTP client live here — the only code
that talks to Flowable (§8.2) and to the ACL respectively. -->
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Big.Application\Big.Application.csproj" />
</ItemGroup>
<ItemGroup>
<!-- The external-task job worker is a hosted BackgroundService (ADR-0009); it logs and
resolves a per-tick scope. Abstractions only — the host (Api) brings the implementations. -->
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.0" />
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,110 @@
using System.Net.Http.Headers;
using System.Net.Http.Json;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using Big.Application;
using Big.Domain;
namespace Big.Infrastructure;
/// <summary>
/// The Workflow Client — the only code that talks to Flowable (§8.2). It starts registratie process
/// instances and drives the <c>OpenZaakAanmaken</c> external-worker jobs over Flowable's REST API
/// (start: <c>service/runtime/process-instances</c>; acquire/complete: <c>external-job-api/…</c>).
/// The REST contract here is the one verified against a live flowable-rest engine (ADR-0009).
/// </summary>
public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions options)
: IWorkflowClient, IExternalWorkerClient
{
private const string Topic = "OpenZaakAanmaken";
private const string ProcessDefinitionKey = "registratie";
private const string RegistrationIdVariable = "registrationId";
private const string ZaakUrlVariable = "zaakUrl";
public async Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default)
{
var request = new StartProcessRequest(
ProcessDefinitionKey,
[new Variable(RegistrationIdVariable, "string", registrationId.ToString())]);
var created = await PostAsync<StartProcessRequest, ProcessInstance>(
"service/runtime/process-instances", request, ct)
?? throw new InvalidOperationException("Flowable returned an empty process-instance response.");
return created.Id;
}
public async Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
{
var request = new AcquireJobsRequest(Topic, options.LockDuration, maxJobs, options.WorkerId);
var jobs = await PostAsync<AcquireJobsRequest, List<AcquiredJob>>(
"external-job-api/acquire/jobs", request, ct) ?? [];
return [.. jobs.Select(job => new OpenZaakJob(job.Id, RegistrationId.Parse(job.RegistrationId())))];
}
public async Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
{
var request = new CompleteJobRequest(
options.WorkerId,
[new Variable(ZaakUrlVariable, "string", zaakUrl.ToString())]);
using var response = await SendAsync(
$"external-job-api/acquire/jobs/{jobId}/complete", request, ct);
response.EnsureSuccessStatusCode();
}
private async Task<TResponse?> PostAsync<TRequest, TResponse>(string path, TRequest body, CancellationToken ct)
{
using var response = await SendAsync(path, body, ct);
response.EnsureSuccessStatusCode();
return await response.Content.ReadFromJsonAsync<TResponse>(ct);
}
private Task<HttpResponseMessage> SendAsync<TRequest>(string path, TRequest body, CancellationToken ct)
{
var message = new HttpRequestMessage(HttpMethod.Post, new Uri(options.BaseUrl, path))
{
Content = JsonContent.Create(body),
};
message.Headers.Authorization = new AuthenticationHeaderValue("Basic", BasicCredentials());
return http.SendAsync(message, ct);
}
private string BasicCredentials()
=> Convert.ToBase64String(Encoding.ASCII.GetBytes($"{options.Username}:{options.Password}"));
private sealed record StartProcessRequest(
[property: JsonPropertyName("processDefinitionKey")] string ProcessDefinitionKey,
[property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables);
private sealed record AcquireJobsRequest(
[property: JsonPropertyName("topic")] string Topic,
[property: JsonPropertyName("lockDuration")] string LockDuration,
[property: JsonPropertyName("numberOfTasks")] int NumberOfTasks,
[property: JsonPropertyName("workerId")] string WorkerId);
private sealed record CompleteJobRequest(
[property: JsonPropertyName("workerId")] string WorkerId,
[property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables);
private sealed record Variable(
[property: JsonPropertyName("name")] string Name,
[property: JsonPropertyName("type")] string Type,
[property: JsonPropertyName("value")] string Value);
private sealed record ProcessInstance([property: JsonPropertyName("id")] string Id);
private sealed record AcquiredJob(
[property: JsonPropertyName("id")] string Id,
[property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables)
{
/// <summary>The registration id this job carries as a process variable.</summary>
public string RegistrationId() =>
// Stryker disable once Linq : equivalent — Single and SingleOrDefault both throw on a job
// with no registrationId variable (the only untested branch), so the mutant is indistinguishable.
Variables.SingleOrDefault(v => v.Name == "registrationId")?.Value
?? throw new InvalidOperationException($"OpenZaakAanmaken job {Id} carries no registrationId variable.");
}
}

View File

@@ -0,0 +1,18 @@
using Big.Application;
namespace Big.Infrastructure;
/// <summary>
/// The worker-facing side of the Workflow Client: acquiring and completing Flowable external-worker
/// jobs (ADR-0009). Kept separate from the Application's <see cref="IWorkflowClient"/> because job
/// acquisition is a Flowable-specific polling mechanic the application never needs to know about.
/// Implemented by <see cref="FlowableWorkflowClient"/> — the only code that talks to Flowable (§8.2).
/// </summary>
public interface IExternalWorkerClient
{
/// <summary>Acquire and lock up to <paramref name="maxJobs"/> <c>OpenZaakAanmaken</c> jobs.</summary>
Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default);
/// <summary>Complete an acquired job, passing the opened zaak URL back into the process.</summary>
Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default);
}

View File

@@ -0,0 +1,25 @@
using System.Collections.Concurrent;
using Big.Application;
using Big.Domain;
namespace Big.Infrastructure;
/// <summary>
/// In-memory <see cref="IRegistrationStore"/> for the minimal slice (ADR-0009). The walking
/// skeleton's read path is the projection (S-06), not this store, so durable domain persistence is a
/// documented follow-up. Registered as a singleton so the submit endpoint and the worker share it.
/// </summary>
public sealed class InMemoryRegistrationStore : IRegistrationStore
{
private readonly ConcurrentDictionary<RegistrationId, Registration> _byId = new();
public Task SaveAsync(Registration registration, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(registration);
_byId[registration.Id] = registration;
return Task.CompletedTask;
}
public Task<Registration?> GetAsync(RegistrationId id, CancellationToken ct = default)
=> Task.FromResult(_byId.GetValueOrDefault(id));
}

View File

@@ -0,0 +1,39 @@
using Big.Application;
using Microsoft.Extensions.Logging;
namespace Big.Infrastructure;
/// <summary>
/// One poll tick of the external-task worker (ADR-0009): acquire the parked <c>OpenZaakAanmaken</c>
/// jobs, hand each to the <see cref="OpenZaakWorker"/> to open a zaak via the ACL, and complete the
/// Flowable job with the resulting zaak URL. A job that fails is logged and left un-completed so
/// Flowable redelivers it (§8.6). Split out from the hosted <see cref="OpenZaakJobPump"/> so the
/// acquire→process→complete logic is unit-testable without a running host.
/// </summary>
public sealed class OpenZaakJobProcessor(
IExternalWorkerClient client,
OpenZaakWorker worker,
ILogger<OpenZaakJobProcessor> logger)
{
/// <summary>Acquire and process up to <paramref name="maxJobs"/> jobs. Returns the number acquired.</summary>
public async Task<int> PumpOnceAsync(int maxJobs, CancellationToken ct = default)
{
var jobs = await client.AcquireOpenZaakJobsAsync(maxJobs, ct);
foreach (var job in jobs)
{
try
{
var zaakUrl = await worker.HandleAsync(job, ct);
await client.CompleteOpenZaakJobAsync(job.JobId, zaakUrl, ct);
}
catch (Exception ex)
{
// Leave the job un-completed: its lock expires and Flowable redelivers it (§8.6).
logger.LogError(ex, "OpenZaakAanmaken job {JobId} failed; leaving it for redelivery.", job.JobId);
}
}
return jobs.Count;
}
}

View File

@@ -0,0 +1,48 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace Big.Infrastructure;
/// <summary>
/// The hosted polling loop of the external-task job worker (ADR-0009): on an interval it resolves a
/// scoped <see cref="OpenZaakJobProcessor"/> and asks it to drain the parked <c>OpenZaakAanmaken</c>
/// jobs. A deliberately thin shell — all acquire/process/complete logic lives in the processor, which
/// is unit-tested; this class only owns the timer, the per-tick scope, and loop resilience.
/// </summary>
public sealed class OpenZaakJobPump(
IServiceScopeFactory scopeFactory,
FlowableOptions options,
ILogger<OpenZaakJobPump> logger) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
using var scope = scopeFactory.CreateScope();
var processor = scope.ServiceProvider.GetRequiredService<OpenZaakJobProcessor>();
await processor.PumpOnceAsync(options.MaxJobsPerPoll, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
break;
}
catch (Exception ex)
{
// A transient fault (e.g. Flowable briefly unreachable) must not kill the loop.
logger.LogError(ex, "OpenZaakAanmaken job poll failed; retrying after the poll interval.");
}
try
{
await Task.Delay(options.PollInterval, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
break;
}
}
}
}

View File

@@ -0,0 +1,29 @@
namespace Big.Infrastructure;
/// <summary>Configuration for the Flowable Workflow Client. <see cref="BaseUrl"/> is the flowable-rest
/// root and must end with a slash (e.g. <c>http://flowable-rest:8080/flowable-rest/</c>) so the
/// <c>service/…</c> and <c>external-job-api/…</c> sub-paths resolve correctly.</summary>
public sealed class FlowableOptions
{
public Uri BaseUrl { get; set; } = null!;
public string Username { get; set; } = "";
public string Password { get; set; } = "";
/// <summary>The id this worker locks jobs under, so two workers don't process the same job.</summary>
public string WorkerId { get; set; } = "big-domain-worker";
/// <summary>How long an acquired job stays locked to this worker (ISO-8601 duration).</summary>
public string LockDuration { get; set; } = "PT5M";
/// <summary>How many OpenZaakAanmaken jobs to acquire per poll.</summary>
public int MaxJobsPerPoll { get; set; } = 5;
/// <summary>How often the worker polls Flowable for new jobs.</summary>
public TimeSpan PollInterval { get; set; } = TimeSpan.FromSeconds(2);
}
/// <summary>Configuration for the ACL HTTP client. <see cref="BaseUrl"/> is the ACL service root.</summary>
public sealed class AclOptions
{
public Uri BaseUrl { get; set; } = null!;
}

View File

@@ -0,0 +1,44 @@
using System.Net;
using Big.Infrastructure;
namespace Big.Tests;
public class AclHttpClientTests
{
private static AclHttpClient Client(StubHandler handler) => new(
new HttpClient(handler), new AclOptions { BaseUrl = new("http://acl/") });
[Fact]
public async Task Opens_a_zaak_by_posting_the_bsn_and_returns_the_zaak_url()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK,
"""{"zaakUrl":"http://openzaak/zaken/api/v1/zaken/abc"}"""));
var url = await client.OpenZaakAsync("123456782");
Assert.Equal("http://openzaak/zaken/api/v1/zaken/abc", url.ToString());
Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
Assert.Equal("http://acl/zaken", capture.Seen.RequestUri!.ToString());
Assert.Contains("\"bsn\":\"123456782\"", capture.Body);
}
[Fact]
public async Task Throws_when_the_acl_rejects_the_request()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.BadGateway));
await Assert.ThrowsAsync<HttpRequestException>(() => client.OpenZaakAsync("123456782"));
}
[Fact]
public async Task Throws_when_the_acl_returns_an_empty_body()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK, "null"));
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => client.OpenZaakAsync("123456782"));
Assert.Contains("empty", ex.Message, StringComparison.OrdinalIgnoreCase);
}
}

View File

@@ -0,0 +1,27 @@
<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="..\Big.Domain\Big.Domain.csproj" />
<ProjectReference Include="..\Big.Application\Big.Application.csproj" />
<ProjectReference Include="..\Big.Infrastructure\Big.Infrastructure.csproj" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,24 @@
using Microsoft.Extensions.Logging;
namespace Big.Tests;
/// <summary>A minimal <see cref="ILogger{T}"/> that records the level and rendered message of each
/// entry, so tests can assert that (for example) a failing job was logged.</summary>
internal sealed class CapturingLogger<T> : ILogger<T>
{
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
=> Entries.Add((logLevel, formatter(state, exception)));
private sealed class NullScope : IDisposable
{
public static readonly NullScope Instance = new();
public void Dispose() { }
}
}

View File

@@ -0,0 +1,61 @@
using Big.Application;
using Big.Domain;
namespace Big.Tests;
/// <summary>An in-memory <see cref="IRegistrationStore"/> for the application-layer tests. Upserts
/// keyed on the registration id, mirroring the production store (kept distinct from the production
/// <c>InMemoryRegistrationStore</c> so tests can seed and inspect save counts).</summary>
internal sealed class FakeRegistrationStore : IRegistrationStore
{
private readonly Dictionary<RegistrationId, Registration> _byId = [];
public int SaveCount { get; private set; }
public Task SaveAsync(Registration registration, CancellationToken ct = default)
{
SaveCount++;
_byId[registration.Id] = registration;
return Task.CompletedTask;
}
public Task<Registration?> GetAsync(RegistrationId id, CancellationToken ct = default)
=> Task.FromResult(_byId.GetValueOrDefault(id));
public void Seed(Registration registration) => _byId[registration.Id] = registration;
}
/// <summary>A fake Workflow Client that records the registration it was asked to start a process for
/// and returns a fixed process-instance id. An optional callback runs at start time, letting a test
/// assert ordering (e.g. that the registration was persisted before the process started).</summary>
internal sealed class FakeWorkflowClient(string processInstanceId = "proc-1", Action<RegistrationId>? onStart = null)
: IWorkflowClient
{
public RegistrationId? StartedFor { get; private set; }
public Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default)
{
onStart?.Invoke(registrationId);
StartedFor = registrationId;
return Task.FromResult(processInstanceId);
}
}
/// <summary>A fake ACL client that records the bsn it was asked to open a zaak for and returns a
/// fixed zaak URL.</summary>
internal sealed class FakeAclClient(Uri? zaakUrl = null) : IAclClient
{
public static readonly Uri DefaultZaakUrl = new("http://openzaak/zaken/api/v1/zaken/abc");
private readonly Uri _zaakUrl = zaakUrl ?? DefaultZaakUrl;
public string? OpenedForBsn { get; private set; }
public int CallCount { get; private set; }
public Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default)
{
CallCount++;
OpenedForBsn = bsn;
return Task.FromResult(_zaakUrl);
}
}

View File

@@ -0,0 +1,139 @@
using System.Net;
using System.Text;
using Big.Domain;
using Big.Infrastructure;
namespace Big.Tests;
public class FlowableWorkflowClientTests
{
private static readonly Uri Base = new("http://flowable/flowable-rest/");
private static FlowableWorkflowClient Client(StubHandler handler) => new(
new HttpClient(handler),
new FlowableOptions { BaseUrl = Base, Username = "rest-admin", Password = "test", WorkerId = "worker-x" });
private static string DecodeBasic(HttpRequestMessage request)
=> Encoding.ASCII.GetString(Convert.FromBase64String(request.Headers.Authorization!.Parameter!));
[Fact]
public async Task Start_posts_the_registration_id_variable_and_returns_the_instance_id()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.Created, """{"id":"pi-1"}"""));
var rid = RegistrationId.New();
var pid = await client.StartRegistrationProcessAsync(rid);
Assert.Equal("pi-1", pid);
Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
Assert.Equal("http://flowable/flowable-rest/service/runtime/process-instances",
capture.Seen.RequestUri!.ToString());
Assert.Equal("Basic", capture.Seen.Headers.Authorization!.Scheme);
Assert.Equal("rest-admin:test", DecodeBasic(capture.Seen));
Assert.Contains("\"processDefinitionKey\":\"registratie\"", capture.Body);
Assert.Contains("\"name\":\"registrationId\"", capture.Body);
Assert.Contains("\"type\":\"string\"", capture.Body);
Assert.Contains($"\"value\":\"{rid}\"", capture.Body);
}
[Fact]
public async Task Start_uses_the_configured_worker_credentials_and_defaults()
{
var capture = new RequestCapture();
// Only BaseUrl set — the worker id and (empty) credentials must come from the defaults.
var client = new FlowableWorkflowClient(
new HttpClient(capture.Responds(HttpStatusCode.OK, "[]")),
new FlowableOptions { BaseUrl = Base });
await client.AcquireOpenZaakJobsAsync(1);
Assert.Equal(":", DecodeBasic(capture.Seen!));
Assert.Contains("\"workerId\":\"big-domain-worker\"", capture.Body);
Assert.Contains("\"lockDuration\":\"PT5M\"", capture.Body);
}
[Fact]
public async Task Acquire_posts_the_topic_and_parses_jobs_with_their_registration_id()
{
var rid = RegistrationId.New();
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK,
$$"""[{"id":"job-1","variables":[{"name":"registrationId","type":"string","value":"{{rid}}"}]}]"""));
var jobs = await client.AcquireOpenZaakJobsAsync(3);
var job = Assert.Single(jobs);
Assert.Equal("job-1", job.JobId);
Assert.Equal(rid, job.RegistrationId);
Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs",
capture.Seen!.RequestUri!.ToString());
Assert.Contains("\"topic\":\"OpenZaakAanmaken\"", capture.Body);
Assert.Contains("\"numberOfTasks\":3", capture.Body);
Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
Assert.Contains("\"lockDuration\":\"PT5M\"", capture.Body);
}
[Theory]
[InlineData("[]")]
[InlineData("null")]
public async Task Acquire_returns_empty_when_flowable_has_no_parked_jobs(string body)
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK, body));
var jobs = await client.AcquireOpenZaakJobsAsync(1);
Assert.NotNull(capture.Seen);
Assert.Empty(jobs);
}
[Fact]
public async Task Acquire_throws_when_a_job_carries_no_registration_id()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK, """[{"id":"job-1","variables":[]}]"""));
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => client.AcquireOpenZaakJobsAsync(1));
Assert.Contains("job-1", ex.Message);
Assert.Contains("registrationId", ex.Message);
}
[Fact]
public async Task Complete_posts_the_zaak_url_variable_to_the_job_complete_endpoint()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.NoContent));
await client.CompleteOpenZaakJobAsync("job-1", new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
Assert.NotNull(capture.Seen);
Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs/job-1/complete",
capture.Seen.RequestUri!.ToString());
Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
Assert.Contains("\"name\":\"zaakUrl\"", capture.Body);
Assert.Contains("\"type\":\"string\"", capture.Body);
Assert.Contains("\"value\":\"http://openzaak/zaken/api/v1/zaken/abc\"", capture.Body);
}
[Fact]
public async Task Start_throws_when_flowable_rejects_the_request()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.InternalServerError));
await Assert.ThrowsAsync<HttpRequestException>(
() => client.StartRegistrationProcessAsync(RegistrationId.New()));
}
[Fact]
public async Task Complete_throws_when_flowable_rejects_the_request()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.InternalServerError));
await Assert.ThrowsAsync<HttpRequestException>(
() => client.CompleteOpenZaakJobAsync("job-1", new Uri("http://openzaak/zaken/api/v1/zaken/abc")));
}
}

View File

@@ -0,0 +1,30 @@
using System.Net;
namespace Big.Tests;
/// <summary>A test double for <see cref="HttpMessageHandler"/> that records the last request and
/// returns a scripted response — the same approach the ACL gateway tests use.</summary>
internal sealed class StubHandler(Func<HttpRequestMessage, Task<HttpResponseMessage>> onSend) : HttpMessageHandler
{
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken ct)
=> onSend(request);
}
/// <summary>Captures the request a client sent, including the (buffered) body.</summary>
internal sealed class RequestCapture
{
public HttpRequestMessage? Seen { get; private set; }
public string? Body { get; private set; }
/// <summary>A handler that records the request, then replies with <paramref name="status"/> and
/// <paramref name="json"/>.</summary>
public StubHandler Responds(HttpStatusCode status, string? json = null) => new(async req =>
{
Seen = req;
Body = req.Content is null ? null : await req.Content.ReadAsStringAsync();
var response = new HttpResponseMessage(status);
if (json is not null)
response.Content = new StringContent(json, System.Text.Encoding.UTF8, "application/json");
return response;
});
}

View File

@@ -0,0 +1,44 @@
using Big.Domain;
using Big.Infrastructure;
namespace Big.Tests;
public class InMemoryRegistrationStoreTests
{
[Fact]
public async Task Saves_and_reads_back_a_registration_by_id()
{
var store = new InMemoryRegistrationStore();
var registration = Registration.Submit("123456782");
await store.SaveAsync(registration);
var loaded = await store.GetAsync(registration.Id);
Assert.NotNull(loaded);
Assert.Equal(registration.Id, loaded.Id);
Assert.Null(await store.GetAsync(RegistrationId.New()));
}
[Fact]
public async Task Saving_the_same_id_upserts()
{
var store = new InMemoryRegistrationStore();
var registration = Registration.Submit("123456782");
await store.SaveAsync(registration);
registration.AttachZaak(new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
await store.SaveAsync(registration);
var loaded = await store.GetAsync(registration.Id);
Assert.NotNull(loaded);
Assert.Equal("http://openzaak/zaken/api/v1/zaken/abc", loaded.ZaakUrl!.ToString());
}
[Fact]
public async Task Saving_a_null_registration_is_rejected()
{
var store = new InMemoryRegistrationStore();
await Assert.ThrowsAsync<ArgumentNullException>(() => store.SaveAsync(null!));
}
}

View File

@@ -0,0 +1,83 @@
using Big.Application;
using Big.Domain;
using Big.Infrastructure;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Big.Tests;
public class OpenZaakJobProcessorTests
{
/// <summary>A fake external-worker client: scripts the jobs to acquire and records completions.</summary>
private sealed class FakeExternalWorkerClient(params OpenZaakJob[] jobs) : IExternalWorkerClient
{
public int AcquireCount { get; private set; }
public List<(string JobId, Uri ZaakUrl)> Completed { get; } = [];
public Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
{
AcquireCount++;
return Task.FromResult<IReadOnlyList<OpenZaakJob>>(jobs.Take(maxJobs).ToList());
}
public Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
{
Completed.Add((jobId, zaakUrl));
return Task.CompletedTask;
}
}
private static OpenZaakJobProcessor Processor(IExternalWorkerClient client, IRegistrationStore store, FakeAclClient acl)
=> new(client, new OpenZaakWorker(store, acl), NullLogger<OpenZaakJobProcessor>.Instance);
[Fact]
public async Task Acquires_a_job_opens_the_zaak_and_completes_it()
{
var registration = Registration.Submit("123456782");
var store = new FakeRegistrationStore();
store.Seed(registration);
var client = new FakeExternalWorkerClient(new OpenZaakJob("job-1", registration.Id));
var acl = new FakeAclClient();
var acquired = await Processor(client, store, acl).PumpOnceAsync(5);
Assert.Equal(1, acquired);
var completed = Assert.Single(client.Completed);
Assert.Equal("job-1", completed.JobId);
Assert.Equal(FakeAclClient.DefaultZaakUrl, completed.ZaakUrl);
Assert.Equal(FakeAclClient.DefaultZaakUrl, (await store.GetAsync(registration.Id))!.ZaakUrl);
}
[Fact]
public async Task A_failing_job_is_left_uncompleted_for_flowable_to_redeliver()
{
var store = new FakeRegistrationStore();
var acl = new FakeAclClient();
// The job correlates to a registration that is not in the store, so the worker throws.
var client = new FakeExternalWorkerClient(new OpenZaakJob("job-1", RegistrationId.New()));
var logger = new CapturingLogger<OpenZaakJobProcessor>();
var processor = new OpenZaakJobProcessor(client, new OpenZaakWorker(store, acl), logger);
var acquired = await processor.PumpOnceAsync(5);
Assert.Equal(1, acquired);
Assert.Empty(client.Completed);
// The failure is logged (and the job left for redelivery), not swallowed silently.
var error = Assert.Single(logger.Entries, e => e.Level == LogLevel.Error);
Assert.Contains("job-1", error.Message);
}
[Fact]
public async Task Does_nothing_but_poll_when_there_are_no_jobs()
{
var store = new FakeRegistrationStore();
var acl = new FakeAclClient();
var client = new FakeExternalWorkerClient();
var acquired = await Processor(client, store, acl).PumpOnceAsync(5);
Assert.Equal(0, acquired);
Assert.Equal(1, client.AcquireCount);
Assert.Empty(client.Completed);
}
}

View File

@@ -0,0 +1,70 @@
using Big.Application;
using Big.Domain;
namespace Big.Tests;
public class OpenZaakWorkerTests
{
private static Registration Submitted(string bsn = "123456782")
{
var registration = Registration.Submit(bsn);
registration.RecordProcessStarted("proc-1");
return registration;
}
[Fact]
public async Task Handling_a_job_opens_a_zaak_via_the_acl_and_attaches_it()
{
var registration = Submitted();
var store = new FakeRegistrationStore();
store.Seed(registration);
var acl = new FakeAclClient();
var worker = new OpenZaakWorker(store, acl);
var zaakUrl = await worker.HandleAsync(new OpenZaakJob("job-1", registration.Id));
Assert.Equal(FakeAclClient.DefaultZaakUrl, zaakUrl);
Assert.Equal("123456782", acl.OpenedForBsn);
var saved = await store.GetAsync(registration.Id);
Assert.Equal(FakeAclClient.DefaultZaakUrl, saved!.ZaakUrl);
Assert.Equal(1, store.SaveCount);
}
[Fact]
public async Task Handling_a_null_job_is_rejected()
{
var worker = new OpenZaakWorker(new FakeRegistrationStore(), new FakeAclClient());
await Assert.ThrowsAsync<ArgumentNullException>(() => worker.HandleAsync(null!));
}
[Fact]
public async Task Handling_a_job_for_an_unknown_registration_throws_and_opens_no_zaak()
{
var store = new FakeRegistrationStore();
var acl = new FakeAclClient();
var worker = new OpenZaakWorker(store, acl);
var job = new OpenZaakJob("job-1", RegistrationId.New());
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => worker.HandleAsync(job));
Assert.Contains(job.RegistrationId.ToString(), ex.Message);
Assert.Equal(0, acl.CallCount);
}
[Fact]
public async Task Handling_a_redelivered_job_is_idempotent_and_does_not_open_a_second_zaak()
{
var registration = Submitted();
var store = new FakeRegistrationStore();
store.Seed(registration);
var acl = new FakeAclClient();
var worker = new OpenZaakWorker(store, acl);
var job = new OpenZaakJob("job-1", registration.Id);
var first = await worker.HandleAsync(job);
var second = await worker.HandleAsync(job);
Assert.Equal(first, second);
Assert.Equal(1, acl.CallCount);
}
}

View File

@@ -0,0 +1,90 @@
using Big.Domain;
namespace Big.Tests;
public class RegistrationTests
{
[Fact]
public void Submitting_a_registration_starts_in_ingediend()
{
var registration = Registration.Submit("123456782");
Assert.Equal(RegistrationStatus.Ingediend, registration.Status);
Assert.Equal("123456782", registration.Bsn);
Assert.NotEqual(Guid.Empty, registration.Id.Value);
Assert.Null(registration.ZaakUrl);
Assert.Null(registration.ProcessInstanceId);
}
[Theory]
[InlineData("")]
[InlineData(" ")]
[InlineData(null)]
public void Submitting_without_a_bsn_is_rejected(string? bsn)
=> Assert.ThrowsAny<ArgumentException>(() => Registration.Submit(bsn!));
[Fact]
public void Recording_the_started_process_keeps_its_instance_id()
{
var registration = Registration.Submit("123456782");
registration.RecordProcessStarted("proc-42");
Assert.Equal("proc-42", registration.ProcessInstanceId);
}
[Theory]
[InlineData("")]
[InlineData(" ")]
[InlineData(null)]
public void Recording_an_empty_process_instance_id_is_rejected(string? processInstanceId)
{
var registration = Registration.Submit("123456782");
Assert.ThrowsAny<ArgumentException>(() => registration.RecordProcessStarted(processInstanceId!));
}
[Fact]
public void Attaching_a_zaak_records_its_url_and_keeps_the_status_ingediend()
{
var registration = Registration.Submit("123456782");
var zaak = new Uri("http://openzaak/zaken/api/v1/zaken/abc");
registration.AttachZaak(zaak);
Assert.Equal(zaak, registration.ZaakUrl);
Assert.Equal(RegistrationStatus.Ingediend, registration.Status);
}
[Fact]
public void Attaching_a_null_zaak_is_rejected()
{
var registration = Registration.Submit("123456782");
Assert.Throws<ArgumentNullException>(() => registration.AttachZaak(null!));
}
[Fact]
public void Re_attaching_the_same_zaak_is_idempotent()
{
var registration = Registration.Submit("123456782");
var zaak = new Uri("http://openzaak/zaken/api/v1/zaken/abc");
registration.AttachZaak(zaak);
registration.AttachZaak(zaak);
Assert.Equal(zaak, registration.ZaakUrl);
}
[Fact]
public void Attaching_a_conflicting_zaak_is_rejected()
{
var registration = Registration.Submit("123456782");
registration.AttachZaak(new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
var ex = Assert.Throws<InvalidOperationException>(
() => registration.AttachZaak(new Uri("http://openzaak/zaken/api/v1/zaken/other")));
Assert.Contains("/zaken/abc", ex.Message);
Assert.Contains("/zaken/other", ex.Message);
}
}

View File

@@ -0,0 +1,66 @@
using Big.Application;
using Big.Domain;
namespace Big.Tests;
public class SubmitRegistrationTests
{
[Fact]
public async Task Submitting_persists_an_ingediend_registration_and_starts_the_process()
{
var store = new FakeRegistrationStore();
var workflow = new FakeWorkflowClient(processInstanceId: "proc-42");
var handler = new SubmitRegistration(store, workflow);
var id = await handler.HandleAsync(new SubmitRegistrationCommand("123456782"));
var saved = await store.GetAsync(id);
Assert.NotNull(saved);
Assert.Equal("123456782", saved.Bsn);
Assert.Equal(RegistrationStatus.Ingediend, saved.Status);
Assert.Equal("proc-42", saved.ProcessInstanceId);
Assert.Equal(id, workflow.StartedFor);
Assert.Null(saved.ZaakUrl);
// Persisted twice: once before starting the process, once after recording the instance id.
Assert.Equal(2, store.SaveCount);
}
[Fact]
public async Task Rejects_a_null_command_without_touching_the_store_or_workflow()
{
var store = new FakeRegistrationStore();
var workflow = new FakeWorkflowClient();
var handler = new SubmitRegistration(store, workflow);
await Assert.ThrowsAsync<ArgumentNullException>(() => handler.HandleAsync(null!));
Assert.Equal(0, store.SaveCount);
Assert.Null(workflow.StartedFor);
}
[Fact]
public async Task Submitting_persists_the_registration_before_starting_the_process()
{
// The worker correlates the OpenZaakAanmaken job back to the aggregate by id, so the
// aggregate must already be persisted when the process starts (ADR-0009).
var store = new FakeRegistrationStore();
Registration? visibleAtStart = null;
var workflow = new FakeWorkflowClient(onStart: id => visibleAtStart = store.GetAsync(id).Result);
var handler = new SubmitRegistration(store, workflow);
await handler.HandleAsync(new SubmitRegistrationCommand("123456782"));
Assert.NotNull(visibleAtStart);
}
[Fact]
public async Task Submitting_without_a_bsn_is_rejected_and_starts_no_process()
{
var store = new FakeRegistrationStore();
var workflow = new FakeWorkflowClient();
var handler = new SubmitRegistration(store, workflow);
await Assert.ThrowsAnyAsync<ArgumentException>(
() => handler.HandleAsync(new SubmitRegistrationCommand("")));
Assert.Null(workflow.StartedFor);
}
}

7
services/domain/Big.slnx Normal file
View File

@@ -0,0 +1,7 @@
<Solution>
<Project Path="Big.Domain/Big.Domain.csproj" />
<Project Path="Big.Application/Big.Application.csproj" />
<Project Path="Big.Infrastructure/Big.Infrastructure.csproj" />
<Project Path="Big.Api/Big.Api.csproj" />
<Project Path="Big.Tests/Big.Tests.csproj" />
</Solution>

View File

@@ -0,0 +1,34 @@
# Multi-stage build for the BIG Domain Service (.NET 10).
# Build context is services/domain (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 Big.Api/Big.Api.csproj Big.Api/
COPY Big.Application/Big.Application.csproj Big.Application/
COPY Big.Infrastructure/Big.Infrastructure.csproj Big.Infrastructure/
COPY Big.Domain/Big.Domain.csproj Big.Domain/
RUN dotnet restore Big.Api/Big.Api.csproj
COPY Big.Api/ Big.Api/
COPY Big.Application/ Big.Application/
COPY Big.Infrastructure/ Big.Infrastructure/
COPY Big.Domain/ Big.Domain/
RUN dotnet publish Big.Api/Big.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=10s --retries=5 \
CMD curl -fsS http://localhost:8080/health || exit 1
ENTRYPOINT ["dotnet", "Big.Api.dll"]

View File

@@ -0,0 +1,15 @@
{
"stryker-config": {
"solution": "Big.slnx",
"test-projects": ["Big.Tests/Big.Tests.csproj"],
"reporters": ["progress", "html"],
"mutate": [
"!**/OpenZaakJobPump.cs"
],
"thresholds": {
"high": 95,
"low": 90,
"break": 90
}
}
}

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,8 @@
<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" />
<ProjectReference Include="..\..\services\domain\Big.Application\Big.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,18 @@
# language: en
# Drives S-05 (#6). A zorgprofessional submits a registration; the Domain Service starts the
# registratie process and, via the external-task worker, opens a zaak through the ACL and records
# it on the aggregate (ADR-0009). This scenario exercises the use cases against in-memory stand-ins
# for the Workflow Client and the ACL; real Flowable + ACL + OpenZaak delivery is verified by the
# live-stack check (verify-domain).
Feature: Een registratie indienen
Als zorgprofessional wil ik mijn registratie indienen
zodat er een registratieproces start en een zaak wordt geopend.
Scenario: Indienen start het registratieproces en opent een zaak
Given a zorgprofessional submits a registration for BSN "123456782"
When the domain service handles the submission
Then a registratie process is started for the registration
And the registration has status "INGEDIEND"
When the OpenZaakAanmaken external task is handled for the registration
Then a zaak is opened via the ACL for BSN "123456782"
And the zaak is recorded on the registration

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,54 @@
using Acceptance.Support;
using Big.Application;
using Big.Domain;
using Reqnroll;
using Xunit;
namespace Acceptance.Steps;
/// <summary>Bindings for <c>RegistratieIndienen.feature</c> (S-05). Drives the SubmitRegistration
/// and OpenZaakWorker use cases against in-memory ports; one instance per scenario.</summary>
[Binding]
public sealed class RegistratieIndienenSteps
{
private readonly InMemoryWorkflowClient _workflow = new();
private readonly InMemoryAclClient _acl = new();
private readonly InMemoryRegistrationStore _store = new();
private string _bsn = "";
private RegistrationId _id;
[Given("a zorgprofessional submits a registration for BSN \"(.*)\"")]
public void GivenAZorgprofessionalSubmitsARegistration(string bsn) => _bsn = bsn;
[When("the domain service handles the submission")]
public async Task WhenTheDomainServiceHandlesTheSubmission()
=> _id = await new SubmitRegistration(_store, _workflow).HandleAsync(new SubmitRegistrationCommand(_bsn));
[Then("a registratie process is started for the registration")]
public void ThenARegistratieProcessIsStarted()
=> Assert.Equal(_id, _workflow.StartedFor);
[Then("the registration has status \"(.*)\"")]
public async Task ThenTheRegistrationHasStatus(string expected)
{
var registration = await _store.GetAsync(_id);
Assert.NotNull(registration);
Assert.Equal(expected, registration.Status.ToString().ToUpperInvariant());
Assert.Equal(InMemoryWorkflowClient.StartedProcessInstanceId, registration.ProcessInstanceId);
}
[When("the OpenZaakAanmaken external task is handled for the registration")]
public async Task WhenTheExternalTaskIsHandled()
=> await new OpenZaakWorker(_store, _acl).HandleAsync(new OpenZaakJob("job-acc-1", _id));
[Then("a zaak is opened via the ACL for BSN \"(.*)\"")]
public void ThenAZaakIsOpenedViaTheAcl(string bsn)
=> Assert.Equal(bsn, _acl.OpenedForBsn);
[Then("the zaak is recorded on the registration")]
public async Task ThenTheZaakIsRecordedOnTheRegistration()
{
var registration = await _store.GetAsync(_id);
Assert.Equal(InMemoryAclClient.OpenedZaakUrl, registration!.ZaakUrl);
}
}

View File

@@ -0,0 +1,50 @@
using Big.Application;
using Big.Domain;
namespace Acceptance.Support;
/// <summary>An in-memory Workflow Client for the domain acceptance scenario: it records which
/// registration a process was started for and returns a fixed instance id. The acceptance test
/// drives the use case without a running Flowable (live verification is the verify-domain check).</summary>
public sealed class InMemoryWorkflowClient : IWorkflowClient
{
public const string StartedProcessInstanceId = "proc-acc-1";
public RegistrationId? StartedFor { get; private set; }
public Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default)
{
StartedFor = registrationId;
return Task.FromResult(StartedProcessInstanceId);
}
}
/// <summary>An in-memory ACL stand-in: records the bsn it opened a zaak for and returns a fixed URL,
/// so the scenario verifies the domain crosses the ACL boundary (§8.1) without a running OpenZaak.</summary>
public sealed class InMemoryAclClient : IAclClient
{
public static readonly Uri OpenedZaakUrl = new("http://openzaak/zaken/api/v1/zaken/acc-zaak");
public string? OpenedForBsn { get; private set; }
public Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default)
{
OpenedForBsn = bsn;
return Task.FromResult(OpenedZaakUrl);
}
}
/// <summary>An in-memory registration store for the domain acceptance scenario.</summary>
public sealed class InMemoryRegistrationStore : IRegistrationStore
{
private readonly Dictionary<RegistrationId, Registration> _byId = [];
public Task SaveAsync(Registration registration, CancellationToken ct = default)
{
_byId[registration.Id] = registration;
return Task.CompletedTask;
}
public Task<Registration?> GetAsync(RegistrationId id, CancellationToken ct = default)
=> Task.FromResult(_byId.GetValueOrDefault(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)];
}