diff --git a/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs b/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs index 71ade5a..1610787 100644 --- a/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs +++ b/services/event-subscriber/EventSubscriber.Application/NotificationProjector.cs @@ -8,12 +8,31 @@ namespace EventSubscriber.Application; 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; + public async Task HandleAsync(Notification notification, CancellationToken ct = default) + { + if (!notification.IsZaakCreated) + return; + + var recorded = new RecordedNotification(notification.IdempotencyKey, notification.Actie, notification.ZaakId); + + // 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); + } /// Rebuild the projection from the durable notification log (PRD §8.4). - public Task RebuildAsync(CancellationToken ct = default) - // RED: not yet implemented. - => Task.CompletedTask; + 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); + } + + /// The minimal projection row for an accepted notification. A zaak-create maps to + /// INGEDIEND; bsn/naam are deferred (ADR-0008). + private static RegisterEntry ToEntry(RecordedNotification recorded) + => new(recorded.ZaakId, RegistrationStatus.Ingediend); }