using Fake.EventBus.Distributed;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Oms.Application.Services.Orders.Issues;
namespace Oms.Application.Messaging.OrderSideEffects;
///
/// 优先 Fake.EventBus.RabbitMQ;发布失败则同进程同步执行 Handler。
///
public sealed class OrderSideEffectEventPublisher(
IDistributedEventBus eventBus,
IServiceScopeFactory scopeFactory,
ILogger logger) : IOrderSideEffectEventPublisher
{
public async Task PublishLifecycleTrackAppendAsync(
int orderId,
string orderNo,
string trackCode,
string? authorization = null,
CancellationToken cancellationToken = default)
{
if (orderId <= 0 || string.IsNullOrWhiteSpace(orderNo) || string.IsNullOrWhiteSpace(trackCode))
{
return;
}
var @event = new OrderLifecycleTrackAppendIntegrationEvent
{
OrderId = orderId,
OrderNo = orderNo.Trim(),
TrackCode = trackCode.Trim(),
Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim()
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
public async Task PublishAutoTriggerFireAsync(
int orderId,
AutoTriggerTiming timing,
int operatorUserId,
CancellationToken cancellationToken = default)
{
if (orderId <= 0)
{
return;
}
var @event = new OrderAutoTriggerFireIntegrationEvent
{
OrderId = orderId,
Timing = timing.ToString(),
OperatorUserId = operatorUserId
};
if (await TryPublishAsync(@event, cancellationToken))
{
return;
}
await ExecuteInScopeAsync(
h => h.HandleAsync(@event, cancellationToken),
cancellationToken);
}
private async Task TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken)
{
try
{
await eventBus.PublishAsync(@event, cancellationToken);
logger.LogDebug(
"Published side-effect event {EventName} Id={EventId}",
@event.GetType().Name,
@event.Id);
return true;
}
catch (Exception ex)
{
logger.LogWarning(
ex,
"Publish side-effect event {EventName} failed, fallback to sync. EventId={EventId}",
@event.GetType().Name,
@event.Id);
return false;
}
}
private async Task ExecuteInScopeAsync(
Func action,
CancellationToken cancellationToken)
where THandler : notnull
{
try
{
using var scope = scopeFactory.CreateScope();
var handler = scope.ServiceProvider.GetRequiredService();
await action(handler);
}
catch (Exception ex)
{
logger.LogWarning(ex, "Sync fallback for {Handler} failed", typeof(THandler).Name);
}
}
}