using ECMJobRunner.Application.Common.Constants;
using ECMJobRunner.Application.Common.Dtos;
using ECMJobRunner.Application.Common.Exceptions;
using ECMJobRunner.Application.Common.Interfaces;
using ECMJobRunner.Application.Common.Options;
using ECMJobRunner.Application.Profiles.Commands;
using MediatR;
using Microsoft.Extensions.Options;
using System;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
namespace ECMJobRunner.Application.Profiles.Commands.Behaviors
{
///
/// Pipeline behavior that executes the check SQL query for DEX jobs
/// Runs after MainQueryExecutionBehavior and validates query results
///
/// The request type
/// The response type
public class CheckQueryExecutionBehavior : IPipelineBehavior
where TRequest : notnull
{
private readonly ISQLExecutor _executor;
private readonly DexJobOptions _options;
///
/// Initializes a new instance of CheckQueryExecutionBehavior
///
/// SQL executor for query execution
/// DEX job configuration options
public CheckQueryExecutionBehavior(ISQLExecutor executor, IOptions options)
{
_executor = executor;
_options = options.Value;
}
///
/// Handles the pipeline behavior
/// Executes check query if request is TriggeringDEXJobCommand
///
#if NET48
public async Task Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate next)
#else
public async Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken)
#endif
{
if (request is TriggeringProfileJobCommand command)
{
await ExecuteCheckQueryAsync(command, cancellationToken);
}
return await next();
}
private async Task ExecuteCheckQueryAsync(TriggeringProfileJobCommand command, CancellationToken cancel)
{
if (!string.IsNullOrWhiteSpace(command.Job.SqlCheckQuery))
{
var sqlCheckQuery = Regex.Replace(
command.Job.SqlCheckQuery!,
_options.Placeholders.BatchId.Pattern,
command.BatchId,
_options.Placeholders.BatchId.RegexOptions);
try
{
var result = await _executor.ExecuteQueryAsync(sqlCheckQuery, cancel);
if (_options.Error.CheckQuery.OnUnexpectedResult == ErrorAction.Stop)
{
if (result is null)
throw new JobSqlException(
profileId: command.Job.ProfileId,
jobName: "Triggering DEX",
processName: "Check Query",
batchId: command.BatchId,
reason: "Check Query returned nothing.",
query: sqlCheckQuery,
innerException: null);
else if (result.ReturnValue <= 0)
throw new JobSqlException(
profileId: command.Job.ProfileId,
jobName: "Triggering DEX",
processName: "Check Query",
batchId: command.BatchId,
reason: $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was any value greater than 0.",
query: sqlCheckQuery,
innerException: null);
}
}
catch (Exception ex)
{
if (_options.Error.CheckQuery.OnExecution == ErrorAction.Stop)
throw new JobSqlException(
profileId: command.Job.ProfileId,
jobName: "Triggering DEX",
processName: "Check Query",
batchId: command.BatchId,
reason: null,
query: sqlCheckQuery,
innerException: ex
);
}
}
else if (_options.Error.CheckQuery.IfNullOrWhiteSpace == ErrorAction.Stop)
{
throw new JobSqlException(
profileId: command.Job.ProfileId,
jobName:"Triggering DEX",
processName:"Check Query",
batchId:command.BatchId,
reason: "SQL Check Query is null or empty",
query: null,
innerException: null
);
}
}
}
}