feat: add change data capture pattern

README.md

@@ -473,15 +473,15 @@ var cachedRemoteProxy = Proxy<int, string>.Create(id => remoteProxy.Execute(id))
 ---
 
 ## Patterns Table
-PatternKit currently tracks 117 production-readiness patterns. Each catalog pattern is represented in tests, documentation, real-world examples, IoC integration, and the BenchmarkDotNet coverage matrix.
+PatternKit currently tracks 118 production-readiness patterns. Each catalog pattern is represented in tests, documentation, real-world examples, IoC integration, and the BenchmarkDotNet coverage matrix.
 
 | Category | Count | Patterns |
 | --- | ---: | --- |
 | Application Architecture | 27 | Activity Tracker, Aggregate Root, Anti-Corruption Layer, Audit Log, Bounded Context, Context Map, CQRS, Data Mapper, Domain Event, Domain Service, Event Sourcing, Eventual Consistency Monitor, Feature Toggle, Identity Map, Lazy Load, Manual Task Gate, Materialized View, Repository, Service Layer, Snapshot / Checkpoint Management, Specification, Table Data Gateway, Timeout Manager, Transaction Script, Unit of Work, Value Object, Workflow Orchestration |
 | Behavioral | 12 | Chain of Responsibility, Command, Interpreter, Iterator, Mediator, Memento, Null Object, Observer, State, Strategy, Template Method, Visitor |
 | Cloud Architecture | 20 | Ambassador, Backends for Frontends, Bulkhead, Cache-Aside, Cache Stampede Protection, Circuit Breaker, External Configuration Store, Gateway Aggregation, Gateway Routing, Health Endpoint Monitoring, Leader Election, Priority Queue, Queue-Based Load Leveling, Rate Limiting, Read-Through Cache, Retry, Scheduler Agent Supervisor, Sidecar, Strangler Fig, Write-Through Cache |
 | Creational | 6 | Abstract Factory, Builder, Factory Method, Object Pool, Prototype, Singleton |
-| Enterprise Integration | 41 | Aggregator, Canonical Data Model, Channel Adapter, Channel Purger, Claim Check, Competing Consumers, Content Enricher, Content-Based Router, Control Bus, Correlation Identifier, Dead Letter Channel, Durable Subscriber, Dynamic Router, Event Notification, Event-Carried State Transfer, Event-Driven Consumer, Guaranteed Delivery, Invalid Message Channel, Mailbox, Message Bus, Message Channel, Message Envelope, Message Expiration, Message Filter, Message History, Message Store, Message Translator, Messaging Bridge, Messaging Gateway, Pipes and Filters, Polling Consumer, Publish-Subscribe, Recipient List, Request-Reply, Resequencer, Routing Slip, Saga / Process Manager, Scatter-Gather, Service Activator, Splitter, Wire Tap |
+| Enterprise Integration | 42 | Aggregator, Canonical Data Model, Change Data Capture, Channel Adapter, Channel Purger, Claim Check, Competing Consumers, Content Enricher, Content-Based Router, Control Bus, Correlation Identifier, Dead Letter Channel, Durable Subscriber, Dynamic Router, Event Notification, Event-Carried State Transfer, Event-Driven Consumer, Guaranteed Delivery, Invalid Message Channel, Mailbox, Message Bus, Message Channel, Message Envelope, Message Expiration, Message Filter, Message History, Message Store, Message Translator, Messaging Bridge, Messaging Gateway, Pipes and Filters, Polling Consumer, Publish-Subscribe, Recipient List, Request-Reply, Resequencer, Routing Slip, Saga / Process Manager, Scatter-Gather, Service Activator, Splitter, Wire Tap |
 | Messaging Reliability | 4 | Backpressure, Idempotent Receiver, Inbox, Outbox |
 | Structural | 7 | Adapter, Bridge, Composite, Decorator, Facade, Flyweight, Proxy |
 
@@ -535,6 +535,8 @@ BenchmarkDotNet guidance is documented in [docs/guides/benchmarks.md](docs/guide
 | Cache-Aside | Execution | 216.50 ns | 1,048 B | 208.60 ns | 1,048 B | Same allocation; generated was slightly faster for the miss-then-hit workflow. |
 | Read-Through / Write-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
 | Read-Through / Write-Through Cache | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
+| Change Data Capture | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
+| Change Data Capture | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
 | Read-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |
 | Read-Through Cache | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |
 | Write-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |

benchmarks/PatternKit.Benchmarks/Messaging/ChangeDataCaptureBenchmarks.cs

@@ -0,0 +1,43 @@
+using BenchmarkDotNet.Attributes;
+using PatternKit.Examples.ChangeDataCaptureDemo;
+using PatternKit.Messaging.ChangeDataCapture;
+
+namespace PatternKit.Benchmarks.Messaging;
+
+[BenchmarkCategory("EnterpriseIntegration", "ChangeDataCapture")]
+public class ChangeDataCaptureBenchmarks
+{
+    private static readonly ProductMutation Mutation = new("sku-1", "Desk", 125m, 1);
+
+    [Benchmark(Baseline = true, Description = "Fluent: create CDC pipeline")]
+    [BenchmarkCategory("Fluent", "Construction")]
+    public ChangeDataCapturePipeline<ProductMutation, ProductChanged> Fluent_Create()
+        => ProductCatalogChangeDataCapturePolicies.CreateFluent(
+            new InMemoryChangeDataCaptureStore<ProductMutation, ProductChanged>(),
+            new InMemoryProductChangePublisher());
+
+    [Benchmark(Description = "Generated: create CDC pipeline")]
+    [BenchmarkCategory("Generated", "Construction")]
+    public ChangeDataCapturePipeline<ProductMutation, ProductChanged> Generated_Create()
+        => GeneratedProductCatalogChangeDataCapture.CreateGenerated(
+            static (_, _) => default,
+            new InMemoryChangeDataCaptureStore<ProductMutation, ProductChanged>());
+
+    [Benchmark(Description = "Fluent: capture and publish mutation")]
+    [BenchmarkCategory("Fluent", "Execution")]
+    public async ValueTask<ChangeDataCapturePublishSummary> Fluent_Capture_And_Publish()
+    {
+        var pipeline = Fluent_Create();
+        await pipeline.CaptureAsync(Mutation).ConfigureAwait(false);
+        return await pipeline.PublishPendingAsync().ConfigureAwait(false);
+    }
+
+    [Benchmark(Description = "Generated: capture and publish mutation")]
+    [BenchmarkCategory("Generated", "Execution")]
+    public async ValueTask<ChangeDataCapturePublishSummary> Generated_Capture_And_Publish()
+    {
+        var pipeline = Generated_Create();
+        await pipeline.CaptureAsync(Mutation).ConfigureAwait(false);
+        return await pipeline.PublishPendingAsync().ConfigureAwait(false);
+    }
+}

docs/examples/index.md

@@ -60,6 +60,9 @@ Welcome! This section collects small, focused demos that show **how to compose b
 * **Customer Profile Lazy Load**
   A Generic Host importable deferred profile lookup with fluent and source-generated routes, `IServiceCollection` registration, TTL caching, and invalidation. See [Customer Profile Lazy Load](customer-profile-lazy-load.md).
 
+* **Product Catalog Change Data Capture**
+  A Generic Host importable mutation capture workflow with fluent and source-generated routes, ordered pending records, and post-commit publication. See [Product Catalog Change Data Capture](product-catalog-change-data-capture.md).
+
 * **Minimal Web Request Router**
   A tiny "API gateway" that separates **first-match middleware** (side effects/logging/auth) from **first-match routes** and **content negotiation**. A crisp example of Strategy patterns in an HTTP-ish setting.
 

docs/examples/product-catalog-change-data-capture.md

@@ -0,0 +1,18 @@
+# Product Catalog Change Data Capture
+
+The product catalog CDC example shows how a catalog service can capture product mutations as ordered integration events and publish them after the mutation is stored.
+
+Import it into a host:
+
+```csharp
+services.AddProductCatalogChangeDataCaptureDemo();
+```
+
+The demo registers:
+
+- `IChangeDataCaptureStore<ProductMutation,ProductChanged>` for pending capture records.
+- `IProductChangePublisher` for event handoff.
+- `ChangeDataCapturePipeline<ProductMutation,ProductChanged>` as the application-owned CDC pipeline.
+- `ProductCatalogChangeDataCaptureService` as the host-facing workflow.
+
+The source-generated route emits the same pipeline factory while letting the application supply its real publisher and durable store.

docs/examples/toc.yml

@@ -211,6 +211,9 @@
 - name: Customer Profile Lazy Load
   href: customer-profile-lazy-load.md
 
+- name: Product Catalog Change Data Capture
+  href: product-catalog-change-data-capture.md
+
 - name: Order Transaction Script Pattern
   href: order-transaction-script-pattern.md
 

docs/generators/change-data-capture.md

@@ -0,0 +1,33 @@
+# Change Data Capture Generator
+
+`[GenerateChangeDataCapture]` emits a typed factory for `ChangeDataCapturePipeline<TMutation,TEvent>` from a mapper method.
+
+```csharp
+[GenerateChangeDataCapture(
+    typeof(ProductMutation),
+    typeof(ProductChanged),
+    FactoryMethodName = "CreatePipeline",
+    MapperMethodName = "Map",
+    PipelineName = "product-catalog-cdc")]
+public static partial class ProductCatalogCdc
+{
+    public static ProductChanged Map(ProductMutation mutation, long sequence)
+        => new(sequence, mutation.Sku, mutation.Name);
+}
+```
+
+## Diagnostics
+
+| ID | Severity | Message |
+| --- | --- | --- |
+| `PKCDC001` | Error | The host type must be partial. |
+| `PKCDC002` | Error | Factory and mapper method names must be valid C# identifiers. |
+| `PKCDC003` | Error | Mutation and event types are required. |
+
+Register generated pipelines by supplying the application publisher and durable store:
+
+```csharp
+services.AddSingleton(sp => ProductCatalogCdc.CreatePipeline(
+    (changed, ct) => sp.GetRequiredService<IProductChangePublisher>().PublishAsync(changed, ct),
+    sp.GetRequiredService<IChangeDataCaptureStore<ProductMutation, ProductChanged>>()));
+```

docs/generators/index.md

@@ -125,6 +125,7 @@ PatternKit includes a Roslyn incremental generator package (`PatternKit.Generato
 | [**Saga**](messaging.md#generated-saga) | Typed process-manager transition factories | `[GenerateSaga]` |
 | [**Mailbox**](messaging.md#generated-mailbox) | Serialized in-process inbox factories | `[GenerateMailbox]` |
 | [**Backpressure**](backpressure.md) | Admission-control policy factories for saturated work boundaries | `[GenerateBackpressurePolicy]` |
+| [**Change Data Capture**](change-data-capture.md) | Ordered mutation capture pipeline factories | `[GenerateChangeDataCapture]` |
 | [**Reliability Pipeline**](messaging.md#generated-reliability-pipeline) | Idempotent receiver, inbox, and outbox factories | `[GenerateReliabilityPipeline]` |
 | [**Backplane Topology**](messaging.md#generated-backplane-topology) | Request/reply routes and publish/subscribe endpoint topology | `[GenerateBackplaneTopology]` |
 
@@ -290,6 +291,10 @@ public static partial class OrderMailbox { }
 [GenerateBackpressurePolicy(typeof(CheckoutAdmission), Capacity = 8, Mode = "Wait", WaitTimeoutMilliseconds = 50)]
 public static partial class CheckoutBackpressurePolicy { }
 
+// Change data capture - generated mutation capture pipeline
+[GenerateChangeDataCapture(typeof(ProductMutation), typeof(ProductChanged), MapperMethodName = "Map")]
+public static partial class ProductCatalogCdc { }
+
 // Reliability pipeline - generated idempotent receiver, inbox, and outbox factories
 [GenerateReliabilityPipeline(typeof(AcceptOrder), typeof(string), typeof(OrderAccepted))]
 public static partial class OrderReliability { }

docs/generators/toc.yml

@@ -31,6 +31,9 @@
 - name: Backpressure
   href: backpressure.md
 
+- name: Change Data Capture
+  href: change-data-capture.md
+
 - name: Cache-Aside
   href: cache-aside.md
 

docs/guides/benchmark-results.md

@@ -59,6 +59,8 @@ The latest measured timings below were captured on Windows 11, Intel Core i9-149
 | Cache-Aside | Execution | 216.50 ns | 1,048 B | 208.60 ns | 1,048 B | Same allocation; generated was slightly faster for the miss-then-hit workflow. |
 | Read-Through / Write-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
 | Read-Through / Write-Through Cache | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
+| Change Data Capture | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
+| Change Data Capture | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix; publish measured values after the next benchmark refresh. |
 | Read-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |
 | Read-Through Cache | Execution | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |
 | Write-Through Cache | Construction | Pending | Pending | Pending | Pending | Covered by the BenchmarkDotNet matrix through the read/write-through cache route. |
@@ -256,19 +258,19 @@ The latest measured timings below were captured on Windows 11, Intel Core i9-149
 
 ## Coverage Matrix Summary
 
-The coverage matrix currently publishes 117 catalog patterns and 468 pattern route results. Each pattern has four BenchmarkDotNet routes: fluent construction, fluent execution, source-generated construction, and source-generated execution. The reusable hosting integration matrix publishes 12 reusable hosting integration route results for package-level `IServiceCollection` registrations.
+The coverage matrix currently publishes 118 catalog patterns and 472 pattern route results. Each pattern has four BenchmarkDotNet routes: fluent construction, fluent execution, source-generated construction, and source-generated execution. The reusable hosting integration matrix publishes 12 reusable hosting integration route results for package-level `IServiceCollection` registrations.
 
 | Category | Patterns | Published route results |
 | --- | ---: | ---: |
 | Application Architecture | 27 | 108 |
 | Behavioral | 12 | 48 |
 | Cloud Architecture | 20 | 80 |
 | Creational | 6 | 24 |
-| Enterprise Integration | 41 | 164 |
+| Enterprise Integration | 42 | 168 |
 | Messaging Reliability | 4 | 16 |
 | Structural | 7 | 28 |
 
-The generator matrix currently publishes 112 generator source route results.
+The generator matrix currently publishes 113 generator source route results.
 
 ## Hosting Integration Matrix Results
 
@@ -358,6 +360,7 @@ The generator matrix currently publishes 112 generator source route results.
 | Creational | Singleton | Covered | Covered | Covered | Covered |
 | Enterprise Integration | Aggregator | Covered | Covered | Covered | Covered |
 | Enterprise Integration | Canonical Data Model | Covered | Covered | Covered | Covered |
+| Enterprise Integration | Change Data Capture | Covered | Covered | Covered | Covered |
 | Enterprise Integration | Channel Adapter | Covered | Covered | Covered | Covered |
 | Enterprise Integration | Channel Purger | Covered | Covered | Covered | Covered |
 | Enterprise Integration | Invalid Message Channel | Covered | Covered | Covered | Covered |
@@ -466,6 +469,7 @@ The generator matrix currently publishes 112 generator source route results.
 | NullObjectGenerator | `src/PatternKit.Generators/NullObject/NullObjectGenerator.cs` | Covered |
 | BackplaneTopologyGenerator | `src/PatternKit.Generators/Messaging/BackplaneTopologyGenerator.cs` | Covered |
 | ChannelAdapterGenerator | `src/PatternKit.Generators/Messaging/ChannelAdapterGenerator.cs` | Covered |
+| ChangeDataCaptureGenerator | `src/PatternKit.Generators/ChangeDataCapture/ChangeDataCaptureGenerator.cs` | Covered |
 | ChannelPurgerGenerator | `src/PatternKit.Generators/Messaging/ChannelPurgerGenerator.cs` | Covered |
 | InvalidMessageChannelGenerator | `src/PatternKit.Generators/Messaging/InvalidMessageChannelGenerator.cs` | Covered |
 | ClaimCheckGenerator | `src/PatternKit.Generators/Messaging/ClaimCheckGenerator.cs` | Covered |

docs/index.md

@@ -66,15 +66,15 @@ if (parser.Execute("123", out var value))
 
 ## 📚 Available Patterns
 
-PatternKit covers 117 production-readiness patterns with fluent APIs, source-generated routes where applicable, IoC integration examples, TinyBDD coverage, and BenchmarkDotNet coverage-matrix validation:
+PatternKit covers 118 production-readiness patterns with fluent APIs, source-generated routes where applicable, IoC integration examples, TinyBDD coverage, and BenchmarkDotNet coverage-matrix validation:
 
 | Category | Count | Patterns |
 | --- | ---: | --- |
 | Application Architecture | 27 | Activity Tracker, Aggregate Root, Anti-Corruption Layer, Audit Log, Bounded Context, Context Map, CQRS, Data Mapper, Domain Event, Domain Service, Event Sourcing, Eventual Consistency Monitor, Feature Toggle, Identity Map, Lazy Load, Manual Task Gate, Materialized View, Repository, Service Layer, Snapshot / Checkpoint Management, Specification, Table Data Gateway, Timeout Manager, Transaction Script, Unit of Work, Value Object, Workflow Orchestration |
 | Behavioral | 12 | Chain of Responsibility, Command, Interpreter, Iterator, Mediator, Memento, Null Object, Observer, State, Strategy, Template Method, Visitor |
 | Cloud Architecture | 20 | Ambassador, Backends for Frontends, Bulkhead, Cache-Aside, Cache Stampede Protection, Circuit Breaker, External Configuration Store, Gateway Aggregation, Gateway Routing, Health Endpoint Monitoring, Leader Election, Priority Queue, Queue-Based Load Leveling, Rate Limiting, Read-Through Cache, Retry, Scheduler Agent Supervisor, Sidecar, Strangler Fig, Write-Through Cache |
 | Creational | 6 | Abstract Factory, Builder, Factory Method, Object Pool, Prototype, Singleton |
-| Enterprise Integration | 41 | Aggregator, Canonical Data Model, Channel Adapter, Channel Purger, Claim Check, Competing Consumers, Content Enricher, Content-Based Router, Control Bus, Correlation Identifier, Dead Letter Channel, Durable Subscriber, Dynamic Router, Event Notification, Event-Carried State Transfer, Event-Driven Consumer, Guaranteed Delivery, Invalid Message Channel, Mailbox, Message Bus, Message Channel, Message Envelope, Message Expiration, Message Filter, Message History, Message Store, Message Translator, Messaging Bridge, Messaging Gateway, Pipes and Filters, Polling Consumer, Publish-Subscribe, Recipient List, Request-Reply, Resequencer, Routing Slip, Saga / Process Manager, Scatter-Gather, Service Activator, Splitter, Wire Tap |
+| Enterprise Integration | 42 | Aggregator, Canonical Data Model, Change Data Capture, Channel Adapter, Channel Purger, Claim Check, Competing Consumers, Content Enricher, Content-Based Router, Control Bus, Correlation Identifier, Dead Letter Channel, Durable Subscriber, Dynamic Router, Event Notification, Event-Carried State Transfer, Event-Driven Consumer, Guaranteed Delivery, Invalid Message Channel, Mailbox, Message Bus, Message Channel, Message Envelope, Message Expiration, Message Filter, Message History, Message Store, Message Translator, Messaging Bridge, Messaging Gateway, Pipes and Filters, Polling Consumer, Publish-Subscribe, Recipient List, Request-Reply, Resequencer, Routing Slip, Saga / Process Manager, Scatter-Gather, Service Activator, Splitter, Wire Tap |
 | Messaging Reliability | 4 | Backpressure, Idempotent Receiver, Inbox, Outbox |
 | Structural | 7 | Adapter, Bridge, Composite, Decorator, Facade, Flyweight, Proxy |