plan-forge messaging.instructions.md

.NET Messaging & Pub/Sub Patterns

Messaging Strategy

Dapr Pub/Sub (Recommended for Service Mesh)

// Publishing events
public class OrderService(DaprClient daprClient)
{
    public async Task PlaceOrderAsync(Order order, CancellationToken ct = default)
    {
        await _repository.SaveAsync(order, ct);
        await daprClient.PublishEventAsync(
            "pubsub",            // component name
            "order-placed",      // topic
            new OrderPlacedEvent(order.Id, order.TenantId, DateTimeOffset.UtcNow),
            ct);
    }
}

// Subscribing to events
[ApiController]
public class OrderEventsController : ControllerBase
{
    [Topic("pubsub", "order-placed")]
    [HttpPost("/events/order-placed")]
    public async Task<IActionResult> HandleOrderPlaced(
        OrderPlacedEvent evt, [FromServices] IOrderProcessor processor, CancellationToken ct)
    {
        await processor.ProcessAsync(evt, ct);
        return Ok();
    }
}

NATS JetStream (Direct)

// Publishing
await js.PublishAsync("orders.placed", new OrderPlacedEvent { ... }, cancellationToken: ct);

// Subscribing (durable consumer)
var consumer = await js.CreateOrUpdateConsumerAsync("orders", new ConsumerConfig
{
    DurableName = "order-processor",
    FilterSubject = "orders.placed",
    AckPolicy = ConsumerConfigAckPolicy.Explicit,
});

MassTransit (RabbitMQ / Azure Service Bus)

// Program.cs
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(builder.Configuration.GetConnectionString("RabbitMQ"));
        cfg.ConfigureEndpoints(context);
    });
});

// Consumer
public class OrderPlacedConsumer : IConsumer<OrderPlacedEvent>
{
    public async Task Consume(ConsumeContext<OrderPlacedEvent> context)
    {
        var evt = context.Message;
        // Process event
    }
}

Event Schema (CloudEvents)

// Always use typed events — never raw strings
public record OrderPlacedEvent(
    Guid OrderId,
    string TenantId,
    DateTimeOffset OccurredAt);

// Include tenant_id in ALL events for multi-tenant routing