feat(workflow): beoordeling escalation to teamlead after 14 days (S-14, closes #15) #99
77
docs/architecture/adr-0015-beoordeling-escalation.md
Normal file
77
docs/architecture/adr-0015-beoordeling-escalation.md
Normal file
@@ -0,0 +1,77 @@
|
|||||||
|
# ADR-0015: Beoordeling escalation reassigns via an external-worker task
|
||||||
|
|
||||||
|
- **Status:** Accepted
|
||||||
|
- **Date:** 2026-07-17
|
||||||
|
- **Deciders:** Respellion engineering
|
||||||
|
- **Relates to:** S-14 (#15); proposal #98. Builds on ADR-0009 (external-task worker / Workflow
|
||||||
|
Client), ADR-0013 (behandel-portal wiring, the `Beoordelen` user task), ADR-0014 (the boundary-event
|
||||||
|
pattern on `Beoordelen`).
|
||||||
|
|
||||||
|
## Context
|
||||||
|
|
||||||
|
S-14 escalates a beoordeling that a behandelaar does not pick up in time: after 14 days the case must
|
||||||
|
move to the `teamlead` role (PRD §5, flow 5). The `Beoordelen` user task already exists, claimable by
|
||||||
|
the `behandelaar` candidate group; the teamlead role is seeded in the medewerker realm.
|
||||||
|
|
||||||
|
Two forces shape this.
|
||||||
|
|
||||||
|
1. **The task must stay open.** Escalation changes *who may claim* an unclaimed beoordeling, not the
|
||||||
|
work itself — so the timer must be **non-interrupting**: the `Beoordelen` task keeps running while
|
||||||
|
escalation happens alongside it.
|
||||||
|
2. **Reassigning an open task's candidate group needs code.** Flowable cannot rewrite the candidate
|
||||||
|
groups of an already-open user task from BPMN XML alone — that requires either a Java delegate/listener
|
||||||
|
embedded in the engine, or an out-of-process actor driving the REST API. The repository has held a
|
||||||
|
"stock Flowable image, no custom jars; the Workflow Client is the only code that talks to Flowable
|
||||||
|
(§8.2)" posture since ADR-0009.
|
||||||
|
|
||||||
|
## Decision
|
||||||
|
|
||||||
|
**A non-interrupting `P14D` boundary timer on `Beoordelen` fires an external-worker task
|
||||||
|
(`BeoordelingEscaleren`); the Workflow Client reassigns the still-open `Beoordelen` task from the
|
||||||
|
behandelaar group to teamlead.**
|
||||||
|
|
||||||
|
- **Modelled in BPMN, driven by an external worker.** The timer routes a parallel token to an
|
||||||
|
`external-worker` service task on the `BeoordelingEscaleren` topic, ending at a dedicated "Beoordeling
|
||||||
|
geëscaleerd" end event. The model owns *when* escalation happens; the Workflow Client — the only code
|
||||||
|
that talks to Flowable (§8.2) — owns *how* the reassignment is applied, exactly as `OpenZaakAanmaken`
|
||||||
|
delegates the ZGW call (ADR-0009). No custom code runs inside Flowable.
|
||||||
|
- **Reassignment is a candidate-group swap.** The escalation worker finds the still-open `Beoordelen`
|
||||||
|
task in the escalating instance (task query by `processInstanceId` + `taskDefinitionKey`), adds
|
||||||
|
`teamlead` as a candidate group via the task identity links, then removes `behandelaar`. The task now
|
||||||
|
belongs to the teamlead; its history and variables are untouched.
|
||||||
|
- **Best-effort, mirroring beoordeling and withdrawal.** If the task is no longer open — the behandelaar
|
||||||
|
completed it in the window before the timer fired — the reassignment is a no-op. A failed reassignment
|
||||||
|
leaves the escalation job un-completed so Flowable redelivers it (§8.6), consistent with the
|
||||||
|
`OpenZaakAanmaken` worker.
|
||||||
|
- **Segregated interface.** The escalation methods live on `IBeoordelingEscalatieClient`, separate from
|
||||||
|
the `OpenZaakAanmaken` worker's `IExternalWorkerClient`, so the OpenZaak worker never sees escalation
|
||||||
|
(interface segregation). Both are implemented by the one `FlowableWorkflowClient`.
|
||||||
|
|
||||||
|
## Consequences
|
||||||
|
|
||||||
|
**Positive**
|
||||||
|
|
||||||
|
- The escalation trigger is visible in `registratie.bpmn`; Flowable stays a stock image, and the
|
||||||
|
Workflow Client remains the sole Flowable client (§8.2 upheld, not bent).
|
||||||
|
- Reuses the external-worker mechanics (topic acquire/complete, hosted pump, per-tick scope,
|
||||||
|
redelivery-on-failure) wholesale — the new code is one client capability, one processor, one pump.
|
||||||
|
- Escalation latency is bounded by the worker's poll interval (seconds) — negligible against a 14-day
|
||||||
|
timer.
|
||||||
|
|
||||||
|
**Negative / costs**
|
||||||
|
|
||||||
|
- Escalation is two REST hops (add teamlead, remove behandelaar) rather than one atomic update; between
|
||||||
|
them the task is briefly claimable by both groups. Harmless at these volumes, and the pair is idempotent
|
||||||
|
on redelivery.
|
||||||
|
- The Flowable identity-link and management-job REST shapes are validated live (verify-domain fires the
|
||||||
|
timer early via the management API), not in the Workflow Client's unit tests, which stub the HTTP
|
||||||
|
exchange and assert only the request shape — consistent with ADR-0009 and ADR-0014.
|
||||||
|
|
||||||
|
## Alternatives considered
|
||||||
|
|
||||||
|
- **Flowable timer/task listener (Java delegate).** Reassign in-engine when the timer fires. Rejected:
|
||||||
|
it needs a custom jar in Flowable, breaking the stock-image, REST-only posture and adding a build/deploy
|
||||||
|
surface to the engine for no capability the external-worker route lacks.
|
||||||
|
- **Interrupting timer that re-creates the task for teamlead.** Cancel `Beoordelen` and start a fresh
|
||||||
|
teamlead task. Rejected: it loses the task's identity/history and complicates correlation, where a
|
||||||
|
candidate-group swap on the same task expresses "the same work, now the teamlead's" directly.
|
||||||
@@ -298,3 +298,48 @@ interrupting boundary event ends it → the werkbak drops the case.
|
|||||||
> DigiD submit → trek aanvraag in → ingetrokken is the Playwright happy path
|
> DigiD submit → trek aanvraag in → ingetrokken is the Playwright happy path
|
||||||
> (`tests/e2e/withdrawal.spec.ts`); the owner-scoping + workflow cancellation are covered by the
|
> (`tests/e2e/withdrawal.spec.ts`); the owner-scoping + workflow cancellation are covered by the
|
||||||
> `Een registratie intrekken` acceptance scenarios and the domain live check.
|
> `Een registratie intrekken` acceptance scenarios and the domain live check.
|
||||||
|
|
||||||
|
## S-14 — Beoordeling escalation: 14 days unclaimed → teamlead (#15, ADR-0015)
|
||||||
|
|
||||||
|
A beoordeling a behandelaar does not pick up within 14 days escalates to the teamlead. A
|
||||||
|
non-interrupting boundary timer on the `Beoordelen` task fires a `BeoordelingEscaleren` external task;
|
||||||
|
the domain's escalation worker reassigns the still-open task's candidate group from `behandelaar` to
|
||||||
|
`teamlead`, so it moves from the behandelaar werkbak into the teamlead's. The `Beoordelen` task keeps
|
||||||
|
its identity throughout — only who may claim it changes.
|
||||||
|
|
||||||
|
The timer is 14 days, so the demo fires it early through Flowable's management API (exactly what the
|
||||||
|
verify-domain check automates):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 1. Submit at the self-service portal (http://localhost:8140/, jan-burger / test123). The case
|
||||||
|
# parks at Beoordelen, visible in the behandelaar werkbak (http://localhost:8142/, merel-behandelaar)
|
||||||
|
# but NOT claimed.
|
||||||
|
#
|
||||||
|
# 2. Find the parked instance and its Beoordelen task, then fire the boundary timer early:
|
||||||
|
FL=http://localhost:8090/flowable-rest/service
|
||||||
|
PID=$(curl -s -u rest-admin:test -X POST "$FL/query/tasks" -H 'Content-Type: application/json' \
|
||||||
|
-d '{"processDefinitionKey":"registratie","taskDefinitionKey":"Beoordelen"}' \
|
||||||
|
| python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["processInstanceId"])')
|
||||||
|
TID=$(curl -s -u rest-admin:test -X POST "$FL/query/tasks" -H 'Content-Type: application/json' \
|
||||||
|
-d '{"processDefinitionKey":"registratie","taskDefinitionKey":"Beoordelen"}' \
|
||||||
|
| python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])')
|
||||||
|
TJ=$(curl -s -u rest-admin:test "$FL/management/timer-jobs?processInstanceId=$PID" \
|
||||||
|
| python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])')
|
||||||
|
curl -s -u rest-admin:test -X POST "$FL/management/timer-jobs/$TJ" \
|
||||||
|
-H 'Content-Type: application/json' -d '{"action":"move"}'
|
||||||
|
AJ=$(curl -s -u rest-admin:test "$FL/management/jobs?processInstanceId=$PID" \
|
||||||
|
| python3 -c 'import sys,json;print(json.load(sys.stdin)["data"][0]["id"])')
|
||||||
|
curl -s -u rest-admin:test -X POST "$FL/management/jobs/$AJ" \
|
||||||
|
-H 'Content-Type: application/json' -d '{"action":"execute"}'
|
||||||
|
#
|
||||||
|
# 3. Within a couple of poll cycles the task's candidate group flips to teamlead:
|
||||||
|
curl -s -u rest-admin:test "$FL/runtime/tasks/$TID/identitylinks" # → [{"group":"teamlead","type":"candidate"}]
|
||||||
|
```
|
||||||
|
|
||||||
|
**The path:** BPMN non-interrupting `P14D` boundary timer on `Beoordelen` → `BeoordelingEscaleren`
|
||||||
|
external task → domain escalation worker (`BeoordelingEscalatiePump`) → Workflow Client swaps the task's
|
||||||
|
candidate group behandelaar → teamlead (§8.2).
|
||||||
|
|
||||||
|
> Both branches (escalate after 14 days; no-op when completed in time) are covered by the
|
||||||
|
> `Een beoordeling escaleren` acceptance scenarios and the Workflow Client unit tests; the timer firing
|
||||||
|
> and reassignment are asserted live by the verify-domain check.
|
||||||
|
|||||||
@@ -157,4 +157,82 @@ for _ in $(seq 1 15); do
|
|||||||
done
|
done
|
||||||
[ -n "$gone" ] || { echo "FAIL — Beoordelen task for $reg_id2 still active after withdrawal" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; }
|
[ -n "$gone" ] || { echo "FAIL — Beoordelen task for $reg_id2 still active after withdrawal" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; }
|
||||||
echo "OK — withdrawal cancelled the Beoordelen task; the registratie process ended (ingetrokken)"
|
echo "OK — withdrawal cancelled the Beoordelen task; the registratie process ended (ingetrokken)"
|
||||||
|
|
||||||
|
# ── S-14: escalation. A third registration parks at Beoordelen. We fire its 14-day boundary timer
|
||||||
|
# early via Flowable's management API (the timer job is moved to executable and run), which routes a
|
||||||
|
# parallel token to the BeoordelingEscaleren external task. The domain's escalation worker acquires
|
||||||
|
# it and reassigns the still-open Beoordelen task from the behandelaar group to teamlead (ADR-0015). ─
|
||||||
|
echo ">> submitting a third registration to escalate"
|
||||||
|
loc3="$(docker run --rm --network "$net" curlimages/curl:latest \
|
||||||
|
-fsS -D - -o /dev/null -X POST "http://$dom_ip:8080/registrations" \
|
||||||
|
-H 'Content-Type: application/json' -d '{"bsn":"123456782"}' \
|
||||||
|
| sed -n 's/\r$//; s/^[Ll]ocation: //p' | head -1)"
|
||||||
|
[ -n "$loc3" ] || { echo "FAIL — third POST /registrations returned no Location" >&2; exit 1; }
|
||||||
|
reg_id3="${loc3##*/}"
|
||||||
|
echo ">> third registration $reg_id3"
|
||||||
|
|
||||||
|
# Extracts "<taskId> <processInstanceId>" for a registration from a task-query response on stdin.
|
||||||
|
task_and_pid_for_reg() { REG_ID="$1" python3 -c "import os,sys,json
|
||||||
|
try:
|
||||||
|
d=json.load(sys.stdin)
|
||||||
|
except Exception:
|
||||||
|
d={}
|
||||||
|
rid=os.environ['REG_ID']
|
||||||
|
t=next((t for t in (d.get('data') or [])
|
||||||
|
if any(v.get('name')=='registrationId' and v.get('value')==rid for v in (t.get('variables') or []))), None)
|
||||||
|
print(f\"{t['id']} {t['processInstanceId']}\" if t else '')"; }
|
||||||
|
|
||||||
|
# The candidate groups on a task (space-separated, sorted) from a runtime identitylinks response.
|
||||||
|
candidate_groups() { python3 -c "import sys,json
|
||||||
|
try:
|
||||||
|
links=json.load(sys.stdin)
|
||||||
|
except Exception:
|
||||||
|
links=[]
|
||||||
|
print(' '.join(sorted(l.get('group') or '' for l in links if l.get('type')=='candidate' and l.get('group'))))"; }
|
||||||
|
|
||||||
|
# The first job id in a management jobs/timer-jobs response on stdin.
|
||||||
|
first_job_id() { python3 -c "import sys,json
|
||||||
|
try:
|
||||||
|
d=json.load(sys.stdin)
|
||||||
|
except Exception:
|
||||||
|
d={}
|
||||||
|
print(((d.get('data') or [{}])[0]).get('id',''))"; }
|
||||||
|
|
||||||
|
echo ">> polling Flowable for its Beoordelen task"
|
||||||
|
task_id3=""; pid3=""
|
||||||
|
for _ in $(seq 1 30); do
|
||||||
|
resp="$(flcurl -X POST "$fl_base/query/tasks" -H 'Content-Type: application/json' -d "$query" 2>/dev/null || true)"
|
||||||
|
read -r task_id3 pid3 <<<"$(printf '%s' "$resp" | task_and_pid_for_reg "$reg_id3")"
|
||||||
|
[ -n "$task_id3" ] && break
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
[ -n "$task_id3" ] || { echo "FAIL — no Beoordelen task appeared for registration $reg_id3" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; }
|
||||||
|
echo ">> Beoordelen task $task_id3 (instance $pid3) is waiting for the behandelaar"
|
||||||
|
|
||||||
|
echo ">> asserting the task starts out claimable by the behandelaar group"
|
||||||
|
before="$(flcurl "$fl_base/runtime/tasks/$task_id3/identitylinks" | candidate_groups)"
|
||||||
|
[ "$before" = "behandelaar" ] || { echo "FAIL — expected candidate group 'behandelaar', got '$before'" >&2; exit 1; }
|
||||||
|
|
||||||
|
echo ">> firing the 14-day boundary timer early via the management API"
|
||||||
|
timer_id="$(flcurl "$fl_base/management/timer-jobs?processInstanceId=$pid3" | first_job_id)"
|
||||||
|
[ -n "$timer_id" ] || { echo "FAIL — no timer job found for instance $pid3" >&2; exit 1; }
|
||||||
|
# Move the timer job to an executable async job. Flowable's async executor (running in flowable-rest)
|
||||||
|
# then picks it up and fires the non-interrupting boundary event. It may run the job before we can
|
||||||
|
# look, so executing it explicitly is a best-effort nudge — tolerate the job already being gone.
|
||||||
|
flcurl -X POST "$fl_base/management/timer-jobs/$timer_id" -H 'Content-Type: application/json' -d '{"action":"move"}' >/dev/null
|
||||||
|
async_id="$(flcurl "$fl_base/management/jobs?processInstanceId=$pid3" 2>/dev/null | first_job_id || true)"
|
||||||
|
if [ -n "$async_id" ]; then
|
||||||
|
flcurl -X POST "$fl_base/management/jobs/$async_id" -H 'Content-Type: application/json' -d '{"action":"execute"}' >/dev/null 2>&1 || true
|
||||||
|
fi
|
||||||
|
echo ">> timer fired; the BeoordelingEscaleren token is parked for the domain worker"
|
||||||
|
|
||||||
|
echo ">> polling until the escalation worker reassigns the beoordeling to the teamlead"
|
||||||
|
escalated=""
|
||||||
|
for _ in $(seq 1 30); do
|
||||||
|
groups="$(flcurl "$fl_base/runtime/tasks/$task_id3/identitylinks" 2>/dev/null | candidate_groups || true)"
|
||||||
|
[ "$groups" = "teamlead" ] && { escalated=1; break; }
|
||||||
|
sleep 2
|
||||||
|
done
|
||||||
|
[ -n "$escalated" ] || { echo "FAIL — Beoordelen task not reassigned to teamlead (candidate groups: '$groups')" >&2; docker logs "$dom" 2>&1 | tail -15 >&2; exit 1; }
|
||||||
|
echo "OK — the 14-day timer escalated the still-open Beoordelen task to the teamlead"
|
||||||
exit 0
|
exit 0
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ builder.Services.AddHttpClient<FlowableWorkflowClient>();
|
|||||||
builder.Services.AddTransient<IWorkflowClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
builder.Services.AddTransient<IWorkflowClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
||||||
builder.Services.AddTransient<IExternalWorkerClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
builder.Services.AddTransient<IExternalWorkerClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
||||||
builder.Services.AddTransient<IUserTaskClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
builder.Services.AddTransient<IUserTaskClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
||||||
|
builder.Services.AddTransient<IBeoordelingEscalatieClient>(sp => sp.GetRequiredService<FlowableWorkflowClient>());
|
||||||
builder.Services.AddHttpClient<IAclClient, AclHttpClient>();
|
builder.Services.AddHttpClient<IAclClient, AclHttpClient>();
|
||||||
|
|
||||||
builder.Services.AddScoped<SubmitRegistration>();
|
builder.Services.AddScoped<SubmitRegistration>();
|
||||||
@@ -30,9 +31,13 @@ builder.Services.AddScoped<WithdrawRegistration>();
|
|||||||
builder.Services.AddScoped<Werkbak>();
|
builder.Services.AddScoped<Werkbak>();
|
||||||
builder.Services.AddScoped<OpenZaakWorker>();
|
builder.Services.AddScoped<OpenZaakWorker>();
|
||||||
builder.Services.AddScoped<OpenZaakJobProcessor>();
|
builder.Services.AddScoped<OpenZaakJobProcessor>();
|
||||||
|
builder.Services.AddScoped<BeoordelingEscalatieProcessor>();
|
||||||
|
|
||||||
// The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion.
|
// The hosted external-task job worker polls Flowable and drives OpenZaakAanmaken to completion.
|
||||||
builder.Services.AddHostedService<OpenZaakJobPump>();
|
builder.Services.AddHostedService<OpenZaakJobPump>();
|
||||||
|
// The escalation worker polls the BeoordelingEscaleren jobs the 14-day timer parks and reassigns
|
||||||
|
// each overdue beoordeling to the teamlead (S-14).
|
||||||
|
builder.Services.AddHostedService<BeoordelingEscalatiePump>();
|
||||||
|
|
||||||
var app = builder.Build();
|
var app = builder.Build();
|
||||||
|
|
||||||
|
|||||||
@@ -86,3 +86,10 @@ public interface IRegistrationStore
|
|||||||
/// it) and the registration id it carries as a process variable.
|
/// it) and the registration id it carries as a process variable.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public sealed record OpenZaakJob(string JobId, RegistrationId RegistrationId);
|
public sealed record OpenZaakJob(string JobId, RegistrationId RegistrationId);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// An acquired <c>BeoordelingEscaleren</c> escalation job (S-14): the Flowable job id and the process
|
||||||
|
/// instance whose still-open <c>Beoordelen</c> task must be reassigned from behandelaar to teamlead
|
||||||
|
/// once the 14-day boundary timer fires (ADR-0015).
|
||||||
|
/// </summary>
|
||||||
|
public sealed record EscalatieJob(string JobId, string ProcessInstanceId);
|
||||||
|
|||||||
@@ -0,0 +1,37 @@
|
|||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
|
||||||
|
namespace Big.Infrastructure;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// One poll tick of the beoordeling-escalation worker (S-14, ADR-0015): acquire the parked
|
||||||
|
/// <c>BeoordelingEscaleren</c> jobs — the tokens the 14-day boundary timer on <c>Beoordelen</c> spawns
|
||||||
|
/// — reassign each instance's still-open <c>Beoordelen</c> task to the teamlead, and complete the job.
|
||||||
|
/// A job that fails is logged and left un-completed so Flowable redelivers it (§8.6). Split out from
|
||||||
|
/// the hosted pump so the acquire→reassign→complete logic is unit-testable without a running host.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class BeoordelingEscalatieProcessor(
|
||||||
|
IBeoordelingEscalatieClient client,
|
||||||
|
ILogger<BeoordelingEscalatieProcessor> logger)
|
||||||
|
{
|
||||||
|
/// <summary>Acquire and process up to <paramref name="maxJobs"/> escalations. Returns the number acquired.</summary>
|
||||||
|
public async Task<int> PumpOnceAsync(int maxJobs, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
var jobs = await client.AcquireBeoordelingEscalatieJobsAsync(maxJobs, ct);
|
||||||
|
|
||||||
|
foreach (var job in jobs)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await client.ReassignBeoordelingToTeamleadAsync(job.ProcessInstanceId, ct);
|
||||||
|
await client.CompleteBeoordelingEscalatieJobAsync(job.JobId, ct);
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
// Leave the job un-completed: its lock expires and Flowable redelivers it (§8.6).
|
||||||
|
logger.LogError(ex, "BeoordelingEscaleren job {JobId} failed; leaving it for redelivery.", job.JobId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return jobs.Count;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,49 @@
|
|||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using Microsoft.Extensions.Hosting;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
|
||||||
|
namespace Big.Infrastructure;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The hosted polling loop of the beoordeling-escalation worker (S-14, ADR-0015): on an interval it
|
||||||
|
/// resolves a scoped <see cref="BeoordelingEscalatieProcessor"/> and asks it to drain the parked
|
||||||
|
/// <c>BeoordelingEscaleren</c> jobs. A deliberately thin shell — all acquire/reassign/complete logic
|
||||||
|
/// lives in the processor, which is unit-tested; this class only owns the timer, the per-tick scope,
|
||||||
|
/// and loop resilience. Structurally identical to <see cref="OpenZaakJobPump"/>.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class BeoordelingEscalatiePump(
|
||||||
|
IServiceScopeFactory scopeFactory,
|
||||||
|
FlowableOptions options,
|
||||||
|
ILogger<BeoordelingEscalatiePump> logger) : BackgroundService
|
||||||
|
{
|
||||||
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
|
{
|
||||||
|
while (!stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
using var scope = scopeFactory.CreateScope();
|
||||||
|
var processor = scope.ServiceProvider.GetRequiredService<BeoordelingEscalatieProcessor>();
|
||||||
|
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, "BeoordelingEscaleren job poll failed; retrying after the poll interval.");
|
||||||
|
}
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await Task.Delay(options.PollInterval, stoppingToken);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -15,11 +15,14 @@ namespace Big.Infrastructure;
|
|||||||
/// The REST contract here is the one verified against a live flowable-rest engine (ADR-0009).
|
/// The REST contract here is the one verified against a live flowable-rest engine (ADR-0009).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions options)
|
public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions options)
|
||||||
: IWorkflowClient, IExternalWorkerClient, IUserTaskClient
|
: IWorkflowClient, IExternalWorkerClient, IUserTaskClient, IBeoordelingEscalatieClient
|
||||||
{
|
{
|
||||||
private const string Topic = "OpenZaakAanmaken";
|
private const string Topic = "OpenZaakAanmaken";
|
||||||
|
private const string EscalatieTopic = "BeoordelingEscaleren";
|
||||||
private const string ProcessDefinitionKey = "registratie";
|
private const string ProcessDefinitionKey = "registratie";
|
||||||
private const string BeoordelenTaskKey = "Beoordelen";
|
private const string BeoordelenTaskKey = "Beoordelen";
|
||||||
|
private const string BehandelaarGroup = "behandelaar";
|
||||||
|
private const string TeamleadGroup = "teamlead";
|
||||||
private const string RegistrationIdVariable = "registrationId";
|
private const string RegistrationIdVariable = "registrationId";
|
||||||
private const string ZaakUrlVariable = "zaakUrl";
|
private const string ZaakUrlVariable = "zaakUrl";
|
||||||
private const string BesluitVariable = "besluit";
|
private const string BesluitVariable = "besluit";
|
||||||
@@ -106,6 +109,47 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti
|
|||||||
response.EnsureSuccessStatusCode();
|
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)
|
private async Task<TResponse?> GetAsync<TResponse>(string path, CancellationToken ct)
|
||||||
{
|
{
|
||||||
var message = new HttpRequestMessage(HttpMethod.Get, new Uri(options.BaseUrl, path));
|
var message = new HttpRequestMessage(HttpMethod.Get, new Uri(options.BaseUrl, path));
|
||||||
@@ -115,6 +159,14 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti
|
|||||||
return await response.Content.ReadFromJsonAsync<TResponse>(ct);
|
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)
|
private async Task<TResponse?> PostAsync<TRequest, TResponse>(string path, TRequest body, CancellationToken ct)
|
||||||
{
|
{
|
||||||
using var response = await SendAsync(path, body, ct);
|
using var response = await SendAsync(path, body, ct);
|
||||||
@@ -154,6 +206,14 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti
|
|||||||
[property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey,
|
[property: JsonPropertyName("taskDefinitionKey")] string TaskDefinitionKey,
|
||||||
[property: JsonPropertyName("includeProcessVariables")] bool IncludeProcessVariables);
|
[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(
|
private sealed record ClaimTaskRequest(
|
||||||
[property: JsonPropertyName("action")] string Action,
|
[property: JsonPropertyName("action")] string Action,
|
||||||
[property: JsonPropertyName("assignee")] string Assignee);
|
[property: JsonPropertyName("assignee")] string Assignee);
|
||||||
@@ -193,6 +253,10 @@ public sealed class FlowableWorkflowClient(HttpClient http, FlowableOptions opti
|
|||||||
|
|
||||||
private sealed record ExecutionDto([property: JsonPropertyName("id")] string Id);
|
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(
|
private sealed record AcquiredJob(
|
||||||
[property: JsonPropertyName("id")] string Id,
|
[property: JsonPropertyName("id")] string Id,
|
||||||
[property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables)
|
[property: JsonPropertyName("variables")] IReadOnlyList<Variable> Variables)
|
||||||
|
|||||||
@@ -16,3 +16,23 @@ public interface IExternalWorkerClient
|
|||||||
/// <summary>Complete an acquired job, passing the opened zaak URL back into the process.</summary>
|
/// <summary>Complete an acquired job, passing the opened zaak URL back into the process.</summary>
|
||||||
Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default);
|
Task CompleteOpenZaakJobAsync(string jobId, Uri zaakUrl, CancellationToken ct = default);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The escalation side of the Workflow Client (S-14): the <c>BeoordelingEscaleren</c> external-worker
|
||||||
|
/// jobs parked by the 14-day boundary timer on <c>Beoordelen</c>, and the reassignment they drive.
|
||||||
|
/// Kept separate from <see cref="IExternalWorkerClient"/> (interface segregation) so the OpenZaak
|
||||||
|
/// worker never sees escalation. Implemented by <see cref="FlowableWorkflowClient"/> — the only code
|
||||||
|
/// that talks to Flowable (§8.2, ADR-0015).
|
||||||
|
/// </summary>
|
||||||
|
public interface IBeoordelingEscalatieClient
|
||||||
|
{
|
||||||
|
/// <summary>Acquire and lock up to <paramref name="maxJobs"/> <c>BeoordelingEscaleren</c> jobs.</summary>
|
||||||
|
Task<IReadOnlyList<EscalatieJob>> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default);
|
||||||
|
|
||||||
|
/// <summary>Reassign the still-open <c>Beoordelen</c> task in the given process instance from the
|
||||||
|
/// behandelaar group to teamlead. Best-effort no-op if the task is no longer open.</summary>
|
||||||
|
Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default);
|
||||||
|
|
||||||
|
/// <summary>Complete an acquired escalation job so its token reaches the escalation end event.</summary>
|
||||||
|
Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default);
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,82 @@
|
|||||||
|
using Big.Application;
|
||||||
|
using Big.Infrastructure;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
|
||||||
|
namespace Big.Tests;
|
||||||
|
|
||||||
|
// S-14 (#15): the escalation drain loop. Mirrors OpenZaakJobProcessor — acquire the parked
|
||||||
|
// BeoordelingEscaleren jobs, reassign each instance's Beoordelen task to teamlead, then complete the
|
||||||
|
// job. A job whose reassignment fails is logged and left un-completed for Flowable to redeliver (§8.6).
|
||||||
|
public class BeoordelingEscalatieProcessorTests
|
||||||
|
{
|
||||||
|
/// <summary>A fake escalation client: scripts the jobs to acquire, records reassignments and
|
||||||
|
/// completions, and can be told to throw on reassigning a given instance.</summary>
|
||||||
|
private sealed class FakeEscalatieClient(params EscalatieJob[] jobs) : IBeoordelingEscalatieClient
|
||||||
|
{
|
||||||
|
public int AcquireCount { get; private set; }
|
||||||
|
public List<string> Reassigned { get; } = [];
|
||||||
|
public List<string> Completed { get; } = [];
|
||||||
|
public string? ThrowOnInstance { get; set; }
|
||||||
|
|
||||||
|
public Task<IReadOnlyList<EscalatieJob>> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
AcquireCount++;
|
||||||
|
return Task.FromResult<IReadOnlyList<EscalatieJob>>(jobs.Take(maxJobs).ToList());
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
if (processInstanceId == ThrowOnInstance)
|
||||||
|
throw new InvalidOperationException("reassign failed");
|
||||||
|
Reassigned.Add(processInstanceId);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
Completed.Add(jobId);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Acquires_an_escalation_reassigns_to_teamlead_and_completes_the_job()
|
||||||
|
{
|
||||||
|
var client = new FakeEscalatieClient(new EscalatieJob("job-9", "pi-1"));
|
||||||
|
|
||||||
|
var acquired = await new BeoordelingEscalatieProcessor(
|
||||||
|
client, NullLogger<BeoordelingEscalatieProcessor>.Instance).PumpOnceAsync(5);
|
||||||
|
|
||||||
|
Assert.Equal(1, acquired);
|
||||||
|
Assert.Equal("pi-1", Assert.Single(client.Reassigned));
|
||||||
|
Assert.Equal("job-9", Assert.Single(client.Completed));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task A_failing_reassign_is_left_uncompleted_for_flowable_to_redeliver()
|
||||||
|
{
|
||||||
|
var client = new FakeEscalatieClient(new EscalatieJob("job-9", "pi-1")) { ThrowOnInstance = "pi-1" };
|
||||||
|
var logger = new CapturingLogger<BeoordelingEscalatieProcessor>();
|
||||||
|
|
||||||
|
var acquired = await new BeoordelingEscalatieProcessor(client, logger).PumpOnceAsync(5);
|
||||||
|
|
||||||
|
Assert.Equal(1, acquired);
|
||||||
|
Assert.Empty(client.Completed);
|
||||||
|
var error = Assert.Single(logger.Entries, e => e.Level == LogLevel.Error);
|
||||||
|
Assert.Contains("job-9", error.Message);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Does_nothing_but_poll_when_there_are_no_escalations()
|
||||||
|
{
|
||||||
|
var client = new FakeEscalatieClient();
|
||||||
|
|
||||||
|
var acquired = await new BeoordelingEscalatieProcessor(
|
||||||
|
client, NullLogger<BeoordelingEscalatieProcessor>.Instance).PumpOnceAsync(5);
|
||||||
|
|
||||||
|
Assert.Equal(0, acquired);
|
||||||
|
Assert.Equal(1, client.AcquireCount);
|
||||||
|
Assert.Empty(client.Completed);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -306,4 +306,108 @@ public class FlowableWorkflowClientTests
|
|||||||
await Assert.ThrowsAsync<HttpRequestException>(
|
await Assert.ThrowsAsync<HttpRequestException>(
|
||||||
() => client.CompleteBeoordelingAsync("task-1", BeoordelingsBesluit.Goedkeuren));
|
() => client.CompleteBeoordelingAsync("task-1", BeoordelingsBesluit.Goedkeuren));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── S-14 (#15): 14-day beoordeling escalation → reassign to teamlead (ADR-0015) ───────────────
|
||||||
|
// The BPMN parks a parallel escalation token on a non-interrupting P14D boundary timer, surfaced as
|
||||||
|
// an external-worker job on the BeoordelingEscaleren topic. The worker reassigns the still-open
|
||||||
|
// Beoordelen task from the behandelaar group to teamlead, then completes the escalation job.
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Acquire_escalation_jobs_posts_the_escalation_topic_and_parses_the_process_instance()
|
||||||
|
{
|
||||||
|
var capture = new RequestCapture();
|
||||||
|
var client = Client(capture.Responds(HttpStatusCode.OK,
|
||||||
|
"""[{"id":"job-9","processInstanceId":"pi-1"}]"""));
|
||||||
|
|
||||||
|
var jobs = await client.AcquireBeoordelingEscalatieJobsAsync(3);
|
||||||
|
|
||||||
|
var job = Assert.Single(jobs);
|
||||||
|
Assert.Equal("job-9", job.JobId);
|
||||||
|
Assert.Equal("pi-1", job.ProcessInstanceId);
|
||||||
|
Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs",
|
||||||
|
capture.Seen!.RequestUri!.ToString());
|
||||||
|
Assert.Contains("\"topic\":\"BeoordelingEscaleren\"", capture.Body);
|
||||||
|
Assert.Contains("\"numberOfTasks\":3", capture.Body);
|
||||||
|
Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Theory]
|
||||||
|
[InlineData("[]")]
|
||||||
|
[InlineData("null")]
|
||||||
|
public async Task Acquire_escalation_jobs_returns_empty_when_none_are_parked(string body)
|
||||||
|
{
|
||||||
|
var capture = new RequestCapture();
|
||||||
|
var client = Client(capture.Responds(HttpStatusCode.OK, body));
|
||||||
|
|
||||||
|
Assert.Empty(await client.AcquireBeoordelingEscalatieJobsAsync(1));
|
||||||
|
Assert.NotNull(capture.Seen);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Reassign_moves_the_open_beoordelen_task_from_behandelaar_to_teamlead()
|
||||||
|
{
|
||||||
|
var requests = new List<(HttpMethod Method, string Url, string? Body)>();
|
||||||
|
var client = Client(new StubHandler(async req =>
|
||||||
|
{
|
||||||
|
requests.Add((req.Method, req.RequestUri!.ToString(),
|
||||||
|
req.Content is null ? null : await req.Content.ReadAsStringAsync()));
|
||||||
|
// The task query returns the still-open Beoordelen task parked in this instance.
|
||||||
|
return req.RequestUri!.AbsoluteUri.EndsWith("service/query/tasks")
|
||||||
|
? new HttpResponseMessage(HttpStatusCode.OK)
|
||||||
|
{
|
||||||
|
Content = new StringContent("""{"data":[{"id":"task-7"}],"total":1}""",
|
||||||
|
Encoding.UTF8, "application/json"),
|
||||||
|
}
|
||||||
|
: new HttpResponseMessage(HttpStatusCode.OK);
|
||||||
|
}));
|
||||||
|
|
||||||
|
await client.ReassignBeoordelingToTeamleadAsync("pi-1");
|
||||||
|
|
||||||
|
// 1. Find the still-open Beoordelen task in this process instance.
|
||||||
|
var query = requests.Single(r => r.Url.EndsWith("service/query/tasks"));
|
||||||
|
Assert.Equal(HttpMethod.Post, query.Method);
|
||||||
|
Assert.Contains("\"processInstanceId\":\"pi-1\"", query.Body);
|
||||||
|
Assert.Contains("\"taskDefinitionKey\":\"Beoordelen\"", query.Body);
|
||||||
|
// 2. Add teamlead as a candidate group on that task.
|
||||||
|
var add = requests.Single(r => r.Method == HttpMethod.Post
|
||||||
|
&& r.Url.EndsWith("service/runtime/tasks/task-7/identitylinks"));
|
||||||
|
Assert.Contains("\"group\":\"teamlead\"", add.Body);
|
||||||
|
Assert.Contains("\"type\":\"candidate\"", add.Body);
|
||||||
|
// 3. Remove behandelaar as a candidate group — the task now belongs to teamlead.
|
||||||
|
Assert.Contains(requests, r => r.Method == HttpMethod.Delete
|
||||||
|
&& r.Url.EndsWith("service/runtime/tasks/task-7/identitylinks/groups/behandelaar/candidate"));
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Reassign_is_a_no_op_when_the_beoordelen_task_is_no_longer_open()
|
||||||
|
{
|
||||||
|
// The behandelaar completed it just before the timer fired: nothing to reassign, no throw.
|
||||||
|
var methods = new List<HttpMethod>();
|
||||||
|
var client = Client(new StubHandler(req =>
|
||||||
|
{
|
||||||
|
methods.Add(req.Method);
|
||||||
|
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK)
|
||||||
|
{
|
||||||
|
Content = new StringContent("""{"data":[],"total":0}""", Encoding.UTF8, "application/json"),
|
||||||
|
});
|
||||||
|
}));
|
||||||
|
|
||||||
|
await client.ReassignBeoordelingToTeamleadAsync("pi-1");
|
||||||
|
|
||||||
|
Assert.DoesNotContain(HttpMethod.Delete, methods);
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task Complete_escalation_job_posts_to_the_job_complete_endpoint()
|
||||||
|
{
|
||||||
|
var capture = new RequestCapture();
|
||||||
|
var client = Client(capture.Responds(HttpStatusCode.NoContent));
|
||||||
|
|
||||||
|
await client.CompleteBeoordelingEscalatieJobAsync("job-9");
|
||||||
|
|
||||||
|
Assert.Equal(HttpMethod.Post, capture.Seen!.Method);
|
||||||
|
Assert.Equal("http://flowable/flowable-rest/external-job-api/acquire/jobs/job-9/complete",
|
||||||
|
capture.Seen.RequestUri!.ToString());
|
||||||
|
Assert.Contains("\"workerId\":\"worker-x\"", capture.Body);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,7 +4,8 @@
|
|||||||
"test-projects": ["Big.Tests/Big.Tests.csproj"],
|
"test-projects": ["Big.Tests/Big.Tests.csproj"],
|
||||||
"reporters": ["progress", "html"],
|
"reporters": ["progress", "html"],
|
||||||
"mutate": [
|
"mutate": [
|
||||||
"!**/OpenZaakJobPump.cs"
|
"!**/OpenZaakJobPump.cs",
|
||||||
|
"!**/BeoordelingEscalatiePump.cs"
|
||||||
],
|
],
|
||||||
"thresholds": {
|
"thresholds": {
|
||||||
"high": 95,
|
"high": 95,
|
||||||
|
|||||||
@@ -22,6 +22,9 @@
|
|||||||
<ProjectReference Include="..\..\services\acl\Acl.Infrastructure\Acl.Infrastructure.csproj" />
|
<ProjectReference Include="..\..\services\acl\Acl.Infrastructure\Acl.Infrastructure.csproj" />
|
||||||
<ProjectReference Include="..\..\services\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
|
<ProjectReference Include="..\..\services\event-subscriber\EventSubscriber.Application\EventSubscriber.Application.csproj" />
|
||||||
<ProjectReference Include="..\..\services\domain\Big.Application\Big.Application.csproj" />
|
<ProjectReference Include="..\..\services\domain\Big.Application\Big.Application.csproj" />
|
||||||
|
<!-- The beoordeling-escalation scenario drives the escalation worker (Infrastructure), as the
|
||||||
|
ACL scenario drives Acl.Infrastructure — escalation has no domain-aggregate surface (S-14). -->
|
||||||
|
<ProjectReference Include="..\..\services\domain\Big.Infrastructure\Big.Infrastructure.csproj" />
|
||||||
<ProjectReference Include="..\..\services\bff\Bff.Api\Bff.Api.csproj" />
|
<ProjectReference Include="..\..\services\bff\Bff.Api\Bff.Api.csproj" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
|
|||||||
22
tests/acceptance/Features/EenBeoordelingEscaleren.feature
Normal file
22
tests/acceptance/Features/EenBeoordelingEscaleren.feature
Normal file
@@ -0,0 +1,22 @@
|
|||||||
|
# language: en
|
||||||
|
# Drives S-14 (#15). A beoordeling a behandelaar does not pick up within 14 days escalates to the
|
||||||
|
# teamlead: a non-interrupting boundary timer parks a BeoordelingEscaleren job (ADR-0015) which the
|
||||||
|
# escalation worker drains, reassigning the still-open Beoordelen task's candidate group to teamlead.
|
||||||
|
# A beoordeling completed before the timer fires does not escalate. This scenario exercises the
|
||||||
|
# escalation worker against an in-memory Flowable stand-in; the timer firing live is verify-domain.
|
||||||
|
Feature: Een beoordeling escaleren
|
||||||
|
Als teamleider wil ik dat een beoordeling die na 14 dagen niet is opgepakt naar mij escaleert
|
||||||
|
zodat aanvragen niet blijven liggen.
|
||||||
|
|
||||||
|
Scenario: Na 14 dagen zonder oppakken escaleert de beoordeling naar de teamlead
|
||||||
|
Given a registration parked at the Beoordelen task for the behandelaar
|
||||||
|
When the 14-day escalation timer fires
|
||||||
|
And the escalation worker runs
|
||||||
|
Then the beoordeling is reassigned to the teamlead
|
||||||
|
|
||||||
|
Scenario: Een tijdig afgeronde beoordeling escaleert niet
|
||||||
|
Given a registration parked at the Beoordelen task for the behandelaar
|
||||||
|
When the behandelaar completes the beoordeling before the timer fires
|
||||||
|
And the 14-day escalation timer fires
|
||||||
|
And the escalation worker runs
|
||||||
|
Then the beoordeling stays with the behandelaar
|
||||||
44
tests/acceptance/Steps/EenBeoordelingEscalerenSteps.cs
Normal file
44
tests/acceptance/Steps/EenBeoordelingEscalerenSteps.cs
Normal file
@@ -0,0 +1,44 @@
|
|||||||
|
using Acceptance.Support;
|
||||||
|
using Big.Infrastructure;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Reqnroll;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Acceptance.Steps;
|
||||||
|
|
||||||
|
/// <summary>Bindings for <c>EenBeoordelingEscaleren.feature</c> (S-14). Drives the escalation worker
|
||||||
|
/// (<see cref="BeoordelingEscalatieProcessor"/>) against an in-memory Flowable stand-in; one instance
|
||||||
|
/// per scenario. Escalation has no domain-aggregate surface — it only reassigns who may claim the
|
||||||
|
/// still-open Beoordelen task — so the scenario asserts on the task's candidate group.</summary>
|
||||||
|
[Binding]
|
||||||
|
[Scope(Feature = "Een beoordeling escaleren")]
|
||||||
|
public sealed class EenBeoordelingEscalerenSteps
|
||||||
|
{
|
||||||
|
private readonly InMemoryEscalatieClient _flowable = new();
|
||||||
|
private string _processInstanceId = "";
|
||||||
|
|
||||||
|
[Given("a registration parked at the Beoordelen task for the behandelaar")]
|
||||||
|
public void GivenARegistrationParkedForTheBehandelaar()
|
||||||
|
=> _processInstanceId = _flowable.ParkBeoordeling();
|
||||||
|
|
||||||
|
[When("the 14-day escalation timer fires")]
|
||||||
|
public void WhenTheEscalationTimerFires()
|
||||||
|
=> _flowable.FireEscalationTimer(_processInstanceId);
|
||||||
|
|
||||||
|
[When("the behandelaar completes the beoordeling before the timer fires")]
|
||||||
|
public void WhenTheBehandelaarCompletesBeforeTheTimer()
|
||||||
|
=> _flowable.CompleteBeoordeling(_processInstanceId);
|
||||||
|
|
||||||
|
[When("the escalation worker runs")]
|
||||||
|
public async Task WhenTheEscalationWorkerRuns()
|
||||||
|
=> await new BeoordelingEscalatieProcessor(
|
||||||
|
_flowable, NullLogger<BeoordelingEscalatieProcessor>.Instance).PumpOnceAsync(5);
|
||||||
|
|
||||||
|
[Then("the beoordeling is reassigned to the teamlead")]
|
||||||
|
public void ThenTheBeoordelingIsReassignedToTheTeamlead()
|
||||||
|
=> Assert.Equal("teamlead", _flowable.CandidateGroupFor(_processInstanceId));
|
||||||
|
|
||||||
|
[Then("the beoordeling stays with the behandelaar")]
|
||||||
|
public void ThenTheBeoordelingStaysWithTheBehandelaar()
|
||||||
|
=> Assert.Equal("behandelaar", _flowable.CandidateGroupFor(_processInstanceId));
|
||||||
|
}
|
||||||
@@ -1,5 +1,6 @@
|
|||||||
using Big.Application;
|
using Big.Application;
|
||||||
using Big.Domain;
|
using Big.Domain;
|
||||||
|
using Big.Infrastructure;
|
||||||
|
|
||||||
namespace Acceptance.Support;
|
namespace Acceptance.Support;
|
||||||
|
|
||||||
@@ -73,6 +74,59 @@ public sealed class InMemoryUserTaskClient : IUserTaskClient
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>An in-memory Flowable stand-in for the beoordeling-escalation scenario (S-14): it models
|
||||||
|
/// one Beoordelen task per process instance — its candidate group and whether it is still open — and
|
||||||
|
/// the escalation jobs the non-interrupting 14-day boundary timer parks. It drives the escalation
|
||||||
|
/// worker's behaviour without a running Flowable; the timer firing live is the verify-domain check.</summary>
|
||||||
|
public sealed class InMemoryEscalatieClient : IBeoordelingEscalatieClient
|
||||||
|
{
|
||||||
|
private sealed class ParkedTask
|
||||||
|
{
|
||||||
|
public string CandidateGroup { get; set; } = "behandelaar";
|
||||||
|
public bool IsOpen { get; set; } = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
private readonly Dictionary<string, ParkedTask> _tasks = [];
|
||||||
|
private readonly List<EscalatieJob> _parked = [];
|
||||||
|
private int _seq;
|
||||||
|
|
||||||
|
/// <summary>A registration parks at Beoordelen, claimable by the behandelaar group.</summary>
|
||||||
|
public string ParkBeoordeling()
|
||||||
|
{
|
||||||
|
var pid = $"pi-{++_seq}";
|
||||||
|
_tasks[pid] = new ParkedTask();
|
||||||
|
return pid;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>The behandelaar completes the beoordeling before the timer fires: the task closes.</summary>
|
||||||
|
public void CompleteBeoordeling(string processInstanceId) => _tasks[processInstanceId].IsOpen = false;
|
||||||
|
|
||||||
|
/// <summary>The 14-day boundary timer fires: a non-interrupting token parks an escalation job.</summary>
|
||||||
|
public void FireEscalationTimer(string processInstanceId)
|
||||||
|
=> _parked.Add(new EscalatieJob($"job-{++_seq}", processInstanceId));
|
||||||
|
|
||||||
|
/// <summary>The candidate group that may currently pick the task up.</summary>
|
||||||
|
public string CandidateGroupFor(string processInstanceId) => _tasks[processInstanceId].CandidateGroup;
|
||||||
|
|
||||||
|
public Task<IReadOnlyList<EscalatieJob>> AcquireBeoordelingEscalatieJobsAsync(int maxJobs, CancellationToken ct = default)
|
||||||
|
=> Task.FromResult<IReadOnlyList<EscalatieJob>>(_parked.Take(maxJobs).ToList());
|
||||||
|
|
||||||
|
public Task ReassignBeoordelingToTeamleadAsync(string processInstanceId, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
// No-op if the behandelaar already completed it — the timer/completion race (§8.6).
|
||||||
|
var task = _tasks[processInstanceId];
|
||||||
|
if (task.IsOpen)
|
||||||
|
task.CandidateGroup = "teamlead";
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default)
|
||||||
|
{
|
||||||
|
_parked.RemoveAll(j => j.JobId == jobId);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>An in-memory registration store for the domain acceptance scenario.</summary>
|
/// <summary>An in-memory registration store for the domain acceptance scenario.</summary>
|
||||||
public sealed class InMemoryRegistrationStore : IRegistrationStore
|
public sealed class InMemoryRegistrationStore : IRegistrationStore
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -15,7 +15,12 @@
|
|||||||
S-11 adds withdrawal: while parked at Beoordelen the citizen can trek de aanvraag in — an
|
S-11 adds withdrawal: while parked at Beoordelen the citizen can trek de aanvraag in — an
|
||||||
interrupting message boundary event (RegistratieIngetrokken) cancels the task and ends the
|
interrupting message boundary event (RegistratieIngetrokken) cancels the task and ends the
|
||||||
process via a dedicated "ingetrokken" end (ADR-0014). The Workflow Client delivers the message
|
process via a dedicated "ingetrokken" end (ADR-0014). The Workflow Client delivers the message
|
||||||
to the task's execution; the BPMN owns the cancellation path. -->
|
to the task's execution; the BPMN owns the cancellation path.
|
||||||
|
S-14 adds escalation: a NON-interrupting boundary timer (P14D) on Beoordelen. If a behandelaar
|
||||||
|
has not picked the task up within 14 days it fires a parallel token to an external-worker task
|
||||||
|
(BeoordelingEscaleren); the Workflow Client reassigns the still-open Beoordelen task from the
|
||||||
|
behandelaar group to teamlead (ADR-0015). The Beoordelen task stays open throughout — the timer
|
||||||
|
only changes who may claim it. -->
|
||||||
<message id="Message_Ingetrokken" name="RegistratieIngetrokken"/>
|
<message id="Message_Ingetrokken" name="RegistratieIngetrokken"/>
|
||||||
|
|
||||||
<process id="registratie" name="Registratie ontvangen" isExecutable="true">
|
<process id="registratie" name="Registratie ontvangen" isExecutable="true">
|
||||||
@@ -45,6 +50,25 @@
|
|||||||
<sequenceFlow id="flow4" sourceRef="Ingetrokken" targetRef="endIngetrokken"/>
|
<sequenceFlow id="flow4" sourceRef="Ingetrokken" targetRef="endIngetrokken"/>
|
||||||
|
|
||||||
<endEvent id="endIngetrokken" name="Registratie ingetrokken"/>
|
<endEvent id="endIngetrokken" name="Registratie ingetrokken"/>
|
||||||
|
|
||||||
|
<!-- Escalation (S-14): non-interrupting P14D boundary timer on Beoordelen. On timeout a parallel
|
||||||
|
token runs EscaleerBeoordeling, an external-worker task the Workflow Client picks up to
|
||||||
|
reassign the still-open Beoordelen task from behandelaar to teamlead (ADR-0015). -->
|
||||||
|
<boundaryEvent id="EscaleerTimer" attachedToRef="Beoordelen" cancelActivity="false">
|
||||||
|
<timerEventDefinition>
|
||||||
|
<timeDuration>P14D</timeDuration>
|
||||||
|
</timerEventDefinition>
|
||||||
|
</boundaryEvent>
|
||||||
|
|
||||||
|
<sequenceFlow id="flow5" sourceRef="EscaleerTimer" targetRef="EscaleerBeoordeling"/>
|
||||||
|
|
||||||
|
<serviceTask id="EscaleerBeoordeling" name="Beoordeling escaleren"
|
||||||
|
flowable:type="external-worker"
|
||||||
|
flowable:topic="BeoordelingEscaleren"/>
|
||||||
|
|
||||||
|
<sequenceFlow id="flow6" sourceRef="EscaleerBeoordeling" targetRef="endEscaleren"/>
|
||||||
|
|
||||||
|
<endEvent id="endEscaleren" name="Beoordeling geëscaleerd"/>
|
||||||
</process>
|
</process>
|
||||||
|
|
||||||
<bpmndi:BPMNDiagram id="diagram">
|
<bpmndi:BPMNDiagram id="diagram">
|
||||||
@@ -83,6 +107,24 @@
|
|||||||
<omgdi:waypoint x="450" y="165"/>
|
<omgdi:waypoint x="450" y="165"/>
|
||||||
<omgdi:waypoint x="450" y="220"/>
|
<omgdi:waypoint x="450" y="220"/>
|
||||||
</bpmndi:BPMNEdge>
|
</bpmndi:BPMNEdge>
|
||||||
|
<bpmndi:BPMNShape id="s_escaleerTimer" bpmnElement="EscaleerTimer">
|
||||||
|
<omgdc:Bounds x="480" y="70" width="30" height="30"/>
|
||||||
|
</bpmndi:BPMNShape>
|
||||||
|
<bpmndi:BPMNShape id="s_escaleerBeoordeling" bpmnElement="EscaleerBeoordeling">
|
||||||
|
<omgdc:Bounds x="580" y="20" width="120" height="60"/>
|
||||||
|
</bpmndi:BPMNShape>
|
||||||
|
<bpmndi:BPMNShape id="s_endEscaleren" bpmnElement="endEscaleren">
|
||||||
|
<omgdc:Bounds x="760" y="35" width="30" height="30"/>
|
||||||
|
</bpmndi:BPMNShape>
|
||||||
|
<bpmndi:BPMNEdge id="e_flow5" bpmnElement="flow5">
|
||||||
|
<omgdi:waypoint x="495" y="70"/>
|
||||||
|
<omgdi:waypoint x="495" y="50"/>
|
||||||
|
<omgdi:waypoint x="580" y="50"/>
|
||||||
|
</bpmndi:BPMNEdge>
|
||||||
|
<bpmndi:BPMNEdge id="e_flow6" bpmnElement="flow6">
|
||||||
|
<omgdi:waypoint x="700" y="50"/>
|
||||||
|
<omgdi:waypoint x="760" y="50"/>
|
||||||
|
</bpmndi:BPMNEdge>
|
||||||
</bpmndi:BPMNPlane>
|
</bpmndi:BPMNPlane>
|
||||||
</bpmndi:BPMNDiagram>
|
</bpmndi:BPMNDiagram>
|
||||||
</definitions>
|
</definitions>
|
||||||
|
|||||||
Reference in New Issue
Block a user