feat(workflow): document-wait task + 30-day timeout cancellation (S-10a, closes #102) #105

Merged
not merged 23 commits from feat/102-document-wait-timeout into main 2026-07-20 09:42:03 +00:00
4 changed files with 99 additions and 0 deletions
Showing only changes of commit c536c965de - Show all commits
@@ -0,0 +1,34 @@
using Big.Domain;
namespace Big.Application;
/// <summary>
/// Handles one acquired <c>RegistratieVerlopen</c> external-worker job (S-10a, ADR-0017): load the
/// registration the job correlates to and expire it to VERLOPEN — the 30-day document-wait timer fired
/// before the documents arrived, so the case is cancelled. Pure application logic over ports; it knows
/// nothing of Flowable. The polling loop that feeds it jobs lives in Infrastructure. Mirrors
/// <see cref="OpenZaakWorker"/>.
/// </summary>
public sealed class ExpireRegistrationWorker(IRegistrationStore store)
{
/// <summary>
/// Process the job. Idempotent: a redelivered job whose registration is already VERLOPEN is a
/// no-op — not persisted again (§8.6, at-least-once delivery). An unknown registration is an error:
/// it throws, leaving the job un-completed for Flowable to redeliver.
/// </summary>
public async Task HandleAsync(RegistratieVerlopenJob job, CancellationToken ct = default)
{
ArgumentNullException.ThrowIfNull(job);
var registration = await store.GetAsync(job.RegistrationId, ct)
?? throw new InvalidOperationException(
$"No registration {job.RegistrationId} for RegistratieVerlopen job {job.JobId}.");
// A redelivered job whose registration is already VERLOPEN completes without persisting again.
if (registration.Status == RegistrationStatus.Verlopen)
return;
registration.Expire();
await store.SaveAsync(registration, ct);
}
}
+8
View File
@@ -95,3 +95,11 @@ public sealed record OpenZaakJob(string JobId, RegistrationId RegistrationId);
/// once the 14-day boundary timer fires (ADR-0015).
/// </summary>
public sealed record EscalatieJob(string JobId, string ProcessInstanceId);
/// <summary>
/// An acquired <c>RegistratieVerlopen</c> job (S-10a): the Flowable job id and the registration id it
/// carries as a process variable. The 30-day boundary timer on <c>WachtOpDocumenten</c> spawns it when
/// the required documents were not supplied in time; expiring the correlated registration to VERLOPEN
/// cancels the case (ADR-0017).
/// </summary>
public sealed record RegistratieVerlopenJob(string JobId, RegistrationId RegistrationId);
@@ -36,3 +36,19 @@ public interface IBeoordelingEscalatieClient
/// <summary>Complete an acquired escalation job so its token reaches the escalation end event.</summary>
Task CompleteBeoordelingEscalatieJobAsync(string jobId, CancellationToken ct = default);
}
/// <summary>
/// The document-timeout side of the Workflow Client (S-10a): the <c>RegistratieVerlopen</c>
/// external-worker jobs parked by the 30-day boundary timer on <c>WachtOpDocumenten</c>. Kept separate
/// from the other worker ports (interface segregation) so neither the OpenZaak nor escalation worker
/// sees expiry. Implemented by <see cref="FlowableWorkflowClient"/> — the only code that talks to
/// Flowable (§8.2, ADR-0017).
/// </summary>
public interface IRegistratieVerlopenClient
{
/// <summary>Acquire and lock up to <paramref name="maxJobs"/> <c>RegistratieVerlopen</c> jobs.</summary>
Task<IReadOnlyList<RegistratieVerlopenJob>> AcquireRegistratieVerlopenJobsAsync(int maxJobs, CancellationToken ct = default);
/// <summary>Complete an acquired expiry job so its token reaches the <c>endVerlopen</c> end event.</summary>
Task CompleteRegistratieVerlopenJobAsync(string jobId, CancellationToken ct = default);
}
@@ -0,0 +1,41 @@
using Big.Application;
using Microsoft.Extensions.Logging;
namespace Big.Infrastructure;
/// <summary>
/// One poll tick of the document-timeout worker (S-10a, ADR-0017): acquire the parked
/// <c>RegistratieVerlopen</c> jobs — the tokens the 30-day boundary timer on <c>WachtOpDocumenten</c>
/// spawns — expire each correlated registration via the <see cref="ExpireRegistrationWorker"/>, and
/// complete the job so its token reaches <c>endVerlopen</c>. 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→expire→complete logic is unit-testable without a running host. Mirrors
/// <see cref="OpenZaakJobProcessor"/> and <see cref="BeoordelingEscalatieProcessor"/>.
/// </summary>
public sealed class RegistratieVerlopenProcessor(
IRegistratieVerlopenClient client,
ExpireRegistrationWorker worker,
ILogger<RegistratieVerlopenProcessor> logger)
{
/// <summary>Acquire and process up to <paramref name="maxJobs"/> jobs. Returns the number acquired.</summary>
public async Task<int> PumpOnceAsync(int maxJobs, CancellationToken ct = default)
{
var jobs = await client.AcquireRegistratieVerlopenJobsAsync(maxJobs, ct);
foreach (var job in jobs)
{
try
{
await worker.HandleAsync(job, ct);
await client.CompleteRegistratieVerlopenJobAsync(job.JobId, ct);
}
catch (Exception ex)
{
// Leave the job un-completed: its lock expires and Flowable redelivers it (§8.6).
logger.LogError(ex, "RegistratieVerlopen job {JobId} failed; leaving it for redelivery.", job.JobId);
}
}
return jobs.Count;
}
}