feat(projection): persist the read projection and expose webhook + read APIs (refs #7)

Add the projection persistence and the two services around it:

- Projection.ReadModel: a shared EF Core (Npgsql) read model owning the projection
  schema — register_projection + the subscriber's processed_notifications log — plus
  EfProjectionStore / EfNotificationLog (atomic record-or-skip on the PK for idempotency)
  and the initial migration. One rebuildable store, written by the subscriber and read
  by projection-api (ADR-0008).
- EventSubscriber.Api: POST /notifications NRC callback (enforces the abonnement bearer,
  401 without it per ADR-0007), POST /admin/rebuild, /health. Migrates on start.
- ProjectionApi.Api: GET /register, GET /register/{id}, /health — the read side.

dotnet-ef pinned as a local tool for migrations; NuGetAuditMode=direct so EF's
design-time-only tooling transitive doesn't flag the shipped build.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 14:55:04 +02:00
parent 017cd5e66b
commit 7ef63c7ae9
17 changed files with 576 additions and 0 deletions

View File

@@ -8,6 +8,13 @@
"dotnet-stryker" "dotnet-stryker"
], ],
"rollForward": false "rollForward": false
},
"dotnet-ef": {
"version": "10.0.0",
"commands": [
"dotnet-ef"
],
"rollForward": false
} }
} }
} }

View File

@@ -12,9 +12,14 @@
<Project Path="services/bff/Bff.Tests/Bff.Tests.csproj" /> <Project Path="services/bff/Bff.Tests/Bff.Tests.csproj" />
</Folder> </Folder>
<Folder Name="/services/event-subscriber/"> <Folder Name="/services/event-subscriber/">
<Project Path="services/event-subscriber/EventSubscriber.Api/EventSubscriber.Api.csproj" />
<Project Path="services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj" /> <Project Path="services/event-subscriber/EventSubscriber.Application/EventSubscriber.Application.csproj" />
<Project Path="services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj" /> <Project Path="services/event-subscriber/EventSubscriber.Tests/EventSubscriber.Tests.csproj" />
</Folder> </Folder>
<Folder Name="/services/projection-api/">
<Project Path="services/projection-api/Projection.ReadModel/Projection.ReadModel.csproj" />
<Project Path="services/projection-api/ProjectionApi.Api/ProjectionApi.Api.csproj" />
</Folder>
<Folder Name="/tests/"> <Folder Name="/tests/">
<Project Path="tests/acceptance/Acceptance.csproj" /> <Project Path="tests/acceptance/Acceptance.csproj" />
</Folder> </Folder>

View File

@@ -0,0 +1,14 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<ItemGroup>
<ProjectReference Include="..\EventSubscriber.Application\EventSubscriber.Application.csproj" />
<ProjectReference Include="..\..\projection-api\Projection.ReadModel\Projection.ReadModel.csproj" />
</ItemGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,56 @@
using EventSubscriber.Application;
using Projection.ReadModel;
var builder = WebApplication.CreateBuilder(args);
var connectionString = builder.Configuration.GetConnectionString("Projection")
?? throw new InvalidOperationException("Missing connection string 'ConnectionStrings:Projection'");
// The exact Authorization header value Open Notificaties sends on each abonnement callback.
// NRC probes the callback during registration and refuses it unless it returns 401 without
// this value (ADR-0007), so the webhook enforces it.
var webhookToken = builder.Configuration["EventSubscriber:Webhook:AuthToken"]
?? throw new InvalidOperationException("Missing configuration 'EventSubscriber:Webhook:AuthToken'");
builder.Services.AddProjectionReadModel(connectionString);
builder.Services.AddProjectionWriteSide();
var app = builder.Build();
// Apply migrations on start so a fresh stack reaches a usable schema unattended.
await app.Services.MigrateProjectionAsync();
app.MapGet("/health", () => "Healthy");
// The NRC abonnement callback. Open Notificaties POSTs a notification here; we project it.
// Auth-on-callback is mandatory: without the configured Authorization, return 401 (NRC's
// registration probe depends on this — ADR-0007).
app.MapPost("/notifications", async (
HttpRequest request,
NotificationDto body,
NotificationProjector projector,
CancellationToken ct) =>
{
if (request.Headers.Authorization != webhookToken)
return Results.Unauthorized();
await projector.HandleAsync(body.ToNotification(), ct);
return Results.NoContent();
});
// Admin: rebuild the projection from the durable notification log (PRD §8.4 — rebuildable).
app.MapPost("/admin/rebuild", async (NotificationProjector projector, CancellationToken ct) =>
{
await projector.RebuildAsync(ct);
return Results.NoContent();
});
await app.RunAsync();
/// <summary>The NRC notification body, as Open Notificaties POSTs it. Only the fields the
/// projection needs are bound; <c>aanmaakdatum</c>/<c>kenmerken</c> are ignored for the minimal slice.</summary>
public sealed record NotificationDto(string Kanaal, string Resource, string Actie, Uri ResourceUrl)
{
public Notification ToNotification() => new(Kanaal, Resource, Actie, ResourceUrl);
}
public partial class Program;

View File

@@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}

View File

@@ -0,0 +1,18 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Design;
namespace Projection.ReadModel;
/// <summary>Lets <c>dotnet ef migrations</c> build the context at design time without a running
/// database or the host's DI. The connection string is a placeholder — migrations only need the
/// provider to emit Postgres-shaped SQL.</summary>
public sealed class DesignTimeProjectionDbContextFactory : IDesignTimeDbContextFactory<ProjectionDbContext>
{
public ProjectionDbContext CreateDbContext(string[] args)
{
var options = new DbContextOptionsBuilder<ProjectionDbContext>()
.UseNpgsql("Host=localhost;Database=projection;Username=projection;Password=projection")
.Options;
return new ProjectionDbContext(options);
}
}

View File

@@ -0,0 +1,39 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>EF Core implementation of the notification log. Idempotency is enforced atomically
/// by the primary key on <c>key</c>: a duplicate insert raises a unique violation, which is
/// caught and reported as "already recorded" rather than failing the request.</summary>
public sealed class EfNotificationLog(ProjectionDbContext db) : INotificationLog
{
public async Task<bool> TryRecordAsync(RecordedNotification notification, CancellationToken ct = default)
{
db.ProcessedNotifications.Add(new ProcessedNotificationRow
{
Key = notification.Key,
Actie = notification.Actie,
ZaakId = notification.ZaakId,
ReceivedAt = DateTimeOffset.UtcNow,
});
try
{
await db.SaveChangesAsync(ct);
return true;
}
catch (DbUpdateException)
{
// Already recorded by an earlier (or concurrent) delivery — drop this duplicate.
db.ChangeTracker.Clear();
return false;
}
}
public async Task<IReadOnlyList<RecordedNotification>> AllAsync(CancellationToken ct = default)
=> await db.ProcessedNotifications
.OrderBy(r => r.ReceivedAt)
.Select(r => new RecordedNotification(r.Key, r.Actie, r.ZaakId))
.ToListAsync(ct);
}

View File

@@ -0,0 +1,40 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>EF Core implementation of the projection store over <see cref="ProjectionDbContext"/>.</summary>
public sealed class EfProjectionStore(ProjectionDbContext db) : IProjectionStore
{
public async Task UpsertAsync(RegisterEntry entry, CancellationToken ct = default)
{
var row = await db.RegisterEntries.FindAsync([entry.Id], ct);
if (row is null)
{
db.RegisterEntries.Add(new RegisterEntryRow
{
Id = entry.Id,
Status = entry.Status,
Bsn = entry.Bsn,
NaamPlaceholder = entry.NaamPlaceholder,
});
}
else
{
row.Status = entry.Status;
row.Bsn = entry.Bsn;
row.NaamPlaceholder = entry.NaamPlaceholder;
}
await db.SaveChangesAsync(ct);
}
public async Task ClearAsync(CancellationToken ct = default)
=> await db.RegisterEntries.ExecuteDeleteAsync(ct);
public async Task<IReadOnlyList<RegisterEntry>> AllAsync(CancellationToken ct = default)
=> await db.RegisterEntries
.OrderBy(r => r.Id)
.Select(r => new RegisterEntry(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.ToListAsync(ct);
}

View File

@@ -0,0 +1,79 @@
// <auto-generated />
using System;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Migrations;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
using Projection.ReadModel;
#nullable disable
namespace Projection.ReadModel.Migrations
{
[DbContext(typeof(ProjectionDbContext))]
[Migration("20260630125055_InitialProjection")]
partial class InitialProjection
{
/// <inheritdoc />
protected override void BuildTargetModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
.HasAnnotation("ProductVersion", "10.0.0")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
modelBuilder.Entity("Projection.ReadModel.ProcessedNotificationRow", b =>
{
b.Property<string>("Key")
.HasColumnType("text")
.HasColumnName("key");
b.Property<string>("Actie")
.IsRequired()
.HasColumnType("text")
.HasColumnName("actie");
b.Property<DateTimeOffset>("ReceivedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("received_at");
b.Property<string>("ZaakId")
.IsRequired()
.HasColumnType("text")
.HasColumnName("zaak_id");
b.HasKey("Key");
b.ToTable("processed_notifications", (string)null);
});
modelBuilder.Entity("Projection.ReadModel.RegisterEntryRow", b =>
{
b.Property<string>("Id")
.HasColumnType("text")
.HasColumnName("id");
b.Property<string>("Bsn")
.HasColumnType("text")
.HasColumnName("bsn");
b.Property<string>("NaamPlaceholder")
.HasColumnType("text")
.HasColumnName("naam_placeholder");
b.Property<string>("Status")
.IsRequired()
.HasColumnType("text")
.HasColumnName("status");
b.HasKey("Id");
b.ToTable("register_projection", (string)null);
});
#pragma warning restore 612, 618
}
}
}

View File

@@ -0,0 +1,53 @@
using System;
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace Projection.ReadModel.Migrations
{
/// <inheritdoc />
public partial class InitialProjection : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.CreateTable(
name: "processed_notifications",
columns: table => new
{
key = table.Column<string>(type: "text", nullable: false),
actie = table.Column<string>(type: "text", nullable: false),
zaak_id = table.Column<string>(type: "text", nullable: false),
received_at = table.Column<DateTimeOffset>(type: "timestamp with time zone", nullable: false)
},
constraints: table =>
{
table.PrimaryKey("PK_processed_notifications", x => x.key);
});
migrationBuilder.CreateTable(
name: "register_projection",
columns: table => new
{
id = table.Column<string>(type: "text", nullable: false),
status = table.Column<string>(type: "text", nullable: false),
bsn = table.Column<string>(type: "text", nullable: true),
naam_placeholder = table.Column<string>(type: "text", nullable: true)
},
constraints: table =>
{
table.PrimaryKey("PK_register_projection", x => x.id);
});
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropTable(
name: "processed_notifications");
migrationBuilder.DropTable(
name: "register_projection");
}
}
}

View File

@@ -0,0 +1,76 @@
// <auto-generated />
using System;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Infrastructure;
using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata;
using Projection.ReadModel;
#nullable disable
namespace Projection.ReadModel.Migrations
{
[DbContext(typeof(ProjectionDbContext))]
partial class ProjectionDbContextModelSnapshot : ModelSnapshot
{
protected override void BuildModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder
.HasAnnotation("ProductVersion", "10.0.0")
.HasAnnotation("Relational:MaxIdentifierLength", 63);
NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder);
modelBuilder.Entity("Projection.ReadModel.ProcessedNotificationRow", b =>
{
b.Property<string>("Key")
.HasColumnType("text")
.HasColumnName("key");
b.Property<string>("Actie")
.IsRequired()
.HasColumnType("text")
.HasColumnName("actie");
b.Property<DateTimeOffset>("ReceivedAt")
.HasColumnType("timestamp with time zone")
.HasColumnName("received_at");
b.Property<string>("ZaakId")
.IsRequired()
.HasColumnType("text")
.HasColumnName("zaak_id");
b.HasKey("Key");
b.ToTable("processed_notifications", (string)null);
});
modelBuilder.Entity("Projection.ReadModel.RegisterEntryRow", b =>
{
b.Property<string>("Id")
.HasColumnType("text")
.HasColumnName("id");
b.Property<string>("Bsn")
.HasColumnType("text")
.HasColumnName("bsn");
b.Property<string>("NaamPlaceholder")
.HasColumnType("text")
.HasColumnName("naam_placeholder");
b.Property<string>("Status")
.IsRequired()
.HasColumnType("text")
.HasColumnName("status");
b.HasKey("Id");
b.ToTable("register_projection", (string)null);
});
#pragma warning restore 612, 618
}
}
}

View File

@@ -0,0 +1,28 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<!-- Audit only the packages we choose directly. The one flagged transitive
(System.Security.Cryptography.Xml, NU1903) comes via EF Core's design-time
migration tooling (Microsoft.EntityFrameworkCore.Design, PrivateAssets=all) and
is never shipped at runtime — it is out of our control to upgrade. -->
<NuGetAuditMode>direct</NuGetAuditMode>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="10.0.0" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="10.0.0" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="10.0.0">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="10.0.0" />
</ItemGroup>
</Project>

View File

@@ -0,0 +1,59 @@
using Microsoft.EntityFrameworkCore;
namespace Projection.ReadModel;
/// <summary>
/// The read projection's persistence. This single store is the system's rebuildable read model
/// (PRD §8.4): the Event Subscriber writes to it (projector) and the projection-api reads it
/// (query). Both processes are the one "Read Projection" bounded context and share this schema —
/// not a cross-domain DB share (ADR-0008 reconciles this with CLAUDE.md §8.5).
/// </summary>
public sealed class ProjectionDbContext(DbContextOptions<ProjectionDbContext> options) : DbContext(options)
{
/// <summary>The public-facing register rows (public-safe filtering is tightened in S-09).</summary>
public DbSet<RegisterEntryRow> RegisterEntries => Set<RegisterEntryRow>();
/// <summary>The Event Subscriber's durable log of accepted notifications — idempotency guard and rebuild source.</summary>
public DbSet<ProcessedNotificationRow> ProcessedNotifications => Set<ProcessedNotificationRow>();
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder.Entity<RegisterEntryRow>(e =>
{
e.ToTable("register_projection");
e.HasKey(r => r.Id);
e.Property(r => r.Id).HasColumnName("id");
e.Property(r => r.Status).HasColumnName("status").IsRequired();
e.Property(r => r.Bsn).HasColumnName("bsn");
e.Property(r => r.NaamPlaceholder).HasColumnName("naam_placeholder");
});
modelBuilder.Entity<ProcessedNotificationRow>(e =>
{
e.ToTable("processed_notifications");
e.HasKey(r => r.Key);
e.Property(r => r.Key).HasColumnName("key");
e.Property(r => r.Actie).HasColumnName("actie").IsRequired();
e.Property(r => r.ZaakId).HasColumnName("zaak_id").IsRequired();
e.Property(r => r.ReceivedAt).HasColumnName("received_at");
});
}
}
/// <summary>A register projection row. <c>Bsn</c>/<c>NaamPlaceholder</c> are deferred (ADR-0008).</summary>
public sealed class RegisterEntryRow
{
public required string Id { get; set; }
public required string Status { get; set; }
public string? Bsn { get; set; }
public string? NaamPlaceholder { get; set; }
}
/// <summary>An accepted notification, retained so the projection can be rebuilt without OpenZaak (§8.1).</summary>
public sealed class ProcessedNotificationRow
{
public required string Key { get; set; }
public required string Actie { get; set; }
public required string ZaakId { get; set; }
public DateTimeOffset ReceivedAt { get; set; }
}

View File

@@ -0,0 +1,30 @@
using EventSubscriber.Application;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
namespace Projection.ReadModel;
public static class ServiceCollectionExtensions
{
/// <summary>Register the projection <see cref="ProjectionDbContext"/> against Postgres.</summary>
public static IServiceCollection AddProjectionReadModel(this IServiceCollection services, string connectionString)
=> services.AddDbContext<ProjectionDbContext>(o => o.UseNpgsql(connectionString));
/// <summary>Register the write-side ports (projector store + notification log) used by the Event Subscriber.</summary>
public static IServiceCollection AddProjectionWriteSide(this IServiceCollection services)
{
services.AddScoped<IProjectionStore, EfProjectionStore>();
services.AddScoped<INotificationLog, EfNotificationLog>();
services.AddScoped<NotificationProjector>();
return services;
}
/// <summary>Apply any pending EF migrations. Called once on service start so a fresh stack
/// reaches a usable schema without a manual migration step (DoD: compose up reaches green).</summary>
public static async Task MigrateProjectionAsync(this IServiceProvider services, CancellationToken ct = default)
{
await using var scope = services.CreateAsyncScope();
var db = scope.ServiceProvider.GetRequiredService<ProjectionDbContext>();
await db.Database.MigrateAsync(ct);
}
}

View File

@@ -0,0 +1,43 @@
using Microsoft.EntityFrameworkCore;
using Projection.ReadModel;
var builder = WebApplication.CreateBuilder(args);
var connectionString = builder.Configuration.GetConnectionString("Projection")
?? throw new InvalidOperationException("Missing connection string 'ConnectionStrings:Projection'");
builder.Services.AddProjectionReadModel(connectionString);
var app = builder.Build();
// Ensure the schema exists before serving reads. EF serialises concurrent migrators via the
// migrations-history lock, so it is safe that the Event Subscriber migrates too.
await app.Services.MigrateProjectionAsync();
app.MapGet("/health", () => "Healthy");
// The read side of the projection. Public-safe field filtering is tightened in S-09; for now
// the minimal projection only carries id + status (bsn/naam deferred — ADR-0008).
app.MapGet("/register", async (ProjectionDbContext db, CancellationToken ct) =>
{
var rows = await db.RegisterEntries
.OrderBy(r => r.Id)
.Select(r => new RegisterEntryDto(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.ToListAsync(ct);
return Results.Ok(rows);
});
app.MapGet("/register/{id}", async (string id, ProjectionDbContext db, CancellationToken ct) =>
{
var row = await db.RegisterEntries
.Where(r => r.Id == id)
.Select(r => new RegisterEntryDto(r.Id, r.Status, r.Bsn, r.NaamPlaceholder))
.SingleOrDefaultAsync(ct);
return row is null ? Results.NotFound() : Results.Ok(row);
});
await app.RunAsync();
public sealed record RegisterEntryDto(string Id, string Status, string? Bsn, string? NaamPlaceholder);
public partial class Program;

View File

@@ -0,0 +1,13 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<ItemGroup>
<ProjectReference Include="..\Projection.ReadModel\Projection.ReadModel.csproj" />
</ItemGroup>
<PropertyGroup>
<TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
</Project>

View File

@@ -0,0 +1,8 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}