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) <noreply@anthropic.com>

services/domain/Big.Api/Program.cs

@@ -22,6 +22,7 @@ builder.Services.AddTransient<IWorkflowClient>(sp => sp.GetRequiredService<Flowa
 builder.Services.AddTransient<IExternalWorkerClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
 builder.Services.AddTransient<IUserTaskClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
 builder.Services.AddTransient<IBeoordelingEscalatieClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
+builder.Services.AddTransient<IRegistratieVerlopenClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
 builder.Services.AddHttpClient<IAclClient, AclHttpClient>();
 
 builder.Services.AddScoped<SubmitRegistration>();
@@ -32,12 +33,17 @@ builder.Services.AddScoped<Werkbak>();
 builder.Services.AddScoped<OpenZaakWorker>();
 builder.Services.AddScoped<OpenZaakJobProcessor>();
 builder.Services.AddScoped<BeoordelingEscalatieProcessor>();
+builder.Services.AddScoped<ExpireRegistrationWorker>();
+builder.Services.AddScoped<RegistratieVerlopenProcessor>();
 
 // The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion.
 builder.Services.AddHostedService<OpenZaakJobPump>();
 // 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<BeoordelingEscalatiePump>();
+// 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<RegistratieVerlopenPump>();
 
 var app = builder.Build();
 

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.
     /// </summary>
     Task WithdrawProcessAsync(string processInstanceId, CancellationToken ct = default);
+
+    /// <summary>
+    /// Signal that the required documents have arrived (S-10a): complete the <c>WachtOpDocumenten</c>
+    /// 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.
+    /// </summary>
+    Task CompleteDocumentWaitAsync(string processInstanceId, CancellationToken ct = default);
 }
 
 /// <summary>

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).
 /// </summary>
 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<TaskByInstanceQueryRequest, TaskQueryResult>(
+            "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<IReadOnlyList<EscalatieJob>> 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<IReadOnlyList<RegistratieVerlopenJob>> AcquireRegistratieVerlopenJobsAsync(int maxJobs, CancellationToken ct = default)
+    {
+        var request = new AcquireJobsRequest(VerlopenTopic, options.LockDuration, maxJobs, options.WorkerId);
+
+        var jobs = await PostAsync<AcquireJobsRequest, List<AcquiredJob>>(
+            "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<TResponse?> GetAsync<TResponse>(string path, CancellationToken ct)
     {
         var message = new HttpRequestMessage(HttpMethod.Get, new Uri(options.BaseUrl, path));

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;
+
+/// <summary>
+/// The hosted polling loop of the document-timeout worker (S-10a, ADR-0017): on an interval it
+/// resolves a scoped <see cref="RegistratieVerlopenProcessor"/> and asks it to drain the parked
+/// <c>RegistratieVerlopen</c> 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 <see cref="BeoordelingEscalatiePump"/>.
+/// </summary>
+public sealed class RegistratieVerlopenPump(
+    IServiceScopeFactory scopeFactory,
+    FlowableOptions options,
+    ILogger<RegistratieVerlopenPump> logger) : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                using var scope = scopeFactory.CreateScope();
+                var processor = scope.ServiceProvider.GetRequiredService<RegistratieVerlopenProcessor>();
+                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;
+            }
+        }
+    }
+}

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<string> 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;
+    }
 }
 
 /// <summary>A fake user-task client for the werkbak/decision use cases: returns a scripted set of

services/domain/stryker-config.json

@@ -5,7 +5,8 @@
     "reporters": ["progress", "html"],
     "mutate": [
       "!**/OpenZaakJobPump.cs",
-      "!**/BeoordelingEscalatiePump.cs"
+      "!**/BeoordelingEscalatiePump.cs",
+      "!**/RegistratieVerlopenPump.cs"
     ],
     "thresholds": {
       "high": 95,

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<string> 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;
+    }
 }
 
 /// <summary>An in-memory ACL stand-in: records the bsn it opened a zaak for and returns a fixed URL,