register-referentie/services/domain/Big.Infrastructure/FlowableWorkflowClient.cs

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;

namespace Big.Infrastructure;

/// <summary>
/// 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
{
    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)
    {
        var request = new StartProcessRequest(
            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 async Task<IReadOnlyList<OpenZaakJob>> AcquireOpenZaakJobsAsync(int maxJobs, CancellationToken ct = default)
    {
        var request = new AcquireJobsRequest(Topic, options.LockDuration, maxJobs, options.WorkerId);

        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 async Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default)
    {
        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<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.");
    }
}