feat(event-subscriber): NRC event subscriber + rebuildable read projection (closes #7) #59
@@ -11,6 +11,10 @@
|
||||
<Project Path="services/bff/Bff.Api/Bff.Api.csproj" />
|
||||
<Project Path="services/bff/Bff.Tests/Bff.Tests.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/services/event-subscriber/">
|
||||
<Project Path="services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj" />
|
||||
<Project Path="services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/tests/">
|
||||
<Project Path="tests/acceptance/Acceptance.csproj" />
|
||||
</Folder>
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<Nullable>enable</Nullable>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
</PropertyGroup>
|
||||
|
||||
</Project>
|
||||
@@ -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}";
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
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 Task HandleAsync(Notification notification, CancellationToken ct = default)
|
||||
// RED: not yet implemented — the smallest change to make the tests pass comes next.
|
||||
=> Task.CompletedTask;
|
||||
|
||||
/// <summary>Rebuild the projection from the durable notification log (PRD §8.4).</summary>
|
||||
public Task RebuildAsync(CancellationToken ct = default)
|
||||
// RED: not yet implemented.
|
||||
=> Task.CompletedTask;
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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";
|
||||
}
|
||||
@@ -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>
|
||||
@@ -0,0 +1,37 @@
|
||||
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 = [];
|
||||
|
||||
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]);
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
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_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);
|
||||
}
|
||||
}
|
||||
@@ -18,6 +18,7 @@
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\services\acl\Acl.Application\Acl.Application.csproj" />
|
||||
<ProjectReference Include="..\..\services\acl\Acl.Infrastructure\Acl.Infrastructure.csproj" />
|
||||
<ProjectReference Include="..\..\services\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
19
tests/acceptance/Features/RegisterProjectieBijwerken.feature
Normal file
19
tests/acceptance/Features/RegisterProjectieBijwerken.feature
Normal 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"
|
||||
48
tests/acceptance/Steps/RegisterProjectieBijwerkenSteps.cs
Normal file
48
tests/acceptance/Steps/RegisterProjectieBijwerkenSteps.cs
Normal 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));
|
||||
}
|
||||
}
|
||||
40
tests/acceptance/Support/InMemoryProjectionStores.cs
Normal file
40
tests/acceptance/Support/InMemoryProjectionStores.cs
Normal 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)];
|
||||
}
|
||||
Reference in New Issue
Block a user