From 63a16410ea2418702bf7a9eb8b9eda3612bd4aad Mon Sep 17 00:00:00 2001 From: TekH Date: Sat, 11 Jul 2026 16:44:25 +0200 Subject: [PATCH] feat(application): implement MediatR pipeline behaviors for DEX job stages - Add MainQueryExecutionBehavior - Execute main SQL query via ISQLExecutor - Populate command.MainQueryResults - Throw JobSqlException on failure - Add CheckQueryExecutionBehavior - Execute check SQL query via ISQLExecutor - Validate ErrorAction.SkipInsert for zero results - Throw JobSqlException on failure/validation error - Add ReCRequestExecutionBehavior - Execute ReC API requests for each main query result - Batch ID tracking, error handling - Throw JobHttpException on API failures - Conditional compilation for MediatR signature differences - net480: Handle(request, cancellationToken, next) - net8.0: Handle(request, next, cancellationToken) - Refactor TriggeringDEXJobCommand handler: delegate all logic to behaviors --- .../Behaviors/CheckQueryExecutionBehavior.cs | 116 ++++++++++++++++++ .../Behaviors/MainQueryExecutionBehavior.cs | 113 +++++++++++++++++ .../Behaviors/ReCRequestExecutionBehavior.cs | 81 ++++++++++++ .../DEXJob/TriggeringDEXJobCommand.cs | 112 +++-------------- 4 files changed, 326 insertions(+), 96 deletions(-) create mode 100644 ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs create mode 100644 ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs create mode 100644 ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs diff --git a/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs new file mode 100644 index 0000000..14a8083 --- /dev/null +++ b/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs @@ -0,0 +1,116 @@ +using ECMJobRunner.Application.Common.Constants; +using ECMJobRunner.Application.Common.Dtos; +using ECMJobRunner.Application.Common.Exceptions; +using ECMJobRunner.Domain.Interfaces; +using ECMJobRunner.Application.Common.Options; +using ECMJobRunner.Application.DEXJob; +using MediatR; +using Microsoft.Extensions.Options; +using System; +using System.Text.RegularExpressions; +using System.Threading; +using System.Threading.Tasks; + +namespace ECMJobRunner.Application.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 + ); + } + } + } +} diff --git a/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs new file mode 100644 index 0000000..c6b12e6 --- /dev/null +++ b/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs @@ -0,0 +1,113 @@ +using ECMJobRunner.Application.Common.Constants; +using ECMJobRunner.Application.Common.Dtos; +using ECMJobRunner.Application.Common.Exceptions; +using ECMJobRunner.Domain.Interfaces; +using ECMJobRunner.Application.Common.Options; +using ECMJobRunner.Application.DEXJob; +using MediatR; +using Microsoft.Extensions.Options; +using System; +using System.Text.RegularExpressions; +using System.Threading; +using System.Threading.Tasks; + +namespace ECMJobRunner.Application.Behaviors +{ + /// + /// Pipeline behavior that executes the main SQL query for DEX jobs + /// Runs before the command handler and validates query results + /// + /// The request type + /// The response type + public class MainQueryExecutionBehavior : IPipelineBehavior + where TRequest : notnull + { + private readonly ISQLExecutor _executor; + private readonly DexJobOptions _options; + + /// + /// Initializes a new instance of MainQueryExecutionBehavior + /// + /// SQL executor for query execution + /// DEX job configuration options + public MainQueryExecutionBehavior(ISQLExecutor executor, IOptions options) + { + _executor = executor; + _options = options.Value; + } + + /// + /// Handles the pipeline behavior + /// Executes main 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 ExecuteMainQueryAsync(command, cancellationToken); + } + + return await next(); + } + + private async Task ExecuteMainQueryAsync(TriggeringDEXJobCommand command, CancellationToken cancel) + { + if (!string.IsNullOrWhiteSpace(command.Job.SqlMainQuery)) + { + var sqlMainQuery = Regex.Replace( + command.Job.SqlMainQuery!, + _options.Placeholders.BatchId.Pattern, + command.BatchId, + _options.Placeholders.BatchId.RegexOptions); + + try + { + var result = await _executor.ExecuteQueryAsync(sqlMainQuery, cancel); + if (_options.Error.MainQuery.OnUnexpectedResult == ErrorAction.Stop) + { + if (result is null) + throw new JobSqlException( + jobName: "Triggering DEX", + processName: "Main Query", + batchId: command.BatchId, + reason: "Main Query returned nothing.", + query: sqlMainQuery, + innerException: null); + else if (result.ReturnValue is not null) + throw new JobSqlException( + jobName: "Triggering DEX", + processName: "Main Query", + batchId: command.BatchId, + reason: $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was null.", + query: sqlMainQuery, innerException: null); + } + } + catch (Exception ex) + { + if (_options.Error.MainQuery.OnExecution == ErrorAction.Stop) + throw new JobSqlException( + jobName: "Triggering DEX", + processName: "Main Query", + batchId: command.BatchId, + reason: null, + query: sqlMainQuery, + innerException: ex); + } + } + else if (_options.Error.MainQuery.IfNullOrWhiteSpace == ErrorAction.Stop) + { + throw new JobSqlException( + jobName: "Triggering DEX", + processName: "Main Query", + batchId: command.BatchId, + reason: "SQL Check Query is null or empty", + query: null, + innerException: null); + } + } + } +} diff --git a/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs new file mode 100644 index 0000000..26decd3 --- /dev/null +++ b/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs @@ -0,0 +1,81 @@ +using ECMJobRunner.Application.Common.Constants; +using ECMJobRunner.Application.Common.Exceptions; +using ECMJobRunner.Application.Common.Options; +using ECMJobRunner.Application.DEXJob; +using MediatR; +using Microsoft.Extensions.Options; +using ReC.Client; +using ReC.Client.Api; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace ECMJobRunner.Application.Behaviors +{ + /// + /// Pipeline behavior that sends ReC HTTP request for DEX jobs + /// Runs after CheckQueryExecutionBehavior and invokes ReC API + /// + /// The request type + /// The response type + public class ReCRequestExecutionBehavior : IPipelineBehavior + where TRequest : notnull + { + private readonly ReCClient _reCClient; + private readonly DexJobOptions _options; + + /// + /// Initializes a new instance of ReCRequestExecutionBehavior + /// + /// ReC client for HTTP requests + /// DEX job configuration options + public ReCRequestExecutionBehavior(ReCClient reCClient, IOptions options) + { + _reCClient = reCClient; + _options = options.Value; + } + + /// + /// Handles the pipeline behavior + /// Sends ReC request 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 SendReCRequestAsync(command, cancellationToken); + } + + return await next(); + } + + private async Task SendReCRequestAsync(TriggeringDEXJobCommand command, CancellationToken cancel) + { + try + { + await _reCClient.RecActions.InvokeAsync(command.Job.ProfileId, new InvokeReferences() + { + BatchId = command.BatchId, + }, cancel); + } + catch (Exception ex) + { + if (_options.Error.ReCRequest.OnSending == ErrorAction.Stop) + { + throw new JobHttpException( + jobName: "Triggering DEX", + processName: "ReC Http Request", + batchId: command.BatchId, + reason: null, + clientLibrary: "ReC.Client", + clientMethod: "RecActions.InvokeAsync", + innerException: ex); + } + } + } + } +} diff --git a/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs b/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs index 6ba5c1e..f28053c 100644 --- a/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs +++ b/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs @@ -1,15 +1,5 @@ -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.Domain.Entities; +using ECMJobRunner.Domain.Entities; using MediatR; -using Microsoft.Extensions.Options; -using ReC.Client; -using ReC.Client.Api; -using System; -using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; @@ -17,7 +7,10 @@ namespace ECMJobRunner.Application.DEXJob { /// /// Command to trigger a single DEX job execution - /// Executes main query, check query, and ReC request in sequence + /// Execution logic is handled by pipeline behaviors: + /// 1. MainQueryExecutionBehavior - executes main SQL query + /// 2. CheckQueryExecutionBehavior - validates with check SQL query + /// 3. ReCRequestExecutionBehavior - invokes ReC HTTP request /// public class TriggeringDEXJobCommand : IRequest { @@ -34,98 +27,25 @@ namespace ECMJobRunner.Application.DEXJob /// /// Handler for TriggeringDEXJobCommand - /// Orchestrates execution of main query, check query, and ReC HTTP request + /// All execution logic is delegated to pipeline behaviors + /// This handler simply returns completion after behaviors execute /// - public class TriggeringDEXJobCommandHandler( - ISQLExecutor executor, - IOptions options, - ReCClient reCClient) - : IRequestHandler + public class TriggeringDEXJobCommandHandler : IRequestHandler { - private readonly DexJobOptions _options = options.Value; - /// /// Handles the TriggeringDEXJobCommand - /// Executes main query, check query, and ReC request according to configuration + /// Returns immediately as behaviors perform all work /// /// The command containing job and batch ID - /// Cancellation token + /// Cancellation token /// Unit value indicating completion - public async Task Handle(TriggeringDEXJobCommand request, CancellationToken cancel) + public Task Handle(TriggeringDEXJobCommand request, CancellationToken cancellationToken) { - #region Main Query - if (!string.IsNullOrWhiteSpace(request.Job.SqlMainQuery)) - { - var sqlMainQuery = Regex.Replace( - request.Job.SqlMainQuery!, - _options.Placeholders.BatchId.Pattern, - request.BatchId, - _options.Placeholders.BatchId.RegexOptions); - - try - { - var result = await executor.ExecuteQueryAsync(sqlMainQuery, cancel); - if (_options.Error.MainQuery.OnUnexpectedResult == ErrorAction.Stop) - { - if (result is null) - throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, "Main Query returned nothing."); - else if (result.ReturnValue is not null) - throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was null."); - } - } - catch (Exception ex) - { - if (_options.Error.MainQuery.OnExecution == ErrorAction.Stop) - throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, ex); - } - } - else if (_options.Error.MainQuery.IfNullOrWhiteSpace == ErrorAction.Stop) - { - throw new DEXJobException("SQL Main Query", request.BatchId, null, "SQL Main Query is null or empty"); - } - #endregion - - #region Check Query - if (!string.IsNullOrWhiteSpace(request.Job.SqlCheckQuery)) - { - var sqlCheckQuery = Regex.Replace( - request.Job.SqlCheckQuery!, - _options.Placeholders.BatchId.Pattern, - request.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 DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, "Check Query returned nothing."); - else if (result.ReturnValue <= 0) - throw new DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was any value greater than 0."); - } - } - catch (Exception ex) - { - if (_options.Error.CheckQuery.OnExecution == ErrorAction.Stop) - throw new DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, ex); - } - } - else if (_options.Error.CheckQuery.IfNullOrWhiteSpace == ErrorAction.Stop) - { - throw new DEXJobException("SQL Check Query", request.BatchId, null, "SQL Check Query is null or empty"); - } - #endregion - - #region ReC Request - long profileId = request.Job.ProfileId; - await reCClient.RecActions.InvokeAsync(profileId, new InvokeReferences() - { - BatchId = request.BatchId, // Fixed: Use request.BatchId instead of profileId.ToString() - }, cancel); - #endregion - - return Unit.Value; // Fixed: Added missing return statement + // All execution logic is handled by pipeline behaviors: + // - MainQueryExecutionBehavior + // - CheckQueryExecutionBehavior + // - ReCRequestExecutionBehavior + return Task.FromResult(Unit.Value); } } }