test(domain): Workflow Client, ACL client, store and job processor (refs #6)

Failing infrastructure unit tests (stub HttpMessageHandler, fakes):
- FlowableWorkflowClient starts a process with the registrationId variable and
  returns the instance id; acquires OpenZaakAanmaken jobs (topic/workerId/lock)
  and parses their registrationId; completes a job with the zaakUrl variable —
  request URIs match flowable-rest's service/ and external-job-api/ paths.
- AclHttpClient POSTs the bsn to the ACL and returns the zaak URL.
- InMemoryRegistrationStore saves/reads/upserts by id.
- OpenZaakJobProcessor acquires, opens a zaak, completes the job; leaves a failing
  job uncompleted for redelivery; polls harmlessly when idle.

Adapters are stubs so the tests compile and fail on their assertions; the green
commit implements them against the REST contract verified on a live Flowable.

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

services/domain/Big.Tests/AclHttpClientTests.cs

@@ -0,0 +1,43 @@
+using System.Net;
+using Big.Infrastructure;
+
+namespace Big.Tests;
+
+public class AclHttpClientTests
+{
+    private static AclHttpClient Client(StubHandler handler) => new(
+        new HttpClient(handler), new AclOptions { BaseUrl = new("http://acl/") });
+
+    [Fact]
+    public async Task Opens_a_zaak_by_posting_the_bsn_and_returns_the_zaak_url()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.OK,
+            """{"zaakUrl":"http://openzaak/zaken/api/v1/zaken/abc"}"""));
+
+        var url = await client.OpenZaakAsync("123456782");
+
+        Assert.Equal("http://openzaak/zaken/api/v1/zaken/abc", url.ToString());
+        Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
+        Assert.Equal("http://acl/zaken", capture.Seen.RequestUri!.ToString());
+        Assert.Contains("\"bsn\":\"123456782\"", capture.Body);
+    }
+
+    [Fact]
+    public async Task Throws_when_the_acl_rejects_the_request()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.BadGateway));
+
+        await Assert.ThrowsAsync<HttpRequestException>(() => client.OpenZaakAsync("123456782"));
+    }
+
+    [Fact]
+    public async Task Throws_when_the_acl_returns_an_empty_body()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.OK, "null"));
+
+        await Assert.ThrowsAsync<InvalidOperationException>(() => client.OpenZaakAsync("123456782"));
+    }
+}

services/domain/Big.Tests/Big.Tests.csproj

@@ -21,6 +21,7 @@
   <ItemGroup>
     <ProjectReference Include="..\Big.Domain\Big.Domain.csproj" />
     <ProjectReference Include="..\Big.Application\Big.Application.csproj" />
+    <ProjectReference Include="..\Big.Infrastructure\Big.Infrastructure.csproj" />
   </ItemGroup>
 
 </Project>

services/domain/Big.Tests/Fakes.cs

@@ -4,8 +4,9 @@ using Big.Domain;
 namespace Big.Tests;
 
 /// <summary>An in-memory <see cref="IRegistrationStore"/> for the application-layer tests. Upserts
-/// keyed on the registration id, mirroring the production in-memory store.</summary>
-internal sealed class InMemoryRegistrationStore : IRegistrationStore
+/// keyed on the registration id, mirroring the production store (kept distinct from the production
+/// <c>InMemoryRegistrationStore</c> so tests can seed and inspect save counts).</summary>
+internal sealed class FakeRegistrationStore : IRegistrationStore
 {
     private readonly Dictionary<RegistrationId, Registration> _byId = [];
 

services/domain/Big.Tests/FlowableWorkflowClientTests.cs

@@ -0,0 +1,93 @@
+using System.Net;
+using Big.Domain;
+using Big.Infrastructure;
+
+namespace Big.Tests;
+
+public class FlowableWorkflowClientTests
+{
+    private static readonly Uri Base = new("http://flowable/flowable-rest/");
+
+    private static FlowableWorkflowClient Client(StubHandler handler) => new(
+        new HttpClient(handler),
+        new FlowableOptions { BaseUrl = Base, Username = "rest-admin", Password = "test", WorkerId = "worker-x" });
+
+    [Fact]
+    public async Task Start_posts_the_registration_id_variable_and_returns_the_instance_id()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.Created, """{"id":"pi-1"}"""));
+        var rid = RegistrationId.New();
+
+        var pid = await client.StartRegistrationProcessAsync(rid);
+
+        Assert.Equal("pi-1", pid);
+        Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
+        Assert.Equal("http://flowable/flowable-rest/service/runtime/process-instances",
+            capture.Seen.RequestUri!.ToString());
+        Assert.Equal("Basic", capture.Seen.Headers.Authorization!.Scheme);
+        Assert.Contains("\"processDefinitionKey\":\"registratie\"", capture.Body);
+        Assert.Contains("\"name\":\"registrationId\"", capture.Body);
+        Assert.Contains($"\"value\":\"{rid}\"", capture.Body);
+    }
+
+    [Fact]
+    public async Task Acquire_posts_the_topic_and_parses_jobs_with_their_registration_id()
+    {
+        var rid = RegistrationId.New();
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.OK,
+            $$"""[{"id":"job-1","variables":[{"name":"registrationId","type":"string","value":"{{rid}}"}]}]"""));
+
+        var jobs = await client.AcquireOpenZaakJobsAsync(3);
+
+        var job = Assert.Single(jobs);
+        Assert.Equal("job-1", job.JobId);
+        Assert.Equal(rid, job.RegistrationId);
+        Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs",
+            capture.Seen!.RequestUri!.ToString());
+        Assert.Contains("\"topic\":\"OpenZaakAanmaken\"", capture.Body);
+        Assert.Contains("\"numberOfTasks\":3", capture.Body);
+        Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
+        Assert.Contains("\"lockDuration\":\"PT5M\"", capture.Body);
+    }
+
+    [Fact]
+    public async Task Acquire_returns_empty_when_flowable_has_no_parked_jobs()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.OK, "[]"));
+
+        var jobs = await client.AcquireOpenZaakJobsAsync(1);
+
+        Assert.NotNull(capture.Seen);
+        Assert.Empty(jobs);
+    }
+
+    [Fact]
+    public async Task Complete_posts_the_zaak_url_variable_to_the_job_complete_endpoint()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.NoContent));
+
+        await client.CompleteOpenZaakJobAsync("job-1", new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
+
+        Assert.NotNull(capture.Seen);
+        Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
+        Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs/job-1/complete",
+            capture.Seen.RequestUri!.ToString());
+        Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
+        Assert.Contains("\"name\":\"zaakUrl\"", capture.Body);
+        Assert.Contains("\"value\":\"http://openzaak/zaken/api/v1/zaken/abc\"", capture.Body);
+    }
+
+    [Fact]
+    public async Task Start_throws_when_flowable_rejects_the_request()
+    {
+        var capture = new RequestCapture();
+        var client = Client(capture.Responds(HttpStatusCode.InternalServerError));
+
+        await Assert.ThrowsAsync<HttpRequestException>(
+            () => client.StartRegistrationProcessAsync(RegistrationId.New()));
+    }
+}

services/domain/Big.Tests/HttpStubs.cs

@@ -0,0 +1,30 @@
+using System.Net;
+
+namespace Big.Tests;
+
+/// <summary>A test double for <see cref="HttpMessageHandler"/> that records the last request and
+/// returns a scripted response — the same approach the ACL gateway tests use.</summary>
+internal sealed class StubHandler(Func<HttpRequestMessage, Task<HttpResponseMessage>> onSend) : HttpMessageHandler
+{
+    protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken ct)
+        => onSend(request);
+}
+
+/// <summary>Captures the request a client sent, including the (buffered) body.</summary>
+internal sealed class RequestCapture
+{
+    public HttpRequestMessage? Seen { get; private set; }
+    public string? Body { get; private set; }
+
+    /// <summary>A handler that records the request, then replies with <paramref name="status"/> and
+    /// <paramref name="json"/>.</summary>
+    public StubHandler Responds(HttpStatusCode status, string? json = null) => new(async req =>
+    {
+        Seen = req;
+        Body = req.Content is null ? null : await req.Content.ReadAsStringAsync();
+        var response = new HttpResponseMessage(status);
+        if (json is not null)
+            response.Content = new StringContent(json, System.Text.Encoding.UTF8, "application/json");
+        return response;
+    });
+}

services/domain/Big.Tests/InMemoryRegistrationStoreTests.cs

@@ -0,0 +1,36 @@
+using Big.Domain;
+using Big.Infrastructure;
+
+namespace Big.Tests;
+
+public class InMemoryRegistrationStoreTests
+{
+    [Fact]
+    public async Task Saves_and_reads_back_a_registration_by_id()
+    {
+        var store = new InMemoryRegistrationStore();
+        var registration = Registration.Submit("123456782");
+
+        await store.SaveAsync(registration);
+
+        var loaded = await store.GetAsync(registration.Id);
+        Assert.NotNull(loaded);
+        Assert.Equal(registration.Id, loaded.Id);
+        Assert.Null(await store.GetAsync(RegistrationId.New()));
+    }
+
+    [Fact]
+    public async Task Saving_the_same_id_upserts()
+    {
+        var store = new InMemoryRegistrationStore();
+        var registration = Registration.Submit("123456782");
+        await store.SaveAsync(registration);
+
+        registration.AttachZaak(new Uri("http://openzaak/zaken/api/v1/zaken/abc"));
+        await store.SaveAsync(registration);
+
+        var loaded = await store.GetAsync(registration.Id);
+        Assert.NotNull(loaded);
+        Assert.Equal("http://openzaak/zaken/api/v1/zaken/abc", loaded.ZaakUrl!.ToString());
+    }
+}

services/domain/Big.Tests/OpenZaakJobProcessorTests.cs

@@ -0,0 +1,77 @@
+using Big.Application;
+using Big.Domain;
+using Big.Infrastructure;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Big.Tests;
+
+public class OpenZaakJobProcessorTests
+{
+    /// <summary>A fake external-worker client: scripts the jobs to acquire and records completions.</summary>
+    private sealed class FakeExternalWorkerClient(params OpenZaakJob[] jobs) : IExternalWorkerClient
+    {
+        public int AcquireCount { get; private set; }
+        public List<(string JobId, Uri ZaakUrl)> Completed { get; } = [];
+
+        public Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
+        {
+            AcquireCount++;
+            return Task.FromResult<IReadOnlyList<OpenZaakJob>>(jobs.Take(maxJobs).ToList());
+        }
+
+        public Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
+        {
+            Completed.Add((jobId, zaakUrl));
+            return Task.CompletedTask;
+        }
+    }
+
+    private static OpenZaakJobProcessor Processor(IExternalWorkerClient client, IRegistrationStore store, FakeAclClient acl)
+        => new(client, new OpenZaakWorker(store, acl), NullLogger<OpenZaakJobProcessor>.Instance);
+
+    [Fact]
+    public async Task Acquires_a_job_opens_the_zaak_and_completes_it()
+    {
+        var registration = Registration.Submit("123456782");
+        var store = new FakeRegistrationStore();
+        store.Seed(registration);
+        var client = new FakeExternalWorkerClient(new OpenZaakJob("job-1", registration.Id));
+        var acl = new FakeAclClient();
+
+        var acquired = await Processor(client, store, acl).PumpOnceAsync(5);
+
+        Assert.Equal(1, acquired);
+        var completed = Assert.Single(client.Completed);
+        Assert.Equal("job-1", completed.JobId);
+        Assert.Equal(FakeAclClient.DefaultZaakUrl, completed.ZaakUrl);
+        Assert.Equal(FakeAclClient.DefaultZaakUrl, (await store.GetAsync(registration.Id))!.ZaakUrl);
+    }
+
+    [Fact]
+    public async Task A_failing_job_is_left_uncompleted_for_flowable_to_redeliver()
+    {
+        var store = new FakeRegistrationStore();
+        var acl = new FakeAclClient();
+        // The job correlates to a registration that is not in the store, so the worker throws.
+        var client = new FakeExternalWorkerClient(new OpenZaakJob("job-1", RegistrationId.New()));
+
+        var acquired = await Processor(client, store, acl).PumpOnceAsync(5);
+
+        Assert.Equal(1, acquired);
+        Assert.Empty(client.Completed);
+    }
+
+    [Fact]
+    public async Task Does_nothing_but_poll_when_there_are_no_jobs()
+    {
+        var store = new FakeRegistrationStore();
+        var acl = new FakeAclClient();
+        var client = new FakeExternalWorkerClient();
+
+        var acquired = await Processor(client, store, acl).PumpOnceAsync(5);
+
+        Assert.Equal(0, acquired);
+        Assert.Equal(1, client.AcquireCount);
+        Assert.Empty(client.Completed);
+    }
+}

services/domain/Big.Tests/OpenZaakWorkerTests.cs

@@ -16,7 +16,7 @@ public class OpenZaakWorkerTests
     public async Task Handling_a_job_opens_a_zaak_via_the_acl_and_attaches_it()
     {
         var registration = Submitted();
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         store.Seed(registration);
         var acl = new FakeAclClient();
         var worker = new OpenZaakWorker(store, acl);
@@ -32,7 +32,7 @@ public class OpenZaakWorkerTests
     [Fact]
     public async Task Handling_a_job_for_an_unknown_registration_throws_and_opens_no_zaak()
     {
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         var acl = new FakeAclClient();
         var worker = new OpenZaakWorker(store, acl);
 
@@ -45,7 +45,7 @@ public class OpenZaakWorkerTests
     public async Task Handling_a_redelivered_job_is_idempotent_and_does_not_open_a_second_zaak()
     {
         var registration = Submitted();
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         store.Seed(registration);
         var acl = new FakeAclClient();
         var worker = new OpenZaakWorker(store, acl);

services/domain/Big.Tests/SubmitRegistrationTests.cs

@@ -8,7 +8,7 @@ public class SubmitRegistrationTests
     [Fact]
     public async Task Submitting_persists_an_ingediend_registration_and_starts_the_process()
     {
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         var workflow = new FakeWorkflowClient(processInstanceId: "proc-42");
         var handler = new SubmitRegistration(store, workflow);
 
@@ -28,7 +28,7 @@ public class SubmitRegistrationTests
     {
         // The worker correlates the OpenZaakAanmaken job back to the aggregate by id, so the
         // aggregate must already be persisted when the process starts (ADR-0009).
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         Registration? visibleAtStart = null;
         var workflow = new FakeWorkflowClient(onStart: id => visibleAtStart = store.GetAsync(id).Result);
         var handler = new SubmitRegistration(store, workflow);
@@ -41,7 +41,7 @@ public class SubmitRegistrationTests
     [Fact]
     public async Task Submitting_without_a_bsn_is_rejected_and_starts_no_process()
     {
-        var store = new InMemoryRegistrationStore();
+        var store = new FakeRegistrationStore();
         var workflow = new FakeWorkflowClient();
         var handler = new SubmitRegistration(store, workflow);