test(event-subscriber): the projection is sourced from register records (refs #153)

Ports, schema and failing tests for the subscriber half of S-19b-2, ahead of the
implementation.

The subscriber now listens on the `objecten` kanaal instead of `zaken`. An Objecten
notification carries no record data — only the object URL — so the record is read back
through the ACL (§8.1), and the zaak-shaped surface goes away: IsZaakCreated /
IsZaakStatusSet / ZaakUrl / ZaakId and ToEntry's `Resource == "status"` mapping are replaced
by IsRegisterRecordWritten + ObjectUrl.

The notification log now holds the projected row itself (register id, status, reference),
so a rebuild is a replay with no mapping rules and no upstream reads. The migration drops
the old columns rather than renaming them — EF scaffolded renames that would have carried
ZGW values into columns meaning something else — and empties both tables, since a
pre-slice row is neither reprojectable nor re-derivable from the new source.

Red: HandleAsync recognises a register write but does not yet read or project it, so the
seven projection assertions fail on an empty store.
This commit is contained in:
not
2026-08-28 12:32:15 +02:00
parent 566ef7dd64
commit 142ed454aa
12 changed files with 336 additions and 184 deletions
@@ -3,28 +3,22 @@ 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.
/// upsert is idempotent on the register id. Rebuilds the projection by replaying the log.
/// </summary>
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. 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>
/// <summary>Handle one inbound notification. Reacts to a register record being written to
/// Objecten (S-19b-2, ADR-0030) and ignores everything else.</summary>
public async Task HandleAsync(Notification notification, CancellationToken ct = default)
{
if (!notification.IsZaakCreated && !notification.IsZaakStatusSet)
ArgumentNullException.ThrowIfNull(notification);
if (!notification.IsRegisterRecordWritten)
return;
var reference = await acl.GetZaakReferenceAsync(notification.ZaakUrl, ct);
var recorded = new RecordedNotification(
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.
if (!await log.TryRecordAsync(recorded, ct))
return;
await store.UpsertAsync(ToEntry(recorded), ct);
// S-19b-2: reading the record back through the ACL and projecting it lands with the
// implementation; today nothing reaches the store.
await Task.CompletedTask;
}
/// <summary>Rebuild the projection from the durable notification log (PRD §8.4).</summary>
@@ -35,11 +29,9 @@ public sealed class NotificationProjector(INotificationLog log, IProjectionStore
await store.UpsertAsync(ToEntry(recorded), ct);
}
/// <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>
/// <summary>The projection row for an accepted notification. The log already holds exactly the
/// row's fields, so a rebuild needs no mapping rules and no upstream reads. bsn/naam stay
/// deferred — the register record is public-safe by construction (ADR-0027).</summary>
private static RegisterEntry ToEntry(RecordedNotification recorded)
=> new(
recorded.ZaakId,
recorded.Resource == "status" ? RegistrationStatus.Ingeschreven : RegistrationStatus.Ingediend,
Reference: recorded.Reference);
=> new(recorded.RegisterId, recorded.Status, recorded.Reference);
}