admin/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs

using System.Data;
using Fake.EntityFrameworkCore;
using Fake.EventBus.Distributed;
using Fake.EventBus.IntegrationEventLog;
using Fake.UnitOfWork;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

namespace Oms.Application.Messaging;

/// <summary>
/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。
/// SqlSugar 与 EF 默认各持连接；若环境事务已由 SqlSugar 占用，则改用 RequiresNew 短事务只写 Outbox。
/// </summary>
internal static class OmsIntegrationEventPublisher
{
    public static async Task PublishAsync(
        IDistributedEventBus eventBus,
        IUnitOfWorkManager unitOfWorkManager,
        ILogger logger,
        IntegrationEvent @event,
        CancellationToken cancellationToken = default)
    {
        if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager))
        {
            try
            {
                await EnsureOutboxDbContextAsync(unitOfWorkManager);
                await eventBus.PublishAsync(@event, cancellationToken);
                logger.LogDebug(
                    "Outbox enqueued {EventName} Id={EventId} (ambient TX)",
                    @event.GetType().Name,
                    @event.Id);
                return;
            }
            catch (Exception ex) when (IsTxConnectionMismatch(ex))
            {
                // SqlSugar 已在另一条连接上开事务时，EF 无法征用该 DbTransaction。
                logger.LogWarning(
                    ex,
                    "Ambient UoW TX incompatible with EF Outbox; using short outbox TX. Event={EventName} Id={EventId}",
                    @event.GetType().Name,
                    @event.Id);
            }
        }
        else
        {
            logger.LogWarning(
                "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX",
                @event.GetType().Name,
                @event.Id);
        }

        using var uow = unitOfWorkManager.Begin(new UnitOfWorkAttribute
        {
            IsTransactional = true,
            RequiresNew = true,
            IsolationLevel = IsolationLevel.ReadCommitted
        });
        await EnsureOutboxDbContextAsync(unitOfWorkManager);
        await eventBus.PublishAsync(@event, cancellationToken);
        await uow.CompleteAsync(cancellationToken);
    }

    public static async Task<bool> TryPublishAsync(
        IDistributedEventBus eventBus,
        IUnitOfWorkManager unitOfWorkManager,
        ILogger logger,
        IntegrationEvent @event,
        CancellationToken cancellationToken = default)
    {
        try
        {
            await PublishAsync(eventBus, unitOfWorkManager, logger, @event, cancellationToken);
            return true;
        }
        catch (Exception ex)
        {
            logger.LogError(
                ex,
                "Outbox publish failed {EventName} Id={EventId}",
                @event.GetType().Name,
                @event.Id);
            return false;
        }
    }

    /// <summary>
    /// 业务写库前先打开 Outbox DbContext，尽量让 EF 先占住环境事务连接，供 SqlSugar 征用。
    /// </summary>
    public static Task EnsureOutboxDbContextAsync(IUnitOfWorkManager unitOfWorkManager)
    {
        var current = unitOfWorkManager.Current
            ?? throw new InvalidOperationException("No ambient unit of work");
        var dbProvider = current.ServiceProvider
            .GetRequiredService<IEfDbContextProvider<IntegrationEventLogContext>>();
        return dbProvider.GetDbContextAsync();
    }

    private static bool IsTxConnectionMismatch(Exception ex)
    {
        for (var e = ex; e != null; e = e.InnerException!)
        {
            if (e.Message.Contains("not associated with the current connection", StringComparison.OrdinalIgnoreCase)
                || e.Message.Contains("associated with the current connection", StringComparison.OrdinalIgnoreCase))
            {
                return true;
            }
        }

        return false;
    }
}