318 lines
13 KiB
C#
318 lines
13 KiB
C#
using System.Text.Json;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using StackExchange.Redis;
|
|
|
|
public class BatchService : IBatchService
|
|
{
|
|
private static readonly Dictionary<BatchStatus, HashSet<BatchStatus>> _allowedTransitions = new()
|
|
{
|
|
[BatchStatus.Uploaded] = new() { BatchStatus.InEntry, BatchStatus.Cancelled },
|
|
[BatchStatus.InEntry] = new() { BatchStatus.PendingVerification, BatchStatus.Cancelled },
|
|
[BatchStatus.PendingVerification] = new() { BatchStatus.Verified, BatchStatus.Rejected, BatchStatus.AwaitingClinicalApproval },
|
|
[BatchStatus.Rejected] = new() { BatchStatus.InEntry, BatchStatus.Cancelled },
|
|
[BatchStatus.Verified] = new() { BatchStatus.Approved },
|
|
[BatchStatus.AwaitingClinicalApproval] = new() { BatchStatus.Approved, BatchStatus.Rejected },
|
|
[BatchStatus.Approved] = new() { BatchStatus.Promoted },
|
|
[BatchStatus.Promoted] = new(),
|
|
[BatchStatus.Cancelled] = new(),
|
|
};
|
|
|
|
private readonly AppDbContext _db;
|
|
private readonly IDocumentStorageService _storage;
|
|
private readonly IConnectionMultiplexer _redis;
|
|
private readonly ILogger<BatchService> _logger;
|
|
|
|
public BatchService(AppDbContext db, IDocumentStorageService storage,
|
|
IConnectionMultiplexer redis, ILogger<BatchService> logger)
|
|
{
|
|
_db = db;
|
|
_storage = storage;
|
|
_redis = redis;
|
|
_logger = logger;
|
|
}
|
|
|
|
public async Task<CreateBatchResult> CreateAsync(
|
|
Stream fileStream, string contentType, BatchType batchType,
|
|
BatchTrack track, Guid? patientId, Guid? supersedesBatchId, Guid actorUserId)
|
|
{
|
|
// Supersession validation: target batch must exist and be in Promoted status
|
|
DigitizationBatch? supersededBatch = null;
|
|
if (supersedesBatchId.HasValue)
|
|
{
|
|
supersededBatch = await _db.DigitizationBatches
|
|
.FirstOrDefaultAsync(b => b.Id == supersedesBatchId.Value);
|
|
|
|
if (supersededBatch is null)
|
|
throw new NotFoundException(
|
|
$"Batch {supersedesBatchId.Value} not found.",
|
|
"SUPERSEDED_BATCH_NOT_FOUND");
|
|
|
|
if (supersededBatch.Status != BatchStatus.Promoted)
|
|
throw new ValidationException(
|
|
"Only promoted batches can be superseded. " +
|
|
$"Batch {supersedesBatchId.Value} is in '{supersededBatch.Status.ToDbString()}' status.",
|
|
"SUPERSEDED_BATCH_NOT_PROMOTED");
|
|
|
|
// Prevent supersession chains: the target batch must not itself be a correction
|
|
// that has already been superseded by another promoted correction
|
|
var existingCorrection = await _db.DigitizationBatches
|
|
.AnyAsync(b => b.SupersedesBatchId == supersedesBatchId.Value
|
|
&& b.Status == BatchStatus.Promoted);
|
|
|
|
if (existingCorrection)
|
|
throw new ConflictException(
|
|
$"Batch {supersedesBatchId.Value} has already been superseded by a promoted correction. " +
|
|
"Create a new correction against the latest promoted batch instead.",
|
|
"BATCH_ALREADY_SUPERSEDED");
|
|
|
|
// Inherit patient context from the superseded batch
|
|
patientId ??= supersededBatch.PatientId;
|
|
}
|
|
|
|
var (objectKey, sha256, fileSize) = await _storage.UploadAsync(fileStream, contentType, Guid.NewGuid());
|
|
|
|
// Duplicate detection: same SHA-256 for same patient within 24 hours
|
|
if (patientId.HasValue)
|
|
{
|
|
// Atomic Redis guard prevents concurrent uploads from racing past the DB check
|
|
var cache = _redis.GetDatabase();
|
|
var dedupKey = $"batch:dedup:{sha256}:{patientId.Value}";
|
|
var acquired = await cache.StringSetAsync(dedupKey, "1",
|
|
TimeSpan.FromHours(24), When.NotExists);
|
|
|
|
if (!acquired)
|
|
throw new ConflictException(
|
|
"A document with the same content was uploaded for this patient within the last 24 hours.",
|
|
"DUPLICATE_DOCUMENT");
|
|
|
|
// DB fallback for dedup entries created before Redis guard was deployed
|
|
var cutoff = DateTimeOffset.UtcNow.AddHours(-24);
|
|
var duplicate = await _db.DigitizationBatches.AnyAsync(b =>
|
|
b.DocumentSha256 == sha256 &&
|
|
b.PatientId == patientId.Value &&
|
|
b.CreatedAt >= cutoff);
|
|
|
|
if (duplicate)
|
|
{
|
|
await cache.KeyDeleteAsync(dedupKey);
|
|
throw new ConflictException(
|
|
"A document with the same content was uploaded for this patient within the last 24 hours.",
|
|
"DUPLICATE_DOCUMENT");
|
|
}
|
|
}
|
|
|
|
var batchId = Guid.NewGuid();
|
|
var batch = new DigitizationBatch
|
|
{
|
|
Id = batchId,
|
|
Status = BatchStatus.Uploaded,
|
|
BatchType = batchType,
|
|
Track = track,
|
|
PatientId = patientId,
|
|
DocumentRef = objectKey,
|
|
DocumentSha256 = sha256,
|
|
EnableRetroactiveAlerts = false,
|
|
SupersedesBatchId = supersedesBatchId,
|
|
CreatedAt = DateTimeOffset.UtcNow,
|
|
UpdatedAt = DateTimeOffset.UtcNow
|
|
};
|
|
|
|
var document = new ScannedDocument
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
BatchId = batchId,
|
|
ObjectKey = objectKey,
|
|
Sha256 = sha256,
|
|
ContentType = contentType,
|
|
FileSizeBytes = fileSize,
|
|
UploadedAt = DateTimeOffset.UtcNow
|
|
};
|
|
|
|
var eventType = supersedesBatchId.HasValue
|
|
? DigitizationEventType.CorrectionUploaded
|
|
: DigitizationEventType.Uploaded;
|
|
var eventMetadata = supersedesBatchId.HasValue
|
|
? JsonSerializer.Serialize(new { supersedesBatchId = supersedesBatchId.Value })
|
|
: null;
|
|
|
|
var evt = new DigitizationEvent
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
BatchId = batchId,
|
|
EventType = eventType,
|
|
ActorUserId = actorUserId,
|
|
OccurredAt = DateTimeOffset.UtcNow,
|
|
MetadataJson = eventMetadata
|
|
};
|
|
|
|
_db.DigitizationBatches.Add(batch);
|
|
_db.ScannedDocuments.Add(document);
|
|
_db.DigitizationEvents.Add(evt);
|
|
await _db.SaveChangesAsync();
|
|
|
|
_logger.LogInformation(
|
|
"Batch {BatchId} created (correction={IsCorrection}, supersedes={SupersedesBatchId})",
|
|
batchId, supersedesBatchId.HasValue, supersedesBatchId);
|
|
|
|
SupersessionInfo? supersession = null;
|
|
if (supersededBatch is not null)
|
|
{
|
|
var obsCount = await _db.Observations
|
|
.CountAsync(o => o.SourceBatchId == supersededBatch.Id);
|
|
supersession = BatchDetailResponse.ToSupersessionInfo(supersededBatch, obsCount);
|
|
}
|
|
|
|
return new CreateBatchResult(batch, supersession);
|
|
}
|
|
|
|
public async Task<DigitizationBatch> GetByIdAsync(Guid id)
|
|
{
|
|
var batch = await _db.DigitizationBatches
|
|
.Include(b => b.Document)
|
|
.Include(b => b.DraftPatient)
|
|
.Include(b => b.DraftEncounter)
|
|
.Include(b => b.DraftObservations)
|
|
.FirstOrDefaultAsync(b => b.Id == id);
|
|
|
|
if (batch is null)
|
|
throw new NotFoundException("Batch not found.", "BATCH_NOT_FOUND");
|
|
|
|
return batch;
|
|
}
|
|
|
|
private static readonly HashSet<string> _allowedSortFields = new(StringComparer.OrdinalIgnoreCase)
|
|
{
|
|
"createdAt", "updatedAt", "status", "batchType", "track"
|
|
};
|
|
|
|
public async Task<PagedResult<DigitizationBatch>> ListAsync(
|
|
BatchStatus? status, BatchType? batchType, Guid? assignedTo, BatchTrack? track,
|
|
int page, int pageSize, string sortBy = "createdAt", string sortDirection = "desc")
|
|
{
|
|
var query = _db.DigitizationBatches.AsQueryable();
|
|
|
|
if (status.HasValue)
|
|
query = query.Where(b => b.Status == status.Value);
|
|
if (batchType.HasValue)
|
|
query = query.Where(b => b.BatchType == batchType.Value);
|
|
if (assignedTo.HasValue)
|
|
query = query.Where(b => b.EnteredByUserId == assignedTo.Value);
|
|
if (track.HasValue)
|
|
query = query.Where(b => b.Track == track.Value);
|
|
|
|
if (!_allowedSortFields.Contains(sortBy))
|
|
throw new ValidationException(
|
|
$"Invalid sortBy field '{sortBy}'. Allowed: {string.Join(", ", _allowedSortFields)}.",
|
|
"INVALID_SORT_FIELD");
|
|
|
|
var total = await query.CountAsync();
|
|
var ordered = ApplySort(query, sortBy, sortDirection);
|
|
var items = await ordered
|
|
.Skip((page - 1) * pageSize)
|
|
.Take(pageSize)
|
|
.ToListAsync();
|
|
|
|
return new PagedResult<DigitizationBatch>(items, page, pageSize, total);
|
|
}
|
|
|
|
private static IOrderedQueryable<DigitizationBatch> ApplySort(
|
|
IQueryable<DigitizationBatch> query, string sortBy, string sortDirection)
|
|
{
|
|
var desc = sortDirection.Equals("desc", StringComparison.OrdinalIgnoreCase);
|
|
|
|
return sortBy.ToLowerInvariant() switch
|
|
{
|
|
"createdat" => desc ? query.OrderByDescending(b => b.CreatedAt) : query.OrderBy(b => b.CreatedAt),
|
|
"updatedat" => desc ? query.OrderByDescending(b => b.UpdatedAt) : query.OrderBy(b => b.UpdatedAt),
|
|
"status" => desc ? query.OrderByDescending(b => b.Status) : query.OrderBy(b => b.Status),
|
|
"batchtype" => desc ? query.OrderByDescending(b => b.BatchType) : query.OrderBy(b => b.BatchType),
|
|
"track" => desc ? query.OrderByDescending(b => b.Track) : query.OrderBy(b => b.Track),
|
|
_ => query.OrderByDescending(b => b.CreatedAt)
|
|
};
|
|
}
|
|
|
|
public async Task<DigitizationBatch> AssignAsync(Guid batchId, Guid entryClerkUserId, Guid actorUserId)
|
|
{
|
|
var batch = await _db.DigitizationBatches.FindAsync(batchId);
|
|
if (batch is null)
|
|
throw new NotFoundException("Batch not found.", "BATCH_NOT_FOUND");
|
|
|
|
if (batch.Status != BatchStatus.Uploaded)
|
|
throw new ConflictException(
|
|
"Only batches in 'uploaded' status can be assigned.",
|
|
"ILLEGAL_STATUS_TRANSITION");
|
|
|
|
// Redis lock to prevent double-assignment
|
|
var cache = _redis.GetDatabase();
|
|
var lockKey = $"batch:assign:{batchId}";
|
|
var acquired = await cache.StringSetAsync(lockKey, entryClerkUserId.ToString(),
|
|
TimeSpan.FromHours(1), When.NotExists);
|
|
|
|
if (!acquired)
|
|
throw new ConflictException(
|
|
"This batch is already assigned to another clerk.",
|
|
"BATCH_ALREADY_ASSIGNED");
|
|
|
|
batch.EnteredByUserId = entryClerkUserId;
|
|
batch.Status = BatchStatus.InEntry;
|
|
batch.UpdatedAt = DateTimeOffset.UtcNow;
|
|
|
|
_db.DigitizationEvents.Add(new DigitizationEvent
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
BatchId = batchId,
|
|
EventType = DigitizationEventType.EntryStarted,
|
|
ActorUserId = actorUserId,
|
|
OccurredAt = DateTimeOffset.UtcNow,
|
|
MetadataJson = JsonSerializer.Serialize(new { assignedTo = entryClerkUserId })
|
|
});
|
|
|
|
await _db.SaveChangesAsync();
|
|
return batch;
|
|
}
|
|
|
|
public async Task<DigitizationBatch> CancelAsync(Guid batchId, string reason, Guid actorUserId)
|
|
{
|
|
var batch = await _db.DigitizationBatches.FindAsync(batchId);
|
|
if (batch is null)
|
|
throw new NotFoundException("Batch not found.", "BATCH_NOT_FOUND");
|
|
|
|
if (!_allowedTransitions.TryGetValue(batch.Status, out var allowed)
|
|
|| !allowed.Contains(BatchStatus.Cancelled))
|
|
throw new ConflictException(
|
|
$"Batch in '{batch.Status.ToDbString()}' status cannot be cancelled.",
|
|
"ILLEGAL_STATUS_TRANSITION");
|
|
|
|
var previousStatus = batch.Status.ToDbString();
|
|
batch.Status = BatchStatus.Cancelled;
|
|
batch.UpdatedAt = DateTimeOffset.UtcNow;
|
|
|
|
_db.DigitizationEvents.Add(new DigitizationEvent
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
BatchId = batchId,
|
|
EventType = DigitizationEventType.Cancelled,
|
|
ActorUserId = actorUserId,
|
|
OccurredAt = DateTimeOffset.UtcNow,
|
|
MetadataJson = JsonSerializer.Serialize(new { previousStatus, reason })
|
|
});
|
|
|
|
// Release Redis assignment lock if one exists
|
|
var cache = _redis.GetDatabase();
|
|
var lockKey = $"batch:assign:{batchId}";
|
|
await cache.KeyDeleteAsync(lockKey);
|
|
|
|
await _db.SaveChangesAsync();
|
|
|
|
_logger.LogInformation(
|
|
"Batch {BatchId} cancelled by {ActorUserId} from status {PreviousStatus}. Reason: {Reason}",
|
|
batchId, actorUserId, previousStatus, reason);
|
|
|
|
return batch;
|
|
}
|
|
|
|
public BatchStatus[] GetAllowedTransitions(BatchStatus current) =>
|
|
_allowedTransitions.TryGetValue(current, out var targets)
|
|
? targets.ToArray()
|
|
: Array.Empty<BatchStatus>();
|
|
} |