From 3f04cb856fa03a02d32161d30cbe7553fb796cb5 Mon Sep 17 00:00:00 2001 From: Niek Otten Date: Mon, 20 Jul 2026 09:48:38 +0200 Subject: [PATCH] feat(infra): Flowable RegistratieVerlopen worker + document-wait completion (refs #102) FlowableWorkflowClient implements IRegistratieVerlopenClient (acquire/complete the RegistratieVerlopen jobs) and CompleteDocumentWaitAsync (complete WachtOpDocumenten, best-effort). Wires the RegistratieVerlopenProcessor + hosted RegistratieVerlopenPump into the domain host and excludes the pump from mutation (like the other pumps). Fakes updated for the new IWorkflowClient member. Co-Authored-By: Claude Opus 4.8 (1M context) --- services/domain/Big.Api/Program.cs | 6 +++ services/domain/Big.Application/Ports.cs | 8 +++ .../FlowableWorkflowClient.cs | 40 ++++++++++++++- .../RegistratieVerlopenPump.cs | 49 +++++++++++++++++++ services/domain/Big.Tests/Fakes.cs | 7 +++ services/domain/stryker-config.json | 3 +- .../acceptance/Support/InMemoryDomainPorts.cs | 7 +++ 7 files changed, 118 insertions(+), 2 deletions(-) create mode 100644 services/domain/Big.Infrastructure/RegistratieVerlopenPump.cs diff --git a/services/domain/Big.Api/Program.cs b/services/domain/Big.Api/Program.cs index ac83c22..627725e 100644 --- a/services/domain/Big.Api/Program.cs +++ b/services/domain/Big.Api/Program.cs @@ -22,6 +22,7 @@ builder.Services.AddTransient(sp => sp.GetRequiredService(sp => sp.GetRequiredService()); builder.Services.AddTransient(sp => sp.GetRequiredService()); builder.Services.AddTransient(sp => sp.GetRequiredService()); +builder.Services.AddTransient(sp => sp.GetRequiredService()); builder.Services.AddHttpClient(); builder.Services.AddScoped(); @@ -32,12 +33,17 @@ builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); +builder.Services.AddScoped(); +builder.Services.AddScoped(); // The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion. builder.Services.AddHostedService(); // The escalation worker polls the BeoordelingEscaleren jobs the 14-day timer parks and reassigns // each overdue beoordeling to the teamlead (S-14). builder.Services.AddHostedService(); +// The document-timeout worker polls the RegistratieVerlopen jobs the 30-day timer on WachtOpDocumenten +// parks and expires each lapsed registration to VERLOPEN (S-10a, ADR-0017). +builder.Services.AddHostedService(); var app = builder.Build(); diff --git a/services/domain/Big.Application/Ports.cs b/services/domain/Big.Application/Ports.cs index 492b232..af12888 100644 --- a/services/domain/Big.Application/Ports.cs +++ b/services/domain/Big.Application/Ports.cs @@ -25,6 +25,14 @@ public interface IWorkflowClient /// ended, or not yet parked) it is a no-op; the aggregate is INGETROKKEN regardless. /// Task WithdrawProcessAsync(string processInstanceId, CancellationToken ct = default); + + /// + /// Signal that the required documents have arrived (S-10a): complete the WachtOpDocumenten + /// user task in the instance so the process leaves the 30-day wait state and continues to + /// beoordeling (ADR-0017). Best-effort โ€” if the instance is not parked at that task (already + /// continued, or timed out) it is a no-op. The upload trigger that calls this is wired in S-10b. + /// + Task CompleteDocumentWaitAsync(string processInstanceId, CancellationToken ct = default); } /// diff --git a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs b/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs index edf379c..b3938b4 100644 --- a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs +++ b/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs @@ -15,12 +15,14 @@ namespace Big.Infrastructure; /// The REST contract here is the one verified against a live flowable-rest engine (ADR-0009). /// public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions options) - : IWorkflowClient, IExternalWorkerClient, IUserTaskClient, IBeoordelingEscalatieClient + : IWorkflowClient, IExternalWorkerClient, IUserTaskClient, IBeoordelingEscalatieClient, IRegistratieVerlopenClient { private const string Topic = "OpenZaakAanmaken"; private const string EscalatieTopic = "BeoordelingEscaleren"; + private const string VerlopenTopic = "RegistratieVerlopen"; private const string ProcessDefinitionKey = "registratie"; private const string BeoordelenTaskKey = "Beoordelen"; + private const string WachtOpDocumentenTaskKey = "WachtOpDocumenten"; private const string BehandelaarGroup = "behandelaar"; private const string TeamleadGroup = "teamlead"; private const string RegistrationIdVariable = "registrationId"; @@ -114,6 +116,25 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti response.EnsureSuccessStatusCode(); } + public async Task CompleteDocumentWaitAsync(string processInstanceId, CancellationToken ct = default) + { + // Find the still-open WachtOpDocumenten task in this instance and complete it, so the process + // leaves the 30-day wait and continues to beoordeling (S-10a, ADR-0017). If the instance is no + // longer parked there (already continued, or the timer already cancelled it) this is a + // best-effort no-op โ€” mirroring the withdrawal/escalation correlation (ยง8.6). + var query = new TaskByInstanceQueryRequest(processInstanceId, WachtOpDocumentenTaskKey); + var page = await PostAsync( + "service/query/tasks", query, ct); + + var task = page?.Data?.FirstOrDefault(); + if (task is null) + return; + + using var response = await SendAsync( + $"service/runtime/tasks/{task.Id}", new CompleteTaskRequest("complete", []), ct); + response.EnsureSuccessStatusCode(); + } + public async Task> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default) { var request = new AcquireJobsRequest(EscalatieTopic, options.LockDuration, maxJobs, options.WorkerId); @@ -155,6 +176,23 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti response.EnsureSuccessStatusCode(); } + public async Task> AcquireRegistratieVerlopenJobsAsync(int maxJobs, CancellationToken ct = default) + { + var request = new AcquireJobsRequest(VerlopenTopic, options.LockDuration, maxJobs, options.WorkerId); + + var jobs = await PostAsync>( + "external-job-api/acquire/jobs", request, ct) ?? []; + + return [.. jobs.Select(job => new RegistratieVerlopenJob(job.Id, RegistrationId.Parse(job.RegistrationId())))]; + } + + public async Task CompleteRegistratieVerlopenJobAsync(string jobId, CancellationToken ct = default) + { + using var response = await SendAsync( + $"external-job-api/acquire/jobs/{jobId}/complete", new CompleteJobRequest(options.WorkerId, []), ct); + response.EnsureSuccessStatusCode(); + } + private async Task GetAsync(string path, CancellationToken ct) { var message = new HttpRequestMessage(HttpMethod.Get, new Uri(options.BaseUrl, path)); diff --git a/services/domain/Big.Infrastructure/RegistratieVerlopenPump.cs b/services/domain/Big.Infrastructure/RegistratieVerlopenPump.cs new file mode 100644 index 0000000..4b2ba82 --- /dev/null +++ b/services/domain/Big.Infrastructure/RegistratieVerlopenPump.cs @@ -0,0 +1,49 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace Big.Infrastructure; + +/// +/// The hosted polling loop of the document-timeout worker (S-10a, ADR-0017): on an interval it +/// resolves a scoped and asks it to drain the parked +/// RegistratieVerlopen jobs. A deliberately thin shell โ€” all acquire/expire/complete logic +/// lives in the processor, which is unit-tested; this class only owns the timer, the per-tick scope, +/// and loop resilience. Structurally identical to . +/// +public sealed class RegistratieVerlopenPump( + IServiceScopeFactory scopeFactory, + FlowableOptions options, + ILogger logger) : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + while (!stoppingToken.IsCancellationRequested) + { + try + { + using var scope = scopeFactory.CreateScope(); + var processor = scope.ServiceProvider.GetRequiredService(); + await processor.PumpOnceAsync(options.MaxJobsPerPoll, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + // A transient fault (e.g. Flowable briefly unreachable) must not kill the loop. + logger.LogError(ex, "RegistratieVerlopen job poll failed; retrying after the poll interval."); + } + + try + { + await Task.Delay(options.PollInterval, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + } + } +} diff --git a/services/domain/Big.Tests/Fakes.cs b/services/domain/Big.Tests/Fakes.cs index e9c6953..f183fda 100644 --- a/services/domain/Big.Tests/Fakes.cs +++ b/services/domain/Big.Tests/Fakes.cs @@ -34,6 +34,7 @@ internal sealed class FakeWorkflowClient(string processInstanceId = "proc-1", Ac public RegistrationId? StartedFor { get; private set; } public DiplomaOrigin? StartedWithOrigin { get; private set; } public string? WithdrawnProcessInstanceId { get; private set; } + public string? CompletedDocumentWaitFor { get; private set; } public Task StartRegistrationProcessAsync( RegistrationId registrationId, DiplomaOrigin diplomaOrigin, CancellationToken ct = default) @@ -49,6 +50,12 @@ internal sealed class FakeWorkflowClient(string processInstanceId = "proc-1", Ac WithdrawnProcessInstanceId = processInstanceId; return Task.CompletedTask; } + + public Task CompleteDocumentWaitAsync(string processInstanceId, CancellationToken ct = default) + { + CompletedDocumentWaitFor = processInstanceId; + return Task.CompletedTask; + } } /// A fake user-task client for the werkbak/decision use cases: returns a scripted set of diff --git a/services/domain/stryker-config.json b/services/domain/stryker-config.json index 0391844..ac22608 100644 --- a/services/domain/stryker-config.json +++ b/services/domain/stryker-config.json @@ -5,7 +5,8 @@ "reporters": ["progress", "html"], "mutate": [ "!**/OpenZaakJobPump.cs", - "!**/BeoordelingEscalatiePump.cs" + "!**/BeoordelingEscalatiePump.cs", + "!**/RegistratieVerlopenPump.cs" ], "thresholds": { "high": 95, diff --git a/tests/acceptance/Support/InMemoryDomainPorts.cs b/tests/acceptance/Support/InMemoryDomainPorts.cs index af07de0..d52bb05 100644 --- a/tests/acceptance/Support/InMemoryDomainPorts.cs +++ b/tests/acceptance/Support/InMemoryDomainPorts.cs @@ -14,6 +14,7 @@ public sealed class InMemoryWorkflowClient : IWorkflowClient public RegistrationId? StartedFor { get; private set; } public DiplomaOrigin? StartedWithOrigin { get; private set; } public string? WithdrawnProcessInstanceId { get; private set; } + public string? CompletedDocumentWaitFor { get; private set; } public Task StartRegistrationProcessAsync( RegistrationId registrationId, DiplomaOrigin diplomaOrigin, CancellationToken ct = default) @@ -28,6 +29,12 @@ public sealed class InMemoryWorkflowClient : IWorkflowClient WithdrawnProcessInstanceId = processInstanceId; return Task.CompletedTask; } + + public Task CompleteDocumentWaitAsync(string processInstanceId, CancellationToken ct = default) + { + CompletedDocumentWaitFor = processInstanceId; + return Task.CompletedTask; + } } /// An in-memory ACL stand-in: records the bsn it opened a zaak for and returns a fixed URL,