csharp-wolverinefx

.NET 向け WolverineFX

このスキルを使う場面

次のような場合にこのスキルを使います：

• Wolverine でメッセージハンドラーまたはコマンドハンドラーを作成する
• WolverineFx.HTTP で HTTP エンドポイントを作成する（Minimal API／MVC の代替）
• Marten と Wolverine でイベントソーシングを実装する
• 信頼性の高いメッセージングのためにトランザクショナルアウトボックスパターンを設定する
• メッセージトランスポート（RabbitMQ、Azure Service Bus、Amazon SQS、TCP）を設定する
• イベントソーシングで CQRS を実装する
• メッセージをバッチ処理する
• テスト可能な純粋関数のハンドラーにカスケードメッセージを使用する
• エラー処理と再試行ポリシーを設定する
• コールドスタートを最適化するためにコードを事前生成する

関連スキル

• efcore-patterns - データアクセスのための Entity Framework Core パターン
• csharp-coding-standards - 最新の C# パターン（レコード、パターンマッチング）
• http-client-resilience - Polly のレジリエンスパターン（補完的）
• background-services - ホステッドサービスとバックグラウンドジョブのパターン
• aspire-configuration - .NET Aspire のオーケストレーション

基本原則

1. 記述量の少ないコード - 純粋関数、メソッドインジェクション、最小限の定型コード
2. カスケードメッセージ - IMessageBus を注入する代わりに、ハンドラーからメッセージを返します
3. トランザクショナルアウトボックス - データベーストランザクションによるメッセージ配信の保証
4. コード生成 - 最適なパフォーマンスを実現するランタイムまたは事前生成コード
5. 垂直スライスアーキテクチャ - 技術レイヤーではなく機能ごとにコードを整理します
6. ビジネスロジックには純粋関数を使用 - インフラストラクチャをビジネスロジックから分離します

必須の NuGet パッケージ

コアメッセージング

<PackageReference Include="Wolverine" />
<PackageReference Include="WolverineFx.Http" />

永続化の統合

<PackageReference Include="WolverineFx.Marten" />

トランスポート

<PackageReference Include="WolverineFx.RabbitMQ" />
<PackageReference Include="WolverineFx.AzureServiceBus" />
<PackageReference Include="WolverineFx.Kafka" />
<PackageReference Include="WolverineFx.AmazonSQS" />

基本設定

Program.cs（ASP.NET Core）

using JasperFx;
using Wolverine;
using Wolverine.Http;

var builder = WebApplication.CreateBuilder(args);

builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
    opts.Policies.UseDurableLocalQueues();
});

builder.Services.AddWolverineHttp();

var app = builder.Build();

app.MapWolverineEndpoints();

return await app.RunJasperFxCommands(args);

メッセージハンドラー

シンプルなメッセージハンドラー

public record DebitAccount(long AccountId, decimal Amount);

public static class DebitAccountHandler
{
    public static void Handle(DebitAccount command, IAccountRepository repository)
    {
        repository.Debit(command.AccountId, command.Amount);
    }
}

カスケードメッセージを使うハンドラー

public record CreateOrder(Guid OrderId, string[] Items);
public record OrderCreated(Guid OrderId);

public static class CreateOrderHandler
{
    public static (OrderCreated, ShipOrder) Handle(
        CreateOrder command,
        IDocumentSession session)
    {
        var order = new Order { Id = command.OrderId, Items = command.Items };
        session.Store(order);
        
        return (
            new OrderCreated(command.OrderId),
            new ShipOrder(command.OrderId)
        );
    }
}

複数のメッセージに OutgoingMessages を使用する

public static OutgoingMessages Handle(ProcessOrder command)
{
    var messages = new OutgoingMessages
    {
        new OrderProcessed(command.OrderId),
        new SendEmail(command.CustomerEmail, "Order processed"),
        new UpdateInventory(command.Items)
    };
    
    messages.Delay(new CleanupOrder(command.OrderId), 5.Minutes());
    
    return messages;
}

HTTP エンドポイント（WolverineFx.HTTP）

基本的な GET エンドポイント

[WolverineGet("/users/{id}")]
public static Task<User?> GetUser(int id, IQuerySession session)
    => session.LoadAsync<User>(id);

メッセージを発行する POST

[WolverinePost("/orders")]
public static async Task<IResult> CreateOrder(
    CreateOrderRequest request,
    IDocumentSession session,
    IMessageBus bus)
{
    var order = new Order { Id = Guid.NewGuid(), Items = request.Items };
    session.Store(order);
    
    await bus.PublishAsync(new OrderCreated(order.Id));
    
    return Results.Created($"/orders/{order.Id}", order);
}

複合ハンドラー（読み込み／検証／処理）

public static class UpdateOrderEndpoint
{
    public static async Task<(Order?, IResult)> LoadAsync(
        UpdateOrder command,
        IDocumentSession session)
    {
        var order = await session.LoadAsync<Order>(command.OrderId);
        return order != null
            ? (order, new WolverineContinue())
            : (order, Results.NotFound());
    }

    [WolverinePut("/orders")]
    public static void Handle(UpdateOrder command, Order order, IDocumentSession session)
    {
        order.Items = command.Items;
        session.Store(order);
    }
}

Marten によるイベントソーシング

集約ハンドラーのワークフロー

public class Order
{
    public Guid Id { get; set; }
    public int Version { get; set; }
    public Dictionary<string, Item> Items { get; set; } = new();
    public DateTimeOffset? Shipped { get; private set; }

    public void Apply(ItemReady ready) => Items[ready.Name].Ready = true;
    public void Apply(IEvent<OrderShipped> shipped) => Shipped = shipped.Timestamp;

    public bool IsReadyToShip() => Shipped == null && Items.Values.All(x => x.Ready);
}

public record MarkItemReady(Guid OrderId, string ItemName, int Version);

[AggregateHandler]
public static IEnumerable<object> Handle(MarkItemReady command, Order order)
{
    if (order.Items.TryGetValue(command.ItemName, out var item))
    {
        item.Ready = true;
        yield return new ItemReady(command.ItemName);
    }
    
    if (order.IsReadyToShip())
    {
        yield return new OrderReady();
    }
}

集約の読み込み（読み取り専用）

[WolverineGet("/orders/{id}")]
public static Order GetOrder([ReadAggregate] Order order) => order;

検証付きで集約を書き込む

public static IEnumerable<object> Handle(
    MarkItemReady command,
    [WriteAggregate(Required = true, OnMissing = OnMissing.ProblemDetailsWith404)] Order order)
{
    order.Items[command.ItemName].Ready = true;
    yield return new ItemReady(command.ItemName);
}

更新された集約を返す

[AggregateHandler]
public static (UpdatedAggregate, Events) Handle(
    MarkItemReady command,
    Order order)
{
    var events = new Events();
    events.Add(new ItemReady(command.ItemName));
    return (new UpdatedAggregate(), events);
}

トランザクショナルアウトボックス

Marten との統合

builder.Services.AddMarten(opts =>
{
    opts.Connection(connectionString);
}).IntegrateWithWolverine();

builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
});

コントローラーでアウトボックスを使う

[HttpPost("/orders")]
public async Task Post(
    [FromBody] CreateOrder command,
    [FromServices] IDocumentSession session,
    [FromServices] IMartenOutbox outbox)
{
    outbox.Enroll(session);
    
    var order = new Order { Id = command.OrderId };
    session.Store(order);
    
    await outbox.PublishAsync(new OrderCreated(command.OrderId));
    
    await session.SaveChangesAsync();
}

トランスポートの設定

RabbitMQ

builder.Host.UseWolverine(opts =>
{
    opts.UseRabbitMq("host=localhost")
        .AutoProvision()
        .AutoPurgeOnStartup();
    
    opts.PublishAllMessages()
        .ToRabbitExchange("wolverine.events", exchange =>
        {
            exchange.ExchangeType = ExchangeType.Topic;
        });
});

Azure Service Bus

builder.Host.UseWolverine(opts =>
{
    opts.UseAzureServiceBus(asbConnectionString)
        .AutoProvision()
        .ConfigureQueue(q => q.MaxDeliveryCount = 5);
});

Amazon SQS

builder.Host.UseWolverine(opts =>
{
    opts.UseAmazonSqs(sqsConfig)
        .AutoProvision();
});

メッセージのバッチ処理

バッチ処理を設定する

builder.Host.UseWolverine(opts =>
{
    opts.BatchMessagesOf<SubTaskCompleted>(batching =>
    {
        batching.BatchSize = 500;
        batching.TriggerTime = 1.Seconds();
    });
});

バッチハンドラー

public static class ItemBatchHandler
{
    public static void Handle(Item[] items, IRepository repository)
    {
        foreach (var item in items)
        {
            repository.Process(item);
        }
    }
}

カスタムバッチ処理戦略

public record SubTaskCompleted(string TaskId, string SubTaskId);
public record SubTaskBatch(string TaskId, string[] SubTaskIds);

public class SubTaskBatcher : IMessageBatcher
{
    public IEnumerable<Envelope> Group(IReadOnlyList<Envelope> envelopes)
    {
        var groups = envelopes
            .GroupBy(x => x.Message!.As<SubTaskCompleted>().TaskId);
        
        foreach (var group in groups)
        {
            var subTaskIds = group
                .Select(x => x.Message)
                .OfType<SubTaskCompleted>()
                .Select(x => x.SubTaskId)
                .ToArray();
            
            yield return new Envelope(
                new SubTaskBatch(group.Key, subTaskIds),
                group);
        }
    }

    public Type BatchMessageType => typeof(SubTaskBatch);
}

エラー処理

再試行ポリシー

builder.Host.UseWolverine(opts =>
{
    opts.Policies.OnException<SqlException>()
        .RetryWithCooldown(50.Milliseconds(), 100.Milliseconds(), 250.Milliseconds());
    
    opts.Policies.OnException<TimeoutException>()
        .RetryTimes(3);
    
    opts.Policies.OnException<InvalidOperationException>()
        .Requeue();
    
    opts.Policies.OnAnyException()
        .MoveToErrorQueue();
});

サーキットブレーカー

opts.ListenToRabbitQueue("incoming")
    .CircuitBreaker(cb =>
    {
        cb.PauseTime = 1.Minutes();
        cb.FailurePercentageThreshold = 50;
        cb.MinimumThreshold = 10;
    });

スケジュール済みメッセージ

遅延メッセージ

public static IEnumerable<object> Handle(OrderCreated command)
{
    yield return new ProcessPayment(command.OrderId);
    yield return new ShipOrder(command.OrderId)
        .DelayedFor(30.Minutes());
}

特定の時刻にスケジュールする

yield return new GenerateReport()
    .ScheduledAt(DateTime.Today.AddDays(1));

OutgoingMessages を使う

var messages = new OutgoingMessages();
messages.Delay(new Reminder(orderId), TimeSpan.FromHours(24));
messages.Schedule(new MonthlyReport(), DateTime.Today.AddMonths(1));

リクエスト／レスポンスパターン

レスポンスを要求して送信する

public async Task<OrderStatus> GetOrderStatus(IMessageBus bus, Guid orderId)
{
    return await bus.InvokeAsync<OrderStatus>(new GetOrder(orderId));
}

レスポンスを返すハンドラー

public record GetOrder(Guid OrderId);
public record OrderStatus(Guid OrderId, string Status);

public static class GetOrderHandler
{
    public static OrderStatus Handle(GetOrder query, IQuerySession session)
    {
        var order = session.Load<Order>(query.OrderId);
        return new OrderStatus(query.OrderId, order?.Status ?? "NotFound");
    }
}

ミドルウェア

カスタムミドルウェア

public class LoggingMiddleware
{
    public void Before(Envelope envelope, ILogger logger)
    {
        logger.LogInformation("Processing {MessageType}", envelope.MessageType);
    }

    public void After(Envelope envelope, ILogger logger)
    {
        logger.LogInformation("Completed {MessageType}", envelope.MessageType);
    }
}

builder.Host.UseWolverine(opts =>
{
    opts.Handlers.AddMiddleware<LoggingMiddleware>();
});

トランザクションミドルウェア

builder.Host.UseWolverine(opts =>
{
    opts.Policies.AutoApplyTransactions();
});

コード生成

型を事前生成する

dotnet run -- codegen write

生成されたコードは ./Internal/Generated/WolverineHandlers/ に出力されます

AOT／トリミング向けの設定

builder.Host.UseWolverine(opts =>
{
    opts.CodeGeneration.TypeLoadMode = TypeLoadMode.Auto;
});

マルチテナント

連結テナンシー

builder.Host.UseWolverine(opts =>
{
    opts.Policies.ConjoinedTenancy(x =>
    {
        x.TenantIdStyle = TenantIdStyle.PerRequest;
    });
});

特定のテナントに発行する

await bus.PublishAsync(new OrderCreated(orderId), 
    new DeliveryOptions { TenantId = "tenant-1" });

Marten の副作用

public static IMartenOp Handle(CreateTodo command)
{
    var todo = new Todo { Name = command.Name };
    return MartenOps.Store(todo);
}

public static IMartenOp Handle(DeleteTodo command)
{
    return MartenOps.Delete<Todo>(command.Id);
}

補助ストア（モジュラーモノリス）

public interface IPlayerStore : IDocumentStore;

builder.Host.UseWolverine(opts =>
{
    opts.Services.AddMartenStore<IPlayerStore>(m =>
    {
        m.Connection(connectionString);
        m.DatabaseSchemaName = "players";
    })
    .IntegrateWithWolverine();
});

[MartenStore(typeof(IPlayerStore))]
public static class PlayerMessageHandler
{
    public static IMartenOp Handle(PlayerMessage message)
    {
        return MartenOps.Store(new Player { Id = message.Id });
    }
}

コマンドラインツール

使用可能なコマンド

dotnet run -- help           # List all commands
dotnet run -- describe       # Application description
dotnet run -- resources      # Resource management
dotnet run -- storage        # Message storage admin
dotnet run -- codegen write  # Pre-generate code

ベストプラクティス

1. 純粋関数を優先する - ビジネスロジックはモックなしでテストできるようにします
2. 連鎖するメッセージを使用する - IMessageBus を注入する代わりに、メッセージを返します
3. コールスタックを短く保つ - 深いサービス階層を避けます
4. コードを事前生成する - 本番環境でのコールドスタートを最適化します
5. 複合ハンドラーを使用する - 読み込み／検証／処理のロジックを分離します
6. エラー処理を構成する - Wolverine にリトライとエラー処理を任せます
7. トランザクショナルアウトボックスを使用する - メッセージの配信を保証します
8. 適切な場合はバッチ処理する - 大量のメッセージのスループットを向上させます

避けるべきアンチパターン

1. コールスタックの深い階層で IMessageBus を注入する - ワークフローの把握が難しくなります
2. コンストラクターインジェクションを過度に使用する - メソッドインジェクションを優先します
3. トランザクショナルアウトボックスを無視する - 障害発生時にメッセージが失われる可能性があります
4. コードを事前生成しない - 本番環境でのコールドスタートが遅くなります
5. 1 つのハンドラーに多くの関心事を混在させる - ハンドラーの役割を明確に保ちます
6. エラー処理を構成しない - メッセージが予期せずエラーキューに入ります