370 lines
14 KiB
C#
370 lines
14 KiB
C#
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,
|
|
string? sessionId = null);
|
|
Task<SimulationSessionState> StartSessionAsync(
|
|
string sessionId, double? speed, string userId, CancellationToken ct);
|
|
bool Cancel(Guid runId);
|
|
bool HasActiveRuns();
|
|
void ClearRegistry();
|
|
}
|
|
|
|
public sealed class SimulationRunner : ISimulationRunner, IHostedService
|
|
{
|
|
private static readonly TimeSpan SessionStagger = TimeSpan.FromMilliseconds(500);
|
|
|
|
private readonly ConcurrentDictionary<Guid, RunContext> _runs = new();
|
|
private readonly ISimulationClientFactory _clientFactory;
|
|
private readonly IScenarioCatalog _catalog;
|
|
private readonly ISessionCatalog _sessions;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly SimulationOptions _options;
|
|
private readonly ILogger<SimulationRunner> _logger;
|
|
|
|
public SimulationRunner(
|
|
ISimulationClientFactory clientFactory,
|
|
IScenarioCatalog catalog,
|
|
ISessionCatalog sessions,
|
|
IServiceScopeFactory scopeFactory,
|
|
IOptions<SimulationOptions> options,
|
|
ILogger<SimulationRunner> logger)
|
|
{
|
|
_clientFactory = clientFactory;
|
|
_catalog = catalog;
|
|
_sessions = sessions;
|
|
_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 bool HasActiveRuns() =>
|
|
_runs.Values.Any(c =>
|
|
c.State.Status is SimulationRunStatus.Pending or SimulationRunStatus.Running);
|
|
|
|
public void ClearRegistry() => _runs.Clear();
|
|
|
|
public async Task<SimulationRunState> StartAsync(
|
|
string scenarioId, double speed, string startedByUserId, CancellationToken ct,
|
|
string? sessionId = null)
|
|
{
|
|
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");
|
|
|
|
return await RegisterAndStartAsync(scenario, speed, startedByUserId, sessionId, ct);
|
|
}
|
|
|
|
public async Task<SimulationSessionState> StartSessionAsync(
|
|
string sessionId, double? speed, string userId, CancellationToken ct)
|
|
{
|
|
if (!_options.Enabled)
|
|
throw new ValidationException("Simulation is disabled.", "SIMULATION_DISABLED");
|
|
|
|
if (string.IsNullOrWhiteSpace(sessionId))
|
|
throw new ValidationException("sessionId is required.", "SIMULATION_SESSION_REQUIRED");
|
|
|
|
var preset = _sessions.GetById(sessionId)
|
|
?? throw new ValidationException(
|
|
$"Unknown session '{sessionId}'.", "SIMULATION_SESSION_UNKNOWN");
|
|
|
|
var resolvedSpeed = speed ?? preset.DefaultSpeed;
|
|
if (resolvedSpeed <= 0 || resolvedSpeed > _options.MaxSpeed)
|
|
throw new ValidationException(
|
|
$"speed must be > 0 and <= {_options.MaxSpeed}.", "SIMULATION_SPEED_INVALID");
|
|
|
|
// Resolve every scenario up front so a typo never half-starts a session.
|
|
var scenarios = new List<ScenarioFile>(preset.Scenarios.Count);
|
|
foreach (var id in preset.Scenarios)
|
|
{
|
|
var scenario = _catalog.GetById(id)
|
|
?? throw new ValidationException(
|
|
$"Unknown scenario '{id}' in session '{preset.Id}'.",
|
|
"SIMULATION_SCENARIO_UNKNOWN");
|
|
scenarios.Add(scenario);
|
|
}
|
|
|
|
// All-or-nothing admission — check capacity before starting anything.
|
|
var activeCount = _runs.Values.Count(c =>
|
|
c.State.Status is SimulationRunStatus.Pending or SimulationRunStatus.Running);
|
|
if (activeCount + scenarios.Count > _options.MaxConcurrentRuns)
|
|
throw new ConflictException(
|
|
$"Session '{preset.Id}' needs {scenarios.Count} runs but only " +
|
|
$"{_options.MaxConcurrentRuns - activeCount} slot(s) remain " +
|
|
$"(max {_options.MaxConcurrentRuns}).",
|
|
"SIMULATION_CONCURRENCY_LIMIT");
|
|
|
|
var startedAt = DateTimeOffset.UtcNow;
|
|
var started = new List<SimulationRunState>(scenarios.Count);
|
|
|
|
try
|
|
{
|
|
for (var i = 0; i < scenarios.Count; i++)
|
|
{
|
|
if (i > 0)
|
|
await Task.Delay(SessionStagger, ct);
|
|
|
|
var state = await RegisterAndStartAsync(
|
|
scenarios[i], resolvedSpeed, userId, preset.Id, ct);
|
|
started.Add(state);
|
|
}
|
|
}
|
|
catch
|
|
{
|
|
foreach (var run in started)
|
|
Cancel(run.RunId);
|
|
throw;
|
|
}
|
|
|
|
return new SimulationSessionState
|
|
{
|
|
SessionId = preset.Id,
|
|
Name = preset.Name,
|
|
StartedAt = startedAt,
|
|
RunIds = started.Select(s => s.RunId).ToList(),
|
|
};
|
|
}
|
|
|
|
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<SimulationRunState> RegisterAndStartAsync(
|
|
ScenarioFile scenario,
|
|
double speed,
|
|
string startedByUserId,
|
|
string? sessionId,
|
|
CancellationToken ct)
|
|
{
|
|
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,
|
|
SessionId = sessionId,
|
|
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();
|
|
}
|
|
|
|
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,
|
|
SessionId = state.SessionId,
|
|
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.SessionId = state.SessionId;
|
|
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;
|
|
}
|
|
}
|