From 0d34d60797226942f062400a4143cc3c655ef104 Mon Sep 17 00:00:00 2001 From: Niek Otten Date: Tue, 30 Jun 2026 17:10:21 +0200 Subject: [PATCH] feat(domain): implement the Flowable Workflow Client and ACL client (refs #6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../Big.Infrastructure/AclHttpClient.cs | 17 +++- .../FlowableWorkflowClient.cs | 95 +++++++++++++++++-- .../InMemoryRegistrationStore.cs | 8 +- .../OpenZaakJobProcessor.cs | 23 ++++- .../Big.Infrastructure/OpenZaakJobPump.cs | 48 ++++++++++ 5 files changed, 169 insertions(+), 22 deletions(-) create mode 100644 services/domain/Big.Infrastructure/OpenZaakJobPump.cs diff --git a/services/domain/Big.Infrastructure/AclHttpClient.cs b/services/domain/Big.Infrastructure/AclHttpClient.cs index 224cf41..293be9c 100644 --- a/services/domain/Big.Infrastructure/AclHttpClient.cs +++ b/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; /// public sealed class AclHttpClient(HttpClient http, AclOptions options) : IAclClient { - public Task OpenZaakAsync(string bsn, CancellationToken ct = default) + public async Task OpenZaakAsync(string bsn, CancellationToken ct = default) { - // STUB (red). - return Task.FromResult(new Uri("about:blank")); + using var response = await http.PostAsJsonAsync( + new Uri(options.BaseUrl, "zaken"), new OpenZaakRequest(bsn), ct); + response.EnsureSuccessStatusCode(); + + var opened = await response.Content.ReadFromJsonAsync(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); } diff --git a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs b/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs index 016a349..ee153ac 100644 --- a/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs +++ b/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 OpenZaakAanmaken external-worker jobs over Flowable's REST API /// (start: service/runtime/process-instances; acquire/complete: external-job-api/…). +/// 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 { - public Task 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 StartRegistrationProcessAsync(RegistrationId registrationId, CancellationToken ct = default) { - // STUB (red). - return Task.FromResult(""); + var request = new StartProcessRequest( + ProcessDefinitionKey, + [new Variable(RegistrationIdVariable, "string", registrationId.ToString())]); + + var created = await PostAsync( + "service/runtime/process-instances", request, ct) + ?? throw new InvalidOperationException("Flowable returned an empty process-instance response."); + return created.Id; } - public Task> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default) + public async Task> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default) { - // STUB (red). - return Task.FromResult>([]); + var request = new AcquireJobsRequest(Topic, options.LockDuration, maxJobs, options.WorkerId); + + var jobs = await PostAsync>( + "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). - return Task.CompletedTask; + var request = new CompleteJobRequest( + 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 PostAsync(string path, TRequest body, CancellationToken ct) + { + using var response = await SendAsync(path, body, ct); + response.EnsureSuccessStatusCode(); + return await response.Content.ReadFromJsonAsync(ct); + } + + private Task SendAsync(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 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 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 Variables) + { + /// The registration id this job carries as a process variable. + public string RegistrationId() => + Variables.SingleOrDefault(v => v.Name == "registrationId")?.Value + ?? throw new InvalidOperationException($"OpenZaakAanmaken job {Id} carries no registrationId variable."); } } diff --git a/services/domain/Big.Infrastructure/InMemoryRegistrationStore.cs b/services/domain/Big.Infrastructure/InMemoryRegistrationStore.cs index 9e83fa9..0acfe94 100644 --- a/services/domain/Big.Infrastructure/InMemoryRegistrationStore.cs +++ b/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 GetAsync(RegistrationId id, CancellationToken ct = default) - { - // STUB (red). - return Task.FromResult(null); - } + => Task.FromResult(_byId.GetValueOrDefault(id)); } diff --git a/services/domain/Big.Infrastructure/OpenZaakJobProcessor.cs b/services/domain/Big.Infrastructure/OpenZaakJobProcessor.cs index 32acf16..cea5576 100644 --- a/services/domain/Big.Infrastructure/OpenZaakJobProcessor.cs +++ b/services/domain/Big.Infrastructure/OpenZaakJobProcessor.cs @@ -15,12 +15,25 @@ public sealed class OpenZaakJobProcessor( OpenZaakWorker worker, ILogger logger) { - /// Process up to jobs. Returns the number completed. + /// Acquire and process up to jobs. Returns the number acquired. public async Task PumpOnceAsync(int maxJobs, CancellationToken ct = default) { - // STUB (red). - await Task.CompletedTask; - _ = logger; - return 0; + var jobs = await client.AcquireOpenZaakJobsAsync(maxJobs, ct); + + foreach (var job in jobs) + { + 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; } } diff --git a/services/domain/Big.Infrastructure/OpenZaakJobPump.cs b/services/domain/Big.Infrastructure/OpenZaakJobPump.cs new file mode 100644 index 0000000..44bcdca --- /dev/null +++ b/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; + +/// +/// The hosted polling loop of the external-task job worker (ADR-0009): on an interval it resolves a +/// scoped and asks it to drain the parked OpenZaakAanmaken +/// 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. +/// +public sealed class OpenZaakJobPump( + 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, "OpenZaakAanmaken job poll failed; retrying after the poll interval."); + } + + try + { + await Task.Delay(options.PollInterval, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + } + } +}