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, IUserTaskClient, IBeoordelingEscalatieClient
{
    private const string Topic = "OpenZaakAanmaken";
    private const string EscalatieTopic = "BeoordelingEscaleren";
    private const string ProcessDefinitionKey = "registratie";
    private const string BeoordelenTaskKey = "Beoordelen";
    private const string BehandelaarGroup = "behandelaar";
    private const string TeamleadGroup = "teamlead";
    private const string RegistrationIdVariable = "registrationId";
    private const string ZaakUrlVariable = "zaakUrl";
    private const string BesluitVariable = "besluit";
    private const string IngetrokkenMessage = "RegistratieIngetrokken";

    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();
    }

    public async Task<IReadOnlyList<BeoordelingTask>> GetOpenBeoordelingenAsync(CancellationToken ct = default)
    {
        var request = new TaskQueryRequest(ProcessDefinitionKey, BeoordelenTaskKey, IncludeProcessVariables: true);

        var page = await PostAsync<TaskQueryRequest, TaskQueryResult>(
            "service/query/tasks", request, ct);

        var tasks = page?.Data ?? [];
        return [.. tasks.Select(t => new BeoordelingTask(t.Id, RegistrationId.Parse(t.RegistrationId())))];
    }

    public async Task ClaimAsync(string taskId, string behandelaar, CancellationToken ct = default)
    {
        using var response = await SendAsync(
            $"service/runtime/tasks/{taskId}", new ClaimTaskRequest("claim", behandelaar), ct);
        response.EnsureSuccessStatusCode();
    }

    public async Task CompleteBeoordelingAsync(string taskId, BeoordelingsBesluit besluit, CancellationToken ct = default)
    {
        var request = new CompleteTaskRequest(
            "complete",
            [new Variable(BesluitVariable, "string", besluit.ToString().ToLowerInvariant())]);

        using var response = await SendAsync($"service/runtime/tasks/{taskId}", request, ct);
        response.EnsureSuccessStatusCode();
    }

    public async Task WithdrawProcessAsync(string processInstanceId, CancellationToken ct = default)
    {
        // Correlate the withdrawal message to the instance: find the execution subscribed to it (the
        // interrupting message event's own execution — NOT the user task's), then deliver
        // messageEventReceived to that execution so the process ends (ADR-0014). If nothing is
        // subscribed (the process is not parked at Beoordelen) this is a best-effort no-op.
        var subscribed = await GetAsync<ExecutionQueryResult>(
            $"service/runtime/executions?messageEventSubscriptionName={IngetrokkenMessage}&processInstanceId={processInstanceId}",
            ct);

        var execution = subscribed?.Data?.FirstOrDefault();
        if (execution is null)
            return;

        var request = new MessageEventRequest("messageEventReceived", IngetrokkenMessage);
        using var response = await SendAsync(
            $"service/runtime/executions/{execution.Id}", request, ct, HttpMethod.Put);
        response.EnsureSuccessStatusCode();
    }

    public async Task<IReadOnlyList<EscalatieJob>> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default)
    {
        var request = new AcquireJobsRequest(EscalatieTopic, options.LockDuration, maxJobs, options.WorkerId);

        var jobs = await PostAsync<AcquireJobsRequest, List<AcquiredEscalatieJob>>(
            "external-job-api/acquire/jobs", request, ct) ?? [];

        return [.. jobs.Select(job => new EscalatieJob(job.Id, job.ProcessInstanceId))];
    }

    public async Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default)
    {
        // The escalation token runs in parallel to the still-open Beoordelen task (non-interrupting
        // boundary timer); find that task in this instance so we can move it to the teamlead. If the
        // behandelaar completed it just before the timer fired there is nothing to reassign — a
        // best-effort no-op (the timer/completion race, cf. §8.6).
        var query = new TaskByInstanceQueryRequest(processInstanceId, BeoordelenTaskKey);
        var page = await PostAsync<TaskByInstanceQueryRequest, TaskQueryResult>(
            "service/query/tasks", query, ct);

        var task = page?.Data?.FirstOrDefault();
        if (task is null)
            return;

        // Add teamlead, then drop behandelaar: the task now belongs to the teamlead group.
        using (var added = await SendAsync(
            $"service/runtime/tasks/{task.Id}/identitylinks",
            new IdentityLinkRequest(TeamleadGroup, "candidate"), ct))
            added.EnsureSuccessStatusCode();

        await DeleteAsync(
            $"service/runtime/tasks/{task.Id}/identitylinks/groups/{BehandelaarGroup}/candidate", ct);
    }

    public async Task CompleteBeoordelingEscalatieJobAsync(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));
        message.Headers.Authorization = new AuthenticationHeaderValue("Basic", BasicCredentials());
        using var response = await http.SendAsync(message, ct);
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<TResponse>(ct);
    }

    private async Task DeleteAsync(string path, CancellationToken ct)
    {
        var message = new HttpRequestMessage(HttpMethod.Delete, new Uri(options.BaseUrl, path));
        message.Headers.Authorization = new AuthenticationHeaderValue("Basic", BasicCredentials());
        using var response = await http.SendAsync(message, 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, HttpMethod? method = null)
    {
        var message = new HttpRequestMessage(method ?? 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 TaskQueryRequest(
        [property: JsonPropertyName("processDefinitionKey")] string ProcessDefinitionKey,
        [property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey,
        [property: JsonPropertyName("includeProcessVariables")] bool IncludeProcessVariables);

    private sealed record TaskByInstanceQueryRequest(
        [property: JsonPropertyName("processInstanceId")] string ProcessInstanceId,
        [property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey);

    private sealed record IdentityLinkRequest(
        [property: JsonPropertyName("group")] string Group,
        [property: JsonPropertyName("type")] string Type);

    private sealed record ClaimTaskRequest(
        [property: JsonPropertyName("action")] string Action,
        [property: JsonPropertyName("assignee")] string Assignee);

    private sealed record CompleteTaskRequest(
        [property: JsonPropertyName("action")] string Action,
        [property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables);

    private sealed record MessageEventRequest(
        [property: JsonPropertyName("action")] string Action,
        [property: JsonPropertyName("messageName")] string MessageName);

    private sealed record TaskQueryResult(
        [property: JsonPropertyName("data")] IReadOnlyList<UserTaskDto>? Data);

    private sealed record UserTaskDto(
        [property: JsonPropertyName("id")] string Id,
        // Flowable's task-query returns the (included) process variables under "variables", not
        // "processVariables"; the request opts in via includeProcessVariables.
        [property: JsonPropertyName("variables")] IReadOnlyList<Variable>? Variables)
    {
        /// <summary>The registration id this task carries as a process variable.</summary>
        public string RegistrationId() =>
            Variables?.SingleOrDefault(v => v.Name == "registrationId")?.Value
            ?? throw new InvalidOperationException($"Beoordelen task {Id} carries no registrationId variable.");
    }

    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 ExecutionQueryResult(
        [property: JsonPropertyName("data")] IReadOnlyList<ExecutionDto>? Data);

    private sealed record ExecutionDto([property: JsonPropertyName("id")] string Id);

    private sealed record AcquiredEscalatieJob(
        [property: JsonPropertyName("id")] string Id,
        [property: JsonPropertyName("processInstanceId")] string ProcessInstanceId);

    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() =>
            // Stryker disable once Linq : equivalent — Single and SingleOrDefault both throw on a job
            // with no registrationId variable (the only untested branch), so the mutant is indistinguishable.
            Variables.SingleOrDefault(v => v.Name == "registrationId")?.Value
            ?? throw new InvalidOperationException($"OpenZaakAanmaken job {Id} carries no registrationId variable.");
    }
}