feat(event-subscriber): enrich the projection with the zaak reference via the ACL (refs #78)

On each notification the subscriber reads the zaak's reference (identificatie)
through the ACL — the only code allowed to talk to ZGW (§8.1) — and persists it on
the register_projection row and in the processed_notifications replay log. Storing
it in the log keeps rebuild log-only (ADR-0008): no ACL/ZGW access on rebuild.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-14 14:46:48 +02:00
parent 7b841d6e62
commit 566fc9bcea
19 changed files with 350 additions and 20 deletions

View File

@@ -0,0 +1,30 @@
using System.Net.Http.Json;
using System.Text.Json.Serialization;
using EventSubscriber.Application;
namespace EventSubscriber.Api;
/// <summary>
/// HTTP client to the ACL service. The subscriber enriches the projection with the zaak's reference
/// (identificatie) by asking the ACL — the only code that may read ZGW (§8.1) — rather than reading
/// OpenZaak itself (adr-proposal #78).
/// </summary>
public sealed class AclHttpClient(HttpClient http) : IAclClient
{
public async Task<string> GetZaakReferenceAsync(Uri zaakUrl, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(zaakUrl);
using var response = await http.PostAsJsonAsync(
new Uri(http.BaseAddress!, "zaken/reference"), new ReferenceRequest(zaakUrl.ToString()), ct);
response.EnsureSuccessStatusCode();
var body = await response.Content.ReadFromJsonAsync<ReferenceResponse>(ct)
?? throw new InvalidOperationException("The ACL returned an empty reference response.");
return body.Reference;
}
private sealed record ReferenceRequest([property: JsonPropertyName("zaakUrl")] string ZaakUrl);
private sealed record ReferenceResponse([property: JsonPropertyName("reference")] string Reference);
}

View File

@@ -11,9 +11,15 @@ var connectionString = builder.Configuration.GetConnectionString("Projection")
// this value (ADR-0007), so the webhook enforces it.
var webhookToken = builder.Configuration["EventSubscriber:Webhook:AuthToken"]
?? throw new InvalidOperationException("Missing configuration 'EventSubscriber:Webhook:AuthToken'");
// The ACL is the only code that may read ZGW (§8.1); the subscriber enriches the projection with the
// zaak reference through it (adr-proposal #78).
var aclBaseUrl = builder.Configuration["Acl:BaseUrl"]
?? throw new InvalidOperationException("Missing configuration 'Acl:BaseUrl'");
builder.Services.AddProjectionReadModel(connectionString);
builder.Services.AddProjectionWriteSide();
builder.Services.AddHttpClient<EventSubscriber.Application.IAclClient, EventSubscriber.Api.AclHttpClient>(
c => c.BaseAddress = new Uri(aclBaseUrl));
var app = builder.Build();

View File

@@ -24,10 +24,12 @@ public sealed record Notification(
public bool IsZaakStatusSet =>
Kanaal == "zaken" && Resource == "status" && Actie == "create";
/// <summary>The zaak UUID used as the projection key. For a status notification the resource URL is
/// the status, so the zaak comes from <c>hoofdObject</c> (which, for a zaak-create, equals the
/// resource URL) — see the NRC payload note.</summary>
public string ZaakId => (HoofdObject ?? ResourceUrl).Segments[^1].Trim('/');
/// <summary>The zaak URL this notification concerns — <c>hoofdObject</c> (the zaak) for a status
/// notification, else the resource URL (which, for a zaak-create, is the zaak).</summary>
public Uri ZaakUrl => HoofdObject ?? ResourceUrl;
/// <summary>The zaak UUID used as the projection key — the trailing segment of <see cref="ZaakUrl"/>.</summary>
public string ZaakId => ZaakUrl.Segments[^1].Trim('/');
/// <summary>
/// A deterministic dedup key. Open Notificaties carries no notification id and may

View File

@@ -5,17 +5,19 @@ namespace EventSubscriber.Application;
/// 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)
public sealed class NotificationProjector(INotificationLog log, IProjectionStore store, IAclClient acl)
{
/// <summary>Handle one inbound notification. Reacts to a zaak being created (INGEDIEND) and a
/// status being set (INGESCHREVEN); ignores everything else.</summary>
/// status being set (INGESCHREVEN); ignores everything else. Enriches the row with the zaak's
/// reference via the ACL (§8.1) and records it so a rebuild needs no ZGW access (#78).</summary>
public async Task HandleAsync(Notification notification, CancellationToken ct = default)
{
if (!notification.IsZaakCreated && !notification.IsZaakStatusSet)
return;
var reference = await acl.GetZaakReferenceAsync(notification.ZaakUrl, ct);
var recorded = new RecordedNotification(
notification.IdempotencyKey, notification.Actie, notification.ZaakId, notification.Resource);
notification.IdempotencyKey, notification.Actie, notification.ZaakId, notification.Resource, reference);
// 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.
@@ -36,7 +38,8 @@ public sealed class NotificationProjector(INotificationLog log, IProjectionStore
/// <summary>The projection row for an accepted notification: a status-set maps to INGESCHREVEN,
/// a zaak-create to INGEDIEND. bsn/naam are deferred (ADR-0008).</summary>
private static RegisterEntry ToEntry(RecordedNotification recorded)
=> new(recorded.ZaakId, recorded.Resource == "status"
? RegistrationStatus.Ingeschreven
: RegistrationStatus.Ingediend);
=> new(
recorded.ZaakId,
recorded.Resource == "status" ? RegistrationStatus.Ingeschreven : RegistrationStatus.Ingediend,
Reference: recorded.Reference);
}

View File

@@ -20,9 +20,20 @@ public interface INotificationLog
}
/// <summary>A notification that has been accepted, retaining what a rebuild needs to recompute its
/// projection row — including the ZGW <c>resource</c>, which distinguishes a zaak-create (INGEDIEND)
/// from a status-set (INGESCHREVEN) so a rebuild reproduces the right status.</summary>
public sealed record RecordedNotification(string Key, string Actie, string ZaakId, string Resource);
/// projection row — the ZGW <c>resource</c> (zaak-create INGEDIEND vs status-set → INGESCHREVEN) and
/// the zaak <c>reference</c> (identificatie), so a rebuild reproduces the row without re-reading ZGW (#78).</summary>
public sealed record RecordedNotification(string Key, string Actie, string ZaakId, string Resource, string? Reference);
/// <summary>
/// Port to the Anti-Corruption Layer. The subscriber enriches the projection with the zaak's
/// public-safe reference (its identificatie) by asking the ACL — the only code that may read ZGW
/// (§8.1) — rather than reading OpenZaak itself (adr-proposal #78).
/// </summary>
public interface IAclClient
{
/// <summary>The zaak's reference (identificatie) for the read projection.</summary>
Task<string> GetZaakReferenceAsync(Uri zaakUrl, CancellationToken ct = default);
}
/// <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>

View File

@@ -9,6 +9,7 @@ namespace EventSubscriber.Application;
public sealed record RegisterEntry(
string Id,
string Status,
string? Reference = null,
string? Bsn = null,
string? NaamPlaceholder = null);

View File

@@ -0,0 +1,86 @@
using System.Net;
using System.Text;
using EventSubscriber.Api;
namespace EventSubscriber.Tests;
/// <summary>
/// Unit tests for the subscriber's ACL client, which reads a zaak's reference (identificatie) through
/// the ACL — the only code allowed to talk to ZGW (§8.1, #78). Uses a scripted message handler so no
/// real ACL is required.
/// </summary>
public class AclHttpClientTests
{
private static AclHttpClient Client(StubHandler handler) =>
new(new HttpClient(handler) { BaseAddress = new Uri("http://acl/") });
[Fact]
public async Task Reads_a_zaak_reference_by_posting_the_zaak_url_and_returns_it()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK, """{"reference":"REG-42"}"""));
var reference = await client.GetZaakReferenceAsync(new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
Assert.Equal("REG-42", reference);
Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
Assert.Equal("http://acl/zaken/reference", capture.Seen.RequestUri!.ToString());
Assert.Contains("\"zaakUrl\":\"http://openzaak/zaken/api/v1/zaken/abc\"", 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.GetZaakReferenceAsync(new Uri("http://openzaak/zaken/api/v1/zaken/abc")));
}
[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.GetZaakReferenceAsync(new Uri("http://openzaak/zaken/api/v1/zaken/abc")));
Assert.Contains("empty", ex.Message, StringComparison.OrdinalIgnoreCase);
}
[Fact]
public async Task Rejects_a_null_zaak_url_without_sending_a_request()
{
var capture = new RequestCapture();
var client = Client(capture.Responds(HttpStatusCode.OK, """{"reference":"REG-1"}"""));
await Assert.ThrowsAsync<ArgumentNullException>(() => client.GetZaakReferenceAsync(null!));
Assert.Null(capture.Seen);
}
}
/// <summary>A test double for <see cref="HttpMessageHandler"/> that records the last request and
/// returns a scripted response — mirrors the ACL/domain gateway tests.</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; }
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, Encoding.UTF8, "application/json");
return response;
});
}

View File

@@ -19,6 +19,7 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\EventSubscriber.Api\EventSubscriber.Api.csproj" />
<ProjectReference Include="..\EventSubscriber.Application\EventSubscriber.Application.csproj" />
</ItemGroup>

View File

@@ -5,6 +5,19 @@ 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>
/// <summary>A fake ACL client that returns a fixed reference derived from the zaak, and records
/// how many times it was called (to prove a rebuild does not re-read via the ACL).</summary>
internal sealed class FakeAclClient : IAclClient
{
public int CallCount { get; private set; }
public Task<string> GetZaakReferenceAsync(Uri zaakUrl, CancellationToken ct = default)
{
CallCount++;
return Task.FromResult("REG-" + zaakUrl.Segments[^1].Trim('/'));
}
}
internal sealed class InMemoryNotificationLog : INotificationLog
{
private readonly Dictionary<string, RecordedNotification> _byKey = [];

View File

@@ -12,8 +12,9 @@ public sealed class NotificationProjectorTests
private readonly InMemoryNotificationLog _log = new();
private readonly InMemoryProjectionStore _store = new();
private readonly FakeAclClient _acl = new();
private NotificationProjector Projector() => new(_log, _store);
private NotificationProjector Projector() => new(_log, _store, _acl);
private static Notification ZaakCreated(string url = ZaakUrl)
=> new("zaken", "zaak", "create", new Uri(url));
@@ -30,6 +31,23 @@ public sealed class NotificationProjectorTests
var entry = Assert.Single(await _store.AllAsync());
Assert.Equal("11111111-1111-1111-1111-111111111111", entry.Id);
Assert.Equal(RegistrationStatus.Ingediend, entry.Status);
// Enriched with the zaak's reference (identificatie), fetched via the ACL (#78).
Assert.Equal("REG-11111111-1111-1111-1111-111111111111", entry.Reference);
}
[Fact]
public async Task rebuild_reproduces_the_reference_without_re_reading_via_the_acl()
{
var projector = Projector();
await projector.HandleAsync(ZaakCreated());
var callsAfterProjection = _acl.CallCount;
await projector.RebuildAsync();
var entry = Assert.Single(await _store.AllAsync());
Assert.Equal("REG-11111111-1111-1111-1111-111111111111", entry.Reference);
// Rebuild replays the log (which stored the reference) — no extra ACL calls (#78, ADR-0008).
Assert.Equal(callsAfterProjection, _acl.CallCount);
}
[Fact]