分布式事务与 Saga 编排

// 单体时代的写法：一个事务保护三张表
using var tx = await db.Database.BeginTransactionAsync(ct);
db.Orders.Add(order);
db.Stock.Add(new StockChange(order.Id, -1));
db.Payments.Add(payment);
await db.SaveChangesAsync(ct);
await tx.CommitAsync(ct);
// 拆分后，这三件事分属三个服务，无法共享事务

// 每一步都实现正向与补偿两个动作
public interface ISagaStep
{
    string Name { get; }
    Task ExecuteAsync(SagaContext ctx, CancellationToken ct);
    Task CompensateAsync(SagaContext ctx, CancellationToken ct);
}

public sealed class ReserveStockStep(IStockService stock) : ISagaStep
{
    public string Name => "reserve-stock";

    public async Task ExecuteAsync(SagaContext ctx, CancellationToken ct)
    {
        var reservationId = await stock.ReserveAsync(ctx.OrderId, ctx.Items, ct);
        ctx.Set("reservationId", reservationId);   // 供补偿使用
    }

    public Task CompensateAsync(SagaContext ctx, CancellationToken ct)
        => stock.ReleaseAsync(ctx.Get<Guid>("reservationId"), ct);
}

// 编排：协调器显式调用每一步，流程一目了然
public sealed class OrderSagaOrchestrator(
    IEnumerable<ISagaStep> steps,
    ISagaRepository repo,
    ILogger<OrderSagaOrchestrator> logger)
{
    public async Task RunAsync(SagaContext ctx, CancellationToken ct)
    {
        var executed = new Stack<ISagaStep>();

        foreach (var step in steps)
        {
            try
            {
                await step.ExecuteAsync(ctx, ct);
                executed.Push(step);
                await repo.SaveAsync(ctx, step.Name, ct);   // 持久化进度
            }
            catch (Exception ex)
            {
                logger.LogError(ex, "步骤 {Step} 失败，开始补偿", step.Name);
                await CompensateAsync(executed, ctx, ct);
                throw;
            }
        }
    }

    private static async Task CompensateAsync(
        Stack<ISagaStep> executed, SagaContext ctx, CancellationToken ct)
    {
        while (executed.TryPop(out var step))
        {
            try { await step.CompensateAsync(ctx, ct); }
            catch (Exception ex)
            {
                // 补偿失败必须记录并交给重试机制，不能吞掉
                ctx.RecordCompensationFailure(step.Name, ex);
            }
        }
    }
}

// 协同：服务各自订阅事件，自发推进，无中心协调者
public sealed class StockEventConsumer(IStockService stock, IEventBus bus)
    : IConsumer<OrderCreated>
{
    public async Task ConsumeAsync(OrderCreated e, CancellationToken ct)
    {
        try
        {
            await stock.ReserveAsync(e.OrderId, e.Items, ct);
            await bus.PublishAsync(new StockReserved(e.OrderId), ct);
        }
        catch (Exception ex)
        {
            await bus.PublishAsync(new StockReservationFailed(e.OrderId, ex.Message), ct);
        }
    }
}

// 幂等处理器：用业务去重键 + 唯一索引保证只生效一次
public sealed class PaymentConsumer(
    AppDbContext db,
    IPaymentGateway gateway) : IConsumer<OrderPlaced>
{
    public async Task ConsumeAsync(OrderPlaced e, CancellationToken ct)
    {
        var key = $"payment:{e.OrderId}";     // 业务去重键

        // 依赖数据库唯一索引：并发下只有一个能插入成功
        if (await db.ProcessedMessages.AnyAsync(m => m.Key == key, ct))
            return;                            // 已处理，直接跳过

        await gateway.ChargeAsync(e.OrderId, e.Amount, key, ct);

        db.ProcessedMessages.Add(new ProcessedMessage(key, DateTimeOffset.UtcNow));
        await db.SaveChangesAsync(ct);         // 唯一索引冲突则说明并发重复
    }
}

// 更稳的做法：业务写入与去重记录放进同一个本地事务
await using var tx = await db.Database.BeginTransactionAsync(ct);
db.Payments.Add(new Payment(e.OrderId, e.Amount, PaymentStatus.Pending));
db.ProcessedMessages.Add(new ProcessedMessage(key, DateTimeOffset.UtcNow));
await db.SaveChangesAsync(ct);              // 业务与去重同时提交，原子
await tx.CommitAsync(ct);
await gateway.ChargeAsync(e.OrderId, e.Amount, ct);   // 外部调用放事务外

// 同一个事务里写业务数据与 outbox 记录
public async Task PlaceOrderAsync(Order order, CancellationToken ct)
{
    await using var tx = await db.Database.BeginTransactionAsync(ct);

    db.Orders.Add(order);
    db.Outbox.Add(new OutboxMessage
    {
        Id = Guid.NewGuid(),
        Type = nameof(OrderPlaced),
        Payload = JsonSerializer.Serialize(new OrderPlaced(order.Id, order.Items)),
        OccurredOn = DateTimeOffset.UtcNow,
        ProcessedOn = null,
    });

    await db.SaveChangesAsync(ct);   // 业务与消息一起提交，原子
    await tx.CommitAsync(ct);
}

// 后台投递器：扫描未投递的 outbox 记录，发送后标记已处理
public sealed class OutboxPublisher(
    IDbContextFactory<AppDbContext> factory,
    IEventBus bus) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(2));
        while (await timer.WaitForNextTickAsync(ct))
        {
            await using var db = await factory.CreateDbContextAsync(ct);
            var batch = await db.Outbox.Where(m => m.ProcessedOn == null)
                .OrderBy(m => m.OccurredOn).Take(100).ToListAsync(ct);

            foreach (var msg in batch)
            {
                await bus.PublishAsync(msg.Type, msg.Payload, ct);   // 至少一次
                msg.ProcessedOn = DateTimeOffset.UtcNow;
            }
            await db.SaveChangesAsync(ct);
        }
    }
}

// MassTransit 状态机：声明状态与事件驱动的转换
public sealed class OrderStateMachine : MassTransitStateMachine<OrderSagaState>
{
    public State StockReserved { get; private set; } = null!;
    public State PaymentCharged { get; private set; } = null!;
    public Event<OrderPlaced> OrderPlaced { get; private set; } = null!;
    public Event<StockReserved> StockReservedEvent { get; private set; } = null!;
    public Event<PaymentFailed> PaymentFailed { get; private set; } = null!;

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);
        Event(() => OrderPlaced, e => e.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => StockReservedEvent, e => e.CorrelateById(ctx => ctx.Message.OrderId));
        Event(() => PaymentFailed, e => e.CorrelateById(ctx => ctx.Message.OrderId));
        Initially(
            When(OrderPlaced)
                .Then(ctx => ctx.Saga.OrderId = ctx.Message.OrderId)
                .TransitionTo(StockReserved)
                .Publish(ctx => new ReserveStock(ctx.Saga.OrderId, ctx.Message.Items)));

        During(StockReserved,
            When(StockReservedEvent)
                .TransitionTo(PaymentCharged)
                .Publish(ctx => new ChargePayment(ctx.Saga.OrderId)),
            When(PaymentFailed)
                .TransitionTo(Compensating)
                .Publish(ctx => new ReleaseStock(ctx.Saga.OrderId)));   // 补偿
    }
}

// 状态持久化：EF Core 存储，Saga 状态随事务落库
services.AddMassTransit(x =>
    x.AddSagaStateMachine<OrderStateMachine, OrderSagaState>()
     .EntityFrameworkRepository(r =>
     {
         r.ExistingDbContext<AppDbContext>();
         r.UsePostgres();
     }));

// 相关性 ID 贯穿消息与日志：从请求头一路透传到每个服务
public async Task InvokeAsync(HttpContext ctx)
{
    var correlationId = ctx.Request.Headers["X-Correlation-Id"].FirstOrDefault()
                        ?? Guid.NewGuid().ToString("N");
    ctx.Response.Headers["X-Correlation-Id"] = correlationId;

    using (LogContext.PushProperty("CorrelationId", correlationId))
    using (Activity.Current?.SetTag("correlation.id", correlationId))
    {
        await next(ctx);
    }
}

// 运维查询：直接回答「这个订单的 Saga 走到哪一步」
var state = await db.OrderSagaStates
    .Where(s => s.OrderId == orderId)
    .Select(s => new { s.CurrentState, s.UpdatedAt, s.CompensationFailures })
    .FirstOrDefaultAsync(ct);