diff --git a/register-referentie.slnx b/register-referentie.slnx index f5b32ab..cd8e0c2 100644 --- a/register-referentie.slnx +++ b/register-referentie.slnx @@ -11,6 +11,10 @@ + + + + diff --git a/services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj b/services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj new file mode 100644 index 0000000..93f5ab4 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj @@ -0,0 +1,9 @@ + + + + net10.0 + enable + enable + + + diff --git a/services/event-subscriber/EventSubscriber.Application/Notification.cs b/services/event-subscriber/EventSubscriber.Application/Notification.cs new file mode 100644 index 0000000..9e99b8f --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Application/Notification.cs @@ -0,0 +1,30 @@ +namespace EventSubscriber.Application; + +/// +/// 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 aanmaakdatum and kenmerken which the +/// minimal projection ignores (bsn is deferred — see ADR-0008). For a zaken/zaak/create +/// notification hoofdObject and resourceUrl are both the created zaak's URL. +/// +public sealed record Notification( + string Kanaal, + string Resource, + string Actie, + Uri ResourceUrl) +{ + /// The only event the walking-skeleton projection reacts to: a zaak being created. + public bool IsZaakCreated => + Kanaal == "zaken" && Resource == "zaak" && Actie == "create"; + + /// The zaak UUID — the trailing path segment of the resource URL — used as the projection key. + public string ZaakId => ResourceUrl.Segments[^1].Trim('/'); + + /// + /// 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.) + /// + public string IdempotencyKey => $"{Kanaal}:{Resource}:{Actie}:{ResourceUrl}"; +} diff --git a/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs b/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs new file mode 100644 index 0000000..71ade5a --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs @@ -0,0 +1,19 @@ +namespace EventSubscriber.Application; + +/// +/// 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. +/// +public sealed class NotificationProjector(INotificationLog log, IProjectionStore store) +{ + /// Handle one inbound notification. Ignores anything that is not a zaak-created event. + public Task HandleAsync(Notification notification, CancellationToken ct = default) + // RED: not yet implemented — the smallest change to make the tests pass comes next. + => Task.CompletedTask; + + /// Rebuild the projection from the durable notification log (PRD §8.4). + public Task RebuildAsync(CancellationToken ct = default) + // RED: not yet implemented. + => Task.CompletedTask; +} diff --git a/services/event-subscriber/EventSubscriber.Application/Ports.cs b/services/event-subscriber/EventSubscriber.Application/Ports.cs new file mode 100644 index 0000000..911ee68 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Application/Ports.cs @@ -0,0 +1,37 @@ +namespace EventSubscriber.Application; + +/// +/// 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. +/// +public interface INotificationLog +{ + /// + /// Record a notification. Returns true if it was newly recorded, false if an + /// entry with the same key already existed (a duplicate delivery). Must be atomic so that + /// concurrent duplicate deliveries cannot both observe true. + /// + Task TryRecordAsync(RecordedNotification notification, CancellationToken ct = default); + + /// Every accepted notification, for rebuilding the projection. + Task> AllAsync(CancellationToken ct = default); +} + +/// A notification that has been accepted, retaining what a rebuild needs to recompute its projection row. +public sealed record RecordedNotification(string Key, string Actie, string ZaakId); + +/// The read projection store. Owned by the projection bounded context (ADR-0008); the +/// subscriber writes to it and the projection-api reads it. +public interface IProjectionStore +{ + /// Insert or update the row for . Idempotent on the id. + Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default); + + /// Remove every row — the first step of a rebuild. + Task ClearAsync(CancellationToken ct = default); + + /// Every projection row (used by tests and the rebuild verification). + Task> AllAsync(CancellationToken ct = default); +} diff --git a/services/event-subscriber/EventSubscriber.Application/RegisterEntry.cs b/services/event-subscriber/EventSubscriber.Application/RegisterEntry.cs new file mode 100644 index 0000000..3f342a3 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Application/RegisterEntry.cs @@ -0,0 +1,20 @@ +namespace EventSubscriber.Application; + +/// +/// A row of the rebuildable read projection (PRD §8.4). For the minimal slice it carries +/// the zaak id and a status; Bsn and NaamPlaceholder 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). +/// +public sealed record RegisterEntry( + string Id, + string Status, + string? Bsn = null, + string? NaamPlaceholder = null); + +/// The projection statuses the walking skeleton knows about. +public static class RegistrationStatus +{ + /// A zaak has been created; the registration is submitted. + public const string Ingediend = "INGEDIEND"; +} diff --git a/services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj b/services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj new file mode 100644 index 0000000..6b32f79 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj @@ -0,0 +1,25 @@ + + + + net10.0 + enable + enable + false + + + + + + + + + + + + + + + + + + diff --git a/services/event-subscriber/EventSubscriber.Tests/InMemoryStores.cs b/services/event-subscriber/EventSubscriber.Tests/InMemoryStores.cs new file mode 100644 index 0000000..4972de2 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Tests/InMemoryStores.cs @@ -0,0 +1,37 @@ +using EventSubscriber.Application; + +namespace EventSubscriber.Tests; + +/// 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). +internal sealed class InMemoryNotificationLog : INotificationLog +{ + private readonly Dictionary _byKey = []; + + public Task TryRecordAsync(RecordedNotification notification, CancellationToken ct = default) + => Task.FromResult(_byKey.TryAdd(notification.Key, notification)); + + public Task> AllAsync(CancellationToken ct = default) + => Task.FromResult>([.. _byKey.Values]); +} + +internal sealed class InMemoryProjectionStore : IProjectionStore +{ + private readonly Dictionary _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> AllAsync(CancellationToken ct = default) + => Task.FromResult>([.. _byId.Values]); +} diff --git a/services/event-subscriber/EventSubscriber.Tests/NotificationProjectorTests.cs b/services/event-subscriber/EventSubscriber.Tests/NotificationProjectorTests.cs new file mode 100644 index 0000000..daca566 --- /dev/null +++ b/services/event-subscriber/EventSubscriber.Tests/NotificationProjectorTests.cs @@ -0,0 +1,69 @@ +using EventSubscriber.Application; + +namespace EventSubscriber.Tests; + +/// 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). +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_notification_on_another_kanaal_is_ignored() + { + await Projector().HandleAsync(new Notification("documenten", "enkelvoudiginformatieobject", "create", new Uri(ZaakUrl))); + + Assert.Empty(await _store.AllAsync()); + } + + [Fact] + public async Task a_zaak_update_is_ignored_by_the_minimal_projection() + { + await Projector().HandleAsync(new Notification("zaken", "zaak", "update", new Uri(ZaakUrl))); + + Assert.Empty(await _store.AllAsync()); + } + + [Fact] + public async Task rebuild_repopulates_the_projection_from_the_notification_log() + { + var projector = Projector(); + await projector.HandleAsync(ZaakCreated()); + await _store.ClearAsync(); + Assert.Empty(await _store.AllAsync()); + + await projector.RebuildAsync(); + + var entry = Assert.Single(await _store.AllAsync()); + Assert.Equal(RegistrationStatus.Ingediend, entry.Status); + } +} diff --git a/tests/acceptance/Acceptance.csproj b/tests/acceptance/Acceptance.csproj index c182441..7b795eb 100644 --- a/tests/acceptance/Acceptance.csproj +++ b/tests/acceptance/Acceptance.csproj @@ -18,6 +18,7 @@ + diff --git a/tests/acceptance/Features/RegisterProjectieBijwerken.feature b/tests/acceptance/Features/RegisterProjectieBijwerken.feature new file mode 100644 index 0000000..f9d1220 --- /dev/null +++ b/tests/acceptance/Features/RegisterProjectieBijwerken.feature @@ -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" diff --git a/tests/acceptance/Steps/RegisterProjectieBijwerkenSteps.cs b/tests/acceptance/Steps/RegisterProjectieBijwerkenSteps.cs new file mode 100644 index 0000000..d65729f --- /dev/null +++ b/tests/acceptance/Steps/RegisterProjectieBijwerkenSteps.cs @@ -0,0 +1,48 @@ +using Acceptance.Support; +using EventSubscriber.Application; +using Reqnroll; +using Xunit; + +namespace Acceptance.Steps; + +/// Bindings for RegisterProjectieBijwerken.feature (S-06). Reqnroll creates +/// one instance per scenario, so instance fields hold scenario-scoped state. +[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)); + } +} diff --git a/tests/acceptance/Support/InMemoryProjectionStores.cs b/tests/acceptance/Support/InMemoryProjectionStores.cs new file mode 100644 index 0000000..67a7fd9 --- /dev/null +++ b/tests/acceptance/Support/InMemoryProjectionStores.cs @@ -0,0 +1,40 @@ +using EventSubscriber.Application; + +namespace Acceptance.Support; + +/// 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 InMemoryZaakGateway approach used for the ACL scenario. +public sealed class InMemoryNotificationLog : INotificationLog +{ + private readonly Dictionary _byKey = []; + + public Task TryRecordAsync(RecordedNotification notification, CancellationToken ct = default) + => Task.FromResult(_byKey.TryAdd(notification.Key, notification)); + + public Task> AllAsync(CancellationToken ct = default) + => Task.FromResult>([.. _byKey.Values]); +} + +public sealed class InMemoryProjectionStore : IProjectionStore +{ + private readonly Dictionary _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> AllAsync(CancellationToken ct = default) + => Task.FromResult>([.. _byId.Values]); + + public IReadOnlyList RowsFor(string id) + => [.. _byId.Values.Where(e => e.Id == id)]; +}