[FA-misc] Add JobStatusUpdateConsumer with upsert logic
This commit is contained in:
@@ -0,0 +1,66 @@
|
||||
using FictionArchive.Service.ReportingService.Models;
|
||||
using FictionArchive.Service.ReportingService.Services;
|
||||
using FictionArchive.Service.Shared.Contracts.Events;
|
||||
using MassTransit;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace FictionArchive.Service.ReportingService.Consumers;
|
||||
|
||||
public class JobStatusUpdateConsumer : IConsumer<IJobStatusUpdate>
|
||||
{
|
||||
private readonly ILogger<JobStatusUpdateConsumer> _logger;
|
||||
private readonly ReportingDbContext _dbContext;
|
||||
|
||||
public JobStatusUpdateConsumer(
|
||||
ILogger<JobStatusUpdateConsumer> logger,
|
||||
ReportingDbContext dbContext)
|
||||
{
|
||||
_logger = logger;
|
||||
_dbContext = dbContext;
|
||||
}
|
||||
|
||||
public async Task Consume(ConsumeContext<IJobStatusUpdate> context)
|
||||
{
|
||||
var message = context.Message;
|
||||
|
||||
var existingJob = await _dbContext.Jobs.FirstOrDefaultAsync(j => j.Id == message.JobId);
|
||||
|
||||
if (existingJob == null)
|
||||
{
|
||||
var job = new Job
|
||||
{
|
||||
Id = message.JobId,
|
||||
ParentJobId = message.ParentJobId,
|
||||
JobType = message.JobType,
|
||||
DisplayName = message.DisplayName,
|
||||
Status = message.Status,
|
||||
ErrorMessage = message.ErrorMessage,
|
||||
Metadata = message.Metadata != null
|
||||
? new Dictionary<string, string>(message.Metadata)
|
||||
: null
|
||||
};
|
||||
|
||||
_dbContext.Jobs.Add(job);
|
||||
_logger.LogInformation("Created job {JobId} of type {JobType}", message.JobId, message.JobType);
|
||||
}
|
||||
else
|
||||
{
|
||||
existingJob.Status = message.Status;
|
||||
existingJob.DisplayName = message.DisplayName;
|
||||
existingJob.ErrorMessage = message.ErrorMessage;
|
||||
|
||||
if (message.Metadata != null)
|
||||
{
|
||||
existingJob.Metadata ??= new Dictionary<string, string>();
|
||||
foreach (var kvp in message.Metadata)
|
||||
{
|
||||
existingJob.Metadata[kvp.Key] = kvp.Value;
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("Updated job {JobId} to status {Status}", message.JobId, message.Status);
|
||||
}
|
||||
|
||||
await _dbContext.SaveChangesAsync();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user