Kanject.Core.ChangeFeed

Turn a DynamoDB row change into a published event, from the table's own stream. You declare what publishes, on which topic, as what payload; the drain reads the stream in order, decodes and diffs the images, stamps one envelope per change, and hands it to your Events publisher. A transactional outbox carries the events that aren't a row's state — so nothing is ever written twice.

Why a stream, not commit-then-publish

Commit, then publish is a dual write: the commit can succeed and the publish fail, and the event is gone. The table's stream records every committed write, in order per item, so an event read from it cannot be lost or reordered behind a later change. ChangeFeed is that pattern as a library — the drain, the diff, the routing and the redrive that teams otherwise copy from service to service.

Install

dotnet add package Kanject.Core.ChangeFeed.Provider.DynamoDbdotnet add package Kanject.Core.ChangeFeed.Provider.DynamoDb.Annotations   # the generatordotnet add package Kanject.Core.ChangeFeed.Provider.Events                 # the sink

Add Kanject.Core.ChangeFeed.Provider.DynamoDb.Outbox for the outbox, and Kanject.Core.ChangeFeed.TestKit (or .Provider.InMemory) for tests. Targets .NET 8, 9 and 10; reflection-free and AOT-clean.

Declare a feed

One small partial class per published entity, in the business assembly. Declare the snapshot type in the same source-generated JSON context as your events.

using Kanject.Core.ChangeFeed.Provider.DynamoDb.Annotations.Attributes;​[ChangeFeed<OrderEntity>(JsonContext = typeof(OrderEventsJsonContext))][OnInsert("order_placed")][OnUpdate("order_updated")]   // only when the snapshot changespublic sealed partial class OrderFeed{    public OrderSnapshot Snapshot(OrderEntity order) =>        new(order.Id, order.Status, order.Total);}​// The payload consumers see. The SDK stamps EventId and Version after mapping.public sealed record OrderSnapshot(string OrderId, string Status, decimal Total){    public string EventId { get; init; } = "";    public long Version { get; init; }}

• [OnUpdate("topic")] with no names publishes only when the snapshot changes — an update to an attribute the snapshot doesn't carry publishes nothing. [OnUpdate("topic", "Status")] publishes when a named property changed.
• Removals publish only with [OnRemove], and TTL expiries only with [OnExpire] — the drain tells the two apart. Both carry a tombstone version: the last version plus one.
• ReadCurrent = true publishes the row as it is now, re-read with a consistent GetItem.

Decide the topic in code

A conditional feed declares a Topic method over the generated change view instead of the attributes. Return a constant topic, or null to publish nothing:

[ChangeFeed<OrderEntity>(JsonContext = typeof(OrderEventsJsonContext))]public sealed partial class OrderStatusFeed{    public static OrderSnapshot Snapshot(OrderEntity order) =>        new(order.Id, order.Status, order.Total);​    // OrderStatusFeedChange is generated: New, Old, IsInsert, IsUpdate,    // IsRemove, IsExpire, SnapshotChanged and Changed.{Property}.    public static string? Topic(OrderStatusFeedChange change) => change switch    {        { IsInsert: true } => "order_placed",        { IsUpdate: true, Changed.Status: true } => "order_status_changed",        _ => null,   // nothing consumers project: publish nothing    };}

Wire the function

// Registration, generated and named after the feeds' DbContext (OrdersDbContext).services.RegisterDynamoDbEntityMappers();             // generated by the data assemblyservices.AddOrdersChangeFeeds().UseEventPublisherSink();​// The stream-triggered function.[LambdaFunction(ResourceName = "ChangeFeedPublisherFunc", MemorySize = 512, Timeout = 60)]public Task<StreamsEventResponse> PublishChanges(DynamoDBEvent streamEvent, ILambdaContext context) =>    ServiceProvider.DrainChangeFeedAsync(streamEvent, context);

The table's stream must be NEW_AND_OLD_IMAGES — HasStream(StreamViewType.NEW_AND_OLD_IMAGES) on the entity model builder creates the table with it — and the event-source mapping must report batch item failures. The host supplies the IEventPublisher, such as BaaS ServerlessEventHub's publisher over SNS; the sink refuses the in-process Events default (KANCFD113) unless you pass allowInProcess: true.

You don't hand-build the infrastructure. The generator compiles a versioned document into the feeds assembly; claim it in the manifest and kanject aws deploy renders the stream's event-source mapping, the S3 failure bucket, the declared topics and the drain's permissions (Kanject CLI 1.0.76 or later). kanject aws change-feeds plan shows what a deploy would do.

"aws": {  "changeFeeds": {    "Orders": { "entry": "tasks" }  }}

The outbox

Some events aren't a row's state — a transfer completed, an email requested. Outbox() on the unit of work is an IEventPublisher, so the generated Publish{Event}Async of any [EventTopic] event works unchanged. It queues the event as a row in the open transaction, and the drain publishes it once the transaction commits:

uow.BeginTransaction();await orders.InsertAsync(order);await uow.Outbox().PublishOrderPlacedAsync(new OrderPlaced(order.Id, order.Lines.Count));await uow.CommitAsync();   // the event row commits with the order, or neither does

// In the drain function: publish the DbContext's outbox rows too.services.AddChangeFeedOutbox<OrdersDbContext>()        .UseEventPublisherSink();

• Published if and only if the write commits, under the id it was written with — consumers deduplicate by it.
• Publishing with no transaction open throws KANCFD105: the row would be a second write.
• A DynamoDB transaction holds at most 100 items, business writes included. Publish a reference to large data, not the data — a row over 400 KB throws KANCFD106.
• Outbox events are unordered and carry no entity version. A consumer that needs order reads a sequence from the payload.
• A published row is marked and, when the DbContext declares a TTL attribute, expires one day later. An unpublished row never expires.

What the drain guarantees

• In order, stopping at the first failure. It reports that record's sequence number and attempts nothing after it, so no later change of the same row overtakes it.
• The same event id on every retry: evt_ plus the stream record's eventID.
• A soft stop one second before the Lambda deadline, set with ConfigureDrain.
• Two ways back. SweepChangeFeedOutboxAsync<TDbContext>() republishes outbox rows left unpublished — run it on a schedule. RedriveChangeFeedFailureAsync drains a batch from the S3 failure destination once its cause is fixed.
• Never silent. A composition it cannot run fails the first record with a KANCFD code, below.

Test it

The TestKit runs your generated feeds through the production drain — no DynamoDB, no AWS. UpdateAsync(before, after), RemoveAsync and ExpireAsync drive the other changes; FailNextPublish() tests a retry.

var host = ChangeFeedTestHost.Create(services =>{    services.RegisterDynamoDbEntityMappers();    services.AddOrdersChangeFeeds();});​await host.InsertAsync(new OrderEntity { Id = "ord_1", Status = "placed", Total = 42m });​Assert.Equal("order_placed", host.Published[0].Topic);

Runtime codes

KANCFD101

No sink is registered

KANCFD102

The stream is not NEW_AND_OLD_IMAGES

KANCFD103

The mapper could not decode an image

KANCFD104

The sink failed

KANCFD105

An outbox publish with no transaction open

KANCFD106

An outbox row over DynamoDB's 400 KB item limit

KANCFD107

A near miss: keys share the prefix but fail the entity's key shape (a warning)

KANCFD108

An entity the DbContext does not map, or keys the table does not have

KANCFD109

No mapper or entity descriptor is registered

KANCFD110

A record from a table no feed is bound to: the mapping is attached to the wrong stream

KANCFD111

An old image could not be decoded, so the update counts as changed (a warning)

KANCFD112

One DbContext has feeds from two assemblies

KANCFD113

The events sink over the in-process Events publisher

KANCFD114

An outbox on a table whose keys are not strings