admin/src/Oms.Application/Services/Orders/Batch/OrderBatchJobProcessor.cs

using Fake.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Oms.Application.Dtos.Orders;
using Oms.Application.Messaging;
using Oms.Domain.OrdersAggregate.Consts;
using Oms.Domain.OrdersAggregate.Entities;
using Oms.Domain.OrdersAggregate.Repositories;

namespace Oms.Application.Services.Orders.Batch;

/// <summary>
/// 消费 RabbitMQ 消息后更新任务状态并执行业务。
/// 正常返回由 Fake EventBus Ack；系统失败可重试，耗尽后标记失败避免子项长期停留在执行中。
/// </summary>
public sealed class OrderBatchJobProcessor(
    IOrderBatchJobRepository batchJobRepository,
    IOrderRepository orderRepository,
    IOrderBatchActionExecutor actionExecutor,
    IOptions<OrderBatchRabbitMqOptions> batchOptions,
    ILogger<OrderBatchJobProcessor> logger) : IScopedDependency
{
    public async Task ProcessAsync(
        OrderBatchJobMessage message,
        CancellationToken cancellationToken = default)
    {
        var maxSystemRetries = Math.Max(1, batchOptions.Value.MaxSystemRetries);
        var item = await ResolveJobItemAsync(message, cancellationToken);
        if (item != null)
        {
            if (item.Status == OrderBatchJobItemStatusCodes.Success)
            {
                return;
            }

            if (item.Status == OrderBatchJobItemStatusCodes.Cancelled)
            {
                logger.LogInformation("Skip cancelled job item {JobItemId}", item.Id);
                return;
            }

            item.Status = OrderBatchJobItemStatusCodes.Running;
            item.StartedAt ??= DateTime.Now;
            item.UpdateTime = DateTime.Now;
            await batchJobRepository.UpdateItemAsync(item, cancellationToken);
        }

        var result = await actionExecutor.ExecuteAsync(message, cancellationToken);

        if (item == null)
        {
            if (!result.Success)
            {
                logger.LogWarning(
                    "Fire-and-forget job failed action={Action} orderId={OrderId} reason={Reason}",
                    message.Action,
                    message.OrderId,
                    result.FailReason);
            }

            return;
        }

        item = await batchJobRepository.GetItemByIdAsync(item.Id, cancellationToken);
        if (item == null)
        {
            logger.LogWarning(
                "Job item {JobItemId} not found after execute, ack message {MessageId}",
                message.JobItemId,
                message.MessageId);
            return;
        }

        item.FinishedAt = DateTime.Now;
        item.UpdateTime = DateTime.Now;
        var action = string.IsNullOrWhiteSpace(item.Action) ? message.Action : item.Action;

        if (result.Success)
        {
            OrderBatchJobItemFeedback.ApplySuccess(item, result, action);
            await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
            await batchJobRepository.UpdateItemAsync(item, cancellationToken);
            await RefreshJobSummaryAsync(item.JobId, cancellationToken);
            return;
        }

        if (result.IsBusinessFailure)
        {
            if (result.CreatedOrderId is > 0)
            {
                item.OrderId = result.CreatedOrderId.Value;
            }

            await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
            OrderBatchJobItemFeedback.ApplyBusinessFailure(item, result, action);
            await batchJobRepository.UpdateItemAsync(item, cancellationToken);

            await RefreshJobSummaryAsync(item.JobId, cancellationToken);
            return;
        }

        await OrderBatchJobItemFeedback.EnrichIdentifiersAsync(item, action, orderRepository, cancellationToken);
        OrderBatchJobItemFeedback.ApplySystemRetry(item, result, action, maxSystemRetries);
        await batchJobRepository.UpdateItemAsync(item, cancellationToken);

        if (item.Status == OrderBatchJobItemStatusCodes.Failed)
        {
            logger.LogError(
                "Batch job item {JobItemId} action={Action} failed after {RetryCount} system retries: {Reason}",
                item.Id,
                action,
                item.RetryCount,
                item.FailReason);
            await RefreshJobSummaryAsync(item.JobId, cancellationToken);
            return;
        }

        throw new InvalidOperationException(item.FailReason ?? result.FailReason ?? "批量任务系统失败");
    }

    private async Task<OrderBatchJobItem?> ResolveJobItemAsync(
        OrderBatchJobMessage message,
        CancellationToken cancellationToken)
    {
        if (message.JobItemId > 0)
        {
            return await batchJobRepository.GetItemByIdAsync((int)message.JobItemId, cancellationToken);
        }

        if (message.JobId > 0 && !string.IsNullOrWhiteSpace(message.MessageId))
        {
            var resolved = await batchJobRepository.GetItemByJobAndMessageIdAsync(
                (int)message.JobId,
                message.MessageId.Trim(),
                cancellationToken);
            if (resolved != null)
            {
                logger.LogWarning(
                    "Resolved job item by messageId jobId={JobId} messageId={MessageId} itemId={JobItemId}",
                    message.JobId,
                    message.MessageId,
                    resolved.Id);
            }

            return resolved;
        }

        return null;
    }

    private async Task RefreshJobSummaryAsync(int jobId, CancellationToken cancellationToken)
    {
        var job = await batchJobRepository.GetByIdAsync(jobId, cancellationToken);
        if (job == null)
        {
            return;
        }

        var items = await batchJobRepository.GetItemsByJobIdAsync(jobId, cancellationToken);
        job.SuccessCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Success);
        job.FailCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Failed);
        job.UpdateTime = DateTime.Now;

        if (items.All(x => x.Status is OrderBatchJobItemStatusCodes.Success
                or OrderBatchJobItemStatusCodes.Failed
                or OrderBatchJobItemStatusCodes.Cancelled))
        {
            var cancelledCount = items.Count(x => x.Status == OrderBatchJobItemStatusCodes.Cancelled);
            if (cancelledCount == items.Count)
            {
                job.Status = OrderBatchJobStatusCodes.Cancelled;
            }
            else
            {
                job.Status = job.FailCount > 0 && job.SuccessCount == 0
                    ? OrderBatchJobStatusCodes.Failed
                    : OrderBatchJobStatusCodes.Completed;
            }

            job.FinishedAt = DateTime.Now;
        }
        else
        {
            job.Status = OrderBatchJobStatusCodes.Running;
        }

        await batchJobRepository.UpdateJobAsync(job, cancellationToken);
    }
}