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
2 changed files with 144 additions and 0 deletions
Showing only changes of commit 5add817c10 - Show all commits
@@ -0,0 +1,65 @@
using Big.Application;
using Big.Domain;
namespace Big.Tests;
// S-10a (#102): the application handler behind the RegistratieVerlopen external-worker job. The 30-day
// document-wait timer fired, so the correlated registration is expired to VERLOPEN. Mirrors
// OpenZaakWorker — pure application logic over ports, idempotent under at-least-once delivery (§8.6).
public class ExpireRegistrationWorkerTests
{
private const string Bsn = "123456782";
private static Registration Submitted(string processInstanceId = "proc-1")
{
var registration = Registration.Submit(Bsn);
registration.RecordProcessStarted(processInstanceId);
return registration;
}
[Fact]
public async Task Expires_the_registration_the_job_correlates_to()
{
var store = new FakeRegistrationStore();
var registration = Submitted();
store.Seed(registration);
await new ExpireRegistrationWorker(store).HandleAsync(
new RegistratieVerlopenJob("job-7", registration.Id));
var saved = await store.GetAsync(registration.Id);
Assert.Equal(RegistrationStatus.Verlopen, saved!.Status);
Assert.Equal(1, store.SaveCount);
}
[Fact]
public async Task An_already_verlopen_registration_is_not_persisted_again()
{
// A redelivered job (§8.6) finds the aggregate already VERLOPEN: a no-op, not saved again.
var store = new FakeRegistrationStore();
var registration = Submitted();
registration.Expire();
store.Seed(registration);
await new ExpireRegistrationWorker(store).HandleAsync(
new RegistratieVerlopenJob("job-7", registration.Id));
Assert.Equal(0, store.SaveCount);
Assert.Equal(RegistrationStatus.Verlopen, (await store.GetAsync(registration.Id))!.Status);
}
[Fact]
public async Task An_unknown_registration_throws_so_the_job_is_redelivered()
{
var store = new FakeRegistrationStore();
await Assert.ThrowsAsync<InvalidOperationException>(() =>
new ExpireRegistrationWorker(store).HandleAsync(
new RegistratieVerlopenJob("job-7", RegistrationId.New())));
}
[Fact]
public async Task Rejects_a_null_job()
=> await Assert.ThrowsAsync<ArgumentNullException>(() =>
new ExpireRegistrationWorker(new FakeRegistrationStore()).HandleAsync(null!));
}
@@ -0,0 +1,79 @@
using Big.Application;
using Big.Infrastructure;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Big.Tests;
// S-10a (#102): the document-timeout drain loop. Mirrors BeoordelingEscalatieProcessor — acquire the
// parked RegistratieVerlopen jobs (the tokens the 30-day boundary timer on WachtOpDocumenten spawns),
// expire each correlated registration via the ExpireRegistrationWorker, then complete the job. A job
// whose expiry fails is logged and left un-completed for Flowable to redeliver (§8.6).
public class RegistratieVerlopenProcessorTests
{
/// <summary>A fake client scripting the jobs to acquire and recording completions.</summary>
private sealed class FakeVerlopenClient(params RegistratieVerlopenJob[] jobs) : IRegistratieVerlopenClient
{
public int AcquireCount { get; private set; }
public List<string> Completed { get; } = [];
public Task<IReadOnlyList<RegistratieVerlopenJob>> AcquireRegistratieVerlopenJobsAsync(int maxJobs, CancellationToken ct = default)
{
AcquireCount++;
return Task.FromResult<IReadOnlyList<RegistratieVerlopenJob>>(jobs.Take(maxJobs).ToList());
}
public Task CompleteRegistratieVerlopenJobAsync(string jobId, CancellationToken ct = default)
{
Completed.Add(jobId);
return Task.CompletedTask;
}
}
private static ExpireRegistrationWorker Worker(FakeRegistrationStore store) => new(store);
[Fact]
public async Task Acquires_a_job_expires_the_registration_and_completes_the_job()
{
var store = new FakeRegistrationStore();
var registration = Domain.Registration.Submit("123456782");
store.Seed(registration);
var client = new FakeVerlopenClient(new RegistratieVerlopenJob("job-9", registration.Id));
var acquired = await new RegistratieVerlopenProcessor(
client, Worker(store), NullLogger<RegistratieVerlopenProcessor>.Instance).PumpOnceAsync(5);
Assert.Equal(1, acquired);
Assert.Equal(Domain.RegistrationStatus.Verlopen, (await store.GetAsync(registration.Id))!.Status);
Assert.Equal("job-9", Assert.Single(client.Completed));
}
[Fact]
public async Task A_failing_expiry_is_left_uncompleted_for_flowable_to_redeliver()
{
// Unknown registration → the worker throws → the job is left for redelivery, error logged.
var store = new FakeRegistrationStore();
var client = new FakeVerlopenClient(new RegistratieVerlopenJob("job-9", Domain.RegistrationId.New()));
var logger = new CapturingLogger<RegistratieVerlopenProcessor>();
var acquired = await new RegistratieVerlopenProcessor(client, Worker(store), 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_jobs()
{
var client = new FakeVerlopenClient();
var acquired = await new RegistratieVerlopenProcessor(
client, Worker(new FakeRegistrationStore()), NullLogger<RegistratieVerlopenProcessor>.Instance).PumpOnceAsync(5);
Assert.Equal(0, acquired);
Assert.Equal(1, client.AcquireCount);
Assert.Empty(client.Completed);
}
}