Katalyst.App.Events 1.2.1

App.Events

Standalone CloudEvents v1.0 pub/sub helpers for .NET: build JSON CloudEvents, publish to Azure Service Bus topics, Amazon EventBridge (PutEvents), or NATS JetStream, consume Service Bus subscriptions or NATS JetStream pull consumers in-process, optional SQLite + EF Core transactional outbox, and best-effort processed-event deduplication.

This project is not referenced by the rest of the WebTemplate; add a ProjectReference when you adopt it.

Packages

• CloudNative.CloudEvents + CloudNative.CloudEvents.SystemTextJson for the envelope and JSON event format.
• Azure.Messaging.ServiceBus for topic publish and subscription processing.
• AWSSDK.EventBridge for EventBridge PutEvents publishing (credentials via the standard AWS SDK chain).
• NATS.Net for JetStream publish and pull consume (competing consumers via a shared durable consumer name).
• EF Core + SQLite for outbox and durable idempotency when UseOutbox is enabled.

Registration

using App.Events.DependencyInjection;

builder.Services.AddAppEvents(builder.Configuration, events =>
{
    events.UseAzureServiceBus(builder.Configuration.GetSection("AppEvents:ServiceBus"));
    events.UseOutbox(builder.Configuration.GetSection("AppEvents:Outbox")); // optional
    events.AddHandler<OrderCreatedV1Handler>();
});

AddAppEvents binds AppEvents from configuration. If you enable the Service Bus consumer or the NATS JetStream consumer, you must register at least one handler. If you call UseOutbox, you must register a transport (UseAzureServiceBus, UseAmazonEventBridge, UseNatsJetStream, or your own IEventTransport) so the outbox dispatcher can publish.

Use only one built-in transport per host; the last UseAzureServiceBus / UseAmazonEventBridge / UseNatsJetStream call wins for IEventTransport.

Example appsettings.json fragments

{
  "AppEvents": {
    "DefaultSource": "https://orders.example.com/service",
    "ServiceBus": {
      "ConnectionString": "...",
      "TopicName": "integration-events",
      "SubscriptionName": "my-service",
      "EnableConsumer": true
    },
    "Outbox": {
      "ConnectionString": "Data Source=app-events-outbox.db"
    },
    "EventBridge": {
      "EventBusName": "default",
      "Region": "us-east-1",
      "FallbackSource": "app.orders"
    },
    "Nats": {
      "Url": "nats://localhost:4222",
      "StreamName": "INTEGRATION",
      "StreamSubjects": [ "integration.>" ],
      "PublishSubject": "integration.events",
      "ConsumerDurableName": "integration-worker",
      "QueueGroup": "api",
      "EnableConsumer": true,
      "MaxAckPending": 1000,
      "AckWait": "00:00:30"
    }
  }
}

See also appsettings.Nats.sample.json for the same structure in one place.

Amazon EventBridge (publish only)

Register the transport and configure IAM credentials (environment variables, shared credentials file, instance profile, etc.) as usual for the AWS SDK for .NET.

builder.Services.AddAppEvents(builder.Configuration, events =>
{
    events.UseAmazonEventBridge(builder.Configuration.GetSection("AppEvents:EventBridge"));
    events.UseOutbox(); // optional — outbox delivers via IEventTransport (EventBridge)
});

Each CloudEvent is sent as one PutEvents entry:

EventBusName is omitted when set to default or empty so the account default bus is used; set a custom bus name or ARN otherwise.

Consuming: EventBridge does not provide a pull API like Service Bus. Typical patterns: a rule targets SQS, Lambda, or an API destination; the target receives detail as the same structured CloudEvent JSON. Decode with CloudEventJson.DecodeStructured (UTF-8 bytes of detail) in your Lambda or worker. This library does not register an EventBridge consumer.

NATS JetStream

Register the transport and bind AppEvents:Nats. The consumer uses a pull JetStream consumer with a durable name shared by every instance of your service so messages are load-balanced (queue-style) without duplicate delivery to each instance. Optional QueueGroup is appended to the durable name to isolate multiple consumer groups on the same stream.

builder.Services.AddAppEvents(builder.Configuration, events =>
{
    events.UseNatsJetStream(builder.Configuration.GetSection("AppEvents:Nats"));
    events.AddHandler<EmployeeCreatedV1Handler>();
});

Publishing requires PublishSubject to match one of StreamSubjects. The hosted consumer creates or updates the stream and consumer on startup, decodes structured CloudEvents JSON (with Content-Type from message headers when present), dispatches handlers inside a DI scope (IServiceScopeFactory), acks after successful processing, negative-acks on handler failure, and ack+terminate when the payload is not valid CloudEvents JSON (to avoid poison retries).

Database migrations (outbox)

Apply from the repository root:

dotnet ef database update --project WebTemplate/App.Events/App.Events.csproj --context EventsDbContext

Or call EventsDbContext.Database.Migrate() once at startup in the adopting application.

Publisher example

public sealed class PlaceOrderWorkflow(IEventPublisher events)
{
    public async Task OnOrderPlacedAsync(Guid orderId, string customerId, CancellationToken ct)
    {
        var payload = new OrderCreatedV1(orderId, customerId, DateTimeOffset.UtcNow);

        await events.PublishAsync(
            type: "com.example.order.created.v1",
            source: "https://orders.example.com/service",
            subject: orderId.ToString(),
            data: payload,
            cancellationToken: ct);

        // Or: await events.PublishViaOutboxAsync(...);  // requires UseOutbox
    }
}

public sealed record OrderCreatedV1(Guid OrderId, string CustomerId, DateTimeOffset OccurredAt);

You can omit source in the call when AppEvents:DefaultSource is set.

Consumer example

Apply [CloudEventType("...")] with the same type string the publisher uses. Implement ICloudEventHandler<TData> with a public HandleAsync(CloudEvent, TData, CancellationToken).

using App.Events.Contracts;
using CloudNative.CloudEvents;

[CloudEventType("com.example.order.created.v1")]
public sealed class OrderCreatedV1Handler : ICloudEventHandler<OrderCreatedV1>
{
    public Task HandleAsync(CloudEvent envelope, OrderCreatedV1 data, CancellationToken cancellationToken)
    {
        // Idempotent side effects (at-least-once delivery).
        return Task.CompletedTask;
    }
}

Reliability notes

• Delivery is at-least-once. Handlers should be idempotent.
• Without UseOutbox, IProcessedEventStore defaults to an in-memory implementation (not shared across instances or restarts). With UseOutbox, processed keys are stored in SQLite alongside the outbox.
• Unknown CloudEvents types are logged and the Service Bus message is completed to avoid endless retries; adjust this policy in your fork if you prefer dead-lettering.
• NATS JetStream: invalid structured CloudEvents payloads are ack+terminated; handler failures trigger a negative ack for redelivery per consumer AckWait / MaxDeliver.
• For atomic “business commit + outbox row” in the same database as your app, share one database and one transaction across your DbContext and EventsDbContext (advanced; not wired by default).

Layout