diff --git a/Directory.Build.props b/Directory.Build.props index 9c939ac..7f7b084 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,6 +1,6 @@ - 8.0.5.18 + 8.0.5.19 diff --git a/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs b/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs index d9bc676..58cff86 100644 --- a/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs +++ b/src/Oms.Application/Dtos/Orders/OrderUpdateInput.cs @@ -25,6 +25,12 @@ public class OrderUpdateInput [StringLength(64)] public string? TrackNo { get; set; } + /// + /// 服务商单号。 + /// + [StringLength(64)] + public string? DownstreamServiceNo { get; set; } + /// /// 渠道编码。 /// diff --git a/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs new file mode 100644 index 0000000..0f12f49 --- /dev/null +++ b/src/Oms.Application/Messaging/OmsIntegrationEventPublisher.cs @@ -0,0 +1,68 @@ +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; + +/// +/// 可靠发布 IntegrationEvent：事务 UoW 内写入 Outbox（由 Fake.EventBus.IntegrationEventLog 投递）。 +/// +internal static class OmsIntegrationEventPublisher +{ + public static async Task PublishAsync( + IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, + ILogger logger, + IntegrationEvent @event, + CancellationToken cancellationToken = default) + { + if (OmsOutboxUnitOfWork.IsAmbientTransactional(unitOfWorkManager)) + { + await eventBus.PublishAsync(@event, cancellationToken); + logger.LogDebug( + "Outbox enqueued {EventName} Id={EventId} (ambient TX)", + @event.GetType().Name, + @event.Id); + return; + } + + // 无环境事务时开启短事务只写 Outbox，并告警：业务落库与事件非原子。 + logger.LogWarning( + "Publish {EventName} Id={EventId} without ambient transactional UoW; opening short outbox TX", + @event.GetType().Name, + @event.Id); + + using var uow = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional()); + // 先打开 Outbox DbContext 以产生可征用的 DbTransaction + var dbProvider = uow.ServiceProvider.GetRequiredService>(); + await dbProvider.GetDbContextAsync(); + await eventBus.PublishAsync(@event, cancellationToken); + await uow.CompleteAsync(cancellationToken); + } + + public static async Task 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; + } + } +} diff --git a/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs b/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs new file mode 100644 index 0000000..dd1c83b --- /dev/null +++ b/src/Oms.Application/Messaging/OmsOutboxUnitOfWork.cs @@ -0,0 +1,22 @@ +using System.Data; +using Fake.UnitOfWork; + +namespace Oms.Application.Messaging; + +/// +/// 集成事件 Outbox 所需的事务性工作单元约定。 +/// +public static class OmsOutboxUnitOfWork +{ + /// + /// 创单/后置事件发布使用的事务 UoW 属性（MySQL ReadCommitted）。 + /// + public static UnitOfWorkAttribute Transactional() => new() + { + IsTransactional = true, + IsolationLevel = IsolationLevel.ReadCommitted + }; + + public static bool IsAmbientTransactional(IUnitOfWorkManager unitOfWorkManager) => + unitOfWorkManager.Current?.Context.IsTransactional == true; +} diff --git a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs index 580e23c..95b3ad4 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobEventHandler.cs @@ -29,7 +29,8 @@ public sealed class OrderBatchJobEventHandler( return; } - using var unitOfWork = unitOfWorkManager.Begin(); + // ForecastBilling 成功后继续写轨迹/面单/自动触发 Outbox，须事务性 UoW + using var unitOfWork = unitOfWorkManager.Begin(OmsOutboxUnitOfWork.Transactional()); await processor.ProcessAsync(@event.ToMessage(), cancellationToken); await unitOfWork.CompleteAsync(cancellationToken); } diff --git a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs index c066098..a007ac4 100644 --- a/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs +++ b/src/Oms.Application/Messaging/OrderBatchJobPublisher.cs @@ -1,14 +1,16 @@ using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Oms.Application.Messaging; /// -/// 通过 Fake 分布式事件总线发布订单批量任务。 +/// 通过 Fake 分布式事件总线（Outbox）发布订单批量任务。 /// public sealed class OrderBatchJobPublisher( IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, IOptions options, ILogger logger) : IOrderBatchJobPublisher { @@ -22,10 +24,11 @@ public sealed class OrderBatchJobPublisher( } var @event = OrderBatchJobIntegrationEvent.FromMessage(message); - await eventBus.PublishAsync(@event, cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); logger.LogDebug( - "Published order batch event {EventId} messageId={MessageId} action={Action}", + "Outbox enqueued order batch event {EventId} messageId={MessageId} action={Action}", @event.Id, message.MessageId, message.Action); diff --git a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs index 078be98..9322fc4 100644 --- a/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs +++ b/src/Oms.Application/Messaging/OrderSideEffects/OrderSideEffectEventPublisher.cs @@ -1,16 +1,16 @@ using Fake.EventBus.Distributed; -using Microsoft.Extensions.DependencyInjection; +using Fake.UnitOfWork; using Microsoft.Extensions.Logging; using Oms.Application.Services.Orders.Issues; namespace Oms.Application.Messaging.OrderSideEffects; /// -/// 优先 Fake.EventBus.RabbitMQ；发布失败则同进程同步执行 Handler。 +/// 订单旁路副作用事件：事务内写 Outbox，Commit 后投递 RabbitMQ。 /// public sealed class OrderSideEffectEventPublisher( IDistributedEventBus eventBus, - IServiceScopeFactory scopeFactory, + IUnitOfWorkManager unitOfWorkManager, ILogger logger) : IOrderSideEffectEventPublisher { public async Task PublishLifecycleTrackAppendAsync( @@ -33,14 +33,8 @@ public sealed class OrderSideEffectEventPublisher( Authorization = string.IsNullOrWhiteSpace(authorization) ? null : authorization.Trim() }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishAutoTriggerFireAsync( @@ -61,52 +55,7 @@ public sealed class OrderSideEffectEventPublisher( 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); - } + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } } diff --git a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs index e265284..cf35e80 100644 --- a/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs +++ b/src/Oms.Application/Messaging/ParcelCreate/OrderParcelCreateEventPublisher.cs @@ -1,4 +1,5 @@ using Fake.EventBus.Distributed; +using Fake.UnitOfWork; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -11,10 +12,12 @@ using Oms.Domain.OrdersAggregate.Entities; namespace Oms.Application.Messaging.ParcelCreate; /// -/// 创单后置事件发布器：优先走 Fake.EventBus.RabbitMQ，失败则同进程同步执行 Handler 逻辑。 +/// 创单后置事件发布器：事务内写 Outbox（Fake.EventBus.IntegrationEventLog），Commit 后投递 RabbitMQ。 +/// 揽收 Publish 失败时由调用方同步 ExecutePickupAttachAsync 回落。 /// public sealed class OrderParcelCreateEventPublisher( IDistributedEventBus eventBus, + IUnitOfWorkManager unitOfWorkManager, IServiceScopeFactory scopeFactory, ILogger logger) : IOrderParcelCreateEventPublisher { @@ -33,14 +36,8 @@ public sealed class OrderParcelCreateEventPublisher( Authorization = request.Authorization }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishLabelGenerateAsync( @@ -54,14 +51,8 @@ public sealed class OrderParcelCreateEventPublisher( OperatorUserId = operatorUserId }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishCustoms9610UpsertAsync( @@ -75,14 +66,8 @@ public sealed class OrderParcelCreateEventPublisher( PlatformCode = platformCode }; - if (await TryPublishAsync(@event, cancellationToken)) - { - return; - } - - await ExecuteInScopeAsync( - h => h.HandleAsync(@event, cancellationToken), - cancellationToken); + await OmsIntegrationEventPublisher.PublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task PublishPickupAttachAsync( @@ -96,8 +81,8 @@ public sealed class OrderParcelCreateEventPublisher( return false; } - // 揽收回落由调用方 ExecutePickupAttachAsync 同步执行，避免重复建单且便于回写单号。 - return await TryPublishAsync(@event, cancellationToken); + return await OmsIntegrationEventPublisher.TryPublishAsync( + eventBus, unitOfWorkManager, logger, @event, cancellationToken); } public async Task ExecutePickupAttachAsync( @@ -155,44 +140,4 @@ public sealed class OrderParcelCreateEventPublisher( ExpectPickupTime = input.ExpectPickupTime.Value }; } - - private async Task TryPublishAsync(IntegrationEvent @event, CancellationToken cancellationToken) - { - try - { - await eventBus.PublishAsync(@event, cancellationToken); - logger.LogDebug( - "Published parcel-create event {EventName} Id={EventId}", - @event.GetType().Name, - @event.Id); - return true; - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "Publish parcel-create 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); - } - } } diff --git a/src/Oms.Application/Oms.Application.csproj b/src/Oms.Application/Oms.Application.csproj index 73214d1..72efa2e 100644 --- a/src/Oms.Application/Oms.Application.csproj +++ b/src/Oms.Application/Oms.Application.csproj @@ -18,6 +18,7 @@ + diff --git a/src/Oms.Application/Rbac/UserProfileRepository.cs b/src/Oms.Application/Rbac/UserProfileRepository.cs index 73f75a3..7115b11 100644 --- a/src/Oms.Application/Rbac/UserProfileRepository.cs +++ b/src/Oms.Application/Rbac/UserProfileRepository.cs @@ -187,6 +187,8 @@ public class UserProfileRepository(FakeRbacDbContext dbContext) : IScopedDepende if (int.TryParse(match.Groups[1].Value, out var number)) { + + max = Math.Max(max, number); } } diff --git a/src/Oms.Application/Services/Operations/InboundRecordService.cs b/src/Oms.Application/Services/Operations/InboundRecordService.cs index 5d6a56d..0770e99 100644 --- a/src/Oms.Application/Services/Operations/InboundRecordService.cs +++ b/src/Oms.Application/Services/Operations/InboundRecordService.cs @@ -1,4 +1,6 @@ +using System.Data; using Fake.Application; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Mvc; using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Shared; @@ -85,6 +87,7 @@ public class InboundRecordService : ApplicationService /// 入参：入库记录创建参数。 /// 入库记录结果。 [HttpPost("/operation/inbound/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateAsync(CreateInboundRecordInput input) { if (string.IsNullOrWhiteSpace(input.WaybillNo)) @@ -141,8 +144,8 @@ public class InboundRecordService : ApplicationService var savedEntity = await InboundRecordRepository.InsertAsync(entity); - // 自动触发改异步，避免拉长扫码 RT - AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Inbound, operatorId); + // 自动触发走 Outbox，避免拉长扫码 RT + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Inbound, operatorId); var refreshed = await OrderRepository.GetByWaybillNoAsync(input.WaybillNo.Trim()); var output = ObjectMapper.Map(savedEntity); diff --git a/src/Oms.Application/Services/Operations/OutboundRecordService.cs b/src/Oms.Application/Services/Operations/OutboundRecordService.cs index b64d4b7..4f856e3 100644 --- a/src/Oms.Application/Services/Operations/OutboundRecordService.cs +++ b/src/Oms.Application/Services/Operations/OutboundRecordService.cs @@ -1,4 +1,6 @@ +using System.Data; using Fake.Application; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Mvc; using Oms.Application.Dtos.Operations; using Oms.Application.Dtos.Shared; @@ -99,6 +101,7 @@ public class OutboundRecordService : ApplicationService /// 入参：出库记录创建参数。 /// 出库记录结果。 [HttpPost("/operation/outbound/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateAsync(CreateOutboundRecordInput input) { if (string.IsNullOrWhiteSpace(input.WaybillNo)) @@ -161,8 +164,8 @@ public class OutboundRecordService : ApplicationService var savedEntity = await OutboundRecordRepository.InsertAsync(entity); - // 自动触发改走 Fake.EventBus，避免拉长出库扫码 RT - AutoTriggerRuntimeService.EnqueueFire(order.Id, AutoTriggerTiming.Outbound, operatorId); + // 自动触发走 Outbox，避免拉长出库扫码 RT + await AutoTriggerRuntimeService.EnqueueFireAsync(order.Id, AutoTriggerTiming.Outbound, operatorId); return ObjectMapper.Map(savedEntity); } diff --git a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs index 9b70856..34d536e 100644 --- a/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs +++ b/src/Oms.Application/Services/Orders/Issues/AutoTriggerRuntimeService.cs @@ -22,17 +22,12 @@ public class AutoTriggerRuntimeService( private const string InterceptOperation = "拦截"; /// - /// 后台异步触发（Fake.EventBus.RabbitMQ），避免拉长入库/出库/创单请求。 + /// 投递自动触发事件（须在事务 UoW 内调用，写入 Outbox）。 /// - public void EnqueueFire(int orderId, AutoTriggerTiming timing, int operatorUserId) - { - if (orderId <= 0) - { - return; - } - - _ = sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId); - } + public Task EnqueueFireAsync(int orderId, AutoTriggerTiming timing, int operatorUserId) => + orderId <= 0 + ? Task.CompletedTask + : sideEffectEventPublisher.PublishAutoTriggerFireAsync(orderId, timing, operatorUserId); public async Task FireAsync(Order order, AutoTriggerTiming timing, int operatorUserId) { diff --git a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs index f37c5d4..e25b977 100644 --- a/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs +++ b/src/Oms.Application/Services/Orders/OrderOperationLogSnapshot.cs @@ -37,21 +37,37 @@ public static class OrderOperationLogSnapshot ["isPaid"] = "是否已支付", ["additionalServices"] = "附加服务", ["sender.name"] = "发件人姓名", + ["sender.company"] = "发件公司", ["sender.phone"] = "发件人电话", + ["sender.telePhone"] = "发件人固话", + ["sender.email"] = "发件人邮箱", ["sender.country"] = "发件国家", ["sender.province"] = "发件省/州", ["sender.city"] = "发件城市", ["sender.zipCode"] = "发件邮编", ["sender.address1"] = "发件地址1", ["sender.address2"] = "发件地址2", + ["sender.address3"] = "发件地址3", + ["sender.houseNumber"] = "发件门牌号", ["receiver.name"] = "收件人姓名", + ["receiver.company"] = "收件公司", ["receiver.phone"] = "收件人电话", + ["receiver.telePhone"] = "收件人固话", + ["receiver.email"] = "收件人邮箱", ["receiver.country"] = "收件国家", ["receiver.province"] = "收件省/州", ["receiver.city"] = "收件城市", ["receiver.zipCode"] = "收件邮编", + ["receiver.streetNumber"] = "收件街道号", ["receiver.address1"] = "收件地址1", ["receiver.address2"] = "收件地址2", + ["receiver.address3"] = "收件地址3", + ["receiver.taxType"] = "收件税号类型", + ["receiver.taxNo"] = "收件税号", + ["receiver.certificateType"] = "收件证件类型", + ["receiver.certificateNo"] = "收件证件号", + ["receiver.eori"] = "收件EORI", + ["receiver.vat"] = "收件VAT", ["fee.receivableTotal"] = "应收合计", ["fee.payableTotal"] = "应付合计", }; @@ -94,13 +110,18 @@ public static class OrderOperationLogSnapshot return new Dictionary(StringComparer.OrdinalIgnoreCase) { ["sender.name"] = Norm(sender.SenderName), + ["sender.company"] = Norm(sender.Company), ["sender.phone"] = Norm(sender.Phone), + ["sender.telePhone"] = Norm(sender.TelePhone), + ["sender.email"] = Norm(sender.Email), ["sender.country"] = FormatInt(sender.Country), ["sender.province"] = Norm(sender.Province), ["sender.city"] = Norm(sender.City), ["sender.zipCode"] = Norm(sender.ZipCode), ["sender.address1"] = Norm(sender.StreetAddress1), ["sender.address2"] = Norm(sender.StreetAddress2), + ["sender.address3"] = Norm(sender.StreetAddress3), + ["sender.houseNumber"] = Norm(sender.HouseNumber), }; } @@ -114,13 +135,24 @@ public static class OrderOperationLogSnapshot return new Dictionary(StringComparer.OrdinalIgnoreCase) { ["receiver.name"] = Norm(recipient.RecipientsName), + ["receiver.company"] = Norm(recipient.CompanyName), ["receiver.phone"] = Norm(recipient.Phone), + ["receiver.telePhone"] = Norm(recipient.TelePhone), + ["receiver.email"] = Norm(recipient.Email), ["receiver.country"] = FormatInt(recipient.Country), ["receiver.province"] = Norm(recipient.Province), ["receiver.city"] = Norm(recipient.City), ["receiver.zipCode"] = Norm(recipient.ZipCode), + ["receiver.streetNumber"] = Norm(recipient.StreetNumber), ["receiver.address1"] = Norm(recipient.StreetAddress1), ["receiver.address2"] = Norm(recipient.StreetAddress2), + ["receiver.address3"] = Norm(recipient.StreetAddress3), + ["receiver.taxType"] = FormatInt(recipient.TaxType), + ["receiver.taxNo"] = Norm(recipient.TaxNo), + ["receiver.certificateType"] = FormatInt(recipient.CertificateType), + ["receiver.certificateNo"] = Norm(recipient.CertificateNo), + ["receiver.eori"] = Norm(recipient.Eori), + ["receiver.vat"] = Norm(recipient.Vat), }; } diff --git a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs index 0a98566..076e3f3 100644 --- a/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs +++ b/src/Oms.Application/Services/Orders/OrderService.OrderEdit.cs @@ -53,6 +53,13 @@ public partial class OrderService order.TrackNo = string.IsNullOrWhiteSpace(input.TrackNo) ? null : input.TrackNo.Trim(); } + if (input.DownstreamServiceNo != null) + { + order.DownstreamServiceNo = string.IsNullOrWhiteSpace(input.DownstreamServiceNo) + ? null + : input.DownstreamServiceNo.Trim(); + } + if (input.ChannelCode != null) { order.ChannelCode = string.IsNullOrWhiteSpace(input.ChannelCode) ? null : input.ChannelCode.Trim(); @@ -138,7 +145,9 @@ public partial class OrderService order = await GetOrderOrThrowAsync(order.Id); var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeParties: true); - var remark = input.Items != null ? "申报信息已更新" : null; + // 前端保存基本信息时总会带 items；仅在无字段级 diff 时用备注兜底，避免掩盖真实字段变更 + var fieldChanges = OrderOperationLogSnapshot.Diff(beforeSnap, afterSnap); + var remark = input.Items != null && fieldChanges.Count == 0 ? "申报信息已更新" : null; await OperationLogWriter.TryWriteAsync( order, OrderOperationActionCodes.EditBasic, diff --git a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs index fbd528a..e2e7fa6 100644 --- a/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs +++ b/src/Oms.Application/Services/Orders/OrderService.ParcelBilling.cs @@ -2382,6 +2382,7 @@ public partial class OrderService try { + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) .OrderBy(x => x.Id) .ToList(); @@ -2419,6 +2420,16 @@ public partial class OrderService order.UpdateTime = DateTime.Now; await OrderRepository.UpdateAsync(order); + // 操作日志同步落库（件重尺已写完）；计费/排货仍走 MQ，不把审计塞进队列 + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.Inbound, + beforeSnap, + afterSnap, + FormatMeasuredDimensionsRemark("收货", length!.Value, width!.Value, height!.Value, actualWeight!.Value)); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.InboundTrackCode); await EnqueueInboundBillingAsync(order); } @@ -2789,7 +2800,7 @@ public partial class OrderService appendInitialTrack: true, operatorUserId); - AutoTriggerRuntimeService.EnqueueFire( + await AutoTriggerRuntimeService.EnqueueFireAsync( order.Id, AutoTriggerTiming.OrderCreate, operatorUserId); @@ -3016,6 +3027,7 @@ public partial class OrderService EnsureBillingProductActive(product); + var beforeSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); var weights = (await OrderRepository.GetWeightsByOrderIdsAsync([order.Id])) .OrderBy(x => x.Id) .ToList(); @@ -3038,9 +3050,28 @@ public partial class OrderService order.UpdateTime = now; await OrderRepository.UpdateAsync(order); + order = await GetOrderOrThrowAsync(order.Id); + var afterSnap = await CaptureOrderAuditSnapshotAsync(order, includeWeights: true); + await OperationLogWriter.TryWriteAsync( + order, + OrderOperationActionCodes.Outbound, + beforeSnap, + afterSnap, + FormatMeasuredDimensionsRemark("发货", length!.Value, width!.Value, height!.Value, actualWeight!.Value)); + await EnqueueLifecycleTrackAppendAsync(order, SmallBagTrackLifecycleEvents.OutboundTrackCode); } + private static string FormatMeasuredDimensionsRemark( + string stage, + decimal length, + decimal width, + decimal height, + decimal actualWeight) + { + return $"{stage} 长×宽×高×重：{length:0.####}×{width:0.####}×{height:0.####}×{actualWeight:0.####}"; + } + private async Task BuildParcelCreateOutputAsync(Order order) { // 预报已成功时，映射/面单/本地存储失败不得回滚订单（常见：容器 /app/wwwroot 不可写）。 diff --git a/src/Oms.Application/Services/Orders/OrderService.cs b/src/Oms.Application/Services/Orders/OrderService.cs index 5e90860..6bcbac0 100644 --- a/src/Oms.Application/Services/Orders/OrderService.cs +++ b/src/Oms.Application/Services/Orders/OrderService.cs @@ -1,7 +1,9 @@ +using System.Data; using System.Globalization; using Fake.Application; using Fake.Domain.Exceptions; using Fake.ObjectStorage; +using Fake.UnitOfWork; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Mvc; using MiniExcelLibs; @@ -265,6 +267,7 @@ public partial class OrderService : ApplicationService /// 入参：订单操作参数。 /// 无返回体，仅完成订单预报处理。 [HttpPost("/order/pre-declare")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task PreDeclareAsync(OrderActionInput input) { var order = await GetOrderOrThrowAsync(input.Id); @@ -605,6 +608,7 @@ public partial class OrderService : ApplicationService /// 入参：订单包裹创建参数。 /// 订单包裹创建结果。 [HttpPost("/order/parcel/create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateParcelAsync(OrderParcelCreateInput input) { var inbound = string.IsNullOrWhiteSpace(input.InboundMethod) @@ -636,6 +640,7 @@ public partial class OrderService : ApplicationService /// /// 客户 OpenAPI 创单：同一客户相同 clientNo 视为幂等，返回已存在订单，不新建。 /// + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task CreateParcelForCustomerOpenApiAsync(OrderParcelCreateInput input) { NormalizeParcelCreateInput(input); @@ -734,6 +739,7 @@ public partial class OrderService : ApplicationService /// 入参：批量创建小包请求集合，每条记录沿用单票创建结构。 /// 批量处理结果，包含批次号、成功失败数量与逐条执行明细。 [HttpPost("/order/parcel/batch-create")] + [UnitOfWork(IsTransactional = true, IsolationLevel = IsolationLevel.ReadCommitted)] public async Task BatchCreateParcelAsync(OrderParcelBatchCreateInput input) { if (input == null || input.Count == 0) diff --git a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs index d6c482b..104d465 100644 --- a/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs +++ b/src/Oms.Domain/OrdersAggregate/Consts/OrderOperationActionCodes.cs @@ -14,6 +14,8 @@ public static class OrderOperationActionCodes public const string PreDeclare = "preDeclare"; public const string DownstreamForward = "downstreamForward"; public const string Print = "print"; + public const string Inbound = "inbound"; + public const string Outbound = "outbound"; public static string GetActionName(string action) => action switch { @@ -29,6 +31,8 @@ public static class OrderOperationActionCodes PreDeclare => "重新预报", DownstreamForward => "预报下游", Print => "打印面单", + Inbound => "入库", + Outbound => "出库", _ => action }; } diff --git a/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs b/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs new file mode 100644 index 0000000..b4a9225 --- /dev/null +++ b/src/Oms.Tests/Messaging/Fakes/RecordingPhysicalDistributedEventBus.cs @@ -0,0 +1,23 @@ +using System.Collections.Concurrent; +using Fake.EventBus; +using Fake.EventBus.Distributed; + +namespace Oms.Tests.Messaging.Fakes; + +public sealed class RecordingPhysicalDistributedEventBus : IPhysicalDistributedEventBus +{ + private readonly ConcurrentQueue _published = new(); + + public IReadOnlyList Published => _published.ToArray(); + + public void Clear() => _published.Clear(); + + public Task PublishAsync(IntegrationEvent @event, CancellationToken cancellationToken = default) + { + _published.Enqueue(@event); + return Task.CompletedTask; + } + + public Task PublishAsync(Event @event, CancellationToken cancellationToken = default) + => PublishAsync((IntegrationEvent)@event, cancellationToken); +} diff --git a/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs b/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs new file mode 100644 index 0000000..8b22af6 --- /dev/null +++ b/src/Oms.Tests/Messaging/OmsOutboxTestModule.cs @@ -0,0 +1,51 @@ +using Fake.Autofac; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Modularity; +using Fake.SyncEx; +using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +[DependsOn(typeof(FakeAutofacModule))] +[DependsOn(typeof(FakeEntityFrameworkCoreModule))] +[DependsOn(typeof(FakeEventBusIntegrationEventLogModule))] +public class OmsOutboxTestModule : FakeModule +{ + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddSingleton(); + context.Services.AddSingleton(sp => + sp.GetRequiredService()); + + context.Services.Configure(o => + { + o.StartupDelay = TimeSpan.FromHours(1); + o.ScanInterval = TimeSpan.FromHours(1); + }); + context.Services.Configure(o => + { + o.StartupDelay = TimeSpan.FromHours(1); + o.CleanupInterval = TimeSpan.FromHours(1); + }); + + var connection = new SqliteConnection("Filename=:memory:"); + connection.Open(); + + context.Services.AddDbContext(builder => + { + builder.UseSqlite(connection); + }, ServiceLifetime.Scoped, ServiceLifetime.Singleton); + } + + public override void PreConfigureApplication(ApplicationConfigureContext context) + { + using var scope = context.ServiceProvider.CreateScope(); + var ctx = scope.ServiceProvider.GetRequiredService(); + SyncContext.Run(async () => await ctx.Database.EnsureCreatedAsync()); + } +} diff --git a/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs new file mode 100644 index 0000000..da907e3 --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderBatchJobOutboxPublisherTests.cs @@ -0,0 +1,90 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Oms.Application.Messaging; +using Oms.Domain.OrdersAggregate.Consts; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderBatchJobOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderBatchJobPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderBatchJobOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderBatchJobPublisher( + ServiceProvider.GetRequiredService(), + _uowManager, + Options.Create(new OrderBatchRabbitMqOptions { Enabled = true }), + NullLogger.Instance); + } + + [Fact] + public async Task ForecastBilling任务消息_事务Commit后投递() + { + _physicalBus.Clear(); + var message = new OrderBatchJobMessage + { + JobId = 1, + JobItemId = 2, + Action = OrderBatchActionCodes.ForecastBilling, + OrderId = 55, + MessageId = "job:1:55:forecast-billing", + OperatorUserId = 9, + PayloadJson = "{\"CouponInstanceId\":1}" + }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishAsync(message); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + var evt = _physicalBus.Published.Should().ContainSingle() + .Which.Should().BeOfType().Subject; + evt.Action.Should().Be(OrderBatchActionCodes.ForecastBilling); + evt.OrderId.Should().Be(55); + evt.MessageId.Should().Be(message.MessageId); + } + + [Fact] + public async Task 队列未启用时_Publish应抛错() + { + var disabled = new OrderBatchJobPublisher( + ServiceProvider.GetRequiredService(), + _uowManager, + Options.Create(new OrderBatchRabbitMqOptions { Enabled = false }), + NullLogger.Instance); + + var act = () => disabled.PublishAsync(new OrderBatchJobMessage + { + Action = OrderBatchActionCodes.ForecastBilling, + OrderId = 1, + MessageId = "x" + }); + + await act.Should().ThrowAsync() + .WithMessage("*disabled*"); + } +} diff --git a/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs new file mode 100644 index 0000000..f4d036e --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderParcelCreateOutboxPublisherTests.cs @@ -0,0 +1,131 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Oms.Application.Integrations.SmallBagTrack; +using Oms.Application.Messaging; +using Oms.Application.Messaging.ParcelCreate; +using Oms.Domain.OrdersAggregate.Entities; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderParcelCreateOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly IDistributedEventBus _distributedEventBus; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IOutboxEventLogService _outbox; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderParcelCreateEventPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderParcelCreateOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _distributedEventBus = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _outbox = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderParcelCreateEventPublisher( + _distributedEventBus, + _uowManager, + ServiceProvider.GetRequiredService(), + NullLogger.Instance); + } + + [Fact] + public void IDistributedEventBus_应为Outbox装饰器() + { + _distributedEventBus.Should().BeOfType(); + } + + [Fact] + public async Task 事务Uow内发布9610_应在Complete后投递物理总线() + { + _physicalBus.Clear(); + var order = new Order { Id = 101, ServiceNo = "SN101" }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishCustoms9610UpsertAsync(order, "lazada"); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().ContainSingle(); + var evt = _physicalBus.Published[0].Should().BeOfType().Subject; + evt.OrderId.Should().Be(101); + evt.PlatformCode.Should().Be("lazada"); + + var pending = (await _outbox.RetrieveEventLogsPendingToPublishAsync(50)).ToList(); + pending.Should().NotContain(e => e.EventId == evt.Id); + } + + [Fact] + public async Task 事务Uow内发布轨迹与面单_Commit后应各投递一次() + { + _physicalBus.Clear(); + var order = new Order { Id = 202, ServiceNo = "SN202" }; + var request = new SmallBagTrackUpsertRequest + { + OrderNo = "SN202", + AppendInitialTrackOnSuccess = true + }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishTrackUpsertAsync(order, request, customerId: 9); + await _publisher.PublishLabelGenerateAsync(order, operatorUserId: 7); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().HaveCount(2); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 202 && e.CustomerId == 9 && e.AppendInitialTrackOnSuccess); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 202 && e.OperatorUserId == 7); + } + + [Fact] + public async Task 回滚后不应投递() + { + _physicalBus.Clear(); + var order = new Order { Id = 303, ServiceNo = "SN303" }; + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishCustoms9610UpsertAsync(order); + await uow.RollbackAsync(); + } + + _physicalBus.Published.Should().BeEmpty(); + } + + [Fact] + public async Task 无环境事务时_短事务仍能写入并投递() + { + _physicalBus.Clear(); + _uowManager.Current.Should().BeNull(); + + var order = new Order { Id = 404, ServiceNo = "SN404" }; + await _publisher.PublishCustoms9610UpsertAsync(order, "shopee"); + + _physicalBus.Published.Should().ContainSingle() + .Which.Should().BeOfType() + .Which.OrderId.Should().Be(404); + } +} diff --git a/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs new file mode 100644 index 0000000..8f3abf8 --- /dev/null +++ b/src/Oms.Tests/Messaging/OrderSideEffectOutboxPublisherTests.cs @@ -0,0 +1,62 @@ +using Fake; +using Fake.EntityFrameworkCore; +using Fake.EventBus.Distributed; +using Fake.EventBus.IntegrationEventLog; +using Fake.Testing; +using Fake.UnitOfWork; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Oms.Application.Messaging; +using Oms.Application.Messaging.OrderSideEffects; +using Oms.Application.Services.Orders.Issues; +using Oms.Tests.Messaging.Fakes; + +namespace Oms.Tests.Messaging; + +public class OrderSideEffectOutboxPublisherTests : ApplicationTestWithTools +{ + private readonly IUnitOfWorkManager _uowManager; + private readonly IDistributedEventBus _distributedEventBus; + private readonly RecordingPhysicalDistributedEventBus _physicalBus; + private readonly IEfDbContextProvider _dbProvider; + private readonly OrderSideEffectEventPublisher _publisher; + + protected override void SetApplicationCreationOptions(FakeApplicationCreationOptions options) + { + options.UseAutofac(); + } + + public OrderSideEffectOutboxPublisherTests() + { + _uowManager = ServiceProvider.GetRequiredService(); + _distributedEventBus = ServiceProvider.GetRequiredService(); + _physicalBus = ServiceProvider.GetRequiredService(); + _dbProvider = ServiceProvider.GetRequiredService>(); + _publisher = new OrderSideEffectEventPublisher( + _distributedEventBus, + _uowManager, + NullLogger.Instance); + } + + [Fact] + public async Task 事务内发布自动触发与生命周期轨迹_Commit后投递() + { + _physicalBus.Clear(); + + using (var uow = _uowManager.Begin(OmsOutboxUnitOfWork.Transactional())) + { + await _dbProvider.GetDbContextAsync(); + await _publisher.PublishAutoTriggerFireAsync(11, AutoTriggerTiming.OrderCreate, 3); + await _publisher.PublishLifecycleTrackAppendAsync(11, "SN11", "A0301"); + _physicalBus.Published.Should().BeEmpty(); + await uow.CompleteAsync(); + } + + _physicalBus.Published.Should().HaveCount(2); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 11 && e.Timing == AutoTriggerTiming.OrderCreate.ToString() && e.OperatorUserId == 3); + _physicalBus.Published.OfType().Should().ContainSingle(e => + e.OrderId == 11 && e.OrderNo == "SN11"); + } +} diff --git a/src/Oms.Tests/Oms.Tests.csproj b/src/Oms.Tests/Oms.Tests.csproj index 19f3343..040ca5b 100644 --- a/src/Oms.Tests/Oms.Tests.csproj +++ b/src/Oms.Tests/Oms.Tests.csproj @@ -13,9 +13,14 @@ all runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + all diff --git a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj index 0143f14..568c1a4 100644 --- a/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj +++ b/src/SimpleAdmin.Api/SimpleAdmin.Api.csproj @@ -15,12 +15,13 @@ + - + diff --git a/src/SimpleAdmin.Api/SimpleAdminApiModule.cs b/src/SimpleAdmin.Api/SimpleAdminApiModule.cs index b24c7ff..19f0d1f 100644 --- a/src/SimpleAdmin.Api/SimpleAdminApiModule.cs +++ b/src/SimpleAdmin.Api/SimpleAdminApiModule.cs @@ -7,6 +7,7 @@ using Fake.AspNetCore.Auditing; using Fake.AspNetCore.ExceptionHandling; using Fake.AspNetCore.Newtonsoft; using Fake.AspNetCore.Swagger; +using Fake.EventBus.IntegrationEventLog; using Fake.FeiShu; using Fake.FileManagement.Application; using Fake.FileManagement.Infrastructure; @@ -37,6 +38,8 @@ namespace SimpleAdmin.Api; typeof(FakeFileManagementApplicationModule), typeof(FakeFileManagementInfrastructureModule), typeof(FakeObjectStorageTencentCosModule), + // 在 RabbitMQ 物理总线之后 Replace IDistributedEventBus → Outbox + typeof(FakeEventBusIntegrationEventLogModule), typeof(FakeAspNetCoreModule), typeof(FakeSerilogModule) )] @@ -80,6 +83,20 @@ public class SimpleAdminApiModule : FakeModule services.AddOmsWeChat(configuration); services.AddSingleton(); + // Outbox / Inbox 与业务同库（EF），通过 DbTransaction 征用 SqlSugar 事务 + services.AddDbContext(options => + { + var connectionString = configuration.GetConnectionString("IntegrationEventLog") + ?? configuration.GetConnectionString("Default") + ?? configuration["ConnectionStrings:Default"] + ?? string.Empty; + var serverVersion = ServerVersion.AutoDetect(connectionString); + options.UseMySql(connectionString, serverVersion); +#if DEBUG + options.EnableSensitiveDataLogging(); +#endif + }); + //Json格式化全局处理 context.Services.AddFakeNewtonsoft(); @@ -188,11 +205,14 @@ public class SimpleAdminApiModule : FakeModule // 端点映射 app.MapControllers(); - // FileManagement 与 RBAC 同库并行（EF）；应用 Pending migrations 建 stored_file + // FileManagement / Outbox 与业务同库并行（EF） using (var scope = app.Services.CreateScope()) { var fileDb = scope.ServiceProvider.GetRequiredService(); fileDb.Database.Migrate(); + + var outboxDb = scope.ServiceProvider.GetRequiredService(); + IntegrationEventLogSchema.EnsureTablesAsync(outboxDb).GetAwaiter().GetResult(); } app.Services.GetRequiredService().Send("Application started"); diff --git a/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs b/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs new file mode 100644 index 0000000..f44e97c --- /dev/null +++ b/src/SimpleAdmin.Api/Storage/IntegrationEventLogSchema.cs @@ -0,0 +1,48 @@ +using Fake.EventBus.IntegrationEventLog; +using Microsoft.EntityFrameworkCore; + +namespace SimpleAdmin.Api.Storage; + +/// +/// Outbox/Inbox 与业务同库时 EnsureCreated 无效（库已存在），用 IF NOT EXISTS 建表。 +/// +internal static class IntegrationEventLogSchema +{ + public static async Task EnsureTablesAsync( + IntegrationEventLogContext db, + CancellationToken cancellationToken = default) + { + await db.Database.ExecuteSqlRawAsync( + """ + CREATE TABLE IF NOT EXISTS `OutboxEventLog` ( + `EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL, + `EventTypeName` varchar(512) NOT NULL, + `State` int NOT NULL, + `TimesSent` int NOT NULL, + `CreationTime` datetime(6) NOT NULL, + `Content` longtext NOT NULL, + `TransactionId` varchar(64) NOT NULL, + `LockExpiresAt` datetime(6) NULL, + PRIMARY KEY (`EventId`), + KEY `IX_OutboxEventLog_State_CreationTime` (`State`, `CreationTime`), + KEY `IX_OutboxEventLog_TransactionId` (`TransactionId`) + ) CHARACTER SET utf8mb4; + """, + cancellationToken); + + await db.Database.ExecuteSqlRawAsync( + """ + CREATE TABLE IF NOT EXISTS `InboxEventLog` ( + `EventId` char(36) CHARACTER SET ascii COLLATE ascii_general_ci NOT NULL, + `EventTypeName` varchar(512) NOT NULL, + `State` int NOT NULL, + `ProcessedTime` datetime(6) NOT NULL, + `Content` longtext NOT NULL, + `ErrorMessage` longtext NULL, + PRIMARY KEY (`EventId`), + KEY `IX_InboxEventLog_ProcessedTime` (`ProcessedTime`) + ) CHARACTER SET utf8mb4; + """, + cancellationToken); + } +} diff --git a/src/SimpleAdmin.Api/appsettings.Development.json b/src/SimpleAdmin.Api/appsettings.Development.json index 1b28af5..03d2137 100644 --- a/src/SimpleAdmin.Api/appsettings.Development.json +++ b/src/SimpleAdmin.Api/appsettings.Development.json @@ -146,5 +146,13 @@ "RabbitMq": { "Enabled": false } + }, + "IntegrationEventLog": { + "OutboxPublisher": { + "StartupDelay": "00:00:05", + "ScanInterval": "00:00:10", + "BatchSize": 100, + "MaxDegreeOfParallelism": 8 + } } }