not
8af09b2c92
HandleAsync reads the record at the notification's object URL through the ACL and writes it to the projection verbatim — the record already carries id, status and reference, so there is no mapping and no enrichment hop. The dedup key is the object plus the state that write projects. It cannot be the object URL alone (the ACL upserts one object per registration, so submit and approval notify about the same URL and the approval would be swallowed), nor include the actie (a retried approval is a second `update`). Keying on the projected row collapses redeliveries and lets genuine state changes through — §8.6.
64 lines
3.3 KiB
C#
64 lines
3.3 KiB
C#
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 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 register record being written to
|
|
/// Objecten (S-19b-2, ADR-0030) and ignores everything else. The notification carries only the
|
|
/// object URL, so the record is read back through the ACL (§8.1) and becomes the row verbatim.</summary>
|
|
public async Task HandleAsync(Notification notification, CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(notification);
|
|
|
|
if (!notification.IsRegisterRecordWritten)
|
|
return;
|
|
|
|
var record = await acl.GetRegisterRecordAsync(notification.ObjectUrl, ct);
|
|
// The object is gone, or holds no register record — nothing to project (§8.6).
|
|
if (record is null)
|
|
return;
|
|
|
|
var recorded = new RecordedNotification(
|
|
KeyFor(notification.ObjectUrl, record), record.Id, record.Status, record.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);
|
|
}
|
|
|
|
/// <summary>
|
|
/// A deterministic dedup key: the object, plus the state that write puts in the projection.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Open Notificaties carries no notification id and may redeliver, so the key is derived from
|
|
/// content. It cannot be the object URL alone — the ACL upserts one object per registration, so
|
|
/// submit and approval both notify about the *same* URL and the approval would be swallowed as a
|
|
/// duplicate. Nor can it include the actie: a retried approval would be a second `update`. Keying
|
|
/// on the projected row means a redelivery collapses and a genuine state change does not, which
|
|
/// is exactly the property §8.6 asks for.
|
|
/// </remarks>
|
|
private static string KeyFor(Uri objectUrl, RegisterRecord record)
|
|
=> $"objecten:object:{objectUrl}:{record.Status}:{record.Reference}";
|
|
|
|
/// <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 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.RegisterId, recorded.Status, recorded.Reference);
|
|
}
|