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.DEXJob.Commands; using MediatR; using Microsoft.Extensions.Options; using System; using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; namespace ECMJobRunner.Application.DEXJob.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 TriggeringDEXJobCommand command) { await ExecuteCheckQueryAsync(command, cancellationToken); } return await next(); } private async Task ExecuteCheckQueryAsync(TriggeringDEXJobCommand 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( 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( 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( 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( jobName:"Triggering DEX", processName:"Check Query", batchId:command.BatchId, reason: "SQL Check Query is null or empty", query: null, innerException: null ); } } } }