feature: In-App Simulation Runner (Backend)
This commit is contained in:
voltsrage
@@ -0,0 +1,269 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Options;
|
||||
using VigilCare.Simulation;
|
||||
|
||||
public interface ISimulationRunner
|
||||
{
|
||||
IReadOnlyList<SimulationRunState> ListRuns();
|
||||
SimulationRunState? GetRun(Guid runId);
|
||||
Task<SimulationRunState> StartAsync(
|
||||
string scenarioId, double speed, string startedByUserId, CancellationToken ct);
|
||||
bool Cancel(Guid runId);
|
||||
}
|
||||
|
||||
public sealed class SimulationRunner : ISimulationRunner, IHostedService
|
||||
{
|
||||
private readonly ConcurrentDictionary<Guid, RunContext> _runs = new();
|
||||
private readonly ISimulationClientFactory _clientFactory;
|
||||
private readonly IScenarioCatalog _catalog;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly SimulationOptions _options;
|
||||
private readonly ILogger<SimulationRunner> _logger;
|
||||
|
||||
public SimulationRunner(
|
||||
ISimulationClientFactory clientFactory,
|
||||
IScenarioCatalog catalog,
|
||||
IServiceScopeFactory scopeFactory,
|
||||
IOptions<SimulationOptions> options,
|
||||
ILogger<SimulationRunner> logger)
|
||||
{
|
||||
_clientFactory = clientFactory;
|
||||
_catalog = catalog;
|
||||
_scopeFactory = scopeFactory;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public IReadOnlyList<SimulationRunState> ListRuns() =>
|
||||
_runs.Values
|
||||
.Select(c => c.State.Snapshot())
|
||||
.OrderByDescending(s => s.StartedAt)
|
||||
.ToList();
|
||||
|
||||
public SimulationRunState? GetRun(Guid runId) =>
|
||||
_runs.TryGetValue(runId, out var ctx) ? ctx.State.Snapshot() : null;
|
||||
|
||||
public async Task<SimulationRunState> StartAsync(
|
||||
string scenarioId, double speed, string startedByUserId, CancellationToken ct)
|
||||
{
|
||||
if (!_options.Enabled)
|
||||
throw new ValidationException("Simulation is disabled.", "SIMULATION_DISABLED");
|
||||
|
||||
if (string.IsNullOrWhiteSpace(scenarioId))
|
||||
throw new ValidationException("scenarioId is required.", "SIMULATION_SCENARIO_REQUIRED");
|
||||
|
||||
if (speed <= 0 || speed > _options.MaxSpeed)
|
||||
throw new ValidationException(
|
||||
$"speed must be > 0 and <= {_options.MaxSpeed}.", "SIMULATION_SPEED_INVALID");
|
||||
|
||||
var scenario = _catalog.GetById(scenarioId)
|
||||
?? throw new ValidationException(
|
||||
$"Unknown scenario '{scenarioId}'.", "SIMULATION_SCENARIO_UNKNOWN");
|
||||
|
||||
var activeCount = _runs.Values.Count(c =>
|
||||
c.State.Status is SimulationRunStatus.Pending or SimulationRunStatus.Running);
|
||||
if (activeCount >= _options.MaxConcurrentRuns)
|
||||
throw new ConflictException(
|
||||
$"Maximum concurrent simulation runs ({_options.MaxConcurrentRuns}) reached.",
|
||||
"SIMULATION_CONCURRENCY_LIMIT");
|
||||
|
||||
var totalOffset = scenario.Events.Count == 0
|
||||
? 0
|
||||
: scenario.Events.Max(e => e.OffsetMinutes);
|
||||
|
||||
var runId = Guid.NewGuid();
|
||||
var startedAt = DateTimeOffset.UtcNow;
|
||||
var state = new SimulationRunState
|
||||
{
|
||||
RunId = runId,
|
||||
ScenarioId = scenario.Scenario.Id,
|
||||
ScenarioName = scenario.Scenario.Name,
|
||||
Speed = speed,
|
||||
StartedByUserId = startedByUserId,
|
||||
StartedAt = startedAt,
|
||||
TotalOffsetMinutes = totalOffset,
|
||||
PatientDisplayName = $"{scenario.Patient.FirstName} {scenario.Patient.LastName}",
|
||||
};
|
||||
|
||||
await PersistNewRunAsync(state, ct);
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
var ctx = new RunContext(state, cts, scenario);
|
||||
if (!_runs.TryAdd(runId, ctx))
|
||||
throw new ConflictException("Failed to register simulation run.", "SIMULATION_REGISTER_FAILED");
|
||||
|
||||
_ = Task.Run(() => ExecuteAsync(ctx, CancellationToken.None), CancellationToken.None);
|
||||
|
||||
return state.Snapshot();
|
||||
}
|
||||
|
||||
public bool Cancel(Guid runId)
|
||||
{
|
||||
if (!_runs.TryGetValue(runId, out var ctx))
|
||||
return false;
|
||||
|
||||
if (ctx.State.Status is SimulationRunStatus.Completed
|
||||
or SimulationRunStatus.Cancelled
|
||||
or SimulationRunStatus.Failed)
|
||||
return true;
|
||||
|
||||
ctx.Cts.Cancel();
|
||||
return true;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var ctx in _runs.Values)
|
||||
{
|
||||
if (ctx.State.Status is SimulationRunStatus.Pending or SimulationRunStatus.Running)
|
||||
ctx.Cts.Cancel();
|
||||
}
|
||||
|
||||
var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
|
||||
while (DateTimeOffset.UtcNow < deadline
|
||||
&& _runs.Values.Any(c =>
|
||||
c.State.Status is SimulationRunStatus.Pending or SimulationRunStatus.Running))
|
||||
{
|
||||
await Task.Delay(50, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ExecuteAsync(RunContext ctx, CancellationToken _)
|
||||
{
|
||||
var runId = ctx.State.RunId;
|
||||
ctx.State.MarkRunning();
|
||||
await UpdateRunRowAsync(ctx.State, SimulationRunStatus.Running);
|
||||
|
||||
try
|
||||
{
|
||||
var client = await _clientFactory.CreateAsync(ctx.Cts.Token);
|
||||
var observer = new RunStateReplayObserver(ctx.State);
|
||||
var engine = new ReplayEngine(
|
||||
client,
|
||||
poller: null,
|
||||
observer,
|
||||
onPatientRegistered: (patientId, ct) => MarkPatientSimulatedAsync(patientId, ct));
|
||||
|
||||
var result = await engine.RunAsync(
|
||||
ctx.Scenario,
|
||||
new ReplayOptions(Speed: ctx.State.Speed, Poll: false),
|
||||
ctx.Cts.Token);
|
||||
|
||||
observer.SyncFromResult(result);
|
||||
ctx.State.MarkTerminal(SimulationRunStatus.Completed);
|
||||
await UpdateRunRowAsync(ctx.State, SimulationRunStatus.Completed);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
ctx.State.MarkTerminal(SimulationRunStatus.Cancelled);
|
||||
await UpdateRunRowAsync(ctx.State, SimulationRunStatus.Cancelled);
|
||||
_logger.LogInformation("Simulation run {RunId} cancelled", runId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
ctx.State.MarkTerminal(SimulationRunStatus.Failed, ex.Message);
|
||||
await UpdateRunRowAsync(ctx.State, SimulationRunStatus.Failed, ex.Message);
|
||||
_logger.LogError(ex, "Simulation run {RunId} failed", runId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
ctx.Cts.Dispose();
|
||||
TrimHistory();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task MarkPatientSimulatedAsync(Guid patientId, CancellationToken ct)
|
||||
{
|
||||
await using var scope = _scopeFactory.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var patient = await db.Patients.FirstOrDefaultAsync(p => p.Id == patientId, ct);
|
||||
if (patient is null)
|
||||
return;
|
||||
|
||||
patient.IsSimulated = true;
|
||||
await db.SaveChangesAsync(ct);
|
||||
}
|
||||
|
||||
private async Task PersistNewRunAsync(SimulationRunState state, CancellationToken ct)
|
||||
{
|
||||
await using var scope = _scopeFactory.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
db.SimulationRuns.Add(new SimulationRun
|
||||
{
|
||||
Id = state.RunId,
|
||||
ScenarioId = state.ScenarioId,
|
||||
ScenarioName = state.ScenarioName,
|
||||
Speed = state.Speed,
|
||||
Status = SimulationRunStatus.Pending,
|
||||
StartedByUserId = state.StartedByUserId,
|
||||
StartedAt = state.StartedAt,
|
||||
TotalOffsetMinutes = state.TotalOffsetMinutes,
|
||||
});
|
||||
await db.SaveChangesAsync(ct);
|
||||
}
|
||||
|
||||
private async Task UpdateRunRowAsync(
|
||||
SimulationRunState state,
|
||||
SimulationRunStatus status,
|
||||
string? failureReason = null)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var scope = _scopeFactory.CreateAsyncScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var row = await db.SimulationRuns.FirstOrDefaultAsync(r => r.Id == state.RunId);
|
||||
if (row is null)
|
||||
return;
|
||||
|
||||
row.Status = status;
|
||||
row.PatientId = state.PatientId;
|
||||
row.EncounterId = state.EncounterId;
|
||||
row.ObservationsSent = state.ObservationsSent;
|
||||
row.MedicationsSent = state.MedicationsSent;
|
||||
row.OrdersPlaced = state.OrdersPlaced;
|
||||
row.LastOffsetMinutes = state.LastOffsetMinutes;
|
||||
row.TotalOffsetMinutes = state.TotalOffsetMinutes;
|
||||
row.FailureReason = failureReason ?? state.FailureReason;
|
||||
if (status is SimulationRunStatus.Completed
|
||||
or SimulationRunStatus.Cancelled
|
||||
or SimulationRunStatus.Failed)
|
||||
{
|
||||
row.CompletedAt = DateTimeOffset.UtcNow;
|
||||
}
|
||||
|
||||
await db.SaveChangesAsync();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to persist simulation run {RunId} status {Status}",
|
||||
state.RunId, status);
|
||||
}
|
||||
}
|
||||
|
||||
private void TrimHistory()
|
||||
{
|
||||
var terminal = _runs.Values
|
||||
.Where(c => c.State.Status is SimulationRunStatus.Completed
|
||||
or SimulationRunStatus.Cancelled
|
||||
or SimulationRunStatus.Failed)
|
||||
.OrderByDescending(c => c.State.StartedAt)
|
||||
.Skip(_options.RunHistoryLimit)
|
||||
.ToList();
|
||||
|
||||
foreach (var old in terminal)
|
||||
_runs.TryRemove(old.State.RunId, out _);
|
||||
}
|
||||
|
||||
private sealed class RunContext(
|
||||
SimulationRunState state,
|
||||
CancellationTokenSource cts,
|
||||
ScenarioFile scenario)
|
||||
{
|
||||
public SimulationRunState State { get; } = state;
|
||||
public CancellationTokenSource Cts { get; } = cts;
|
||||
public ScenarioFile Scenario { get; } = scenario;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user