feat(domain): implement the Flowable Workflow Client and ACL client (refs #6)

FlowableWorkflowClient speaks flowable-rest's REST API (Basic auth): start a
registratie process with the registrationId variable, acquire OpenZaakAanmaken
external-worker jobs and parse their registrationId, complete a job with the
zaakUrl variable — the contract verified against a live engine. AclHttpClient
POSTs the bsn to the ACL and returns the zaak URL. InMemoryRegistrationStore is
a concurrent-dictionary upsert. OpenZaakJobProcessor drains parked jobs, opening
a zaak per job and completing it, leaving failures for redelivery; OpenZaakJobPump
is the hosted polling shell that drives it on an interval (ADR-0009).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

services/domain/Big.Infrastructure/AclHttpClient.cs

@@ -1,3 +1,5 @@
+using System.Net.Http.Json;
+using System.Text.Json.Serialization;
 using Big.Application;
 
 namespace Big.Infrastructure;
@@ -9,9 +11,18 @@ namespace Big.Infrastructure;
 /// </summary>
 public sealed class AclHttpClient(HttpClient http, AclOptions options) : IAclClient
 {
-    public Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default)
+    public async Task<Uri> OpenZaakAsync(string bsn, CancellationToken ct = default)
     {
-        // STUB (red).
+        using var response = await http.PostAsJsonAsync(
-        return Task.FromResult(new Uri("about:blank"));
+            new Uri(options.BaseUrl, "zaken"), new OpenZaakRequest(bsn), ct);
+        response.EnsureSuccessStatusCode();
+
+        var opened = await response.Content.ReadFromJsonAsync<OpenZaakResponse>(ct)
+            ?? throw new InvalidOperationException("The ACL returned an empty zaak response.");
+        return new Uri(opened.ZaakUrl);
     }
+
+    private sealed record OpenZaakRequest([property: JsonPropertyName("bsn")] string Bsn);
+
+    private sealed record OpenZaakResponse([property: JsonPropertyName("zaakUrl")] string ZaakUrl);
 }

services/domain/Big.Infrastructure/FlowableWorkflowClient.cs

@@ -1,3 +1,8 @@
+using System.Net.Http.Headers;
+using System.Net.Http.Json;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Serialization;
 using Big.Application;
 using Big.Domain;
 
@@ -7,25 +12,97 @@ namespace Big.Infrastructure;
 /// The Workflow Client — the only code that talks to Flowable (§8.2). It starts registratie process
 /// instances and drives the <c>OpenZaakAanmaken</c> external-worker jobs over Flowable's REST API
 /// (start: <c>service/runtime/process-instances</c>; acquire/complete: <c>external-job-api/…</c>).
+/// 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
 {
-    public Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default)
+    private const string Topic = "OpenZaakAanmaken";
+    private const string ProcessDefinitionKey = "registratie";
+    private const string RegistrationIdVariable = "registrationId";
+    private const string ZaakUrlVariable = "zaakUrl";
+
+    public async Task<string> StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default)
     {
-        // STUB (red).
+        var request = new StartProcessRequest(
-        return Task.FromResult("");
+            ProcessDefinitionKey,
+            [new Variable(RegistrationIdVariable, "string", registrationId.ToString())]);
+
+        var created = await PostAsync<StartProcessRequest, ProcessInstance>(
+            "service/runtime/process-instances", request, ct)
+            ?? throw new InvalidOperationException("Flowable returned an empty process-instance response.");
+        return created.Id;
     }
 
-    public Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
+    public async Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
     {
-        // STUB (red).
+        var request = new AcquireJobsRequest(Topic, options.LockDuration, maxJobs, options.WorkerId);
-        return Task.FromResult<IReadOnlyList<OpenZaakJob>>([]);
+
+        var jobs = await PostAsync<AcquireJobsRequest, List<AcquiredJob>>(
+            "external-job-api/acquire/jobs", request, ct) ?? [];
+
+        return [.. jobs.Select(job => new OpenZaakJob(job.Id, RegistrationId.Parse(job.RegistrationId())))];
     }
 
-    public Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
+    public async Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
     {
-        // STUB (red).
+        var request = new CompleteJobRequest(
-        return Task.CompletedTask;
+            options.WorkerId,
+            [new Variable(ZaakUrlVariable, "string", zaakUrl.ToString())]);
+
+        using var response = await SendAsync(
+            $"external-job-api/acquire/jobs/{jobId}/complete", request, ct);
+        response.EnsureSuccessStatusCode();
+    }
+
+    private async Task<TResponse?> PostAsync<TRequest, TResponse>(string path, TRequest body, CancellationToken ct)
+    {
+        using var response = await SendAsync(path, body, ct);
+        response.EnsureSuccessStatusCode();
+        return await response.Content.ReadFromJsonAsync<TResponse>(ct);
+    }
+
+    private Task<HttpResponseMessage> SendAsync<TRequest>(string path, TRequest body, CancellationToken ct)
+    {
+        var message = new HttpRequestMessage(HttpMethod.Post, new Uri(options.BaseUrl, path))
+        {
+            Content = JsonContent.Create(body),
+        };
+        message.Headers.Authorization = new AuthenticationHeaderValue("Basic", BasicCredentials());
+        return http.SendAsync(message, ct);
+    }
+
+    private string BasicCredentials()
+        => Convert.ToBase64String(Encoding.ASCII.GetBytes($"{options.Username}:{options.Password}"));
+
+    private sealed record StartProcessRequest(
+        [property: JsonPropertyName("processDefinitionKey")] string ProcessDefinitionKey,
+        [property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables);
+
+    private sealed record AcquireJobsRequest(
+        [property: JsonPropertyName("topic")] string Topic,
+        [property: JsonPropertyName("lockDuration")] string LockDuration,
+        [property: JsonPropertyName("numberOfTasks")] int NumberOfTasks,
+        [property: JsonPropertyName("workerId")] string WorkerId);
+
+    private sealed record CompleteJobRequest(
+        [property: JsonPropertyName("workerId")] string WorkerId,
+        [property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables);
+
+    private sealed record Variable(
+        [property: JsonPropertyName("name")] string Name,
+        [property: JsonPropertyName("type")] string Type,
+        [property: JsonPropertyName("value")] string Value);
+
+    private sealed record ProcessInstance([property: JsonPropertyName("id")] string Id);
+
+    private sealed record AcquiredJob(
+        [property: JsonPropertyName("id")] string Id,
+        [property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables)
+    {
+        /// <summary>The registration id this job carries as a process variable.</summary>
+        public string RegistrationId() =>
+            Variables.SingleOrDefault(v => v.Name == "registrationId")?.Value
+            ?? throw new InvalidOperationException($"OpenZaakAanmaken job {Id} carries no registrationId variable.");
     }
 }

services/domain/Big.Infrastructure/InMemoryRegistrationStore.cs

@@ -15,13 +15,11 @@ public sealed class InMemoryRegistrationStore : IRegistrationStore
 
     public Task SaveAsync(Registration registration, CancellationToken ct = default)
     {
-        // STUB (red).
+        ArgumentNullException.ThrowIfNull(registration);
+        _byId[registration.Id] = registration;
         return Task.CompletedTask;
     }
 
     public Task<Registration?> GetAsync(RegistrationId id, CancellationToken ct = default)
-    {
+        => Task.FromResult(_byId.GetValueOrDefault(id));
-        // STUB (red).
-        return Task.FromResult<Registration?>(null);
-    }
 }

services/domain/Big.Infrastructure/OpenZaakJobProcessor.cs

@@ -15,12 +15,25 @@ public sealed class OpenZaakJobProcessor(
     OpenZaakWorker worker,
     ILogger<OpenZaakJobProcessor> logger)
 {
-    /// <summary>Process up to <paramref name="maxJobs"/> jobs. Returns the number completed.</summary>
+    /// <summary>Acquire and process up to <paramref name="maxJobs"/> jobs. Returns the number acquired.</summary>
     public async Task<int> PumpOnceAsync(int maxJobs, CancellationToken ct = default)
     {
-        // STUB (red).
+        var jobs = await client.AcquireOpenZaakJobsAsync(maxJobs, ct);
-        await Task.CompletedTask;
+
-        _ = logger;
+        foreach (var job in jobs)
-        return 0;
+        {
+            try
+            {
+                var zaakUrl = await worker.HandleAsync(job, ct);
+                await client.CompleteOpenZaakJobAsync(job.JobId, zaakUrl, ct);
+            }
+            catch (Exception ex)
+            {
+                // Leave the job un-completed: its lock expires and Flowable redelivers it (§8.6).
+                logger.LogError(ex, "OpenZaakAanmaken job {JobId} failed; leaving it for redelivery.", job.JobId);
+            }
+        }
+
+        return jobs.Count;
     }
 }

services/domain/Big.Infrastructure/OpenZaakJobPump.cs

@@ -0,0 +1,48 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Big.Infrastructure;
+
+/// <summary>
+/// The hosted polling loop of the external-task job worker (ADR-0009): on an interval it resolves a
+/// scoped <see cref="OpenZaakJobProcessor"/> and asks it to drain the parked <c>OpenZaakAanmaken</c>
+/// jobs. A deliberately thin shell — all acquire/process/complete logic lives in the processor, which
+/// is unit-tested; this class only owns the timer, the per-tick scope, and loop resilience.
+/// </summary>
+public sealed class OpenZaakJobPump(
+    IServiceScopeFactory scopeFactory,
+    FlowableOptions options,
+    ILogger<OpenZaakJobPump> logger) : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                using var scope = scopeFactory.CreateScope();
+                var processor = scope.ServiceProvider.GetRequiredService<OpenZaakJobProcessor>();
+                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, "OpenZaakAanmaken job poll failed; retrying after the poll interval.");
+            }
+
+            try
+            {
+                await Task.Delay(options.PollInterval, stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+        }
+    }
+}